fix: enforce attestation checkpoint ancestry (leanEthereum#833)

* fix: enforce attestation checkpoint ancestry

* fix: address review findings on attestation checkpoint ancestry

- document the ancestry check rationale and list it in the
  validation contract
- expand the ancestry helper docstring with its conservative
  failure behavior
- correct the chain-match docstring wording for three checkpoints
- add positive deep-chain and unknown-parent ancestry unit tests
- cover the head zero-hash and head out-of-bounds rejection branches
- add language-neutral rejection reasons for the ancestry asserts
- add consensus vectors: sibling-fork head and source rejection in
  gossip validation, off-canonical head vote excluded from
  justification in state transition

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Thomas Coratger <60488569+tcoratger@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

packages/testing/src/consensus_testing/rejection.py

@@ -28,6 +28,14 @@
     ("Source checkpoint slot mismatch", RejectionReason.SOURCE_SLOT_MISMATCH),
     ("Target checkpoint slot mismatch", RejectionReason.TARGET_SLOT_MISMATCH),
     ("Head checkpoint slot mismatch", RejectionReason.HEAD_SLOT_MISMATCH),
+    (
+        "Source checkpoint must be ancestor of target",
+        RejectionReason.SOURCE_NOT_ANCESTOR_OF_TARGET,
+    ),
+    (
+        "Target checkpoint must be ancestor of head",
+        RejectionReason.TARGET_NOT_ANCESTOR_OF_HEAD,
+    ),
     ("Attestation too far in future", RejectionReason.ATTESTATION_TOO_FAR_IN_FUTURE),
     ("not found in state", RejectionReason.VALIDATOR_NOT_IN_STATE),
     ("Validator index out of range", RejectionReason.VALIDATOR_INDEX_OUT_OF_RANGE),

src/lean_spec/spec/forks/lstar/fork_choice.py

@@ -36,6 +36,32 @@
 class ForkChoiceMixin(LstarSpecBase):
     """Fork choice and store maintenance for the lstar fork."""
 
+    def _checkpoint_is_ancestor(
+        self,
+        store: LstarStore,
+        ancestor: Checkpoint,
+        descendant: Checkpoint,
+    ) -> bool:
+        """
+        Return whether the ancestor checkpoint lies on the descendant's parent chain.
+
+        Walks parent links from the descendant down to the ancestor's slot.
+        Conservative: a skipped slot or a block missing from the store yields False.
+        """
+        if ancestor.slot > descendant.slot:
+            return False
+
+        current_root = descendant.root
+        while current_root in store.blocks:
+            current_block = store.blocks[current_root]
+            if current_block.slot == ancestor.slot:
+                return current_root == ancestor.root
+            if current_block.slot < ancestor.slot:
+                return False
+            current_root = current_block.parent_root
+
+        return False
+
     def create_store(
         self,
         state: SpecStateType,
@@ -146,7 +172,8 @@ def validate_attestation(self, store: LstarStore, attestation_data: AttestationD
             2. A vote cannot span backwards in time (source > target).
             3. The head must be at least as recent as source and target.
             4. Checkpoint slots must match the actual block slots.
-            5. The vote's slot must have started locally (a small disparity margin is allowed).
+            5. Source, target, and head must lie on one parent chain.
+            6. The vote's slot must have started locally (a small disparity margin is allowed).
 
         Raises:
             AssertionError: If attestation fails validation.
@@ -189,6 +216,17 @@ def validate_attestation(self, store: LstarStore, attestation_data: AttestationD
         assert target_block.slot == target_checkpoint.slot, "Target checkpoint slot mismatch"
         assert head_block.slot == head_checkpoint.slot, "Head checkpoint slot mismatch"
 
+        # Ancestry Check
+        #
+        # Why: fork-choice weight accrues to every ancestor of the attested head.
+        # A sibling head would steer that weight onto a non-canonical branch.
+        assert self._checkpoint_is_ancestor(store, source_checkpoint, target_checkpoint), (
+            "Source checkpoint must be ancestor of target"
+        )
+        assert self._checkpoint_is_ancestor(store, target_checkpoint, head_checkpoint), (
+            "Target checkpoint must be ancestor of head"
+        )
+
         # Time Check
         #
         # Honest validators emit votes only after their slot has begun.

src/lean_spec/spec/forks/lstar/rejection_reason.py

@@ -70,6 +70,12 @@ class RejectionReason(StrEnum):
     HEAD_SLOT_MISMATCH = "HEAD_SLOT_MISMATCH"
     """The head checkpoint slot disagrees with the referenced block."""
 
+    SOURCE_NOT_ANCESTOR_OF_TARGET = "SOURCE_NOT_ANCESTOR_OF_TARGET"
+    """The attestation source checkpoint is not an ancestor of its target."""
+
+    TARGET_NOT_ANCESTOR_OF_HEAD = "TARGET_NOT_ANCESTOR_OF_HEAD"
+    """The attestation target checkpoint is not an ancestor of its head."""
+
     ATTESTATION_TOO_FAR_IN_FUTURE = "ATTESTATION_TOO_FAR_IN_FUTURE"
     """The attestation slot is beyond the store's acceptance horizon."""
 

src/lean_spec/spec/forks/lstar/state_transition.py

@@ -31,39 +31,47 @@ def attestation_data_matches_chain(
     historical_block_hashes: Sequence[Bytes32],
 ) -> bool:
     """
-    Check that source and target checkpoints point to blocks on a chain.
+    Check that attestation checkpoints point to blocks on a chain.
 
     Args:
         attestation_data: The attestation being validated.
         historical_block_hashes: Chain view indexed by slot.
             Empty slots carry the zero hash.
 
     Returns:
-        True when both checkpoint roots match the chain at their slot.
-        False when either root is the zero hash.
-        False when either checkpoint slot is past the end of the chain view.
+        True when all checkpoint roots match the chain at their slot.
+        False when any root is the zero hash.
+        False when any checkpoint slot is past the end of the chain view.
     """
     # Reject zero-hash checkpoints up front.
     #
     # Empty slots carry the zero hash on the chain.
     # A vote whose recorded root equals the zero hash is meaningless.
-    if attestation_data.source.root == ZERO_HASH or attestation_data.target.root == ZERO_HASH:
+    if (
+        attestation_data.source.root == ZERO_HASH
+        or attestation_data.target.root == ZERO_HASH
+        or attestation_data.head.root == ZERO_HASH
+    ):
         return False
 
     # Reject checkpoints whose slot is beyond the chain view.
     #
     # Without this guard, indexed access raises IndexError.
     source_slot = int(attestation_data.source.slot)
     target_slot = int(attestation_data.target.slot)
+    head_slot = int(attestation_data.head.slot)
     if source_slot >= len(historical_block_hashes):
         return False
     if target_slot >= len(historical_block_hashes):
         return False
+    if head_slot >= len(historical_block_hashes):
+        return False
 
-    # Both checkpoint roots must match the chain at their slot.
+    # All checkpoint roots must match the chain at their slot.
     return (
         attestation_data.source.root == historical_block_hashes[source_slot]
         and attestation_data.target.root == historical_block_hashes[target_slot]
+        and attestation_data.head.root == historical_block_hashes[head_slot]
     )
 
 

tests/consensus/lstar/fc/test_gossip_attestation_validation.py

@@ -790,3 +790,105 @@ def test_attestation_unknown_source_block_rejected(
             ),
         ],
     )
+
+
+def test_attestation_head_on_sibling_fork_rejected(
+    fork_choice_test: ForkChoiceTestFiller,
+) -> None:
+    """
+    Attestation whose head sits on a sibling fork of the target is rejected.
+
+    Scenario
+    --------
+    Build a common base at slot 1.
+    Create two competing blocks from that same base at slots 2 and 3.
+    Distinct slots give the siblings distinct roots so they never collide.
+    Vote with source on the base, target on the slot-2 block, and head on the
+    slot-3 block.
+    Every slot and availability check passes, yet target and head diverge.
+
+    Expected:
+        - Validation fails with "Target checkpoint must be ancestor of head"
+    """
+    fork_choice_test(
+        steps=[
+            BlockStep(
+                block=BlockSpec(slot=Slot(1), label="base"),
+                checks=StoreChecks(head_slot=Slot(1)),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(2), parent_label="base", label="fork_left"),
+                checks=StoreChecks(head_slot=Slot(2)),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(3), parent_label="base", label="fork_right"),
+            ),
+            AttestationStep(
+                attestation=GossipAttestationSpec(
+                    validator_index=ValidatorIndex(1),
+                    slot=Slot(3),
+                    target_slot=Slot(2),
+                    target_root_label="fork_left",
+                    head_slot=Slot(3),
+                    head_root_label="fork_right",
+                    source_root_label="base",
+                    valid_signature=False,
+                ),
+                valid=False,
+                expected_error="Target checkpoint must be ancestor of head",
+            ),
+        ],
+    )
+
+
+def test_attestation_source_on_sibling_fork_rejected(
+    fork_choice_test: ForkChoiceTestFiller,
+) -> None:
+    """
+    Attestation whose source sits on a sibling fork of the target is rejected.
+
+    Scenario
+    --------
+    Build a common base at slot 1.
+    Create a slot-2 block on one branch from that base.
+    Create a slot-3 block on a competing branch from that same base.
+    Distinct slots give the branches distinct roots so they never collide.
+    Extend the competing branch with a slot-4 block.
+    Vote with source on the abandoned slot-2 block, target on the slot-4 block,
+    and head on that same slot-4 block.
+    Source slot precedes the target slot, yet source lies off the target chain.
+
+    Expected:
+        - Validation fails with "Source checkpoint must be ancestor of target"
+    """
+    fork_choice_test(
+        steps=[
+            BlockStep(
+                block=BlockSpec(slot=Slot(1), label="base"),
+                checks=StoreChecks(head_slot=Slot(1)),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(2), parent_label="base", label="fork_left"),
+                checks=StoreChecks(head_slot=Slot(2)),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(3), parent_label="base", label="fork_right"),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(4), parent_label="fork_right", label="fork_right_head"),
+            ),
+            AttestationStep(
+                attestation=GossipAttestationSpec(
+                    validator_index=ValidatorIndex(1),
+                    slot=Slot(4),
+                    target_slot=Slot(4),
+                    target_root_label="fork_right_head",
+                    head_root_label="fork_right_head",
+                    source_root_label="fork_left",
+                    valid_signature=False,
+                ),
+                valid=False,
+                expected_error="Source checkpoint must be ancestor of target",
+            ),
+        ],
+    )

