X-Git-Url: https://git.novaco.in/?p=electrum-server.git;a=blobdiff_plain;f=server.py;h=d41bcf35e4af8362455304eb527a67527b60ae5b;hp=6ab05128ecb57176e5f797180d75f7d011676514;hb=4ce69b7ea24ead59ebbcc7ed335ea9762ae3724b;hpb=70f23254388eb74ca7a05009ccf2a5d7a723ec36

diff --git a/server.py b/server.py
index 6ab0512..d41bcf3 100755
--- a/server.py
+++ b/server.py
@@ -15,407 +15,185 @@
 # License along with this program.  If not, see
 # <http://www.gnu.org/licenses/agpl.html>.
 
-"""
-Todo:
-   * server should check and return bitcoind status..
-   * improve txpoint sorting
-   * command to check cache
-
- mempool transactions do not need to be added to the database; it slows it down
-"""
-
-import abe_backend
-
-
-
-
-import time, json, socket, operator, thread, ast, sys, re, traceback
 import ConfigParser
-from json import dumps, loads
-import urllib
+import logging
+import socket
+import sys
+import time
 import threading
+import traceback
 
-config = ConfigParser.ConfigParser()
-# set some defaults, which will be overwritten by the config file
-config.add_section('server')
-config.set('server','banner', 'Welcome to Electrum!')
-config.set('server', 'host', 'localhost')
-config.set('server', 'port', '50000')
-config.set('server', 'password', '')
-config.set('server', 'irc', 'yes')
-config.set('server', 'ircname', 'Electrum server')
-config.add_section('database')
-config.set('database', 'type', 'psycopg2')
-config.set('database', 'database', 'abe')
-
-try:
-    f = open('/etc/electrum.conf','r')
-    config.readfp(f)
-    f.close()
-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')
-stopping = False
-sessions = {}
-
-
-
-def random_string(N):
-    import random, string
-    return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
-
-
-
-def modified_addresses(a_session):
-    #t1 = time.time()
-    import copy
-    session = copy.deepcopy(a_session)
-    addresses = session['addresses']
-    session['last_time'] = time.time()
-    ret = {}
-    k = 0
-    for addr in addresses:
-        status = store.get_status( addr )
-        msg_id, last_status = addresses.get( addr )
-        if last_status != status:
-            addresses[addr] = msg_id, status
-            ret[addr] = status
-
-    #t2 = time.time() - t1 
-    #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2
-    return ret, addresses
-
-
-def poll_session(session_id): 
-    # native
-    session = sessions.get(session_id)
-    if session is None:
-        print time.asctime(), "session not found", session_id
-        return -1, {}
-    else:
-        sessions[session_id]['last_time'] = time.time()
-        ret, addresses = modified_addresses(session)
-        if ret: sessions[session_id]['addresses'] = addresses
-        return repr( (store.block_number,ret))
-
-
-def add_address_to_session(session_id, address):
-    status = store.get_status(address)
-    sessions[session_id]['addresses'][address] = ("", status)
-    sessions[session_id]['last_time'] = time.time()
-    return status
-
-
-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') ) )
-    sessions[session_id]['last_time'] = time.time()
-    return out
-
-
-def update_session(session_id,addresses):
-    """deprecated in 0.42, wad replaced by add_address_to_session"""
-    sessions[session_id]['addresses'] = {}
-    for a in addresses:
-        sessions[session_id]['addresses'][a] = ''
-    sessions[session_id]['last_time'] = time.time()
-    return 'ok'
-
-
-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()
-        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 native_client_thread(ipaddr,conn):
-    #print "client thread", ipaddr
-    try:
-        ipaddr = ipaddr[0]
-        msg = ''
-        while 1:
-            d = conn.recv(1024)
-            msg += d
-            if not d: 
-                break
-            if '#' in msg:
-                msg = msg.split('#', 1)[0]
-                break
-        try:
-            cmd, data = ast.literal_eval(msg)
-        except:
-            print "syntax error", repr(msg), ipaddr
-            conn.close()
-            return
+import json
 
-        out = do_command(cmd, data, ipaddr)
-        if out:
-            #print ipaddr, cmd, len(out)
-            try:
-                conn.send(out)
-            except:
-                print "error, could not send"
+logging.basicConfig()
 
