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/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_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..080d22b8ef 100755 --- a/test/functional/test_runner.py +++ b/test/functional/test_runner.py @@ -273,6 +273,9 @@ 'p2p_v2_transport.py', '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',