chat: roster-bound authorship, replay dedup, and rate limits (Phase 2)
CI / check (push) Successful in 2m59s
CI / check (push) Successful in 2m59s
Chat-hardening plan Phase 2 — only current authenticated room members can create chat UI work, impersonation via the wire name is structurally closed, and no member can monopolize the event channel: - core: new ChatRoster (bounded id -> sanitized-name map, shared) replaces the event task's bare HashSet; upserted on PeerJoined/PeerUpdated, removed on graceful PeerLeft AND terminal grace-expiry eviction (both timer paths). Non-roster chat is dropped before attachment handling; the rendered author label is the roster-bound name — the sender-claimed wire name is never read. - gossip: ChatIngressGate after verify_gossip, before any sanitize work or event send: early known-author gate (live + mid-reconnect peers), exact- replay suppression keyed on the deterministic Ed25519 signature (1024-entry cap + freshness-window TTL, zero new deps vs the plan's BLAKE3 option), then per-author (8 burst, 1/s) and room-wide (32 burst, 8/s) token buckets. Replays are detected before tokens are consumed; a room-bucket reject refunds the author token; rejection logging is squelched per author. - The inner Chat.ts is now ignored entirely; RoomEvent carries the signed envelope timestamp. 550 lib tests (+18), reconnect_eviction +1 (grace keeps chat authority, terminal eviction revokes it), clippy --all-targets -D warnings clean. Tests-green-only: the plan's two-machine field-test section remains open. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
+52
-18
@@ -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<EndpointId>` to
|
||||
- [x] Change the core event task's chat roster from a bare `HashSet<EndpointId>` 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.
|
||||
|
||||
+4
-4
@@ -2003,10 +2003,10 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
// 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,
|
||||
|
||||
@@ -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<Mutex<…>>`) 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<Mutex<HashMap<EndpointId, String>>>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
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()));
|
||||
}
|
||||
}
|
||||
+59
-9
@@ -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<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
||||
ui_tx: &'a mpsc::Sender<UiEvent>,
|
||||
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<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
||||
recovery: Option<RecoveryContext>,
|
||||
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<EndpointId> = 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;
|
||||
|
||||
+518
-3
@@ -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<u64>,
|
||||
}
|
||||
|
||||
/// 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<EndpointId, AuthorGateState>,
|
||||
/// Shared squelch slot for unknown-author rejects (they have no map entry).
|
||||
last_unknown_log_ms: Option<u64>,
|
||||
}
|
||||
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
+4
-3
@@ -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,
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user