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)
476 out = self.safe_sql("SELECT block_height FROM chain_summary WHERE block_id = %d AND in_longest = 1"%block_id)
478 if not out: raise BaseException("block not found")
479 block_height = int(out[0][0])
484 # list all tx in block
485 for row in self.safe_sql("""
486 SELECT DISTINCT tx_id, tx_pos, tx_hash
489 ORDER BY tx_pos""", (block_id,)):
490 _id, _pos, _hash = row
492 if _hash == tx_hash: tx_pos = int(_pos)
495 # TODO: do not compute this on client request, better store the hash tree of each block in a database...
497 merkle = map(decode, merkle)
498 target_hash = decode(tx_hash)
501 while len(merkle) != 1:
502 if len(merkle)%2: merkle.append( merkle[-1] )
505 new_hash = Hash( merkle[0] + merkle[1] )
506 if merkle[0] == target_hash:
507 s.append( encode(merkle[1]))
508 target_hash = new_hash
509 elif merkle[1] == target_hash:
510 s.append( encode(merkle[0]))
511 target_hash = new_hash
517 return {"block_height":block_height, "merkle":s, "pos":tx_pos}
522 def memorypool_update(store):
524 ds = BCDataStream.BCDataStream()
525 postdata = dumps({"method": 'getrawmempool', 'params': [], 'id':'jsonrpc'})
526 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
528 if r['error'] != None:
532 mempool_hashes = r.get('result')
533 for tx_hash in mempool_hashes:
535 if tx_hash in store.known_mempool_hashes: continue
536 store.known_mempool_hashes.append(tx_hash)
538 postdata = dumps({"method": 'getrawtransaction', 'params': [tx_hash], 'id':'jsonrpc'})
539 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
541 if r['error'] != None:
543 hextx = r.get('result')
545 ds.write(hextx.decode('hex'))
546 tx = deserialize.parse_Transaction(ds)
547 tx['hash'] = util.double_sha256(tx['tx'])
549 if store.tx_find_id_and_value(tx):
552 tx_id = store.import_tx(tx, False)
553 store.update_tx_cache(tx_id)
557 store.known_mempool_hashes = mempool_hashes
560 def send_tx(self,tx):
561 postdata = dumps({"method": 'sendrawtransaction', 'params': [tx], 'id':'jsonrpc'})
562 respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
564 if r['error'] != None:
565 msg = r['error'].get('message')
566 out = "error: transaction rejected by memorypool: " + msg + "\n" + tx
572 def main_iteration(store):
575 store.memorypool_update()
576 height = store.get_block_number( store.chain_id )
578 block_header = store.get_block_header( height )
585 # if there is an exception, do rollback and then re-raise the exception
586 for dircfg in store.datadirs:
588 store.catch_up_dir(dircfg)
590 store.log.exception("Failed to catch up %s", dircfg)
597 from processor import Processor
599 class BlockchainProcessor(Processor):
601 def __init__(self, config):
602 Processor.__init__(self)
603 self.store = AbeStore(config)
604 self.watched_addresses = []
607 self.block_header = self.store.main_iteration()
608 self.block_number = self.block_header.get('block_height')
609 print "blockchain: %d blocks"%self.block_number
611 threading.Timer(10, self.run_store_iteration).start()
613 def process(self, request):
614 #print "abe process", request
616 message_id = request['id']
617 method = request['method']
618 params = request.get('params',[])
622 if method == 'blockchain.numblocks.subscribe':
623 result = self.block_number
625 elif method == 'blockchain.headers.subscribe':
626 result = self.block_header
628 elif method == 'blockchain.address.subscribe':
631 result = self.store.get_status(address)
632 self.watch_address(address)
633 except BaseException, e:
634 error = str(e) + ': ' + address
635 print "error:", error
637 elif method == 'blockchain.address.get_history':
640 result = self.store.get_history( address )
641 except BaseException, e:
642 error = str(e) + ': ' + address
643 print "error:", error
645 elif method == 'blockchain.block.get_header':
648 result = self.store.get_block_header( height )
649 except BaseException, e:
650 error = str(e) + ': %d'% height
651 print "error:", error
653 elif method == 'blockchain.block.get_chunk':
656 result = self.store.get_chunk( index )
657 except BaseException, e:
658 error = str(e) + ': %d'% index
659 print "error:", error
661 elif method == 'blockchain.transaction.broadcast':
662 txo = self.store.send_tx(params[0])
663 print "sent tx:", txo
666 elif method == 'blockchain.transaction.get_merkle':
669 result = self.store.get_tx_merkle(tx_hash )
670 except BaseException, e:
671 error = str(e) + ': ' + tx_hash
672 print "error:", error
675 error = "unknown method:%s"%method
679 response = { 'id':message_id, 'error':error }
680 self.push_response(response)
682 response = { 'id':message_id, 'result':result }
683 self.push_response(response)
686 def watch_address(self, addr):
687 if addr not in self.watched_addresses:
688 self.watched_addresses.append(addr)
691 def run_store_iteration(self):
694 block_header = self.store.main_iteration()
696 traceback.print_exc(file=sys.stdout)
700 if self.shared.stopped():
704 if self.block_number != block_header.get('block_height'):
705 self.block_number = block_header.get('block_height')
706 print "block number:", self.block_number
707 self.push_response({ 'id': None, 'method':'blockchain.numblocks.subscribe', 'params':[self.block_number] })
709 if self.block_header != block_header:
710 self.block_header = block_header
711 self.push_response({ 'id': None, 'method':'blockchain.headers.subscribe', 'params':[self.block_header] })
715 addr = self.store.address_queue.get(False)
718 if addr in self.watched_addresses:
719 status = self.store.get_status( addr )
720 self.push_response({ 'id': None, 'method':'blockchain.address.subscribe', 'params':[addr, status] })
722 threading.Timer(10, self.run_store_iteration).start()