fix(security): cap recovery-identity state (Tier C F-01 follow-up)
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 <noreply@anthropic.com>
This commit is contained in:
+94
-18
@@ -134,6 +134,25 @@ type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;
|
||||
type KnownPeers =
|
||||
Arc<std::sync::Mutex<HashMap<[u8; 32], HashMap<EndpointId, EndpointAddr>>>>;
|
||||
|
||||
/// 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<IrohTransport>,
|
||||
/// 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.
|
||||
|
||||
+68
-8
@@ -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<IrohGossipState>) -> (Self, JoinHandle<()>) {
|
||||
pub(super) fn spawn(
|
||||
room_state: Arc<IrohGossipState>,
|
||||
) -> (Self, JoinHandle<()>, mpsc::Receiver<EndpointId>) {
|
||||
Self::spawn_inner(room_state)
|
||||
}
|
||||
|
||||
fn spawn_inner(room_state: Arc<dyn RecoveryRoom>) -> (Self, JoinHandle<()>) {
|
||||
fn spawn_inner(
|
||||
room_state: Arc<dyn RecoveryRoom>,
|
||||
) -> (Self, JoinHandle<()>, mpsc::Receiver<EndpointId>) {
|
||||
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<dyn RecoveryRoom>,
|
||||
active: Arc<Mutex<HashSet<EndpointId>>>,
|
||||
mut rx: mpsc::Receiver<RecoveryCommand>,
|
||||
terminal_tx: mpsc::Sender<EndpointId>,
|
||||
) {
|
||||
let mut entries: HashMap<EndpointId, RecoveryEntry> = 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);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user