From 2c66c21ba1850506ecb430de117188f56cce9305 Mon Sep 17 00:00:00 2001 From: Dan Hepworth Date: Tue, 6 Oct 2026 11:47:20 -0400 Subject: [PATCH 1/2] test: add an opt-in hybrid v2 Python peer Give the functional test framework an independent implementation of the hybrid post-quantum v2 handshake (doc/design/pq-transport.md), built on crypto/mlkem.py and the key recipe in crypto/bip324_cipher.py, and use it for end-to-end tests against the node. - v2_p2p.py: EncryptedP2PState(supports_v2_pq=True) negotiates in both roles. A responder offers an ML-KEM-1024 key from two independent 32-byte draws; an initiator holds its version packet until the responder's arrives, accepts a valid offer and switches; each side's first hybrid packet is a key confirmation, and the peer's must verify. parse_version_contents() mirrors the node's record grammar. Subclasses can override the offer and accept contents, the shared secret and the confirmation, and a peer can withhold its confirmation. - Framework peers stay legacy by default (supports_v2_pq=False). The unconditional assertion that the node's version contents are empty is gone: a legacy peer accepts a non-empty offer and answers with empty contents. Received and sent contents are recorded, so tests of switched-off nodes assert empty contents explicitly. - p2p.py: authenticate_handshake() also returns the bytes it produced (the held version packet, a confirmation), which the driver sends. The handshake completes only after the peer's confirmation verified and ours was sent; send_pq_confirmation() releases a withheld one. - test_node.py: add_p2p_connection() and add_outbound_p2p_connection() take supports_v2_pq, and the outbound helper gains wait_for_v2_handshake. - v2_p2p.py unit tests (feature_framework_unit_tests.py) replay all four vectors of src/test/data/pq_transport_vectors.json through two states, run every role pairing across a rekey, and check withheld and damaged confirmations. p2p_v2_pq.py covers two nodes in both directions (hybrid, the same session id, more than 224 packets each way), the Python peer in both roles (the node draws fresh keys and encapsulation randomness per connection), legacy and v1 peers under each switch setting and each pair of node settings, and the pending status while a peer withholds its confirmation. Work item G1 of #212 (#184 part 7). Co-Authored-By: Claude Opus 5.5 --- .../feature_framework_unit_tests.py | 1 + test/functional/p2p_v2_pq.py | 261 ++++++++++ test/functional/test_framework/p2p.py | 33 +- test/functional/test_framework/test_node.py | 16 +- test/functional/test_framework/v2_p2p.py | 464 +++++++++++++++++- test/functional/test_runner.py | 1 + 6 files changed, 745 insertions(+), 31 deletions(-) create mode 100755 test/functional/p2p_v2_pq.py diff --git a/test/functional/feature_framework_unit_tests.py b/test/functional/feature_framework_unit_tests.py index 1343b93220..3f547b3d7d 100755 --- a/test/functional/feature_framework_unit_tests.py +++ b/test/functional/feature_framework_unit_tests.py @@ -32,6 +32,7 @@ "script", "script_util", "segwit_addr", + "v2_p2p", "wallet_util", ] diff --git a/test/functional/p2p_v2_pq.py b/test/functional/p2p_v2_pq.py new file mode 100755 index 0000000000..cab43c0010 --- /dev/null +++ b/test/functional/p2p_v2_pq.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026-present The qbit core developers +# Distributed under the MIT software license, see the accompanying +# file COPYING or https://opensource.org/license/mit. +"""Test the hybrid post-quantum v2 transport end to end (doc/design/pq-transport.md). + +- test_node_roles: two upgraded nodes switch to hybrid keys in both connection directions, agree on + the session id, and keep exchanging packets across the 224-packet rekey in each direction. +- test_python_roles: the test framework's independent Python implementation, as initiator and as + responder, derives the same keys as the node, and the node draws fresh ML-KEM randomness for + every connection. +- test_legacy_and_switches: legacy v2 peers, v1 peers and every switch combination get the + transport_pq_status each role and setting calls for. +- test_pending_confirmation: a connection whose peer withholds its key confirmation reports pending + with no session id, and hybrid only once the confirmation arrives. +""" + +from test_framework.crypto.chacha20 import REKEY_INTERVAL +from test_framework.crypto.mlkem import EK_SIZE, CT_SIZE, mlkem1024_check_ek +from test_framework.p2p import P2PInterface +from test_framework.test_framework import BitcoinTestFramework +from test_framework.util import ( + assert_equal, + assert_not_equal, +) +from test_framework.v2_p2p import ( + PQ_RECORD_LEN, + ParseKind, + parse_version_contents, +) + +PQ_ON = ["-v2transport=1", "-v2pqtransport=1"] +PQ_OFF = ["-v2transport=1", "-v2pqtransport=0"] +# A v2 ping: an encrypted length (3), the header (1), the short message id (1), the nonce (8) and the tag (16). +V2_PING_BYTES = 29 + + +def node_view(node, p2p): + """The node's getpeerinfo entry for a Python peer, found by the peer's own socket address.""" + sockname = p2p._transport.get_extra_info("socket").getsockname() + infos = [info for info in node.getpeerinfo() if info["addr"] == f"{sockname[0]}:{sockname[1]}"] + assert_equal(len(infos), 1) + return infos[0] + + +class WithholdingPeer(P2PInterface): + """A hybrid Python peer that holds back its key confirmation until send_pq_confirmation().""" + def make_v2_state(self, **kwargs): + state = super().make_v2_state(**kwargs) + state.withhold_confirmation = True + return state + + +class FixedKeyPeer(P2PInterface): + """A hybrid Python responder whose ML-KEM key pair is the same on every connection.""" + def make_v2_state(self, **kwargs): + state = super().make_v2_state(**kwargs) + draws = [bytes([1]) * 32, bytes([2]) * 32] + state.pq_random_bytes = lambda n: draws.pop(0) + return state + + +class P2PV2PQTest(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 2 + self.extra_args = [PQ_ON, PQ_ON] + + def setup_network(self): + self.setup_nodes() + + def run_test(self): + self.test_node_roles() + self.test_python_roles() + self.test_legacy_and_switches() + self.test_pending_confirmation() + + def node_pair(self, a, b): + """Connect node a to node b and return both sides' getpeerinfo entries.""" + self.connect_nodes(a, b) + [outbound] = self.nodes[a].getpeerinfo() + [inbound] = self.nodes[b].getpeerinfo() + assert_equal((outbound["inbound"], inbound["inbound"]), (False, True)) + return outbound, inbound + + def test_node_roles(self): + self.log.info("Two upgraded nodes switch to hybrid keys with either one initiating") + for a, b in ((0, 1), (1, 0)): + with self.nodes[a].assert_debug_log(["v2 pq: switched role=initiator"]), \ + self.nodes[b].assert_debug_log(["v2 pq: switched role=responder"]): + outbound, inbound = self.node_pair(a, b) + for info in (outbound, inbound): + assert_equal(info["transport_protocol_type"], "v2") + assert_equal(info["transport_pq"], True) + assert_equal(info["transport_pq_status"], "hybrid") + assert_equal(len(info["session_id"]), 64) + assert_equal(outbound["session_id"], inbound["session_id"]) + + self.log.info(f"More than {REKEY_INTERVAL} packets flow each way under the hybrid keys (node{a} initiating)") + count = REKEY_INTERVAL + 10 + for sender, peer in ((self.nodes[a], outbound), (self.nodes[b], inbound)): + for _ in range(count): + sender.sendmsgtopeer(peer_id=peer["id"], msg_type="ping", msg="00" * 8) + # Each node received at least `count` pings and `count` pongs: every one of them a packet + # under the hybrid keys, past the first rekey in both directions. + for node in (self.nodes[a], self.nodes[b]): + self.wait_until(lambda node=node: all( + node.getpeerinfo()[0]["bytesrecv_per_msg"].get(msg, 0) >= count * V2_PING_BYTES + for msg in ("ping", "pong"))) + [info] = node.getpeerinfo() + assert_equal(info["transport_pq_status"], "hybrid") + assert_equal(self.nodes[a].getpeerinfo()[0]["session_id"], outbound["session_id"]) + self.disconnect_nodes(a, b) + + def check_hybrid(self, node, peer): + """Both sides report the hybrid keys and the same session id.""" + info = node_view(node, peer) + assert_equal(info["transport_protocol_type"], "v2") + assert_equal(info["transport_pq"], True) + assert_equal(info["transport_pq_status"], "hybrid") + assert_equal(peer.v2_state.pq_status, "hybrid") + assert_equal(info["session_id"], peer.v2_state.peer["session_id"].hex()) + assert_equal(peer.v2_state.hybrid_keys["session_id"], peer.v2_state.peer["session_id"]) + peer.sync_with_ping() + + def test_python_roles(self): + node = self.nodes[0] + self.log.info("A Python hybrid initiator and the node agree on the hybrid keys") + offers = [] + for _ in range(2): + peer = node.add_p2p_connection(P2PInterface(), supports_v2_pq=True) + self.check_hybrid(node, peer) + offer = peer.v2_state.received_version_contents + assert_equal(len(offer), PQ_RECORD_LEN) + kind, ek = parse_version_contents(offer) + assert_equal((kind, len(ek)), (ParseKind.OWN_RECORD, EK_SIZE)) + assert mlkem1024_check_ek(ek) + assert_equal(len(peer.v2_state.sent_version_contents), PQ_RECORD_LEN) + offers.append(offer) + self.log.info("The node offers a fresh encapsulation key on every connection") + assert_not_equal(offers[0], offers[1]) + node.disconnect_p2ps() + + self.log.info("A Python hybrid responder and the node agree on the hybrid keys") + accepts = [] + session_ids = [] + for i in range(2): + # The same encapsulation key both times: the node must still encapsulate to fresh randomness. + peer = node.add_outbound_p2p_connection(FixedKeyPeer(), p2p_idx=i, supports_v2_pq=True) + self.check_hybrid(node, peer) + accept = peer.v2_state.received_version_contents + kind, ct = parse_version_contents(accept) + assert_equal((len(accept), kind, len(ct)), (PQ_RECORD_LEN, ParseKind.OWN_RECORD, CT_SIZE)) + accepts.append(accept) + session_ids.append(peer.v2_state.peer["session_id"]) + assert_equal(len(set(accepts)), 2) + assert_equal(len(set(session_ids)), 2) + node.disconnect_p2ps() + self.wait_until(lambda: len(node.getpeerinfo()) == 0) + + def check_status(self, node, peer, status, received_contents_len): + """The node's status for a legacy or switched-off pairing, and what the Python peer received.""" + info = node_view(node, peer) + assert_equal(info["transport_pq"], False) + assert_equal(info["transport_pq_status"], status) + if status == "v1": + assert_equal(info["transport_protocol_type"], "v1") + return + assert_equal(info["transport_protocol_type"], "v2") + # The ECDH session id, as the Python peer derived it. + assert_equal(info["session_id"], peer.v2_state.peer["session_id"].hex()) + assert_equal(peer.v2_state.hybrid_keys, None) + assert_equal(len(peer.v2_state.received_version_contents), received_contents_len) + peer.sync_with_ping() + + def test_legacy_and_switches(self): + node = self.nodes[0] + for setting, args in (("on", PQ_ON), ("off", PQ_OFF)): + self.restart_node(0, extra_args=args) + pq_on = setting == "on" + self.log.info(f"Switch {setting}: legacy and hybrid Python v2 peers, both roles") + # A legacy Python peer accepts the node's offer (non-empty contents) and answers with empty ones. + peer = node.add_p2p_connection(P2PInterface()) + self.check_status(node, peer, "legacy_peer" if pq_on else "off", PQ_RECORD_LEN if pq_on else 0) + assert_equal(peer.v2_state.sent_version_contents, b"") + peer = node.add_outbound_p2p_connection(P2PInterface(), p2p_idx=0) + self.check_status(node, peer, "legacy_peer" if pq_on else "off", 0) + if not pq_on: + # A switched-off node never offers, and never accepts the offer of a hybrid peer. + peer = node.add_p2p_connection(P2PInterface(), supports_v2_pq=True) + self.check_status(node, peer, "off", 0) + assert_equal(peer.v2_state.pq_status, "legacy_peer") + peer = node.add_outbound_p2p_connection(P2PInterface(), p2p_idx=1, supports_v2_pq=True) + self.check_status(node, peer, "off", 0) + assert_equal(len(peer.v2_state.sent_version_contents), PQ_RECORD_LEN) + assert_equal(peer.v2_state.pq_status, "legacy_peer") + + self.log.info(f"Switch {setting}: v1 Python peers, both roles") + peer = node.add_p2p_connection(P2PInterface(), supports_v2_p2p=False) + self.check_status(node, peer, "v1", None) + peer = node.add_outbound_p2p_connection(P2PInterface(), p2p_idx=2, advertise_v2_p2p=False) + self.check_status(node, peer, "v1", None) + node.disconnect_p2ps() + self.wait_until(lambda: len(node.getpeerinfo()) == 0) + + self.log.info("Two nodes with each combination of switches") + expected = { + # (initiator on, responder on): (initiator status, responder status) + (True, True): ("hybrid", "hybrid"), + (True, False): ("legacy_peer", "off"), + (False, True): ("off", "legacy_peer"), + (False, False): ("off", "off"), + } + for (initiator_on, responder_on), statuses in expected.items(): + self.restart_node(0, extra_args=PQ_ON if initiator_on else PQ_OFF) + self.restart_node(1, extra_args=PQ_ON if responder_on else PQ_OFF) + outbound, inbound = self.node_pair(0, 1) + assert_equal((outbound["transport_pq_status"], inbound["transport_pq_status"]), statuses) + assert_equal(outbound["transport_pq"], statuses[0] == "hybrid") + assert_equal(inbound["transport_pq"], statuses[1] == "hybrid") + assert_equal(outbound["session_id"], inbound["session_id"]) + self.disconnect_nodes(0, 1) + + self.log.info("A v1 connection between two switched-on nodes reports v1") + self.restart_node(1, extra_args=PQ_ON) + self.restart_node(0, extra_args=PQ_ON) + self.connect_nodes(0, 1, peer_advertises_v2=False) + for node in self.nodes: + [info] = node.getpeerinfo() + assert_equal((info["transport_protocol_type"], info["transport_pq"], info["transport_pq_status"]), + ("v1", False, "v1")) + self.disconnect_nodes(0, 1) + + def test_pending_confirmation(self): + node = self.nodes[0] + for role in ("initiator", "responder"): + self.log.info(f"A Python {role} that withholds its key confirmation leaves the node pending") + peer = WithholdingPeer() + if role == "initiator": + node.add_p2p_connection(peer, supports_v2_pq=True, wait_for_v2_handshake=False, + wait_for_verack=False, expect_success=False) + else: + node.add_outbound_p2p_connection(peer, p2p_idx=0, supports_v2_pq=True, wait_for_v2_handshake=False) + # The node switched and its confirmation verified at the Python peer. + peer.wait_for_connect() + peer.wait_until(lambda: peer.v2_state.peer_confirmed) + info = node_view(node, peer) + assert_equal((info["transport_protocol_type"], info["session_id"]), ("detecting", "")) + assert_equal((info["transport_pq"], info["transport_pq_status"]), (False, "pending")) + assert not peer.v2_state.tried_v2_handshake + + self.log.info(f"The connection reports hybrid once the {role}'s confirmation arrives") + peer.send_pq_confirmation() + peer.wait_until(lambda: peer.v2_state.tried_v2_handshake) + peer.wait_for_verack() + self.check_hybrid(node, peer) + node.disconnect_p2ps() + self.wait_until(lambda: len(node.getpeerinfo()) == 0) + + +if __name__ == '__main__': + P2PV2PQTest(__file__).main() diff --git a/test/functional/test_framework/p2p.py b/test/functional/test_framework/p2p.py index c2e773656c..3f6e0aa7bb 100755 --- a/test/functional/test_framework/p2p.py +++ b/test/functional/test_framework/p2p.py @@ -191,25 +191,29 @@ def peer_connect_helper(self, dstaddr, dstport, net, timeout_factor): self.magic_bytes = MAGIC_BYTES[net] self.p2p_connected_to_node = dstport != 0 - def peer_connect(self, dstaddr, dstport, *, net, timeout_factor, supports_v2_p2p): + def peer_connect(self, dstaddr, dstport, *, net, timeout_factor, supports_v2_p2p, supports_v2_pq=False): self.peer_connect_helper(dstaddr, dstport, net, timeout_factor) if supports_v2_p2p: - self.v2_state = EncryptedP2PState(initiating=True, net=net) + self.v2_state = self.make_v2_state(initiating=True, net=net, supports_v2_pq=supports_v2_pq) loop = NetworkThread.network_event_loop logger.debug('Connecting to Bitcoin Node: %s:%d' % (self.dstaddr, self.dstport)) coroutine = loop.create_connection(lambda: self, host=self.dstaddr, port=self.dstport) return lambda: loop.call_soon_threadsafe(loop.create_task, coroutine) - def peer_accept_connection(self, connect_id, connect_cb=lambda: None, *, net, timeout_factor, supports_v2_p2p, reconnect): + def peer_accept_connection(self, connect_id, connect_cb=lambda: None, *, net, timeout_factor, supports_v2_p2p, reconnect, supports_v2_pq=False): self.peer_connect_helper('0', 0, net, timeout_factor) self.reconnect = reconnect if supports_v2_p2p: - self.v2_state = EncryptedP2PState(initiating=False, net=net) + self.v2_state = self.make_v2_state(initiating=False, net=net, supports_v2_pq=supports_v2_pq) logger.debug('Listening for Bitcoin Node with id: {}'.format(connect_id)) return lambda: NetworkThread.listen(self, connect_cb, idx=connect_id) + def make_v2_state(self, *, initiating, net, supports_v2_pq): + """The v2 handshake state of a new connection. Tests of misbehaving peers return a subclass.""" + return EncryptedP2PState(initiating=initiating, net=net, supports_v2_pq=supports_v2_pq) + def peer_disconnect(self): # Connection could have already been closed by other end. NetworkThread.network_event_loop.call_soon_threadsafe(lambda: self._transport and self._transport.abort()) @@ -288,10 +292,18 @@ def _on_data_v2_handshake(self): # is derived in `complete_handshake()`. # so `authenticate_handshake()` which uses the BIP324 derived ciphers gets called after `complete_handshake()`. assert self.v2_state.peer - length, is_mac_auth = self.v2_state.authenticate_handshake(self.recvbuf) + length, is_mac_auth, pending = self.v2_state.authenticate_handshake(self.recvbuf) + # Bytes produced while authenticating (a hybrid initiator's version packet, a key confirmation) go out + # first, even when authentication then failed: the node sees what a real peer would have sent. + if pending: + self.send_raw_message(pending) if not is_mac_auth: raise ValueError("invalid v2 mac tag in handshake authentication") self.recvbuf = self.recvbuf[length:] + self._maybe_finish_v2_handshake() + + def _maybe_finish_v2_handshake(self): + """Start the application phase once the v2 handshake is complete.""" if self.v2_state.tried_v2_handshake: # for v2 outbound connections, send version message immediately after v2 handshake if self.p2p_connected_to_node: @@ -300,6 +312,17 @@ def _on_data_v2_handshake(self): if len(self.recvbuf) > 0: self._on_data() + def send_pq_confirmation(self): + """Send the key confirmation that a hybrid peer with v2_state.withhold_confirmation set held back after + its switch, and complete the handshake if the node's confirmation already verified.""" + def release(): + if not self.is_connected: + return + self.send_raw_message(self.v2_state.release_confirmation()) + self.v2_state.check_handshake_complete() + self._maybe_finish_v2_handshake() + NetworkThread.network_event_loop.call_soon_threadsafe(release) + # Socket read methods def data_received(self, t): diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py index 6edc0a99d7..1a9d232021 100755 --- a/test/functional/test_framework/test_node.py +++ b/test/functional/test_framework/test_node.py @@ -956,7 +956,7 @@ def assert_start_raises_init_error(self, extra_args=None, expected_msg=None, mat assert_msg += "with expected error " + expected_msg self._raise_assertion_error(assert_msg) - def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=True, supports_v2_p2p=None, wait_for_v2_handshake=True, expect_success=True, **kwargs): + def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=True, supports_v2_p2p=None, supports_v2_pq=False, wait_for_v2_handshake=True, expect_success=True, **kwargs): """Add an inbound p2p connection to the node. This method adds the p2p connection to the self.p2ps list and also @@ -968,6 +968,8 @@ def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=Tru - if TestNode doesn't advertise NODE_P2P_V2 service, P2PConnection sends version message and v1 P2P is followed - if TestNode advertises NODE_P2P_V2 service, (and if P2PConnections supports v2 P2P) P2PConnection sends ellswift bytes and v2 P2P is followed + - supports_v2_pq: a v2 P2PConnection also negotiates hybrid post-quantum keys (opt-in; framework + peers are legacy BIP324 peers by default) """ if 'dstport' not in kwargs: kwargs['dstport'] = p2p_port(self.index) @@ -979,7 +981,8 @@ def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=Tru if self.use_v2transport: kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2 supports_v2_p2p = self.use_v2transport and supports_v2_p2p - p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p)() + assert supports_v2_p2p or not supports_v2_pq + p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p, supports_v2_pq=supports_v2_pq)() self.p2ps.append(p2p_conn) if not expect_success: @@ -1015,7 +1018,7 @@ def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=Tru return p2p_conn - def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, wait_for_disconnect=False, p2p_idx, connection_type="outbound-full-relay", supports_v2_p2p=None, advertise_v2_p2p=None, **kwargs): + def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, wait_for_disconnect=False, p2p_idx, connection_type="outbound-full-relay", supports_v2_p2p=None, advertise_v2_p2p=None, supports_v2_pq=False, wait_for_v2_handshake=True, **kwargs): """Add an outbound p2p connection from node. Must be an "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection. @@ -1029,6 +1032,8 @@ def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, wait_fo Parameters: supports_v2_p2p: whether p2p_conn supports v2 P2P or not advertise_v2_p2p: whether p2p_conn is advertised to support v2 P2P or not + supports_v2_pq: whether a v2 p2p_conn also negotiates hybrid post-quantum keys (opt-in) + wait_for_v2_handshake: whether to wait until the v2 handshake, including a hybrid key confirmation, is complete An outbound connection is made from TestNode -------> P2PConnection - if P2PConnection doesn't advertise_v2_p2p, TestNode sends version message and v1 P2P is followed @@ -1056,7 +1061,8 @@ def addconnection_callback(address, port): reconnect = advertise_v2_p2p and not supports_v2_p2p # P2PConnection needs to be advertised to support v2 P2P so that ellswift bytes are sent instead of msg_version supports_v2_p2p = supports_v2_p2p and advertise_v2_p2p - p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p, reconnect=reconnect, **kwargs)() + assert supports_v2_p2p or not supports_v2_pq + p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p, reconnect=reconnect, supports_v2_pq=supports_v2_pq, **kwargs)() if reconnect: p2p_conn.wait_for_reconnect() @@ -1069,6 +1075,8 @@ def addconnection_callback(address, port): p2p_conn.wait_for_connect() self.p2ps.append(p2p_conn) + if supports_v2_p2p and not wait_for_v2_handshake: + return p2p_conn if supports_v2_p2p: p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake) p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg) diff --git a/test/functional/test_framework/v2_p2p.py b/test/functional/test_framework/v2_p2p.py index 87600c36de..fb7f9c35fa 100644 --- a/test/functional/test_framework/v2_p2p.py +++ b/test/functional/test_framework/v2_p2p.py @@ -2,14 +2,36 @@ # Copyright (c) 2022 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. -"""Class for v2 P2P protocol (see BIP 324)""" +"""Class for v2 P2P protocol (see BIP 324), with qbit's opt-in hybrid post-quantum key exchange +(doc/design/pq-transport.md)""" +from enum import Enum +from io import BytesIO +import hashlib +import json +import os +from pathlib import Path import random +import unittest -from .crypto.bip324_cipher import FSChaCha20Poly1305 +from .crypto.bip324_cipher import ( + FSChaCha20Poly1305, + hybrid_ciphers, + hybrid_derive_keys, + hybrid_transcript_input, + ser_compact_size, +) from .crypto.chacha20 import FSChaCha20 from .crypto.ellswift import ellswift_create, ellswift_ecdh_xonly from .crypto.hkdf import hkdf_sha256 +from .crypto.mlkem import ( + CT_SIZE, + EK_SIZE, + mlkem1024_check_ek, + mlkem1024_decaps, + mlkem1024_encaps, + mlkem1024_keygen, +) from .key import TaggedHash from .messages import MAGIC_BYTES @@ -20,6 +42,55 @@ LENGTH_FIELD_LEN = 3 MAX_GARBAGE_LEN = 4095 +# Hybrid post-quantum negotiation (doc/design/pq-transport.md). +PQ_MLKEM1024 = 0xF0 # record header of the ML-KEM-1024 offer (responder) and accept (initiator) +PQ_RECORD_LEN = 1572 # CompactSize(1569) || header || 1,568-byte payload +PQ_CONFIRMATION_LEN = 20 # encrypted length (3) || encrypted header (1) || tag (16) +# The largest packet contents a node accepts (src/net.cpp): a 0x00 byte, a 12-byte message type and +# MAX_PROTOCOL_MESSAGE_LENGTH bytes. It bounds the version packet contents too. +MAX_CONTENTS_LEN = 1 + 12 + 4_000_000 +# ReadCompactSize's range check (MAX_SIZE in src/serialize.h). +MAX_COMPACT_SIZE = 0x02000000 + + +class ParseKind(Enum): + """What parse_version_contents() found, as PQHandshake::ParseContents reports it.""" + NO_RECORD = 0 + INVALID_GRAMMAR = 1 + OWN_RECORD = 2 + + +def ser_pq_record(payload, header=PQ_MLKEM1024): + """A version-contents record: CompactSize(1 + len(payload)) || header || payload.""" + return ser_compact_size(1 + len(payload)) + bytes([header]) + payload + + +def parse_version_contents(contents): + """Parse version packet contents as records, the way a node does. + + The whole contents must parse; a zero, non-canonical, oversized or truncated length anywhere + gives INVALID_GRAMMAR. Otherwise the first own (0xF0) record wins. Returns (kind, payload), + where payload is the first own record's payload for OWN_RECORD and None otherwise.""" + pos = 0 + first = None + while pos < len(contents): + prefix = contents[pos] + size = {0xfd: 2, 0xfe: 4, 0xff: 8}.get(prefix, 0) + if pos + 1 + size > len(contents): + return ParseKind.INVALID_GRAMMAR, None + length = int.from_bytes(contents[pos + 1:pos + 1 + size], 'little') if size else prefix + if size and length < {2: 253, 4: 0x10000, 8: 0x100000000}[size]: + return ParseKind.INVALID_GRAMMAR, None # non-canonical + pos += 1 + size + if length > MAX_COMPACT_SIZE or length == 0 or length > len(contents) - pos: + return ParseKind.INVALID_GRAMMAR, None + if contents[pos] == PQ_MLKEM1024 and first is None: + first = contents[pos + 1:pos + length] + pos += length + if first is None: + return ParseKind.NO_RECORD, None + return ParseKind.OWN_RECORD, first + SHORTID = { 1: b"addr", 2: b"block", @@ -67,6 +138,12 @@ class EncryptedP2PState: net (string): chain used (regtest, signet etc..) + supports_v2_pq (bool): negotiate qbit's hybrid post-quantum keys (doc/design/pq-transport.md). A + responder offers an ML-KEM-1024 encapsulation key in its version packet; an initiator holds its + version packet until the responder's arrives and accepts a valid offer. After a switch, each + side's first packet under the hybrid keys is a key confirmation. Without it (the default), the + peer is a legacy BIP324 peer: it sends empty version contents and ignores the peer's. + Methods: perform an advanced form of diffie-hellman handshake to instantiate the encrypted transport. before exchanging any P2P messages, 2 nodes perform this handshake in order to determine a shared secret that is unique to both @@ -78,12 +155,13 @@ class EncryptedP2PState: encrypt/decrypt v2 P2P messages using v2_enc_packet() and v2_receive_packet(). """ - def __init__(self, *, initiating, net): + def __init__(self, *, initiating, net, supports_v2_pq=False): self.initiating = initiating # True if initiator self.net = net self.peer = {} # object with various BIP324 derived keys and ciphers self.privkey_ours = None self.ellswift_ours = None + self.ellswift_theirs = None self.sent_garbage = b"" self.received_garbage = b"" self.received_prefix = b"" # received ellswift bytes till the first mismatch from 16 bytes v1_prefix @@ -93,6 +171,26 @@ def __init__(self, *, initiating, net): self.contents_len = -1 self.found_garbage_terminator = False self.transport_version = b'' + # Hybrid post-quantum negotiation. Tests read these; subclasses may override the make_* methods. + self.supports_v2_pq = supports_v2_pq + # The version packet contents each side sent: None until sent or received. + self.sent_version_contents = None + self.received_version_contents = None + # The negotiation outcome from this side's view: None while it runs, then "hybrid" or "legacy_peer" + # (or "off" without supports_v2_pq). + self.pq_status = None + self.ecdh_secret = None # the BIP324 ECDH secret, which the hybrid keys also use + self.mlkem_dk = None # responder: the decapsulation key of our offer + self.mlkem_shared_secret = None + self.hybrid_keys = None # hybrid_derive_keys() output after a switch + # The AAD of our next packet: the garbage we sent, until our first packet after the garbage terminator. + self.send_aad = None + # After a switch: our key confirmation, while withheld; whether we sent it; whether the peer's verified. + self.withhold_confirmation = False + self.held_confirmation = None + self.confirmation_sent = False + self.peer_confirmed = False + self.confirmation_len = -1 # the decrypted length of the peer's key confirmation, once received @staticmethod def v2_ecdh(priv, ellswift_theirs, ellswift_ours, initiating): @@ -152,6 +250,9 @@ def complete_handshake(self, response): sends garbage terminator + optional decoy packets + transport version packet. Done by both initiator and responder. + A hybrid initiator (supports_v2_pq) holds its version packet until the responder's arrives, and a + hybrid responder offers an ML-KEM-1024 encapsulation key in its version packet. + Returns: 1. int - length of bytes that were consumed. returns 0 if all 64 bytes from ellswift haven't been received yet. 2. bytes - bytes to be sent to the peer when completing the v2 handshake @@ -160,28 +261,42 @@ def complete_handshake(self, response): # return b"" if we need to receive more bytes if len(ellswift_theirs) != 64: return 0, b"" - ecdh_secret = self.v2_ecdh(self.privkey_ours, ellswift_theirs, self.ellswift_ours, self.initiating) - self.initialize_v2_transport(ecdh_secret) + self.ellswift_theirs = ellswift_theirs + self.ecdh_secret = self.v2_ecdh(self.privkey_ours, ellswift_theirs, self.ellswift_ours, self.initiating) + self.initialize_v2_transport(self.ecdh_secret) # Send garbage terminator msg_to_send = self.peer['send_garbage_terminator'] # Optionally send decoy packets after garbage terminator. - aad = self.sent_garbage - for decoy_content_len in [random.randint(1, 100) for _ in range(random.randint(0, 10))]: - msg_to_send += self.v2_enc_packet(decoy_content_len * b'\x00', aad=aad, ignore=True) - aad = b'' + self.send_aad = self.sent_garbage + msg_to_send += self.make_decoys() + if self.supports_v2_pq and self.initiating: + # Hold the version packet: it answers the responder's offer, if any. + return 64 - len(self.received_prefix), msg_to_send + if self.supports_v2_pq: + # Offer an encapsulation key, from two independent 32-byte draws (d || z). + ek, self.mlkem_dk = mlkem1024_keygen(self.pq_random_bytes(32) + self.pq_random_bytes(32)) + contents = self.make_offer_contents(ser_pq_record(ek)) + else: + contents = self.transport_version # Send version packet. - msg_to_send += self.v2_enc_packet(self.transport_version, aad=aad) + msg_to_send += self.send_version_packet(contents) return 64 - len(self.received_prefix), msg_to_send def authenticate_handshake(self, response): - """ Ensures that the received optional decoy packets and transport version packet are authenticated. + """ Ensures that the received optional decoy packets and transport version packet are authenticated, and + after a switch to hybrid keys, that the peer's key confirmation verifies. Marks the v2 handshake as complete. Done by both initiator and responder. + The handshake is complete after the peer's version packet, or after a switch once the peer's key + confirmation verified and ours was sent (see withhold_confirmation and release_confirmation()). + Returns: 1. int - length of bytes that were processed so that recvbuf can be updated 2. bool - True if the authentication was successful/more bytes need to be received and False otherwise + 3. bytes - bytes to send to the peer: a hybrid initiator's version packet and either side's key confirmation """ processed_length = 0 + to_send = b"" # Detect garbage terminator in the received bytes if not self.found_garbage_terminator: @@ -198,30 +313,170 @@ def authenticate_handshake(self, response): else: # don't update recvbuf since more bytes need to be received if len(response) == 0: - return 0, True + return 0, True, to_send received_garbage += response[:1] processed_length += 1 response = response[1:] else: # disconnect since garbage terminator was not seen after 4 KiB of garbage. - return processed_length, False + return processed_length, False, to_send # Process optional decoy packets and transport version packet - while not self.tried_v2_handshake: + while self.received_version_contents is None: length, contents = self.v2_receive_packet(response, aad=self.received_garbage) if length == -1: - return processed_length, False + return processed_length, False, to_send elif length == 0: - return processed_length, True + return processed_length, True, to_send processed_length += length + response = response[length:] self.received_garbage = b"" - # decoy packets have contents = None. v2 handshake is complete only when version packet - # (can be empty with contents = b"") with contents != None is received. + # decoy packets have contents = None. the version packet (can be empty with contents = b"") has + # contents != None. if contents is not None: - assert contents == b"" # currently TestNode sends an empty version packet - self.tried_v2_handshake = True - return processed_length, True - response = response[length:] + self.received_version_contents = contents + ok, version_bytes = self.process_version_contents(contents) + to_send += version_bytes + if not ok: + return processed_length, False, to_send + + # After a switch, the peer's first packet under the hybrid keys must be its key confirmation. + if self.hybrid_keys is not None and not self.peer_confirmed: + length, ok = self.receive_confirmation(response) + if not ok: + return processed_length, False, to_send + processed_length += length + self.check_handshake_complete() + return processed_length, True, to_send + + def check_handshake_complete(self): + """Mark the handshake complete once nothing is left to exchange before application messages.""" + if self.received_version_contents is None: + return + if self.hybrid_keys is None or (self.peer_confirmed and self.confirmation_sent): + self.tried_v2_handshake = True + + def process_version_contents(self, contents): + """Act on the peer's version packet contents. + + Returns: + 1. bool - False to disconnect + 2. bytes - bytes to send: a held version packet and our key confirmation + """ + if not self.supports_v2_pq: + # A legacy BIP324 peer ignores the contents, whatever they are. + self.pq_status = "off" + return True, b"" + kind, payload = parse_version_contents(contents) + if self.initiating: + if kind != ParseKind.OWN_RECORD: + # A legacy responder (or contents that do not parse): send the held, empty version packet. + self.pq_status = "legacy_peer" + return True, self.send_version_packet(b"") + if len(payload) != EK_SIZE or not mlkem1024_check_ek(payload): + return False, b"" + self.mlkem_shared_secret, ct = mlkem1024_encaps(payload, self.pq_random_bytes(32)) + # The accept travels under the ECDH keys, then both directions switch. + to_send = self.send_version_packet(self.make_accept_contents(ser_pq_record(ct))) + return True, to_send + self.switch_to_hybrid() + if kind != ParseKind.OWN_RECORD: + # A legacy initiator: keep the ECDH keys. + self.pq_status = "legacy_peer" + self.mlkem_dk = None + return True, b"" + if len(payload) != CT_SIZE: + return False, b"" + self.mlkem_shared_secret = mlkem1024_decaps(self.mlkem_dk, payload) + self.mlkem_dk = None + return True, self.switch_to_hybrid() + + def switch_to_hybrid(self): + """Replace all four packet ciphers and the session id with the hybrid ones, and return our key + confirmation (or b"" if it is withheld).""" + if self.initiating: + contents_initiator, contents_responder = self.sent_version_contents, self.received_version_contents + ellswift_initiator, ellswift_responder = self.ellswift_ours, self.ellswift_theirs + else: + contents_initiator, contents_responder = self.received_version_contents, self.sent_version_contents + ellswift_initiator, ellswift_responder = self.ellswift_theirs, self.ellswift_ours + transcript_hash = hashlib.sha256(hybrid_transcript_input( + ellswift_initiator, ellswift_responder, contents_responder, contents_initiator)).digest() + self.hybrid_keys = hybrid_derive_keys(self.ecdh_secret, self.hybrid_shared_secret(), transcript_hash, + MAGIC_BYTES[self.net]) + (self.peer['send_L'], self.peer['send_P'], + self.peer['recv_L'], self.peer['recv_P']) = hybrid_ciphers(self.hybrid_keys, initiator=self.initiating) + self.peer['session_id'] = self.hybrid_keys['session_id'] + self.held_confirmation = self.make_confirmation() + if self.withhold_confirmation: + return b"" + return self.release_confirmation() + + def release_confirmation(self): + """Return our key confirmation to send, after a switch. The caller sends it before any later packet.""" + assert self.held_confirmation is not None and not self.confirmation_sent + confirmation, self.held_confirmation = self.held_confirmation, None + self.confirmation_sent = True + return confirmation + + def receive_confirmation(self, response): + """Check the peer's first packet under the hybrid keys: an empty decoy with empty AAD. + + Returns: + 1. int - number of bytes consumed + 2. bool - False on a failed confirmation""" + if self.confirmation_len == -1: + if len(response) < LENGTH_FIELD_LEN: + return 0, True + self.confirmation_len = int.from_bytes(self.peer['recv_L'].crypt(response[:LENGTH_FIELD_LEN]), 'little') + # A key mismatch decrypts to a random length: fail now rather than wait for its bytes. + if self.confirmation_len != 0: + return 0, False + if len(response) < PQ_CONFIRMATION_LEN: + return 0, True + plaintext = self.peer['recv_P'].decrypt(b"", response[LENGTH_FIELD_LEN:PQ_CONFIRMATION_LEN]) + if plaintext is None or not plaintext[0] & (1 << IGNORE_BIT_POS): + return 0, False + self.peer_confirmed = True + self.pq_status = "hybrid" + return PQ_CONFIRMATION_LEN, True + + def send_version_packet(self, contents): + """Encrypt our version packet under the ECDH keys.""" + self.sent_version_contents = contents + return self.enc_handshake_packet(contents) + + def enc_handshake_packet(self, contents, ignore=False): + """Encrypt a packet before our version packet is sent: the first one authenticates our garbage.""" + packet = self.v2_enc_packet(contents, aad=self.send_aad, ignore=ignore) + self.send_aad = b"" + return packet + + # Overridable parts of the handshake, for tests of misbehaving peers. + + def make_decoys(self): + """Decoy packets to send between our garbage terminator and our version packet.""" + return b"".join(self.enc_handshake_packet(random.randint(1, 100) * b'\x00', ignore=True) + for _ in range(random.randint(0, 10))) + + def make_offer_contents(self, offer): + """The responder's version packet contents, given its offer record.""" + return offer + + def make_accept_contents(self, accept): + """The initiator's version packet contents, given its accept record.""" + return accept + + def hybrid_shared_secret(self): + """The ML-KEM-1024 shared secret the hybrid keys are derived from.""" + return self.mlkem_shared_secret + + def make_confirmation(self): + """Our key confirmation: an empty decoy packet with empty AAD, the first packet under the hybrid keys.""" + return self.v2_enc_packet(b"", aad=b"", ignore=True) + + def pq_random_bytes(self, n): + """Randomness for ML-KEM key generation and encapsulation.""" + return os.urandom(n) def initialize_v2_transport(self, ecdh_secret): """Sets the peer object with various BIP324 derived keys and ciphers.""" @@ -281,3 +536,168 @@ def v2_receive_packet(self, response, aad=b''): length = LENGTH_FIELD_LEN + HEADER_LEN + self.contents_len + CHACHA20POLY1305_EXPANSION self.contents_len = -1 return length, None if (header[0] & (1 << IGNORE_BIT_POS)) else plaintext[HEADER_LEN:] + + +def _drive(state, recvbuf): + """Feed recvbuf to state the way P2PConnection's v2 handshake driver does. + + Returns the unconsumed bytes and the bytes state sends.""" + to_send = b"" + if not state.peer: + if not state.initiating and state.ellswift_ours is None: + length, response = state.respond_v2_handshake(BytesIO(recvbuf)) + recvbuf = recvbuf[length:] + assert response not in (-1, b"") + to_send += response + length, response = state.complete_handshake(BytesIO(recvbuf)) + recvbuf = recvbuf[length:] + if not response: + return recvbuf, to_send + to_send += response + length, ok, pending = state.authenticate_handshake(recvbuf) + assert ok + return recvbuf[length:], to_send + pending + + +def _run_handshake(initiator, responder): + """Run a v2 handshake between two states in memory until neither has anything left to send.""" + to_responder, to_initiator = initiator.initiate_v2_handshake(), b"" + initiator_buf = responder_buf = b"" + while to_responder or to_initiator: + responder_buf += to_responder + responder_buf, to_initiator = _drive(responder, responder_buf) if not responder.tried_v2_handshake else (responder_buf, b"") + initiator_buf += to_initiator + initiator_buf, to_responder = _drive(initiator, initiator_buf) if not initiator.tried_v2_handshake else (initiator_buf, b"") + return initiator_buf, responder_buf + + +class TestFrameworkV2P2P(unittest.TestCase): + def test_parse_version_contents(self): + """The contents parser follows the node's grammar: all records must parse, the first own record wins.""" + offer = ser_pq_record(bytes(EK_SIZE)) + self.assertEqual(offer[:4].hex(), "fd2106f0") + self.assertEqual(len(offer), PQ_RECORD_LEN) + for contents, expected in [ + (b"", (ParseKind.NO_RECORD, None)), + (bytes.fromhex("02f100"), (ParseKind.NO_RECORD, None)), + (bytes.fromhex("0100"), (ParseKind.NO_RECORD, None)), + (offer, (ParseKind.OWN_RECORD, bytes(EK_SIZE))), + (bytes.fromhex("01f0"), (ParseKind.OWN_RECORD, b"")), + (bytes.fromhex("04f0aabbcc"), (ParseKind.OWN_RECORD, bytes.fromhex("aabbcc"))), + (bytes.fromhex("02f111") + bytes.fromhex("02f022") + bytes.fromhex("02f033"), (ParseKind.OWN_RECORD, b"\x22")), + (offer + b"\x00", (ParseKind.INVALID_GRAMMAR, None)), # zero length after a valid record + (offer + b"\x05", (ParseKind.INVALID_GRAMMAR, None)), # truncated trailing record + (bytes.fromhex("00"), (ParseKind.INVALID_GRAMMAR, None)), + (bytes.fromhex("fd0200f000"), (ParseKind.INVALID_GRAMMAR, None)), # non-canonical CompactSize + (bytes.fromhex("fe0000010000"), (ParseKind.INVALID_GRAMMAR, None)), # 65536, but nothing follows + (bytes.fromhex("ff0100000000000000f0"), (ParseKind.INVALID_GRAMMAR, None)), # non-canonical 8-byte + (bytes.fromhex("fe01000002") + bytes(8), (ParseKind.INVALID_GRAMMAR, None)), # above MAX_SIZE + (bytes.fromhex("fd21"), (ParseKind.INVALID_GRAMMAR, None)), # truncated CompactSize + (offer[:-1], (ParseKind.INVALID_GRAMMAR, None)), # truncated payload + ]: + self.assertEqual(parse_version_contents(contents), expected, contents[:8].hex()) + + def test_vectors(self): + """Two hybrid states fed the inputs of src/test/data/pq_transport_vectors.json reproduce its values.""" + vectors_path = Path(__file__).resolve().parents[3] / "src" / "test" / "data" / "pq_transport_vectors.json" + vectors = json.loads(vectors_path.read_text(encoding="utf8"))["vectors"] + self.assertEqual(len(vectors), 4) + for vector in vectors: + v = {key: bytes.fromhex(value) for key, value in vector.items() if key not in ("comment", "chain")} + net = {"main": "mainnet", "regtest": "regtest"}[vector["chain"]] + self.assertEqual(MAGIC_BYTES[net], v["magic"]) + + class FixedState(EncryptedP2PState): + def generate_keypair_and_garbage(self, garbage_len=None): + role = "initiator" if self.initiating else "responder" + self.privkey_ours, self.ellswift_ours = v[f"{role}_privkey"], v[f"{role}_ellswift"] + self.sent_garbage = random.randbytes(random.randrange(1, 64)) + return self.ellswift_ours + self.sent_garbage + + def pq_random_bytes(self, n): + return self.draws.pop(0) + + def make_offer_contents(self, offer): + assert offer == ser_pq_record(v["ek"]) + return v["contents_responder"] + + def make_accept_contents(self, accept): + assert accept == ser_pq_record(v["ct"]) + return v["contents_initiator"] + + initiator = FixedState(initiating=True, net=net, supports_v2_pq=True) + responder = FixedState(initiating=False, net=net, supports_v2_pq=True) + initiator.draws = [v["mlkem_encaps_m"]] + responder.draws = [v["mlkem_keygen_seed"][:32], v["mlkem_keygen_seed"][32:]] + initiator.withhold_confirmation = responder.withhold_confirmation = True + _run_handshake(initiator, responder) + for state in (initiator, responder): + self.assertEqual(state.ecdh_secret, v["ss_ecdh"]) + self.assertEqual(state.mlkem_shared_secret, v["ss_mlkem"]) + self.assertEqual(state.peer["session_id"], v["session_id"]) + for label in ("salt", "ikm", "prk", "initiator_L", "initiator_P", "responder_L", "responder_P"): + self.assertEqual(state.hybrid_keys[label], v[label]) + self.assertEqual(initiator.received_version_contents, v["contents_responder"]) + self.assertEqual(responder.received_version_contents, v["contents_initiator"]) + # Each side's held confirmation is hybrid packet 0, and its first application packet follows. + self.assertEqual(initiator.held_confirmation, v["initiator_confirmation"]) + self.assertEqual(responder.held_confirmation, v["responder_confirmation"]) + self.assertEqual(initiator.v2_enc_packet(v["initiator_first_contents"]), v["initiator_first_packet"]) + self.assertEqual(responder.v2_enc_packet(v["responder_first_contents"]), v["responder_first_packet"]) + + def test_roles(self): + """Hybrid and legacy states interoperate in both roles, and only two hybrid states switch.""" + for initiator_pq, responder_pq in [(True, True), (True, False), (False, True), (False, False)]: + initiator = EncryptedP2PState(initiating=True, net="regtest", supports_v2_pq=initiator_pq) + responder = EncryptedP2PState(initiating=False, net="regtest", supports_v2_pq=responder_pq) + _run_handshake(initiator, responder) + self.assertTrue(initiator.tried_v2_handshake and responder.tried_v2_handshake) + hybrid = initiator_pq and responder_pq + self.assertEqual((initiator.hybrid_keys is not None, responder.hybrid_keys is not None), (hybrid, hybrid)) + self.assertEqual(initiator.peer["session_id"], responder.peer["session_id"]) + self.assertEqual(initiator.pq_status, "hybrid" if hybrid else "legacy_peer" if initiator_pq else "off") + self.assertEqual(responder.pq_status, "hybrid" if hybrid else "legacy_peer" if responder_pq else "off") + # A legacy initiator sends empty contents; so does a hybrid one toward a legacy responder. + self.assertEqual(len(initiator.sent_version_contents), PQ_RECORD_LEN if hybrid else 0) + self.assertEqual(len(responder.sent_version_contents), PQ_RECORD_LEN if responder_pq else 0) + # Packets flow both ways, across a rekey. + for i in range(300): + contents = i.to_bytes(2, "little") + packet = initiator.v2_enc_packet(contents) + self.assertEqual(responder.v2_receive_packet(packet), (len(packet), contents)) + packet = responder.v2_enc_packet(contents) + self.assertEqual(initiator.v2_receive_packet(packet), (len(packet), contents)) + + def test_withheld_confirmation(self): + """A withheld confirmation keeps the handshake open on that side until it is released.""" + initiator = EncryptedP2PState(initiating=True, net="regtest", supports_v2_pq=True) + responder = EncryptedP2PState(initiating=False, net="regtest", supports_v2_pq=True) + initiator.withhold_confirmation = True + initiator_buf, responder_buf = _run_handshake(initiator, responder) + self.assertTrue(initiator.peer_confirmed) + self.assertFalse(initiator.tried_v2_handshake) + self.assertFalse(responder.peer_confirmed) + self.assertFalse(responder.tried_v2_handshake) + confirmation = initiator.release_confirmation() + self.assertEqual(len(confirmation), PQ_CONFIRMATION_LEN) + initiator.check_handshake_complete() + self.assertTrue(initiator.tried_v2_handshake) + self.assertEqual(_drive(responder, responder_buf + confirmation), (b"", b"")) + self.assertTrue(responder.tried_v2_handshake) + + def test_bad_confirmation(self): + """A confirmation with a nonzero length, a bad tag or a clear ignore bit fails.""" + for damage in ("length", "tag", "not_decoy"): + initiator = EncryptedP2PState(initiating=True, net="regtest", supports_v2_pq=True) + responder = EncryptedP2PState(initiating=False, net="regtest", supports_v2_pq=True) + initiator.withhold_confirmation = True + _, responder_buf = _run_handshake(initiator, responder) + initiator.release_confirmation() + if damage == "length": + confirmation = initiator.v2_enc_packet(b"\x00", ignore=True) + elif damage == "tag": + confirmation = initiator.v2_enc_packet(b"", ignore=True) + confirmation = confirmation[:-1] + bytes([confirmation[-1] ^ 1]) + else: + confirmation = initiator.v2_enc_packet(b"", ignore=False) + self.assertEqual(responder.authenticate_handshake(responder_buf + confirmation)[1], False) diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py index eaf80f64fa..14bb2da048 100755 --- a/test/functional/test_runner.py +++ b/test/functional/test_runner.py @@ -273,6 +273,7 @@ 'p2p_v2_transport.py', 'p2p_v2_encrypted.py', 'p2p_v2_misbehaving.py', + 'p2p_v2_pq.py', 'example_test.py', 'feature_minchainwork_requested_block.py', 'mempool_truc.py', From dfd5a22aeb1748b26ea49a3fc272ab83db4a3715 Mon Sep 17 00:00:00 2001 From: Dan Hepworth Date: Tue, 6 Oct 2026 13:02:23 -0400 Subject: [PATCH 2/2] test: cover malformed negotiation, confirmation and fallback end to end Drive every way a hybrid negotiation can go wrong through a real node, using scripted variants of the Python hybrid peer, and check the fallback that follows repeated failures. p2p_v2_pq_misbehaving.py: - test_records: offers (the node initiates) and accepts (the node responds) of every shape: EK and CT one byte short and long, a bad-modulus EK, two own records (the first wins either way), unknown and reserved headers, junk after a record, zero, non-canonical, past-the-end, above-MAX_SIZE and truncated lengths, contents padded to exactly 4,000,013 bytes and one byte more, a damaged version tag, and version packets cut short before EOF. Each case checks that the connection is kept (with its transport_pq_status) or dropped at once, the exact log lines, and the exact per-direction counter deltas. - test_confirmation: a nonzero length, a bad tag and a non-decoy confirmation in both roles, a corrupted CT and a responder that derives another secret: each disconnects at once, long before -peertimeout. - test_decoys_and_partial_writes: decoys before the version packet (the first one authenticating the garbage), decoys after the confirmation, coalesced bursts and a handshake fragmented into 1- to 7-byte writes, in both roles. - test_unsolicited_accept: a switched-off responder ignores own-record bytes it never asked for. - test_load_shedding: a threaded flood of 64-byte keys starts load shedding; while shed, a hybrid initiator sees a legacy responder, an accept is ignored and outbound connections still negotiate; then, with no inbound connection at all, the socket handler alone ends shedding after the quiet period. Shedding counts seconds of the steady clock, so a machine that can't outpace the threshold skips this part, and one that could but didn't shed fails. p2p_v2_pq_fallback.py, with addnode onetry for explicit outbound attempts (addconnection can't open manual connections): - test_three_failures: -test=pq_fail_first_packet on one node; three failed attempts enter the fallback set (fallback = 1, one warning line, the set's entry and ring), and the fourth connection is plain v2 with status fallback. - test_expiry_and_restart: past the window the negotiation is tried again with the next window at 4 h; a restart clears everything under a new instance_id. - test_malformed_offer_fallback: against a responder that always offers a bad-modulus key, the fourth connection completes, because a fallback connection neither parses nor validates offers. - test_inbound_and_success_exclusion: inbound failures, and outbound failures after a hybrid success with the endpoint, never lead to a fallback. - test_endpoint_separation: through a SOCKS5 proxy, hostnames in any letter case with or without the root dot, one hostname on two ports and one proxied address on two ports keep separate histories. This drives ConnectNode's name-proxy endpoint key through the real path. Mutating the node, each of these is killed: dropping the socket handler's shedding update, keying name-proxy endpoints by the proxy, reporting v2 before the confirmation, parsing only up to the first own record, dropping the early confirmation length check, skipping the fallback lookup, the success protection or the window escalation, fixed KEM randomness, counting inbound failures, and parsing contents with the switch off. Work item G2 of #212 (#184 part 7). Co-Authored-By: Claude Opus 5.5 --- test/functional/p2p_v2_pq_fallback.py | 345 +++++++++++++++ test/functional/p2p_v2_pq_misbehaving.py | 520 +++++++++++++++++++++++ test/functional/test_runner.py | 2 + 3 files changed, 867 insertions(+) create mode 100755 test/functional/p2p_v2_pq_fallback.py create mode 100755 test/functional/p2p_v2_pq_misbehaving.py diff --git a/test/functional/p2p_v2_pq_fallback.py b/test/functional/p2p_v2_pq_fallback.py new file mode 100755 index 0000000000..805bdf82b8 --- /dev/null +++ b/test/functional/p2p_v2_pq_fallback.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026-present The qbit core developers +# Distributed under the MIT software license, see the accompanying +# file COPYING or https://opensource.org/license/mit. +"""Test the bounded fallback to plain v2 after repeated hybrid post-quantum failures with one +endpoint (#184, doc/design/pq-transport.md). + +- test_three_failures: three explicit outbound attempts whose key confirmation fails put the + endpoint in the fallback set; the fourth connection is plain v2 with status fallback, and one + warning line says so. +- test_malformed_offer_fallback: against a responder that always sends a malformed offer, the + fourth connection succeeds, because a fallback connection neither parses nor validates offers. +- test_inbound_and_success_exclusion: inbound failures never lead to a fallback, and neither do + outbound failures after a hybrid success with the same endpoint. +- test_expiry_and_restart: the hybrid negotiation is tried again once the window expires (and the + next window is longer), and a restart clears all of it under a new instance_id. +- test_endpoint_separation: through a SOCKS5 proxy, two hostnames (in any letter case, with or + without the root dot), one hostname on two ports, and one address on two ports each keep their + own history: a name-proxy destination is its hostname and port, never the proxy. + +addconnection can't open manual connections, so explicit outbound attempts use addnode onetry. +""" + +import threading + +from test_framework.p2p import P2PInterface +from test_framework.socks5 import Socks5Configuration, Socks5Server +from test_framework.test_framework import BitcoinTestFramework +from test_framework.util import ( + MAX_NODES, + assert_equal, + assert_not_equal, + p2p_port, +) + +PQ_ON = ["-v2transport=1", "-v2pqtransport=1"] +FAIL_FIRST_PACKET = "-test=pq_fail_first_packet" +# Counted failures (the outbound counters that advance a streak) and the fallback windows. +COUNTED = ("outbound_malformed_record", "outbound_first_packet_failed", "outbound_closed_after_switch") +FIRST_WINDOW = 3600 +SECOND_WINDOW = 4 * 3600 +# The fake public address the SOCKS5 proxy maps to Python peers. +PROXIED_IP = "15.61.23.23" + + +def responder_port(idx): + """The port a Python responder with connect_id idx listens on (NetworkThread.listen()).""" + return p2p_port(MAX_NODES - idx) + + +def counters(node): + info = node.getpqtransportinfo()["handshakes"] + return {f"{direction}_{key}": value for direction in ("inbound", "outbound") for key, value in info[direction].items()} + + +def counted_failures(node): + return sum(counters(node)[key] for key in COUNTED) + + +def address_endpoint(address, port): + return {"kind": "address", "network": "ipv4", "address": address, "port": port} + + +def name_endpoint(hostname, port): + return {"kind": "name_proxy", "network": "name_proxy", "address": hostname, "port": port} + + +def bad_modulus_offer(state, offer): + """An offer whose key fails the FIPS 203 modulus check: its first coefficient is 4095.""" + return offer[:4] + b"\xff" + bytes([offer[5] | 0x0f]) + offer[6:] + + +class MalformedOfferPeer(P2PInterface): + """A hybrid Python responder that always offers a key failing the modulus check.""" + def make_v2_state(self, **kwargs): + state = super().make_v2_state(**kwargs) + state.make_offer_contents = bad_modulus_offer.__get__(state) + return state + + +class BadConfirmationPeer(P2PInterface): + """A hybrid Python initiator whose key confirmation is not a decoy.""" + def make_v2_state(self, **kwargs): + state = super().make_v2_state(**kwargs) + state.make_confirmation = lambda: state.v2_enc_packet(b"", aad=b"", ignore=False) + return state + + +class P2PV2PQFallbackTest(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 2 + self.extra_args = [PQ_ON, PQ_ON] + + def setup_network(self): + self.setup_nodes() + + def run_test(self): + self.test_three_failures() + self.test_expiry_and_restart() + self.test_malformed_offer_fallback() + self.test_inbound_and_success_exclusion() + self.test_endpoint_separation() + + def wait_no_peers(self, *nodes): + for node in nodes: + self.wait_until(lambda node=node: len(node.getpeerinfo()) == 0) + + def connect_responder(self, node, peer, *, idx, target=None): + """Let a Python responder listen on its own port and make node open a manual connection to it + (to target, through a proxy that leads there, if given).""" + listening = threading.Event() + peer.peer_accept_connection(connect_cb=lambda addr, port: listening.set(), connect_id=idx, net=node.chain, + timeout_factor=node.timeout_factor, supports_v2_p2p=True, + supports_v2_pq=True, reconnect=False)() + node.p2ps.append(peer) + # The RPC call comes from this thread, not the network thread's callback: the RPC connection is + # not shared across threads. + self.wait_until(listening.is_set) + node.addnode(node=target or f"127.0.0.1:{responder_port(idx)}", command="onetry") + return responder_port(idx) + + def test_three_failures(self): + node_a, node_b = self.nodes + self.log.info("Three failed key confirmations with one endpoint put it in the fallback set") + self.restart_node(0, extra_args=PQ_ON + [FAIL_FIRST_PACKET]) + target = f"127.0.0.1:{p2p_port(1)}" + b_before = counters(node_b) + for attempt in range(1, 4): + expected = ["v2 pq: switched role=initiator conn_type=manual peer="] + unexpected = ["v2 pq: fallback"] + if attempt == 3: + expected.append("v2 pq: fallback cause=") + unexpected = [] + with node_a.assert_debug_log(expected, unexpected_msgs=unexpected): + node_a.addnode(node=target, command="onetry") + self.wait_until(lambda: counted_failures(node_a) == attempt) + self.wait_no_peers(node_a, node_b) + a_counts = counters(node_a) + assert_equal(a_counts["outbound_switched"], 3) + assert_equal(a_counts["outbound_fallback"], 1) + # node B saw node A's damaged confirmation each time; inbound failures never count. + b_counts = counters(node_b) + assert_equal(b_counts["inbound_first_packet_failed"] - b_before["inbound_first_packet_failed"], 3) + assert_equal(b_counts["outbound_fallback"], 0) + + info = node_a.getpqtransportinfo() + assert_equal(info["failure_streaks"], []) + [entry] = info["fallback_set"] + assert_equal(entry["endpoint"], address_endpoint("127.0.0.1", p2p_port(1))) + assert entry["cause"] in ("first_packet_failed", "closed_after_switch") + assert_equal((entry["streak"], entry["window_seconds"]), (3, FIRST_WINDOW)) + assert_equal(entry["expires"] - entry["entered"], FIRST_WINDOW) + ring = info["recent_failures"]["outbound"]["entries"] + assert_equal([e["outcome"] for e in ring][-1], "fallback") + assert_equal(ring[-1]["reason"], entry["cause"]) + assert_equal(ring[-1]["endpoint"], entry["endpoint"]) + assert_equal(len([e for e in ring if e["outcome"] in ("first_packet_failed", "closed_after_switch")]), 3) + with open(node_a.debug_log_path, encoding="utf-8") as log: + warnings = [line for line in log if "v2 pq: fallback cause=" in line] + assert_equal(len(warnings), 1) + assert f"v2 pq: fallback cause={entry['cause']} until=" in warnings[0] + assert "role=initiator conn_type=manual" in warnings[0] + + self.log.info("The fourth connection is plain v2, with status fallback") + before = counters(node_a) + with node_a.assert_debug_log([], unexpected_msgs=["v2 pq: switched", "v2 pq: fallback"]): + self.connect_nodes(0, 1) + [outbound] = node_a.getpeerinfo() + [inbound] = node_b.getpeerinfo() + assert_equal((outbound["transport_protocol_type"], outbound["transport_pq"], outbound["transport_pq_status"]), + ("v2", False, "fallback")) + # A fallback connection sends empty contents, so node B sees a legacy initiator. + assert_equal((inbound["transport_pq"], inbound["transport_pq_status"]), (False, "legacy_peer")) + assert_equal(outbound["session_id"], inbound["session_id"]) + assert_equal(counters(node_a), before) + self.disconnect_nodes(0, 1) + + def test_expiry_and_restart(self): + node_a, node_b = self.nodes + self.log.info("Once the window expires, the next connection tries the hybrid keys again") + [entry] = node_a.getpqtransportinfo()["fallback_set"] + for node in self.nodes: + node.setmocktime(entry["expires"]) + with node_a.assert_debug_log(["v2 pq: switched role=initiator conn_type=manual"], + unexpected_msgs=["v2 pq: fallback cause="]): + node_a.addnode(node=f"127.0.0.1:{p2p_port(1)}", command="onetry") + self.wait_until(lambda: counted_failures(node_a) == 4) + self.wait_no_peers(node_a, node_b) + info = node_a.getpqtransportinfo() + assert_equal(info["fallback_set"], []) + [streak] = info["failure_streaks"] + assert_equal(streak["endpoint"], address_endpoint("127.0.0.1", p2p_port(1))) + # The streak restarted, and the escalation stays: the next window is the second one. + assert_equal((streak["streak"], streak["next_window_seconds"]), (1, SECOND_WINDOW)) + assert_equal(info["handshakes"]["outbound"]["fallback"], 1) + + self.log.info("A restart clears the history and starts a new counting period") + instance_id = info["instance_id"] + for node in self.nodes: + node.setmocktime(0) + self.restart_node(0, extra_args=PQ_ON) + info = node_a.getpqtransportinfo() + assert_not_equal(info["instance_id"], instance_id) + assert_equal(len(info["instance_id"]), 64) + assert_equal((info["fallback_set"], info["failure_streaks"]), ([], [])) + assert_equal(set(counters(node_a).values()), {0}) + self.connect_nodes(0, 1) + assert_equal(node_a.getpeerinfo()[0]["transport_pq_status"], "hybrid") + self.disconnect_nodes(0, 1) + + def test_malformed_offer_fallback(self): + node_a = self.nodes[0] + self.log.info("Three malformed offers from one responder put it in the fallback set") + self.restart_node(0, extra_args=PQ_ON) + port = None + for attempt in range(1, 4): + with node_a.assert_debug_log(["v2 pq: malformed_record reason=ek_modulus role=initiator conn_type=manual"]): + peer = MalformedOfferPeer() + port = self.connect_responder(node_a, peer, idx=1) + self.wait_until(lambda: counters(node_a)["outbound_malformed_record"] == attempt) + peer.wait_for_disconnect() + self.wait_no_peers(node_a) + [entry] = node_a.getpqtransportinfo()["fallback_set"] + assert_equal(entry["endpoint"], address_endpoint("127.0.0.1", port)) + assert_equal((entry["cause"], entry["reason"], entry["streak"]), ("malformed_record", "ek_modulus", 3)) + + self.log.info("The fourth connection ignores the malformed offer and completes as plain v2") + before = counters(node_a) + peer = MalformedOfferPeer() + self.connect_responder(node_a, peer, idx=1) + peer.wait_for_connect() + peer.wait_for_verack() + peer.sync_with_ping() + [info] = node_a.getpeerinfo() + assert_equal((info["transport_protocol_type"], info["transport_pq_status"]), ("v2", "fallback")) + assert_equal((peer.v2_state.pq_status, peer.v2_state.received_version_contents), ("legacy_peer", b"")) + assert_equal(info["session_id"], peer.v2_state.peer["session_id"].hex()) + assert_equal(counters(node_a), before) + node_a.disconnect_p2ps() + self.wait_no_peers(node_a) + + def test_inbound_and_success_exclusion(self): + node_a = self.nodes[0] + self.log.info("Repeated inbound failures never lead to a fallback") + self.restart_node(0, extra_args=PQ_ON) + for attempt in range(1, 5): + peer = node_a.add_p2p_connection(BadConfirmationPeer(), supports_v2_pq=True, wait_for_verack=False, + wait_for_v2_handshake=False, expect_success=False) + self.wait_until(lambda: counters(node_a)["inbound_first_packet_failed"] == attempt) + peer.wait_for_disconnect() + self.wait_no_peers(node_a) + info = node_a.getpqtransportinfo() + assert_equal((info["fallback_set"], info["failure_streaks"]), ([], [])) + assert_equal(counters(node_a)["outbound_fallback"], 0) + peer = node_a.add_p2p_connection(P2PInterface(), supports_v2_pq=True) + assert_equal(node_a.getpeerinfo()[0]["transport_pq_status"], "hybrid") + node_a.disconnect_p2ps() + self.wait_no_peers(node_a) + + self.log.info("Outbound failures after a hybrid success with the same endpoint never lead to a fallback") + peer = P2PInterface() + self.connect_responder(node_a, peer, idx=2) + peer.wait_for_connect() + peer.wait_for_verack() + assert_equal(node_a.getpeerinfo()[0]["transport_pq_status"], "hybrid") + node_a.disconnect_p2ps() + self.wait_no_peers(node_a) + for attempt in range(1, 5): + peer = MalformedOfferPeer() + self.connect_responder(node_a, peer, idx=2) + self.wait_until(lambda: counters(node_a)["outbound_malformed_record"] == attempt) + peer.wait_for_disconnect() + self.wait_no_peers(node_a) + info = node_a.getpqtransportinfo() + assert_equal((info["fallback_set"], info["failure_streaks"]), ([], [])) + assert_equal(counters(node_a)["outbound_fallback"], 0) + + def test_endpoint_separation(self): + node_a = self.nodes[0] + self.log.info("Through a SOCKS5 proxy, every hostname, port and address keeps its own history") + # Every destination leads to the same Python responder: only the endpoint node A accounts the + # outcome to differs. + conf = Socks5Configuration() + conf.addr = ("127.0.0.1", p2p_port(self.num_nodes)) + conf.unauth = conf.auth = True + conf.destinations_factory = lambda addr, port: {"actual_to_addr": "127.0.0.1", "actual_to_port": responder_port(3)} + proxy = Socks5Server(conf) + proxy.start() + self.restart_node(0, extra_args=PQ_ON + [f"-proxy={conf.addr[0]}:{conf.addr[1]}"]) + + attempts = [ + # (what node A connects to, the endpoint it is accounted to) + ("alpha.pqtest:9001", name_endpoint("alpha.pqtest", 9001)), + ("ALPHA.PQtest.:9001", name_endpoint("alpha.pqtest", 9001)), + ("beta.pqtest:9001", name_endpoint("beta.pqtest", 9001)), + ("alpha.pqtest:9002", name_endpoint("alpha.pqtest", 9002)), + (f"{PROXIED_IP}:9001", address_endpoint(PROXIED_IP, 9001)), + (f"{PROXIED_IP}:9001", address_endpoint(PROXIED_IP, 9001)), + (f"{PROXIED_IP}:9002", address_endpoint(PROXIED_IP, 9002)), + ("Alpha.pqtest:9001", name_endpoint("alpha.pqtest", 9001)), + ] + for i, (target, endpoint) in enumerate(attempts, start=1): + self.log.info(f"- {target}") + peer = MalformedOfferPeer() + self.connect_responder(node_a, peer, idx=3, target=target) + self.wait_until(lambda: counters(node_a)["outbound_malformed_record"] == i) + peer.wait_for_disconnect() + self.wait_no_peers(node_a) + if i == 3: + # A third failure, but with two endpoints: no fallback yet. + assert_equal(node_a.getpqtransportinfo()["fallback_set"], []) + + info = node_a.getpqtransportinfo() + [entry] = info["fallback_set"] + assert_equal(entry["endpoint"], name_endpoint("alpha.pqtest", 9001)) + assert_equal((entry["cause"], entry["streak"]), ("malformed_record", 3)) + streaks = sorted((s["endpoint"]["kind"], s["endpoint"]["address"], s["endpoint"]["port"], s["streak"]) + for s in info["failure_streaks"]) + assert_equal(streaks, [ + ("address", PROXIED_IP, 9001, 2), + ("address", PROXIED_IP, 9002, 1), + ("name_proxy", "alpha.pqtest", 9002, 1), + ("name_proxy", "beta.pqtest", 9001, 1), + ]) + + self.log.info("Only the endpoint in the fallback set runs plain v2") + before = counters(node_a) + peer = MalformedOfferPeer() + self.connect_responder(node_a, peer, idx=3, target="alpha.pqtest.:9001") + peer.wait_for_connect() + peer.wait_for_verack() + assert_equal(node_a.getpeerinfo()[0]["transport_pq_status"], "fallback") + node_a.disconnect_p2ps() + self.wait_no_peers(node_a) + assert_equal(counters(node_a), before) + peer = MalformedOfferPeer() + self.connect_responder(node_a, peer, idx=3, target="beta.pqtest:9001") + self.wait_until(lambda: counters(node_a)["outbound_malformed_record"] == len(attempts) + 1) + peer.wait_for_disconnect() + self.wait_no_peers(node_a) + proxy.stop() + + +if __name__ == '__main__': + P2PV2PQFallbackTest(__file__).main() diff --git a/test/functional/p2p_v2_pq_misbehaving.py b/test/functional/p2p_v2_pq_misbehaving.py new file mode 100755 index 0000000000..7dc1e2aa1c --- /dev/null +++ b/test/functional/p2p_v2_pq_misbehaving.py @@ -0,0 +1,520 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026-present The qbit core developers +# Distributed under the MIT software license, see the accompanying +# file COPYING or https://opensource.org/license/mit. +"""Test how a node handles every way a hybrid post-quantum v2 negotiation can go wrong +(doc/design/pq-transport.md). + +- test_records: version contents of every shape, as offers (the node initiates) and as accepts + (the node responds): the connection is kept or dropped as specified, the log gives the exact + reason, and the per-direction counters move by exactly the expected amounts. +- test_confirmation: a damaged key confirmation, or keys that don't match, disconnects at once, + long before -peertimeout. +- test_decoys_and_partial_writes: decoys on both sides of the switch, coalesced writes and writes + fragmented byte by byte all complete the hybrid handshake. +- test_unsolicited_accept: own-header bytes sent to a responder that made no offer are ignored. +- test_load_shedding: an inbound offer flood starts load shedding, a shed responder ignores an + accept, outbound connections are unaffected, and the socket handler alone ends shedding once + the flood stops. Shedding counts real seconds, so a machine too slow to outpace the threshold + skips it. +""" + +from concurrent.futures import ThreadPoolExecutor +import bisect +import os +import random +import socket +import threading +import time +import types + +from test_framework.crypto.mlkem import CT_SIZE +from test_framework.messages import MAGIC_BYTES +from test_framework.p2p import NetworkThread, P2PInterface +from test_framework.test_framework import BitcoinTestFramework +from test_framework.util import ( + MAX_NODES, + assert_approx, + assert_equal, + p2p_port, +) +from test_framework.v2_p2p import ( + MAX_CONTENTS_LEN, + PQ_RECORD_LEN, + ser_pq_record, +) + +PQ_ON = ["-v2transport=1", "-v2pqtransport=1"] +PQ_OFF = ["-v2transport=1", "-v2pqtransport=0"] +# The node's default load shedding threshold (DEFAULT_PQ_SHED_THRESHOLD_PER_S). +SHED_THRESHOLD = 1000 +# Quiet seconds that end shedding (PQ_SHED_QUIET_PERIOD). +SHED_QUIET_SECONDS = 10 +# The flood's connections in flight, and how long it may try to outpace the threshold. +FLOOD_THREADS = 8 +FLOOD_SECONDS = 30 + + +class ScriptedPeer(P2PInterface): + """A Python v2 peer with some of its handshake steps replaced. + + Each keyword names an EncryptedP2PState attribute: a function replaces that method (it gets the + state as its first argument), any other value replaces the attribute.""" + def __init__(self, **overrides): + super().__init__() + self.overrides = overrides + + def make_v2_state(self, **kwargs): + state = super().make_v2_state(**kwargs) + for name, value in self.overrides.items(): + assert hasattr(state, name), name + setattr(state, name, types.MethodType(value, state) if callable(value) else value) + return state + + +class TruncatingPeer(ScriptedPeer): + """Sends only the first part of its version packet, then shuts down its sending side (EOF).""" + def __init__(self, keep, **overrides): + def send_version_packet(state, contents): + state.sent_version_contents = contents + return state.enc_handshake_packet(contents)[:keep] + super().__init__(send_version_packet=send_version_packet, **overrides) + self.closing = False + + def send_raw_message(self, raw_message_bytes): + if self.closing: + return # nothing more after EOF + super().send_raw_message(raw_message_bytes) + if self.v2_state.sent_version_contents is not None: + # Queued after the truncated packet, so the node receives it and then EOF. Reading on, rather + # than closing, keeps the node's later bytes from turning the close into a reset. + self.closing = True + NetworkThread.network_event_loop.call_soon_threadsafe(lambda: self._transport and self._transport.write_eof()) + + +class CoalescingPeer(ScriptedPeer): + """Writes everything it sends within one network event as a single write.""" + def __init__(self, **overrides): + super().__init__(**overrides) + self.pending = b"" + + def send_raw_message(self, raw_message_bytes): + def queue(): + if not self.pending: + NetworkThread.network_event_loop.call_soon(flush) + self.pending += raw_message_bytes + + def flush(): + data, self.pending = self.pending, b"" + if self._transport and not self._transport.is_closing(): + self._transport.write(data) + if not self.is_connected: + raise IOError('Not connected') + NetworkThread.network_event_loop.call_soon_threadsafe(queue) + + +class FragmentingPeer(ScriptedPeer): + """Writes its handshake in pieces of 1, 2, 3 and 7 bytes, a millisecond apart.""" + def __init__(self, **overrides): + super().__init__(**overrides) + self.next_write = 0 + + def send_raw_message(self, raw_message_bytes): + def schedule(): + loop = NetworkThread.network_event_loop + pos = 0 + sizes = [1, 2, 3, 7] + while pos < len(raw_message_bytes): + size = sizes[0] if not self.v2_state.tried_v2_handshake else len(raw_message_bytes) + sizes = sizes[1:] + sizes[:1] + self.next_write = max(self.next_write, loop.time()) + 0.001 + piece = raw_message_bytes[pos:pos + size] + loop.call_at(self.next_write, lambda piece=piece: self._transport and self._transport.write(piece)) + pos += size + if not self.is_connected: + raise IOError('Not connected') + NetworkThread.network_event_loop.call_soon_threadsafe(schedule) + + +def ignore_confirmation(state, response): + """For peers whose keys won't match the node's: never complete, never fail on the node's packets.""" + return 0, True + + +def bad_modulus(ek): + """ek with its first 12-bit coefficient set to 4095, which is not below q = 3329.""" + return b"\xff" + bytes([ek[1] | 0x0f]) + ek[2:] + + +def payload(record): + """The payload of an offer or accept record.""" + assert_equal(len(record), PQ_RECORD_LEN) + return record[4:] + + +def padding(contents_len): + """An unknown 0xf1 record that pads contents of contents_len bytes to exactly MAX_CONTENTS_LEN.""" + remaining = MAX_CONTENTS_LEN - contents_len + # CompactSize (5 bytes for these sizes) || header || payload + record = ser_pq_record(bytes(remaining - 6), header=0xF1) + assert_equal(len(record), remaining) + return record + + +def counters(node): + info = node.getpqtransportinfo()["handshakes"] + return {f"{direction}_{key}": value for direction in ("inbound", "outbound") for key, value in info[direction].items()} + + +class P2PV2PQMisbehavingTest(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 1 + self.extra_args = [PQ_ON] + + def run_test(self): + self.protect_outbound_endpoint() + self.test_records() + self.test_confirmation() + self.test_decoys_and_partial_writes() + self.test_unsolicited_accept() + self.test_load_shedding() + + def protect_outbound_endpoint(self): + """Outbound peers here all listen on one endpoint. A confirmed hybrid handshake protects it until + restart, so that the failures below never put it in the fallback set (p2p_v2_pq_fallback.py + tests that).""" + node = self.nodes[0] + peer = node.add_outbound_p2p_connection(P2PInterface(), p2p_idx=0, supports_v2_pq=True) + assert_equal(peer.v2_state.pq_status, "hybrid") + self.disconnect_python_peers() + + def disconnect_python_peers(self): + node = self.nodes[0] + node.disconnect_p2ps() + self.wait_until(lambda: len(node.getpeerinfo()) == 0) + + def expect(self, before, **deltas): + """Wait until the node's counters moved by exactly deltas (missing keys: 0) since before.""" + node = self.nodes[0] + expected = {key: value + deltas.pop(key, 0) for key, value in before.items()} + assert_equal(deltas, {}) # every key named is a real counter + self.wait_until(lambda: counters(node) == expected) + + def run_case(self, *, inbound, peer, kept, status=None, logs, deltas, **connect_args): + """Connect a scripted Python peer and check the outcome. + + inbound: the Python peer initiates (the node responds); otherwise the node initiates. + kept: the connection survives, with status as the node's transport_pq_status; otherwise the + node drops it at once. + logs: exact debug log lines (or prefixes) the case produces.""" + node = self.nodes[0] + before = counters(node) + with node.assert_debug_log(logs): + if inbound: + node.add_p2p_connection(peer, supports_v2_pq=peer.overrides.pop("pq", True), + wait_for_verack=kept, wait_for_v2_handshake=kept, expect_success=kept, + **connect_args) + elif kept: + node.add_outbound_p2p_connection(peer, p2p_idx=0, supports_v2_pq=peer.overrides.pop("pq", True), + **connect_args) + else: + # Not add_outbound_p2p_connection(): it waits to see the connection open, which a + # connection the node drops at once may never be. + listening = threading.Event() + peer.peer_accept_connection( + connect_cb=lambda addr, port: listening.set(), connect_id=1, net=node.chain, + timeout_factor=node.timeout_factor, supports_v2_p2p=True, + supports_v2_pq=peer.overrides.pop("pq", True), reconnect=False)() + node.p2ps.append(peer) + self.wait_until(listening.is_set) + node.addconnection(f"127.0.0.1:{p2p_port(MAX_NODES - 1)}", "outbound-full-relay", True) + if kept: + peer.sync_with_ping() + [info] = node.getpeerinfo() + assert_equal(info["transport_pq_status"], status) + assert_equal(info["transport_pq"], status == "hybrid") + else: + # The node closes at once: -peertimeout is far longer than this wait. + self.wait_until(lambda: peer.v2_state is not None and peer.v2_state.sent_version_contents is not None, + timeout=10) + peer.wait_for_disconnect(timeout=10) + self.disconnect_python_peers() + self.expect(before, **deltas) + + def test_records(self): + self.log.info("Offers of every shape: the node initiates") + for name, contents, kept, logs, deltas in [ + ("ek one byte short", lambda offer: ser_pq_record(payload(offer)[:-1]), False, + ["V2 transport error: malformed hybrid record (ek_length)", + "v2 pq: malformed_record reason=ek_length role=initiator conn_type=outbound-full-relay peer="], + {"outbound_malformed_record": 1}), + ("ek one byte long", lambda offer: ser_pq_record(payload(offer) + b"\x00"), False, + ["v2 pq: malformed_record reason=ek_length role=initiator"], {"outbound_malformed_record": 1}), + ("ek fails the modulus check", lambda offer: ser_pq_record(bad_modulus(payload(offer))), False, + ["V2 transport error: malformed hybrid record (ek_modulus)", + "v2 pq: malformed_record reason=ek_modulus role=initiator"], {"outbound_malformed_record": 1}), + ("a malformed own record before a valid one", lambda offer: ser_pq_record(payload(offer)[:-1]) + offer, False, + ["v2 pq: malformed_record reason=ek_length role=initiator"], {"outbound_malformed_record": 1}), + ("the first of two own records wins", lambda offer: offer + ser_pq_record(b"\x00"), True, + ["v2 pq: switched role=initiator conn_type=outbound-full-relay peer="], {"outbound_switched": 1}), + ("an unknown record, then the offer", lambda offer: ser_pq_record(b"\x01", header=0xF1) + offer, True, + ["v2 pq: switched role=initiator"], {"outbound_switched": 1}), + ("an offer followed by junk", lambda offer: offer + b"\x05", True, + ["v2 pq: legacy_peer reason=parse_error role=initiator conn_type=outbound-full-relay peer="], + {"outbound_legacy_peer": 1}), + ("a zero record length", lambda offer: b"\x00" + offer, True, + ["v2 pq: legacy_peer reason=parse_error role=initiator"], {"outbound_legacy_peer": 1}), + ("a non-canonical record length", lambda offer: b"\xfd\x02\x00\xf0\x00", True, + ["v2 pq: legacy_peer reason=parse_error role=initiator"], {"outbound_legacy_peer": 1}), + ("a record length past the end", lambda offer: b"\xfd\xff\xff" + offer, True, + ["v2 pq: legacy_peer reason=parse_error role=initiator"], {"outbound_legacy_peer": 1}), + ("a record length above MAX_SIZE", lambda offer: b"\xfe\x01\x00\x00\x02" + offer, True, + ["v2 pq: legacy_peer reason=parse_error role=initiator"], {"outbound_legacy_peer": 1}), + ("a truncated record length", lambda offer: offer + b"\xfd\x21", True, + ["v2 pq: legacy_peer reason=parse_error role=initiator"], {"outbound_legacy_peer": 1}), + ("only an unknown record", lambda offer: ser_pq_record(b"\x01\x02", header=0xF1), True, + ["v2 pq: legacy_peer reason=no_features role=initiator"], {"outbound_legacy_peer": 1}), + ("only a reserved 0x00 record", lambda offer: ser_pq_record(b"", header=0x00), True, + ["v2 pq: legacy_peer reason=no_features role=initiator"], {"outbound_legacy_peer": 1}), + ("an offer padded to the largest contents", lambda offer: offer + padding(len(offer)), True, + ["v2 pq: switched role=initiator"], {"outbound_switched": 1}), + ]: + self.log.info(f"- {name}") + peer = ScriptedPeer(make_offer_contents=lambda state, offer, contents=contents: contents(offer)) + status = None if not kept else "hybrid" if "outbound_switched" in deltas else "legacy_peer" + self.run_case(inbound=False, peer=peer, kept=kept, status=status, logs=logs, deltas=deltas) + + self.log.info("- a version packet longer than the largest contents") + def too_long(state, contents): + state.sent_version_contents = contents + return state.peer['send_L'].crypt((MAX_CONTENTS_LEN + 1).to_bytes(3, 'little')) + self.run_case(inbound=False, peer=ScriptedPeer(send_version_packet=too_long), kept=False, + logs=[f"V2 transport error: packet too large ({MAX_CONTENTS_LEN + 1} bytes)"], deltas={}) + self.log.info("- a truncated version packet, then EOF") + self.run_case(inbound=False, peer=TruncatingPeer(keep=PQ_RECORD_LEN // 2), kept=False, + logs=["socket closed, disconnecting peer="], deltas={}) + + self.log.info("Accepts of every shape: the node responds") + for name, contents, kept, logs, deltas in [ + ("ct one byte short", lambda accept: ser_pq_record(payload(accept)[:-1]), False, + ["V2 transport error: malformed hybrid record (ct_length)", + "v2 pq: malformed_record reason=ct_length role=responder conn_type=inbound peer="], + {"inbound_malformed_record": 1}), + ("ct one byte long", lambda accept: ser_pq_record(payload(accept) + b"\x00"), False, + ["v2 pq: malformed_record reason=ct_length role=responder"], {"inbound_malformed_record": 1}), + ("a malformed own record before a valid one", lambda accept: ser_pq_record(b"") + accept, False, + ["v2 pq: malformed_record reason=ct_length role=responder"], {"inbound_malformed_record": 1}), + ("the first of two own records wins", lambda accept: accept + ser_pq_record(bytes(CT_SIZE)), True, + ["v2 pq: switched role=responder conn_type=inbound peer="], {"inbound_switched": 1}), + ("unknown records around the accept", + lambda accept: ser_pq_record(b"", header=0x01) + accept + ser_pq_record(b"\xaa", header=0xF1), True, + ["v2 pq: switched role=responder"], {"inbound_switched": 1}), + ("an accept padded to the largest contents", lambda accept: accept + padding(len(accept)), True, + ["v2 pq: switched role=responder"], {"inbound_switched": 1}), + ]: + self.log.info(f"- {name}") + peer = ScriptedPeer(make_accept_contents=lambda state, accept, contents=contents: contents(accept)) + if not kept: + peer.overrides["receive_confirmation"] = ignore_confirmation + self.run_case(inbound=True, peer=peer, kept=kept, status="hybrid", logs=logs, deltas=deltas) + + # A peer whose contents don't parse, or carry no accept, is a legacy peer: it keeps the ECDH keys. + accept_like = ser_pq_record(os.urandom(CT_SIZE)) + for name, contents, reason in [ + ("an accept followed by junk", accept_like + b"\x05", "parse_error"), + ("a zero record length", b"\x00", "parse_error"), + ("a non-canonical record length", b"\xfd\x02\x00\xf0\x00", "parse_error"), + ("a record length past the end", b"\xfe\x00\x00\x01\x00" + accept_like, "parse_error"), + ("a record length above MAX_SIZE", b"\xff" + (2**32).to_bytes(8, 'little') + accept_like, "parse_error"), + ("a truncated record", accept_like[:-1], "parse_error"), + ("only unknown records", ser_pq_record(b"", header=0xF1) + ser_pq_record(b"\x00" * 300, header=0x7F), + "no_features"), + ("empty contents", b"", "no_features"), + ]: + self.log.info(f"- {name}") + self.run_case(inbound=True, peer=ScriptedPeer(transport_version=contents, pq=False), kept=True, + status="legacy_peer", + logs=[f"v2 pq: legacy_peer reason={reason} role=responder conn_type=inbound peer="], + deltas={"inbound_legacy_peer": 1}) + + self.log.info("- a version packet longer than the largest contents") + self.run_case(inbound=True, peer=ScriptedPeer(send_version_packet=too_long, pq=False), kept=False, + logs=[f"V2 transport error: packet too large ({MAX_CONTENTS_LEN + 1} bytes)", + "v2 pq: abandoned reason=version_length role=responder conn_type=inbound peer="], + deltas={"inbound_abandoned": 1}) + self.log.info("- a version packet with a damaged tag") + def damaged_tag(state, contents): + state.sent_version_contents = contents + packet = state.enc_handshake_packet(contents) + return packet[:-1] + bytes([packet[-1] ^ 0x80]) + self.run_case(inbound=True, peer=ScriptedPeer(send_version_packet=damaged_tag, pq=False), kept=False, + logs=["V2 transport error: packet decryption failure", + "v2 pq: abandoned reason=version_tag role=responder conn_type=inbound peer="], + deltas={"inbound_abandoned": 1}) + self.log.info("- a truncated version packet, then EOF") + self.run_case(inbound=True, peer=TruncatingPeer(keep=10, pq=False), kept=False, + logs=["v2 pq: abandoned reason=eof role=responder conn_type=inbound peer="], + deltas={"inbound_abandoned": 1}) + + def test_confirmation(self): + def damage(kind): + def make_confirmation(state): + if kind == "length": + return state.v2_enc_packet(b"\x00", aad=b"", ignore=True) + if kind == "not_decoy": + return state.v2_enc_packet(b"", aad=b"", ignore=False) + confirmation = state.v2_enc_packet(b"", aad=b"", ignore=True) + return confirmation[:-1] + bytes([confirmation[-1] ^ 1]) + return make_confirmation + + for inbound, role, conn_type in ((True, "responder", "inbound"), (False, "initiator", "outbound-full-relay")): + direction = "inbound" if inbound else "outbound" + for kind in ("length", "tag", "not_decoy"): + self.log.info(f"A confirmation that fails the {kind} check: the node is the {role}") + self.run_case(inbound=inbound, peer=ScriptedPeer(make_confirmation=damage(kind)), kept=False, + logs=[f"V2 transport error: key confirmation failed ({kind})", + f"v2 pq: first_packet_failed reason={kind} role={role} conn_type={conn_type} peer="], + deltas={f"{direction}_switched": 1, f"{direction}_first_packet_failed": 1}) + + # Keys that don't match decrypt the confirmation's length to a random value, which is not 0. + self.log.info("A corrupted ct: the node decapsulates another secret and fails the confirmation at once") + def corrupt_ct(state, accept): + return accept[:-1] + bytes([accept[-1] ^ 1]) + self.run_case(inbound=True, peer=ScriptedPeer(make_accept_contents=corrupt_ct, + receive_confirmation=ignore_confirmation), kept=False, + logs=["V2 transport error: key confirmation failed (length)", + "v2 pq: first_packet_failed reason=length role=responder"], + deltas={"inbound_switched": 1, "inbound_first_packet_failed": 1}) + self.log.info("A responder that derives another shared secret: the node fails the confirmation at once") + def wrong_secret(state): + return bytes([state.mlkem_shared_secret[0] ^ 0xff]) + state.mlkem_shared_secret[1:] + self.run_case(inbound=False, peer=ScriptedPeer(hybrid_shared_secret=wrong_secret, + receive_confirmation=ignore_confirmation), kept=False, + logs=["V2 transport error: key confirmation failed (length)", + "v2 pq: first_packet_failed reason=length role=initiator"], + deltas={"outbound_switched": 1, "outbound_first_packet_failed": 1}) + + def test_decoys_and_partial_writes(self): + def decoys(state): + # The first one authenticates the garbage (AAD), the later ones and the version packet don't. + return b"".join(state.enc_handshake_packet(bytes(size), ignore=True) for size in (1, 100, 1000)) + + def confirmation_then_decoys(state): + confirmation = state.v2_enc_packet(b"", aad=b"", ignore=True) + return confirmation + b"".join(state.v2_enc_packet(bytes(size), ignore=True) for size in (0, 7, 500)) + + for peer_class, overrides, what in [ + (ScriptedPeer, {"make_decoys": decoys}, "decoys before the version packet (ECDH keys)"), + (ScriptedPeer, {"make_confirmation": confirmation_then_decoys}, + "decoys after the confirmation (hybrid keys)"), + (ScriptedPeer, {"make_decoys": decoys, "make_confirmation": confirmation_then_decoys}, "both"), + (CoalescingPeer, {"make_decoys": decoys}, "each burst coalesced into one write"), + (FragmentingPeer, {"make_decoys": decoys}, "the handshake fragmented into 1- to 7-byte writes"), + ]: + for inbound, role in ((True, "responder"), (False, "initiator")): + self.log.info(f"The node as {role}: {what}") + direction = "inbound" if inbound else "outbound" + self.run_case(inbound=inbound, peer=peer_class(**overrides), kept=True, status="hybrid", + logs=[f"v2 pq: switched role={role}"], deltas={f"{direction}_switched": 1}) + + def test_unsolicited_accept(self): + node = self.nodes[0] + self.log.info("A switched-off responder ignores an accept it never asked for: no decapsulation, no switch") + self.restart_node(0, extra_args=PQ_OFF) + before = counters(node) + with node.assert_debug_log([], unexpected_msgs=["v2 pq:", "malformed hybrid record"]): + peer = node.add_p2p_connection(ScriptedPeer(transport_version=ser_pq_record(os.urandom(CT_SIZE)))) + [info] = node.getpeerinfo() + assert_equal((info["transport_pq"], info["transport_pq_status"]), (False, "off")) + assert_equal(info["session_id"], peer.v2_state.peer["session_id"].hex()) + assert_equal(peer.v2_state.received_version_contents, b"") + self.disconnect_python_peers() + assert_equal(counters(node), before) + self.restart_node(0, extra_args=PQ_ON) + self.protect_outbound_endpoint() + + def flood_offers(self, count): + """Open count inbound v2 connections, several at a time, that each send just a 64-byte public key (the + point at which a responder decides whether to offer), then close and wait for the node to close too. + Returns when each connection ended (time.monotonic()).""" + node = self.nodes[0] + magic = MAGIC_BYTES[node.chain] + + def one_offer(_): + key = bytes([magic[0] ^ 0xff]) + random.randbytes(63) + with socket.create_connection(("127.0.0.1", p2p_port(node.index))) as sock: + sock.sendall(key) + sock.shutdown(socket.SHUT_WR) + while sock.recv(65536): + pass + return time.monotonic() + with ThreadPoolExecutor(max_workers=FLOOD_THREADS) as pool: + return list(pool.map(one_offer, range(count))) + + def test_load_shedding(self): + node = self.nodes[0] + self.log.info(f"More than {SHED_THRESHOLD} offers within one second start load shedding") + # The node counts offers per second of its steady clock, which setmocktime doesn't move: the flood + # has to beat the threshold in real time. Flood in short bursts until the node sheds. + before = counters(node) + ended = [] + deadline = time.monotonic() + FLOOD_SECONDS * self.options.timeout_factor + with node.assert_debug_log([f"v2 pq: load_shedding started threshold_per_s={SHED_THRESHOLD}"]): + while not node.getpqtransportinfo()["load_shedding"]["active"] and time.monotonic() < deadline: + ended += self.flood_offers(SHED_THRESHOLD // 2) + if not node.getpqtransportinfo()["load_shedding"]["active"]: + # Any 1 s window holding more than twice the threshold puts more than the threshold in one + # of the node's seconds; a slower machine can't tell us anything. + ended.sort() + peak = max(bisect.bisect_left(ended, t + 1) - i for i, t in enumerate(ended)) + assert peak <= 2 * SHED_THRESHOLD, f"no shedding although {peak} offers ended within one second" + self.log.warning(f"Skipping the load shedding test: at most {peak} offers ended within one second here") + return + # Each connection was one attempt, offered (then abandoned at EOF) or shed; the node offered the + # first SHED_THRESHOLD of the second it started shedding in. + self.wait_until(lambda: (counters(node)["inbound_abandoned"] + counters(node)["inbound_shed"] + - before["inbound_abandoned"] - before["inbound_shed"]) == len(ended)) + self.wait_until(lambda: len(node.getpeerinfo()) == 0) + flood = counters(node) + assert flood["inbound_abandoned"] - before["inbound_abandoned"] >= SHED_THRESHOLD + assert flood["inbound_shed"] - before["inbound_shed"] >= 1 + assert_equal({key: value for key, value in flood.items() if key not in ("inbound_abandoned", "inbound_shed")}, + {key: value for key, value in before.items() if key not in ("inbound_abandoned", "inbound_shed")}) + shedding = node.getpqtransportinfo()["load_shedding"] + assert_equal((shedding["active"], shedding["threshold_per_s"]), (True, SHED_THRESHOLD)) + assert_approx(shedding["since"], time.time(), vspan=60) + + self.log.info("A shedding responder sends no offer: a hybrid initiator sees a legacy responder") + peer = node.add_p2p_connection(P2PInterface(), supports_v2_pq=True) + assert_equal((peer.v2_state.pq_status, peer.v2_state.received_version_contents), ("legacy_peer", b"")) + [info] = node.getpeerinfo() + assert_equal((info["transport_pq"], info["transport_pq_status"]), (False, "off")) + self.disconnect_python_peers() + self.log.info("A shedding responder ignores an accept: no decapsulation, no switch") + with node.assert_debug_log([], unexpected_msgs=["v2 pq: switched", "malformed hybrid record"]): + peer = node.add_p2p_connection(ScriptedPeer(transport_version=ser_pq_record(os.urandom(CT_SIZE)))) + assert_equal(node.getpeerinfo()[0]["transport_pq_status"], "off") + assert_equal(node.getpeerinfo()[0]["session_id"], peer.v2_state.peer["session_id"].hex()) + self.disconnect_python_peers() + self.log.info("Outbound connections still negotiate while inbound offers are shed") + node.add_outbound_p2p_connection(P2PInterface(), p2p_idx=0, supports_v2_pq=True) + assert_equal(node.getpeerinfo()[0]["transport_pq_status"], "hybrid") + self.disconnect_python_peers() + self.expect(flood, inbound_shed=2, outbound_switched=1) + assert node.getpqtransportinfo()["load_shedding"]["active"] + + self.log.info(f"With no offers at all, the socket handler ends shedding after {SHED_QUIET_SECONDS} quiet seconds") + # Nothing connects from here on: only the socket handler's own clock check can end it. + self.wait_until(lambda: not node.getpqtransportinfo()["load_shedding"]["active"], + timeout=SHED_QUIET_SECONDS + 30) + assert_equal(node.getpqtransportinfo()["load_shedding"], + {"active": False, "threshold_per_s": SHED_THRESHOLD, "since": 0}) + with open(node.debug_log_path, encoding="utf-8") as log: + assert_equal(sum("v2 pq: load_shedding stopped" in line for line in log), 1) + self.log.info("Offers resume") + peer = node.add_p2p_connection(P2PInterface(), supports_v2_pq=True) + assert_equal((peer.v2_state.pq_status, len(peer.v2_state.received_version_contents)), ("hybrid", PQ_RECORD_LEN)) + self.disconnect_python_peers() + +if __name__ == '__main__': + P2PV2PQMisbehavingTest(__file__).main() diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py index 14bb2da048..080d22b8ef 100755 --- a/test/functional/test_runner.py +++ b/test/functional/test_runner.py @@ -274,6 +274,8 @@ 'p2p_v2_encrypted.py', 'p2p_v2_misbehaving.py', 'p2p_v2_pq.py', + 'p2p_v2_pq_misbehaving.py', + 'p2p_v2_pq_fallback.py', 'example_test.py', 'feature_minchainwork_requested_block.py', 'mempool_truc.py',