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