feat(network): add shadowfork block hash filtering for peers (#18361)

Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>

crates/net/network/src/config.rs

@@ -6,6 +6,7 @@ use crate::{
     transactions::TransactionsManagerConfig,
     NetworkHandle, NetworkManager,
 };
+use alloy_primitives::B256;
 use reth_chainspec::{ChainSpecProvider, EthChainSpec, Hardforks};
 use reth_discv4::{Discv4Config, Discv4ConfigBuilder, NatResolver, DEFAULT_DISCOVERY_ADDRESS};
 use reth_discv5::NetworkStackId;
@@ -93,6 +94,9 @@ pub struct NetworkConfig<C, N: NetworkPrimitives = EthNetworkPrimitives> {
     /// This can be overridden to support custom handshake logic via the
     /// [`NetworkConfigBuilder`].
     pub handshake: Arc<dyn EthRlpxHandshake>,
+    /// List of block hashes to check for required blocks.
+    /// If non-empty, peers that don't have these blocks will be filtered out.
+    pub required_block_hashes: Vec<B256>,
 }
 
 // === impl NetworkConfig ===
@@ -220,6 +224,8 @@ pub struct NetworkConfigBuilder<N: NetworkPrimitives = EthNetworkPrimitives> {
     /// The Ethereum P2P handshake, see also:
     /// <https://github.com/ethereum/devp2p/blob/master/rlpx.md#initial-handshake>.
     handshake: Arc<dyn EthRlpxHandshake>,
+    /// List of block hashes to check for required blocks.
+    required_block_hashes: Vec<B256>,
 }
 
 impl NetworkConfigBuilder<EthNetworkPrimitives> {
@@ -260,6 +266,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
             transactions_manager_config: Default::default(),
             nat: None,
             handshake: Arc::new(EthHandshake::default()),
+            required_block_hashes: Vec::new(),
         }
     }
 
@@ -544,6 +551,12 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
         self
     }
 
+    /// Sets the required block hashes for peer filtering.
+    pub fn required_block_hashes(mut self, hashes: Vec<B256>) -> Self {
+        self.required_block_hashes = hashes;
+        self
+    }
+
     /// Sets the block import type.
     pub fn block_import(mut self, block_import: Box<dyn BlockImport<N::NewBlockPayload>>) -> Self {
         self.block_import = Some(block_import);
@@ -606,6 +619,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
             transactions_manager_config,
             nat,
             handshake,
+            required_block_hashes,
         } = self;
 
         let head = head.unwrap_or_else(|| Head {
@@ -674,6 +688,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
             transactions_manager_config,
             nat,
             handshake,
+            required_block_hashes,
         }
     }
 }

crates/net/network/src/lib.rs

@@ -140,6 +140,7 @@ mod listener;
 mod manager;
 mod metrics;
 mod network;
+mod required_block_filter;
 mod session;
 mod state;
 mod swarm;

crates/net/network/src/manager.rs

@@ -29,6 +29,7 @@ use crate::{
     peers::PeersManager,
     poll_nested_stream_with_budget,
     protocol::IntoRlpxSubProtocol,
+    required_block_filter::RequiredBlockFilter,
     session::SessionManager,
     state::NetworkState,
     swarm::{Swarm, SwarmEvent},
@@ -250,6 +251,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
             transactions_manager_config: _,
             nat,
             handshake,
+            required_block_hashes,
         } = config;
 
         let peers_manager = PeersManager::new(peers_config);
@@ -335,6 +337,12 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
             nat,
         );
 
