From dfc7cc9358fc1c4e886b99b81e87c765fd3e7699 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 14:11:05 -0500 Subject: [PATCH 1/3] perf(sync): batch new-message detection to remove the is_new_message N+1 (#341) Replace the per-message get_message existence check in fetch_and_persist with a single existing_message_ids(chunk_ids) lookup per fetched folder chunk, plus a per-pass seen-set to keep duplicate ids within one chunk firing at most once. Fire-once semantics for filter actions are unchanged: a message is new iff it was absent from the store before this sync pass and this is its first occurrence in the current chunk. Refs #325 Co-authored-by: James Maes --- crates/nuncio-store/src/db.rs | 67 +++++++++++++++++ crates/nunciod/src/sync.rs | 135 ++++++++++++++++++++++++++-------- 2 files changed, 171 insertions(+), 31 deletions(-) diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 5dd7ae9..7cba21a 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -1350,6 +1350,37 @@ impl DatabaseEngine { }) } + /// Return the subset of `ids` already present in the `messages` table, as a single + /// `SELECT id FROM messages WHERE id IN (...)` query. + /// + /// Lets a caller classify a whole fetched batch as new-vs-seen with one round trip instead + /// of one existence lookup per id. Passing an empty slice short-circuits to an empty set + /// without querying at all, since `IN ()` is not valid SQL. + pub async fn existing_message_ids( + &self, + ids: &[String], + ) -> Result, DatabaseError> { + if ids.is_empty() { + return Ok(std::collections::HashSet::new()); + } + + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = + sqlx::QueryBuilder::new("SELECT id FROM messages WHERE id IN ("); + let mut separated = builder.separated(", "); + for id in ids { + separated.push_bind(id); + } + builder.push(")"); + + let rows: Vec<(String,)> = builder + .build_query_as() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + Ok(rows.into_iter().map(|(id,)| id).collect()) + } + /// Save a [`nuncio_core::model::CalendarEvent`] to SQLite (INSERT OR REPLACE). /// /// Unlike [`Self::save_email`], calendar event summary/location are never encrypted at @@ -3730,6 +3761,42 @@ mod tests { } } + #[tokio::test] + async fn existing_message_ids_returns_only_the_ids_already_persisted() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + engine + .save_email(&export_test_email("msg-1", "acct-a", "inbox")) + .await + .unwrap(); + engine + .save_email(&export_test_email("msg-2", "acct-a", "inbox")) + .await + .unwrap(); + + let found = engine + .existing_message_ids(&[ + "msg-1".to_string(), + "msg-2".to_string(), + "msg-3-never-saved".to_string(), + ]) + .await + .unwrap(); + + assert_eq!(found.len(), 2); + assert!(found.contains("msg-1")); + assert!(found.contains("msg-2")); + assert!(!found.contains("msg-3-never-saved")); + } + + #[tokio::test] + async fn existing_message_ids_with_empty_input_returns_empty_set_without_querying() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + + let found = engine.existing_message_ids(&[]).await.unwrap(); + + assert!(found.is_empty()); + } + #[tokio::test] async fn list_messages_for_export_with_no_filters_returns_every_message() { let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index bbab07d..98dc78a 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -195,39 +195,26 @@ pub async fn apply_filter_actions( applied } -/// Reports whether `message_id` is not yet present in the store. -/// -/// Sync is not yet incremental (a re-sync re-fetches every message the -/// backend still reports, not just genuinely new ones), so this is the guard -/// that keeps filter actions firing exactly once per message: it must be -/// checked BEFORE `save_email` persists (or re-persists) the message, since -/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the -/// distinction between "arriving for the first time" and "seen again". -/// -/// A lookup failure that is not "not found" (a genuine DB error) is logged -/// and treated as "not new", the conservative choice: firing filters again on -/// a message we can't positively identify as new risks duplicate remote -/// mutations, which is worse than occasionally missing a fire on a message -/// that was in fact new. -async fn is_new_message(db: &nuncio_store::db::DatabaseEngine, message_id: &str) -> bool { - match db.get_message(message_id).await { - Ok(_) => false, - Err(e) if e.is_not_found() => true, - Err(e) => { - tracing::warn!( - "could not determine whether message '{message_id}' is new before sync \ - (treating as not-new to avoid duplicate filter firing): {e}" - ); - false - } - } -} - /// Fetch every folder and every message in every folder from `backend`, /// persisting each message via [`DatabaseEngine::save_email`] and, for /// messages genuinely new to the store, evaluating it against -/// `filter_engine` via [`apply_filter_actions`] (see [`is_new_message`] for -/// why this must be gated rather than run on every persisted message). +/// `filter_engine` via [`apply_filter_actions`]. +/// +/// Sync is not yet incremental (a re-sync re-fetches every message the +/// backend still reports, not just genuinely new ones), so new-vs-seen must +/// be determined BEFORE `save_email` persists (or re-persists) anything -- +/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the +/// distinction between "arriving for the first time" and "seen again". Doing +/// that with one existence lookup per message (`DatabaseEngine::get_message`) +/// is an N+1 query pattern over a fetched chunk; instead, this classifies an +/// entire folder's chunk with a single batched +/// `DatabaseEngine::existing_message_ids` lookup up front, then folds in a +/// per-batch "already seen this pass" set while iterating so a chunk that +/// itself contains a repeated id (e.g. a backend surfacing the same message +/// twice in one fetch) still fires at most once -- exactly as if each +/// message's existence had been checked one at a time immediately before its +/// own `save_email`. +/// /// Returns the total number of messages processed (a message id "processed" /// more than once, e.g. by a backend that returns the same id from multiple /// folders, is counted once per occurrence -- `save_email` itself is @@ -260,8 +247,19 @@ async fn fetch_and_persist( .sync_messages(&folder.id, last_state.as_deref()) .await?; let fetched = emails.len(); + + // One round trip classifies the whole chunk instead of one lookup per + // message. A lookup failure that is not "not found" would be a + // genuine DB error; `existing_message_ids` surfaces those as `Err` + // rather than silently treating the batch as "not new", so callers + // never persist and skip-filter on an unverified assumption. + let chunk_ids: Vec = emails.iter().map(|e| e.id.clone()).collect(); + let already_present = db.existing_message_ids(&chunk_ids).await?; + let mut seen_this_pass = std::collections::HashSet::new(); + for email in emails { - let is_new = is_new_message(db, &email.id).await; + let is_new = + !already_present.contains(&email.id) && seen_this_pass.insert(email.id.clone()); db.save_email(&email).await?; if is_new { apply_filter_actions(db, filter_engine, &email).await; @@ -1304,6 +1302,81 @@ mod tests { ); } + #[tokio::test] + async fn sync_with_backend_fires_filters_only_for_new_messages_in_a_mixed_batch() { + // A single fetched chunk containing a mix of an already-seen message + // (persisted by a prior sync) and genuinely new ones must fire filter + // actions ONLY for the new ones, via the batched + // `existing_message_ids` classification -- not a per-message lookup. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let mark_read_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![mark_read_rule]).expect("compile rule"); + + // Pre-seed one message directly, simulating it having landed in a + // prior sync pass. It must NOT re-fire even though this pass's + // backend still reports it (sync is not yet incremental). + db.save_email(&mock_email( + "m-already-seen", + "inbox", + "Urgent: known issue", + )) + .await + .expect("pre-seed existing message"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 3, + unread_messages: 3, + }); + // Mixed chunk: one already-present id, two genuinely new ids. + mock.add_message(mock_email("m-already-seen", "inbox", "Urgent: known issue")); + mock.add_message(mock_email("m-new-1", "inbox", "Urgent: brand new")); + mock.add_message(mock_email("m-new-2", "inbox", "Urgent: also new")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("sync succeeds"); + assert_eq!(synced, 3, "every fetched message is still persisted"); + + let logs = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs"); + assert_eq!( + logs.len(), + 2, + "filters fire only for the two genuinely new messages in the mixed batch" + ); + let fired_ids: std::collections::HashSet = + logs.iter().map(|l| l.message_id.clone()).collect(); + assert!(fired_ids.contains("m-new-1")); + assert!(fired_ids.contains("m-new-2")); + assert!(!fired_ids.contains("m-already-seen")); + + // The pre-seeded message must still be persisted (INSERT OR REPLACE + // semantics are unaffected by the batched new/seen classification) + // but its read flag must be untouched by this pass's MARK READ rule. + let already_seen = db + .get_message("m-already-seen") + .await + .expect("still persisted"); + assert!( + !already_seen.read, + "a message that already existed before this sync must not re-fire MARK READ" + ); + } + #[tokio::test] async fn sync_with_backend_persists_and_threads_the_folder_checkpoint_across_syncs() { // The regression this story exists to kill: the returned checkpoint -- 2.54.0 From cd859729231b46543fe98a09ed731cb9c55e9559 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 14:12:11 -0500 Subject: [PATCH 2/3] fix(filter): pin webhook dispatch to the checked IP to close DNS-rebind TOCTOU Refs #307 --- crates/nuncio-filter/src/validator.rs | 6 +- crates/nuncio-filter/src/webhook.rs | 162 ++++++++++++++++++++------ 2 files changed, 131 insertions(+), 37 deletions(-) diff --git a/crates/nuncio-filter/src/validator.rs b/crates/nuncio-filter/src/validator.rs index 7975e55..37a058d 100644 --- a/crates/nuncio-filter/src/validator.rs +++ b/crates/nuncio-filter/src/validator.rs @@ -35,8 +35,10 @@ pub struct ValidationOptions { pub allowed_forward_domains: Option>, /// When set, reject webhook targets that point at private, loopback, /// link-local, or metadata address ranges. The validator checks literal-IP - /// hosts; the dispatcher additionally checks addresses a hostname resolves - /// to before connecting. + /// hosts; the dispatcher additionally resolves hostnames, checks every + /// candidate address, and then pins the outbound connection to the + /// checked address so it cannot be swapped for an unchecked one by a + /// second DNS lookup. pub block_private_webhooks: bool, } diff --git a/crates/nuncio-filter/src/webhook.rs b/crates/nuncio-filter/src/webhook.rs index 68e7422..7d0e6da 100644 --- a/crates/nuncio-filter/src/webhook.rs +++ b/crates/nuncio-filter/src/webhook.rs @@ -6,7 +6,7 @@ use hmac::{Hmac, Mac}; use reqwest::Client; use serde_json::json; use sha2::Sha256; -use std::net::IpAddr; +use std::net::{IpAddr, SocketAddr}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use thiserror::Error; use tokio::net::lookup_host; @@ -24,6 +24,38 @@ fn first_blocked_address(addrs: &[IpAddr]) -> Option { .find(|ip| is_blocked_webhook_target(*ip)) } +/// Resolves `host` to its candidate addresses. Kept separate from the policy +/// check so the address list a rebinding DNS server could return is visible +/// to callers before any of it is trusted. +async fn resolve_host(host: &str, port: u16) -> Result, WebhookError> { + let resolved: Vec = lookup_host((host, port)) + .await + .map_err(|e| WebhookError::NetworkError(e.to_string()))? + .map(|addr| addr.ip()) + .collect(); + if resolved.is_empty() { + return Err(WebhookError::NetworkError(format!( + "webhook host '{host}' did not resolve to any address" + ))); + } + Ok(resolved) +} + +/// Builds an HTTP client whose connection to `host` is pinned to `addr`. +/// `reqwest`'s DNS override maps only the transport-level connect address; it +/// still sends the original hostname as the TLS SNI and `Host` header, so +/// certificate validation is unaffected. This is what stops the dispatcher +/// from re-resolving `host` after it has been checked against the SSRF +/// policy: the checked address is the only one the connection can use. +fn pinned_client(host: &str, addr: IpAddr, port: u16) -> Result { + Client::builder() + .timeout(Duration::from_secs(5)) + .redirect(reqwest::redirect::Policy::none()) + .resolve(host, SocketAddr::new(addr, port)) + .build() + .map_err(|e| WebhookError::NetworkError(e.to_string())) +} + /// Errors emitted during HTTP webhook dispatch execution. #[derive(Error, Debug)] pub enum WebhookError { @@ -39,22 +71,18 @@ pub enum WebhookError { } /// Outbound Webhook Dispatcher executing authenticated HTTP POST requests. +/// +/// Each dispatch builds its own client pinned to the address it just +/// resolved and checked (see [`pinned_client`]), so there is no shared +/// client to cache here. pub struct WebhookDispatcher { - client: Client, secret_key: Zeroizing, } impl WebhookDispatcher { /// Create a new `WebhookDispatcher` with an HMAC signing key. pub fn new(secret_key: impl Into) -> Self { - let client = Client::builder() - .timeout(Duration::from_secs(5)) - .redirect(reqwest::redirect::Policy::none()) - .build() - .unwrap_or_default(); - Self { - client, secret_key: Zeroizing::new(secret_key.into()), } } @@ -73,33 +101,36 @@ impl WebhookDispatcher { NsqlValidator::pass6_action_security(&[action], opts) .map_err(|e| WebhookError::SecurityViolation(e.to_string()))?; + // Pass 6 only sees literal-IP hosts. Resolve the target here and, + // when the policy is enabled, reject it if it points into a blocked + // range - closing the DNS-rebind class of SSRF bypass that a + // substring/literal check on the URL cannot catch. The resolved + // address is then pinned below for the actual connection, so the + // dispatch cannot re-resolve the hostname and land on a different, + // unchecked address: the address that was checked is the only one + // `reqwest` is permitted to connect to. + let parsed = reqwest::Url::parse(url) + .map_err(|e| WebhookError::SecurityViolation(format!("invalid webhook URL: {e}")))?; + let host = parsed + .host_str() + .ok_or_else(|| WebhookError::SecurityViolation("webhook URL has no host".to_string()))? + // `host_str` brackets IPv6 literals (`[::1]`); strip them so the + // pair passes to the resolver cleanly. + .trim_start_matches('[') + .trim_end_matches(']') + .to_string(); + let port = parsed.port_or_known_default().unwrap_or(0); + + let resolved = resolve_host(&host, port).await?; if opts.block_private_webhooks { - // Pass 6 only sees literal-IP hosts. Resolve the target here and - // reject it if a hostname points into a blocked range, closing the - // DNS-rebind class of SSRF bypass. This is not airtight: reqwest - // re-resolves and connects on its own after this check, so a narrow - // TOCTOU window remains where DNS could flip between our lookup and - // the actual connection. Fully closing it needs a pinned - // resolver/connector, which is out of scope here. - if let Ok(parsed) = reqwest::Url::parse(url) { - if let Some(host) = parsed.host_str() { - // `host_str` brackets IPv6 literals (`[::1]`); strip them so - // the pair passes to the resolver cleanly. - let host = host.trim_start_matches('[').trim_end_matches(']'); - let port = parsed.port_or_known_default().unwrap_or(0); - let resolved: Vec = lookup_host((host, port)) - .await - .map_err(|e| WebhookError::NetworkError(e.to_string()))? - .map(|addr| addr.ip()) - .collect(); - if let Some(blocked) = first_blocked_address(&resolved) { - return Err(WebhookError::SecurityViolation(format!( - "webhook host '{host}' resolves to blocked address {blocked}" - ))); - } - } + if let Some(blocked) = first_blocked_address(&resolved) { + return Err(WebhookError::SecurityViolation(format!( + "webhook host '{host}' resolves to blocked address {blocked}" + ))); } } + let pinned_addr = resolved[0]; + let client = pinned_client(&host, pinned_addr, port)?; let timestamp = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -122,8 +153,7 @@ impl WebhookDispatcher { mac.update(format!("{timestamp}.{payload_str}").as_bytes()); let signature = hex::encode(mac.finalize().into_bytes()); - let response = self - .client + let response = client .post(url) .header("Content-Type", "application/json") .header( @@ -162,6 +192,68 @@ impl WebhookDispatcher { #[cfg(test)] mod tests { use super::*; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + #[tokio::test] + async fn test_pinned_client_connects_to_checked_ip_not_dns() { + // `.invalid` is reserved by RFC 2606 to never resolve: any real DNS + // lookup for it fails. The mock server only listens on its loopback + // address, which has no relationship to that hostname in any + // resolver. If the request below reaches the mock server, it can + // only be because `pinned_client`'s resolve override - not DNS - + // supplied the connect address, proving the dispatcher connects to + // exactly the address it already checked. + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/hook")) + .respond_with(ResponseTemplate::new(200)) + .mount(&mock_server) + .await; + + let addr = *mock_server.address(); + let host = "checked-ip-pin.invalid"; + let client = pinned_client(host, addr.ip(), addr.port()) + .expect("building a client with a resolve override must succeed"); + + let url = format!("http://{host}:{}/hook", addr.port()); + let response = client + .post(&url) + .send() + .await + .expect("the pinned connection must reach the mock server, not a DNS lookup"); + + assert_eq!(response.status(), 200); + } + + #[tokio::test] + async fn test_dispatch_reaches_pinned_loopback_mock_server() { + // End-to-end proof that `dispatch_with_options` itself (not just the + // `pinned_client` helper) drives the request to the address it + // resolved and checked. `block_private_webhooks` is disabled here + // only to allow the test to target a loopback mock server; it does + // not affect whether the connection is pinned. + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/hook")) + .respond_with(ResponseTemplate::new(202)) + .mount(&mock_server) + .await; + + let dispatcher = WebhookDispatcher::new("secret_key_123"); + let opts = ValidationOptions { + available_folders: None, + allowed_forward_domains: None, + block_private_webhooks: false, + }; + let url = format!("{}/hook", mock_server.uri()); + let status = dispatcher + .dispatch_with_options(&url, "rule_1", "msg_1", "Test", "a@b.com", &opts) + .await + .expect("dispatch to the mock server must succeed"); + + assert_eq!(status, 202); + } #[tokio::test] async fn test_blocked_private_ip_webhook() { -- 2.54.0 From 01e7acd97d36b5120d764d87c41ec57aae9b05ec Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 14:12:44 -0500 Subject: [PATCH 3/3] feat(proto): typed errors (ErrorInfo/ErrorReason) via Status details (#342) * feat(proto): add typed error taxonomy and Status helper Introduce a frozen, machine-readable error taxonomy for the nuncio.v1 contract: an ErrorReason enum and an ErrorInfo message carrying a reason, optional metadata, and a human message. A nuncio-proto errors helper builds a tonic::Status that encodes ErrorInfo into the grpc-status-details-bin trailer with the canonical gRPC code per reason, and decodes the reason back on the client side. Regenerate the committed descriptor golden. Refs #273 * feat(nunciod): return typed ErrorInfo details from error sites Convert a representative set of daemon gRPC error returns to carry a typed ErrorInfo in the status-details trailer via the nuncio-proto errors helper: account/message/event/contact/rule not-found (with the offending id as metadata), NSQL validation failures, and the calendar not-configured precondition. Canonical status codes are unchanged, so existing error-path assertions still hold. Add an offline over-the-wire E2E test that decodes the reason and metadata from a returned Status against the real daemon server. Refs #273 --------- Co-authored-by: James Maes --- Cargo.lock | 1 + crates/nuncio-proto/Cargo.toml | 1 + .../proto/nuncio/v1/descriptor.bin | Bin 12854 -> 13517 bytes .../nuncio-proto/proto/nuncio/v1/nuncio.proto | 58 ++++++ crates/nuncio-proto/src/errors.rs | 153 ++++++++++++++++ crates/nuncio-proto/src/lib.rs | 1 + crates/nunciod/src/grpc.rs | 119 +++++++++--- crates/nunciod/tests/typed_errors_e2e_test.rs | 171 ++++++++++++++++++ 8 files changed, 475 insertions(+), 29 deletions(-) create mode 100644 crates/nuncio-proto/src/errors.rs create mode 100644 crates/nunciod/tests/typed_errors_e2e_test.rs diff --git a/Cargo.lock b/Cargo.lock index df41e19..f94cbed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2594,6 +2594,7 @@ dependencies = [ name = "nuncio-proto" version = "0.1.0" dependencies = [ + "bytes", "prost", "protoc-bin-vendored", "thiserror 1.0.69", diff --git a/crates/nuncio-proto/Cargo.toml b/crates/nuncio-proto/Cargo.toml index ccb1ec7..71359e1 100644 --- a/crates/nuncio-proto/Cargo.toml +++ b/crates/nuncio-proto/Cargo.toml @@ -13,6 +13,7 @@ workspace = true tonic.workspace = true tonic-prost.workspace = true prost.workspace = true +bytes.workspace = true thiserror.workspace = true [build-dependencies] diff --git a/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin b/crates/nuncio-proto/proto/nuncio/v1/descriptor.bin index 1f17b041e64990ea6ac8d64c00bbd63f1cdc35ec..850edc681e2a0d676fc2fee02bcd32376c5c1d8b 100644 GIT binary patch delta 687 zcmZvY%}T>i5QR-+|E5*KMI~yfrBo3Y0lN|f;hNmqi!rG;NzqjZHcHW^kp4hdeFh&z z1h?)5H@<|g;Z3dDSa&nbcg~!d=hjk+ujTbUIZ+aKbJg-+MdYiTPazWY!x8(@AlZR* z$M5yLTVEFW#w`18H;!N_ra}^kcm8M+EX$%2tw3_<-4A@eI+A`hhlvB>apz-u{31eb zCNnC4Oc*nn%Luki!*vX+K~!3!M3WX!8H%xK6YEsNHsxRqQ=K!y9D`NWbPU@yOxvyT zN|O?33pT%e#?v*-(a8)U&bgcu%fhF#%pxj5A(n=%}#!~7x%rpzu zd7)jo metadata = 2; + // Human-readable message, identical to the enclosing `Status` message. + string message = 3; +} + // System service exposes daemon-level introspection RPCs. // // This is the foundation of the versioned gRPC API (see diff --git a/crates/nuncio-proto/src/errors.rs b/crates/nuncio-proto/src/errors.rs new file mode 100644 index 0000000..c13f1c6 --- /dev/null +++ b/crates/nuncio-proto/src/errors.rs @@ -0,0 +1,153 @@ +//! Typed, machine-readable gRPC errors for the `nuncio.v1` contract. +//! +//! A failed RPC should carry a structured [`v1::ErrorInfo`] -- a stable +//! [`v1::ErrorReason`] discriminant, optional metadata, and a human message +//! -- in the gRPC `grpc-status-details-bin` status-details trailer, alongside +//! the canonical gRPC [`Code`] that the reason maps to. Servers build such a +//! `Status` with [`status`] / [`status_with_metadata`]; clients read the +//! reason back with [`error_reason`] / [`error_info`] and branch on it, +//! rather than string-matching the human-readable (English) message text. +//! +//! The reason-to-code mapping is centralized in [`ErrorReason::code`] so the +//! transport-level status code and the application-level reason can never +//! drift apart at a call site. + +use crate::v1::{ErrorInfo, ErrorReason}; +use bytes::Bytes; +use prost::Message; +use tonic::{Code, Status}; + +impl ErrorReason { + /// The canonical gRPC [`Code`] this reason maps to. Both the unspecified + /// zero value and an explicit internal error map to [`Code::Internal`], + /// so an older client that receives a reason it does not recognize (and + /// decodes it as `Unspecified`) still sees an internal-class failure. + #[must_use] + pub fn code(self) -> Code { + match self { + ErrorReason::Unspecified | ErrorReason::Internal => Code::Internal, + ErrorReason::ValidationFailed => Code::InvalidArgument, + ErrorReason::AccountNotFound + | ErrorReason::RuleNotFound + | ErrorReason::MessageNotFound + | ErrorReason::EventNotFound + | ErrorReason::ContactNotFound => Code::NotFound, + ErrorReason::AlreadyExists => Code::AlreadyExists, + ErrorReason::NotConfigured => Code::FailedPrecondition, + ErrorReason::BackendUnreachable => Code::Unavailable, + ErrorReason::AuthRequired => Code::Unauthenticated, + ErrorReason::Unsupported => Code::Unimplemented, + } + } +} + +/// Builds a [`Status`] for `reason` carrying `message` both as the `Status` +/// message and inside a prost-encoded [`v1::ErrorInfo`] in the +/// `grpc-status-details-bin` trailer, with the canonical code from +/// [`ErrorReason::code`]. +pub fn status(reason: ErrorReason, message: impl Into) -> Status { + build(reason, message.into(), Vec::new()) +} + +/// Like [`status`], but also attaches structured `metadata` (e.g. the id of +/// the resource that was not found) to the encoded [`v1::ErrorInfo`]. +pub fn status_with_metadata( + reason: ErrorReason, + message: impl Into, + metadata: impl IntoIterator, +) -> Status { + build(reason, message.into(), metadata.into_iter().collect()) +} + +fn build(reason: ErrorReason, message: String, metadata: Vec<(String, String)>) -> Status { + let info = ErrorInfo { + reason: reason as i32, + metadata: metadata.into_iter().collect(), + message: message.clone(), + }; + Status::with_details(reason.code(), message, Bytes::from(info.encode_to_vec())) +} + +/// Decodes the [`v1::ErrorInfo`] carried in `status`'s details trailer, if +/// one is present and well-formed. Returns `None` when the status carries no +/// details (e.g. an error not built through this module) or the details are +/// not a valid `ErrorInfo`. +#[must_use] +pub fn error_info(status: &Status) -> Option { + let details = status.details(); + if details.is_empty() { + return None; + } + ErrorInfo::decode(details).ok() +} + +/// Decodes just the [`v1::ErrorReason`] from `status`'s details trailer, +/// returning `None` when no typed [`v1::ErrorInfo`] is present. +#[must_use] +pub fn error_reason(status: &Status) -> Option { + error_info(status).map(|info| info.reason()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn each_reason_maps_to_its_canonical_code() { + assert_eq!(ErrorReason::Unspecified.code(), Code::Internal); + assert_eq!(ErrorReason::ValidationFailed.code(), Code::InvalidArgument); + assert_eq!(ErrorReason::AccountNotFound.code(), Code::NotFound); + assert_eq!(ErrorReason::RuleNotFound.code(), Code::NotFound); + assert_eq!(ErrorReason::MessageNotFound.code(), Code::NotFound); + assert_eq!(ErrorReason::EventNotFound.code(), Code::NotFound); + assert_eq!(ErrorReason::ContactNotFound.code(), Code::NotFound); + assert_eq!(ErrorReason::AlreadyExists.code(), Code::AlreadyExists); + assert_eq!(ErrorReason::NotConfigured.code(), Code::FailedPrecondition); + assert_eq!(ErrorReason::BackendUnreachable.code(), Code::Unavailable); + assert_eq!(ErrorReason::AuthRequired.code(), Code::Unauthenticated); + assert_eq!(ErrorReason::Unsupported.code(), Code::Unimplemented); + assert_eq!(ErrorReason::Internal.code(), Code::Internal); + } + + #[test] + fn status_carries_reason_code_and_decodable_details() { + let status = status(ErrorReason::MessageNotFound, "message 'm-1' not found"); + assert_eq!(status.code(), Code::NotFound); + assert_eq!(status.message(), "message 'm-1' not found"); + + let info = error_info(&status).expect("typed details present"); + assert_eq!(info.reason(), ErrorReason::MessageNotFound); + assert_eq!(info.message, "message 'm-1' not found"); + assert!(info.metadata.is_empty()); + + assert_eq!( + error_reason(&status), + Some(ErrorReason::MessageNotFound), + "the reason round-trips through the encoded details" + ); + } + + #[test] + fn status_with_metadata_round_trips_context() { + let status = status_with_metadata( + ErrorReason::AccountNotFound, + "account 'acct-1' not found", + [("account_id".to_string(), "acct-1".to_string())], + ); + assert_eq!(status.code(), Code::NotFound); + + let info = error_info(&status).expect("typed details present"); + assert_eq!(info.reason(), ErrorReason::AccountNotFound); + assert_eq!( + info.metadata.get("account_id").map(String::as_str), + Some("acct-1") + ); + } + + #[test] + fn plain_status_without_details_decodes_to_none() { + let status = Status::not_found("no typed details here"); + assert_eq!(error_info(&status), None); + assert_eq!(error_reason(&status), None); + } +} diff --git a/crates/nuncio-proto/src/lib.rs b/crates/nuncio-proto/src/lib.rs index 6fe756c..6709375 100644 --- a/crates/nuncio-proto/src/lib.rs +++ b/crates/nuncio-proto/src/lib.rs @@ -22,6 +22,7 @@ pub mod addr; pub mod client; +pub mod errors; pub use addr::{grpc_addr_from_env, DEFAULT_GRPC_ADDR, GRPC_ADDR_ENV_VAR}; diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 43c73ef..cb25ab5 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -10,6 +10,7 @@ use nuncio_contacts::ContactsBackend; use nuncio_core::{CoreCommand, CoreEvent, EventBus}; use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, ValidationOptions}; use nuncio_mail::{ImapEngine, MailBackend, MessageSender, SmtpTransportEngine}; +use nuncio_proto::errors; use nuncio_proto::v1::accounts_server::{Accounts, AccountsServer}; use nuncio_proto::v1::audit_server::{Audit, AuditServer}; use nuncio_proto::v1::calendar_server::{Calendar, CalendarServer}; @@ -19,6 +20,7 @@ use nuncio_proto::v1::export_server::{Export, ExportServer}; use nuncio_proto::v1::filters_server::{Filters, FiltersServer}; use nuncio_proto::v1::mail_server::{Mail, MailServer}; use nuncio_proto::v1::system_server::{System, SystemServer}; +use nuncio_proto::v1::ErrorReason; use nuncio_proto::v1::{ export_request, AccountConfig as AccountConfigProto, AccountProtocol as AccountProtocolProto, AddAccountRequest, AddAccountResponse, Attachment as AttachmentProto, @@ -549,7 +551,7 @@ impl Accounts for AccountsGrpcService { let config = map_account_config_from_proto(proto_config)?; config .validate() - .map_err(|e| Status::invalid_argument(e.to_string()))?; + .map_err(|e| errors::status(ErrorReason::ValidationFailed, e.to_string()))?; // Store the password credential in the OS keyring FIRST, keyed by // this account's `keyring_secret_key` -- it is never written to @@ -615,7 +617,7 @@ impl Accounts for AccountsGrpcService { let config = map_account_config_from_proto(proto_config)?; config .validate() - .map_err(|e| Status::invalid_argument(e.to_string()))?; + .map_err(|e| errors::status(ErrorReason::ValidationFailed, e.to_string()))?; // Updating a non-existent account is a client error, not a silent // create: reject it so a typo'd id never conjures a new account row. @@ -672,7 +674,13 @@ impl Accounts for AccountsGrpcService { .get_account(&req.id) .await .map_err(|e| Status::internal(format!("failed to look up account: {e}")))? - .ok_or_else(|| Status::not_found(format!("account '{}' not found", req.id)))?; + .ok_or_else(|| { + errors::status_with_metadata( + ErrorReason::AccountNotFound, + format!("account '{}' not found", req.id), + [("account_id".to_string(), req.id.clone())], + ) + })?; self.db .delete_account(&req.id) @@ -708,7 +716,13 @@ impl Accounts for AccountsGrpcService { .get_account(&req.id) .await .map_err(|e| Status::internal(format!("failed to look up account: {e}")))? - .ok_or_else(|| Status::not_found(format!("account '{}' not found", req.id)))?; + .ok_or_else(|| { + errors::status_with_metadata( + ErrorReason::AccountNotFound, + format!("account '{}' not found", req.id), + [("account_id".to_string(), req.id.clone())], + ) + })?; let password = self .secrets @@ -954,16 +968,19 @@ impl CalendarGrpcService { .collect(); if caldav_accounts.is_empty() { - return Err(Status::failed_precondition(if req.account_id.is_empty() { - "no CalDAV account is configured; add one with protocol CALDAV and a \ - collection_url before syncing" - .to_string() - } else { - format!( - "no CalDAV account with id '{}' is configured", - req.account_id - ) - })); + return Err(errors::status( + ErrorReason::NotConfigured, + if req.account_id.is_empty() { + "no CalDAV account is configured; add one with protocol CALDAV and a \ + collection_url before syncing" + .to_string() + } else { + format!( + "no CalDAV account with id '{}' is configured", + req.account_id + ) + }, + )); } let mut total = 0usize; @@ -1076,7 +1093,13 @@ impl Calendar for CalendarGrpcService { .db .get_calendar_event(&req.event_id) .await - .map_err(|e| Status::not_found(format!("event '{}' not found: {e}", req.event_id)))?; + .map_err(|e| { + errors::status_with_metadata( + ErrorReason::EventNotFound, + format!("event '{}' not found: {e}", req.event_id), + [("event_id".to_string(), req.event_id.clone())], + ) + })?; Ok(Response::new(GetEventResponse { event: Some(map_calendar_event_to_proto(event)), @@ -1183,7 +1206,11 @@ impl Contacts for ContactsGrpcService { } let contact = self.db.get_contact(&req.contact_id).await.map_err(|e| { - Status::not_found(format!("contact '{}' not found: {e}", req.contact_id)) + errors::status_with_metadata( + ErrorReason::ContactNotFound, + format!("contact '{}' not found: {e}", req.contact_id), + [("contact_id".to_string(), req.contact_id.clone())], + ) })?; Ok(Response::new(GetContactResponse { @@ -1358,7 +1385,11 @@ impl Mail for MailGrpcService { } let email = self.db.get_message(&req.message_id).await.map_err(|e| { - Status::not_found(format!("message '{}' not found: {e}", req.message_id)) + errors::status_with_metadata( + ErrorReason::MessageNotFound, + format!("message '{}' not found: {e}", req.message_id), + [("message_id".to_string(), req.message_id.clone())], + ) })?; Ok(Response::new(GetMessageResponse { @@ -1387,7 +1418,11 @@ impl Mail for MailGrpcService { .set_message_read(&req.message_id, req.read) .await .map_err(|e| { - Status::not_found(format!("message '{}' not found: {e}", req.message_id)) + errors::status_with_metadata( + ErrorReason::MessageNotFound, + format!("message '{}' not found: {e}", req.message_id), + [("message_id".to_string(), req.message_id.clone())], + ) })?; self.event_bus.process_command(CoreCommand::MarkRead { @@ -1655,11 +1690,19 @@ impl Filters for FiltersGrpcService { ) -> Result, Status> { let req = request.into_inner(); - let rule = NsqlParser::parse_rule(req.name, req.priority, &req.nsql) - .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?; + let rule = NsqlParser::parse_rule(req.name, req.priority, &req.nsql).map_err(|e| { + errors::status( + ErrorReason::ValidationFailed, + format!("NSQL syntax error: {e}"), + ) + })?; - NsqlValidator::validate(&rule, &ValidationOptions::default()) - .map_err(|e| Status::invalid_argument(format!("NSQL validation error: {e}")))?; + NsqlValidator::validate(&rule, &ValidationOptions::default()).map_err(|e| { + errors::status( + ErrorReason::ValidationFailed, + format!("NSQL validation error: {e}"), + ) + })?; self.db .save_filter_rule(&rule) @@ -1751,8 +1794,12 @@ impl Filters for FiltersGrpcService { ) -> Result, Status> { let req = request.into_inner(); - let rule = NsqlParser::parse_rule("Preview Rule", 0, &req.nsql) - .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?; + let rule = NsqlParser::parse_rule("Preview Rule", 0, &req.nsql).map_err(|e| { + errors::status( + ErrorReason::ValidationFailed, + format!("NSQL syntax error: {e}"), + ) + })?; let preview_engine = FilterEngine::new(vec![rule]) .map_err(|e| Status::internal(format!("failed to build preview engine: {e}")))?; @@ -1791,17 +1838,31 @@ impl Filters for FiltersGrpcService { .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))? .into_iter() .find(|r| r.id == req.id) - .ok_or_else(|| Status::not_found(format!("filter rule '{}' not found", req.id)))?; + .ok_or_else(|| { + errors::status_with_metadata( + ErrorReason::RuleNotFound, + format!("filter rule '{}' not found", req.id), + [("rule_id".to_string(), req.id.clone())], + ) + })?; let name = req.name.unwrap_or(existing.name); let nsql = req.nsql.unwrap_or(existing.nsql_text); let priority = req.priority.unwrap_or(existing.priority); - let mut rule = NsqlParser::parse_rule(name, priority, &nsql) - .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?; + let mut rule = NsqlParser::parse_rule(name, priority, &nsql).map_err(|e| { + errors::status( + ErrorReason::ValidationFailed, + format!("NSQL syntax error: {e}"), + ) + })?; - NsqlValidator::validate(&rule, &ValidationOptions::default()) - .map_err(|e| Status::invalid_argument(format!("NSQL validation error: {e}")))?; + NsqlValidator::validate(&rule, &ValidationOptions::default()).map_err(|e| { + errors::status( + ErrorReason::ValidationFailed, + format!("NSQL validation error: {e}"), + ) + })?; rule.id = existing.id; diff --git a/crates/nunciod/tests/typed_errors_e2e_test.rs b/crates/nunciod/tests/typed_errors_e2e_test.rs new file mode 100644 index 0000000..436f49c --- /dev/null +++ b/crates/nunciod/tests/typed_errors_e2e_test.rs @@ -0,0 +1,171 @@ +//! Offline end-to-end test that typed `ErrorInfo` details survive the gRPC +//! wire. +//! +//! Boots a real `nunciod` daemon gRPC server +//! (`nunciod::grpc::serve_on_listener_with_overrides`) on an ephemeral +//! loopback port, backed by a real temp file-backed [`DatabaseEngine`] and a +//! [`SecretManager::mock`] vault (never the real OS keyring), with every +//! engine override left at its default (no injected backends). +//! +//! Driving the authenticated `nuncio-proto` clients, it triggers several +//! known error conditions and proves a client can machine-read each failure: +//! the returned `tonic::Status` carries the correct canonical gRPC code AND a +//! prost-encoded [`nuncio_proto::v1::ErrorInfo`] in the +//! `grpc-status-details-bin` trailer whose `reason` (and metadata) decode +//! back exactly, so a client never has to string-match the English message. +//! +//! Fully offline: no real network, no real OS keyring, no live server. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use nuncio_core::EventBus; +use nuncio_filter::FilterEngine; +use nuncio_proto::errors; +use nuncio_proto::v1::{ + CalendarSyncRequest, CreateRuleRequest, ErrorReason, GetMessageRequest, RemoveAccountRequest, +}; +use nuncio_store::db::DatabaseEngine; +use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT}; +use nunciod::grpc::{ + serve_on_listener_with_overrides, AccountsEngineOverrides, CalendarEngineOverrides, + ContactsEngineOverrides, MailEngineOverrides, +}; +use std::sync::Arc; +use tokio::net::TcpListener; +use tonic::Code; + +/// Boots the real daemon gRPC server on an ephemeral loopback port with every +/// override at its default. Returns the server address and minted bearer +/// token; the `TempDir` guard is returned so the caller controls its lifetime +/// for the whole test (the spawned server keeps using the db). +async fn boot_daemon() -> (std::net::SocketAddr, String, tempfile::TempDir) { + let secrets = Arc::new(SecretManager::mock()); + let dir = tempfile::tempdir().expect("create temp dir for file-backed db"); + let db_path = dir.path().join("typed_errors_e2e.db"); + let db = Arc::new( + DatabaseEngine::connect_file(&db_path, &secrets) + .await + .expect("open temp file-backed database"), + ); + + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("mint gRPC bearer token from mock vault"), + ); + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral loopback port"); + let addr = listener.local_addr().expect("listener has local addr"); + + let event_bus = Arc::new(EventBus::new()); + let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")); + let server_secrets = secrets.clone(); + let server_token = token.clone(); + tokio::spawn(async move { + let _ = serve_on_listener_with_overrides( + listener, + event_bus, + db, + filter_engine, + server_secrets, + server_token, + MailEngineOverrides::default(), + CalendarEngineOverrides::default(), + ContactsEngineOverrides::default(), + AccountsEngineOverrides::default(), + ) + .await; + }); + + (addr, token, dir) +} + +#[tokio::test] +async fn typed_error_reasons_decode_from_status_details_over_the_wire() { + let (addr, token, _db_dir) = boot_daemon().await; + let addr_str = addr.to_string(); + + // ---- account-not-found: NotFound + ACCOUNT_NOT_FOUND + account_id ---- + let mut accounts = nuncio_proto::client::connect_accounts(&addr_str, &token) + .await + .expect("accounts client connects"); + let status = accounts + .remove_account(RemoveAccountRequest { + id: "no-such-account".to_string(), + }) + .await + .expect_err("removing a nonexistent account must fail"); + assert_eq!(status.code(), Code::NotFound); + let info = errors::error_info(&status).expect("typed ErrorInfo present in details trailer"); + assert_eq!(info.reason(), ErrorReason::AccountNotFound); + assert_eq!( + info.metadata.get("account_id").map(String::as_str), + Some("no-such-account"), + "the missing id is carried as machine-readable metadata" + ); + assert_eq!(info.message, status.message()); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::AccountNotFound) + ); + + // ---- message-not-found: NotFound + MESSAGE_NOT_FOUND + message_id ---- + let mut mail = nuncio_proto::client::connect_mail(&addr_str, &token) + .await + .expect("mail client connects"); + let status = mail + .get_message(GetMessageRequest { + message_id: "no-such-message".to_string(), + }) + .await + .expect_err("fetching a nonexistent message must fail"); + assert_eq!(status.code(), Code::NotFound); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::MessageNotFound) + ); + let info = errors::error_info(&status).expect("typed ErrorInfo present"); + assert_eq!( + info.metadata.get("message_id").map(String::as_str), + Some("no-such-message") + ); + + // ---- not-configured: FailedPrecondition + NOT_CONFIGURED ---- + let mut calendar = nuncio_proto::client::connect_calendar(&addr_str, &token) + .await + .expect("calendar client connects"); + let status = calendar + .sync(CalendarSyncRequest { + account_id: String::new(), + calendar_id: "personal".to_string(), + start_window: 0, + end_window: 0, + }) + .await + .expect_err("syncing with no CalDAV account configured must fail"); + assert_eq!(status.code(), Code::FailedPrecondition); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::NotConfigured) + ); + + // ---- validation-failed: InvalidArgument + VALIDATION_FAILED ---- + let mut filters = nuncio_proto::client::connect_filters(&addr_str, &token) + .await + .expect("filters client connects"); + let status = filters + .create_rule(CreateRuleRequest { + name: "bad rule".to_string(), + nsql: "this is not valid nsql".to_string(), + priority: 0, + }) + .await + .expect_err("creating a rule from unparseable NSQL must fail"); + assert_eq!(status.code(), Code::InvalidArgument); + assert_eq!( + errors::error_reason(&status), + Some(ErrorReason::ValidationFailed) + ); +} -- 2.54.0