test: end-to-end hybrid transport tests, Python hybrid peer, misbehaving peers and fallback (#184 part 7)

.github/workflows/ci-pq.yml

@@ -15,8 +15,9 @@
 #   says needs them.
 # The workflow_dispatch trigger below stays for when this file is on main itself;
 # until then GitHub rejects gh workflow run ci-pq.yml.
-# Jobs: aarch64-unit, s390x-unit (big-endian, emulated), macos-arm64-unit and
-# macos-x86_64-unit.
+# Jobs: aarch64-unit, s390x-unit (big-endian, emulated), macos-arm64-unit,
+# macos-x86_64-unit and pq-functional (the whole functional suite with
+# --v2pqtransport).
 # The source is resolved once to a commit, every job checks out and verifies
 # that commit, and the summaries record it.
 
@@ -32,7 +33,7 @@ on:
         type: string
         default: ''
       jobs:
-        description: 'Comma-separated jobs to run (aarch64-unit, s390x-unit, macos-arm64-unit, macos-x86_64-unit); empty runs all'
+        description: 'Comma-separated jobs to run (aarch64-unit, s390x-unit, macos-arm64-unit, macos-x86_64-unit, pq-functional); empty runs all'
         required: false
         type: string
         default: ''
@@ -44,7 +45,7 @@ on:
         type: string
         default: ''
       jobs:
-        description: 'Comma-separated jobs to run (aarch64-unit, s390x-unit, macos-arm64-unit, macos-x86_64-unit); empty runs all'
+        description: 'Comma-separated jobs to run (aarch64-unit, s390x-unit, macos-arm64-unit, macos-x86_64-unit, pq-functional); empty runs all'
         required: false
         type: string
         default: ''
@@ -66,6 +67,7 @@ jobs:
       selected: ${{ steps.select.outputs.selected }}
       run_macos_arm64_unit: ${{ steps.select.outputs.run_macos_arm64_unit }}
       run_macos_x86_64_unit: ${{ steps.select.outputs.run_macos_x86_64_unit }}
+      run_pq_functional: ${{ steps.select.outputs.run_pq_functional }}
     steps:
       - name: Resolve source
         id: resolve
@@ -190,7 +192,7 @@ jobs:
               }
             ]
           # Selectable jobs that are not matrix entries; each gets a run_<job> output.
-          NON_MATRIX_JOBS: 'macos-arm64-unit macos-x86_64-unit'
+          NON_MATRIX_JOBS: 'macos-arm64-unit macos-x86_64-unit pq-functional'
         run: |
           # Turn the comma-separated jobs input into the matrix: empty selects
           # every job; an empty, unknown or repeated name fails the run.
@@ -547,3 +549,52 @@ jobs:
       - *REQUIRE_DEPENDS_BUILD
       - *REQUIRE_NATIVE_MLKEM
       - *MACOS_REPORT_EVIDENCE
+
+  # The whole functional suite with hybrid post-quantum v2 transport between all
+  # nodes (--v2pqtransport, #184 part 7), so the feature can't quietly break
+  # unrelated behavior. It shares unit's source checks; the test runner's result
+  # is the job's.
+  pq-functional:
+    name: Functional tests with hybrid transport
+    needs: resolve-source
+    if: ${{ needs.resolve-source.outputs.run_pq_functional == 'true' }}
+    runs-on: ubuntu-24.04
+    # Twice a measured local run, rounded up: a cold build at -j 4 on 4 cores
+    # (475 s) plus the whole suite at -j 4 (727 s) is 1,202 s; x 2 = 40.1 min.
+    timeout-minutes: 45
+    env:
+      CI_FAILFAST_TEST_LEAVE_DANGLING: 1
+      DANGER_CI_ON_HOST_FOLDERS: 1
+      FILE_ENV: ./ci/test/00_setup_env_native_pq_functional.sh
+      MATRIX_NAME: Functional tests with hybrid transport
+      TEST_RUNNER_FAILFAST: 'false'
+      MAKEJOBS: '-j4'
+      TEST_RUNNER_JOBS: '4'
+      CTEST_JOBS: '4'
+      CI_PROFILE: public
+      CI_IMAGE_REGISTRY_PREFIX: docker.io/library
+      CI_ENFORCE_INTERNAL_REGISTRY: '0'
+      CI_BUILDKIT_DNS_NAMESERVERS: '1.1.1.1,8.8.8.8'
+      CI_BUILDKIT_DNS_SEARCH_DOMAINS: ''
+    steps:
+      - *REQUIRE_SOURCE
+      - *CHECKOUT_SOURCE
+      - *VERIFY_SOURCE
+
+      - name: Configure environment
+        uses: ./.github/actions/configure-environment
+
+      - name: Restore caches
+        id: restore-cache
+        uses: ./.github/actions/restore-caches
+
+      - name: Configure Docker
+        uses: ./.github/actions/configure-docker
+        with:
+          cache-provider: gha
+
+      - name: CI script
+        run: ./ci/test_run_all.sh
+
+      - name: Save caches
+        uses: ./.github/actions/save-caches