-    finally:
-        conn.close()
+if sys.maxsize <= 2**32:
+    print "Warning: it looks like you are using a 32bit system. You may experience crashes caused by mmap"
 
 
-def timestr():
-    return time.strftime("[%d/%m/%Y-%H:%M:%S]")
-
-# used by the native handler
-def do_command(cmd, data, ipaddr):
+def attempt_read_config(config, filename):
+    try:
+        with open(filename, 'r') as f:
+            config.readfp(f)
+    except IOError:
+        pass
+
+
+def create_config():
+    config = ConfigParser.ConfigParser()
+    # set some defaults, which will be overwritten by the config file
+    config.add_section('server')
+    config.set('server', 'banner', 'Welcome to Electrum!')
+    config.set('server', 'host', 'localhost')
+    config.set('server', 'report_host', '')
+    config.set('server', 'stratum_tcp_port', '50001')
+    config.set('server', 'stratum_http_port', '8081')
+    config.set('server', 'stratum_tcp_ssl_port', '')
+    config.set('server', 'stratum_http_ssl_port', '')
+    config.set('server', 'report_stratum_tcp_port', '')
+    config.set('server', 'report_stratum_http_port', '')
+    config.set('server', 'report_stratum_tcp_ssl_port', '')
+    config.set('server', 'report_stratum_http_ssl_port', '')
+    config.set('server', 'ssl_certfile', '')
+    config.set('server', 'ssl_keyfile', '')
+    config.set('server', 'password', '')
+    config.set('server', 'irc', 'yes')
+    config.set('server', 'irc_nick', '')
+    config.set('server', 'coin', '')
+    config.set('server', 'datadir', '')
+
+    # use leveldb as default
+    config.set('server', 'backend', 'leveldb')
+    config.add_section('leveldb')
+    config.set('leveldb', 'path', '/dev/shm/electrum_db')
+    config.set('leveldb', 'pruning_limit', '100')
+
+    for path in ('/etc/', ''):
+        filename = path + 'electrum.conf'
+        attempt_read_config(config, filename)
 
-    if cmd=='b':
-        out = "%d"%block_number
+    try:
+        with open('/etc/electrum.banner', 'r') as f:
+            config.set('server', 'banner', f.read())
+    except IOError:
+        pass
 
-    elif cmd in ['session','new_session']:
-        try:
-            if cmd == 'session':
-                addresses = ast.literal_eval(data)
-                version = "old"
-            else:
-                version, addresses = ast.literal_eval(data)
-                if version[0]=="0": version = "v" + version
-        except:
-            print "error", data
-            return None
-        print timestr(), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
-        out = new_session(version, addresses)
+    return config
 
-    elif cmd=='address.subscribe':
-        try:
-            session_id, addr = ast.literal_eval(data)
-        except:
-            traceback.print_exc(file=sys.stdout)
-            print data
-            return None
-        out = add_address_to_session(session_id,addr)
 
-    elif cmd=='update_session':
-        try:
-            session_id, addresses = ast.literal_eval(data)
-        except:
-            traceback.print_exc(file=sys.stdout)
-            return None
-        print timestr(), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
-        out = update_session(session_id,addresses)
-            
-    elif cmd=='poll': 
-        out = poll_session(data)
+def run_rpc_command(command, stratum_tcp_port):
+    try:
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.connect((host, int(stratum_tcp_port)))
+    except:
+        print "cannot connect to server."
+        return
+
+    method = 'server.' + command
+    params = [password]
+    if len(sys.argv) > 2:
+        params.append(sys.argv[2])
+    
+    request = json.dumps({'id': 0, 'method': method, 'params': params})
+    s.send(request + '\n')
+    msg = ''
+    while True:
+        o = s.recv(1024)
+        if not o: break
+        msg += o
+        if msg.find('\n') != -1:
+            break
+    s.close()
+    r = json.loads(msg).get('result')
+
+    if command == 'info':
+        now = time.time()
+        print 'type           address         sub  version  time'
+        for item in r:
+            print '%4s   %21s   %3s  %7s  %.2f' % (item.get('name'),
+                                                   item.get('address'),
+                                                   item.get('subscriptions'),
+                                                   item.get('version'),
+                                                   (now - item.get('time')),
+                                                   )
+    else:
+        print r
 
