StratumJSONRPCServer.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
  29 try:
  30     import fcntl
  31 except ImportError:
  32     # For Windows
  33     fcntl = None
  34
  35 import json
  36
  37 def get_version(request):
  38     # must be a dict
  39     if 'jsonrpc' in request.keys():
  40         return 2.0
  41     if 'id' in request.keys():
  42         return 1.0
  43     return None
  44
  45 def validate_request(request):
  46     if type(request) is not types.DictType:
  47         fault = Fault(
  48             -32600, 'Request must be {}, not %s.' % type(request)
  49         )
  50         return fault
  51     rpcid = request.get('id', None)
  52     version = get_version(request)
  53     if not version:
  54         fault = Fault(-32600, 'Request %s invalid.' % request, rpcid=rpcid)
  55         return fault
  56     request.setdefault('params', [])
  57     method = request.get('method', None)
  58     params = request.get('params')
  59     param_types = (types.ListType, types.DictType, types.TupleType)
  60     if not method or type(method) not in types.StringTypes or \
  61         type(params) not in param_types:
  62         fault = Fault(
  63             -32600, 'Invalid request parameters or method.', rpcid=rpcid
  64         )
  65         return fault
  66     return True
  67
  68 class StratumJSONRPCDispatcher(SimpleXMLRPCServer.SimpleXMLRPCDispatcher):
  69
  70     def __init__(self, encoding=None):
  71         SimpleXMLRPCServer.SimpleXMLRPCDispatcher.__init__(self,
  72                                         allow_none=True,
  73                                         encoding=encoding)
  74
  75     def _marshaled_dispatch(self, data, dispatch_method = None):
  76         response = None
  77         try:
  78             request = jsonrpclib.loads(data)
  79         except Exception, e:
  80             fault = Fault(-32700, 'Request %s invalid. (%s)' % (data, e))
  81             response = fault.response()
  82             return response
  83
  84         responses = []
  85         if type(request) is not types.ListType:
  86             request = [ request ]
  87
  88         for req_entry in request:
  89             result = validate_request(req_entry)
  90             if type(result) is Fault:
  91                 responses.append(result.response())
  92                 continue
  93             resp_entry = self._marshaled_single_dispatch(req_entry)
  94             if resp_entry is not None:
  95                 responses.append(resp_entry)
  96
  97         # poll
  98         r = self._marshaled_single_dispatch({'method':'session.poll', 'params':[], 'id':'z' })
  99         r = jsonrpclib.loads(r)
 100         r = r.get('result')
 101         for item in r:
 102             responses.append(json.dumps(item))
 103
 104         if len(responses) > 1:
 105             response = '[%s]' % ','.join(responses)
 106         elif len(responses) == 1:
 107             response = responses[0]
 108         else:
 109             response = ''
 110
 111         return response
 112
 113     def _marshaled_single_dispatch(self, request):
 114         # TODO - Use the multiprocessing and skip the response if
 115         # it is a notification
 116         # Put in support for custom dispatcher here
 117         # (See SimpleXMLRPCServer._marshaled_dispatch)
 118         method = request.get('method')
 119         params = request.get('params')
 120         if params is None: params=[]
 121         params = [ self.session_id, request['id'] ] + params
 122         #print method, params
 123         try:
 124             response = self._dispatch(method, params)
 125         except:
 126             exc_type, exc_value, exc_tb = sys.exc_info()
 127             fault = Fault(-32603, '%s:%s' % (exc_type, exc_value))
 128             return fault.response()
 129         if 'id' not in request.keys() or request['id'] == None:
 130             # It's a notification
 131             return None
 132
 133         try:
 134             response = jsonrpclib.dumps(response,
 135                                         methodresponse=True,
 136                                         rpcid=request['id']
 137                                         )
 138             return response
 139         except:
 140             exc_type, exc_value, exc_tb = sys.exc_info()
 141             fault = Fault(-32603, '%s:%s' % (exc_type, exc_value))
 142             return fault.response()
 143
 144     def _dispatch(self, method, params):
 145         func = None
 146         try:
 147             func = self.funcs[method]
 148         except KeyError:
 149             if self.instance is not None:
 150                 if hasattr(self.instance, '_dispatch'):
 151                     return self.instance._dispatch(method, params)
 152                 else:
 153                     try:
 154                         func = SimpleXMLRPCServer.resolve_dotted_attribute(
 155                             self.instance,
 156                             method,
 157                             True
 158                             )
 159                     except AttributeError:
 160                         pass
 161         if func is not None:
 162             try:
 163                 if type(params) is types.ListType:
 164                     response = func(*params)
 165                 else:
 166                     response = func(**params)
 167                 return response
 168             except TypeError:
 169                 return Fault(-32602, 'Invalid parameters.')
 170             except:
 171                 err_lines = traceback.format_exc().splitlines()
 172                 trace_string = '%s | %s' % (err_lines[-3], err_lines[-1])
 173                 fault = jsonrpclib.Fault(-32603, 'Server error: %s' %
 174                                          trace_string)
 175                 return fault
 176         else:
 177             return Fault(-32601, 'Method %s not supported.' % method)
 178
 179 class StratumJSONRPCRequestHandler(
 180         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 181
 182     def do_GET(self):
 183         if not self.is_rpc_path_valid():
 184             self.report_404()
 185             return
 186         try:
 187             self.server.session_id = None
 188             c = self.headers.get('cookie')
 189             if c:
 190                 if c[0:8]=='SESSION=':
 191                     #print "found cookie", c[8:]
 192                     self.server.session_id = c[8:]
 193
 194             if self.server.session_id is None:
 195                 r = self.server._marshaled_single_dispatch({'method':'session.create', 'params':[], 'id':'z' })
 196                 r = jsonrpclib.loads(r)
 197                 self.server.session_id = r.get('result')
 198                 #print "setting cookie", self.server.session_id
 199
 200             data = json.dumps([])
 201             response = self.server._marshaled_dispatch(data)
 202             self.send_response(200)
 203         except Exception, e:
 204             self.send_response(500)
 205             err_lines = traceback.format_exc().splitlines()
 206             trace_string = '%s | %s' % (err_lines[-3], err_lines[-1])
 207             fault = jsonrpclib.Fault(-32603, 'Server error: %s' % trace_string)
 208             response = fault.response()
 209             print "500", trace_string
 210         if response == None:
 211             response = ''
 212
 213         if hasattr(self.server, 'session_id'):
 214             if self.server.session_id:
 215                 self.send_header("Set-Cookie", "SESSION=%s"%self.server.session_id)
 216                 self.session_id = None
 217
 218         self.send_header("Content-type", "application/json-rpc")
 219         self.send_header("Content-length", str(len(response)))
 220         self.end_headers()
 221         self.wfile.write(response)
 222         self.wfile.flush()
 223         self.connection.shutdown(1)
 224
 225
 226     def do_POST(self):
 227         if not self.is_rpc_path_valid():
 228             self.report_404()
 229             return
 230         try:
 231             max_chunk_size = 10*1024*1024
 232             size_remaining = int(self.headers["content-length"])
 233             L = []
 234             while size_remaining:
 235                 chunk_size = min(size_remaining, max_chunk_size)
 236                 L.append(self.rfile.read(chunk_size))
 237                 size_remaining -= len(L[-1])
 238             data = ''.join(L)
 239
 240             self.server.session_id = None
 241             c = self.headers.get('cookie')
 242             if c:
 243                 if c[0:8]=='SESSION=':
 244                     #print "found cookie", c[8:]
 245                     self.server.session_id = c[8:]
 246
 247             if self.server.session_id is None:
 248                 r = self.server._marshaled_single_dispatch({'method':'session.create', 'params':[], 'id':'z' })
 249                 r = jsonrpclib.loads(r)
 250                 self.server.session_id = r.get('result')
 251                 #print "setting cookie", self.server.session_id
 252
 253             response = self.server._marshaled_dispatch(data)
 254             self.send_response(200)
 255         except Exception, e:
 256             self.send_response(500)
 257             err_lines = traceback.format_exc().splitlines()
 258             trace_string = '%s | %s' % (err_lines[-3], err_lines[-1])
 259             fault = jsonrpclib.Fault(-32603, 'Server error: %s' % trace_string)
 260             response = fault.response()
 261             print "500", trace_string
 262         if response == None:
 263             response = ''
 264
 265         if hasattr(self.server, 'session_id'):
 266             if self.server.session_id:
 267                 self.send_header("Set-Cookie", "SESSION=%s"%self.server.session_id)
 268                 self.session_id = None
 269
 270         self.send_header("Content-type", "application/json-rpc")
 271         self.send_header("Content-length", str(len(response)))
 272         self.end_headers()
 273         self.wfile.write(response)
 274         self.wfile.flush()
 275         self.connection.shutdown(1)
 276
 277
 278 class StratumJSONRPCServer(SocketServer.TCPServer, StratumJSONRPCDispatcher):
 279
 280     allow_reuse_address = True
 281
 282     def __init__(self, addr, requestHandler=StratumJSONRPCRequestHandler,
 283                  logRequests=False, encoding=None, bind_and_activate=True,
 284                  address_family=socket.AF_INET):
 285         self.logRequests = logRequests
 286         StratumJSONRPCDispatcher.__init__(self, encoding)
 287         # TCPServer.__init__ has an extra parameter on 2.6+, so
 288         # check Python version and decide on how to call it
 289         vi = sys.version_info
 290         self.address_family = address_family
 291         if USE_UNIX_SOCKETS and address_family == socket.AF_UNIX:
 292             # Unix sockets can't be bound if they already exist in the
 293             # filesystem. The convention of e.g. X11 is to unlink
 294             # before binding again.
 295             if os.path.exists(addr):
 296                 try:
 297                     os.unlink(addr)
 298                 except OSError:
 299                     logging.warning("Could not unlink socket %s", addr)
 300         # if python 2.5 and lower
 301         if vi[0] < 3 and vi[1] < 6:
 302             SocketServer.TCPServer.__init__(self, addr, requestHandler)
 303         else:
 304             SocketServer.TCPServer.__init__(self, addr, requestHandler,
 305                 bind_and_activate)
 306         if fcntl is not None and hasattr(fcntl, 'FD_CLOEXEC'):
 307             flags = fcntl.fcntl(self.fileno(), fcntl.F_GETFD)
 308             flags |= fcntl.FD_CLOEXEC
 309             fcntl.fcntl(self.fileno(), fcntl.F_SETFD, flags)
 310
 311