1 from Abe.util import hash_to_address, decode_check_address
2 from Abe.DataStore import DataStore as Datastore_class
3 from Abe import DataStore, readconf, BCDataStream, deserialize, util, base58
7 import thread, traceback, sys, urllib, operator
8 from json import dumps, loads
9 from Queue import Queue
10 import time, threading
14 encode = lambda x: x[::-1].encode('hex')
15 decode = lambda x: x.decode('hex')[::-1]
16 Hash = lambda x: hashlib.sha256(hashlib.sha256(x).digest()).digest()
19 return s.decode('hex')[::-1].encode('hex')
21 def int_to_hex(i, length=1):
22 s = hex(i)[2:].rstrip('L')
23 s = "0"*(2*length - len(s)) + s
26 def header_to_string(res):
27 s = int_to_hex(res.get('version'),4) \
28 + rev_hex(res.get('prev_block_hash')) \
29 + rev_hex(res.get('merkle_root')) \
30 + int_to_hex(int(res.get('timestamp')),4) \
31 + int_to_hex(int(res.get('bits')),4) \
32 + int_to_hex(int(res.get('nonce')),4)
36 class AbeStore(Datastore_class):
38 def __init__(self, config):
39 conf = DataStore.CONFIG_DEFAULTS
40 args, argv = readconf.parse_argv( [], conf)
41 args.dbtype = config.get('database','type')
42 if args.dbtype == 'sqlite3':
43 args.connect_args = { 'database' : config.get('database','database') }
44 elif args.dbtype == 'MySQLdb':
45 args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
46 elif args.dbtype == 'psycopg2':
47 args.connect_args = { 'database' : config.get('database','database') }
49 coin = config.get('server', 'coin')
51 if coin == 'litecoin':
52 print 'Litecoin settings:'
53 datadir = config.get('server','datadir')
54 print ' datadir = ' + datadir
55 args.datadir = [{"dirname":datadir,"chain":"Litecoin","code3":"LTC","address_version":"\u0030"}]
56 print ' addrtype = 48'
59 Datastore_class.__init__(self,args)
61 # Use 1 (Bitcoin) if chain_id is not sent
62 self.chain_id = self.datadirs[0]["chain_id"] or 1
63 print 'Coin chain_id = %d' % self.chain_id
65 self.sql_limit = int( config.get('database','limit') )
68 self.bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port'))
70 self.address_queue = Queue()
72 self.dblock = thread.allocate_lock()
74 self.known_mempool_hashes = []
77 def import_tx(self, tx, is_coinbase):
78 tx_id = super(AbeStore, self).import_tx(tx, is_coinbase)
79 self.last_tx_id = tx_id
85 def import_block(self, b, chain_ids=frozenset()):
87 block_id = super(AbeStore, self).import_block(b, chain_ids)
88 for pos in xrange(len(b['transactions'])):
89 tx = b['transactions'][pos]
91 tx['hash'] = util.double_sha256(tx['tx'])
92 tx_id = self.tx_find_id_and_value(tx)
94 self.update_tx_cache(tx_id)
96 print "error: import_block: no tx_id"
100 def update_tx_cache(self, txid):
101 inrows = self.get_tx_inputs(txid, False)
103 _hash = self.binout(row[6])
105 #print "WARNING: missing tx_in for tx", txid
108 address = hash_to_address(chr(self.addrtype), _hash)
109 if self.tx_cache.has_key(address):
110 print "cache: invalidating", address
111 self.tx_cache.pop(address)
112 self.address_queue.put(address)
114 outrows = self.get_tx_outputs(txid, False)
116 _hash = self.binout(row[6])
118 #print "WARNING: missing tx_out for tx", txid
121 address = hash_to_address(chr(self.addrtype), _hash)
122 if self.tx_cache.has_key(address):
123 print "cache: invalidating", address
124 self.tx_cache.pop(address)
125 self.address_queue.put(address)
127 def safe_sql(self,sql, params=(), lock=True):
131 if lock: self.dblock.acquire()
132 ret = self.selectall(sql,params)
135 traceback.print_exc(file=sys.stdout)
137 if lock: self.dblock.release()
140 raise BaseException('sql error')
145 def get_tx_outputs(self, tx_id, lock=True):
146 return self.safe_sql("""SELECT
148 txout.txout_scriptPubKey,
155 LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
156 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
157 LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
158 WHERE txout.tx_id = %d
159 ORDER BY txout.txout_pos
160 """%(tx_id), (), lock)
162 def get_tx_inputs(self, tx_id, lock=True):
163 return self.safe_sql(""" SELECT
167 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
169 COALESCE(txout.txout_pos, u.txout_pos),
172 LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
173 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
174 LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
175 LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
176 WHERE txin.tx_id = %d
177 ORDER BY txin.txin_pos
178 """%(tx_id,), (), lock)
181 def get_address_out_rows(self, dbhash):
182 out = self.safe_sql(""" SELECT
192 FROM chain_candidate cc
193 JOIN block b ON (b.block_id = cc.block_id)
194 JOIN block_tx ON (block_tx.block_id = b.block_id)
195 JOIN tx ON (tx.tx_id = block_tx.tx_id)
196 JOIN txin ON (txin.tx_id = tx.tx_id)
197 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
198 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
199 WHERE pubkey.pubkey_hash = ?
201 AND cc.in_longest = 1
202 LIMIT ? """, (dbhash, self.chain_id, self.sql_limit))
204 if len(out)==self.sql_limit:
205 raise BaseException('limit reached')
208 def get_address_out_rows_memorypool(self, dbhash):
209 out = self.safe_sql(""" SELECT
216 JOIN txin ON (txin.tx_id = tx.tx_id)
217 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
218 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
219 WHERE pubkey.pubkey_hash = ?
220 LIMIT ? """, (dbhash,self.sql_limit))
222 if len(out)==self.sql_limit:
223 raise BaseException('limit reached')
226 def get_address_in_rows(self, dbhash):
227 out = self.safe_sql(""" SELECT
237 FROM chain_candidate cc
238 JOIN block b ON (b.block_id = cc.block_id)
239 JOIN block_tx ON (block_tx.block_id = b.block_id)
240 JOIN tx ON (tx.tx_id = block_tx.tx_id)
241 JOIN txout ON (txout.tx_id = tx.tx_id)
242 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
243 WHERE pubkey.pubkey_hash = ?
245 AND cc.in_longest = 1
246 LIMIT ? """, (dbhash, self.chain_id, self.sql_limit))
248 if len(out)==self.sql_limit:
249 raise BaseException('limit reached')
252 def get_address_in_rows_memorypool(self, dbhash):
253 out = self.safe_sql( """ SELECT
260 JOIN txout ON (txout.tx_id = tx.tx_id)
261 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
262 WHERE pubkey.pubkey_hash = ?
263 LIMIT ? """, (dbhash,self.sql_limit))
265 if len(out)==self.sql_limit:
266 raise BaseException('limit reached')
269 def get_history(self, addr):
271 cached_version = self.tx_cache.get( addr )
272 if cached_version is not None:
273 return cached_version
275 version, binaddr = decode_check_address(addr)
279 dbhash = self.binin(binaddr)
281 rows += self.get_address_out_rows( dbhash )
282 rows += self.get_address_in_rows( dbhash )
289 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
291 print "cannot unpack row", row
293 tx_hash = self.hashout_hex(tx_hash)
295 "timestamp": int(nTime),
296 "height": int(height),
297 "is_input": int(is_in),
298 "block_hash": self.hashout_hex(blk_hash),
305 txpoints.append(txpoint)
306 known_tx.append(self.hashout_hex(tx_hash))
309 # todo: sort them really...
310 txpoints = sorted(txpoints, key=operator.itemgetter("timestamp"))
314 rows += self.get_address_in_rows_memorypool( dbhash )
315 rows += self.get_address_out_rows_memorypool( dbhash )
316 address_has_mempool = False
319 is_in, tx_hash, tx_id, pos, value = row
320 tx_hash = self.hashout_hex(tx_hash)
321 if tx_hash in known_tx:
324 # discard transactions that are too old
325 if self.last_tx_id - tx_id > 50000:
326 print "discarding tx id", tx_id
329 # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
330 address_has_mempool = True
332 #print "mempool", tx_hash
336 "is_input": int(is_in),
337 "block_hash": 'mempool',
343 txpoints.append(txpoint)
346 for txpoint in txpoints:
347 tx_id = txpoint['tx_id']
350 inrows = self.get_tx_inputs(tx_id)
352 _hash = self.binout(row[6])
354 #print "WARNING: missing tx_in for tx", tx_id, addr
356 address = hash_to_address(chr(self.addrtype), _hash)
357 txinputs.append(address)
358 txpoint['inputs'] = txinputs
360 outrows = self.get_tx_outputs(tx_id)
362 _hash = self.binout(row[6])
364 #print "WARNING: missing tx_out for tx", tx_id, addr
366 address = hash_to_address(chr(self.addrtype), _hash)
367 txoutputs.append(address)
368 txpoint['outputs'] = txoutputs
370 # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
371 if not txpoint['is_input']:
372 # detect if already redeemed...
374 if row[6] == dbhash: break
377 #row = self.get_tx_output(tx_id,dbhash)
378 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
379 # if not redeemed, we add the script
381 if not row[4]: txpoint['raw_output_script'] = row[1]
386 # do not cache mempool results because statuses are ambiguous
387 if not address_has_mempool:
388 self.tx_cache[addr] = txpoints
393 def get_status(self,addr):
394 # get address status, i.e. the last block for that address.
395 tx_points = self.get_history(addr)
399 lastpoint = tx_points[-1]
400 status = lastpoint['block_hash']
401 # this is a temporary hack; move it up once old clients have disappeared
402 if status == 'mempool': # and session['version'] != "old":
403 status = status + ':%d'% len(tx_points)
407 def get_block_header(self, block_height):
408 out = self.safe_sql("""
412 block_hashMerkleRoot,
420 WHERE block_height = %d AND in_longest = 1"""%block_height)
422 if not out: raise BaseException("block not found")
424 (block_hash, block_version, hashMerkleRoot, nTime, nBits, nNonce, height,prev_block_hash, block_id) \
425 = ( self.hashout_hex(row[0]), int(row[1]), self.hashout_hex(row[2]), int(row[3]), int(row[4]), int(row[5]), int(row[6]), self.hashout_hex(row[7]), int(row[8]) )
427 out = {"block_height":block_height, "version":block_version, "prev_block_hash":prev_block_hash,
428 "merkle_root":hashMerkleRoot, "timestamp":nTime, "bits":nBits, "nonce":nNonce}
432 def get_chunk(self, index):
437 block_hashMerkleRoot,
445 WHERE block_height >= %d AND block_height< %d AND in_longest = 1"""%(index*2016, (index+1)*2016)
447 out = self.safe_sql(sql)
450 (block_hash, block_version, hashMerkleRoot, nTime, nBits, nNonce, height, prev_block_hash, block_height) \
451 = ( self.hashout_hex(row[0]), int(row[1]), self.hashout_hex(row[2]), int(row[3]), int(row[4]), int(row[5]), int(row[6]), self.hashout_hex(row[7]), int(row[8]) )
452 h = {"block_height":block_height, "version":block_version, "prev_block_hash":prev_block_hash,
453 "merkle_root":hashMerkleRoot, "timestamp":nTime, "bits":nBits, "nonce":nNonce}
455 if h.get('block_height')==0: h['prev_block_hash'] = "0"*64
456 msg += header_to_string(h)
458 #print "hash", encode(Hash(msg.decode('hex')))
459 #if h.get('block_height')==1:break
461 print "get_chunk", index, len(msg)
466 def get_tx_merkle(self, tx_hash):
468 out = self.safe_sql("""
469 SELECT block_tx.block_id FROM tx
470 JOIN block_tx on tx.tx_id = block_tx.tx_id
471 JOIN chain_summary on chain_summary.block_id = block_tx.block_id
472 WHERE tx_hash='%s' AND in_longest = 1"""%tx_hash)
474 if not out: raise BaseException("not in a block")
475 block_id = int(out[0][0])
478 out = self.safe_sql("SELECT block_height FROM chain_summary WHERE block_id = %d AND in_longest = 1"%block_id)
480 if not out: raise BaseException("block not found")
481 block_height = int(out[0][0])
486 # list all tx in block
487 for row in self.safe_sql("""
488 SELECT DISTINCT tx_id, tx_pos, tx_hash
491 ORDER BY tx_pos""", (block_id,)):
492 _id, _pos, _hash = row
494 if _hash == tx_hash: tx_pos = int(_pos)
497 # TODO: do not compute this on client request, better store the hash tree of each block in a database...
499 merkle = map(decode, merkle)
500 target_hash = decode(tx_hash)
503 while len(merkle) != 1:
504 if len(merkle)%2: merkle.append( merkle[-1] )
507 new_hash = Hash( merkle[0] + merkle[1] )
508 if merkle[0] == target_hash:
509 s.append( encode(merkle[1]))
510 target_hash = new_hash
511 elif merkle[1] == target_hash:
512 s.append( encode(merkle[0]))
513 target_hash = new_hash
519 return {"block_height":block_height, "merkle":s, "pos":tx_pos}
524 def memorypool_update(store):
526 ds = BCDataStream.BCDataStream()
527 postdata = dumps({"method": 'getrawmempool', 'params': [], 'id':'jsonrpc'})
528 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
530 if r['error'] != None:
534 mempool_hashes = r.get('result')
535 for tx_hash in mempool_hashes:
537 if tx_hash in store.known_mempool_hashes: continue
538 store.known_mempool_hashes.append(tx_hash)
540 postdata = dumps({"method": 'getrawtransaction', 'params': [tx_hash], 'id':'jsonrpc'})
541 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
543 if r['error'] != None:
545 hextx = r.get('result')
547 ds.write(hextx.decode('hex'))
548 tx = deserialize.parse_Transaction(ds)
549 tx['hash'] = util.double_sha256(tx['tx'])
551 if store.tx_find_id_and_value(tx):
554 tx_id = store.import_tx(tx, False)
555 store.update_tx_cache(tx_id)
559 store.known_mempool_hashes = mempool_hashes
562 def send_tx(self,tx):
563 postdata = dumps({"method": 'sendrawtransaction', 'params': [tx], 'id':'jsonrpc'})
564 respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
566 if r['error'] != None:
567 msg = r['error'].get('message')
568 out = "error: transaction rejected by memorypool: " + msg + "\n" + tx
574 def main_iteration(store):
577 store.memorypool_update()
578 height = store.get_block_number( store.chain_id )
580 block_header = store.get_block_header( height )
587 # if there is an exception, do rollback and then re-raise the exception
588 for dircfg in store.datadirs:
590 store.catch_up_dir(dircfg)
592 store.log.exception("Failed to catch up %s", dircfg)
599 from processor import Processor
601 class BlockchainProcessor(Processor):
603 def __init__(self, config):
604 Processor.__init__(self)
605 self.store = AbeStore(config)
606 self.watched_addresses = []
609 self.block_header = self.store.main_iteration()
610 self.block_number = self.block_header.get('block_height')
611 print "blockchain: %d blocks"%self.block_number
613 threading.Timer(10, self.run_store_iteration).start()
615 def process(self, request):
616 #print "abe process", request
618 message_id = request['id']
619 method = request['method']
620 params = request.get('params',[])
624 if method == 'blockchain.numblocks.subscribe':
625 result = self.block_number
627 elif method == 'blockchain.headers.subscribe':
628 result = self.block_header
630 elif method == 'blockchain.address.subscribe':
633 result = self.store.get_status(address)
634 self.watch_address(address)
635 except BaseException, e:
636 error = str(e) + ': ' + address
637 print "error:", error
639 elif method == 'blockchain.address.get_history':
642 result = self.store.get_history( address )
643 except BaseException, e:
644 error = str(e) + ': ' + address
645 print "error:", error
647 elif method == 'blockchain.block.get_header':
650 result = self.store.get_block_header( height )
651 except BaseException, e:
652 error = str(e) + ': %d'% height
653 print "error:", error
655 elif method == 'blockchain.block.get_chunk':
658 result = self.store.get_chunk( index )
659 except BaseException, e:
660 error = str(e) + ': %d'% index
661 print "error:", error
663 elif method == 'blockchain.transaction.broadcast':
664 txo = self.store.send_tx(params[0])
665 print "sent tx:", txo
668 elif method == 'blockchain.transaction.get_merkle':
671 result = self.store.get_tx_merkle(tx_hash )
672 except BaseException, e:
673 error = str(e) + ': ' + tx_hash
674 print "error:", error
677 error = "unknown method:%s"%method
681 response = { 'id':message_id, 'error':error }
682 self.push_response(response)
684 response = { 'id':message_id, 'result':result }
685 self.push_response(response)
688 def watch_address(self, addr):
689 if addr not in self.watched_addresses:
690 self.watched_addresses.append(addr)
693 def run_store_iteration(self):
697 block_header = self.store.main_iteration()
698 t2 = time.time() - t1
700 traceback.print_exc(file=sys.stdout)
704 if self.shared.stopped():
708 if self.block_number != block_header.get('block_height'):
709 self.block_number = block_header.get('block_height')
710 print "block number: %d (%.3f seconds)"%(self.block_number, t2)
711 self.push_response({ 'id': None, 'method':'blockchain.numblocks.subscribe', 'params':[self.block_number] })
713 if self.block_header != block_header:
714 self.block_header = block_header
715 self.push_response({ 'id': None, 'method':'blockchain.headers.subscribe', 'params':[self.block_header] })
720 addr = self.store.address_queue.get(False)
723 if addr in self.watched_addresses:
724 status = self.store.get_status( addr )
725 self.push_response({ 'id': None, 'method':'blockchain.address.subscribe', 'params':[addr, status] })
727 threading.Timer(10, self.run_store_iteration).start()