Bugfix: authorization: Handle utf8 decoding outside of generic modules, safe if passw...
[bitcoin:eloipool.git] / jsonrpcserver.py
1 # Eloipool - Python Bitcoin pool server
2 # Copyright (C) 2011-2013  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 from binascii import b2a_hex
18 import httpserver
19 import json
20 import logging
21 import networkserver
22 import socket
23 from time import time
24 import traceback
25
26 WithinLongpoll = httpserver.AsyncRequest
27
28 class _SentJSONError(BaseException):
29         def __init__(self, rv):
30                 self.rv = rv
31
32 class JSONRPCHandler(httpserver.HTTPHandler):
33         default_quirks = {
34                 'NELH': None,  # FIXME: identify which clients have a problem with this
35         }
36         
37         LPHeaders = {
38                 'X-Long-Polling': None,
39         }
40         
41         JSONRPCURIs = (b'/', b'/LP', b'/LP/')
42         
43         logger = logging.getLogger('JSONRPCHandler')
44         
45         def final_init(server):
46                 pass
47         
48         def __init__(self, *a, **ka):
49                 super().__init__(*a, **ka)
50                 self.UA = None
51         
52         def sendReply(self, status=200, body=b'', headers=None, *a, **ka):
53                 headers = dict(headers) if headers else {}
54                 if body and body[0] == 123:  # b'{'
55                         headers.setdefault('Content-Type', 'application/json')
56                 if status == 200 and self.path in self.JSONRPCURIs:
57                         if not body:
58                                 headers.setdefault('Content-Type', 'application/json')
59                         headers.setdefault('X-Long-Polling', '/LP')
60                 return super().sendReply(status, body, headers, *a, **ka)
61         
62         def fmtError(self, reason = '', code = 100):
63                 reason = json.dumps(reason)
64                 reason = r'{"result":null,"id":null,"error":{"name":"JSONRPCError","code":%d,"message":%s}}' % (code, reason)
65                 reason = reason.encode('utf8')
66                 return reason
67         
68         def doError(self, reason = '', code = 100):
69                 reason = self.fmtError(reason, code)
70                 return self.sendReply(500, reason)
71         
72         _MidstateNotAdv = (b'phoenix', b'poclbm', b'gminor')
73         def doHeader_user_agent(self, value):
74                 self.reqinfo['UA'] = value
75                 self.UA = value.decode('latin-1')  # technically ASCII, but latin-1 ignores errors
76                 quirks = self.quirks
77                 (UA, v, *x) = value.split(b'/', 1) + [None]
78                 
79                 # Temporary HACK to keep working with older gmp-proxy
80                 # NOTE: This will go away someday.
81                 if UA == b'AuthServiceProxy':
82                         # SubmitBlock Boolean
83                         quirks['SBB'] = None
84                 
85                 try:
86                         if v[0] == b'v': v = v[1:]
87                         v = tuple(map(int, v.split(b'.'))) + (0,0,0)
88                 except:
89                         pass
90                 if UA in self._MidstateNotAdv:
91                         if UA == b'phoenix':
92                                 if v != (1, 50, 0):
93                                         quirks['midstate'] = None
94                                 if v[0] < 2 and v[1] < 8 and v[2] < 1:
95                                         quirks['NELH'] = None
96                         else:
97                                 quirks['midstate'] = None
98         
99         def doHeader_x_minimum_wait(self, value):
100                 self.reqinfo['MinWait'] = int(value)
101         
102         def doHeader_x_mining_extensions(self, value):
103                 self.extensions = value.decode('ascii').lower().split(' ')
104         
105         def processLP(self, lpid):
106                 lpw = self.server.LPId
107                 if isinstance(lpid, str):
108                         if lpw != lpid:
109                                 return
110                 self.doLongpoll()
111         
112         def doLongpoll(self, *a):
113                 timeNow = time()
114                 
115                 self._LP = True
116                 self._LPCall = a
117                 if 'NELH' not in self.quirks:
118                         # [NOT No] Early Longpoll Headers
119                         self.sendReply(200, body=None, headers=self.LPHeaders)
120                         self.push(b"1\r\n{\r\n")
121                         self.changeTask(self._chunkedKA, timeNow + 45)
122                 else:
123                         self.changeTask(None)
124                 
125                 waitTime = self.reqinfo.get('MinWait', 15)  # TODO: make default configurable
126                 self.waitTime = waitTime + timeNow
127                 
128                 totfromme = self.LPTrack()
129                 self.server._LPClients[id(self)] = self
130                 self.logger.debug("New LP client; %d total; %d from %s" % (len(self.server._LPClients), totfromme, self.remoteHost))
131                 
132                 raise WithinLongpoll
133         
134         def _chunkedKA(self):
135                 # Keepalive via chunked transfer encoding
136                 self.push(b"1\r\n \r\n")
137                 self.changeTask(self._chunkedKA, time() + 45)
138         
139         def LPTrack(self):
140                 myip = self.remoteHost
141                 if myip not in self.server.LPTracking:
142                         self.server.LPTracking[myip] = 0
143                 self.server.LPTracking[myip] += 1
144                 
145                 myuser = self.Username
146                 if myuser not in self.server.LPTrackingByUser:
147                         self.server.LPTrackingByUser[myuser] = 0
148                 self.server.LPTrackingByUser[myuser] += 1
149                 
150                 return self.server.LPTracking[myip]
151         
152         def LPUntrack(self):
153                 self.server.LPTracking[self.remoteHost] -= 1
154                 self.server.LPTrackingByUser[self.Username] -= 1
155         
156         def cleanupLP(self):
157                 # Called when the connection is closed
158                 if not self._LP:
159                         return
160                 self.changeTask(None)
161                 try:
162                         del self.server._LPClients[id(self)]
163                 except KeyError:
164                         pass
165                 self.LPUntrack()
166         
167         def wakeLongpoll(self, wantClear = False):
168                 now = time()
169                 if now < self.waitTime:
170                         self.changeTask(lambda: self.wakeLongpoll(wantClear), self.waitTime)
171                         return
172                 else:
173                         self.changeTask(None)
174                 
175                 self.LPUntrack()
176                 
177                 self.server.tls.wantClear = wantClear
178                 try:
179                         rv = self._doJSON_i(*self._LPCall, longpoll=True)
180                 except WithinLongpoll:
181                         # Not sure why this would happen right now, but handle it sanely...
182                         return
183                 finally:
184                         self.server.tls.wantClear = False
185                 if 'NELH' not in self.quirks:
186                         rv = rv[1:]  # strip the '{' we already sent
187                         self.push(('%x' % len(rv)).encode('utf8') + b"\r\n" + rv + b"\r\n0\r\n\r\n")
188                         self.reset_request()
189                         return
190                 
191                 try:
192                         self.sendReply(200, body=rv, headers=self.LPHeaders, tryCompression=False)
193                         raise httpserver.RequestNotHandled
194                 except httpserver.RequestHandled:
195                         # Expected
196                         pass
197                 finally:
198                         self.reset_request()
199         
200         def _doJSON_i(self, reqid, method, params, longpoll = False):
201                 try:
202                         rv = getattr(self, method)(*params)
203                 except WithinLongpoll:
204                         self._LPCall = (reqid, method, params)
205                         raise
206                 except Exception as e:
207                         self.logger.error(("Error during JSON-RPC call (UA=%s, IP=%s): %s%s\n" % (self.reqinfo.get('UA'), self.remoteHost, method, params)) + traceback.format_exc())
208                         efun = self.fmtError if longpoll else self.doError
209                         return efun(r'Service error: %s' % (e,))
210                 rv = {'id': reqid, 'error': None, 'result': rv}
211                 try:
212                         rv = json.dumps(rv)
213                 except:
214                         efun = self.fmtError if longpoll else self.doError
215                         return efun(r'Error encoding reply in JSON')
216                 rv = rv.encode('utf8')
217                 return rv if longpoll else self.sendReply(200, rv, headers=self._JSONHeaders)
218         
219         def doJSON(self, data, longpoll = False):
220                 # TODO: handle JSON errors
221                 try:
222                         data = data.decode('utf8')
223                 except UnicodeDecodeError as e:
224                         return self.doError(str(e))
225                 if longpoll and not data:
226                         self.JSONRPCId = jsonid = 1
227                         self.JSONRPCMethod = 'getwork'
228                         self._JSONHeaders = {}
229                         return self.doLongpoll(1, 'doJSON_getwork', ())
230                 try:
231                         data = json.loads(data)
232                         method = str(data['method']).lower()
233                         self.JSONRPCId = jsonid = data['id']
234                         self.JSONRPCMethod = method
235                         method = 'doJSON_' + method
236                 except ValueError:
237                         return self.doError(r'Parse error')
238                 except TypeError:
239                         return self.doError(r'Bad call')
240                 if not hasattr(self, method):
241                         return self.doError(r'Procedure not found')
242                 # TODO: handle errors as JSON-RPC
243                 self._JSONHeaders = {}
244                 params = data.setdefault('params', ())
245                 procfun = self._doJSON_i
246                 if longpoll and not params:
247                         procfun = self.doLongpoll
248                 return procfun(jsonid, method, params)
249         
250         def handle_close(self):
251                 self.cleanupLP()
252                 super().handle_close()
253         
254         def handle_request(self):
255                 if not self.method in (b'GET', b'POST'):
256                         return self.sendReply(405)
257                 if not self.path in self.JSONRPCURIs:
258                         if isinstance(self.path, bytes) and self.path[:5] == b'/src/':
259                                 return self.handle_src_request()
260                         return self.sendReply(404)
261                 if not self.Username:
262                         return self.doAuthenticate()
263                 try:
264                         data = b''.join(self.incoming)
265                         return self.doJSON(data, self.path[:3] == b'/LP')
266                 except socket.error:
267                         raise
268                 except WithinLongpoll:
269                         raise
270                 except httpserver.RequestHandled:
271                         raise
272                 except:
273                         self.logger.error(traceback.format_exc())
274                         return self.doError('uncaught error')
275         
276         def reset_request(self):
277                 self._LP = False
278                 self.JSONRPCMethod = None
279                 super().reset_request()
280         
281 setattr(JSONRPCHandler, 'doHeader_user-agent', JSONRPCHandler.doHeader_user_agent);
282 setattr(JSONRPCHandler, 'doHeader_x-minimum-wait', JSONRPCHandler.doHeader_x_minimum_wait);
283 setattr(JSONRPCHandler, 'doHeader_x-mining-extensions', JSONRPCHandler.doHeader_x_mining_extensions);
284
285 JSONRPCListener = networkserver.NetworkListener
286
287 class JSONRPCServer(networkserver.AsyncSocketServer):
288         logger = logging.getLogger('JSONRPCServer')
289         
290         waker = True
291         
292         def __init__(self, *a, **ka):
293                 ka.setdefault('RequestHandlerClass', JSONRPCHandler)
294                 super().__init__(*a, **ka)
295                 
296                 self.SecretUser = None
297                 self.ShareTarget = 0x00000000ffffffffffffffffffffffffffffffffffffffffffffffffffffffff
298                 
299                 self._LPId = 0
300                 self.LPId = '%d' % (time(),)
301                 self.LPRequest = False
302                 self._LPClients = {}
303                 self._LPWaitTime = time() + 15
304                 
305                 self.LPTracking = {}
306                 self.LPTrackingByUser = {}
307         
308         def checkAuthentication(self, username, password):
309                 return True
310         
311         def final_init(self):
312                 JSONRPCHandler.final_init(self)
313         
314         def pre_schedule(self):
315                 if self.LPRequest == 1:
316                         self._LPsch()
317         
318         def wakeLongpoll(self, wantClear = False):
319                 if self.LPRequest:
320                         self.logger.info('Ignoring longpoll attempt while another is waiting')
321                         return
322                 self._LPId += 1
323                 self.LPId = '%d %d' % (time(), self._LPId)
324                 self._LPWantClear = wantClear
325                 self.LPRequest = 1
326                 self.wakeup()
327         
328         def _LPsch(self):
329                 now = time()
330                 if self._LPWaitTime > now:
331                         delay = self._LPWaitTime - now
332                         self.logger.info('Waiting %.3g seconds to longpoll' % (delay,))
333                         self.schedule(self._actualLP, self._LPWaitTime)
334                         self.LPRequest = 2
335                 else:
336                         self._actualLP()
337         
338         def _actualLP(self):
339                 self.LPRequest = False
340                 C = tuple(self._LPClients.values())
341                 self._LPClients = {}
342                 if not C:
343                         self.logger.info('Nobody to longpoll')
344                         return
345                 OC = len(C)
346                 self.logger.debug("%d clients to wake up..." % (OC,))
347                 
348                 now = time()
349                 
350                 for ic in C:
351                         self.lastHandler = ic
352                         try:
353                                 ic.wakeLongpoll(self._LPWantClear)
354                         except socket.error:
355                                 OC -= 1
356                                 # Ignore socket errors; let the main event loop take care of them later
357                         except:
358                                 OC -= 1
359                                 self.logger.debug('Error waking longpoll handler:\n' + traceback.format_exc())
360                 
361                 self._LPWaitTime = time()
362                 self.logger.info('Longpoll woke up %d clients in %.3f seconds' % (OC, self._LPWaitTime - now))
363                 self._LPWaitTime += 5  # TODO: make configurable: minimum time between longpolls
364         
365         def TopLPers(self, n = 0x10):
366                 tmp = list(self.LPTracking.keys())
367                 tmp.sort(key=lambda k: self.LPTracking[k])
368                 for jerk in map(lambda k: (k, self.LPTracking[k]), tmp[-n:]):
369                         print(jerk)
370         
371         def TopLPersByUser(self, n = 0x10):
372                 tmp = list(self.LPTrackingByUser.keys())
373                 tmp.sort(key=lambda k: self.LPTrackingByUser[k])
374                 for jerk in map(lambda k: (k, self.LPTrackingByUser[k]), tmp[-n:]):
375                         print(jerk)