From 4d6595588b6e1fa33897f1a476c6cc427d37e5ed Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Mon, 12 Oct 2026 09:12:25 +0100 Subject: [PATCH] fix: bound IRC state, read recording metadata from the file tail, cap the audit log - IRC: channel state is kept only for rooms the client itself joined, so JOIN/PART/KICK/NAMES for other rooms can't grow it; users per channel are capped at 10 000 and the view keeps at most 200 conversations, dropping the oldest idle query (SEC-20261012-014, F14). - Recordings: the list reads the header line and the last 64 KiB for the duration instead of whole files of up to 250 MiB, falling back to a full scan only when the final event is longer than the window (SEC-20261012-020, F9). - Audit log: keeps the newest 10 000 entries (F17). --- rust/app/src/irc_view.rs | 12 ++++++++ rust/core/src/clients/irc.rs | 41 +++++++++++++++++++++++++-- rust/core/src/recording.rs | 55 +++++++++++++++++++++++++++--------- rust/core/src/storage.rs | 31 +++++++++++++++++++- 4 files changed, 122 insertions(+), 17 deletions(-) diff --git a/rust/app/src/irc_view.rs b/rust/app/src/irc_view.rs index 70107ff..2541e84 100644 --- a/rust/app/src/irc_view.rs +++ b/rust/app/src/irc_view.rs @@ -16,6 +16,9 @@ use std::sync::Arc; const SERVER: &str = "server"; const MAX_LINES: usize = 2000; +/// Open conversations; past this the oldest idle query is dropped, so a flood of messages from +/// unique nicks can't grow the view without bound. +const MAX_TARGETS: usize = 200; const SIDEBAR_W: f32 = 230.0; struct Line { @@ -172,6 +175,15 @@ impl IrcView { let i = match self.targets.iter().position(|t| t.name == name) { Some(i) => i, None => { + if self.targets.len() >= MAX_TARGETS { + // ponytail: evicts the oldest-opened idle target, not least-recently-used. + let (active, joined) = (&self.active, &self.joined); + if let Some(i) = self.targets.iter().position(|t| t.name != SERVER && t.name != *active && !joined.iter().any(|r| r.name == t.name)) { + let gone = self.targets.remove(i); + self.unread.remove(&gone.name); + self.users.remove(&gone.name); + } + } self.targets.push(Target { name: name.into(), lines: VecDeque::new() }); self.targets.len() - 1 } diff --git a/rust/core/src/clients/irc.rs b/rust/core/src/clients/irc.rs index 80b7c2e..3b00ab5 100644 --- a/rust/core/src/clients/irc.rs +++ b/rust/core/src/clients/irc.rs @@ -25,6 +25,8 @@ const PING_DELAY: Duration = Duration::from_secs(20); const PING_TIMEOUT: Duration = Duration::from_secs(60); const PING_GRACE: Duration = Duration::from_secs(30); const MAX_LINE_BYTES: usize = 1 << 20; +/// Users tracked per channel; a hostile server can't grow the client past this. +const MAX_CHANNEL_USERS: usize = 10_000; /// A bidirectional byte stream. Reads must not block indefinitely: they should return /// `WouldBlock` or `TimedOut` after a short interval (tens of milliseconds) so the client's @@ -602,6 +604,12 @@ impl Client { fold(nick) == fold(&self.st().nick) } + /// Whether we're in `room`: state is only kept for channels the client itself joined, so a + /// server can't grow it with JOINs for rooms we never asked for. + fn tracks(&self, room: &str) -> bool { + self.st().channels.contains_key(&fold(room)) + } + fn handle_message(&self, m: &Message) { match m.command.as_str() { "001" => self.handle_welcome(m), @@ -734,12 +742,16 @@ impl Client { return; } let from = m.source.clone(); + let me = self.is_me(&from); + if !me && !self.tracks(&room) { + return; + } if !from.is_empty() { let mut st = self.st(); let ch = st.channels.entry(fold(&room)).or_insert_with(|| Channel { name: room.clone(), topic: String::new(), users: Vec::new() }); add_user(ch, &from); } - if self.is_me(&from) { + if me { self.remember_room_membership(&room, true); } self.emit(Event { kind: "join".into(), target: room.clone(), from: from.clone(), text: format!("{from} joined {room}"), ..Default::default() }); @@ -754,6 +766,9 @@ impl Client { } let from = m.source.clone(); let me = self.is_me(&from); + if !self.tracks(&room) { + return; + } if !from.is_empty() { let mut st = self.st(); if me { @@ -777,6 +792,9 @@ impl Client { } let (room, kicked) = (m.param(0).to_string(), m.param(1).to_string()); let me = self.is_me(&kicked); + if !self.tracks(&room) { + return; + } { let mut st = self.st(); if me { @@ -846,7 +864,7 @@ impl Client { } fn handle_names(&self, m: &Message) { - if m.params.len() < 4 { + if m.params.len() < 4 || !self.tracks(m.param(2)) { return; } { @@ -943,7 +961,7 @@ fn fold(s: &str) -> String { } fn add_user(ch: &mut Channel, nick: &str) { - if !ch.users.iter().any(|u| fold(u) == fold(nick)) { + if ch.users.len() < MAX_CHANNEL_USERS && !ch.users.iter().any(|u| fold(u) == fold(nick)) { ch.users.push(nick.to_string()); } } @@ -1048,6 +1066,23 @@ mod tests { assert_eq!(client.st().room_names, vec!["#two"]); } + #[test] + fn only_joined_channels_are_tracked_and_users_are_capped() { + let client = Client::new(Config::default()); + client.st().nick = "me".into(); + let feed = |l: &str| client.handle_message(&parse_line(l).unwrap()); + for n in 0..20_000 { + feed(&format!(":x{n}!u@h JOIN #room-{n}")); + } + assert!(client.st().channels.is_empty(), "JOINs by others for rooms we're not in are ignored"); + feed(":me!u@h JOIN #ours"); + feed(":other!u@h JOIN #ours"); + assert_eq!(client.st().channels[&fold("#ours")].users, vec!["me", "other"]); + let mut ch = Channel { name: "#big".into(), topic: String::new(), users: (0..MAX_CHANNEL_USERS).map(|n| format!("u{n}")).collect() }; + add_user(&mut ch, "one-too-many"); + assert_eq!(ch.users.len(), MAX_CHANNEL_USERS); + } + #[test] fn parses_and_encodes_lines() { let m = parse_line("@time=x :nick!u@h PRIVMSG #chan :hello there").unwrap(); diff --git a/rust/core/src/recording.rs b/rust/core/src/recording.rs index 5d2180b..9c3fe03 100644 --- a/rust/core/src/recording.rs +++ b/rust/core/src/recording.rs @@ -7,7 +7,7 @@ use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fs::File; -use std::io::{Read, Write}; +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; use std::time::Instant; @@ -328,22 +328,32 @@ impl Store { Ok(self.dir.join(format!("{id}.cast"))) } - /// Header fields plus the last event's timestamp as the duration. + /// Header fields plus the last event's timestamp as the duration. Reads the header line and + /// the file's tail rather than the whole recording (they can be hundreds of MiB). fn read_meta(&self, filename: &str) -> Result { + const TAIL: u64 = 64 << 10; let path = self.dir.join(filename); - let mut f = File::open(&path)?; - let size = f.metadata()?.len(); - let mut raw = Vec::new(); - f.read_to_end(&mut raw)?; - let mut values = serde_json::Deserializer::from_slice(&raw).into_iter::(); - let hdr: Header = serde_json::from_value(values.next().ok_or("recording: empty file")??)?; - let mut duration = 0.0; - for v in values { - let Ok(v) = v else { break }; - if let Some(e) = v.get(0).and_then(|e| e.as_f64()) { - duration = e; + let mut f = BufReader::new(File::open(&path)?); + let size = f.get_ref().metadata()?.len(); + let mut head = Vec::new(); + f.by_ref().take(1 << 20).read_until(b'\n', &mut head)?; + if head.iter().all(u8::is_ascii_whitespace) { + return Err("recording: empty file".into()); + } + let hdr: Header = serde_json::from_slice(&head)?; + let body = head.len() as u64; + let mut duration = None; + // The tail first; if one event is longer than the window, fall back to the whole body. + for start in [size.saturating_sub(TAIL).max(body), body] { + let mut tail = Vec::new(); + f.seek(SeekFrom::Start(start))?; + f.read_to_end(&mut tail)?; + duration = tail.split(|b| *b == b'\n').rev().find_map(|l| serde_json::from_slice::(l).ok()?.get(0)?.as_f64()); + if duration.is_some() || start == body { + break; } } + let duration = duration.unwrap_or(0.0); let id = filename.trim_end_matches(".cast").to_string(); let created_at = chrono::DateTime::from_timestamp(hdr.timestamp, 0).ok_or("recording: invalid timestamp")?.with_timezone(&chrono::Local).fixed_offset(); Ok(RecordingMeta { @@ -431,6 +441,25 @@ mod tests { std::fs::remove_dir_all(dir).unwrap(); } + #[test] + fn duration_comes_from_the_tail_even_with_long_events() { + let dir = tmp("tail"); + std::fs::create_dir_all(&dir).unwrap(); + let hdr = r#"{"version":2,"width":80,"height":24,"timestamp":1791633486}"#; + let mut many = format!("{hdr}\n"); + for i in 0..20_000 { + many.push_str(&format!("[{}.5, \"o\", \"line {i}\\r\\n\"]\n", i)); + } + std::fs::write(dir.join("many.cast"), many).unwrap(); + // The last event alone is bigger than the tail window. + std::fs::write(dir.join("long.cast"), format!("{hdr}\n[1.0, \"o\", \"a\"]\n[7.25, \"o\", \"{}\"]\n", "x".repeat(200 << 10))).unwrap(); + std::fs::write(dir.join("bare.cast"), format!("{hdr}\n")).unwrap(); + let metas = Store::new(&dir).list().unwrap(); + let d = |id: &str| metas.iter().find(|m| m.id == id).unwrap().duration; + assert_eq!((d("many"), d("long"), d("bare")), (19_999.5, 7.25, 0.0)); + std::fs::remove_dir_all(dir).unwrap(); + } + #[test] fn stops_at_size_limit() { let dir = tmp("limit"); diff --git a/rust/core/src/storage.rs b/rust/core/src/storage.rs index 27f5b14..518d076 100644 --- a/rust/core/src/storage.rs +++ b/rust/core/src/storage.rs @@ -34,6 +34,9 @@ pub fn parse_time(s: &str) -> Time { DateTime::parse_from_rfc3339(s).unwrap_or_else(|_| zero_time()) } +/// Audit-log entries kept; older ones are pruned as new ones arrive. +pub const MAX_AUDIT_ROWS: i64 = 10_000; + pub struct Store { db: Mutex, } @@ -318,11 +321,14 @@ impl Store { Ok(()) } + /// Records an action; the log keeps the newest [`MAX_AUDIT_ROWS`] entries. pub fn add_audit(&self, action: &str, profile_id: &str, session_id: &str, details: &str) { - let _ = self.db.lock().execute( + let db = self.db.lock(); + let _ = db.execute( "INSERT INTO audit_log (id, timestamp, action, profile_id, session_id, details) VALUES (?, ?, ?, ?, ?, ?)", params![new_id(), fmt_time(&now()), action, profile_id, session_id, details], ); + let _ = db.execute("DELETE FROM audit_log WHERE rowid <= (SELECT MAX(rowid) FROM audit_log) - ?", [MAX_AUDIT_ROWS]); } pub fn list_audit(&self, limit: i64) -> Result> { @@ -433,6 +439,29 @@ pub fn dedup_rooms(groups: &[&[String]]) -> Vec { mod tests { use super::*; + #[test] + fn audit_log_keeps_the_newest_entries() { + let tmp = std::env::temp_dir().join(format!("nexus-audit-{}.db", std::process::id())); + let s = Store::open(&tmp).unwrap(); + { + let mut db = s.db.lock(); + let tx = db.transaction().unwrap(); + for n in 0..MAX_AUDIT_ROWS { + tx.execute("INSERT INTO audit_log (id, timestamp, action) VALUES (?, '2020-01-01T00:00:00Z', 'old')", [n.to_string()]).unwrap(); + } + tx.commit().unwrap(); + } + for _ in 0..3 { + s.add_audit("new", "", "", ""); + } + let count = |action: &str| s.db.lock().query_row("SELECT COUNT(*) FROM audit_log WHERE action = ?", [action], |r| r.get::<_, i64>(0)).unwrap(); + assert_eq!((count("old"), count("new")), (MAX_AUDIT_ROWS - 3, 3)); + drop(s); + for ext in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{}{ext}", tmp.display())); + } + } + #[test] fn opens_go_database_and_migrates() { let fx = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/link.db");