From 42cf98ecec1d1cce4d33869721ee53290260c88a Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 12 Sep 2026 13:09:06 +0200 Subject: [PATCH] fix(peer): charge discovery quota to packet origin Carry the mDNS response source independently of its advertised A or AAAA target. Active and cooling candidates now charge the observed host, so rotating peer IDs, ports, and target addresses cannot escape the eight-candidate origin budget. Test Plan: - just test - just clippy - rotating-advertised-target regression - git diff --check --- .../lanspread-peer/src/services/discovery.rs | 133 +++++++++++------- 1 file changed, 79 insertions(+), 54 deletions(-) diff --git a/crates/lanspread-peer/src/services/discovery.rs b/crates/lanspread-peer/src/services/discovery.rs index 76bb6da..b21b580 100644 --- a/crates/lanspread-peer/src/services/discovery.rs +++ b/crates/lanspread-peer/src/services/discovery.rs @@ -1,7 +1,7 @@ //! mDNS peer discovery and discovery-time protocol negotiation. use std::{ - collections::{HashSet, VecDeque}, + collections::{HashMap, VecDeque}, future::Future, thread::JoinHandle, time::Duration, @@ -11,10 +11,7 @@ use eyre::WrapErr as _; use futures::{StreamExt as _, stream::FuturesUnordered}; use lanspread_mdns::{LANSPREAD_SERVICE_TYPE, MdnsBrowser, MdnsService, MdnsServicePoll}; use lanspread_proto::{PROTOCOL_VERSION, PeerEndpoint, PeerId}; -use tokio::sync::{ - mpsc, - oneshot, -}; +use tokio::sync::{mpsc, oneshot}; use tokio_util::sync::CancellationToken; use crate::{ @@ -29,30 +26,35 @@ use crate::{ }; const MAX_ACTIVE_DISCOVERY_CANDIDATES: usize = 64; -/// Upper bound on active-or-cooling candidates advertised from one source IP. -/// mDNS lets a single LAN host claim arbitrarily many peer IDs and ports, so -/// without this bound one host could occupy every discovery slot and starve -/// genuinely new peers. Legitimate hosts run one peer, occasionally a few. +/// Upper bound on active-or-cooling candidates observed from one response IP. +/// mDNS lets a single LAN host claim arbitrarily many target addresses, peer +/// IDs, and ports, so accounting must use packet provenance rather than the +/// advertised A/AAAA record. Legitimate hosts run one peer, occasionally a few. const MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP: usize = 8; const MAX_PENDING_MDNS_SERVICES: usize = 64; const DISCOVERY_CANDIDATE_COOLDOWN: Duration = Duration::from_secs(5); #[derive(Default)] struct RecentCandidates { - entries: VecDeque<(PeerEndpoint, tokio::time::Instant)>, + entries: VecDeque<(PeerEndpoint, std::net::IpAddr, tokio::time::Instant)>, } impl RecentCandidates { - fn try_record(&mut self, candidate: PeerEndpoint, now: tokio::time::Instant) -> bool { + fn try_record( + &mut self, + candidate: PeerEndpoint, + source_ip: std::net::IpAddr, + now: tokio::time::Instant, + ) -> bool { self.expire(now); - if self.entries.iter().any(|(endpoint, _)| { + if self.entries.iter().any(|(endpoint, _, _)| { endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr }) || self.entries.len() >= MAX_ACTIVE_DISCOVERY_CANDIDATES { return false; } self.entries - .push_back((candidate, now + DISCOVERY_CANDIDATE_COOLDOWN)); + .push_back((candidate, source_ip, now + DISCOVERY_CANDIDATE_COOLDOWN)); true } @@ -60,7 +62,7 @@ impl RecentCandidates { while self .entries .front() - .is_some_and(|(_, deadline)| *deadline <= now) + .is_some_and(|(_, _, deadline)| *deadline <= now) { self.entries.pop_front(); } @@ -71,13 +73,13 @@ impl RecentCandidates { fn count_from_ip_outside( &mut self, ip: std::net::IpAddr, - active: &HashSet, + active: &HashMap, now: tokio::time::Instant, ) -> usize { self.expire(now); self.entries .iter() - .filter(|(endpoint, _)| endpoint.addr.ip() == ip && !active.contains(endpoint)) + .filter(|(endpoint, source_ip, _)| *source_ip == ip && !active.contains_key(endpoint)) .count() } @@ -89,6 +91,7 @@ impl RecentCandidates { struct MdnsPeerInfo { addr: std::net::SocketAddr, + source_ip: std::net::IpAddr, peer_id: Option, proto_ver: Option, } @@ -203,7 +206,7 @@ pub async fn run_peer_discovery( let service_shutdown = ctx.shutdown.child_token(); let mut worker = DiscoveryWorker::spawn(service_type, service_tx, service_shutdown.clone())?; let mut negotiations = FuturesUnordered::new(); - let mut active_candidates = HashSet::new(); + let mut active_candidates = HashMap::new(); let mut recent_candidates = RecentCandidates::default(); let mut mismatch_emitted = false; let mut state_sync = Box::pin(run_state_sync( @@ -256,6 +259,7 @@ pub async fn run_peer_discovery( &active_candidates, &mut recent_candidates, endpoint, + info.source_ip, tokio::time::Instant::now(), ) { log::warn!( @@ -273,7 +277,7 @@ pub async fn run_peer_discovery( continue; } }; - active_candidates.insert(endpoint); + active_candidates.insert(endpoint, info.source_ip); negotiations.push(run_protocol_negotiation(ProtocolNegotiation { endpoint, handshake, @@ -312,39 +316,42 @@ pub async fn run_peer_discovery( } } -fn candidate_conflicts(active: &HashSet, candidate: PeerEndpoint) -> bool { +fn candidate_conflicts( + active: &HashMap, + candidate: PeerEndpoint, +) -> bool { active - .iter() + .keys() .any(|endpoint| endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr) } /// Returns whether admitting `candidate` would exceed the per-source-IP /// discovery budget across active negotiations and cooling-down attempts. fn source_ip_is_saturated( - active: &HashSet, + active: &HashMap, recent: &mut RecentCandidates, - candidate: PeerEndpoint, + source_ip: std::net::IpAddr, now: tokio::time::Instant, ) -> bool { - let ip = candidate.addr.ip(); let active_from_ip = active - .iter() - .filter(|endpoint| endpoint.addr.ip() == ip) + .values() + .filter(|observed_ip| **observed_ip == source_ip) .count(); - active_from_ip + recent.count_from_ip_outside(ip, active, now) + active_from_ip + recent.count_from_ip_outside(source_ip, active, now) >= MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP } fn candidate_is_admissible( - active: &HashSet, + active: &HashMap, recent: &mut RecentCandidates, candidate: PeerEndpoint, + source_ip: std::net::IpAddr, now: tokio::time::Instant, ) -> bool { active.len() < MAX_ACTIVE_DISCOVERY_CANDIDATES && !candidate_conflicts(active, candidate) - && !source_ip_is_saturated(active, recent, candidate, now) - && recent.try_record(candidate, now) + && !source_ip_is_saturated(active, recent, source_ip, now) + && recent.try_record(candidate, source_ip, now) } fn run_mdns_browser( @@ -409,6 +416,7 @@ async fn wait_for_local_peer_addr(ctx: &NetworkServiceCtx) -> bool { fn parse_mdns_peer(service: &MdnsService) -> MdnsPeerInfo { MdnsPeerInfo { addr: service.addr, + source_ip: service.source_ip, peer_id: service .properties .get("peer_id") @@ -496,8 +504,8 @@ where #[cfg(test)] mod tests { use std::{ - collections::HashSet, - net::SocketAddr, + collections::HashMap, + net::{IpAddr, SocketAddr}, sync::{ Arc, atomic::{AtomicBool, AtomicUsize, Ordering}, @@ -562,6 +570,7 @@ mod tests { let peer_id = Some(PeerId::from_bytes([7; 32])); let unicast = MdnsPeerInfo { addr: "192.168.1.50:42424".parse().expect("test address parses"), + source_ip: "192.168.1.50".parse().expect("test source parses"), peer_id, proto_ver: Some(lanspread_proto::PROTOCOL_VERSION), }; @@ -569,6 +578,7 @@ mod tests { let multicast = MdnsPeerInfo { addr: "224.0.0.251:42424".parse().expect("test address parses"), + source_ip: "192.168.1.50".parse().expect("test source parses"), peer_id, proto_ver: Some(lanspread_proto::PROTOCOL_VERSION), }; @@ -576,6 +586,7 @@ mod tests { let zero_port = MdnsPeerInfo { addr: "192.168.1.50:0".parse().expect("test address parses"), + source_ip: "192.168.1.50".parse().expect("test source parses"), peer_id, proto_ver: Some(lanspread_proto::PROTOCOL_VERSION), }; @@ -585,7 +596,7 @@ mod tests { #[test] fn active_candidate_keys_bound_both_claimed_identity_and_address() { let active_endpoint = endpoint(1, 12001); - let mut active = HashSet::from([active_endpoint]); + let mut active = HashMap::from([(active_endpoint, active_endpoint.addr.ip())]); assert!(candidate_conflicts(&active, endpoint(1, 12002))); assert!(candidate_conflicts(&active, endpoint(2, 12001))); @@ -601,22 +612,26 @@ mod tests { let now = tokio::time::Instant::now(); let first = endpoint(1, 12001); let mut recent = RecentCandidates::default(); - assert!(recent.try_record(first, now)); + assert!(recent.try_record(first, first.addr.ip(), now)); for port in 12002..12102 { - assert!(!recent.try_record(endpoint(1, port), now)); + let candidate = endpoint(1, port); + assert!(!recent.try_record(candidate, candidate.addr.ip(), now)); } for seed in 2..=101 { - assert!(!recent.try_record(endpoint(seed, 12001), now)); + let candidate = endpoint(seed, 12001); + assert!(!recent.try_record(candidate, candidate.addr.ip(), now)); } for seed in 2..=u8::try_from(MAX_ACTIVE_DISCOVERY_CANDIDATES).expect("test bound fits u8") { - assert!(recent.try_record(endpoint(seed, 13000 + u16::from(seed)), now)); + let candidate = endpoint(seed, 13000 + u16::from(seed)); + assert!(recent.try_record(candidate, candidate.addr.ip(), now)); } assert_eq!(recent.len(), MAX_ACTIVE_DISCOVERY_CANDIDATES); - assert!(!recent.try_record(endpoint(200, 14000), now)); + let last = endpoint(200, 14000); + assert!(!recent.try_record(last, last.addr.ip(), now)); let after_cooldown = now + DISCOVERY_CANDIDATE_COOLDOWN; - assert!(recent.try_record(endpoint(200, 14000), after_cooldown)); + assert!(recent.try_record(last, last.addr.ip(), after_cooldown)); assert_eq!(recent.len(), 1); } @@ -625,46 +640,52 @@ mod tests { } #[test] - fn one_source_ip_cannot_occupy_more_than_its_candidate_budget() { + fn rotating_advertised_targets_share_the_observed_source_budget() { let now = tokio::time::Instant::now(); - let mut active = HashSet::new(); + let mut active = HashMap::new(); let mut recent = RecentCandidates::default(); - let flood_ip = [192, 168, 1, 66]; + let flood_source: IpAddr = [192, 168, 1, 66].into(); - // A single host rotating peer IDs and ports fills only its own budget. + // One responder can rotate advertised A records, peer IDs, and ports, + // but all candidates remain charged to the observed datagram source. for index in 0..MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP { let seed = u8::try_from(index + 1).expect("test index fits u8"); - let candidate = endpoint_at(seed, flood_ip, 20000 + u16::from(seed)); + let candidate = endpoint_at(seed, [10, 0, seed, 2], 20000 + u16::from(seed)); assert!(candidate_is_admissible( &active, &mut recent, candidate, + flood_source, now )); - active.insert(candidate); + active.insert(candidate, flood_source); } assert!(!candidate_is_admissible( &active, &mut recent, - endpoint_at(100, flood_ip, 30000), + endpoint_at(100, [10, 1, 1, 2], 30000), + flood_source, now )); // Other hosts are unaffected while the flood host is saturated. + let other_source: IpAddr = [192, 168, 1, 67].into(); assert!(candidate_is_admissible( &active, &mut recent, - endpoint_at(101, [192, 168, 1, 67], 42424), + endpoint_at(101, [10, 1, 1, 3], 42424), + other_source, now )); // Completed negotiations keep counting while they cool down ... - let completed = endpoint_at(1, flood_ip, 20001); + let completed = endpoint_at(1, [10, 0, 1, 2], 20001); active.remove(&completed); assert!(!candidate_is_admissible( &active, &mut recent, - endpoint_at(102, flood_ip, 30001), + endpoint_at(102, [10, 1, 1, 4], 30001), + flood_source, now )); // ... and free the budget once the cooldown expires. @@ -672,7 +693,8 @@ mod tests { assert!(candidate_is_admissible( &active, &mut recent, - endpoint_at(102, flood_ip, 30001), + endpoint_at(102, [10, 1, 1, 4], 30001), + flood_source, after_cooldown )); } @@ -685,12 +707,13 @@ mod tests { let mut active = (0..MAX_ACTIVE_DISCOVERY_CANDIDATES) .map(|index| { let seed = u8::try_from(index + 1).expect("test index fits u8"); - endpoint_at(seed, [10, 0, seed, 1], 15000 + u16::from(seed)) + let candidate = endpoint_at(seed, [10, 0, seed, 1], 15000 + u16::from(seed)); + (candidate, candidate.addr.ip()) }) - .collect::>(); + .collect::>(); let mut recent = RecentCandidates::default(); - for candidate in &active { - assert!(recent.try_record(*candidate, now)); + for (candidate, source_ip) in &active { + assert!(recent.try_record(*candidate, *source_ip, now)); } let after_cooldown = now + DISCOVERY_CANDIDATE_COOLDOWN; @@ -699,16 +722,18 @@ mod tests { &active, &mut recent, next, + next.addr.ip(), after_cooldown, )); assert_eq!(active.len(), MAX_ACTIVE_DISCOVERY_CANDIDATES); - let completed = *active.iter().next().expect("active set should be nonempty"); + let completed = *active.keys().next().expect("active set should be nonempty"); active.remove(&completed); assert!(candidate_is_admissible( &active, &mut recent, next, + next.addr.ip(), after_cooldown, )); }