From 7621dac937eba5352b245972d027d74c39722d5d Mon Sep 17 00:00:00 2001 From: ThomasV Date: Thu, 29 Mar 2012 14:10:07 +0400 Subject: [PATCH] move files in modules dir --- abe_backend.py | 436 --------------------------------------------- begin.py | 8 - irc.py | 86 --------- modules/abe/__init__.py | 443 ++++++++++++++++++++++++++++++++++++++++++++-- modules/abe/__init__.pyc | Bin 1105 -> 14150 bytes modules/irc/__init__.py | 86 +++++++++ server.py | 5 +- 7 files changed, 518 insertions(+), 546 deletions(-) delete mode 100644 abe_backend.py delete mode 100644 begin.py delete mode 100644 irc.py create mode 100644 modules/irc/__init__.py diff --git a/abe_backend.py b/abe_backend.py deleted file mode 100644 index dc7056a..0000000 --- a/abe_backend.py +++ /dev/null @@ -1,436 +0,0 @@ -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 -from Queue import Queue -import time - -class AbeStore(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()): - #print "import block" - block_id = super(AbeStore, 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 = self.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 get_status(self,addr): - # get address status, i.e. the last block for that address. - tx_points = self.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 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 - - -from processor import Processor - -class AbeProcessor(Processor): - - def __init__(self, config): - Processor.__init__(self) - self.store = AbeStore(config) - self.block_number = -1 - self.watched_addresses = [] - - def process(self, request): - message_id = request['id'] - method = request['method'] - params = request.get('params',[]) - result = '' - if method == 'blockchain.numblocks.subscribe': - result = self.block_number - elif method == 'blockchain.address.subscribe': - address = params[0] - self.watch_address(address) - status = self.store.get_status(address) - result = status - elif method == 'blockchain.address.get_history': - address = params[0] - result = self.store.get_history( address ) - elif method == 'blockchain.transaction.broadcast': - txo = self.store.send_tx(params[0]) - print "sent tx:", txo - result = txo - else: - print "unknown method", request - - if result != '': - response = { 'id':message_id, 'method':method, 'params':params, 'result':result } - self.push_response(response) - - - def watch_address(self, addr): - if addr not in self.watched_addresses: - self.watched_addresses.append(addr) - - - def run(self): - - old_block_number = None - while not self.shared.stopped(): - self.block_number = self.store.main_iteration() - - if self.block_number != old_block_number: - old_block_number = self.block_number - self.push_response({ 'method':'blockchain.numblocks.subscribe', 'result':self.block_number }) - - while True: - try: - addr = self.store.address_queue.get(False) - except: - break - if addr in self.watched_addresses: - status = self.get_status( addr ) - self.push_response({ 'method':'blockchain.address.subscribe', 'params':[addr], 'result':status }) - - time.sleep(10) - - - diff --git a/begin.py b/begin.py deleted file mode 100644 index 04eeef7..0000000 --- a/begin.py +++ /dev/null @@ -1,8 +0,0 @@ -import sys -import stratum - -if __name__ == "__main__": - backend = __import__("modules." + sys.argv[1], fromlist=["run"]) - stratum_frontend = stratum.Stratum() - backend.run(stratum_frontend) - diff --git a/irc.py b/irc.py deleted file mode 100644 index 908089a..0000000 --- a/irc.py +++ /dev/null @@ -1,86 +0,0 @@ -#################################################################### - -import threading, socket, traceback, time - -def random_string(N): - import random, string - return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N)) - -from processor import Processor - -class ServerProcessor(Processor): - - def __init__(self, config): - Processor.__init__(self) - self.daemon = True - self.peers = {} - self.banner = config.get('server','banner') - self.host = config.get('server','host') - self.nick = config.get('server','ircname') - self.irc = config.get('server','irc') == 'yes' - - def get_peers(self): - return self.peers.values() - - def run(self): - if not self.irc: - return - NICK = 'E_'+random_string(10) - while not self.shared.stopped(): - try: - s = socket.socket() - s.connect(('irc.freenode.net', 6667)) - s.send('USER electrum 0 * :'+self.host+' '+self.nick+'\n') - s.send('NICK '+NICK+'\n') - s.send('JOIN #electrum\n') - sf = s.makefile('r', 0) - t = 0 - while not self.shared.stopped(): - line = sf.readline() - line = line.rstrip('\r\n') - line = line.split() - if line[0]=='PING': - s.send('PONG '+line[1]+'\n') - elif '353' in line: # answer to /names - k = line.index('353') - for item in line[k+1:]: - if item[0:2] == 'E_': - s.send('WHO %s\n'%item) - elif '352' in line: # answer to /who - # warning: this is a horrible hack which apparently works - k = line.index('352') - ip = line[k+4] - ip = socket.gethostbyname(ip) - name = line[k+6] - host = line[k+9] - self.peers[name] = (ip,host) - if time.time() - t > 5*60: - self.push_response({'method':'server.peers', 'result':[self.get_peers()]}) - s.send('NAMES #electrum\n') - t = time.time() - self.peers = {} - except: - traceback.print_exc(file=sys.stdout) - finally: - sf.close() - s.close() - - - - def process(self, request): - method = request['method'] - - result = '' - if method == 'server.banner': - result = self.banner.replace('\\n','\n') - elif method == 'server.peers.subscribe': - result = self.get_peers() - else: - print "unknown method", request - - if result!='': - response = { 'id':request['id'], 'method':method, 'params':request['params'], 'result':result } - self.push_response(response) - - - diff --git a/modules/abe/__init__.py b/modules/abe/__init__.py index 1a5cc49..dc7056a 100644 --- a/modules/abe/__init__.py +++ b/modules/abe/__init__.py @@ -1,19 +1,436 @@ -import stratum +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 -class AbeProcessor(stratum.Processor): +import psycopg2, binascii - def __init__(self): - stratum.Processor.__init__(self) +import thread, traceback, sys, urllib, operator +from json import dumps, loads +from Queue import Queue +import time - def stop(self): - pass +class AbeStore(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()): + #print "import block" + block_id = super(AbeStore, 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 = self.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 get_status(self,addr): + # get address status, i.e. the last block for that address. + tx_points = self.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 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 + + +from processor import Processor + +class AbeProcessor(Processor): + + def __init__(self, config): + Processor.__init__(self) + self.store = AbeStore(config) + self.block_number = -1 + self.watched_addresses = [] + + def process(self, request): + message_id = request['id'] + method = request['method'] + params = request.get('params',[]) + result = '' + if method == 'blockchain.numblocks.subscribe': + result = self.block_number + elif method == 'blockchain.address.subscribe': + address = params[0] + self.watch_address(address) + status = self.store.get_status(address) + result = status + elif method == 'blockchain.address.get_history': + address = params[0] + result = self.store.get_history( address ) + elif method == 'blockchain.transaction.broadcast': + txo = self.store.send_tx(params[0]) + print "sent tx:", txo + result = txo + else: + print "unknown method", request + + if result != '': + response = { 'id':message_id, 'method':method, 'params':params, 'result':result } + self.push_response(response) + + + def watch_address(self, addr): + if addr not in self.watched_addresses: + self.watched_addresses.append(addr) + + + def run(self): + + old_block_number = None + while not self.shared.stopped(): + self.block_number = self.store.main_iteration() + + if self.block_number != old_block_number: + old_block_number = self.block_number + self.push_response({ 'method':'blockchain.numblocks.subscribe', 'result':self.block_number }) + + while True: + try: + addr = self.store.address_queue.get(False) + except: + break + if addr in self.watched_addresses: + status = self.get_status( addr ) + self.push_response({ 'method':'blockchain.address.subscribe', 'params':[addr], 'result':status }) + + time.sleep(10) - def process(self, session): - request = session.pop_request() - print request - # session.push_response(response) -def run(stratum): - processor = AbeProcessor() - stratum.start(processor) diff --git a/modules/abe/__init__.pyc b/modules/abe/__init__.pyc index e16b48d38612395168fc61f3b9da4159dd023cb1..344ae06e153c362d48681c214d59883a71689c16 100644 GIT binary patch literal 14150 zcmd^GTW=)Cbw1rQ9CC(ta>>0&ibk?m8VhT#ELl>rS~-?T?n=>$D>YZ9G?6lxOE1~v zOwX{phf4|^N6_lUP=LgNkRU)HNDg8r27&+qazPCEArJWt3E-CmK_2qrhrGx6zEj;Z z&C4=g(Mp13mtCx`t~yn9>fGv_!_2>r7an^ex>=X_Uk1Owi6^{Zi8SzINmHa{N!gN? zEoECQw%XFnNGmJlA@iP*W=`L;(i)cXu(U>`JR+^Ul=IRmNVy=bQ7Mm_-dSmmNqJ1+ zqdqR>@uWT><%y&|DdoweJ|*R;q&_X>>7+g*4wHuG>V3X8@hp4ZF=vyQ5Nsr$ZJLfP^*USGrzoI zRy4$(-Bu@zSZlVcjnGWY*EhOu*Nr&Ui#1nQaDV__!iVrP6t3{xap1raj^51>Qy_^C zK@V9{0o#Y9F)40N8X5Tz6avhmp4C1hS|8GSUK%;=Um*55a5Tm~DIu>dAcKNdKm&?$ z62O?Y;Pl6}f}@|%%8WRxr75hkPQ-Cr!*~|)gdLzb64|t*XG>(uW=7UU_HDUq%R9F8 zGSbV+z9pO31oo%x`x&VW3Cc}uC?i|81P$@5R5y#m(f?U?&58d((mSiY|37c)I=(Ii7P<%XtF4vd6=Ez z1>sd^AH0qq{5rwc+gX_>#8Ck?;`CV`muNyZCzDrY%+(TkU`2_l0lFqJDtShG9T%&_ zscbjB$bDMtukEeAaidYg(SV+6I;ab(_N%R!rtzY!ooX1~X$K7@aD8c*_HeJ>?yNr* z&H}>(cyt#7imFtzI zpKzK*4;=Kf+cfTWKZy^@75#s2~`Y`TSGE+rQXEi zbs?3C=X+75qA$^IrCxpULUI^@`P?Vbep&J1Eq?$u5Bt{vV_f*$CaHD8bpim5^9&*o#?XL z#0GoN{y#|3!OdZVGNkS6aEqkvGUeAYtEXq9>%lG-}aQ9 zN9!y0m|Z6}NfexS5l{FD3b1ShhJ!o?%PO53mIc4=?h8%;%LKK81<&6FGg`9rZX@Po zw(V!-d=JcvkwdaIEJ0Hs{5ErZ1xIrXYqnhVm5w)u`Ht2ey5uDIE~h3pfln>5wx6OH zyZ$nMuYtQ(6`d=3NKsJr!)iV9+P*;@Wxo=0v>QHzowz~J4z3hE9BvR*YCMfAMZaCd z*?Wx=WxtB$u-kD1heBP!rn~EL?P4C#9Ff&@eMiro9Dhm(!iouGL^ks=jab{~ zK?Z7k-|0riF1S{0LQJdi;Q8%tJZKP{GJ zK|Wmv0bhT49Vhpe9v?}4C8R8XW-O}2&?9voLzAMtTHo$^fvY?U5e_L_Z8r5wfHx>u z6-kDhxpKf!JVR?$TShRcO5ituIR!dZafQ{}kU87U>zqnC*veQrnF3N-j^GU(%*pr= zrHkx&76lncEl1#Bo&p-2Ns-EqE8!DB#p?2n<;7ct;-P=hE@)vv%L-)Cr4u`Z^}y>y zuXk&&x_c-0RQY(ci|_76(JrP_vOMX#*EsqWbnas;MCq|B?R0D4p#}3xCpfg3YtGHr zpdql~LmzG|U%OTOwVSW3;H&2sZ>|(e@m%N&NnL!w42$vPHNJ9B0A_Xtad15;DDetU z0u${P4X)CQF)JD1@CbmlmzSO8{!&f%i;d#J#olz5mYw2DZ} zN<+#6&K0(;uno`W)ln~D=o@%K60lggiK7&K0Uy#7jeF!#qW%;_UA!k%pGYYKLNXM+ z>a8C~Z3E7@2RSd^eDTKe>f&;#6SzAmx-NEON{I>mD@UOZMjS`!eZYs=vkw^ub{y27 zlLb7#f7(VKq&WcFj;U^dfS7s}tE87Bq*ok3_{Fh+LDGT67@|&N@7}!F^_!l*8* zmP4!r3vX3@7%SmVPvD{2f-zM6TVBgOO2GB{LSoh){g{m1aJ}`7=;-&a9c}6lfu(=+ zYl^uJo&wNpz8%L#13o`iVSJTjG@x;1@R=#f)QqG-ftUUe(CGl$vE$>B2^u&c0YYQ+h_X6X(HR;~VBBLKN`@F` zF;k`l0_Hu4&wI>#vUo^9+)GCq#x!5ksvfid>Y-C9zPPf4vswVVH`}=8g%LV^O`UW` zv5`@XaYy61QKPAR3eBeXL|y?BX95MK8svbUu;|SFvgpT9#P?Ej$FPw(LGI*UO!NgZ z3Mlc=8T_B6wc>Gf}dCR@%+DtHoh3SN{9I|=bjR)KTMH8UP<5qv@kxPK9--r zkz&kfc&XrHp;KGEEbugKk-psdm`Y|vBz|dnvf+bc~2?O3NcuWCch(R;s z8)rKlsOl|$ZqTHzb#!GD`wnY4BD?=WC;HO6KapNu_AbhPUf#(|uOPiq>5WNmTzV7I zo0Q&^^rodZBWW*L9$vE7eM{crr0~ch&V!~L+uy%3K!C|d_hPGFB$WK@D*ON zh^3ibfGxp(KhAJxQyBLntxcmV0L{qN?H^^ff1KI= z$i~mw{uv4XAp->R{4mlok*ZLPjld&-2?+vFmrL^{ry!N7q}hI)BZbR-Nl! z2w%UtZfni0fu}w8`WScFuK=WWR(nHz^YBt?KtVmtaez|846zemqX)wn2f56;M^70d zs~{2)OQ$raBi^&0~WNc0Kc*Umn^-+ z*oaPF?Q~p3ba^8R5nFRjIJk{CS`|T84ca2g+=4WX5L$Z^-JDexS5fHHw}P(gJi;cr zC!7~p;1TB=EM7uU;xf&}i#y|WNrN03TS08i4MGI|6g%lUcEPS{1n5k9w;BP2Xt#qf zG(s+X+49?WeB7X&WzO;~er9vxtZ>%Y{1%J1S(I6*W0NQ4{5lGqIffaLFp>)$a7L9u zGhAWPzrX`>mN@hpi|ZWEn1h-}`tfDFmGFE{hpC4V^%Jc%K%CygMRF3 zN{q9oF`@v-lW1j-XWA-Yw$xjufH|jR#Li@MRzV1R%2q8(^5Y?^7A3;_tOu!MR4BsP z4TzSE_|H*`GSbZK5zT)&&+sDxk0xK_O|2vPXaXGY3ZLaL0PTm5HD}12?7NsPBxzr{ z*f37)9lSVI7EhrV1a}+=gXGlU3RhB628gOrH`L=qszIb*PIHnpCwP*f!sQJzWPXj< zJiNSJ0FuY7vsRsqOCIeb?iyMmm05RzDTr&-6@8G&s8LFqXliyGtk&fXaO{YW&gw| z(@$xWK#n463s@FwNQZ%Thr-LVFMcc}QEWmV8>dW8LFcvW| zILnI3$#{?}m)6n4iC01S%L9OE$Hgs?uKy}!l7 zo4~tRU$)G#jYB}zDH?4&BuhP@k5GAUNA|w1$X9`Z4XCKVF~l{ivCpfR5$(SwGsJ;P$c0OhcwFMPO9GU61 zYDnWF*5YZ5W0*8>WX1!puHQk){^RSeZokn>vv&wRTy=+m4f z&#t9uPp}h_C)4(;JkkfzELKKnltABs`^K#yaAyXF#e_X!n<;5@>GmrXrYs_{kE*j$ zF`03dil)Np_HGC5_go)2HqMjip(4y7R@K9=)}}E-)%aCI^%lRX-KyrmdweBwo!@8i z8H&NSwRZoBdeq&-rBXo7&Yzu}%+F31#-_(+@^ks|{F(e%KFCkzAI^`OoMH0V>p{DY zv}V}Y9E!Z*enTDa`j5m<16o%EQJ&5n23dn$%di1V)-VV?qb);B<0Xmp9b4p|MS`If z1mQ-P266(asUgCYJ@8&pbApvD@0YB=4xkF<_!X0!OZ$2;9aaw#>qZnjOtaBA z0g4tJ5~g9jl`PyGjYZ9rYFdT_0?Uwn#^8xDCKeWH>?>H(dce$OTQN!n8#0OpbP^M^ zN(jA9E%=Z?j5US}VYkLRY0Xs*Pj!jUeb7N=o^r-9y;Ap_u!sp$T~g|7p%x(Dz7CUI zPaY~GLKOIl>Wi+g7eHt^wdLLikP^v5@5Y8KxjiMzgeVQksw7aEKuMq`VyBBgIiMAT zJXE(tqB7T3irm0OvM^Gf+1di*R@Ytl)Ki-IeRjXjVi$#qUt9|NhLM8AhtGR0mqZ0F;IR37=tVb(c@2MrOvkKX?!H>wfok%sl}3-sn;EQF0}0Ot>8~>e?@{rcRx!tSLp1oAg!_A$>%R{;=t~9g~6iI6u;uX*U}Oj2XR& zE5DpVFDR=8UH=nO5MBQQ-^GkIgKKu)dIVX}8S6YA`bH+LOSZ9VzlUzlAF)tP_UEjT z44uD3F<6Y&?q5@VA3%1v9Eu`X1iAbJ`JvK-cu@>P+bLGz-}J92%pH@~u(|N@yc}N9RL;<47S=t_`7)dTki{n~K47tqLQja-T!#xY zwmyl$`3n}bV`3vRwoQz#=Wb4m5?$X_W6*sAn2~TE#XPQO3-&qtDn!+kHD=HMOuQ-k FKLI1n)dm0n literal 1105 zcmbtT!A{#i5S_IHB{ZlU%AqF?Tzo)S;@S$8AOxpSZBFE}3dvF&CARTSq*e+iIQ4h? z0qvV{;&3Hcne~i!_RZTjllbr6?w_yu&4`vyjQ4j~IK?u?&(M@;04E|(G%<8Xo*v>4lE~GN=QD1@Z^-ZtBImL^W9Re) zw?VM(STEtHhB z6nRBbvb4IDjdxV8*>h3Ky;jwQgfMEVA0Sv5nrQQ$pjt#%*RSYEB7ekE7Nx8FU zA(dXOCVH6_x%#UN9N17Wg(>R^k-9--Ak#5ANXsVW!95yX{C~Fc-T(}*QGX7OB?g8ZpE44 KYu`H2m%ab0^4I?W diff --git a/modules/irc/__init__.py b/modules/irc/__init__.py new file mode 100644 index 0000000..908089a --- /dev/null +++ b/modules/irc/__init__.py @@ -0,0 +1,86 @@ +#################################################################### + +import threading, socket, traceback, time + +def random_string(N): + import random, string + return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N)) + +from processor import Processor + +class ServerProcessor(Processor): + + def __init__(self, config): + Processor.__init__(self) + self.daemon = True + self.peers = {} + self.banner = config.get('server','banner') + self.host = config.get('server','host') + self.nick = config.get('server','ircname') + self.irc = config.get('server','irc') == 'yes' + + def get_peers(self): + return self.peers.values() + + def run(self): + if not self.irc: + return + NICK = 'E_'+random_string(10) + while not self.shared.stopped(): + try: + s = socket.socket() + s.connect(('irc.freenode.net', 6667)) + s.send('USER electrum 0 * :'+self.host+' '+self.nick+'\n') + s.send('NICK '+NICK+'\n') + s.send('JOIN #electrum\n') + sf = s.makefile('r', 0) + t = 0 + while not self.shared.stopped(): + line = sf.readline() + line = line.rstrip('\r\n') + line = line.split() + if line[0]=='PING': + s.send('PONG '+line[1]+'\n') + elif '353' in line: # answer to /names + k = line.index('353') + for item in line[k+1:]: + if item[0:2] == 'E_': + s.send('WHO %s\n'%item) + elif '352' in line: # answer to /who + # warning: this is a horrible hack which apparently works + k = line.index('352') + ip = line[k+4] + ip = socket.gethostbyname(ip) + name = line[k+6] + host = line[k+9] + self.peers[name] = (ip,host) + if time.time() - t > 5*60: + self.push_response({'method':'server.peers', 'result':[self.get_peers()]}) + s.send('NAMES #electrum\n') + t = time.time() + self.peers = {} + except: + traceback.print_exc(file=sys.stdout) + finally: + sf.close() + s.close() + + + + def process(self, request): + method = request['method'] + + result = '' + if method == 'server.banner': + result = self.banner.replace('\\n','\n') + elif method == 'server.peers.subscribe': + result = self.get_peers() + else: + print "unknown method", request + + if result!='': + response = { 'id':request['id'], 'method':method, 'params':request['params'], 'result':result } + self.push_response(response) + + + diff --git a/server.py b/server.py index 2005d8b..9d2ff84 100755 --- a/server.py +++ b/server.py @@ -54,13 +54,12 @@ from processor import Dispatcher from transports.stratum_http import HttpServer from transports.stratum_tcp import TcpServer from transports.native import NativeServer -from irc import ServerProcessor -from abe_backend import AbeProcessor +from modules.irc import ServerProcessor if use_libbitcoin: from modules.python_bitcoin import LibBitcoinProcessor as BlockchainProcessor else: - from abe_backend import AbeProcessor as BlockchainProcessor + from modules.abe import AbeProcessor as BlockchainProcessor if __name__ == '__main__': -- 1.7.1