forkchoice: skip STARK aggregate verify on proofs the filler just produced

packages/testing/src/consensus_testing/test_fixtures/fork_choice.py

@@ -276,9 +276,14 @@ def make_fixture(self) -> Self:
 
                         # Process the block through Store.
                         # This validates, applies state transition, and updates the store's head.
+                        #
+                        # trust_proofs mirrors the spec's valid_signature flag.
+                        # True: fresh proof, skip STARK verify.
+                        # False: explicit invalid test, keep verify to reject.
                         store = store.on_block(
                             signed_block,
                             scheme=LEAN_ENV_TO_SCHEMES[self.lean_env],
+                            trust_proofs=step.block.valid_signature,
                         )
 
                     case AttestationStep():
@@ -306,7 +311,13 @@ def make_fixture(self) -> Self:
                             key_manager,
                         )
                         step._filled_attestation = signed_aggregated
-                        store = store.on_gossip_aggregated_attestation(signed_aggregated)
+                        # trust_proof mirrors the spec's valid_signature flag.
+                        # True: fresh proof, skip STARK verify.
+                        # False: explicit invalid test, keep verify to reject.
+                        store = store.on_gossip_aggregated_attestation(
+                            signed_aggregated,
+                            trust_proof=step.attestation.valid_signature,
+                        )
 
                     case _:
                         raise ValueError(f"Step {i}: unknown step type {type(step).__name__}")

src/lean_spec/subspecs/containers/block/block.py

@@ -98,6 +98,8 @@ def verify_signatures(
         self,
         validators: Validators,
         scheme: GeneralizedXmssScheme = TARGET_SIGNATURE_SCHEME,
+        *,
+        skip_aggregate_verify: bool = False,
     ) -> bool:
         """
         Verify all XMSS signatures in this signed block.
@@ -110,6 +112,10 @@ def verify_signatures(
         Args:
             validators: Validator registry providing public keys for verification.
             scheme: XMSS signature scheme for verification.
+            skip_aggregate_verify: Skip the STARK aggregate verify when True.
+                Intended for fixture generation on fresh proofs.
+                Structural checks and proposer verify still run.
+                Defaults to False.
 
         Returns:
             True if all signatures are valid.
@@ -137,14 +143,19 @@ def verify_signatures(
             # The aggregation bits encode validator indices as a bitfield.
             validator_ids = aggregated_attestation.aggregation_bits.to_validator_indices()
 
-            # The signed message is the attestation data root.
-            # All validators in this group signed this exact data.
-            attestation_data_root = hash_tree_root(aggregated_attestation.data)
-
             for validator_id in validator_ids:
                 num_validators = Uint64(len(validators))
                 assert validator_id.is_valid(num_validators), "Validator index out of range"
 
+            # Trusted path: caller vouches for this fresh proof.
+            # Structural checks above have already run.
+            if skip_aggregate_verify:
+                continue
+
+            # The signed message is the attestation data root.
+            # All validators in this group signed this exact data.
+            attestation_data_root = hash_tree_root(aggregated_attestation.data)
+
             # Collect attestation public keys for all participating validators.
             # Order matters: must match the order in the aggregated signature.
             public_keys = [validators[vid].get_attestation_pubkey() for vid in validator_ids]

src/lean_spec/subspecs/forkchoice/store.py

@@ -396,7 +396,10 @@ def on_gossip_attestation(
             )
 
     def on_gossip_aggregated_attestation(
-        self, signed_attestation: SignedAggregatedAttestation
+        self,
+        signed_attestation: SignedAggregatedAttestation,
+        *,
+        trust_proof: bool = False,
     ) -> "Store":
         """
         Process a signed aggregated attestation received via aggregation topic
