fix: validate gossip aggregated attestations before storing

packages/testing/src/consensus_testing/__init__.py

@@ -19,6 +19,8 @@
     BlockSpec,
     BlockStep,
     ForkChoiceStep,
+    GossipAggregatedAttestationSpec,
+    GossipAggregatedAttestationStep,
     GossipAttestationSpec,
     StateExpectation,
     StoreChecks,
@@ -34,6 +36,7 @@
 __all__ = [
     # Public API
     "AggregatedAttestationSpec",
+    "GossipAggregatedAttestationSpec",
     "GossipAttestationSpec",
     "BlockSpec",
     "forks",
@@ -50,6 +53,7 @@
     "TickStep",
     "BlockStep",
     "AttestationStep",
+    "GossipAggregatedAttestationStep",
     "ForkChoiceStep",
     "StateExpectation",
     "StoreChecks",

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

@@ -21,6 +21,10 @@
     AttestationData,
     SignedAttestation,
 )
+from lean_spec.subspecs.containers.attestation.aggregation_bits import AggregationBits
+from lean_spec.subspecs.containers.attestation.attestation import (
+    SignedAggregatedAttestation,
+)
 from lean_spec.subspecs.containers.block import (
     Block,
     BlockBody,
@@ -34,11 +38,12 @@
 from lean_spec.subspecs.containers.slot import Slot
 from lean_spec.subspecs.containers.state import Validators
 from lean_spec.subspecs.containers.state.state import State
-from lean_spec.subspecs.containers.validator import ValidatorIndex
+from lean_spec.subspecs.containers.validator import ValidatorIndex, ValidatorIndices
 from lean_spec.subspecs.forkchoice import Store
 from lean_spec.subspecs.ssz import hash_tree_root
+from lean_spec.subspecs.xmss.aggregation import AggregatedSignatureProof
 from lean_spec.subspecs.xmss.containers import Signature
-from lean_spec.types import Bytes32, Uint64
+from lean_spec.types import ByteListMiB, Bytes32, Uint64
 
 from ..keys import (
     LEAN_ENV_TO_SCHEMES,
@@ -51,6 +56,8 @@
     BlockSpec,
     BlockStep,
     ForkChoiceStep,
+    GossipAggregatedAttestationSpec,
+    GossipAggregatedAttestationStep,
     GossipAttestationSpec,
     TickStep,
 )
@@ -156,6 +163,8 @@ def set_max_slot_default(self) -> Self:
                     max_slot_value = max(max_slot_value, step.block.slot)
                 elif isinstance(step, AttestationStep):
                     max_slot_value = max(max_slot_value, step.attestation.slot)
+                elif isinstance(step, GossipAggregatedAttestationStep):
+                    max_slot_value = max(max_slot_value, step.attestation.slot)
 
             self.max_slot = max_slot_value
 
@@ -309,6 +318,16 @@ def make_fixture(self) -> Self:
                             scheme=LEAN_ENV_TO_SCHEMES[self.lean_env],
                         )
 
+                    case GossipAggregatedAttestationStep():
+                        signed_aggregated = self._build_signed_aggregated_attestation_from_spec(
+                            step.attestation,
+                            store,
+                            self._block_registry,
+                            key_manager,
+                        )
+                        step._filled_attestation = signed_aggregated
+                        store = store.on_gossip_aggregated_attestation(signed_aggregated)
+
                     case _:
                         raise ValueError(f"Step {i}: unknown step type {type(step).__name__}")
 
@@ -789,3 +808,193 @@ def _build_overridden_attestation_data(
             target=target,
             source=source,
         )
