X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=42b6b9f487a3cf52384b955f21ab69517995ca9b;hb=d9bbde33c1faca005c47bce650650907991d9c51;hp=417382837266013e9154fded47cbfb83f36d7c8c;hpb=11f7b3515070354092ac6ffd9e6afd3aead32d34;p=electrum-server.git diff --git a/server.py b/server.py index 4173828..42b6b9f 100755 --- a/server.py +++ b/server.py @@ -25,15 +25,22 @@ Todo: """ -import time, socket, operator, thread, ast, sys,re +import time, json, socket, operator, thread, ast, sys,re import psycopg2, binascii -import bitcoinrpc from Abe.abe import hash_to_address, decode_check_address from Abe.DataStore import DataStore as Datastore_class from Abe import DataStore, readconf, BCDataStream, deserialize, util, base58 import ConfigParser +from json import dumps, loads +import urllib + +# we need to import electrum +sys.path.append('../client/') +from wallet import Wallet +from interface import Interface + config = ConfigParser.ConfigParser() # set some defaults, which will be overwritten by the config file @@ -56,12 +63,31 @@ try: except: print "Could not read electrum.conf. I will use the default values." +try: + f = open('/etc/electrum.banner','r') + config.set('server','banner', f.read()) + f.close() +except: + pass + +password = config.get('server','password') +bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port')) + stopping = False block_number = -1 +old_block_number = -1 sessions = {} +sessions_sub_numblocks = [] # sessions that have subscribed to the service + dblock = thread.allocate_lock() peer_list = {} +wallets = {} # for ultra-light clients such as bccapi + +from Queue import Queue +input_queue = Queue() +output_queue = Queue() +address_queue = Queue() class MyStore(Datastore_class): @@ -75,15 +101,18 @@ class MyStore(Datastore_class): _hash = store.binout(row[6]) address = hash_to_address(chr(0), _hash) if self.tx_cache.has_key(address): - #print "cache: invalidating", address + print "cache: invalidating", address self.tx_cache.pop(address) + address_queue.put(address) + outrows = self.get_tx_outputs(txid, False) for row in outrows: _hash = store.binout(row[6]) address = hash_to_address(chr(0), _hash) if self.tx_cache.has_key(address): - #print "cache: invalidating", address + print "cache: invalidating", address self.tx_cache.pop(address) + address_queue.put(address) def safe_sql(self,sql, params=(), lock=True): try: @@ -196,16 +225,17 @@ class MyStore(Datastore_class): JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id) WHERE pubkey.pubkey_hash = ? """, (dbhash,)) - def get_txpoints(self, addr): + def get_history(self, addr): if config.get('server','cache') == 'yes': - cached_version = self.tx_cache.get( addr ) - if cached_version is not None: + cached_version = self.tx_cache.get( addr ) + if cached_version is not None: return cached_version version, binaddr = decode_check_address(addr) if binaddr is None: - return "err" + return None + dbhash = self.binin(binaddr) rows = [] rows += self.get_address_out_rows( dbhash ) @@ -223,7 +253,6 @@ class MyStore(Datastore_class): tx_hash = self.hashout_hex(tx_hash) txpoint = { "nTime": int(nTime), - #"chain_id": int(chain_id), "height": int(height), "is_in": int(is_in), "blk_hash": self.hashout_hex(blk_hash), @@ -251,14 +280,17 @@ class MyStore(Datastore_class): tx_hash = self.hashout_hex(tx_hash) if tx_hash in known_tx: continue + + # this means that pending transactions were added to the db, even if they are not returned by getmemorypool + address_has_mempool = True + + # this means pending transactions are returned by getmemorypool if tx_hash not in self.mempool_keys: continue - address_has_mempool = True #print "mempool", tx_hash txpoint = { "nTime": 0, - #"chain_id": 1, "height": 0, "is_in": int(is_in), "blk_hash": 'mempool', @@ -309,31 +341,181 @@ class MyStore(Datastore_class): +class Direct_Interface(Interface): + def __init__(self): + pass + + def handler(self, method, params = ''): + cmds = {'session.new':new_session, + 'session.poll':poll_session, + 'session.update':update_session, + 'blockchain.transaction.broadcast':send_tx, + 'blockchain.address.get_history':store.get_history + } + func = cmds[method] + return func( params ) + + def send_tx(tx): - import bitcoinrpc - conn = bitcoinrpc.connect_to_local() - try: - v = conn.importtransaction(tx) - except: - v = "error: transaction rejected by memorypool" - return v + postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'}) + respdata = urllib.urlopen(bitcoind_url, postdata).read() + r = loads(respdata) + if r['error'] != None: + out = "error: transaction rejected by memorypool\n"+tx + else: + out = r['result'] + return out + + + +def random_string(N): + import random, string + return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N)) + -def listen_thread(store): +def cmd_stop(data): + global stopping + if password == data: + stopping = True + return 'ok' + else: + return 'wrong password' + +def cmd_load(pw): + if password == pw: + return repr( len(sessions) ) + else: + return 'wrong password' + + +def clear_cache(pw): + if password == pw: + store.tx_cache = {} + return 'ok' + else: + return 'wrong password' + +def get_cache(pw,addr): + if password == pw: + return store.tx_cache.get(addr) + else: + return 'wrong password' + + +def poll_session(session_id): + session = sessions.get(session_id) + if session is None: + print time.asctime(), "session not found", session_id + out = repr( (-1, {})) + else: + t1 = time.time() + addresses = session['addresses'] + session['last_time'] = time.time() + ret = {} + k = 0 + for addr in addresses: + if store.tx_cache.get( addr ) is not None: k += 1 + status = get_address_status( addr ) + last_status = addresses.get( addr ) + if last_status != status: + addresses[addr] = status + ret[addr] = status + if ret: + sessions[session_id]['addresses'] = addresses + out = repr( (block_number, ret ) ) + t2 = time.time() - t1 + if t2 > 10: + print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 + + return out + + +def do_update_address(addr): + # an address was involved in a transaction; we check if it was subscribed to in a session + # the address can be subscribed in several sessions; the cache should ensure that we don't do redundant requests + for session_id in sessions.keys(): + session = sessions[session_id] + if session.get('type') != 'subscribe': continue + addresses = session['addresses'].keys() + + if addr in addresses: + print "address ", addr, "found in session", session_id + status = get_address_status( addr ) + print "new_status:", status + last_status = session['addresses'][addr] + print "last_status", last_status + if last_status != status: + print "status is new", addr + send_status(session_id,addr,status) + sessions[session_id]['addresses'][addr] = status + + +def get_address_status(addr): + # get address status, i.e. the last block for that address. + tx_points = store.get_history(addr) + if not tx_points: + status = None + else: + lastpoint = tx_points[-1] + status = lastpoint['blk_hash'] + # this is a temporary hack; move it up once old clients have disappeared + if status == 'mempool': # and session['version'] != "old": + status = status + ':%d'% len(tx_points) + return status + + +def send_numblocks(session_id): + out = json.dumps( {'method':'numblocks.subscribe', 'result':block_number} ) + output_queue.put((session_id, out)) + +def send_status(session_id, address, status): + out = json.dumps( { 'method':'address.subscribe', 'address':address, 'status':status } ) + output_queue.put((session_id, out)) + +def subscribe_to_numblocks(session_id): + sessions_sub_numblocks.append(session_id) + send_numblocks(session_id) + +def subscribe_to_address(session_id, address): + status = get_address_status(address) + sessions[session_id]['type'] = 'subscribe' + sessions[session_id]['addresses'][address] = status + sessions[session_id]['last_time'] = time.time() + send_status(session_id, address, status) + +def new_session(version, addresses): + session_id = random_string(10) + sessions[session_id] = { 'addresses':{}, 'version':version } + for a in addresses: + sessions[session_id]['addresses'][a] = '' + out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) ) + sessions[session_id]['last_time'] = time.time() + return out + +def update_session(session_id,addresses): + sessions[session_id]['addresses'] = {} + for a in addresses: + sessions[session_id]['addresses'][a] = '' + sessions[session_id]['last_time'] = time.time() + return 'ok' + +def native_server_thread(): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((config.get('server','host'), config.getint('server','port'))) s.listen(1) while not stopping: conn, addr = s.accept() - thread.start_new_thread(client_thread, (addr, conn,)) + try: + thread.start_new_thread(native_client_thread, (addr, conn,)) + except: + # can't start new thread if there is no memory.. + traceback.print_exc(file=sys.stdout) -def random_string(N): - import random, string - return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N)) -def client_thread(ipaddr,conn): +def native_client_thread(ipaddr,conn): #print "client thread", ipaddr try: ipaddr = ipaddr[0] @@ -343,180 +525,267 @@ def client_thread(ipaddr,conn): msg += d if not d: break - if d[-1]=='#': + if '#' in msg: + msg = msg.split('#', 1)[0] break - - #print msg - try: - cmd, data = ast.literal_eval(msg[:-1]) + cmd, data = ast.literal_eval(msg) except: - print "syntax error", repr(msg) + print "syntax error", repr(msg), ipaddr conn.close() return - if cmd=='b': - out = "%d"%block_number - - elif cmd in ['session','new_session']: - session_id = random_string(10) + out = do_command(cmd, data, ipaddr) + if out: + #print ipaddr, cmd, len(out) try: - if cmd == 'session': - addresses = ast.literal_eval(data) - version = "old" - else: - version, addresses = ast.literal_eval(data) - version = "v"+version + conn.send(out) except: - print "error", data - conn.close() - return + print "error, could not send" - print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version + finally: + conn.close() - sessions[session_id] = { 'addresses':{}, 'version':version, 'ip':ipaddr } - for a in addresses: - sessions[session_id]['addresses'][a] = '' - out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) ) - sessions[session_id]['last_time'] = time.time() - elif cmd=='update_session': - try: - session_id, addresses = ast.literal_eval(data) - except: - print "error" - conn.close() - return - - print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses) - - sessions[session_id]['addresses'] = {} - for a in addresses: - sessions[session_id]['addresses'][a] = '' - out = 'ok' - sessions[session_id]['last_time'] = time.time() - - elif cmd=='poll': - session_id = data - session = sessions.get(session_id) - if session is None: - print time.asctime(), "session not found", session_id, ipaddr - out = repr( (-1, {})) - else: - t1 = time.time() - addresses = session['addresses'] - session['last_time'] = time.time() - ret = {} - k = 0 - for addr in addresses: - if store.tx_cache.get( addr ) is not None: k += 1 - - # get addtess status, i.e. the last block for that address. - tx_points = store.get_txpoints(addr) - if not tx_points: - status = None - else: - lastpoint = tx_points[-1] - status = lastpoint['blk_hash'] - # this is a temporary hack; move it up once old clients have disappeared - if status == 'mempool' and session['version'] != "old": - status = status + ':%s'% lastpoint['tx_hash'] - - last_status = addresses.get( addr ) - if last_status != status: - addresses[addr] = status - ret[addr] = status - if ret: - sessions[session_id]['addresses'] = addresses - out = repr( (block_number, ret ) ) - t2 = time.time() - t1 - if t2 > 10: - print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 - - elif cmd == 'h': - # history - address = data - out = repr( store.get_txpoints( address ) ) - elif cmd == 'load': - if config.get('server','password') == data: - out = repr( len(sessions) ) - else: - out = 'wrong password' +# used by the native handler +def do_command(cmd, data, ipaddr): - elif cmd =='tx': - out = send_tx(data) - print "sent tx:", out + timestr = time.strftime("[%d/%m/%Y-%H:%M:%S]") - elif cmd =='clear_cache': - if config.get('server','password') == data: - self.tx_cache = {} - out = 'ok' - else: - out = 'wrong password' + if cmd=='b': + out = "%d"%block_number - elif cmd =='get_cache': - try: - pw, addr = data - except: - addr = None - if addr: - if config.get('server','password') == pw: - out = store.tx_cache.get(addr) - out = repr(out) - else: - out = 'wrong password' + elif cmd in ['session','new_session']: + try: + if cmd == 'session': + addresses = ast.literal_eval(data) + version = "old" else: - out = "error: "+ repr(data) + version, addresses = ast.literal_eval(data) + if version[0]=="0": version = "v" + version + except: + print "error", data + return None + print timestr, "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version + out = new_session(version, addresses) - elif cmd == 'stop': - global stopping - if config.get('server','password') == data: - stopping = True - out = 'ok' - else: - out = 'wrong password' + elif cmd=='update_session': + try: + session_id, addresses = ast.literal_eval(data) + except: + print "error" + return None + print timestr, "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses) + out = update_session(session_id,addresses) + + elif cmd == 'bccapi_login': + import electrum + print "data",data + v, k = ast.literal_eval(data) + master_public_key = k.decode('hex') # todo: sanitize. no need to decode twice... + print master_public_key + wallet_id = random_string(10) + w = Wallet( Direct_Interface() ) + w.master_public_key = master_public_key.decode('hex') + w.synchronize() + wallets[wallet_id] = w + out = wallet_id + print "wallets", wallets + + elif cmd == 'bccapi_getAccountInfo': + from wallet import int_to_hex + v, wallet_id = ast.literal_eval(data) + w = wallets.get(wallet_id) + if w is not None: + num = len(w.addresses) + c, u = w.get_balance() + out = int_to_hex(num,4) + int_to_hex(c,8) + int_to_hex( c+u, 8 ) + out = out.decode('hex') + else: + print "error",data + out = "error" + + elif cmd == 'bccapi_getAccountStatement': + from wallet import int_to_hex + v, wallet_id = ast.literal_eval(data) + w = wallets.get(wallet_id) + if w is not None: + num = len(w.addresses) + c, u = w.get_balance() + total_records = num_records = 0 + 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 ) + out = out.decode('hex') + else: + print "error",data + out = "error" - elif cmd == 'peers': - out = repr(peer_list.values()) + elif cmd == 'bccapi_getSendCoinForm': + out = '' - else: - out = None + elif cmd == 'bccapi_submitTransaction': + out = '' + + elif cmd=='poll': + out = poll_session(data) + elif cmd == 'h': + # history + address = data + out = repr( store.get_history( address ) ) + + elif cmd == 'load': + out = cmd_load(data) + + elif cmd =='tx': + out = send_tx(data) + print timestr, "sent tx:", ipaddr, out + + elif cmd == 'stop': + out = cmd_stop(data) + + elif cmd == 'peers': + out = repr(peer_list.values()) + + else: + out = None + + return out + + + +#################################################################### + +def tcp_server_thread(): + thread.start_new_thread(process_input_queue, ()) + thread.start_new_thread(process_output_queue, ()) + + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + s.bind((config.get('server','host'), 50001)) + s.listen(1) + while not stopping: + conn, addr = s.accept() + try: + thread.start_new_thread(tcp_client_thread, (addr, conn,)) + except: + # can't start new thread if there is no memory.. + traceback.print_exc(file=sys.stdout) + + +def close_session(session_id): + print "lost connection", session_id + sessions.pop(session_id) + if session_id in sessions_sub_numblocks: + sessions_sub_numblocks.remove(session_id) + + +# one thread per client. put requests in a queue. +def tcp_client_thread(ipaddr,conn): + """ use a persistent connection. put commands in a queue.""" + print "persistent client thread", ipaddr + global sessions + + session_id = random_string(10) + sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown' } + + ipaddr = ipaddr[0] + msg = '' + + while not stopping: + d = conn.recv(1024) + msg += d + if not d: + close_session(session_id) + break + + while True: + s = msg.find('\n') + if s ==-1: + break + else: + c = msg[0:s] + msg = msg[s+1:] + c = json.loads(c) + try: + cmd = c['method'] + data = c['params'] + except: + print "syntax error", repr(c), ipaddr + continue + + # add to queue + input_queue.put((session_id, cmd, data)) + + +# read commands from the input queue. perform requests, etc. this should be called from the main thread. +def process_input_queue(): + while not stopping: + session_id, cmd, data = input_queue.get() + if session_id not in sessions.keys(): + continue + out = None + if cmd == 'address.subscribe': + subscribe_to_address(session_id,data) + elif cmd == 'numblocks.subscribe': + subscribe_to_numblocks(session_id) + elif cmd == 'client.version': + sessions[session_id]['version'] = data + elif cmd == 'server.banner': + out = json.dumps( { 'method':'server.banner', 'result':config.get('server','banner').replace('\\n','\n') } ) + elif cmd == 'server.peers': + out = json.dumps( { 'method':'server.peers', 'result':peer_list.values() } ) + elif cmd == 'address.get_history': + address = data + out = json.dumps( { 'method':'address.get_history', 'address':address, 'result':store.get_history( address ) } ) + elif cmd == 'transaction.broadcast': + txo = send_tx(data) + print "sent tx:", txo + out = json.dumps( { 'method':'transaction.broadcast', 'result':txo } ) + else: + print "unknown command", cmd if out: - #print ipaddr, cmd, len(out) + output_queue.put((session_id, out)) + +# this is a separate thread +def process_output_queue(): + while not stopping: + session_id, out = output_queue.get() + session = sessions.get(session_id) + if session: try: - conn.send(out) + conn = session.get('conn') + conn.send(out+'\n') except: - print "error, could not send" - - finally: - conn.close() - + close_session(session_id) + -ds = BCDataStream.BCDataStream() +#################################################################### def memorypool_update(store): + ds = BCDataStream.BCDataStream() store.mempool_keys = [] - conn = bitcoinrpc.connect_to_local() - try: - v = conn.getmemorypool() - except: - print "cannot contact bitcoin daemon" + + postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'}) + respdata = urllib.urlopen(bitcoind_url, postdata).read() + r = loads(respdata) + if r['error'] != None: return - v = v['transactions'] + + v = r['result'].get('transactions') for hextx in v: ds.clear() ds.write(hextx.decode('hex')) tx = deserialize.parse_Transaction(ds) tx['hash'] = util.double_sha256(tx['tx']) - store.mempool_keys.append(tx['hash'][::-1].encode('hex')) + tx_hash = tx['hash'][::-1].encode('hex') + store.mempool_keys.append(tx_hash) if store.tx_find_id_and_value(tx): pass - else: + else: store.import_tx(tx, False) store.commit() @@ -528,10 +797,11 @@ def clean_session_thread(): time.sleep(30) t = time.time() for k,s in sessions.items(): + if s.get('type') == 'subscribe': continue t0 = s['last_time'] if t - t0 > 5*60: - print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "end session", s['ip'] sessions.pop(k) + print "lost session", k def irc_thread(): @@ -576,37 +846,54 @@ def irc_thread(): s.close() + +def http_server_thread(store): + # see http://code.google.com/p/jsonrpclib/ + from SocketServer import ThreadingMixIn + from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer + class SimpleThreadedJSONRPCServer(ThreadingMixIn, SimpleJSONRPCServer): pass + server = SimpleThreadedJSONRPCServer(( config.get('server','host'), 8081)) + server.register_function(lambda : peer_list.values(), 'peers') + server.register_function(cmd_stop, 'stop') + server.register_function(cmd_load, 'load') + server.register_function(lambda : block_number, 'blocks') + server.register_function(clear_cache, 'clear_cache') + server.register_function(get_cache, 'get_cache') + server.register_function(send_tx, 'blockchain.transaction.broadcast') + server.register_function(store.get_history, 'blockchain.address.get_history') + server.register_function(new_session, 'session.new') + server.register_function(update_session, 'session.update') + server.register_function(poll_session, 'session.poll') + server.serve_forever() + + import traceback if __name__ == '__main__': if len(sys.argv)>1: + import jsonrpclib + server = jsonrpclib.Server('http://%s:8081'%config.get('server','host')) cmd = sys.argv[1] if cmd == 'load': - request = "('load','%s')#"%config.get('server','password') + out = server.load(password) elif cmd == 'peers': - request = "('peers','')#" + out = server.peers() elif cmd == 'stop': - request = "('stop','%s')#"%config.get('server','password') + out = server.stop(password) elif cmd == 'clear_cache': - request = "('clear_cache','%s')#"%config.get('server','password') + out = server.clear_cache(password) elif cmd == 'get_cache': - request = "('get_cache',('%s','%s'))#"%(config.get('server','password'),sys.argv[2]) + out = server.get_cache(password,sys.argv[2]) elif cmd == 'h': - request = "('h','%s')#"%sys.argv[2] + out = server.blockchain.address.get_history(sys.argv[2]) + elif cmd == 'tx': + out = server.blockchain.transaction.broadcast(sys.argv[2]) elif cmd == 'b': - request = "('b','')#" - - s = socket.socket( socket.AF_INET, socket.SOCK_STREAM) - s.connect((config.get('server','host'), config.getint('server','port'))) - s.send( request ) - out = '' - while 1: - msg = s.recv(1024) - if msg: out += msg - else: break - s.close() + out = server.blocks() + else: + out = "Unknown command: '%s'" % cmd print out sys.exit(0) @@ -627,8 +914,13 @@ if __name__ == '__main__': store.tx_cache = {} store.mempool_keys = {} - thread.start_new_thread(listen_thread, (store,)) + # supported protocols + thread.start_new_thread(native_server_thread, ()) + thread.start_new_thread(tcp_server_thread, ()) + thread.start_new_thread(http_server_thread, (store,)) + thread.start_new_thread(clean_session_thread, ()) + if (config.get('server','irc') == 'yes' ): thread.start_new_thread(irc_thread, ()) @@ -638,9 +930,29 @@ if __name__ == '__main__': store.catch_up() memorypool_update(store) block_number = store.get_block_number(1) - dblock.release() + + if block_number != old_block_number: + old_block_number = block_number + for session_id in sessions_sub_numblocks: + send_numblocks(session_id) + + except IOError: + print "IOError: cannot reach bitcoind" + block_number = 0 except: traceback.print_exc(file=sys.stdout) + block_number = 0 + finally: + dblock.release() + + # do addresses + while True: + try: + addr = address_queue.get(False) + except: + break + do_update_address(addr) + time.sleep(10) print "server stopped"