From 1a3c481f4c6160e26f3643926046d7523dba430b Mon Sep 17 00:00:00 2001 From: Mollusk Date: Tue, 23 Jun 2026 16:20:45 -0400 Subject: [PATCH] fix(security): cap recovery-identity state (Tier C F-01 follow-up) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the remaining insider resource-exhaustion vector the Codex Tier C audit flagged: the active-roster cap bounds the live peer map, but a member could join (<=32), drop the link without a signed Leave, let the grace timer expire, and repeat with a fresh identity. Each abandoned identity grew two unbounded structures and kept doing periodic work forever: - known_peers[topic] (the retained rejoin/recovery dial table) was only pruned on a signed PeerLeft, so grace-evicted ghosts accumulated. - the recovery coordinator's active set + entries map had no identity cap and no terminal retry budget — backoff saturated at 60s and re-dialed a never-returning peer indefinitely. Two non-breaking, dependency-free bounds (no wire/protocol change): - MAX_RETAINED_PEERS=64 per topic via pure admit_retained() — refreshing a tracked peer always succeeds, a brand-new identity is rejected when full. Set above MAX_ACTIVE_PEERS=32 so legitimate rooms never hit it. - RECOVERY_TERMINAL_ATTEMPTS=12 (~7 min) via pure recovery_is_terminal(): the coordinator gives up, frees the active slot, and signals a new terminal channel; a small drain task forgets the retained address (so the table self-drains), scrubs seen-connected state, and emits PeerConnectionFailed. Giving up never blocks a legitimate reconnect: a peer returning after a long outage still rejoins on its own via a gossip announce — terminal eviction only stops us from dialing a peer that is not coming back, which was a latent leak even absent an attacker. +2 pure-seam unit tests (admit_retained, recovery_is_terminal); 418 lib tests green, clippy --all-targets clean. Co-Authored-By: Claude Opus 4.8 --- src/core/mod.rs | 112 ++++++++++++++++++++++++++++++++++++------- src/core/recovery.rs | 76 +++++++++++++++++++++++++---- 2 files changed, 162 insertions(+), 26 deletions(-) diff --git a/src/core/mod.rs b/src/core/mod.rs index 86e3f96..2f329e6 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -134,6 +134,25 @@ type SeenConnected = Arc>>; type KnownPeers = Arc>>>; +/// Per-topic cap on the retained rejoin-bootstrap / recovery target table +/// (Tier C recovery-identity cap). Set comfortably above the live-roster cap +/// (`gossip::MAX_ACTIVE_PEERS`, 32) so a legitimate room — even one where every +/// member drops at once during a relay outage — never hits it, while an insider +/// who grace-cycles distinct identities (join, drop without a signed Leave, +/// repeat) cannot grow the table without bound. Combined with the recovery +/// terminal budget (which forgets a retained address when it gives up), abandoned +/// identities drain on their own, so this cap is a deterministic ceiling rather +/// than a pinnable slot pool. +const MAX_RETAINED_PEERS: usize = 64; + +/// Whether a peer may be inserted into a retained-target table at `len` entries. +/// An update to an id already present is always allowed (it only refreshes an +/// address); a brand-new id is admitted only while below the cap. Mirrors the +/// gossip roster's `admit_into_roster` reject-when-full admission. +fn admit_retained(len: usize, is_new_id: bool, cap: usize) -> bool { + !is_new_id || len < cap +} + #[derive(Clone)] struct RecoveryContext { coordinator: RecoveryCoordinator, @@ -522,6 +541,7 @@ struct ActiveSession { event_task: tokio::task::JoinHandle<()>, conn_event_task: tokio::task::JoinHandle<()>, recovery_task: tokio::task::JoinHandle<()>, + recovery_terminal_task: tokio::task::JoinHandle<()>, grace_timers: GraceTimers, transport: Arc, /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. @@ -557,6 +577,7 @@ impl ActiveSession { handle.abort(); } self.recovery_task.abort(); + self.recovery_terminal_task.abort(); crate::log_msg("Aborted tasks"); let audio_backend_clone = audio_backend.clone(); @@ -1842,7 +1863,7 @@ async fn run_core_loop( // The topic of the room this event loop serves, so peer add/remove // updates the right per-topic bucket in `known_peers` (A8 archive). let room_topic = topic_id; - let (recovery_coordinator, recovery_task) = + let (recovery_coordinator, recovery_task, recovery_terminal_rx) = RecoveryCoordinator::spawn(room_state.clone()); let recovery_context = RecoveryContext { coordinator: recovery_coordinator, @@ -1851,6 +1872,28 @@ async fn run_core_loop( topic_id, }; let recovery_events = recovery_context.clone(); + // Drain the recovery coordinator's terminal-eviction signals (Tier C + // recovery-identity cap). When background recovery exhausts its budget + // for a peer, forget its retained dial target so the per-topic retain + // table drains, scrub residual seen-connected state, and surface the + // failure. A peer that later returns can still rejoin via a gossip + // announce, so giving up never blocks a legitimate reconnect. + let recovery_terminal_ctx = recovery_context.clone(); + let seen_connected_terminal = seen_connected.clone(); + let ui_tx_terminal = ui_tx.clone(); + let recovery_terminal_task = tokio::spawn(async move { + let mut terminal_rx = recovery_terminal_rx; + while let Some(peer_id) = terminal_rx.recv().await { + crate::log_msg(&format!( + "Background recovery gave up on peer {peer_id:?}; forgetting retained target" + )); + recovery_terminal_ctx.forget(peer_id); + seen_connected_terminal.lock().unwrap().remove(&peer_id); + let _ = ui_tx_terminal + .send(UiEvent::PeerConnectionFailed { id: peer_id }) + .await; + } + }); // Friends store + ui sender, so a connected peer who is a friend has // their saved address auto-healed (W7) — populates `last_addr` so the // presence scheduler can reach them later. @@ -1898,13 +1941,22 @@ async fn run_core_loop( .await; } // Retain this peer under this room's topic as a - // future rejoin bootstrap target (A8). - known_peers_events - .lock() - .unwrap() - .entry(room_topic) - .or_default() - .insert(peer_id, state.addr.clone()); + // future rejoin bootstrap target (A8), bounded by the + // per-topic retain cap (Tier C recovery-identity cap): + // refreshing a peer we already track is always allowed, + // a brand-new identity only while below the cap. + { + let mut kp = known_peers_events.lock().unwrap(); + let bucket = kp.entry(room_topic).or_default(); + let is_new_id = !bucket.contains_key(&peer_id); + if admit_retained(bucket.len(), is_new_id, MAX_RETAINED_PEERS) { + bucket.insert(peer_id, state.addr.clone()); + } else { + crate::log_msg(&format!( + "Retain table full ({MAX_RETAINED_PEERS}); not retaining new peer {peer_id:?} for rejoin" + )); + } + } // If a multitrack recording is live, give this peer // its own stem track (silence-padded back to t=0). if is_multitrack_events.load(Ordering::Relaxed) @@ -1955,13 +2007,21 @@ async fn run_core_loop( .await; } // Refresh this room's retained rejoin target with the - // fresh addr (A8). - known_peers_events - .lock() - .unwrap() - .entry(room_topic) - .or_default() - .insert(peer_id, state.addr.clone()); + // fresh addr (A8), under the per-topic retain cap. A + // re-announce from a peer we already track always + // refreshes; a new identity is bounded by the cap. + { + let mut kp = known_peers_events.lock().unwrap(); + let bucket = kp.entry(room_topic).or_default(); + let is_new_id = !bucket.contains_key(&peer_id); + if admit_retained(bucket.len(), is_new_id, MAX_RETAINED_PEERS) { + bucket.insert(peer_id, state.addr.clone()); + } else { + crate::log_msg(&format!( + "Retain table full ({MAX_RETAINED_PEERS}); not retaining new peer {peer_id:?} for rejoin" + )); + } + } let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await; } RoomEvent::ChatMessage { from, name, text, ts: _, attachment } => { @@ -2079,6 +2139,7 @@ async fn run_core_loop( event_task, conn_event_task, recovery_task, + recovery_terminal_task, grace_timers, transport: transport.clone(), #[cfg(target_os = "linux")] @@ -2665,11 +2726,26 @@ async fn run_core_loop( #[cfg(test)] mod tests { use super::{ - apply_volume, audio_datagram_len_ok, frame_level, mix_frames, mix_stereo_frames, - next_game_change, should_auto_fetch, stereo_to_mono, KnownPeers, MicLevelMeter, - PeerSpeakTicket, MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES, + admit_retained, apply_volume, audio_datagram_len_ok, frame_level, mix_frames, + mix_stereo_frames, next_game_change, should_auto_fetch, stereo_to_mono, KnownPeers, + MicLevelMeter, PeerSpeakTicket, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, + MIC_LEVEL_REPORT_SAMPLES, }; + #[test] + fn admit_retained_rejects_only_new_ids_at_the_cap() { + // Below the cap, a brand-new identity is retained. + assert!(admit_retained(0, true, MAX_RETAINED_PEERS)); + assert!(admit_retained(MAX_RETAINED_PEERS - 1, true, MAX_RETAINED_PEERS)); + // At the cap, a brand-new identity is refused — this is the bound that stops + // an insider grace-cycling distinct identities from growing the retain table. + assert!(!admit_retained(MAX_RETAINED_PEERS, true, MAX_RETAINED_PEERS)); + // A peer already tracked always refreshes, even at (or past) the cap: it only + // updates an existing address and never adds a slot. + assert!(admit_retained(MAX_RETAINED_PEERS, false, MAX_RETAINED_PEERS)); + assert!(admit_retained(MAX_RETAINED_PEERS + 5, false, MAX_RETAINED_PEERS)); + } + #[test] fn auto_fetch_only_for_roster_images_not_already_inflight() { // The happy path: a roster author's brand-new image attachment. diff --git a/src/core/recovery.rs b/src/core/recovery.rs index 587c28c..63fd5d2 100644 --- a/src/core/recovery.rs +++ b/src/core/recovery.rs @@ -22,6 +22,27 @@ fn recovery_delay(attempt: usize) -> Duration { RECOVERY_DELAYS[attempt.min(RECOVERY_DELAYS.len() - 1)] } +/// Terminal retry budget for background recovery. After this many failed attempts +/// the coordinator gives up: it drops the entry, frees the active slot, and signals +/// the event task to forget the retained address (Tier C recovery-identity cap). +/// +/// With the [`RECOVERY_DELAYS`] backoff this is roughly seven minutes of dialing +/// (1+2+4+8+15+30+60s, then 60s steps), far beyond any normal transient outage. A +/// genuine peer returning after a longer outage still rejoins on its own via a +/// gossip announce, so giving up only stops us from dialing a peer that is not +/// coming back — it does not break legitimate reconnect-after-outage. +const RECOVERY_TERMINAL_ATTEMPTS: usize = 12; + +/// Capacity of the terminal-eviction notification channel. Bounded; on the rare +/// event of saturation the entry is still removed (the dial work stops) and only +/// the retained-address forget is skipped, which the per-topic retain cap bounds. +const RECOVERY_TERMINAL_CAPACITY: usize = 64; + +/// Whether `attempt` completed recoveries have exhausted the terminal budget. +fn recovery_is_terminal(attempt: usize, max_attempts: usize) -> bool { + attempt >= max_attempts +} + enum RecoveryCommand { Start { peer_id: EndpointId, @@ -60,19 +81,24 @@ pub(super) struct RecoveryCoordinator { } impl RecoveryCoordinator { - pub(super) fn spawn(room_state: Arc) -> (Self, JoinHandle<()>) { + pub(super) fn spawn( + room_state: Arc, + ) -> (Self, JoinHandle<()>, mpsc::Receiver) { Self::spawn_inner(room_state) } - fn spawn_inner(room_state: Arc) -> (Self, JoinHandle<()>) { + fn spawn_inner( + room_state: Arc, + ) -> (Self, JoinHandle<()>, mpsc::Receiver) { let (tx, rx) = mpsc::channel(RECOVERY_COMMAND_CAPACITY); + let (terminal_tx, terminal_rx) = mpsc::channel(RECOVERY_TERMINAL_CAPACITY); let active = Arc::new(Mutex::new(HashSet::new())); let handle = Self { tx, active: active.clone(), }; - let task = tokio::spawn(run_coordinator(room_state, active, rx)); - (handle, task) + let task = tokio::spawn(run_coordinator(room_state, active, rx, terminal_tx)); + (handle, task, terminal_rx) } /// Reserve one recovery slot before grace-expiry teardown begins. Returns @@ -116,6 +142,7 @@ async fn run_coordinator( room_state: Arc, active: Arc>>, mut rx: mpsc::Receiver, + terminal_tx: mpsc::Sender, ) { let mut entries: HashMap = HashMap::new(); @@ -155,9 +182,24 @@ async fn run_coordinator( entries.remove(&peer_id); continue; } - if let Some(entry) = entries.get_mut(&peer_id) { + // Advance the backoff, then check the terminal budget. + // `attempt` counts completed attempts, so the delay + // uses the current value before it is incremented. + let terminal = if let Some(entry) = entries.get_mut(&peer_id) { entry.next_attempt = scheduled_at + recovery_delay(entry.attempt); entry.attempt = entry.attempt.saturating_add(1); + recovery_is_terminal(entry.attempt, RECOVERY_TERMINAL_ATTEMPTS) + } else { + false + }; + if terminal { + // Give up on a peer that has not returned within the + // budget: drop its entry, free the active slot, and + // signal the event task to forget its retained + // address so the per-topic retain table drains. + entries.remove(&peer_id); + active.lock().unwrap().remove(&peer_id); + let _ = terminal_tx.try_send(peer_id); } } } @@ -210,6 +252,23 @@ mod tests { assert_eq!(actual, vec![1, 2, 4, 8, 15, 30, 60, 60, 60, 60]); } + #[test] + fn recovery_budget_is_terminal_only_at_or_past_the_cap() { + assert!(!recovery_is_terminal(0, RECOVERY_TERMINAL_ATTEMPTS)); + assert!(!recovery_is_terminal( + RECOVERY_TERMINAL_ATTEMPTS - 1, + RECOVERY_TERMINAL_ATTEMPTS + )); + assert!(recovery_is_terminal( + RECOVERY_TERMINAL_ATTEMPTS, + RECOVERY_TERMINAL_ATTEMPTS + )); + assert!(recovery_is_terminal( + RECOVERY_TERMINAL_ATTEMPTS + 5, + RECOVERY_TERMINAL_ATTEMPTS + )); + } + #[test] fn recovery_slots_are_deduplicated_and_cancel_immediately() { let (tx, mut rx) = mpsc::channel(4); @@ -244,9 +303,10 @@ mod tests { #[tokio::test] async fn coordinator_attempts_rebootstrap_immediately() { let (attempts_tx, mut attempts_rx) = mpsc::unbounded_channel(); - let (coordinator, task) = RecoveryCoordinator::spawn_inner(Arc::new(RecordingRoom { - attempts: attempts_tx, - })); + let (coordinator, task, _terminal_rx) = + RecoveryCoordinator::spawn_inner(Arc::new(RecordingRoom { + attempts: attempts_tx, + })); let peer_id = SecretKey::generate().public(); let addr = EndpointAddr::from(peer_id);