first jsonrpc commands
[electrum-server.git] / server.py
1 #!/usr/bin/env python
2 # Copyright(C) 2011 thomasv@gitorious
3
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as
6 # published by the Free Software Foundation, either version 3 of the
7 # License, or (at your option) any later version.
8 #
9 # This program is distributed in the hope that it will be useful, but
10 # WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
12 # Affero General Public License for more details.
13 #
14 # You should have received a copy of the GNU Affero General Public
15 # License along with this program.  If not, see
16 # <http://www.gnu.org/licenses/agpl.html>.
17
18 """
19 Todo:
20    * server should check and return bitcoind status..
21    * improve txpoint sorting
22    * command to check cache
23
24  mempool transactions do not need to be added to the database; it slows it down
25 """
26
27
28 import time, socket, operator, thread, ast, sys,re
29 import psycopg2, binascii
30 import bitcoinrpc
31
32 from Abe.abe import hash_to_address, decode_check_address
33 from Abe.DataStore import DataStore as Datastore_class
34 from Abe import DataStore, readconf, BCDataStream,  deserialize, util, base58
35
36 import ConfigParser
37
38 config = ConfigParser.ConfigParser()
39 # set some defaults, which will be overwritten by the config file
40 config.add_section('server')
41 config.set('server','banner', 'Welcome to Electrum!')
42 config.set('server', 'host', 'localhost')
43 config.set('server', 'port', 50000)
44 config.set('server', 'password', '')
45 config.set('server', 'irc', 'yes')
46 config.set('server', 'cache', 'no') 
47 config.set('server', 'ircname', 'Electrum server')
48 config.add_section('database')
49 config.set('database', 'type', 'psycopg2')
50 config.set('database', 'database', 'abe')
51
52 try:
53     f = open('/etc/electrum.conf','r')
54     config.readfp(f)
55     f.close()
56 except:
57     print "Could not read electrum.conf. I will use the default values."
58
59 password = config.get('server','password')
60
61 stopping = False
62 block_number = -1
63 sessions = {}
64 dblock = thread.allocate_lock()
65 peer_list = {}
66
67 wallets = {} # for ultra-light clients such as bccapi
68
69 class MyStore(Datastore_class):
70
71     def import_tx(self, tx, is_coinbase):
72         tx_id = super(MyStore, self).import_tx(tx, is_coinbase)
73         if config.get('server', 'cache') == 'yes': self.update_tx_cache(tx_id)
74
75     def update_tx_cache(self, txid):
76         inrows = self.get_tx_inputs(txid, False)
77         for row in inrows:
78             _hash = store.binout(row[6])
79             address = hash_to_address(chr(0), _hash)
80             if self.tx_cache.has_key(address):
81                 print "cache: invalidating", address
82                 self.tx_cache.pop(address)
83         outrows = self.get_tx_outputs(txid, False)
84         for row in outrows:
85             _hash = store.binout(row[6])
86             address = hash_to_address(chr(0), _hash)
87             if self.tx_cache.has_key(address):
88                 print "cache: invalidating", address
89                 self.tx_cache.pop(address)
90
91     def safe_sql(self,sql, params=(), lock=True):
92         try:
93             if lock: dblock.acquire()
94             ret = self.selectall(sql,params)
95             if lock: dblock.release()
96             return ret
97         except:
98             print "sql error", sql
99             return []
100
101     def get_tx_outputs(self, tx_id, lock=True):
102         return self.safe_sql("""SELECT
103                 txout.txout_pos,
104                 txout.txout_scriptPubKey,
105                 txout.txout_value,
106                 nexttx.tx_hash,
107                 nexttx.tx_id,
108                 txin.txin_pos,
109                 pubkey.pubkey_hash
110               FROM txout
111               LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
112               LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
113               LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
114              WHERE txout.tx_id = %d 
115              ORDER BY txout.txout_pos
116         """%(tx_id), (), lock)
117
118     def get_tx_inputs(self, tx_id, lock=True):
119         return self.safe_sql(""" SELECT
120                 txin.txin_pos,
121                 txin.txin_scriptSig,
122                 txout.txout_value,
123                 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
124                 prevtx.tx_id,
125                 COALESCE(txout.txout_pos, u.txout_pos),
126                 pubkey.pubkey_hash
127               FROM txin
128               LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
129               LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
130               LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
131               LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
132              WHERE txin.tx_id = %d
133              ORDER BY txin.txin_pos
134              """%(tx_id,), (), lock)
135
136     def get_address_out_rows(self, dbhash):
137         return self.safe_sql(""" SELECT
138                 b.block_nTime,
139                 cc.chain_id,
140                 b.block_height,
141                 1,
142                 b.block_hash,
143                 tx.tx_hash,
144                 tx.tx_id,
145                 txin.txin_pos,
146                 -prevout.txout_value
147               FROM chain_candidate cc
148               JOIN block b ON (b.block_id = cc.block_id)
149               JOIN block_tx ON (block_tx.block_id = b.block_id)
150               JOIN tx ON (tx.tx_id = block_tx.tx_id)
151               JOIN txin ON (txin.tx_id = tx.tx_id)
152               JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
153               JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
154              WHERE pubkey.pubkey_hash = ?
155                AND cc.in_longest = 1""", (dbhash,))
156
157     def get_address_out_rows_memorypool(self, dbhash):
158         return self.safe_sql(""" SELECT
159                 1,
160                 tx.tx_hash,
161                 tx.tx_id,
162                 txin.txin_pos,
163                 -prevout.txout_value
164               FROM tx 
165               JOIN txin ON (txin.tx_id = tx.tx_id)
166               JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
167               JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
168              WHERE pubkey.pubkey_hash = ? """, (dbhash,))
169
170     def get_address_in_rows(self, dbhash):
171         return self.safe_sql(""" SELECT
172                 b.block_nTime,
173                 cc.chain_id,
174                 b.block_height,
175                 0,
176                 b.block_hash,
177                 tx.tx_hash,
178                 tx.tx_id,
179                 txout.txout_pos,
180                 txout.txout_value
181               FROM chain_candidate cc
182               JOIN block b ON (b.block_id = cc.block_id)
183               JOIN block_tx ON (block_tx.block_id = b.block_id)
184               JOIN tx ON (tx.tx_id = block_tx.tx_id)
185               JOIN txout ON (txout.tx_id = tx.tx_id)
186               JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
187              WHERE pubkey.pubkey_hash = ?
188                AND cc.in_longest = 1""", (dbhash,))
189
190     def get_address_in_rows_memorypool(self, dbhash):
191         return self.safe_sql( """ SELECT
192                 0,
193                 tx.tx_hash,
194                 tx.tx_id,
195                 txout.txout_pos,
196                 txout.txout_value
197               FROM tx
198               JOIN txout ON (txout.tx_id = tx.tx_id)
199               JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
200              WHERE pubkey.pubkey_hash = ? """, (dbhash,))
201
202     def get_history(self, addr):
203         
204         if config.get('server','cache') == 'yes':
205             cached_version = self.tx_cache.get( addr ) 
206             if cached_version is not None: 
207                 return cached_version
208
209         version, binaddr = decode_check_address(addr)
210         if binaddr is None:
211             return None
212
213         dbhash = self.binin(binaddr)
214         rows = []
215         rows += self.get_address_out_rows( dbhash )
216         rows += self.get_address_in_rows( dbhash )
217
218         txpoints = []
219         known_tx = []
220
221         for row in rows:
222             try:
223                 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
224             except:
225                 print "cannot unpack row", row
226                 break
227             tx_hash = self.hashout_hex(tx_hash)
228             txpoint = {
229                     "nTime":    int(nTime),
230                     #"chain_id": int(chain_id),
231                     "height":   int(height),
232                     "is_in":    int(is_in),
233                     "blk_hash": self.hashout_hex(blk_hash),
234                     "tx_hash":  tx_hash,
235                     "tx_id":    int(tx_id),
236                     "pos":      int(pos),
237                     "value":    int(value),
238                     }
239
240             txpoints.append(txpoint)
241             known_tx.append(self.hashout_hex(tx_hash))
242
243
244         # todo: sort them really...
245         txpoints = sorted(txpoints, key=operator.itemgetter("nTime"))
246
247         # read memory pool
248         rows = []
249         rows += self.get_address_in_rows_memorypool( dbhash )
250         rows += self.get_address_out_rows_memorypool( dbhash )
251         address_has_mempool = False
252
253         for row in rows:
254             is_in, tx_hash, tx_id, pos, value = row
255             tx_hash = self.hashout_hex(tx_hash)
256             if tx_hash in known_tx:
257                 continue
258
259             # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
260             address_has_mempool = True
261
262             # this means pending transactions are returned by getmemorypool
263             if tx_hash not in self.mempool_keys:
264                 continue
265
266             #print "mempool", tx_hash
267             txpoint = {
268                     "nTime":    0,
269                     #"chain_id": 1,
270                     "height":   0,
271                     "is_in":    int(is_in),
272                     "blk_hash": 'mempool', 
273                     "tx_hash":  tx_hash,
274                     "tx_id":    int(tx_id),
275                     "pos":      int(pos),
276                     "value":    int(value),
277                     }
278             txpoints.append(txpoint)
279
280
281         for txpoint in txpoints:
282             tx_id = txpoint['tx_id']
283             
284             txinputs = []
285             inrows = self.get_tx_inputs(tx_id)
286             for row in inrows:
287                 _hash = self.binout(row[6])
288                 address = hash_to_address(chr(0), _hash)
289                 txinputs.append(address)
290             txpoint['inputs'] = txinputs
291             txoutputs = []
292             outrows = self.get_tx_outputs(tx_id)
293             for row in outrows:
294                 _hash = self.binout(row[6])
295                 address = hash_to_address(chr(0), _hash)
296                 txoutputs.append(address)
297             txpoint['outputs'] = txoutputs
298
299             # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
300             if not txpoint['is_in']:
301                 # detect if already redeemed...
302                 for row in outrows:
303                     if row[6] == dbhash: break
304                 else:
305                     raise
306                 #row = self.get_tx_output(tx_id,dbhash)
307                 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
308                 # if not redeemed, we add the script
309                 if row:
310                     if not row[4]: txpoint['raw_scriptPubKey'] = row[1]
311
312         # cache result
313         if config.get('server','cache') == 'yes' and not address_has_mempool:
314             self.tx_cache[addr] = txpoints
315         
316         return txpoints
317
318
319
320 def send_tx(tx):
321     import bitcoinrpc
322     conn = bitcoinrpc.connect_to_local()
323     try:
324         v = conn.importtransaction(tx)
325     except:
326         v = "error: transaction rejected by memorypool"
327     return v
328
329
330 def listen_thread(store):
331     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
332     s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
333     s.bind((config.get('server','host'), config.getint('server','port')))
334     s.listen(1)
335     while not stopping:
336         conn, addr = s.accept()
337         thread.start_new_thread(client_thread, (addr, conn,))
338
339 def random_string(N):
340     import random, string
341     return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
342
343 def client_thread(ipaddr,conn):
344     #print "client thread", ipaddr
345     try:
346         ipaddr = ipaddr[0]
347         msg = ''
348         while 1:
349             d = conn.recv(1024)
350             msg += d
351             if not d: 
352                 break
353             if '#' in msg:
354                 msg = msg.split('#', 1)[0]
355                 break
356         try:
357             cmd, data = ast.literal_eval(msg)
358         except:
359             print "syntax error", repr(msg), ipaddr
360             conn.close()
361             return
362
363         out = do_command(cmd, data, ipaddr)
364         if out:
365             #print ipaddr, cmd, len(out)
366             try:
367                 conn.send(out)
368             except:
369                 print "error, could not send"
370
371     finally:
372         conn.close()
373     
374
375 def cmd_stop(data):
376     global stopping
377     if password == data:
378         stopping = True
379         return 'ok'
380     else:
381         return 'wrong password'
382
383 def cmd_load(pw):
384     if password == pw:
385         return repr( len(sessions) )
386     else:
387         return 'wrong password'
388
389
390 def clear_cache(pw):
391     if password == pw:
392         store.tx_cache = {}
393         return 'ok'
394     else:
395         return 'wrong password'
396
397 def get_cache(pw,addr):
398     if password == pw:
399         return store.tx_cache.get(addr)
400     else:
401         return 'wrong password'
402
403
404 def do_command(cmd, data, ipaddr):
405
406     if cmd=='b':
407         out = "%d"%block_number
408
409     elif cmd in ['session','new_session']:
410         session_id = random_string(10)
411         try:
412             if cmd == 'session':
413                 addresses = ast.literal_eval(data)
414                 version = "old"
415             else:
416                 version, addresses = ast.literal_eval(data)
417                 if version[0]=="0": version = "v" + version
418         except:
419             print "error", data
420             return None
421
422         print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
423
424         sessions[session_id] = { 'addresses':{}, 'version':version, 'ip':ipaddr }
425         for a in addresses:
426             sessions[session_id]['addresses'][a] = ''
427         out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) )
428         sessions[session_id]['last_time'] = time.time()
429
430     elif cmd=='update_session':
431         try:
432             session_id, addresses = ast.literal_eval(data)
433         except:
434             print "error"
435             return None
436
437         print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
438
439         sessions[session_id]['addresses'] = {}
440         for a in addresses:
441             sessions[session_id]['addresses'][a] = ''
442         out = 'ok'
443         sessions[session_id]['last_time'] = time.time()
444
445     elif cmd == 'bccapi_login':
446         import electrum
447         print "data",data
448         v, k = ast.literal_eval(data)
449         master_public_key = k.decode('hex') # todo: sanitize. no need to decode twice...
450         print master_public_key
451         wallet_id = random_string(10)
452         w = electrum.Wallet()
453         w.master_public_key = master_public_key.decode('hex')
454         w.synchronize()
455         wallets[wallet_id] = w
456         out = wallet_id
457         print "wallets", wallets
458
459     elif cmd == 'bccapi_getAccountInfo':
460         from electrum import int_to_hex
461         v, wallet_id = ast.literal_eval(data)
462         w = wallets.get(wallet_id)
463         if w is not None:
464             num = len(w.addresses)
465             c, u = w.get_balance()
466             out = int_to_hex(num,4) + int_to_hex(c,8) + int_to_hex( c+u, 8 )
467             out = out.decode('hex')
468         else:
469             print "error",data
470             out = "error"
471
472     elif cmd == 'bccapi_getAccountStatement':
473         from electrum import int_to_hex
474         v, wallet_id = ast.literal_eval(data)
475         w = wallets.get(wallet_id)
476         if w is not None:
477             num = len(w.addresses)
478             c, u = w.get_balance()
479             total_records = num_records = 0
480             out = int_to_hex(num,4) + int_to_hex(c,8) + int_to_hex( c+u, 8 ) + int_to_hex( total_records ) + int_to_hex( num_records )
481             out = out.decode('hex')
482         else:
483             print "error",data
484             out = "error"
485
486     elif cmd == 'bccapi_getSendCoinForm':
487         out = ''
488
489     elif cmd == 'bccapi_submitTransaction':
490         out = ''
491             
492     elif cmd=='poll': 
493         session_id = data
494         session = sessions.get(session_id)
495         if session is None:
496             print time.asctime(), "session not found", session_id, ipaddr
497             out = repr( (-1, {}))
498         else:
499             t1 = time.time()
500             addresses = session['addresses']
501             session['last_time'] = time.time()
502             ret = {}
503             k = 0
504             for addr in addresses:
505                 if store.tx_cache.get( addr ) is not None: k += 1
506
507                 # get addtess status, i.e. the last block for that address.
508                 tx_points = store.get_history(addr)
509                 if not tx_points:
510                     status = None
511                 else:
512                     lastpoint = tx_points[-1]
513                     status = lastpoint['blk_hash']
514                     # this is a temporary hack; move it up once old clients have disappeared
515                     if status == 'mempool' and session['version'] != "old":
516                         status = status + ':%d'% len(tx_points)
517
518                 last_status = addresses.get( addr )
519                 if last_status != status:
520                     addresses[addr] = status
521                     ret[addr] = status
522             if ret:
523                 sessions[session_id]['addresses'] = addresses
524             out = repr( (block_number, ret ) )
525             t2 = time.time() - t1 
526             if t2 > 10:
527                 print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2
528
529     elif cmd == 'h': 
530         # history
531         address = data
532         out = repr( store.get_history( address ) )
533
534     elif cmd == 'load': 
535         out = cmd_load(data)
536
537     elif cmd =='tx':
538         out = send_tx(data)
539         print "sent tx:", out
540
541     elif cmd == 'stop':
542         out = cmd_stop(data)
543
544     elif cmd == 'peers':
545         out = repr(peer_list.values())
546
547     else:
548         out = None
549
550     return out
551
552
553
554
555 def memorypool_update(store):
556     ds = BCDataStream.BCDataStream()
557     store.mempool_keys = []
558     conn = bitcoinrpc.connect_to_local()
559     try:
560         v = conn.getmemorypool()
561     except:
562         print "cannot contact bitcoin daemon"
563         return
564     v = v['transactions']
565     for hextx in v:
566         ds.clear()
567         ds.write(hextx.decode('hex'))
568         tx = deserialize.parse_Transaction(ds)
569         tx['hash'] = util.double_sha256(tx['tx'])
570         tx_hash = tx['hash'][::-1].encode('hex')
571         store.mempool_keys.append(tx_hash)
572         if store.tx_find_id_and_value(tx):
573             pass
574         else:
575             store.import_tx(tx, False)
576
577     store.commit()
578
579
580
581 def clean_session_thread():
582     while not stopping:
583         time.sleep(30)
584         t = time.time()
585         for k,s in sessions.items():
586             t0 = s['last_time']
587             if t - t0 > 5*60:
588                 print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "end session", s['ip']
589                 sessions.pop(k)
590             
591
592 def irc_thread():
593     global peer_list
594     NICK = 'E_'+random_string(10)
595     while not stopping:
596         try:
597             s = socket.socket()
598             s.connect(('irc.freenode.net', 6667))
599             s.send('USER electrum 0 * :'+config.get('server','host')+' '+config.get('server','ircname')+'\n')
600             s.send('NICK '+NICK+'\n')
601             s.send('JOIN #electrum\n')
602             sf = s.makefile('r', 0)
603             t = 0
604             while not stopping:
605                 line = sf.readline()
606                 line = line.rstrip('\r\n')
607                 line = line.split()
608                 if line[0]=='PING': 
609                     s.send('PONG '+line[1]+'\n')
610                 elif '353' in line: # answer to /names
611                     k = line.index('353')
612                     for item in line[k+1:]:
613                         if item[0:2] == 'E_':
614                             s.send('WHO %s\n'%item)
615                 elif '352' in line: # answer to /who
616                     # warning: this is a horrible hack which apparently works
617                     k = line.index('352')
618                     ip = line[k+4]
619                     ip = socket.gethostbyname(ip)
620                     name = line[k+6]
621                     host = line[k+9]
622                     peer_list[name] = (ip,host)
623                 if time.time() - t > 5*60:
624                     s.send('NAMES #electrum\n')
625                     t = time.time()
626                     peer_list = {}
627         except:
628             traceback.print_exc(file=sys.stdout)
629         finally:
630             sf.close()
631             s.close()
632
633
634
635 def jsonrpc_thread(store):
636     # see http://code.google.com/p/jsonrpclib/
637     from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
638     server = SimpleJSONRPCServer(('localhost', 8080))
639     server.register_function(store.get_history, 'history')
640     server.register_function(lambda : peer_list.values(), 'peers')
641     server.register_function(cmd_stop, 'stop')
642     server.register_function(cmd_load, 'load')
643     server.register_function(lambda : block_number, 'blocks')
644     server.register_function(clear_cache, 'clear_cache')
645     server.register_function(get_cache, 'get_cache')
646     server.serve_forever()
647
648
649 import traceback
650
651
652 if __name__ == '__main__':
653
654     if len(sys.argv)>1:
655         import jsonrpclib
656         server = jsonrpclib.Server('http://localhost:8080')
657         cmd = sys.argv[1]
658         if cmd == 'load':
659             out = server.load(password)
660         elif cmd == 'peers':
661             out = server.peers()
662         elif cmd == 'stop':
663             out = server.stop(password)
664         elif cmd == 'clear_cache':
665             out = server.clear_cache(password)
666         elif cmd == 'get_cache':
667             out = server.get_cache(password,sys.argv[2])
668         elif cmd == 'h':
669             out = server.history(sys.argv[2])
670         elif cmd == 'b':
671             out = server.blocks()
672         print out
673         sys.exit(0)
674
675
676     print "starting Electrum server"
677     print "cache:", config.get('server', 'cache')
678
679     conf = DataStore.CONFIG_DEFAULTS
680     args, argv = readconf.parse_argv( [], conf)
681     args.dbtype= config.get('database','type')
682     if args.dbtype == 'sqlite3':
683         args.connect_args = { 'database' : config.get('database','database') }
684     elif args.dbtype == 'MySQLdb':
685         args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
686     elif args.dbtype == 'psycopg2':
687         args.connect_args = { 'database' : config.get('database','database') }
688     store = MyStore(args)
689     store.tx_cache = {}
690     store.mempool_keys = {}
691
692     thread.start_new_thread(listen_thread, (store,))
693     thread.start_new_thread(jsonrpc_thread, (store,))
694     thread.start_new_thread(clean_session_thread, ())
695     if (config.get('server','irc') == 'yes' ):
696         thread.start_new_thread(irc_thread, ())
697
698     while not stopping:
699         try:
700             dblock.acquire()
701             store.catch_up()
702             memorypool_update(store)
703             block_number = store.get_block_number(1)
704             dblock.release()
705         except:
706             traceback.print_exc(file=sys.stdout)
707         time.sleep(10)
708
709     print "server stopped"
710