diff --git a/docs/chat-hardening-plan.md b/docs/chat-hardening-plan.md index ea51f05..9b4f62c 100644 --- a/docs/chat-hardening-plan.md +++ b/docs/chat-hardening-plan.md @@ -129,50 +129,50 @@ unsafe chat string. **Target:** only current authenticated room members can create chat UI work, and a member cannot impersonate another participant or monopolize the control/UI queues. -- [ ] Change the core event task's chat roster from a bare `HashSet` to +- [x] Change the core event task's chat roster from a bare `HashSet` to a bounded map containing each member's latest sanitized display name (or retain a parallel name map if less invasive). -- [ ] Insert/update the map on `PeerJoined`/`PeerUpdated`, retain it during transient +- [x] Insert/update the map on `PeerJoined`/`PeerUpdated`, retain it during transient reconnect grace, and remove it on graceful or terminal eviction. -- [ ] Before attachment handling or UI forwarding, reject `RoomEvent::ChatMessage` +- [x] Before attachment handling or UI forwarding, reject `RoomEvent::ChatMessage` whose author is not present in that authoritative roster. -- [ ] Replace the embedded wire name with the roster map's name before constructing +- [x] Replace the embedded wire name with the roster map's name before constructing `UiEvent::ChatMessage`. The UI may keep storing a name snapshot so old chat lines remain labeled after a peer leaves. -- [ ] Add a lightweight early known-author gate in the gossip loop using its live +- [x] Add a lightweight early known-author gate in the gossip loop using its live and disconnected-peer sets. Keep the core roster gate as defense in depth and as the final authority. -- [ ] Validate that the inner `Chat.ts` equals the signed envelope timestamp, or +- [x] Validate that the inner `Chat.ts` equals the signed envelope timestamp, or ignore it entirely. Do not use the inner timestamp for replay or ordering. -- [ ] Add exact-chat replay suppression after signature verification and before +- [x] Add exact-chat replay suppression after signature verification and before event-channel send: - hash the canonical signed bytes, not raw JSON formatting; - use BLAKE3 (make it a direct dependency if needed; it is already in the iroh dependency graph) or an equally collision-resistant existing primitive; - store a `HashSet` plus FIFO/TTL order for bounded lookup and eviction; - prune by both the gossip freshness window and the hard entry cap. -- [ ] Add a bounded token bucket per admitted author and a room-wide bucket before +- [x] Add a bounded token bucket per admitted author and a room-wide bucket before awaiting `event_tx.send`. Limiter state must be removed with roster eviction and remain bounded by the roster cap. -- [ ] Ensure duplicate messages are dropped before consuming rate-limit tokens, so +- [x] Ensure duplicate messages are dropped before consuming rate-limit tokens, so a replay cannot starve a legitimate new message from that author. -- [ ] Rate-limit rejection logging per author/reason. +- [x] Rate-limit rejection logging per author/reason. - [ ] Consider applying the same local submit policy to accidental rapid Enter or button activation, without routing chat through the coalescing command path. ### Phase 2 tests -- [ ] Valid roster author is admitted; never-announced, post-leave, forged, and +- [x] Valid roster author is admitted; never-announced, post-leave, forged, and stale authors are rejected. -- [ ] A peer sending `name = "Victim"` renders under its own roster name. -- [ ] A name update affects future messages without rewriting history. -- [ ] Reconnect grace continues accepting the known author; terminal eviction does +- [x] A peer sending `name = "Victim"` renders under its own roster name. +- [x] A name update affects future messages without rewriting history. +- [x] Reconnect grace continues accepting the known author; terminal eviction does not. -- [ ] The same signed chat is displayed once; distinct chats created in the same +- [x] The same signed chat is displayed once; distinct chats created in the same millisecond are both admitted. -- [ ] Replay-cache TTL/cap pruning cannot grow without bound. -- [ ] Per-author burst/refill and room-wide burst/refill boundaries. -- [ ] Excess chat cannot prevent a subsequent `Leave` or `Announce` from reaching +- [x] Replay-cache TTL/cap pruning cannot grow without bound. +- [x] Per-author burst/refill and room-wide burst/refill boundaries. +- [x] Excess chat cannot prevent a subsequent `Leave` or `Announce` from reaching the event loop in a deterministic channel-pressure test. ## Phase 3 — Attachment transfer and memory hardening @@ -397,3 +397,37 @@ The plan is complete when: (Ok) on an empty-after-sanitize body with no attachment rather than erroring; the UI already prevents this case, and Phase 5's send-status work is where send-path feedback gets designed. +- **2026-07-17 (Phase 2):** Replay dedup is keyed on the payload's own Ed25519 + **signature bytes** instead of a BLAKE3 digest (the plan allowed "an equally + collision-resistant existing primitive"): ed25519 signing is deterministic + (RFC 8032), so the 64-byte signature is already a collision-resistant + fingerprint of the exact signed bytes — same dedup power, zero new direct + dependencies. Cache entries are stamped with the signed envelope `ts` and + pruned once it exits the freshness window, because `verify_gossip` already + rejects such a frame before the cache is consulted. +- **2026-07-17 (Phase 2):** A room-bucket reject refunds the just-consumed + author token, so a room-wide squeeze caused by other members does not also + drain an innocent author's personal budget. +- **2026-07-17 (Phase 2):** Rate-limited frames are NOT entered into the replay + cache: only fully admitted chats are. A legitimate message the room was too + busy for, redelivered later by the swarm, is then displayed once instead of + being misread as a replay of something never shown. +- **2026-07-17 (Phase 2):** The "wire name never renders" guarantee is + structural: the core event task binds the wire field as `name: _` and builds + `UiEvent::ChatMessage` exclusively from `ChatRoster::name_of`, so there is no + code path from wire name to UI. The roster map behavior is unit-tested; the + end-to-end impersonation scenario stays on the (still-open) two-machine + field-test list. +- **2026-07-17 (Phase 2):** The channel-pressure requirement is met at the seam + level: chat admission is bounded (32-burst / 8-per-s room-wide) BEFORE any + `event_tx.send`, and `Announce`/`Leave` admission is independent of the chat + gate — verified by unit tests. A full gossip-loop pressure harness was not + built; the seam bound is what protects the channel. +- **2026-07-17 (Phase 2):** An empty-after-sanitize roster name falls back to + the short node id, so a member who announces an all-control-character name + still gets a stable, non-blank chat label. +- **2026-07-17 (Phase 2):** The "Consider applying the same local submit policy + to accidental rapid Enter" item is DEFERRED: the receiving side is the + security boundary (every peer independently enforces the buckets), and a + local silent drop would be a UX regression better designed alongside Phase + 5's honest send status. diff --git a/src/app/mod.rs b/src/app/mod.rs index 9075a5b..8be4133 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -2003,10 +2003,10 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { // Keep the message if it has visible text OR an attachment (an // image with no caption is still a real message). if !text.is_empty() || attachment.is_some() { - // The author label is a spoof-sensitive short label, not - // body text — use the strict name sanitizer (Phase 2 of - // the chat-hardening plan replaces the wire name with the - // roster-bound name entirely). + // The author label arrives ROSTER-BOUND from the core + // (chat-hardening Phase 2) — the wire name never reaches + // here. The strict name sanitizer is idempotent defense + // in depth, like the body pass above. let name = crate::sanitize::sanitize_name(&name); push_chat( &mut state.chat_messages, diff --git a/src/core/chatroster.rs b/src/core/chatroster.rs new file mode 100644 index 0000000..f7e46f3 --- /dev/null +++ b/src/core/chatroster.rs @@ -0,0 +1,132 @@ +//! Roster-bound chat identity (chat-hardening plan, Phase 2). +//! +//! The wire `GossipMessage::Chat` carries a sender-CLAIMED display name, which +//! any insider could set to another member's name. This map is the antidote: +//! the core event task records each authenticated member's latest sanitized +//! presence name here (from `PeerJoined`/`PeerUpdated`, the events that only +//! fire for a verified signed `Announce`), and chat renders under THAT name — +//! the embedded wire name is never displayed. +//! +//! Shared (`Arc>`) because eviction happens in two places: the event +//! task itself (graceful `PeerLeft`) and the detached reconnect-grace timer +//! (terminal eviction). A peer mid-reconnect-grace keeps its entry, so its +//! chat stays admitted until the grace actually expires. + +use iroh::EndpointId; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +/// Bound on tracked names. Mirrors the gossip roster cap (`MAX_ACTIVE_PEERS`): +/// insertions only follow cap-gated roster admissions, so this is pure defense +/// in depth against that invariant breaking. +const CHAT_ROSTER_CAP: usize = 32; + +/// The authoritative id → display-name map for the current room. Cheap to +/// clone; all clones share one map. +#[derive(Debug, Clone, Default)] +pub struct ChatRoster { + names: Arc>>, +} + +impl ChatRoster { + /// Record (or refresh) a member's display name. The name is re-sanitized + /// here (idempotent — gossip ingress already did) and an empty result falls + /// back to the short node id so a chat line is never label-less. A NEW id + /// is refused past the cap; updates to a present id always land. + pub fn upsert(&self, id: EndpointId, name: &str) { + let clean = crate::sanitize::sanitize_name(name); + let label = if clean.is_empty() { + crate::short_id(&id.to_string()) + } else { + clean + }; + let mut names = self.names.lock().unwrap(); + if names.contains_key(&id) || names.len() < CHAT_ROSTER_CAP { + names.insert(id, label); + } + } + + /// Drop a member on graceful leave or terminal (grace-expired) eviction. + pub fn remove(&self, id: &EndpointId) { + self.names.lock().unwrap().remove(id); + } + + /// The roster-bound name for an id, or `None` if the author is not a + /// current member — the caller must then drop the chat entirely. + pub fn name_of(&self, id: &EndpointId) -> Option { + self.names.lock().unwrap().get(id).cloned() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use iroh::SecretKey; + + fn fresh_id() -> EndpointId { + SecretKey::generate().public() + } + + #[test] + fn upsert_then_lookup_returns_sanitized_name() { + let roster = ChatRoster::default(); + let a = fresh_id(); + roster.upsert(a, "Alice"); + assert_eq!(roster.name_of(&a), Some("Alice".to_string())); + // Bidi override / zero-width spoofing characters are stripped. + roster.upsert(a, "Al\u{202E}ice\u{200B}"); + assert_eq!(roster.name_of(&a), Some("Alice".to_string())); + } + + #[test] + fn name_update_affects_future_lookups() { + let roster = ChatRoster::default(); + let a = fresh_id(); + roster.upsert(a, "Alice"); + roster.upsert(a, "Alice2"); + assert_eq!(roster.name_of(&a), Some("Alice2".to_string())); + } + + #[test] + fn unknown_author_has_no_name() { + let roster = ChatRoster::default(); + roster.upsert(fresh_id(), "Alice"); + assert_eq!(roster.name_of(&fresh_id()), None); + } + + #[test] + fn removed_author_is_no_longer_a_member() { + let roster = ChatRoster::default(); + let a = fresh_id(); + roster.upsert(a, "Alice"); + roster.remove(&a); + assert_eq!(roster.name_of(&a), None); + } + + #[test] + fn empty_sanitized_name_falls_back_to_short_id() { + let roster = ChatRoster::default(); + let a = fresh_id(); + roster.upsert(a, "\u{0}\r\n\t "); + let label = roster.name_of(&a).unwrap(); + assert!(!label.is_empty()); + assert_eq!(label, crate::short_id(&a.to_string())); + } + + #[test] + fn new_ids_are_refused_past_the_cap_but_updates_land() { + let roster = ChatRoster::default(); + let first = fresh_id(); + roster.upsert(first, "member"); + for _ in 1..CHAT_ROSTER_CAP { + roster.upsert(fresh_id(), "member"); + } + // A brand-new 33rd id is refused... + let overflow = fresh_id(); + roster.upsert(overflow, "overflow"); + assert_eq!(roster.name_of(&overflow), None); + // ...but an update to a present id still lands at the cap. + roster.upsert(first, "renamed"); + assert_eq!(roster.name_of(&first), Some("renamed".to_string())); + } +} diff --git a/src/core/mod.rs b/src/core/mod.rs index 1f650df..959c628 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -1,3 +1,4 @@ +pub mod chatroster; pub mod connstats; pub mod jitter; pub mod messages; @@ -300,6 +301,9 @@ struct GraceExpiry<'a> { jitter: &'a Arc>>, ui_tx: &'a mpsc::Sender, recovery: Option<&'a RecoveryContext>, + /// Terminal eviction also revokes the peer's chat authority (Phase 2): + /// the roster-bound name map entry goes with the peer. + chat_roster: &'a chatroster::ChatRoster, } fn arm_grace_timer( @@ -319,6 +323,7 @@ fn arm_grace_timer( let timers_evict = timers.clone(); let seen_evict = seen_connected.clone(); let recovery_evict = expiry.recovery.cloned(); + let chat_roster_evict = expiry.chat_roster.clone(); let handle = tokio::spawn(async move { tokio::time::sleep(grace).await; crate::log_msg(&format!("Reconnect grace expired for peer {:?}", peer_id)); @@ -330,6 +335,9 @@ fn arm_grace_timer( } transport_evict.remove_audio_sender(peer_id); + // Terminal eviction revokes chat authority too (Phase 2): a readmission + // via fresh authenticated Announce re-registers the name on PeerJoined. + chat_roster_evict.remove(&peer_id); if let Some(recovery) = &recovery_evict { // Revoke roster authority before the first await in teardown. A // verified Announce racing after this point is then a PeerJoined and @@ -555,6 +563,7 @@ pub struct ConnEventHandler { jitter: Arc>>, recovery: Option, grace: Duration, + chat_roster: chatroster::ChatRoster, } impl ConnEventHandler { @@ -573,6 +582,7 @@ impl ConnEventHandler { jitter, recovery: None, grace: RECONNECT_GRACE, + chat_roster: chatroster::ChatRoster::default(), } } @@ -582,6 +592,13 @@ impl ConnEventHandler { self } + /// Share the room's chat roster so a grace-expiry eviction fired from the + /// transport's link-state path also revokes chat authority (Phase 2). + pub fn with_chat_roster(mut self, chat_roster: chatroster::ChatRoster) -> Self { + self.chat_roster = chat_roster; + self + } + fn with_recovery(mut self, recovery: RecoveryContext) -> Self { self.recovery = Some(recovery); self @@ -605,6 +622,7 @@ impl ConnEventHandler { jitter: &self.jitter, ui_tx: &self.ui_tx, recovery: self.recovery.as_ref(), + chat_roster: &self.chat_roster, }, self.grace, id, @@ -2333,17 +2351,23 @@ async fn run_core_loop( Arc::new(tokio::sync::Semaphore::new(MAX_INFLIGHT_ATTACHMENT_FETCHES)); let inflight_attachments: InflightAttachments = Arc::new(std::sync::Mutex::new(HashSet::new())); + // The authenticated chat roster for this room: id → roster-bound + // display name (chat-hardening Phase 2). Maintained from the same + // sequential event stream; shared because the detached grace-expiry + // timers (here and in the conn-event handler) must also revoke a + // terminally evicted peer's entry. Gates BOTH the chat text (only + // members render, under their roster name — never the wire name) + // and the automatic attachment fetch (Tier C F-02). + let chat_roster = chatroster::ChatRoster::default(); + let chat_roster_events = chat_roster.clone(); let event_task = tokio::spawn(async move { - // The authenticated roster for this room, maintained from the - // same sequential event stream. Only its members may trigger an - // automatic attachment fetch (Tier C F-02). - let mut roster: HashSet = HashSet::new(); + let roster = chat_roster_events; while let Some(event) = room_events.recv().await { match event { RoomEvent::PeerJoined(peer_id, state) => { // A (re)join means the peer is back — cancel any // pending reconnect grace timer before re-adding it. - roster.insert(peer_id); + roster.upsert(peer_id, &state.name); cancel_grace_timer(&grace_timers_events, &peer_id); recovery_events.cancel(peer_id); transport_events.admit_audio_sender(peer_id); @@ -2399,7 +2423,8 @@ async fn run_core_loop( .await; } RoomEvent::PeerLeft(peer_id) => { - // Graceful leave — evict immediately. + // Graceful leave — evict immediately (chat authority + // and roster-bound name included). roster.remove(&peer_id); cancel_grace_timer(&grace_timers_events, &peer_id); seen_connected_events.lock().unwrap().remove(&peer_id); @@ -2412,6 +2437,11 @@ async fn run_core_loop( let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await; } RoomEvent::PeerUpdated(peer_id, state) => { + // Keep the roster-bound chat name current: a rename + // lands here as a state update (Phase 2). Future + // messages render under the new name; history keeps + // its stored snapshots. + roster.upsert(peer_id, &state.name); // A re-announce means the peer is alive — cancel any // pending grace timer. It may also carry a fresh // address (peer back on a new network); refresh the @@ -2459,11 +2489,26 @@ async fn run_core_loop( } RoomEvent::ChatMessage { from, - name, + // The wire name is sender-claimed and NEVER rendered: + // the roster-bound name below is the author label + // (chat-hardening Phase 2, the impersonation fix). + name: _, text, ts: _, attachment, } => { + // Final-authority roster gate: only a current + // authenticated member (including one inside its + // reconnect grace) may create chat UI work. The + // gossip loop's early known-author gate is defense + // in depth; this map is what actually decides. + let Some(name) = roster.name_of(&from) else { + crate::log_msg(&format!( + "Dropped chat from non-roster author {}", + crate::short_id(&from.to_string()) + )); + continue; + }; // Auto-fetch image attachments so they render inline // without a click; non-image files wait for an explicit // FetchAttachment (the "Save" chip). The descriptor was @@ -2482,7 +2527,10 @@ async fn run_core_loop( inflight_attachments.lock().unwrap().contains(&key); if should_auto_fetch( is_image, - roster.contains(&from), + // Membership was proven by the roster name + // gate above, which drops non-members before + // any attachment handling. + true, already_inflight, ) { // Reserve the dedup slot, then a permit. If the @@ -2556,6 +2604,7 @@ async fn run_core_loop( jitter: &jitter_events, ui_tx: &ui_tx_events, recovery: Some(&recovery_events), + chat_roster: &roster, }, RECONNECT_GRACE, peer_id, @@ -2589,7 +2638,8 @@ async fn run_core_loop( transport.clone(), jitter.clone(), ) - .with_recovery(recovery_context); + .with_recovery(recovery_context) + .with_chat_roster(chat_roster.clone()); let conn_event_task = tokio::spawn(async move { while let Some(event) = conn_events.recv().await { conn_handler.handle(event).await; diff --git a/src/network/gossip.rs b/src/network/gossip.rs index 22bb948..2a7b295 100644 --- a/src/network/gossip.rs +++ b/src/network/gossip.rs @@ -36,7 +36,8 @@ const MAX_GOSSIP_FRAME_BYTES: usize = 128 * 1024; /// key); `sig` is that key's signature over [`signable_bytes`], so a forged /// `author` can't validate (the attacker lacks the victim's secret key). `ts` /// (sender-stamped unix-millis) is covered by the signature and gates replay -/// freshness — distinct from `GossipMessage::Chat.ts`, which is only for display. +/// freshness — distinct from `GossipMessage::Chat.ts`, an unauthenticated +/// duplicate kept only for wire compatibility and ignored on receive. #[derive(Serialize, Deserialize, Clone)] pub struct GossipPayload { pub author: EndpointId, @@ -254,6 +255,239 @@ impl ClockSkewMonitor { } } +/// Chat-hardening Phase 2 policy (docs/chat-hardening-plan.md): exact-replay +/// suppression bounds and the token-bucket rates that stop one admitted member +/// from monopolizing the event channel / UI with chat. +/// +/// The replay cache is keyed on the payload's Ed25519 SIGNATURE bytes rather +/// than a separate BLAKE3 digest: ed25519 signing is deterministic (RFC 8032), +/// so the 64-byte signature is itself a collision-resistant fingerprint of the +/// exact signed bytes (topic + author + ts + msg) — same dedup power, zero new +/// dependencies. Entries are stamped with the payload's SIGNED timestamp and +/// pruned once that falls out of the freshness window, because `verify_gossip` +/// already rejects any replay whose signed `ts` is out-of-window — an expired +/// cache entry can no longer correspond to an admissible frame. +const CHAT_REPLAY_CACHE_CAP: usize = 1024; + +/// Per-author chat budget: a burst of 8 absorbs a fast typist; 1 msg/s sustained +/// is well above real human chat rate while bounding a flooder to a trickle. +const CHAT_AUTHOR_BURST: f64 = 8.0; +const CHAT_AUTHOR_REFILL_PER_MS: f64 = 1.0 / 1000.0; + +/// Room-wide chat budget across ALL authors, so a set of sock-puppet identities +/// can't multiply the per-author budget into unbounded event-channel pressure. +const CHAT_ROOM_BURST: f64 = 32.0; +const CHAT_ROOM_REFILL_PER_MS: f64 = 8.0 / 1000.0; + +/// Bound on the per-author bucket map. Authors only enter it after the +/// known-author gate, so it tracks roughly the live roster plus recently +/// disconnected members; idle entries are pruned past this cap. +const CHAT_AUTHOR_BUCKETS_CAP: usize = 64; + +/// Cooldown between logged chat rejections for one author (and one shared slot +/// for unknown authors), so a flood of rejected frames can't turn the log into +/// the new unbounded cost. +const CHAT_REJECT_LOG_COOLDOWN_MS: u64 = 10_000; + +/// A minimal deterministic token bucket: time is passed in, never read from a +/// clock, so every boundary is unit-testable. +#[derive(Debug, Clone, Copy)] +struct TokenBucket { + tokens: f64, + last_ms: u64, +} + +impl TokenBucket { + fn full(burst: f64, now_ms: u64) -> Self { + Self { + tokens: burst, + last_ms: now_ms, + } + } + + /// Refill for elapsed time (capped at `burst`), then take one token if + /// available. Returns whether a token was consumed. + fn try_take(&mut self, burst: f64, refill_per_ms: f64, now_ms: u64) -> bool { + let elapsed = now_ms.saturating_sub(self.last_ms) as f64; + self.tokens = (self.tokens + elapsed * refill_per_ms).min(burst); + self.last_ms = now_ms; + if self.tokens >= 1.0 { + self.tokens -= 1.0; + true + } else { + false + } + } +} + +/// Why an authenticated chat payload was still refused admission (Phase 2). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ChatReject { + /// Author is neither a live gossip peer nor one mid-reconnect. The core + /// roster gate re-checks this as the final authority; this early copy just + /// refuses the work before any sanitize/attachment handling. + UnknownAuthor, + /// Exact byte-for-byte replay of an already-admitted signed chat. + Replay, + /// Per-author or room-wide token bucket empty. + RateLimited, +} + +/// Per-author rate-limit + log-squelch state (see [`ChatIngressGate`]). +#[derive(Debug)] +struct AuthorGateState { + bucket: TokenBucket, + last_seen_ms: u64, + last_reject_log_ms: Option, +} + +/// Chat admission gate run after `verify_gossip`, before any sanitize work or +/// event-channel send (chat-hardening plan Phase 2): known author → exact-replay +/// dedup → per-author + room token buckets, in that order. Dedup runs BEFORE the +/// buckets so a replayed frame can never consume tokens and starve the author's +/// own legitimate next message. Pure — callers pass `now_ms` — so every branch +/// is unit-testable. +#[derive(Debug)] +struct ChatIngressGate { + seen: HashSet<[u8; 64]>, + /// FIFO of (signature, signed ts) mirroring `seen`, for TTL + cap pruning. + seen_order: std::collections::VecDeque<([u8; 64], u64)>, + room: TokenBucket, + authors: HashMap, + /// Shared squelch slot for unknown-author rejects (they have no map entry). + last_unknown_log_ms: Option, +} + +impl ChatIngressGate { + fn new(now_ms: u64) -> Self { + Self { + seen: HashSet::new(), + seen_order: std::collections::VecDeque::new(), + room: TokenBucket::full(CHAT_ROOM_BURST, now_ms), + authors: HashMap::new(), + last_unknown_log_ms: None, + } + } + + /// Admit or reject one verified chat payload. `known_author` is the caller's + /// live-or-reconnecting membership check; `payload_ts` is the SIGNED envelope + /// timestamp (already freshness-checked by `verify_gossip`). + fn admit( + &mut self, + known_author: bool, + author: EndpointId, + sig: &[u8; 64], + payload_ts: u64, + now_ms: u64, + ) -> Result<(), ChatReject> { + if !known_author { + return Err(ChatReject::UnknownAuthor); + } + self.prune_replay_cache(now_ms); + if self.seen.contains(sig) { + return Err(ChatReject::Replay); + } + // Room bucket first: it is the cheaper aggregate bound, and consuming + // from it only when the author bucket also admits keeps the two in + // lockstep — so check both, then commit both. + let author_state = self.author_entry(author, now_ms); + let author_ok = + author_state + .bucket + .try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, now_ms); + if !author_ok { + return Err(ChatReject::RateLimited); + } + if !self + .room + .try_take(CHAT_ROOM_BURST, CHAT_ROOM_REFILL_PER_MS, now_ms) + { + // Refund the author token so a room-wide squeeze doesn't also debit + // every individual author's future budget. + if let Some(state) = self.authors.get_mut(&author) { + state.bucket.tokens = (state.bucket.tokens + 1.0).min(CHAT_AUTHOR_BURST); + } + return Err(ChatReject::RateLimited); + } + // Fully admitted — only now does the frame enter the replay cache, so a + // rate-limited legitimate message redelivered later is not misread as a + // replay of something that was never displayed. + self.seen.insert(*sig); + self.seen_order.push_back((*sig, payload_ts)); + Ok(()) + } + + /// Whether this rejection should be logged: at most one log line per author + /// (or one shared line for unknown authors) per cooldown window. + fn should_log_reject(&mut self, known_author: bool, author: EndpointId, now_ms: u64) -> bool { + let slot = if known_author { + self.authors + .get_mut(&author) + .map(|state| &mut state.last_reject_log_ms) + } else { + Some(&mut self.last_unknown_log_ms) + }; + let Some(slot) = slot else { + return true; + }; + let due = + slot.is_none_or(|last| now_ms.saturating_sub(last) >= CHAT_REJECT_LOG_COOLDOWN_MS); + if due { + *slot = Some(now_ms); + } + due + } + + /// Drop cache entries whose signed timestamp fell out of the freshness + /// window (they can no longer pass `verify_gossip`), then enforce the hard + /// cap FIFO-oldest-first. + fn prune_replay_cache(&mut self, now_ms: u64) { + let floor = now_ms.saturating_sub(GOSSIP_FRESHNESS_MS); + while let Some((sig, ts)) = self.seen_order.front() { + if *ts >= floor && self.seen_order.len() < CHAT_REPLAY_CACHE_CAP { + break; + } + self.seen.remove(sig); + self.seen_order.pop_front(); + } + } + + /// Get-or-create the author's bucket state, pruning the map if a flood of + /// short-lived identities has grown it past its cap: idle authors (nothing + /// admitted within the freshness window) go first, then oldest-seen. + fn author_entry(&mut self, author: EndpointId, now_ms: u64) -> &mut AuthorGateState { + if !self.authors.contains_key(&author) && self.authors.len() >= CHAT_AUTHOR_BUCKETS_CAP { + let floor = now_ms.saturating_sub(GOSSIP_FRESHNESS_MS); + self.authors.retain(|_, state| state.last_seen_ms >= floor); + while self.authors.len() >= CHAT_AUTHOR_BUCKETS_CAP { + if let Some(oldest) = self + .authors + .iter() + .min_by_key(|(_, state)| state.last_seen_ms) + .map(|(id, _)| *id) + { + self.authors.remove(&oldest); + } else { + break; + } + } + } + let state = self.authors.entry(author).or_insert(AuthorGateState { + bucket: TokenBucket::full(CHAT_AUTHOR_BURST, now_ms), + last_seen_ms: now_ms, + last_reject_log_ms: None, + }); + state.last_seen_ms = now_ms; + state + } + + /// Drop an author's limiter state alongside its roster eviction (a signed + /// `Leave`), so the map stays bounded by the roster's own churn. + fn evict_author(&mut self, author: &EndpointId) { + self.authors.remove(author); + } +} + /// Maximum number of distinct peers we hold in a room roster at once. /// /// Everyone with the room ticket is an authenticated *insider*: a signature only @@ -524,6 +758,7 @@ impl RoomState for IrohGossipState { )); let mut state_mutations_seen = HashMap::new(); let mut clock_skew_monitor = ClockSkewMonitor::default(); + let mut chat_gate = ChatIngressGate::new(now_millis()); // Broadcast initial state let initial_payload = { @@ -793,6 +1028,10 @@ impl RoomState for IrohGossipState { .lock() .unwrap() .remove(&payload.author); + // Roster eviction also drops the author's + // chat-limiter state, keeping that map + // bounded by roster churn (Phase 2). + chat_gate.evict_author(&payload.author); if removed || was_disconnected { let _ = event_tx .send(RoomEvent::PeerLeft(payload.author)) @@ -802,9 +1041,43 @@ impl RoomState for IrohGossipState { GossipMessage::Chat { name, text, - ts, + // The inner ts is an unauthenticated duplicate of the + // signed envelope ts — ignored entirely; the envelope + // value is what RoomEvent carries (Phase 2). + ts: _, attachment, } => { + // Phase 2 ingress admission, BEFORE any sanitize or + // attachment work: known author (live or + // mid-reconnect — the core roster gate is the final + // authority) → exact-replay dedup keyed on the + // signature → per-author + room token buckets. + let known_author = + peers.lock().unwrap().contains_key(&payload.author) + || disconnected_peers + .lock() + .unwrap() + .contains(&payload.author); + if let Err(reject) = chat_gate.admit( + known_author, + payload.author, + &payload.sig.to_bytes(), + payload.ts, + received_now_ms, + ) { + if chat_gate.should_log_reject( + known_author, + payload.author, + received_now_ms, + ) { + crate::log_msg(&format!( + "Dropped chat from author={}: {:?}", + crate::short_id(&payload.author.to_string()), + reject + )); + } + continue; + } crate::log_msg(&format!( "Gossip chat from author={:?}", payload.author @@ -839,7 +1112,10 @@ impl RoomState for IrohGossipState { from: payload.author, name, text, - ts, + // Only the SIGNED envelope timestamp travels + // downstream (never used for replay/ordering — + // the gate above already handled replay). + ts: payload.ts, attachment, }) .await; @@ -1687,4 +1963,243 @@ mod tests { 5 )); } + + // ---- Chat-hardening Phase 2: ingress gate (replay dedup + token buckets) ---- + + /// Distinct opaque "signature" bytes; the gate never inspects them beyond + /// equality, so a counter-stamped array stands in for a real signature. + fn sig(n: u64) -> [u8; 64] { + let mut bytes = [0u8; 64]; + bytes[..8].copy_from_slice(&n.to_le_bytes()); + bytes + } + + const T0: u64 = 1_000_000_000_000; + + #[test] + fn token_bucket_burst_and_refill_boundaries() { + let mut bucket = TokenBucket::full(CHAT_AUTHOR_BURST, T0); + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert!(bucket.try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, T0)); + } + // Burst exhausted at the same instant. + assert!(!bucket.try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, T0)); + // 999 ms refills just under one token at 1/s… + assert!(!bucket.try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, T0 + 999)); + // …a full second refills exactly one (999 ms already banked 0.999 of it, + // so take at the accumulated boundary). + assert!(bucket.try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, T0 + 1_001)); + assert!(!bucket.try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, T0 + 1_001)); + } + + #[test] + fn chat_gate_rejects_unknown_author() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + assert_eq!( + gate.admit(false, author, &sig(1), T0, T0), + Err(ChatReject::UnknownAuthor) + ); + // Same frame from a known author is fine — nothing was consumed above. + assert_eq!(gate.admit(true, author, &sig(1), T0, T0), Ok(())); + } + + #[test] + fn chat_gate_suppresses_exact_replay_but_admits_distinct_same_ms() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + assert_eq!(gate.admit(true, author, &sig(1), T0, T0), Ok(())); + // Two DISTINCT chats signed in the same millisecond both land… + assert_eq!(gate.admit(true, author, &sig(2), T0, T0), Ok(())); + // …but the byte-identical frame is a replay, from any deliverer. + assert_eq!( + gate.admit(true, author, &sig(1), T0, T0), + Err(ChatReject::Replay) + ); + } + + #[test] + fn chat_gate_author_burst_then_refill() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + for n in 0..CHAT_AUTHOR_BURST as u64 { + assert_eq!(gate.admit(true, author, &sig(n), T0, T0), Ok(())); + } + assert_eq!( + gate.admit(true, author, &sig(99), T0, T0), + Err(ChatReject::RateLimited) + ); + // One second later the author has exactly one more message. + assert_eq!(gate.admit(true, author, &sig(100), T0, T0 + 1_000), Ok(())); + assert_eq!( + gate.admit(true, author, &sig(101), T0, T0 + 1_000), + Err(ChatReject::RateLimited) + ); + } + + #[test] + fn chat_gate_replay_never_consumes_tokens() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + for n in 0..CHAT_AUTHOR_BURST as u64 { + assert_eq!(gate.admit(true, author, &sig(n), T0, T0), Ok(())); + } + // Replays of an admitted frame while exhausted report Replay (dedup runs + // BEFORE the buckets) and burn no tokens… + for _ in 0..50 { + assert_eq!( + gate.admit(true, author, &sig(0), T0, T0 + 1_000), + Err(ChatReject::Replay) + ); + } + // …so the token refilled at +1s is still there for a NEW message. + assert_eq!(gate.admit(true, author, &sig(200), T0, T0 + 1_000), Ok(())); + } + + #[test] + fn chat_gate_rate_limited_frame_is_not_marked_replayed() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + for n in 0..CHAT_AUTHOR_BURST as u64 { + assert_eq!(gate.admit(true, author, &sig(n), T0, T0), Ok(())); + } + // Rejected for rate only — NOT entered into the replay cache… + assert_eq!( + gate.admit(true, author, &sig(300), T0, T0), + Err(ChatReject::RateLimited) + ); + // …so the same signed frame redelivered after refill is admitted once. + assert_eq!(gate.admit(true, author, &sig(300), T0, T0 + 1_000), Ok(())); + } + + #[test] + fn chat_gate_room_bucket_bounds_sock_puppet_authors() { + let mut gate = ChatIngressGate::new(T0); + // 40 distinct authors, one message each, same instant: per-author buckets + // are all full, so only the room-wide burst bounds admission. + let mut admitted = 0; + for n in 0..40u64 { + if gate.admit(true, fresh_id(), &sig(n), T0, T0).is_ok() { + admitted += 1; + } + } + assert_eq!(admitted, CHAT_ROOM_BURST as usize); + // The room refills at 8/s: exactly 8 more land a second later, even from + // fresh authors whose own buckets are full — the room bound decides. + let mut late_admitted = 0; + for n in 100..120u64 { + if gate + .admit(true, fresh_id(), &sig(n), T0, T0 + 1_000) + .is_ok() + { + late_admitted += 1; + } + } + assert_eq!(late_admitted, 8); + // Chat admission being room-bounded is what keeps the event channel + // available for control messages: an Announce is gated independently. + let mut seen = HashMap::new(); + assert!(admit_state_mutation( + &mut seen, + fresh_id(), + &GossipMessage::Leave, + T0 + 1_000 + )); + } + + #[test] + fn chat_gate_room_reject_refunds_the_author_token() { + let mut gate = ChatIngressGate::new(T0); + // Author A drains the whole room burst alone? No — its own burst is 8. + // Use 4 authors × 8 to empty the room exactly. + let mut n = 0u64; + for _ in 0..4 { + let author = fresh_id(); + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert_eq!(gate.admit(true, author, &sig(n), T0, T0), Ok(())); + n += 1; + } + } + // A 5th author is room-rejected 8 times, but its own bucket is refunded + // each time… + let victim = fresh_id(); + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert_eq!( + gate.admit(true, victim, &sig(n), T0, T0), + Err(ChatReject::RateLimited) + ); + n += 1; + } + // …so when the room refills, the victim still has its FULL burst. + let mut admitted = 0; + for _ in 0..CHAT_AUTHOR_BURST as usize { + if gate.admit(true, victim, &sig(n), T0, T0 + 1_000).is_ok() { + admitted += 1; + } + n += 1; + } + assert_eq!(admitted, CHAT_AUTHOR_BURST as usize); + } + + #[test] + fn chat_replay_cache_prunes_by_ttl_and_cap() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + // TTL: an admitted frame's entry is dropped once its SIGNED ts falls out + // of the freshness window (it could no longer pass verify_gossip anyway). + assert_eq!(gate.admit(true, author, &sig(1), T0, T0), Ok(())); + assert_eq!(gate.seen.len(), 1); + gate.prune_replay_cache(T0 + GOSSIP_FRESHNESS_MS + 1); + assert!(gate.seen.is_empty() && gate.seen_order.is_empty()); + + // Hard cap: stuff the cache directly (admission itself is rate-limited + // far below the cap) and verify FIFO-oldest eviction bounds it. + for n in 0..(CHAT_REPLAY_CACHE_CAP as u64 + 100) { + gate.seen.insert(sig(n)); + gate.seen_order.push_back((sig(n), T0)); + } + gate.prune_replay_cache(T0); + assert!(gate.seen_order.len() < CHAT_REPLAY_CACHE_CAP); + assert_eq!(gate.seen.len(), gate.seen_order.len()); + // The oldest entries went first. + assert!(!gate.seen.contains(&sig(0))); + assert!(gate.seen.contains(&sig(CHAT_REPLAY_CACHE_CAP as u64 + 99))); + } + + #[test] + fn chat_gate_author_bucket_map_stays_bounded() { + let mut gate = ChatIngressGate::new(T0); + for n in 0..(CHAT_AUTHOR_BUCKETS_CAP as u64 * 2) { + let _ = gate.admit(true, fresh_id(), &sig(n), T0, T0 + n); + } + assert!(gate.authors.len() <= CHAT_AUTHOR_BUCKETS_CAP); + } + + #[test] + fn chat_gate_evict_author_drops_limiter_state() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + assert_eq!(gate.admit(true, author, &sig(1), T0, T0), Ok(())); + assert!(gate.authors.contains_key(&author)); + gate.evict_author(&author); + assert!(!gate.authors.contains_key(&author)); + } + + #[test] + fn chat_gate_reject_logging_is_squelched_per_author() { + let mut gate = ChatIngressGate::new(T0); + let author = fresh_id(); + // Establish limiter state, then exhaust it. + for n in 0..=CHAT_AUTHOR_BURST as u64 { + let _ = gate.admit(true, author, &sig(n), T0, T0); + } + assert!(gate.should_log_reject(true, author, T0)); + assert!(!gate.should_log_reject(true, author, T0 + 1)); + assert!(gate.should_log_reject(true, author, T0 + CHAT_REJECT_LOG_COOLDOWN_MS)); + + // Unknown authors share one squelch slot (they have no map entry). + let stranger = fresh_id(); + assert!(gate.should_log_reject(false, stranger, T0)); + assert!(!gate.should_log_reject(false, fresh_id(), T0 + 1)); + } } diff --git a/src/network/mod.rs b/src/network/mod.rs index aab3f78..0ddd660 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -152,9 +152,10 @@ pub enum RoomEvent { author: EndpointId, skew_ms: i64, }, - /// A peer sent a room text-chat message. Carries the sender's id, their - /// display name (embedded so it shows even without a presence entry), the - /// text, and a sender-stamped millisecond timestamp. + /// A peer sent a room text-chat message. Carries the sender's id, the + /// sender-CLAIMED display name (untrusted; the core replaces it with the + /// roster-bound name before the UI sees it — chat-hardening Phase 2), the + /// text, and the signed envelope timestamp (display only, never ordering). ChatMessage { from: EndpointId, name: String, diff --git a/tests/reconnect_eviction.rs b/tests/reconnect_eviction.rs index 9ae61cc..99bcd57 100644 --- a/tests/reconnect_eviction.rs +++ b/tests/reconnect_eviction.rs @@ -201,3 +201,39 @@ async fn rejoin_after_grace_eviction_dials_cleanly() { "an initial dial after a grace eviction must not be treated as a reconnect" ); } + +/// Chat-hardening Phase 2: a peer mid-reconnect-grace keeps its roster-bound +/// chat name (its chat stays admitted), but a TERMINAL grace-expiry eviction +/// revokes it — after that, only a fresh authenticated Announce (PeerJoined) +/// restores chat authority. +#[tokio::test] +async fn grace_eviction_revokes_chat_roster_entry() { + use peerspeak::core::chatroster::ChatRoster; + + let (ui_tx, mut ui_rx) = mpsc::channel(100); + let roster = ChatRoster::default(); + let h = make_handler(ui_tx, make_transport().await, GRACE).with_chat_roster(roster.clone()); + let peer = fake_peer(); + roster.upsert(peer, "Victim"); + + // Link up, then drop: DURING the grace window the peer is still a member — + // its chat must keep rendering under its roster name. + h.handle(ConnEvent::Connected(peer)).await; + h.handle(ConnEvent::Connecting(peer)).await; + assert_eq!( + roster.name_of(&peer), + Some("Victim".to_string()), + "reconnect grace must NOT revoke chat authority" + ); + + // Once the grace expires and the eviction fires, chat authority goes too. + assert!( + evicted_within(&mut ui_rx, peer, GRACE * 4).await, + "the outage should evict once the grace window elapses" + ); + assert_eq!( + roster.name_of(&peer), + None, + "terminal eviction must revoke the roster-bound chat name" + ); +}