X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=e0edb78244c06ed7eb8624b88bb094f83333f215;hb=ba0b0d9f23ec14a0a3552cc8146f0ebce77eefbc;hp=24ef4367a95b856c7e8bd4bc7f8837858ec63f35;hpb=c0505efec333e8d7ad12067dbe12b74f787d07b8;p=electrum-server.git diff --git a/server.py b/server.py index 24ef436..e0edb78 100755 --- a/server.py +++ b/server.py @@ -25,7 +25,7 @@ Todo: """ -import time, socket, operator, thread, ast, sys,re +import time, json, socket, operator, thread, ast, sys,re import psycopg2, binascii from Abe.abe import hash_to_address, decode_check_address @@ -36,6 +36,12 @@ 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 config.add_section('server') @@ -44,7 +50,6 @@ config.set('server', 'host', 'localhost') config.set('server', 'port', 50000) config.set('server', 'password', '') config.set('server', 'irc', 'yes') -config.set('server', 'cache', 'no') config.set('server', 'ircname', 'Electrum server') config.add_section('database') config.set('database', 'type', 'psycopg2') @@ -57,22 +62,52 @@ 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 + +m_sessions = [{}] # served by http + 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): - def import_tx(self, tx, is_coinbase): - tx_id = super(MyStore, self).import_tx(tx, is_coinbase) - if config.get('server', 'cache') == 'yes': self.update_tx_cache(tx_id) + def import_block(self, b, chain_ids=frozenset()): + block_id = super(MyStore, self).import_block(b, chain_ids) + #print "block", block_id + for pos in xrange(len(b['transactions'])): + tx = b['transactions'][pos] + if 'hash' not in tx: + tx['hash'] = util.double_sha256(tx['tx']) + tx_id = store.tx_find_id_and_value(tx) + if tx_id: + self.update_tx_cache(tx_id) + else: + print "error: import_block: no tx_id" + return block_id + def update_tx_cache(self, txid): inrows = self.get_tx_inputs(txid, False) @@ -82,6 +117,8 @@ class MyStore(Datastore_class): if self.tx_cache.has_key(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]) @@ -89,6 +126,7 @@ class MyStore(Datastore_class): if self.tx_cache.has_key(address): print "cache: invalidating", address self.tx_cache.pop(address) + address_queue.put(address) def safe_sql(self,sql, params=(), lock=True): try: @@ -203,10 +241,9 @@ class MyStore(Datastore_class): def get_history(self, addr): - if config.get('server','cache') == 'yes': - cached_version = self.tx_cache.get( addr ) - if cached_version is not None: - return cached_version + 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: @@ -229,7 +266,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), @@ -268,7 +304,6 @@ class MyStore(Datastore_class): #print "mempool", tx_hash txpoint = { "nTime": 0, - #"chain_id": 1, "height": 0, "is_in": int(is_in), "blk_hash": 'mempool', @@ -312,19 +347,38 @@ class MyStore(Datastore_class): if not row[4]: txpoint['raw_scriptPubKey'] = row[1] # cache result - if config.get('server','cache') == 'yes' and not address_has_mempool: + if not address_has_mempool: self.tx_cache[addr] = txpoints return txpoints +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, + 'transaction.broadcast':send_tx, + 'address.get_history':store.get_history + } + func = cmds[method] + return func( params ) + + def send_tx(tx): postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'}) respdata = urllib.urlopen(bitcoind_url, postdata).read() r = loads(respdata) - return r + if r['error'] != None: + out = "error: transaction rejected by memorypool\n"+tx + else: + out = r['result'] + return out @@ -334,41 +388,39 @@ def random_string(N): -def cmd_stop(data): +def cmd_stop(_,__,pw): global stopping - if password == data: + if password == pw: stopping = True return 'ok' else: return 'wrong password' -def cmd_load(pw): +def cmd_load(_,__,pw): if password == pw: return repr( len(sessions) ) else: return 'wrong password' -def clear_cache(pw): +def clear_cache(_,__,pw): if password == pw: store.tx_cache = {} return 'ok' else: return 'wrong password' -def get_cache(pw,addr): +def get_cache(_,__,pw,addr): if password == pw: return store.tx_cache.get(addr) else: return 'wrong password' -def cmd_poll(session_id): - session = sessions.get(session_id) - if session is None: - print time.asctime(), "session not found", session_id - out = repr( (-1, {})) - else: + + +def modified_addresses(session): + if 1: t1 = time.time() addresses = session['addresses'] session['last_time'] = time.time() @@ -376,60 +428,178 @@ def cmd_poll(session_id): 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_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) - - last_status = addresses.get( addr ) + status = get_address_status( addr ) + msg_id, last_status = addresses.get( addr ) if last_status != status: - addresses[addr] = status + addresses[addr] = msg_id, 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 + #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 + return ret, addresses + + +def poll_session(session_id): + # native + session = sessions.get(session_id) + if session is None: + print time.asctime(), "session not found", session_id + return -1, {} + else: + ret, addresses = modified_addresses(session) + if ret: sessions[session_id]['addresses'] = addresses + return repr( (block_number,ret)) + + +def poll_session_json(session_id, message_id): + session = m_sessions[0].get(session_id) + if session is None: + raise BaseException("session not found %s"%session_id) + else: + print "poll: session found", session_id + out = [] + ret, addresses = modified_addresses(session) + if ret: + m_sessions[0][session_id]['addresses'] = addresses + for addr in ret: + msg_id, status = addresses[addr] + out.append( { 'id':msg_id, 'result':status } ) + + msg_id, last_nb = session.get('numblocks') + if last_nb: + if last_nb != block_number: + m_sessions[0][session_id]['numblocks'] = msg_id, block_number + out.append( {'id':msg_id, 'result':block_number} ) return out -def new_session(addresses, version): +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') != 'persistent': continue + addresses = session['addresses'].keys() + + if addr in addresses: + status = get_address_status( addr ) + message_id, last_status = session['addresses'][addr] + if last_status != status: + #print "sending new status for %s:"%addr, status + send_status(session_id,message_id,addr,status) + sessions[session_id]['addresses'][addr] = (message_id,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): + message_id = sessions_sub_numblocks[session_id] + out = json.dumps( {'id':message_id, 'result':block_number} ) + output_queue.put((session_id, out)) + +def send_status(session_id, message_id, address, status): + out = json.dumps( { 'id':message_id, 'result':status } ) + output_queue.put((session_id, out)) + +def address_get_history_json(_,message_id,address): + return store.get_history(address) + +def subscribe_to_numblocks(session_id, message_id): + sessions_sub_numblocks[session_id] = message_id + send_numblocks(session_id) + +def subscribe_to_numblocks_json(session_id, message_id): + global m_sessions + m_sessions[0][session_id]['numblocks'] = message_id,block_number + return block_number + +def subscribe_to_address(session_id, message_id, address): + status = get_address_status(address) + sessions[session_id]['addresses'][address] = (message_id, status) + sessions[session_id]['last_time'] = time.time() + send_status(session_id, message_id, address, status) + +def add_address_to_session_json(session_id, message_id, address): + global m_sessions + sessions = m_sessions[0] + status = get_address_status(address) + sessions[session_id]['addresses'][address] = (message_id, status) + sessions[session_id]['last_time'] = time.time() + m_sessions[0] = sessions + return status + +def add_address_to_session(session_id, address): + status = get_address_status(address) + sessions[session_id]['addresses'][addr] = ("", status) + sessions[session_id]['last_time'] = time.time() + return 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] = '' + 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 client_version_json(session_id, _, version): + global m_sessions + sessions = m_sessions[0] + sessions[session_id]['version'] = version + m_sessions[0] = sessions + +def create_session_json(_, __): + sessions = m_sessions[0] + session_id = random_string(10) + print "creating session", session_id + sessions[session_id] = { 'addresses':{}, 'numblocks':('','') } + sessions[session_id]['last_time'] = time.time() + m_sessions[0] = sessions + return session_id + + + +def get_banner(_,__): + return config.get('server','banner').replace('\\n','\n') + def update_session(session_id,addresses): + """deprecated in 0.42""" sessions[session_id]['addresses'] = {} for a in addresses: sessions[session_id]['addresses'][a] = '' sessions[session_id]['last_time'] = time.time() return 'ok' -def listen_thread(store): +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 client_thread(ipaddr,conn): +def native_client_thread(ipaddr,conn): #print "client thread", ipaddr try: ipaddr = ipaddr[0] @@ -461,6 +631,10 @@ def client_thread(ipaddr,conn): conn.close() +def timestr(): + return time.strftime("[%d/%m/%Y-%H:%M:%S]") + +# used by the native handler def do_command(cmd, data, ipaddr): if cmd=='b': @@ -477,8 +651,16 @@ def do_command(cmd, data, ipaddr): except: print "error", data return None - print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version - out = new_session(addresses, version) + print timestr(), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version + out = new_session(version, addresses) + + elif cmd=='address.subscribe': + try: + session_id, addr = ast.literal_eval(data) + except: + print "error" + return None + out = add_address_to_session(session_id,addr) elif cmd=='update_session': try: @@ -486,7 +668,7 @@ def do_command(cmd, data, ipaddr): except: print "error" return None - print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses) + print timestr(), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses) out = update_session(session_id,addresses) elif cmd == 'bccapi_login': @@ -496,7 +678,7 @@ def do_command(cmd, data, ipaddr): master_public_key = k.decode('hex') # todo: sanitize. no need to decode twice... print master_public_key wallet_id = random_string(10) - w = electrum.Wallet() + w = Wallet( Direct_Interface() ) w.master_public_key = master_public_key.decode('hex') w.synchronize() wallets[wallet_id] = w @@ -504,7 +686,7 @@ def do_command(cmd, data, ipaddr): print "wallets", wallets elif cmd == 'bccapi_getAccountInfo': - from electrum import int_to_hex + from wallet import int_to_hex v, wallet_id = ast.literal_eval(data) w = wallets.get(wallet_id) if w is not None: @@ -517,7 +699,7 @@ def do_command(cmd, data, ipaddr): out = "error" elif cmd == 'bccapi_getAccountStatement': - from electrum import int_to_hex + from wallet import int_to_hex v, wallet_id = ast.literal_eval(data) w = wallets.get(wallet_id) if w is not None: @@ -537,7 +719,7 @@ def do_command(cmd, data, ipaddr): out = '' elif cmd=='poll': - out = cmd_poll(data) + out = poll_session(data) elif cmd == 'h': # history @@ -548,12 +730,8 @@ def do_command(cmd, data, ipaddr): out = cmd_load(data) elif cmd =='tx': - r = send_tx(data) - if r['error'] != None: - out = "error: transaction rejected by memorypool" - else: - out = r['result'] - print "sent tx:", out + out = send_tx(data) + print timestr(), "sent tx:", ipaddr, out elif cmd == 'stop': out = cmd_stop(data) @@ -568,9 +746,139 @@ def do_command(cmd, data, ipaddr): +#################################################################### + +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.pop(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 timestr(), "TCP session", ipaddr + global sessions + + session_id = random_string(10) + sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown', 'type':'persistent' } + + ipaddr = ipaddr[0] + msg = '' + + while not stopping: + try: + d = conn.recv(1024) + except socket.error: + d = '' + if not d: + close_session(session_id) + break + + msg += d + while True: + s = msg.find('\n') + if s ==-1: + break + else: + c = msg[0:s].strip() + msg = msg[s+1:] + if c == 'quit': + conn.close() + close_session(session_id) + return + try: + c = json.loads(c) + except: + print "json error", repr(c) + continue + try: + message_id = c.get('id') + method = c.get('method') + params = c.get('params') + except: + print "syntax error", repr(c), ipaddr + continue + + # add to queue + input_queue.put((session_id, message_id, method, params)) + + + +# 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, message_id, method, data = input_queue.get() + if session_id not in sessions.keys(): + continue + out = None + if method == 'address.subscribe': + address = data[0] + subscribe_to_address(session_id,message_id,address) + elif method == 'numblocks.subscribe': + subscribe_to_numblocks(session_id,message_id) + elif method == 'client.version': + sessions[session_id]['version'] = data[0] + elif method == 'server.banner': + out = { 'result':config.get('server','banner').replace('\\n','\n') } + elif method == 'server.peers': + out = { 'result':peer_list.values() } + elif method == 'address.get_history': + address = data[0] + out = { 'result':store.get_history( address ) } + elif method == 'transaction.broadcast': + postdata = dumps({"method": 'importtransaction', 'params': [data], 'id':'jsonrpc'}) + txo = urllib.urlopen(bitcoind_url, postdata).read() + print "sent tx:", txo + out = json.loads(txo) + else: + print "unknown command", method + if out: + out['id'] = message_id + out = json.dumps( 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 = session.get('conn') + conn.send(out+'\n') + except: + close_session(session_id) + + + + +#################################################################### + def memorypool_update(store): + ds = BCDataStream.BCDataStream() + previous_transactions = store.mempool_keys store.mempool_keys = [] postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'}) @@ -585,25 +893,28 @@ def memorypool_update(store): ds.write(hextx.decode('hex')) tx = deserialize.parse_Transaction(ds) tx['hash'] = util.double_sha256(tx['tx']) - tx_hash = tx['hash'][::-1].encode('hex') + tx_hash = store.hashin(tx['hash']) + store.mempool_keys.append(tx_hash) if store.tx_find_id_and_value(tx): pass else: - store.import_tx(tx, False) + tx_id = store.import_tx(tx, False) + store.update_tx_cache(tx_id) store.commit() - def clean_session_thread(): while not stopping: time.sleep(30) t = time.time() for k,s in sessions.items(): + if s.get('type') == 'persistent': continue t0 = s['last_time'] if t - t0 > 5*60: sessions.pop(k) + print "lost session", k def irc_thread(): @@ -648,22 +959,28 @@ def irc_thread(): s.close() +def get_peers_json(_,__): + return peer_list.values() -def jsonrpc_thread(store): +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') + from StratumJSONRPCServer import StratumJSONRPCServer + class StratumThreadedJSONRPCServer(ThreadingMixIn, StratumJSONRPCServer): pass + server = StratumThreadedJSONRPCServer(( config.get('server','host'), 8081)) + server.register_function(get_peers_json, 'server.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, 'new_session') + server.register_function(get_banner, 'server.banner') + server.register_function(lambda a,b,c: send_tx(c), 'transaction.broadcast') + server.register_function(address_get_history_json, 'address.get_history') + server.register_function(add_address_to_session_json, 'address.subscribe') + server.register_function(subscribe_to_numblocks_json, 'numblocks.subscribe') + server.register_function(client_version_json, 'client.version') + server.register_function(create_session_json, 'session.create') # internal message (not part of protocol) + server.register_function(poll_session_json, 'session.poll') # internal message (not part of protocol) server.serve_forever() @@ -679,7 +996,7 @@ if __name__ == '__main__': if cmd == 'load': out = server.load(password) elif cmd == 'peers': - out = server.peers() + out = server.server.peers() elif cmd == 'stop': out = server.stop(password) elif cmd == 'clear_cache': @@ -687,17 +1004,18 @@ if __name__ == '__main__': elif cmd == 'get_cache': out = server.get_cache(password,sys.argv[2]) elif cmd == 'h': - out = server.blockchain.address.get_history(sys.argv[2]) + out = server.address.get_history(sys.argv[2]) elif cmd == 'tx': - out = server.blockchain.transaction.broadcast(sys.argv[2]) + out = server.transaction.broadcast(sys.argv[2]) elif cmd == 'b': - out = server.blocks() + out = server.numblocks.subscribe() + else: + out = "Unknown command: '%s'" % cmd print out sys.exit(0) print "starting Electrum server" - print "cache:", config.get('server', 'cache') conf = DataStore.CONFIG_DEFAULTS args, argv = readconf.parse_argv( [], conf) @@ -712,9 +1030,13 @@ if __name__ == '__main__': store.tx_cache = {} store.mempool_keys = {} - thread.start_new_thread(listen_thread, (store,)) - thread.start_new_thread(jsonrpc_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, ()) @@ -723,10 +1045,30 @@ if __name__ == '__main__': dblock.acquire() 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.keys(): + 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"