-    elif cmd == 'h': 
-        address = data
-        out = repr( store.get_history( address ) )
 
-    elif cmd =='tx':
-        out = store.send_tx(data)
-        print timestr(), "sent tx:", ipaddr, out
+if __name__ == '__main__':
+    config = create_config()
+    password = config.get('server', 'password')
+    host = config.get('server', 'host')
+    stratum_tcp_port = config.get('server', 'stratum_tcp_port')
+    stratum_http_port = config.get('server', 'stratum_http_port')
+    stratum_tcp_ssl_port = config.get('server', 'stratum_tcp_ssl_port')
+    stratum_http_ssl_port = config.get('server', 'stratum_http_ssl_port')
+    ssl_certfile = config.get('server', 'ssl_certfile')
+    ssl_keyfile = config.get('server', 'ssl_keyfile')
+
+    if stratum_tcp_ssl_port or stratum_http_ssl_port:
+        assert ssl_certfile and ssl_keyfile
+
+    if len(sys.argv) > 1:
+        run_rpc_command(sys.argv[1], stratum_tcp_port)
+        sys.exit(0)
 
-    elif cmd == 'peers':
-        out = repr(irc.get_peers())
+    from processor import Dispatcher, print_log
+    from backends.irc import ServerProcessor
 
+    backend_name = config.get('server', 'backend')
+    if backend_name == 'libbitcoin':
+        from backends.libbitcoin import BlockchainProcessor
+    elif backend_name == 'leveldb':
+        from backends.bitcoind import BlockchainProcessor
     else:
-        out = None
-
-    return out
-
-
-def clean_session_thread():
-    while not stopping:
-        time.sleep(30)
-        t = time.time()
-        for k,s in sessions.items():
-            if s.get('type') == 'persistent': continue
-            t0 = s['last_time']
-            if t - t0 > 5*60:
-                sessions.pop(k)
-                print "lost session", k
-            
-
-####################################################################
-
-
-from processor import Shared, Processor, Dispatcher
-from stratum_http import HttpServer
-from stratum import TcpServer
-
-class AbeProcessor(Processor):
-    def process(self,request):
-        message_id = request['id']
-        method = request['method']
-        params = request.get('params',[])
-        #print request
-
-        result = ''
-        if method == 'numblocks.subscribe':
-            result = store.block_number
-        elif method == 'address.subscribe':
-            address = params[0]
-            store.watch_address(address)
-            status = store.get_status(address)
-            result = status
-        elif method == 'client.version':
-            #session.version = params[0]
-            pass
-        elif method == 'server.banner':
-            result = config.get('server','banner').replace('\\n','\n')
-        elif method == 'server.peers':
-            result = irc.get_peers()
-        elif method == 'address.get_history':
-            address = params[0]
-            result = store.get_history( address ) 
-        elif method == 'transaction.broadcast':
-            txo = store.send_tx(params[0])
-            print "sent tx:", txo
-            result = txo 
-        else:
-            print "unknown method", request
-
-        if result!='':
-            response = { 'id':message_id, 'method':method, 'params':params, 'result':result }
-            self.push_response(response)
-
-    def get_status(self,addr):
-        return store.get_status(addr)
-
-
-
-####################################################################
-
-
-
-class Irc(threading.Thread):
-
-    def __init__(self, processor):
-        self.processor = processor
-        threading.Thread.__init__(self)
-        self.daemon = True
-        self.peers = {}
-
-    def get_peers(self):
-        return self.peers.values()
-
-    def run(self):
-        NICK = 'E_'+random_string(10)
-        while not self.processor.shared.stopped():
-            try:
-                s = socket.socket()
-                s.connect(('irc.freenode.net', 6667))
-                s.send('USER electrum 0 * :'+config.get('server','host')+' '+config.get('server','ircname')+'\n')
-                s.send('NICK '+NICK+'\n')
-                s.send('JOIN #electrum\n')
-                sf = s.makefile('r', 0)
-                t = 0
-                while not self.processor.shared.stopped():
-                    line = sf.readline()
-                    line = line.rstrip('\r\n')
-                    line = line.split()
-                    if line[0]=='PING': 
-                        s.send('PONG '+line[1]+'\n')
-                    elif '353' in line: # answer to /names
-                        k = line.index('353')
-                        for item in line[k+1:]:
-                            if item[0:2] == 'E_':
-                                s.send('WHO %s\n'%item)
-                    elif '352' in line: # answer to /who
-                        # warning: this is a horrible hack which apparently works
-                        k = line.index('352')
-                        ip = line[k+4]
-                        ip = socket.gethostbyname(ip)
-                        name = line[k+6]
-                        host = line[k+9]
-                        self.peers[name] = (ip,host)
-                    if time.time() - t > 5*60:
-                        self.processor.push_response({'method':'server.peers', 'result':[self.get_peers()]})
-                        s.send('NAMES #electrum\n')
-                        t = time.time()
-                        self.peers = {}
-            except:
-                traceback.print_exc(file=sys.stdout)
-            finally:
-                sf.close()
-                s.close()
+        print "Unknown backend '%s' specified\n" % backend_name
+        sys.exit(1)
 
