read password once
[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
376 def do_command(cmd, data, ipaddr):
377
378     if cmd=='b':
379         out = "%d"%block_number
380
381     elif cmd in ['session','new_session']:
382         session_id = random_string(10)
383         try:
384             if cmd == 'session':
385                 addresses = ast.literal_eval(data)
386                 version = "old"
387             else:
388                 version, addresses = ast.literal_eval(data)
389                 if version[0]=="0": version = "v" + version
390         except:
391             print "error", data
392             return None
393
394         print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
395
396         sessions[session_id] = { 'addresses':{}, 'version':version, 'ip':ipaddr }
397         for a in addresses:
398             sessions[session_id]['addresses'][a] = ''
399         out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) )
400         sessions[session_id]['last_time'] = time.time()
401
402     elif cmd=='update_session':
403         try:
404             session_id, addresses = ast.literal_eval(data)
405         except:
406             print "error"
407             return None
408
409         print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
410
411         sessions[session_id]['addresses'] = {}
412         for a in addresses:
413             sessions[session_id]['addresses'][a] = ''
414         out = 'ok'
415         sessions[session_id]['last_time'] = time.time()
416
417     elif cmd == 'bccapi_login':
418         import electrum
419         print "data",data
420         v, k = ast.literal_eval(data)
421         master_public_key = k.decode('hex') # todo: sanitize. no need to decode twice...
422         print master_public_key
423         wallet_id = random_string(10)
424         w = electrum.Wallet()
425         w.master_public_key = master_public_key.decode('hex')
426         w.synchronize()
427         wallets[wallet_id] = w
428         out = wallet_id
429         print "wallets", wallets
430
431     elif cmd == 'bccapi_getAccountInfo':
432         from electrum import int_to_hex
433         v, wallet_id = ast.literal_eval(data)
434         w = wallets.get(wallet_id)
435         if w is not None:
436             num = len(w.addresses)
437             c, u = w.get_balance()
438             out = int_to_hex(num,4) + int_to_hex(c,8) + int_to_hex( c+u, 8 )
439             out = out.decode('hex')
440         else:
441             print "error",data
442             out = "error"
443
444     elif cmd == 'bccapi_getAccountStatement':
445         from electrum import int_to_hex
446         v, wallet_id = ast.literal_eval(data)
447         w = wallets.get(wallet_id)
448         if w is not None:
449             num = len(w.addresses)
450             c, u = w.get_balance()
451             total_records = num_records = 0
452             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 )
453             out = out.decode('hex')
454         else:
455             print "error",data
456             out = "error"
457
458     elif cmd == 'bccapi_getSendCoinForm':
459         out = ''
460
461     elif cmd == 'bccapi_submitTransaction':
462         out = ''
463             
464     elif cmd=='poll': 
465         session_id = data
466         session = sessions.get(session_id)
467         if session is None:
468             print time.asctime(), "session not found", session_id, ipaddr
469             out = repr( (-1, {}))
470         else:
471             t1 = time.time()
472             addresses = session['addresses']
473             session['last_time'] = time.time()
474             ret = {}
475             k = 0
476             for addr in addresses:
477                 if store.tx_cache.get( addr ) is not None: k += 1
478
479                 # get addtess status, i.e. the last block for that address.
480                 tx_points = store.get_history(addr)
481                 if not tx_points:
482                     status = None
483                 else:
484                     lastpoint = tx_points[-1]
485                     status = lastpoint['blk_hash']
486                     # this is a temporary hack; move it up once old clients have disappeared
487                     if status == 'mempool' and session['version'] != "old":
488                         status = status + ':%d'% len(tx_points)
489
490                 last_status = addresses.get( addr )
491                 if last_status != status:
492                     addresses[addr] = status
493                     ret[addr] = status
494             if ret:
495                 sessions[session_id]['addresses'] = addresses
496             out = repr( (block_number, ret ) )
497             t2 = time.time() - t1 
498             if t2 > 10:
499                 print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2
500
501     elif cmd == 'h': 
502         # history
503         address = data
504         out = repr( store.get_history( address ) )
505
506     elif cmd == 'load': 
507         if password == data:
508             out = repr( len(sessions) )
509         else:
510             out = 'wrong password'
511
512     elif cmd =='tx':
513         out = send_tx(data)
514         print "sent tx:", out
515
516     elif cmd =='clear_cache':
517         if password == data:
518             store.tx_cache = {}
519             out = 'ok'
520         else:
521             out = 'wrong password'
522
523     elif cmd =='get_cache':
524         try:
525             pw, addr = data
526         except:
527             addr = None
528         if addr:
529             if password == pw:
530                 out = store.tx_cache.get(addr)
531                 out = repr(out)
532             else:
533                 out = 'wrong password'
534         else:
535             out = "error: "+ repr(data)
536
537     elif cmd == 'stop':
538         global stopping
539         if password == data:
540             stopping = True
541             out = 'ok'
542         else:
543             out = 'wrong password'
544
545     elif cmd == 'peers':
546         out = repr(peer_list.values())
547
548     else:
549         out = None
550
551     return out
552
553
554
555
556
557
558 def memorypool_update(store):
559     ds = BCDataStream.BCDataStream()
560     store.mempool_keys = []
561     conn = bitcoinrpc.connect_to_local()
562     try:
563         v = conn.getmemorypool()
564     except:
565         print "cannot contact bitcoin daemon"
566         return
567     v = v['transactions']
568     for hextx in v:
569         ds.clear()
570         ds.write(hextx.decode('hex'))
571         tx = deserialize.parse_Transaction(ds)
572         tx['hash'] = util.double_sha256(tx['tx'])
573         tx_hash = tx['hash'][::-1].encode('hex')
574         store.mempool_keys.append(tx_hash)
575         if store.tx_find_id_and_value(tx):
576             pass
577         else:
578             store.import_tx(tx, False)
579
580     store.commit()
581
582
583
584 def clean_session_thread():
585     while not stopping:
586         time.sleep(30)
587         t = time.time()
588         for k,s in sessions.items():
589             t0 = s['last_time']
590             if t - t0 > 5*60:
591                 print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "end session", s['ip']
592                 sessions.pop(k)
593             
594
595 def irc_thread():
596     global peer_list
597     NICK = 'E_'+random_string(10)
598     while not stopping:
599         try:
600             s = socket.socket()
601             s.connect(('irc.freenode.net', 6667))
602             s.send('USER electrum 0 * :'+config.get('server','host')+' '+config.get('server','ircname')+'\n')
603             s.send('NICK '+NICK+'\n')
604             s.send('JOIN #electrum\n')
605             sf = s.makefile('r', 0)
606             t = 0
607             while not stopping:
608                 line = sf.readline()
609                 line = line.rstrip('\r\n')
610                 line = line.split()
611                 if line[0]=='PING': 
612                     s.send('PONG '+line[1]+'\n')
613                 elif '353' in line: # answer to /names
614                     k = line.index('353')
615                     for item in line[k+1:]:
616                         if item[0:2] == 'E_':
617                             s.send('WHO %s\n'%item)
618                 elif '352' in line: # answer to /who
619                     # warning: this is a horrible hack which apparently works
620                     k = line.index('352')
621                     ip = line[k+4]
622                     ip = socket.gethostbyname(ip)
623                     name = line[k+6]
624                     host = line[k+9]
625                     peer_list[name] = (ip,host)
626                 if time.time() - t > 5*60:
627                     s.send('NAMES #electrum\n')
628                     t = time.time()
629                     peer_list = {}
630         except:
631             traceback.print_exc(file=sys.stdout)
632         finally:
633             sf.close()
634             s.close()
635
636
637 import traceback
638
639
640 if __name__ == '__main__':
641
642     if len(sys.argv)>1:
643         cmd = sys.argv[1]
644         if cmd == 'load':
645             request = "('load','%s')#"%password
646         elif cmd == 'peers':
647             request = "('peers','')#"
648         elif cmd == 'stop':
649             request = "('stop','%s')#"%password
650         elif cmd == 'clear_cache':
651             request = "('clear_cache','%s')#"%password
652         elif cmd == 'get_cache':
653             request = "('get_cache',('%s','%s'))#"%(password,sys.argv[2])
654         elif cmd == 'h':
655             request = "('h','%s')#"%sys.argv[2]
656         elif cmd == 'b':
657             request = "('b','')#"
658
659         s = socket.socket( socket.AF_INET, socket.SOCK_STREAM)
660         s.connect((config.get('server','host'), config.getint('server','port')))
661         s.send( request )
662         out = ''
663         while 1:
664             msg = s.recv(1024)
665             if msg: out += msg
666             else: break
667         s.close()
668         print out
669         sys.exit(0)
670
671
672     print "starting Electrum server"
673     print "cache:", config.get('server', 'cache')
674
675     conf = DataStore.CONFIG_DEFAULTS
676     args, argv = readconf.parse_argv( [], conf)
677     args.dbtype= config.get('database','type')
678     if args.dbtype == 'sqlite3':
679         args.connect_args = { 'database' : config.get('database','database') }
680     elif args.dbtype == 'MySQLdb':
681         args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
682     elif args.dbtype == 'psycopg2':
683         args.connect_args = { 'database' : config.get('database','database') }
684     store = MyStore(args)
685     store.tx_cache = {}
686     store.mempool_keys = {}
687
688     thread.start_new_thread(listen_thread, (store,))
689     thread.start_new_thread(clean_session_thread, ())
690     if (config.get('server','irc') == 'yes' ):
691         thread.start_new_thread(irc_thread, ())
692
693     while not stopping:
694         try:
695             dblock.acquire()
696             store.catch_up()
697             memorypool_update(store)
698             block_number = store.get_block_number(1)
699             dblock.release()
700         except:
701             traceback.print_exc(file=sys.stdout)
702         time.sleep(10)
703
704     print "server stopped"
705