Inside Bitcoin

251        # don't display warning if reconnection needs to be attempted using v1 P2P
252        if exc and not self.reconnect:
253            logger.warning("Connection lost to {}:{} due to {}".format(self.dstaddr, self.dstport, exc))
254        else:
255            logger.debug("Closed connection to: %s:%d" % (self.dstaddr, self.dstport))
256        self._transport = None
257        self.recvbuf = b""
258        self.on_close()
259
260    # v2 handshake method
261    def _on_data_v2_handshake(self):
262        """v2 handshake performed before P2P messages are exchanged (see BIP324). P2PConnection is the initiator
263        (in inbound connections to TestNode) and the responder (in outbound connections from TestNode).
264        Performed by:
265            * initiator using `initiate_v2_handshake()`, `complete_handshake()` and `authenticate_handshake()`
266            * responder using `respond_v2_handshake()`, `complete_handshake()` and `authenticate_handshake()`
267
268        `initiate_v2_handshake()` is immediately done by the initiator when the connection is established in
269        `connection_made()`. The rest of the initial v2 handshake functions are handled here.
270        """
271        if not self.v2_state.peer:
272            if not self.v2_state.initiating and not self.v2_state.sent_garbage:
273                # if the responder hasn't sent garbage yet, the responder is still reading ellswift bytes
274                # reads ellswift bytes till the first mismatch from 12 bytes V1_PREFIX
275                length, send_handshake_bytes = self.v2_state.respond_v2_handshake(BytesIO(self.recvbuf))
276                self.recvbuf = self.recvbuf[length:]
277                if send_handshake_bytes == -1:
278                    self.v2_state = None
279                    return
280                elif send_handshake_bytes:
281                    logger.debug(f"sending {len(self.v2_state.sent_garbage)} bytes of garbage data")
282                    self.send_raw_message(send_handshake_bytes)
283                elif send_handshake_bytes == b"":
284                    return  # only after send_handshake_bytes are sent can `complete_handshake()` be done
285
286            # `complete_handshake()` reads the remaining ellswift bytes from recvbuf
287            # and sends response after deriving shared ECDH secret using received ellswift bytes
288            length, response = self.v2_state.complete_handshake(BytesIO(self.recvbuf))
289            self.recvbuf = self.recvbuf[length:]
290            if response:
291                self.send_raw_message(response)
292            else:
293                return  # only after response is sent can `authenticate_handshake()` be done
294
295        # `self.v2_state.peer` is instantiated only after shared ECDH secret/BIP324 derived keys and ciphers
296        # is derived in `complete_handshake()`.
297        # so `authenticate_handshake()` which uses the BIP324 derived ciphers gets called after `complete_handshake()`.
298        assert self.v2_state.peer
299        length, is_mac_auth = self.v2_state.authenticate_handshake(self.recvbuf)
300        if not is_mac_auth:
301            raise ValueError("invalid v2 mac tag in handshake authentication")
302        self.recvbuf = self.recvbuf[length:]
303        if self.v2_state.tried_v2_handshake:
304            # for v2 outbound connections, send version message immediately after v2 handshake
305            if self.p2p_connected_to_node:
306                self.send_version()
307            # process post-v2-handshake data immediately, if available
308            if len(self.recvbuf) > 0:
309                self._on_data()
310
311    # Socket read methods
312
313    def data_received(self, t):
314        """asyncio callback when data is read from the socket."""
315        if len(t) > 0:
316            self.recvbuf += t
317            if self.supports_v2_p2p and not self.v2_state.tried_v2_handshake:
318                self._on_data_v2_handshake()
319            else:
320                self._on_data()
321
322    def _on_data(self):
323        """Try to read P2P messages from the recv buffer.
324
325        This method reads data from the buffer in a loop. It deserializes,
326        parses and verifies the P2P header, then passes the P2P payload to
327        the on_message callback for processing."""
328        try:
329            while True:
330                if self.supports_v2_p2p:
331                    # v2 P2P messages are read
332                    msglen, msg = self.v2_state.v2_receive_packet(self.recvbuf)
333                    if msglen == -1:
334                        raise ValueError("invalid v2 mac tag " + repr(self.recvbuf))
335                    elif msglen == 0:  # need to receive more bytes in recvbuf
336                        return
337                    self.recvbuf = self.recvbuf[msglen:]
338
339                    if msg is None:  # ignore decoy messages
340                        return
341                    assert msg  # application layer messages (which aren't decoy messages) are non-empty
342                    shortid = msg[0]  # 1-byte short message type ID
343                    if shortid == 0:
344                        # next 12 bytes are interpreted as ASCII message type if shortid is b'\x00'
345                        if len(msg) < 13:
346                            raise IndexError("msg needs minimum required length of 13 bytes")
347                        msgtype = msg[1:13].rstrip(b'\x00')
348                        msg = msg[13:]  # msg is set to be payload
349                    else:
350                        # a 1-byte short message type ID
351                        msgtype = SHORTID.get(shortid, f"unknown-{shortid}")
352                        msg = msg[1:]
353                else:
354                    # v1 P2P messages are read
355                    if len(self.recvbuf) < 4:
356                        return
357                    if self.recvbuf[:4] != self.magic_bytes:
358                        raise ValueError("magic bytes mismatch: {} != {}".format(repr(self.magic_bytes), repr(self.recvbuf)))
359                    if len(self.recvbuf) < 4 + 12 + 4 + 4:
360                        return
361                    msgtype = self.recvbuf[4:4+12].split(b"\x00", 1)[0]
362                    msglen = struct.unpack("<i", self.recvbuf[4+12:4+12+4])[0]
363                    checksum = self.recvbuf[4+12+4:4+12+4+4]
364                    if len(self.recvbuf) < 4 + 12 + 4 + 4 + msglen:
365                        return
366                    msg = self.recvbuf[4+12+4+4:4+12+4+4+msglen]
367                    th = sha256(msg)
368                    h = sha256(th)
369                    if checksum != h[:4]:
370                        raise ValueError("got bad checksum " + repr(self.recvbuf))
371                    self.recvbuf = self.recvbuf[4+12+4+4+msglen:]
372                if msgtype not in MESSAGEMAP:
373                    raise ValueError("Received unknown msgtype from %s:%d: '%s' %s" % (self.dstaddr, self.dstport, msgtype, repr(msg)))
374                f = BytesIO(msg)
375                t = MESSAGEMAP[msgtype]()
376                t.deserialize(f)
377                self._log_message("receive", t)
378                self.on_message(t)
379        except Exception as e:
380            if not self.reconnect:
381                logger.exception(f"Error reading message: {repr(e)}")
382            raise
383
384    def on_message(self, message):
385        """Callback for processing a P2P payload. Must be overridden by derived class."""
386        raise NotImplementedError
387
388    # Socket write methods
389
390    def send_without_ping(self, message, is_decoy=False):
391        """Send a P2P message over the socket.
392
393        This method takes a P2P payload, builds the P2P header and adds
394        the message to the send buffer to be sent over the socket.
395
396        When a message does not lead to a disconnect, send_and_ping is usually
397        preferred to send a message. This can help to reduce intermittent test
398        failures due to a missing sync. Also, it includes a call to
399        sync_with_ping, allowing for concise test code.
400        """
401        with self._send_lock:
402            tmsg = self.build_message(message, is_decoy)
403            self._log_message("send", message)
404            return self.send_raw_message(tmsg)
405
406    def send_raw_message(self, raw_message_bytes):
407        if not self.is_connected:
408            raise IOError('Not connected')
409
410        def maybe_write():
411            if not self._transport:
412                return
413            if self._transport.is_closing():
414                return
415            self._transport.write(raw_message_bytes)
416        NetworkThread.network_event_loop.call_soon_threadsafe(maybe_write)
417
418    # Class utility methods
419
420    def build_message(self, message, is_decoy=False):
421        """Build a serialized P2P message"""
422        msgtype = message.msgtype
423        data = message.serialize()
424        if self.supports_v2_p2p:
425            if msgtype in SHORTID.values():
426                tmsg = MSGTYPE_TO_SHORTID.get(msgtype).to_bytes(1, 'big')
427            else:
428                tmsg = b"\x00"
429                tmsg += msgtype
430                tmsg += b"\x00" * (12 - len(msgtype))
431            tmsg += data
432            return self.v2_state.v2_enc_packet(tmsg, ignore=is_decoy)
433        else:
434            tmsg = self.magic_bytes
435            tmsg += msgtype
436            tmsg += b"\x00" * (12 - len(msgtype))
437            tmsg += len(data).to_bytes(4, "little")
438            th = sha256(data)
439            h = sha256(th)
440            tmsg += h[:4]
441            tmsg += data
442            return tmsg
443
444    def _log_message(self, direction, msg):
445        """Logs a message being sent or received over the connection."""
446        if direction == "send":
447            log_message = "Send message to "
448        elif direction == "receive":
449            log_message = "Received message from "
450        log_message += "%s:%d: %s" % (self.dstaddr, self.dstport, repr(msg)[:500])
451        if len(log_message) > 500:
452            log_message += "... (msg truncated)"
453        logger.debug(log_message)
454
455
456class P2PInterface(P2PConnection):
457    """A high-level P2P interface class for communicating with a Bitcoin node.
458
459    This class provides high-level callbacks for processing P2P message
460    payloads, as well as convenience methods for interacting with the
461    node over P2P.
462
463    Individual testcases should subclass this and override the on_* methods
464    if they want to alter message handling behaviour."""
465    def __init__(self, support_addrv2=False, wtxidrelay=True):
466        super().__init__()
467
468        # Track number of messages of each type received.
469        # Should be read-only in a test.
470        self.message_count = defaultdict(int)
471
472        # Track the most recent message of each type.
473        # To wait for a message to be received, pop that message from
474        # this and use self.wait_until.
475        self.last_message = {}
476
477        # A count of the number of ping messages we've sent to the node
478        self.ping_counter = 1
479
480        # The network services received from the peer
481        self.nServices = 0
482
483        self.support_addrv2 = support_addrv2
484
485        # If the peer supports wtxid-relay
486        self.wtxidrelay = wtxidrelay
487
488    def peer_connect_send_version(self, services):
489        # Send a version msg
490        vt = msg_version()
491        vt.nVersion = P2P_VERSION
492        vt.strSubVer = P2P_SUBVERSION
493        vt.relay = P2P_VERSION_RELAY
494        vt.nServices = services
495        vt.addrTo.ip = self.dstaddr
496        vt.addrTo.port = self.dstport
497        vt.addrFrom.ip = "0.0.0.0"
498        vt.addrFrom.port = 0
499        self.on_connection_send_msg = vt  # Will be sent in connection_made callback
500