From e0060c9e6e68c9152a3e7f05f8fd31925bf151a6 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Mon, 28 Sep 2026 17:52:33 -0700 Subject: [PATCH] DLP at DATA: block, warn and override over SMTP and JMAP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2f of the DLP and mail flow rules spec: the rules now run on mail an authenticated sender submits, after the DATA system script and before headers and DKIM signing (§2.1). - smtp/inbound/mailflow.rs: builds what the rules look at from the message (subject, the text version of each body, one level of attached messages, attachment text via the extractor, 10 MB of text at most) and the envelope (sender's groups and tenant; each recipient local or not, and its groups). Skipped entirely when no enabled rule applies to outgoing mail. Rules that can't be loaded refuse with a 451: nothing unchecked leaves. - Block: 550 5.7.1 with the rule's notice. Warn: 550 5.7.1 with the notice and how to override: "[override: reason]" at the start of the subject, taken out before the message goes on (settled answer 1). Until phase 3, a hold rule blocks rather than let mail through. - JMAP: EmailSubmission takes inbuxa:dlpOverride {reason}; a refusal comes back as inbuxa:dlpWarning or inbuxa:dlpBlocked with each rule's name and notice (description too, for older clients). - Audit: one record per DLP match, the sender as actor, action create, target a message: the recipient domains, each rule with its detectors' counts, the outcome, an override's reason. Never the matched text. No new audit action: an older node that meets one fails its daily clean-up, which would make rolling back unsafe (spec §2.7 updated). Tests: mail_rules_tests gains the DLP flow over JMAP (no rules, warning with rule and notice, local recipient not warned, override with a reason, block that no reason passes, the subject tag stripped from the delivered message, audit records with no card or key text). smtp inbound tests pass; system_tests passed twice after one timeout in the email delivery tests that didn't recur. --- crates/features/src/mailflow/engine.rs | 8 +- crates/jmap-proto/src/error/set.rs | 28 ++ .../jmap-proto/src/object/email_submission.rs | 7 + crates/jmap/src/submission/set.rs | 47 +- crates/smtp/src/core/mod.rs | 20 + crates/smtp/src/inbound/data.rs | 14 + crates/smtp/src/inbound/mailflow.rs | 427 ++++++++++++++++++ crates/smtp/src/inbound/mod.rs | 3 + docs/spec/features/dlp-and-mail-flow-rules.md | 21 +- tests/src/system/mail_rules.rs | 346 +++++++++++++- 10 files changed, 899 insertions(+), 22 deletions(-) create mode 100644 crates/smtp/src/inbound/mailflow.rs diff --git a/crates/features/src/mailflow/engine.rs b/crates/features/src/mailflow/engine.rs index e858a5b..e17b72a 100644 --- a/crates/features/src/mailflow/engine.rs +++ b/crates/features/src/mailflow/engine.rs @@ -46,7 +46,7 @@ pub struct Recipient<'a> { pub struct Attachment<'a> { pub name: Option<&'a str>, /// Declared type, or detected where the caller knows better. - pub content_type: &'a str, + pub content_type: Cow<'a, str>, pub size: u64, pub extracted: Extracted, } @@ -606,13 +606,15 @@ mod tests { attachments: vec![ Attachment { name: Some("plan.docx"), - content_type: "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + content_type: + "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + .into(), size: 40_000, extracted: Extracted::Text("IBAN GB29 NWBK 6016 1331 9268 19".into()), }, Attachment { name: Some("scan.pdf"), - content_type: "application/pdf", + content_type: "application/pdf".into(), size: 900_000, extracted: Extracted::NotInspectable(Why::Pdf), }, diff --git a/crates/jmap-proto/src/error/set.rs b/crates/jmap-proto/src/error/set.rs index 51c820f..fb96379 100644 --- a/crates/jmap-proto/src/error/set.rs +++ b/crates/jmap-proto/src/error/set.rs @@ -47,6 +47,18 @@ struct SetErrorInner { #[serde(skip_serializing_if = "Vec::is_empty")] #[serde(rename = "validationErrors")] validation_errors: Vec, + + // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.5): each rule that + // warned or blocked, with its notice + #[serde(skip_serializing_if = "Vec::is_empty")] + rules: Vec, +} + +/// inbuxa: a DLP rule named in an `inbuxa:dlpWarning` or `inbuxa:dlpBlocked`. +#[derive(Debug, Clone, serde::Serialize)] +pub struct DlpRule { + pub name: String, + pub notice: String, } #[derive(Debug, Clone)] @@ -127,6 +139,12 @@ pub enum SetErrorType { // inbuxa: a create that couldn't run (ai-explain spec: busy, timeout, …) #[serde(rename = "serverFail")] ServerFail, + // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.5): a warning the + // sender may answer with inbuxa:dlpOverride, and a block + #[serde(rename = "inbuxa:dlpWarning")] + DlpWarning, + #[serde(rename = "inbuxa:dlpBlocked")] + DlpBlocked, } impl SetErrorType { @@ -166,6 +184,8 @@ impl SetErrorType { SetErrorType::PrimaryKeyViolation => "primaryKeyViolation", SetErrorType::ValidationFailed => "validationFailed", SetErrorType::ServerFail => "serverFail", + SetErrorType::DlpWarning => "inbuxa:dlpWarning", + SetErrorType::DlpBlocked => "inbuxa:dlpBlocked", } } } @@ -180,9 +200,16 @@ impl SetError { object_id: None, linked_objects: Vec::new(), validation_errors: Vec::new(), + rules: Vec::new(), })) } + /// inbuxa: the DLP rules behind a warning or block. + pub fn with_dlp_rules(mut self, rules: Vec) -> Self { + self.0.rules = rules; + self + } + pub fn with_description(mut self, description: impl Into>) -> Self { self.0.description = description.into().into(); self @@ -353,6 +380,7 @@ impl From for SetError { object_id: None, linked_objects: Vec::new(), validation_errors: Vec::new(), + rules: Vec::new(), })) } } diff --git a/crates/jmap-proto/src/object/email_submission.rs b/crates/jmap-proto/src/object/email_submission.rs index c52780f..f16c4c7 100644 --- a/crates/jmap-proto/src/object/email_submission.rs +++ b/crates/jmap-proto/src/object/email_submission.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{ @@ -40,6 +42,9 @@ pub enum EmailSubmissionProperty { Displayed, DsnBlobIds, MdnBlobIds, + // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.5): `{"reason": ...}` + // to send despite a warning + DlpOverride, Pointer(JsonPointer), } @@ -90,6 +95,7 @@ impl Property for EmailSubmissionProperty { EmailSubmissionProperty::Id => "id", EmailSubmissionProperty::IdentityId => "identityId", EmailSubmissionProperty::MdnBlobIds => "mdnBlobIds", + EmailSubmissionProperty::DlpOverride => "inbuxa:dlpOverride", EmailSubmissionProperty::SendAt => "sendAt", EmailSubmissionProperty::ThreadId => "threadId", EmailSubmissionProperty::UndoStatus => "undoStatus", @@ -181,6 +187,7 @@ impl EmailSubmissionProperty { "displayed" => EmailSubmissionProperty::Displayed, "dsnBlobIds" => EmailSubmissionProperty::DsnBlobIds, "mdnBlobIds" => EmailSubmissionProperty::MdnBlobIds, + "inbuxa:dlpOverride" => EmailSubmissionProperty::DlpOverride, ) .or_else(|| { if allow_patch && value.contains('/') { diff --git a/crates/jmap/src/submission/set.rs b/crates/jmap/src/submission/set.rs index 90d8b5c..7c1a6fc 100644 --- a/crates/jmap/src/submission/set.rs +++ b/crates/jmap/src/submission/set.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use common::{ @@ -16,7 +18,7 @@ use email::{ submission::{Address, Delivered, DeliveryStatus, EmailSubmission, UndoStatus}, }; use jmap_proto::{ - error::set::{SetError, SetErrorType}, + error::set::{DlpRule, SetError, SetErrorType}, method::set::{SetRequest, SetResponse}, object::email_submission::{self, EmailSubmissionProperty, EmailSubmissionValue}, references::resolve::ResolveCreatedReference, @@ -379,6 +381,8 @@ impl EmailSubmissionSet for Server { }; let mut mail_from: Option>> = None; let mut rcpt_to: Vec>> = Vec::new(); + // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.5) + let mut dlp_override: Option = None; for (property, mut value) in object.into_expanded_object() { if let Err(err) = response.resolve_self_references(&mut value, 0, false) { @@ -493,6 +497,25 @@ impl EmailSubmissionSet for Server { (Key::Property(EmailSubmissionProperty::UndoStatus), Value::Element(_)) => { continue; } + // inbuxa: the sender's reason to send despite a DLP warning + (Key::Property(EmailSubmissionProperty::DlpOverride), Value::Object(value)) => { + let reason = value + .iter() + .find(|(key, _)| key.to_string() == "reason") + .and_then(|(_, value)| value.as_str().map(|r| r.trim().to_string())) + .filter(|r| !r.is_empty()); + match reason { + Some(reason) => dlp_override = Some(reason.chars().take(500).collect()), + None => { + return Ok(Err(SetError::invalid_properties() + .with_property(EmailSubmissionProperty::DlpOverride) + .with_description("An override needs a reason."))); + } + } + } + (Key::Property(EmailSubmissionProperty::DlpOverride), Value::Null) => { + continue; + } _ => { return Ok(Err(SetError::invalid_properties() .with_property(property.into_owned()) @@ -700,6 +723,7 @@ impl EmailSubmissionSet for Server { 0, ), ); + session.data.dlp_override = dlp_override; // Spawn SMTP session to avoid overflowing the stack let handle = tokio::spawn(async move { @@ -730,6 +754,27 @@ impl EmailSubmissionSet for Server { let response = session.queue_message().await; if let smtp::core::State::Accepted(queue_id) = session.state { Ok((responses, Some(queue_id))) + } else if let Some(refusal) = session.data.dlp_refusal.take() { + // inbuxa: DLP (§2.5): which rules, and what they say + let description = refusal + .rules + .iter() + .map(|(_, notice)| notice.as_str()) + .collect::>() + .join(" "); + Err(SetError::new(if refusal.blocked { + SetErrorType::DlpBlocked + } else { + SetErrorType::DlpWarning + }) + .with_description(description) + .with_dlp_rules( + refusal + .rules + .into_iter() + .map(|(name, notice)| DlpRule { name, notice }) + .collect(), + )) } else { Err( SetError::new(SetErrorType::ForbiddenToSend).with_description(format!( diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index 485b3c3..866e6cc 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{inbound::auth::SaslToken, queue::QueueId}; @@ -92,6 +94,20 @@ pub struct SessionData { pub spf_ehlo: Option, pub spf_mail_from: Option, pub dnsbl_error: Option>, + + // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.5): the reason a JMAP + // sender gave to send despite a warning, and why DATA refused a + // message, for the submission to report + pub dlp_override: Option, + pub dlp_refusal: Option, +} + +/// inbuxa: a DATA refusal by DLP rules: blocked, or a warning the sender +/// may override, with each rule's name and notice. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DlpRefusal { + pub blocked: bool, + pub rules: Vec<(String, String)>, } #[derive(Clone, Debug)] @@ -168,6 +184,8 @@ impl SessionData { spf_ehlo: None, spf_mail_from: None, dnsbl_error: None, + dlp_override: None, + dlp_refusal: None, } } } @@ -291,6 +309,8 @@ impl SessionData { spf_ehlo: None, spf_mail_from: None, dnsbl_error: None, + dlp_override: None, + dlp_refusal: None, } } } diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index 6c30dc6..f6970ca 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -738,6 +738,20 @@ impl Session { } } + // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.1): after the system + // script, before headers and signing + match self + .check_mail_rules(edited_message.as_deref().unwrap_or(raw_message.as_slice())) + .await + { + super::mailflow::Checked::Accept => {} + super::mailflow::Checked::Replace(message) => edited_message = Some(message), + super::mailflow::Checked::Refuse(reply, refusal) => { + self.data.dlp_refusal = refusal; + return reply.into(); + } + } + // Build message let mail_from = self.data.mail_from.clone().unwrap(); let rcpt_to = std::mem::take(&mut self.data.rcpt_to); diff --git a/crates/smtp/src/inbound/mailflow.rs b/crates/smtp/src/inbound/mailflow.rs new file mode 100644 index 0000000..7229a1f --- /dev/null +++ b/crates/smtp/src/inbound/mailflow.rs @@ -0,0 +1,427 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! inbuxa: DLP at DATA (dlp-and-mail-flow-rules spec, §2.1, §2.4–§2.7). +//! +//! Runs after the DATA system script and before headers and DKIM signing, +//! on mail an authenticated sender submits over SMTP or JMAP. The rules and +//! the detectors are `inbuxa_features::mailflow`; this is the glue: build +//! what they look at from the message, apply the decision, record it. + +use crate::core::{DlpRefusal, Session}; +use common::network::SessionStream; +use inbuxa_features::{ + audit::{Action, Actor, Outcome, Record, Target}, + mailflow::{ + cache, + engine::{ + Attachment, Content, Decision, Envelope, Outcome as RulesOutcome, Recipient, RuleRef, + }, + extract::{self, Extracted, Limits}, + rules::Kind, + }, +}; +use mail_parser::{HeaderName, Message, MessageParser, MimeHeaders, PartType}; +use std::{borrow::Cow, time::SystemTime}; + +/// How much text one message is read for; past it, the rest counts as +/// "can't be inspected" (§2.3). +const INSPECTION_LIMIT: usize = 10 * 1024 * 1024; + +/// What the check decided. +pub enum Checked { + /// Go on, with the message unchanged. + Accept, + /// Go on with this message instead (the override tag taken out). + Replace(Vec), + /// Refuse, with this SMTP reply, and for a JMAP submission, why. + Refuse(Vec, Option), +} + +/// `[override: reason]` at the start of a subject: the reason, and the +/// subject without it. +pub fn override_tag(subject: &str) -> Option<(String, String)> { + let trimmed = subject.trim_start(); + let head = trimmed.get(..10)?; + if !head.eq_ignore_ascii_case("[override:") { + return None; + } + let close = trimmed.find(']')?; + let reason = trimmed[10..close].trim(); + if reason.is_empty() { + return None; + } + Some(( + reason.chars().take(500).collect(), + trimmed[close + 1..].trim_start().to_string(), + )) +} + +/// One line of an SMTP reply: no line breaks, a sane length. +fn reply_text(text: &str) -> String { + text.split_whitespace() + .collect::>() + .join(" ") + .chars() + .take(400) + .collect() +} + +fn notices(rules: &[RuleRef]) -> String { + let mut seen = Vec::new(); + for rule in rules { + let notice = reply_text(&rule.notice); + if !seen.contains(¬ice) { + seen.push(notice); + } + } + seen.join(" ") +} + +fn refusal(blocked: bool, rules: &[RuleRef]) -> DlpRefusal { + DlpRefusal { + blocked, + rules: rules + .iter() + .map(|r| (r.name.clone(), r.notice.clone())) + .collect(), + } +} + +/// A message and the messages attached to it, one level down. +fn collect<'x>(message: &'x Message<'x>, content: &mut Content<'x>, budget: &mut usize, depth: u8) { + let add = |text: Cow<'x, str>, content: &mut Content<'x>, budget: &mut usize| { + if *budget == 0 { + content.truncated = true; + return; + } + if text.len() > *budget { + let mut cut = *budget; + while !text.is_char_boundary(cut) { + cut -= 1; + } + content.bodies.push(Cow::Owned(text[..cut].to_string())); + content.truncated = true; + *budget = 0; + } else { + *budget -= text.len(); + content.bodies.push(text); + } + }; + // The text version of each body (an HTML-only one converted), not both + // versions of the same alternative, so words aren't counted twice + for part in message.text_bodies() { + match &part.body { + PartType::Text(text) => add(Cow::Borrowed(text.as_ref()), content, budget), + PartType::Html(html) => add( + Cow::Owned(mail_parser::decoders::html::html_to_text(html)), + content, + budget, + ), + _ => {} + } + } + for part in message.attachments() { + if let (Some(inner), true) = (part.message(), depth == 0) { + if let Some(subject) = inner.subject() { + add(Cow::Borrowed(subject), content, budget); + } + collect(inner, content, budget, depth + 1); + continue; + } + let content_type = part + .content_type() + .map(|ct| match ct.subtype() { + Some(sub) => format!("{}/{}", ct.ctype(), sub), + None => ct.ctype().to_string(), + }) + .unwrap_or_default(); + let bytes = part.contents(); + let mut extracted = extract::extract( + &content_type, + part.attachment_name(), + bytes, + &Limits::default(), + ); + if let Extracted::Text(text) = &extracted { + if text.len() > *budget { + extracted = Extracted::NotInspectable(extract::Why::TooLarge); + } else { + *budget -= text.len(); + } + } + content.attachments.push(Attachment { + name: part.attachment_name(), + content_type: content_type.into(), + size: bytes.len() as u64, + extracted, + }); + } +} + +impl Session { + /// DLP on an outgoing message (§2.4). `message` is what the DATA stage + /// has so far (the script's replacement, if it made one). + pub async fn check_mail_rules(&self, message: &[u8]) -> Checked { + let Some(sender) = self.data.authenticated_as.as_ref() else { + // DLP checks outgoing mail only (settled) + return Checked::Accept; + }; + let (account_id, account) = (sender.account_id, sender.account.clone()); + let rules = match cache::compiled(self.server.store()).await { + Ok(rules) => rules, + Err(err) => { + trc::error!( + err.span_id(self.data.session_id) + .caused_by(trc::location!()) + .details("Failed to load mail rules") + ); + // Fail closed: a message nobody could check doesn't leave + return Checked::Refuse( + b"451 4.3.0 This message couldn't be checked against the server's rules. Try again later.\r\n" + .to_vec(), + None, + ); + } + }; + if !rules.applies_to(true) { + return Checked::Accept; + } + + let parsed = MessageParser::new().parse(message); + let subject = parsed + .as_ref() + .and_then(|m| m.subject()) + .unwrap_or_default(); + let jmap_override = self.data.dlp_override.clone(); + let tag = override_tag(subject); + let checked_subject = tag.as_ref().map_or(subject, |(_, rest)| rest.as_str()); + + let mut content = Content { + subject: checked_subject, + size: message.len() as u64, + ..Default::default() + }; + let mut budget = INSPECTION_LIMIT; + match &parsed { + Some(parsed) => { + content.headers = parsed + .headers() + .iter() + .filter_map(|h| h.value.as_text().map(|v| (h.name.as_str(), v))) + .collect(); + collect(parsed, &mut content, &mut budget, 0); + } + // Nothing a rule could read: say so, rather than pass it + None => content.truncated = true, + } + let mut recipient_groups = Vec::with_capacity(self.data.rcpt_to.len()); + for rcpt in &self.data.rcpt_to { + let local = self + .server + .domain(&rcpt.domain) + .await + .ok() + .flatten() + .is_some(); + let groups = if local { + match self + .server + .account_id_from_email(&rcpt.address_lcase, false) + .await + { + Ok(Some(id)) => self + .server + .account(id) + .await + .map(|a| a.id_member_of.to_vec()) + .unwrap_or_default(), + _ => Vec::new(), + } + } else { + Vec::new() + }; + recipient_groups.push((local, groups)); + } + let sender_address = self + .data + .mail_from + .as_ref() + .map(|m| m.address_lcase.clone()) + .unwrap_or_default(); + let envelope = Envelope { + outgoing: true, + sender: &sender_address, + sender_groups: &account.id_member_of, + sender_tenant: account.id_tenant, + recipients: self + .data + .rcpt_to + .iter() + .zip(&recipient_groups) + .map(|(rcpt, (local, groups))| Recipient { + address: &rcpt.address_lcase, + local: *local, + groups, + }) + .collect(), + }; + + let outcome = rules.evaluate(&envelope, &content); + let override_reason = + jmap_override.or_else(|| tag.as_ref().map(|(reason, _)| reason.clone())); + let decision = outcome.decision(override_reason.is_some()); + let mut domains: Vec<&str> = self + .data + .rcpt_to + .iter() + .map(|r| r.domain.as_str()) + .collect(); + domains.sort_unstable(); + domains.dedup(); + let domains = domains.join(", "); + drop(envelope); + + self.record_dlp( + account_id, + &account, + &outcome, + &decision, + override_reason.as_deref(), + &domains, + ) + .await; + + match decision { + Decision::Block(rules) | Decision::Hold { rules, .. } => { + // Hold for review is phase 3: until then a hold rule blocks, + // rather than let the message through unreviewed + let refusal = refusal(true, &rules); + Checked::Refuse(format!("550 5.7.1 {}\r\n", notices(&rules)).into_bytes(), Some(refusal)) + } + Decision::Warn(rules) => Checked::Refuse( + format!( + "550 5.7.1 {} To send anyway, start the subject with [override: your reason]\r\n", + notices(&rules) + ) + .into_bytes(), + Some(refusal(false, &rules)), + ), + Decision::Pass => match (tag, &parsed) { + // The tag was an instruction to the server, not part of the + // subject: it doesn't go out + (Some((_, rest)), Some(parsed)) => parsed + .headers() + .iter() + .find(|h| h.name == HeaderName::Subject) + .map_or(Checked::Accept, |header| { + let mut out = Vec::with_capacity(message.len()); + out.extend_from_slice(&message[..header.offset_field as usize]); + out.extend_from_slice(b"Subject: "); + out.extend_from_slice(rest.as_bytes()); + out.extend_from_slice(b"\r\n"); + out.extend_from_slice(&message[header.offset_end as usize..]); + Checked::Replace(out) + }), + _ => Checked::Accept, + }, + } + } + + /// One audit record per message a DLP rule matched (§2.7): who sent it, + /// where to, which rules and each detector's count, what happened, and + /// an override's reason. Never the matched text. + async fn record_dlp( + &self, + account_id: u32, + account: &common::auth::AccountCache, + outcome: &RulesOutcome, + decision: &Decision, + override_reason: Option<&str>, + domains: &str, + ) { + let dlp: Vec<_> = outcome + .matched + .iter() + .filter(|m| m.kind == Kind::Dlp) + .collect(); + if dlp.is_empty() { + return; + } + let rules = dlp + .iter() + .map(|m| { + let counts = m + .counts + .iter() + .map(|(id, n)| format!("{id} {n}")) + .collect::>() + .join(", "); + if counts.is_empty() { + format!("\"{}\"", m.name) + } else { + format!("\"{}\" ({counts})", m.name) + } + }) + .collect::>() + .join("; "); + let (what, outcome, reason) = match decision { + Decision::Block(_) | Decision::Hold { .. } => { + ("blocked", Outcome::refused("inbuxa:dlpBlocked", None), None) + } + Decision::Warn(_) => ("warned", Outcome::refused("inbuxa:dlpWarning", None), None), + Decision::Pass => ( + "sent after a warning", + Outcome::success(), + override_reason.map(str::to_string), + ), + }; + let at = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map_or(0, |d| d.as_millis() as u64); + self.server + .audit_note(Record { + at, + actor: Actor::account(account_id, account.name.to_string(), account.id_tenant), + via: None, + remote_ip: Some(self.data.remote_ip), + action: Action::Create, + target: Target { + kind: "message".into(), + id: None, + name: None, + account_id: Some(account_id), + tenant_id: account.id_tenant, + }, + changes: vec![], + details: Some(format!("DLP {what}, to {domains}: {rules}")), + reason, + outcome, + }) + .await; + } +} + +#[cfg(test)] +mod tests { + use super::override_tag; + + #[test] + fn override_tags() { + assert_eq!( + override_tag("[override: client asked for it] Card details"), + Some(("client asked for it".into(), "Card details".into())) + ); + assert_eq!( + override_tag(" [OVERRIDE:yes]x"), + Some(("yes".into(), "x".into())) + ); + assert_eq!(override_tag("[override: ] x"), None); + assert_eq!(override_tag("Re: [override: no] x"), None); + assert_eq!(override_tag("[override: unclosed"), None); + assert_eq!(override_tag(""), None); + } +} diff --git a/crates/smtp/src/inbound/mod.rs b/crates/smtp/src/inbound/mod.rs index 99585af..78d5447 100644 --- a/crates/smtp/src/inbound/mod.rs +++ b/crates/smtp/src/inbound/mod.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use mail_auth::{DkimResult, DmarcResult, IprevResult, SpfResult, dmarc::Policy}; @@ -13,6 +15,7 @@ pub mod dkim; pub mod ehlo; pub mod hooks; pub mod mail; +pub mod mailflow; // inbuxa: DLP and mail flow rules pub mod milter; pub mod rcpt; pub mod session; diff --git a/docs/spec/features/dlp-and-mail-flow-rules.md b/docs/spec/features/dlp-and-mail-flow-rules.md index ad13146..b017619 100644 --- a/docs/spec/features/dlp-and-mail-flow-rules.md +++ b/docs/spec/features/dlp-and-mail-flow-rules.md @@ -293,13 +293,24 @@ once it's held (the webmail says so). Held messages count against no one's quota. Each held message and each decision is in the audit log. +**Until phase 3** a hold rule blocks, with its notice, rather than let the +message through unreviewed. + ### 2.7 What's recorded -Every DLP match writes one audit record: actor **DLP** (a system actor), -target the message (queue id, sender, recipient domains), the rules and each -detector's count, the action, and for an override the sender's reason. -**Never the matched text**: the log would otherwise become a second copy of -what the policy was keeping in. A card number isn't written, even masked. +Every DLP match writes one audit record, and **never the matched text**: +the log would otherwise become a second copy of what the policy was keeping +in. A card number isn't written, even masked. + +**As built (phase 2f).** The actor is the sender (they sent it; filtering by +sender is what a reviewer wants), the action `create`, the target kind +`message`. The details say what happened, where to, and each rule with its +detectors' counts: `DLP warned, to elsewhere.org: "Cards leaving" +(payment-card 1)`. A block or an unanswered warning is recorded as refused +(`inbuxa:dlpBlocked`, `inbuxa:dlpWarning`); an override as a success, with +the sender's reason. No new audit action was added: an older node reading a +record with an action it doesn't know fails its daily clean-up, so a new +action would make rolling back unsafe. Transport rules that change a message record the rule and action the same way. Unmatched mail writes nothing. diff --git a/tests/src/system/mail_rules.rs b/tests/src/system/mail_rules.rs index 38b9c26..5f69cb1 100644 --- a/tests/src/system/mail_rules.rs +++ b/tests/src/system/mail_rules.rs @@ -21,6 +21,8 @@ use serde_json::{Value, json}; const USING: &[&str] = &[ "urn:ietf:params:jmap:core", + "urn:ietf:params:jmap:mail", + "urn:ietf:params:jmap:submission", "urn:inbuxa:jmap", "urn:inbuxa:jmap:registry", ]; @@ -29,13 +31,18 @@ async fn call(account: &Account, method: &str, mut arguments: Value) -> (String, if arguments.get("accountId").is_none() { arguments["accountId"] = account.id_string().into(); } - let response = account.jmap_request(USING, json!([[method, arguments, "0"]])).await; + let response = account + .jmap_request(USING, json!([[method, arguments, "0"]])) + .await; let call = response .0 .pointer("/methodResponses/0") .cloned() .unwrap_or_else(|| panic!("{method}: {}", response.0)); - (call[0].as_str().unwrap_or_default().to_string(), call[1].clone()) + ( + call[0].as_str().unwrap_or_default().to_string(), + call[1].clone(), + ) } fn dlp_rule() -> Value { @@ -94,10 +101,16 @@ pub async fn test(test: &mut TestServer) { let (_, response) = call(&admin, "inbuxa:MailRule/get", json!({"ids": [dlp_id]})).await; let rule = &response["list"][0]; - assert_eq!(rule["conditions"][1]["detectors"][0]["atLeast"], 5, "{rule}"); + assert_eq!( + rule["conditions"][1]["detectors"][0]["atLeast"], 5, + "{rule}" + ); assert_eq!(rule["actions"][0]["notifySender"], true); assert_eq!(rule["createdBy"], "admin@example.com"); - assert!(rule["createdAt"].as_str().is_some_and(|d| d.ends_with('Z')), "{rule}"); + assert!( + rule["createdAt"].as_str().is_some_and(|d| d.ends_with('Z')), + "{rule}" + ); // Checked when written let mut inbound = dlp_rule(); @@ -110,9 +123,15 @@ pub async fn test(test: &mut TestServer) { json!({"create": {"a": inbound, "b": unknown, "c": {"name": "x", "kind": "dlp"}}}), ) .await; - assert_eq!(response["notCreated"]["a"]["properties"][0], "direction", "{response}"); + assert_eq!( + response["notCreated"]["a"]["properties"][0], "direction", + "{response}" + ); assert!( - response["notCreated"]["b"]["description"].as_str().unwrap().contains("no-such-detector"), + response["notCreated"]["b"]["description"] + .as_str() + .unwrap() + .contains("no-such-detector"), "{response}" ); assert!(response["notCreated"].get("c").is_some(), "{response}"); @@ -127,15 +146,26 @@ pub async fn test(test: &mut TestServer) { }}), ) .await; - assert!(response["updated"].get(transport_id.as_str()).is_some(), "{response}"); - assert_eq!(response["notUpdated"][dlp_id.as_str()]["properties"][0], "createdBy", "{response}"); + assert!( + response["updated"].get(transport_id.as_str()).is_some(), + "{response}" + ); + assert_eq!( + response["notUpdated"][dlp_id.as_str()]["properties"][0], + "createdBy", + "{response}" + ); assert_eq!(names(&admin).await, vec!["Cards leaving", "Footer"]); // A compliance officer sees DLP rules, not mail flow rules, and changes // neither (settled answer 4) let mut officer_role = None; for id in admin - .registry_query_ids(ObjectType::Role, Vec::<(&str, &str)>::new(), Vec::<&str>::new()) + .registry_query_ids( + ObjectType::Role, + Vec::<(&str, &str)>::new(), + Vec::<&str>::new(), + ) .await { let role = admin.registry_get::(id).await; @@ -144,7 +174,13 @@ pub async fn test(test: &mut TestServer) { } } let officer = admin - .create_user_account("rules-officer@example.com", "officer-secret-7731", "Officer", &[], vec![]) + .create_user_account( + "rules-officer@example.com", + "officer-secret-7731", + "Officer", + &[], + vec![], + ) .await; admin .registry_update_object( @@ -156,8 +192,12 @@ pub async fn test(test: &mut TestServer) { ) .await; assert_eq!(names(&officer).await, vec!["Cards leaving"]); - let (name, response) = - call(&officer, "inbuxa:MailRule/set", json!({"create": {"d": dlp_rule()}})).await; + let (name, response) = call( + &officer, + "inbuxa:MailRule/set", + json!({"create": {"d": dlp_rule()}}), + ) + .await; assert_eq!(name, "error", "the officer created a DLP rule: {response}"); // Deleted @@ -167,7 +207,11 @@ pub async fn test(test: &mut TestServer) { json!({"destroy": [transport_id]}), ) .await; - assert_eq!(response["destroyed"][0], transport_id.as_str(), "{response}"); + assert_eq!( + response["destroyed"][0], + transport_id.as_str(), + "{response}" + ); assert_eq!(names(&admin).await, vec!["Cards leaving"]); // Every change is in the audit log @@ -183,6 +227,281 @@ pub async fn test(test: &mut TestServer) { ); } +/// A draft from `sender`, submitted; the submission call's response. +async fn submit( + sender: &Account, + identity: &str, + mailbox: &str, + to: &[&str], + subject: &str, + body: &str, + dlp_override: Option<&str>, +) -> Value { + let (_, response) = call( + sender, + "Email/set", + json!({"create": {"e": { + "mailboxIds": {mailbox: true}, + "from": [{"email": sender.name()}], + "to": to.iter().map(|a| json!({"email": a})).collect::>(), + "subject": subject, + "bodyValues": {"b": {"value": body}}, + "textBody": [{"partId": "b", "type": "text/plain"}] + }}}), + ) + .await; + let email = response["created"]["e"]["id"] + .as_str() + .unwrap_or_else(|| panic!("draft: {response}")) + .to_string(); + let mut create = json!({"emailId": email, "identityId": identity}); + if let Some(reason) = dlp_override { + create["inbuxa:dlpOverride"] = json!({"reason": reason}); + } + call( + sender, + "EmailSubmission/set", + json!({"create": {"s": create}}), + ) + .await + .1 +} + +/// DLP at DATA (§2.4–§2.7): a warning answered with a reason, a block no +/// reason answers, the subject tag, and records that name the rules and +/// counts but never what was found. +pub async fn dlp(test: &mut TestServer) { + println!("Running DLP sending tests..."); + let admin = test.account("admin@example.com"); + let sender = admin + .create_user_account( + "dlp-sender@example.com", + "dlp-sender-secret-5501", + "DLP sender", + &[], + vec![], + ) + .await; + let (_, response) = call( + &sender, + "Identity/set", + json!({"create": {"i": {"name": "Sender", "email": "dlp-sender@example.com"}}}), + ) + .await; + let identity = response["created"]["i"]["id"] + .as_str() + .unwrap_or_else(|| panic!("{response}")) + .to_string(); + let (_, response) = call( + &sender, + "Mailbox/set", + json!({"create": {"m": {"name": "DLP drafts"}}}), + ) + .await; + let mailbox = response["created"]["m"]["id"].as_str().unwrap().to_string(); + let outside = ["someone@elsewhere.org"]; + let card = "Card 4242 4242 4242 4242, expires 12/31"; + + // No rules: sent + let response = submit( + &sender, &identity, &mailbox, &outside, "Numbers", card, None, + ) + .await; + assert!( + response["created"].get("s").is_some(), + "no rules: {response}" + ); + + // A warning: refused with the rule and its notice, then sent with a reason + let (_, response) = call( + &admin, + "inbuxa:MailRule/set", + json!({"create": {"w": { + "name": "Cards leaving", "kind": "dlp", "direction": "outgoing", + "conditions": [{"type": "recipientOutside"}, + {"type": "detected", "detectors": [{"id": "payment-card"}]}], + "actions": [{"type": "warn", "notice": "This looks like a card number."}] + }}}), + ) + .await; + let warn_rule = response["created"]["w"]["id"] + .as_str() + .unwrap_or_else(|| panic!("{response}")) + .to_string(); + let response = submit( + &sender, &identity, &mailbox, &outside, "Numbers", card, None, + ) + .await; + let refused = &response["notCreated"]["s"]; + assert_eq!(refused["type"], "inbuxa:dlpWarning", "{response}"); + assert_eq!(refused["rules"][0]["name"], "Cards leaving"); + assert_eq!(refused["description"], "This looks like a card number."); + // Inside the server: no warning + let response = submit( + &sender, + &identity, + &mailbox, + &["dlp-sender@example.com"], + "Numbers", + card, + None, + ) + .await; + assert!( + response["created"].get("s").is_some(), + "local recipient: {response}" + ); + let response = submit( + &sender, + &identity, + &mailbox, + &outside, + "Numbers", + card, + Some("The client asked for it"), + ) + .await; + assert!( + response["created"].get("s").is_some(), + "overridden: {response}" + ); + + // A block: no reason gets past it + let (_, response) = call( + &admin, + "inbuxa:MailRule/set", + json!({"create": {"b": { + "name": "Keys", "kind": "dlp", "direction": "outgoing", + "conditions": [{"type": "detected", "detectors": [{"id": "private-key"}]}], + "actions": [{"type": "block", "notice": "Private keys don't leave by mail."}] + }}}), + ) + .await; + assert!(response["created"].get("b").is_some(), "{response}"); + let key = "-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQ\n-----END OPENSSH PRIVATE KEY-----"; + let response = submit( + &sender, + &identity, + &mailbox, + &outside, + "Key", + key, + Some("Please"), + ) + .await; + assert_eq!( + response["notCreated"]["s"]["type"], "inbuxa:dlpBlocked", + "{response}" + ); + + // The subject tag overrides, and doesn't go out: the sender's own copy + // arrives without it + let response = submit( + &sender, + &identity, + &mailbox, + &["someone@elsewhere.org", "dlp-sender@example.com"], + "[override: Agreed with finance] Tagged numbers", + card, + None, + ) + .await; + assert!( + response["created"].get("s").is_some(), + "tag override: {response}" + ); + let mut delivered = Vec::new(); + for _ in 0..50 { + let (_, response) = call( + &sender, + "Email/query", + json!({"filter": {"text": "Tagged"}}), + ) + .await; + let ids = response["ids"].clone(); + let (_, response) = call( + &sender, + "Email/get", + json!({"ids": ids, "properties": ["subject", "mailboxIds"]}), + ) + .await; + delivered = response["list"] + .as_array() + .unwrap() + .iter() + .filter(|e| !e["mailboxIds"].as_object().unwrap().contains_key(&mailbox)) + .map(|e| e["subject"].as_str().unwrap_or_default().to_string()) + // The bounce for the unreachable outside address quotes it + .filter(|subject| !subject.starts_with("Failed to deliver")) + .collect(); + if !delivered.is_empty() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + assert_eq!( + delivered, + vec!["Tagged numbers".to_string()], + "delivered subject" + ); + + // Recorded: sender, what happened, rules and counts, the reason; never + // the number itself + let (_, response) = call( + &admin, + "inbuxa:AuditEvent/query", + json!({"filter": {"targetKind": "message"}}), + ) + .await; + let ids = response["ids"].clone(); + let (_, response) = call(&admin, "inbuxa:AuditEvent/get", json!({"ids": ids})).await; + let events = response["list"].as_array().unwrap(); + let details: Vec<&str> = events + .iter() + .filter_map(|e| e["details"].as_str()) + .collect(); + assert!( + details + .iter() + .any(|d| d + .starts_with("DLP warned, to elsewhere.org: \"Cards leaving\" (payment-card 1)")), + "{details:?}" + ); + assert!( + details.iter().any(|d| d.starts_with("DLP blocked")), + "{details:?}" + ); + assert!( + events.iter().any( + |e| e["reason"] == "The client asked for it" && e["outcome"]["status"] == "success" + ), + "{response}" + ); + assert!( + events.iter().any(|e| e["reason"] == "Agreed with finance"), + "{response}" + ); + assert!( + events + .iter() + .all(|e| e["actor"]["name"] == "dlp-sender@example.com"), + "{response}" + ); + let all = response.to_string(); + assert!( + !all.contains("4242") && !all.contains("b3BlbnNz"), + "matched text in the audit log" + ); + + // Rules off again for the tests that follow + call( + &admin, + "inbuxa:MailRule/set", + json!({"destroy": [warn_rule]}), + ) + .await; +} + #[ignore] #[tokio::test(flavor = "multi_thread")] pub async fn mail_rules_tests() { @@ -195,6 +514,7 @@ pub async fn mail_rules_tests() { let admin = test.create_admin_account("admin@example.com").await; test.insert_account(admin); self::test(&mut test).await; + self::dlp(&mut test).await; if test.is_reset() { test.temp_dir.delete(); }