2 Generic message-based protocol used by Bitcoin and P2Pool for P2P communication
8 from twisted.internet import protocol
9 from twisted.python import log
12 from p2pool.util import datachunker, variable
14 class TooLong(Exception):
17 class Protocol(protocol.Protocol):
18 def __init__(self, message_prefix, max_payload_length, traffic_happened=variable.Event()):
19 self._message_prefix = message_prefix
20 self._max_payload_length = max_payload_length
21 self.dataReceived2 = datachunker.DataChunker(self.dataReceiver())
22 self.paused_var = variable.Variable(False)
23 self.traffic_happened = traffic_happened
25 def connectionMade(self):
26 self.transport.registerProducer(self, True)
28 def pauseProducing(self):
29 self.paused_var.set(True)
31 def resumeProducing(self):
32 self.paused_var.set(False)
34 def stopProducing(self):
37 def dataReceived(self, data):
38 self.traffic_happened.happened('p2p/in', len(data))
39 self.dataReceived2(data)
41 def dataReceiver(self):
44 while start != self._message_prefix:
45 start = (start + (yield 1))[-len(self._message_prefix):]
47 command = (yield 12).rstrip('\0')
48 length, = struct.unpack('<I', (yield 4))
49 if length > self._max_payload_length:
50 print 'length too large'
53 payload = yield length
55 if hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] != checksum:
56 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')
59 type_ = getattr(self, 'message_' + command, None)
62 print 'no type for', repr(command)
66 self.packetReceived(command, type_.unpack(payload))
68 print 'RECV', command, payload[:100].encode('hex') + ('...' if len(payload) > 100 else '')
69 log.err(None, 'Error handling message: (see RECV line)')
70 self.transport.loseConnection()
72 def packetReceived(self, command, payload2):
73 handler = getattr(self, 'handle_' + command, None)
76 print 'no handler for', repr(command)
81 def badPeerHappened(self):
82 self.transport.loseConnection()
84 def sendPacket(self, command, payload2):
85 if len(command) >= 12:
86 raise ValueError('command too long')
87 type_ = getattr(self, 'message_' + command, None)
89 raise ValueError('invalid command')
90 #print 'SEND', command, repr(payload2)[:500]
91 payload = type_.pack(payload2)
92 if len(payload) > self._max_payload_length:
93 raise TooLong('payload too long')
94 data = self._message_prefix + struct.pack('<12sI', command, len(payload)) + hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] + payload
95 self.traffic_happened.happened('p2p/out', len(data))
96 self.transport.write(data)
97 return self.paused_var.get_when_satisfies(lambda paused: not paused)
99 def __getattr__(self, attr):
101 if attr.startswith(prefix):
102 command = attr[len(prefix):]
103 return lambda **payload2: self.sendPacket(command, payload2)
104 #return protocol.Protocol.__getattr__(self, attr)
105 raise AttributeError(attr)