diff --git a/Cargo.lock b/Cargo.lock index ad178f6..b0ac32b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2604,7 +2604,6 @@ dependencies = [ "thiserror 1.0.69", "tokio", "tracing", - "tracing-subscriber", "wiremock", "zeroize", ] diff --git a/crates/nuncio-core/src/config.rs b/crates/nuncio-core/src/config.rs index 8c5fc99..245ad11 100644 --- a/crates/nuncio-core/src/config.rs +++ b/crates/nuncio-core/src/config.rs @@ -154,6 +154,21 @@ pub struct AccountConfig { pub keyring_secret_key: String, /// Background sync polling interval in seconds (minimum 10s). pub sync_interval_secs: u64, + /// Whether this daemon executes filter rules for this account. + /// + /// **Defaults to `false`**, which is the conservative answer to a problem + /// the engine cannot solve on its own. Every daemon syncing an account + /// decides independently that a message is new, so every daemon fires the + /// same rules against it. `MOVE`/`COPY`/`FLAG` survive that -- ten moves + /// converge on one outcome -- but `FORWARD` and `CALL WEBHOOK` land on a + /// third party with no shared state to compare against, so N daemons send + /// N forwards and N webhook calls. + /// + /// Enabling this on exactly one daemon per account makes filter execution + /// single-owner. It is a deliberate, documented limitation rather than a + /// silent duplication bug; a server-side claim is the eventual fix where + /// the server supports one. + pub filters_enabled: bool, /// The account's single transport. pub transport: Transport, } @@ -343,6 +358,7 @@ mod tests { email_address: "user@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-123".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::ImapSmtp(valid_imap_smtp_transport()), } } @@ -354,6 +370,7 @@ mod tests { email_address: "user@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-jmap-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::Jmap(JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -367,6 +384,7 @@ mod tests { email_address: "user@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-cal-1".to_string(), sync_interval_secs: 300, + filters_enabled: false, transport: Transport::Dav(DavTransport { collection_url: "https://caldav.example.com/calendars/user/work/".to_string(), }), diff --git a/crates/nuncio-filter/src/webhook.rs b/crates/nuncio-filter/src/webhook.rs index 0a18320..cadd6e1 100644 --- a/crates/nuncio-filter/src/webhook.rs +++ b/crates/nuncio-filter/src/webhook.rs @@ -189,9 +189,26 @@ impl WebhookDispatcher { mac.update(format!("{timestamp}.{payload_str}").as_bytes()); let signature = hex::encode(mac.finalize().into_bytes()); + // Stable across every retry of this rule/message pair, so a + // receiver can discard a redelivery caused by a lost response + // rather than acting twice. It is NOT yet stable across daemons: + // `rule_id` is random per daemon, so two daemons executing the + // same rule produce different keys. Single-owner filter execution + // is what prevents that case today; a rule key derived from the + // normalized NSQL would make it dedupable directly. + let idempotency_key = { + let mut hasher = ::new(); + for field in [rule_id, message_id] { + sha2::Digest::update(&mut hasher, (field.len() as u64).to_le_bytes()); + sha2::Digest::update(&mut hasher, field.as_bytes()); + } + hex::encode(sha2::Digest::finalize(hasher)) + }; + let send_result = client .post(url) .header("Content-Type", "application/json") + .header("Idempotency-Key", &idempotency_key) .header( "X-Nuncio-Signature", format!("t={timestamp},v1={signature}"), diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 10f874d..9aa7253 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -858,7 +858,13 @@ impl DatabaseEngine { smtp_port INTEGER, imap_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls', smtp_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls', - collection_url TEXT + collection_url TEXT, + -- Defaults to 0 (off), including for accounts written before + -- the column existed. Filter execution is not safe to run on + -- more than one daemon per account, and defaulting a shared + -- account to "every daemon forwards" would be the wrong way + -- to be wrong. + filters_enabled INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS filter_rules ( @@ -1184,8 +1190,8 @@ impl DatabaseEngine { sqlx::query( r#" INSERT OR REPLACE INTO accounts - (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url, filters_enabled) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#, ) .bind(&config.id) @@ -1208,6 +1214,7 @@ impl DatabaseEngine { .bind(tls_mode_to_db(imap_tls_mode)) .bind(tls_mode_to_db(smtp_tls_mode)) .bind(&collection_url) + .bind(i64::from(config.filters_enabled)) .execute(&self.pool) .await .map_err(DatabaseError::Query)?; @@ -1253,9 +1260,10 @@ impl DatabaseEngine { String, String, Option, + i64, )> = sqlx::query_as( r#" - SELECT id, name, email_address, protocol, server_host, server_port, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url + SELECT id, name, email_address, protocol, server_host, server_port, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url, filters_enabled FROM accounts "#, ) @@ -1280,6 +1288,7 @@ impl DatabaseEngine { imap_tls_mode, smtp_tls_mode, collection_url, + filters_enabled, )| { let protocol = serde_json::from_str(&protocol_str) .unwrap_or(nuncio_core::AccountProtocol::ImapSmtp); @@ -1326,6 +1335,7 @@ impl DatabaseEngine { email_address, keyring_secret_key, sync_interval_secs: sync_interval_secs as u64, + filters_enabled: filters_enabled != 0, transport, } }, @@ -3292,6 +3302,7 @@ mod tests { email_address: "contended@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-contended-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -3921,6 +3932,7 @@ mod tests { email_address: "work@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-test-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -3965,6 +3977,7 @@ mod tests { email_address: "mixed@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-tls-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 143, @@ -4007,6 +4020,7 @@ mod tests { email_address: "cal@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-caldav-store-1".to_string(), sync_interval_secs: 300, + filters_enabled: false, transport: nuncio_core::Transport::Dav(nuncio_core::DavTransport { collection_url: "https://dav.example.com/calendars/user/work/".to_string(), }), @@ -4039,6 +4053,7 @@ mod tests { email_address: "imap@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-mix-imap".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 143, @@ -4054,6 +4069,7 @@ mod tests { email_address: "jmap@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-mix-jmap".to_string(), sync_interval_secs: 120, + filters_enabled: false, transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -4064,6 +4080,7 @@ mod tests { email_address: "dav@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-mix-dav".to_string(), sync_interval_secs: 300, + filters_enabled: false, transport: nuncio_core::Transport::Dav(nuncio_core::DavTransport { collection_url: "https://dav.example.com/cal/".to_string(), }), @@ -4103,6 +4120,7 @@ mod tests { email_address: "del@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-del-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -4631,6 +4649,7 @@ mod tests { email_address: "reset@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-reset".to_string(), sync_interval_secs: 60, + filters_enabled: true, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -4678,6 +4697,13 @@ mod tests { let accounts = engine.list_accounts().await.expect("list accounts"); assert_eq!(accounts.len(), 1, "accounts must be preserved"); assert_eq!(accounts[0].id, "acct-reset"); + // Including which daemon owns filter execution: a rebuild that dropped + // it would silently switch the owner off, or -- worse, if it defaulted + // the other way -- switch a bystander on. + assert!( + accounts[0].filters_enabled, + "filter ownership must survive an identity rebuild" + ); // The version is recorded, so the next open is a no-op. let (version,): (i64,) = sqlx::query_as("PRAGMA user_version") diff --git a/crates/nuncio-store/src/recovery.rs b/crates/nuncio-store/src/recovery.rs index 177793b..64d82a2 100644 --- a/crates/nuncio-store/src/recovery.rs +++ b/crates/nuncio-store/src/recovery.rs @@ -453,6 +453,13 @@ impl SqliteRecoveryEngine { email_address, keyring_secret_key, sync_interval_secs: sync_interval_secs as u64, + // Salvage deliberately does not recover this flag. + // The column may be among what the corruption took, + // and re-enabling filter execution on a guess would + // start forwarding mail from a daemon the user never + // chose as the owner. Off is recoverable by one + // setting; wrongly on is not recoverable at all. + filters_enabled: false, transport, } }, @@ -618,6 +625,7 @@ mod tests { email_address: "work@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-test-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -679,6 +687,7 @@ mod tests { email_address: "salvage@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-salvage-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -816,6 +825,7 @@ mod tests { email_address: "starttls@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-starttls-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 143, diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 1a9b8be..ce0986a 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -490,6 +490,7 @@ fn map_account_config_from_proto( email_address: config.email_address, keyring_secret_key: config.keyring_secret_key, sync_interval_secs: nuncio_proto::time::duration_to_secs(&sync_interval), + filters_enabled: false, transport, }) } @@ -2069,12 +2070,29 @@ impl Mail for MailGrpcService { } let synced = if let Some(backend) = &self.overrides.mail_backend { + // Filter execution is owned by whichever daemon the account opts + // in (see `AccountConfig::filters_enabled`). Resolve it from the + // stored account rather than assuming: an injected backend must + // take the same path production does, or the gate would only ever + // be exercised in production. + let filters_enabled = match &account_id { + Some(id) => self + .db + .list_accounts() + .await + .map_err(|e| Status::internal(format!("sync failed: {e}")))? + .into_iter() + .find(|a| &a.id == id) + .is_some_and(|a| a.filters_enabled), + None => false, + }; crate::sync::sync_with_backend( &self.db, &self.event_bus, backend.as_ref(), &self.filter_engine, account_id, + filters_enabled, ) .await .map_err(|e| Status::internal(format!("sync failed: {e}")))? @@ -2612,7 +2630,34 @@ impl Filters for FiltersGrpcService { req.chunk_size as usize }; - tracing::info!(chunk_size, "Filters: triage rescan started"); + // Triage produces exactly the side effects a live sync produces, so it + // is bound by the same single-owner rule: only accounts this daemon + // owns filter execution for may be acted on. Resolved once here rather + // than per message. + let owning_accounts: std::collections::HashSet = self + .db + .list_accounts() + .await + .map_err(|e| Status::internal(format!("failed to read accounts for triage: {e}")))? + .into_iter() + .filter(|a| a.filters_enabled) + .map(|a| a.id) + .collect(); + + if owning_accounts.is_empty() { + // An honest refusal beats streaming a scan that reports progress + // and applies nothing, which would read as "no rules matched". + return Err(Status::failed_precondition( + "no account on this daemon has filter execution enabled; \ + enable filters_enabled on exactly one daemon per account", + )); + } + + tracing::info!( + chunk_size, + owning_accounts = owning_accounts.len(), + "Filters: triage rescan started" + ); let (tx, rx) = tokio::sync::mpsc::channel::>(16); let db = self.db.clone(); @@ -2643,27 +2688,36 @@ impl Filters for FiltersGrpcService { let is_last_page = batch.len() < chunk_size; for email in &batch { - // Rules can ask where a message is, so a retroactive - // pass evaluates each mailbox the message occupies -- - // the same fan-out live sync does. The fire-once claim - // inside `apply_filter_actions` keeps a message in two - // folders from acting twice. - let placements = match db.placements_of(&email.id).await { - Ok(placements) => placements, - Err(e) => { - let _ = tx - .send(Err(Status::internal(format!( - "failed to read placements during triage: {e}" - )))) - .await; - return; - } - }; + // Messages belonging to an account this daemon does not + // own are counted as scanned but never acted on, so the + // progress total still reflects the whole store. A + // non-owner does not even look up the placements: it has + // nothing to do with them, and leaving the fire-once + // claim untouched keeps it available to the owner. let mut actions_for_email = 0usize; - for placement in &placements { - actions_for_email += - crate::sync::apply_filter_actions(&db, &engine, email, placement) - .await; + if owning_accounts.contains(&email.account_id) { + // Rules can ask where a message is, so a retroactive + // pass evaluates each mailbox the message occupies -- + // the same fan-out live sync does. The fire-once claim + // inside `apply_filter_actions` keeps a message in two + // folders from acting twice. + let placements = match db.placements_of(&email.id).await { + Ok(placements) => placements, + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!( + "failed to read placements during triage: {e}" + )))) + .await; + return; + } + }; + for placement in &placements { + actions_for_email += crate::sync::apply_filter_actions( + &db, &engine, email, placement, + ) + .await; + } } scanned += 1; applied += actions_for_email as u64; @@ -3893,6 +3947,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -4654,6 +4709,7 @@ mod tests { email_address: "rotate@nuncio.mx".to_string(), keyring_secret_key: key.to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -4697,6 +4753,7 @@ mod tests { email_address: "rotate2@nuncio.mx".to_string(), keyring_secret_key: key.to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -4779,6 +4836,9 @@ mod tests { email_address: format!("{}@nuncio.mx", placement.account_id), keyring_secret_key: format!("nuncio/{}", placement.account_id), sync_interval_secs: 60, + // This daemon owns filter execution for the seeded account, so + // triage tests exercise the acting path rather than the refusal. + filters_enabled: true, transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -5597,6 +5657,7 @@ mod tests { email_address: "sender@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-send-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -5770,6 +5831,7 @@ mod tests { email_address: "sender@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-injected-send-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -5886,6 +5948,7 @@ mod tests { email_address: "sender@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-injected-send-fail-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nunciod/src/orchestrator.rs b/crates/nunciod/src/orchestrator.rs index 24af9aa..9b5bf5d 100644 --- a/crates/nunciod/src/orchestrator.rs +++ b/crates/nunciod/src/orchestrator.rs @@ -154,6 +154,7 @@ mod tests { email_address: "orch@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-orch-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nunciod/src/send.rs b/crates/nunciod/src/send.rs index de07d15..27a24d2 100644 --- a/crates/nunciod/src/send.rs +++ b/crates/nunciod/src/send.rs @@ -292,6 +292,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index 13b1309..3b9dbe4 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -289,6 +289,7 @@ async fn fetch_and_persist( backend: &dyn MailBackend, filter_engine: &FilterEngine, account_id: Option<&str>, + filters_enabled: bool, ) -> Result { let folders = backend.sync_folders().await?; let mut synced = 0usize; @@ -326,7 +327,14 @@ async fn fetch_and_persist( let is_new = !already_present.contains(&key) && seen_this_pass.insert(key); db.save_email_at(&message.email, message.source, &message.placement) .await?; - if is_new { + // Filter execution is single-owner per account (see + // `AccountConfig::filters_enabled`). A daemon that is not the owner + // still syncs and stores the message; it just does not act on it, + // because the side-effecting actions -- FORWARD, CALL WEBHOOK -- + // are not idempotent across daemons. The gate sits outside the + // fire-once claim on purpose: a non-owner must not consume the + // claim it is declining to act on. + if is_new && filters_enabled { apply_filter_actions(db, filter_engine, &message.email, &message.placement).await; } synced += 1; @@ -407,6 +415,7 @@ pub async fn sync_with_backend( backend: &dyn MailBackend, filter_engine: &FilterEngine, account_id: Option, + filters_enabled: bool, ) -> Result { match &account_id { Some(id) => event_bus.process_command(CoreCommand::SyncAccount { @@ -415,8 +424,15 @@ pub async fn sync_with_backend( None => event_bus.process_command(CoreCommand::SyncAll), } - let result = - fetch_and_persist(db, event_bus, backend, filter_engine, account_id.as_deref()).await; + let result = fetch_and_persist( + db, + event_bus, + backend, + filter_engine, + account_id.as_deref(), + filters_enabled, + ) + .await; if let Err(e) = &result { event_bus.process_command(CoreCommand::ReportError { @@ -488,18 +504,20 @@ pub async fn run_account_sync( let setup = async { let config = find_account(db, account_id).await?; let password = secrets.get_secret(&config.keyring_secret_key)?; - build_mail_backend(&config, &password) + let filters_enabled = config.filters_enabled; + build_mail_backend(&config, &password).map(|backend| (backend, filters_enabled)) } .await; match setup { - Ok(backend) => { + Ok((backend, filters_enabled)) => { sync_with_backend( db, event_bus, backend.as_ref(), filter_engine, Some(account_id.to_string()), + filters_enabled, ) .await } @@ -540,6 +558,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: host.to_string(), }), @@ -626,6 +645,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -714,6 +734,7 @@ mod tests { &mock, &filter_engine, Some("acct-mock-1".to_string()), + true, ) .await .expect("sync succeeds"); @@ -773,7 +794,7 @@ mod tests { }); let filter_engine = empty_filter_engine(); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 0); @@ -816,6 +837,7 @@ mod tests { &mock, &filter_engine, Some("acct-fail".to_string()), + true, ) .await .expect_err("sync fails"); @@ -1036,7 +1058,7 @@ mod tests { }); mock.add_message(mock_placed("m-urgent", "inbox", "Urgent: server down")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 1); @@ -1063,6 +1085,96 @@ mod tests { ); } + #[tokio::test] + async fn a_daemon_that_does_not_own_filters_stores_the_mail_but_fires_nothing() { + // The single-owner rule. A non-owning daemon must still sync and store + // -- it is a full mail client -- while producing none of the side + // effects that would be duplicated across the fleet. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let rule = nuncio_filter::NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = FilterEngine::new(vec![rule]).expect("compile rule"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_placed("m-urgent", "inbox", "Urgent: server down")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, false) + .await + .expect("sync succeeds"); + + assert_eq!(synced, 1, "the message must still be synced and stored"); + assert!( + !read_flag_of(&db, "m-urgent").await, + "a non-owning daemon must not apply MARK READ" + ); + assert!( + db.list_filter_execution_logs(10) + .await + .expect("list logs") + .is_empty(), + "no rule ran, so nothing may be logged as executed" + ); + assert!( + db.list_pending_mutations(10) + .await + .expect("list mutations") + .is_empty(), + "a non-owning daemon must enqueue no remote mutation" + ); + } + + #[tokio::test] + async fn run_account_sync_takes_filter_ownership_from_the_stored_account() { + // The flag has to come from the account row, not a caller's guess -- + // otherwise the gate is only ever exercised by tests that pass it. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + + let mut account = sample_imap_account("acct-owner"); + assert!( + !account.filters_enabled, + "a freshly constructed account must default to not owning filters" + ); + db.save_account(&account).await.expect("save account"); + let reloaded = db + .list_accounts() + .await + .expect("list accounts") + .into_iter() + .find(|a| a.id == "acct-owner") + .expect("account round-trips"); + assert!(!reloaded.filters_enabled); + + account.filters_enabled = true; + db.save_account(&account).await.expect("update account"); + let reloaded = db + .list_accounts() + .await + .expect("list accounts") + .into_iter() + .find(|a| a.id == "acct-owner") + .expect("account round-trips"); + assert!( + reloaded.filters_enabled, + "opting one daemon in must persist" + ); + } + #[tokio::test] async fn sync_with_backend_enqueues_outbox_mutation_for_remote_action_on_match() { let (db, _dir) = DatabaseEngine::connect_ephemeral() @@ -1087,7 +1199,7 @@ mod tests { }); mock.add_message(mock_placed("m-archive", "inbox", "Please Archive Me")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 1); @@ -1137,7 +1249,7 @@ mod tests { }); mock.add_message(mock_placed("m-plain", "inbox", "Just a normal update")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 1); @@ -1201,7 +1313,7 @@ mod tests { mock.add_message(repeat); // First sync: the message is genuinely new, so both rules must fire. - let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("first sync succeeds"); assert_eq!(synced_first, 1); @@ -1229,7 +1341,7 @@ mod tests { // Second sync: the mock backend reports the SAME message again // (unchanged id), simulating a non-incremental re-sync. - let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("second sync succeeds"); assert_eq!(synced_second, 1); @@ -1303,7 +1415,7 @@ mod tests { mock.add_message(mock_placed("m-new-1", "inbox", "Urgent: brand new")); mock.add_message(mock_placed("m-new-2", "inbox", "Urgent: also new")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 3, "every fetched message is still persisted"); @@ -1363,6 +1475,7 @@ mod tests { &mock, &filter_engine, Some("acct-inc-1".to_string()), + true, ) .await .expect("first sync succeeds"); @@ -1384,6 +1497,7 @@ mod tests { &mock, &filter_engine, Some("acct-inc-1".to_string()), + true, ) .await .expect("second sync succeeds"); @@ -1436,6 +1550,7 @@ mod tests { &backend, &filter_engine, Some(MockMailBackend::SHARED_ACCOUNT_ID), + true, ) .await .expect("sync succeeds"); @@ -1489,6 +1604,7 @@ mod tests { &backend, &filter_engine, Some(MockMailBackend::SHARED_ACCOUNT_ID), + true, ) .await .expect("sync succeeds"); @@ -1602,7 +1718,7 @@ mod tests { // `None` account: the pass has nowhere to key a checkpoint and cannot // run the `present` diff, but the removal is still actionable. - let synced = fetch_and_persist(&db, &event_bus, &mock, &filter_engine, None) + let synced = fetch_and_persist(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 0, "nothing new arrived"); diff --git a/crates/nunciod/src/sync_dispatcher.rs b/crates/nunciod/src/sync_dispatcher.rs index 4526da0..52bdd6c 100644 --- a/crates/nunciod/src/sync_dispatcher.rs +++ b/crates/nunciod/src/sync_dispatcher.rs @@ -806,6 +806,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::Jmap(JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), diff --git a/crates/nunciod/tests/filters_triage_e2e_test.rs b/crates/nunciod/tests/filters_triage_e2e_test.rs index 57680d0..4ece4a2 100644 --- a/crates/nunciod/tests/filters_triage_e2e_test.rs +++ b/crates/nunciod/tests/filters_triage_e2e_test.rs @@ -54,6 +54,27 @@ fn sample_email(id: &str, subject: &str) -> Email { } } +/// The account the seeded messages belong to, opted in as this daemon's +/// filter-execution owner. +fn owning_account() -> nuncio_core::AccountConfig { + nuncio_core::AccountConfig { + id: ACCOUNT_ID.to_string(), + name: "Triage E2E".to_string(), + email_address: "triage-e2e@nuncio.mx".to_string(), + keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"), + sync_interval_secs: 60, + filters_enabled: true, + transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { + imap_host: "imap.nuncio.mx".to_string(), + imap_port: 993, + imap_tls_mode: nuncio_core::TlsMode::ImplicitTls, + smtp_host: "smtp.nuncio.mx".to_string(), + smtp_port: 465, + smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls, + }), + } +} + /// The INBOX occupancy a seeded message sits in. `id` doubles as the UID, so /// distinct messages occupy distinct rows. fn sample_placement(id: &str) -> Placement { @@ -111,6 +132,13 @@ async fn boot_daemon( .expect("open temp file-backed database"), ); + // Triage only acts on accounts this daemon owns filter execution for, so + // the account must exist and be opted in -- otherwise the RPC honestly + // refuses rather than scanning and applying nothing. + db.save_account(&owning_account()) + .await + .expect("seed the triage account"); + let token_bytes = secrets .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) .expect("mint gRPC bearer token from mock vault"); @@ -292,3 +320,54 @@ async fn triage_streams_multiple_chunks_for_a_store_bigger_than_one_chunk() { prev_scanned = update.scanned_count; } } + +/// A daemon that owns filter execution for no account must refuse `Triage` +/// outright. Streaming a scan that reports progress and applies nothing would +/// be indistinguishable, from the caller's side, from "no rules matched" -- +/// and the whole point of single-owner execution is that the non-owner stays +/// visibly inert rather than quietly so. +#[tokio::test] +async fn triage_refuses_when_this_daemon_owns_no_account() { + let secrets = Arc::new(SecretManager::mock()); + let rule = NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = Arc::new(FilterEngine::new(vec![rule]).expect("compile rule")); + + let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await; + + // Take ownership away: the account exists, but this daemon is not its + // filter owner. + let mut account = owning_account(); + account.filters_enabled = false; + db.save_account(&account) + .await + .expect("opt the account out"); + + seed(&db, "m-1", "Urgent: server down").await; + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("connect to the daemon"); + let status = client + .triage(authed_request(&token, TriageRequest::default())) + .await + .expect_err("triage must refuse without an owning account"); + + assert_eq!(status.code(), tonic::Code::FailedPrecondition); + assert!( + status.message().contains("filter execution"), + "the refusal must say why, got: {}", + status.message() + ); + + // And it must genuinely have applied nothing: the occupancy the rule + // would have marked read is still unread. + assert!( + !read_flag_of(&db, "m-1").await, + "a refused triage must not apply any action" + ); +} diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs index f16b4a9..318a1bb 100644 --- a/crates/nunciod/tests/outbox_executor_e2e_test.rs +++ b/crates/nunciod/tests/outbox_executor_e2e_test.rs @@ -101,6 +101,7 @@ fn sample_account() -> nuncio_core::AccountConfig { email_address: "owner@nuncio.mx".to_string(), keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993,