X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=transports%2Fstratum_http.py;h=f60b3209890b27b118099f526828aee30a5c33e0;hb=6acbe2b434894de08ec5a79d5e414c9d62dd50b2;hp=ef738e36d56600a638dd3a15fc2d0649ba8730bd;hpb=289ccdf9d4eb9b686d26b6482992cd4ad078df9f;p=electrum-server.git diff --git a/transports/stratum_http.py b/transports/stratum_http.py index ef738e3..f60b320 100644 --- a/transports/stratum_http.py +++ b/transports/stratum_http.py @@ -22,11 +22,13 @@ import SimpleXMLRPCServer import SocketServer import socket import logging -import os +import os, time import types import traceback import sys, threading +from OpenSSL import SSL + try: import fcntl except ImportError: @@ -47,7 +49,7 @@ from the processor point of view: """ -from processor import random_string +from processor import random_string, print_log def get_version(request): @@ -100,6 +102,7 @@ class StratumJSONRPCDispatcher(SimpleXMLRPCServer.SimpleXMLRPCDispatcher): session = self.dispatcher.get_session_by_address(session_id) if not session: return 'Error: session not found' + session.time = time.time() responses = [] if type(request) is not types.ListType: @@ -111,7 +114,7 @@ class StratumJSONRPCDispatcher(SimpleXMLRPCServer.SimpleXMLRPCDispatcher): responses.append(result.response()) continue - self.dispatcher.process(session, req_entry) + self.dispatcher.do_dispatch(session, req_entry) if req_entry['method'] == 'server.stop': return json.dumps({'result':'ok'}) @@ -130,10 +133,35 @@ class StratumJSONRPCDispatcher(SimpleXMLRPCServer.SimpleXMLRPCDispatcher): return response + def create_session(self): + session_id = random_string(10) + session = HttpSession(session_id) + self.dispatcher.add_session(session) + return session_id + + def poll_session(self, session): + q = session.pending_responses + responses = [] + while not q.empty(): + r = q.get() + responses.append(r) + #print "poll: %d responses"%len(responses) + return responses + + + class StratumJSONRPCRequestHandler( SimpleXMLRPCServer.SimpleXMLRPCRequestHandler): - + + def do_OPTIONS(self): + self.send_response(200) + self.send_header('Allow', 'GET, POST, OPTIONS') + self.send_header('Access-Control-Allow-Origin', '*') + self.send_header('Access-Control-Allow-Headers', '*') + self.send_header('Content-Length', '0') + self.end_headers() + def do_GET(self): if not self.is_rpc_path_valid(): self.report_404() @@ -172,7 +200,7 @@ class StratumJSONRPCRequestHandler( self.end_headers() self.wfile.write(response) self.wfile.flush() - self.connection.shutdown(1) + self.shutdown_connection() def do_POST(self): @@ -221,10 +249,39 @@ class StratumJSONRPCRequestHandler( self.end_headers() self.wfile.write(response) self.wfile.flush() + self.shutdown_connection() + + def shutdown_connection(self): self.connection.shutdown(1) -class StratumJSONRPCServer(SocketServer.TCPServer, StratumJSONRPCDispatcher): +class SSLRequestHandler(StratumJSONRPCRequestHandler): + def setup(self): + self.connection = self.request + self.rfile = socket._fileobject(self.request, "rb", self.rbufsize) + self.wfile = socket._fileobject(self.request, "wb", self.wbufsize) + + def shutdown_connection(self): + self.connection.shutdown() + + +class SSLTCPServer(SocketServer.TCPServer): + def __init__(self, server_address, certfile, keyfile, RequestHandlerClass, bind_and_activate=True): + SocketServer.BaseServer.__init__(self, server_address, RequestHandlerClass) + ctx = SSL.Context(SSL.SSLv3_METHOD) + ctx.use_privatekey_file(keyfile) + ctx.use_certificate_file(certfile) + self.socket = SSL.Connection(ctx, socket.socket(self.address_family, self.socket_type)) + if bind_and_activate: + self.server_bind() + self.server_activate() + + def shutdown_request(self,request): + #request.shutdown() + pass + + +class StratumHTTPServer(SocketServer.TCPServer, StratumJSONRPCDispatcher): allow_reuse_address = True @@ -246,12 +303,42 @@ class StratumJSONRPCServer(SocketServer.TCPServer, StratumJSONRPCDispatcher): os.unlink(addr) except OSError: logging.warning("Could not unlink socket %s", addr) - # if python 2.5 and lower - if vi[0] < 3 and vi[1] < 6: - SocketServer.TCPServer.__init__(self, addr, requestHandler) - else: - SocketServer.TCPServer.__init__(self, addr, requestHandler, - bind_and_activate) + + SocketServer.TCPServer.__init__(self, addr, requestHandler, bind_and_activate) + + if fcntl is not None and hasattr(fcntl, 'FD_CLOEXEC'): + flags = fcntl.fcntl(self.fileno(), fcntl.F_GETFD) + flags |= fcntl.FD_CLOEXEC + fcntl.fcntl(self.fileno(), fcntl.F_SETFD, flags) + + +class StratumHTTPSSLServer(SSLTCPServer, StratumJSONRPCDispatcher): + + allow_reuse_address = True + + def __init__(self, addr, certfile, keyfile, + requestHandler=SSLRequestHandler, + logRequests=False, encoding=None, bind_and_activate=True, + address_family=socket.AF_INET): + + self.logRequests = logRequests + StratumJSONRPCDispatcher.__init__(self, encoding) + # TCPServer.__init__ has an extra parameter on 2.6+, so + # check Python version and decide on how to call it + vi = sys.version_info + self.address_family = address_family + if USE_UNIX_SOCKETS and address_family == socket.AF_UNIX: + # Unix sockets can't be bound if they already exist in the + # filesystem. The convention of e.g. X11 is to unlink + # before binding again. + if os.path.exists(addr): + try: + os.unlink(addr) + except OSError: + logging.warning("Could not unlink socket %s", addr) + + SSLTCPServer.__init__(self, addr, certfile, keyfile, requestHandler, bind_and_activate) + if fcntl is not None and hasattr(fcntl, 'FD_CLOEXEC'): flags = fcntl.fcntl(self.fileno(), fcntl.F_GETFD) flags |= fcntl.FD_CLOEXEC @@ -259,20 +346,7 @@ class StratumJSONRPCServer(SocketServer.TCPServer, StratumJSONRPCDispatcher): - def create_session(self): - session_id = random_string(10) - session = HttpSession(session_id) - self.dispatcher.add_session(session) - return session_id - def poll_session(self, session): - q = session.pending_responses - responses = [] - while not q.empty(): - r = q.get() - responses.append(r) - #print "poll: %d responses"%len(responses) - return responses from processor import Session @@ -284,32 +358,47 @@ class HttpSession(Session): Session.__init__(self) self.pending_responses = Queue.Queue() self.address = session_id - self.name = "HTTP session" + self.name = "HTTP" def send_response(self, response): raw_response = json.dumps(response) self.pending_responses.put(response) + def stopped(self): + with self.lock: + if time.time() - self.time > 60: + self._stopped = True + return self._stopped + class HttpServer(threading.Thread): - def __init__(self, dispatcher, host, port): + def __init__(self, dispatcher, host, port, use_ssl, certfile, keyfile): self.shared = dispatcher.shared self.dispatcher = dispatcher.request_dispatcher threading.Thread.__init__(self) self.daemon = True self.host = host self.port = port + self.use_ssl = use_ssl + self.certfile = certfile + self.keyfile = keyfile self.lock = threading.Lock() + def run(self): # see http://code.google.com/p/jsonrpclib/ from SocketServer import ThreadingMixIn - class StratumThreadedJSONRPCServer(ThreadingMixIn, StratumJSONRPCServer): pass + if self.use_ssl: + class StratumThreadedServer(ThreadingMixIn, StratumHTTPSSLServer): pass + self.server = StratumThreadedServer(( self.host, self.port), self.certfile, self.keyfile) + print_log( "HTTPS server started.") + else: + class StratumThreadedServer(ThreadingMixIn, StratumHTTPServer): pass + self.server = StratumThreadedServer(( self.host, self.port)) + print_log( "HTTP server started.") - self.server = StratumThreadedJSONRPCServer(( self.host, self.port)) self.server.dispatcher = self.dispatcher self.server.register_function(None, 'server.stop') self.server.register_function(None, 'server.info') - print "HTTP server started." self.server.serve_forever()