Compare commits
6
Commits
v0.4.0
...
3b640726d7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3b640726d7 | ||
|
|
381e00bc0e | ||
|
|
1a3c481f4c | ||
|
|
f927567105 | ||
|
|
5c11947bd7 | ||
|
|
7349744d16 |
@@ -8,7 +8,7 @@ it once, then you and I connect directly to each other.
|
||||
|
||||
## 1. Install it
|
||||
|
||||
1. Double-click **`peerspeak-0.3.0-setup.exe`** (the file I sent you).
|
||||
1. Double-click **`peerspeak-0.4.0-setup.exe`** (the file I sent you).
|
||||
|
||||
2. **Windows will probably show a blue "Windows protected your PC" warning.**
|
||||
This is normal — it shows up for any app that isn't from a big company with a
|
||||
|
||||
@@ -12,7 +12,7 @@ runtime, so there are no extra DLLs to bundle. The installer payload is just the
|
||||
## Version compatibility
|
||||
|
||||
The installer version tracks the crate version in `Cargo.toml` (currently
|
||||
**0.3.0**) — keep `MyAppVersion` in `peerspeak.iss` in sync when it changes.
|
||||
**0.4.0**) — keep `MyAppVersion` in `peerspeak.iss` in sync when it changes.
|
||||
|
||||
Per `VERSIONING.md`, a **MINOR** bump in `0.x` is a **breaking wire change**:
|
||||
peers on different MINOR versions can't connect (they fail fast at the
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed).
|
||||
|
||||
#define MyAppName "PeerSpeak"
|
||||
#define MyAppVersion "0.3.0"
|
||||
#define MyAppVersion "0.4.0"
|
||||
#define MyAppPublisher "mollusk"
|
||||
#define MyAppExeName "peerspeak.exe"
|
||||
|
||||
|
||||
+346
-128
@@ -23,7 +23,7 @@ use iced::{
|
||||
Point, Rectangle, Renderer, Size,
|
||||
};
|
||||
use iroh::EndpointId;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::collections::{HashMap, HashSet, VecDeque};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
@@ -123,8 +123,8 @@ struct ChatEntry {
|
||||
/// for any future system-generated lines.
|
||||
from: Option<String>,
|
||||
/// Optional file attachment descriptor. The bytes (if fetched) live in
|
||||
/// `AppState.attachment_data` keyed by `attachment.id`; the entry only holds
|
||||
/// the descriptor so history stays cheap.
|
||||
/// `AppState.attachments` keyed by `(author, attachment.id)`; the entry only
|
||||
/// holds the descriptor so history stays cheap.
|
||||
attachment: Option<crate::files::ChatAttachment>,
|
||||
}
|
||||
|
||||
@@ -137,6 +137,91 @@ enum AttachmentState {
|
||||
Failed(String),
|
||||
}
|
||||
|
||||
/// Identifies one fetched attachment by BOTH the authoring peer and the
|
||||
/// attachment id. The id is attacker-chosen, so a malicious peer can reuse a
|
||||
/// victim's id to alias displayed/saved content; keying on the author too means
|
||||
/// each chat line resolves only its own sender's bytes (Tier C F-12).
|
||||
type AttachmentKey = (EndpointId, crate::files::AttachmentId);
|
||||
|
||||
/// Cap on retained attachment results so an insider streaming distinct
|
||||
/// attachments can't grow the cache without bound (Tier C F-02). Sized well
|
||||
/// above any realistic on-screen image working set.
|
||||
const ATTACHMENT_CACHE_CAP: usize = 64;
|
||||
|
||||
/// One cached attachment result: its fetch state plus, for ready images, the
|
||||
/// pre-built iced image handle (built once on arrival, not per redraw — the
|
||||
/// e917c53 flicker fix). `handle` is `None` for files and failures.
|
||||
#[derive(Debug, Clone)]
|
||||
struct AttachmentEntry {
|
||||
state: AttachmentState,
|
||||
handle: Option<iced::widget::image::Handle>,
|
||||
}
|
||||
|
||||
/// Bounded store of fetched chat-attachment results, keyed by [`AttachmentKey`].
|
||||
///
|
||||
/// Eviction is insertion-order (oldest first), NOT true LRU: iced's `view`
|
||||
/// borrows `&self`, so the render read path cannot reorder an access-ordered
|
||||
/// cache. With a generous cap the newest entries — the ones actually on screen —
|
||||
/// are always retained, so a normal session never evicts; only an abusive stream
|
||||
/// of distinct attachments hits the bound (Tier C F-02).
|
||||
///
|
||||
/// Construct via [`AttachmentCache::new`] — there is deliberately no `Default`,
|
||||
/// because a zero cap would make `insert` evict endlessly.
|
||||
#[derive(Debug)]
|
||||
struct AttachmentCache {
|
||||
entries: HashMap<AttachmentKey, AttachmentEntry>,
|
||||
/// Keys in insertion order; the front is the eviction candidate. Holds
|
||||
/// exactly the present keys (one entry each), so it is bounded by `cap`.
|
||||
order: VecDeque<AttachmentKey>,
|
||||
cap: usize,
|
||||
}
|
||||
|
||||
impl AttachmentCache {
|
||||
fn new(cap: usize) -> Self {
|
||||
Self { entries: HashMap::new(), order: VecDeque::new(), cap: cap.max(1) }
|
||||
}
|
||||
|
||||
/// Insert or replace an entry. A brand-new key evicts the oldest entries
|
||||
/// until there is room; replacing an existing key keeps its position (and so
|
||||
/// its age), only updating the value.
|
||||
fn insert(&mut self, key: AttachmentKey, state: AttachmentState, handle: Option<iced::widget::image::Handle>) {
|
||||
if !self.entries.contains_key(&key) {
|
||||
while self.entries.len() >= self.cap {
|
||||
match self.order.pop_front() {
|
||||
Some(old) => {
|
||||
self.entries.remove(&old);
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
self.order.push_back(key);
|
||||
}
|
||||
self.entries.insert(key, AttachmentEntry { state, handle });
|
||||
}
|
||||
|
||||
fn get(&self, key: &AttachmentKey) -> Option<&AttachmentState> {
|
||||
self.entries.get(key).map(|e| &e.state)
|
||||
}
|
||||
|
||||
fn handle(&self, key: &AttachmentKey) -> Option<&iced::widget::image::Handle> {
|
||||
self.entries.get(key).and_then(|e| e.handle.as_ref())
|
||||
}
|
||||
|
||||
fn is_ready(&self, key: &AttachmentKey) -> bool {
|
||||
matches!(self.entries.get(key), Some(AttachmentEntry { state: AttachmentState::Ready(_), .. }))
|
||||
}
|
||||
|
||||
fn clear(&mut self) {
|
||||
self.entries.clear();
|
||||
self.order.clear();
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn len(&self) -> usize {
|
||||
self.entries.len()
|
||||
}
|
||||
}
|
||||
|
||||
/// Cap on retained chat history so a long call can't grow it without bound.
|
||||
const CHAT_HISTORY_MAX: usize = 300;
|
||||
|
||||
@@ -289,12 +374,15 @@ pub enum AppMessage {
|
||||
PickAttachmentFile,
|
||||
/// Result of the attach picker: (filename, bytes), or None if cancelled.
|
||||
AttachmentFilePicked(Option<(String, Vec<u8>)>),
|
||||
/// Save (downloading first if needed) a received attachment to disk.
|
||||
SaveAttachment(crate::files::AttachmentId),
|
||||
/// Save (downloading first if needed) a received attachment to disk. Carries
|
||||
/// the full `(author, id)` key so the correct sender's bytes are fetched and
|
||||
/// saved even if another peer reused the same attachment id (Tier C F-12).
|
||||
SaveAttachment(AttachmentKey),
|
||||
/// Result of the async save dialog: a status line to show, or None if cancelled.
|
||||
AttachmentSaved(Option<String>),
|
||||
/// Fetch (if needed) and start an inline audio attachment.
|
||||
PlayAudio(crate::files::AttachmentId),
|
||||
/// Fetch (if needed) and start an inline audio attachment. Carries the full
|
||||
/// `(author, id)` key (see [`AppMessage::SaveAttachment`]).
|
||||
PlayAudio(AttachmentKey),
|
||||
PauseAudio,
|
||||
ResumeAudio,
|
||||
SeekAudio(crate::files::AttachmentId, f32),
|
||||
@@ -451,15 +539,14 @@ pub struct AppState {
|
||||
/// Room text-chat history (newest last) and the pending input line.
|
||||
chat_messages: Vec<ChatEntry>,
|
||||
chat_input: String,
|
||||
/// Fetched/failed state for chat attachments, keyed by attachment id.
|
||||
/// Session-only (cleared on leave); never persisted.
|
||||
attachment_data: HashMap<crate::files::AttachmentId, AttachmentState>,
|
||||
/// Cached iced image handles for ready image attachments, keyed by id, so we
|
||||
/// don't re-upload to the GPU every redraw (the e917c53 avatar flicker fix).
|
||||
image_handle_cache: HashMap<crate::files::AttachmentId, iced::widget::image::Handle>,
|
||||
/// Attachment ids the user asked to save before the bytes arrived; when the
|
||||
/// fetch completes a save dialog is opened for them.
|
||||
pending_saves: std::collections::HashSet<crate::files::AttachmentId>,
|
||||
/// Fetched/failed bytes + cached image handles for chat attachments, keyed by
|
||||
/// `(author, id)` and bounded. Session-only (cleared on leave); never
|
||||
/// persisted. (Tier C F-02 bound + F-12 author keying.)
|
||||
attachments: AttachmentCache,
|
||||
/// Attachments the user asked to save before the bytes arrived; when the
|
||||
/// fetch completes a save dialog is opened for them. Keyed by `(author, id)`
|
||||
/// so a same-id attachment from a different sender can't trigger the save.
|
||||
pending_saves: HashSet<AttachmentKey>,
|
||||
/// Clip ids waiting for the existing attachment fetch path to return bytes.
|
||||
pending_plays: HashSet<crate::files::AttachmentId>,
|
||||
/// Filename-hinted audio whose fetched bytes or decoder validation failed;
|
||||
@@ -536,8 +623,7 @@ impl AppState {
|
||||
self.locally_muted.clear();
|
||||
self.chat_messages.clear();
|
||||
self.chat_input.clear();
|
||||
self.attachment_data.clear();
|
||||
self.image_handle_cache.clear();
|
||||
self.attachments.clear();
|
||||
self.pending_saves.clear();
|
||||
self.pending_plays.clear();
|
||||
self.invalid_audio.clear();
|
||||
@@ -658,9 +744,8 @@ impl Default for AppState {
|
||||
recording: false,
|
||||
recording_started: None,
|
||||
chat_messages: Vec::new(),
|
||||
attachment_data: HashMap::new(),
|
||||
image_handle_cache: HashMap::new(),
|
||||
pending_saves: std::collections::HashSet::new(),
|
||||
attachments: AttachmentCache::new(ATTACHMENT_CACHE_CAP),
|
||||
pending_saves: HashSet::new(),
|
||||
pending_plays: HashSet::new(),
|
||||
invalid_audio: HashSet::new(),
|
||||
clip_player,
|
||||
@@ -1186,31 +1271,31 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
});
|
||||
}
|
||||
}
|
||||
UiEvent::AttachmentReady { id, data } => {
|
||||
// Bytes arrived. For images we can cache the iced handle now
|
||||
// (built once, not per redraw). If the user was waiting to save
|
||||
// this file, the save dialog is opened from update() below by
|
||||
// checking pending_saves — done lazily so this arm stays simple.
|
||||
if crate::files::validate_image_bytes(&data).is_some() {
|
||||
state.image_handle_cache.insert(
|
||||
id,
|
||||
iced::widget::image::Handle::from_bytes(data.clone()),
|
||||
);
|
||||
}
|
||||
let needs_save = state.pending_saves.remove(&id);
|
||||
UiEvent::AttachmentReady { from, id, data } => {
|
||||
// Bytes arrived for this specific (author, id). For images we
|
||||
// can cache the iced handle now (built once, not per redraw).
|
||||
// If the user was waiting to save this file, the save dialog
|
||||
// is opened from update() below by checking pending_saves —
|
||||
// done lazily so this arm stays simple.
|
||||
let key = (from, id);
|
||||
let handle = crate::files::validate_image_bytes(&data)
|
||||
.is_some()
|
||||
.then(|| iced::widget::image::Handle::from_bytes(data.clone()));
|
||||
let needs_save = state.pending_saves.remove(&key);
|
||||
let needs_play = state.pending_plays.remove(&id);
|
||||
state.attachment_data.insert(id, AttachmentState::Ready(data));
|
||||
state.attachments.insert(key, AttachmentState::Ready(data), handle);
|
||||
if needs_play {
|
||||
play_ready_audio(state, id);
|
||||
play_ready_audio(state, key);
|
||||
}
|
||||
if needs_save {
|
||||
return save_attachment_task(state, id);
|
||||
return save_attachment_task(state, key);
|
||||
}
|
||||
}
|
||||
UiEvent::AttachmentFailed { id, error } => {
|
||||
state.pending_saves.remove(&id);
|
||||
UiEvent::AttachmentFailed { from, id, error } => {
|
||||
let key = (from, id);
|
||||
state.pending_saves.remove(&key);
|
||||
state.pending_plays.remove(&id);
|
||||
state.attachment_data.insert(id, AttachmentState::Failed(error.clone()));
|
||||
state.attachments.insert(key, AttachmentState::Failed(error.clone()), None);
|
||||
state.status_message = format!("Attachment failed: {error}");
|
||||
}
|
||||
UiEvent::ScreenShareStarted => {
|
||||
@@ -1870,17 +1955,18 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
id,
|
||||
};
|
||||
// Keep our own bytes locally so we see our own attachment inline
|
||||
// immediately (others fetch it off the file plane).
|
||||
if kind == crate::files::AttachmentKind::Image
|
||||
&& crate::files::validate_image_bytes(&bytes).is_some()
|
||||
{
|
||||
// immediately (others fetch it off the file plane). Cache under
|
||||
// our own (author, id) key so the render path — which keys the
|
||||
// line by `from` = self_id — finds them.
|
||||
if let Ok(self_eid) = state.self_id.parse::<EndpointId>() {
|
||||
let key = (self_eid, id);
|
||||
let handle = (kind == crate::files::AttachmentKind::Image
|
||||
&& crate::files::validate_image_bytes(&bytes).is_some())
|
||||
.then(|| iced::widget::image::Handle::from_bytes(bytes.clone()));
|
||||
state
|
||||
.image_handle_cache
|
||||
.insert(id, iced::widget::image::Handle::from_bytes(bytes.clone()));
|
||||
.attachments
|
||||
.insert(key, AttachmentState::Ready(bytes.clone()), handle);
|
||||
}
|
||||
state
|
||||
.attachment_data
|
||||
.insert(id, AttachmentState::Ready(bytes.clone()));
|
||||
push_chat(
|
||||
&mut state.chat_messages,
|
||||
ChatEntry {
|
||||
@@ -1898,21 +1984,18 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
});
|
||||
}
|
||||
}
|
||||
AppMessage::SaveAttachment(id) => {
|
||||
AppMessage::SaveAttachment(key) => {
|
||||
// If we already have the bytes, save now; otherwise fetch from the
|
||||
// sender and save when AttachmentReady arrives (pending_saves).
|
||||
if matches!(state.attachment_data.get(&id), Some(AttachmentState::Ready(_))) {
|
||||
return save_attachment_task(state, id);
|
||||
} else if let Some((from, att)) = find_attachment_source(state, id) {
|
||||
if let Ok(eid) = from.parse::<EndpointId>() {
|
||||
state.pending_saves.insert(id);
|
||||
state.status_message = format!("Downloading {}…", att.name);
|
||||
let _ = state
|
||||
.controller
|
||||
.send(CoreCommand::FetchAttachment { from: eid, attachment: att });
|
||||
} else {
|
||||
state.status_message = "Can't download: unknown sender.".to_string();
|
||||
}
|
||||
// sender and save when AttachmentReady arrives (pending_saves). The
|
||||
// key's author half is the exact sender of the clicked line.
|
||||
if state.attachments.is_ready(&key) {
|
||||
return save_attachment_task(state, key);
|
||||
} else if let Some(att) = find_attachment_source(state, key) {
|
||||
state.pending_saves.insert(key);
|
||||
state.status_message = format!("Downloading {}…", att.name);
|
||||
let _ = state
|
||||
.controller
|
||||
.send(CoreCommand::FetchAttachment { from: key.0, attachment: att });
|
||||
}
|
||||
}
|
||||
AppMessage::AttachmentSaved(msg) => {
|
||||
@@ -1920,21 +2003,18 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
state.status_message = m;
|
||||
}
|
||||
}
|
||||
AppMessage::PlayAudio(id) => {
|
||||
if matches!(state.attachment_data.get(&id), Some(AttachmentState::Ready(_))) {
|
||||
play_ready_audio(state, id);
|
||||
} else if let Some((from, att)) = find_attachment_source(state, id) {
|
||||
if let Ok(eid) = from.parse::<EndpointId>() {
|
||||
// Repeated clicks while the transfer is pending must not
|
||||
// launch duplicate fetches.
|
||||
if state.pending_plays.insert(id) {
|
||||
state.status_message = format!("Loading {}…", att.name);
|
||||
let _ = state
|
||||
.controller
|
||||
.send(CoreCommand::FetchAttachment { from: eid, attachment: att });
|
||||
}
|
||||
} else {
|
||||
state.status_message = "Can't play: unknown sender.".to_string();
|
||||
AppMessage::PlayAudio(key) => {
|
||||
if state.attachments.is_ready(&key) {
|
||||
play_ready_audio(state, key);
|
||||
} else if let Some(att) = find_attachment_source(state, key) {
|
||||
// Repeated clicks while the transfer is pending must not launch
|
||||
// duplicate fetches. (pending_plays is id-keyed — it's coupled to
|
||||
// the id-keyed clip player; same-id collisions are cosmetic.)
|
||||
if state.pending_plays.insert(key.1) {
|
||||
state.status_message = format!("Loading {}…", att.name);
|
||||
let _ = state
|
||||
.controller
|
||||
.send(CoreCommand::FetchAttachment { from: key.0, attachment: att });
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2192,17 +2272,35 @@ fn push_chat(messages: &mut Vec<ChatEntry>, entry: ChatEntry) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Find the sender id + descriptor for a received attachment by its id, so a
|
||||
/// fetch can be addressed. Returns `None` for our own attachments or an unknown
|
||||
/// id.
|
||||
/// Pick the default save-dialog filename for an attachment, matched by the FULL
|
||||
/// `(author, id)` key — not the bare id — so a peer reusing another sender's id
|
||||
/// can't supply the filename (and extension) for a different line (Tier C F-12
|
||||
/// metadata residual). Falls back to "download" if the line is gone. Matches own
|
||||
/// and received lines alike (our own `from` = `self_id` parses to the key author).
|
||||
fn attachment_default_name(messages: &[ChatEntry], key: AttachmentKey) -> String {
|
||||
messages
|
||||
.iter()
|
||||
.find_map(|m| {
|
||||
let att = m.attachment.as_ref()?;
|
||||
let from = m.from.as_ref()?.parse::<EndpointId>().ok()?;
|
||||
(att.id == key.1 && from == key.0).then(|| att.name.clone())
|
||||
})
|
||||
.unwrap_or_else(|| "download".to_string())
|
||||
}
|
||||
|
||||
/// Find the chat-attachment descriptor for an exact `(author, id)` key among
|
||||
/// received (non-own) messages. Matching on the author too — not just the id —
|
||||
/// means a peer reusing another sender's id can't redirect the fetch to the
|
||||
/// wrong line (Tier C F-12).
|
||||
fn find_attachment_source(
|
||||
state: &AppState,
|
||||
id: crate::files::AttachmentId,
|
||||
) -> Option<(String, crate::files::ChatAttachment)> {
|
||||
key: AttachmentKey,
|
||||
) -> Option<crate::files::ChatAttachment> {
|
||||
state.chat_messages.iter().find_map(|m| {
|
||||
let att = m.attachment.as_ref()?;
|
||||
if att.id == id && !m.mine {
|
||||
Some((m.from.clone()?, att.clone()))
|
||||
let from = m.from.as_ref()?;
|
||||
if att.id == key.1 && !m.mine && from.parse::<EndpointId>().ok() == Some(key.0) {
|
||||
Some(att.clone())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
@@ -2211,13 +2309,15 @@ fn find_attachment_source(
|
||||
|
||||
/// Validate cached bytes and hand them to the independent clip player. A false
|
||||
/// filename hint falls back to the generic file chip without reaching rodio.
|
||||
fn play_ready_audio(state: &mut AppState, id: crate::files::AttachmentId) {
|
||||
let Some(AttachmentState::Ready(data)) = state.attachment_data.get(&id) else {
|
||||
fn play_ready_audio(state: &mut AppState, key: AttachmentKey) {
|
||||
let Some(AttachmentState::Ready(data)) = state.attachments.get(&key) else {
|
||||
return;
|
||||
};
|
||||
let id = key.1;
|
||||
if crate::files::is_probably_audio(data) {
|
||||
let bytes = data.clone();
|
||||
state.invalid_audio.remove(&id);
|
||||
state.clip_player.play(id, data.clone());
|
||||
state.clip_player.play(id, bytes);
|
||||
} else {
|
||||
state.invalid_audio.insert(id);
|
||||
state.status_message = "This attachment is not valid supported audio.".to_string();
|
||||
@@ -2235,21 +2335,12 @@ fn play_ready_audio(state: &mut AppState, id: crate::files::AttachmentId) {
|
||||
/// the Linux `xdg-desktop-portal`/GTK backend wedges the dialog — Save/Cancel
|
||||
/// stop responding. The file *picker* paths already use the async variant; this
|
||||
/// is the one save path that must match.
|
||||
fn save_attachment_task(state: &AppState, id: crate::files::AttachmentId) -> Task<AppMessage> {
|
||||
let Some(AttachmentState::Ready(data)) = state.attachment_data.get(&id) else {
|
||||
fn save_attachment_task(state: &AppState, key: AttachmentKey) -> Task<AppMessage> {
|
||||
let Some(AttachmentState::Ready(data)) = state.attachments.get(&key) else {
|
||||
return Task::none();
|
||||
};
|
||||
let data = data.clone();
|
||||
let default_name = state
|
||||
.chat_messages
|
||||
.iter()
|
||||
.find_map(|m| {
|
||||
m.attachment
|
||||
.as_ref()
|
||||
.filter(|a| a.id == id)
|
||||
.map(|a| a.name.clone())
|
||||
})
|
||||
.unwrap_or_else(|| "download".to_string());
|
||||
let default_name = attachment_default_name(&state.chat_messages, key);
|
||||
Task::perform(
|
||||
async move {
|
||||
let handle = rfd::AsyncFileDialog::new()
|
||||
@@ -4382,7 +4473,14 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
);
|
||||
// Attachment row (indented under the message), if any.
|
||||
if let Some(att) = &m.attachment {
|
||||
let data = state.attachment_data.get(&att.id);
|
||||
// This line's cache key is (its author, the attachment id).
|
||||
// `None` only for a system line or an unparseable author.
|
||||
let key: Option<AttachmentKey> = m
|
||||
.from
|
||||
.as_ref()
|
||||
.and_then(|f| f.parse::<EndpointId>().ok())
|
||||
.map(|eid| (eid, att.id));
|
||||
let data = key.as_ref().and_then(|k| state.attachments.get(k));
|
||||
let elem: Element<'_, AppMessage> =
|
||||
if let Some(AttachmentState::Failed(e)) = data {
|
||||
text(format!("⚠ {} — {e}", att.name))
|
||||
@@ -4390,7 +4488,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
.color(color_red)
|
||||
.into()
|
||||
} else if att.kind == crate::files::AttachmentKind::Image {
|
||||
match state.image_handle_cache.get(&att.id) {
|
||||
match key.as_ref().and_then(|k| state.attachments.handle(k)) {
|
||||
Some(handle) => iced::widget::image(handle.clone())
|
||||
.width(iced::Length::Fixed(260.0))
|
||||
.into(),
|
||||
@@ -4419,7 +4517,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
button(text("Pause").size(12)).on_press(AppMessage::PauseAudio)
|
||||
} else {
|
||||
button(text("Play").size(12))
|
||||
.on_press(AppMessage::PlayAudio(att.id))
|
||||
.on_press_maybe(key.map(AppMessage::PlayAudio))
|
||||
}
|
||||
.style(b_style(
|
||||
color_blue,
|
||||
@@ -4447,7 +4545,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
"Download"
|
||||
})
|
||||
.size(12))
|
||||
.on_press(AppMessage::SaveAttachment(att.id))
|
||||
.on_press_maybe(key.map(AppMessage::SaveAttachment))
|
||||
.style(b_style(
|
||||
color_surface,
|
||||
color_overlay,
|
||||
@@ -4493,7 +4591,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
.size(12)
|
||||
.color(color_text),
|
||||
button(text(btn_label).size(12))
|
||||
.on_press(AppMessage::SaveAttachment(att.id))
|
||||
.on_press_maybe(key.map(AppMessage::SaveAttachment))
|
||||
.style(b_style(
|
||||
color_blue,
|
||||
color_lavender,
|
||||
@@ -5526,30 +5624,35 @@ fn avatar_badge<'a>(name: &str, key: &str, size: f32) -> Element<'a, AppMessage>
|
||||
.into()
|
||||
}
|
||||
|
||||
/// Maximum distinct avatar images we keep handles for. Each avatar is bounded to
|
||||
/// 48 KiB / 256×256 at ingest, so a 64-entry LRU caps this cache at a few MB
|
||||
/// regardless of how many distinct avatars peers publish over time (Tier C F-03).
|
||||
const AVATAR_CACHE_CAP: usize = 64;
|
||||
|
||||
thread_local! {
|
||||
/// Cache of avatar image handles, keyed by a hash of the PNG bytes, so the
|
||||
/// SAME `image::Handle` (and thus the same GPU texture id) is reused across
|
||||
/// Bounded cache of avatar image handles, keyed by PNG content, so the SAME
|
||||
/// `image::Handle` (and thus the same GPU texture id) is reused across
|
||||
/// redraws. `image::Handle::from_bytes` mints a fresh *unique* id on every
|
||||
/// call, so building handles inline in `view()` made iced re-upload the
|
||||
/// texture on every repaint — including the redraws fired on each mouse move —
|
||||
/// which showed up as constant flicker. Lives on the (single) UI thread.
|
||||
static AVATAR_HANDLE_CACHE: std::cell::RefCell<HashMap<u64, iced::widget::image::Handle>> =
|
||||
std::cell::RefCell::new(HashMap::new());
|
||||
/// which showed up as constant flicker. A peer can publish an unbounded stream
|
||||
/// of distinct valid avatars over a session, so the cache is an LRU (bounded +
|
||||
/// byte-equality keyed) rather than a plain map (Tier C F-03). Lives on the
|
||||
/// (single) UI thread.
|
||||
static AVATAR_HANDLE_CACHE:
|
||||
std::cell::RefCell<crate::avatar::ByteLru<iced::widget::image::Handle>> =
|
||||
std::cell::RefCell::new(crate::avatar::ByteLru::new(AVATAR_CACHE_CAP));
|
||||
}
|
||||
|
||||
/// A stable image handle for these exact PNG bytes (cached by content hash), so
|
||||
/// it keeps the same id across redraws — see [`AVATAR_HANDLE_CACHE`].
|
||||
/// A stable image handle for these exact PNG bytes (cached by content), so it
|
||||
/// keeps the same id across redraws — see [`AVATAR_HANDLE_CACHE`].
|
||||
fn cached_image_handle(bytes: bytes::Bytes) -> iced::widget::image::Handle {
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
bytes.as_ref().hash(&mut hasher);
|
||||
let key = hasher.finish();
|
||||
AVATAR_HANDLE_CACHE.with(|cache| {
|
||||
cache
|
||||
.borrow_mut()
|
||||
.entry(key)
|
||||
.or_insert_with(|| iced::widget::image::Handle::from_bytes(bytes))
|
||||
.clone()
|
||||
.get_or_insert(bytes.as_ref(), || {
|
||||
iced::widget::image::Handle::from_bytes(bytes.clone())
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5794,12 +5897,127 @@ impl Program<AppMessage> for Icon {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
format_duration, initial_window_position, reconnect_attempt_chime, reconnected_chime,
|
||||
set_peer_gate_config, set_peer_volume_config, AppConfig, AppState, AttachmentState,
|
||||
ChatEntry, GateMeter, METER_MAX,
|
||||
attachment_default_name, format_duration, initial_window_position, reconnect_attempt_chime,
|
||||
reconnected_chime, set_peer_gate_config, set_peer_volume_config, AppConfig, AppState,
|
||||
AttachmentCache, AttachmentState, ChatEntry, GateMeter, METER_MAX,
|
||||
};
|
||||
use iroh::SecretKey;
|
||||
|
||||
/// Two distinct (peer, id) keys for cache tests; `id` may be shared to model
|
||||
/// a malicious peer reusing a victim's attachment id (Tier C F-12).
|
||||
fn peer_key(id: [u8; 32]) -> super::AttachmentKey {
|
||||
(SecretKey::generate().public(), id)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attachment_cache_evicts_oldest_when_full() {
|
||||
let mut cache = AttachmentCache::new(2);
|
||||
let k1 = peer_key([1u8; 32]);
|
||||
let k2 = peer_key([2u8; 32]);
|
||||
let k3 = peer_key([3u8; 32]);
|
||||
cache.insert(k1, AttachmentState::Ready(vec![1]), None);
|
||||
cache.insert(k2, AttachmentState::Ready(vec![2]), None);
|
||||
assert_eq!(cache.len(), 2);
|
||||
// Inserting a third NEW key evicts the oldest (k1), not the newest.
|
||||
cache.insert(k3, AttachmentState::Ready(vec![3]), None);
|
||||
assert_eq!(cache.len(), 2);
|
||||
assert!(cache.get(&k1).is_none(), "oldest should be evicted");
|
||||
assert!(cache.get(&k2).is_some());
|
||||
assert!(cache.get(&k3).is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attachment_cache_replace_keeps_position_and_count() {
|
||||
let mut cache = AttachmentCache::new(2);
|
||||
let k1 = peer_key([1u8; 32]);
|
||||
let k2 = peer_key([2u8; 32]);
|
||||
cache.insert(k1, AttachmentState::Failed("pending".into()), None);
|
||||
cache.insert(k2, AttachmentState::Ready(vec![2]), None);
|
||||
// Replacing k1 (Failed -> Ready) must NOT bump it to newest; it stays the
|
||||
// eviction candidate, and the count is unchanged (no order leak).
|
||||
cache.insert(k1, AttachmentState::Ready(vec![1]), None);
|
||||
assert_eq!(cache.len(), 2);
|
||||
let k3 = peer_key([3u8; 32]);
|
||||
cache.insert(k3, AttachmentState::Ready(vec![3]), None);
|
||||
assert!(cache.get(&k1).is_none(), "replaced entry kept its old age");
|
||||
assert!(cache.get(&k2).is_some());
|
||||
assert!(cache.get(&k3).is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attachment_cache_same_id_distinct_authors_do_not_alias() {
|
||||
// The F-12 core property: two peers sending the SAME attachment id keep
|
||||
// separate bytes — one never overwrites or aliases the other.
|
||||
let mut cache = AttachmentCache::new(8);
|
||||
let shared_id = [7u8; 32];
|
||||
let victim = peer_key(shared_id);
|
||||
let attacker = peer_key(shared_id);
|
||||
cache.insert(victim, AttachmentState::Ready(vec![1, 1, 1]), None);
|
||||
cache.insert(attacker, AttachmentState::Ready(vec![9, 9, 9]), None);
|
||||
assert_eq!(cache.len(), 2);
|
||||
assert!(matches!(cache.get(&victim), Some(AttachmentState::Ready(b)) if b == &[1, 1, 1]));
|
||||
assert!(matches!(cache.get(&attacker), Some(AttachmentState::Ready(b)) if b == &[9, 9, 9]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attachment_cache_is_ready_and_handle_and_clear() {
|
||||
let mut cache = AttachmentCache::new(4);
|
||||
let k = peer_key([5u8; 32]);
|
||||
cache.insert(
|
||||
k,
|
||||
AttachmentState::Ready(vec![1]),
|
||||
Some(iced::widget::image::Handle::from_bytes(vec![1])),
|
||||
);
|
||||
assert!(cache.is_ready(&k));
|
||||
assert!(cache.handle(&k).is_some());
|
||||
let failed = peer_key([6u8; 32]);
|
||||
cache.insert(failed, AttachmentState::Failed("nope".into()), None);
|
||||
assert!(!cache.is_ready(&failed));
|
||||
assert!(cache.handle(&failed).is_none());
|
||||
cache.clear();
|
||||
assert_eq!(cache.len(), 0);
|
||||
assert!(cache.get(&k).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attachment_cache_cap_zero_is_clamped_to_one() {
|
||||
let mut cache = AttachmentCache::new(0);
|
||||
let k = peer_key([1u8; 32]);
|
||||
cache.insert(k, AttachmentState::Ready(vec![1]), None);
|
||||
assert_eq!(cache.len(), 1);
|
||||
assert!(cache.get(&k).is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attachment_default_name_matches_full_key_not_bare_id() {
|
||||
// Two chat lines carry the SAME attachment id but come from different
|
||||
// senders. The save-dialog default filename must be the one belonging to
|
||||
// the clicked (author, id) — not whichever line happens to match the bare
|
||||
// id first (Tier C F-12 metadata residual).
|
||||
let victim = SecretKey::generate().public();
|
||||
let attacker = SecretKey::generate().public();
|
||||
let shared_id = [7u8; 32];
|
||||
let mk = |from: iroh::EndpointId, fname: &str| ChatEntry {
|
||||
name: "Peer".to_string(),
|
||||
text: String::new(),
|
||||
mine: false,
|
||||
from: Some(from.to_string()),
|
||||
attachment: Some(crate::files::ChatAttachment {
|
||||
name: fname.to_string(),
|
||||
size: 3,
|
||||
kind: crate::files::AttachmentKind::File,
|
||||
id: shared_id,
|
||||
}),
|
||||
};
|
||||
// Attacker's line is FIRST in history, so a bare-id scan would pick it.
|
||||
let messages = vec![mk(attacker, "evil.sh"), mk(victim, "report.pdf")];
|
||||
assert_eq!(attachment_default_name(&messages, (victim, shared_id)), "report.pdf");
|
||||
assert_eq!(attachment_default_name(&messages, (attacker, shared_id)), "evil.sh");
|
||||
// Unknown line → safe fallback.
|
||||
let unknown = SecretKey::generate().public();
|
||||
assert_eq!(attachment_default_name(&messages, (unknown, shared_id)), "download");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reset_room_state_clears_all_room_scoped_state() {
|
||||
let mut state = AppState::default();
|
||||
@@ -5825,12 +6043,13 @@ mod tests {
|
||||
attachment: None,
|
||||
});
|
||||
state.chat_input = "draft".to_string();
|
||||
state.attachment_data.insert(attachment_id, AttachmentState::Ready(vec![1]));
|
||||
state.image_handle_cache.insert(
|
||||
attachment_id,
|
||||
iced::widget::image::Handle::from_bytes(vec![1]),
|
||||
let att_key = (peer, attachment_id);
|
||||
state.attachments.insert(
|
||||
att_key,
|
||||
AttachmentState::Ready(vec![1]),
|
||||
Some(iced::widget::image::Handle::from_bytes(vec![1])),
|
||||
);
|
||||
state.pending_saves.insert(attachment_id);
|
||||
state.pending_saves.insert(att_key);
|
||||
state.pending_plays.insert(attachment_id);
|
||||
state.invalid_audio.insert(attachment_id);
|
||||
state.connecting.insert(peer);
|
||||
@@ -5849,8 +6068,7 @@ mod tests {
|
||||
assert!(state.locally_muted.is_empty());
|
||||
assert!(state.chat_messages.is_empty());
|
||||
assert!(state.chat_input.is_empty());
|
||||
assert!(state.attachment_data.is_empty());
|
||||
assert!(state.image_handle_cache.is_empty());
|
||||
assert!(state.attachments.len() == 0);
|
||||
assert!(state.pending_saves.is_empty());
|
||||
assert!(state.pending_plays.is_empty());
|
||||
assert!(state.invalid_audio.is_empty());
|
||||
|
||||
+119
@@ -185,10 +185,129 @@ pub fn initials(name: &str) -> String {
|
||||
}
|
||||
}
|
||||
|
||||
/// A small content-addressed LRU cache mapping image bytes to a built value
|
||||
/// (e.g. an `iced` image handle), so the SAME value is reused across redraws
|
||||
/// instead of rebuilt every frame. Two Tier C F-03 properties beyond a plain
|
||||
/// hash map:
|
||||
///
|
||||
/// 1. **Bounded** — at most `cap` entries, evicting the least-recently-used on
|
||||
/// overflow, so a peer can't grow the cache without limit by publishing an
|
||||
/// endless stream of distinct valid avatars.
|
||||
/// 2. **Collision-safe** — a hit requires full byte equality, not just a matching
|
||||
/// 64-bit hash, so a hash collision can never return a different image's value.
|
||||
///
|
||||
/// Linear scan; intended for small `cap` (tens of entries).
|
||||
pub struct ByteLru<V> {
|
||||
cap: usize,
|
||||
/// `(content hash, content bytes, value)`; back = most recently used.
|
||||
entries: Vec<(u64, Vec<u8>, V)>,
|
||||
}
|
||||
|
||||
impl<V: Clone> ByteLru<V> {
|
||||
/// Create an LRU holding at most `cap` entries (`cap` is clamped to >= 1).
|
||||
pub fn new(cap: usize) -> Self {
|
||||
Self { cap: cap.max(1), entries: Vec::new() }
|
||||
}
|
||||
|
||||
/// Return the cached value for these exact `bytes`, building and inserting it
|
||||
/// on a miss (evicting the least-recently-used entry once over `cap`). A hit
|
||||
/// verifies full byte equality, so a 64-bit hash collision never returns the
|
||||
/// wrong value. A hit also refreshes the entry's recency.
|
||||
pub fn get_or_insert(&mut self, bytes: &[u8], build: impl FnOnce() -> V) -> V {
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
bytes.hash(&mut hasher);
|
||||
self.get_or_insert_hashed(hasher.finish(), bytes, build)
|
||||
}
|
||||
|
||||
/// Inner seam with the content `hash` supplied explicitly. Production callers
|
||||
/// use [`get_or_insert`]; tests use this to force a hash collision (different
|
||||
/// bytes, same hash) and exercise the byte-equality guard.
|
||||
fn get_or_insert_hashed(&mut self, hash: u64, bytes: &[u8], build: impl FnOnce() -> V) -> V {
|
||||
if let Some(idx) = self
|
||||
.entries
|
||||
.iter()
|
||||
.position(|(h, b, _)| *h == hash && b.as_slice() == bytes)
|
||||
{
|
||||
// LRU touch: move the hit entry to the back (most recent).
|
||||
let entry = self.entries.remove(idx);
|
||||
let val = entry.2.clone();
|
||||
self.entries.push(entry);
|
||||
return val;
|
||||
}
|
||||
|
||||
let val = build();
|
||||
if self.entries.len() >= self.cap {
|
||||
self.entries.remove(0); // evict least-recently-used
|
||||
}
|
||||
self.entries.push((hash, bytes.to_vec(), val.clone()));
|
||||
val
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn len(&self) -> usize {
|
||||
self.entries.len()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn byte_lru_reuses_value_for_identical_bytes() {
|
||||
let mut lru: ByteLru<u32> = ByteLru::new(4);
|
||||
let mut next = 0u32;
|
||||
let mut build = |lru: &mut ByteLru<u32>, b: &[u8]| {
|
||||
lru.get_or_insert(b, || {
|
||||
next += 1;
|
||||
next
|
||||
})
|
||||
};
|
||||
// Same bytes → same value, built only once.
|
||||
assert_eq!(build(&mut lru, b"alice"), 1);
|
||||
assert_eq!(build(&mut lru, b"alice"), 1);
|
||||
// Different bytes → a freshly built value.
|
||||
assert_eq!(build(&mut lru, b"bob"), 2);
|
||||
assert_eq!(lru.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn byte_lru_evicts_least_recently_used() {
|
||||
let mut lru: ByteLru<u32> = ByteLru::new(2);
|
||||
let mut n = 0u32;
|
||||
let mut ins = |lru: &mut ByteLru<u32>, b: &[u8]| {
|
||||
lru.get_or_insert(b, || {
|
||||
n += 1;
|
||||
n
|
||||
})
|
||||
};
|
||||
ins(&mut lru, b"a"); // -> 1
|
||||
ins(&mut lru, b"b"); // -> 2, cache = [a, b]
|
||||
ins(&mut lru, b"a"); // touch a, cache = [b, a]
|
||||
ins(&mut lru, b"c"); // evicts LRU (b), cache = [a, c]
|
||||
assert_eq!(lru.len(), 2);
|
||||
// `a` survived (recently touched) → still value 1, not rebuilt.
|
||||
assert_eq!(ins(&mut lru, b"a"), 1);
|
||||
// `b` was evicted → rebuilt with a new value.
|
||||
assert_eq!(ins(&mut lru, b"b"), 4);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn byte_lru_byte_equality_survives_a_hash_collision() {
|
||||
// Force the SAME 64-bit hash for two DIFFERENT byte strings (the case a
|
||||
// bare-hash cache would alias — Tier C F-03 collision bug).
|
||||
let mut lru: ByteLru<u32> = ByteLru::new(4);
|
||||
assert_eq!(lru.get_or_insert_hashed(42, b"alice", || 1), 1);
|
||||
// `bob` collides on the hash but differs in bytes → a MISS, built fresh,
|
||||
// NOT aliased to alice's value.
|
||||
assert_eq!(lru.get_or_insert_hashed(42, b"bob", || 2), 2);
|
||||
// Both coexist; each re-lookup returns its own value (build closure unused).
|
||||
assert_eq!(lru.get_or_insert_hashed(42, b"alice", || 99), 1);
|
||||
assert_eq!(lru.get_or_insert_hashed(42, b"bob", || 99), 2);
|
||||
assert_eq!(lru.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn initials_takes_first_two_words() {
|
||||
assert_eq!(initials("Alice"), "A");
|
||||
|
||||
@@ -133,11 +133,13 @@ pub enum UiEvent {
|
||||
/// string, used to key their avatar (W4).
|
||||
ChatMessage { from: String, name: String, text: String, attachment: Option<crate::files::ChatAttachment> },
|
||||
/// An attachment's bytes are now available (auto-fetched for images, or
|
||||
/// fetched on demand for files). Keyed by attachment id so the UI can match
|
||||
/// it to the chat entry.
|
||||
AttachmentReady { id: crate::files::AttachmentId, data: Vec<u8> },
|
||||
/// fetched on demand for files). Keyed by `(from, id)`: the id is
|
||||
/// attacker-chosen, so a malicious peer can reuse a victim's id — the author
|
||||
/// disambiguates whose bytes these are and stops content aliasing (Tier C
|
||||
/// F-12).
|
||||
AttachmentReady { from: EndpointId, id: crate::files::AttachmentId, data: Vec<u8> },
|
||||
/// An attachment fetch failed (sender gone, too large, decode error, etc.).
|
||||
AttachmentFailed { id: crate::files::AttachmentId, error: String },
|
||||
AttachmentFailed { from: EndpointId, id: crate::files::AttachmentId, error: String },
|
||||
/// Our own screen share started; the UI flips the Share button to "Stop".
|
||||
ScreenShareStarted,
|
||||
/// Our own screen share stopped (or failed to start).
|
||||
|
||||
+213
-32
@@ -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();
|
||||
@@ -744,25 +765,70 @@ async fn build_net_stack(
|
||||
})
|
||||
}
|
||||
|
||||
/// Maximum number of *automatic* chat-attachment fetches in flight at once.
|
||||
///
|
||||
/// Auto-fetch (inline image preview) is triggered by an untrusted peer's chat
|
||||
/// message, and each fetch is a detached task that can spend up to ~60s dialing
|
||||
/// and reading. Without a bound, a room insider could spam attachment-carrying
|
||||
/// chat to accumulate arbitrary pending tasks/dials (Tier C F-02). When the bound
|
||||
/// is reached we simply skip the auto-fetch; the descriptor still renders and the
|
||||
/// user can fetch it on demand (which is not rate-limited here).
|
||||
const MAX_INFLIGHT_ATTACHMENT_FETCHES: usize = 4;
|
||||
|
||||
/// In-flight `(author, attachment_id)` markers for bounded, deduplicated auto-
|
||||
/// fetches (Tier C F-02). Bounded by [`MAX_INFLIGHT_ATTACHMENT_FETCHES`].
|
||||
type InflightAttachments =
|
||||
Arc<std::sync::Mutex<HashSet<(EndpointId, crate::files::AttachmentId)>>>;
|
||||
|
||||
/// RAII bookkeeping for one bounded auto-fetch: holds the concurrency permit for
|
||||
/// the task's lifetime and clears the in-flight `(author, id)` marker when the
|
||||
/// fetch finishes (success OR failure), so the same image can be retried later.
|
||||
struct AutoFetchGuard {
|
||||
_permit: tokio::sync::OwnedSemaphorePermit,
|
||||
inflight: InflightAttachments,
|
||||
key: (EndpointId, crate::files::AttachmentId),
|
||||
}
|
||||
|
||||
impl Drop for AutoFetchGuard {
|
||||
fn drop(&mut self) {
|
||||
self.inflight.lock().unwrap().remove(&self.key);
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether to AUTO-fetch a chat image attachment. Only authenticated roster
|
||||
/// authors qualify (closing the non-roster injection vector), and a `(author,
|
||||
/// id)` already being fetched is skipped (dedup). The concurrency bound itself is
|
||||
/// enforced separately by the permit. Pure → unit-testable (Tier C F-02).
|
||||
fn should_auto_fetch(is_image: bool, author_in_roster: bool, already_inflight: bool) -> bool {
|
||||
is_image && author_in_roster && !already_inflight
|
||||
}
|
||||
|
||||
/// Fetch a chat attachment's bytes from `from` over the file plane in a detached
|
||||
/// task, then report the result to the UI via [`UiEvent::AttachmentReady`] /
|
||||
/// [`UiEvent::AttachmentFailed`] keyed by the attachment id. For images
|
||||
/// [`UiEvent::AttachmentFailed`], tagged with `from` so the UI keys the bytes by
|
||||
/// `(author, id)` and can't alias a same-id attachment from another sender. For images
|
||||
/// (`is_image`) the bytes are defensively re-validated (decodable + within pixel
|
||||
/// limits) before being handed to the renderer; an "Image" that doesn't decode is
|
||||
/// reported as a failure rather than rendered.
|
||||
/// reported as a failure rather than rendered. `guard` is `Some` for bounded
|
||||
/// auto-fetches and `None` for user-initiated fetches; it is dropped when the
|
||||
/// task ends, releasing the concurrency permit and the dedup marker.
|
||||
fn spawn_attachment_fetch(
|
||||
transport: Arc<IrohTransport>,
|
||||
ui_tx: mpsc::Sender<UiEvent>,
|
||||
from: EndpointId,
|
||||
att: crate::files::ChatAttachment,
|
||||
is_image: bool,
|
||||
guard: Option<AutoFetchGuard>,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
// Held for the whole fetch; dropped here on completion (Tier C F-02).
|
||||
let _guard = guard;
|
||||
match transport.fetch_attachment(from, &att).await {
|
||||
Ok(data) => {
|
||||
if is_image && crate::files::validate_image_bytes(&data).is_none() {
|
||||
let _ = ui_tx
|
||||
.send(UiEvent::AttachmentFailed {
|
||||
from,
|
||||
id: att.id,
|
||||
error: "received image failed to decode".to_string(),
|
||||
})
|
||||
@@ -770,12 +836,12 @@ fn spawn_attachment_fetch(
|
||||
return;
|
||||
}
|
||||
let _ = ui_tx
|
||||
.send(UiEvent::AttachmentReady { id: att.id, data })
|
||||
.send(UiEvent::AttachmentReady { from, id: att.id, data })
|
||||
.await;
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = ui_tx
|
||||
.send(UiEvent::AttachmentFailed { id: att.id, error: e.to_string() })
|
||||
.send(UiEvent::AttachmentFailed { from, id: att.id, error: e.to_string() })
|
||||
.await;
|
||||
}
|
||||
}
|
||||
@@ -1799,7 +1865,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,
|
||||
@@ -1808,17 +1874,51 @@ 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.
|
||||
let friends_events = friends.clone();
|
||||
let friends_read_only_events = friends_read_only;
|
||||
// Bounded, deduplicated auto-fetch of chat image attachments (Tier C
|
||||
// F-02): the permit pool caps concurrent fetch tasks; the in-flight
|
||||
// set dedups identical (author, id) pairs.
|
||||
let attachment_limiter =
|
||||
Arc::new(tokio::sync::Semaphore::new(MAX_INFLIGHT_ATTACHMENT_FETCHES));
|
||||
let inflight_attachments: InflightAttachments =
|
||||
Arc::new(std::sync::Mutex::new(HashSet::new()));
|
||||
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();
|
||||
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);
|
||||
cancel_grace_timer(&grace_timers_events, &peer_id);
|
||||
recovery_events.cancel(peer_id);
|
||||
transport_events.admit_audio_sender(peer_id);
|
||||
@@ -1843,13 +1943,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)
|
||||
@@ -1862,6 +1971,7 @@ async fn run_core_loop(
|
||||
}
|
||||
RoomEvent::PeerLeft(peer_id) => {
|
||||
// Graceful leave — evict immediately.
|
||||
roster.remove(&peer_id);
|
||||
cancel_grace_timer(&grace_timers_events, &peer_id);
|
||||
seen_connected_events.lock().unwrap().remove(&peer_id);
|
||||
// A signed Leave cancels background recovery and
|
||||
@@ -1899,13 +2009,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 } => {
|
||||
@@ -1913,16 +2031,47 @@ async fn run_core_loop(
|
||||
// without a click; non-image files wait for an explicit
|
||||
// FetchAttachment (the "Save" chip). The descriptor was
|
||||
// already filename-sanitized + size-capped on ingest.
|
||||
if let Some(att) = attachment.clone()
|
||||
&& att.kind == crate::files::AttachmentKind::Image
|
||||
{
|
||||
spawn_attachment_fetch(
|
||||
transport_events.clone(),
|
||||
ui_tx_events.clone(),
|
||||
from,
|
||||
att,
|
||||
true,
|
||||
);
|
||||
//
|
||||
// The auto path is an untrusted-peer-triggered detached
|
||||
// task, so it is gated (Tier C F-02): only roster authors
|
||||
// qualify, identical (author,id) pairs are deduped, and a
|
||||
// permit pool caps concurrent fetch tasks. The chat TEXT
|
||||
// is always forwarded (it's sanitized at the UI edge);
|
||||
// only the fetch is bounded.
|
||||
if let Some(att) = attachment.clone() {
|
||||
let is_image = att.kind == crate::files::AttachmentKind::Image;
|
||||
let key = (from, att.id);
|
||||
let already_inflight =
|
||||
inflight_attachments.lock().unwrap().contains(&key);
|
||||
if should_auto_fetch(is_image, roster.contains(&from), already_inflight) {
|
||||
// Reserve the dedup slot, then a permit. If the
|
||||
// pool is exhausted, drop the auto-fetch (and the
|
||||
// dedup marker) — the descriptor still shows and
|
||||
// the user can fetch on demand.
|
||||
inflight_attachments.lock().unwrap().insert(key);
|
||||
match attachment_limiter.clone().try_acquire_owned() {
|
||||
Ok(permit) => {
|
||||
spawn_attachment_fetch(
|
||||
transport_events.clone(),
|
||||
ui_tx_events.clone(),
|
||||
from,
|
||||
att,
|
||||
true,
|
||||
Some(AutoFetchGuard {
|
||||
_permit: permit,
|
||||
inflight: inflight_attachments.clone(),
|
||||
key,
|
||||
}),
|
||||
);
|
||||
}
|
||||
Err(_) => {
|
||||
inflight_attachments.lock().unwrap().remove(&key);
|
||||
crate::log_msg(
|
||||
"Chat attachment auto-fetch limit reached; skipping (fetch on demand)",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = ui_tx_events.send(UiEvent::ChatMessage {
|
||||
from: from.to_string(),
|
||||
@@ -1992,6 +2141,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")]
|
||||
@@ -2467,12 +2617,15 @@ async fn run_core_loop(
|
||||
CoreCommand::FetchAttachment { from, attachment } => {
|
||||
if let Some(session) = &active_session {
|
||||
let is_image = attachment.kind == crate::files::AttachmentKind::Image;
|
||||
// User-initiated (the "Save" chip): not bounded here — a human
|
||||
// click rate-limits it. The auto path (F-02) passes a guard.
|
||||
spawn_attachment_fetch(
|
||||
session.transport.clone(),
|
||||
ui_tx.clone(),
|
||||
from,
|
||||
attachment,
|
||||
is_image,
|
||||
None,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -2575,11 +2728,39 @@ 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, 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.
|
||||
assert!(should_auto_fetch(true, true, false));
|
||||
// A non-image (generic file) never auto-fetches — it waits for "Save".
|
||||
assert!(!should_auto_fetch(false, true, false));
|
||||
// A non-roster author (e.g. a sock puppet that never announced) is rejected,
|
||||
// closing the F-02 unbounded-task vector.
|
||||
assert!(!should_auto_fetch(true, false, false));
|
||||
// An identical (author,id) already being fetched is deduped.
|
||||
assert!(!should_auto_fetch(true, true, true));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retained_peers_are_shared_by_topic_across_restamped_tickets() {
|
||||
let topic_id = [23u8; 32];
|
||||
|
||||
+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);
|
||||
|
||||
|
||||
+225
-9
@@ -1,11 +1,11 @@
|
||||
use crate::network::{RoomState, NetError, PeerState, RoomEvent, PeerSpeakTicket};
|
||||
use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, Signature};
|
||||
use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, Signature, TransportAddr};
|
||||
use iroh_gossip::net::Gossip;
|
||||
use iroh_gossip::proto::TopicId;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::sync::mpsc::Receiver;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::collections::{BTreeSet, HashMap, HashSet};
|
||||
use async_trait::async_trait;
|
||||
use tokio_stream::StreamExt;
|
||||
use serde::{Serialize, Deserialize};
|
||||
@@ -131,6 +131,89 @@ fn admit_state_mutation(
|
||||
true
|
||||
}
|
||||
|
||||
/// Size at which we prune stale entries from the replay-tracking map (Tier C
|
||||
/// F-01 audit). `admit_state_mutation` records `(author, kind)` for every signed
|
||||
/// mutation, so an insider sending validly signed `Leave`s from unlimited
|
||||
/// generated keys would otherwise grow it for the room's lifetime. A mutation
|
||||
/// older than the freshness window can never be the deciding `last_ts` for an
|
||||
/// in-window message — `verify_gossip`'s timestamp check rejects such a replay
|
||||
/// first — so dropping those entries cannot weaken replay protection; it bounds
|
||||
/// the map to roughly the authors seen within one freshness window.
|
||||
const STATE_MUTATIONS_SOFT_CAP: usize = 256;
|
||||
|
||||
/// Drop replay-tracking entries whose timestamp is older than `window_ms` before
|
||||
/// `now_ms` (see [`STATE_MUTATIONS_SOFT_CAP`]). Pure → unit-testable.
|
||||
fn prune_stale_mutations(
|
||||
seen: &mut HashMap<(EndpointId, StateMutationKind), u64>,
|
||||
now_ms: u64,
|
||||
window_ms: u64,
|
||||
) {
|
||||
let floor = now_ms.saturating_sub(window_ms);
|
||||
seen.retain(|_, last_ts| *last_ts >= floor);
|
||||
}
|
||||
|
||||
/// 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
|
||||
/// proves ownership of the generated keypair it was made with, not that the
|
||||
/// author is a distinct human. A malicious member can therefore mint many valid
|
||||
/// signed identities. Voice is full-mesh (each peer dials every other), so a real
|
||||
/// room is realistically well under this bound; the cap exists purely so a flood
|
||||
/// of sock-puppet `Announce`s can't grow our peer map / audio supervisors / dials
|
||||
/// without limit (Tier C F-01).
|
||||
const MAX_ACTIVE_PEERS: usize = 32;
|
||||
|
||||
/// Maximum transport addresses we retain from a single peer announce. iroh
|
||||
/// normally advertises a handful (a few LAN/WAN IP candidates plus one home
|
||||
/// relay); the cap stops an insider stuffing a large unique address set into each
|
||||
/// announce to inflate the address lookup and the dialer's candidate list.
|
||||
const MAX_PEER_ADDRS: usize = 8;
|
||||
|
||||
/// Maximum byte length of a relay URL we accept inside a peer address. A relay
|
||||
/// URL is normal-length; anything longer is dropped rather than retained.
|
||||
const MAX_RELAY_URL_LEN: usize = 256;
|
||||
|
||||
/// Bound an untrusted peer's advertised address set before we retain it / hand it
|
||||
/// to the address lookup and dialer (Tier C F-01). Drops transport kinds we never
|
||||
/// use (`Custom`) and over-long relay URLs, then truncates to at most
|
||||
/// [`MAX_PEER_ADDRS`] addresses. `BTreeSet` iteration is deterministic, so the
|
||||
/// kept subset is stable. Pure → unit-testable.
|
||||
fn sanitize_endpoint_addr(addr: &EndpointAddr) -> EndpointAddr {
|
||||
let addrs: BTreeSet<TransportAddr> = addr
|
||||
.addrs
|
||||
.iter()
|
||||
.filter(|a| match a {
|
||||
TransportAddr::Relay(url) => url.as_str().len() <= MAX_RELAY_URL_LEN,
|
||||
TransportAddr::Ip(_) => true,
|
||||
// `TransportAddr` is #[non_exhaustive]; we only speak IP + relay, so
|
||||
// anything else (Custom / future kinds) is dropped, not retained.
|
||||
_ => false,
|
||||
})
|
||||
.take(MAX_PEER_ADDRS)
|
||||
.cloned()
|
||||
.collect();
|
||||
EndpointAddr { id: addr.id, addrs }
|
||||
}
|
||||
|
||||
/// Whether an `Announce` may enter the roster. Only a brand-new author
|
||||
/// (`subject_to_cap`) is gated by [`MAX_ACTIVE_PEERS`]; updates to an
|
||||
/// already-present peer AND re-announces from a peer mid-reconnect (which
|
||||
/// already held a slot) always pass — exempting reconnects keeps a full room
|
||||
/// from rejecting a legitimately reconnecting member and orphaning its recovery
|
||||
/// state (Tier C F-01 audit). Pure → unit-testable.
|
||||
fn admit_into_roster(roster_len: usize, subject_to_cap: bool, max_peers: usize) -> bool {
|
||||
!subject_to_cap || roster_len < max_peers
|
||||
}
|
||||
|
||||
/// Whether a received `Announce`'s author is gated by the roster cap. A peer
|
||||
/// already in the roster (`is_new == false`, an ordinary update) or one
|
||||
/// mid-reconnect (`is_reconnecting`, it already held a slot) is exempt; only a
|
||||
/// brand-new author counts against [`MAX_ACTIVE_PEERS`] (Tier C F-01 audit).
|
||||
/// Pure → unit-testable.
|
||||
fn announce_subject_to_cap(is_new: bool, is_reconnecting: bool) -> bool {
|
||||
is_new && !is_reconnecting
|
||||
}
|
||||
|
||||
fn peer_state_for_log(state: &PeerState) -> String {
|
||||
format!(
|
||||
"name={:?}, muted={}, addr_id={}, addrs={}, sharing={}, game={:?}",
|
||||
@@ -390,6 +473,18 @@ impl RoomState for IrohGossipState {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Keep the replay-tracking map bounded: prune entries
|
||||
// older than the freshness window once it grows past the
|
||||
// soft cap (Tier C F-01 audit). Stale entries can't gate
|
||||
// an in-window message, so this never weakens replay
|
||||
// protection.
|
||||
if state_mutations_seen.len() > STATE_MUTATIONS_SOFT_CAP {
|
||||
prune_stale_mutations(
|
||||
&mut state_mutations_seen,
|
||||
now_millis(),
|
||||
GOSSIP_FRESHNESS_MS,
|
||||
);
|
||||
}
|
||||
if !admit_state_mutation(
|
||||
&mut state_mutations_seen,
|
||||
payload.author,
|
||||
@@ -436,16 +531,49 @@ impl RoomState for IrohGossipState {
|
||||
let cleaned = crate::sanitize::sanitize_game_label(&g);
|
||||
(!cleaned.is_empty()).then_some(cleaned)
|
||||
});
|
||||
disconnected_peers.lock().unwrap().remove(&payload.author);
|
||||
let (is_new, state_changed) = {
|
||||
// Bound an insider's advertised address set
|
||||
// before we retain it / hand it to the dialer
|
||||
// (Tier C F-01).
|
||||
state.addr = sanitize_endpoint_addr(&state.addr);
|
||||
// A peer reconnecting from a transient drop sits
|
||||
// in `disconnected_peers` (not the live roster);
|
||||
// it already held a slot, so it must be re-admitted
|
||||
// regardless of the cap, and its disconnect marker
|
||||
// cleared ONLY once re-admitted — clearing it before
|
||||
// a possible reject would orphan its recovery state
|
||||
// (Tier C F-01 audit).
|
||||
let is_reconnecting =
|
||||
disconnected_peers.lock().unwrap().contains(&payload.author);
|
||||
let admitted = {
|
||||
let mut peer_map = peers.lock().unwrap();
|
||||
let is_new = !peer_map.contains_key(&payload.author);
|
||||
let state_changed = peer_map.get(&payload.author) != Some(&state);
|
||||
if is_new || state_changed {
|
||||
peer_map.insert(payload.author, state.clone());
|
||||
// Cap the roster so a flood of signed
|
||||
// sock-puppet identities can't grow our
|
||||
// memory/tasks/dials without bound (Tier C
|
||||
// F-01). Existing-peer updates and reconnects
|
||||
// are exempt; only brand-new authors are gated.
|
||||
let subject_to_cap = announce_subject_to_cap(is_new, is_reconnecting);
|
||||
if !admit_into_roster(peer_map.len(), subject_to_cap, MAX_ACTIVE_PEERS) {
|
||||
None
|
||||
} else {
|
||||
let state_changed = peer_map.get(&payload.author) != Some(&state);
|
||||
if is_new || state_changed {
|
||||
peer_map.insert(payload.author, state.clone());
|
||||
}
|
||||
Some((is_new, state_changed))
|
||||
}
|
||||
(is_new, state_changed)
|
||||
};
|
||||
let Some((is_new, state_changed)) = admitted else {
|
||||
crate::log_msg(&format!(
|
||||
"Gossip roster full ({MAX_ACTIVE_PEERS}); rejecting new peer {}",
|
||||
crate::short_id(&payload.author.to_string())
|
||||
));
|
||||
continue;
|
||||
};
|
||||
// Admitted — now it is safe to clear any reconnect
|
||||
// marker (a rejected announce above leaves it intact
|
||||
// so a later signed Leave still cleans up).
|
||||
disconnected_peers.lock().unwrap().remove(&payload.author);
|
||||
|
||||
if is_new {
|
||||
crate::log_msg(&format!(
|
||||
@@ -453,7 +581,12 @@ impl RoomState for IrohGossipState {
|
||||
crate::short_id(&payload.author.to_string()),
|
||||
peer_state_for_log(&state)
|
||||
));
|
||||
address_lookup.add_endpoint_info(state.addr.clone());
|
||||
// Replace (not union) the lookup's record for
|
||||
// this id with the authenticated, sanitized
|
||||
// address set, so leave/re-announce cycles
|
||||
// can't accumulate attacker-supplied history
|
||||
// (Tier C F-01).
|
||||
let _ = address_lookup.set_endpoint_info(state.addr.clone());
|
||||
let _ = event_tx.send(RoomEvent::PeerJoined(payload.author, state)).await;
|
||||
} else if state_changed {
|
||||
crate::log_msg(&format!(
|
||||
@@ -466,6 +599,11 @@ impl RoomState for IrohGossipState {
|
||||
}
|
||||
GossipMessage::Leave => {
|
||||
crate::log_msg(&format!("Gossip peer leave request from author={:?}", payload.author));
|
||||
// Drop this id's address-lookup entry so cycling
|
||||
// distinct identities through Announce→Leave can't
|
||||
// grow the lookup for the room's lifetime (Tier C
|
||||
// F-01 audit). Re-announce re-populates it.
|
||||
let _ = address_lookup.remove_endpoint_info(payload.author);
|
||||
let removed = peers.lock().unwrap().remove(&payload.author).is_some();
|
||||
let was_disconnected = disconnected_peers
|
||||
.lock()
|
||||
@@ -768,6 +906,84 @@ mod tests {
|
||||
assert!(!bootstrap.contains(&me));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admit_into_roster_caps_new_authors_but_not_updates() {
|
||||
// New authors are admitted while there's room...
|
||||
assert!(admit_into_roster(0, true, 3));
|
||||
assert!(admit_into_roster(2, true, 3));
|
||||
// ...rejected once the roster is full...
|
||||
assert!(!admit_into_roster(3, true, 3));
|
||||
assert!(!admit_into_roster(10, true, 3));
|
||||
// ...but an existing peer's update always passes, even at/over the cap.
|
||||
assert!(admit_into_roster(3, false, 3));
|
||||
assert!(admit_into_roster(99, false, 3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reconnecting_and_existing_peers_are_exempt_from_the_cap() {
|
||||
// A brand-new author counts against the cap...
|
||||
assert!(announce_subject_to_cap(/* is_new */ true, /* is_reconnecting */ false));
|
||||
// ...but an ordinary update from an in-roster peer does not...
|
||||
assert!(!announce_subject_to_cap(false, false));
|
||||
// ...and neither does a re-announce from a peer mid-reconnect, even
|
||||
// though it was removed from the live roster (the F-01-audit fix: a full
|
||||
// room must not reject a legitimately reconnecting member).
|
||||
assert!(!announce_subject_to_cap(true, true));
|
||||
// Combined with admit_into_roster: a reconnecting author passes at a full
|
||||
// roster, a brand-new one does not.
|
||||
assert!(admit_into_roster(3, announce_subject_to_cap(true, true), 3));
|
||||
assert!(!admit_into_roster(3, announce_subject_to_cap(true, false), 3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prune_stale_mutations_drops_only_out_of_window_entries() {
|
||||
let a = fresh_id();
|
||||
let b = fresh_id();
|
||||
let mut seen = HashMap::new();
|
||||
seen.insert((a, StateMutationKind::Announce), 10_000u64);
|
||||
seen.insert((b, StateMutationKind::Leave), 250_000u64);
|
||||
// now = 300_000, window = 120_000 → floor 180_000. The 10_000 entry is
|
||||
// stale (and could never gate an in-window message), the 250_000 is live.
|
||||
prune_stale_mutations(&mut seen, 300_000, GOSSIP_FRESHNESS_MS);
|
||||
assert_eq!(seen.len(), 1);
|
||||
assert!(seen.contains_key(&(b, StateMutationKind::Leave)));
|
||||
assert!(!seen.contains_key(&(a, StateMutationKind::Announce)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sanitize_endpoint_addr_caps_address_count() {
|
||||
use std::net::SocketAddr;
|
||||
let id = fresh_id();
|
||||
// An insider stuffs far more addresses than MAX_PEER_ADDRS into one announce.
|
||||
let many: Vec<TransportAddr> = (0..(MAX_PEER_ADDRS as u16 + 50))
|
||||
.map(|i| TransportAddr::Ip(SocketAddr::from(([127, 0, 0, 1], 1000 + i))))
|
||||
.collect();
|
||||
let addr = EndpointAddr::from_parts(id, many);
|
||||
let out = sanitize_endpoint_addr(&addr);
|
||||
assert_eq!(out.id, id);
|
||||
assert_eq!(out.addrs.len(), MAX_PEER_ADDRS);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sanitize_endpoint_addr_drops_overlong_relay_url() {
|
||||
use std::str::FromStr;
|
||||
let id = fresh_id();
|
||||
let short = iroh::RelayUrl::from_str("https://relay.example/").unwrap();
|
||||
let long = iroh::RelayUrl::from_str(&format!(
|
||||
"https://relay.example/{}",
|
||||
"a".repeat(MAX_RELAY_URL_LEN)
|
||||
))
|
||||
.unwrap();
|
||||
assert!(long.as_str().len() > MAX_RELAY_URL_LEN);
|
||||
let addr = EndpointAddr::from_parts(
|
||||
id,
|
||||
[TransportAddr::Relay(short.clone()), TransportAddr::Relay(long)],
|
||||
);
|
||||
let out = sanitize_endpoint_addr(&addr);
|
||||
let relays: Vec<_> = out.relay_urls().cloned().collect();
|
||||
assert_eq!(relays, vec![short], "over-long relay URL must be dropped");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_gossip_message_leave_round_trip() {
|
||||
let original = GossipMessage::Leave;
|
||||
|
||||
Reference in New Issue
Block a user