sqlchain-electrum

#!/usr/bin/env python

from time import sleep
from urllib2 import urlopen, URLError

import os, sys, getopt, signal, json, daemon

from gevent import socket, monkey, spawn
from gevent.server import StreamServer
from gevent.queue import Queue

from sqlchain.version import version
from sqlchain.util import dotdict, loadcfg, savecfg, drop2user, logts, getChunk

__builtins__.sqc = dotdict()  # container for super globals

sqc.cfg = { 'log':sys.argv[0]+'.log', 'listen':'localhost:8081', 'www':'www', 'api':'http://localhost:8085/api',
        'banner':'docs/electrum.banner', 'path':'/var/data/sqlchain' }

srvinfo = { 'version':version, 'banner':'', 'block':0, 'header':{} }
subs = { 'numblocks':{}, 'headers':{}, 'address':{}, '_ip_':{} }

def ReqHandler():
    while True:
        #resp = None
        fp,req = sqc.reqQ.get()
        print 'REQ', req
        args = req['method'].split('.')
        val = req['params'][0] if len(req['params']) > 0 else 1
        if args[-1] == 'subscribe':
            if args[1] in subs and not getSubs(args[1], val, fp):
                addSub(args[1], val, fp)
            respSub(args[1], fp, req)
        elif args[0] == 'server':
            sqc.respQ.put((fp, req['id'], srvinfo[args[1]] if args[1] in srvinfo else {}))
        elif req['method'] in reqFuncs:
            spawn(reqFuncs[req['method']], fp, req)
        else:
            logts("Bad Req %s:%d - %s" % (subs['_ip_'][fp][0]+(req['method'],)))

def RespHandler():
    while True:
        fp,reqid,resp = sqc.respQ.get()
        resp = json.dumps({ 'id':reqid, 'result':resp } if resp is None or not 'error' in resp else { 'id':reqid,'error':resp['error'] })
        print "RESP", reqid, resp
        fp.write(resp+'\n')
        fp.flush()

def SyncHandler():
    sync_id = 0
    while True:
        resp = apicall('/sync/'+str(sync_id))
        if resp and 'error' in resp:
            sleep(30)
        elif resp:
            if resp['block'] != srvinfo['block']:
                srvinfo['block'] = resp['block']
                pubSubs('numblocks', msg=resp['block'])
            hdr = apicall('/block-index/'+str(resp['block'])+'/electrum')
            if hdr != srvinfo['header']:
                srvinfo['header'] = hdr
                pubSubs('headers', msg=hdr)
            if len(resp['txs']) > 0:
                for tx in resp['txs']:
                    pubSubs('address', addrs=getAddrs(tx))
            sync_id = resp['sync_id']

def TcpHandler(sock, address):
    fp = sock.makefile()
    addSub('_ip_', address, fp)
    while True:
        line = fp.readline()
        if line:
            sqc.reqQ.put((fp, json.loads(line)))
        else:
            break
    delSubs(fp)
    sock.shutdown(socket.SHUT_WR)
    sock.close()

def pubSubs(sub, msg=None, addrs=None):
    if addrs:
        for addr in addrs:
            fps = getSubs(sub, addr)
            if len(fps) > 0:
                data = apicall('/history/'+addr+'/status')
                for fp in fps:
                    sqc.respQ.put((fp, None, data))
    if msg:
        for fp in getSubs(sub):
            sqc.respQ.put((fp, None, msg))

def getSubs(sub, val=1, key=None):
    if key:
        return key in subs[sub] and val in subs[sub][key]
    if val == 1:
        return subs[sub].keys()
    fps = []
    for k in subs[sub]:
        if val in subs[sub][k]:
            fps.append(k)
    return fps

def addSub(sub, val, key):
    if key in subs[sub]:
        subs[sub][key].add(val)
    else:
        subs[sub][key] = set(val)

def delSubs(key):
    for sub in subs:
        if key in subs[sub]:
            del subs[sub][key]

def respSub(to, fp, req):
    if to == 'address':
        spawn(addrHistory, fp, req, '/status')
    elif to == 'numblocks':
        sqc.respQ.put((fp, req['id'], srvinfo['block']))
    elif to == 'headers':
        sqc.respQ.put((fp, req['id'], srvinfo['header']))
    else:
        sqc.respQ.put((fp, req['id'], []))

def addrHistory(fp, req, args=''):
    data = apicall('/history/'+req['params'][0] + args)
    sqc.respQ.put((fp, req['id'], data if args else data['txs'] if len(data['txs']) > 0 else None ))

def addrBalance(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/history/'+req['params'][0]+'/balance')))

def addrMemPool(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/history/'+req['params'][0]+'/uncfmd')))

def addrUnspent(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/history/'+req['params'][0]+'/utxo')))

def addrProof(fp, req): # pylint:disable=unused-argument
    pass

def blkHeader(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/block-index/'+req['params'][0]+'/electrum') ))

def blkChunk(fp, req):
    sqc.respQ.put((fp, req['id'], getChunk(int(req['params'][0]), sqc.cfg).encode('hex') ))

def utxoAddress(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/tx/'+req['params'][0]+'/output/%d' % req['params'][1]) ))

def txGet(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/tx/'+req['params'][0]+'/raw') ))

def txSend(fp, req):
    #logts("Tx Sent: %s" % txid)
    sqc.respQ.put((fp, req['id'], apicall('/tx/send', {'rawtx':req['params'][0]}) ))

def txMerkle(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/merkle/'+req['params'][0]) ))

def feeEstimate(fp, req):
    sqc.respQ.put((fp, req['id'], apicall('/util/estimatefee/'+req['params'][0]) ))

reqFuncs = { 'blockchain.address.get_history':addrHistory, 'blockchain.address.get_balance':addrBalance,
             'blockchain.address.get_mempool':addrMemPool, 'blockchain.address.get_proof':addrProof,
             'blockchain.address.listunspent':addrUnspent, 'blockchain.utxo.get_address':utxoAddress,
             'blockchain.block.get_header':blkHeader,      'blockchain.block.get_chunk':blkChunk,
             'blockchain.transaction.broadcast':txSend,    'blockchain.transaction.get_merkle':txMerkle,
             'blockchain.transaction.get':txGet,           'blockchain.estimatefee':feeEstimate }

def getAddrs(tx):
    addrs = []
    for vi in tx['inputs']:
        if 'addr' in vi['prev_out']:
            addrs.append(vi['prev_out']['addr'])
    for vo in tx['out']:
        addrs.append(vo['addr'])
    return addrs

def options(cfg): # pylint:disable=too-many-branches
    try:
        opts,_ = getopt.getopt(sys.argv[1:], "hvb:p:c:d:l:w:p:s:a:u:b:",
            ["help", "version", "debug", "db=", "log=", "listen=", "www=", "user=", "banner=", "defaults" ])
    except getopt.GetoptError:
        usage()
    for opt,arg in opts:
        if opt in ("-h", "--help"):
            usage()
        elif opt in ("-v", "--version"):
            sys.exit(sys.argv[0]+': '+version)
        elif opt in ("-d", "--db"):
            cfg['db'] = arg
        elif opt in ("-l", "--log"):
            cfg['log'] = arg
        elif opt in ("-w", "--www"):
            cfg['www'] = arg
        elif opt in ("-p", "--path"):
            cfg['path'] = arg
        elif opt in ("-s", "--listen"):
            cfg['listen'] = arg
        elif opt in ("-a", "--api"):
            cfg['api'] = arg
        elif opt in ("-u", "--user"):
            cfg['user'] = arg
        elif opt in ("-b", "--banner"):
            cfg['banner'] = arg
        elif opt in "--defaults":
            savecfg(cfg)
            sys.exit("%s updated" % (sys.argv[0]+'.cfg'))
        elif opt in "--debug":
            cfg['debug'] = True

def usage():
    print """Usage: {0} [options...][cfg file]\nCommand options are:\n-h,--help\tShow this help info\n-v,--version\tShow version info
--debug\t\tRun in foreground with logging to console
--defaults\tUpdate cfg and exit\nDefault files are {0}.cfg, {0}.log
\nThese options get saved in cfg file as defaults.
-s,--listen\tSet host:port for Electrum server\n-w,--www\tWeb server root directory\n-u,--user\tSet user to run as
-p,--path\tSet path for header data files (/var/data/sqlchain)
-b,--banner\tSet file path for banner text\n-a,--api\tSet host:port for API connection\n-l,--log\tSet log file path""".format(sys.argv[0])
    sys.exit(2)

def apicall(url, post=None):
    try:
        data = urlopen(sqc.cfg['api']+url, post).read()
    except URLError:
        logts("Error: sqlchain-api not at %s" % sqc.cfg['api'])
        return { 'error':'No api connection' }
    return json.loads(data)

def sigterm_handler(_signo, _stack_frame):
    logts("Shutting down.")
    if not sqc.cfg['debug']:
        os.unlink(sqc.cfg['pid'] if 'pid' in sqc.cfg else sys.argv[0]+'.pid')
    sys.exit(0)

def sighup_handler(_signo, _stack_frame):
    path = sqc.cfg['log'] if 'log' in sqc.cfg else sys.argv[0]+'.log'
    sys.stdout.close()
    sys.stdout=open(path,'a')
    sys.stderr.close()
    sys.stderr=open(path,'a')
    logts("SIGHUP Log reopened")

def run():
    monkey.patch_socket()

    with open(sqc.cfg['banner']) as bf:
        srvinfo['banner'] = bf.read()

    hdr = apicall('/block-index/latest/electrum')
    if 'error' in hdr:
        sys.exit(1)
    srvinfo['block'],srvinfo['header'] = hdr['block_height'],hdr

    sqc.reqQ = Queue()
    sqc.respQ = Queue()
    spawn(ReqHandler)
    spawn(RespHandler)
    spawn(SyncHandler)

    logts("Starting on %s" % sqc.cfg['listen'])
    host,port = sqc.cfg['listen'].split(':')
    cert = {'certfile':sqc.cfg['ssl']} if ('ssl' in sqc.cfg) and (sqc.cfg['ssl'] != '') else {}
    server = StreamServer((host, int(port)), TcpHandler, **cert)

    drop2user(sqc.cfg, chown=True)

    server.serve_forever()

if __name__ == '__main__':

    loadcfg(sqc.cfg)
    options(sqc.cfg)

    if sqc.cfg['debug']:
        signal.signal(signal.SIGINT, sigterm_handler)
        run()
    else:
        logpath = sqc.cfg['log'] if 'log' in sqc.cfg else sys.argv[0]+'.log'
        pidpath = sqc.cfg['pid'] if 'pid' in sqc.cfg else sys.argv[0]+'.pid'
        with daemon.DaemonContext(working_directory='.', umask=0002, stdout=open(logpath,'a'), stderr=open(logpath,'a'),
                signal_map={signal.SIGTERM:sigterm_handler, signal.SIGHUP:sighup_handler } ):
            with file(pidpath,'w') as f:
                f.write(str(os.getpid()))
            run()