+    for i in xrange(5):
+        print ""
+    print_log("Starting Electrum server on", host)
 
+    # Create hub
+    dispatcher = Dispatcher(config)
+    shared = dispatcher.shared
 
+    # Create and register processors
+    chain_proc = BlockchainProcessor(config, shared)
+    dispatcher.register('blockchain', chain_proc)
 
-if __name__ == '__main__':
-
-    if len(sys.argv)>1:
-        import jsonrpclib
-        server = jsonrpclib.Server('http://%s:8081'%config.get('server','host'))
-        cmd = sys.argv[1]
-        if cmd == 'load':
-            out = server.load(password)
-        elif cmd == 'peers':
-            out = server.server.peers()
-        elif cmd == 'stop':
-            out = server.stop(password)
-        elif cmd == 'clear_cache':
-            out = server.clear_cache(password)
-        elif cmd == 'get_cache':
-            out = server.get_cache(password,sys.argv[2])
-        elif cmd == 'h':
-            out = server.address.get_history(sys.argv[2])
-        elif cmd == 'tx':
-            out = server.transaction.broadcast(sys.argv[2])
-        elif cmd == 'b':
-            out = server.numblocks.subscribe()
-        else:
-            out = "Unknown command: '%s'" % cmd
-        print out
-        sys.exit(0)
+    server_proc = ServerProcessor(config)
+    dispatcher.register('server', server_proc)
 
-    # backend
-    store = abe_backend.AbeStore(config)
-
-    # old protocol
-    thread.start_new_thread(native_server_thread, ())
-    thread.start_new_thread(clean_session_thread, ())
-
-    processor = AbeProcessor()
-    shared = Shared()
-    # Bind shared to processor since constructor is user defined
-    processor.shared = shared
-    processor.start()
-    # dispatcher
-    dispatcher = Dispatcher(shared, processor)
-    dispatcher.start()
+    transports = []
     # Create various transports we need
-    transports = [ TcpServer(shared, processor, "ecdsa.org",50001),
-                   HttpServer(shared, processor, "ecdsa.org",8081)
-                   ]
+    if stratum_tcp_port:
+        from transports.stratum_tcp import TcpServer
+        tcp_server = TcpServer(dispatcher, host, int(stratum_tcp_port), False, None, None)
+        transports.append(tcp_server)
+
+    if stratum_tcp_ssl_port:
+        from transports.stratum_tcp import TcpServer
+        tcp_server = TcpServer(dispatcher, host, int(stratum_tcp_ssl_port), True, ssl_certfile, ssl_keyfile)
+        transports.append(tcp_server)
+
+    if stratum_http_port:
+        from transports.stratum_http import HttpServer
+        http_server = HttpServer(dispatcher, host, int(stratum_http_port), False, None, None)
+        transports.append(http_server)
+
+    if stratum_http_ssl_port:
+        from transports.stratum_http import HttpServer
+        http_server = HttpServer(dispatcher, host, int(stratum_http_ssl_port), True, ssl_certfile, ssl_keyfile)
+        transports.append(http_server)
+
     for server in transports:
         server.start()
 
+    while not shared.stopped():
+        try:
+            time.sleep(1)
+        except:
+            shared.stop()
 
-    if (config.get('server','irc') == 'yes' ):
-	irc = Irc(processor)
-        irc.start()
-
-
-    print "starting Electrum server"
-    store.run(processor)
-    print "server stopped"
-
+    print_log("Electrum Server stopped")