diff --git a/crates/nuncio-mail/src/backend.rs b/crates/nuncio-mail/src/backend.rs index a766d4d..5c9bb96 100644 --- a/crates/nuncio-mail/src/backend.rs +++ b/crates/nuncio-mail/src/backend.rs @@ -5,6 +5,60 @@ use nuncio_core::model::{Attachment, Email, Folder}; use crate::parser::MailError; +/// The remote change a [`RemoteMutationSpec`] applies to one addressed message. +/// +/// This is the closed vocabulary of genuinely-remote filter actions that the +/// outbox drains against a real server. `MARK READ`/`MARK UNREAD` are applied +/// locally to the store when a filter matches, so they are deliberately NOT +/// represented here -- only actions that mutate server-side state are. +/// `FORWARD` and `CALL WEBHOOK` are also excluded: they are not backend +/// mutations (one composes an outbound message, the other calls an HTTP +/// endpoint), so they are executed through the send/webhook seams instead. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RemoteMutationKind { + /// Add (`value == true`) or remove (`value == false`) the "flagged"/ + /// starred keyword: IMAP `\Flagged`, JMAP `$flagged`. Covers `FLAG`/`UNFLAG`. + SetFlagged { + /// Whether the message should end up flagged. + value: bool, + }, + /// Move the message from its source folder to `to_folder`. + Move { + /// Destination folder (IMAP mailbox name / JMAP mailbox id). + to_folder: String, + }, + /// Copy the message into `to_folder`, leaving the source copy in place. + Copy { + /// Destination folder (IMAP mailbox name / JMAP mailbox id). + to_folder: String, + }, + /// Delete the message (IMAP `\Deleted` + `EXPUNGE`; JMAP `destroy`). + Delete, +} + +/// A fully-addressed remote mutation for a single message: which message +/// (protocol-native id), the folder it currently lives in, that folder's +/// stored sync checkpoint (carrying IMAP UIDVALIDITY), and the change to +/// apply. +/// +/// `folder_checkpoint` is the value persisted by a prior sync +/// (`"{uidvalidity}:{uidnext}"` for IMAP). The IMAP backend uses it to refuse +/// to act when the mailbox's current UIDVALIDITY no longer matches -- a +/// renumbered mailbox reassigns UIDs, so the stored UID would otherwise +/// address the wrong message. Protocols that do not use UIDVALIDITY (JMAP) +/// ignore it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemoteMutationSpec { + /// Protocol-native message id (IMAP `imap-uid-{uid}`, JMAP object id). + pub message_id: String, + /// Folder the message currently lives in. + pub folder_id: String, + /// The source folder's stored sync checkpoint, if any. + pub folder_checkpoint: Option, + /// The change to apply. + pub kind: RemoteMutationKind, +} + /// Protocol-agnostic mail backend engine trait implemented by JMAP and IMAP engines. #[async_trait] pub trait MailBackend: Send + Sync { @@ -18,6 +72,13 @@ pub trait MailBackend: Send + Sync { folder_id: &str, since_state: Option<&str>, ) -> Result<(Vec, String), MailError>; + + /// Apply a remote mutation (move/copy/flag/delete) to a single message + /// against the real server. MUST return `Ok(())` only when the server + /// genuinely applied the change -- implementations must never fabricate + /// success, and must refuse to act (returning an error) when the target + /// message cannot be safely identified (e.g. an IMAP UIDVALIDITY mismatch). + async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError>; } /// A composed outbound email message ready to send over SMTP. Deliberately diff --git a/crates/nuncio-mail/src/imap.rs b/crates/nuncio-mail/src/imap.rs index 2b2f50c..adfbf63 100644 --- a/crates/nuncio-mail/src/imap.rs +++ b/crates/nuncio-mail/src/imap.rs @@ -31,9 +31,36 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); /// honest timeout rather than hanging the sync indefinitely. const FETCH_ITEM_TIMEOUT: Duration = Duration::from_secs(60); -use crate::backend::MailBackend; +use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec}; use crate::parser::{MailError, MimeParserAdapter}; +/// Prefix an IMAP [`Email::id`] carries in front of its raw numeric UID (see +/// [`ImapEngine::sync_folder_messages_with_session`], which formats every id as +/// `"imap-uid-{uid}"`). A remote mutation must recover the UID from the id to +/// address the message, so the prefix is defined once here. +const IMAP_UID_ID_PREFIX: &str = "imap-uid-"; + +/// Recover the raw IMAP UID from an [`Email::id`] of the form +/// `"imap-uid-{uid}"`. Returns `None` for any id that does not carry the +/// expected prefix and a valid non-zero `u32` UID, so a mutation against a +/// malformed/foreign id fails honestly rather than addressing UID 0. +fn parse_imap_uid(message_id: &str) -> Option { + let raw = message_id.strip_prefix(IMAP_UID_ID_PREFIX)?; + let uid = raw.trim().parse::().ok()?; + (uid >= 1).then_some(uid) +} + +/// Extract the UIDVALIDITY component of a stored folder checkpoint +/// (`"{uidvalidity}:{uidnext}"`). Returns `None` for a missing, malformed, or +/// legacy bare-number checkpoint -- the caller treats an absent stored +/// UIDVALIDITY as "cannot prove the mailbox has not been renumbered" and +/// refuses to act. +fn checkpoint_uid_validity(checkpoint: Option<&str>) -> Option { + let raw = checkpoint?; + let (validity, _uid) = raw.split_once(CHECKPOINT_DELIM)?; + validity.trim().parse::().ok() +} + /// Delimiter between the two components of a folder sync checkpoint. The /// checkpoint state string is `"{uidvalidity}{DELIM}{uidnext}"` (e.g. /// `"42:105"`): the mailbox's UIDVALIDITY at the time it was written, then the @@ -759,6 +786,222 @@ impl ImapEngine { result } + /// Apply a remote mutation over an active IMAP session, selecting the + /// source folder and verifying its UIDVALIDITY against the stored + /// checkpoint BEFORE issuing any command that could touch the wrong + /// message. + /// + /// The UIDVALIDITY guard (RFC 3501 s2.3.1.1) is the safety-critical step: a + /// stored IMAP UID is only meaningful under the UIDVALIDITY it was captured + /// with. If the mailbox was recreated its UIDVALIDITY changes and UIDs are + /// reassigned, so acting on the old UID would mutate a different message. A + /// mismatch -- or a stored checkpoint that carries no UIDVALIDITY to + /// compare against, or a server that reports none -- is refused with an + /// honest [`MailError::ImapError`]; the mutation is never applied blind. + /// + /// `MOVE` uses `UID MOVE` (RFC 6851) when the server advertises the `MOVE` + /// capability, else falls back to the equivalent `UID COPY` + `UID STORE + /// +FLAGS (\Deleted)` + UID-scoped `UID EXPUNGE`. `DELETE` is `UID STORE + /// +FLAGS (\Deleted)` + `UID EXPUNGE`. The expunge is always scoped to the + /// target UID via RFC 4315 (never a mailbox-wide `EXPUNGE`, which would + /// destroy other `\Deleted` messages) and requires `UIDPLUS`, failing + /// closed otherwise. Each command's tagged result is checked, so only a + /// genuine server success returns `Ok(())`. + pub async fn apply_mutation_with_session( + &self, + spec: &RemoteMutationSpec, + session: &mut async_imap::Session, + ) -> Result<(), MailError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + let uid = parse_imap_uid(&spec.message_id).ok_or_else(|| { + MailError::ImapError(format!( + "cannot parse an IMAP UID from message id '{}'", + spec.message_id + )) + })?; + + let mailbox = session.select(&spec.folder_id).await.map_err(|e| { + MailError::ImapError(format!( + "failed to select folder '{}': {}", + spec.folder_id, e + )) + })?; + + // Refuse to act unless the mailbox's current UIDVALIDITY provably + // matches the one the stored UID was captured under. + let stored_validity = checkpoint_uid_validity(spec.folder_checkpoint.as_deref()); + match (stored_validity, mailbox.uid_validity) { + (Some(stored), Some(current)) if stored == current => {} + (stored, current) => { + return Err(MailError::ImapError(format!( + "refusing to mutate message '{}' in folder '{}': UIDVALIDITY guard failed \ + (stored {stored:?}, current {current:?}); the stored UID may no longer \ + address the intended message", + spec.message_id, spec.folder_id + ))); + } + } + + let uid_set = uid.to_string(); + match &spec.kind { + RemoteMutationKind::SetFlagged { value } => { + let op = if *value { "+FLAGS" } else { "-FLAGS" }; + self.uid_store_flags(session, &uid_set, op, "\\Flagged") + .await + } + RemoteMutationKind::Copy { to_folder } => { + session.uid_copy(&uid_set, to_folder).await.map_err(|e| { + MailError::ImapError(format!( + "UID COPY of '{}' to '{}' failed: {}", + spec.message_id, to_folder, e + )) + }) + } + RemoteMutationKind::Move { to_folder } => { + self.uid_move(session, &uid_set, to_folder, &spec.message_id) + .await + } + RemoteMutationKind::Delete => { + self.uid_store_flags(session, &uid_set, "+FLAGS", "\\Deleted") + .await?; + self.uid_expunge_scoped(session, &uid_set).await + } + } + } + + /// Issue a `UID STORE {uid} {op} ({flag})` and drain its untagged `FETCH` + /// responses, surfacing any protocol error. `op` is `+FLAGS`/`-FLAGS`. + async fn uid_store_flags( + &self, + session: &mut async_imap::Session, + uid_set: &str, + op: &str, + flag: &str, + ) -> Result<(), MailError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + let query = format!("{op} ({flag})"); + let updates = session.uid_store(uid_set, &query).await.map_err(|e| { + MailError::ImapError(format!("UID STORE {query} for uid {uid_set} failed: {e}")) + })?; + tokio::pin!(updates); + while let Some(item) = updates.next().await { + item.map_err(|e| { + MailError::ImapError(format!("reading UID STORE response failed: {e}")) + })?; + } + Ok(()) + } + + /// Permanently remove ONLY the target `uid_set` from the selected folder + /// via RFC 4315 `UID EXPUNGE`, draining its untagged `EXPUNGE` responses. + /// + /// A bare `EXPUNGE` (RFC 3501) removes EVERY `\Deleted` message in the + /// mailbox, so it would collaterally destroy any other message a different + /// client had already flagged `\Deleted`. `UID EXPUNGE` is scoped to + /// exactly the given UIDs, so it is the only safe expunge in a shared + /// mailbox. It requires the `UIDPLUS` capability; if the server does not + /// advertise it, this fails closed with an honest error rather than + /// falling back to the destructive mailbox-wide `EXPUNGE`. + async fn uid_expunge_scoped( + &self, + session: &mut async_imap::Session, + uid_set: &str, + ) -> Result<(), MailError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + let caps = session + .capabilities() + .await + .map_err(|e| MailError::ImapError(format!("CAPABILITY query failed: {e}")))?; + let has_uidplus = caps.has_str("UIDPLUS"); + drop(caps); + if !has_uidplus { + return Err(MailError::ImapError(format!( + "refusing to expunge uid {uid_set}: server does not advertise UIDPLUS, so a \ + UID-scoped EXPUNGE (RFC 4315) is unavailable; a bare EXPUNGE could destroy \ + other \\Deleted messages in the folder" + ))); + } + + let expunged = session + .uid_expunge(uid_set) + .await + .map_err(|e| MailError::ImapError(format!("UID EXPUNGE {uid_set} failed: {e}")))?; + tokio::pin!(expunged); + while let Some(item) = expunged.next().await { + item.map_err(|e| { + MailError::ImapError(format!("reading UID EXPUNGE response failed: {e}")) + })?; + } + Ok(()) + } + + /// Move a message by UID, preferring RFC 6851 `UID MOVE` and falling back + /// to `UID COPY` + `UID STORE +FLAGS (\Deleted)` + UID-scoped `UID EXPUNGE` + /// when the server does not advertise the `MOVE` capability. The fallback's + /// expunge is scoped to the target UID (never a mailbox-wide `EXPUNGE`) and + /// requires `UIDPLUS`, failing closed otherwise. + async fn uid_move( + &self, + session: &mut async_imap::Session, + uid_set: &str, + to_folder: &str, + message_id: &str, + ) -> Result<(), MailError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + Debug + 'static, + { + let caps = session + .capabilities() + .await + .map_err(|e| MailError::ImapError(format!("CAPABILITY query failed: {e}")))?; + let has_move = caps.has_str("MOVE"); + drop(caps); + + if has_move { + return session.uid_mv(uid_set, to_folder).await.map_err(|e| { + MailError::ImapError(format!( + "UID MOVE of '{message_id}' to '{to_folder}' failed: {e}" + )) + }); + } + + session.uid_copy(uid_set, to_folder).await.map_err(|e| { + MailError::ImapError(format!( + "UID COPY (move fallback) of '{message_id}' to '{to_folder}' failed: {e}" + )) + })?; + self.uid_store_flags(session, uid_set, "+FLAGS", "\\Deleted") + .await?; + self.uid_expunge_scoped(session, uid_set).await + } + + /// Apply a remote mutation over a freshly authenticated session. See + /// [`Self::apply_mutation_with_session`] for the UIDVALIDITY guard and the + /// per-action protocol commands. + pub async fn apply_remote_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + let (username, password) = match (&self.username, &self.password) { + (Some(u), Some(p)) => (u.as_str(), p.as_str()), + _ => { + return Err(MailError::AuthError( + "IMAP remote mutation requires credentials".to_string(), + )) + } + }; + let mut session = self + .socket_manager + .connect_session(username, password) + .await?; + let result = self.apply_mutation_with_session(spec, &mut session).await; + let _ = session.logout().await; + result + } + /// Listen for real-time IMAP IDLE notification events on an active IMAP session. pub async fn listen_idle_session( &self, @@ -900,6 +1143,10 @@ impl MailBackend for ImapEngine { ) -> Result<(Vec, String), MailError> { self.sync_folder_messages(folder_id, since_state).await } + + async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + self.apply_remote_mutation(spec).await + } } #[cfg(test)] @@ -1327,6 +1574,346 @@ mod tests { server.abort(); } + #[test] + fn parse_imap_uid_recovers_uid_and_rejects_malformed_ids() { + assert_eq!(parse_imap_uid("imap-uid-42"), Some(42)); + assert_eq!(parse_imap_uid("imap-uid-1"), Some(1)); + // UID 0 is not a valid IMAP UID; a bare/foreign id yields nothing. + assert_eq!(parse_imap_uid("imap-uid-0"), None); + assert_eq!(parse_imap_uid("42"), None); + assert_eq!(parse_imap_uid("jmap-object-id"), None); + } + + #[test] + fn checkpoint_uid_validity_extracts_only_a_well_formed_validity() { + assert_eq!(checkpoint_uid_validity(Some("7:105")), Some(7)); + assert_eq!(checkpoint_uid_validity(Some(" 7 : 105 ")), Some(7)); + assert_eq!(checkpoint_uid_validity(None), None); + // A legacy bare-number checkpoint carries no validity to compare. + assert_eq!(checkpoint_uid_validity(Some("105")), None); + assert_eq!(checkpoint_uid_validity(Some("x:105")), None); + } + + /// Drive a mutation over a scripted IMAP server (in-memory duplex), + /// capturing every command line the client issued so the test can assert + /// on the exact protocol commands. `advertise_move`/`advertise_uidplus` + /// control whether the scripted `CAPABILITY` response includes `MOVE`/ + /// `UIDPLUS`; `select_uid_validity` is the UIDVALIDITY the scripted + /// `SELECT` reports. + async fn run_scripted_mutation( + spec: &RemoteMutationSpec, + advertise_move: bool, + advertise_uidplus: bool, + select_uid_validity: u32, + ) -> (Result<(), MailError>, Vec) { + let (client_io, server_io) = tokio::io::duplex(8192); + let commands = Arc::new(std::sync::Mutex::new(Vec::::new())); + let server_commands = commands.clone(); + + let server = tokio::spawn(async move { + let mut io = server_io; + while let Some(line) = read_scripted_line(&mut io).await { + if let Ok(mut g) = server_commands.lock() { + g.push(line.clone()); + } + let tag = line.split_whitespace().next().unwrap_or("").to_string(); + let upper = line.to_ascii_uppercase(); + if upper.contains("LOGIN") { + let _ = io + .write_all(format!("{tag} OK Logged in\r\n").as_bytes()) + .await; + } else if upper.contains("SELECT") { + let resp = format!( + "* FLAGS (\\Seen \\Flagged \\Deleted)\r\n* 3 EXISTS\r\n* 0 RECENT\r\n\ + * OK [UIDVALIDITY {select_uid_validity}] ok\r\n* OK [UIDNEXT 105] ok\r\n\ + {tag} OK [READ-WRITE] SELECT done\r\n" + ); + let _ = io.write_all(resp.as_bytes()).await; + } else if upper.contains("CAPABILITY") { + let mut caps = String::from("IMAP4rev1"); + if advertise_move { + caps.push_str(" MOVE"); + } + if advertise_uidplus { + caps.push_str(" UIDPLUS"); + } + let _ = io + .write_all( + format!("* CAPABILITY {caps}\r\n{tag} OK CAPABILITY done\r\n") + .as_bytes(), + ) + .await; + } else if upper.contains("UID EXPUNGE") { + let _ = io + .write_all(format!("{tag} OK UID EXPUNGE completed\r\n").as_bytes()) + .await; + } else if upper.contains("UID STORE") { + let _ = io + .write_all(format!("{tag} OK STORE completed\r\n").as_bytes()) + .await; + } else if upper.contains("UID COPY") { + let _ = io + .write_all(format!("{tag} OK COPY completed\r\n").as_bytes()) + .await; + } else if upper.contains("UID MOVE") { + let _ = io + .write_all(format!("{tag} OK MOVE completed\r\n").as_bytes()) + .await; + } else if upper.contains("EXPUNGE") { + let _ = io + .write_all(format!("{tag} OK EXPUNGE completed\r\n").as_bytes()) + .await; + } else if upper.contains("LOGOUT") { + let _ = io + .write_all(format!("* BYE\r\n{tag} OK LOGOUT\r\n").as_bytes()) + .await; + break; + } + } + }); + + let client = async_imap::Client::new(client_io); + let mut session = client + .login("user", "pass") + .await + .map_err(|(e, _)| e) + .expect("scripted login succeeds"); + let engine = ImapEngine::new("acct-1", "example.test", 993); + let result = engine.apply_mutation_with_session(spec, &mut session).await; + drop(session); + let _ = server.await; + + let captured = commands.lock().expect("lock captured commands").clone(); + (result, captured) + } + + /// True if `line` is a bare mailbox-wide `EXPUNGE` (RFC 3501), NOT a + /// UID-scoped `UID EXPUNGE` (RFC 4315). Lets a test prove the destructive + /// mailbox-wide form is never issued (a plain `contains("EXPUNGE")` would + /// be satisfied by the safe `UID EXPUNGE` too). + fn is_bare_expunge(line: &str) -> bool { + let u = line.to_ascii_uppercase(); + u.contains("EXPUNGE") && !u.contains("UID EXPUNGE") + } + + fn imap_spec(kind: RemoteMutationKind, checkpoint: Option<&str>) -> RemoteMutationSpec { + RemoteMutationSpec { + message_id: "imap-uid-42".to_string(), + folder_id: "INBOX".to_string(), + folder_checkpoint: checkpoint.map(str::to_string), + kind, + } + } + + #[tokio::test] + async fn imap_apply_mutation_flag_issues_uid_store_add_flags() { + let (result, commands) = run_scripted_mutation( + &imap_spec( + RemoteMutationKind::SetFlagged { value: true }, + Some("1:105"), + ), + false, + true, + 1, + ) + .await; + result.expect("flag mutation succeeds"); + assert!( + commands + .iter() + .any(|c| c.contains("UID STORE 42 +FLAGS (\\Flagged)")), + "expected a UID STORE +FLAGS (\\Flagged), got: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_move_uses_uid_move_when_server_advertises_move() { + let (result, commands) = run_scripted_mutation( + &imap_spec( + RemoteMutationKind::Move { + to_folder: "Archive".to_string(), + }, + Some("1:105"), + ), + true, + true, + 1, + ) + .await; + result.expect("move mutation succeeds"); + assert!( + commands.iter().any(|c| c.contains("UID MOVE 42")), + "expected a UID MOVE, got: {commands:?}" + ); + assert!( + !commands.iter().any(|c| c.contains("UID COPY")), + "must not fall back to COPY when MOVE is advertised: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_move_falls_back_to_copy_store_uid_expunge_without_move_capability() + { + let (result, commands) = run_scripted_mutation( + &imap_spec( + RemoteMutationKind::Move { + to_folder: "Archive".to_string(), + }, + Some("1:105"), + ), + false, + true, + 1, + ) + .await; + result.expect("move fallback succeeds"); + assert!(commands.iter().any(|c| c.contains("UID COPY 42"))); + assert!(commands + .iter() + .any(|c| c.contains("UID STORE 42 +FLAGS (\\Deleted)"))); + // The expunge MUST be scoped to the target UID, never a mailbox-wide + // bare EXPUNGE that could destroy other \Deleted messages. + assert!( + commands.iter().any(|c| c.contains("UID EXPUNGE 42")), + "fallback must issue a UID-scoped UID EXPUNGE 42, got: {commands:?}" + ); + assert!( + !commands.iter().any(|c| is_bare_expunge(c)), + "fallback must NOT issue a bare mailbox-wide EXPUNGE, got: {commands:?}" + ); + assert!( + !commands.iter().any(|c| c.contains("UID MOVE")), + "must not issue UID MOVE without the capability: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_delete_stores_deleted_and_uid_expunges_the_target_only() { + let (result, commands) = run_scripted_mutation( + &imap_spec(RemoteMutationKind::Delete, Some("1:105")), + false, + true, + 1, + ) + .await; + result.expect("delete mutation succeeds"); + assert!(commands + .iter() + .any(|c| c.contains("UID STORE 42 +FLAGS (\\Deleted)"))); + // The expunge MUST target exactly the deleted UID (RFC 4315), never a + // mailbox-wide bare EXPUNGE that would also destroy other client's + // \Deleted messages in the same folder. + assert!( + commands.iter().any(|c| c.contains("UID EXPUNGE 42")), + "delete must issue a UID-scoped UID EXPUNGE 42, got: {commands:?}" + ); + assert!( + !commands.iter().any(|c| is_bare_expunge(c)), + "delete must NOT issue a bare mailbox-wide EXPUNGE, got: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_delete_fails_closed_without_uidplus_and_issues_no_bare_expunge() { + // Without UIDPLUS there is no UID-scoped expunge; the op MUST fail + // closed rather than fall back to a destructive mailbox-wide EXPUNGE. + let (result, commands) = run_scripted_mutation( + &imap_spec(RemoteMutationKind::Delete, Some("1:105")), + false, + false, + 1, + ) + .await; + assert!( + matches!(result, Err(MailError::ImapError(_))), + "delete without UIDPLUS must fail closed, got: {result:?}" + ); + assert!( + !commands + .iter() + .any(|c| c.to_ascii_uppercase().contains("EXPUNGE")), + "no expunge of any kind may be issued when UIDPLUS is unavailable, got: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_move_fallback_fails_closed_without_uidplus() { + // The no-MOVE-capability move fallback also relies on a UID-scoped + // expunge; without UIDPLUS it must fail closed and issue no bare + // EXPUNGE (the preceding COPY/STORE are non-destructive to others). + let (result, commands) = run_scripted_mutation( + &imap_spec( + RemoteMutationKind::Move { + to_folder: "Archive".to_string(), + }, + Some("1:105"), + ), + false, + false, + 1, + ) + .await; + assert!( + matches!(result, Err(MailError::ImapError(_))), + "move fallback without UIDPLUS must fail closed, got: {result:?}" + ); + assert!( + !commands + .iter() + .any(|c| c.to_ascii_uppercase().contains("EXPUNGE")), + "no expunge of any kind may be issued when UIDPLUS is unavailable, got: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_refuses_to_act_on_a_uidvalidity_mismatch() { + // The stored checkpoint captured the UID under UIDVALIDITY 1, but the + // mailbox now reports 2 (renumbered). The mutation MUST be refused and + // NO mutating command (STORE/COPY/MOVE/EXPUNGE) may be issued. + let (result, commands) = run_scripted_mutation( + &imap_spec(RemoteMutationKind::Delete, Some("1:105")), + false, + true, + 2, + ) + .await; + let err = result.expect_err("a UIDVALIDITY mismatch must be refused"); + assert!(matches!(err, MailError::ImapError(_))); + assert!( + !commands.iter().any(|c| { + let u = c.to_ascii_uppercase(); + u.contains("UID STORE") + || u.contains("UID COPY") + || u.contains("UID MOVE") + || u.contains("EXPUNGE") + }), + "no mutating command may be issued on a UIDVALIDITY mismatch, got: {commands:?}" + ); + } + + #[tokio::test] + async fn imap_apply_mutation_refuses_when_no_checkpoint_proves_uidvalidity() { + // With no stored checkpoint there is no UIDVALIDITY to prove the UID + // still addresses the intended message, so the op is refused. + let (result, _commands) = + run_scripted_mutation(&imap_spec(RemoteMutationKind::Delete, None), false, true, 1) + .await; + assert!(matches!( + result.expect_err("missing checkpoint must be refused"), + MailError::ImapError(_) + )); + } + + #[tokio::test] + async fn imap_apply_remote_mutation_requires_credentials() { + let engine = ImapEngine::new("acct-1", "example.test", 993); + let err = engine + .apply_remote_mutation(&imap_spec(RemoteMutationKind::Delete, Some("1:105"))) + .await + .expect_err("credential-less mutation must fail"); + assert!(matches!(err, MailError::AuthError(_))); + } + #[tokio::test] async fn with_tls_mode_records_mode_and_resolves_default_ports() { // A `0` port resolves to the conventional default for each mode. diff --git a/crates/nuncio-mail/src/jmap.rs b/crates/nuncio-mail/src/jmap.rs index 76fad28..b7a1e0a 100644 --- a/crates/nuncio-mail/src/jmap.rs +++ b/crates/nuncio-mail/src/jmap.rs @@ -5,7 +5,7 @@ use nuncio_core::model::{Email, Folder}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use crate::backend::MailBackend; +use crate::backend::{MailBackend, RemoteMutationKind, RemoteMutationSpec}; use crate::parser::MailError; /// JMAP Session Object (RFC 8620 Section 2). @@ -203,6 +203,100 @@ impl JmapEngine { }) } + /// Build a JSON-RPC `Email/set` request applying `spec` to a single + /// message (RFC 8621 s4.6). A `Move`/`Copy`/`SetFlagged` becomes an + /// `update` patch keyed by the JMAP object id; a `Delete` becomes a + /// `destroy`. + /// + /// `Move` replaces the message's mailbox set with the destination (removing + /// the source), whereas `Copy` adds the destination via a `mailboxIds/{id}` + /// patch, leaving the existing membership intact. Flag toggles patch the + /// `$flagged` keyword (`true` to set, `null` to clear), the JMAP equivalent + /// of IMAP `\Flagged`. + pub fn build_email_set_request(account_id: &str, spec: &RemoteMutationSpec) -> Value { + let id = &spec.message_id; + let (update, destroy) = match &spec.kind { + RemoteMutationKind::SetFlagged { value } => { + let keyword = if *value { json!(true) } else { Value::Null }; + (json!({ id: { "keywords/$flagged": keyword } }), json!([])) + } + RemoteMutationKind::Move { to_folder } => ( + json!({ id: { "mailboxIds": { to_folder: true } } }), + json!([]), + ), + RemoteMutationKind::Copy { to_folder } => ( + json!({ id: { format!("mailboxIds/{to_folder}"): true } }), + json!([]), + ), + RemoteMutationKind::Delete => (json!({}), json!([id])), + }; + + json!({ + "using": ["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"], + "methodCalls": [ + [ + "Email/set", + { + "accountId": account_id, + "update": update, + "destroy": destroy + }, + "c1" + ] + ] + }) + } + + /// Confirm a JMAP `Email/set` response genuinely applied the mutation to + /// `message_id`: the id must appear in `updated`/`destroyed` and must NOT + /// appear in `notUpdated`/`notDestroyed`. Anything else -- a server-side + /// rejection, or an id the server silently ignored -- surfaces as an honest + /// error so the outbox never marks a rejected mutation completed. + pub fn confirm_email_set_applied( + raw_json: &str, + message_id: &str, + expect_destroy: bool, + ) -> Result<(), MailError> { + let val: Value = serde_json::from_str(raw_json) + .map_err(|e| MailError::ParseFailed(format!("invalid JMAP JSON response: {e}")))?; + let payload = Self::extract_method_response_payload(&val)?; + + // A rejection is reported in notUpdated/notDestroyed keyed by id; treat + // its presence as the authoritative failure signal. + let rejected = payload + .get(if expect_destroy { + "notDestroyed" + } else { + "notUpdated" + }) + .and_then(|v| v.as_object()) + .is_some_and(|m| m.contains_key(message_id)); + if rejected { + return Err(MailError::OperationFailed(format!( + "JMAP Email/set rejected mutation for message '{message_id}'" + ))); + } + + let applied = if expect_destroy { + payload + .get("destroyed") + .and_then(|v| v.as_array()) + .is_some_and(|ids| ids.iter().any(|v| v.as_str() == Some(message_id))) + } else { + payload + .get("updated") + .and_then(|v| v.as_object()) + .is_some_and(|m| m.contains_key(message_id)) + }; + if applied { + Ok(()) + } else { + Err(MailError::OperationFailed(format!( + "JMAP Email/set did not confirm the mutation for message '{message_id}'" + ))) + } + } + /// Build JSON-RPC request invocation for `Email/changes` (differential sync). pub fn build_email_changes_request(account_id: &str, since_state: &str) -> Value { json!({ @@ -451,6 +545,21 @@ impl MailBackend for JmapEngine { let get_raw = self.post_jmap(&session.api_url, &get_request).await?; self.parse_email_get_response(&get_raw) } + + async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + if !self.has_credentials() { + return Err(MailError::AuthError( + "JMAP remote mutation requires credentials".to_string(), + )); + } + let session = self.discover_session().await?; + let account_id = self.resolve_account_id(&session); + + let request = Self::build_email_set_request(&account_id, spec); + let raw = self.post_jmap(&session.api_url, &request).await?; + let expect_destroy = matches!(spec.kind, RemoteMutationKind::Delete); + Self::confirm_email_set_applied(&raw, &spec.message_id, expect_destroy) + } } #[cfg(test)] @@ -578,6 +687,49 @@ mod tests { assert_eq!(email.body_snippet, Some("hello world".to_string())); } + fn spec(kind: RemoteMutationKind) -> RemoteMutationSpec { + RemoteMutationSpec { + message_id: "m-1".to_string(), + folder_id: "mb-inbox".to_string(), + folder_checkpoint: None, + kind, + } + } + + #[test] + fn build_email_set_request_copy_adds_target_without_removing_source() { + let req = JmapEngine::build_email_set_request( + "acct-1", + &spec(RemoteMutationKind::Copy { + to_folder: "mb-archive".to_string(), + }), + ); + let call = &req["methodCalls"][0]; + assert_eq!(call[0], "Email/set"); + // A copy patches a single mailboxIds/{id} pointer to true, leaving the + // rest of the message's mailbox membership untouched. + assert_eq!(call[1]["update"]["m-1"]["mailboxIds/mb-archive"], true); + } + + #[test] + fn build_email_set_request_unflag_clears_keyword_with_null() { + let req = JmapEngine::build_email_set_request( + "acct-1", + &spec(RemoteMutationKind::SetFlagged { value: false }), + ); + assert!(req["methodCalls"][0][1]["update"]["m-1"]["keywords/$flagged"].is_null()); + } + + #[test] + fn confirm_email_set_applied_requires_the_id_in_the_updated_map() { + let ok = r#"{"methodResponses":[["Email/set",{"updated":{"m-1":null}},"c1"]]}"#; + assert!(JmapEngine::confirm_email_set_applied(ok, "m-1", false).is_ok()); + + // An empty updated map is NOT success -- the server silently ignored it. + let ignored = r#"{"methodResponses":[["Email/set",{"updated":{}},"c1"]]}"#; + assert!(JmapEngine::confirm_email_set_applied(ignored, "m-1", false).is_err()); + } + #[tokio::test] async fn uncredentialed_engine_requires_credentials_to_sync() { // Without credentials there is no server to discover a session from; @@ -597,5 +749,11 @@ mod tests { .await .expect_err("credential-less sync_messages must fail"); assert!(matches!(err, MailError::AuthError(_))); + + let err = engine + .apply_mutation(&spec(RemoteMutationKind::Delete)) + .await + .expect_err("credential-less apply_mutation must fail"); + assert!(matches!(err, MailError::AuthError(_))); } } diff --git a/crates/nuncio-mail/src/lib.rs b/crates/nuncio-mail/src/lib.rs index 3d2b6aa..cc8bfc9 100644 --- a/crates/nuncio-mail/src/lib.rs +++ b/crates/nuncio-mail/src/lib.rs @@ -10,7 +10,9 @@ pub mod mock; pub mod parser; pub mod smtp; -pub use backend::{MailBackend, MessageSender, OutboundMessage}; +pub use backend::{ + MailBackend, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec, +}; pub use imap::{IdleSocketState, ImapDualSocketManager, ImapEngine}; pub use jmap::JmapEngine; pub use mock::{MockMailBackend, MockMessageSender}; diff --git a/crates/nuncio-mail/src/mock.rs b/crates/nuncio-mail/src/mock.rs index ec2d4f3..4a581d5 100644 --- a/crates/nuncio-mail/src/mock.rs +++ b/crates/nuncio-mail/src/mock.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use nuncio_core::model::{Email, Folder}; use std::sync::{Arc, Mutex}; -use crate::backend::{MailBackend, MessageSender, OutboundMessage}; +use crate::backend::{MailBackend, MessageSender, OutboundMessage, RemoteMutationSpec}; use crate::parser::MailError; /// Thread-safe mock mail backend for offline testing. @@ -19,6 +19,10 @@ pub struct MockMailBackend { since_state_calls: Arc>>>, /// The checkpoint token this mock returns from `sync_messages`. returned_state: Arc>, + /// Every [`RemoteMutationSpec`] passed to `apply_mutation`, in call order, + /// so a daemon E2E can prove the outbox executor genuinely invoked the + /// backend op (not a fabricated completion). + applied_mutations: Arc>>, } impl MockMailBackend { @@ -70,6 +74,17 @@ impl MockMailBackend { guard.push(email); } } + + /// Every [`RemoteMutationSpec`] this mock's `apply_mutation` was called + /// with, in call order. A failing configuration (`set_should_fail(true)`) + /// records nothing, so a test can prove a failed op is never counted as + /// applied. + pub fn applied_mutations(&self) -> Vec { + self.applied_mutations + .lock() + .map(|guard| guard.clone()) + .unwrap_or_default() + } } #[async_trait] @@ -135,6 +150,26 @@ impl MailBackend for MockMailBackend { .collect(); Ok((matches, returned_state)) } + + async fn apply_mutation(&self, spec: &RemoteMutationSpec) -> Result<(), MailError> { + let should_fail = self + .should_fail + .lock() + .map_err(|e| MailError::ParseFailed(e.to_string()))?; + if *should_fail { + return Err(MailError::ImapError( + "simulated remote mutation failure".to_string(), + )); + } + // Record only genuinely-applied mutations, so a test can distinguish a + // real op from a fabricated completion. + let mut applied = self + .applied_mutations + .lock() + .map_err(|e| MailError::ParseFailed(e.to_string()))?; + applied.push(spec.clone()); + Ok(()) + } } /// Deterministic mock [`MessageSender`] for offline testing of the outbound diff --git a/crates/nuncio-mail/tests/wiremock_jmap_test.rs b/crates/nuncio-mail/tests/wiremock_jmap_test.rs index fabf20e..f7773af 100644 --- a/crates/nuncio-mail/tests/wiremock_jmap_test.rs +++ b/crates/nuncio-mail/tests/wiremock_jmap_test.rs @@ -2,9 +2,9 @@ #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] -use nuncio_mail::{JmapEngine, MailBackend}; +use nuncio_mail::{JmapEngine, MailBackend, RemoteMutationKind, RemoteMutationSpec}; use serde_json::json; -use wiremock::matchers::{method, path}; +use wiremock::matchers::{body_partial_json, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; #[tokio::test] @@ -237,3 +237,144 @@ async fn jmap_engine_sync_folders_surfaces_session_discovery_failure() { .expect_err("session discovery failure must not fall back to canned folders"); assert!(err.to_string().contains("network I/O error")); } + +/// Mount JMAP session discovery so `apply_mutation` can reach the mock `/jmap/api`. +async fn mount_session(mock_server: &MockServer) { + let session_body = json!({ + "username": "james.maes@kof22.com", + "primaryAccounts": { "urn:ietf:params:jmap:mail": "acct-100" }, + "apiUrl": format!("{}/jmap/api", mock_server.uri()), + "state": "session-state-1" + }); + Mock::given(method("GET")) + .and(path("/.well-known/jmap")) + .respond_with(ResponseTemplate::new(200).set_body_json(&session_body)) + .mount(mock_server) + .await; +} + +fn jmap_engine(mock_server: &MockServer) -> JmapEngine { + JmapEngine::with_credentials( + "acct-100", + &mock_server.uri(), + "james.maes@kof22.com", + "wiremock-password", + ) +} + +fn spec(kind: RemoteMutationKind) -> RemoteMutationSpec { + RemoteMutationSpec { + message_id: "jmap-msg-1".to_string(), + folder_id: "mb-inbox".to_string(), + folder_checkpoint: None, + kind, + } +} + +#[tokio::test] +async fn jmap_apply_mutation_flag_issues_keywords_patch_and_confirms_success() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + + // Assert the exact Email/set update patch reaches the server: a $flagged + // keyword set to true on the target id. + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [[ + "Email/set", + { "update": { "jmap-msg-1": { "keywords/$flagged": true } } }, + "c1" + ]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/set", {"updated": {"jmap-msg-1": null}}, "c1"]] + }))) + .mount(&mock_server) + .await; + + let engine = jmap_engine(&mock_server); + engine + .apply_mutation(&spec(RemoteMutationKind::SetFlagged { value: true })) + .await + .expect("flag mutation applied"); +} + +#[tokio::test] +async fn jmap_apply_mutation_move_replaces_mailbox_ids() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [[ + "Email/set", + { "update": { "jmap-msg-1": { "mailboxIds": { "mb-archive": true } } } }, + "c1" + ]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/set", {"updated": {"jmap-msg-1": null}}, "c1"]] + }))) + .mount(&mock_server) + .await; + + let engine = jmap_engine(&mock_server); + engine + .apply_mutation(&spec(RemoteMutationKind::Move { + to_folder: "mb-archive".to_string(), + })) + .await + .expect("move mutation applied"); +} + +#[tokio::test] +async fn jmap_apply_mutation_delete_issues_destroy_and_confirms_success() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + + Mock::given(method("POST")) + .and(path("/jmap/api")) + .and(body_partial_json(json!({ + "methodCalls": [["Email/set", { "destroy": ["jmap-msg-1"] }, "c1"]] + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/set", {"destroyed": ["jmap-msg-1"]}, "c1"]] + }))) + .mount(&mock_server) + .await; + + let engine = jmap_engine(&mock_server); + engine + .apply_mutation(&spec(RemoteMutationKind::Delete)) + .await + .expect("delete mutation applied"); +} + +/// A server that reports the id under `notUpdated` genuinely rejected the +/// mutation; `apply_mutation` MUST surface that as an error so the outbox never +/// marks a rejected op completed. +#[tokio::test] +async fn jmap_apply_mutation_surfaces_server_rejection_as_error() { + let mock_server = MockServer::start().await; + mount_session(&mock_server).await; + + Mock::given(method("POST")) + .and(path("/jmap/api")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "methodResponses": [["Email/set", { + "updated": {}, + "notUpdated": { "jmap-msg-1": { "type": "notFound" } } + }, "c1"]] + }))) + .mount(&mock_server) + .await; + + let engine = jmap_engine(&mock_server); + let err = engine + .apply_mutation(&spec(RemoteMutationKind::SetFlagged { value: false })) + .await + .expect_err("a server-rejected mutation must not be reported as success"); + assert!(err.to_string().contains("jmap-msg-1")); +} diff --git a/crates/nuncio-store/src/vault.rs b/crates/nuncio-store/src/vault.rs index 19d8165..ca60afd 100644 --- a/crates/nuncio-store/src/vault.rs +++ b/crates/nuncio-store/src/vault.rs @@ -27,6 +27,14 @@ pub const LEDGER_KEY_ACCOUNT: &str = "ledger-hmac-key"; /// and hex-encoded for use as an `authorization: Bearer ` header value. pub const GRPC_TOKEN_ACCOUNT: &str = "grpc-bearer-token"; +/// Keyring account name for the HMAC-SHA256 key that signs outbound filter +/// `CALL WEBHOOK` payloads (the `X-Nuncio-Signature` header produced by +/// `nuncio_filter::WebhookDispatcher`). Minted on first use via +/// [`SecretManager::get_or_create_key_bytes`] and hex-encoded; a single +/// daemon-wide signing key lets a webhook receiver verify that a delivery +/// genuinely originated from this daemon. +pub const WEBHOOK_SIGNING_KEY_ACCOUNT: &str = "webhook-signing-key"; + /// Zeroized sensitive string wrapper ensuring heap bytes are wiped on drop. #[derive(Debug, Clone, PartialEq, Eq, Zeroize, ZeroizeOnDrop)] pub struct ZeroizingSecret(String); diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs index 7144aa8..94daff7 100644 --- a/crates/nunciod/src/lib.rs +++ b/crates/nunciod/src/lib.rs @@ -6,6 +6,7 @@ pub mod calendar_sync; pub mod contacts_sync; pub mod grpc; pub mod orchestrator; +pub mod outbox; pub mod send; pub mod sync; pub use orchestrator::SelfHealingSyncOrchestrator; diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index ecd4543..21b07bd 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -3,7 +3,7 @@ //! filter automation engine, outbox retries, and the multi-client gRPC API. use nuncio_core::{CoreCommand, CoreEvent, EventBus}; -use nuncio_filter::{FilterEngine, OutboxManager}; +use nuncio_filter::FilterEngine; use std::sync::Arc; #[tokio::main] @@ -98,38 +98,32 @@ async fn main() -> Result<(), Box> { ); } - // Background Outbox Worker Task + // Background Outbox Worker Task. Each tick drains the pending remote + // mutations the filter engine enqueued (move/copy/flag/unflag/delete on the + // real mail server, forward via SMTP, or a signed webhook call) and applies + // them against the resolved account's backend. A mutation is marked + // "completed" ONLY when the real operation genuinely succeeded; a transient + // failure is retried on later ticks and a permanent one (gone message, + // unknown action) fails honestly -- success is never fabricated. let db_outbox = db.clone(); + let outbox_env = Arc::new(nunciod::outbox::ProductionExecutionEnv::new( + db.clone(), + account_secrets.clone(), + )); let _outbox_task = tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); loop { interval.tick().await; - if let Ok(pending) = db_outbox.list_pending_mutations(50).await { - for item in pending { - let next_retry = item.retry_count + 1; - if next_retry > OutboxManager::MAX_RETRIES { - let _ = db_outbox - .update_mutation_status(&item.id, "failed", next_retry) - .await; - continue; - } - let backoff_ms = OutboxManager::calculate_backoff_ms(item.retry_count); - tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await; - // Outbound IMAP/JMAP remote mutation execution (actually - // applying a filter action like move/delete/flag on the - // real mail server) is not yet wired to a real mail - // server. This worker deliberately does NOT mark - // mutations "completed": doing so would fabricate - // success for work that never happened. It only bumps - // the retry counter (so an item that keeps failing still - // eventually flips to "failed" once retries are - // exhausted) and leaves the item "pending" so it is - // retried on the next poll tick once a real sender is - // wired in. - let _ = db_outbox - .update_mutation_status(&item.id, "pending", next_retry) - .await; - } + let summary = + nunciod::outbox::execute_pending_mutations(&db_outbox, outbox_env.as_ref(), 50) + .await; + if summary.completed > 0 || summary.failed > 0 { + tracing::info!( + "outbox drain: {} completed, {} retried, {} failed", + summary.completed, + summary.retried, + summary.failed + ); } } }); diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs new file mode 100644 index 0000000..e010fd3 --- /dev/null +++ b/crates/nunciod/src/outbox.rs @@ -0,0 +1,466 @@ +//! Outbox remote-mutation executor. +//! +//! The filter engine enqueues a [`nuncio_filter::PendingRemoteMutation`] for +//! every genuinely-remote action a rule matches (`MOVE`/`COPY`/`FLAG`/ +//! `UNFLAG`/`DELETE`/`FORWARD`/`CALL WEBHOOK`); `MARK READ`/`MARK UNREAD` are +//! applied to the local store inline and are never enqueued. This module +//! drains that outbox against the real world -- the resolved account's +//! [`MailBackend`] for the mailbox mutations, a [`MessageSender`] for +//! `FORWARD`, and the SSRF-hardened webhook dispatcher for `CALL WEBHOOK`. +//! +//! # The one inviolable rule +//! +//! A mutation is marked `completed` ONLY when the real remote operation +//! genuinely succeeded. A transient failure keeps the item `pending` (bumping +//! its retry count) until [`nuncio_filter::OutboxManager::MAX_RETRIES`] is +//! exceeded, at which point it flips to `failed`. A permanent failure (the +//! message no longer exists, a malformed payload, an unknown action) flips to +//! `failed` immediately -- there is nothing a retry could fix. Success is +//! never fabricated. +//! +//! # Testability +//! +//! [`execute_pending_mutations`] takes a `&dyn RemoteExecutionEnv`, so an +//! offline E2E injects mock backends/senders and a loopback-allowed webhook +//! dispatcher, proving the full sync -> filter -> outbox -> execute vertical +//! without any live network. Production wires [`ProductionExecutionEnv`], +//! which resolves real per-account engines from the store and the OS keyring. + +use async_trait::async_trait; +use nuncio_core::AccountConfig; +use nuncio_filter::{ + MutationPayload, OutboxManager, PendingRemoteMutation, ValidationOptions, WebhookDispatcher, + WebhookError, +}; +use nuncio_mail::{ + MailBackend, MailError, MessageSender, OutboundMessage, RemoteMutationKind, RemoteMutationSpec, + SmtpTransportEngine, +}; +use nuncio_store::db::{DatabaseEngine, DatabaseError}; +use nuncio_store::vault::{SecretManager, VaultError, WEBHOOK_SIGNING_KEY_ACCOUNT}; +use std::sync::Arc; +use thiserror::Error; + +/// Errors that can occur while resolving the resources needed to execute a +/// single outbox mutation. These are surfaced from a [`RemoteExecutionEnv`] +/// and treated as transient (retryable) by the executor unless it has already +/// determined the failure is permanent. +#[derive(Debug, Error)] +pub enum OutboxExecuteError { + /// No persisted account configuration matches the resolved account id. + #[error("account '{0}' is not configured")] + AccountNotFound(String), + /// Failed to read the account's credential from the secret vault. + #[error("failed to read account credential from the secret vault: {0}")] + Vault(#[from] VaultError), + /// A mail backend / transport construction or operation error. + #[error("mail error: {0}")] + Mail(#[from] MailError), + /// A database error occurred while resolving the account/message. + #[error("database error: {0}")] + Store(#[from] DatabaseError), +} + +/// Injectable seam for the resources a mutation needs: the per-account mail +/// backend and outbound sender, plus the webhook dispatcher. Production uses +/// [`ProductionExecutionEnv`]; tests inject mocks. +#[async_trait] +pub trait RemoteExecutionEnv: Send + Sync { + /// Build the [`MailBackend`] for the account that owns the target message. + async fn mail_backend( + &self, + account_id: &str, + ) -> Result, OutboxExecuteError>; + + /// Build the [`MessageSender`] used to deliver a `FORWARD`. + async fn message_sender( + &self, + account_id: &str, + ) -> Result, OutboxExecuteError>; + + /// Dispatch a signed `CALL WEBHOOK` POST, returning the HTTP status code on + /// delivery. The implementation owns the SSRF egress policy -- production + /// keeps the secure default (private targets blocked); a test env may relax + /// it to reach a loopback mock server. + async fn dispatch_webhook( + &self, + url: &str, + rule_id: &str, + message_id: &str, + subject: &str, + sender: &str, + ) -> Result; +} + +/// The disposition of a single mutation attempt. +enum Disposition { + /// The real operation genuinely succeeded; mark `completed`. + Completed, + /// A transient failure; keep retrying until `MAX_RETRIES` is exceeded. + Retry(String), + /// A permanent failure nothing could retry away; mark `failed` now. + Permanent(String), +} + +/// Tally of what a drain pass did, for logging and test assertions. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct ExecutionSummary { + /// Mutations marked `completed` on genuine success. + pub completed: usize, + /// Mutations left `pending` for a later retry. + pub retried: usize, + /// Mutations flipped to `failed` (permanent, or retries exhausted). + pub failed: usize, +} + +/// Drain up to `limit` pending mutations, executing each against `env` and +/// persisting the honest outcome. Never marks a mutation `completed` unless the +/// real operation succeeded. Returns a tally of the pass. +pub async fn execute_pending_mutations( + db: &DatabaseEngine, + env: &dyn RemoteExecutionEnv, + limit: usize, +) -> ExecutionSummary { + let pending = match db.list_pending_mutations(limit).await { + Ok(items) => items, + Err(e) => { + tracing::warn!("outbox: failed to list pending mutations: {e}"); + return ExecutionSummary::default(); + } + }; + + let mut summary = ExecutionSummary::default(); + for item in pending { + let disposition = execute_one(db, env, &item).await; + let next_retry = item.retry_count + 1; + let (status, retry, tally): (&str, i32, fn(&mut ExecutionSummary)) = match disposition { + Disposition::Completed => ("completed", item.retry_count, |s| s.completed += 1), + Disposition::Permanent(reason) => { + tracing::warn!( + mutation_id = %item.id, + "outbox: permanently failing mutation: {reason}" + ); + ("failed", next_retry, |s| s.failed += 1) + } + Disposition::Retry(reason) if next_retry > OutboxManager::MAX_RETRIES => { + tracing::warn!( + mutation_id = %item.id, + "outbox: mutation failed after exhausting retries: {reason}" + ); + ("failed", next_retry, |s| s.failed += 1) + } + Disposition::Retry(reason) => { + tracing::debug!( + mutation_id = %item.id, + "outbox: mutation will be retried: {reason}" + ); + ("pending", next_retry, |s| s.retried += 1) + } + }; + + if let Err(e) = db.update_mutation_status(&item.id, status, retry).await { + // A status-write failure must not crash the worker; the item stays + // in its prior state and is reconsidered on the next poll. + tracing::warn!(mutation_id = %item.id, "outbox: failed to persist status: {e}"); + } else { + tally(&mut summary); + } + } + summary +} + +/// Execute a single mutation and report its honest disposition. Performs no +/// status writes itself -- the caller persists the outcome. +async fn execute_one( + db: &DatabaseEngine, + env: &dyn RemoteExecutionEnv, + mutation: &PendingRemoteMutation, +) -> Disposition { + // Resolve the message to recover its account and source folder. A message + // that no longer exists can never be acted on -- fail permanently. + let email = match db.get_message(&mutation.message_id).await { + Ok(email) => email, + Err(e) if e.is_not_found() => { + return Disposition::Permanent(format!( + "message '{}' no longer exists in the store", + mutation.message_id + )); + } + Err(e) => { + return Disposition::Retry(format!("failed to load message: {e}")); + } + }; + + let payload = match serde_json::from_str::(&mutation.payload) { + Ok(payload) => payload, + Err(e) => { + return Disposition::Permanent(format!("malformed mutation payload: {e}")); + } + }; + + match mutation.mutation_type.as_str() { + "MOVE" | "COPY" | "FLAG" | "UNFLAG" | "DELETE" => { + dispatch_mailbox_mutation(db, env, mutation, &email, &payload).await + } + "FORWARD" => dispatch_forward(db, env, &email, &payload).await, + "WEBHOOK" => dispatch_webhook(env, mutation, &email, &payload).await, + other => Disposition::Permanent(format!("unknown mutation type '{other}'")), + } +} + +/// Execute a mailbox-affecting mutation (move/copy/flag/unflag/delete) against +/// the account's mail backend, threading the source folder's stored checkpoint +/// so the backend can enforce its UIDVALIDITY guard. +async fn dispatch_mailbox_mutation( + db: &DatabaseEngine, + env: &dyn RemoteExecutionEnv, + mutation: &PendingRemoteMutation, + email: &nuncio_core::model::Email, + payload: &MutationPayload, +) -> Disposition { + let kind = match mutation.mutation_type.as_str() { + "FLAG" => RemoteMutationKind::SetFlagged { value: true }, + "UNFLAG" => RemoteMutationKind::SetFlagged { value: false }, + "DELETE" => RemoteMutationKind::Delete, + "MOVE" => match &payload.target { + Some(folder) => RemoteMutationKind::Move { + to_folder: folder.clone(), + }, + None => { + return Disposition::Permanent("MOVE mutation has no target folder".to_string()) + } + }, + "COPY" => match &payload.target { + Some(folder) => RemoteMutationKind::Copy { + to_folder: folder.clone(), + }, + None => { + return Disposition::Permanent("COPY mutation has no target folder".to_string()) + } + }, + other => return Disposition::Permanent(format!("unexpected mailbox mutation '{other}'")), + }; + + let folder_checkpoint = match db + .get_folder_sync_state(&email.account_id, &email.folder_id) + .await + { + Ok(checkpoint) => checkpoint, + Err(e) => return Disposition::Retry(format!("failed to read folder checkpoint: {e}")), + }; + + let backend = match env.mail_backend(&email.account_id).await { + Ok(backend) => backend, + Err(e) => return Disposition::Retry(format!("failed to build mail backend: {e}")), + }; + + let spec = RemoteMutationSpec { + message_id: email.id.clone(), + folder_id: email.folder_id.clone(), + folder_checkpoint, + kind, + }; + + match backend.apply_mutation(&spec).await { + Ok(()) => Disposition::Completed, + Err(e) => Disposition::Retry(format!("remote mutation failed: {e}")), + } +} + +/// Compose and send a forward of `email` to the payload's target address via +/// the account's outbound transport. +async fn dispatch_forward( + db: &DatabaseEngine, + env: &dyn RemoteExecutionEnv, + email: &nuncio_core::model::Email, + payload: &MutationPayload, +) -> Disposition { + let Some(to) = payload.target.clone() else { + return Disposition::Permanent("FORWARD mutation has no target address".to_string()); + }; + + // The From: is the forwarding account's own address, resolved from its + // persisted configuration rather than trusting the stored envelope. + let from = match find_account_config(db, &email.account_id).await { + Ok(config) => config.email_address, + Err(OutboxExecuteError::AccountNotFound(id)) => { + return Disposition::Permanent(format!("forwarding account '{id}' is not configured")); + } + Err(e) => return Disposition::Retry(format!("failed to resolve forwarding account: {e}")), + }; + + let sender = match env.message_sender(&email.account_id).await { + Ok(sender) => sender, + Err(e) => return Disposition::Retry(format!("failed to build message sender: {e}")), + }; + + let outbound = OutboundMessage { + from, + to, + cc: None, + subject: format!("Fwd: {}", email.subject), + body_plain: email.body_plain.clone(), + body_html: email.body_html.clone(), + attachments: email.attachments.clone(), + }; + + match sender.send(&outbound).await { + Ok(()) => Disposition::Completed, + Err(e) => Disposition::Retry(format!("forward send failed: {e}")), + } +} + +/// Deliver a signed webhook for the matched message. A 2xx/3xx response is a +/// genuine delivery; a 4xx/5xx is treated as a retryable failure so a transient +/// receiver outage does not silently drop the notification. +async fn dispatch_webhook( + env: &dyn RemoteExecutionEnv, + mutation: &PendingRemoteMutation, + email: &nuncio_core::model::Email, + payload: &MutationPayload, +) -> Disposition { + let Some(url) = payload.target.as_deref() else { + return Disposition::Permanent("WEBHOOK mutation has no target URL".to_string()); + }; + + match env + .dispatch_webhook( + url, + &mutation.rule_id, + &mutation.message_id, + &email.subject, + &email.sender, + ) + .await + { + Ok(status) if (200..400).contains(&status) => Disposition::Completed, + Ok(status) => Disposition::Retry(format!("webhook returned HTTP {status}")), + // A security-policy violation (blocked SSRF target) is permanent: the + // URL will never become allowable on retry. + Err(WebhookError::SecurityViolation(reason)) => { + Disposition::Permanent(format!("webhook blocked by security policy: {reason}")) + } + Err(e) => Disposition::Retry(format!("webhook dispatch failed: {e}")), + } +} + +/// Look up an account's persisted configuration by id. +async fn find_account_config( + db: &DatabaseEngine, + account_id: &str, +) -> Result { + db.list_accounts() + .await? + .into_iter() + .find(|a| a.id == account_id) + .ok_or_else(|| OutboxExecuteError::AccountNotFound(account_id.to_string())) +} + +/// Production [`RemoteExecutionEnv`]: resolves real per-account IMAP/JMAP mail +/// backends and SMTP senders from the store plus the OS keyring, and lazily +/// builds a single SSRF-hardened [`WebhookDispatcher`] keyed by a daemon-wide +/// signing secret provisioned from the vault. +/// +/// The webhook signing key is provisioned lazily -- only the first time a +/// `CALL WEBHOOK` mutation is actually dispatched -- so a vault issue with that +/// one key can never disable the mailbox-mutation and forward paths, which do +/// not need it. +pub struct ProductionExecutionEnv { + db: Arc, + secrets: Arc, + webhook: std::sync::OnceLock, +} + +impl ProductionExecutionEnv { + /// Build the production environment. Infallible: no vault access happens + /// here, so the outbox worker can always start; the webhook signing key is + /// provisioned on first webhook dispatch instead (see the struct doc). + pub fn new(db: Arc, secrets: Arc) -> Self { + Self { + db, + secrets, + webhook: std::sync::OnceLock::new(), + } + } + + async fn account(&self, account_id: &str) -> Result { + find_account_config(&self.db, account_id).await + } + + /// Return the lazily-provisioned webhook dispatcher, minting (or loading) + /// the daemon-wide signing key from the vault on first use. Fails closed if + /// the vault is unavailable, so a `CALL WEBHOOK` is never dispatched + /// unsigned -- and, being lazy, that failure is isolated to webhook + /// dispatch and never disables the other mutation paths. + fn webhook_dispatcher(&self) -> Result<&WebhookDispatcher, WebhookError> { + if let Some(dispatcher) = self.webhook.get() { + return Ok(dispatcher); + } + let key_bytes = self + .secrets + .get_or_create_key_bytes(WEBHOOK_SIGNING_KEY_ACCOUNT, 32) + .map_err(|e| { + WebhookError::SigningError(format!( + "failed to provision webhook signing key from the vault: {e}" + )) + })?; + // A concurrent caller may win the race to set it; either way the stored + // dispatcher is authoritative. + let _ = self + .webhook + .set(WebhookDispatcher::new(hex::encode(key_bytes))); + self.webhook.get().ok_or_else(|| { + WebhookError::SigningError("webhook dispatcher unavailable after provisioning".into()) + }) + } +} + +#[async_trait] +impl RemoteExecutionEnv for ProductionExecutionEnv { + async fn mail_backend( + &self, + account_id: &str, + ) -> Result, OutboxExecuteError> { + let config = self.account(account_id).await?; + let password = self.secrets.get_secret(&config.keyring_secret_key)?; + Ok(crate::sync::build_mail_backend(&config, &password)) + } + + async fn message_sender( + &self, + account_id: &str, + ) -> Result, OutboxExecuteError> { + let config = self.account(account_id).await?; + let password = self.secrets.get_secret(&config.keyring_secret_key)?; + let engine = SmtpTransportEngine::new( + &config.smtp_host, + config.smtp_port, + config.smtp_tls_mode, + &config.email_address, + &password, + )?; + Ok(Box::new(engine)) + } + + async fn dispatch_webhook( + &self, + url: &str, + rule_id: &str, + message_id: &str, + subject: &str, + sender: &str, + ) -> Result { + // Production keeps the secure default egress policy (private/loopback + // targets are rejected). + self.webhook_dispatcher()? + .dispatch_with_options( + url, + rule_id, + message_id, + subject, + sender, + &ValidationOptions::default(), + ) + .await + } +} diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index a3c4522..02265a1 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -339,7 +339,7 @@ async fn find_account( /// keyring by the caller). This is the ONLY place a concrete protocol engine /// is chosen; every other function in this module only ever sees `&dyn /// MailBackend`. -fn build_mail_backend(config: &AccountConfig, password: &str) -> Box { +pub(crate) fn build_mail_backend(config: &AccountConfig, password: &str) -> Box { match config.protocol { AccountProtocol::ImapSmtp => Box::new(ImapEngine::with_credentials( &config.id, diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs new file mode 100644 index 0000000..855a4c9 --- /dev/null +++ b/crates/nunciod/tests/outbox_executor_e2e_test.rs @@ -0,0 +1,329 @@ +//! Offline end-to-end test for the outbox remote-mutation executor. +//! +//! Proves the full `sync -> filter -> outbox -> execute` vertical without any +//! live network: a matched filter rule enqueues a real +//! `PendingRemoteMutation`, and `execute_pending_mutations` drains it against +//! injected mocks -- a `MockMailBackend` that records each applied mutation, a +//! `MockMessageSender` for `FORWARD`, and a loopback-allowed webhook dispatcher +//! hitting a wiremock server. The safety-critical property under test is that a +//! mutation is marked `completed` ONLY when the real op genuinely succeeded: a +//! failing backend never records an op and never completes the item. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use async_trait::async_trait; +use nuncio_core::model::Email; +use nuncio_filter::{FilterEngine, NsqlParser, ValidationOptions, WebhookDispatcher, WebhookError}; +use nuncio_mail::{ + MailBackend, MessageSender, MockMailBackend, MockMessageSender, RemoteMutationKind, +}; +use nuncio_store::db::DatabaseEngine; +use nunciod::outbox::{execute_pending_mutations, RemoteExecutionEnv}; +use nunciod::sync::apply_filter_actions; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const ACCOUNT_ID: &str = "acct-outbox-e2e"; +const FOLDER_ID: &str = "INBOX"; +const CHECKPOINT: &str = "1:105"; + +/// Test [`RemoteExecutionEnv`] injecting recording mocks and a webhook +/// dispatcher whose SSRF egress policy is relaxed so it can reach a loopback +/// wiremock server (production keeps the secure default). +struct MockEnv { + backend: MockMailBackend, + sender: MockMessageSender, + webhook: WebhookDispatcher, +} + +impl MockEnv { + fn new() -> Self { + Self { + backend: MockMailBackend::new(), + sender: MockMessageSender::new(), + webhook: WebhookDispatcher::new("test-signing-key"), + } + } +} + +#[async_trait] +impl RemoteExecutionEnv for MockEnv { + async fn mail_backend( + &self, + _account_id: &str, + ) -> Result, nunciod::outbox::OutboxExecuteError> { + // Clones share the same Arc-backed recording storage, so the returned + // boxed backend records into the `self.backend` the test inspects. + Ok(Box::new(self.backend.clone())) + } + + async fn message_sender( + &self, + _account_id: &str, + ) -> Result, nunciod::outbox::OutboxExecuteError> { + Ok(Box::new(self.sender.clone())) + } + + async fn dispatch_webhook( + &self, + url: &str, + rule_id: &str, + message_id: &str, + subject: &str, + sender: &str, + ) -> Result { + let opts = ValidationOptions { + block_private_webhooks: false, + ..Default::default() + }; + self.webhook + .dispatch_with_options(url, rule_id, message_id, subject, sender, &opts) + .await + } +} + +fn sample_account() -> nuncio_core::AccountConfig { + nuncio_core::AccountConfig { + id: ACCOUNT_ID.to_string(), + name: "Outbox E2E Account".to_string(), + email_address: "owner@nuncio.mx".to_string(), + protocol: nuncio_core::AccountProtocol::ImapSmtp, + server_host: "imap.nuncio.mx".to_string(), + server_port: 993, + smtp_host: "smtp.nuncio.mx".to_string(), + smtp_port: 465, + use_tls: true, + imap_tls_mode: nuncio_core::TlsMode::ImplicitTls, + smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls, + keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"), + sync_interval_secs: 60, + } +} + +fn sample_email(subject: &str) -> Email { + Email { + id: "imap-uid-42".to_string(), + account_id: ACCOUNT_ID.to_string(), + folder_id: FOLDER_ID.to_string(), + subject: subject.to_string(), + sender: "alice@nuncio.mx".to_string(), + recipient: "owner@nuncio.mx".to_string(), + received_at: 1_700_000_000, + read: false, + body_plain: Some("original body".to_string()), + body_html: None, + attachments: Vec::new(), + } +} + +/// Seed the account, source-folder checkpoint, and a matching message, then run +/// the given rule's actions through `apply_filter_actions` so the outbox is +/// populated exactly as a real sync would populate it. +async fn seed_and_enqueue(db: &DatabaseEngine, rule_nsql: &str, subject: &str) { + db.save_account(&sample_account()) + .await + .expect("save account"); + db.save_folder_sync_state(ACCOUNT_ID, FOLDER_ID, CHECKPOINT) + .await + .expect("save folder checkpoint"); + + let email = sample_email(subject); + db.save_email(&email).await.expect("save email"); + + let rule = NsqlParser::parse_rule("Outbox Rule", 1, rule_nsql).expect("parse rule"); + let engine = FilterEngine::new(vec![rule]).expect("compile rule"); + apply_filter_actions(db, &engine, &email).await; + + let pending = db.list_pending_mutations(10).await.expect("list pending"); + assert_eq!( + pending.len(), + 1, + "the rule must enqueue exactly one mutation" + ); +} + +#[tokio::test] +async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_checkpoint() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'", + "Please Archive", + ) + .await; + + let env = MockEnv::new(); + let summary = execute_pending_mutations(&db, &env, 50).await; + + assert_eq!(summary.completed, 1); + assert_eq!(summary.failed, 0); + + let applied = env.backend.applied_mutations(); + assert_eq!(applied.len(), 1, "the backend op must genuinely run"); + assert_eq!(applied[0].message_id, "imap-uid-42"); + assert_eq!(applied[0].folder_id, FOLDER_ID); + // The source folder's stored checkpoint (carrying UIDVALIDITY) is threaded + // through so the backend can enforce its guard. + assert_eq!(applied[0].folder_checkpoint.as_deref(), Some(CHECKPOINT)); + assert_eq!( + applied[0].kind, + RemoteMutationKind::Move { + to_folder: "Archive".to_string() + } + ); + + // A completed mutation is no longer pending. + assert!(db.list_pending_mutations(10).await.unwrap().is_empty()); +} + +#[tokio::test] +async fn flag_rule_executes_as_set_flagged() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Star' ACTION FLAG", + "Star this", + ) + .await; + + let env = MockEnv::new(); + let summary = execute_pending_mutations(&db, &env, 50).await; + + assert_eq!(summary.completed, 1); + let applied = env.backend.applied_mutations(); + assert_eq!(applied.len(), 1); + assert_eq!( + applied[0].kind, + RemoteMutationKind::SetFlagged { value: true } + ); +} + +#[tokio::test] +async fn delete_rule_executes_as_delete() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Trash' ACTION DELETE", + "Trash me", + ) + .await; + + let env = MockEnv::new(); + let summary = execute_pending_mutations(&db, &env, 50).await; + + assert_eq!(summary.completed, 1); + let applied = env.backend.applied_mutations(); + assert_eq!(applied.len(), 1); + assert_eq!(applied[0].kind, RemoteMutationKind::Delete); +} + +#[tokio::test] +async fn a_failing_backend_op_is_never_marked_completed() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'", + "Please Archive", + ) + .await; + + let env = MockEnv::new(); + env.backend.set_should_fail(true); + let summary = execute_pending_mutations(&db, &env, 50).await; + + // The op failed: nothing recorded, nothing completed, and the item is kept + // for retry (still pending, with a bumped retry count) -- never faked. + assert_eq!(summary.completed, 0); + assert_eq!(summary.retried, 1); + assert!(env.backend.applied_mutations().is_empty()); + + let pending = db.list_pending_mutations(10).await.unwrap(); + assert_eq!(pending.len(), 1, "a failed op must remain pending"); + assert_eq!(pending[0].retry_count, 1, "the retry count must be bumped"); +} + +#[tokio::test] +async fn forward_rule_sends_via_the_message_sender() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + seed_and_enqueue( + &db, + "WHERE subject CONTAINS 'Forward' ACTION FORWARD TO 'boss@nuncio.mx'", + "Please Forward", + ) + .await; + + let env = MockEnv::new(); + let summary = execute_pending_mutations(&db, &env, 50).await; + + assert_eq!(summary.completed, 1); + let sent = env.sender.sent_messages(); + assert_eq!(sent.len(), 1, "the forward must genuinely be sent"); + assert_eq!(sent[0].to, "boss@nuncio.mx"); + // The From: is the forwarding account's own configured address. + assert_eq!(sent[0].from, "owner@nuncio.mx"); + assert!(sent[0].subject.contains("Please Forward")); +} + +#[tokio::test] +async fn webhook_rule_dispatches_a_signed_post_to_the_receiver() { + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/hooks/nuncio")) + .and(wiremock::matchers::header( + "Content-Type", + "application/json", + )) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&mock_server) + .await; + + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + let url = format!("{}/hooks/nuncio", mock_server.uri()); + seed_and_enqueue( + &db, + &format!("WHERE subject CONTAINS 'Ping' ACTION CALL WEBHOOK '{url}'"), + "Ping me", + ) + .await; + + let env = MockEnv::new(); + let summary = execute_pending_mutations(&db, &env, 50).await; + + // Completion is genuine: wiremock's `.expect(1)` verifies the POST actually + // arrived when the server drops. + assert_eq!(summary.completed, 1); + assert!(db.list_pending_mutations(10).await.unwrap().is_empty()); +} + +#[tokio::test] +async fn a_mutation_for_a_vanished_message_fails_permanently() { + let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db"); + db.save_account(&sample_account()) + .await + .expect("save account"); + + // Enqueue a mutation whose target message was never persisted. + let mutation = nuncio_filter::OutboxManager::create_mutation( + "rule-x", + "imap-uid-999", + "MOVE", + Some("Archive".to_string()), + ); + db.save_pending_mutation(&mutation) + .await + .expect("save mutation"); + + let env = MockEnv::new(); + let summary = execute_pending_mutations(&db, &env, 50).await; + + // Cannot act on a message that no longer exists -- fail honestly, never + // fabricate success, and never touch the backend. + assert_eq!(summary.failed, 1); + assert_eq!(summary.completed, 0); + assert!(env.backend.applied_mutations().is_empty()); + assert!( + db.list_pending_mutations(10).await.unwrap().is_empty(), + "a permanently-failed mutation is no longer pending" + ); +}