From 44415e69f718c21fa3223497bc5aae1dcbd9daa6 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:16:27 +0200 Subject: [PATCH 01/12] fix(core): make SemVer::parse fail-closed on integer overflow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A peer-controlled version string with an over-long digit run (e.g. from peerinfo/dkg-sync over the wire) overflowed usize::parse and hit .expect("invalid regex"), panicking the per-connection handler — a remote-triggerable DoS. Map the parse failure to SemVerError::InvalidFormat instead; callers already handle Err. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/core/src/version.rs | 43 +++++++++++++++++++++++++++++++++++--- 1 file changed, 40 insertions(+), 3 deletions(-) diff --git a/crates/core/src/version.rs b/crates/core/src/version.rs index fd4f439a..475614d5 100644 --- a/crates/core/src/version.rs +++ b/crates/core/src/version.rs @@ -135,15 +135,28 @@ impl SemVer { .filter(|matches| matches.len() == 5) .ok_or(SemVerError::InvalidFormat)?; - let major = matches[1].parse().expect("invalid regex"); - let minor = matches[2].parse().expect("invalid regex"); + // The regex guarantees these capture groups are non-empty ASCII digits, + // so the only possible parse failure is integer overflow (e.g. a very + // long digit run from a malicious peer). Fail closed with InvalidFormat + // rather than panicking. NB: this is an intentional, safe divergence + // from Charon's Parse (app/version/version.go @ v1.7.1), which discards + // the strconv.Atoi error and silently yields 0 on overflow. + let major = matches[1] + .parse() + .map_err(|_| SemVerError::InvalidFormat)?; + let minor = matches[2] + .parse() + .map_err(|_| SemVerError::InvalidFormat)?; let mut patch = 0; let mut pre_release = ""; let mut sem_ver_type = SemVerType::Minor; if let Some(m) = matches.get(3) { - patch = m.as_str().parse().expect("invalid regex"); + patch = m + .as_str() + .parse() + .map_err(|_| SemVerError::InvalidFormat)?; sem_ver_type = SemVerType::Patch; } @@ -346,6 +359,12 @@ mod tests { version: "12-dev", expected: Err(SemVerError::InvalidFormat), }, + ParseTestCase { + name: "Overflow major", + // 30 nines: far exceeds usize::MAX on any target, must not panic. + version: "v999999999999999999999999999999.0", + expected: Err(SemVerError::InvalidFormat), + }, ]; for test in tc { @@ -353,4 +372,22 @@ mod tests { assert_eq!(actual, test.expected, "parse: `{}`", test.name); } } + + #[test] + fn parse_overflow_is_fail_closed() { + // Long digit runs that overflow usize must return Err, never panic. + let overflow = "9".repeat(40); + for version in [ + format!("v{overflow}.0"), // major overflow + format!("v0.{overflow}"), // minor overflow + format!("v0.0.{overflow}"), // patch overflow + format!("v0.0.{overflow}-rc1"), // patch overflow with pre-release + ] { + assert_eq!( + SemVer::parse(&version), + Err(SemVerError::InvalidFormat), + "expected fail-closed for `{version}`" + ); + } + } } From 1b4e7993106d870d9ff615b356f9af09478aa8ff Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:20:06 +0200 Subject: [PATCH 02/12] fix(cluster): cap num_validators for legacy definitions Legacy definition versions v1.0-v1.4 route an untrusted num_validators (u64 from JSON, reachable over the network) through repeat_v_addresses, an unbounded clone loop, before any length/signature check. A value up to u64::MAX drives unbounded allocation / OOM. Cap at SSZ_MAX_VALIDATORS (65536, the CompositeList[65536] bound) inside repeat_v_addresses before the loop, returning DefinitionError::NumValidatorsTooLarge. Hardening beyond Charon (whose signed int + SSZ bound implicitly guard this); it cannot reject any definition that could round-trip SSZ hashing. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/cluster/src/definition.rs | 92 +++++++++++++++++++++++++++++--- 1 file changed, 86 insertions(+), 6 deletions(-) diff --git a/crates/cluster/src/definition.rs b/crates/cluster/src/definition.rs index c2069f2f..cba253a2 100644 --- a/crates/cluster/src/definition.rs +++ b/crates/cluster/src/definition.rs @@ -9,7 +9,7 @@ use crate::{ }, helpers::from_0x_hex_str, operator::{Operator, OperatorV1X1, OperatorV1X2OrLater}, - ssz::{SSZError, hash_definition}, + ssz::{SSZ_MAX_VALIDATORS, SSZError, hash_definition}, version::{CURRENT_VERSION, DKG_ALGO, versions::*}, }; use chrono::{DateTime, Timelike, Utc}; @@ -256,6 +256,15 @@ pub enum DefinitionError { #[error("Failed to convert length")] FailedToConvertLength, + /// num_validators exceeds the maximum allowed validators (SSZ_MAX_VALIDATORS). + #[error("num_validators {num_validators} exceeds maximum {max}")] + NumValidatorsTooLarge { + /// The offending value from the definition. + num_validators: u64, + /// The enforced maximum (SSZ_MAX_VALIDATORS). + max: u64, + }, + /// Failed to convert hex string #[error("Failed to convert hex string")] FailedToConvertHexString(#[from] hex::FromHexError), @@ -921,7 +930,7 @@ impl TryFrom for Definition { }; let validator_addresses = - repeat_v_addresses(validator_addresses, definition.num_validators); + repeat_v_addresses(validator_addresses, definition.num_validators)?; Ok(Self { name: definition.name, @@ -1034,7 +1043,7 @@ impl TryFrom for Definition { }; let validator_addresses = - repeat_v_addresses(validator_addresses, definition.num_validators); + repeat_v_addresses(validator_addresses, definition.num_validators)?; Ok(Self { name: definition.name, @@ -1153,7 +1162,7 @@ impl TryFrom for Definition { }; let validator_addresses = - repeat_v_addresses(validator_addresses, definition.num_validators); + repeat_v_addresses(validator_addresses, definition.num_validators)?; Ok(Self { name: definition.name, @@ -1609,12 +1618,28 @@ impl From for Definition { } } -fn repeat_v_addresses(addr: ValidatorAddresses, num_validators: u64) -> Vec { +fn repeat_v_addresses( + addr: ValidatorAddresses, + num_validators: u64, +) -> Result, DefinitionError> { + // `num_validators` is an untrusted u64 from legacy-definition JSON. Cap it at + // SSZ_MAX_VALIDATORS (the CompositeList[65536] bound already enforced on the + // validator-addresses field) *before* the clone loop, so a value up to + // u64::MAX cannot drive an unbounded allocation / OOM. This is defensive + // hardening consistent with the SSZ bound Charon enforces; it cannot reject + // any definition that could round-trip through SSZ hashing. + if num_validators > SSZ_MAX_VALIDATORS as u64 { + return Err(DefinitionError::NumValidatorsTooLarge { + num_validators, + max: SSZ_MAX_VALIDATORS as u64, + }); + } + let mut validator_addresses = Vec::new(); for _ in 0..num_validators { validator_addresses.push(addr.clone()); } - validator_addresses + Ok(validator_addresses) } #[cfg(test)] @@ -1653,6 +1678,61 @@ mod tests { EthClient::new("http://127.0.0.1:8545").await.unwrap() } + fn legacy_definition_json(num_validators: u64) -> String { + format!( + r#"{{ + "name": "test", + "operators": [], + "uuid": "0194fdc2-fa2f-4cc0-81d3-ff12045b73c8", + "version": "v1.0.0", + "timestamp": "", + "num_validators": {num_validators}, + "threshold": 0, + "fee_recipient_address": "0x0000000000000000000000000000000000000000", + "withdrawal_address": "0x0000000000000000000000000000000000000000", + "dkg_algorithm": "default", + "fork_version": "0x00000000", + "config_hash": "", + "definition_hash": "" + }}"# + ) + } + + #[test] + fn legacy_definition_rejects_oversized_num_validators() { + // v1.0.0 definition with num_validators = u64::MAX must be rejected + // promptly (before the clone loop), never allocated. + let json = legacy_definition_json(u64::MAX); + + let result = serde_json::from_str::(&json); + assert!( + result.is_err(), + "u64::MAX num_validators must be rejected, not allocated" + ); + // The deserialize dispatch renders the TryFrom error via Debug + // ("Conversion error: NumValidatorsTooLarge {{ .. }}"), so assert on the + // variant name and the enforced max, both of which appear in the Debug repr. + let msg = format!("{}", result.unwrap_err()); + assert!( + msg.contains("NumValidatorsTooLarge") && msg.contains("65536"), + "unexpected error message: {msg}" + ); + } + + #[test] + fn legacy_definition_num_validators_boundary() { + // Exactly the max is accepted. + let at_max = serde_json::from_str::(&legacy_definition_json(65536)); + assert!(at_max.is_ok(), "65536 must be accepted: {:?}", at_max.err()); + assert_eq!(at_max.unwrap().validator_addresses.len(), 65536); + + // One over the max is rejected (guards `>` vs `>=`). + assert!( + serde_json::from_str::(&legacy_definition_json(65537)).is_err(), + "65537 must be rejected" + ); + } + #[test] fn cluster_definition_v1_10_0_fields() { let definition = serde_json::from_str::(include_str!( From de2255c65b03e338d124666c5c31e51e553bb599 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:21:13 +0200 Subject: [PATCH 03/12] fix(ssz): reject all-zero final byte in parse_bitlist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An SSZ bitlist whose final byte is 0x00 (missing the length-sentinel bit) underflowed msb to 255, then evaluated 1u8 << 255 — a debug-mode shift-overflow panic (masked to 1<<7 in release). Reachable via the public HashWalker::put_bitlist on attacker-influenced input. Guard the zero-final-byte case up front and return HasherError::InvalidBitlistDelimiter. Deliberate fail-closed divergence from fastssz (which wraps to a bogus length); a well-formed bitlist always has a non-zero final byte. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/ssz/src/hasher.rs | 62 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/crates/ssz/src/hasher.rs b/crates/ssz/src/hasher.rs index a1954ac8..822d1213 100644 --- a/crates/ssz/src/hasher.rs +++ b/crates/ssz/src/hasher.rs @@ -103,6 +103,9 @@ pub enum HasherError { /// Declared limit. limit: usize, }, + /// Bitlist final (delimiter) byte is zero — missing the length sentinel bit. + #[error("Invalid bitlist: final byte is zero (missing length delimiter bit)")] + InvalidBitlistDelimiter, } /// SSZ hasher for calculating Merkle roots. @@ -408,6 +411,15 @@ fn parse_bitlist(tmp: &mut Vec, buf: &[u8]) -> Result { } let last_byte = buf[buf.len().wrapping_sub(1)]; + if last_byte == 0 { + // A valid SSZ bitlist's final byte carries the length-delimiter (sentinel) + // bit and is therefore always non-zero. A zero final byte would underflow + // `msb` to 255 and trigger `1u8 << 255` (debug panic / masked-shift in + // release), so reject it as malformed input. + return Err(HasherError::InvalidBitlistDelimiter); + } + + // `last_byte != 0` => leading_zeros() in 0..=7 => msb in 0..=7, no overflow. let msb = 8u8 .wrapping_sub(last_byte.leading_zeros() as u8) .wrapping_sub(1); @@ -430,3 +442,53 @@ fn parse_bitlist(tmp: &mut Vec, buf: &[u8]) -> Result { Ok(size) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_bitlist_rejects_zero_final_byte_single() { + // Single 0x00 byte: previously underflowed msb to 255 and panicked on + // `1u8 << 255` in debug builds. Must now return an error, never panic. + let mut tmp = Vec::new(); + let err = parse_bitlist(&mut tmp, &[0x00]).unwrap_err(); + assert!(matches!(err, HasherError::InvalidBitlistDelimiter)); + } + + #[test] + fn parse_bitlist_rejects_zero_final_byte_multibyte() { + // Multi-byte buffer whose *final* byte is zero is also malformed. + let mut tmp = Vec::new(); + let err = parse_bitlist(&mut tmp, &[0x05, 0x00]).unwrap_err(); + assert!(matches!(err, HasherError::InvalidBitlistDelimiter)); + } + + #[test] + fn parse_bitlist_empty_buffer_is_invalid_length() { + let mut tmp = Vec::new(); + let err = parse_bitlist(&mut tmp, &[]).unwrap_err(); + assert!(matches!(err, HasherError::InvalidBufferLength)); + } + + #[test] + fn parse_bitlist_valid_single_byte_ok() { + // 0b0000_0011 -> sentinel bit at index 1, one data bit set at index 0. + // msb = 1, size = 1; the sentinel bit is cleared, leaving 0b0000_0001, + // which is non-zero so not truncated away. + let mut tmp = Vec::new(); + let size = parse_bitlist(&mut tmp, &[0b0000_0011]).unwrap(); + assert_eq!(size, 1); + assert_eq!(tmp, vec![0b0000_0001]); + } + + #[test] + fn parse_bitlist_valid_sentinel_only_truncates() { + // 0b0000_0001 -> sentinel at index 0, no data bits. msb = 0, size = 0, + // clearing the sentinel yields a trailing zero byte which is truncated. + let mut tmp = Vec::new(); + let size = parse_bitlist(&mut tmp, &[0b0000_0001]).unwrap(); + assert_eq!(size, 0); + assert!(tmp.is_empty()); + } +} From b6b7484f1e1b39008fa19f9ce56420c451c73706 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:23:54 +0200 Subject: [PATCH 04/12] fix(p2p): bound PeerStore inactive_peers and peer_addresses growth PeerStore accumulated a Peer per closed connection in inactive_peers (never trimmed) and a peer_addresses entry per identify (never removed). On a relay (gater defaults open) a remote rotating self-asserted PeerIds grows both without bound. Cap inactive_peers at MAX_INACTIVE_PEERS=1024 with oldest-first eviction that never drops known cluster peers (tracked via a companion VecDeque), and drop peer_addresses on disconnect for non-known peers with no remaining connections. Pluto-specific structure with no Charon analogue; defensive hardening. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/p2p/src/conn_logger.rs | 25 ++++-- crates/p2p/src/p2p_context.rs | 154 ++++++++++++++++++++++++++++++++-- 2 files changed, 169 insertions(+), 10 deletions(-) diff --git a/crates/p2p/src/conn_logger.rs b/crates/p2p/src/conn_logger.rs index 4486a5cf..ac08f4cb 100644 --- a/crates/p2p/src/conn_logger.rs +++ b/crates/p2p/src/conn_logger.rs @@ -249,11 +249,26 @@ impl NetworkBehaviour for ConnectionLogger } }; // Update peer store - self.p2p_context.peer_store_write_lock().remove_peer(Peer { - id: event.peer_id, - connection_id: event.connection_id, - remote_addr: addr.clone(), - }); + { + let known = self.p2p_context.known_peers().clone(); + let mut store = self.p2p_context.peer_store_write_lock(); + store.remove_peer( + Peer { + id: event.peer_id, + connection_id: event.connection_id, + remote_addr: addr.clone(), + }, + &known, + ); + // Drop cached identify addresses once the peer has no active + // connections and is not a known cluster peer, to bound + // `peer_addresses` growth. + if store.connections_to_peer(&event.peer_id).is_empty() + && !known.contains(&event.peer_id) + { + store.remove_peer_addresses(&event.peer_id); + } + } // Decrement the connection count based on the endpoint address self.decrement_connection(event.peer_id, &addr); } diff --git a/crates/p2p/src/p2p_context.rs b/crates/p2p/src/p2p_context.rs index 26e7d6a9..69f2b678 100644 --- a/crates/p2p/src/p2p_context.rs +++ b/crates/p2p/src/p2p_context.rs @@ -1,11 +1,17 @@ use std::{ - collections::{HashMap, HashSet}, + collections::{HashMap, HashSet, VecDeque}, sync::{Arc, OnceLock, RwLock, RwLockReadGuard, RwLockWriteGuard}, }; use libp2p::{Multiaddr, PeerId, swarm::ConnectionId}; use tracing::error; +/// Maximum number of inactive peers retained in the [`PeerStore`]. +/// +/// Disconnected, non-cluster peers are evicted oldest-first once this many are +/// stored, bounding memory growth from connection churn by remote peers. +const MAX_INACTIVE_PEERS: usize = 1024; + /// Global context shared across P2P components. /// /// This struct provides thread-safe access to shared state including: @@ -94,6 +100,11 @@ pub struct PeerStore { /// Inactive peers. inactive_peers: HashSet, + /// Insertion order of `inactive_peers`, oldest at the front. Used to evict + /// the oldest entries once `MAX_INACTIVE_PEERS` is exceeded. Membership is + /// kept consistent with `inactive_peers`. + inactive_order: VecDeque, + /// Known addresses for each peer (populated from identify protocol). peer_addresses: HashMap>, } @@ -101,14 +112,47 @@ pub struct PeerStore { impl PeerStore { /// Adds a peer to the peer store. pub fn add_peer(&mut self, peer: Peer) { - self.inactive_peers.remove(&peer); + if self.inactive_peers.remove(&peer) { + self.inactive_order.retain(|p| p != &peer); + } self.active_peers.insert(peer); } - /// Removes a peer from the peer store. - pub fn remove_peer(&mut self, peer: Peer) { + /// Marks a peer connection as inactive. + /// + /// `known_peers` is the set of cluster peer IDs; entries whose peer ID is in + /// this set are never evicted by the `MAX_INACTIVE_PEERS` cap. + pub fn remove_peer(&mut self, peer: Peer, known_peers: &HashSet) { self.active_peers.remove(&peer); - self.inactive_peers.insert(peer.clone()); + + // Avoid duplicate ordering entries if the same Peer is removed twice. + if self.inactive_peers.insert(peer.clone()) { + self.inactive_order.push_back(peer); + } + + self.evict_inactive(known_peers); + } + + /// Evicts oldest inactive, non-known peers until `inactive_peers` is within + /// `MAX_INACTIVE_PEERS`. Known cluster peers are re-queued (kept) so they are + /// never dropped. Bounded: each call scans at most `inactive_order.len()` entries. + fn evict_inactive(&mut self, known_peers: &HashSet) { + // Bound the number of scan iterations to the current queue length so a + // queue made entirely of known peers cannot loop forever. + let mut scanned = 0usize; + let max_scan = self.inactive_order.len(); + while self.inactive_peers.len() > MAX_INACTIVE_PEERS && scanned < max_scan { + scanned += 1; + let Some(candidate) = self.inactive_order.pop_front() else { + break; + }; + if known_peers.contains(&candidate.id) { + // Never evict known cluster peers; keep them, re-queue at back. + self.inactive_order.push_back(candidate); + continue; + } + self.inactive_peers.remove(&candidate); + } } /// Returns the active peers. @@ -157,4 +201,104 @@ impl PeerStore { pub fn peer_addresses(&self, peer_id: &PeerId) -> Option<&Vec> { self.peer_addresses.get(peer_id) } + + /// Removes the stored addresses for a peer. Returns the removed addresses, if any. + pub fn remove_peer_addresses(&mut self, peer_id: &PeerId) -> Option> { + self.peer_addresses.remove(peer_id) + } + + /// Number of peers with stored addresses. Test/diagnostic helper. + #[cfg(test)] + pub(crate) fn peer_addresses_count(&self) -> usize { + self.peer_addresses.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(n: usize) -> Peer { + Peer { + id: PeerId::random(), + connection_id: ConnectionId::new_unchecked(n), + remote_addr: "/ip4/127.0.0.1/tcp/9000".parse().unwrap(), + } + } + + #[test] + fn inactive_peers_evicts_oldest_beyond_cap() { + let mut store = PeerStore::default(); + let known = HashSet::new(); + + let peers: Vec = (0..(MAX_INACTIVE_PEERS + 10)).map(test_peer).collect(); + for peer in &peers { + store.remove_peer(peer.clone(), &known); + } + + assert_eq!(store.inactive_count(), MAX_INACTIVE_PEERS); + // Ordering queue membership stays consistent with the set. + assert_eq!(store.inactive_order.len(), store.inactive_count()); + + let all: Vec = store.all_peers(); + // The 10 oldest are evicted, the most recent are retained. + for old in &peers[..10] { + assert!(!all.contains(old), "oldest peer should have been evicted"); + } + for recent in &peers[peers.len() - 10..] { + assert!(all.contains(recent), "recent peer should be retained"); + } + } + + #[test] + fn known_peers_are_not_evicted() { + let mut store = PeerStore::default(); + let known_peer = test_peer(0); + let known: HashSet = std::iter::once(known_peer.id).collect(); + + // Insert the known peer first (oldest), then flood with non-known peers. + store.remove_peer(known_peer.clone(), &known); + for n in 1..=(MAX_INACTIVE_PEERS + 5) { + store.remove_peer(test_peer(n), &known); + } + + let all: Vec = store.all_peers(); + assert!( + all.contains(&known_peer), + "known peer must never be evicted" + ); + } + + #[test] + fn add_peer_then_remove_keeps_order_consistent() { + let mut store = PeerStore::default(); + let known = HashSet::new(); + let peer = test_peer(1); + + // Inactive, then re-activated: must be removed from the inactive set and + // the ordering queue so it is not double-counted when it goes inactive again. + store.remove_peer(peer.clone(), &known); + assert_eq!(store.inactive_count(), 1); + store.add_peer(peer.clone()); + assert_eq!(store.inactive_count(), 0); + assert_eq!(store.inactive_order.len(), 0); + + store.remove_peer(peer.clone(), &known); + assert_eq!(store.inactive_count(), 1); + assert_eq!(store.inactive_order.len(), 1); + } + + #[test] + fn remove_peer_addresses_removes_entry() { + let mut store = PeerStore::default(); + let peer = PeerId::random(); + let addrs: Vec = vec!["/ip4/127.0.0.1/tcp/9000".parse().unwrap()]; + + store.set_peer_addresses(peer, addrs.clone()); + assert_eq!(store.peer_addresses_count(), 1); + + assert_eq!(store.remove_peer_addresses(&peer), Some(addrs)); + assert!(store.peer_addresses(&peer).is_none()); + assert_eq!(store.remove_peer_addresses(&peer), None); + } } From d2061c7ca991322058907c56e4d28112ed4fdeda Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:27:42 +0200 Subject: [PATCH 05/12] fix(p2p): gate identify address writes to known peers and cap count Every identify::Event::Received stored the peer's self-asserted listen_addrs for any peer with no count bound. On a relay a remote can flood addresses that are never used, since all peer_addresses consumers look up known cluster peers only. Gate the write behind is_known_peer (dropping unknown-peer addrs without cloning) and truncate stored addresses to MAX_PEER_ADDRESSES=32. Hardening beyond Charon (peer_addresses is Pluto-specific); the gate bounds the key space to the finite cluster set. Builds on the T04 disconnect prune. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/p2p/src/p2p.rs | 79 ++++++++++++++++++++++++++++++++--- crates/p2p/src/p2p_context.rs | 46 +++++++++++++++++++- 2 files changed, 118 insertions(+), 7 deletions(-) diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index 907fb54e..c40ff1fb 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -589,16 +589,19 @@ impl Node { /// Handles a swarm event to update metrics and logging. fn handle_event(&mut self, event: &SwarmEvent>) { match event { - // Identify - update peer addresses in the peer store + // Identify - update peer addresses in the peer store. + // + // Only store addresses for known cluster peers: the addresses are + // attacker-controlled and the only consumers (quic_upgrade, + // force_direct, qbft/p2p, priority) look up addresses for known + // peers exclusively, so storing addresses for unknown peers is pure + // unbounded growth. SwarmEvent::Behaviour(PlutoBehaviourEvent::Identify(identify::Event::Received { peer_id, info, .. })) => { - // The peer addresses will be available in the next poll of the node. - self.p2p_context - .peer_store_write_lock() - .set_peer_addresses(*peer_id, info.listen_addrs.clone()); + store_identify_addrs(&self.p2p_context, peer_id, &info.listen_addrs); } // Ping metrics @@ -693,3 +696,69 @@ impl FusedStream for Node { false } } + +/// Stores identify-reported listen addresses for a peer, gated to known cluster +/// peers only. Addresses from unknown peers are dropped (and not cloned), since +/// the only consumers of `peer_addresses` look up known peers exclusively — so +/// storing them would be pure unbounded growth. The per-peer count is bounded by +/// [`PeerStore::set_peer_addresses`]. +fn store_identify_addrs(ctx: &P2PContext, peer_id: &PeerId, addrs: &[Multiaddr]) { + if ctx.is_known_peer(peer_id) { + // The peer addresses will be available in the next poll of the node. + ctx.peer_store_write_lock() + .set_peer_addresses(*peer_id, addrs.to_vec()); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn random_peer_id() -> PeerId { + libp2p::identity::Keypair::generate_ed25519() + .public() + .to_peer_id() + } + + fn addrs(n: usize) -> Vec { + (0..n) + .map(|i| format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)).parse().unwrap()) + .collect() + } + + #[test] + fn identify_addrs_stored_for_known_peer() { + let known = random_peer_id(); + let ctx = P2PContext::new([known]); + + store_identify_addrs(&ctx, &known, &addrs(2)); + + let store = ctx.peer_store_lock(); + assert_eq!(store.peer_addresses(&known).map(Vec::len), Some(2)); + } + + #[test] + fn identify_addrs_dropped_for_unknown_peer() { + let known = random_peer_id(); + let unknown = random_peer_id(); + let ctx = P2PContext::new([known]); + + store_identify_addrs(&ctx, &unknown, &addrs(3)); + + assert!(ctx.peer_store_lock().peer_addresses(&unknown).is_none()); + } + + #[test] + fn identify_addrs_capped_for_known_peer() { + let known = random_peer_id(); + let ctx = P2PContext::new([known]); + + store_identify_addrs(&ctx, &known, &addrs(crate::p2p_context::MAX_PEER_ADDRESSES + 1)); + + let store = ctx.peer_store_lock(); + assert_eq!( + store.peer_addresses(&known).map(Vec::len), + Some(crate::p2p_context::MAX_PEER_ADDRESSES) + ); + } +} diff --git a/crates/p2p/src/p2p_context.rs b/crates/p2p/src/p2p_context.rs index 69f2b678..23846d48 100644 --- a/crates/p2p/src/p2p_context.rs +++ b/crates/p2p/src/p2p_context.rs @@ -12,6 +12,14 @@ use tracing::error; /// stored, bounding memory growth from connection churn by remote peers. const MAX_INACTIVE_PEERS: usize = 1024; +/// Maximum number of listen addresses stored per peer (from the identify +/// protocol). `listen_addrs` is attacker-controlled in both content and +/// length; legitimate cluster peers advertise only a handful of addresses, so +/// this cap drops nothing real while bounding per-peer memory. Excess +/// addresses beyond the cap are discarded (the first `MAX_PEER_ADDRESSES` are +/// kept). +pub(crate) const MAX_PEER_ADDRESSES: usize = 32; + /// Global context shared across P2P components. /// /// This struct provides thread-safe access to shared state including: @@ -142,7 +150,7 @@ impl PeerStore { let mut scanned = 0usize; let max_scan = self.inactive_order.len(); while self.inactive_peers.len() > MAX_INACTIVE_PEERS && scanned < max_scan { - scanned += 1; + scanned = scanned.saturating_add(1); let Some(candidate) = self.inactive_order.pop_front() else { break; }; @@ -193,7 +201,13 @@ impl PeerStore { } /// Sets the known addresses for a peer (from identify protocol). - pub fn set_peer_addresses(&mut self, peer_id: PeerId, addrs: Vec) { + /// + /// The address list is truncated to [`MAX_PEER_ADDRESSES`] before storing, + /// because `addrs` is attacker-controlled and otherwise unbounded. + pub fn set_peer_addresses(&mut self, peer_id: PeerId, mut addrs: Vec) { + if addrs.len() > MAX_PEER_ADDRESSES { + addrs.truncate(MAX_PEER_ADDRESSES); + } self.peer_addresses.insert(peer_id, addrs); } @@ -288,6 +302,34 @@ mod tests { assert_eq!(store.inactive_order.len(), 1); } + #[test] + fn set_peer_addresses_truncates_to_cap() { + let mut store = PeerStore::default(); + let peer = PeerId::random(); + let addrs: Vec = (0..(MAX_PEER_ADDRESSES + 50)) + .map(|i| format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)).parse().unwrap()) + .collect(); + + store.set_peer_addresses(peer, addrs.clone()); + + let stored = store.peer_addresses(&peer).unwrap(); + assert_eq!(stored.len(), MAX_PEER_ADDRESSES); + // truncate keeps the prefix + assert_eq!(stored.as_slice(), &addrs[..MAX_PEER_ADDRESSES]); + } + + #[test] + fn set_peer_addresses_keeps_under_cap() { + let mut store = PeerStore::default(); + let peer = PeerId::random(); + let addrs: Vec = (0..3) + .map(|i| format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)).parse().unwrap()) + .collect(); + + store.set_peer_addresses(peer, addrs.clone()); + assert_eq!(store.peer_addresses(&peer), Some(&addrs)); + } + #[test] fn remove_peer_addresses_removes_entry() { let mut store = PeerStore::default(); From c88fea6e2ca72d6e3365a900675794c90118e578 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:31:23 +0200 Subject: [PATCH 06/12] fix(relay-server): restore rate limiters and add per-IP reservation cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit create_relay_config built relay::Config with a struct literal that hard-coded reservation_rate_limiters and circuit_src_rate_limiters to empty, discarding rust-libp2p's default per-peer (30/2min) and per-IP (60/min) token-bucket limiters — leaving the public relay open to reservation/circuit floods. Build from relay::Config::default() (via functional-update) to retain the four defaults, and append a per-IP reservation rate limiter sized from max_res_per_peer, mirroring Charon's relay.DefaultResources() + MaxReservationsPerIP = MaxResPerPeer. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/relay-server/src/config.rs | 92 ++++++++++++++++++++++++++++--- 1 file changed, 85 insertions(+), 7 deletions(-) diff --git a/crates/relay-server/src/config.rs b/crates/relay-server/src/config.rs index 56310d23..440a84c2 100644 --- a/crates/relay-server/src/config.rs +++ b/crates/relay-server/src/config.rs @@ -1,4 +1,4 @@ -use std::{path::PathBuf, time::Duration}; +use std::{num::NonZeroU32, path::PathBuf, time::Duration}; use bon::Builder; use libp2p::relay; @@ -11,6 +11,14 @@ pub const ONE_HOUR_SECONDS: u64 = 60 * 60; pub const ONE_MINUTE_SECONDS: u64 = 60; /// 32 MB in bytes. pub const MB_32: u64 = 32 * 1024 * 1024; +/// Per-IP reservation rate limit (token-bucket capacity). +/// +/// rust-libp2p's relay has no per-IP reservation *count* cap (unlike Charon's +/// go-libp2p `MaxReservationsPerIP`), so we mirror Charon's intent with the +/// per-IP reservation *rate* limiter instead. `relay::Config::default()` uses +/// 60 reservations / minute per IP; we keep that default capacity as the +/// fallback when the operator leaves `max_res_per_peer` at 0. +pub const RESERVATIONS_PER_IP_PER_MINUTE: u32 = 60; /// External host resolve interval. pub const EXTERNAL_HOST_RESOLVE_INTERVAL: Duration = Duration::from_secs(5 * 60); @@ -47,18 +55,88 @@ pub struct Config { } pub(crate) fn create_relay_config(config: &Config) -> relay::Config { - relay::Config { + // Start from rust-libp2p defaults so the per-peer and per-IP reservation / + // circuit-source rate limiters are preserved (they guard against + // reservation/circuit floods from a single peer or IP). The defaults are: + // - reservation per-peer: 30 / 2min - reservation per-IP: 60 / 1min + // - circuit-src per-peer: 30 / 2min - circuit-src per-IP: 60 / 1min + // We only override the count / duration / byte limits below, matching + // Charon's `relay.DefaultResources()` + selective overrides + // (charon@v1.7.1 cmd/relay/p2p.go:61-67). + let relay_config = relay::Config { max_reservations: config.max_conns, max_reservations_per_peer: config.max_res_per_peer, reservation_duration: Duration::from_secs(ONE_HOUR_SECONDS), - reservation_rate_limiters: vec![], - // todo(varex83): check if this is correct, since it's aligned with the original - // implementation, but I'm not sure if it's the correct way to do it. - // Would it be better to use max_res_per_peer * max_conns? max_circuits: config.max_res_per_peer, max_circuits_per_peer: config.max_res_per_peer, max_circuit_duration: Duration::from_secs(ONE_MINUTE_SECONDS), max_circuit_bytes: MB_32, - circuit_src_rate_limiters: vec![], + // Restore the default rate limiters dropped by the previous + // struct-literal construction. + ..relay::Config::default() + }; + + // Charon sets MaxReservationsPerIP = MaxResPerPeer. rust-libp2p has no per-IP + // reservation *count* cap, so we approximate it with an additional per-IP + // reservation *rate* limiter sized from `max_res_per_peer` (falling back to + // the default capacity when the operator leaves it at 0). This appends a + // second per-IP reservation limiter on top of the default one; both must + // pass, so the effective per-IP rate becomes min(60/min, max_res_per_peer/min). + let per_ip_limit = u32::try_from(config.max_res_per_peer) + .ok() + .and_then(NonZeroU32::new) + .unwrap_or_else(|| { + NonZeroU32::new(RESERVATIONS_PER_IP_PER_MINUTE).expect("60 > 0") + }); + + relay_config + .reservation_rate_per_ip(per_ip_limit, Duration::from_secs(ONE_MINUTE_SECONDS)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_config(max_conns: usize, max_res_per_peer: usize) -> Config { + Config::builder() + .p2p_config(P2PConfig::default()) + .max_conns(max_conns) + .max_res_per_peer(max_res_per_peer) + .build() + } + + #[test] + fn restored_rate_limiters_present() { + let relay_config = create_relay_config(&test_config(64, 8)); + // 2 default reservation limiters (per-peer, per-IP) + 1 added per-IP. + assert_eq!(relay_config.reservation_rate_limiters.len(), 3); + // 2 default circuit-src limiters (per-peer, per-IP) preserved. + assert_eq!(relay_config.circuit_src_rate_limiters.len(), 2); + } + + #[test] + fn count_limits_from_config() { + let relay_config = create_relay_config(&test_config(64, 8)); + assert_eq!(relay_config.max_reservations, 64); + assert_eq!(relay_config.max_reservations_per_peer, 8); + assert_eq!(relay_config.max_circuits, 8); + assert_eq!(relay_config.max_circuits_per_peer, 8); + assert_eq!( + relay_config.reservation_duration, + Duration::from_secs(ONE_HOUR_SECONDS) + ); + assert_eq!( + relay_config.max_circuit_duration, + Duration::from_secs(ONE_MINUTE_SECONDS) + ); + assert_eq!(relay_config.max_circuit_bytes, MB_32); + } + + #[test] + fn zero_max_res_per_peer_uses_default_per_ip_capacity() { + // max_res_per_peer = 0 must not panic (NonZeroU32 fallback path) and + // still yields the restored + added reservation rate limiters. + let relay_config = create_relay_config(&test_config(64, 0)); + assert_eq!(relay_config.reservation_rate_limiters.len(), 3); } } From d88bdeeb8a2bc86d841d481ff3978504f6e7bb5a Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:34:14 +0200 Subject: [PATCH 07/12] fix(eth2util): cap concurrent keystore decryption to 10 workers load_files_unordered spawned one async task per keystore file with no cap and ran the CPU/memory-heavy scrypt KDF directly on the async runtime; a directory of many keystores could spawn unbounded tasks and starve the reactor. Gate both loaders with a 10-permit Semaphore (acquired before spawn, matching Charon's loadStoreWorkers=10) and move the KDF in load_files_unordered onto spawn_blocking. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/eth2util/src/keystore/load.rs | 81 ++++++++++++++++++++++++++-- 1 file changed, 78 insertions(+), 3 deletions(-) diff --git a/crates/eth2util/src/keystore/load.rs b/crates/eth2util/src/keystore/load.rs index 62a299ce..ff358d8c 100644 --- a/crates/eth2util/src/keystore/load.rs +++ b/crates/eth2util/src/keystore/load.rs @@ -11,6 +11,13 @@ use super::{ store::Keystore, }; +/// Maximum number of keystore files decrypted concurrently. +/// +/// Mirrors Charon's `loadStoreWorkers` (`eth2util/keystore/keystore.go`), +/// bounding the blocking key-derivation work so a directory containing many +/// keystore files cannot saturate the async runtime / blocking thread pool. +const LOAD_STORE_WORKERS: usize = 10; + /// Wraps a list of key files with convenience functions. #[derive(Debug)] pub struct KeyFiles(Vec); @@ -84,6 +91,10 @@ pub struct KeyFile { pub async fn load_files_unordered(dir: impl AsRef) -> Result { let mut read_dir = tokio::fs::read_dir(dir.as_ref()).await?; let mut set = tokio::task::JoinSet::new(); + // Cap concurrency at LOAD_STORE_WORKERS (matching Charon's loadStoreWorkers). + // Acquire an owned permit *before* spawning so we never park thousands of + // tasks up front for a large directory. + let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(LOAD_STORE_WORKERS)); while let Some(entry) = read_dir.next_entry().await? { let path = entry.path(); @@ -95,14 +106,28 @@ pub async fn load_files_unordered(dir: impl AsRef) -> Result { continue; } + // `Semaphore` is never closed, so acquire cannot fail. + let permit = std::sync::Arc::clone(&sem) + .acquire_owned() + .await + .expect("semaphore not closed"); + set.spawn(async move { + let _permit = permit; // released when this task completes + let b = tokio::fs::read_to_string(&path).await?; let store: Keystore = serde_json::from_str(&b)?; - let password = super::store::load_password(&path).await?; - let private_key = super::store::decrypt(&store, &password)?; let file_index = extract_file_index(path.to_string_lossy())?; + // `decrypt` runs scrypt/PBKDF2 (CPU- and memory-heavy); run it on the + // blocking pool so it never blocks an async reactor thread. + let (private_key, path) = tokio::task::spawn_blocking(move || { + let key = super::store::decrypt(&store, &password)?; + Ok::<_, KeystoreError>((key, path)) + }) + .await??; + Ok::(KeyFile { private_key, filename: path, @@ -181,9 +206,10 @@ pub async fn load_files_recursively(dir: impl AsRef) -> Result { passwords_map.insert(filepath.clone(), b); } - // Step 4: Decrypt keystores concurrently. + // Step 4: Decrypt keystores concurrently, capped at LOAD_STORE_WORKERS. let mut set = tokio::task::JoinSet::new(); let passwords_map = std::sync::Arc::new(passwords_map); + let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(LOAD_STORE_WORKERS)); for filepath in valid_files { let store = @@ -197,9 +223,18 @@ pub async fn load_files_recursively(dir: impl AsRef) -> Result { let password_file = filepath.with_extension("txt"); let passwords = std::sync::Arc::clone(&passwords_map); + // Acquire a permit before spawning so no more than LOAD_STORE_WORKERS + // blocking KDF tasks are in flight at once. `Semaphore` is never closed, + // so acquire cannot fail. + let permit = std::sync::Arc::clone(&sem) + .acquire_owned() + .await + .expect("semaphore not closed"); + // `decrypt` is CPU-intensive (key derivation), so use `spawn_blocking` to avoid // blocking the async runtime. The closure has no `.await` calls. set.spawn_blocking(move || { + let _permit = permit; // released when this blocking task finishes // First try the password file that matches the keystore file. let mut err = None; @@ -520,6 +555,46 @@ mod tests { assert_eq!(expected, actual, "keys mismatch"); } + #[tokio::test] + async fn load_unordered_many_files_capped() { + // More files than LOAD_STORE_WORKERS exercises the semaphore-gated path. + let dir = TempDir::new().unwrap(); + let mut expected = std::collections::HashSet::new(); + for i in 0..25usize { + let target = dir.path().join(format!("keystore-{i}.json")); + expected.insert(store_new_key_for_test(&target).await); + } + let key_files = load_files_unordered(dir.path()).await.unwrap(); + assert_eq!(key_files.len(), 25); + for kf in key_files.iter() { + assert!(expected.contains(&kf.private_key)); + } + } + + #[tokio::test] + async fn load_recursively_many_files_capped() { + // More files than LOAD_STORE_WORKERS, spread across nested dirs. + let dir = TempDir::new().unwrap(); + let nested = dir.path().join("nested"); + std::fs::create_dir(&nested).unwrap(); + + let mut expected = std::collections::HashSet::new(); + for i in 0..15usize { + let target = if i % 2 == 0 { + dir.path().join(format!("keystore-{i}.json")) + } else { + nested.join(format!("keystore-{i}.json")) + }; + expected.insert(store_new_key_for_test(&target).await); + } + + let key_files = load_files_recursively(dir.path()).await.unwrap(); + assert_eq!(key_files.len(), 15); + for kf in key_files.iter() { + assert!(expected.contains(&kf.private_key)); + } + } + #[tokio::test] async fn load_files_recursively_test() { let dir = TempDir::new().unwrap(); From 76dc7c9f2b61815fb745fc17ddfb13802715ecf8 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:37:54 +0200 Subject: [PATCH 08/12] fix(consensus): cap justification count per inbound QBFT message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consensus::handle verified every entry in pb_msg.justification (a secp256k1 recovery each), bounded only by the 32MB wire cap — a large message could pack many small entries into O(many) recoveries (Byzantine-insider CPU amplification). Reject before the loop when the count exceeds MAX_JUSTIFICATIONS_PER_NODE * node_count(); honest QBFT never exceeds O(node_count). Defensive hardening beyond Charon; sized as a superset of every honest justification set. Part 1 of T08. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/consensus/src/qbft/component.rs | 74 ++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/crates/consensus/src/qbft/component.rs b/crates/consensus/src/qbft/component.rs index f3638186..73130b19 100644 --- a/crates/consensus/src/qbft/component.rs +++ b/crates/consensus/src/qbft/component.rs @@ -32,6 +32,15 @@ use super::{ runner, }; +/// Maximum justification entries accepted per inbound consensus message, +/// expressed as a multiple of the cluster node count. Honest QBFT produces at +/// most one justification per peer per justified transition, so a small +/// multiple of `node_count()` is a strict superset of every legitimate set +/// while bounding per-message secp256k1 recovery work. Charon has no explicit +/// cap (it is implicitly bounded by quorum/peer count + transport size); this +/// is defensive hardening that cannot reject a legitimate justification set. +const MAX_JUSTIFICATIONS_PER_NODE: usize = 4; + /// Result returned by outbound QBFT broadcasting. pub type BroadcastResult = std::result::Result<(), Box>; @@ -164,6 +173,15 @@ pub enum Error { #[error("qbft justification duty differs from message duty")] JustificationDutyDiffers, + /// Inbound message carried more justifications than the per-message cap. + #[error("too many justifications: {count} exceeds maximum {max}")] + TooManyJustifications { + /// Number of justifications in the message. + count: usize, + /// The enforced maximum (MAX_JUSTIFICATIONS_PER_NODE * node_count()). + max: usize, + }, + /// Inbound Any could not be decoded. #[error("unmarshal any")] UnmarshalAny, @@ -355,6 +373,17 @@ impl Consensus { return Err(Error::InvalidDuty); } + // Bound the number of justifications before any secp256k1 recovery runs: + // a 32MB message could otherwise pack a huge number of small entries, + // each forcing a recovery. Honest QBFT never exceeds O(node_count). + let max_justifications = MAX_JUSTIFICATIONS_PER_NODE.saturating_mul(self.node_count()); + if pb_msg.justification.len() > max_justifications { + return Err(Error::TooManyJustifications { + count: pb_msg.justification.len(), + max: max_justifications, + }); + } + for justification in &pb_msg.justification { self.verify_msg(justification) .map_err(|err| Error::InvalidJustification(Box::new(err)))?; @@ -930,6 +959,51 @@ pub(crate) mod tests { assert_eq!(recv_rx.try_recv().unwrap().justification().len(), 1); } + #[tokio::test] + async fn handle_rejects_too_many_justifications() { + let consensus = consensus(0, true); + let max = MAX_JUSTIFICATIONS_PER_NODE * consensus.node_count(); + + let mut justification = unsigned_msg(0); + justification.r#type = i64::from(qbft::MSG_ROUND_CHANGE); + justification.value_hash = Bytes::new(); + let signed = sign_for_peer(justification, 0); + + let mut outer = valid_consensus_msg(0); + outer.justification = std::iter::repeat_n(signed, max + 1).collect(); + + let err = consensus + .handle(outer, &CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!(err, Error::TooManyJustifications { .. })); + assert!(err.to_string().contains("too many justifications")); + } + + #[tokio::test] + async fn handle_accepts_max_justifications() { + // Exactly `max` justifications must not trip the cap (guards `>` vs `>=`). + let consensus = consensus(0, true); + let inst = consensus.get_instance_io(duty()); + let max = MAX_JUSTIFICATIONS_PER_NODE * consensus.node_count(); + + let mut justification = unsigned_msg(0); + justification.r#type = i64::from(qbft::MSG_ROUND_CHANGE); + justification.value_hash = Bytes::new(); + let signed = sign_for_peer(justification, 0); + + let mut outer = valid_consensus_msg(0); + outer.justification = std::iter::repeat_n(signed, max).collect(); + + consensus + .handle(outer, &CancellationToken::new()) + .await + .unwrap(); + + let mut recv_rx = inst.take_recv_rx().unwrap(); + assert_eq!(recv_rx.try_recv().unwrap().justification().len(), max); + } + #[test] fn values_by_hash_rejects_invalid_type_url() { let err = values_by_hash(&[Any { From 531e592197702da9fbe06954588647bf577e3cf7 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:51:46 +0200 Subject: [PATCH 09/12] fix: bound HTTP response bodies and add per-request timeouts Several clients read remote bodies with no size cap (and some no timeout), letting a hostile/slow upstream exhaust memory or stall: - obolapi http_get: stream body capped at 16MB; error bodies truncated - cluster fetch_definition: stream body capped at 16MB before decode - eth1wrap ERC-1271 call: wrap in a 10s per-call timeout - p2p bootnode relay query: client timeout + 1MB body cap - CLI mev/beacon diagnostics: shared timeout-configured client + body caps Body caps are hardening beyond Charon, each sized well above any real response so no legitimate body is rejected; timeouts are parity-aligned. Part 2 of T08. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- Cargo.lock | 1 + crates/app/src/obolapi/client.rs | 58 ++++++++++++++++++++-- crates/app/src/obolapi/error.rs | 7 +++ crates/cli/src/commands/test/beacon.rs | 20 ++++++-- crates/cli/src/commands/test/mev.rs | 40 +++++++++++++--- crates/cli/src/error.rs | 4 ++ crates/cluster/Cargo.toml | 3 +- crates/cluster/src/helpers.rs | 66 +++++++++++++++++++++++++- crates/eth1wrap/Cargo.toml | 1 + crates/eth1wrap/src/lib.rs | 23 +++++++-- crates/p2p/Cargo.toml | 2 +- crates/p2p/src/bootnode.rs | 52 ++++++++++++++++++-- 12 files changed, 249 insertions(+), 28 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8f13ed09..96491a56 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5558,6 +5558,7 @@ name = "pluto-cluster" version = "1.7.1" dependencies = [ "chrono", + "futures", "hex", "k256", "libp2p", diff --git a/crates/app/src/obolapi/client.rs b/crates/app/src/obolapi/client.rs index 84eb2c4c..02d45ae9 100644 --- a/crates/app/src/obolapi/client.rs +++ b/crates/app/src/obolapi/client.rs @@ -6,6 +6,7 @@ use std::time::Duration; use bon::Builder; +use futures::StreamExt; use pluto_cluster::lock::Lock; use reqwest::{Method, StatusCode}; use url::Url; @@ -15,6 +16,56 @@ use crate::obolapi::error::{Error, Result}; /// Default HTTP request timeout if not specified. const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); +/// Maximum body size read from a successful Obol API response (16 MB). A +/// full-exit response holds up to `share_count` hex BLS signatures; 16 MB is +/// orders of magnitude above any real cluster, so this never rejects a +/// legitimate response while bounding memory against a hostile upstream. +const OBOLAPI_MAX_BODY: usize = 16 * 1024 * 1024; + +/// Maximum body size read from an HTTP error response (diagnostics only). +const OBOLAPI_MAX_ERROR_BODY: usize = 64 * 1024; + +/// Reads a response body, failing with [`Error::BodyTooLarge`] if it would +/// exceed `max` bytes. Streams so the cap is enforced before full buffering. +async fn read_body_capped(response: reqwest::Response, max: usize) -> Result> { + // Fast-path reject if the server advertised an oversized length. + if let Some(len) = response.content_length() + && len > max as u64 + { + return Err(Error::BodyTooLarge { limit: max }); + } + + let mut buf = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + if buf.len().saturating_add(chunk.len()) > max { + return Err(Error::BodyTooLarge { limit: max }); + } + buf.extend_from_slice(&chunk); + } + Ok(buf) +} + +/// Reads an HTTP error body for diagnostics, truncating at `max` bytes so a +/// hostile error response cannot exhaust memory. Never fails (best-effort). +async fn read_error_body(response: reqwest::Response, max: usize) -> String { + let mut buf = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(Ok(chunk)) = stream.next().await { + let remaining = max.saturating_sub(buf.len()); + if remaining == 0 { + break; + } + let take = remaining.min(chunk.len()); + buf.extend_from_slice(&chunk[..take]); + if buf.len() >= max { + break; + } + } + String::from_utf8_lossy(&buf).into_owned() +} + /// REST client for Obol API requests. #[derive(Debug, Clone)] pub struct Client { @@ -98,7 +149,7 @@ impl Client { let status = response.status(); if !status.is_success() { - let body_text = response.text().await.unwrap_or_default(); + let body_text = read_error_body(response, OBOLAPI_MAX_ERROR_BODY).await; return Err(Error::HttpError { method: Method::POST, @@ -133,7 +184,7 @@ impl Client { return Err(Error::NoExit); } - let body_text = response.text().await.unwrap_or_default(); + let body_text = read_error_body(response, OBOLAPI_MAX_ERROR_BODY).await; return Err(Error::HttpError { method: Method::GET, @@ -142,8 +193,7 @@ impl Client { }); } - let body_bytes = response.bytes().await?.to_vec(); - Ok(body_bytes) + read_body_capped(response, OBOLAPI_MAX_BODY).await } /// Makes an HTTP DELETE request. diff --git a/crates/app/src/obolapi/error.rs b/crates/app/src/obolapi/error.rs index 5ad99b01..cb1ffa04 100644 --- a/crates/app/src/obolapi/error.rs +++ b/crates/app/src/obolapi/error.rs @@ -31,6 +31,13 @@ pub enum Error { #[error("HTTP client error: {0}")] Reqwest(#[from] reqwest::Error), + /// Response body exceeded the allowed size. + #[error("response body exceeds {limit} bytes")] + BodyTooLarge { + /// The enforced maximum body size in bytes. + limit: usize, + }, + /// JSON serialization/deserialization error. #[error("JSON error: {0}")] Json(#[from] serde_json::Error), diff --git a/crates/cli/src/commands/test/beacon.rs b/crates/cli/src/commands/test/beacon.rs index 6635fe43..2a549741 100644 --- a/crates/cli/src/commands/test/beacon.rs +++ b/crates/cli/src/commands/test/beacon.rs @@ -29,6 +29,18 @@ use tokio::{ }; use tokio_util::sync::CancellationToken; +/// Per-request timeout for beacon-node diagnostic HTTP calls, so a hostile or +/// slow endpoint cannot stall a diagnostic indefinitely. +const BEACON_HTTP_TIMEOUT: StdDuration = StdDuration::from_secs(10); + +/// Builds a diagnostic HTTP client with a request timeout. +fn http_client() -> reqwest::Client { + reqwest::Client::builder() + .timeout(BEACON_HTTP_TIMEOUT) + .build() + .unwrap_or_default() +} + const THRESHOLD_BEACON_MEASURE_AVG: StdDuration = StdDuration::from_millis(40); const THRESHOLD_BEACON_MEASURE_POOR: StdDuration = StdDuration::from_millis(100); const THRESHOLD_BEACON_LOAD_AVG: StdDuration = StdDuration::from_millis(40); @@ -448,7 +460,7 @@ async fn beacon_version_test( let mut res = TestResult::new("Version"); let url = format!("{target}/eth/v1/node/version"); - let client = reqwest::Client::new(); + let client = http_client(); let resp = match client.get(&url).send().await { Ok(r) => r, Err(e) => return res.fail(e), @@ -498,7 +510,7 @@ async fn beacon_is_synced_test( let mut res = TestResult::new("Synced"); let url = format!("{target}/eth/v1/node/syncing"); - let client = reqwest::Client::new(); + let client = http_client(); let resp = match client.get(&url).send().await { Ok(r) => r, Err(e) => return res.fail(e), @@ -543,7 +555,7 @@ async fn beacon_peer_count_test( let mut res = TestResult::new("PeerCount"); let url = format!("{target}/eth/v1/node/peers?state=connected"); - let client = reqwest::Client::new(); + let client = http_client(); let resp = match client.get(&url).send().await { Ok(r) => r, Err(e) => return res.fail(e), @@ -1551,7 +1563,7 @@ async fn sync_committee_message_duty( async fn get_current_slot(target: &str) -> CliResult { let url = format!("{target}/eth/v1/node/syncing"); - let client = reqwest::Client::new(); + let client = http_client(); let resp = client.get(&url).send().await?; // More strict than the Charon check, which requires the status code to be > diff --git a/crates/cli/src/commands/test/mev.rs b/crates/cli/src/commands/test/mev.rs index 4207f1f3..4a61d38a 100644 --- a/crates/cli/src/commands/test/mev.rs +++ b/crates/cli/src/commands/test/mev.rs @@ -21,6 +21,32 @@ use crate::{ }; use clap::Args; +/// Per-request timeout for MEV/beacon diagnostic HTTP calls. +const MEV_HTTP_TIMEOUT: Duration = Duration::from_secs(10); +/// Maximum diagnostic response body read from a beacon/relay endpoint (16 MB). +const BN_MAX_BODY: usize = 16 * 1024 * 1024; + +/// Builds a diagnostic HTTP client with a request timeout so a hostile/slow +/// endpoint cannot stall a diagnostic indefinitely. +fn http_client() -> reqwest::Client { + reqwest::Client::builder() + .timeout(MEV_HTTP_TIMEOUT) + .build() + .unwrap_or_default() +} + +/// Reads a response body, rejecting bodies that exceed [`BN_MAX_BODY`]. Uses the +/// advertised `Content-Length` for the fast-path reject; the client timeout +/// bounds a slow/absent-length body. +async fn read_body_capped(resp: reqwest::Response) -> Result> { + if let Some(len) = resp.content_length() + && len > BN_MAX_BODY as u64 + { + return Err(MevTestError::BodyTooLarge(BN_MAX_BODY).into()); + } + Ok(resp.bytes().await?.to_vec()) +} + /// Thresholds for MEV ping measure test. const THRESHOLD_MEV_MEASURE_AVG: Duration = Duration::from_millis(40); /// Threshold for poor MEV ping measure. @@ -233,7 +259,7 @@ async fn test_single_mev( async fn mev_ping_test(target: &str, _conf: &TestMevArgs) -> TestResult { let test_res = TestResult::new("Ping"); let url = format!("{target}/eth/v1/builder/status"); - let client = reqwest::Client::new(); + let client = http_client(); let resp = match client.get(&url).send().await { Ok(r) => r, @@ -459,8 +485,8 @@ struct BuilderBidResponse { async fn latest_beacon_block(endpoint: &str) -> Result { let url = format!("{endpoint}/eth/v2/beacon/blocks/head"); - let resp = reqwest::Client::new().get(&url).send().await?; - let body = resp.bytes().await?; + let resp = http_client().get(&url).send().await?; + let body = read_body_capped(resp).await?; let block: BeaconBlock = serde_json::from_slice(&body)?; Ok(block.data.message) @@ -471,8 +497,8 @@ async fn fetch_proposers_for_epoch( epoch: i64, ) -> Result> { let url = format!("{beacon_endpoint}/eth/v1/validator/duties/proposer/{epoch}"); - let resp = reqwest::Client::new().get(&url).send().await?; - let body = resp.bytes().await?; + let resp = http_client().get(&url).send().await?; + let body = read_body_capped(resp).await?; let duties: ProposerDuties = serde_json::from_slice(&body)?; Ok(duties.data) @@ -497,7 +523,7 @@ async fn get_block_header( let start = Instant::now(); - let resp = reqwest::Client::new().get(&url).send().await?; + let resp = http_client().get(&url).send().await?; let rtt = start.elapsed(); @@ -505,7 +531,7 @@ async fn get_block_header( return Err(MevTestError::StatusCodeNot200.into()); } - let body = resp.bytes().await?; + let body = read_body_capped(resp).await?; let bid: BuilderBidResponse = serde_json::from_slice(&body)?; diff --git a/crates/cli/src/error.rs b/crates/cli/src/error.rs index 3a1f35f3..e9cb5350 100644 --- a/crates/cli/src/error.rs +++ b/crates/cli/src/error.rs @@ -142,6 +142,10 @@ pub enum MevTestError { #[error("status code {0}")] HttpStatus(u16), + /// Response body exceeded the allowed size. + #[error("response body exceeds {0} bytes")] + BodyTooLarge(usize), + /// Beacon node endpoint required but not provided. #[error("beacon-node-endpoint required when load-test enabled")] BeaconNodeEndpointRequired, diff --git a/crates/cluster/Cargo.toml b/crates/cluster/Cargo.toml index 32bdb442..1da9e673 100644 --- a/crates/cluster/Cargo.toml +++ b/crates/cluster/Cargo.toml @@ -28,7 +28,8 @@ pluto-k1util.workspace = true pluto-ssz.workspace = true k256.workspace = true tokio.workspace = true -reqwest = { workspace = true, features = ["json"] } +futures.workspace = true +reqwest = { workspace = true, features = ["json", "stream"] } # Workaround to use test code from different crate. # See: https://github.com/NethermindEth/pluto/pull/285 pluto-testutil = { workspace = true, optional = true } diff --git a/crates/cluster/src/helpers.rs b/crates/cluster/src/helpers.rs index abc19e3a..28fcdbf5 100644 --- a/crates/cluster/src/helpers.rs +++ b/crates/cluster/src/helpers.rs @@ -37,6 +37,12 @@ pub fn verify_sig( Ok(expected_addr == actual_addr) } +/// Maximum cluster-definition response body read from a remote URI (16 MB). A +/// definition with the SSZ-max 65536 validators is well under this, so the cap +/// never rejects a legitimate definition while bounding memory against a +/// hostile upstream. +const DEFINITION_MAX_BODY: usize = 16 * 1024 * 1024; + /// Error type returned by `fetch_definition`. #[derive(Debug, thiserror::Error)] pub enum FetchError { @@ -47,6 +53,14 @@ pub enum FetchError { /// HTTP error while fetching the definition. #[error("HTTP error {0}")] Http(#[from] reqwest::Error), + + /// Response body exceeded the allowed size. + #[error("definition body exceeds {0} bytes")] + BodyTooLarge(usize), + + /// JSON decode error after the capped read. + #[error("decode definition: {0}")] + Decode(#[from] serde_json::Error), } /// Fetch cluster definition file from a remote URI. @@ -55,13 +69,42 @@ pub async fn fetch_definition( ) -> std::result::Result { let definition = tokio::time::timeout(std::time::Duration::from_secs(10), async { let response = reqwest::get(url).await?.error_for_status()?; - response.json::().await + let buf = read_body_capped(response, DEFINITION_MAX_BODY).await?; + Ok::(serde_json::from_slice::(&buf)?) }) .await??; Ok(definition) } +/// Reads a response body, failing with [`FetchError::BodyTooLarge`] if it would +/// exceed `max` bytes. Streams so the cap bounds memory even without a +/// trustworthy `Content-Length` header. +async fn read_body_capped( + response: reqwest::Response, + max: usize, +) -> std::result::Result, FetchError> { + use futures::StreamExt; + + // Reject early if the server advertised an oversized body. + if let Some(len) = response.content_length() + && len > max as u64 + { + return Err(FetchError::BodyTooLarge(max)); + } + + let mut buf = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + if buf.len().saturating_add(chunk.len()) > max { + return Err(FetchError::BodyTooLarge(max)); + } + buf.extend_from_slice(&chunk); + } + Ok(buf) +} + /// Creates a new directory for validator keys. /// If the directory "validator_keys" exists, it checks if the directory is /// empty. @@ -253,7 +296,26 @@ mod tests { let response = super::fetch_definition(format!("{}/invalidDef", &server.uri())).await; - assert!(matches!(response, Err(super::FetchError::Http(e)) if e.is_decode())); + assert!(matches!(response, Err(super::FetchError::Decode(_)))); + } + + #[tokio::test] + async fn read_body_capped_rejects_oversized() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .and(wiremock::matchers::path("/big")) + .respond_with( + wiremock::ResponseTemplate::new(200) + .set_body_raw(vec![b'x'; 100], "application/json"), + ) + .mount(&server) + .await; + + let response = reqwest::get(format!("{}/big", &server.uri())) + .await + .unwrap(); + let result = super::read_body_capped(response, 10).await; + assert!(matches!(result, Err(super::FetchError::BodyTooLarge(10)))); } #[tokio::test] diff --git a/crates/eth1wrap/Cargo.toml b/crates/eth1wrap/Cargo.toml index 625295bc..93ab42e9 100644 --- a/crates/eth1wrap/Cargo.toml +++ b/crates/eth1wrap/Cargo.toml @@ -10,6 +10,7 @@ publish.workspace = true alloy.workspace = true url.workspace = true thiserror.workspace = true +tokio.workspace = true [dev-dependencies] tokio.workspace = true diff --git a/crates/eth1wrap/src/lib.rs b/crates/eth1wrap/src/lib.rs index b0457ec1..b8022ad6 100644 --- a/crates/eth1wrap/src/lib.rs +++ b/crates/eth1wrap/src/lib.rs @@ -13,6 +13,13 @@ sol!( "src/build/IERC1271.abi" ); +/// Per-call timeout (seconds) for the ERC-1271 `isValidSignature` request. +/// +/// The provider carries a `RetryBackoffLayer` (`MAX_RETRY = 10`) but no request +/// timeout, so a hostile/slow EL endpoint could otherwise stall the call +/// indefinitely. `tokio::time::timeout` wraps the whole retried operation here. +const ERC1271_CALL_TIMEOUT_SECS: u64 = 10; + type Result = std::result::Result; /// Defines errors that can occur when interacting with the Ethereum client. @@ -37,6 +44,10 @@ pub enum EthClientError { /// No execution engine endpoint was configured. #[error("execution engine endpoint is not set")] NoExecutionEngineAddr, + + /// The ERC-1271 verification call did not complete within the timeout. + #[error("ERC-1271 call timed out")] + CallTimeout, } /// Defines the interface for the Ethereum EL RPC client. @@ -95,10 +106,14 @@ impl EthClient { let instance = IERC1271::new(address, provider); - let call = instance - .isValidSignature(hash.into(), sig.to_vec().into()) - .call() - .await?; + let call = tokio::time::timeout( + std::time::Duration::from_secs(ERC1271_CALL_TIMEOUT_SECS), + instance + .isValidSignature(hash.into(), sig.to_vec().into()) + .call(), + ) + .await + .map_err(|_| EthClientError::CallTimeout)??; Ok(call == MAGIC_VALUE) } diff --git a/crates/p2p/Cargo.toml b/crates/p2p/Cargo.toml index 07554fb2..e9dab79b 100644 --- a/crates/p2p/Cargo.toml +++ b/crates/p2p/Cargo.toml @@ -28,7 +28,7 @@ serde_json.workspace = true pluto-core.workspace = true backon.workspace = true pin-project.workspace = true -reqwest.workspace = true +reqwest = { workspace = true, features = ["stream"] } unsigned-varint.workspace = true url.workspace = true diff --git a/crates/p2p/src/bootnode.rs b/crates/p2p/src/bootnode.rs index b35e8125..2cc51b0e 100644 --- a/crates/p2p/src/bootnode.rs +++ b/crates/p2p/src/bootnode.rs @@ -26,6 +26,14 @@ const BOOTNODE_RESOLVE_TIMEOUT: Duration = Duration::from_secs(60); /// Interval for checking bootnode resolution status. const BOOTNODE_CHECK_INTERVAL: Duration = Duration::from_secs(1); +/// Per-request timeout for relay address queries. +const RELAY_QUERY_TIMEOUT: Duration = Duration::from_secs(10); + +/// Maximum relay-address response body read from a relay (1 MB). The response +/// is an ENR string or a small JSON list of multiaddrs; 1 MB is generous while +/// bounding memory against a hostile relay. +const RELAY_MAX_BODY: usize = 1024 * 1024; + /// Bootnode error. #[derive(Debug, thiserror::Error)] pub enum BootnodeError { @@ -80,6 +88,10 @@ pub enum BootnodeError { /// ENR does not have TCP nor UDP port. #[error("enr does not have TCP nor UDP port")] EnrNoPort, + + /// Relay address response body exceeded the allowed size. + #[error("relay address body exceeds {0} bytes")] + BodyTooLarge(usize), } /// Result type for bootnode operations. @@ -163,7 +175,10 @@ async fn resolve_relay( mutable: MutablePeer, ) { let mut prev_addrs = String::new(); - let client = reqwest::Client::new(); + let client = reqwest::Client::builder() + .timeout(RELAY_QUERY_TIMEOUT) + .build() + .unwrap_or_default(); loop { if cancel.is_cancelled() { @@ -265,10 +280,7 @@ async fn query_relay_addrs( return Err(BootnodeError::InvalidRelayUrl); } - let body = resp.text().await.map_err(|e| { - tracing::warn!(err = %e, "Failure reading relay addresses (will try again)"); - BootnodeError::NewRequest(e) - })?; + let body = read_relay_body_capped(resp, RELAY_MAX_BODY).await?; if body.starts_with("enr:") { match multi_addr_from_enr_str(&body) { @@ -307,6 +319,36 @@ async fn query_relay_addrs( fetch.retry(backoff).when(retry_condition).await } +/// Reads a relay response body as UTF-8, failing with +/// [`BootnodeError::BodyTooLarge`] if it would exceed `max` bytes. Streams so +/// the cap bounds memory even without a trustworthy `Content-Length` header. +async fn read_relay_body_capped(resp: reqwest::Response, max: usize) -> Result { + use futures::StreamExt; + + if let Some(len) = resp.content_length() + && len > max as u64 + { + tracing::warn!(len, max, "Relay address body too large (will try again)"); + return Err(BootnodeError::BodyTooLarge(max)); + } + + let mut buf = Vec::new(); + let mut stream = resp.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| { + tracing::warn!(err = %e, "Failure reading relay addresses (will try again)"); + BootnodeError::NewRequest(e) + })?; + if buf.len().saturating_add(chunk.len()) > max { + tracing::warn!(max, "Relay address body too large (will try again)"); + return Err(BootnodeError::BodyTooLarge(max)); + } + buf.extend_from_slice(&chunk); + } + + Ok(String::from_utf8_lossy(&buf).into_owned()) +} + /// Returns multiaddrs from an ENR string. /// /// Creates QUIC-v1 multiaddr if UDP port is present, and TCP multiaddr if TCP From dd4da88534519dada5f793771e9201e481ba6dd2 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:54:36 +0200 Subject: [PATCH 10/12] fix: bound peer-supplied metric label length peerinfo's nickname and version Prometheus labels are set verbatim from peer-controlled values with no length bound (the SemVer pre-release group is unbounded, so a v1.2.3- version parses and is used as a label); monitoringapi's beacon_node_version label comes from the raw BN version string. Truncate all three to 64 bytes on a char boundary at the metric boundary. The peerinfo prev-nickname reset uses the same truncated form; monitoringapi still passes the full version to the compatibility check. git_hash and wordlist-based labels are already bounded. Hardening beyond Charon; truncation only affects pathological inputs. Part 3 of T08. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/monitoringapi/checker.rs | 37 +++++++++++++++- crates/peerinfo/src/protocol.rs | 59 ++++++++++++++++++++++--- 2 files changed, 89 insertions(+), 7 deletions(-) diff --git a/crates/app/src/monitoringapi/checker.rs b/crates/app/src/monitoringapi/checker.rs index 2f90994a..eaac1163 100644 --- a/crates/app/src/monitoringapi/checker.rs +++ b/crates/app/src/monitoringapi/checker.rs @@ -105,19 +105,42 @@ async fn set_beacon_node_version(beacon_node: &EthBeaconNodeApiClient) { } }; + // The BN version string is upstream-controlled and unbounded in length; + // truncate it for the metric label (a hostile/buggy BN could return a + // multi-MB string). The reset loop compares against the same truncated form + // so the stale series is cleared correctly. + let label = truncate_label(&version); + // Emulate Charon's `beaconNodeVersionGauge.Reset`: vise's `Family` cannot // delete series, so clear any previously-reported version before setting the // current one. for (previous, gauge) in MONITORING_METRICS.beacon_node_version.to_entries() { - if previous != version { + if previous != label { gauge.set(0); } } - MONITORING_METRICS.beacon_node_version[&version].set(1); + MONITORING_METRICS.beacon_node_version[&label].set(1); + // The semantic compatibility check uses the FULL (untruncated) version. check_beacon_node_version(&version); } +/// Maximum length (in bytes) for the upstream-supplied beacon-node version +/// metric label. A real BN version (e.g. "Lighthouse/v5.1.3-...") is short. +const MAX_METRIC_LABEL_LEN: usize = 64; + +/// Truncates a label value to [`MAX_METRIC_LABEL_LEN`] on a UTF-8 char boundary. +fn truncate_label(s: &str) -> String { + if s.len() <= MAX_METRIC_LABEL_LEN { + return s.to_string(); + } + let mut end = MAX_METRIC_LABEL_LEN; + while end > 0 && !s.is_char_boundary(end) { + end = end.saturating_sub(1); + } + s[..end].to_string() +} + async fn fetch_node_version( beacon_node: &EthBeaconNodeApiClient, ) -> Result { @@ -449,6 +472,16 @@ mod tests { Keypair::generate_secp256k1().public().to_peer_id() } + #[test] + fn truncate_label_bounds_length() { + assert_eq!(truncate_label("Lighthouse/v5.1.3"), "Lighthouse/v5.1.3"); + let long = "x".repeat(MAX_METRIC_LABEL_LEN + 500); + assert_eq!(truncate_label(&long).len(), MAX_METRIC_LABEL_LEN); + // Multibyte input truncates on a char boundary without panicking. + let mb = "é".repeat(MAX_METRIC_LABEL_LEN); + assert!(truncate_label(&mb).len() <= MAX_METRIC_LABEL_LEN); + } + fn pubkey(byte: u8) -> PubKey { PubKey::from([byte; 48]) } diff --git a/crates/peerinfo/src/protocol.rs b/crates/peerinfo/src/protocol.rs index 0dab9915..fe485f09 100644 --- a/crates/peerinfo/src/protocol.rs +++ b/crates/peerinfo/src/protocol.rs @@ -34,6 +34,25 @@ static GIT_HASH_RE: LazyLock = const PEERINFO_MAX_MESSAGE_SIZE: usize = 64 * 1024; +/// Maximum length (in bytes) for a peer-supplied metric label value. Real +/// Charon nicknames/versions are short; this bounds pathological input (e.g. a +/// nickname or a `v1.2.3-` pre-release version that parses fine) +/// without rejecting any legitimate peer. +const MAX_METRIC_LABEL_LEN: usize = 64; + +/// Truncates a peer-supplied label value to [`MAX_METRIC_LABEL_LEN`] on a UTF-8 +/// char boundary so the resulting label stays valid. +fn truncate_label(s: &str) -> String { + if s.len() <= MAX_METRIC_LABEL_LEN { + return s.to_string(); + } + let mut end = MAX_METRIC_LABEL_LEN; + while end > 0 && !s.is_char_boundary(end) { + end = end.saturating_sub(1); + } + s[..end].to_string() +} + /// State of the protocol. pub struct ProtocolState { /// The peer ID. @@ -206,11 +225,17 @@ impl ProtocolState { ) { let peer_name = pluto_p2p::name::peer_name(&self.peer_id); + // nickname/version are peer-supplied and unbounded in length; truncate at + // the metric boundary so a pathological value cannot blow up label memory. + // Reset the previous series under the SAME truncated form it was stored. + let nickname = truncate_label(nickname); + // Reset previous peer nickname if it changed if let Some(prev) = prev_nickname { - PEERINFO_METRICS.nickname[&PeerNicknameLabels::new(&peer_name, prev)].set(0); + let prev = truncate_label(prev); + PEERINFO_METRICS.nickname[&PeerNicknameLabels::new(&peer_name, &prev)].set(0); } - PEERINFO_METRICS.nickname[&PeerNicknameLabels::new(&peer_name, nickname)].set(1); + PEERINFO_METRICS.nickname[&PeerNicknameLabels::new(&peer_name, &nickname)].set(1); // Clamp clock offset to [-1 hour, 1 hour] let one_hour = chrono::Duration::hours(1); @@ -231,11 +256,11 @@ impl ProtocolState { // Handle version - use "unknown" if empty let version = if version.is_empty() { - "unknown" + "unknown".to_string() } else { - version + truncate_label(version) }; - PEERINFO_METRICS.version[&PeerVersionLabels::new(&peer_name, version)].set(1); + PEERINFO_METRICS.version[&PeerVersionLabels::new(&peer_name, &version)].set(1); // Handle git hash - use "unknown" if empty let git_hash = if git_hash.is_empty() { @@ -457,6 +482,30 @@ mod tests { assert_eq!(decoded, expected); } + #[test] + fn truncate_label_short_unchanged() { + let s = "short-nickname"; + assert_eq!(truncate_label(s), s); + } + + #[test] + fn truncate_label_long_ascii() { + let s = "a".repeat(MAX_METRIC_LABEL_LEN + 100); + let out = truncate_label(&s); + assert_eq!(out.len(), MAX_METRIC_LABEL_LEN); + } + + #[test] + fn truncate_label_multibyte_char_boundary() { + // Each 'é' is 2 bytes; a long run must truncate on a char boundary and + // stay valid UTF-8 (no panic). + let s = "é".repeat(MAX_METRIC_LABEL_LEN); + let out = truncate_label(&s); + assert!(out.len() <= MAX_METRIC_LABEL_LEN); + // Valid UTF-8: length is even (whole 'é' chars), never splitting one. + assert_eq!(out.len() % 2, 0); + } + #[test] fn encode_minimal() { let msg = make_minimal_peerinfo(); From 95c3e6457cd5af5f58cf19ea396785c1f755c680 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Wed, 1 Jul 2026 13:33:28 +0200 Subject: [PATCH 11/12] style: apply rustfmt to DoS-hardening changes Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/monitoringapi/checker.rs | 3 ++- crates/cli/src/commands/test/mev.rs | 4 ++-- crates/cluster/src/definition.rs | 3 ++- crates/core/src/version.rs | 13 +++---------- crates/p2p/src/p2p.rs | 16 ++++++++++++---- crates/p2p/src/p2p_context.rs | 24 +++++++++++++++++------- crates/relay-server/src/config.rs | 7 ++----- crates/ssz/src/hasher.rs | 3 ++- 8 files changed, 42 insertions(+), 31 deletions(-) diff --git a/crates/app/src/monitoringapi/checker.rs b/crates/app/src/monitoringapi/checker.rs index eaac1163..58b65ec5 100644 --- a/crates/app/src/monitoringapi/checker.rs +++ b/crates/app/src/monitoringapi/checker.rs @@ -129,7 +129,8 @@ async fn set_beacon_node_version(beacon_node: &EthBeaconNodeApiClient) { /// metric label. A real BN version (e.g. "Lighthouse/v5.1.3-...") is short. const MAX_METRIC_LABEL_LEN: usize = 64; -/// Truncates a label value to [`MAX_METRIC_LABEL_LEN`] on a UTF-8 char boundary. +/// Truncates a label value to [`MAX_METRIC_LABEL_LEN`] on a UTF-8 char +/// boundary. fn truncate_label(s: &str) -> String { if s.len() <= MAX_METRIC_LABEL_LEN { return s.to_string(); diff --git a/crates/cli/src/commands/test/mev.rs b/crates/cli/src/commands/test/mev.rs index 4a61d38a..329d3b53 100644 --- a/crates/cli/src/commands/test/mev.rs +++ b/crates/cli/src/commands/test/mev.rs @@ -35,8 +35,8 @@ fn http_client() -> reqwest::Client { .unwrap_or_default() } -/// Reads a response body, rejecting bodies that exceed [`BN_MAX_BODY`]. Uses the -/// advertised `Content-Length` for the fast-path reject; the client timeout +/// Reads a response body, rejecting bodies that exceed [`BN_MAX_BODY`]. Uses +/// the advertised `Content-Length` for the fast-path reject; the client timeout /// bounds a slow/absent-length body. async fn read_body_capped(resp: reqwest::Response) -> Result> { if let Some(len) = resp.content_length() diff --git a/crates/cluster/src/definition.rs b/crates/cluster/src/definition.rs index cba253a2..4f105c53 100644 --- a/crates/cluster/src/definition.rs +++ b/crates/cluster/src/definition.rs @@ -256,7 +256,8 @@ pub enum DefinitionError { #[error("Failed to convert length")] FailedToConvertLength, - /// num_validators exceeds the maximum allowed validators (SSZ_MAX_VALIDATORS). + /// num_validators exceeds the maximum allowed validators + /// (SSZ_MAX_VALIDATORS). #[error("num_validators {num_validators} exceeds maximum {max}")] NumValidatorsTooLarge { /// The offending value from the definition. diff --git a/crates/core/src/version.rs b/crates/core/src/version.rs index 475614d5..3ed3817e 100644 --- a/crates/core/src/version.rs +++ b/crates/core/src/version.rs @@ -141,22 +141,15 @@ impl SemVer { // rather than panicking. NB: this is an intentional, safe divergence // from Charon's Parse (app/version/version.go @ v1.7.1), which discards // the strconv.Atoi error and silently yields 0 on overflow. - let major = matches[1] - .parse() - .map_err(|_| SemVerError::InvalidFormat)?; - let minor = matches[2] - .parse() - .map_err(|_| SemVerError::InvalidFormat)?; + let major = matches[1].parse().map_err(|_| SemVerError::InvalidFormat)?; + let minor = matches[2].parse().map_err(|_| SemVerError::InvalidFormat)?; let mut patch = 0; let mut pre_release = ""; let mut sem_ver_type = SemVerType::Minor; if let Some(m) = matches.get(3) { - patch = m - .as_str() - .parse() - .map_err(|_| SemVerError::InvalidFormat)?; + patch = m.as_str().parse().map_err(|_| SemVerError::InvalidFormat)?; sem_ver_type = SemVerType::Patch; } diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index c40ff1fb..4cf88307 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -700,8 +700,8 @@ impl FusedStream for Node { /// Stores identify-reported listen addresses for a peer, gated to known cluster /// peers only. Addresses from unknown peers are dropped (and not cloned), since /// the only consumers of `peer_addresses` look up known peers exclusively — so -/// storing them would be pure unbounded growth. The per-peer count is bounded by -/// [`PeerStore::set_peer_addresses`]. +/// storing them would be pure unbounded growth. The per-peer count is bounded +/// by [`PeerStore::set_peer_addresses`]. fn store_identify_addrs(ctx: &P2PContext, peer_id: &PeerId, addrs: &[Multiaddr]) { if ctx.is_known_peer(peer_id) { // The peer addresses will be available in the next poll of the node. @@ -722,7 +722,11 @@ mod tests { fn addrs(n: usize) -> Vec { (0..n) - .map(|i| format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)).parse().unwrap()) + .map(|i| { + format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)) + .parse() + .unwrap() + }) .collect() } @@ -753,7 +757,11 @@ mod tests { let known = random_peer_id(); let ctx = P2PContext::new([known]); - store_identify_addrs(&ctx, &known, &addrs(crate::p2p_context::MAX_PEER_ADDRESSES + 1)); + store_identify_addrs( + &ctx, + &known, + &addrs(crate::p2p_context::MAX_PEER_ADDRESSES + 1), + ); let store = ctx.peer_store_lock(); assert_eq!( diff --git a/crates/p2p/src/p2p_context.rs b/crates/p2p/src/p2p_context.rs index 23846d48..66a16477 100644 --- a/crates/p2p/src/p2p_context.rs +++ b/crates/p2p/src/p2p_context.rs @@ -128,8 +128,8 @@ impl PeerStore { /// Marks a peer connection as inactive. /// - /// `known_peers` is the set of cluster peer IDs; entries whose peer ID is in - /// this set are never evicted by the `MAX_INACTIVE_PEERS` cap. + /// `known_peers` is the set of cluster peer IDs; entries whose peer ID is + /// in this set are never evicted by the `MAX_INACTIVE_PEERS` cap. pub fn remove_peer(&mut self, peer: Peer, known_peers: &HashSet) { self.active_peers.remove(&peer); @@ -142,8 +142,9 @@ impl PeerStore { } /// Evicts oldest inactive, non-known peers until `inactive_peers` is within - /// `MAX_INACTIVE_PEERS`. Known cluster peers are re-queued (kept) so they are - /// never dropped. Bounded: each call scans at most `inactive_order.len()` entries. + /// `MAX_INACTIVE_PEERS`. Known cluster peers are re-queued (kept) so they + /// are never dropped. Bounded: each call scans at most + /// `inactive_order.len()` entries. fn evict_inactive(&mut self, known_peers: &HashSet) { // Bound the number of scan iterations to the current queue length so a // queue made entirely of known peers cannot loop forever. @@ -216,7 +217,8 @@ impl PeerStore { self.peer_addresses.get(peer_id) } - /// Removes the stored addresses for a peer. Returns the removed addresses, if any. + /// Removes the stored addresses for a peer. Returns the removed addresses, + /// if any. pub fn remove_peer_addresses(&mut self, peer_id: &PeerId) -> Option> { self.peer_addresses.remove(peer_id) } @@ -307,7 +309,11 @@ mod tests { let mut store = PeerStore::default(); let peer = PeerId::random(); let addrs: Vec = (0..(MAX_PEER_ADDRESSES + 50)) - .map(|i| format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)).parse().unwrap()) + .map(|i| { + format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)) + .parse() + .unwrap() + }) .collect(); store.set_peer_addresses(peer, addrs.clone()); @@ -323,7 +329,11 @@ mod tests { let mut store = PeerStore::default(); let peer = PeerId::random(); let addrs: Vec = (0..3) - .map(|i| format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)).parse().unwrap()) + .map(|i| { + format!("/ip4/127.0.0.1/tcp/{}", 9000usize.saturating_add(i)) + .parse() + .unwrap() + }) .collect(); store.set_peer_addresses(peer, addrs.clone()); diff --git a/crates/relay-server/src/config.rs b/crates/relay-server/src/config.rs index 440a84c2..d77596a9 100644 --- a/crates/relay-server/src/config.rs +++ b/crates/relay-server/src/config.rs @@ -85,12 +85,9 @@ pub(crate) fn create_relay_config(config: &Config) -> relay::Config { let per_ip_limit = u32::try_from(config.max_res_per_peer) .ok() .and_then(NonZeroU32::new) - .unwrap_or_else(|| { - NonZeroU32::new(RESERVATIONS_PER_IP_PER_MINUTE).expect("60 > 0") - }); + .unwrap_or_else(|| NonZeroU32::new(RESERVATIONS_PER_IP_PER_MINUTE).expect("60 > 0")); - relay_config - .reservation_rate_per_ip(per_ip_limit, Duration::from_secs(ONE_MINUTE_SECONDS)) + relay_config.reservation_rate_per_ip(per_ip_limit, Duration::from_secs(ONE_MINUTE_SECONDS)) } #[cfg(test)] diff --git a/crates/ssz/src/hasher.rs b/crates/ssz/src/hasher.rs index 822d1213..27cac11a 100644 --- a/crates/ssz/src/hasher.rs +++ b/crates/ssz/src/hasher.rs @@ -103,7 +103,8 @@ pub enum HasherError { /// Declared limit. limit: usize, }, - /// Bitlist final (delimiter) byte is zero — missing the length sentinel bit. + /// Bitlist final (delimiter) byte is zero — missing the length sentinel + /// bit. #[error("Invalid bitlist: final byte is zero (missing length delimiter bit)")] InvalidBitlistDelimiter, } From ee657fab8a0d41ceb4def8f4bb24fb8bb3ddfaf0 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:46:10 +0200 Subject: [PATCH 12/12] chore(deps): bump quick-xml to 0.41 to fix RUSTSEC advisory quick-xml 0.39.4 is flagged by RUSTSEC for unbounded namespace allocation (fixed in >=0.41.0). Bump the workspace dependency and dedupe the lockfile to a single quick-xml 0.41.0. Co-authored-by: varex83 --- Cargo.lock | 16 +++------------- Cargo.toml | 2 +- 2 files changed, 4 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1c726d5a..df102b8d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2236,7 +2236,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 2.0.118", + "syn 1.0.109", ] [[package]] @@ -5155,7 +5155,7 @@ dependencies = [ "oas3", "prettyplease", "proc-macro2", - "quick-xml 0.41.0", + "quick-xml", "quote", "regex", "reqwest 0.13.4", @@ -5579,7 +5579,7 @@ dependencies = [ "pluto-ssz", "pluto-testutil", "pluto-tracing", - "quick-xml 0.39.4", + "quick-xml", "rand 0.8.6", "reqwest 0.13.4", "serde", @@ -6414,16 +6414,6 @@ dependencies = [ "unsigned-varint 0.8.0", ] -[[package]] -name = "quick-xml" -version = "0.39.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e" -dependencies = [ - "memchr", - "serde", -] - [[package]] name = "quick-xml" version = "0.41.0" diff --git a/Cargo.toml b/Cargo.toml index 4264df4f..86f0f46b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -116,7 +116,7 @@ flate2 = "1.1" wiremock = "0.6" tower = "0.5" sysinfo = "0.33" -quick-xml = { version = "0.39", features = ["serialize"] } +quick-xml = { version = "0.41", features = ["serialize"] } # Crates in the workspace pluto-app = { path = "crates/app" }