Fix checkpoint sync and verify all incoming signatures

lean_client/containers/src/state.rs

@@ -342,11 +342,6 @@ impl State {
 
         let state = self.process_block_header(block)?;
 
-        ensure!(
-            !AggregatedAttestation::has_duplicate_data(&block.body.attestations),
-            "block contains duplicate attestation data"
-        );
-
         state.process_attestations(&block.body.attestations)
     }
 

lean_client/fork_choice/src/handlers.rs

@@ -1,4 +1,4 @@
-use anyhow::{Result, anyhow, bail, ensure};
+use anyhow::{Context, Result, anyhow, bail, ensure};
 use containers::{
     AttestationData, SignatureKey, SignedAggregatedAttestation, SignedAttestation,
     SignedBlockWithAttestation,
@@ -62,17 +62,25 @@ fn validate_attestation_data(store: &Store, data: &AttestationData) -> Result<()
         data.target.slot.0
     );
 
-    // Validate checkpoint slots match block slots
-    // Per devnet-2, store.blocks now contains Block (not SignedBlockWithAttestation)
+    // Validate checkpoint slots match block slots.
+    // Skip the source slot-match when the root is the store's known justified or
+    // finalized checkpoint: after checkpoint sync the anchor block sits at
+    // anchor_slot in store.blocks but the checkpoint carries the actual historical
+    // slot from the downloaded state (e.g. 100994 vs anchor 101002).
     let source_block = &store.blocks[&data.source.root];
     let target_block = &store.blocks[&data.target.root];
 
-    ensure!(
-        source_block.slot == data.source.slot,
-        "Source checkpoint slot mismatch: checkpoint {} vs block {}",
-        data.source.slot.0,
-        source_block.slot.0
-    );
+    let source_is_trusted_checkpoint = data.source.root == store.latest_justified.root
+        || data.source.root == store.latest_finalized.root;
+
+    if !source_is_trusted_checkpoint {
+        ensure!(
+            source_block.slot == data.source.slot,
+            "Source checkpoint slot mismatch: checkpoint {} vs block {}",
+            data.source.slot.0,
+            source_block.slot.0
+        );
+    }
 
     ensure!(
         target_block.slot == data.target.slot,
@@ -152,8 +160,35 @@ pub fn on_gossip_attestation(
         });
     })?;
 
-    // Store signature for later lookup during block building
+    // Verify individual XMSS signature against the validator's public key.
+    // State is available: the pending-block check above confirmed target.root is in the store,
+    // and states are stored 1:1 with blocks in process_block_internal.
+    let key_state = store
+        .states
+        .get(&attestation_data.target.root)
+        .ok_or_else(|| anyhow!("no state for target block {}", attestation_data.target.root))?;
+
+    ensure!(
+        validator_id < key_state.validators.len_u64(),
+        "validator {} out of range (max {})",
+        validator_id,
+        key_state.validators.len_u64()
+    );
+
+    let pubkey = key_state
+        .validators
+        .get(validator_id)
+        .map(|v| v.pubkey.clone())
+        .map_err(|e| anyhow!("{e}"))?;
+
     let data_root = attestation_data.hash_tree_root();
+
+    signed_attestation
+        .signature
+        .verify(&pubkey, attestation_data.slot.0 as u32, data_root)
+        .context("individual attestation signature verification failed")?;
+
+    // Store verified signature for later lookup during block building
     let sig_key = SignatureKey::new(signed_attestation.validator_id, data_root);
     store
         .gossip_signatures
