ddidderr/lanspread
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(peer): charge discovery quota to packet origin

Browse Source 
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
This commit is contained in:
ddidderr
2026-09-12 13:09:06 +02:00
parent 4d5881d6b0
commit 42cf98ecec
1 changed files with 79 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer/src/services/discovery.rs
+79 -54
View File
@@ -1,7 +1,7 @@
//! mDNS peer discovery and discovery-time protocol negotiation. //! mDNS peer discovery and discovery-time protocol negotiation.
use std::{ use std::{
collections::{HashSet, VecDeque}, collections::{HashMap, VecDeque},
future::Future, future::Future,
thread::JoinHandle, thread::JoinHandle,
time::Duration, time::Duration,
@@ -11,10 +11,7 @@ use eyre::WrapErr as _;
use futures::{StreamExt as _, stream::FuturesUnordered}; use futures::{StreamExt as _, stream::FuturesUnordered};
use lanspread_mdns::{LANSPREAD_SERVICE_TYPE, MdnsBrowser, MdnsService, MdnsServicePoll}; use lanspread_mdns::{LANSPREAD_SERVICE_TYPE, MdnsBrowser, MdnsService, MdnsServicePoll};
use lanspread_proto::{PROTOCOL_VERSION, PeerEndpoint, PeerId}; use lanspread_proto::{PROTOCOL_VERSION, PeerEndpoint, PeerId};
use tokio::sync::{ use tokio::sync::{mpsc, oneshot};
mpsc,
oneshot,
};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use crate::{ use crate::{
@@ -29,30 +26,35 @@ use crate::{
}; };
const MAX_ACTIVE_DISCOVERY_CANDIDATES: usize = 64; const MAX_ACTIVE_DISCOVERY_CANDIDATES: usize = 64;
/// Upper bound on active-or-cooling candidates advertised from one source IP. /// Upper bound on active-or-cooling candidates observed from one response IP.
/// mDNS lets a single LAN host claim arbitrarily many peer IDs and ports, so /// mDNS lets a single LAN host claim arbitrarily many target addresses, peer
/// without this bound one host could occupy every discovery slot and starve /// IDs, and ports, so accounting must use packet provenance rather than the
/// genuinely new peers. Legitimate hosts run one peer, occasionally a few. /// advertised A/AAAA record. Legitimate hosts run one peer, occasionally a few.
const MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP: usize = 8; const MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP: usize = 8;
const MAX_PENDING_MDNS_SERVICES: usize = 64; const MAX_PENDING_MDNS_SERVICES: usize = 64;
const DISCOVERY_CANDIDATE_COOLDOWN: Duration = Duration::from_secs(5); const DISCOVERY_CANDIDATE_COOLDOWN: Duration = Duration::from_secs(5);
#[derive(Default)] #[derive(Default)]
struct RecentCandidates { struct RecentCandidates {
entries: VecDeque<(PeerEndpoint, tokio::time::Instant)>, entries: VecDeque<(PeerEndpoint, std::net::IpAddr, tokio::time::Instant)>,
} }
impl RecentCandidates { 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); 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 endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr
}) || self.entries.len() >= MAX_ACTIVE_DISCOVERY_CANDIDATES }) || self.entries.len() >= MAX_ACTIVE_DISCOVERY_CANDIDATES
{ {
return false; return false;
} }
self.entries self.entries
.push_back((candidate, now + DISCOVERY_CANDIDATE_COOLDOWN)); .push_back((candidate, source_ip, now + DISCOVERY_CANDIDATE_COOLDOWN));
true true
} }
@@ -60,7 +62,7 @@ impl RecentCandidates {
while self while self
.entries .entries
.front() .front()
.is_some_and(|(_, deadline)| *deadline <= now) .is_some_and(|(_, _, deadline)| *deadline <= now)
{ {
self.entries.pop_front(); self.entries.pop_front();
} }
@@ -71,13 +73,13 @@ impl RecentCandidates {
fn count_from_ip_outside( fn count_from_ip_outside(
&mut self, &mut self,
ip: std::net::IpAddr, ip: std::net::IpAddr,
active: &HashSet<PeerEndpoint>, active: &HashMap<PeerEndpoint, std::net::IpAddr>,
now: tokio::time::Instant, now: tokio::time::Instant,
) -> usize { ) -> usize {
self.expire(now); self.expire(now);
self.entries self.entries
.iter() .iter()
.filter(|(endpoint, _)| endpoint.addr.ip() == ip && !active.contains(endpoint)) .filter(|(endpoint, source_ip, _)| *source_ip == ip && !active.contains_key(endpoint))
.count() .count()
} }
@@ -89,6 +91,7 @@ impl RecentCandidates {
struct MdnsPeerInfo { struct MdnsPeerInfo {
addr: std::net::SocketAddr, addr: std::net::SocketAddr,
source_ip: std::net::IpAddr,
peer_id: Option<PeerId>, peer_id: Option<PeerId>,
proto_ver: Option<u32>, proto_ver: Option<u32>,
} }
@@ -203,7 +206,7 @@ pub async fn run_peer_discovery(
let service_shutdown = ctx.shutdown.child_token(); let service_shutdown = ctx.shutdown.child_token();
let mut worker = DiscoveryWorker::spawn(service_type, service_tx, service_shutdown.clone())?; let mut worker = DiscoveryWorker::spawn(service_type, service_tx, service_shutdown.clone())?;
let mut negotiations = FuturesUnordered::new(); 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 recent_candidates = RecentCandidates::default();
let mut mismatch_emitted = false; let mut mismatch_emitted = false;
let mut state_sync = Box::pin(run_state_sync( let mut state_sync = Box::pin(run_state_sync(
@@ -256,6 +259,7 @@ pub async fn run_peer_discovery(
&active_candidates, &active_candidates,
&mut recent_candidates, &mut recent_candidates,
endpoint, endpoint,
info.source_ip,
tokio::time::Instant::now(), tokio::time::Instant::now(),
) { ) {
log::warn!( log::warn!(
@@ -273,7 +277,7 @@ pub async fn run_peer_discovery(
continue; continue;
} }
}; };
active_candidates.insert(endpoint); active_candidates.insert(endpoint, info.source_ip);
negotiations.push(run_protocol_negotiation(ProtocolNegotiation { negotiations.push(run_protocol_negotiation(ProtocolNegotiation {
endpoint, endpoint,
handshake, handshake,
@@ -312,39 +316,42 @@ pub async fn run_peer_discovery(
} }
} }
fn candidate_conflicts(active: &HashSet<PeerEndpoint>, candidate: PeerEndpoint) -> bool { fn candidate_conflicts(
active: &HashMap<PeerEndpoint, std::net::IpAddr>,
candidate: PeerEndpoint,
) -> bool {
active active
.iter() .keys()
.any(|endpoint| endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr) .any(|endpoint| endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr)
} }
/// Returns whether admitting `candidate` would exceed the per-source-IP /// Returns whether admitting `candidate` would exceed the per-source-IP
/// discovery budget across active negotiations and cooling-down attempts. /// discovery budget across active negotiations and cooling-down attempts.
fn source_ip_is_saturated( fn source_ip_is_saturated(
active: &HashSet<PeerEndpoint>, active: &HashMap<PeerEndpoint, std::net::IpAddr>,
recent: &mut RecentCandidates, recent: &mut RecentCandidates,
candidate: PeerEndpoint, source_ip: std::net::IpAddr,
now: tokio::time::Instant, now: tokio::time::Instant,
) -> bool { ) -> bool {
let ip = candidate.addr.ip();
let active_from_ip = active let active_from_ip = active
.iter() .values()
.filter(|endpoint| endpoint.addr.ip() == ip) .filter(|observed_ip| **observed_ip == source_ip)
.count(); .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 >= MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP
} }
fn candidate_is_admissible( fn candidate_is_admissible(
active: &HashSet<PeerEndpoint>, active: &HashMap<PeerEndpoint, std::net::IpAddr>,
recent: &mut RecentCandidates, recent: &mut RecentCandidates,
candidate: PeerEndpoint, candidate: PeerEndpoint,
source_ip: std::net::IpAddr,
now: tokio::time::Instant, now: tokio::time::Instant,
) -> bool { ) -> bool {
active.len() < MAX_ACTIVE_DISCOVERY_CANDIDATES active.len() < MAX_ACTIVE_DISCOVERY_CANDIDATES
&& !candidate_conflicts(active, candidate) && !candidate_conflicts(active, candidate)
&& !source_ip_is_saturated(active, recent, candidate, now) && !source_ip_is_saturated(active, recent, source_ip, now)
&& recent.try_record(candidate, now) && recent.try_record(candidate, source_ip, now)
} }
fn run_mdns_browser( 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 { fn parse_mdns_peer(service: &MdnsService) -> MdnsPeerInfo {
MdnsPeerInfo { MdnsPeerInfo {
addr: service.addr, addr: service.addr,
source_ip: service.source_ip,
peer_id: service peer_id: service
.properties .properties
.get("peer_id") .get("peer_id")
@@ -496,8 +504,8 @@ where
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use std::{ use std::{
collections::HashSet, collections::HashMap,
net::SocketAddr, net::{IpAddr, SocketAddr},
sync::{ sync::{
Arc, Arc,
atomic::{AtomicBool, AtomicUsize, Ordering}, atomic::{AtomicBool, AtomicUsize, Ordering},
@@ -562,6 +570,7 @@ mod tests {
let peer_id = Some(PeerId::from_bytes([7; 32])); let peer_id = Some(PeerId::from_bytes([7; 32]));
let unicast = MdnsPeerInfo { let unicast = MdnsPeerInfo {
addr: "192.168.1.50:42424".parse().expect("test address parses"), addr: "192.168.1.50:42424".parse().expect("test address parses"),
source_ip: "192.168.1.50".parse().expect("test source parses"),
peer_id, peer_id,
proto_ver: Some(lanspread_proto::PROTOCOL_VERSION), proto_ver: Some(lanspread_proto::PROTOCOL_VERSION),
}; };
@@ -569,6 +578,7 @@ mod tests {
let multicast = MdnsPeerInfo { let multicast = MdnsPeerInfo {
addr: "224.0.0.251:42424".parse().expect("test address parses"), addr: "224.0.0.251:42424".parse().expect("test address parses"),
source_ip: "192.168.1.50".parse().expect("test source parses"),
peer_id, peer_id,
proto_ver: Some(lanspread_proto::PROTOCOL_VERSION), proto_ver: Some(lanspread_proto::PROTOCOL_VERSION),
}; };
@@ -576,6 +586,7 @@ mod tests {
let zero_port = MdnsPeerInfo { let zero_port = MdnsPeerInfo {
addr: "192.168.1.50:0".parse().expect("test address parses"), addr: "192.168.1.50:0".parse().expect("test address parses"),
source_ip: "192.168.1.50".parse().expect("test source parses"),
peer_id, peer_id,
proto_ver: Some(lanspread_proto::PROTOCOL_VERSION), proto_ver: Some(lanspread_proto::PROTOCOL_VERSION),
}; };
@@ -585,7 +596,7 @@ mod tests {
#[test] #[test]
fn active_candidate_keys_bound_both_claimed_identity_and_address() { fn active_candidate_keys_bound_both_claimed_identity_and_address() {
let active_endpoint = endpoint(1, 12001); 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(1, 12002)));
assert!(candidate_conflicts(&active, endpoint(2, 12001))); assert!(candidate_conflicts(&active, endpoint(2, 12001)));
@@ -601,22 +612,26 @@ mod tests {
let now = tokio::time::Instant::now(); let now = tokio::time::Instant::now();
let first = endpoint(1, 12001); let first = endpoint(1, 12001);
let mut recent = RecentCandidates::default(); 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 { 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 { 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") { 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_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; 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); assert_eq!(recent.len(), 1);
} }
@@ -625,46 +640,52 @@ mod tests {
} }
#[test] #[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 now = tokio::time::Instant::now();
let mut active = HashSet::new(); let mut active = HashMap::new();
let mut recent = RecentCandidates::default(); 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 { for index in 0..MAX_DISCOVERY_CANDIDATES_PER_SOURCE_IP {
let seed = u8::try_from(index + 1).expect("test index fits u8"); 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( assert!(candidate_is_admissible(
&active, &active,
&mut recent, &mut recent,
candidate, candidate,
flood_source,
now now
)); ));
active.insert(candidate); active.insert(candidate, flood_source);
} }
assert!(!candidate_is_admissible( assert!(!candidate_is_admissible(
&active, &active,
&mut recent, &mut recent,
endpoint_at(100, flood_ip, 30000), endpoint_at(100, [10, 1, 1, 2], 30000),
flood_source,
now now
)); ));
// Other hosts are unaffected while the flood host is saturated. // Other hosts are unaffected while the flood host is saturated.
let other_source: IpAddr = [192, 168, 1, 67].into();
assert!(candidate_is_admissible( assert!(candidate_is_admissible(
&active, &active,
&mut recent, &mut recent,
endpoint_at(101, [192, 168, 1, 67], 42424), endpoint_at(101, [10, 1, 1, 3], 42424),
other_source,
now now
)); ));
// Completed negotiations keep counting while they cool down ... // 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); active.remove(&completed);
assert!(!candidate_is_admissible( assert!(!candidate_is_admissible(
&active, &active,
&mut recent, &mut recent,
endpoint_at(102, flood_ip, 30001), endpoint_at(102, [10, 1, 1, 4], 30001),
flood_source,
now now
)); ));
// ... and free the budget once the cooldown expires. // ... and free the budget once the cooldown expires.
@@ -672,7 +693,8 @@ mod tests {
assert!(candidate_is_admissible( assert!(candidate_is_admissible(
&active, &active,
&mut recent, &mut recent,
endpoint_at(102, flood_ip, 30001), endpoint_at(102, [10, 1, 1, 4], 30001),
flood_source,
after_cooldown after_cooldown
)); ));
} }
@@ -685,12 +707,13 @@ mod tests {
let mut active = (0..MAX_ACTIVE_DISCOVERY_CANDIDATES) let mut active = (0..MAX_ACTIVE_DISCOVERY_CANDIDATES)
.map(|index| { .map(|index| {
let seed = u8::try_from(index + 1).expect("test index fits u8"); 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::<HashSet<_>>(); .collect::<HashMap<_, _>>();
let mut recent = RecentCandidates::default(); let mut recent = RecentCandidates::default();
for candidate in &active { for (candidate, source_ip) in &active {
assert!(recent.try_record(*candidate, now)); assert!(recent.try_record(*candidate, *source_ip, now));
} }
let after_cooldown = now + DISCOVERY_CANDIDATE_COOLDOWN; let after_cooldown = now + DISCOVERY_CANDIDATE_COOLDOWN;
@@ -699,16 +722,18 @@ mod tests {
&active, &active,
&mut recent, &mut recent,
next, next,
next.addr.ip(),
after_cooldown, after_cooldown,
)); ));
assert_eq!(active.len(), MAX_ACTIVE_DISCOVERY_CANDIDATES); 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); active.remove(&completed);
assert!(candidate_is_admissible( assert!(candidate_is_admissible(
&active, &active,
&mut recent, &mut recent,
next, next,
next.addr.ip(),
after_cooldown, after_cooldown,
)); ));
} }