+        // Spawn required block peer filter if configured
+        if !required_block_hashes.is_empty() {
+            let filter = RequiredBlockFilter::new(handle.clone(), required_block_hashes);
+            filter.spawn();
+        }
+
         Ok(Self {
             swarm,
             handle,

crates/net/network/src/required_block_filter.rs

@@ -0,0 +1,179 @@
+//! Required block peer filtering implementation.
+//!
+//! This module provides functionality to filter out peers that don't have
+//! specific required blocks (primarily used for shadowfork testing).
+
+use alloy_primitives::B256;
+use futures::StreamExt;
+use reth_eth_wire_types::{GetBlockHeaders, HeadersDirection};
+use reth_network_api::{
+    NetworkEvent, NetworkEventListenerProvider, PeerRequest, Peers, ReputationChangeKind,
+};
+use tokio::sync::oneshot;
+use tracing::{debug, info, trace};
+
+/// Task that filters peers based on required block hashes.
+///
+/// This task listens for new peer sessions and checks if they have the required
+/// block hashes. Peers that don't have these blocks are banned.
+pub struct RequiredBlockFilter<N> {
+    /// Network handle for listening to events and managing peer reputation.
+    network: N,
+    /// List of block hashes that peers must have to be considered valid.
+    block_hashes: Vec<B256>,
+}
+
+impl<N> RequiredBlockFilter<N>
+where
+    N: NetworkEventListenerProvider + Peers + Clone + Send + Sync + 'static,
+{
+    /// Creates a new required block peer filter.
+    pub const fn new(network: N, block_hashes: Vec<B256>) -> Self {
+        Self { network, block_hashes }
+    }
+
+    /// Spawns the required block peer filter task.
+    ///
+    /// This task will run indefinitely, monitoring new peer sessions and filtering
+    /// out peers that don't have the required blocks.
+    pub fn spawn(self) {
+        if self.block_hashes.is_empty() {
+            debug!(target: "net::filter", "No required block hashes configured, skipping peer filtering");
+            return;
+        }
+
+        info!(target: "net::filter", "Starting required block peer filter with {} block hashes", self.block_hashes.len());
+
+        tokio::spawn(async move {
+            self.run().await;
+        });
+    }
+
+    /// Main loop for the required block peer filter.
+    async fn run(self) {
+        let mut event_stream = self.network.event_listener();
+
+        while let Some(event) = event_stream.next().await {
+            if let NetworkEvent::ActivePeerSession { info, messages } = event {
+                let peer_id = info.peer_id;
+                debug!(target: "net::filter", "New peer session established: {}", peer_id);
+
+                // Spawn a task to check this peer's blocks
+                let network = self.network.clone();
+                let block_hashes = self.block_hashes.clone();
+
+                tokio::spawn(async move {
+                    Self::check_peer_blocks(network, peer_id, messages, block_hashes).await;
+                });
+            }
+        }
+    }
+
+    /// Checks if a peer has the required blocks and bans them if not.
+    async fn check_peer_blocks(
+        network: N,
+        peer_id: reth_network_api::PeerId,
+        messages: reth_network_api::PeerRequestSender<PeerRequest<N::Primitives>>,
+        block_hashes: Vec<B256>,
+    ) {
+        for block_hash in block_hashes {
+            trace!(target: "net::filter", "Checking if peer {} has block {}", peer_id, block_hash);
+
+            // Create a request for block headers
+            let request = GetBlockHeaders {
+                start_block: block_hash.into(),
+                limit: 1,
+                skip: 0,
+                direction: HeadersDirection::Rising,
+            };
+
+            let (tx, rx) = oneshot::channel();
+            let peer_request = PeerRequest::GetBlockHeaders { request, response: tx };
+
+            // Send the request to the peer
+            if let Err(e) = messages.try_send(peer_request) {
+                debug!(target: "net::filter", "Failed to send block header request to peer {}: {:?}", peer_id, e);
+                continue;
+            }
+
+            // Wait for the response
+            let response = match rx.await {
+                Ok(response) => response,
+                Err(e) => {
+                    debug!(
+                        target: "net::filter",
+                        "Channel error getting block {} from peer {}: {:?}",
+                        block_hash, peer_id, e
+                    );
+                    continue;
+                }
+            };
+
+            let headers = match response {
+                Ok(headers) => headers,
+                Err(e) => {
+                    debug!(target: "net::filter", "Error getting block {} from peer {}: {:?}", block_hash, peer_id, e);
+                    // Ban the peer if they fail to respond properly
+                    network.reputation_change(peer_id, ReputationChangeKind::BadProtocol);
+                    return;
+                }
+            };
+
+            if headers.0.is_empty() {
+                info!(
+                    target: "net::filter",
+                    "Peer {} does not have required block {}, banning",
+                    peer_id, block_hash
+                );
+                network.reputation_change(peer_id, ReputationChangeKind::BadProtocol);
+                return; // No need to check more blocks if one is missing
+            }
+
+            trace!(target: "net::filter", "Peer {} has required block {}", peer_id, block_hash);
+        }
+
+        debug!(target: "net::filter", "Peer {} has all required blocks", peer_id);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_primitives::{b256, B256};
+    use reth_network_api::noop::NoopNetwork;
+
+    #[test]
+    fn test_required_block_filter_creation() {
+        let network = NoopNetwork::default();
+        let block_hashes = vec![
+            b256!("0x1111111111111111111111111111111111111111111111111111111111111111"),
+            b256!("0x2222222222222222222222222222222222222222222222222222222222222222"),
+        ];
+
+        let filter = RequiredBlockFilter::new(network, block_hashes.clone());
+        assert_eq!(filter.block_hashes.len(), 2);
+        assert_eq!(filter.block_hashes, block_hashes);
+    }
+
+    #[test]
+    fn test_required_block_filter_empty_hashes_does_not_spawn() {
+        let network = NoopNetwork::default();
+        let block_hashes = vec![];
+
+        let filter = RequiredBlockFilter::new(network, block_hashes);
+        // This should not panic and should exit early when spawn is called
+        filter.spawn();
+    }
+
+    #[tokio::test]
+    async fn test_required_block_filter_with_mock_peer() {
+        // This test would require a more complex setup with mock network components
+        // For now, we ensure the basic structure is correct
+        let network = NoopNetwork::default();
+        let block_hashes = vec![B256::default()];
+
+        let filter = RequiredBlockFilter::new(network, block_hashes);
+        // Verify the filter can be created and basic properties are set
+        assert_eq!(filter.block_hashes.len(), 1);
+    }
+}

crates/node/core/src/args/network.rs

@@ -1,5 +1,6 @@
 //! clap [Args](clap::Args) for network related arguments.
 
+use alloy_primitives::B256;
 use std::{
     net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
     ops::Not,
@@ -178,6 +179,11 @@ pub struct NetworkArgs {
         help = "Transaction propagation mode (sqrt, all, max:<number>)"
     )]
     pub propagation_mode: TransactionPropagationMode,
+
+    /// Comma separated list of required block hashes.
+    /// Peers that don't have these blocks will be filtered out.
+    #[arg(long = "required-block-hashes", value_delimiter = ',')]
+    pub required_block_hashes: Vec<B256>,
 }
 
 impl NetworkArgs {
@@ -290,6 +296,7 @@ impl NetworkArgs {
                 self.discovery.port,
             ))
             .disable_tx_gossip(self.disable_tx_gossip)
+            .required_block_hashes(self.required_block_hashes.clone())
     }
 
     /// If `no_persist_peers` is false then this returns the path to the persistent peers file path.
@@ -363,6 +370,7 @@ impl Default for NetworkArgs {
             tx_propagation_policy: TransactionPropagationKind::default(),
             disable_tx_gossip: false,
             propagation_mode: TransactionPropagationMode::Sqrt,
+            required_block_hashes: vec![],
         }
     }
 }