@@ -281,14 +316,7 @@ pub fn on_attestation(
 /// `latest_new_aggregated_payloads`. At interval 3, these are merged with
 /// `latest_known_aggregated_payloads` (from blocks) to compute safe target.
 ///
-/// # Signature Verification Strategy (TODO for production)
-///
-/// Currently, this function validates attestation data but does NOT verify the
-/// aggregated XMSS signature. This is intentional for devnet-3 performance testing.
-///
-/// For production, signature verification should be added:
-/// 1. Verify the `AggregatedSignatureProof` against the aggregation bits
-/// 2. Consider async/batched verification for throughput
+/// Verifies the aggregated XMSS proof against participant public keys before storing.
 #[inline]
 pub fn on_aggregated_attestation(
     store: &mut Store,
@@ -310,7 +338,6 @@ pub fn on_aggregated_attestation(
     }
 
     // Validate attestation data (slot bounds, target validity, etc.)
-    // TODO(production): Add signature verification here or in caller
     validate_attestation_data(store, &attestation_data)?;
 
     // Store attestation data indexed by hash for later extraction
@@ -319,7 +346,38 @@ pub fn on_aggregated_attestation(
         .attestation_data_by_root
         .insert(data_root, attestation_data.clone());
 
-    // Per leanSpec: Store the proof in latest_new_aggregated_payloads
+    // Verify aggregated XMSS proof against participant public keys.
+    // State is available: the pending-block check above confirmed target.root is in the store,
+    // and states are stored 1:1 with blocks in process_block_internal.
+    let key_state = store
+        .states
+        .get(&attestation_data.target.root)
+        .ok_or_else(|| anyhow!("no state for target block {}", attestation_data.target.root))?;
+
+    // Guard before calling to_validator_indices() which panics on an empty bitfield.
+    ensure!(
+        proof.participants.0.iter().any(|b| *b),
+        "aggregated attestation has empty participants bitfield"
+    );
+
+    let validator_ids = proof.participants.to_validator_indices();
+
+    let public_keys = validator_ids
+        .iter()
+        .map(|&id| {
+            key_state
+                .validators
+                .get(id)
+                .map(|v| v.pubkey.clone())
+                .map_err(Into::into)
+        })
+        .collect::<Result<Vec<_>>>()?;
+
+    proof
+        .verify(public_keys, data_root, attestation_data.slot.0 as u32)
+        .context("aggregated attestation proof verification failed")?;
+
+    // Per leanSpec: Store the verified proof in latest_new_aggregated_payloads
     // Each participating validator gets an entry via their SignatureKey
     for (bit_idx, bit) in proof.participants.0.iter().enumerate() {
         if *bit {
@@ -452,7 +510,12 @@ fn process_block_internal(
         "Processing block - parent state info"
     );
 
-    // Execute state transition to get post-state
+    // Verify block signatures against parent state before executing the state transition.
+    // If any signature is invalid the error propagates and the block is rejected;
+    // it never enters store.blocks or store.states.
+    signed_block.verify_signatures(state.clone())?;
+
+    // Execute state transition to get post-state (signatures verified above)
     let new_state = state.state_transition(signed_block.clone(), true)?;
 
     // Debug: Log new state checkpoints after transition
@@ -516,6 +579,7 @@ fn process_block_internal(
             "Store finalized checkpoint updated!"
         );
         store.latest_finalized = new_state.latest_finalized.clone();
+        store.finalized_ever_updated = true;
         METRICS.get().map(|metrics| {
             let Some(slot) = new_state.latest_finalized.slot.0.try_into().ok() else {
                 warn!("unable to set latest_finalized slot in metrics");
@@ -530,6 +594,7 @@ fn process_block_internal(
             .0
             .saturating_sub(STATE_PRUNE_BUFFER);
         store.states.retain(|_, state| state.slot.0 >= keep_from);
+        store.blocks.retain(|_, block| block.slot.0 >= keep_from);
     }
 
     if !justified_updated && !finalized_updated {
@@ -583,9 +648,9 @@ fn process_block_internal(
             .set(store.latest_known_aggregated_payloads.len() as i64);
     });
 
-    // Process each aggregated attestation's validators for fork choice
-    // Signature verification is done in verify_signatures() before on_block()
-    // Per Devnet-2, we process attestation data directly (not SignedAttestation)
+    // Process each aggregated attestation's validators for fork choice.
+    // Signatures have already been verified above via verify_signatures().
+    // Per Devnet-2, we process attestation data directly (not SignedAttestation).
     for aggregated_attestation in aggregated_attestations.into_iter() {
         let validator_ids: Vec<u64> = aggregated_attestation
             .aggregation_bits

lean_client/fork_choice/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod block_cache;
 pub mod handlers;
 pub mod store;
+pub mod sync_state;
 
 // dirty hack to avoid issues compiling grandine dependencies. by default, bls
 // crate has no features enabled, and thus compilation fails (as exactly one

lean_client/fork_choice/src/store.rs

@@ -2,9 +2,9 @@ use std::collections::{HashMap, HashSet};
 
 use anyhow::{Result, anyhow, ensure};
 use containers::{
-    AggregatedSignatureProof, Attestation, AttestationData, Block, BlockHeader, Checkpoint, Config,
-    SignatureKey, SignedAggregatedAttestation, SignedAttestation, SignedBlockWithAttestation, Slot,
-    State,
+    AggregatedSignatureProof, Attestation, AttestationData, Block, BlockHeader,
+    Checkpoint, Config, SignatureKey, SignedAggregatedAttestation, SignedAttestation,
+    SignedBlockWithAttestation, Slot, State,
 };
 use metrics::{METRICS, set_gauge_u64};
 use ssz::{H256, SszHash};
@@ -34,12 +34,22 @@ pub struct Store {
     pub latest_finalized: Checkpoint,
 
     /// Set to `true` the first time `on_block` drives a justified checkpoint
-    /// update beyond the initial anchor value. Validator duties (attestation,
-    /// block proposal) must not run while this is `false` — the store's
-    /// `latest_justified` is still the placeholder anchor checkpoint and using
-    /// it as an attestation source would produce wrong source checkpoints.
+    /// update beyond the initial checkpoint-sync value. Validator duties must
+    /// not run while this is `false` — the node has not yet observed real
+    /// justification progress and its attestations would reference a stale source.
     pub justified_ever_updated: bool,
 
+    /// Set to `true` the first time `on_block` drives a finalized checkpoint
+    /// update beyond the initial anchor value.
+    ///
+    /// The `/states/finalized` endpoint must return 503 while this is `false`.
+    /// A checkpoint-synced node that has not yet seen real finalization holds
+    /// the anchor block (head slot, not finalized slot) as `latest_finalized`.
+    /// Serving that state poisons downstream checkpoint syncs: the receiving
+    /// node anchors at the head slot, which exceeds the network's justified
+    /// slot, causing the justified-ever-updated gate to never fire.
+    pub finalized_ever_updated: bool,
+
     pub blocks: HashMap<H256, Block>,
 
     pub states: HashMap<H256, State>,
@@ -170,9 +180,11 @@ pub fn get_forkchoice_store(
     let block_slot = block.slot;
 
     // Compute block root differently for genesis vs checkpoint sync:
-    // - Genesis (slot 0): Use block.hash_tree_root() directly
-    // - Checkpoint sync (slot > 0): Use BlockHeader from state.latest_block_header
-    //   because we have the correct body_root there but may have synthetic empty body in Block
+    // - Genesis (slot 0): Use block.hash_tree_root() directly — block and state are consistent.
+    // - Checkpoint sync (slot > 0): Reconstruct BlockHeader from state.latest_block_header,
+    //   using anchor_state.hash_tree_root() as state_root.  This guarantees the root stored
+    //   as the key in store.blocks / store.states is the canonical one committed to by the
+    //   downloaded state, independent of what the real block's state_root field contains.
     let block_root = if block_slot.0 == 0 {
         block.hash_tree_root()
     } else {
@@ -186,25 +198,18 @@ pub fn get_forkchoice_store(
         block_header.hash_tree_root()
     };
 
-    // Per checkpoint sync: always use anchor block's root and slot for checkpoints.
-    // The original checkpoint roots point to blocks that don't exist in our store.
-    // We only have the anchor block, so both root and slot must refer to it.
-    //
-    // Using the state's justified.slot with the anchor root creates an inconsistency:
-    // validate_attestation_data requires store.blocks[source.root].slot == source.slot,
-    // which fails when the chain has progressed beyond the last justified block
-    // (e.g., state downloaded at slot 2291, last justified at slot 2285).
-    //
-    // The first real justification event from on_block will replace these values
-    // with the correct ones, so the anchor slot is only used for the initial period.
+    // Per leanSpec: substitute anchor_root for the checkpoint roots (the
+    // historical justified/finalized blocks are not in our store), but keep
+    // the actual slots from the downloaded state.  validate_attestation_data
+    // skips the block-slot match when the source root is a known checkpoint.
     let latest_justified = Checkpoint {
         root: block_root,
-        slot: block_slot,
+        slot: anchor_state.latest_justified.slot,
     };
 
     let latest_finalized = Checkpoint {
         root: block_root,
-        slot: block_slot,
+        slot: anchor_state.latest_finalized.slot,
     };
 
     // Store the original anchor_state - do NOT modify it
@@ -218,8 +223,17 @@ pub fn get_forkchoice_store(
         latest_justified,
         latest_finalized,
         justified_ever_updated: false,
-        blocks: [(block_root, block)].into(),
-        states: [(block_root, anchor_state)].into(),
+        finalized_ever_updated: false,
+        blocks: {
+            let mut m = HashMap::new();
+            m.insert(block_root, block);
+            m
+        },
+        states: {
+            let mut m = HashMap::new();
+            m.insert(block_root, anchor_state);
+            m
+        },
         latest_known_attestations: HashMap::new(),
         latest_new_attestations: HashMap::new(),
         gossip_signatures: HashMap::new(),

lean_client/fork_choice/src/sync_state.rs

@@ -0,0 +1,25 @@
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum SyncState {
+    #[default]
+    Idle,
+    Syncing,
+    Synced,
+}
+
+impl SyncState {
+    pub fn accepts_gossip(self) -> bool {
+        matches!(self, SyncState::Syncing | SyncState::Synced)
+    }
+
+    pub fn is_idle(self) -> bool {
+        self == SyncState::Idle
+    }
+
+    pub fn is_syncing(self) -> bool {
+        self == SyncState::Syncing
+    }
+
+    pub fn is_synced(self) -> bool {
+        self == SyncState::Synced
+    }
+}

lean_client/http_api/src/handlers.rs

@@ -23,6 +23,10 @@ pub async fn health() -> impl IntoResponse {
 pub async fn states_finalized(State(store): State<SharedStore>) -> Result<Response, StatusCode> {
     let store = store.read();
 
+    if !store.finalized_ever_updated {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+
     let finalized_root = store.latest_finalized.root;
 
     let state = store