X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=5d1b1e01b031ec7f7c1ff2646188b62a5073b507;hb=b328090bcd9b05790a638f2a99ef7fcc5056a6c8;hp=117d61ad8823195d90addf1746e96f8f1fc93672;hpb=4070854996c87ea805b8312255b0b836af606fa0;p=electrum-server.git diff --git a/server.py b/server.py index 117d61a..5d1b1e0 100755 --- a/server.py +++ b/server.py @@ -25,18 +25,367 @@ Todo: """ -import time, json, socket, operator, thread, ast, sys,re +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 psycopg2, binascii +import thread, traceback, sys, urllib, operator +from json import dumps, loads + + +class MyStore(Datastore_class): + + def __init__(self, config): + conf = DataStore.CONFIG_DEFAULTS + args, argv = readconf.parse_argv( [], conf) + args.dbtype = config.get('database','type') + if args.dbtype == 'sqlite3': + args.connect_args = { 'database' : config.get('database','database') } + elif args.dbtype == 'MySQLdb': + args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') } + elif args.dbtype == 'psycopg2': + args.connect_args = { 'database' : config.get('database','database') } + + Datastore_class.__init__(self,args) + + self.tx_cache = {} + self.mempool_keys = {} + self.bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port')) + + self.address_queue = Queue() + + self.dblock = thread.allocate_lock() + + + + def import_block(self, b, chain_ids=frozenset()): + block_id = super(MyStore, self).import_block(b, chain_ids) + 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) + for row in inrows: + _hash = self.binout(row[6]) + address = hash_to_address(chr(0), _hash) + if self.tx_cache.has_key(address): + print "cache: invalidating", address + self.tx_cache.pop(address) + self.address_queue.put(address) + + outrows = self.get_tx_outputs(txid, False) + for row in outrows: + _hash = self.binout(row[6]) + address = hash_to_address(chr(0), _hash) + if self.tx_cache.has_key(address): + print "cache: invalidating", address + self.tx_cache.pop(address) + self.address_queue.put(address) + + def safe_sql(self,sql, params=(), lock=True): + try: + if lock: self.dblock.acquire() + ret = self.selectall(sql,params) + if lock: self.dblock.release() + return ret + except: + print "sql error", sql + return [] + + def get_tx_outputs(self, tx_id, lock=True): + return self.safe_sql("""SELECT + txout.txout_pos, + txout.txout_scriptPubKey, + txout.txout_value, + nexttx.tx_hash, + nexttx.tx_id, + txin.txin_pos, + pubkey.pubkey_hash + FROM txout + LEFT JOIN txin ON (txin.txout_id = txout.txout_id) + LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id) + LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id) + WHERE txout.tx_id = %d + ORDER BY txout.txout_pos + """%(tx_id), (), lock) + + def get_tx_inputs(self, tx_id, lock=True): + return self.safe_sql(""" SELECT + txin.txin_pos, + txin.txin_scriptSig, + txout.txout_value, + COALESCE(prevtx.tx_hash, u.txout_tx_hash), + prevtx.tx_id, + COALESCE(txout.txout_pos, u.txout_pos), + pubkey.pubkey_hash + FROM txin + LEFT JOIN txout ON (txout.txout_id = txin.txout_id) + LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id) + LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id) + LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id) + WHERE txin.tx_id = %d + ORDER BY txin.txin_pos + """%(tx_id,), (), lock) + + def get_address_out_rows(self, dbhash): + return self.safe_sql(""" SELECT + b.block_nTime, + cc.chain_id, + b.block_height, + 1, + b.block_hash, + tx.tx_hash, + tx.tx_id, + txin.txin_pos, + -prevout.txout_value + FROM chain_candidate cc + JOIN block b ON (b.block_id = cc.block_id) + JOIN block_tx ON (block_tx.block_id = b.block_id) + JOIN tx ON (tx.tx_id = block_tx.tx_id) + JOIN txin ON (txin.tx_id = tx.tx_id) + JOIN txout prevout ON (txin.txout_id = prevout.txout_id) + JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id) + WHERE pubkey.pubkey_hash = ? + AND cc.in_longest = 1""", (dbhash,)) + + def get_address_out_rows_memorypool(self, dbhash): + return self.safe_sql(""" SELECT + 1, + tx.tx_hash, + tx.tx_id, + txin.txin_pos, + -prevout.txout_value + FROM tx + JOIN txin ON (txin.tx_id = tx.tx_id) + JOIN txout prevout ON (txin.txout_id = prevout.txout_id) + JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id) + WHERE pubkey.pubkey_hash = ? """, (dbhash,)) + + def get_address_in_rows(self, dbhash): + return self.safe_sql(""" SELECT + b.block_nTime, + cc.chain_id, + b.block_height, + 0, + b.block_hash, + tx.tx_hash, + tx.tx_id, + txout.txout_pos, + txout.txout_value + FROM chain_candidate cc + JOIN block b ON (b.block_id = cc.block_id) + JOIN block_tx ON (block_tx.block_id = b.block_id) + JOIN tx ON (tx.tx_id = block_tx.tx_id) + JOIN txout ON (txout.tx_id = tx.tx_id) + JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id) + WHERE pubkey.pubkey_hash = ? + AND cc.in_longest = 1""", (dbhash,)) + + def get_address_in_rows_memorypool(self, dbhash): + return self.safe_sql( """ SELECT + 0, + tx.tx_hash, + tx.tx_id, + txout.txout_pos, + txout.txout_value + FROM tx + JOIN txout ON (txout.tx_id = tx.tx_id) + JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id) + WHERE pubkey.pubkey_hash = ? """, (dbhash,)) + + def get_history(self, addr): + + 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 None + + dbhash = self.binin(binaddr) + rows = [] + rows += self.get_address_out_rows( dbhash ) + rows += self.get_address_in_rows( dbhash ) + + txpoints = [] + known_tx = [] + + for row in rows: + try: + nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row + except: + print "cannot unpack row", row + break + tx_hash = self.hashout_hex(tx_hash) + txpoint = { + "nTime": int(nTime), + "height": int(height), + "is_in": int(is_in), + "blk_hash": self.hashout_hex(blk_hash), + "tx_hash": tx_hash, + "tx_id": int(tx_id), + "pos": int(pos), + "value": int(value), + } + + txpoints.append(txpoint) + known_tx.append(self.hashout_hex(tx_hash)) + + + # todo: sort them really... + txpoints = sorted(txpoints, key=operator.itemgetter("nTime")) + + # read memory pool + rows = [] + rows += self.get_address_in_rows_memorypool( dbhash ) + rows += self.get_address_out_rows_memorypool( dbhash ) + address_has_mempool = False + + for row in rows: + is_in, tx_hash, tx_id, pos, value = row + 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 + + #print "mempool", tx_hash + txpoint = { + "nTime": 0, + "height": 0, + "is_in": int(is_in), + "blk_hash": 'mempool', + "tx_hash": tx_hash, + "tx_id": int(tx_id), + "pos": int(pos), + "value": int(value), + } + txpoints.append(txpoint) + + + for txpoint in txpoints: + tx_id = txpoint['tx_id'] + + txinputs = [] + inrows = self.get_tx_inputs(tx_id) + for row in inrows: + _hash = self.binout(row[6]) + address = hash_to_address(chr(0), _hash) + txinputs.append(address) + txpoint['inputs'] = txinputs + txoutputs = [] + outrows = self.get_tx_outputs(tx_id) + for row in outrows: + _hash = self.binout(row[6]) + address = hash_to_address(chr(0), _hash) + txoutputs.append(address) + txpoint['outputs'] = txoutputs + + # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address) + if not txpoint['is_in']: + # detect if already redeemed... + for row in outrows: + if row[6] == dbhash: break + else: + raise + #row = self.get_tx_output(tx_id,dbhash) + # pos, script, value, o_hash, o_id, o_pos, binaddr = row + # if not redeemed, we add the script + if row: + if not row[4]: txpoint['raw_scriptPubKey'] = row[1] + + # cache result + if not address_has_mempool: + self.tx_cache[addr] = txpoints + + return txpoints + + + + def memorypool_update(store): + + ds = BCDataStream.BCDataStream() + previous_transactions = store.mempool_keys + store.mempool_keys = [] + + postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'}) + + respdata = urllib.urlopen(store.bitcoind_url, postdata).read() + r = loads(respdata) + if r['error'] != None: + return + + 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']) + tx_hash = store.hashin(tx['hash']) + + store.mempool_keys.append(tx_hash) + if store.tx_find_id_and_value(tx): + pass + else: + tx_id = store.import_tx(tx, False) + store.update_tx_cache(tx_id) + + store.commit() + + + def send_tx(self,tx): + postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'}) + respdata = urllib.urlopen(self.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 main_iteration(store): + try: + store.dblock.acquire() + store.catch_up() + store.memorypool_update() + block_number = store.get_block_number(1) + + except IOError: + print "IOError: cannot reach bitcoind" + block_number = 0 + except: + traceback.print_exc(file=sys.stdout) + block_number = 0 + finally: + store.dblock.release() + + return block_number + + + +import time, json, socket, operator, thread, ast, sys, re, traceback 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 @@ -70,7 +419,6 @@ password = config.get('server','password') stopping = False block_number = -1 -old_block_number = -1 sessions = {} sessions_sub_numblocks = {} # sessions that have subscribed to the service @@ -83,37 +431,7 @@ wallets = {} # for ultra-light clients such as bccapi from Queue import Queue input_queue = Queue() output_queue = Queue() -address_queue = Queue() - - - - - -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) - if r['error'] != None: - out = "error: transaction rejected by memorypool\n"+tx - else: - out = r['result'] - return out @@ -141,23 +459,24 @@ def cmd_load(_,__,pw): -def modified_addresses(session): - if 1: - t1 = time.time() - addresses = session['addresses'] - session['last_time'] = time.time() - ret = {} - k = 0 - for addr in addresses: - status = get_address_status( addr ) - msg_id, last_status = addresses.get( addr ) - if last_status != status: - addresses[addr] = msg_id, status - ret[addr] = status +def modified_addresses(a_session): + #t1 = time.time() + import copy + session = copy.deepcopy(a_session) + addresses = session['addresses'] + session['last_time'] = time.time() + ret = {} + k = 0 + for addr in addresses: + status = get_address_status( addr ) + msg_id, last_status = addresses.get( addr ) + if last_status != status: + addresses[addr] = msg_id, status + ret[addr] = status - t2 = time.time() - t1 - #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 - return ret, addresses + #t2 = time.time() - t1 + #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 + return ret, addresses def poll_session(session_id): @@ -167,6 +486,7 @@ def poll_session(session_id): print time.asctime(), "session not found", session_id return -1, {} else: + sessions[session_id]['last_time'] = time.time() ret, addresses = modified_addresses(session) if ret: sessions[session_id]['addresses'] = addresses return repr( (block_number,ret)) @@ -177,6 +497,7 @@ def poll_session_json(session_id, message_id): if session is None: raise BaseException("session not found %s"%session_id) else: + m_sessions[0][session_id]['last_time'] = time.time() out = [] ret, addresses = modified_addresses(session) if ret: @@ -263,7 +584,7 @@ def add_address_to_session_json(session_id, message_id, address): def add_address_to_session(session_id, address): status = get_address_status(address) - sessions[session_id]['addresses'][addr] = ("", status) + sessions[session_id]['addresses'][address] = ("", status) sessions[session_id]['last_time'] = time.time() return status @@ -378,7 +699,8 @@ def do_command(cmd, data, ipaddr): try: session_id, addr = ast.literal_eval(data) except: - print "error" + traceback.print_exc(file=sys.stdout) + print data return None out = add_address_to_session(session_id,addr) @@ -386,57 +708,10 @@ def do_command(cmd, data, ipaddr): try: session_id, addresses = ast.literal_eval(data) except: - print "error" + traceback.print_exc(file=sys.stdout) 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 == 'bccapi_getSendCoinForm': - out = '' - - elif cmd == 'bccapi_submitTransaction': - out = '' elif cmd=='poll': out = poll_session(data) @@ -447,10 +722,10 @@ def do_command(cmd, data, ipaddr): out = repr( store.get_history( address ) ) elif cmd == 'load': - out = cmd_load(data) + out = cmd_load(None,None,data) elif cmd =='tx': - out = send_tx(data) + out = store.send_tx(data) print timestr(), "sent tx:", ipaddr, out elif cmd == 'stop': @@ -566,10 +841,9 @@ def process_input_queue(): 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() + txo = store.send_tx(data[0]) print "sent tx:", txo - out = json.loads(txo) + out = {'result':txo } else: print "unknown command", method if out: @@ -664,7 +938,7 @@ def http_server_thread(): server.register_function(cmd_stop, 'stop') server.register_function(cmd_load, 'load') server.register_function(get_banner, 'server.banner') - server.register_function(lambda a,b,c: send_tx(c), 'transaction.broadcast') + server.register_function(lambda a,b,c: store.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') @@ -674,9 +948,6 @@ def http_server_thread(): server.serve_forever() -import traceback - - if __name__ == '__main__': if len(sys.argv)>1: @@ -704,11 +975,9 @@ if __name__ == '__main__': print out sys.exit(0) - # backend - import db - store = db.MyStore(config,address_queue) - + # from db import MyStore + store = MyStore(config) # supported protocols thread.start_new_thread(native_server_thread, ()) @@ -721,7 +990,7 @@ if __name__ == '__main__': print "starting Electrum server" - + old_block_number = None while not stopping: block_number = store.main_iteration() @@ -729,10 +998,9 @@ if __name__ == '__main__': old_block_number = block_number for session_id in sessions_sub_numblocks.keys(): send_numblocks(session_id) - # do addresses while True: try: - addr = address_queue.get(False) + addr = store.address_queue.get(False) except: break do_update_address(addr)