jsonrpcserver.py

   1 # Eloipool - Python Bitcoin pool server
   2 # Copyright (C) 2011-2012  Luke Dashjr <luke-jr+eloipool@utopios.org>
   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,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 import httpserver
  18 import json
  19 import logging
  20 import networkserver
  21 import socket
  22 from time import time
  23 import traceback
  24
  25 WithinLongpoll = httpserver.AsyncRequest
  26
  27 class _SentJSONError(BaseException):
  28         def __init__(self, rv):
  29                 self.rv = rv
  30
  31 class JSONRPCHandler(httpserver.HTTPHandler):
  32         default_quirks = {
  33                 'NELH': None,  # FIXME: identify which clients have a problem with this
  34         }
  35
  36         LPHeaders = {
  37                 'X-Long-Polling': None,
  38         }
  39
  40         JSONRPCURIs = (b'/', b'/LP', b'/LP/')
  41
  42         logger = logging.getLogger('JSONRPCHandler')
  43
  44         def sendReply(self, status=200, body=b'', headers=None):
  45                 headers = dict(headers) if headers else {}
  46                 if body and body[0] == 123:  # b'{'
  47                         headers.setdefault('Content-Type', 'application/json')
  48                 if status == 200 and self.path in self.JSONRPCURIs:
  49                         headers.setdefault('X-Long-Polling', '/LP')
  50                         if self.JSONRPCMethod == 'getwork':
  51                                 # FIXME: Move this to jsonrpc_getwork somehow
  52                                 headers.setdefault('X-Roll-NTime', 'expire=120')
  53                 return super().sendReply(status, body, headers)
  54
  55         def fmtError(self, reason = '', code = 100):
  56                 reason = json.dumps(reason)
  57                 reason = r'{"result":null,"id":null,"error":{"name":"JSONRPCError","code":%d,"message":%s}}' % (code, reason)
  58                 reason = reason.encode('utf8')
  59                 return reason
  60
  61         def doError(self, reason = '', code = 100):
  62                 reason = self.fmtError(reason, code)
  63                 return self.sendReply(500, reason)
  64
  65         def checkAuthentication(self, un, pw):
  66                 return bool(un)
  67
  68         _MidstateNotAdv = (b'phoenix', b'poclbm', b'gminor')
  69         def doHeader_user_agent(self, value):
  70                 self.reqinfo['UA'] = value
  71                 quirks = self.quirks
  72                 (UA, v, *x) = value.split(b'/', 1) + [None]
  73                 try:
  74                         if v[0] == b'v': v = v[1:]
  75                         v = tuple(map(int, v.split(b'.'))) + (0,0,0)
  76                 except:
  77                         pass
  78                 if UA in self._MidstateNotAdv:
  79                         if UA == b'phoenix':
  80                                 if v != (1, 50, 0):
  81                                         quirks['midstate'] = None
  82                                 if v[0] < 2 and v[1] < 8 and v[2] < 1:
  83                                         quirks['NELH'] = None
  84                         else:
  85                                 quirks['midstate'] = None
  86
  87         def doHeader_x_minimum_wait(self, value):
  88                 self.reqinfo['MinWait'] = int(value)
  89
  90         def doHeader_x_mining_extensions(self, value):
  91                 self.extensions = value.decode('ascii').lower().split(' ')
  92
  93         def doLongpoll(self, *a):
  94                 timeNow = time()
  95
  96                 self._LP = True
  97                 self._LPCall = a
  98                 if 'NELH' not in self.quirks:
  99                         # [NOT No] Early Longpoll Headers
 100                         self.sendReply(200, body=None, headers=self.LPHeaders)
 101                         self.push(b"1\r\n{\r\n")
 102                         self.changeTask(self._chunkedKA, timeNow + 45)
 103                 else:
 104                         self.changeTask(None)
 105
 106                 waitTime = self.reqinfo.get('MinWait', 15)  # TODO: make default configurable
 107                 self.waitTime = waitTime + timeNow
 108
 109                 totfromme = self.LPTrack()
 110                 self.server._LPClients[id(self)] = self
 111                 self.logger.debug("New LP client; %d total; %d from %s" % (len(self.server._LPClients), totfromme, self.remoteHost))
 112
 113                 raise WithinLongpoll
 114
 115         def _chunkedKA(self):
 116                 # Keepalive via chunked transfer encoding
 117                 self.push(b"1\r\n \r\n")
 118                 self.changeTask(self._chunkedKA, time() + 45)
 119
 120         def LPTrack(self):
 121                 myip = self.remoteHost
 122                 if myip not in self.server.LPTracking:
 123                         self.server.LPTracking[myip] = 0
 124                 self.server.LPTracking[myip] += 1
 125                 return self.server.LPTracking[myip]
 126
 127         def LPUntrack(self):
 128                 self.server.LPTracking[self.remoteHost] -= 1
 129
 130         def cleanupLP(self):
 131                 # Called when the connection is closed
 132                 if not self._LP:
 133                         return
 134                 self.changeTask(None)
 135                 try:
 136                         del self.server._LPClients[id(self)]
 137                 except KeyError:
 138                         pass
 139                 self.LPUntrack()
 140
 141         def wakeLongpoll(self):
 142                 now = time()
 143                 if now < self.waitTime:
 144                         self.changeTask(self.wakeLongpoll, self.waitTime)
 145                         return
 146                 else:
 147                         self.changeTask(None)
 148
 149                 self.LPUntrack()
 150
 151                 rv = self._doJSON_i(*self._LPCall, longpoll=True)
 152                 if 'NELH' not in self.quirks:
 153                         rv = rv[1:]  # strip the '{' we already sent
 154                         self.push(('%x' % len(rv)).encode('utf8') + b"\r\n" + rv + b"\r\n0\r\n\r\n")
 155                 else:
 156                         self.sendReply(200, body=rv, headers=self.LPHeaders)
 157
 158                 self.reset_request()
 159
 160         def _doJSON_i(self, reqid, method, params, longpoll = False):
 161                 try:
 162                         rv = getattr(self, method)(*params)
 163                 except Exception as e:
 164                         self.logger.error(("Error during JSON-RPC call: %s%s\n" % (method, params)) + traceback.format_exc())
 165                         efun = self.fmtError if longpoll else self.doError
 166                         return efun(r'Service error: %s' % (e,))
 167                 if rv is None:
 168                         # response was already sent (eg, authentication request)
 169                         return
 170                 try:
 171                         rv.setdefault('submitold', True)
 172                 except:
 173                         pass
 174                 rv = {'id': reqid, 'error': None, 'result': rv}
 175                 try:
 176                         rv = json.dumps(rv)
 177                 except:
 178                         efun = self.fmtError if longpoll else self.doError
 179                         return efun(r'Error encoding reply in JSON')
 180                 rv = rv.encode('utf8')
 181                 return rv if longpoll else self.sendReply(200, rv, headers=self._JSONHeaders)
 182
 183         def doJSON(self, data, longpoll = False):
 184                 # TODO: handle JSON errors
 185                 data = data.decode('utf8')
 186                 if longpoll and not data:
 187                         return self.doLongpoll(1, 'doJSON_getwork', ())
 188                 try:
 189                         data = json.loads(data)
 190                         method = str(data['method']).lower()
 191                         self.JSONRPCMethod = method
 192                         method = 'doJSON_' + method
 193                 except ValueError:
 194                         return self.doError(r'Parse error')
 195                 except TypeError:
 196                         return self.doError(r'Bad call')
 197                 if not hasattr(self, method):
 198                         return self.doError(r'Procedure not found')
 199                 # TODO: handle errors as JSON-RPC
 200                 self._JSONHeaders = {}
 201                 params = data.setdefault('params', ())
 202                 procfun = self._doJSON_i
 203                 if longpoll and not params:
 204                         procfun = self.doLongpoll
 205                 return procfun(data['id'], method, params)
 206
 207         def handle_close(self):
 208                 self.cleanupLP()
 209                 super().handle_close()
 210
 211         def handle_request(self):
 212                 if not self.method in (b'GET', b'POST'):
 213                         return self.sendReply(405)
 214                 if not self.path in self.JSONRPCURIs:
 215                         if self.path[:5] == b'/src/':
 216                                 return self.handle_src_request()
 217                         return self.sendReply(404)
 218                 if not self.Username:
 219                         return self.doAuthenticate()
 220                 try:
 221                         data = b''.join(self.incoming)
 222                         return self.doJSON(data, self.path[:3] == b'/LP')
 223                 except socket.error:
 224                         raise
 225                 except WithinLongpoll:
 226                         raise
 227                 except:
 228                         self.logger.error(traceback.format_exc())
 229                         return self.doError('uncaught error')
 230
 231         def reset_request(self):
 232                 self._LP = False
 233                 self.JSONRPCMethod = None
 234                 super().reset_request()
 235
 236 setattr(JSONRPCHandler, 'doHeader_user-agent', JSONRPCHandler.doHeader_user_agent);
 237 setattr(JSONRPCHandler, 'doHeader_x-minimum-wait', JSONRPCHandler.doHeader_x_minimum_wait);
 238 setattr(JSONRPCHandler, 'doHeader_x-mining-extensions', JSONRPCHandler.doHeader_x_mining_extensions);
 239
 240 JSONRPCListener = networkserver.NetworkListener
 241
 242 class JSONRPCServer(networkserver.AsyncSocketServer):
 243         logger = logging.getLogger('JSONRPCServer')
 244
 245         waker = True
 246
 247         def __init__(self, *a, **ka):
 248                 ka.setdefault('RequestHandlerClass', JSONRPCHandler)
 249                 super().__init__(*a, **ka)
 250
 251                 self.SecretUser = None
 252
 253                 self.LPRequest = False
 254                 self._LPClients = {}
 255                 self._LPWaitTime = time() + 15
 256
 257                 self.LPTracking = {}
 258
 259         def pre_schedule(self):
 260                 if self.LPRequest == 1:
 261                         self._LPsch()
 262
 263         def wakeLongpoll(self):
 264                 if self.LPRequest:
 265                         self.logger.info('Ignoring longpoll attempt while another is waiting')
 266                         return
 267                 self.LPRequest = 1
 268                 self.wakeup()
 269
 270         def _LPsch(self):
 271                 now = time()
 272                 if self._LPWaitTime > now:
 273                         delay = self._LPWaitTime - now
 274                         self.logger.info('Waiting %.3g seconds to longpoll' % (delay,))
 275                         self.schedule(self._actualLP, self._LPWaitTime)
 276                         self.LPRequest = 2
 277                 else:
 278                         self._actualLP()
 279
 280         def _actualLP(self):
 281                 self.LPRequest = False
 282                 C = tuple(self._LPClients.values())
 283                 self._LPClients = {}
 284                 if not C:
 285                         self.logger.info('Nobody to longpoll')
 286                         return
 287                 OC = len(C)
 288                 self.logger.debug("%d clients to wake up..." % (OC,))
 289
 290                 now = time()
 291
 292                 for ic in C:
 293                         try:
 294                                 ic.wakeLongpoll()
 295                         except socket.error:
 296                                 OC -= 1
 297                                 # Ignore socket errors; let the main event loop take care of them later
 298                         except:
 299                                 OC -= 1
 300                                 self.logger.debug('Error waking longpoll handler:\n' + traceback.format_exc())
 301
 302                 self._LPWaitTime = time()
 303                 self.logger.info('Longpoll woke up %d clients in %.3f seconds' % (OC, self._LPWaitTime - now))
 304                 self._LPWaitTime += 5  # TODO: make configurable: minimum time between longpolls
 305
 306         def TopLPers(self, n = 0x10):
 307                 tmp = list(self.LPTracking.keys())
 308                 tmp.sort(key=lambda k: self.LPTracking[k])
 309                 for jerk in map(lambda k: (k, self.LPTracking[k]), tmp[-n:]):
 310                         print(jerk)