From bdb61488e65e0aaf43acc3615e60949a445e6693 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 18:32:06 -0500 Subject: [PATCH] fix(filter): surface rule-eval timeouts and instrument engine + webhook egress Replace the silent unwrap_or_default() on the ReDoS evaluation timeout with an explicit, per-rule timeout that logs a warn! carrying the rule's identity instead of quietly folding a timed-out evaluation into "no match". The per-rule evaluation now runs on a blocking-pool task so the timeout can actually preempt a slow synchronous evaluation, rather than racing a timer a non-yielding future would never let fire; a timed-out rule is documented and logged as non-matching, and the rest of the batch still runs. Add DEBUG rule-fire logging (rule id/name + action kinds, never the destination data an action carries) and INFO on reload_rules. Wrap webhook dispatch in a tracing span logging the target host, and log every SSRF/CIDR/DNS-rebind block decision as warn! with host/blocked-ip and reason; allowed egress logs at debug. No message bodies, secrets, or full URLs are logged. Refs #360 --- crates/nuncio-filter/src/engine.rs | 192 ++++++++++++++++-- crates/nuncio-filter/src/webhook.rs | 300 ++++++++++++++++++++++------ 2 files changed, 414 insertions(+), 78 deletions(-) diff --git a/crates/nuncio-filter/src/engine.rs b/crates/nuncio-filter/src/engine.rs index 6037db1..d5d1230 100644 --- a/crates/nuncio-filter/src/engine.rs +++ b/crates/nuncio-filter/src/engine.rs @@ -11,6 +11,7 @@ use regex::Regex; use sha2::Sha256; use std::sync::Arc; use std::time::{Duration, Instant}; +use tracing::{debug, info, warn}; /// Pre-compiled single filter rule optimizing regex and predicate execution. #[derive(Clone)] @@ -233,6 +234,26 @@ impl CompiledFilterSet { } } +/// Map each action to its variant name for observability logging, deliberately +/// dropping any destination data (folder name, forward address, webhook URL) +/// that the action variants carry. +fn action_kinds(actions: &[RuleAction]) -> Vec<&'static str> { + actions + .iter() + .map(|action| match action { + RuleAction::MoveTo(_) => "MOVE_TO", + RuleAction::CopyTo(_) => "COPY_TO", + RuleAction::MarkRead => "MARK_READ", + RuleAction::MarkUnread => "MARK_UNREAD", + RuleAction::Flag => "FLAG", + RuleAction::Unflag => "UNFLAG", + RuleAction::Delete => "DELETE", + RuleAction::ForwardTo(_) => "FORWARD_TO", + RuleAction::CallWebhook(_) => "CALL_WEBHOOK", + }) + .collect() +} + /// Lock-free Filter Engine wrapping `ArcSwap`. pub struct FilterEngine { cache: Arc>, @@ -250,7 +271,9 @@ impl FilterEngine { /// Swap / reload active filter rules atomically (<5ns latencies). pub fn reload_rules(&self, rules: Vec) -> Result<(), String> { let set = CompiledFilterSet::new(rules)?; + let rule_count = set.filters.len(); self.cache.store(Arc::new(set)); + info!(rule_count, "filter engine rules reloaded"); Ok(()) } @@ -266,6 +289,12 @@ impl FilterEngine { for filter in &guard.filters { if filter.rule.matches_account(&email.account_id) && filter.evaluate_condition(email) { + debug!( + rule_id = %filter.rule.id, + rule_name = %filter.rule.name, + actions = ?action_kinds(&filter.rule.actions), + "rule matched; dispatching actions" + ); results.push((filter.rule.clone(), filter.rule.actions.clone())); } } @@ -274,28 +303,66 @@ impl FilterEngine { } /// Evaluate with a Tokio hard timeout for ReDoS safety. + /// + /// Each rule's condition evaluation is timed independently (on a + /// blocking-pool task, so a real timeout can actually preempt a + /// long-running synchronous evaluation rather than merely racing a timer + /// that a purely synchronous future would never yield to). A rule whose + /// evaluation exceeds `timeout_duration` is treated as **non-matching** + /// for this message — the documented, fail-closed outcome — but unlike a + /// bare `unwrap_or_default()`, the timeout is never silent: it is logged + /// with the rule's identity via `warn!`, and evaluation continues with + /// the remaining rules rather than aborting the whole batch. pub async fn evaluate_with_timeout( &self, email: &Email, timeout_duration: Duration, ) -> Vec<(FilterRule, Vec)> { - let email_clone = email.clone(); - let engine_cache = self.cache.clone(); + let guard = self.cache.load(); + let mut results = Vec::new(); - tokio::time::timeout(timeout_duration, async move { - let guard = engine_cache.load(); - let mut results = Vec::new(); - for filter in &guard.filters { - if filter.rule.matches_account(&email_clone.account_id) - && filter.evaluate_condition(&email_clone) - { + for filter in &guard.filters { + if !filter.rule.matches_account(&email.account_id) { + continue; + } + + let filter_for_task = filter.clone(); + let email_for_task = email.clone(); + let eval_task = tokio::task::spawn_blocking(move || { + filter_for_task.evaluate_condition(&email_for_task) + }); + + match tokio::time::timeout(timeout_duration, eval_task).await { + Ok(Ok(true)) => { + debug!( + rule_id = %filter.rule.id, + rule_name = %filter.rule.name, + actions = ?action_kinds(&filter.rule.actions), + "rule matched; dispatching actions" + ); results.push((filter.rule.clone(), filter.rule.actions.clone())); } + Ok(Ok(false)) => {} + Ok(Err(join_error)) => { + warn!( + rule_id = %filter.rule.id, + rule_name = %filter.rule.name, + error = %join_error, + "rule evaluation task failed to complete; treating rule as non-matching" + ); + } + Err(_elapsed) => { + warn!( + rule_id = %filter.rule.id, + rule_name = %filter.rule.name, + timeout_ms = timeout_duration.as_millis(), + "rule evaluation timed out (ReDoS guard); treating rule as non-matching" + ); + } } - results - }) - .await - .unwrap_or_default() + } + + results } /// Dry-run preview evaluation returning detailed microsecond traces. @@ -367,6 +434,51 @@ impl FilterEngine { mod tests { use super::*; use nuncio_core::model::Email; + use std::sync::Mutex; + use tracing::field::{Field, Visit}; + use tracing::span; + + /// Minimal `tracing::Subscriber` that records a formatted line per event + /// so tests can assert on emitted level + fields without pulling in + /// `tracing-subscriber`'s registry machinery. + struct CapturingSubscriber { + events: Arc>>, + } + + struct LineVisitor<'a>(&'a mut String); + + impl Visit for LineVisitor<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.push_str(&format!(" {}={:?}", field.name(), value)); + } + } + + impl tracing::Subscriber for CapturingSubscriber { + fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { + true + } + + fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id { + span::Id::from_u64(1) + } + + fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {} + + fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {} + + fn event(&self, event: &tracing::Event<'_>) { + let mut line = format!("{}", event.metadata().level()); + let mut visitor = LineVisitor(&mut line); + event.record(&mut visitor); + if let Ok(mut events) = self.events.lock() { + events.push(line); + } + } + + fn enter(&self, _span: &span::Id) {} + + fn exit(&self, _span: &span::Id) {} + } fn test_email(account_id: &str, subject: &str, folder_id: &str) -> Email { Email { @@ -576,4 +688,58 @@ mod tests { let email = test_email("acct-1", "Anything", "inbox"); assert_eq!(engine.evaluate(&email).len(), 1); } + + #[tokio::test] + async fn test_evaluate_with_timeout_logs_warn_and_treats_timeout_as_non_match() { + // A crafted, slow-to-evaluate condition (large body scanned by a + // CONTAINS check) paired with a near-zero timeout budget forces the + // per-rule timeout path deterministically: the evaluation runs on a + // real blocking-pool thread while the timer races it on the async + // task, so the timer reliably wins for any evaluation slower than a + // few microseconds. + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let subscriber = CapturingSubscriber { + events: events.clone(), + }; + let _guard = tracing::subscriber::set_default(subscriber); + + let rule = FilterRule { + id: "slow-rule".to_string(), + name: "Slow Rule".to_string(), + target_account: "*".to_string(), + priority: 1, + enabled: true, + nsql_text: String::new(), + conditions: ConditionNode::Leaf(ConditionLeaf { + field: FilterField::Body, + operator: FilterOperator::Contains, + value: FilterValue::String("nonexistent-token".to_string()), + }), + actions: vec![RuleAction::MarkRead], + created_at: 0, + updated_at: 0, + }; + let engine = FilterEngine::new(vec![rule]).unwrap(); + + let mut email = test_email("acct-1", "Anything", "inbox"); + email.body_plain = Some("A".repeat(60_000_000)); + + let results = engine + .evaluate_with_timeout(&email, Duration::from_millis(1)) + .await; + + assert!( + results.is_empty(), + "documented outcome: a timed-out rule must be treated as non-matching, \ + not fabricated as a match" + ); + + let captured = events.lock().unwrap(); + assert!( + captured.iter().any(|line| line.contains("WARN") + && line.contains("timed out") + && line.contains("slow-rule")), + "expected a WARN log carrying the rule id for the timed-out evaluation, got: {captured:?}" + ); + } } diff --git a/crates/nuncio-filter/src/webhook.rs b/crates/nuncio-filter/src/webhook.rs index 7d0e6da..123f82e 100644 --- a/crates/nuncio-filter/src/webhook.rs +++ b/crates/nuncio-filter/src/webhook.rs @@ -10,8 +10,20 @@ use std::net::{IpAddr, SocketAddr}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use thiserror::Error; use tokio::net::lookup_host; +use tracing::{debug, warn, Instrument}; use zeroize::Zeroizing; +/// Extract just the host for logging (never the path/query, which may carry +/// caller-supplied tokens), falling back to a placeholder when the URL itself +/// cannot be parsed. +fn webhook_host_for_log(url: &str) -> String { + reqwest::Url::parse(url) + .ok() + .and_then(|parsed| parsed.host_str().map(str::to_string)) + .map(|h| h.trim_start_matches('[').trim_end_matches(']').to_string()) + .unwrap_or_else(|| "unparsable-url".to_string()) +} + type HmacSha256 = Hmac; /// Returns the first address in `addrs` that is a blocked webhook target, if @@ -97,75 +109,111 @@ impl WebhookDispatcher { sender: &str, opts: &ValidationOptions, ) -> Result { - let action = RuleAction::CallWebhook(url.to_string()); - NsqlValidator::pass6_action_security(&[action], opts) - .map_err(|e| WebhookError::SecurityViolation(e.to_string()))?; + let host_for_log = webhook_host_for_log(url); + let span = + tracing::info_span!("webhook_dispatch", host = %host_for_log, rule_id = %rule_id); - // 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); + async move { + let action = RuleAction::CallWebhook(url.to_string()); + if let Err(e) = NsqlValidator::pass6_action_security(&[action], opts) { + warn!( + host = %host_for_log, + reason = %e, + "webhook egress blocked by SSRF/security policy" + ); + return Err(WebhookError::SecurityViolation(e.to_string())); + } - let resolved = resolve_host(&host, port).await?; - if opts.block_private_webhooks { - if let Some(blocked) = first_blocked_address(&resolved) { - return Err(WebhookError::SecurityViolation(format!( - "webhook host '{host}' resolves to blocked address {blocked}" - ))); + // 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 { + if let Some(blocked) = first_blocked_address(&resolved) { + warn!( + host = %host, + blocked_ip = %blocked, + "webhook egress blocked: resolved address is in a disallowed range (DNS-rebind guard)" + ); + return Err(WebhookError::SecurityViolation(format!( + "webhook host '{host}' resolves to blocked address {blocked}" + ))); + } + } + let pinned_addr = resolved[0]; + debug!( + host = %host, + resolved_ip = %pinned_addr, + "webhook egress allowed by SSRF/security policy" + ); + let client = pinned_client(&host, pinned_addr, port)?; + + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + + let payload = json!({ + "event": "nuncio.filter.matched", + "timestamp": timestamp, + "rule_id": rule_id, + "message_id": message_id, + "subject": subject, + "sender": sender, + }); + + let payload_str = payload.to_string(); + + let mut mac = HmacSha256::new_from_slice(self.secret_key.as_bytes()) + .map_err(|e| WebhookError::SigningError(e.to_string()))?; + mac.update(format!("{timestamp}.{payload_str}").as_bytes()); + let signature = hex::encode(mac.finalize().into_bytes()); + + let send_result = client + .post(url) + .header("Content-Type", "application/json") + .header( + "X-Nuncio-Signature", + format!("t={timestamp},v1={signature}"), + ) + .body(payload_str) + .send() + .await; + + match send_result { + Ok(response) => { + let status = response.status().as_u16(); + debug!(host = %host, status, "webhook dispatched"); + Ok(status) + } + Err(e) => { + warn!(host = %host, error = %e, "webhook dispatch network error"); + Err(WebhookError::NetworkError(e.to_string())) + } } } - let pinned_addr = resolved[0]; - let client = pinned_client(&host, pinned_addr, port)?; - - let timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - - let payload = json!({ - "event": "nuncio.filter.matched", - "timestamp": timestamp, - "rule_id": rule_id, - "message_id": message_id, - "subject": subject, - "sender": sender, - }); - - let payload_str = payload.to_string(); - - let mut mac = HmacSha256::new_from_slice(self.secret_key.as_bytes()) - .map_err(|e| WebhookError::SigningError(e.to_string()))?; - mac.update(format!("{timestamp}.{payload_str}").as_bytes()); - let signature = hex::encode(mac.finalize().into_bytes()); - - let response = client - .post(url) - .header("Content-Type", "application/json") - .header( - "X-Nuncio-Signature", - format!("t={timestamp},v1={signature}"), - ) - .body(payload_str) - .send() - .await - .map_err(|e| WebhookError::NetworkError(e.to_string()))?; - - Ok(response.status().as_u16()) + .instrument(span) + .await } /// Dispatch a `CALL WEBHOOK` action with default production security options. @@ -192,9 +240,54 @@ impl WebhookDispatcher { #[cfg(test)] mod tests { use super::*; + use std::sync::Mutex; + use tracing::field::{Field, Visit}; + use tracing::span; use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; + /// Minimal `tracing::Subscriber` that records a formatted line per event + /// so tests can assert on emitted level + fields without pulling in + /// `tracing-subscriber`'s registry machinery. + struct CapturingSubscriber { + events: std::sync::Arc>>, + } + + struct LineVisitor<'a>(&'a mut String); + + impl Visit for LineVisitor<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.push_str(&format!(" {}={:?}", field.name(), value)); + } + } + + impl tracing::Subscriber for CapturingSubscriber { + fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { + true + } + + fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id { + span::Id::from_u64(1) + } + + fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {} + + fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {} + + fn event(&self, event: &tracing::Event<'_>) { + let mut line = format!("{}", event.metadata().level()); + let mut visitor = LineVisitor(&mut line); + event.record(&mut visitor); + if let Ok(mut events) = self.events.lock() { + events.push(line); + } + } + + fn enter(&self, _span: &span::Id) {} + + fn exit(&self, _span: &span::Id) {} + } + #[tokio::test] async fn test_pinned_client_connects_to_checked_ip_not_dns() { // `.invalid` is reserved by RFC 2606 to never resolve: any real DNS @@ -297,4 +390,81 @@ mod tests { ]; assert_eq!(first_blocked_address(&addrs), None); } + + #[tokio::test] + async fn test_blocked_webhook_logs_warn_with_host_and_reason() { + let events: std::sync::Arc>> = + std::sync::Arc::new(Mutex::new(Vec::new())); + let subscriber = CapturingSubscriber { + events: events.clone(), + }; + let _guard = tracing::subscriber::set_default(subscriber); + + let dispatcher = WebhookDispatcher::new("secret_key_123"); + let result = dispatcher + .dispatch( + "http://169.254.169.254/latest/meta-data", + "rule_1", + "msg_1", + "Test", + "a@b.com", + ) + .await; + + assert!(matches!(result, Err(WebhookError::SecurityViolation(_)))); + + let captured = events.lock().unwrap(); + assert!( + captured.iter().any(|line| line.contains("WARN") + && line.contains("blocked") + && line.contains("169.254.169.254")), + "expected a WARN log carrying the blocked host and reason, got: {captured:?}" + ); + assert!( + !captured.iter().any(|line| line.contains("secret_key_123")), + "the webhook HMAC secret must never appear in a log line" + ); + } + + #[tokio::test] + async fn test_allowed_webhook_logs_debug_not_warn() { + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/hook")) + .respond_with(ResponseTemplate::new(200)) + .mount(&mock_server) + .await; + + let events: std::sync::Arc>> = + std::sync::Arc::new(Mutex::new(Vec::new())); + let subscriber = CapturingSubscriber { + events: events.clone(), + }; + let _guard = tracing::subscriber::set_default(subscriber); + + let dispatcher = WebhookDispatcher::new("secret_key_123"); + let opts = ValidationOptions { + available_folders: None, + allowed_forward_domains: None, + block_private_webhooks: false, + }; + let url = format!("{}/hook", mock_server.uri()); + let status = dispatcher + .dispatch_with_options(&url, "rule_1", "msg_1", "Test", "a@b.com", &opts) + .await + .expect("dispatch to the mock server must succeed"); + assert_eq!(status, 200); + + let captured = events.lock().unwrap(); + assert!( + captured + .iter() + .any(|line| line.contains("DEBUG") && line.contains("allowed")), + "expected a DEBUG log for the allowed egress decision, got: {captured:?}" + ); + assert!( + !captured.iter().any(|line| line.contains("WARN")), + "an allowed dispatch must not emit a WARN, got: {captured:?}" + ); + } } -- 2.54.0