+
+    def _build_attestation_data_from_gossip_aggregated_spec(
+        self,
+        spec: GossipAggregatedAttestationSpec,
+        block_registry: dict[str, Block],
+        state: State,
+    ) -> AttestationData:
+        """
+        Build attestation data for a gossip aggregated attestation.
+
+        Resolves the three checkpoints (target, head, source) from the spec.
+
+        The state provides fallback values for the source checkpoint
+        when neither an explicit root nor a label is specified.
+
+        Args:
+            spec: Aggregated attestation specification with checkpoint overrides.
+            block_registry: Labeled blocks for checkpoint resolution.
+            state: Head state providing the latest justified checkpoint as source fallback.
+
+        Returns:
+            Attestation data with all three checkpoints resolved.
+
+        Raises:
+            ValueError: If the target has neither an explicit root nor
+                a label.
+        """
+        # Resolve the target checkpoint.
+        #
+        # An explicit root takes highest priority.
+        # A label triggers lookup in the block registry.
+        # Unlike the other two checkpoints, the target is mandatory.
+        if spec.target_root is not None:
+            target = Checkpoint(root=spec.target_root, slot=spec.target_slot)
+        elif spec.target_root_label is not None:
+            target = self._resolve_checkpoint(
+                spec.target_root_label, spec.target_slot, block_registry
+            )
+        else:
+            raise ValueError("gossip aggregated attestation spec requires a target root")
+
+        # Resolve the head checkpoint.
+        #
+        # Priority: explicit root > label > target checkpoint.
+        # - When using an explicit root without a slot, the target slot is used.
+        # - When no head information is provided at all, the head mirrors the target.
+        #
+        # This matches honest validator behavior.
+        if spec.head_root is not None:
+            head = Checkpoint(
+                root=spec.head_root,
+                slot=spec.head_slot if spec.head_slot is not None else spec.target_slot,
+            )
+        elif spec.head_root_label is not None:
+            head = self._resolve_checkpoint(spec.head_root_label, spec.head_slot, block_registry)
+        else:
+            head = Checkpoint(
+                root=target.root,
+                slot=spec.head_slot if spec.head_slot is not None else target.slot,
+            )
+
+        # Resolve the source checkpoint.
+        #
+        # Priority: explicit root > label > latest justified from state.
+        # The source represents the most recent justified checkpoint the attester is aware of.
+        # When not overridden, the state's latest justified checkpoint provides the correct default.
+        if spec.source_root is not None:
+            source = Checkpoint(
+                root=spec.source_root,
+                slot=(
+                    spec.source_slot
+                    if spec.source_slot is not None
+                    else state.latest_justified.slot
+                ),
+            )
+        elif spec.source_root_label is not None:
+            source = self._resolve_checkpoint(
+                spec.source_root_label, spec.source_slot, block_registry
+            )
+        else:
+            source = Checkpoint(
+                root=state.latest_justified.root,
+                slot=(
+                    spec.source_slot
+                    if spec.source_slot is not None
+                    else state.latest_justified.slot
+                ),
+            )
+
+        # Assemble the final attestation data from resolved checkpoints.
+        return AttestationData(
+            slot=spec.slot,
+            head=head,
+            target=target,
+            source=source,
+        )
+
+    def _build_signed_aggregated_attestation_from_spec(
+        self,
+        spec: GossipAggregatedAttestationSpec,
+        store: Store,
+        block_registry: dict[str, Block],
+        key_manager: XmssKeyManager,
+    ) -> SignedAggregatedAttestation:
+        """
+        Build a signed aggregated attestation from a gossip specification.
+
+        Args:
+            spec: Aggregated attestation specification with participant and checkpoint
+                configuration.
+            store: Fork choice store providing head state for checkpoint resolution.
+            block_registry: Labeled blocks for checkpoint resolution.
+            key_manager: XMSS key manager for signing attestation data.
+
+        Returns:
+            Signed aggregated attestation ready for gossip processing.
+        """
+        # Resolve checkpoints into concrete attestation data.
+        attestation_data = self._build_attestation_data_from_gossip_aggregated_spec(
+            spec,
+            block_registry,
+            store.states[store.head],
+        )
+
+        # Separate "claimed" from "actual" participants.
+        #
+        # - Claimed validators appear in the proof's participant bitfield.
+        # - Actual signers produce the cryptographic material.
+        # They default to the same set for honest attestations.
+        validator_ids = spec.validator_ids
+        signer_ids = spec.signer_ids or spec.validator_ids
+
+        # Path 1: Invalid signature.
+        #
+        # Build a proof with the correct participant bitfield but zeroed-out proof bytes.
+        # This exercises signature verification rejection.
+        if not spec.valid_signature:
+            proof = AggregatedSignatureProof(
+                participants=AggregationBits.from_validator_indices(
+                    ValidatorIndices(data=validator_ids)
+                ),
+                proof_data=ByteListMiB(data=b"\x00" * 32),
+            )
+            return SignedAggregatedAttestation(data=attestation_data, proof=proof)
+
+        # Path 2: Valid signature.
+        #
+        # Collect each signer's public key and individual signature over the attestation data.
+        # These feed into the aggregation step.
+        raw_xmss = [
+            (
+                key_manager.get_public_keys(vid)[0],
+                key_manager.sign_attestation_data(vid, attestation_data),
+            )
+            for vid in signer_ids
+        ]
+
+        # Encode which validators actually produced signatures.
+        xmss_participants = AggregationBits.from_validator_indices(
+            ValidatorIndices(data=signer_ids)
+        )
+
+        # Aggregate individual signatures into a single proof.
+        #
+        # The proof combines all signer contributions so the verifier
+        # can check them in one pass against the attestation data root.
+        proof = AggregatedSignatureProof.aggregate(
+            xmss_participants=xmss_participants,
+            children=[],
+            raw_xmss=raw_xmss,
+            message=attestation_data.data_root_bytes(),
+            slot=attestation_data.slot,
+        )
+
+        # Path 3: Participant mismatch.
+        #
+        # Replace the participant bitfield with different validator indices
+        # while keeping the original proof bytes intact.
+        # The proof is cryptographically valid for the actual signers,
+        # but the claimed participants no longer match.
+        # The store must detect and reject this inconsistency.
+        if spec.signer_ids and spec.signer_ids != spec.validator_ids:
+            proof = AggregatedSignatureProof(
+                participants=AggregationBits.from_validator_indices(
+                    ValidatorIndices(data=validator_ids)
+                ),
+                proof_data=proof.proof_data,
+            )
+
+        return SignedAggregatedAttestation(data=attestation_data, proof=proof)