tests/consensus/lstar/state_transition/test_justification.py

@@ -800,6 +800,64 @@ def test_attestation_with_target_root_not_in_historical_hashes_is_skipped(
     )
 
 
+def test_attestation_with_off_canonical_head_does_not_justify_target(
+    state_transition_test: StateTransitionTestFiller,
+) -> None:
+    """
+    Test that a vote with an off-canonical head cannot justify its target.
+
+    Scenario
+    --------
+    1. Start from genesis with 4 validators
+    2. Process block_1 at slot 1 and block_2 at slot 2 on the canonical chain
+    3. Force a supermajority attestation into block_2 whose source and target
+       match the canonical chain but whose head points to a sibling root at a
+       slot that already holds a canonical block
+
+    Expected Behavior
+    -----------------
+    1. The source and target roots match the canonical chain at their slots
+    2. The head root does not match the canonical block at its slot
+    3. The whole vote is skipped before any tally is recorded
+    4. latest_justified_slot stays at genesis with no pending votes
+    """
+    state_transition_test(
+        blocks=[
+            BlockSpec(slot=Slot(1), label="block_1"),
+            BlockSpec(
+                slot=Slot(2),
+                parent_label="block_1",
+                forced_attestations=[
+                    # Source genesis and target block_1 both sit on the
+                    # canonical chain.
+                    # The head root is a sibling at slot 1, where the canonical
+                    # block is block_1.
+                    # Threshold: 3 of 4 would justify slot 1 if the head matched.
+                    AggregatedAttestationSpec(
+                        validator_indices=[
+                            ValidatorIndex(0),
+                            ValidatorIndex(1),
+                            ValidatorIndex(2),
+                        ],
+                        slot=Slot(2),
+                        target_slot=Slot(1),
+                        target_root_label="block_1",
+                        head_root=Bytes32(b"\x99" * 32),
+                        head_slot=Slot(1),
+                    ),
+                ],
+            ),
+        ],
+        post=StateExpectation(
+            slot=Slot(2),
+            latest_justified_slot=Slot(0),
+            latest_finalized_slot=Slot(0),
+            justifications_roots=JustificationRoots(data=[]),
+            justifications_validators=JustificationValidators(data=[]),
+        ),
+    )
+
+
 def test_justification_clears_only_the_resolved_target_votes(
     state_transition_test: StateTransitionTestFiller,
 ) -> None: