X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=2d4118a60fe27fd365e3cca979023d7c3334ea57;hb=bb925d1cf82639af3808483b22ee810144a3f2f9;hp=59c3d0ae3ad9fc77585472384c00f9141ad26336;hpb=e24e733abb95471235065e66730ecff54af96e41;p=electrum-server.git

diff --git a/server.py b/server.py
index 59c3d0a..2d4118a 100755
--- a/server.py
+++ b/server.py
@@ -25,15 +25,22 @@ Todo:
 """
 
 
-import time, socket, operator, thread, ast, sys,re
+import time, json, socket, operator, thread, ast, sys,re
 import psycopg2, binascii
-import bitcoinrpc
 
 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 ConfigParser
+from json import dumps, loads
+import urllib
+
+# we need to import electrum
+sys.path.append('../client/')
+from wallet import Wallet
+from interface import Interface
+
 
 config = ConfigParser.ConfigParser()
 # set some defaults, which will be overwritten by the config file
@@ -56,16 +63,32 @@ try:
 except:
     print "Could not read electrum.conf. I will use the default values."
 
+try:
+    f = open('/etc/electrum.banner','r')
+    config.set('server','banner', f.read())
+    f.close()
+except:
+    pass
+
 password = config.get('server','password')
+bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port'))
 
 stopping = False
 block_number = -1
+old_block_number = -1
 sessions = {}
+sessions_sub_numblocks = [] # sessions that have subscribed to the service
+
 dblock = thread.allocate_lock()
 peer_list = {}
 
 wallets = {} # for ultra-light clients such as bccapi
 
+from Queue import Queue
+input_queue = Queue()
+output_queue = Queue()
+address_queue = Queue()
+
 class MyStore(Datastore_class):
 
     def import_tx(self, tx, is_coinbase):
@@ -80,6 +103,8 @@ class MyStore(Datastore_class):
             if self.tx_cache.has_key(address):
                 print "cache: invalidating", address
                 self.tx_cache.pop(address)
+            address_queue.put(address)
+
         outrows = self.get_tx_outputs(txid, False)
         for row in outrows:
             _hash = store.binout(row[6])
@@ -87,6 +112,7 @@ class MyStore(Datastore_class):
             if self.tx_cache.has_key(address):
                 print "cache: invalidating", address
                 self.tx_cache.pop(address)
+            address_queue.put(address)
 
     def safe_sql(self,sql, params=(), lock=True):
         try:
@@ -202,8 +228,8 @@ class MyStore(Datastore_class):
     def get_history(self, addr):
         
         if config.get('server','cache') == 'yes':
-            cached_version = self.tx_cache.get( addr ) 
-            if cached_version is not None: 
+            cached_version = self.tx_cache.get( addr )
+            if cached_version is not None:
                 return cached_version
 
         version, binaddr = decode_check_address(addr)
@@ -227,7 +253,6 @@ class MyStore(Datastore_class):
             tx_hash = self.hashout_hex(tx_hash)
             txpoint = {
                     "nTime":    int(nTime),
-                    #"chain_id": int(chain_id),
                     "height":   int(height),
                     "is_in":    int(is_in),
                     "blk_hash": self.hashout_hex(blk_hash),
@@ -266,7 +291,6 @@ class MyStore(Datastore_class):
             #print "mempool", tx_hash
             txpoint = {
                     "nTime":    0,
-                    #"chain_id": 1,
                     "height":   0,
                     "is_in":    int(is_in),
                     "blk_hash": 'mempool', 
@@ -317,14 +341,33 @@ class MyStore(Datastore_class):
 
 
 
+class Direct_Interface(Interface):
+    def __init__(self):
+        pass
+
+    def handler(self, method, params = ''):
+        cmds = {'session.new':new_session,
+                'session.poll':poll_session,
+                'session.update':update_session,
+                'blockchain.transaction.broadcast':send_tx,
+                'blockchain.address.get_history':store.get_history
+                }
+        func = cmds[method]
+        return func( params )
+
+
+
 def send_tx(tx):
-    import bitcoinrpc
-    conn = bitcoinrpc.connect_to_local()
-    try:
-        v = conn.importtransaction(tx)
-    except:
-        v = "error: transaction rejected by memorypool"
-    return v
+    postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'})
+    respdata = urllib.urlopen(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 random_string(N):
     import random, string
@@ -361,10 +404,10 @@ def get_cache(pw,addr):
         return 'wrong password'
 
 
-def cmd_poll(session_id):
+def poll_session(session_id):
     session = sessions.get(session_id)
     if session is None:
-        print time.asctime(), "session not found", session_id, ipaddr
+        print time.asctime(), "session not found", session_id
         out = repr( (-1, {}))
     else:
         t1 = time.time()
@@ -374,18 +417,7 @@ def cmd_poll(session_id):
         k = 0
         for addr in addresses:
             if store.tx_cache.get( addr ) is not None: k += 1
-
-            # get addtess status, i.e. the last block for that address.
-            tx_points = store.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)
-
+            status = get_address_status( addr )
             last_status = addresses.get( addr )
             if last_status != status:
                 addresses[addr] = status
@@ -400,12 +432,62 @@ def cmd_poll(session_id):
         return out
 
 
-def new_session(addresses, version, ipaddr):
-    session_id = random_string(10)
+def do_update_address(addr):
+    # an address was involved in a transaction; we check if it was subscribed to in a session
+    # the address can be subscribed in several sessions; the cache should ensure that we don't do redundant requests
+    for session_id in sessions.keys():
+        session = sessions[session_id]
+        if session.get('type') != 'subscribe': continue
+        addresses = session['addresses'].keys()
+
+        if addr in addresses:
+            print "address ", addr, "found in session", session_id
+            status = get_address_status( addr )
+            print "new_status:", status
+            last_status = session['addresses'][addr]
+            print "last_status", last_status
+            if last_status != status:
+                print "status is new", addr
+                send_status(session_id,addr,status)
+                sessions[session_id]['addresses'][addr] = status
+
 
-    print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
+def get_address_status(addr):
+    # get address status, i.e. the last block for that address.
+    tx_points = store.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 send_numblocks(session_id):
+    out = json.dumps( {'method':'numblocks.subscribe', 'result':block_number} )
+    output_queue.put((session_id, out))
+
+def send_status(session_id, address, status):
+    out = json.dumps( { 'method':'address.subscribe', 'address':address, 'status':status } )
+    output_queue.put((session_id, out))
+
+def subscribe_to_numblocks(session_id):
+    sessions_sub_numblocks.append(session_id)
+    send_numblocks(session_id)
+
+def subscribe_to_address(session_id, address):
+    status = get_address_status(address)
+    sessions[session_id]['type'] = 'subscribe'
+    sessions[session_id]['addresses'][address] = status
+    sessions[session_id]['last_time'] = time.time()
+    send_status(session_id, address, status)
 
-    sessions[session_id] = { 'addresses':{}, 'version':version, 'ip':ipaddr }
+def new_session(version, addresses):
+    session_id = random_string(10)
+    sessions[session_id] = { 'addresses':{}, 'version':version }
     for a in addresses:
         sessions[session_id]['addresses'][a] = ''
     out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) )
@@ -413,26 +495,27 @@ def new_session(addresses, version, ipaddr):
     return out
 
 def update_session(session_id,addresses):
-    print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
     sessions[session_id]['addresses'] = {}
     for a in addresses:
         sessions[session_id]['addresses'][a] = ''
-    out = 'ok'
     sessions[session_id]['last_time'] = time.time()
+    return 'ok'
 
-
-def listen_thread(store):
+def native_server_thread():
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     s.bind((config.get('server','host'), config.getint('server','port')))
     s.listen(1)
     while not stopping:
         conn, addr = s.accept()
-        thread.start_new_thread(client_thread, (addr, conn,))
-
+        try:
+            thread.start_new_thread(native_client_thread, (addr, conn,))
+        except:
+            # can't start new thread if there is no memory..
+            traceback.print_exc(file=sys.stdout)
 
 
-def client_thread(ipaddr,conn):
+def native_client_thread(ipaddr,conn):
     #print "client thread", ipaddr
     try:
         ipaddr = ipaddr[0]
@@ -464,8 +547,12 @@ def client_thread(ipaddr,conn):
         conn.close()
 
 
+
+# used by the native handler
 def do_command(cmd, data, ipaddr):
 
+    timestr = time.strftime("[%d/%m/%Y-%H:%M:%S]")
+
     if cmd=='b':
         out = "%d"%block_number
 
@@ -480,7 +567,8 @@ def do_command(cmd, data, ipaddr):
         except:
             print "error", data
             return None
-        out = new_session(addresses, version, ipaddr)
+        print timestr, "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
+        out = new_session(version, addresses)
 
     elif cmd=='update_session':
         try:
@@ -488,9 +576,9 @@ def do_command(cmd, data, ipaddr):
         except:
             print "error"
             return None
+        print timestr, "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
         out = update_session(session_id,addresses)
 
-
     elif cmd == 'bccapi_login':
         import electrum
         print "data",data
@@ -498,7 +586,7 @@ def do_command(cmd, data, ipaddr):
         master_public_key = k.decode('hex') # todo: sanitize. no need to decode twice...
         print master_public_key
         wallet_id = random_string(10)
-        w = electrum.Wallet()
+        w = Wallet( Direct_Interface() )
         w.master_public_key = master_public_key.decode('hex')
         w.synchronize()
         wallets[wallet_id] = w
@@ -506,7 +594,7 @@ def do_command(cmd, data, ipaddr):
         print "wallets", wallets
 
     elif cmd == 'bccapi_getAccountInfo':
-        from electrum import int_to_hex
+        from wallet import int_to_hex
         v, wallet_id = ast.literal_eval(data)
         w = wallets.get(wallet_id)
         if w is not None:
@@ -519,7 +607,7 @@ def do_command(cmd, data, ipaddr):
             out = "error"
 
     elif cmd == 'bccapi_getAccountStatement':
-        from electrum import int_to_hex
+        from wallet import int_to_hex
         v, wallet_id = ast.literal_eval(data)
         w = wallets.get(wallet_id)
         if w is not None:
@@ -539,7 +627,7 @@ def do_command(cmd, data, ipaddr):
         out = ''
             
     elif cmd=='poll': 
-        out = cmd_poll(data)
+        out = poll_session(data)
 
     elif cmd == 'h': 
         # history
@@ -551,7 +639,7 @@ def do_command(cmd, data, ipaddr):
 
     elif cmd =='tx':
         out = send_tx(data)
-        print "sent tx:", out
+        print timestr, "sent tx:", ipaddr, out
 
     elif cmd == 'stop':
         out = cmd_stop(data)
@@ -566,17 +654,123 @@ def do_command(cmd, data, ipaddr):
 
 
 
+####################################################################
+
+def tcp_server_thread():
+    thread.start_new_thread(process_input_queue, ())
+    thread.start_new_thread(process_output_queue, ())
+
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.bind((config.get('server','host'), 50001))
+    s.listen(1)
+    while not stopping:
+        conn, addr = s.accept()
+        try:
+            thread.start_new_thread(tcp_client_thread, (addr, conn,))
+        except:
+            # can't start new thread if there is no memory..
+            traceback.print_exc(file=sys.stdout)
+
+
+def close_sesion(session_id):
+    print "lost connection", session_id
+    sessions.pop(session_id)
+    sessions_sub_numblocks.remove(session_id)
+
+
+# one thread per client. put requests in a queue.
+def tcp_client_thread(ipaddr,conn):
+    """ use a persistent connection. put commands in a queue."""
+    print "persistent client thread", ipaddr
+    global sessions
+
+    session_id = random_string(10)
+    sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown' }
+
+    ipaddr = ipaddr[0]
+    msg = ''
+
+    while not stopping:
+        d = conn.recv(1024)
+        msg += d
+        if not d:
+            close_sesion(session_id)
+            break
+
+        while True:
+            s = msg.find('\n')
+            if s ==-1:
+                break
+            else:
+                c = msg[0:s]
+                msg = msg[s+1:]
+                c = json.loads(c)
+                try:
+                    cmd = c['method']
+                    data = c['params']
+                except:
+                    print "syntax error", repr(c), ipaddr
+                    continue
+
+                # add to queue
+                input_queue.put((session_id, cmd, data))
+
+
+# read commands from the input queue. perform requests, etc. this should be called from the main thread.
+def process_input_queue():
+    while not stopping:
+        session_id, cmd, data = input_queue.get()
+        out = None
+        if cmd == 'address.subscribe':
+            subscribe_to_address(session_id,data)
+        elif cmd == 'numblocks.subscribe':
+            subscribe_to_numblocks(session_id)
+        elif cmd == 'client.version':
+            sessions[session_id]['version'] = data
+        elif cmd == 'server.banner':
+            out = json.dumps( { 'method':'server.banner', 'result':config.get('server','banner').replace('\\n','\n') } )
+        elif cmd == 'address.get_history':
+            address = data
+            out = json.dumps( { 'method':'address.get_history', 'address':address, 'result':store.get_history( address ) } )
+        elif cmd == 'transaction.broadcast':
+            txo = send_tx(data)
+            print "sent tx:", txo
+            out = json.dumps( { 'method':'transaction.broadcast', 'result':txo } )
+        else:
+            print "unknown command", cmd
+        if out:
+            output_queue.put((session_id, out))
+
+# this is a separate thread
+def process_output_queue():
+    while not stopping:
+        session_id, out = output_queue.get()
+        session = sessions.get(session_id)
+        if session: 
+            try:
+                conn = session.get('conn')
+                conn.send(out+'\n')
+            except:
+                close_session(session_id)
+                
+
+
+
+####################################################################
+
 
 def memorypool_update(store):
     ds = BCDataStream.BCDataStream()
     store.mempool_keys = []
-    conn = bitcoinrpc.connect_to_local()
-    try:
-        v = conn.getmemorypool()
-    except:
-        print "cannot contact bitcoin daemon"
+
+    postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'})
+    respdata = urllib.urlopen(bitcoind_url, postdata).read()
+    r = loads(respdata)
+    if r['error'] != None:
         return
-    v = v['transactions']
+
+    v = r['result'].get('transactions')
     for hextx in v:
         ds.clear()
         ds.write(hextx.decode('hex'))
@@ -598,10 +792,11 @@ def clean_session_thread():
         time.sleep(30)
         t = time.time()
         for k,s in sessions.items():
+            if s.get('type') == 'subscribe': continue
             t0 = s['last_time']
             if t - t0 > 5*60:
-                print time.strftime("[%d/%m/%Y-%H:%M:%S]"), "end session", s['ip']
                 sessions.pop(k)
+                print "lost session", k
             
 
 def irc_thread():
@@ -647,17 +842,23 @@ def irc_thread():
 
 
 
-def jsonrpc_thread(store):
+def http_server_thread(store):
     # see http://code.google.com/p/jsonrpclib/
+    from SocketServer import ThreadingMixIn
     from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
-    server = SimpleJSONRPCServer(('localhost', 8080))
-    server.register_function(store.get_history, 'history')
+    class SimpleThreadedJSONRPCServer(ThreadingMixIn, SimpleJSONRPCServer): pass
+    server = SimpleThreadedJSONRPCServer(( config.get('server','host'), 8081))
     server.register_function(lambda : peer_list.values(), 'peers')
     server.register_function(cmd_stop, 'stop')
     server.register_function(cmd_load, 'load')
     server.register_function(lambda : block_number, 'blocks')
     server.register_function(clear_cache, 'clear_cache')
     server.register_function(get_cache, 'get_cache')
+    server.register_function(send_tx, 'blockchain.transaction.broadcast')
+    server.register_function(store.get_history, 'blockchain.address.get_history')
+    server.register_function(new_session, 'session.new')
+    server.register_function(update_session, 'session.update')
+    server.register_function(poll_session, 'session.poll')
     server.serve_forever()
 
 
@@ -668,7 +869,7 @@ if __name__ == '__main__':
 
     if len(sys.argv)>1:
         import jsonrpclib
-        server = jsonrpclib.Server('http://localhost:8080')
+        server = jsonrpclib.Server('http://%s:8081'%config.get('server','host'))
         cmd = sys.argv[1]
         if cmd == 'load':
             out = server.load(password)
@@ -681,9 +882,13 @@ if __name__ == '__main__':
         elif cmd == 'get_cache':
             out = server.get_cache(password,sys.argv[2])
         elif cmd == 'h':
-            out = server.history(sys.argv[2])
+            out = server.blockchain.address.get_history(sys.argv[2])
+        elif cmd == 'tx':
+            out = server.blockchain.transaction.broadcast(sys.argv[2])
         elif cmd == 'b':
             out = server.blocks()
+        else:
+            out = "Unknown command: '%s'" % cmd
         print out
         sys.exit(0)
 
@@ -704,9 +909,13 @@ if __name__ == '__main__':
     store.tx_cache = {}
     store.mempool_keys = {}
 
-    thread.start_new_thread(listen_thread, (store,))
-    thread.start_new_thread(jsonrpc_thread, (store,))
+    # supported protocols
+    thread.start_new_thread(native_server_thread, ())
+    thread.start_new_thread(tcp_server_thread, ())
+    thread.start_new_thread(http_server_thread, (store,))
+
     thread.start_new_thread(clean_session_thread, ())
+
     if (config.get('server','irc') == 'yes' ):
 	thread.start_new_thread(irc_thread, ())
 
@@ -716,9 +925,29 @@ if __name__ == '__main__':
             store.catch_up()
             memorypool_update(store)
             block_number = store.get_block_number(1)
-            dblock.release()
+
+            if block_number != old_block_number:
+                old_block_number = block_number
+                for session_id in sessions_sub_numblocks:
+                    send_numblocks(session_id)
+
+        except IOError:
+            print "IOError: cannot reach bitcoind"
+            block_number = 0
         except:
             traceback.print_exc(file=sys.stdout)
+            block_number = 0
+        finally:
+            dblock.release()
+
+        # do addresses
+        while True:
+            try:
+                addr = address_queue.get(False)
+            except:
+                break
+            do_update_address(addr)
+
         time.sleep(10)
 
     print "server stopped"