fix(consensus): validate block retrieval responses

aptos-core/consensus/consensus-types/src/block_retrieval.rs

@@ -160,8 +160,11 @@ impl BlockRetrievalResponse {
                     commit_block_id
                 )
             })?;
+            let allows_epoch_change_suffix_number = ledger_info.ends_epoch() &&
+                ledger_info.commit_info().epoch_block_info().is_some() &&
+                *block_number >= ledger_info.block_number();
             ensure!(
-                *block_number == ledger_info.block_number(),
+                *block_number == ledger_info.block_number() || allows_epoch_change_suffix_number,
                 "ledger info block number mismatch for block {}: expected {}, got {}",
                 commit_block_id,
                 block_number,
@@ -176,27 +179,53 @@ impl BlockRetrievalResponse {
         retrieval_request: BlockRetrievalRequest,
         sig_verifier: &ValidatorVerifier,
     ) -> anyhow::Result<()> {
+        ensure!(
+            !matches!(
+                self.status,
+                BlockRetrievalStatus::Succeeded | BlockRetrievalStatus::SucceededWithTarget
+            ) || !self.blocks.is_empty(),
+            "successful block retrieval response must not be empty",
+        );
         ensure!(
             self.status != BlockRetrievalStatus::Succeeded ||
                 self.blocks.len() as u64 == retrieval_request.num_blocks(),
             "not enough blocks returned, expect {}, get {}",
             retrieval_request.num_blocks(),
             self.blocks.len(),
         );
-        self.blocks
-            .iter()
-            .try_fold(retrieval_request.block_id(), |expected_id, (block, _, _)| {
+        let mut expected_id = (retrieval_request.block_id() != HashValue::zero())
+            .then_some(retrieval_request.block_id());
+        for (block, _, _) in &self.blocks {
+            if let Some(expected_id) = expected_id {
                 ensure!(
                     block.id() == expected_id,
                     "blocks doesn't form a chain: expect {}, get {}",
                     expected_id,
                     block.id()
                 );
-                block.validate_signature(sig_verifier)?;
-                block.verify_well_formed()?;
-                Ok(block.parent_id())
-            })
-            .map(|_| ())
+            }
+            block.validate_signature(sig_verifier)?;
+            block.verify_well_formed()?;
+            expected_id = Some(block.parent_id());
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use gaptos::aptos_types::validator_verifier::random_validator_verifier;
+
+    #[test]
+    fn verify_rejects_empty_successful_response() {
+        let (_signers, verifier) = random_validator_verifier(1, None, false);
+        let request = BlockRetrievalRequest::new(HashValue::zero(), 0);
+
+        for status in [BlockRetrievalStatus::Succeeded, BlockRetrievalStatus::SucceededWithTarget] {
+            let response = BlockRetrievalResponse::new(status, vec![], vec![], vec![]);
+            assert!(response.verify(request.clone(), &verifier).is_err());
+        }
     }
 }
 

aptos-core/consensus/src/block_storage/sync_manager.rs

@@ -19,7 +19,7 @@ use crate::{
     payload_manager::TPayloadManager,
     persistent_liveness_storage::PersistentLivenessStorage,
 };
-use anyhow::{anyhow, bail};
+use anyhow::{anyhow, bail, ensure};
 use aptos_consensus_types::{
     block::Block,
     block_retrieval::{
@@ -408,7 +408,16 @@ impl BlockStore {
         }
 
         for (i, (block, _, _)) in blocks.iter().enumerate() {
-            assert_eq!(block.id(), quorum_certs[i].certified_block().id());
+            let qc = quorum_certs.get(i).ok_or_else(|| {
+                anyhow!("Missing quorum cert for retrieved block {} at index {}", block.id(), i,)
+            })?;
+            ensure!(
+                block.id() == qc.certified_block().id(),
+                "Retrieved block and quorum cert mismatch at index {}: block {}, qc certifies {}",
+                i,
+                block.id(),
+                qc.certified_block().id(),
+            );
             if let Some(payload) = block.payload() {
                 payload_manager.prefetch_payload_data(payload, block.timestamp_usecs());
             }
@@ -513,24 +522,37 @@ impl BlockStore {
             )
             .await?;
 
-        assert_eq!(
-            blocks.first().expect("blocks are empty").0.id(),
-            highest_quorum_cert.certified_block().id(),
+        let first_block = blocks
+            .first()
+            .ok_or_else(|| anyhow!("Empty block retrieval response during fast-forward sync"))?;
+        ensure!(
+            first_block.0.id() == highest_quorum_cert.certified_block().id(),
             "Expecting in the retrieval response, first block should be {}, but got {}",
             highest_quorum_cert.certified_block().id(),
-            blocks.first().expect("blocks are empty").0.id(),
+            first_block.0.id(),
         );
 
         let mut quorum_certs = vec![highest_quorum_cert.clone()];
         quorum_certs.extend(
             blocks.iter().take(blocks.len() - 1).map(|(block, _, _)| block.quorum_cert().clone()),
         );
-        assert_eq!(blocks.len(), quorum_certs.len());
+        ensure!(
+            blocks.len() == quorum_certs.len(),
+            "Retrieved block/quorum cert length mismatch: blocks {}, quorum_certs {}",
+            blocks.len(),
+            quorum_certs.len(),
+        );
         info!("[FastForwardSync] Fetched {} blocks. Requested num_blocks {}. Initial block hash {:?}, target block hash {:?}",
             blocks.len(), num_blocks, highest_quorum_cert.certified_block().id(), highest_commit_cert.commit_info().id()
         );
-        for (i, (block, _, _)) in blocks.iter().enumerate() {
-            assert_eq!(block.id(), quorum_certs[i].certified_block().id());
+        for (i, ((block, _, _), qc)) in blocks.iter().zip(quorum_certs.iter()).enumerate() {
+            ensure!(
+                block.id() == qc.certified_block().id(),
+                "Retrieved block and quorum cert mismatch at index {}: block {}, qc certifies {}",
+                i,
+                block.id(),
+                qc.certified_block().id(),
+            );
         }
         let block_numbers = blocks
             .iter()
@@ -1145,7 +1167,13 @@ impl BlockRetriever {
                         }
                     }
                     progress += batch.len() as u64;
-                    last_block_id = batch.last().expect("Batch should not be empty").0.parent_id();
+                    let last_block = batch.last().ok_or_else(|| {
+                        anyhow!(
+                            "Empty successful block retrieval batch starting from {}",
+                            last_block_id
+                        )
+                    })?;
+                    last_block_id = last_block.0.parent_id();
                     CUR_BLOCK_SYNC_BLOCK_SUM_GAUGE.with_label_values(&[]).add(batch.len() as i64);
                     result_blocks.extend(batch);
                     ledger_infos.extend(result.ledger_infos().clone());
@@ -1227,7 +1255,13 @@ impl BlockRetriever {
                         payload_manager.prefetch_payload_data(payload, block.timestamp_usecs());
                     }
                 }
-                let last_block_id = batch.last().expect("Batch should not be empty").0.parent_id();
+                let last_block_id = batch
+                    .last()
+                    .ok_or_else(|| {
+                        anyhow!("Empty successful epoch block retrieval batch for epoch {}", epoch)
+                    })?
+                    .0
+                    .parent_id();
                 result_blocks.extend(batch);
                 ledger_infos.extend(result.ledger_infos().clone());
                 quorum_certs.extend(result.quorum_certs().clone());

aptos-core/consensus/src/network.rs

@@ -41,7 +41,6 @@ use gaptos::{
     aptos_channels::{self, aptos_channel, message_queues::QueueStyle},
     aptos_config::network_id::{NetworkId, PeerNetworkId},
     aptos_consensus::counters,
-    aptos_crypto::HashValue,
     aptos_logger::prelude::*,
     aptos_network::{
         application::interface::{NetworkClient, NetworkClientInterface, NetworkServiceEvents},
@@ -257,32 +256,22 @@ impl NetworkSender {
             ConsensusMsg::BlockRetrievalResponse(resp) => *resp,
             _ => return Err(anyhow!("Invalid response to request")),
         };
-        // TODO(next release): Re-enable this after LedgerInfo exposes the committed suffix block
-        // number separately from the epoch-change anchor number. For an epoch-change suffix LI,
-        // `LedgerInfo::block_number()` intentionally returns `EpochBlockInfo::block_number`, while
-        // the retrieval response associates the LI's commit block ID with the suffix block number.
-        // The current verifier therefore rejects valid responses (for example, suffix block 2557
-        // carrying epoch-change anchor 2556), which prevents nodes from catching up across an epoch
-        // boundary during a rolling upgrade.
-        //
-        // response.verify_ledger_infos(&retrieval_request, &self.validators).map_err(|e| {
-        //     error!(
-        //         SecurityEvent::InvalidRetrievedBlock,
-        //         request_block_response = response,
-        //         error = ?e,
-        //     );
-        //     e
-        // })?;
-        if retrieval_request.block_id() != HashValue::zero() {
-            response.verify(retrieval_request, &self.validators).map_err(|e| {
-                error!(
-                    SecurityEvent::InvalidRetrievedBlock,
-                    request_block_response = response,
-                    error = ?e,
-                );
-                e
-            })?;
-        }
+        response.verify(retrieval_request.clone(), &self.validators).map_err(|e| {
+            error!(
+                SecurityEvent::InvalidRetrievedBlock,
+                request_block_response = response,
+                error = ?e,
+            );
+            e
+        })?;
+        response.verify_ledger_infos(&retrieval_request, &self.validators).map_err(|e| {
+            error!(
+                SecurityEvent::InvalidRetrievedBlock,
+                request_block_response = response,
+                error = ?e,
+            );
+            e
+        })?;
 
         Ok(response)
     }