ci/checks/test_ci_pq_contract.py

@@ -83,8 +83,10 @@
     MACOS_JOB: ("macos-15", "arm64", "aarch64", "native AArch64"),
     MACOS_X86_64_JOB: ("macos-15-intel", "x86_64", "x86_64", "native x86_64"),
 }
+PQ_FUNCTIONAL_JOB = "pq-functional"
 PQ_MATRIX_JOBS = [AARCH64_JOB, S390X_JOB]
-ALL_PQ_JOBS = {PQ_UNIT_JOB: PQ_MATRIX_JOBS, MACOS_JOB: [], MACOS_X86_64_JOB: []}
+ALL_PQ_JOBS = {PQ_UNIT_JOB: PQ_MATRIX_JOBS, MACOS_JOB: [], MACOS_X86_64_JOB: [], PQ_FUNCTIONAL_JOB: []}
+I686_ENV = "ci/test/00_setup_env_i686_no_ipc.sh"
 PQ_SUITES = ["mlkem_tests", "bip324_tests", "net_tests"]
 AARCH64_SUITES = PQ_SUITES
 NIGHTLY_MATRIX_JOBS = ["arm32-unit-1", "arm32-unit-2", "previous-releases", "fuzz"]
@@ -504,6 +506,7 @@ def test_named_dispatch_runs_only_selected_jobs(self) -> None:
             # What Full Validation's pq-unit job passes.
             (PQ, {"jobs": f"{AARCH64_JOB},{MACOS_JOB},{MACOS_X86_64_JOB}"},
              {PQ_UNIT_JOB: [AARCH64_JOB], MACOS_JOB: [], MACOS_X86_64_JOB: []}),
+            (PQ, {"jobs": PQ_FUNCTIONAL_JOB}, {PQ_FUNCTIONAL_JOB: []}),
             (PQ, {"jobs": ""}, ALL_PQ_JOBS),
             (NIGHTLY, {"run_test_matrix": True, "run_scanners": False, "jobs": "previous-releases"},
              {"nightly-matrix": ["previous-releases"]}),
@@ -779,7 +782,7 @@ def group(context: dict[str, Any]) -> str:
     def test_rejects_unknown_repeated_and_empty_job_names(self) -> None:
         bad = {
             PQ: ["aarch64", "AARCH64-UNIT", f"{AARCH64_JOB},{AARCH64_JOB}", f"{AARCH64_JOB},", ",", "s390x", "macos-arm64",
-                 f"{MACOS_JOB},{MACOS_JOB}", "macos-x86_64", "macos-intel"],
+                 f"{MACOS_JOB},{MACOS_JOB}", "macos-x86_64", "macos-intel", "pq_functional", "functional"],
             NIGHTLY: ["previous_releases", "previous-releases,previous-releases", "previous-releases,,fuzz", "unit"],
         }
         for key, values in bad.items():
@@ -835,11 +838,11 @@ def test_every_pq_job_checks_out_the_resolved_commit(self) -> None:
         context = self.resolve(PQ, self.dispatch(PQ, source_ref=svc.MAINTAINED_BRANCH, jobs=AARCH64_JOB))
         refs = svc.check_source_wiring(workflow, context)
         self.assertEqual(refs, {job: fixture.maintained_sha for job in ALL_PQ_JOBS})
-        # The macOS jobs share the unit job's guard, checkout and verification steps.
-        for macos_job in MACOS_JOBS:
+        # The macOS and functional jobs share the unit job's guard, checkout and verification steps.
+        for job_id in (*MACOS_JOBS, PQ_FUNCTIONAL_JOB):
             for step_name in (svc.GUARD_STEP, svc.CHECKOUT_STEP, svc.VERIFY_STEP):
-                self.assertEqual(svc.find_step(workflow["jobs"][macos_job], step_name),
-                                 svc.find_step(workflow["jobs"][PQ_UNIT_JOB], step_name), (macos_job, step_name))
+                self.assertEqual(svc.find_step(workflow["jobs"][job_id], step_name),
+                                 svc.find_step(workflow["jobs"][PQ_UNIT_JOB], step_name), (job_id, step_name))
 
         mutated = copy.deepcopy(workflow)
         checkout = svc.find_step(mutated["jobs"][PQ_UNIT_JOB], svc.CHECKOUT_STEP)
