Check known parent on rpc blob process (#5893)

* Check known parent on rpc blob process

* fix test

* Merge branch 'unstable' of https://github.com/sigp/lighthouse into blob-unknown-parent

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -3073,6 +3073,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlockError::BlockIsAlreadyKnown(block_root));
         }
 
+        // Reject RPC blobs referencing unknown parents. Otherwise we allow potentially invalid data
+        // into the da_checker, where invalid = descendant of invalid blocks.
+        // Note: blobs should have at least one item and all items have the same parent root.
+        if let Some(parent_root) = blobs
+            .iter()
+            .filter_map(|b| b.as_ref().map(|b| b.block_parent_root()))
+            .next()
+        {
+            if !self
+                .canonical_head
+                .fork_choice_read_lock()
+                .contains_block(&parent_root)
+            {
+                return Err(BlockError::ParentUnknown { parent_root });
+            }
+        }
+
         if let Some(event_handler) = self.event_handler.as_ref() {
             if event_handler.has_blob_sidecar_subscribers() {
                 for blob in blobs.iter().filter_map(|maybe_blob| maybe_blob.as_ref()) {
@@ -3122,6 +3139,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlockError::BlockIsAlreadyKnown(block_root));
         }
 
+        // Reject RPC columns referencing unknown parents. Otherwise we allow potentially invalid data
+        // into the da_checker, where invalid = descendant of invalid blocks.
+        // Note: custody_columns should have at least one item and all items have the same parent root.
+        if let Some(parent_root) = custody_columns.iter().map(|c| c.block_parent_root()).next() {
+            if !self
+                .canonical_head
+                .fork_choice_read_lock()
+                .contains_block(&parent_root)
+            {
+                return Err(BlockError::ParentUnknown { parent_root });
+            }
+        }
+
         let r = self
             .check_rpc_custody_columns_availability_and_import(slot, block_root, custody_columns)
             .await;

beacon_node/beacon_chain/tests/events.rs

@@ -62,13 +62,17 @@ async fn blob_sidecar_event_on_process_rpc_blobs() {
     let kzg = harness.chain.kzg.as_ref().unwrap();
     let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64);
 
-    let blob_1 = BlobSidecar::random_valid(&mut rng, kzg)
+    let mut blob_1 = BlobSidecar::random_valid(&mut rng, kzg).unwrap();
-        .map(Arc::new)
+    let mut blob_2 = BlobSidecar {
-        .unwrap();
-    let blob_2 = Arc::new(BlobSidecar {
         index: 1,
         ..BlobSidecar::random_valid(&mut rng, kzg).unwrap()
-    });
+    };
+    let parent_root = harness.chain.head().head_block_root();
+    blob_1.signed_block_header.message.parent_root = parent_root;
+    blob_2.signed_block_header.message.parent_root = parent_root;
+    let blob_1 = Arc::new(blob_1);
+    let blob_2 = Arc::new(blob_2);
+
     let blobs = FixedBlobSidecarList::from(vec![Some(blob_1.clone()), Some(blob_2.clone())]);
     let expected_sse_blobs = vec![
         SseBlobSidecar::from_blob_sidecar(blob_1.as_ref()),

beacon_node/network/src/sync/block_lookups/single_block_lookup.rs

@@ -1,4 +1,3 @@
-use super::common::ResponseType;
 use super::{BlockComponent, PeerId, SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS};
 use crate::sync::block_lookups::common::RequestState;
 use crate::sync::network_context::{
@@ -188,7 +187,6 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
             .state
             .peek_downloaded_data()
             .cloned();
-        let block_is_processed = self.block_request_state.state.is_processed();
         let request = R::request_state_mut(self);
 
         // Attempt to progress awaiting downloads
@@ -241,12 +239,7 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
         // Otherwise, attempt to progress awaiting processing
         // If this request is awaiting a parent lookup to be processed, do not send for processing.
         // The request will be rejected with unknown parent error.
-        //
+        } else if !awaiting_parent {
-        // TODO: The condition `block_is_processed || Block` can be dropped after checking for
-        // unknown parent root when import RPC blobs
-        } else if !awaiting_parent
-            && (block_is_processed || matches!(R::response_type(), ResponseType::Block))
-        {
             // maybe_start_processing returns Some if state == AwaitingProcess. This pattern is
             // useful to conditionally access the result data.
             if let Some(result) = request.get_state_mut().maybe_start_processing() {