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