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
13 class AbeStore(Datastore_class):
15 def __init__(self, config):
16 conf = DataStore.CONFIG_DEFAULTS
17 args, argv = readconf.parse_argv( [], conf)
18 args.dbtype = config.get('database','type')
19 if args.dbtype == 'sqlite3':
20 args.connect_args = { 'database' : config.get('database','database') }
21 elif args.dbtype == 'MySQLdb':
22 args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
23 elif args.dbtype == 'psycopg2':
24 args.connect_args = { 'database' : config.get('database','database') }
26 coin = config.get('server', 'coin')
28 if coin == 'litecoin':
29 print 'Litecoin settings:'
30 datadir = config.get('server','datadir')
31 print ' datadir = ' + datadir
32 args.datadir = [{"dirname":datadir,"chain":"Litecoin","code3":"LTC","address_version":"\u0030"}]
33 print ' addrtype = 48'
36 Datastore_class.__init__(self,args)
38 # Use 1 (Bitcoin) if chain_id is not sent
39 self.chain_id = self.datadirs[0]["chain_id"] or 1
40 print 'Coin chain_id = %d' % self.chain_id
42 self.sql_limit = int( config.get('database','limit') )
45 self.bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port'))
47 self.address_queue = Queue()
49 self.dblock = thread.allocate_lock()
51 self.known_mempool_hashes = []
54 def import_tx(self, tx, is_coinbase):
55 tx_id = super(AbeStore, self).import_tx(tx, is_coinbase)
56 self.last_tx_id = tx_id
62 def import_block(self, b, chain_ids=frozenset()):
64 block_id = super(AbeStore, self).import_block(b, chain_ids)
65 for pos in xrange(len(b['transactions'])):
66 tx = b['transactions'][pos]
68 tx['hash'] = util.double_sha256(tx['tx'])
69 tx_id = self.tx_find_id_and_value(tx)
71 self.update_tx_cache(tx_id)
73 print "error: import_block: no tx_id"
77 def update_tx_cache(self, txid):
78 inrows = self.get_tx_inputs(txid, False)
80 _hash = self.binout(row[6])
82 #print "WARNING: missing tx_in for tx", txid
85 address = hash_to_address(chr(self.addrtype), _hash)
86 if self.tx_cache.has_key(address):
87 print "cache: invalidating", address
88 self.tx_cache.pop(address)
89 self.address_queue.put(address)
91 outrows = self.get_tx_outputs(txid, False)
93 _hash = self.binout(row[6])
95 #print "WARNING: missing tx_out for tx", txid
98 address = hash_to_address(chr(self.addrtype), _hash)
99 if self.tx_cache.has_key(address):
100 print "cache: invalidating", address
101 self.tx_cache.pop(address)
102 self.address_queue.put(address)
104 def safe_sql(self,sql, params=(), lock=True):
108 if lock: self.dblock.acquire()
109 ret = self.selectall(sql,params)
112 traceback.print_exc(file=sys.stdout)
114 if lock: self.dblock.release()
117 raise BaseException('sql error')
122 def get_tx_outputs(self, tx_id, lock=True):
123 return self.safe_sql("""SELECT
125 txout.txout_scriptPubKey,
132 LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
133 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
134 LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
135 WHERE txout.tx_id = %d
136 ORDER BY txout.txout_pos
137 """%(tx_id), (), lock)
139 def get_tx_inputs(self, tx_id, lock=True):
140 return self.safe_sql(""" SELECT
144 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
146 COALESCE(txout.txout_pos, u.txout_pos),
149 LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
150 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
151 LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
152 LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
153 WHERE txin.tx_id = %d
154 ORDER BY txin.txin_pos
155 """%(tx_id,), (), lock)
158 def get_address_out_rows(self, dbhash):
159 out = self.safe_sql(""" SELECT
169 FROM chain_candidate cc
170 JOIN block b ON (b.block_id = cc.block_id)
171 JOIN block_tx ON (block_tx.block_id = b.block_id)
172 JOIN tx ON (tx.tx_id = block_tx.tx_id)
173 JOIN txin ON (txin.tx_id = tx.tx_id)
174 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
175 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
176 WHERE pubkey.pubkey_hash = ?
178 AND cc.in_longest = 1
179 LIMIT ? """, (dbhash, self.chain_id, self.sql_limit))
181 if len(out)==self.sql_limit:
182 raise BaseException('limit reached')
185 def get_address_out_rows_memorypool(self, dbhash):
186 out = self.safe_sql(""" SELECT
193 JOIN txin ON (txin.tx_id = tx.tx_id)
194 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
195 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
196 WHERE pubkey.pubkey_hash = ?
197 LIMIT ? """, (dbhash,self.sql_limit))
199 if len(out)==self.sql_limit:
200 raise BaseException('limit reached')
203 def get_address_in_rows(self, dbhash):
204 out = self.safe_sql(""" SELECT
214 FROM chain_candidate cc
215 JOIN block b ON (b.block_id = cc.block_id)
216 JOIN block_tx ON (block_tx.block_id = b.block_id)
217 JOIN tx ON (tx.tx_id = block_tx.tx_id)
218 JOIN txout ON (txout.tx_id = tx.tx_id)
219 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
220 WHERE pubkey.pubkey_hash = ?
222 AND cc.in_longest = 1
223 LIMIT ? """, (dbhash, self.chain_id, self.sql_limit))
225 if len(out)==self.sql_limit:
226 raise BaseException('limit reached')
229 def get_address_in_rows_memorypool(self, dbhash):
230 out = self.safe_sql( """ SELECT
237 JOIN txout ON (txout.tx_id = tx.tx_id)
238 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
239 WHERE pubkey.pubkey_hash = ?
240 LIMIT ? """, (dbhash,self.sql_limit))
242 if len(out)==self.sql_limit:
243 raise BaseException('limit reached')
246 def get_history(self, addr):
248 cached_version = self.tx_cache.get( addr )
249 if cached_version is not None:
250 return cached_version
252 version, binaddr = decode_check_address(addr)
256 dbhash = self.binin(binaddr)
258 rows += self.get_address_out_rows( dbhash )
259 rows += self.get_address_in_rows( dbhash )
266 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
268 print "cannot unpack row", row
270 tx_hash = self.hashout_hex(tx_hash)
272 "timestamp": int(nTime),
273 "height": int(height),
274 "is_input": int(is_in),
275 "block_hash": self.hashout_hex(blk_hash),
282 txpoints.append(txpoint)
283 known_tx.append(self.hashout_hex(tx_hash))
286 # todo: sort them really...
287 txpoints = sorted(txpoints, key=operator.itemgetter("timestamp"))
291 rows += self.get_address_in_rows_memorypool( dbhash )
292 rows += self.get_address_out_rows_memorypool( dbhash )
293 address_has_mempool = False
296 is_in, tx_hash, tx_id, pos, value = row
297 tx_hash = self.hashout_hex(tx_hash)
298 if tx_hash in known_tx:
301 # discard transactions that are too old
302 if self.last_tx_id - tx_id > 50000:
303 print "discarding tx id", tx_id
306 # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
307 address_has_mempool = True
309 #print "mempool", tx_hash
313 "is_input": int(is_in),
314 "block_hash": 'mempool',
320 txpoints.append(txpoint)
323 for txpoint in txpoints:
324 tx_id = txpoint['tx_id']
327 inrows = self.get_tx_inputs(tx_id)
329 _hash = self.binout(row[6])
331 #print "WARNING: missing tx_in for tx", tx_id, addr
333 address = hash_to_address(chr(self.addrtype), _hash)
334 txinputs.append(address)
335 txpoint['inputs'] = txinputs
337 outrows = self.get_tx_outputs(tx_id)
339 _hash = self.binout(row[6])
341 #print "WARNING: missing tx_out for tx", tx_id, addr
343 address = hash_to_address(chr(self.addrtype), _hash)
344 txoutputs.append(address)
345 txpoint['outputs'] = txoutputs
347 # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
348 if not txpoint['is_input']:
349 # detect if already redeemed...
351 if row[6] == dbhash: break
354 #row = self.get_tx_output(tx_id,dbhash)
355 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
356 # if not redeemed, we add the script
358 if not row[4]: txpoint['raw_output_script'] = row[1]
363 # do not cache mempool results because statuses are ambiguous
364 if not address_has_mempool:
365 self.tx_cache[addr] = txpoints
370 def get_status(self,addr):
371 # get address status, i.e. the last block for that address.
372 tx_points = self.get_history(addr)
376 lastpoint = tx_points[-1]
377 status = lastpoint['block_hash']
378 # this is a temporary hack; move it up once old clients have disappeared
379 if status == 'mempool': # and session['version'] != "old":
380 status = status + ':%d'% len(tx_points)
384 def get_block_header(self, block_height):
385 out = self.safe_sql("""
389 block_hashMerkleRoot,
397 WHERE block_height = %d AND in_longest = 1"""%block_height)
399 if not out: raise BaseException("block not found")
401 (block_hash, block_version, hashMerkleRoot, nTime, nBits, nNonce, height,prev_block_hash, block_id) \
402 = ( 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]) )
404 out = {"block_height":block_height, "version":block_version, "prev_block_hash":prev_block_hash,
405 "merkle_root":hashMerkleRoot, "timestamp":nTime, "bits":nBits, "nonce":nNonce}
409 def get_tx_merkle(self, tx_hash):
411 out = self.safe_sql("""
412 SELECT block_tx.block_id FROM tx
413 JOIN block_tx on tx.tx_id = block_tx.tx_id
414 JOIN chain_summary on chain_summary.block_id = block_tx.block_id
415 WHERE tx_hash='%s' AND in_longest = 1"""%tx_hash)
419 out = self.safe_sql("SELECT block_height FROM chain_summary WHERE block_id = %d AND in_longest = 1"%block_id)
421 if not out: raise BaseException("block not found")
422 block_height = int(out[0][0])
425 # list all tx in block
426 for row in self.safe_sql("""
427 SELECT DISTINCT tx_id, tx_pos, tx_hash
430 ORDER BY tx_pos""", (block_id,)):
431 tx_id, tx_pos, tx_h = row
435 # TODO: do not compute this on client request, better store the hash tree of each block in a database...
437 encode = lambda x: x[::-1].encode('hex')
438 decode = lambda x: x.decode('hex')[::-1]
439 Hash = lambda x: hashlib.sha256(hashlib.sha256(x).digest()).digest()
441 merkle = map(decode, merkle)
442 target_hash = decode(tx_hash)
445 while len(merkle) != 1:
446 if len(merkle)%2: merkle.append( merkle[-1] )
449 if merkle[0] == target_hash:
450 s.append( "L" + encode(merkle[1]))
451 n.append( target_hash )
452 elif merkle[1] == target_hash:
453 s.append( "R" + encode(merkle[0]))
454 n.append( target_hash)
456 n.append( Hash( merkle[0] + merkle[1] ) )
461 return {"block_height":block_height,"merkle":s}
466 def memorypool_update(store):
468 ds = BCDataStream.BCDataStream()
469 postdata = dumps({"method": 'getrawmempool', 'params': [], 'id':'jsonrpc'})
470 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
472 if r['error'] != None:
476 mempool_hashes = r.get('result')
477 for tx_hash in mempool_hashes:
479 if tx_hash in store.known_mempool_hashes: continue
480 store.known_mempool_hashes.append(tx_hash)
482 postdata = dumps({"method": 'getrawtransaction', 'params': [tx_hash], 'id':'jsonrpc'})
483 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
485 if r['error'] != None:
487 hextx = r.get('result')
489 ds.write(hextx.decode('hex'))
490 tx = deserialize.parse_Transaction(ds)
491 tx['hash'] = util.double_sha256(tx['tx'])
493 if store.tx_find_id_and_value(tx):
496 tx_id = store.import_tx(tx, False)
497 store.update_tx_cache(tx_id)
501 store.known_mempool_hashes = mempool_hashes
504 def send_tx(self,tx):
505 postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'})
506 respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
508 if r['error'] != None:
509 msg = r['error'].get('message')
510 out = "error: transaction rejected by memorypool: " + msg + "\n" + tx
516 def main_iteration(store):
519 store.memorypool_update()
520 block_number = store.get_block_number(store.chain_id)
527 # if there is an exception, do rollback and then re-raise the exception
528 for dircfg in store.datadirs:
530 store.catch_up_dir(dircfg)
532 store.log.exception("Failed to catch up %s", dircfg)
539 from processor import Processor
541 class BlockchainProcessor(Processor):
543 def __init__(self, config):
544 Processor.__init__(self)
545 self.store = AbeStore(config)
546 self.block_number = -1
547 self.watched_addresses = []
550 n = self.store.main_iteration()
551 print "blockchain: %d blocks"%n
553 threading.Timer(10, self.run_store_iteration).start()
555 def process(self, request):
556 #print "abe process", request
558 message_id = request['id']
559 method = request['method']
560 params = request.get('params',[])
564 if method == 'blockchain.numblocks.subscribe':
565 result = self.block_number
567 elif method == 'blockchain.address.subscribe':
570 result = self.store.get_status(address)
571 self.watch_address(address)
572 except BaseException, e:
573 error = str(e) + ': ' + address
574 print "error:", error
576 elif method == 'blockchain.address.get_history':
579 result = self.store.get_history( address )
580 except BaseException, e:
581 error = str(e) + ': ' + address
582 print "error:", error
584 elif method == 'blockchain.block.get_header':
587 result = self.store.get_block_header( height )
588 except BaseException, e:
589 error = str(e) + ': %d'% height
590 print "error:", error
592 elif method == 'blockchain.transaction.broadcast':
593 txo = self.store.send_tx(params[0])
594 print "sent tx:", txo
597 elif method == 'blockchain.transaction.get_merkle':
600 result = self.store.get_tx_merkle(tx_hash )
601 except BaseException, e:
602 error = str(e) + ': ' + tx_hash
603 print "error:", error
606 error = "unknown method:%s"%method
610 response = { 'id':message_id, 'error':error }
611 self.push_response(response)
613 response = { 'id':message_id, 'result':result }
614 self.push_response(response)
617 def watch_address(self, addr):
618 if addr not in self.watched_addresses:
619 self.watched_addresses.append(addr)
622 def run_store_iteration(self):
625 block_number = self.store.main_iteration()
627 traceback.print_exc(file=sys.stdout)
631 if self.shared.stopped():
635 if self.block_number != block_number:
636 self.block_number = block_number
637 print "block number:", self.block_number
638 self.push_response({ 'id': None, 'method':'blockchain.numblocks.subscribe', 'params':[self.block_number] })
642 addr = self.store.address_queue.get(False)
645 if addr in self.watched_addresses:
646 status = self.store.get_status( addr )
647 self.push_response({ 'id': None, 'method':'blockchain.address.subscribe', 'params':[addr, status] })
649 threading.Timer(10, self.run_store_iteration).start()