packages/testing/src/consensus_testing/test_types/__init__.py

@@ -3,20 +3,23 @@
 from .aggregated_attestation_spec import AggregatedAttestationSpec
 from .block_spec import BlockSpec
 from .genesis import generate_pre_state
+from .gossip_aggregated_attestation_spec import GossipAggregatedAttestationSpec
 from .gossip_attestation_spec import GossipAttestationSpec
 from .state_expectation import StateExpectation
 from .step_types import (
     AttestationStep,
     BaseForkChoiceStep,
     BlockStep,
     ForkChoiceStep,
+    GossipAggregatedAttestationStep,
     TickStep,
 )
 from .store_checks import AggregatedAttestationCheck, AttestationCheck, StoreChecks
 
 __all__ = [
     "AggregatedAttestationSpec",
     "GossipAttestationSpec",
+    "GossipAggregatedAttestationSpec",
     "StateExpectation",
     "StoreChecks",
     "AttestationCheck",
@@ -27,5 +30,6 @@
     "BlockStep",
     "AttestationStep",
     "ForkChoiceStep",
+    "GossipAggregatedAttestationStep",
     "generate_pre_state",
 ]

packages/testing/src/consensus_testing/test_types/gossip_aggregated_attestation_spec.py

@@ -0,0 +1,54 @@
+"""Lightweight aggregated-gossip attestation specification."""
+
+from __future__ import annotations
+
+from lean_spec.subspecs.containers.slot import Slot
+from lean_spec.subspecs.containers.validator import ValidatorIndex
+from lean_spec.types import Bytes32, CamelModel
+
+
+class GossipAggregatedAttestationSpec(CamelModel):
+    """
+    Specification for an aggregated attestation received via gossip.
+
+    The spec allows overriding head/source checkpoints to exercise validation logic.
+    """
+
+    validator_ids: list[ValidatorIndex]
+    """Claimed validators participating in the aggregation."""
+
+    slot: Slot
+    """Slot of the attestation."""
+
+    target_slot: Slot
+    """Slot of the attestation target checkpoint."""
+
+    target_root_label: str | None = None
+    """Label referencing the target block root."""
+
+    target_root: Bytes32 | None = None
+    """Optional explicit target root (bypasses label lookup)."""
+
+    head_root_label: str | None = None
+    """Optional label for the head checkpoint."""
+
+    head_root: Bytes32 | None = None
+    """Optional explicit head root."""
+
+    head_slot: Slot | None = None
+    """Optional override for the head checkpoint slot."""
+
+    source_root_label: str | None = None
+    """Optional label for the source checkpoint."""
+
+    source_root: Bytes32 | None = None
+    """Optional explicit source root."""
+
+    source_slot: Slot | None = None
+    """Optional override for the source checkpoint slot."""
+
+    valid_signature: bool = True
+    """Whether the aggregated proof should be generated with valid signatures."""
+
+    signer_ids: list[ValidatorIndex] | None = None
+    """Optional override for which validators actually produce the signatures."""