X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=p2pool%2Fbitcoin%2Fp2p.py;h=83a3403f0f94695db3153d5e8419594e26f21aaf;hb=0a3493d6873cfef4fb189d39e64dfbc6e162e2a7;hp=46c104fd864e2f0d0d6b1b7475b3f66f6f8c6312;hpb=151b088435d17d4ffe24c5f37febfd37594faa28;p=p2pool.git diff --git a/p2pool/bitcoin/p2p.py b/p2pool/bitcoin/p2p.py index 46c104f..83a3403 100644 --- a/p2pool/bitcoin/p2p.py +++ b/p2pool/bitcoin/p2p.py @@ -2,95 +2,19 @@ Implementation of Bitcoin's p2p protocol ''' -from __future__ import division - -import hashlib import random -import struct +import sys import time -from twisted.internet import defer, protocol, reactor, task -from twisted.python import log +from twisted.internet import protocol import p2pool from . import data as bitcoin_data -from p2pool.util import datachunker, deferral, pack, variable - -class TooLong(Exception): - pass - -class BaseProtocol(protocol.Protocol): - def __init__(self, message_prefix, max_payload_length): - self._message_prefix = message_prefix - self._max_payload_length = max_payload_length - self.dataReceived = datachunker.DataChunker(self.dataReceiver()) - - def dataReceiver(self): - while True: - start = '' - while start != self._message_prefix: - start = (start + (yield 1))[-len(self._message_prefix):] - - command = (yield 12).rstrip('\0') - length, = struct.unpack(' self._max_payload_length: - print 'length too large' - continue - checksum = yield 4 - payload = yield length - - if hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] != checksum: - print 'invalid hash for', self.transport.getPeer().host, repr(command), length, checksum.encode('hex'), hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4].encode('hex'), payload.encode('hex') - continue - - type_ = getattr(self, 'message_' + command, None) - if type_ is None: - if p2pool.DEBUG: - print 'no type for', repr(command) - continue - - try: - self.packetReceived(command, type_.unpack(payload)) - except: - print 'RECV', command, payload[:100].encode('hex') + ('...' if len(payload) > 100 else '') - log.err(None, 'Error handling message: (see RECV line)') - self.badPeerHappened() - - def packetReceived(self, command, payload2): - handler = getattr(self, 'handle_' + command, None) - if handler is None: - if p2pool.DEBUG: - print 'no handler for', repr(command) - return - - handler(**payload2) - - def badPeerHappened(self): - self.transport.loseConnection() - - def sendPacket(self, command, payload2): - if len(command) >= 12: - raise ValueError('command too long') - type_ = getattr(self, 'message_' + command, None) - if type_ is None: - raise ValueError('invalid command') - #print 'SEND', command, repr(payload2)[:500] - payload = type_.pack(payload2) - if len(payload) > self._max_payload_length: - raise TooLong('payload too long') - self.transport.write(self._message_prefix + struct.pack('<12sI', command, len(payload)) + hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] + payload) - - def __getattr__(self, attr): - prefix = 'send_' - if attr.startswith(prefix): - command = attr[len(prefix):] - return lambda **payload2: self.sendPacket(command, payload2) - #return protocol.Protocol.__getattr__(self, attr) - raise AttributeError(attr) +from p2pool.util import deferral, p2protocol, pack, variable -class Protocol(BaseProtocol): +class Protocol(p2protocol.Protocol): def __init__(self, net): - BaseProtocol.__init__(self, net.P2P_PREFIX, 1000000) + p2protocol.Protocol.__init__(self, net.P2P_PREFIX, 1000000, ignore_trailing_payload=True) def connectionMade(self): self.send_version( @@ -108,7 +32,7 @@ class Protocol(BaseProtocol): port=self.transport.getHost().port, ), nonce=random.randrange(2**64), - sub_version_num='', + sub_version_num='/P2Pool:%s/' % (p2pool.__version__,), start_height=0, ) @@ -129,34 +53,33 @@ class Protocol(BaseProtocol): def handle_verack(self): self.get_block = deferral.ReplyMatcher(lambda hash: self.send_getdata(requests=[dict(type='block', hash=hash)])) self.get_block_header = deferral.ReplyMatcher(lambda hash: self.send_getheaders(version=1, have=[], last=hash)) - self.get_tx = deferral.ReplyMatcher(lambda hash: self.send_getdata(requests=[dict(type='tx', hash=hash)])) if hasattr(self.factory, 'resetDelay'): self.factory.resetDelay() if hasattr(self.factory, 'gotConnection'): self.factory.gotConnection(self) - self.pinger = task.LoopingCall(self.send_ping) + self.pinger = deferral.RobustLoopingCall(self.send_ping) self.pinger.start(30) message_inv = pack.ComposedType([ ('invs', pack.ListType(pack.ComposedType([ - ('type', pack.EnumType(pack.IntType(32), {'tx': 1, 'block': 2})), + ('type', pack.EnumType(pack.IntType(32), {1: 'tx', 2: 'block'})), ('hash', pack.IntType(256)), ]))), ]) def handle_inv(self, invs): for inv in invs: if inv['type'] == 'tx': - self.factory.new_tx.happened(inv['hash']) + self.send_getdata(requests=[inv]) elif inv['type'] == 'block': self.factory.new_block.happened(inv['hash']) else: - print 'Unknown inv type', item + print 'Unknown inv type', inv message_getdata = pack.ComposedType([ ('requests', pack.ListType(pack.ComposedType([ - ('type', pack.EnumType(pack.IntType(32), {'tx': 1, 'block': 2})), + ('type', pack.EnumType(pack.IntType(32), {1: 'tx', 2: 'block'})), ('hash', pack.IntType(256)), ]))), ]) @@ -186,7 +109,7 @@ class Protocol(BaseProtocol): ('tx', bitcoin_data.tx_type), ]) def handle_tx(self, tx): - self.get_tx.got_response(bitcoin_data.hash256(bitcoin_data.tx_type.pack(tx)), tx) + self.factory.new_tx.happened(tx) message_block = pack.ComposedType([ ('block', bitcoin_data.block_type), @@ -214,14 +137,15 @@ class Protocol(BaseProtocol): ('signature', pack.VarStrType()), ]) def handle_alert(self, message, signature): - print 'ALERT:', (message, signature) + pass # print 'ALERT:', (message, signature) def connectionLost(self, reason): if hasattr(self.factory, 'gotConnection'): self.factory.gotConnection(None) if hasattr(self, 'pinger'): self.pinger.stop() - print 'Bitcoin connection lost. Reason:', reason.getErrorMessage() + if p2pool.DEBUG: + print >>sys.stderr, 'Bitcoin connection lost. Reason:', reason.getErrorMessage() class ClientFactory(protocol.ReconnectingClientFactory): protocol = Protocol @@ -246,20 +170,3 @@ class ClientFactory(protocol.ReconnectingClientFactory): def getProtocol(self): return self.conn.get_not_none() - -if __name__ == '__main__': - from . import networks - factory = ClientFactory(networks.BitcoinMainnet) - reactor.connectTCP('127.0.0.1', 8333, factory) - - @repr - @apply - @defer.inlineCallbacks - def think(): - try: - print (yield (yield factory.getProtocol()).get_block(0x000000000000003aaaf7638f9f9c0d0c60e8b0eb817dcdb55fd2b1964efc5175)) - except defer.TimeoutError: - print "timeout" - reactor.stop() - - reactor.run()