X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=973f932f8edca0f92548ff6975da24843491c1ea;hb=2b521da6a9f9fb386706d86a3d412986c58eba18;hp=cd70018aff95cd9b19219192def0ff20642f825b;hpb=e4ca4884db0857d4a63b324807f0c5639d9459e1;p=electrum-server.git diff --git a/server.py b/server.py index cd70018..973f932 100755 --- a/server.py +++ b/server.py @@ -27,13 +27,20 @@ Todo: import time, 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 @@ -57,6 +64,7 @@ except: print "Could not read electrum.conf. I will use the default values." 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 @@ -317,16 +325,124 @@ 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" + 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 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 + + # 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 ) + 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 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 listen_thread(store): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -336,9 +452,7 @@ def listen_thread(store): conn, addr = s.accept() thread.start_new_thread(client_thread, (addr, conn,)) -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): #print "client thread", ipaddr @@ -370,16 +484,16 @@ def client_thread(ipaddr,conn): finally: conn.close() - def do_command(cmd, data, ipaddr): + timestr = time.strftime("[%d/%m/%Y-%H:%M:%S]") + if cmd=='b': out = "%d"%block_number elif cmd in ['session','new_session']: - session_id = random_string(10) try: if cmd == 'session': addresses = ast.literal_eval(data) @@ -390,14 +504,8 @@ 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 - - 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() + print timestr, "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version + out = new_session(version, addresses) elif cmd=='update_session': try: @@ -405,14 +513,8 @@ 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) - - sessions[session_id]['addresses'] = {} - for a in addresses: - sessions[session_id]['addresses'][a] = '' - out = 'ok' - sessions[session_id]['last_time'] = time.time() + 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 @@ -421,7 +523,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 @@ -429,7 +531,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: @@ -442,7 +544,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: @@ -462,41 +564,7 @@ def do_command(cmd, data, ipaddr): out = '' 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_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 ) - 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 + out = poll_session(data) elif cmd == 'h': # history @@ -504,43 +572,14 @@ def do_command(cmd, data, ipaddr): out = repr( store.get_history( address ) ) elif cmd == 'load': - if password == data: - out = repr( len(sessions) ) - else: - out = 'wrong password' + out = cmd_load(data) elif cmd =='tx': out = send_tx(data) - print "sent tx:", out - - elif cmd =='clear_cache': - if password == data: - store.tx_cache = {} - out = 'ok' - else: - out = 'wrong password' - - elif cmd =='get_cache': - try: - pw, addr = data - except: - addr = None - if addr: - if password == pw: - out = store.tx_cache.get(addr) - out = repr(out) - else: - out = 'wrong password' - else: - out = "error: "+ repr(data) + print timestr, "sent tx:", ipaddr, out elif cmd == 'stop': - global stopping - if password == data: - stopping = True - out = 'ok' - else: - out = 'wrong password' + out = cmd_stop(data) elif cmd == 'peers': out = repr(peer_list.values()) @@ -553,18 +592,17 @@ def do_command(cmd, data, ipaddr): - - 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')) @@ -588,7 +626,6 @@ def clean_session_thread(): for k,s in sessions.items(): 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) @@ -634,37 +671,52 @@ def irc_thread(): s.close() + +def jsonrpc_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')#"%password + out = server.load(password) elif cmd == 'peers': - request = "('peers','')#" + out = server.peers() elif cmd == 'stop': - request = "('stop','%s')#"%password + out = server.stop(password) elif cmd == 'clear_cache': - request = "('clear_cache','%s')#"%password + out = server.clear_cache(password) elif cmd == 'get_cache': - request = "('get_cache',('%s','%s'))#"%(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() print out sys.exit(0) @@ -686,6 +738,7 @@ if __name__ == '__main__': store.mempool_keys = {} thread.start_new_thread(listen_thread, (store,)) + thread.start_new_thread(jsonrpc_thread, (store,)) thread.start_new_thread(clean_session_thread, ()) if (config.get('server','irc') == 'yes' ): thread.start_new_thread(irc_thread, ()) @@ -696,9 +749,14 @@ if __name__ == '__main__': store.catch_up() memorypool_update(store) block_number = store.get_block_number(1) - dblock.release() + except IOError: + print "IOError: cannot reach bitcoind" + block_number = -1 except: traceback.print_exc(file=sys.stdout) + block_number = -1 + finally: + dblock.release() time.sleep(10) print "server stopped"