p2pool/bitcoin/p2p.py

   1 '''
   2 Implementation of Bitcoin's p2p protocol
   3 '''
   4
   5 import random
   6 import sys
   7 import time
   8
   9 from twisted.internet import protocol, task
  10
  11 import p2pool
  12 from . import data as bitcoin_data
  13 from p2pool.util import deferral, p2protocol, pack, variable
  14
  15 class Protocol(p2protocol.Protocol):
  16     def __init__(self, net):
  17         p2protocol.Protocol.__init__(self, net.P2P_PREFIX, 1000000)
  18
  19     def connectionMade(self):
  20         self.send_version(
  21             version=32200,
  22             services=1,
  23             time=int(time.time()),
  24             addr_to=dict(
  25                 services=1,
  26                 address=self.transport.getPeer().host,
  27                 port=self.transport.getPeer().port,
  28             ),
  29             addr_from=dict(
  30                 services=1,
  31                 address=self.transport.getHost().host,
  32                 port=self.transport.getHost().port,
  33             ),
  34             nonce=random.randrange(2**64),
  35             sub_version_num='',
  36             start_height=0,
  37         )
  38
  39     message_version = pack.ComposedType([
  40         ('version', pack.IntType(32)),
  41         ('services', pack.IntType(64)),
  42         ('time', pack.IntType(64)),
  43         ('addr_to', bitcoin_data.address_type),
  44         ('addr_from', bitcoin_data.address_type),
  45         ('nonce', pack.IntType(64)),
  46         ('sub_version_num', pack.VarStrType()),
  47         ('start_height', pack.IntType(32)),
  48     ])
  49     def handle_version(self, version, services, time, addr_to, addr_from, nonce, sub_version_num, start_height):
  50         self.send_verack()
  51
  52     message_verack = pack.ComposedType([])
  53     def handle_verack(self):
  54         self.get_block = deferral.ReplyMatcher(lambda hash: self.send_getdata(requests=[dict(type='block', hash=hash)]))
  55         self.get_block_header = deferral.ReplyMatcher(lambda hash: self.send_getheaders(version=1, have=[], last=hash))
  56         self.get_tx = deferral.ReplyMatcher(lambda hash: self.send_getdata(requests=[dict(type='tx', hash=hash)]))
  57
  58         if hasattr(self.factory, 'resetDelay'):
  59             self.factory.resetDelay()
  60         if hasattr(self.factory, 'gotConnection'):
  61             self.factory.gotConnection(self)
  62
  63         self.pinger = task.LoopingCall(self.send_ping)
  64         self.pinger.start(30)
  65
  66     message_inv = pack.ComposedType([
  67         ('invs', pack.ListType(pack.ComposedType([
  68             ('type', pack.EnumType(pack.IntType(32), {1: 'tx', 2: 'block'})),
  69             ('hash', pack.IntType(256)),
  70         ]))),
  71     ])
  72     def handle_inv(self, invs):
  73         for inv in invs:
  74             if inv['type'] == 'tx':
  75                 self.factory.new_tx.happened(inv['hash'])
  76             elif inv['type'] == 'block':
  77                 self.factory.new_block.happened(inv['hash'])
  78             else:
  79                 print 'Unknown inv type', item
  80
  81     message_getdata = pack.ComposedType([
  82         ('requests', pack.ListType(pack.ComposedType([
  83             ('type', pack.EnumType(pack.IntType(32), {1: 'tx', 2: 'block'})),
  84             ('hash', pack.IntType(256)),
  85         ]))),
  86     ])
  87     message_getblocks = pack.ComposedType([
  88         ('version', pack.IntType(32)),
  89         ('have', pack.ListType(pack.IntType(256))),
  90         ('last', pack.PossiblyNoneType(0, pack.IntType(256))),
  91     ])
  92     message_getheaders = pack.ComposedType([
  93         ('version', pack.IntType(32)),
  94         ('have', pack.ListType(pack.IntType(256))),
  95         ('last', pack.PossiblyNoneType(0, pack.IntType(256))),
  96     ])
  97     message_getaddr = pack.ComposedType([])
  98
  99     message_addr = pack.ComposedType([
 100         ('addrs', pack.ListType(pack.ComposedType([
 101             ('timestamp', pack.IntType(32)),
 102             ('address', bitcoin_data.address_type),
 103         ]))),
 104     ])
 105     def handle_addr(self, addrs):
 106         for addr in addrs:
 107             pass
 108
 109     message_tx = pack.ComposedType([
 110         ('tx', bitcoin_data.tx_type),
 111     ])
 112     def handle_tx(self, tx):
 113         self.get_tx.got_response(bitcoin_data.hash256(bitcoin_data.tx_type.pack(tx)), tx)
 114
 115     message_block = pack.ComposedType([
 116         ('block', bitcoin_data.block_type),
 117     ])
 118     def handle_block(self, block):
 119         block_hash = bitcoin_data.hash256(bitcoin_data.block_header_type.pack(block['header']))
 120         self.get_block.got_response(block_hash, block)
 121         self.get_block_header.got_response(block_hash, block['header'])
 122
 123     message_headers = pack.ComposedType([
 124         ('headers', pack.ListType(bitcoin_data.block_type)),
 125     ])
 126     def handle_headers(self, headers):
 127         for header in headers:
 128             header = header['header']
 129             self.get_block_header.got_response(bitcoin_data.hash256(bitcoin_data.block_header_type.pack(header)), header)
 130         self.factory.new_headers.happened([header['header'] for header in headers])
 131
 132     message_ping = pack.ComposedType([])
 133     def handle_ping(self):
 134         pass
 135
 136     message_alert = pack.ComposedType([
 137         ('message', pack.VarStrType()),
 138         ('signature', pack.VarStrType()),
 139     ])
 140     def handle_alert(self, message, signature):
 141         pass # print 'ALERT:', (message, signature)
 142
 143     def connectionLost(self, reason):
 144         if hasattr(self.factory, 'gotConnection'):
 145             self.factory.gotConnection(None)
 146         if hasattr(self, 'pinger'):
 147             self.pinger.stop()
 148         if p2pool.DEBUG:
 149             print >>sys.stderr, 'Bitcoin connection lost. Reason:', reason.getErrorMessage()
 150
 151 class ClientFactory(protocol.ReconnectingClientFactory):
 152     protocol = Protocol
 153
 154     maxDelay = 1
 155
 156     def __init__(self, net):
 157         self.net = net
 158         self.conn = variable.Variable(None)
 159
 160         self.new_block = variable.Event()
 161         self.new_tx = variable.Event()
 162         self.new_headers = variable.Event()
 163
 164     def buildProtocol(self, addr):
 165         p = self.protocol(self.net)
 166         p.factory = self
 167         return p
 168
 169     def gotConnection(self, conn):
 170         self.conn.set(conn)
 171
 172     def getProtocol(self):
 173         return self.conn.get_not_none()