Revert "broadcast shares in serial", strongly suspected of causing a memory leak
[p2pool.git] / p2pool / util / p2protocol.py
1 '''
2 Generic message-based protocol used by Bitcoin and P2Pool for P2P communication
3 '''
4
5 import hashlib
6 import struct
7
8 from twisted.internet import protocol
9 from twisted.python import log
10
11 import p2pool
12 from p2pool.util import datachunker, variable
13
14 class TooLong(Exception):
15     pass
16
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.traffic_happened = traffic_happened
23     
24     def dataReceived(self, data):
25         self.traffic_happened.happened('p2p/in', len(data))
26         self.dataReceived2(data)
27     
28     def dataReceiver(self):
29         while True:
30             start = ''
31             while start != self._message_prefix:
32                 start = (start + (yield 1))[-len(self._message_prefix):]
33             
34             command = (yield 12).rstrip('\0')
35             length, = struct.unpack('<I', (yield 4))
36             if length > self._max_payload_length:
37                 print 'length too large'
38                 continue
39             checksum = yield 4
40             payload = yield length
41             
42             if hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] != checksum:
43                 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')
44                 self.badPeerHappened()
45                 continue
46             
47             type_ = getattr(self, 'message_' + command, None)
48             if type_ is None:
49                 if p2pool.DEBUG:
50                     print 'no type for', repr(command)
51                 continue
52             
53             try:
54                 self.packetReceived(command, type_.unpack(payload))
55             except:
56                 print 'RECV', command, payload[:100].encode('hex') + ('...' if len(payload) > 100 else '')
57                 log.err(None, 'Error handling message: (see RECV line)')
58                 self.transport.loseConnection()
59     
60     def packetReceived(self, command, payload2):
61         handler = getattr(self, 'handle_' + command, None)
62         if handler is None:
63             if p2pool.DEBUG:
64                 print 'no handler for', repr(command)
65             return
66         
67         if getattr(self, 'connected', True) and not getattr(self, 'disconnecting', False):
68             handler(**payload2)
69     
70     def badPeerHappened(self):
71         self.transport.loseConnection()
72     
73     def sendPacket(self, command, payload2):
74         if len(command) >= 12:
75             raise ValueError('command too long')
76         type_ = getattr(self, 'message_' + command, None)
77         if type_ is None:
78             raise ValueError('invalid command')
79         #print 'SEND', command, repr(payload2)[:500]
80         payload = type_.pack(payload2)
81         if len(payload) > self._max_payload_length:
82             raise TooLong('payload too long')
83         data = self._message_prefix + struct.pack('<12sI', command, len(payload)) + hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] + payload
84         self.traffic_happened.happened('p2p/out', len(data))
85         self.transport.write(data)
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)