Merge pull request 'DLP at DATA: block, warn and override over SMTP and JMAP' (#104) from feature/dlp-data-stage into main

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),
                 },

crates/jmap-proto/src/error/set.rs

@@ -47,6 +47,18 @@ struct SetErrorInner<P: Property> {
     #[serde(skip_serializing_if = "Vec::is_empty")]
     #[serde(rename = "validationErrors")]
     validation_errors: Vec<ValidationError>,
+
+    // 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<DlpRule>,
+}
+
+/// 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<T: Property> SetError<T> {
             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<DlpRule>) -> Self {
+        self.0.rules = rules;
+        self
+    }
+
     pub fn with_description(mut self, description: impl Into<Cow<'static, str>>) -> Self {
         self.0.description = description.into().into();
         self
@@ -353,6 +380,7 @@ impl From<PatchError> for SetError<registry::schema::properties::Property> {
             object_id: None,
             linked_objects: Vec::new(),
             validation_errors: Vec::new(),
+            rules: Vec::new(),
         }))
     }
 }

crates/jmap-proto/src/object/email_submission.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * 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<EmailSubmissionProperty>),
 }
@@ -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('/') {

crates/jmap/src/submission/set.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * 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<MailFrom<Cow<'_, str>>> = None;
         let mut rcpt_to: Vec<RcptTo<Cow<'_, str>>> = Vec::new();
+        // inbuxa: DLP (dlp-and-mail-flow-rules spec, §2.5)
+        let mut dlp_override: Option<String> = 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::<Vec<_>>()
+                        .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!(

crates/smtp/src/core/mod.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * 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<SpfOutput>,
     pub spf_mail_from: Option<SpfOutput>,
     pub dnsbl_error: Option<Vec<u8>>,
+
+    // 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<String>,
+    pub dlp_refusal: Option<DlpRefusal>,
+}
+
+/// 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,
         }
     }
 }

crates/smtp/src/inbound/data.rs

@@ -738,6 +738,20 @@ impl<T: SessionStream> Session<T> {
             }
         }
 
+        // 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);

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<u8>),
+    /// Refuse, with this SMTP reply, and for a JMAP submission, why.
+    Refuse(Vec<u8>, Option<DlpRefusal>),
+}
+
+/// `[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::<Vec<_>>()
+        .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(&notice) {
+            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<T: SessionStream> Session<T> {
+    /// 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::<Vec<_>>()
+                    .join(", ");
+                if counts.is_empty() {
+                    format!("\"{}\"", m.name)
+                } else {
+                    format!("\"{}\" ({counts})", m.name)
+                }
+            })
+            .collect::<Vec<_>>()
+            .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);
+    }
+}

crates/smtp/src/inbound/mod.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * 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;

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.

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"], "[email protected]");
-    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::<Role>(id).await;
@@ -144,7 +174,13 @@ pub async fn test(test: &mut TestServer) {
         }
     }
     let officer = admin
-        .create_user_account("[email protected]", "officer-secret-7731", "Officer", &[], vec![])
+        .create_user_account(
+            "[email protected]",
+            "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::<Vec<_>>(),
+            "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("[email protected]");
+    let sender = admin
+        .create_user_account(
+            "[email protected]",
+            "dlp-sender-secret-5501",
+            "DLP sender",
+            &[],
+            vec![],
+        )
+        .await;
+    let (_, response) = call(
+        &sender,
+        "Identity/set",
+        json!({"create": {"i": {"name": "Sender", "email": "[email protected]"}}}),
+    )
+    .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 = ["[email protected]"];
+    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,
+        &["[email protected]"],
+        "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,
+        &["[email protected]", "[email protected]"],
+        "[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"] == "[email protected]"),
+        "{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("[email protected]").await;
     test.insert_account(admin);
     self::test(&mut test).await;
+    self::dlp(&mut test).await;
     if test.is_reset() {
         test.temp_dir.delete();
     }