@@ -407,6 +410,10 @@ def on_gossip_aggregated_attestation(
 
         Args:
             signed_attestation: The signed aggregated attestation from committee aggregation.
+            trust_proof: Skip the STARK aggregate verify when True.
+                Intended for fixture generation on fresh proofs.
+                Structural checks still run.
+                Defaults to False.
 
         Returns:
             New Store with aggregation processed and stored.
@@ -437,20 +444,22 @@ def on_gossip_aggregated_attestation(
                 f"Validator {validator_id} not found in state {data.target.root.hex()}"
             )
 
-        # Prepare public keys for verification
-        public_keys = [validators[vid].get_attestation_pubkey() for vid in validator_ids]
-
-        # Verify the leanVM aggregated proof
-        try:
-            proof.verify(
-                public_keys=public_keys,
-                message=hash_tree_root(data),
-                slot=data.slot,
-            )
-        except AggregationError as exc:
-            raise AssertionError(
-                f"Committee aggregation signature verification failed: {exc}"
-            ) from exc
+        # Trusted path: caller vouches for this fresh proof.
+        if not trust_proof:
+            # Prepare public keys for verification
+            public_keys = [validators[vid].get_attestation_pubkey() for vid in validator_ids]
+
+            # Verify the leanVM aggregated proof
+            try:
+                proof.verify(
+                    public_keys=public_keys,
+                    message=hash_tree_root(data),
+                    slot=data.slot,
+                )
+            except AggregationError as exc:
+                raise AssertionError(
+                    f"Committee aggregation signature verification failed: {exc}"
+                ) from exc
 
         # Shallow-copy the dict and its inner sets to preserve immutability.
         new_aggregated_payloads = {
@@ -469,6 +478,8 @@ def on_block(
         self,
         signed_block: SignedBlock,
         scheme: GeneralizedXmssScheme = TARGET_SIGNATURE_SCHEME,
+        *,
+        trust_proofs: bool = False,
     ) -> "Store":
         """
         Process a new block and update the forkchoice state.
@@ -482,6 +493,10 @@ def on_block(
         Args:
             signed_block: Complete signed block.
             scheme: XMSS signature scheme to use for signature verification.
+            trust_proofs: Skip the STARK aggregate verify when True.
+                Intended for fixture generation on fresh proofs.
+                Proposer-signature verify and structural checks still run.
+                Defaults to False.
 
         Returns:
             New Store with block integrated and head updated.
@@ -508,7 +523,11 @@ def on_block(
             )
 
             # Validate cryptographic signatures
-            valid_signatures = signed_block.verify_signatures(parent_state.validators, scheme)
+            valid_signatures = signed_block.verify_signatures(
+                parent_state.validators,
+                scheme,
+                skip_aggregate_verify=trust_proofs,
+            )
 
             # Execute state transition function to compute post-block state
             post_state = parent_state.state_transition(block, valid_signatures)

tests/lean_spec/subspecs/containers/block/test_verify_signatures_skip.py

@@ -0,0 +1,66 @@
+"""Tests for the skip_aggregate_verify flag on SignedBlock.verify_signatures."""
+
+from __future__ import annotations
+
+import pytest
+from consensus_testing.keys import XmssKeyManager
+
+from lean_spec.subspecs.containers.block.block import BlockSignatures
+from lean_spec.subspecs.containers.slot import Slot
+from lean_spec.subspecs.containers.validator import ValidatorIndex
+from lean_spec.subspecs.xmss.aggregation import AggregatedSignatureProof
+from lean_spec.types import Bytes32
+from tests.lean_spec.helpers import (
+    make_aggregated_proof,
+    make_signed_block_from_store,
+    make_store,
+)
+
+
+def test_skip_aggregate_verify_bypasses_stark(
+    key_manager: XmssKeyManager, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """skip_aggregate_verify=True must not invoke AggregatedSignatureProof.verify."""
+    base = make_store(num_validators=3, key_manager=key_manager)
+    data = base.produce_attestation_data(Slot(1))
+    proof = make_aggregated_proof(key_manager, [ValidatorIndex(1), ValidatorIndex(2)], data)
+    producer = base.model_copy(update={"latest_known_aggregated_payloads": {data: {proof}}})
+    consumer, signed_block = make_signed_block_from_store(
+        producer, key_manager, Slot(1), ValidatorIndex(1)
+    )
+
+    def _raise_if_called(*args: object, **kwargs: object) -> None:
+        pytest.fail("AggregatedSignatureProof.verify must not be called")
+
+    monkeypatch.setattr(AggregatedSignatureProof, "verify", _raise_if_called)
+
+    validators = consumer.states[consumer.head].validators
+    assert signed_block.verify_signatures(validators, skip_aggregate_verify=True) is True
+
+
+def test_skip_aggregate_verify_preserves_proposer_verify(key_manager: XmssKeyManager) -> None:
+    """Skipping the aggregate verify must not skip the proposer signature verify."""
+    base = make_store(num_validators=3, key_manager=key_manager)
+    data = base.produce_attestation_data(Slot(1))
+    proof = make_aggregated_proof(key_manager, [ValidatorIndex(1), ValidatorIndex(2)], data)
+    producer = base.model_copy(update={"latest_known_aggregated_payloads": {data: {proof}}})
+    consumer, signed_block = make_signed_block_from_store(
+        producer, key_manager, Slot(1), ValidatorIndex(1)
+    )
+
+    # Sign with a different validator: shape-valid XMSS, wrong proposer.
+    bad_sig = key_manager.sign_block_root(
+        ValidatorIndex(2), signed_block.block.slot, Bytes32.zero()
+    )
+    corrupted = signed_block.model_copy(
+        update={
+            "signature": BlockSignatures(
+                attestation_signatures=signed_block.signature.attestation_signatures,
+                proposer_signature=bad_sig,
+            )
+        }
+    )
+
+    validators = consumer.states[consumer.head].validators
+    with pytest.raises(AssertionError, match="Proposer block signature verification failed"):
+        corrupted.verify_signatures(validators, skip_aggregate_verify=True)

tests/lean_spec/subspecs/forkchoice/test_store_attestations.py

@@ -65,6 +65,30 @@ def test_on_block_processes_multi_validator_aggregations(key_manager: XmssKeyMan
     assert extracted_attestations[ValidatorIndex(2)] == attestation_data
 
 
+def test_on_block_trust_proofs_skips_stark_verify(
+    key_manager: XmssKeyManager, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """trust_proofs=True must not invoke AggregatedSignatureProof.verify."""
+    base_store = make_store(num_validators=3, key_manager=key_manager)
+    attestation_data = base_store.produce_attestation_data(Slot(1))
+    proof = make_aggregated_proof(
+        key_manager, [ValidatorIndex(1), ValidatorIndex(2)], attestation_data
+    )
+    producer_store = base_store.model_copy(
+        update={"latest_known_aggregated_payloads": {attestation_data: {proof}}}
+    )
+    consumer_store, signed_block = make_signed_block_from_store(
+        producer_store, key_manager, Slot(1), ValidatorIndex(1)
+    )
+
+    def _raise_if_called(*args: object, **kwargs: object) -> None:
+        pytest.fail("AggregatedSignatureProof.verify must not be called")
+
+    monkeypatch.setattr(AggregatedSignatureProof, "verify", _raise_if_called)
+
+    consumer_store.on_block(signed_block, trust_proofs=True)
+
+
 def test_on_block_preserves_immutability_of_aggregated_payloads(
     key_manager: XmssKeyManager,
 ) -> None:
@@ -394,6 +418,26 @@ def test_invalid_proof_rejected(self, key_manager: XmssKeyManager) -> None:
         with pytest.raises(AssertionError, match="signature verification failed"):
             store.on_gossip_aggregated_attestation(signed_aggregated)
 
+    def test_trust_proof_skips_stark_verify(
+        self, key_manager: XmssKeyManager, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """trust_proof=True must not invoke AggregatedSignatureProof.verify."""
+        store, attestation_data = make_store_with_attestation_data(
+            key_manager, num_validators=4, validator_id=ValidatorIndex(0)
+        )
+        proof = make_aggregated_proof(
+            key_manager, [ValidatorIndex(1), ValidatorIndex(2)], attestation_data
+        )
+        signed_aggregated = SignedAggregatedAttestation(data=attestation_data, proof=proof)
+
+        def _raise_if_called(*args: object, **kwargs: object) -> None:
+            pytest.fail("AggregatedSignatureProof.verify must not be called")
+
+        monkeypatch.setattr(AggregatedSignatureProof, "verify", _raise_if_called)
+
+        updated = store.on_gossip_aggregated_attestation(signed_aggregated, trust_proof=True)
+        assert attestation_data in updated.latest_new_aggregated_payloads
+
     def test_multiple_proofs_accumulate(self, key_manager: XmssKeyManager) -> None:
         """
         Multiple aggregated proofs for same validator accumulate.