From efb4e2e3e4df89f3834ee11eccf9f174b529498c Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Thu, 16 Jul 2026 10:16:53 +0700 Subject: [PATCH] test(export): cover share stream and signer recovery --- .testnet/scenarios/export-suite.yml | 6 + .../export/export_committee_recovery.py | 126 ++++++++++++ .testnet/scenarios/export/export_helpers.py | 77 ++++++- .../export/export_share_subscription.py | 193 ++++++++++++++++++ 4 files changed, 396 insertions(+), 6 deletions(-) create mode 100644 .testnet/scenarios/export/export_committee_recovery.py create mode 100644 .testnet/scenarios/export/export_share_subscription.py diff --git a/.testnet/scenarios/export-suite.yml b/.testnet/scenarios/export-suite.yml index 159b8e00c..1fc6b9631 100644 --- a/.testnet/scenarios/export-suite.yml +++ b/.testnet/scenarios/export-suite.yml @@ -53,6 +53,12 @@ tests: - rng_poll_ms=333 - n4:no_export_sig_hash=true + - name: export_share_subscription + script: .testnet/scenarios/export/export_share_subscription.py + + - name: export_two_member_committee_recovery + script: .testnet/scenarios/export/export_committee_recovery.py + # CE + Export: 1 node suppressed, 4/5 = 80% quorum, should succeed - name: export_ce_one_node_down script: .testnet/scenarios/export/export_quorum.py diff --git a/.testnet/scenarios/export/export_committee_recovery.py b/.testnet/scenarios/export/export_committee_recovery.py new file mode 100644 index 000000000..bb1f958d1 --- /dev/null +++ b/.testnet/scenarios/export/export_committee_recovery.py @@ -0,0 +1,126 @@ +""":descr: a 2-of-5 Export committee recovers without a new intent + +The intent selects only validators n0 and n4, so qC is 2. Validator n4 is +stopped before admission: the network still validates the intent with 4/5, but +one selected share cannot form a witness. Restarting n4 must republish its share +for the same live latch and complete the witness without resubmitting Export. +""" + +from __future__ import annotations + +from export_helpers import ( + EXPORT_RETRY_LEDGER_WINDOW, + assert_export_latch, + bitmap_positions, + find_export_signature_witness, + find_export_txns, + require_export, + submit_direct_export, + wait_for_export_signature_witness, +) + + +async def scenario(ctx, log): + await require_export(ctx, log) + await ctx.fund_accounts({"alice": 10000, "bob": 1000}) + alice = ctx.account("alice") + bob = ctx.account("bob") + + if not ctx.stop_node(4): + raise AssertionError("Failed to stop selected validator n4") + await ctx.wait_for_nodes_down(nodes=[4], timeout=30) + log("Stopped selected validator n4; selected committee is n0+n4") + + current = ctx.validated_ledger_index(0) + result = await submit_direct_export( + ctx, + log, + { + "TransactionType": "Export", + "Fee": "1000000", + "ExportedTxn": { + "TransactionType": "Payment", + "Account": alice.address, + "Destination": bob.address, + "Amount": "1000000", + "Fee": "10", + "Sequence": 0, + "TicketSequence": 1, + "FirstLedgerSequence": current + 1, + "LastLedgerSequence": current + EXPORT_RETRY_LEDGER_WINDOW, + "Flags": 2147483648, + "SigningPubKey": "", + }, + }, + alice.wallet, + committee_node_ids=[0, 4], + ) + if result.get("engine_result") != "tesSUCCESS": + raise AssertionError(f"Export intent failed: {result}") + + origin = result.get("hash") + origin_seq = int(result.get("ledger_index")) + if not origin: + raise AssertionError(f"Validated Export missing hash: {result}") + + latches = assert_export_latch( + ctx, + alice.address, + log, + origin_hash=origin, + expect_witness=False, + ) + selected = bitmap_positions(latches[0]["ExportCommittee"]) + if len(selected) != 2: + raise AssertionError(f"Expected a 2-member committee, got {selected}") + + await ctx.wait_for_ledger(origin_seq + 1, node_id=0, timeout=30) + if find_export_signature_witness(ctx, origin_seq + 1, origin): + raise AssertionError("Witness formed while one of two selected signers was down") + assert_export_latch( + ctx, + alice.address, + log, + origin_hash=origin, + expect_witness=False, + ) + log("No witness with only one selected signer; original latch remains pending") + + if not ctx.start_node(4): + raise AssertionError("Failed to restart selected validator n4") + await ctx.wait_for_nodes( + lambda node_id: ctx.rpc.server_info(node_id) is not None, + nodes=[4], + timeout=30, + poll_interval=0.5, + name="selected-export-validator-up", + ) + log("Restarted selected validator n4") + + witness = await wait_for_export_signature_witness( + ctx, log, origin, after_ledger=origin_seq + ) + contributors = bitmap_positions(witness["EntropyContributors"]) + if contributors != selected: + raise AssertionError( + f"Recovered witness contributors {contributors} != selected {selected}" + ) + if len(witness["_WitnessSigners"]) != 2: + raise AssertionError( + f"Recovered witness has {len(witness['_WitnessSigners'])} signers, need 2" + ) + witness_seq = int(witness["LedgerSequence"]) + for seq in range(origin_seq + 1, witness_seq + 1): + if any(tx.get("Account") == alice.address for tx in find_export_txns(ctx, seq)): + raise AssertionError( + f"A second Export was submitted before recovery in ledger {seq}" + ) + assert_export_latch( + ctx, + alice.address, + log, + origin_hash=origin, + expect_witness=True, + ) + log(f"Same Export {origin} completed after selected validator recovery") + log("PASS") diff --git a/.testnet/scenarios/export/export_helpers.py b/.testnet/scenarios/export/export_helpers.py index f3763c4f3..a31f47d34 100644 --- a/.testnet/scenarios/export/export_helpers.py +++ b/.testnet/scenarios/export/export_helpers.py @@ -2,7 +2,13 @@ from __future__ import annotations -from xahaud_scripts.testnet.config import _unl_report_index, feature_name_to_hash +import json + +from xahaud_scripts.testnet.config import ( + _decode_node_public_key, + _unl_report_index, + feature_name_to_hash, +) EXPORT_RETRY_LEDGER_WINDOW = 5 EXPORT_PUBLICATION_LEDGER_WINDOW = 5 @@ -65,7 +71,29 @@ def find_export_txns(ctx, seq): return [tx for tx in txns if tx.get("TransactionType") == "Export"] -def export_authority(ctx, *, require_unl_report=True): +def _validator_master_keys_by_node(ctx): + """Return generated validator master keys keyed by testnet node id.""" + network = json.loads((ctx.base_dir / "network.json").read_text()) + return { + int(node["id"]): _decode_node_public_key(node["public_key"]) + for node in network["nodes"] + } + + +def bitmap_positions(bitmap): + """Return the selected bit positions from a hex bitmap or bytes.""" + raw = bytes.fromhex(bitmap) if isinstance(bitmap, str) else bytes(bitmap) + return { + byte_index * 8 + bit_index + for byte_index, byte in enumerate(raw) + for bit_index in range(8) + if byte & (1 << bit_index) + } + + +def export_authority( + ctx, *, require_unl_report=True, committee_node_ids=None +): """Build the explicit authority declaration for a direct Export. Direct clients pin the validated parent ledger and select members from its @@ -92,8 +120,34 @@ def export_authority(ctx, *, require_unl_report=True): raise AssertionError(f"UNLReport active universe unavailable: {report}") active = [{}] - committee = bytearray((len(active) + 7) // 8) - for index in range(len(active)): + active_keys = set() + for entry in active: + validator = entry.get("ActiveValidator", entry) + key = validator.get("PublicKey") + if not key: + raise AssertionError(f"Malformed UNLReport validator entry: {entry}") + active_keys.add(key.upper()) + active_keys = sorted(active_keys, key=bytes.fromhex) + + if committee_node_ids is None: + selected_positions = range(len(active_keys)) + else: + masters = _validator_master_keys_by_node(ctx) + selected_positions = [] + for node_id in committee_node_ids: + if node_id not in masters: + raise AssertionError(f"Unknown testnet validator node n{node_id}") + try: + selected_positions.append(active_keys.index(masters[node_id])) + except ValueError as exc: + raise AssertionError( + f"Validator n{node_id} is absent from the active UNLReport" + ) from exc + if not selected_positions: + raise AssertionError("Export committee must select at least one validator") + + committee = bytearray((len(active_keys) + 7) // 8) + for index in selected_positions: committee[index // 8] |= 1 << (index % 8) return { @@ -175,7 +229,16 @@ async def wait_for_validated_transaction( raise AssertionError(f"Transaction {tx_hash} did not validate by ledger {target}") -async def submit_direct_export(ctx, log, tx, wallet, *, timeout=60, max_rebases=2): +async def submit_direct_export( + ctx, + log, + tx, + wallet, + *, + timeout=60, + max_rebases=2, + committee_node_ids=None, +): """Submit a direct Export, rebasing after a validated parent mismatch.""" for attempt in range(max_rebases + 1): current = ctx.validated_ledger_index(0) @@ -183,7 +246,9 @@ async def submit_direct_export(ctx, log, tx, wallet, *, timeout=60, max_rebases= raise AssertionError("Validated ledger unavailable before Export") candidate = dict(tx) - candidate.update(export_authority(ctx)) + candidate.update( + export_authority(ctx, committee_node_ids=committee_node_ids) + ) candidate["LastLedgerSequence"] = current + EXPORT_RETRY_LEDGER_WINDOW result = await ctx.submit_and_wait(candidate, wallet, timeout=timeout) if result.get("engine_result") != "tecEXPORT_UNIVERSE_MISMATCH": diff --git a/.testnet/scenarios/export/export_share_subscription.py b/.testnet/scenarios/export/export_share_subscription.py new file mode 100644 index 000000000..d2cfcb3e9 --- /dev/null +++ b/.testnet/scenarios/export/export_share_subscription.py @@ -0,0 +1,193 @@ +""":descr: subscribe over a real WebSocket to post-validation Export shares + +The subscriber opens before the Export is submitted. It proves the stream +publishes a quorum of independently attributable shares for the exact validated +origin and that those same signature records form the later ledger witness. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json + +import websockets +from xahaud_scripts.testnet.config import _decode_node_public_key + +from export_helpers import ( + EXPORT_RETRY_LEDGER_WINDOW, + assert_export_latch, + bitmap_positions, + require_export, + submit_direct_export, + wait_for_export_signature_witness, +) + + +def _witness_records(witness): + positions = sorted(bitmap_positions(witness["EntropyContributors"])) + signers = witness["_WitnessSigners"] + if len(positions) != len(signers): + raise AssertionError("Witness bitmap and signer count differ") + return { + ( + position, + signer["SigningPubKey"].upper(), + signer["TxnSignature"].upper(), + ) + for position, signer in zip(positions, signers, strict=True) + } + + +async def scenario(ctx, log): + await require_export(ctx, log) + await ctx.fund_accounts({"alice": 10000, "bob": 1000}) + alice = ctx.account("alice") + bob = ctx.account("bob") + + network = json.loads((ctx.base_dir / "network.json").read_text()) + node0 = next(node for node in network["nodes"] if int(node["id"]) == 0) + ws_url = f"ws://127.0.0.1:{node0['port_ws']}" + events = [] + + async with websockets.connect(ws_url, open_timeout=10) as websocket: + await websocket.send( + json.dumps( + { + "id": 1, + "command": "subscribe", + "streams": ["export_signatures"], + } + ) + ) + ack = json.loads(await asyncio.wait_for(websocket.recv(), timeout=10)) + if ack.get("status") != "success": + raise AssertionError(f"export_signatures subscription failed: {ack}") + log("Subscribed to export_signatures over WebSocket") + + async def receive_events(): + while True: + event = json.loads(await websocket.recv()) + if event.get("stream") == "export_signatures": + events.append(event) + + reader = asyncio.create_task(receive_events()) + try: + current = ctx.validated_ledger_index(0) + result = await submit_direct_export( + ctx, + log, + { + "TransactionType": "Export", + "Fee": "1000000", + "ExportedTxn": { + "TransactionType": "Payment", + "Account": alice.address, + "Destination": bob.address, + "Amount": "1000000", + "Fee": "10", + "Sequence": 0, + "TicketSequence": 1, + "FirstLedgerSequence": current + 1, + "LastLedgerSequence": current + + EXPORT_RETRY_LEDGER_WINDOW, + "Flags": 2147483648, + "SigningPubKey": "", + }, + }, + alice.wallet, + ) + if result.get("engine_result") != "tesSUCCESS": + raise AssertionError(f"Export failed: {result}") + + origin = result.get("hash") + origin_seq = int(result.get("ledger_index")) + if not origin: + raise AssertionError(f"Validated Export missing hash: {result}") + origin_ledger = ctx.ledger(origin_seq) or {} + origin_hash = origin_ledger.get("ledger_hash") or origin_ledger.get( + "ledger", {} + ).get("hash") + if not origin_hash: + raise AssertionError( + f"Validated origin ledger {origin_seq} missing hash" + ) + + latches = assert_export_latch( + ctx, + alice.address, + log, + origin_hash=origin, + expect_witness=False, + ) + selected = bitmap_positions(latches[0]["ExportCommittee"]) + quorum = (4 * len(selected) + 4) // 5 + + witness = await wait_for_export_signature_witness( + ctx, log, origin, after_ledger=origin_seq + ) + expected_records = _witness_records(witness) + + deadline = asyncio.get_running_loop().time() + 10 + while asyncio.get_running_loop().time() < deadline: + matching = [event for event in events if event.get("origin_txid") == origin] + observed_records = { + ( + int(event["universe_position"]), + _decode_node_public_key(event["signing_key"]), + event["signature"].upper(), + ) + for event in matching + } + if expected_records <= observed_records: + break + await asyncio.sleep(0.1) + else: + raise AssertionError( + "WebSocket stream did not publish every signature used by " + f"the witness: expected={expected_records}, " + f"observed={observed_records}" + ) + + unique_positions = set() + for event in matching: + if event.get("type") != "exportSignatureReceived": + raise AssertionError(f"Unexpected Export stream event: {event}") + if event.get("version") != 1: + raise AssertionError(f"Unexpected Export share version: {event}") + if event.get("owner") != alice.address: + raise AssertionError(f"Export stream owner mismatch: {event}") + if int(event.get("origin_ledger_seq", 0)) != origin_seq: + raise AssertionError(f"Export stream origin sequence mismatch: {event}") + if event.get("origin_ledger_hash") != origin_hash: + raise AssertionError(f"Export stream origin hash mismatch: {event}") + if event.get("trigger_txid") != origin: + raise AssertionError(f"Export stream trigger mismatch: {event}") + position = int(event.get("universe_position", -1)) + if position not in selected: + raise AssertionError(f"Unselected validator streamed a share: {event}") + unique_positions.add(position) + + if len(unique_positions) < quorum: + raise AssertionError( + f"Stream exposed only {len(unique_positions)} selected positions; " + f"need quorum {quorum}" + ) + witness_seq = int(witness["LedgerSequence"]) + if not any( + int(event.get("ledger_index", 0)) < witness_seq + for event in matching + ): + raise AssertionError( + "No Export share stream event preceded the ledger witness" + ) + log( + f"WebSocket exposed {len(unique_positions)} selected shares; " + f"witness used {len(expected_records)}" + ) + finally: + reader.cancel() + with contextlib.suppress(asyncio.CancelledError): + await reader + + log("PASS")