@@ -1191,6 +1194,38 @@ def check_macos_depends(self, workflow: dict[str, Any], job_id: str, runner_host
                     self.assertIn("::error::", result.stderr)
                     self.assertIn(error.format(prefix=prefix), result.stderr)
 
+    def test_functional_job_runs_the_whole_suite_with_hybrid_transport(self) -> None:
+        workflow = self.workflows[PQ]
+        job = workflow["jobs"][PQ_FUNCTIONAL_JOB]
+        self.assertNotIn(PQ_FUNCTIONAL_JOB, [entry["job"] for entry in catalog(workflow)])
+        self.assertEqual(job["runs-on"], "ubuntu-24.04")
+        context = self.resolve(PQ, self.dispatch(PQ, jobs=PQ_FUNCTIONAL_JOB))
+        env = svc.job_env(workflow, job, context)
+        self.assertEqual(env["MATRIX_NAME"], job["name"])
+        self.assertEqual(env["CI_IMAGE_REGISTRY_PREFIX"], "docker.io/library")
+        self.assertEqual(env["CI_ENFORCE_INTERNAL_REGISTRY"], "0")
+        self.assertEqual(env["CI_PROFILE"], "public")
+        env_file = env["FILE_ENV"]
+        self.assertTrue(os.access(REPO_ROOT / env_file, os.X_OK), env_file)
+        self.assertLess(svc.step_index(job, svc.VERIFY_STEP), svc.step_index(job, "CI script"))
+        self.assertEqual(svc.find_step(job, "CI script")["run"], "./ci/test_run_all.sh")
+
+        names = ["RUN_UNIT_TESTS", "RUN_FUNCTIONAL_TESTS", "TEST_RUNNER_EXTRA", "CI_BUILD_TARGET", "NO_DEPENDS", "BITCOIN_CONFIG",
+                 "CI_IMAGE_NAME_TAG"]
+        effective = source_env(env_file, names, {key: env[key] for key in ("CI_IMAGE_REGISTRY_PREFIX", "CI_ENFORCE_INTERNAL_REGISTRY")})
+        # The whole suite (no test selection), with hybrid transport between all nodes.
+        self.assertEqual(effective["TEST_RUNNER_EXTRA"].split(), ["--v2pqtransport"])
+        self.assertEqual((effective["RUN_FUNCTIONAL_TESTS"], effective["RUN_UNIT_TESTS"]), ("true", "false"))
+        self.assertEqual(effective["CI_BUILD_TARGET"], "", "every binary the suite runs is built")
+        self.assertIn("-DWITH_ZMQ=ON", effective["BITCOIN_CONFIG"].split())
+        self.assertEqual(effective["CI_IMAGE_NAME_TAG"], "docker.io/library/ubuntu:24.04")
+
+        # The i686 job of ci.yml keeps covering plain v2: --v2transport without --v2pqtransport, for which
+        # the test framework passes -v2pqtransport=0 explicitly.
+        extra = source_env(I686_ENV, ["TEST_RUNNER_EXTRA"])["TEST_RUNNER_EXTRA"].split()
+        self.assertIn("--v2transport", extra)
+        self.assertNotIn("--v2pqtransport", extra)
+
     def test_contexts_are_available_where_used(self) -> None:
         """Mirror actionlint's context-availability rule for the places these workflows use expressions.
 

ci/test/00_setup_env_i686_no_ipc.sh

@@ -17,6 +17,8 @@ export PACKAGES="llvm clang g++-multilib"
 export DEP_OPTS="DEBUG=1 NO_IPC=1"
 export GOAL="install"
 export CI_LIMIT_STACK_SIZE=1
+# Plain v2: without --v2pqtransport, the test framework starts every node with
+# -v2pqtransport=0, so this job keeps covering plain v2 whatever the default.
 export TEST_RUNNER_EXTRA="--v2transport --usecli"
 export BITCOIN_CONFIG="\
  -DCMAKE_BUILD_TYPE=Debug \

ci/test/00_setup_env_native_pq_functional.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+#
+# Copyright (c) 2026-present The qbit core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+export LC_ALL=C.UTF-8
+
+# shellcheck source=ci/test/00_setup_env_base_image.sh
+source "$( dirname "${BASH_SOURCE[0]}" )/00_setup_env_base_image.sh"
+
+# The pq-functional job of .github/workflows/ci-pq.yml: the whole functional
+# suite with hybrid post-quantum v2 transport between all nodes. Framework Python
+# peers stay legacy BIP324 peers, so the suite pays no pure-Python KEM cost;
+# p2p_v2_pq*.py exercise the Python hybrid peer in every run.
+export CONTAINER_NAME=ci_native_pq_functional
+ci_set_base_image_name_tag "ubuntu:24.04"
+export PACKAGES="python3-zmq libevent-dev libboost-dev libzmq3-dev libsqlite3-dev"
+export NO_DEPENDS=1
+export GOAL="install"
+export RUN_UNIT_TESTS=false
+export RUN_FUNCTIONAL_TESTS=true
+export TEST_RUNNER_EXTRA="--v2pqtransport"
+export BITCOIN_CONFIG="\
+ -DWITH_ZMQ=ON -DBUILD_GUI=OFF -DENABLE_IPC=OFF -DREDUCE_EXPORTS=ON \
+"

src/common/args.cpp

@@ -747,6 +747,8 @@ const std::vector<std::string> TEST_OPTIONS_DOC{
     "disable_witness_pruning (disable witness pruning in regtest)",
     "pq_fail_first_packet (corrupt this node's ML-KEM shared secret on every hybrid post-quantum v2 handshake, so the peer's key confirmation fails)",
     "mlkem_portable (use the portable ML-KEM-1024 code instead of native code)",
+    "pq_low_shed_threshold (start shedding inbound hybrid post-quantum v2 offers above 5 attempts per second instead of 1000)",
+    "mock_steady_clock (setmocktime also sets the steady clock that durations such as load shedding's seconds are measured on; setmocktime 0 clears both)",
 };
 
 bool HasTestOption(const ArgsManager& args, const std::string& test_option)

src/init.cpp

@@ -2207,6 +2207,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     connOptions.whitelist_forcerelay = args.GetBoolArg("-whitelistforcerelay", DEFAULT_WHITELISTFORCERELAY);
     connOptions.whitelist_relay = args.GetBoolArg("-whitelistrelay", DEFAULT_WHITELISTRELAY);
     connOptions.m_pq = g_pq_transport_config;
+    if (HasTestOption(args, "pq_low_shed_threshold")) connOptions.pq_shed_threshold_per_s = TEST_PQ_SHED_THRESHOLD_PER_S;
 
     // Port to bind to if `-bind=addr` is provided without a `:port` suffix.
     const uint16_t default_bind_port =

src/net.h

@@ -902,6 +902,8 @@ inline constexpr size_t PQ_LOCAL_FAULT_THRESHOLD{8};
 /** Inbound offers attempted within one second above which responders stop offering. R3's flood
  *  measurement (#222) sets the final value; CConnman::Options can override it. */
 inline constexpr uint64_t DEFAULT_PQ_SHED_THRESHOLD_PER_S{1000};
+/** The threshold of -test=pq_low_shed_threshold, which functional tests reach with a few connections. */
+inline constexpr uint64_t TEST_PQ_SHED_THRESHOLD_PER_S{5};
 /** Consecutive seconds with fewer than half the threshold's attempts that end load shedding. */
 inline constexpr std::chrono::seconds PQ_SHED_QUIET_PERIOD{10};
 

src/rpc/node.cpp

@@ -6,6 +6,7 @@
 #include <bitcoin-build-config.h> // IWYU pragma: keep
 
 #include <chainparams.h>
+#include <common/args.h>
 #include <httpserver.h>
 #include <index/blockfilterindex.h>
 #include <index/coinstatsindex.h>
@@ -64,6 +65,13 @@ static RPCHelpMan setmocktime()
     }
 
     SetMockTime(time);
+    if (HasTestOption(EnsureAnyArgsman(request.context), "mock_steady_clock")) {
+        if (time) {
+            MockableSteadyClock::SetMockTime(std::chrono::seconds{time});
+        } else {
+            MockableSteadyClock::ClearMockTime();
+        }
+    }
     const NodeContext& node_context{EnsureAnyNodeContext(request.context)};
     for (const auto& chain_client : node_context.chain_clients) {
         chain_client->setMockTime(time);

test/functional/feature_framework_unit_tests.py

@@ -32,6 +32,7 @@
     "script",
     "script_util",
     "segwit_addr",
+    "v2_p2p",
     "wallet_util",
 ]
 

test/functional/p2p_v2_misbehaving.py

@@ -14,6 +14,8 @@
 from test_framework.v2_p2p import (
     EncryptedP2PState,
     MAX_GARBAGE_LEN,
+    ParseKind,
+    parse_version_contents,
 )
 
 
@@ -27,13 +29,16 @@ class TestType(Enum):
     4. WRONG_GARBAGE - Disconnection happens when garbage bytes that is sent is different from what the peer receives
     5. SEND_NO_AAD - Disconnection happens when AAD of first encrypted packet after the garbage terminator is not filled
     6. SEND_NON_EMPTY_VERSION_PACKET - non-empty version packet is simply ignored
+    7. SEND_OWN_RECORD_VERSION_PACKET - a version packet whose contents parse as a malformed hybrid accept
+    (04 f0 00 00 00) disconnects when the node offered hybrid keys, and is ignored otherwise
     """
     EARLY_KEY_RESPONSE = 0
     EXCESS_GARBAGE = 1
     WRONG_GARBAGE_TERMINATOR = 2
     WRONG_GARBAGE = 3
     SEND_NO_AAD = 4
     SEND_NON_EMPTY_VERSION_PACKET = 5
+    SEND_OWN_RECORD_VERSION_PACKET = 6
 
 
 class EarlyKeyResponseState(EncryptedP2PState):
@@ -91,7 +96,18 @@ def complete_handshake(self, response):
 class NonEmptyVersionPacketState(EncryptedP2PState):
     """"Add option for sending non-empty transport version packet."""
     def complete_handshake(self, response):
+        # Redraw contents that parse as a hybrid accept: a node that offered would reject them as
+        # malformed (SEND_OWN_RECORD_VERSION_PACKET tests that).
         self.transport_version = random.randbytes(5)
+        while parse_version_contents(self.transport_version)[0] == ParseKind.OWN_RECORD:
+            self.transport_version = random.randbytes(5)
+        return super().complete_handshake(response)
+
+
+class OwnRecordVersionPacketState(EncryptedP2PState):
+    """Send version packet contents that parse as a hybrid accept with a 3-byte ciphertext."""
+    def complete_handshake(self, response):
+        self.transport_version = bytes.fromhex("04f0000000")
         return super().complete_handshake(response)
 
 
@@ -112,8 +128,10 @@ def connection_made(self, transport):
             self.v2_state = WrongGarbageState(initiating=True, net='regtest')
         elif self.test_type == TestType.SEND_NO_AAD:
             self.v2_state = NoAADState(initiating=True, net='regtest')
-        elif TestType.SEND_NON_EMPTY_VERSION_PACKET:
+        elif self.test_type == TestType.SEND_NON_EMPTY_VERSION_PACKET:
             self.v2_state = NonEmptyVersionPacketState(initiating=True, net='regtest')
+        elif self.test_type == TestType.SEND_OWN_RECORD_VERSION_PACKET:
+            self.v2_state = OwnRecordVersionPacketState(initiating=True, net='regtest')
         super().connection_made(transport)
 
     def data_received(self, t):
@@ -166,13 +184,23 @@ def test_v2disconnection(self):
             ["V2 transport error: packet decryption failure"],  # WRONG_GARBAGE
             ["V2 transport error: packet decryption failure"],  # SEND_NO_AAD
             [],  # SEND_NON_EMPTY_VERSION_PACKET
+            ["V2 transport error: malformed hybrid record (ct_length)"],  # SEND_OWN_RECORD_VERSION_PACKET
         ]
+        # Whether the node offers hybrid keys depends on the run mode (--v2pqtransport) and on its default.
+        pq_enabled = node0.getpqtransportinfo()["enabled"]
         for test_type in TestType:
             if test_type == TestType.EARLY_KEY_RESPONSE:
                 continue
-            elif test_type == TestType.SEND_NON_EMPTY_VERSION_PACKET:
+            elif test_type == TestType.SEND_NON_EMPTY_VERSION_PACKET or (
+                    test_type == TestType.SEND_OWN_RECORD_VERSION_PACKET and not pq_enabled):
                 node0.add_p2p_connection(MisbehavingV2Peer(test_type), wait_for_verack=True, send_version=True, supports_v2_p2p=True)
                 self.log.info(f"No disconnection for {test_type.name}")
+            elif test_type == TestType.SEND_OWN_RECORD_VERSION_PACKET:
+                with node0.assert_debug_log(expected_debug_message[test_type.value]):
+                    peer1 = node0.add_p2p_connection(MisbehavingV2Peer(test_type), wait_for_verack=False, send_version=False, supports_v2_p2p=True, expect_success=False)
+                    # At once: the node offered, and the accept is malformed.
+                    peer1.wait_for_disconnect()
+                self.log.info(f"Expected disconnection for {test_type.name}")
             else:
                 with node0.assert_debug_log(expected_debug_message[test_type.value], timeout=5):
                     node0.setmocktime(int(time.time()))