diff --git a/packages/testing/src/consensus_testing/test_fixtures/verify_signatures.py b/packages/testing/src/consensus_testing/test_fixtures/verify_signatures.py index 77fd9e86..11be0ec5 100644 --- a/packages/testing/src/consensus_testing/test_fixtures/verify_signatures.py +++ b/packages/testing/src/consensus_testing/test_fixtures/verify_signatures.py @@ -6,6 +6,7 @@ from pydantic import Field +from lean_spec.spec.crypto.merkleization import hash_tree_root from lean_spec.spec.forks import AggregationBits, Checkpoint, Slot, ValidatorIndex from lean_spec.spec.forks.lstar.containers import ( AggregatedAttestation, @@ -77,6 +78,9 @@ class VerifySignaturesTest(BaseConsensusFixture): signing so the block root differs. Exercises the per-component message binding that prevents reusing an honest proof under a different message. + - `{"operation": "swap_first_two_attestations"}`: Swap the first + two body attestations and re-sign only the proposer. Exercises + body/proof ordering without relying on a block-root mismatch. Tampered blocks bypass the builder's structural invariants. The resulting fixture pins the exact rejection a client must raise when @@ -214,4 +218,45 @@ def _apply_tamper(self, signed_block: SignedBlock) -> SignedBlock: signed_block.block.state_root = Bytes32(b"\xff" * 32) return signed_block + if operation == "swap_first_two_attestations": + body = signed_block.block.body + original = body.attestations.data + if len(original) < 2: + raise ValueError("swap_first_two_attestations requires at least two attestations") + assert self.anchor_state is not None + + key_manager = XmssKeyManager.shared() + original_attestation_proofs = [ + key_manager.sign_and_aggregate( + list(attestation.aggregation_bits.to_validator_indices()), + attestation.data, + ) + for attestation in original + ] + + swapped_body = body.model_copy( + update={ + "attestations": AggregatedAttestations( + data=[original[1], original[0], *original[2:]] + ) + } + ) + swapped_block = signed_block.block.model_copy(update={"body": swapped_body}) + + # Keep the block root honestly signed; only the attestation + # proof order remains mismatched with the body order. + post_state = LstarSpec().process_slots(self.anchor_state, swapped_block.slot) + post_state = LstarSpec().process_block(post_state, swapped_block) + swapped_block = swapped_block.model_copy( + update={"state_root": hash_tree_root(post_state)} + ) + + return self.block._sign_block( + swapped_block, + original_attestation_proofs, + swapped_block.proposer_index, + key_manager, + self.anchor_state, + ) + raise ValueError(f"Unknown tamper operation: {operation!r}") diff --git a/tests/consensus/lstar/verify_signatures/test_structural_rejections.py b/tests/consensus/lstar/verify_signatures/test_structural_rejections.py index 76a80efc..e847ae00 100644 --- a/tests/consensus/lstar/verify_signatures/test_structural_rejections.py +++ b/tests/consensus/lstar/verify_signatures/test_structural_rejections.py @@ -10,12 +10,15 @@ import pytest from consensus_testing import ( + AggregatedAttestationSpec, BlockSpec, VerifySignaturesTestFiller, + build_anchor, generate_pre_state, ) -from lean_spec.spec.forks import Slot +from lean_spec.spec.crypto.merkleization import hash_tree_root +from lean_spec.spec.forks import Slot, ValidatorIndex pytestmark = pytest.mark.valid_until("Lstar") @@ -115,3 +118,39 @@ def test_proof_reused_under_different_message_rejected( tamper={"operation": "mutate_state_root"}, expect_exception=AssertionError, ) + + +def test_attestation_proof_order_mismatch_rejected( + verify_signatures_test: VerifySignaturesTestFiller, +) -> None: + """A block whose body order no longer matches proof order is rejected.""" + anchor_state, anchor_block = build_anchor(num_validators=4, anchor_slot=Slot(2)) + parent_root = hash_tree_root(anchor_block) + + verify_signatures_test( + anchor_state=anchor_state, + block=BlockSpec( + slot=Slot(3), + parent_root=parent_root, + attestations=[ + AggregatedAttestationSpec( + validator_indices=[ValidatorIndex(0)], + slot=Slot(3), + target_slot=Slot(1), + target_root=anchor_state.historical_block_hashes[1], + head_root=parent_root, + head_slot=Slot(2), + ), + AggregatedAttestationSpec( + validator_indices=[ValidatorIndex(2)], + slot=Slot(3), + target_slot=Slot(2), + target_root=parent_root, + head_root=parent_root, + head_slot=Slot(2), + ), + ], + ), + tamper={"operation": "swap_first_two_attestations"}, + expect_exception=AssertionError, + )