p2pool/bitcoin/p2p.py

   1 '''
   2 Implementation of Bitcoin's p2p protocol
   3 '''
   4
   5 from __future__ import division
   6
   7 import hashlib
   8 import random
   9 import struct
  10 import time
  11 import traceback
  12
  13 from twisted.internet import protocol, reactor
  14
  15 from . import data as bitcoin_data
  16 import p2pool.util
  17
  18 class BaseProtocol(protocol.Protocol):
  19     def connectionMade(self):
  20         self.dataReceived = p2pool.util.DataChunker(self.dataReceiver())
  21
  22     def dataReceiver(self):
  23         while True:
  24             start = ''
  25             while start != self._prefix:
  26                 start = (start + (yield 1))[-len(self._prefix):]
  27
  28             command = (yield 12).rstrip('\0')
  29             length, = struct.unpack('<I', (yield 4))
  30
  31             if self.use_checksum:
  32                 checksum = yield 4
  33             else:
  34                 checksum = None
  35
  36             payload = yield length
  37
  38             if checksum is not None:
  39                 if hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] != checksum:
  40                     print 'RECV', command, checksum.encode('hex') if checksum is not None else None, repr(payload.encode('hex')), len(payload)
  41                     print 'INVALID HASH'
  42                     continue
  43
  44             type_ = getattr(self, "message_" + command, None)
  45             if type_ is None:
  46                 print 'RECV', command, checksum.encode('hex') if checksum is not None else None, repr(payload.encode('hex')), len(payload)
  47                 print 'NO TYPE FOR', repr(command)
  48                 continue
  49
  50             try:
  51                 payload2 = type_.unpack(payload)
  52             except:
  53                 print 'RECV', command, checksum.encode('hex') if checksum is not None else None, repr(payload.encode('hex')), len(payload)
  54                 traceback.print_exc()
  55                 continue
  56
  57             handler = getattr(self, 'handle_' + command, None)
  58             if handler is None:
  59                 print 'RECV', command, checksum.encode('hex') if checksum is not None else None, repr(payload.encode('hex')), len(payload)
  60                 print 'NO HANDLER FOR', command
  61                 continue
  62
  63             #print 'RECV', command, payload2
  64
  65             try:
  66                 handler(**payload2)
  67             except:
  68                 print 'RECV', command, checksum.encode('hex') if checksum is not None else None, repr(payload.encode('hex')), len(payload)
  69                 traceback.print_exc()
  70                 continue
  71
  72     def sendPacket(self, command, payload2={}):
  73         type_ = getattr(self, "message_" + command, None)
  74         if type_ is None:
  75             raise ValueError('invalid command')
  76         payload = type_.pack(payload2)
  77         if len(command) >= 12:
  78             raise ValueError('command too long')
  79         if self.use_checksum:
  80             checksum = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4]
  81         else:
  82             checksum = ''
  83         data = self._prefix + struct.pack('<12sI', command, len(payload)) + checksum + payload
  84         self.transport.write(data)
  85         #print 'SEND', command, payload2
  86
  87     def __getattr__(self, attr):
  88         prefix = 'send_'
  89         if attr.startswith(prefix):
  90             command = attr[len(prefix):]
  91             return lambda **payload2: self.sendPacket(command, payload2)
  92         #return protocol.Protocol.__getattr__(self, attr)
  93         raise AttributeError(attr)
  94
  95 class Protocol(BaseProtocol):
  96     def __init__(self, testnet=False):
  97         if testnet:
  98             self._prefix = 'fabfb5da'.decode('hex')
  99         else:
 100             self._prefix = 'f9beb4d9'.decode('hex')
 101
 102     version = 0
 103
 104     @property
 105     def use_checksum(self):
 106         return self.version >= 209
 107
 108     message_version = bitcoin_data.ComposedType([
 109         ('version', bitcoin_data.StructType('<I')),
 110         ('services', bitcoin_data.StructType('<Q')),
 111         ('time', bitcoin_data.StructType('<Q')),
 112         ('addr_to', bitcoin_data.address_type),
 113         ('addr_from', bitcoin_data.address_type),
 114         ('nonce', bitcoin_data.StructType('<Q')),
 115         ('sub_version_num', bitcoin_data.VarStrType()),
 116         ('start_height', bitcoin_data.StructType('<I')),
 117     ])
 118     message_verack = bitcoin_data.ComposedType([])
 119     message_addr = bitcoin_data.ComposedType([
 120         ('addrs', bitcoin_data.ListType(bitcoin_data.ComposedType([
 121             ('timestamp', bitcoin_data.StructType('<I')),
 122             ('address', bitcoin_data.address_type),
 123         ]))),
 124     ])
 125     message_inv = bitcoin_data.ComposedType([
 126         ('invs', bitcoin_data.ListType(bitcoin_data.ComposedType([
 127             ('type', bitcoin_data.EnumType(bitcoin_data.StructType('<I'), {'tx': 1, 'block': 2})),
 128             ('hash', bitcoin_data.HashType()),
 129         ]))),
 130     ])
 131     message_getdata = bitcoin_data.ComposedType([
 132         ('requests', bitcoin_data.ListType(bitcoin_data.ComposedType([
 133             ('type', bitcoin_data.EnumType(bitcoin_data.StructType('<I'), {'tx': 1, 'block': 2})),
 134             ('hash', bitcoin_data.HashType()),
 135         ]))),
 136     ])
 137     message_getblocks = bitcoin_data.ComposedType([
 138         ('version', bitcoin_data.StructType('<I')),
 139         ('have', bitcoin_data.ListType(bitcoin_data.HashType())),
 140         ('last', bitcoin_data.HashType()),
 141     ])
 142     message_getheaders = bitcoin_data.ComposedType([
 143         ('version', bitcoin_data.StructType('<I')),
 144         ('have', bitcoin_data.ListType(bitcoin_data.HashType())),
 145         ('last', bitcoin_data.HashType()),
 146     ])
 147     message_tx = bitcoin_data.ComposedType([
 148         ('tx', bitcoin_data.tx_type),
 149     ])
 150     message_block = bitcoin_data.ComposedType([
 151         ('block', bitcoin_data.block_type),
 152     ])
 153     message_headers = bitcoin_data.ComposedType([
 154         ('headers', bitcoin_data.ListType(bitcoin_data.block_header_type)),
 155     ])
 156     message_getaddr = bitcoin_data.ComposedType([])
 157     message_checkorder = bitcoin_data.ComposedType([
 158         ('id', bitcoin_data.HashType()),
 159         ('order', bitcoin_data.FixedStrType(60)), # XXX
 160     ])
 161     message_submitorder = bitcoin_data.ComposedType([
 162         ('id', bitcoin_data.HashType()),
 163         ('order', bitcoin_data.FixedStrType(60)), # XXX
 164     ])
 165     message_reply = bitcoin_data.ComposedType([
 166         ('hash', bitcoin_data.HashType()),
 167         ('reply',  bitcoin_data.EnumType(bitcoin_data.StructType('<I'), {'success': 0, 'failure': 1, 'denied': 2})),
 168         ('script', bitcoin_data.VarStrType()),
 169     ])
 170     message_ping = bitcoin_data.ComposedType([])
 171     message_alert = bitcoin_data.ComposedType([
 172         ('message', bitcoin_data.VarStrType()),
 173         ('signature', bitcoin_data.VarStrType()),
 174     ])
 175
 176     null_order = '\0'*60
 177
 178     def connectionMade(self):
 179         BaseProtocol.connectionMade(self)
 180
 181         self.send_version(
 182             version=32200,
 183             services=1,
 184             time=int(time.time()),
 185             addr_to=dict(
 186                 services=1,
 187                 address='::ffff:' + self.transport.getPeer().host,
 188                 port=self.transport.getPeer().port,
 189             ),
 190             addr_from=dict(
 191                 services=1,
 192                 address='::ffff:' + self.transport.getHost().host,
 193                 port=self.transport.getHost().port,
 194             ),
 195             nonce=random.randrange(2**64),
 196             sub_version_num='',
 197             start_height=0,
 198         )
 199
 200     def handle_version(self, version, services, time, addr_to, addr_from, nonce, sub_version_num, start_height):
 201         #print 'VERSION', locals()
 202         self.version_after = version
 203         self.send_verack()
 204
 205     def handle_verack(self):
 206         self.version = self.version_after
 207
 208         # connection ready
 209         self.check_order = p2pool.util.GenericDeferrer(2**256, lambda id, order: self.send_checkorder(id=id, order=order))
 210         self.submit_order = p2pool.util.GenericDeferrer(2**256, lambda id, order: self.send_submitorder(id=id, order=order))
 211         self.get_block = p2pool.util.ReplyMatcher(lambda hash: self.send_getdata(requests=[dict(type='block', hash=hash)]))
 212         self.get_block_header = p2pool.util.ReplyMatcher(lambda hash: self.send_getdata(requests=[dict(type='block', hash=hash)]))
 213
 214         if hasattr(self.factory, 'resetDelay'):
 215             self.factory.resetDelay()
 216         if hasattr(self.factory, 'gotConnection'):
 217             self.factory.gotConnection(self)
 218
 219     def handle_inv(self, invs):
 220         for inv in invs:
 221             #print 'INV', item['type'], hex(item['hash'])
 222             self.send_getdata(requests=[inv])
 223
 224     def handle_addr(self, addrs):
 225         for addr in addrs:
 226             pass#print 'ADDR', addr
 227
 228     def handle_reply(self, hash, reply, script):
 229         self.check_order.got_response(hash, dict(reply=reply, script=script))
 230         self.submit_order.got_response(hash, dict(reply=reply, script=script))
 231
 232     def handle_tx(self, tx):
 233         #print 'TX', hex(merkle_hash([tx])), tx
 234         self.factory.new_tx.happened(tx)
 235
 236     def handle_block(self, block):
 237         self.get_block.got_response(bitcoin_data.block_hash(block['header']), block)
 238         self.factory.new_block.happened(block)
 239
 240     def handle_ping(self):
 241         pass
 242
 243     def connectionLost(self, reason):
 244         if hasattr(self.factory, 'gotConnection'):
 245             self.factory.gotConnection(None)
 246
 247 class ClientFactory(protocol.ReconnectingClientFactory):
 248     protocol = Protocol
 249
 250     maxDelay = 15
 251
 252     def __init__(self, testnet=False):
 253         self.testnet = testnet
 254         self.conn = p2pool.util.Variable(None)
 255
 256         self.new_block = p2pool.util.Event()
 257         self.new_tx = p2pool.util.Event()
 258
 259     def buildProtocol(self, addr):
 260         p = self.protocol(self.testnet)
 261         p.factory = self
 262         return p
 263
 264     def gotConnection(self, conn):
 265         self.conn.set(conn)
 266
 267     def getProtocol(self):
 268         return self.conn.get_not_none()
 269
 270 if __name__ == '__main__':
 271     factory = ClientFactory()
 272     reactor.connectTCP('127.0.0.1', 8333, factory)
 273
 274     reactor.run()