From 887e2456381ca64ed0da3a032e2687c705c4d018 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Wed, 2 Sep 2026 22:30:28 +0200 Subject: [PATCH] fix(peer): honour change hints only from the claimed peer's address Security audit findings NET-01 (partial) and Codex #9 ("unauthenticated hints can make the victim pull arbitrary known peers"). Inbound QUIC connections are intentionally anonymous: only the responder is authenticated, so any LAN host can open a stream and send `LibraryChanged`/`CallToPlayChanged` hints naming any `claimed_peer_id`. If the claimed peer was known and the forged session or revision did not match the cached snapshot, state sync scheduled a full pinned Hello pull to that peer. Sending one small forged hint to every node on the LAN therefore made all of them pull a victim's complete snapshot at once (reflected amplification), bounded only by the 5-second per-peer coalesce window. Full mutual TLS would bind hints to a verified identity but is a larger protocol change than this application warrants. Instead the hint now carries the source IP of the anonymous connection, and `hint_requires_pull_from_snapshot` discards any hint whose source IP differs from the address at which the claimed peer was last authenticated. On a LAN a QUIC connection cannot be established from a spoofed IP, so a third host can no longer select which peer this node pulls. A genuine peer whose address changed loses only the hint fast path; mDNS rediscovery and pinned liveness reconciliation still pick it up. `PeerEndpointGeneration::for_tests` is added under cfg(test) so unit tests can build a `PeerRevisionSnapshot`. Test plan: `just test`. The new test accepts a hint from the peer's address, rejects the same hint from another IP or with no address, and keeps the revision comparison for matching sources. Manual: with two peer-cli containers, adding a game on one still triggers the other to refresh its library promptly. Claude-Session: https://claude.ai/code/session_017C3Nbgwpdm3YNwZhhFLHwg --- crates/lanspread-peer/src/peer_db.rs | 5 + .../lanspread-peer/src/services/state_sync.rs | 98 ++++++++++++++++++- crates/lanspread-peer/src/services/stream.rs | 29 ++++-- 3 files changed, 122 insertions(+), 10 deletions(-) diff --git a/crates/lanspread-peer/src/peer_db.rs b/crates/lanspread-peer/src/peer_db.rs index 48bc8b3..faba5a1 100644 --- a/crates/lanspread-peer/src/peer_db.rs +++ b/crates/lanspread-peer/src/peer_db.rs @@ -30,6 +30,11 @@ impl PeerEndpointGeneration { pub const fn get(self) -> u64 { self.0 } + + #[cfg(test)] + pub(crate) const fn for_tests(value: u64) -> Self { + Self(value) + } } /// One endpoint generation retired by an authenticated topology transition. diff --git a/crates/lanspread-peer/src/services/state_sync.rs b/crates/lanspread-peer/src/services/state_sync.rs index 7d0ff4f..1584108 100644 --- a/crates/lanspread-peer/src/services/state_sync.rs +++ b/crates/lanspread-peer/src/services/state_sync.rs @@ -1,6 +1,6 @@ //! Bounded scheduling for responder-owned state pulls and change-hint fanout. -use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc, time::Duration}; +use std::{collections::HashMap, future::Future, net::IpAddr, pin::Pin, sync::Arc, time::Duration}; use futures::{StreamExt as _, stream::FuturesUnordered}; use lanspread_proto::{ChangeHint, PeerId}; @@ -37,6 +37,9 @@ struct LocalRevisions { struct HintTrigger { domain: StateDomain, hint: ChangeHint, + /// IP address the hint arrived from. Inbound connections are anonymous, + /// so this is the only evidence tying `hint.claimed_peer_id` to a sender. + source_ip: Option, } struct StateSyncInbox { @@ -98,11 +101,25 @@ impl StateSyncHandle { } /// Treats an inbound message only as a lossy invalidation hint. - pub(crate) fn schedule_hint(&self, domain: StateDomain, hint: ChangeHint) { + /// + /// `source_ip` is the address of the anonymous connection that delivered + /// the hint. It is later compared against the authenticated address of + /// the claimed peer so a third party cannot make this node pull from + /// arbitrary known peers. + pub(crate) fn schedule_hint( + &self, + domain: StateDomain, + hint: ChangeHint, + source_ip: Option, + ) { if hint.claimed_peer_id == self.local_peer_id { return; } - match self.hint_tx.try_send(HintTrigger { domain, hint }) { + match self.hint_tx.try_send(HintTrigger { + domain, + hint, + source_ip, + }) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(_)) => { log::trace!("Coalescing remote state hint because the bounded queue is full"); @@ -368,6 +385,14 @@ async fn hint_requires_pull(ctx: &NetworkServiceCtx, trigger: HintTrigger) -> bo hint_requires_pull_from_snapshot(ctx.peer_id, trigger, snapshot.as_ref()) } +/// Decides whether an untrusted hint justifies a pinned pull. +/// +/// Requester connections are not authenticated, so the claimed peer ID in a +/// hint proves nothing by itself. The hint is honoured only when it arrived +/// from the IP address at which the claimed peer was last authenticated; +/// anything else is discarded as a probable forgery. A genuine peer whose +/// address changed is still picked up by mDNS rediscovery and pinned +/// liveness reconciliation, which do not depend on hints. fn hint_requires_pull_from_snapshot( local_peer_id: PeerId, trigger: HintTrigger, @@ -379,6 +404,15 @@ fn hint_requires_pull_from_snapshot( let Some(snapshot) = snapshot else { return false; }; + if trigger.source_ip != Some(snapshot.endpoint.addr.ip()) { + log::debug!( + "Ignoring change hint for {} from {:?}; peer is authenticated at {}", + trigger.hint.claimed_peer_id, + trigger.source_ip, + snapshot.endpoint.addr + ); + return false; + } if snapshot.runtime_session_id != trigger.hint.runtime_session_id { return true; } @@ -658,6 +692,7 @@ mod tests { runtime_session_id: session(1), revision: 99, }, + source_ip: Some(IpAddr::from([192, 168, 1, 2])), }; if hint_requires_pull_from_snapshot(local, trigger, None) { let _ = try_enqueue_peer(&mut slots, claimed_peer_id, false); @@ -666,6 +701,63 @@ mod tests { assert!(slots.is_empty()); } + #[test] + fn hints_are_honoured_only_from_the_claimed_peers_authenticated_address() { + use lanspread_proto::PeerEndpoint; + + use crate::peer_db::PeerEndpointGeneration; + + let local = peer(1); + let claimed = peer(2); + let peer_addr: std::net::SocketAddr = "192.168.1.20:42424".parse().expect("addr"); + let snapshot = PeerRevisionSnapshot { + endpoint: PeerEndpoint::new(claimed, peer_addr), + generation: PeerEndpointGeneration::for_tests(1), + runtime_session_id: session(1), + library_revision: Some(3), + call_to_play_revision: Some(0), + }; + let trigger = |source_ip: Option| HintTrigger { + domain: StateDomain::Library, + hint: ChangeHint { + claimed_peer_id: claimed, + runtime_session_id: session(1), + revision: 4, + }, + source_ip, + }; + + assert!(hint_requires_pull_from_snapshot( + local, + trigger(Some(peer_addr.ip())), + Some(&snapshot) + )); + // A third host claiming the peer's identity must not trigger a pull. + assert!(!hint_requires_pull_from_snapshot( + local, + trigger(Some(IpAddr::from([192, 168, 1, 99]))), + Some(&snapshot) + )); + assert!(!hint_requires_pull_from_snapshot( + local, + trigger(None), + Some(&snapshot) + )); + // The revision check still applies for a matching source. + let same_revision = HintTrigger { + hint: ChangeHint { + revision: 3, + ..trigger(None).hint + }, + ..trigger(Some(peer_addr.ip())) + }; + assert!(!hint_requires_pull_from_snapshot( + local, + same_revision, + Some(&snapshot) + )); + } + #[tokio::test] async fn local_revision_watch_coalesces_to_latest_value() { let handle = StateSyncHandle::new(peer(1)); diff --git a/crates/lanspread-peer/src/services/stream.rs b/crates/lanspread-peer/src/services/stream.rs index 2428c84..de6d43c 100644 --- a/crates/lanspread-peer/src/services/stream.rs +++ b/crates/lanspread-peer/src/services/stream.rs @@ -67,6 +67,7 @@ pub(super) async fn handle_peer_stream( let mut framed_tx = FramedWrite::new(tx, control_codec()); log::trace!("{remote_addr:?} peer stream opened"); + let source_ip = remote_addr.map(|addr| addr.ip()); let first_frame = read_expected_frame(&mut framed_rx, &stream_shutdown).await; let mut control_permit = Some(control_permit); let mut _bulk_permit = None; @@ -87,9 +88,14 @@ pub(super) async fn handle_peer_stream( drop(control_permit.take()); if let Ok(permit) = bulk_permit { _bulk_permit = Some(permit); - let dispatched = - dispatch_request(&ctx, request, framed_tx, &stream_shutdown) - .await; + let dispatched = dispatch_request( + &ctx, + request, + source_ip, + framed_tx, + &stream_shutdown, + ) + .await; framed_tx = dispatched.writer; response_reset = dispatched.response_reset; } else { @@ -99,8 +105,14 @@ pub(super) async fn handle_peer_stream( response_reset = true; } } else { - let dispatched = - dispatch_request(&ctx, request, framed_tx, &stream_shutdown).await; + let dispatched = dispatch_request( + &ctx, + request, + source_ip, + framed_tx, + &stream_shutdown, + ) + .await; framed_tx = dispatched.writer; response_reset = dispatched.response_reset; } @@ -214,6 +226,7 @@ async fn read_expected_eof( async fn dispatch_request( ctx: &PeerCtx, request: Request, + source_ip: Option, framed_tx: ResponseWriter, stream_shutdown: &CancellationToken, ) -> DispatchResult { @@ -274,11 +287,13 @@ async fn dispatch_request( } } Request::LibraryChanged(hint) => { - ctx.state_sync.schedule_hint(StateDomain::Library, hint); + ctx.state_sync + .schedule_hint(StateDomain::Library, hint, source_ip); DispatchResult::close(framed_tx) } Request::CallToPlayChanged(hint) => { - ctx.state_sync.schedule_hint(StateDomain::CallToPlay, hint); + ctx.state_sync + .schedule_hint(StateDomain::CallToPlay, hint, source_ip); DispatchResult::close(framed_tx) } Request::GetGameFileChunk {