stratumserver.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 from binascii import b2a_hex
  18 from copy import deepcopy
  19 import json
  20 import logging
  21 import networkserver
  22 #import re
  23 import struct
  24 from time import time
  25 #import traceback
  26 from util import RejectedShare, swap32
  27
  28 class StratumError(BaseException):
  29         def __init__(self, errno, msg):
  30                 self.StratumErrNo = errno
  31                 self.StratumErrMsg = msg
  32
  33 StratumCodes = {
  34         'stale-prevblk': 21,
  35         'stale-work': 21,
  36         'duplicate': 22,
  37         'H-not-zero': 23,
  38         'high-hash': 23,
  39 }
  40
  41 class StratumHandler(networkserver.SocketHandler):
  42         logger = logging.getLogger('StratumHandler')
  43
  44         def __init__(self, *a, **ka):
  45                 super().__init__(*a, **ka)
  46                 self.remoteHost = self.addr[0]
  47                 self.changeTask(None)
  48                 self.set_terminator(b"\n")
  49                 self.Usernames = {}
  50
  51         def sendReply(self, ob):
  52                 return self.push(json.dumps(ob).encode('ascii') + b"\n")
  53
  54         def found_terminator(self):
  55                 inbuf = b"".join(self.incoming).decode('ascii')
  56                 self.incoming = []
  57
  58                 if not inbuf:
  59                         return
  60
  61                 rpc = json.loads(inbuf)
  62                 funcname = '_stratum_%s' % (rpc['method'].replace('.', '_'),)
  63                 if not hasattr(self, funcname):
  64                         self.sendReply({
  65                                 'error': [-3, "Method '%s' not found" % (rpc['method'],), None],
  66                                 'id': rpc['id'],
  67                                 'result': None,
  68                         })
  69                         return
  70
  71                 try:
  72                         rv = getattr(self, funcname)(*rpc['params'])
  73                 except StratumError as e:
  74                         self.sendReply({
  75                                 'error': (e.StratumErrNo, e.StratumErrMsg, None),
  76                                 'id': rpc['id'],
  77                                 'result': None,
  78                         })
  79                         return
  80
  81                 if rpc['id'] is None:
  82                         return
  83
  84                 self.sendReply({
  85                         'error': None,
  86                         'id': rpc['id'],
  87                         'result': rv,
  88                 })
  89
  90         def sendJob(self):
  91                 self.push(self.server.JobBytes)
  92
  93         def _stratum_mining_subscribe(self):
  94                 xid = struct.pack('@P', id(self))
  95                 self.extranonce1 = xid
  96                 xid = b2a_hex(xid).decode('ascii')
  97                 self.server._Clients[id(self)] = self
  98                 self.changeTask(self.sendJob, 0)
  99                 return [
 100                         [
 101                                 ['mining.notify', '%s1' % (xid,)],
 102                                 ['mining.set_difficulty', '%s2' % (xid,)],
 103                         ],
 104                         xid,
 105                         4,
 106                 ]
 107
 108         def handle_close(self):
 109                 try:
 110                         del self.server._Clients[id(self)]
 111                 except:
 112                         pass
 113                 super().handle_close()
 114
 115         def _stratum_mining_submit(self, username, jobid, extranonce2, ntime, nonce):
 116                 if username not in self.Usernames:
 117                         pass # TODO
 118                 share = {
 119                         'username': username,
 120                         'remoteHost': self.remoteHost,
 121                         'jobid': jobid,
 122                         'extranonce1': self.extranonce1,
 123                         'extranonce2': bytes.fromhex(extranonce2),
 124                         'ntime': bytes.fromhex(ntime),
 125                         'nonce': bytes.fromhex(nonce),
 126                 }
 127                 try:
 128                         self.server.receiveShare(share)
 129                 except RejectedShare as rej:
 130                         rej = str(rej)
 131                         errno = StratumCodes.get(rej, 20)
 132                         raise StratumError(errno, rej)
 133                 return True
 134
 135         def checkAuthentication(self, username, password):
 136                 return True
 137
 138         def _stratum_mining_authorize(self, username, password = None):
 139                 try:
 140                         valid = self.checkAuthentication(username, password)
 141                 except:
 142                         valid = False
 143                 if valid:
 144                         self.Usernames[username] = None
 145                 return valid
 146
 147 class StratumServer(networkserver.AsyncSocketServer):
 148         logger = logging.getLogger('StratumServer')
 149
 150         waker = True
 151
 152         extranonce1null = struct.pack('@P', 0)
 153
 154         def __init__(self, *a, **ka):
 155                 ka.setdefault('RequestHandlerClass', StratumHandler)
 156                 super().__init__(*a, **ka)
 157
 158                 self._Clients = {}
 159                 self._JobId = 0
 160                 self.JobId = '%d' % (time(),)
 161                 self.WakeRequest = None
 162
 163         def updateJob(self, wantClear = False):
 164                 self._JobId += 1
 165                 JobId = '%d %d' % (time(), self._JobId)
 166                 (MC, wld) = self.getStratumJob(JobId, wantClear=wantClear)
 167                 (height, merkleTree, cb, prevBlock, bits) = MC[:5]
 168
 169                 if len(cb) > 96 - len(self.extranonce1null) - 4:
 170                         self.logger.warning('Coinbase too big for Stratum: GIVING CLIENTS INVALID JOBS')
 171                         # TODO: shutdown stratum
 172                         # TODO: restart automatically when coinbase works?
 173
 174                 txn = deepcopy(merkleTree.data[0])
 175                 cb += self.extranonce1null + b'Eloi'
 176                 txn.setCoinbase(cb)
 177                 txn.assemble()
 178                 pos = txn.data.index(cb) + len(cb)
 179
 180                 steps = list(b2a_hex(h).decode('ascii') for h in merkleTree._steps)
 181
 182                 self.JobBytes = json.dumps({
 183                         'id': None,
 184                         'method': 'mining.notify',
 185                         'params': [
 186                                 JobId,
 187                                 b2a_hex(swap32(prevBlock)).decode('ascii'),
 188                                 b2a_hex(txn.data[:pos - len(self.extranonce1null) - 4]).decode('ascii'),
 189                                 b2a_hex(txn.data[pos:]).decode('ascii'),
 190                                 steps,
 191                                 '00000002',
 192                                 b2a_hex(bits[::-1]).decode('ascii'),
 193                                 b2a_hex(struct.pack('>L', int(time()))).decode('ascii'),
 194                                 False
 195                         ],
 196                 }).encode('ascii') + b"\n"
 197                 self.JobId = JobId
 198
 199                 self.WakeRequest = 1
 200                 self.wakeup()
 201
 202         def pre_schedule(self):
 203                 if self.WakeRequest:
 204                         self._wakeNodes()
 205
 206         def _wakeNodes(self):
 207                 self.WakeRequest = None
 208                 C = self._Clients
 209                 if not C:
 210                         self.logger.info('Nobody to wake up')
 211                         return
 212                 OC = len(C)
 213                 self.logger.debug("%d clients to wake up..." % (OC,))
 214
 215                 now = time()
 216
 217                 for ic in list(C.values()):
 218                         try:
 219                                 ic.sendJob()
 220                         except socket.error:
 221                                 OC -= 1
 222                                 # Ignore socket errors; let the main event loop take care of them later
 223                         except:
 224                                 OC -= 1
 225                                 self.logger.debug('Error sending new job:\n' + traceback.format_exc())
 226
 227                 self.logger.info('New job sent to %d clients in %.3f seconds' % (OC, time() - now))