diff --git a/Cargo.lock b/Cargo.lock index 3af7c652b99..b9b77db8d8a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4182,6 +4182,7 @@ dependencies = [ "hkdf", "hmac", "indexmap 2.14.0", + "ipnet", "lazy_static", "lru 0.16.4", "prometheus", diff --git a/Cargo.toml b/Cargo.toml index 97b345d7527..32e09f5b8f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -152,6 +152,7 @@ tempfile = "3.8" uuid = { version = "1.18.1", features = ["v4"] } tower-http = { version = "0.6.2", features = ["cors"] } indexmap = { version = "2.11.4" } +ipnet = "2.12.0" k256 = "0.13.4" anyhow = "1.0.86" diff --git a/cmd/ethrex/cli.rs b/cmd/ethrex/cli.rs index 73350d6bb1b..afe802de9b8 100644 --- a/cmd/ethrex/cli.rs +++ b/cmd/ethrex/cli.rs @@ -18,6 +18,7 @@ use ethrex_blockchain::{ use ethrex_common::types::{Block, DEFAULT_BUILDER_GAS_CEIL, Genesis, validate_block_body}; use ethrex_p2p::{ discovery::INITIAL_LOOKUP_INTERVAL_MS, + netrestrict::IpNet, peer_table::TARGET_PEERS, sync::{HistoryChain, SyncMode}, tx_broadcaster::BROADCAST_INTERVAL_MS, @@ -484,6 +485,18 @@ pub struct Options { env = "ETHREX_P2P_LOOKUP_INTERVAL" )] pub lookup_interval: f64, + #[arg( + long = "p2p.netrestrict", + value_parser = clap::value_parser!(IpNet), + value_name = "CIDR_LIST", + value_delimiter = ',', + num_args = 1.., + help = "Restrict P2P traffic to the given IP networks (comma separated CIDRs).", + long_help = "Comma separated IP networks in CIDR notation, e.g. 10.0.0.0/8,172.16.0.0/12. Nodes discovered outside these networks are ignored, bootnodes outside are dropped, and inbound TCP and UDP from outside is discarded. Meant for private devnets and other closed networks. Unrestricted when not set.", + help_heading = "P2P options", + env = "ETHREX_P2P_NETRESTRICT" + )] + pub netrestrict: Vec, #[arg( long = "builder.extra-data", default_value = get_minimal_client_version(), @@ -636,6 +649,7 @@ impl Default for Options { tx_broadcasting_time_interval: Default::default(), target_peers: Default::default(), lookup_interval: INITIAL_LOOKUP_INTERVAL_MS, + netrestrict: Default::default(), extra_data: get_minimal_client_version(), gas_limit: DEFAULT_BUILDER_GAS_CEIL, max_blobs_per_block: None, diff --git a/cmd/ethrex/initializers.rs b/cmd/ethrex/initializers.rs index 0c13f8b0f97..194a82138df 100644 --- a/cmd/ethrex/initializers.rs +++ b/cmd/ethrex/initializers.rs @@ -17,6 +17,7 @@ use ethrex_metrics::rpc::initialize_rpc_metrics; use ethrex_p2p::rlpx::initiator::RLPxInitiator; use ethrex_p2p::{ DiscoveryConfig, + netrestrict::NetRestrict, network::P2PContext, peer_handler::PeerHandler, peer_table::{PeerTable, PeerTableServer}, @@ -466,10 +467,16 @@ pub async fn init_network( let bootnodes = get_bootnodes(opts, network, datadir); + let netrestrict = NetRestrict::new(opts.netrestrict.clone()); + if !netrestrict.is_unrestricted() { + info!(%netrestrict, "P2P traffic restricted to the configured networks"); + } + let discovery_config = DiscoveryConfig { discv4_enabled: opts.discv4_enabled, discv5_enabled: opts.discv5_enabled, nat_extip_set: opts.nat_extip.is_some(), + netrestrict, }; ethrex_p2p::start_network(context, bootnodes, discovery_config, shared_local_node) @@ -926,8 +933,12 @@ pub async fn init_l1( record: local_node_record, })); - let peer_table = - PeerTableServer::spawn(local_p2p_node.node_id(), opts.target_peers, store.clone()); + let peer_table = PeerTableServer::spawn( + local_p2p_node.node_id(), + opts.target_peers, + store.clone(), + NetRestrict::new(opts.netrestrict.clone()), + ); // TODO: Check every module starts properly. let tracker = TaskTracker::new(); @@ -946,6 +957,7 @@ pub async fn init_l1( None, opts.tx_broadcasting_time_interval, opts.lookup_interval, + NetRestrict::new(opts.netrestrict.clone()), ) .expect("P2P context could not be created"); diff --git a/cmd/ethrex/l2/initializers.rs b/cmd/ethrex/l2/initializers.rs index 9c086e043d7..7adb1f8aefd 100644 --- a/cmd/ethrex/l2/initializers.rs +++ b/cmd/ethrex/l2/initializers.rs @@ -16,6 +16,7 @@ use ethrex_common::types::fee_config::{FeeConfig, L1FeeConfig, OperatorFeeConfig use ethrex_l2::sequencer::block_producer::{self, block_producer_protocol}; use ethrex_l2::sequencer::l1_committer::{self, l1_committer_protocol, regenerate_state}; use ethrex_p2p::{ + netrestrict::NetRestrict, network::P2PContext, peer_handler::PeerHandler, peer_table::PeerTableServer, @@ -303,6 +304,7 @@ pub async fn init_l2( local_p2p_node.node_id(), opts.node_opts.target_peers, store.clone(), + NetRestrict::new(opts.node_opts.netrestrict.clone()), ); let p2p_context = P2PContext::new( local_p2p_node.clone(), @@ -332,6 +334,7 @@ pub async fn init_l2( }), opts.node_opts.tx_broadcasting_time_interval, opts.node_opts.lookup_interval, + NetRestrict::new(opts.node_opts.netrestrict.clone()), ) .expect("P2P context could not be created"); let initiator = RLPxInitiator::spawn(p2p_context.clone()); diff --git a/crates/networking/p2p/Cargo.toml b/crates/networking/p2p/Cargo.toml index 6e964e965ec..bc110e68962 100644 --- a/crates/networking/p2p/Cargo.toml +++ b/crates/networking/p2p/Cargo.toml @@ -37,6 +37,7 @@ spawned-concurrency.workspace = true sha2.workspace = true futures.workspace = true indexmap.workspace = true +ipnet.workspace = true rustc-hash.workspace = true rocksdb = { workspace = true, optional = true } prometheus = "0.14.0" diff --git a/crates/networking/p2p/discovery/discv4_handlers.rs b/crates/networking/p2p/discovery/discv4_handlers.rs index d5f031dad6d..b6e73e8b3d3 100644 --- a/crates/networking/p2p/discovery/discv4_handlers.rs +++ b/crates/networking/p2p/discovery/discv4_handlers.rs @@ -129,11 +129,39 @@ impl DiscoveryServer { } // Remove finished lookups + let had_active = self + .discv4 + .as_ref() + .is_some_and(|s| !s.active_lookups.is_empty()); self.discv4 .as_mut() .expect("discv4 state must exist") .active_lookups .retain(|(l, _)| !l.is_finished()); + let just_finished = had_active + && self + .discv4 + .as_ref() + .is_some_and(|s| s.active_lookups.is_empty()); + + // A lookup just ended. If it added nothing to the peer table, the + // network is saturated: rather than start the next one on this same + // tick, let the backoff in `get_lookup_interval` space them out. + if just_finished { + let count = self.peer_table.discovered_count().await?; + let discv4 = self.discv4.as_mut().expect("discv4 state must exist"); + if count > discv4.lookup_started_at_count { + discv4.empty_lookups_in_a_row = 0; + } else { + discv4.empty_lookups_in_a_row = discv4.empty_lookups_in_a_row.saturating_add(1); + trace!( + protocol = "discv4", + empty_in_a_row = discv4.empty_lookups_in_a_row, + "Lookup found nothing new, backing off" + ); + return Ok(()); + } + } // If a lookup is already active, advance it instead of starting a new // one. Lookups are timer-driven: each tick sends the next alpha queries. @@ -180,7 +208,9 @@ impl DiscoveryServer { let mut buf = BytesMut::new(); msg.encode_with_header(&mut buf, &self.signer); + let started_at_count = self.peer_table.discovered_count().await?; let discv4 = self.discv4.as_mut().expect("discv4 state must exist"); + discv4.lookup_started_at_count = started_at_count; discv4.active_lookups.push((lookup, buf)); // Fire the initial queries for the new lookup @@ -338,18 +368,15 @@ impl DiscoveryServer { { self.discv4_send_ping(&node).await?; } else { + // A contact we hold no record for counts as seq 0, as on the discv5 + // side. Otherwise a discv4 node's ENR, and with it the fork-id verdict + // that keeps it from being dialed on the wrong chain, would only ever + // be fetched by the slow random ENR lookup. let node_id = node_id(&sender_public_key); - let stored_enr_seq = self - .peer_table - .get_contact(node_id) - .await? - .and_then(|c| c.record) - .map(|r| r.seq); - - let received_enr_seq = ping_message.enr_seq; - - if let (Some(received), Some(stored)) = (received_enr_seq, stored_enr_seq) - && received > stored + if let Some(contact) = self.peer_table.get_contact(node_id).await? + && contact.was_validated() + && let Some(received) = ping_message.enr_seq + && received > contact.record.as_ref().map_or(0, |r| r.seq) { self.discv4_send_enr_request(&node).await?; } @@ -369,10 +396,10 @@ impl DiscoveryServer { let ping_id = Bytes::copy_from_slice(message.ping_hash.as_bytes()); self.peer_table.record_pong_received(node_id, ping_id)?; - let stored_enr_seq = contact.record.map(|r| r.seq); - let received_enr_seq = message.enr_seq; - if let (Some(received), Some(stored)) = (received_enr_seq, stored_enr_seq) - && received > stored + // No record yet counts as seq 0 (see `discv4_handle_ping`). + let stored_enr_seq = contact.record.as_ref().map_or(0, |r| r.seq); + if let Some(received) = message.enr_seq + && received > stored_enr_seq { self.discv4_send_enr_request(&contact.node).await?; } diff --git a/crates/networking/p2p/discovery/discv5_handlers.rs b/crates/networking/p2p/discovery/discv5_handlers.rs index a96700d0545..3c320794d9f 100644 --- a/crates/networking/p2p/discovery/discv5_handlers.rs +++ b/crates/networking/p2p/discovery/discv5_handlers.rs @@ -296,11 +296,39 @@ impl DiscoveryServer { } // Remove finished lookups + let had_active = self + .discv5 + .as_ref() + .is_some_and(|s| !s.active_lookups.is_empty()); self.discv5 .as_mut() .expect("discv5 state must exist") .active_lookups .retain(|l| !l.is_finished()); + let just_finished = had_active + && self + .discv5 + .as_ref() + .is_some_and(|s| s.active_lookups.is_empty()); + + // A lookup just ended. If it added nothing to the peer table, the + // network is saturated: rather than start the next one on this same + // tick, let the backoff in `get_lookup_interval` space them out. + if just_finished { + let count = self.peer_table.discovered_count().await?; + let discv5 = self.discv5.as_mut().expect("discv5 state must exist"); + if count > discv5.lookup_started_at_count { + discv5.empty_lookups_in_a_row = 0; + } else { + discv5.empty_lookups_in_a_row = discv5.empty_lookups_in_a_row.saturating_add(1); + trace!( + protocol = "discv5", + empty_in_a_row = discv5.empty_lookups_in_a_row, + "Lookup found nothing new, backing off" + ); + return Ok(()); + } + } // If a lookup is already active, advance it instead of starting a new // one. Lookups are timer-driven: each tick sends the next alpha queries. @@ -338,7 +366,9 @@ impl DiscoveryServer { "Starting new iterative lookup" ); let lookup = IterativeLookup::new(target_id, seed); + let started_at_count = self.peer_table.discovered_count().await?; let discv5 = self.discv5.as_mut().expect("discv5 state must exist"); + discv5.lookup_started_at_count = started_at_count; discv5.active_lookups.push(lookup); // Fire the initial queries for the new lookup diff --git a/crates/networking/p2p/discovery/mod.rs b/crates/networking/p2p/discovery/mod.rs index 8ff3162b7d1..3c67dd0ef74 100644 --- a/crates/networking/p2p/discovery/mod.rs +++ b/crates/networking/p2p/discovery/mod.rs @@ -20,6 +20,7 @@ pub mod server; pub use ip_predictor::IpPredictor; pub use server::{DiscoveryServer, DiscoveryServerError, is_discv4_packet}; +use crate::netrestrict::NetRestrict; use std::time::Duration; /// Configuration for which discovery protocols to enable. @@ -30,6 +31,9 @@ pub struct DiscoveryConfig { /// Set to true when `--nat extip:` was supplied; locks the IP predictor /// from overwriting the user-specified external address. pub nat_extip_set: bool, + /// IP networks peers must fall in. Packets from outside are dropped before + /// they are decoded, and bootnodes outside are not contacted. + pub netrestrict: NetRestrict, } /// Lookup interval bounds for the RLPx initiator's connection attempts. The @@ -53,3 +57,50 @@ pub fn lookup_interval_function(progress: f64, lower_limit: f64, upper_limit: f6 (1000f64 * (ease_in_out_cubic * (upper_limit - lower_limit) + lower_limit)).round() as u64, ) } + +/// Interval before the *next* lookup starts, once the previous one has finished. +/// +/// `empty_lookups_in_a_row` counts finished lookups that added nothing to the +/// peer table. Each one doubles the wait from `lower_limit`, up to +/// `upper_limit`: a network whose every node is already known stops being asked +/// the same question twice a second. The completion-based pacing still applies, +/// so the result is never shorter than [`lookup_interval_function`] alone. +pub fn next_lookup_interval( + progress: f64, + empty_lookups_in_a_row: u32, + lower_limit: f64, + upper_limit: f64, +) -> Duration { + let paced = lookup_interval_function(progress, lower_limit, upper_limit); + if empty_lookups_in_a_row == 0 { + return paced; + } + // 2^16 already dwarfs any sane upper limit; the clamp only keeps `powi` finite. + let backoff_ms = + (lower_limit * 2f64.powi(empty_lookups_in_a_row.min(16) as i32)).min(upper_limit); + paced.max(Duration::from_micros((1000f64 * backoff_ms).round() as u64)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn saturation_backoff_doubles_from_the_lower_limit_and_caps() { + let at = |empty| next_lookup_interval(0.0, empty, 500.0, 10_000.0); + assert_eq!(at(0), Duration::from_millis(500)); + assert_eq!(at(1), Duration::from_millis(1_000)); + assert_eq!(at(2), Duration::from_millis(2_000)); + assert_eq!(at(3), Duration::from_millis(4_000)); + assert_eq!(at(4), Duration::from_millis(8_000)); + assert_eq!(at(5), Duration::from_millis(10_000)); + assert_eq!(at(1_000), Duration::from_millis(10_000)); + } + + #[test] + fn saturation_backoff_never_undercuts_completion_pacing() { + let paced = lookup_interval_function(0.9, 500.0, 10_000.0); + assert!(paced > Duration::from_millis(1_000)); + assert_eq!(next_lookup_interval(0.9, 1, 500.0, 10_000.0), paced); + } +} diff --git a/crates/networking/p2p/discovery/server.rs b/crates/networking/p2p/discovery/server.rs index 368981d27eb..68595f79178 100644 --- a/crates/networking/p2p/discovery/server.rs +++ b/crates/networking/p2p/discovery/server.rs @@ -36,7 +36,9 @@ use tokio::sync::watch; use tokio_util::udp::UdpFramed; use tracing::{debug, error, info, trace, warn}; -use super::{DiscoveryConfig, codec::DiscriminatingCodec, lookup_interval_function}; +use super::{ + DiscoveryConfig, codec::DiscriminatingCodec, lookup_interval_function, next_lookup_interval, +}; /// Minimum packet size for a valid discv4 packet. /// hash (32) + signature (65) + type (1) = 98 bytes @@ -146,6 +148,21 @@ impl DiscoveryServer { ) -> Result<(), DiscoveryServerError> { debug!("Starting discovery server"); + let bootnodes: Vec = bootnodes + .into_iter() + .filter(|node| { + let allowed = config.netrestrict.allows(node.ip); + if !allowed { + warn!( + node = %node, + netrestrict = %config.netrestrict, + "Dropping bootnode outside --p2p.netrestrict" + ); + } + allowed + }) + .collect(); + let discv4 = if config.discv4_enabled { info!( protocol = "discv4", @@ -170,6 +187,18 @@ impl DiscoveryServer { None }; + // With both protocols off the bootnodes are the whole network this node + // will ever know. Hand them to the dialer anyway, so disabling discovery + // doubles as a static-peers mode instead of a node with no peers at all. + // The protocol tag only steers revalidation, which is not running. + if discv4.is_none() && discv5.is_none() && !bootnodes.is_empty() { + info!( + count = bootnodes.len(), + "Discovery disabled, using bootnodes as static peers" + ); + peer_table.new_contacts(bootnodes.clone(), DiscoveryProtocol::Discv4)?; + } + let ip_override_locked = config.nat_extip_set; let mut server = Self { local_node: local_node.clone(), @@ -302,7 +331,7 @@ impl DiscoveryServer { let _ = self.discv4_lookup().await.inspect_err( |e| error!(protocol = "discv4", err=?e, "Error performing Discovery lookup"), ); - let interval = self.get_lookup_interval().await; + let interval = self.get_lookup_interval(DiscoveryProtocol::Discv4).await; send_after(interval, ctx.clone(), discovery_server_protocol::LookupV4); } @@ -316,7 +345,7 @@ impl DiscoveryServer { let _ = self.discv5_lookup().await.inspect_err( |e| error!(protocol = "discv5", err=?e, "Error performing Discovery lookup"), ); - let interval = self.get_lookup_interval().await; + let interval = self.get_lookup_interval(DiscoveryProtocol::Discv5).await; send_after(interval, ctx.clone(), discovery_server_protocol::LookupV5); } @@ -330,7 +359,7 @@ impl DiscoveryServer { let _ = self.discv4_enr_lookup().await.inspect_err( |e| error!(protocol = "discv4", err=?e, "Error performing Discovery lookup"), ); - let interval = self.get_lookup_interval().await; + let interval = self.get_lookup_interval(DiscoveryProtocol::Discv4).await; send_after(interval, ctx.clone(), discovery_server_protocol::EnrLookup); } @@ -355,6 +384,10 @@ impl DiscoveryServer { // --- Shared logic --- async fn route_packet(&mut self, data: &[u8], from: SocketAddr) { + if !self.config.netrestrict.allows(from.ip()) { + trace!(%from, "Dropping UDP packet from outside --p2p.netrestrict"); + return; + } if is_discv4_packet(data) { self.route_to_discv4(data, from).await; } else { @@ -467,17 +500,43 @@ impl DiscoveryServer { guard.record = self.local_node_record.clone(); } - pub(crate) async fn get_lookup_interval(&self) -> Duration { + /// How long until the next lookup tick for `protocol`. + /// + /// While a lookup is in flight the tick follows the completion-based pacing, + /// so the lookup finishes promptly. Between lookups the saturation backoff + /// applies as well: a network that keeps answering with nodes we already + /// know is asked less and less often, down to the steady-state rate. + pub(crate) async fn get_lookup_interval(&self, protocol: DiscoveryProtocol) -> Duration { let peer_completion = self .peer_table .target_peers_completion() .await .unwrap_or_default(); - lookup_interval_function( - peer_completion, - ITERATIVE_LOOKUP_INITIAL_MS, - ITERATIVE_LOOKUP_INTERVAL_MS, - ) + let (lookup_in_flight, empty_lookups_in_a_row) = match protocol { + DiscoveryProtocol::Discv4 => self + .discv4 + .as_ref() + .map(|s| (!s.active_lookups.is_empty(), s.empty_lookups_in_a_row)), + DiscoveryProtocol::Discv5 => self + .discv5 + .as_ref() + .map(|s| (!s.active_lookups.is_empty(), s.empty_lookups_in_a_row)), + } + .unwrap_or((false, 0)); + if lookup_in_flight { + lookup_interval_function( + peer_completion, + ITERATIVE_LOOKUP_INITIAL_MS, + ITERATIVE_LOOKUP_INTERVAL_MS, + ) + } else { + next_lookup_interval( + peer_completion, + empty_lookups_in_a_row, + ITERATIVE_LOOKUP_INITIAL_MS, + ITERATIVE_LOOKUP_INTERVAL_MS, + ) + } } /// Republish the ENR when the network layer reports a new fork id. /// @@ -553,6 +612,7 @@ impl DiscoveryServer { discv4_enabled: false, discv5_enabled: true, nat_extip_set: false, + netrestrict: crate::netrestrict::NetRestrict::default(), }, discv4: None, discv5: Some(Discv5State::default()), @@ -592,9 +652,12 @@ mod tests { record: local_node_record.clone(), })); let udp_socket = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap()); - let peer_table = PeerTableServer::spawn(H256::random(), TARGET_PEERS, { - ethrex_storage::Store::new("", ethrex_storage::EngineType::InMemory).unwrap() - }); + let peer_table = PeerTableServer::spawn( + H256::random(), + TARGET_PEERS, + ethrex_storage::Store::new("", ethrex_storage::EngineType::InMemory).unwrap(), + crate::netrestrict::NetRestrict::default(), + ); DiscoveryServer { local_node, local_node_record, @@ -605,6 +668,7 @@ mod tests { discv4_enabled: false, discv5_enabled: false, nat_extip_set: ip_override_locked, + netrestrict: crate::netrestrict::NetRestrict::default(), }, discv4: None, discv5: None, diff --git a/crates/networking/p2p/discv4/server.rs b/crates/networking/p2p/discv4/server.rs index e3c89a4f4d4..225e4f18717 100644 --- a/crates/networking/p2p/discv4/server.rs +++ b/crates/networking/p2p/discv4/server.rs @@ -17,6 +17,12 @@ pub struct Discv4State { pub pending_find_node: HashMap, /// Currently active iterative lookups, each with its cached signed FindNode message. pub active_lookups: Vec<(IterativeLookup, BytesMut)>, + /// The peer table's `discovered_count` when the current lookup started, to + /// tell on completion whether it found any node the table did not already hold. + pub lookup_started_at_count: u64, + /// Finished lookups in a row that found nothing new. Each doubles the wait + /// before the next lookup starts; see `discovery::next_lookup_interval`. + pub empty_lookups_in_a_row: u32, } #[derive(Debug, Clone)] diff --git a/crates/networking/p2p/discv5/server.rs b/crates/networking/p2p/discv5/server.rs index 844708018d7..ce3c655a3f2 100644 --- a/crates/networking/p2p/discv5/server.rs +++ b/crates/networking/p2p/discv5/server.rs @@ -66,6 +66,12 @@ pub struct Discv5State { /// split across up to `total` packets sharing one request id; they expire /// via `PENDING_FINDNODE_TIMEOUT` instead. pub pending_findnodes: FxHashMap<(H256, Bytes), Instant>, + /// The peer table's `discovered_count` when the current lookup started, to + /// tell on completion whether it found any node the table did not already hold. + pub lookup_started_at_count: u64, + /// Finished lookups in a row that found nothing new. Each doubles the wait + /// before the next lookup starts; see `discovery::next_lookup_interval`. + pub empty_lookups_in_a_row: u32, } impl Default for Discv5State { @@ -83,6 +89,8 @@ impl Default for Discv5State { whoareyou_global_window_start: Instant::now(), session_ips: Default::default(), active_lookups: Vec::new(), + lookup_started_at_count: 0, + empty_lookups_in_a_row: 0, } } } diff --git a/crates/networking/p2p/netrestrict.rs b/crates/networking/p2p/netrestrict.rs new file mode 100644 index 00000000000..91e680c7158 --- /dev/null +++ b/crates/networking/p2p/netrestrict.rs @@ -0,0 +1,96 @@ +//! `--p2p.netrestrict`: confine every peer interaction to a set of IP networks. +//! +//! A node on a private devnet has no business talking to the public internet, +//! and an operator behind a carrier-grade NAT can get rate-limited by their +//! provider if it does. With a restriction in place, discovered nodes outside +//! the allowed networks are never stored, pinged or dialed, and inbound TCP and +//! UDP from outside is dropped before any handshake. Mirrors geth's +//! `--netrestrict`. + +use std::{fmt, net::IpAddr, sync::Arc}; + +pub use ipnet::IpNet; + +/// IP networks this node may talk to. Empty means unrestricted. +/// +/// Consulted on every inbound packet and every discovered node, from several +/// actors, so the list is shared rather than copied on clone. +#[derive(Clone, Debug, Default)] +pub struct NetRestrict(Arc<[IpNet]>); + +impl NetRestrict { + pub fn new(nets: Vec) -> Self { + Self(nets.into()) + } + + /// Whether `ip` may be contacted, or accepted as a source. Always true when + /// no restriction is configured. + pub fn allows(&self, ip: IpAddr) -> bool { + self.0.is_empty() || self.0.iter().any(|net| net.contains(&ip)) + } + + pub fn is_unrestricted(&self) -> bool { + self.0.is_empty() + } +} + +impl From> for NetRestrict { + fn from(nets: Vec) -> Self { + Self::new(nets) + } +} + +impl fmt::Display for NetRestrict { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + if self.0.is_empty() { + return f.write_str("unrestricted"); + } + for (i, net) in self.0.iter().enumerate() { + if i > 0 { + f.write_str(",")?; + } + write!(f, "{net}")?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{Ipv4Addr, Ipv6Addr}; + + fn restrict(nets: &[&str]) -> NetRestrict { + NetRestrict::new(nets.iter().map(|n| n.parse().unwrap()).collect()) + } + + #[test] + fn unrestricted_allows_everything() { + let unrestricted = NetRestrict::default(); + assert!(unrestricted.is_unrestricted()); + assert!(unrestricted.allows(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)))); + assert!(unrestricted.allows(IpAddr::V6(Ipv6Addr::LOCALHOST))); + assert_eq!(unrestricted.to_string(), "unrestricted"); + } + + #[test] + fn only_addresses_inside_a_listed_network_are_allowed() { + let devnet = restrict(&["10.0.0.0/8", "172.16.0.0/12"]); + assert!(!devnet.is_unrestricted()); + assert!(devnet.allows(IpAddr::V4(Ipv4Addr::new(10, 200, 1, 1)))); + assert!(devnet.allows(IpAddr::V4(Ipv4Addr::new(172, 31, 255, 254)))); + assert!(!devnet.allows(IpAddr::V4(Ipv4Addr::new(172, 32, 0, 1)))); + assert!(!devnet.allows(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); + // A v4-only list says nothing about v6, so v6 sources are rejected too. + assert!(!devnet.allows(IpAddr::V6(Ipv6Addr::LOCALHOST))); + assert_eq!(devnet.to_string(), "10.0.0.0/8,172.16.0.0/12"); + } + + #[test] + fn v6_networks_are_matched_as_well() { + let local = restrict(&["fd00::/8", "127.0.0.0/8"]); + assert!(local.allows("fd12::1".parse().unwrap())); + assert!(!local.allows("2001:db8::1".parse().unwrap())); + assert!(local.allows(IpAddr::V4(Ipv4Addr::LOCALHOST))); + } +} diff --git a/crates/networking/p2p/network.rs b/crates/networking/p2p/network.rs index 1c59f5772fd..927a9b08807 100644 --- a/crates/networking/p2p/network.rs +++ b/crates/networking/p2p/network.rs @@ -6,6 +6,7 @@ pub struct P2PBasedContext; use crate::{ discovery::{DiscoveryConfig, DiscoveryServer, DiscoveryServerError}, metrics::{CurrentStepValue, METRICS}, + netrestrict::NetRestrict, peer_table::{PeerData, PeerTable, PeerTableServerProtocol as _}, rlpx::{ connection::server::{PeerConnBroadcastSender, PeerConnection}, @@ -52,6 +53,10 @@ pub struct P2PContext { pub based_context: Option, pub tx_broadcaster: ActorRef, pub initial_lookup_interval: f64, + /// IP networks peers must fall in. Inbound connections from outside are + /// dropped before the handshake; discovery applies the same list to what it + /// stores and dials. + pub netrestrict: NetRestrict, /// Caps concurrent INBOUND connections (including pre-handshake) so a flood of inbound /// dials can't accumulate connection actors/sockets without bound. Each inbound connection /// actor holds a permit for its lifetime; the permit is released when the actor is dropped. @@ -76,6 +81,7 @@ impl P2PContext { based_context: Option, tx_broadcasting_time_interval: u64, lookup_interval: f64, + netrestrict: NetRestrict, ) -> Result { let (channel_broadcast_send_end, _) = tokio::sync::broadcast::channel::<( tokio::task::Id, @@ -108,6 +114,7 @@ impl P2PContext { based_context, tx_broadcaster, initial_lookup_interval: lookup_interval, + netrestrict, inbound_admission: Arc::new(tokio::sync::Semaphore::new(MAX_INBOUND_CONNECTIONS)), }) } @@ -217,6 +224,11 @@ pub(crate) async fn serve_p2p_requests(context: P2PContext) { continue; } + if !context.netrestrict.allows(peer_addr.ip()) { + tracing::debug!(peer = %peer_addr, "Dropping inbound connection from outside --p2p.netrestrict"); + continue; + } + // Bound concurrent inbound connections: if we're at capacity, drop this one instead // of letting connection actors/sockets accumulate. The permit is moved into the // connection actor and released when the actor is dropped. diff --git a/crates/networking/p2p/p2p.rs b/crates/networking/p2p/p2p.rs index 825fb92456c..9d964f248b1 100644 --- a/crates/networking/p2p/p2p.rs +++ b/crates/networking/p2p/p2p.rs @@ -67,6 +67,7 @@ pub mod discovery; pub mod discv4; pub mod discv5; pub(crate) mod metrics; +pub mod netrestrict; pub mod network; pub mod peer_filter; pub mod peer_handler; @@ -83,5 +84,6 @@ pub mod utils; pub mod test_utils; pub use discovery::DiscoveryConfig; +pub use netrestrict::NetRestrict; pub use network::periodically_show_peer_stats; pub use network::start_network; diff --git a/crates/networking/p2p/peer_table.rs b/crates/networking/p2p/peer_table.rs index 4130dedabf5..1b0bb2aa4dd 100644 --- a/crates/networking/p2p/peer_table.rs +++ b/crates/networking/p2p/peer_table.rs @@ -11,6 +11,7 @@ use crate::{ metrics::METRICS, + netrestrict::NetRestrict, peer_filter::{EthForkIdFilter, PeerFilter}, rlpx::{connection::server::PeerConnection, p2p::Capability}, types::{Node, NodeRecord}, @@ -23,7 +24,7 @@ use indexmap::IndexMap; use rand::distributions::WeightedIndex; use rand::prelude::Distribution; use rand::seq::{IteratorRandom, SliceRandom}; -use rustc_hash::{FxHashMap, FxHashSet}; +use rustc_hash::FxHashMap; use spawned_concurrency::{ actor, error::ActorError, @@ -64,6 +65,25 @@ const MAX_REPLACEMENTS_PER_BUCKET: usize = 10; /// structure allows (256 × 16 = 4,096 vs this larger capacity). /// 10K matches what Reth and Nethermind use for their candidate pools. const MAX_CONNECTION_POOL_SIZE: usize = 10_000; +/// How long a dial candidate is left alone after its first failed attempt. Geth +/// uses the same figure (`dialHistoryExpiration`, 35s) for the same purpose. +const DIAL_BACKOFF_BASE: Duration = Duration::from_secs(35); +/// Ceiling for the per-node dial backoff. Doubling from the base, a node that +/// never answers is retried a sixth time roughly 19 minutes after the fifth, and +/// every 30 minutes from then on, instead of once per sweep of the pool. +const DIAL_BACKOFF_MAX: Duration = Duration::from_secs(30 * 60); + +/// Wait imposed after `failures` consecutive failed dials: 35s, 70s, 140s, …, +/// capped at [`DIAL_BACKOFF_MAX`]. No failures, no wait. +fn dial_backoff(failures: u32) -> Duration { + if failures == 0 { + return Duration::ZERO; + } + DIAL_BACKOFF_BASE + .checked_mul(1u32.checked_shl(failures - 1).unwrap_or(u32::MAX)) + .unwrap_or(DIAL_BACKOFF_MAX) + .min(DIAL_BACKOFF_MAX) +} /// A single k-bucket in the Kademlia routing table. /// Each bucket stores contacts at a specific XOR distance range from the local node. @@ -296,6 +316,43 @@ impl Contact { } } +/// A dial candidate in the flat connection pool, with the little state the RLPx +/// initiator needs to stop hammering it. +#[derive(Debug, Clone)] +struct PoolEntry { + node: Node, + /// When this node was last handed out for dialing. + last_dial_attempt: Option, + /// Dials since the last successful connection. Each one doubles the wait + /// before the next; see [`dial_backoff`]. + dial_failures: u32, +} + +impl PoolEntry { + fn new(node: Node) -> Self { + Self { + node, + last_dial_attempt: None, + dial_failures: 0, + } + } + + /// Whether the backoff from previous failed dials has elapsed. + fn dial_allowed(&self, now: Instant) -> bool { + match self.last_dial_attempt { + None => true, + Some(at) => now.saturating_duration_since(at) >= dial_backoff(self.dial_failures), + } + } + + /// Record that the node was handed out for dialing. Counted as a failure up + /// front; a connection that succeeds clears it again. + fn record_dial(&mut self, now: Instant) { + self.last_dial_attempt = Some(now); + self.dial_failures = self.dial_failures.saturating_add(1); + } +} + #[derive(Debug, Clone)] pub struct PeerData { pub node: Node, @@ -453,6 +510,7 @@ pub trait PeerTableServerProtocol: Send + Sync { fn target_reached(&self) -> Response; fn target_peers_reached(&self) -> Response; fn target_peers_completion(&self) -> Response; + fn discovered_count(&self) -> Response; fn get_contact_to_initiate(&self) -> Response>>; fn get_contact_for_enr_lookup(&self) -> Response>>; fn get_closest_from_pool(&self, target: H256, count: usize) -> Response>; @@ -498,7 +556,6 @@ pub struct PeerTableServer { local_node_id: H256, buckets: Vec, peers: IndexMap, - already_tried_peers: FxHashSet, target_peers: usize, /// What this consumer requires of a discovered peer. Judged as each ENR /// arrives, over either discovery protocol; the answer is cached on the @@ -512,7 +569,13 @@ pub struct PeerTableServer { /// has access to a much larger candidate pool than the k-bucket structure /// allows (k-buckets: 256 × 16 = 4,096 max; this pool: up to 50,000). /// K-buckets are still used for all Kademlia protocol operations. - connection_pool: IndexMap, + connection_pool: IndexMap, + /// Monotonic count of nodes ever added to the connection pool. Discovery + /// compares it before and after a lookup to tell whether the lookup found + /// anything this table did not already know. + discovered_count: u64, + /// IP networks a node must fall in to be stored at all. See [`NetRestrict`]. + netrestrict: NetRestrict, } // Hand-written because `Box` is not `Debug`, and requiring that @@ -522,8 +585,8 @@ impl std::fmt::Debug for PeerTableServer { f.debug_struct("PeerTableServer") .field("local_node_id", &self.local_node_id) .field("peers", &self.peers) - .field("already_tried_peers", &self.already_tried_peers) .field("target_peers", &self.target_peers) + .field("netrestrict", &self.netrestrict) .field("sessions", &self.sessions) .field("connection_pool", &self.connection_pool) .finish_non_exhaustive() @@ -538,32 +601,45 @@ impl PeerTableServer { /// A contact discovered without an ENR is never screened and stays dialable, /// so bootnodes are usable before they have published anything. See /// [`Self::spawn_with_filter`] for consumers on other networks. - pub fn spawn(local_node_id: H256, target_peers: usize, store: Store) -> PeerTable { - Self::spawn_with_filter(local_node_id, target_peers, EthForkIdFilter::new(store)) + pub fn spawn( + local_node_id: H256, + target_peers: usize, + store: Store, + netrestrict: NetRestrict, + ) -> PeerTable { + Self::spawn_with_filter( + local_node_id, + target_peers, + EthForkIdFilter::new(store), + netrestrict, + ) } pub fn spawn_with_filter( local_node_id: H256, target_peers: usize, filter: impl PeerFilter + 'static, + netrestrict: NetRestrict, ) -> PeerTable { - PeerTableServer::new(local_node_id, target_peers, Box::new(filter)).start() + PeerTableServer::new(local_node_id, target_peers, Box::new(filter), netrestrict).start() } pub(crate) fn new( local_node_id: H256, target_peers: usize, filter: Box, + netrestrict: NetRestrict, ) -> Self { Self { local_node_id, buckets: vec![KBucket::default(); NUMBER_OF_BUCKETS], peers: Default::default(), - already_tried_peers: Default::default(), target_peers, filter, sessions: Default::default(), connection_pool: IndexMap::with_capacity(MAX_CONNECTION_POOL_SIZE), + discovered_count: 0, + netrestrict, } } @@ -611,6 +687,7 @@ impl PeerTableServer { msg.negotiated_eth, ); new_peer.is_connection_inbound = msg.is_inbound; + self.record_connected(&new_peer_id); self.peers.insert(new_peer_id, new_peer); } @@ -843,6 +920,15 @@ impl PeerTableServer { self.peers.len() as f64 / self.target_peers as f64 } + #[request_handler] + async fn handle_discovered_count( + &mut self, + _msg: peer_table_server_protocol::DiscoveredCount, + _ctx: &Context, + ) -> u64 { + self.discovered_count + } + #[request_handler] async fn handle_get_contact_to_initiate( &mut self, @@ -997,6 +1083,9 @@ impl PeerTableServer { msg: peer_table_server_protocol::InsertIfNew, _ctx: &Context, ) -> bool { + if !self.netrestrict.allows(msg.node.ip) { + return false; + } let node_id = msg.node.node_id(); // Always add to the connection pool self.insert_to_connection_pool(node_id, msg.node.clone()); @@ -1180,7 +1269,16 @@ impl PeerTableServer { if self.connection_pool.len() >= MAX_CONNECTION_POOL_SIZE { self.connection_pool.shift_remove_index(0); } - self.connection_pool.insert(node_id, node); + self.connection_pool.insert(node_id, PoolEntry::new(node)); + self.discovered_count += 1; + } + + /// A connection to this node succeeded, so it is reachable: the next dial, + /// should it disconnect, starts from a clean slate. + fn record_connected(&mut self, node_id: &H256) { + if let Some(entry) = self.connection_pool.get_mut(node_id) { + entry.dial_failures = 0; + } } /// Look up a contact by node ID in either the main or replacement list. @@ -1294,11 +1392,20 @@ impl PeerTableServer { // --- Contact operations --- - /// Prune disposable contacts from both main and replacement lists. - /// When a main contact is removed, a replacement is automatically promoted. - /// Pruned contacts remain in the connection pool so they can be retried - /// later — the RLPx handshake will reject them if they're truly bad. + /// Prune disposable contacts from both main and replacement lists, and from + /// the connection pool. When a main contact is removed, a replacement is + /// automatically promoted. fn prune(&mut self) { + // A node discovery could not even reach is no dial candidate either. Left + // in the pool, the RLPx initiator would keep offering it after every backoff. + let unreachable: Vec = self + .iter_contacts() + .filter(|(_, c)| c.disposable) + .map(|(id, _)| *id) + .collect(); + for node_id in &unreachable { + self.connection_pool.shift_remove(node_id); + } for bucket in &mut self.buckets { // Collect disposable contacts from main list let main_disposable: Vec = bucket @@ -1328,26 +1435,34 @@ impl PeerTableServer { return None; } + let now = Instant::now(); let start = rand::random::() % pool_len; for offset in 0..pool_len { let idx = (start + offset) % pool_len; - let Some((node_id, node)) = self.connection_pool.get_index(idx) else { + let Some((node_id, entry)) = self.connection_pool.get_index(idx) else { continue; }; let node_id = *node_id; + // Skip what is connected, what is still backing off from a failed + // dial, and what discovery has already ruled out. A node the k-buckets + // never had room for has no contact to consult and stays dialable. if self.peers.contains_key(&node_id) - || self.already_tried_peers.contains(&node_id) + || !entry.dial_allowed(now) || self .get_contact_or_replacement(&node_id) - .map(|c| !c.knows_us || c.unwanted || c.passes_filter == Some(false)) + .map(|c| { + !c.knows_us || c.unwanted || c.disposable || c.passes_filter == Some(false) + }) .unwrap_or(false) { continue; } - let node = node.clone(); - self.already_tried_peers.insert(node_id); + let node = entry.node.clone(); + if let Some((_, entry)) = self.connection_pool.get_index_mut(idx) { + entry.record_dial(now); + } let contact = self .get_contact_or_replacement(&node_id) .cloned() @@ -1355,9 +1470,6 @@ impl PeerTableServer { return Some(contact); } - // Exhausted all candidates — reset tried set for next cycle. - tracing::trace!("Resetting list of tried peers."); - self.already_tried_peers.clear(); None } @@ -1365,15 +1477,15 @@ impl PeerTableServer { fn do_get_closest_from_pool(&self, target: H256, count: usize) -> Vec<(H256, Node)> { let mut nodes: Vec<(H256, Node, H256)> = Vec::with_capacity(count); - for (node_id, node) in &self.connection_pool { + for (node_id, entry) in &self.connection_pool { let dist = xor_distance(&target, node_id); if nodes.len() < count { - nodes.push((*node_id, node.clone(), dist)); + nodes.push((*node_id, entry.node.clone(), dist)); } else if let Some((farthest_idx, _)) = nodes.iter().enumerate().max_by_key(|(_, (_, _, d))| *d) && dist < nodes[farthest_idx].2 { - nodes[farthest_idx] = (*node_id, node.clone(), dist); + nodes[farthest_idx] = (*node_id, entry.node.clone(), dist); } } @@ -1482,6 +1594,10 @@ impl PeerTableServer { if node_id == self.local_node_id { continue; } + if !self.netrestrict.allows(node.ip) { + tracing::trace!(node = %node, "Ignoring node outside --p2p.netrestrict"); + continue; + } #[cfg(feature = "metrics")] let insert_start = std::time::Instant::now(); @@ -1541,6 +1657,10 @@ impl PeerTableServer { if node_id == self.local_node_id { continue; } + if !self.netrestrict.allows(node.ip) { + tracing::trace!(node = %node, "Ignoring node outside --p2p.netrestrict"); + continue; + } // Always add to the connection pool (regardless of k-bucket capacity) self.insert_to_connection_pool(node_id, node.clone()); @@ -1690,7 +1810,17 @@ mod tests { } fn table_with(filter: impl PeerFilter + 'static) -> PeerTableServer { - PeerTableServer::new(H256::zero(), 10, Box::new(filter)) + PeerTableServer::new(H256::zero(), 10, Box::new(filter), NetRestrict::default()) + } + + /// A table that only keeps nodes inside `cidr`. + fn table_restricted_to(cidr: &str) -> PeerTableServer { + PeerTableServer::new( + H256::zero(), + 10, + Box::new(FixedAnswer(true)), + NetRestrict::new(vec![cidr.parse().unwrap()]), + ) } /// A signed record for `seed`'s node at sequence number `seq`. @@ -2038,4 +2168,144 @@ mod tests { remote.0[0] = 0x80; assert_eq!(bucket_index(&local, &remote), Some(255)); } + + // --- dial backoff --- + + #[test] + fn dial_backoff_doubles_from_the_base_and_caps() { + assert_eq!(dial_backoff(0), Duration::ZERO); + assert_eq!(dial_backoff(1), DIAL_BACKOFF_BASE); + assert_eq!(dial_backoff(2), DIAL_BACKOFF_BASE * 2); + assert_eq!(dial_backoff(3), DIAL_BACKOFF_BASE * 4); + assert_eq!(dial_backoff(6), DIAL_BACKOFF_BASE * 32); + assert_eq!(dial_backoff(7), DIAL_BACKOFF_MAX); + assert_eq!(dial_backoff(u32::MAX), DIAL_BACKOFF_MAX); + } + + #[tokio::test] + async fn a_dialed_candidate_is_not_offered_again_inside_its_backoff() { + // Before, the only thing between two dials of the same dead node was one + // sweep of the pool: with three dead nodes that meant one SYN each every + // 300ms, forever. + let mut table = table_with(FixedAnswer(true)); + let (_, contact) = dummy_contact(11); + table + .do_new_contacts(vec![contact.node], DiscoveryProtocol::Discv4) + .await; + + assert!(table.do_get_contact_to_initiate().is_some()); + assert!(table.do_get_contact_to_initiate().is_none()); + } + + #[tokio::test] + async fn a_candidate_is_offered_again_once_its_backoff_elapsed() { + let mut table = table_with(FixedAnswer(true)); + let (node_id, contact) = dummy_contact(12); + table + .do_new_contacts(vec![contact.node], DiscoveryProtocol::Discv4) + .await; + assert!(table.do_get_contact_to_initiate().is_some()); + + let entry = table.connection_pool.get_mut(&node_id).unwrap(); + let elapsed = dial_backoff(entry.dial_failures); + entry.last_dial_attempt = Instant::now().checked_sub(elapsed); + + assert!(table.do_get_contact_to_initiate().is_some()); + assert_eq!(table.connection_pool[&node_id].dial_failures, 2); + } + + #[tokio::test] + async fn a_successful_connection_resets_the_dial_failures() { + let mut table = table_with(FixedAnswer(true)); + let (node_id, contact) = dummy_contact(13); + table + .do_new_contacts(vec![contact.node], DiscoveryProtocol::Discv4) + .await; + assert!(table.do_get_contact_to_initiate().is_some()); + assert_eq!(table.connection_pool[&node_id].dial_failures, 1); + + table.record_connected(&node_id); + + assert_eq!(table.connection_pool[&node_id].dial_failures, 0); + } + + #[tokio::test] + async fn an_unreachable_contact_is_neither_dialed_nor_kept_in_the_pool() { + let mut table = table_with(FixedAnswer(true)); + let (node_id, contact) = dummy_contact(14); + table + .do_new_contacts(vec![contact.node], DiscoveryProtocol::Discv4) + .await; + table.get_contact_mut(&node_id).unwrap().disposable = true; + + assert!(table.do_get_contact_to_initiate().is_none()); + + table.prune(); + + assert!(!table.connection_pool.contains_key(&node_id)); + assert!(table.get_contact(&node_id).is_none()); + } + + // --- netrestrict --- + + #[tokio::test] + async fn nodes_outside_the_netrestrict_are_never_stored() { + let mut table = table_restricted_to("10.0.0.0/8"); + let inside = Node::new( + IpAddr::V4(Ipv4Addr::new(10, 1, 2, 3)), + 30303, + 30303, + H512::from_low_u64_be(21), + ); + let outside = Node::new( + IpAddr::V4(Ipv4Addr::new(127, 0, 0, 22)), + 30303, + 30303, + H512::from_low_u64_be(22), + ); + + table + .do_new_contacts( + vec![inside.clone(), outside.clone()], + DiscoveryProtocol::Discv4, + ) + .await; + + assert!(table.get_contact(&inside.node_id()).is_some()); + assert!(table.get_contact(&outside.node_id()).is_none()); + assert!(!table.connection_pool.contains_key(&outside.node_id())); + assert_eq!(table.discovered_count, 1); + } + + #[tokio::test] + async fn records_outside_the_netrestrict_are_never_stored() { + let mut table = table_restricted_to("10.0.0.0/8"); + // `record_for` puts the node on 127.0.0.x. + let (node_id, record) = record_for(23, 1); + + table.do_new_contact_records(vec![record]).await; + + assert!(table.get_contact(&node_id).is_none()); + assert!(table.connection_pool.is_empty()); + assert_eq!(table.discovered_count, 0); + } + + // --- discovered_count --- + + #[tokio::test] + async fn discovered_count_only_grows_on_nodes_new_to_the_pool() { + let mut table = table_with(FixedAnswer(true)); + let (_, a) = dummy_contact(31); + let (_, b) = dummy_contact(32); + + table + .do_new_contacts(vec![a.node.clone()], DiscoveryProtocol::Discv4) + .await; + assert_eq!(table.discovered_count, 1); + + table + .do_new_contacts(vec![a.node, b.node], DiscoveryProtocol::Discv4) + .await; + assert_eq!(table.discovered_count, 2); + } } diff --git a/crates/networking/p2p/rlpx/initiator.rs b/crates/networking/p2p/rlpx/initiator.rs index 2f339415a81..eb5b9749ef6 100644 --- a/crates/networking/p2p/rlpx/initiator.rs +++ b/crates/networking/p2p/rlpx/initiator.rs @@ -74,16 +74,20 @@ impl RLPxInitiator { _msg: rlpx_initiator_protocol::LookForPeer, ctx: &Context, ) { - let _ = self + let dialed = self .do_look_for_peer() .await - .inspect_err(|e| error!(err=?e, "Error looking for peers")); + .inspect_err(|e| error!(err=?e, "Error looking for peers")) + .unwrap_or(false); - send_after( - self.get_lookup_interval().await, - ctx.clone(), - rlpx_initiator_protocol::LookForPeer, - ); + // With nothing to dial (target reached, or every candidate connected or + // backing off) there is no point polling the table ten times a second. + let interval = if dialed { + self.get_lookup_interval().await + } else { + Duration::from_millis(LOOKUP_INTERVAL_MS as u64) + }; + send_after(interval, ctx.clone(), rlpx_initiator_protocol::LookForPeer); } #[send_handler] @@ -105,16 +109,18 @@ impl RLPxInitiator { ctx.stop(); } - async fn do_look_for_peer(&mut self) -> Result<(), RLPxInitiatorError> { + /// Dials one candidate if the table has one to offer. Returns whether it did. + async fn do_look_for_peer(&mut self) -> Result { if !self.context.table.target_peers_reached().await? { if let Some(contact) = self.context.table.get_contact_to_initiate().await? { PeerConnection::spawn_as_initiator(self.context.clone(), &contact.node); METRICS.record_new_rlpx_conn_attempt().await; - }; + return Ok(true); + } } else { debug!("Target peer connections reached, no need to initiate new connections."); } - Ok(()) + Ok(false) } // We use the same lookup intervals as Discovery to try to get both process to check at the same rate diff --git a/crates/networking/rpc/test_utils.rs b/crates/networking/rpc/test_utils.rs index a66e602e4a6..dea11de36ba 100644 --- a/crates/networking/rpc/test_utils.rs +++ b/crates/networking/rpc/test_utils.rs @@ -383,7 +383,12 @@ pub async fn dummy_sync_manager() -> SyncManager { /// Creates a dummy PeerHandler for tests where interacting with peers is not needed /// This should only be used in tests as it won't be able to interact with the node's connected peers pub async fn dummy_peer_handler(store: Store) -> PeerHandler { - let peer_table = PeerTableServer::spawn(H256::random(), TARGET_PEERS, store); + let peer_table = PeerTableServer::spawn( + H256::random(), + TARGET_PEERS, + store, + ethrex_p2p::netrestrict::NetRestrict::default(), + ); PeerHandler::new(peer_table.clone(), dummy_actor(peer_table).await) } @@ -417,6 +422,7 @@ pub async fn dummy_p2p_context(peer_table: PeerTable) -> P2PContext { None, 1000, 100.0, + ethrex_p2p::netrestrict::NetRestrict::default(), ) .unwrap() } diff --git a/docs/CLI.md b/docs/CLI.md index 479c24d5915..9b8d17e189b 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -233,6 +233,11 @@ P2P options: [env: ETHREX_P2P_LOOKUP_INTERVAL=] [default: 100] + --p2p.netrestrict ... + Comma separated IP networks in CIDR notation, e.g. 10.0.0.0/8,172.16.0.0/12. Nodes discovered outside these networks are ignored, bootnodes outside are dropped, and inbound TCP and UDP from outside is discarded. Meant for private devnets and other closed networks. Unrestricted when not set. + + [env: ETHREX_P2P_NETRESTRICT=] + --blob-sampling Enable EIP-8070 PeerDAS blob sampling (sampler/provider state machine). Disabled by default; when off the node always acts as provider (p=1.0). diff --git a/docs/l1/running/startup.md b/docs/l1/running/startup.md index d0be3dec2c4..c8790cf28ac 100644 --- a/docs/l1/running/startup.md +++ b/docs/l1/running/startup.md @@ -14,6 +14,15 @@ These are the only names `--network` accepts. Any other value is treated as a pa For more information about sync modes please read the [sync modes document](../fundamentals/sync_modes.md). Snap syncing is the default; to switch to full sync use the flag `--syncmode full`. Full sync is not possible on a fresh mainnet or sepolia database, since ethrex only executes post-merge blocks. +## Private networks and devnets + +Two flags keep a node on a private network from talking to anything outside it: + +- `--p2p.netrestrict ` confines all P2P traffic to the given IP networks. Nodes discovered outside them are ignored, bootnodes outside them are dropped, and inbound TCP and UDP from outside is discarded. For example, `--p2p.netrestrict 10.0.0.0/8,172.16.0.0/12`. +- `--p2p.discv4=false --p2p.discv5=false` turns discovery off. The node then connects only to its `--bootnodes`, which act as static peers. + +Setting `--p2p.target-peers` to the number of nodes in the network also helps: connection attempts stop once the target is reached, and discovery lookups slow down as it is approached. + ## Run an Ethereum node This guide will assume that you already [installed ethrex](../../getting-started/installation/) and you know how to set up a [consensus client](./consensus_client.md) to communicate with ethrex.