@@ -650,4 +658,30 @@ mod tests {
 
         assert_eq!(args, default_args);
     }
+
+    #[test]
+    fn parse_required_block_hashes() {
+        let args = CommandParser::<NetworkArgs>::parse_from([
+            "reth",
+            "--required-block-hashes",
+            "0x1111111111111111111111111111111111111111111111111111111111111111,0x2222222222222222222222222222222222222222222222222222222222222222",
+        ])
+        .args;
+
+        assert_eq!(args.required_block_hashes.len(), 2);
+        assert_eq!(
+            args.required_block_hashes[0].to_string(),
+            "0x1111111111111111111111111111111111111111111111111111111111111111"
+        );
+        assert_eq!(
+            args.required_block_hashes[1].to_string(),
+            "0x2222222222222222222222222222222222222222222222222222222222222222"
+        );
+    }
+
+    #[test]
+    fn parse_empty_required_block_hashes() {
+        let args = CommandParser::<NetworkArgs>::parse_from(["reth"]).args;
+        assert!(args.required_block_hashes.is_empty());
+    }
 }

docs/vocs/docs/pages/cli/reth/node.mdx

@@ -245,6 +245,9 @@ Networking:
 
           [default: sqrt]
 
+      --required-block-hashes <REQUIRED_BLOCK_HASHES>
+          Comma separated list of required block hashes. Peers that don't have these blocks will be filtered out
+
 RPC:
       --http
           Enable the HTTP-RPC server

docs/vocs/docs/pages/cli/reth/p2p/body.mdx

@@ -203,6 +203,9 @@ Networking:
 
           [default: sqrt]
 
+      --required-block-hashes <REQUIRED_BLOCK_HASHES>
+          Comma separated list of required block hashes. Peers that don't have these blocks will be filtered out
+
 Datadir:
       --datadir <DATA_DIR>
           The path to the data dir for all reth files and subdirectories.

docs/vocs/docs/pages/cli/reth/p2p/header.mdx

@@ -203,6 +203,9 @@ Networking:
 
           [default: sqrt]
 
+      --required-block-hashes <REQUIRED_BLOCK_HASHES>
+          Comma separated list of required block hashes. Peers that don't have these blocks will be filtered out
+
 Datadir:
       --datadir <DATA_DIR>
           The path to the data dir for all reth files and subdirectories.

docs/vocs/docs/pages/cli/reth/stage/run.mdx

@@ -299,6 +299,9 @@ Networking:
 
           [default: sqrt]
 
+      --required-block-hashes <REQUIRED_BLOCK_HASHES>
+          Comma separated list of required block hashes. Peers that don't have these blocks will be filtered out
+
 Logging:
       --log.stdout.format <FORMAT>
           The format to use for logs written to stdout