transports/stratum_http.py

   1 #!/usr/bin/env python
   2 # Copyright(C) 2012 thomasv@gitorious
   3
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as
   6 # published by the Free Software Foundation, either version 3 of the
   7 # License, or (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful, but
  10 # WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public
  15 # License along with this program.  If not, see
  16 # <http://www.gnu.org/licenses/agpl.html>.
  17
  18 import jsonrpclib
  19 from jsonrpclib import Fault
  20 from jsonrpclib.jsonrpc import USE_UNIX_SOCKETS
  21 import SimpleXMLRPCServer
  22 import SocketServer
  23 import socket
  24 import logging
  25 import os
  26 import types
  27 import traceback
  28 import sys, threading
  29
  30 try:
  31     import fcntl
  32 except ImportError:
  33     # For Windows
  34     fcntl = None
  35
  36 import json
  37
  38
  39 """
  40 sessions are identified with cookies
  41  - each session has a buffer of responses to requests
  42
  43
  44 from the processor point of view:
  45  - the user only defines process() ; the rest is session management.  thus sessions should not belong to processor
  46
  47 """
  48
  49
  50 def random_string(N):
  51     import random, string
  52     return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
  53
  54
  55
  56
  57 def get_version(request):
  58     # must be a dict
  59     if 'jsonrpc' in request.keys():
  60         return 2.0
  61     if 'id' in request.keys():
  62         return 1.0
  63     return None
  64
  65 def validate_request(request):
  66     if type(request) is not types.DictType:
  67         fault = Fault(
  68             -32600, 'Request must be {}, not %s.' % type(request)
  69         )
  70         return fault
  71     rpcid = request.get('id', None)
  72     version = get_version(request)
  73     if not version:
  74         fault = Fault(-32600, 'Request %s invalid.' % request, rpcid=rpcid)
  75         return fault
  76     request.setdefault('params', [])
  77     method = request.get('method', None)
  78     params = request.get('params')
  79     param_types = (types.ListType, types.DictType, types.TupleType)
  80     if not method or type(method) not in types.StringTypes or \
  81         type(params) not in param_types:
  82         fault = Fault(
  83             -32600, 'Invalid request parameters or method.', rpcid=rpcid
  84         )
  85         return fault
  86     return True
  87
  88 class StratumJSONRPCDispatcher(SimpleXMLRPCServer.SimpleXMLRPCDispatcher):
  89
  90     def __init__(self, encoding=None):
  91         SimpleXMLRPCServer.SimpleXMLRPCDispatcher.__init__(self,
  92                                         allow_none=True,
  93                                         encoding=encoding)
  94
  95     def _marshaled_dispatch(self, session_id, data, dispatch_method = None):
  96         response = None
  97         try:
  98             request = jsonrpclib.loads(data)
  99         except Exception, e:
 100             fault = Fault(-32700, 'Request %s invalid. (%s)' % (data, e))
 101             response = fault.response()
 102             return response
 103
 104         responses = []
 105         if type(request) is not types.ListType:
 106             request = [ request ]
 107
 108         for req_entry in request:
 109             result = validate_request(req_entry)
 110             if type(result) is Fault:
 111                 responses.append(result.response())
 112                 continue
 113
 114             session = self.sessions.get(session_id)
 115             if session:
 116                 self.dispatcher.process(session, req_entry)
 117                 if req_entry['method'] == 'server.stop':
 118                     return json.dumps({'result':'ok'})
 119
 120
 121         r = self.poll_session(session_id)
 122         for item in r:
 123             responses.append(json.dumps(item))
 124
 125         if len(responses) > 1:
 126             response = '[%s]' % ','.join(responses)
 127         elif len(responses) == 1:
 128             response = responses[0]
 129         else:
 130             response = ''
 131
 132         return response
 133
 134
 135
 136 class StratumJSONRPCRequestHandler(
 137         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 138
 139     def do_GET(self):
 140         if not self.is_rpc_path_valid():
 141             self.report_404()
 142             return
 143         try:
 144             session_id = None
 145             c = self.headers.get('cookie')
 146             if c:
 147                 if c[0:8]=='SESSION=':
 148                     #print "found cookie", c[8:]
 149                     session_id = c[8:]
 150
 151             if session_id is None:
 152                 session_id = self.server.create_session()
 153                 #print "setting cookie", session_id
 154
 155             data = json.dumps([])
 156             response = self.server._marshaled_dispatch(session_id, data)
 157             self.send_response(200)
 158         except Exception, e:
 159             self.send_response(500)
 160             err_lines = traceback.format_exc().splitlines()
 161             trace_string = '%s | %s' % (err_lines[-3], err_lines[-1])
 162             fault = jsonrpclib.Fault(-32603, 'Server error: %s' % trace_string)
 163             response = fault.response()
 164             print "500", trace_string
 165         if response == None:
 166             response = ''
 167
 168         if session_id:
 169             self.send_header("Set-Cookie", "SESSION=%s"%session_id)
 170
 171         self.send_header("Content-type", "application/json-rpc")
 172         self.send_header("Content-length", str(len(response)))
 173         self.end_headers()
 174         self.wfile.write(response)
 175         self.wfile.flush()
 176         self.connection.shutdown(1)
 177
 178
 179     def do_POST(self):
 180         if not self.is_rpc_path_valid():
 181             self.report_404()
 182             return
 183         try:
 184             max_chunk_size = 10*1024*1024
 185             size_remaining = int(self.headers["content-length"])
 186             L = []
 187             while size_remaining:
 188                 chunk_size = min(size_remaining, max_chunk_size)
 189                 L.append(self.rfile.read(chunk_size))
 190                 size_remaining -= len(L[-1])
 191             data = ''.join(L)
 192
 193             session_id = None
 194             c = self.headers.get('cookie')
 195             if c:
 196                 if c[0:8]=='SESSION=':
 197                     print "found cookie", c[8:]
 198                     session_id = c[8:]
 199
 200             if session_id is None:
 201                 session_id = self.server.create_session()
 202                 print "setting cookie", session_id
 203
 204             response = self.server._marshaled_dispatch(session_id, data)
 205             self.send_response(200)
 206         except Exception, e:
 207             self.send_response(500)
 208             err_lines = traceback.format_exc().splitlines()
 209             trace_string = '%s | %s' % (err_lines[-3], err_lines[-1])
 210             fault = jsonrpclib.Fault(-32603, 'Server error: %s' % trace_string)
 211             response = fault.response()
 212             print "500", trace_string
 213         if response == None:
 214             response = ''
 215
 216         if session_id:
 217             self.send_header("Set-Cookie", "SESSION=%s"%session_id)
 218
 219         self.send_header("Content-type", "application/json-rpc")
 220         self.send_header("Content-length", str(len(response)))
 221         self.end_headers()
 222         self.wfile.write(response)
 223         self.wfile.flush()
 224         self.connection.shutdown(1)
 225
 226
 227 class StratumJSONRPCServer(SocketServer.TCPServer, StratumJSONRPCDispatcher):
 228
 229     allow_reuse_address = True
 230
 231     def __init__(self, addr, requestHandler=StratumJSONRPCRequestHandler,
 232                  logRequests=False, encoding=None, bind_and_activate=True,
 233                  address_family=socket.AF_INET):
 234         self.logRequests = logRequests
 235         StratumJSONRPCDispatcher.__init__(self, encoding)
 236         # TCPServer.__init__ has an extra parameter on 2.6+, so
 237         # check Python version and decide on how to call it
 238         vi = sys.version_info
 239         self.address_family = address_family
 240         if USE_UNIX_SOCKETS and address_family == socket.AF_UNIX:
 241             # Unix sockets can't be bound if they already exist in the
 242             # filesystem. The convention of e.g. X11 is to unlink
 243             # before binding again.
 244             if os.path.exists(addr):
 245                 try:
 246                     os.unlink(addr)
 247                 except OSError:
 248                     logging.warning("Could not unlink socket %s", addr)
 249         # if python 2.5 and lower
 250         if vi[0] < 3 and vi[1] < 6:
 251             SocketServer.TCPServer.__init__(self, addr, requestHandler)
 252         else:
 253             SocketServer.TCPServer.__init__(self, addr, requestHandler,
 254                 bind_and_activate)
 255         if fcntl is not None and hasattr(fcntl, 'FD_CLOEXEC'):
 256             flags = fcntl.fcntl(self.fileno(), fcntl.F_GETFD)
 257             flags |= fcntl.FD_CLOEXEC
 258             fcntl.fcntl(self.fileno(), fcntl.F_SETFD, flags)
 259
 260         self.sessions = {}
 261
 262
 263
 264     def create_session(self):
 265         session_id = random_string(10)
 266         self.sessions[session_id] = HttpSession(session_id)
 267         return session_id
 268
 269     def poll_session(self,session_id):
 270         responses = self.sessions[session_id].pending_responses[:]
 271         self.sessions[session_id].pending_responses = []
 272         print "poll: %d responses"%len(responses)
 273         return responses
 274
 275
 276 from processor import Session
 277
 278 class HttpSession(Session):
 279
 280     def __init__(self, session_id):
 281         Session.__init__(self)
 282         self.pending_responses = []
 283         print "new http session", session_id
 284
 285     def send_response(self, response):
 286         raw_response = json.dumps(response)
 287         self.pending_responses.append(response)
 288
 289 class HttpServer(threading.Thread):
 290     def __init__(self, dispatcher, host, port):
 291         self.shared = dispatcher.shared
 292         self.dispatcher = dispatcher.request_dispatcher
 293         threading.Thread.__init__(self)
 294         self.daemon = True
 295         self.host = host
 296         self.port = port
 297         self.lock = threading.Lock()
 298
 299     def run(self):
 300         # see http://code.google.com/p/jsonrpclib/
 301         from SocketServer import ThreadingMixIn
 302         class StratumThreadedJSONRPCServer(ThreadingMixIn, StratumJSONRPCServer): pass
 303
 304         self.server = StratumThreadedJSONRPCServer(( self.host, self.port))
 305         self.server.dispatcher = self.dispatcher
 306         self.server.register_function(None, 'server.stop')
 307
 308         print "HTTP server started."
 309         self.server.serve_forever()
 310