diff --git a/crates/lanspread-peer/src/services/server.rs b/crates/lanspread-peer/src/services/server.rs index a5f2c4a..82a7094 100644 --- a/crates/lanspread-peer/src/services/server.rs +++ b/crates/lanspread-peer/src/services/server.rs @@ -1,7 +1,7 @@ //! QUIC server accept loop. use std::{ - collections::HashMap, + collections::{HashMap, VecDeque}, future::Future, net::{IpAddr, SocketAddr}, panic::AssertUnwindSafe, @@ -24,6 +24,7 @@ use tokio::{ use tokio_util::sync::CancellationToken; use crate::{ + config::QUIC_HANDSHAKE_TIMEOUT, context::PeerCtx, library::prime_library_manifests, quic_runtime::{quic_congestion_controller, quic_server_limits, tracked_quic_io}, @@ -37,6 +38,11 @@ use crate::{ /// Limits unauthenticated handshake memory before QUIC admission completes. const MAX_INFLIGHT_HANDSHAKES: usize = 64; +/// One unverified datagram source may start only this many handshakes during a +/// full handshake-timeout window, before application connection leases exist. +const MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN: usize = 8; +/// Bounds the unauthenticated source-rate ledger itself under spoofed-IP churn. +const MAX_TRACKED_HANDSHAKE_ORIGINS: usize = 256; /// Limits established connection scopes owned by the application accept loop. const MAX_ESTABLISHED_CONNECTIONS: usize = 64; /// Mirrors the transport stream limit and bounds application stream futures. @@ -305,6 +311,38 @@ impl ServerAdmission { struct BoundedEndpointLimits { inner: endpoint_limits::Default, + handshakes: HandshakeOriginLimiter, +} + +#[derive(Default)] +struct HandshakeOriginLimiter { + attempts: HashMap>, +} + +impl HandshakeOriginLimiter { + fn admit(&mut self, origin: IpAddr, now: Duration) -> bool { + for attempts in self.attempts.values_mut() { + while attempts + .front() + .is_some_and(|started| now.saturating_sub(*started) >= QUIC_HANDSHAKE_TIMEOUT) + { + attempts.pop_front(); + } + } + self.attempts.retain(|_, attempts| !attempts.is_empty()); + + if !self.attempts.contains_key(&origin) + && self.attempts.len() >= MAX_TRACKED_HANDSHAKE_ORIGINS + { + return false; + } + let attempts = self.attempts.entry(origin).or_default(); + if attempts.len() >= MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN { + return false; + } + attempts.push_back(now); + true + } } impl endpoint_limits::Limiter for BoundedEndpointLimits { @@ -315,7 +353,22 @@ impl endpoint_limits::Limiter for BoundedEndpointLimits { if !endpoint_connection_capacity_available(info.connection_count) { return endpoint_limits::Outcome::close(); } - endpoint_limits::Limiter::on_connection_attempt(&mut self.inner, info) + let default_outcome = + endpoint_limits::Limiter::on_connection_attempt(&mut self.inner, info); + if default_outcome != endpoint_limits::Outcome::allow() { + return default_outcome; + } + let remote_addr = SocketAddr::from(&info.remote_address); + let origin = canonical_origin_ip(remote_addr.ip()); + if !self + .handshakes + .admit(origin, info.timestamp.duration_since_start()) + { + // A valid Retry token bypasses s2n's connection-attempt callback on + // the next Initial, so origin exhaustion must not mint one. + return endpoint_limits::Outcome::drop(); + } + default_outcome } } @@ -328,6 +381,7 @@ fn bounded_endpoint_limits() -> eyre::Result { inner: endpoint_limits::Default::builder() .with_inflight_handshake_limit(MAX_INFLIGHT_HANDSHAKES)? .build()?, + handshakes: HandshakeOriginLimiter::default(), }) } @@ -670,7 +724,7 @@ async fn drain_joined_child_tasks(children: &mut JoinSet>, labe #[cfg(test)] mod tests { use std::{ - net::SocketAddr, + net::{IpAddr, Ipv6Addr, SocketAddr}, panic::{AssertUnwindSafe, catch_unwind}, sync::{ Arc, @@ -687,6 +741,7 @@ mod tests { use super::{ CONNECTION_NO_STREAM_IDLE_TIMEOUT, + HandshakeOriginLimiter, MAX_BULK_TRANSFER_TASKS_PER_ORIGIN, MAX_CONTROL_STREAM_TASKS, MAX_CONTROL_STREAM_TASKS_PER_ORIGIN, @@ -695,8 +750,10 @@ mod tests { MAX_GLOBAL_BULK_TRANSFER_TASKS, MAX_GLOBAL_CONTROL_STREAM_TASKS, MAX_GLOBAL_STREAM_INSTALL_TASKS, + MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN, MAX_INFLIGHT_HANDSHAKES, MAX_STREAM_INSTALL_TASKS_PER_ORIGIN, + MAX_TRACKED_HANDSHAKE_ORIGINS, ObservedOrigin, OriginAdmission, OriginAdmissionClass, @@ -711,6 +768,7 @@ mod tests { #[test] fn unauthenticated_runtime_bounds_are_explicit_and_closed_at_capacity() { assert_eq!(MAX_INFLIGHT_HANDSHAKES, 64); + assert_eq!(MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN, 8); assert_eq!(MAX_ESTABLISHED_CONNECTIONS, 64); assert_eq!(MAX_CONTROL_STREAM_TASKS, 32); assert_eq!(MAX_GLOBAL_CONTROL_STREAM_TASKS, 16); @@ -729,6 +787,38 @@ mod tests { assert!(!has_child_capacity(32, MAX_CONTROL_STREAM_TASKS)); } + #[test] + fn handshake_attempt_window_preserves_capacity_for_other_origins() { + let mut limiter = HandshakeOriginLimiter::default(); + let now = Duration::from_secs(1); + let flood = IpAddr::from([192, 0, 2, 10]); + let healthy = IpAddr::from([192, 0, 2, 11]); + + for _ in 0..MAX_INFLIGHT_HANDSHAKE_ATTEMPTS_PER_ORIGIN { + assert!(limiter.admit(flood, now)); + } + assert!(!limiter.admit(flood, now)); + assert!(limiter.admit(healthy, now)); + assert!(limiter.admit( + flood, + now + crate::config::QUIC_HANDSHAKE_TIMEOUT + )); + } + + #[test] + fn handshake_origin_ledger_rejects_spoofed_source_churn_at_capacity() { + let mut limiter = HandshakeOriginLimiter::default(); + let now = Duration::from_secs(1); + for index in 0..MAX_TRACKED_HANDSHAKE_ORIGINS { + let origin = IpAddr::V6(Ipv6Addr::from( + u128::try_from(index + 1).expect("test origin index should fit u128"), + )); + assert!(limiter.admit(origin, now)); + } + assert!(!limiter.admit(IpAddr::V6(Ipv6Addr::from(1_000_u128)), now)); + assert_eq!(limiter.attempts.len(), MAX_TRACKED_HANDSHAKE_ORIGINS); + } + #[test] fn endpoint_limiter_rejects_before_the_internal_accept_queue_can_exceed_the_cap() { bounded_endpoint_limits().expect("endpoint limiter should build");