From ac017ba3c4baaf763191fbdc47c34de34c55e7ca Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 10 Aug 2026 13:57:12 -0500 Subject: [PATCH 1/2] test(mail): add a stateful multi-connection IMAP server for offline tests The IMAP client had no mock. wiremock is HTTP-only, so JMAP and the DAV protocols have one and IMAP does not; what stood in for it was a scripted responder rewritten inline in each test, holding one connection and no mailbox state. That shape cannot express the property the sync work needs to assert: two clients observing each other's writes through shared server state. It also cannot reproduce the response forms that decide whether a mutation actually applied -- COPYUID, VANISHED, and a tagged [MODIFIED ...] -- so a client that reports success for a no-op cannot currently be caught by a test. The server binds a real loopback listener, shares mailbox state across every accepted connection, and speaks IMAP4rev1 plus the parts of RFC 7162, RFC 4315, and RFC 6851 the engine exercises. Capability advertisement is selectable, because the sync ladder branches on it and a suite that only ever runs against a QRESYNC server proves nothing about the path Gmail and Exchange take. Behind a `test-support` feature so it stays out of production builds while remaining available to dependent crates' integration suites. Covers: shared state across connections; a move of an already-moved UID returning OK with no COPYUID; UNCHANGEDSINCE reporting MODIFIED in the tagged response; QRESYNC SELECT reporting VANISHED; QRESYNC refused without ENABLE and on a CONDSTORE-only server; VANISHED versus EXPUNGE on delete. --- crates/nuncio-mail/Cargo.toml | 5 + crates/nuncio-mail/src/lib.rs | 6 + crates/nuncio-mail/src/test_server.rs | 1257 +++++++++++++++++++++++++ 3 files changed, 1268 insertions(+) create mode 100644 crates/nuncio-mail/src/test_server.rs diff --git a/crates/nuncio-mail/Cargo.toml b/crates/nuncio-mail/Cargo.toml index 1045a59..72a10cc 100644 --- a/crates/nuncio-mail/Cargo.toml +++ b/crates/nuncio-mail/Cargo.toml @@ -9,6 +9,11 @@ repository.workspace = true [lints] workspace = true +[features] +# Exposes `test_server`, a stateful loopback IMAP server, to dependent crates' +# test suites. Never enabled in a production build. +test-support = [] + [dependencies] nuncio-core = { path = "../nuncio-core" } tokio.workspace = true diff --git a/crates/nuncio-mail/src/lib.rs b/crates/nuncio-mail/src/lib.rs index cc8bfc9..4689ea8 100644 --- a/crates/nuncio-mail/src/lib.rs +++ b/crates/nuncio-mail/src/lib.rs @@ -10,6 +10,12 @@ pub mod mock; pub mod parser; pub mod smtp; +// A real loopback IMAP server for offline tests. Behind a feature so it never +// enters a production build, but visible to this crate's own tests and to +// dependents that opt in (integration suites in `nunciod`). +#[cfg(any(test, feature = "test-support"))] +pub mod test_server; + pub use backend::{ MailBackend, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec, }; diff --git a/crates/nuncio-mail/src/test_server.rs b/crates/nuncio-mail/src/test_server.rs new file mode 100644 index 0000000..bb659f3 --- /dev/null +++ b/crates/nuncio-mail/src/test_server.rs @@ -0,0 +1,1257 @@ +//! A stateful, multi-connection IMAP server for offline tests. +//! +//! `wiremock` only speaks HTTP, so JMAP and the DAV protocols have a mock and +//! IMAP does not. What existed instead was a scripted responder re-written +//! inline in each test: one connection, no mailbox state, and a hand-rolled +//! answer per command. That shape cannot express the behaviour this module +//! exists to test -- two clients observing each other's writes through shared +//! server state. +//! +//! The server binds a real loopback `TcpListener` and speaks enough of +//! IMAP4rev1 (plus RFC 7162 CONDSTORE/QRESYNC, RFC 4315 UIDPLUS, and RFC 6851 +//! MOVE) for a client to be driven end to end over plain TCP. Mailbox state is +//! shared across every accepted connection, so a mutation issued on one +//! connection is visible to the next command on another -- which is the only +//! way to reproduce, offline, what happens when several engines sync the same +//! account. +//! +//! Capability advertisement is selectable ([`ServerProfile`]) because the sync +//! ladder branches on it: Fastmail/Cyrus and Dovecot advertise QRESYNC, Gmail +//! advertises CONDSTORE without QRESYNC, and Exchange advertises neither. A +//! test that only ever runs against a QRESYNC server proves nothing about the +//! path most real accounts take. +//! +//! Deliberately not a conformant server. It implements the commands the engine +//! issues, in the response shapes the relevant RFCs require, and answers `BAD` +//! to everything else. + +use std::collections::{BTreeMap, BTreeSet}; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex, MutexGuard}; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; + +/// Which capabilities the server advertises, and therefore which rung of the +/// sync ladder a client can negotiate. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ServerProfile { + /// No CONDSTORE and no QRESYNC: full UID-set reconciliation is the only + /// option. Models Exchange. + Basic, + /// CONDSTORE but no QRESYNC: `CHANGEDSINCE` works, `SELECT (QRESYNC ...)` + /// does not. Models Gmail, which is the common case and the one a + /// QRESYNC-only test suite silently skips. + Condstore, + /// CONDSTORE and QRESYNC. Models Fastmail/Cyrus and Dovecot. + Qresync, +} + +impl ServerProfile { + /// The `CAPABILITY` atom list for this profile. `UIDPLUS` and `MOVE` are + /// advertised by all three because every profile we model supports them. + fn capability_line(self) -> String { + let mut caps = vec!["IMAP4rev1", "UIDPLUS", "MOVE", "ENABLE", "LIST-STATUS"]; + match self { + ServerProfile::Basic => {} + ServerProfile::Condstore => caps.push("CONDSTORE"), + ServerProfile::Qresync => { + caps.push("CONDSTORE"); + caps.push("QRESYNC"); + } + } + caps.join(" ") + } + + fn supports_condstore(self) -> bool { + matches!(self, ServerProfile::Condstore | ServerProfile::Qresync) + } + + fn supports_qresync(self) -> bool { + matches!(self, ServerProfile::Qresync) + } +} + +/// One message as the server holds it. +#[derive(Debug, Clone)] +pub struct MockMessage { + pub uid: u32, + pub modseq: u64, + pub flags: BTreeSet, + /// Full RFC822 octets served for `BODY[]`. + pub body: String, +} + +/// A mailbox: its UID space, its mod-sequence, and the tombstones needed to +/// answer `VANISHED (EARLIER)`. +#[derive(Debug, Clone)] +pub struct MockMailbox { + pub uid_validity: u32, + pub uid_next: u32, + pub highest_modseq: u64, + pub messages: Vec, + /// `(uid, modseq_at_removal)` for every message removed from this mailbox. + /// QRESYNC needs removal history; a client resyncing from an older modseq + /// must be told what disappeared, which is exactly the information a + /// high-water-mark sync can never recover. + pub vanished: Vec<(u32, u64)>, +} + +impl MockMailbox { + fn new(uid_validity: u32) -> Self { + Self { + uid_validity, + uid_next: 1, + highest_modseq: 1, + messages: Vec::new(), + vanished: Vec::new(), + } + } + + fn next_modseq(&mut self) -> u64 { + self.highest_modseq += 1; + self.highest_modseq + } + + fn append(&mut self, body: &str, flags: &[&str]) -> u32 { + let uid = self.uid_next; + self.uid_next += 1; + let modseq = self.next_modseq(); + self.messages.push(MockMessage { + uid, + modseq, + flags: flags.iter().map(|f| (*f).to_string()).collect(), + body: body.to_string(), + }); + uid + } + + /// Remove `uid`, recording a tombstone at a fresh modseq. + fn remove(&mut self, uid: u32) -> bool { + let Some(pos) = self.messages.iter().position(|m| m.uid == uid) else { + return false; + }; + let modseq = self.next_modseq(); + self.messages.remove(pos); + self.vanished.push((uid, modseq)); + true + } + + /// 1-based sequence number of `uid`, as untagged responses require. + fn seq_of(&self, uid: u32) -> usize { + self.messages + .iter() + .position(|m| m.uid == uid) + .map_or(0, |p| p + 1) + } +} + +/// Server state shared by every connection. +#[derive(Debug)] +pub struct ServerState { + pub profile: ServerProfile, + pub mailboxes: BTreeMap, + /// Every command line received, across all connections, in arrival order. + /// Tests assert on this to prove *which* protocol path ran -- e.g. that a + /// second sync narrowed its FETCH rather than re-fetching everything. + pub commands: Vec, + next_uid_validity: u32, +} + +impl ServerState { + fn new(profile: ServerProfile) -> Self { + Self { + profile, + mailboxes: BTreeMap::new(), + commands: Vec::new(), + next_uid_validity: 1000, + } + } +} + +/// Handle to a running mock server. Dropping it aborts the accept loop; the +/// bound port is released with it. +#[derive(Debug)] +pub struct MockImapServer { + addr: SocketAddr, + state: Arc>, + accept_task: tokio::task::JoinHandle<()>, +} + +impl Drop for MockImapServer { + fn drop(&mut self) { + self.accept_task.abort(); + } +} + +/// Lock the shared state, recovering from a poisoned mutex rather than +/// panicking: a panic in one connection task must not cascade into every other +/// assertion in the test. +fn lock_state(state: &Arc>) -> MutexGuard<'_, ServerState> { + match state.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +impl MockImapServer { + /// Bind an ephemeral loopback port and start accepting connections. + pub async fn start(profile: ServerProfile) -> std::io::Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let state = Arc::new(Mutex::new(ServerState::new(profile))); + + let accept_state = Arc::clone(&state); + let accept_task = tokio::spawn(async move { + loop { + let Ok((socket, _)) = listener.accept().await else { + return; + }; + let conn_state = Arc::clone(&accept_state); + tokio::spawn(async move { + // A dropped client mid-command is normal in these tests; + // the connection task simply ends. + let _ = handle_connection(socket, conn_state).await; + }); + } + }); + + Ok(Self { + addr, + state, + accept_task, + }) + } + + pub fn port(&self) -> u16 { + self.addr.port() + } + + pub fn host(&self) -> &'static str { + "127.0.0.1" + } + + /// Create a mailbox with a fresh UIDVALIDITY and return its name. + pub fn create_mailbox(&self, name: &str) { + let mut state = lock_state(&self.state); + let uid_validity = state.next_uid_validity; + state.next_uid_validity += 1; + state + .mailboxes + .insert(name.to_string(), MockMailbox::new(uid_validity)); + } + + /// Append a message, creating the mailbox if absent. Returns its UID. + pub fn append_message(&self, mailbox: &str, body: &str, flags: &[&str]) -> u32 { + let mut state = lock_state(&self.state); + let uid_validity = state.next_uid_validity; + let entry = state.mailboxes.entry(mailbox.to_string()); + let mailbox = entry.or_insert_with(|| { + // Only consumes a UIDVALIDITY when the mailbox is actually new. + MockMailbox::new(uid_validity) + }); + let fresh = mailbox.uid_next == 1 && mailbox.messages.is_empty(); + let uid = mailbox.append(body, flags); + if fresh { + state.next_uid_validity += 1; + } + uid + } + + /// Simulate another client removing a message -- the case a forward-only + /// high-water-mark sync can never observe. + pub fn expunge(&self, mailbox: &str, uid: u32) -> bool { + let mut state = lock_state(&self.state); + state + .mailboxes + .get_mut(mailbox) + .is_some_and(|mb| mb.remove(uid)) + } + + /// Force a UIDVALIDITY change, as a server does when a mailbox is + /// recreated. UIDs restart at 1. + pub fn bump_uid_validity(&self, mailbox: &str) { + let mut state = lock_state(&self.state); + if let Some(mb) = state.mailboxes.get_mut(mailbox) { + mb.uid_validity += 1; + mb.uid_next = 1; + mb.messages.clear(); + mb.vanished.clear(); + } + } + + /// UIDs currently present in `mailbox`. + pub fn uids(&self, mailbox: &str) -> Vec { + let state = lock_state(&self.state); + state + .mailboxes + .get(mailbox) + .map(|mb| mb.messages.iter().map(|m| m.uid).collect()) + .unwrap_or_default() + } + + /// Flags currently set on `uid`, or `None` if it is not present. + pub fn flags(&self, mailbox: &str, uid: u32) -> Option> { + let state = lock_state(&self.state); + state.mailboxes.get(mailbox).and_then(|mb| { + mb.messages + .iter() + .find(|m| m.uid == uid) + .map(|m| m.flags.iter().cloned().collect()) + }) + } + + /// Every command line the server has received, in arrival order. + pub fn commands(&self) -> Vec { + lock_state(&self.state).commands.clone() + } + + /// Commands containing `needle` (case-insensitive). + pub fn commands_matching(&self, needle: &str) -> Vec { + let upper = needle.to_ascii_uppercase(); + self.commands() + .into_iter() + .filter(|c| c.to_ascii_uppercase().contains(&upper)) + .collect() + } +} + +/// Per-connection session state. Only `selected` and `qresync_enabled` are +/// connection-scoped; everything else lives in the shared `ServerState`. +struct Session { + selected: Option, + qresync_enabled: bool, +} + +async fn handle_connection( + mut socket: TcpStream, + state: Arc>, +) -> std::io::Result<()> { + let greeting = { + let st = lock_state(&state); + format!( + "* OK [CAPABILITY {}] Nuncio mock IMAP ready\r\n", + st.profile.capability_line() + ) + }; + socket.write_all(greeting.as_bytes()).await?; + + let mut session = Session { + selected: None, + qresync_enabled: false, + }; + + while let Some(line) = read_line(&mut socket).await { + if line.trim().is_empty() { + continue; + } + let response = { + let mut st = lock_state(&state); + st.commands.push(line.clone()); + dispatch(&line, &mut session, &mut st) + }; + socket.write_all(response.text.as_bytes()).await?; + if response.close { + break; + } + } + Ok(()) +} + +struct Reply { + text: String, + close: bool, +} + +impl Reply { + fn open(text: String) -> Self { + Self { text, close: false } + } + fn closing(text: String) -> Self { + Self { text, close: true } + } +} + +fn dispatch(line: &str, session: &mut Session, st: &mut ServerState) -> Reply { + let mut parts = line.split_whitespace(); + let tag = parts.next().unwrap_or("*").to_string(); + let command = parts.next().unwrap_or("").to_ascii_uppercase(); + let rest: Vec = parts.map(str::to_string).collect(); + + match command.as_str() { + "CAPABILITY" => Reply::open(format!( + "* CAPABILITY {}\r\n{tag} OK CAPABILITY completed\r\n", + st.profile.capability_line() + )), + "LOGIN" => Reply::open(format!( + "{tag} OK [CAPABILITY {}] Logged in\r\n", + st.profile.capability_line() + )), + "ENABLE" => handle_enable(&tag, &rest, session, st), + "NOOP" => Reply::open(format!("{tag} OK NOOP completed\r\n")), + "LIST" => handle_list(&tag, st), + "STATUS" => handle_status(&tag, &rest, st), + "SELECT" | "EXAMINE" => handle_select(&tag, line, &rest, session, st), + "UID" => handle_uid(&tag, line, &rest, session, st), + "LOGOUT" => Reply::closing(format!( + "* BYE Logging out\r\n{tag} OK LOGOUT completed\r\n" + )), + _ => Reply::open(format!("{tag} BAD unsupported command\r\n")), + } +} + +fn handle_enable(tag: &str, rest: &[String], session: &mut Session, st: &ServerState) -> Reply { + let wants_qresync = rest.iter().any(|a| a.eq_ignore_ascii_case("QRESYNC")); + if wants_qresync && st.profile.supports_qresync() { + session.qresync_enabled = true; + return Reply::open(format!( + "* ENABLED QRESYNC\r\n{tag} OK ENABLE completed\r\n" + )); + } + Reply::open(format!("{tag} OK ENABLE completed\r\n")) +} + +fn handle_list(tag: &str, st: &ServerState) -> Reply { + let mut out = String::new(); + for name in st.mailboxes.keys() { + out.push_str(&format!("* LIST () \"/\" {name}\r\n")); + } + out.push_str(&format!("{tag} OK LIST completed\r\n")); + Reply::open(out) +} + +fn handle_status(tag: &str, rest: &[String], st: &ServerState) -> Reply { + let Some(name) = rest.first().map(|n| n.trim_matches('"').to_string()) else { + return Reply::open(format!("{tag} BAD STATUS needs a mailbox\r\n")); + }; + let Some(mb) = st.mailboxes.get(&name) else { + return Reply::open(format!("{tag} NO no such mailbox\r\n")); + }; + let mut items = format!( + "MESSAGES {} UIDNEXT {} UIDVALIDITY {}", + mb.messages.len(), + mb.uid_next, + mb.uid_validity + ); + if st.profile.supports_condstore() { + items.push_str(&format!(" HIGHESTMODSEQ {}", mb.highest_modseq)); + } + Reply::open(format!( + "* STATUS {name} ({items})\r\n{tag} OK STATUS completed\r\n" + )) +} + +fn handle_select( + tag: &str, + line: &str, + rest: &[String], + session: &mut Session, + st: &ServerState, +) -> Reply { + let Some(name) = rest.first().map(|n| n.trim_matches('"').to_string()) else { + return Reply::open(format!("{tag} BAD SELECT needs a mailbox\r\n")); + }; + let Some(mb) = st.mailboxes.get(&name) else { + return Reply::open(format!("{tag} NO no such mailbox\r\n")); + }; + + let upper = line.to_ascii_uppercase(); + let asked_qresync = upper.contains("QRESYNC"); + // RFC 7162 3.2.3: a server MUST reply BAD to a QRESYNC SELECT parameter + // when the client has not issued `ENABLE QRESYNC` on this connection. + // Enforced so a client that skips ENABLE fails here rather than silently + // degrading in production. + if asked_qresync && !session.qresync_enabled { + return Reply::open(format!("{tag} BAD QRESYNC not enabled\r\n")); + } + + session.selected = Some(name.clone()); + + let mut out = String::new(); + out.push_str(&format!("* {} EXISTS\r\n", mb.messages.len())); + out.push_str("* 0 RECENT\r\n"); + out.push_str("* FLAGS (\\Seen \\Flagged \\Deleted \\Draft)\r\n"); + out.push_str("* OK [PERMANENTFLAGS (\\Seen \\Flagged \\Deleted \\Draft \\*)] Limited\r\n"); + out.push_str(&format!( + "* OK [UIDVALIDITY {}] UIDs valid\r\n", + mb.uid_validity + )); + out.push_str(&format!( + "* OK [UIDNEXT {}] Predicted next\r\n", + mb.uid_next + )); + if st.profile.supports_condstore() { + out.push_str(&format!( + "* OK [HIGHESTMODSEQ {}] Highest\r\n", + mb.highest_modseq + )); + } + + if asked_qresync { + if let Some(since) = parse_qresync_modseq(line) { + let gone: Vec = mb + .vanished + .iter() + .filter(|(_, m)| *m > since) + .map(|(u, _)| u.to_string()) + .collect(); + if !gone.is_empty() { + out.push_str(&format!("* VANISHED (EARLIER) {}\r\n", gone.join(","))); + } + for msg in mb.messages.iter().filter(|m| m.modseq > since) { + out.push_str(&fetch_line(mb, msg, false)); + } + } + } + + out.push_str(&format!("{tag} OK [READ-WRITE] SELECT completed\r\n")); + Reply::open(out) +} + +/// Pull the mod-sequence out of `SELECT x (QRESYNC (uidvalidity modseq ...))`. +fn parse_qresync_modseq(line: &str) -> Option { + let upper = line.to_ascii_uppercase(); + let at = upper.find("QRESYNC")?; + let tail = &line[at + "QRESYNC".len()..]; + let nums: Vec = tail + .split(|c: char| !c.is_ascii_digit()) + .filter(|s| !s.is_empty()) + .filter_map(|s| s.parse().ok()) + .collect(); + // (uidvalidity modseq [known-uids...]) -- the second number is the modseq. + nums.get(1).copied() +} + +fn handle_uid( + tag: &str, + line: &str, + rest: &[String], + session: &Session, + st: &mut ServerState, +) -> Reply { + let sub = rest + .first() + .map(|s| s.to_ascii_uppercase()) + .unwrap_or_default(); + let Some(selected) = session.selected.clone() else { + return Reply::open(format!("{tag} BAD no mailbox selected\r\n")); + }; + match sub.as_str() { + "FETCH" => handle_uid_fetch(tag, line, rest, &selected, st), + "STORE" => handle_uid_store(tag, line, rest, &selected, st), + "COPY" => handle_uid_copy(tag, rest, &selected, st, false), + "MOVE" => handle_uid_copy(tag, rest, &selected, st, true), + "EXPUNGE" => handle_uid_expunge(tag, rest, &selected, session, st), + _ => Reply::open(format!("{tag} BAD unsupported UID subcommand\r\n")), + } +} + +/// Expand an IMAP `uid-set` (`1,3:5,10:*`) against the UIDs present. +fn expand_uid_set(spec: &str, present: &[u32], uid_next: u32) -> Vec { + let mut out = Vec::new(); + for part in spec.split(',') { + if let Some((lo, hi)) = part.split_once(':') { + let lo: u32 = lo.trim().parse().unwrap_or(0); + let hi: u32 = if hi.trim() == "*" { + uid_next.saturating_sub(1).max(lo) + } else { + hi.trim().parse().unwrap_or(0) + }; + let (lo, hi) = if lo <= hi { (lo, hi) } else { (hi, lo) }; + out.extend(present.iter().copied().filter(|u| *u >= lo && *u <= hi)); + } else if let Ok(single) = part.trim().parse::() { + // RFC 3501 6.4.8: a non-existent UID is silently ignored, never an + // error. Filtering against `present` reproduces that exactly -- + // which is the behaviour that makes an unverified mutation look + // like a success. + out.extend(present.iter().copied().filter(|u| *u == single)); + } + } + out.sort_unstable(); + out.dedup(); + out +} + +fn format_flags(flags: &BTreeSet) -> String { + flags.iter().cloned().collect::>().join(" ") +} + +fn fetch_line(mb: &MockMailbox, msg: &MockMessage, include_body: bool) -> String { + let seq = mb.seq_of(msg.uid); + let mut items = format!( + "UID {} FLAGS ({}) MODSEQ ({})", + msg.uid, + format_flags(&msg.flags), + msg.modseq + ); + if include_body { + items.push_str(&format!(" BODY[] {{{}}}\r\n{}", msg.body.len(), msg.body)); + } + format!("* {seq} FETCH ({items})\r\n") +} + +fn handle_uid_fetch( + tag: &str, + line: &str, + rest: &[String], + selected: &str, + st: &mut ServerState, +) -> Reply { + let condstore = st.profile.supports_condstore(); + let Some(mb) = st.mailboxes.get(selected) else { + return Reply::open(format!("{tag} NO mailbox vanished\r\n")); + }; + let Some(spec) = rest.get(1) else { + return Reply::open(format!("{tag} BAD UID FETCH needs a set\r\n")); + }; + + let upper = line.to_ascii_uppercase(); + let include_body = upper.contains("BODY[]") || upper.contains("RFC822"); + let changed_since = if condstore { + parse_changed_since(&upper) + } else { + None + }; + + let present: Vec = mb.messages.iter().map(|m| m.uid).collect(); + let wanted = expand_uid_set(spec, &present, mb.uid_next); + + let mut out = String::new(); + for msg in mb.messages.iter().filter(|m| wanted.contains(&m.uid)) { + if let Some(since) = changed_since { + if msg.modseq <= since { + continue; + } + } + out.push_str(&fetch_line(mb, msg, include_body)); + } + + // RFC 7162 3.2.6: with CHANGEDSINCE ... VANISHED the server reports + // removals in the same pass. This is the whole point of the mechanism -- + // it is the only way a client learns a message is gone. + if changed_since.is_some() && upper.contains("VANISHED") { + if let Some(since) = changed_since { + let gone: Vec = mb + .vanished + .iter() + .filter(|(_, m)| *m > since) + .map(|(u, _)| u.to_string()) + .collect(); + if !gone.is_empty() { + out.push_str(&format!("* VANISHED (EARLIER) {}\r\n", gone.join(","))); + } + } + } + + out.push_str(&format!("{tag} OK UID FETCH completed\r\n")); + Reply::open(out) +} + +fn parse_changed_since(upper: &str) -> Option { + let at = upper.find("CHANGEDSINCE")?; + upper[at + "CHANGEDSINCE".len()..] + .split(|c: char| !c.is_ascii_digit()) + .find(|s| !s.is_empty()) + .and_then(|s| s.parse().ok()) +} + +fn parse_unchanged_since(upper: &str) -> Option { + let at = upper.find("UNCHANGEDSINCE")?; + upper[at + "UNCHANGEDSINCE".len()..] + .split(|c: char| !c.is_ascii_digit()) + .find(|s| !s.is_empty()) + .and_then(|s| s.parse().ok()) +} + +fn handle_uid_store( + tag: &str, + line: &str, + rest: &[String], + selected: &str, + st: &mut ServerState, +) -> Reply { + let condstore = st.profile.supports_condstore(); + let Some(mb) = st.mailboxes.get_mut(selected) else { + return Reply::open(format!("{tag} NO mailbox vanished\r\n")); + }; + let Some(spec) = rest.get(1) else { + return Reply::open(format!("{tag} BAD UID STORE needs a set\r\n")); + }; + + let upper = line.to_ascii_uppercase(); + let silent = upper.contains(".SILENT"); + let removing = line.contains("-FLAGS"); + let unchanged_since = if condstore { + parse_unchanged_since(&upper) + } else { + None + }; + + let flags = parse_flag_list(line); + let present: Vec = mb.messages.iter().map(|m| m.uid).collect(); + let targets = expand_uid_set(spec, &present, mb.uid_next); + + let mut modified: Vec = Vec::new(); + let mut touched: Vec = Vec::new(); + for uid in targets { + let Some(idx) = mb.messages.iter().position(|m| m.uid == uid) else { + continue; + }; + // RFC 7162 3.1.3: a message changed since the client's mod-sequence is + // reported in MODIFIED and left untouched -- the compare-and-swap that + // makes a lost update detectable instead of silent. + if let Some(since) = unchanged_since { + if mb.messages[idx].modseq > since { + modified.push(uid); + continue; + } + } + let next = mb.next_modseq(); + if let Some(msg) = mb.messages.get_mut(idx) { + for flag in &flags { + if removing { + msg.flags.remove(flag); + } else { + msg.flags.insert(flag.clone()); + } + } + msg.modseq = next; + } + touched.push(uid); + } + + let mut out = String::new(); + if !silent { + for uid in &touched { + if let Some(msg) = mb.messages.iter().find(|m| m.uid == *uid) { + out.push_str(&fetch_line(mb, msg, false)); + } + } + } + // RFC 7162 3.1.3 puts MODIFIED in the *tagged* response -- the response + // whose code async-imap currently discards. + if modified.is_empty() { + out.push_str(&format!("{tag} OK UID STORE completed\r\n")); + } else { + let list: Vec = modified.iter().map(u32::to_string).collect(); + out.push_str(&format!( + "{tag} OK [MODIFIED {}] Conditional STORE failed\r\n", + list.join(",") + )); + } + Reply::open(out) +} + +/// Extract flag atoms from the parenthesised argument of a STORE. +fn parse_flag_list(line: &str) -> Vec { + let Some(open) = line.find('(') else { + return Vec::new(); + }; + let Some(close) = line[open..].find(')') else { + return Vec::new(); + }; + line[open + 1..open + close] + .split_whitespace() + .map(str::to_string) + .collect() +} + +fn handle_uid_copy( + tag: &str, + rest: &[String], + selected: &str, + st: &mut ServerState, + is_move: bool, +) -> Reply { + let qresync = st.profile.supports_qresync(); + let (Some(spec), Some(dest_raw)) = (rest.get(1), rest.get(2)) else { + return Reply::open(format!("{tag} BAD UID COPY needs a set and a mailbox\r\n")); + }; + let dest = dest_raw.trim_matches('"').to_string(); + if !st.mailboxes.contains_key(&dest) { + return Reply::open(format!("{tag} NO [TRYCREATE] no such destination\r\n")); + } + + let (source_uids, source_validity, bodies) = { + let Some(mb) = st.mailboxes.get(selected) else { + return Reply::open(format!("{tag} NO mailbox vanished\r\n")); + }; + let present: Vec = mb.messages.iter().map(|m| m.uid).collect(); + let uids = expand_uid_set(spec, &present, mb.uid_next); + let bodies: Vec<(u32, String, BTreeSet)> = mb + .messages + .iter() + .filter(|m| uids.contains(&m.uid)) + .map(|m| (m.uid, m.body.clone(), m.flags.clone())) + .collect(); + (uids, mb.uid_validity, bodies) + }; + + // RFC 3501 6.4.8 again: a UID set that matches nothing is a successful + // no-op. No COPYUID is emitted, and the client cannot distinguish this + // from a copy it did not observe -- which is precisely why an absent + // COPYUID must mean "unknown", never "applied" and never "conflict". + if source_uids.is_empty() { + return Reply::open(format!("{tag} OK UID {} completed\r\n", verb(is_move))); + } + + let mut dest_uids = Vec::new(); + let dest_validity = { + let Some(dst) = st.mailboxes.get_mut(&dest) else { + return Reply::open(format!("{tag} NO destination vanished\r\n")); + }; + for (_, body, flags) in &bodies { + let refs: Vec<&str> = flags.iter().map(String::as_str).collect(); + dest_uids.push(dst.append(body, &refs)); + } + dst.uid_validity + }; + + let mut out = String::new(); + // RFC 6851 4.3 advises COPYUID in an *untagged* OK for MOVE, and RFC 4315 + // 3 defines the code. A client reading only the tagged OK misses it. + out.push_str(&format!( + "* OK [COPYUID {} {} {}] Copied\r\n", + dest_validity, + join_uids(&source_uids), + join_uids(&dest_uids) + )); + + if is_move { + let Some(src) = st.mailboxes.get_mut(selected) else { + return Reply::open(format!("{tag} NO mailbox vanished\r\n")); + }; + let mut seqs = Vec::new(); + for uid in &source_uids { + seqs.push(src.seq_of(*uid)); + src.remove(*uid); + } + if qresync { + // RFC 6851 4.4: under QRESYNC the server reports VANISHED rather + // than EXPUNGE. A client proving its move by counting EXPUNGE + // responses sees zero and wrongly concludes it failed. + out.push_str(&format!("* VANISHED {}\r\n", join_uids(&source_uids))); + } else { + // Descending, so earlier removals do not renumber later ones. + seqs.sort_unstable(); + for seq in seqs.iter().rev() { + out.push_str(&format!("* {seq} EXPUNGE\r\n")); + } + } + let _ = source_validity; + } + + out.push_str(&format!("{tag} OK UID {} completed\r\n", verb(is_move))); + Reply::open(out) +} + +fn verb(is_move: bool) -> &'static str { + if is_move { + "MOVE" + } else { + "COPY" + } +} + +fn join_uids(uids: &[u32]) -> String { + uids.iter() + .map(|u| u.to_string()) + .collect::>() + .join(",") +} + +fn handle_uid_expunge( + tag: &str, + rest: &[String], + selected: &str, + session: &Session, + st: &mut ServerState, +) -> Reply { + let qresync = st.profile.supports_qresync(); + let Some(mb) = st.mailboxes.get_mut(selected) else { + return Reply::open(format!("{tag} NO mailbox vanished\r\n")); + }; + let Some(spec) = rest.get(1) else { + return Reply::open(format!("{tag} BAD UID EXPUNGE needs a set\r\n")); + }; + + let present: Vec = mb.messages.iter().map(|m| m.uid).collect(); + // UID EXPUNGE removes only \Deleted messages within the given set. + let targets: Vec = expand_uid_set(spec, &present, mb.uid_next) + .into_iter() + .filter(|uid| { + mb.messages + .iter() + .any(|m| m.uid == *uid && m.flags.iter().any(|f| f == "\\Deleted")) + }) + .collect(); + + let mut out = String::new(); + let mut seqs = Vec::new(); + for uid in &targets { + seqs.push(mb.seq_of(*uid)); + mb.remove(*uid); + } + + if qresync && session.qresync_enabled { + if !targets.is_empty() { + out.push_str(&format!("* VANISHED {}\r\n", join_uids(&targets))); + } + } else { + seqs.sort_unstable(); + for seq in seqs.iter().rev() { + out.push_str(&format!("* {seq} EXPUNGE\r\n")); + } + } + out.push_str(&format!("{tag} OK UID EXPUNGE completed\r\n")); + Reply::open(out) +} + +/// Read one CRLF-terminated command line, returning `None` at EOF. +async fn read_line(socket: &mut TcpStream) -> Option { + let mut buf = Vec::new(); + let mut byte = [0u8; 1]; + loop { + match socket.read(&mut byte).await { + Ok(0) => { + return if buf.is_empty() { + None + } else { + Some(String::from_utf8_lossy(&buf).into_owned()) + } + } + Ok(_) => { + if byte[0] == b'\n' { + return Some(String::from_utf8_lossy(&buf).into_owned()); + } + if byte[0] != b'\r' { + buf.push(byte[0]); + } + } + Err(_) => return None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::BufReader; + + /// A raw line-oriented IMAP client. The harness is tested at the wire + /// level on purpose: `async-imap` cannot express ENABLE, a QRESYNC SELECT, + /// or read a tagged response code, so an engine-driven test could not + /// observe most of what this server exists to provide. + struct RawClient { + reader: BufReader, + writer: tokio::net::tcp::OwnedWriteHalf, + counter: usize, + } + + impl RawClient { + async fn connect(server: &MockImapServer) -> Self { + let stream = TcpStream::connect((server.host(), server.port())) + .await + .expect("connect to mock server"); + let (r, w) = stream.into_split(); + let mut client = Self { + reader: BufReader::new(r), + writer: w, + counter: 0, + }; + let greeting = client.read_line().await; + assert!(greeting.starts_with("* OK"), "greeting was {greeting:?}"); + client + } + + async fn read_line(&mut self) -> String { + use tokio::io::AsyncBufReadExt; + let mut line = String::new(); + let _ = self.reader.read_line(&mut line).await; + line.trim_end().to_string() + } + + /// Send a command and collect every response line up to the tagged one. + async fn send(&mut self, command: &str) -> Vec { + self.counter += 1; + let tag = format!("a{}", self.counter); + let _ = self + .writer + .write_all(format!("{tag} {command}\r\n").as_bytes()) + .await; + let mut lines = Vec::new(); + loop { + let line = self.read_line().await; + if line.is_empty() { + break; + } + let done = line.starts_with(&format!("{tag} ")); + lines.push(line); + if done { + break; + } + } + lines + } + + async fn login_select(&mut self, mailbox: &str) -> Vec { + let _ = self.send("LOGIN user pass").await; + self.send(&format!("SELECT {mailbox}")).await + } + } + + fn joined(lines: &[String]) -> String { + lines.join("\n") + } + + #[tokio::test] + async fn two_connections_observe_each_others_writes_through_shared_state() { + // The property the old inline duplex responders could not express, and + // the reason this harness exists: engine B must see what engine A did. + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.create_mailbox("Archive"); + let uid = server.append_message("INBOX", "Subject: one\r\n\r\nbody", &[]); + + let mut engine_a = RawClient::connect(&server).await; + let mut engine_b = RawClient::connect(&server).await; + + engine_a.login_select("INBOX").await; + engine_b.login_select("INBOX").await; + + // B sees the message before A moves it. + let before = engine_b.send("UID FETCH 1:* (FLAGS)").await; + assert!( + joined(&before).contains(&format!("UID {uid}")), + "B should see the message first: {before:?}" + ); + + let moved = engine_a.send(&format!("UID MOVE {uid} Archive")).await; + assert!( + joined(&moved).contains("COPYUID"), + "a real move must emit COPYUID: {moved:?}" + ); + + // The same command on B now returns nothing -- shared state, not a + // per-connection script. + let after = engine_b.send("UID FETCH 1:* (FLAGS)").await; + assert!( + !joined(&after).contains(&format!("UID {uid} ")), + "B must not still see the moved message: {after:?}" + ); + assert_eq!(server.uids("INBOX"), Vec::::new()); + assert_eq!(server.uids("Archive").len(), 1); + } + + #[tokio::test] + async fn a_move_of_an_already_moved_uid_succeeds_silently_without_copyuid() { + // RFC 3501 6.4.8. This is the exact race that makes an unverified + // mutation report success: the losing engine's MOVE returns OK and + // does nothing. The harness must reproduce it, or no test can catch it. + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + server.create_mailbox("Archive"); + server.create_mailbox("Trash"); + let uid = server.append_message("INBOX", "Subject: contested\r\n\r\nx", &[]); + + let mut engine_a = RawClient::connect(&server).await; + let mut engine_b = RawClient::connect(&server).await; + engine_a.login_select("INBOX").await; + engine_b.login_select("INBOX").await; + + let winner = engine_a.send(&format!("UID MOVE {uid} Archive")).await; + assert!(joined(&winner).contains("COPYUID")); + + let loser = engine_b.send(&format!("UID MOVE {uid} Trash")).await; + let loser_text = joined(&loser); + assert!( + loser_text.contains("OK UID MOVE completed"), + "the losing move must report OK: {loser:?}" + ); + assert!( + !loser_text.contains("COPYUID"), + "a no-op move must NOT emit COPYUID: {loser:?}" + ); + assert!(server.uids("Trash").is_empty(), "nothing moved to Trash"); + } + + #[tokio::test] + async fn unchangedsince_reports_modified_in_the_tagged_response() { + // RFC 7162 3.1.3. MODIFIED rides on the tagged OK -- the response whose + // code async-imap discards, which is why chunk 3 needs a raw layer. + let server = MockImapServer::start(ServerProfile::Condstore) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + let uid = server.append_message("INBOX", "Subject: flags\r\n\r\nx", &[]); + + let mut engine_a = RawClient::connect(&server).await; + let mut engine_b = RawClient::connect(&server).await; + engine_a.login_select("INBOX").await; + engine_b.login_select("INBOX").await; + + // B captures the current modseq, then A changes the message. + let observed = engine_b.send(&format!("UID FETCH {uid} (FLAGS)")).await; + let modseq = observed + .iter() + .find_map(|l| { + let at = l.find("MODSEQ (")?; + l[at + "MODSEQ (".len()..] + .split(')') + .next()? + .parse::() + .ok() + }) + .expect("a MODSEQ is reported"); + + let _ = engine_a + .send(&format!("UID STORE {uid} +FLAGS (\\Seen)")) + .await; + + let conflicted = engine_b + .send(&format!( + "UID STORE {uid} (UNCHANGEDSINCE {modseq}) +FLAGS (\\Flagged)" + )) + .await; + assert!( + joined(&conflicted).contains(&format!("[MODIFIED {uid}]")), + "B's conditional store must be refused: {conflicted:?}" + ); + let flags = server.flags("INBOX", uid).unwrap_or_default(); + assert!( + !flags.iter().any(|f| f == "\\Flagged"), + "the refused store must not have applied: {flags:?}" + ); + } + + #[tokio::test] + async fn qresync_select_reports_vanished_for_messages_removed_since_the_modseq() { + // The mechanism that fixes ghost messages. A forward-only high-water + // mark cannot learn this; VANISHED states it directly. + let server = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + let keep = server.append_message("INBOX", "Subject: keep\r\n\r\nx", &[]); + let drop = server.append_message("INBOX", "Subject: drop\r\n\r\nx", &[]); + + let mut client = RawClient::connect(&server).await; + let _ = client.send("LOGIN user pass").await; + let _ = client.send("ENABLE QRESYNC").await; + let selected = client.send("SELECT INBOX").await; + let modseq = selected + .iter() + .find_map(|l| { + let at = l.find("HIGHESTMODSEQ ")?; + l[at + "HIGHESTMODSEQ ".len()..] + .split(']') + .next()? + .trim() + .parse::() + .ok() + }) + .expect("HIGHESTMODSEQ is advertised"); + + // Another client removes a message while we are away. + assert!(server.expunge("INBOX", drop)); + + let resync = client + .send(&format!("SELECT INBOX (QRESYNC (1000 {modseq}))")) + .await; + let text = joined(&resync); + assert!( + text.contains("VANISHED (EARLIER)") && text.contains(&drop.to_string()), + "the removal must be reported: {resync:?}" + ); + assert_eq!(server.uids("INBOX"), vec![keep]); + } + + #[tokio::test] + async fn qresync_select_is_rejected_without_enable_and_on_a_condstore_only_server() { + // RFC 7162 3.2.3 requires BAD when QRESYNC was not enabled. Modelling + // Gmail (CONDSTORE, no QRESYNC) keeps the ladder's rung 2 honest. + let qresync = MockImapServer::start(ServerProfile::Qresync) + .await + .expect("server starts"); + qresync.create_mailbox("INBOX"); + let mut client = RawClient::connect(&qresync).await; + let _ = client.send("LOGIN user pass").await; + let no_enable = client.send("SELECT INBOX (QRESYNC (1000 1))").await; + assert!( + joined(&no_enable).contains("BAD"), + "QRESYNC without ENABLE must be BAD: {no_enable:?}" + ); + + let gmail = MockImapServer::start(ServerProfile::Condstore) + .await + .expect("server starts"); + gmail.create_mailbox("INBOX"); + let mut gclient = RawClient::connect(&gmail).await; + let caps = gclient.send("CAPABILITY").await; + let caps_text = joined(&caps); + assert!(caps_text.contains("CONDSTORE"), "{caps:?}"); + assert!( + !caps_text.contains("QRESYNC"), + "the Gmail profile must not advertise QRESYNC: {caps:?}" + ); + let _ = gclient.send("LOGIN user pass").await; + let enabled = gclient.send("ENABLE QRESYNC").await; + assert!( + !joined(&enabled).contains("ENABLED QRESYNC"), + "a CONDSTORE-only server must not enable QRESYNC: {enabled:?}" + ); + } + + #[tokio::test] + async fn expunge_reports_vanished_under_qresync_and_expunge_otherwise() { + // RFC 6851 4.4. A client proving a delete by counting EXPUNGE lines + // sees none under QRESYNC and would wrongly infer failure. + for (profile, expect_vanished) in [ + (ServerProfile::Qresync, true), + (ServerProfile::Condstore, false), + ] { + let server = MockImapServer::start(profile).await.expect("server starts"); + server.create_mailbox("INBOX"); + let uid = server.append_message("INBOX", "Subject: gone\r\n\r\nx", &["\\Deleted"]); + + let mut client = RawClient::connect(&server).await; + let _ = client.send("LOGIN user pass").await; + if expect_vanished { + let _ = client.send("ENABLE QRESYNC").await; + } + let _ = client.send("SELECT INBOX").await; + let out = client.send(&format!("UID EXPUNGE {uid}")).await; + let text = joined(&out); + + if expect_vanished { + assert!(text.contains("* VANISHED"), "expected VANISHED: {out:?}"); + assert!(!text.contains("EXPUNGE\r"), "{out:?}"); + } else { + assert!(text.contains("EXPUNGE"), "expected EXPUNGE: {out:?}"); + assert!(!text.contains("VANISHED"), "{out:?}"); + } + assert!(server.uids("INBOX").is_empty()); + } + } + + #[tokio::test] + async fn the_command_log_records_every_connections_commands_in_order() { + let server = MockImapServer::start(ServerProfile::Basic) + .await + .expect("server starts"); + server.create_mailbox("INBOX"); + let mut a = RawClient::connect(&server).await; + let mut b = RawClient::connect(&server).await; + a.login_select("INBOX").await; + let _ = b.send("LOGIN user pass").await; + let _ = b.send("CAPABILITY").await; + + let fetches = server.commands_matching("SELECT"); + assert_eq!(fetches.len(), 1, "{fetches:?}"); + assert!(server.commands().len() >= 4, "{:?}", server.commands()); + } +} -- 2.54.0 From 83636efcdb4255666ce96c8c43c3c1ee6ead578e Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 10 Aug 2026 14:14:41 -0500 Subject: [PATCH 2/2] fix(filter): stop the webhook log-capture tests failing at random The two webhook telemetry tests installed a hand-rolled tracing::Subscriber with set_default and asserted on what it recorded. Roughly three runs in ten they captured nothing at all and failed. Callsite Interest is cached process-globally. A raw Subscriber installed thread-locally does not participate in the way tracing_subscriber's registry handles that cache, so whichever sibling test reached the dispatch callsites first decided whether these two could ever see them -- an order dependency, and therefore a coin flip under a parallel test harness. Two narrower fixes were tried and measured, and both made it worse: rebuild_interest_cache() after set_default failed five runs in six, because the rebuild is itself global and concurrent capture tests clobber each other; returning Interest::sometimes() from register_callsite failed nine in ten, because callsite registration consults the global dispatcher and never asks a thread-local subscriber at all. Adopt the registry-backed capture helper nuncio-cal and nuncio-contacts already use, which does not exhibit the problem. The two tests become plain #[test] driving a current-thread runtime built inside the capture scope, since a #[tokio::test] creates its runtime outside it. Assertions now match on level and named fields rather than substrings of a formatted line, so "the host was logged" no longer passes because the host happened to appear in some other field. Measured: 0 failures in 12 stress runs, against 3 in 10 before. --- crates/nuncio-filter/Cargo.toml | 1 + crates/nuncio-filter/src/lib.rs | 3 + crates/nuncio-filter/src/test_tracing.rs | 166 +++++++++++++++++++++++ crates/nuncio-filter/src/webhook.rs | 161 ++++++++-------------- 4 files changed, 229 insertions(+), 102 deletions(-) create mode 100644 crates/nuncio-filter/src/test_tracing.rs diff --git a/crates/nuncio-filter/Cargo.toml b/crates/nuncio-filter/Cargo.toml index 3eb1acb..341443a 100644 --- a/crates/nuncio-filter/Cargo.toml +++ b/crates/nuncio-filter/Cargo.toml @@ -31,3 +31,4 @@ aes-gcm.workspace = true [dev-dependencies] tempfile.workspace = true wiremock.workspace = true +tracing-subscriber.workspace = true diff --git a/crates/nuncio-filter/src/lib.rs b/crates/nuncio-filter/src/lib.rs index 7056052..2314110 100644 --- a/crates/nuncio-filter/src/lib.rs +++ b/crates/nuncio-filter/src/lib.rs @@ -11,6 +11,9 @@ pub mod parser; pub mod validator; pub mod webhook; +#[cfg(test)] +mod test_tracing; + pub use ast::*; pub use codegen::*; pub use engine::*; diff --git a/crates/nuncio-filter/src/test_tracing.rs b/crates/nuncio-filter/src/test_tracing.rs new file mode 100644 index 0000000..03729b2 --- /dev/null +++ b/crates/nuncio-filter/src/test_tracing.rs @@ -0,0 +1,166 @@ +//! Shared in-test `tracing` capture helper. +//! +//! Installs a thread-local subscriber that records every span (with its +//! fields) and every event (with its level, fields, and message) into an +//! in-memory buffer, so tests can assert on what the code logged -- including +//! the negative assertion that a secret never appears in any recorded field. +//! Deterministic and fully offline: no global subscriber, no I/O. +//! +//! This mirrors the identical helper in `nuncio-cal` and `nuncio-contacts`. +//! The webhook tests previously used a hand-rolled `tracing::Subscriber` +//! installed with `set_default`; that captured nothing whenever a sibling test +//! reached the same callsites first, because callsite `Interest` is cached +//! process-globally and a raw subscriber does not participate in the +//! `tracing_subscriber` registry's interest handling. The result was an +//! order-dependent empty capture that failed roughly three runs in ten. + +#![cfg(test)] + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use tracing::field::{Field, Visit}; +use tracing::span::Attributes; +use tracing::{Event, Id, Level, Subscriber}; +use tracing_subscriber::layer::Context as LayerContext; +use tracing_subscriber::prelude::*; +use tracing_subscriber::registry::LookupSpan; + +/// A captured span: its name and the string-rendered fields it opened with. +#[derive(Debug, Clone)] +pub(crate) struct RecordedSpan { + #[allow(dead_code)] + pub name: String, + pub fields: HashMap, +} + +/// A captured event: its level and its string-rendered fields (including the +/// `message`). +#[derive(Debug, Clone)] +pub(crate) struct RecordedEvent { + pub level: Level, + pub fields: HashMap, +} + +impl RecordedEvent { + /// The event's `message` field, or the empty string if it had none. + pub fn message(&self) -> &str { + self.fields.get("message").map(String::as_str).unwrap_or("") + } + + /// The string-rendered value of `name`, if the event carried that field. + pub fn field(&self, name: &str) -> Option<&str> { + self.fields.get(name).map(String::as_str) + } +} + +/// In-memory sink of everything the subscriber captured during a test. +#[derive(Debug, Default)] +pub(crate) struct Recorder { + spans: Mutex>, + events: Mutex>, +} + +impl Recorder { + /// A snapshot of every captured span. + #[allow(dead_code)] + pub fn spans(&self) -> Vec { + self.spans.lock().unwrap_or_else(|e| e.into_inner()).clone() + } + + /// A snapshot of every captured event. + pub fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + + /// Every field value recorded across all spans and events, so a test can + /// assert that a secret never appears anywhere in the telemetry. + pub fn all_field_values(&self) -> Vec { + let mut values = Vec::new(); + for span in self.spans().iter() { + values.extend(span.fields.values().cloned()); + } + for event in self.events().iter() { + values.extend(event.fields.values().cloned()); + } + values + } +} + +struct FieldVisitor(HashMap); + +impl Visit for FieldVisitor { + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().to_string(), value.to_string()); + } + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0 + .entry(field.name().to_string()) + .or_insert_with(|| format!("{value:?}")); + } +} + +struct CaptureLayer(Arc); + +impl tracing_subscriber::Layer for CaptureLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, _id: &Id, _ctx: LayerContext<'_, S>) { + let mut visitor = FieldVisitor(HashMap::new()); + attrs.record(&mut visitor); + self.0 + .spans + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedSpan { + name: attrs.metadata().name().to_string(), + fields: visitor.0, + }); + } + + fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { + let mut visitor = FieldVisitor(HashMap::new()); + event.record(&mut visitor); + self.0 + .events + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(RecordedEvent { + level: *event.metadata().level(), + fields: visitor.0, + }); + } +} + +/// Runs `f` with a thread-local capturing subscriber installed and returns the +/// recorder alongside `f`'s result. Because the subscriber is thread-local, +/// any async work driven inside `f` must run on the calling thread (e.g. a +/// current-thread runtime's `block_on`) for its telemetry to be captured -- +/// see [`block_on_captured`]. +pub(crate) fn with_recorder(f: impl FnOnce() -> T) -> (Arc, T) { + let recorder = Arc::new(Recorder::default()); + let subscriber = tracing_subscriber::registry().with(CaptureLayer(recorder.clone())); + let value = tracing::subscriber::with_default(subscriber, f); + (recorder, value) +} + +/// Drives `future` to completion on a current-thread runtime built inside the +/// capture scope, so every event it emits lands on the thread holding the +/// subscriber. This is the async counterpart to [`with_recorder`]; a +/// `#[tokio::test]` cannot be used here because its runtime is created outside +/// the capture scope. +pub(crate) fn block_on_captured(future: F) -> (Arc, F::Output) { + with_recorder(|| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build(); + match runtime { + Ok(rt) => rt.block_on(future), + Err(e) => panic!("building a current-thread runtime must succeed: {e}"), + } + }) +} diff --git a/crates/nuncio-filter/src/webhook.rs b/crates/nuncio-filter/src/webhook.rs index 123f82e..3cf543f 100644 --- a/crates/nuncio-filter/src/webhook.rs +++ b/crates/nuncio-filter/src/webhook.rs @@ -240,54 +240,11 @@ impl WebhookDispatcher { #[cfg(test)] mod tests { use super::*; - use std::sync::Mutex; - use tracing::field::{Field, Visit}; - use tracing::span; + use crate::test_tracing::block_on_captured; + use tracing::Level; use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; - /// Minimal `tracing::Subscriber` that records a formatted line per event - /// so tests can assert on emitted level + fields without pulling in - /// `tracing-subscriber`'s registry machinery. - struct CapturingSubscriber { - events: std::sync::Arc>>, - } - - struct LineVisitor<'a>(&'a mut String); - - impl Visit for LineVisitor<'_> { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - self.0.push_str(&format!(" {}={:?}", field.name(), value)); - } - } - - impl tracing::Subscriber for CapturingSubscriber { - fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { - true - } - - fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id { - span::Id::from_u64(1) - } - - fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {} - - fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {} - - fn event(&self, event: &tracing::Event<'_>) { - let mut line = format!("{}", event.metadata().level()); - let mut visitor = LineVisitor(&mut line); - event.record(&mut visitor); - if let Ok(mut events) = self.events.lock() { - events.push(line); - } - } - - fn enter(&self, _span: &span::Id) {} - - fn exit(&self, _span: &span::Id) {} - } - #[tokio::test] async fn test_pinned_client_connects_to_checked_ip_not_dns() { // `.invalid` is reserved by RFC 2606 to never resolve: any real DNS @@ -391,80 +348,80 @@ mod tests { assert_eq!(first_blocked_address(&addrs), None); } - #[tokio::test] - async fn test_blocked_webhook_logs_warn_with_host_and_reason() { - let events: std::sync::Arc>> = - std::sync::Arc::new(Mutex::new(Vec::new())); - let subscriber = CapturingSubscriber { - events: events.clone(), - }; - let _guard = tracing::subscriber::set_default(subscriber); - - let dispatcher = WebhookDispatcher::new("secret_key_123"); - let result = dispatcher - .dispatch( - "http://169.254.169.254/latest/meta-data", - "rule_1", - "msg_1", - "Test", - "a@b.com", - ) - .await; + // Not `#[tokio::test]`: the capture subscriber is thread-local, so the + // runtime has to be built inside the capture scope for the dispatch's + // telemetry to land on the thread holding it. + #[test] + fn test_blocked_webhook_logs_warn_with_host_and_reason() { + let (recorder, result) = block_on_captured(async { + let dispatcher = WebhookDispatcher::new("secret_key_123"); + dispatcher + .dispatch( + "http://169.254.169.254/latest/meta-data", + "rule_1", + "msg_1", + "Test", + "a@b.com", + ) + .await + }); assert!(matches!(result, Err(WebhookError::SecurityViolation(_)))); - let captured = events.lock().unwrap(); + let events = recorder.events(); assert!( - captured.iter().any(|line| line.contains("WARN") - && line.contains("blocked") - && line.contains("169.254.169.254")), - "expected a WARN log carrying the blocked host and reason, got: {captured:?}" + events.iter().any(|e| e.level == Level::WARN + && e.message().contains("blocked") + && e.field("host") + .is_some_and(|h| h.contains("169.254.169.254"))), + "expected a WARN log carrying the blocked host and reason, got: {events:?}" ); assert!( - !captured.iter().any(|line| line.contains("secret_key_123")), + !recorder + .all_field_values() + .iter() + .any(|v| v.contains("secret_key_123")), "the webhook HMAC secret must never appear in a log line" ); } - #[tokio::test] - async fn test_allowed_webhook_logs_debug_not_warn() { - let mock_server = MockServer::start().await; - Mock::given(method("POST")) - .and(path("/hook")) - .respond_with(ResponseTemplate::new(200)) - .mount(&mock_server) - .await; + // See the sibling capture test for why this is not a `#[tokio::test]`. + // The wiremock server is started inside the same runtime so the whole + // exchange runs on the capturing thread. + #[test] + fn test_allowed_webhook_logs_debug_not_warn() { + let (recorder, status) = block_on_captured(async { + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/hook")) + .respond_with(ResponseTemplate::new(200)) + .mount(&mock_server) + .await; - let events: std::sync::Arc>> = - std::sync::Arc::new(Mutex::new(Vec::new())); - let subscriber = CapturingSubscriber { - events: events.clone(), - }; - let _guard = tracing::subscriber::set_default(subscriber); - - let dispatcher = WebhookDispatcher::new("secret_key_123"); - let opts = ValidationOptions { - available_folders: None, - allowed_forward_domains: None, - block_private_webhooks: false, - }; - let url = format!("{}/hook", mock_server.uri()); - let status = dispatcher - .dispatch_with_options(&url, "rule_1", "msg_1", "Test", "a@b.com", &opts) - .await - .expect("dispatch to the mock server must succeed"); + let dispatcher = WebhookDispatcher::new("secret_key_123"); + let opts = ValidationOptions { + available_folders: None, + allowed_forward_domains: None, + block_private_webhooks: false, + }; + let url = format!("{}/hook", mock_server.uri()); + dispatcher + .dispatch_with_options(&url, "rule_1", "msg_1", "Test", "a@b.com", &opts) + .await + .expect("dispatch to the mock server must succeed") + }); assert_eq!(status, 200); - let captured = events.lock().unwrap(); + let events = recorder.events(); assert!( - captured + events .iter() - .any(|line| line.contains("DEBUG") && line.contains("allowed")), - "expected a DEBUG log for the allowed egress decision, got: {captured:?}" + .any(|e| e.level == Level::DEBUG && e.message().contains("allowed")), + "expected a DEBUG log for the allowed egress decision, got: {events:?}" ); assert!( - !captured.iter().any(|line| line.contains("WARN")), - "an allowed dispatch must not emit a WARN, got: {captured:?}" + !events.iter().any(|e| e.level == Level::WARN), + "an allowed dispatch must not emit a WARN, got: {events:?}" ); } } -- 2.54.0