Cargo.lock

@@ -4000,9 +4000,11 @@ dependencies = [
  "http-body-util",
  "hyper",
  "hyper-util",
+ "inbuxa-outbound-tls",
  "inbuxa-relay-proto",
  "rcgen",
  "rustls",
+ "rustls-native-certs",
  "rustls-pki-types",
  "rustls-webpki",
  "serde",

crates/outbound-tls/src/dnssec.rs

@@ -49,6 +49,16 @@ pub enum TlsaResult {
 /// An answer, and until when it may be cached: `None` when it mustn't be.
 pub type Cacheable<T> = (T, Option<Instant>);
 
+/// Whether `resolver` validates DNSSEC: the root's DNSKEY comes back
+/// proved secure. Through a resolver that strips signatures, DANE can't
+/// work, and every lookup would look insecure.
+pub async fn validates(resolver: &TokioResolver) -> bool {
+    resolver
+        .lookup(Name::root(), RecordType::DNSKEY)
+        .await
+        .is_ok_and(|lookup| lookup.answers().iter().any(|r| r.proof.is_secure()))
+}
+
 /// MX records for `name`, a fully qualified name.
 pub async fn mx(
     resolvers: &Resolvers<'_>,

crates/relay-proto/src/types.rs

@@ -65,6 +65,8 @@ pub struct RelaySettings {
     pub outbound_retry: Vec<u32>,
     /// Seconds before outbound mail is given up.
     pub outbound_expiry: u32,
+    /// TLS where neither DANE nor MTA-STS asks for it.
+    pub outbound_tls: OutboundTls,
     /// The STARTTLS certificate for port 25.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub smtp_certificate: Option<SmtpCertificate>,
@@ -91,6 +93,7 @@ impl Default for RelaySettings {
             spool_max_bytes: 2 * 1024 * 1024 * 1024,
             outbound_retry: vec![120, 300, 600, 1800, 3600, 7200, 14_400],
             outbound_expiry: 3 * 86_400,
+            outbound_tls: OutboundTls::default(),
             smtp_certificate: None,
             sealing_key: None,
             recipient_key: None,
@@ -149,6 +152,19 @@ impl RelaySettings {
     }
 }
 
+/// TLS for outbound mail where the destination publishes no DANE or
+/// MTA-STS policy; those are always enforced (spec "Outbound").
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum OutboundTls {
+    /// STARTTLS when offered, without checking the certificate: it could
+    /// be stripped anyway.
+    #[default]
+    Opportunistic,
+    /// No delivery without STARTTLS; the certificate still isn't checked.
+    Required,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase", default)]
 pub struct RateLimits {

crates/relay/Cargo.toml

@@ -13,6 +13,8 @@ path = "src/main.rs"
 # small, and most of it the server already uses.
 [dependencies]
 inbuxa-relay-proto = { path = "../relay-proto" }
+inbuxa-outbound-tls = { path = "../outbound-tls" }
+rustls-native-certs = "0.8"
 aws-lc-rs = { version = "1" }
 smtp-proto = { version = "0.2.3" }
 tokio = { version = "1.53", features = ["rt-multi-thread", "net", "io-util", "time", "sync", "macros", "signal", "fs"] }
@@ -22,7 +24,7 @@ rustls-pki-types = { version = "1", features = ["std"] }
 rustls-webpki = { version = "0.103", default-features = false, features = ["std", "aws-lc-rs"] }
 rcgen = { version = "0.14", default-features = false, features = ["crypto", "pem", "aws_lc_rs"] }
 x509-parser = { version = "0.18" }
-hyper = { version = "1.11.1", features = ["server", "http1"] }
+hyper = { version = "1.11.1", features = ["server", "client", "http1"] }
 hyper-util = { version = "0.1.20", features = ["tokio"] }
 http-body-util = "0.1.5"
 tokio-tungstenite = "0.30"

crates/relay/src/app.rs

@@ -25,7 +25,7 @@ use std::{
         atomic::{AtomicU32, AtomicU64, Ordering},
     },
 };
-use tokio::sync::{broadcast, watch};
+use tokio::sync::{Notify, broadcast, watch};
 
 pub const VERSION: &str = env!("CARGO_PKG_VERSION");
 
@@ -45,6 +45,8 @@ pub struct App {
     pub recipients: RwLock<Recipients>,
     pub spool: Mutex<Spool>,
     pub outbox: Mutex<Outbox>,
+    /// Wakes the delivery engine when mail is handed over.
+    pub outbound_wake: Notify,
     pub events: Mutex<Events>,
     pub push: broadcast::Sender<Changed>,
     /// Link connections, and those with push on for `Inbound`.
@@ -79,6 +81,7 @@ impl App {
             recipients: RwLock::new(recipients),
             spool: Mutex::new(spool),
             outbox: Mutex::new(outbox),
+            outbound_wake: Notify::new(),
             events: Mutex::new(events),
             push,
             link_clients: AtomicU32::new(0),
@@ -113,6 +116,11 @@ impl App {
         let _ = self.push.send(Changed { type_name, state });
     }
 
+    pub fn outbound_changed(&self) {
+        let state = self.outbox.lock().expect("outbox").state();
+        self.notify(type_name::OUTBOUND, state);
+    }
+
     pub fn inbound_changed(&self) {
         let state = self.spool.lock().expect("spool").state();
         self.notify(type_name::INBOUND, state);

crates/relay/src/deliver/client.rs

@@ -0,0 +1,592 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! One SMTP client session to one address of one MX host (RFC 5321): the
+//! greeting, EHLO, STARTTLS as the TLS mode asks, and one transaction.
+
+use super::tls::{self, TlsMode, TlsOutcome};
+use inbuxa_relay_proto::types::{Address, TlsInfo};
+use std::{net::SocketAddr, path::Path, time::Duration};
+use tokio::{
+    io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader},
+    net::TcpStream,
+};
+
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
+/// RFC 5321 §4.5.3.2 asks for five minutes for most replies.
+const COMMAND_TIMEOUT: Duration = Duration::from_secs(300);
+/// And ten for the reply to the final dot.
+const DATA_TIMEOUT: Duration = Duration::from_secs(600);
+const LINE_MAX: u64 = 4096;
+const REPLY_LINES_MAX: usize = 100;
+
+/// An SMTP reply, all its lines joined.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Reply {
+    pub code: u16,
+    pub text: String,
+}
+
+impl Reply {
+    pub fn class(&self) -> u16 {
+        self.code / 100
+    }
+
+    /// As `deliveryStatus` shows it: `250 2.0.0 OK`.
+    pub fn line(&self) -> String {
+        if self.text.is_empty() {
+            self.code.to_string()
+        } else {
+            format!("{} {}", self.code, self.text.replace('\n', " "))
+        }
+    }
+}
+
+/// What a session left each recipient with.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RcptOutcome {
+    Delivered(String),
+    Refused(String),
+    Deferred(String),
+}
+
+/// A session that couldn't reach a transaction: the next address or host
+/// may do better. `tls` is set when a TLS policy failed, for TLS-RPT.
+#[derive(Debug)]
+pub struct HostFailure {
+    pub reason: String,
+    pub tls: Option<tls::TlsFailure>,
+}
+
+fn host_failure(reason: impl Into<String>) -> HostFailure {
+    HostFailure {
+        reason: reason.into(),
+        tls: None,
+    }
+}
+
+pub struct Session<'a> {
+    pub addr: SocketAddr,
+    /// The MX host, for SNI and certificate checks.
+    pub mx: &'a str,
+    /// The recipient domain, a DANE reference identifier too.
+    pub domain: &'a str,
+    /// This relay's name, for EHLO.
+    pub helo: &'a str,
+    pub mode: TlsMode,
+    pub mail_from: &'a Address,
+    pub rcpt_to: &'a [&'a Address],
+    pub message: &'a Path,
+    pub size: u64,
+}
+
+/// What came of a session that reached MAIL FROM.
+#[derive(Debug)]
+pub struct Delivered {
+    pub outcomes: Vec<RcptOutcome>,
+    pub tls: Option<TlsInfo>,
+}
+
+impl Session<'_> {
+    pub async fn run(&self) -> Result<Delivered, HostFailure> {
+        match self.attempt(self.mode.clone()).await {
+            // Opportunistic TLS that fails to set up: once more in the
+            // clear, as other MTAs do; nothing was promised.
+            Err(HostFailure {
+                tls: Some(failure), ..
+            }) if self.mode == TlsMode::Opportunistic && failure.handshake => {
+                self.attempt(TlsMode::None).await
+            }
+            other => other,
+        }
+    }
+
+    async fn attempt(&self, mode: TlsMode) -> Result<Delivered, HostFailure> {
+        let tcp = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(self.addr))
+            .await
+            .map_err(|_| host_failure(format!("Connection to {} timed out", self.mx)))?
+            .map_err(|e| host_failure(format!("Connection to {} failed: {e}", self.mx)))?;
+        let mut conn = BufReader::new(tcp);
+        let greeting = read_reply(&mut conn, COMMAND_TIMEOUT).await?;
+        if greeting.class() != 2 {
+            return Err(host_failure(format!(
+                "{} greeted with {}",
+                self.mx,
+                greeting.line()
+            )));
+        }
+        let ehlo = self.ehlo(&mut conn).await?;
+        let wants_tls = mode != TlsMode::None;
+        if !ehlo.starttls {
+            if mode.is_required() {
+                return Err(HostFailure {
+                    reason: format!("{} doesn't offer STARTTLS", self.mx),
+                    tls: Some(tls::TlsFailure::starttls_not_supported()),
+                });
+            }
+            return self.transaction(conn, &ehlo, None).await;
+        }
+        if !wants_tls {
+            return self.transaction(conn, &ehlo, None).await;
+        }
+        command(&mut conn, "STARTTLS\r\n").await?;
+        let reply = read_reply(&mut conn, COMMAND_TIMEOUT).await?;
+        if reply.class() != 2 {
+            return Err(HostFailure {
+                reason: format!("{} refused STARTTLS: {}", self.mx, reply.line()),
+                tls: mode
+                    .is_required()
+                    .then(tls::TlsFailure::starttls_not_supported),
+            });
+        }
+        // Anything sent before the handshake would be read as if it had
+        // come over TLS (CVE-2011-0411).
+        if !conn.buffer().is_empty() {
+            return Err(host_failure(format!(
+                "{} sent data after accepting STARTTLS",
+                self.mx
+            )));
+        }
+        let TlsOutcome { stream, info } = tls::handshake(
+            conn.into_inner(),
+            self.mx,
+            self.domain,
+            &mode,
+            COMMAND_TIMEOUT,
+        )
+        .await
+        .map_err(|failure| HostFailure {
+            reason: format!("TLS with {} failed: {}", self.mx, failure.reason),
+            tls: Some(failure),
+        })?;
+        let mut conn = BufReader::new(stream);
+        let ehlo = self.ehlo(&mut conn).await?;
+        self.transaction(conn, &ehlo, Some(info)).await
+    }
+
+    async fn ehlo<S: AsyncRead + AsyncWrite + Unpin>(
+        &self,
+        conn: &mut BufReader<S>,
+    ) -> Result<Ehlo, HostFailure> {
+        command(conn, &format!("EHLO {}\r\n", self.helo)).await?;
+        let reply = read_reply(conn, COMMAND_TIMEOUT).await?;
+        if reply.class() != 2 {
+            return Err(host_failure(format!(
+                "{} refused EHLO: {}",
+                self.mx,
+                reply.line()
+            )));
+        }
+        Ok(Ehlo::parse(&reply))
+    }
+
+    async fn transaction<S: AsyncRead + AsyncWrite + Unpin>(
+        &self,
+        mut conn: BufReader<S>,
+        ehlo: &Ehlo,
+        tls: Option<TlsInfo>,
+    ) -> Result<Delivered, HostFailure> {
+        let all = |outcome: fn(String) -> RcptOutcome, reply: String| Delivered {
+            outcomes: self
+                .rcpt_to
+                .iter()
+                .map(|_| outcome(reply.clone()))
+                .collect(),
+            tls: tls.clone(),
+        };
+        let needs_utf8 =
+            !self.mail_from.email.is_ascii() || self.rcpt_to.iter().any(|r| !r.email.is_ascii());
+        if needs_utf8 && !ehlo.smtputf8 {
+            let _ = quit(&mut conn).await;
+            return Ok(all(
+                RcptOutcome::Refused,
+                format!(
+                    "553 5.6.7 {} doesn't accept internationalized addresses",
+                    self.mx
+                ),
+            ));
+        }
+        if ehlo.size.is_some_and(|max| max > 0 && self.size > max) {
+            let _ = quit(&mut conn).await;
+            return Ok(all(
+                RcptOutcome::Refused,
+                format!("552 5.3.4 The message is larger than {} accepts", self.mx),
+            ));
+        }
+
+        command(
+            &mut conn,
+            &mail_from(self.mail_from, ehlo, self.size, needs_utf8),
+        )
+        .await?;
+        let reply = read_reply(&mut conn, COMMAND_TIMEOUT).await?;
+        match reply.class() {
+            2 => {}
+            5 => {
+                let _ = quit(&mut conn).await;
+                return Ok(all(RcptOutcome::Refused, reply.line()));
+            }
+            _ => {
+                let _ = quit(&mut conn).await;
+                return Ok(all(RcptOutcome::Deferred, reply.line()));
+            }
+        }
+
+        let mut outcomes = Vec::with_capacity(self.rcpt_to.len());
+        let mut accepted = Vec::new();
+        for (i, rcpt) in self.rcpt_to.iter().enumerate() {
+            command(&mut conn, &rcpt_to(rcpt, ehlo)).await?;
+            let reply = read_reply(&mut conn, COMMAND_TIMEOUT).await?;
+            outcomes.push(match reply.class() {
+                2 => {
+                    accepted.push(i);
+                    RcptOutcome::Delivered(reply.line())
+                }
+                5 => RcptOutcome::Refused(reply.line()),
+                _ => RcptOutcome::Deferred(reply.line()),
+            });
+        }
+        if accepted.is_empty() {
+            let _ = quit(&mut conn).await;
+            return Ok(Delivered { outcomes, tls });
+        }
+
+        command(&mut conn, "DATA\r\n").await?;
+        let reply = read_reply(&mut conn, COMMAND_TIMEOUT).await?;
+        let final_reply = if reply.code == 354 {
+            send_message(&mut conn, self.message)
+                .await
+                .map_err(|e| host_failure(format!("Sending to {} failed: {e}", self.mx)))?;
+            read_reply(&mut conn, DATA_TIMEOUT).await?
+        } else {
+            reply
+        };
+        for i in accepted {
+            outcomes[i] = match final_reply.class() {
+                2 => RcptOutcome::Delivered(final_reply.line()),
+                5 => RcptOutcome::Refused(final_reply.line()),
+                _ => RcptOutcome::Deferred(final_reply.line()),
+            };
+        }
+        let _ = quit(&mut conn).await;
+        Ok(Delivered { outcomes, tls })
+    }
+}
+
+/// What the server's EHLO reply offers.
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct Ehlo {
+    pub starttls: bool,
+    pub size: Option<u64>,
+    pub eight_bit: bool,
+    pub smtputf8: bool,
+    pub dsn: bool,
+}
+
+impl Ehlo {
+    fn parse(reply: &Reply) -> Ehlo {
+        let mut ehlo = Ehlo::default();
+        // The first line is the greeting; the rest are keywords.
+        for line in reply.text.split('\n').skip(1) {
+            let mut words = line.split_ascii_whitespace();
+            let Some(keyword) = words.next() else {
+                continue;
+            };
+            match keyword.to_ascii_uppercase().as_str() {
+                "STARTTLS" => ehlo.starttls = true,
+                "SIZE" => ehlo.size = Some(words.next().and_then(|n| n.parse().ok()).unwrap_or(0)),
+                "8BITMIME" => ehlo.eight_bit = true,
+                "SMTPUTF8" => ehlo.smtputf8 = true,
+                "DSN" => ehlo.dsn = true,
+                _ => {}
+            }
+        }
+        ehlo
+    }
+}
+
+/// The value of an SMTP parameter, if the downstream gave it.
+fn parameter<'a>(address: &'a Address, name: &str) -> Option<Option<&'a str>> {
+    address
+        .parameters
+        .as_ref()?
+        .iter()
+        .find(|(k, _)| k.eq_ignore_ascii_case(name))
+        .map(|(_, v)| v.as_deref())
+}
+
+/// MAIL FROM, with only the parameters this server takes (RFC 1870, 6152,
+/// 6531, 3461). The size is the relay's own count.
+fn mail_from(from: &Address, ehlo: &Ehlo, size: u64, utf8: bool) -> String {
+    let mut line = format!("MAIL FROM:<{}>", from.email);
+    if ehlo.size.is_some() {
+        line.push_str(&format!(" SIZE={size}"));
+    }
+    if ehlo.eight_bit
+        && let Some(Some(body)) = parameter(from, "BODY")
+        && body.eq_ignore_ascii_case("8BITMIME")
+    {
+        line.push_str(" BODY=8BITMIME");
+    }
+    if utf8 {
+        line.push_str(" SMTPUTF8");
+    }
+    if ehlo.dsn {
+        for name in ["RET", "ENVID"] {
+            if let Some(Some(value)) = parameter(from, name) {
+                line.push_str(&format!(" {name}={value}"));
+            }
+        }
+    }
+    line.push_str("\r\n");
+    line
+}
+
+fn rcpt_to(rcpt: &Address, ehlo: &Ehlo) -> String {
+    let mut line = format!("RCPT TO:<{}>", rcpt.email);
+    if ehlo.dsn {
+        for name in ["NOTIFY", "ORCPT"] {
+            if let Some(Some(value)) = parameter(rcpt, name) {
+                line.push_str(&format!(" {name}={value}"));
+            }
+        }
+    }
+    line.push_str("\r\n");
+    line
+}
+
+async fn command<S: AsyncRead + AsyncWrite + Unpin>(
+    conn: &mut BufReader<S>,
+    line: &str,
+) -> Result<(), HostFailure> {
+    let stream = conn.get_mut();
+    tokio::time::timeout(COMMAND_TIMEOUT, async {
+        stream.write_all(line.as_bytes()).await?;
+        stream.flush().await
+    })
+    .await
+    .map_err(|_| host_failure("The server stopped reading"))?
+    .map_err(|e| host_failure(format!("The connection dropped: {e}")))
+}
+
+async fn quit<S: AsyncRead + AsyncWrite + Unpin>(
+    conn: &mut BufReader<S>,
+) -> Result<(), HostFailure> {
+    command(conn, "QUIT\r\n").await?;
+    tokio::time::timeout(Duration::from_secs(10), read_reply(conn, COMMAND_TIMEOUT))
+        .await
+        .map_err(|_| host_failure("no reply to QUIT"))?
+        .map(|_| ())
+}
+
+/// One reply, multi-line or not; the lines after the first are joined
+/// with `\n`, so EHLO can read them.
+pub async fn read_reply<S: AsyncRead + Unpin>(
+    conn: &mut BufReader<S>,
+    timeout: Duration,
+) -> Result<Reply, HostFailure> {
+    tokio::time::timeout(timeout, read_reply_inner(conn))
+        .await
+        .map_err(|_| host_failure("The server didn't answer in time"))?
+}
+
+async fn read_reply_inner<S: AsyncRead + Unpin>(
+    conn: &mut BufReader<S>,
+) -> Result<Reply, HostFailure> {
+    let mut code = None;
+    let mut lines: Vec<String> = Vec::new();
+    loop {
+        let mut raw = Vec::new();
+        (&mut *conn)
+            .take(LINE_MAX)
+            .read_until(b'\n', &mut raw)
+            .await
+            .map_err(|e| host_failure(format!("The connection dropped: {e}")))?;
+        if raw.is_empty() {
+            return Err(host_failure("The server closed the connection"));
+        }
+        if raw.last() != Some(&b'\n') {
+            return Err(host_failure("The server sent a line that is too long"));
+        }
+        let line = String::from_utf8_lossy(&raw);
+        let line = line.trim_end_matches(['\r', '\n']);
+        let this_code = line
+            .get(..3)
+            .and_then(|c| c.parse::<u16>().ok())
+            .filter(|c| (200..600).contains(c))
+            .ok_or_else(|| host_failure("The server sent something that isn't a reply"))?;
+        if code.is_some_and(|c| c != this_code) {
+            return Err(host_failure("The server mixed reply codes in one reply"));
+        }
+        code = Some(this_code);
+        let rest = line.get(4..).unwrap_or("");
+        lines.push(rest.replace(|c: char| c.is_control(), " "));
+        let last = line.as_bytes().get(3) != Some(&b'-');
+        if last {
+            return Ok(Reply {
+                code: this_code,
+                text: lines.join("\n"),
+            });
+        }
+        if lines.len() >= REPLY_LINES_MAX {
+            return Err(host_failure("The server's reply is too long"));
+        }
+    }
+}
+
+/// The message as stored, dot-stuffed (RFC 5321 §4.5.2), then the final
+/// dot. Its bytes are otherwise sent as they are: the downstream signed
+/// them.
+async fn send_message<S: AsyncRead + AsyncWrite + Unpin>(
+    conn: &mut BufReader<S>,
+    path: &Path,
+) -> std::io::Result<()> {
+    let mut file = tokio::fs::File::open(path).await?;
+    let stream = conn.get_mut();
+    let mut buf = vec![0u8; 64 * 1024];
+    let mut out = Vec::with_capacity(buf.len() + 1024);
+    let mut stuffer = DotStuffer::default();
+    loop {
+        let n = file.read(&mut buf).await?;
+        if n == 0 {
+            break;
+        }
+        out.clear();
+        stuffer.push(&buf[..n], &mut out);
+        tokio::time::timeout(COMMAND_TIMEOUT, stream.write_all(&out))
+            .await
+            .map_err(|_| std::io::Error::other("the server stopped reading"))??;
+    }
+    out.clear();
+    stuffer.finish(&mut out);
+    stream.write_all(&out).await?;
+    stream.flush().await
+}
+
+/// Doubles a dot that starts a line, across chunk boundaries.
+struct DotStuffer {
+    at_line_start: bool,
+    last: Option<u8>,
+}
+
+impl Default for DotStuffer {
+    fn default() -> Self {
+        DotStuffer {
+            at_line_start: true,
+            last: None,
+        }
+    }
+}
+
+impl DotStuffer {
+    fn push(&mut self, chunk: &[u8], out: &mut Vec<u8>) {
+        for &b in chunk {
+            if self.at_line_start && b == b'.' {
+                out.push(b'.');
+            }
+            out.push(b);
+            self.at_line_start = b == b'\n';
+            self.last = Some(b);
+        }
+    }
+
+    fn finish(&self, out: &mut Vec<u8>) {
+        match self.last {
+            Some(b'\n') | None => {}
+            Some(b'\r') => out.push(b'\n'),
+            Some(_) => out.extend_from_slice(b"\r\n"),
+        }
+        out.extend_from_slice(b".\r\n");
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn stuff(chunks: &[&[u8]]) -> Vec<u8> {
+        let mut s = DotStuffer::default();
+        let mut out = Vec::new();
+        for c in chunks {
+            s.push(c, &mut out);
+        }
+        s.finish(&mut out);
+        out
+    }
+
+    #[test]
+    fn dots_are_stuffed_across_chunks() {
+        assert_eq!(
+            stuff(&[b".a\r\nb\r", b"\n.c\r\n"]),
+            b"..a\r\nb\r\n..c\r\n.\r\n"
+        );
+        assert_eq!(stuff(&[b"a\r\n", b".\r\n"]), b"a\r\n..\r\n.\r\n");
+        assert_eq!(stuff(&[b"no end"]), b"no end\r\n.\r\n");
+        assert_eq!(stuff(&[b"x\r"]), b"x\r\n.\r\n");
+    }
+
+    #[tokio::test]
+    async fn replies_parse() {
+        let mut r = BufReader::new(
+            &b"250-mx.example.org hi\r\n250-SIZE 1000\r\n250-starttls\r\n250 DSN\r\n"[..],
+        );
+        let reply = read_reply_inner(&mut r).await.unwrap();
+        assert_eq!(reply.code, 250);
+        let ehlo = Ehlo::parse(&reply);
+        assert!(ehlo.starttls && ehlo.dsn && !ehlo.smtputf8);
+        assert_eq!(ehlo.size, Some(1000));
+
+        let mut r = BufReader::new(&b"550 5.1.1 no\x07 such user\r\n"[..]);
+        assert_eq!(
+            read_reply_inner(&mut r).await.unwrap().line(),
+            "550 5.1.1 no  such user"
+        );
+        for bad in [
+            &b"250-a\r\n251 b\r\n"[..],
+            b"hello\r\n",
+            b"250-a\r\n",
+            b"099 x\r\n",
+        ] {
+            let mut r = BufReader::new(bad);
+            assert!(read_reply_inner(&mut r).await.is_err(), "{bad:?}");
+        }
+    }
+
+    #[test]
+    fn only_offered_parameters_are_sent() {
+        let mut from = Address::new("[email protected]");
+        from.parameters = Some(
+            [
+                ("BODY".to_string(), Some("8BITMIME".to_string())),
+                ("RET".to_string(), Some("HDRS".to_string())),
+                ("AUTH".to_string(), Some("x".to_string())),
+            ]
+            .into(),
+        );
+        let bare = Ehlo::default();
+        assert_eq!(
+            mail_from(&from, &bare, 10, false),
+            "MAIL FROM:<[email protected]>\r\n"
+        );
+        let full = Ehlo {
+            size: Some(0),
+            eight_bit: true,
+            dsn: true,
+            ..Default::default()
+        };
+        assert_eq!(
+            mail_from(&from, &full, 10, true),
+            "MAIL FROM:<[email protected]> SIZE=10 BODY=8BITMIME SMTPUTF8 RET=HDRS\r\n"
+        );
+        let mut rcpt = Address::new("[email protected]");
+        rcpt.parameters = Some([("NOTIFY".to_string(), Some("NEVER".to_string()))].into());
+        assert_eq!(rcpt_to(&rcpt, &bare), "RCPT TO:<[email protected]>\r\n");
+        assert_eq!(
+            rcpt_to(&rcpt, &full),
+            "RCPT TO:<[email protected]> NOTIFY=NEVER\r\n"
+        );
+    }
+}

crates/relay/src/deliver/https.rs

@@ -0,0 +1,55 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! The one HTTPS request the relay makes: an MTA-STS policy (RFC 8461
+//! §3.3). WebPKI checked, no redirects, small.
+
+use http_body_util::{BodyExt, Empty, Limited};
+use hyper::{Request, StatusCode, body::Bytes};
+use hyper_util::rt::TokioIo;
+use rustls_pki_types::ServerName;
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio_rustls::TlsConnector;
+
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+
+pub async fn get(host: &str, path: &str, max: usize) -> Result<Vec<u8>, String> {
+    let name = ServerName::try_from(host.to_string()).map_err(|e| e.to_string())?;
+    let tcp = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect((host, 443)))
+        .await
+        .map_err(|_| "connection timed out".to_string())?
+        .map_err(|e| e.to_string())?;
+    let tls = TlsConnector::from(super::tls::verified())
+        .connect(name, tcp)
+        .await
+        .map_err(|e| e.to_string())?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(tls))
+        .await
+        .map_err(|e| e.to_string())?;
+    tokio::spawn(conn);
+    let request = Request::get(path)
+        .header("Host", host)
+        .header(
+            "User-Agent",
+            concat!("inbuxa-relay/", env!("CARGO_PKG_VERSION")),
+        )
+        .body(Empty::<Bytes>::new())
+        .map_err(|e| e.to_string())?;
+    let response = sender
+        .send_request(request)
+        .await
+        .map_err(|e| e.to_string())?;
+    // RFC 8461 §3.3: a redirect is not followed.
+    if response.status() != StatusCode::OK {
+        return Err(format!("the server answered {}", response.status()));
+    }
+    Limited::new(response.into_body(), max)
+        .collect()
+        .await
+        .map(|body| body.to_bytes().to_vec())
+        .map_err(|_| "the policy is too large".to_string())
+}

crates/relay/src/deliver/mod.rs

@@ -0,0 +1,383 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Outbound delivery (spec "Outbound"): each message handed over is sent
+//! by MX, one domain at a time, with DANE where the destination's TLSA
+//! records are signed, MTA-STS where it enforces a policy, and otherwise
+//! TLS per `outboundTls`. After each attempt the message's
+//! `deliveryStatus` is updated and pushed; the downstream writes any DSN.
+//! TLS results go up as `outbound.tls` events, for TLS-RPT.
+
+pub mod client;
+pub mod https;
+pub mod route;
+#[cfg(test)]
+mod tests;
+pub mod tls;
+
+use crate::{app::App, outbound::Meta};
+use client::{RcptOutcome, Session};
+use inbuxa_outbound_tls::{Mode, Policy, Tlsa, TlsaMatching, dnssec::TlsaResult};
+use inbuxa_relay_proto::{
+    time,
+    types::{Address, Delivered, DeliveryStatus, Displayed, OutboundTls, RelaySettings},
+};
+use route::{LookupError, Mx, Resolve};
+use serde_json::{Value, json};
+use std::{collections::BTreeMap, net::SocketAddr, path::Path, sync::Arc, time::Duration};
+use tls::{TlsFailure, TlsMode};
+use tokio::sync::Semaphore;
+
+/// Messages delivered at once.
+const CONCURRENCY: usize = 8;
+/// MX hosts tried per domain, and addresses per host.
+const MX_MAX: usize = 5;
+const ADDRESSES_PER_HOST: usize = 2;
+/// A finished message the downstream never cleared goes after a week.
+const FINISHED_KEEP: u64 = 7 * 86_400;
+const TICK: Duration = Duration::from_secs(5);
+
+pub async fn run<R: Resolve>(app: Arc<App>, resolve: Arc<R>) {
+    let running = Arc::new(Semaphore::new(CONCURRENCY));
+    let mut tick = tokio::time::interval(TICK);
+    loop {
+        tokio::select! {
+            _ = tick.tick() => {}
+            _ = app.outbound_wake.notified() => {}
+        }
+        let now = time::now();
+        housekeeping(&app, now);
+        let Some(settings) = app.settings().filter(|s| s.problems().is_empty()) else {
+            continue;
+        };
+        let free = running.available_permits();
+        if free == 0 {
+            continue;
+        }
+        let due = app.outbox.lock().expect("outbox").take_due(now, free);
+        for meta in due {
+            let Ok(permit) = running.clone().try_acquire_owned() else {
+                app.outbox.lock().expect("outbox").release(&meta.id);
+                continue;
+            };
+            let (app, resolve, settings) = (app.clone(), resolve.clone(), settings.clone());
+            tokio::spawn(async move {
+                attempt(&app, &*resolve, &settings, meta).await;
+                drop(permit);
+            });
+        }
+    }
+}
+
+/// Gives up on what has expired, and forgets what was never cleared.
+fn housekeeping(app: &App, now: u64) {
+    let (expired, swept) = {
+        let mut outbox = app.outbox.lock().expect("outbox");
+        let expired = outbox.expire(now);
+        (expired, outbox.sweep_finished(now, FINISHED_KEEP))
+    };
+    let expired = match expired {
+        Ok(expired) => expired,
+        Err(e) => {
+            app.set_error(format!("could not write to the outbound spool: {e}"));
+            return;
+        }
+    };
+    for (id, recipients) in &expired {
+        app.record(
+            "outbound.expired",
+            crate::events::data([("id", json!(id)), ("recipients", json!(recipients))]),
+        );
+    }
+    if !expired.is_empty() || swept > 0 {
+        app.outbound_changed();
+        app.relay_changed();
+    }
+}
+
+async fn attempt<R: Resolve>(app: &App, resolve: &R, settings: &RelaySettings, meta: Meta) {
+    let path = app.outbox.lock().expect("outbox").eml_path(&meta.id);
+    let mut by_domain: BTreeMap<String, Vec<&Address>> = BTreeMap::new();
+    for rcpt in meta.queued() {
+        let domain = rcpt
+            .email
+            .rsplit_once('@')
+            .map(|(_, d)| d.trim_end_matches('.').to_lowercase())
+            .unwrap_or_default();
+        by_domain.entry(domain).or_default().push(rcpt);
+    }
+    let mut statuses = Vec::new();
+    for (domain, rcpts) in &by_domain {
+        let target = Target {
+            app,
+            resolve,
+            settings,
+            meta: &meta,
+            path: &path,
+            domain,
+            rcpts,
+        };
+        statuses.extend(target.deliver().await);
+    }
+    let recorded = app.outbox.lock().expect("outbox").record(
+        &meta.id,
+        statuses,
+        &settings.outbound_retry,
+        time::now(),
+    );
+    match recorded {
+        Ok(true) => app.outbound_changed(),
+        Ok(false) => {}
+        Err(e) => app.set_error(format!("could not write to the outbound spool: {e}")),
+    }
+}
+
+/// One message's recipients at one domain.
+struct Target<'a, R> {
+    app: &'a App,
+    resolve: &'a R,
+    settings: &'a RelaySettings,
+    meta: &'a Meta,
+    path: &'a Path,
+    domain: &'a str,
+    rcpts: &'a [&'a Address],
+}
+
+fn status(delivered: Delivered, smtp_reply: String) -> DeliveryStatus {
+    DeliveryStatus {
+        smtp_reply,
+        delivered,
+        displayed: Displayed::Unknown,
+    }
+}
+
+impl<R: Resolve> Target<'_, R> {
+    fn every(&self, delivered: Delivered, reply: String) -> Vec<(String, DeliveryStatus)> {
+        self.rcpts
+            .iter()
+            .map(|r| (r.email.clone(), status(delivered, reply.clone())))
+            .collect()
+    }
+
+    async fn deliver(&self) -> Vec<(String, DeliveryStatus)> {
+        let domain = self.domain;
+        let (hosts, mx_secure) = match self.resolve.mx(domain).await {
+            Ok(Mx::Hosts { hosts, secure }) => (hosts, secure),
+            Ok(Mx::Null) => {
+                return self.every(
+                    Delivered::No,
+                    format!("556 5.1.10 {domain} accepts no mail (null MX)"),
+                );
+            }
+            Err(LookupError::Permanent(e)) => {
+                return self.every(Delivered::No, format!("550 5.1.2 {domain}: {e}"));
+            }
+            Err(LookupError::Temporary(e)) => {
+                return self.every(
+                    Delivered::Queued,
+                    format!("451 4.4.3 Looking up {domain} failed: {e}"),
+                );
+            }
+        };
+        let sts = self
+            .resolve
+            .mta_sts(domain)
+            .await
+            .filter(|p| p.mode != Mode::None);
+        let enforce = sts.as_ref().is_some_and(|p| p.enforce());
+        let mut last = format!("451 4.4.0 No MX host of {domain} could be reached");
+
+        for host in hosts.iter().take(MX_MAX) {
+            if let Some(policy) = &sts
+                && !policy.verify(host)
+            {
+                self.tls_event(
+                    host,
+                    None,
+                    &TlsMode::Verified,
+                    Some(policy),
+                    Some(&TlsFailure {
+                        reason: "the MX host isn't in the MTA-STS policy".into(),
+                        result_type: "validation-failure",
+                        handshake: false,
+                    }),
+                );
+                if enforce {
+                    last = format!("451 4.7.5 {host} isn't in {domain}'s MTA-STS policy");
+                    continue;
+                }
+            }
+            let (ips, ips_secure) = match self.resolve.addresses(host, mx_secure).await {
+                Ok(found) => found,
+                Err(LookupError::Permanent(e) | LookupError::Temporary(e)) => {
+                    last = format!("451 4.4.3 Looking up {host} failed: {e}");
+                    continue;
+                }
+            };
+            let fallback = if enforce {
+                TlsMode::Verified
+            } else if self.settings.outbound_tls == OutboundTls::Required {
+                TlsMode::Required
+            } else {
+                TlsMode::Opportunistic
+            };
+            // DANE needs the MX answer and the host's addresses signed
+            // (RFC 7672 §2.2), and then takes precedence over MTA-STS.
+            let mode = if mx_secure && ips_secure {
+                match self.resolve.tlsa(host).await {
+                    Ok(TlsaResult::Secure(tlsa))
+                        if tlsa.has_end_entities || tlsa.has_intermediates =>
+                    {
+                        TlsMode::Dane(tlsa)
+                    }
+                    Ok(TlsaResult::Secure(_)) => TlsMode::DaneEncrypt,
+                    Ok(TlsaResult::Bogus) => {
+                        self.tls_event(
+                            host,
+                            None,
+                            &TlsMode::DaneEncrypt,
+                            None,
+                            Some(&TlsFailure {
+                                reason: "the TLSA records failed DNSSEC validation".into(),
+                                result_type: "dnssec-invalid",
+                                handshake: false,
+                            }),
+                        );
+                        last = format!("451 4.7.5 {host}'s TLSA records failed DNSSEC validation");
+                        continue;
+                    }
+                    Ok(TlsaResult::Missing) => fallback,
+                    Err(LookupError::Permanent(e) | LookupError::Temporary(e)) => {
+                        last = format!("451 4.4.3 Looking up TLSA for {host} failed: {e}");
+                        continue;
+                    }
+                }
+            } else {
+                fallback
+            };
+
+            for ip in ips.iter().take(ADDRESSES_PER_HOST) {
+                let session = Session {
+                    addr: SocketAddr::new(*ip, self.resolve.port()),
+                    mx: host,
+                    domain,
+                    helo: &self.settings.hostname,
+                    mode: mode.clone(),
+                    mail_from: &self.meta.envelope.mail_from,
+                    rcpt_to: self.rcpts,
+                    message: self.path,
+                    size: self.meta.size,
+                };
+                match session.run().await {
+                    Ok(delivered) => {
+                        if delivered.tls.is_some() || mode.is_required() {
+                            self.tls_event(host, Some(*ip), &mode, sts.as_deref(), None);
+                        }
+                        let mut counts = [0usize; 3];
+                        let statuses = self
+                            .rcpts
+                            .iter()
+                            .zip(delivered.outcomes)
+                            .map(|(rcpt, outcome)| {
+                                let (delivered, reply, n) = match outcome {
+                                    RcptOutcome::Delivered(r) => (Delivered::Yes, r, 0),
+                                    RcptOutcome::Refused(r) => (Delivered::No, r, 1),
+                                    RcptOutcome::Deferred(r) => (Delivered::Queued, r, 2),
+                                };
+                                counts[n] += 1;
+                                (rcpt.email.clone(), status(delivered, reply))
+                            })
+                            .collect();
+                        self.app.record(
+                            "outbound.attempt",
+                            crate::events::data([
+                                ("id", json!(self.meta.id)),
+                                ("domain", json!(domain)),
+                                ("mx", json!(host)),
+                                ("ip", json!(ip.to_string())),
+                                (
+                                    "tls",
+                                    delivered.tls.as_ref().map_or(
+                                        Value::Null,
+                                        |t| json!({ "version": t.version, "cipher": t.cipher }),
+                                    ),
+                                ),
+                                ("delivered", json!(counts[0])),
+                                ("refused", json!(counts[1])),
+                                ("deferred", json!(counts[2])),
+                            ]),
+                        );
+                        return statuses;
+                    }
+                    Err(failure) => {
+                        if let Some(tls) = &failure.tls {
+                            self.tls_event(host, Some(*ip), &mode, sts.as_deref(), Some(tls));
+                        }
+                        last = format!("451 4.4.1 {}", failure.reason);
+                    }
+                }
+            }
+        }
+        self.every(Delivered::Queued, last)
+    }
+
+    /// One session's TLS result, for the downstream's TLS-RPT reports
+    /// (RFC 8460): the policy that applied and what came of it.
+    fn tls_event(
+        &self,
+        mx: &str,
+        ip: Option<std::net::IpAddr>,
+        mode: &TlsMode,
+        sts: Option<&Policy>,
+        failure: Option<&TlsFailure>,
+    ) {
+        let policy_strings: Vec<String> = match mode {
+            TlsMode::Dane(tlsa) => tlsa_strings(tlsa),
+            TlsMode::Verified => sts
+                .map(|p| p.to_string().lines().map(str::to_string).collect())
+                .unwrap_or_default(),
+            _ => Vec::new(),
+        };
+        let mut data = crate::events::data([
+            ("domain", json!(self.domain)),
+            ("mx", json!(mx)),
+            ("policyType", json!(mode.policy_type())),
+            ("policyStrings", json!(policy_strings)),
+            (
+                "result",
+                json!(failure.map_or("success", |f| f.result_type)),
+            ),
+        ]);
+        if let Some(ip) = ip {
+            data.insert("ip".into(), json!(ip.to_string()));
+        }
+        if let Some(failure) = failure {
+            data.insert("reason".into(), json!(failure.reason));
+        }
+        self.app.record("outbound.tls", data);
+    }
+}
+
+/// TLSA records in presentation form (RFC 6698 §2.2), as TLS-RPT reports
+/// them.
+fn tlsa_strings(tlsa: &Tlsa) -> Vec<String> {
+    tlsa.entries
+        .iter()
+        .map(|e| {
+            let hex: String = e.data.iter().map(|b| format!("{b:02x}")).collect();
+            format!(
+                "{} {} {} {hex}",
+                if e.is_end_entity { 3 } else { 2 },
+                u8::from(e.is_spki),
+                match e.matching {
+                    TlsaMatching::Full => 0,
+                    TlsaMatching::Sha256 => 1,
+                    TlsaMatching::Sha512 => 2,
+                }
+            )
+        })
+        .collect()
+}

crates/relay/src/deliver/route.rs

@@ -0,0 +1,303 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Where a domain's mail goes, and what TLS it asks for: MX, addresses,
+//! TLSA and MTA-STS, through the server's own lookups (RL-30).
+
+use inbuxa_outbound_tls::{
+    Policy,
+    dnssec::{self, DnssecStatus, Resolvers, TlsaResult},
+    mail_auth::{
+        self,
+        common::parse::TxtRecordParser,
+        hickory_resolver::{
+            TokioResolver,
+            config::{CLOUDFLARE, ResolverConfig, ResolverOpts},
+            net::runtime::TokioRuntimeProvider,
+            proto::op::ResponseCode,
+        },
+        mta_sts::MtaSts,
+    },
+};
+use std::{
+    collections::HashMap,
+    future::Future,
+    net::IpAddr,
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+
+/// A domain's MX hosts, best first, and whether the answer was signed.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Mx {
+    Hosts {
+        hosts: Vec<String>,
+        secure: bool,
+    },
+    /// RFC 7505: the domain takes no mail.
+    Null,
+}
+
+/// A lookup failure: for good (the domain doesn't exist) or for now.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum LookupError {
+    Permanent(String),
+    Temporary(String),
+}
+
+pub trait Resolve: Send + Sync + 'static {
+    fn mx(&self, domain: &str) -> impl Future<Output = Result<Mx, LookupError>> + Send;
+
+    /// The host's addresses, and whether they were signed. `secure` asks
+    /// for the DNSSEC status; without it, it may be reported as false.
+    fn addresses(
+        &self,
+        host: &str,
+        secure: bool,
+    ) -> impl Future<Output = Result<(Vec<IpAddr>, bool), LookupError>> + Send;
+
+    fn tlsa(&self, host: &str) -> impl Future<Output = Result<TlsaResult, LookupError>> + Send;
+
+    /// The domain's MTA-STS policy, if it publishes one.
+    fn mta_sts(&self, domain: &str) -> impl Future<Output = Option<Arc<Policy>>> + Send;
+
+    /// Port 25, except in tests.
+    fn port(&self) -> u16 {
+        25
+    }
+}
+
+const MTA_STS_TIMEOUT: Duration = Duration::from_secs(30);
+const MTA_STS_MAX: usize = 64 * 1024;
+
+/// The real thing: the system's resolvers, one validating.
+pub struct Dns {
+    dnssec: TokioResolver,
+    plain: TokioResolver,
+    /// The validating resolver works; without it there is no DANE.
+    validates: bool,
+    sts: Mutex<HashMap<String, (Arc<Policy>, u64)>>,
+}
+
+impl Dns {
+    pub async fn new() -> Dns {
+        let (config, opts) = mail_auth::hickory_resolver::system_conf::read_system_conf()
+            .unwrap_or_else(|e| {
+                eprintln!(
+                    "inbuxa-relay: no usable resolv.conf ({e}); using Cloudflare's resolvers"
+                );
+                (
+                    ResolverConfig::udp_and_tcp(&CLOUDFLARE),
+                    ResolverOpts::default(),
+                )
+            });
+        let mut validating = opts.clone();
+        validating.validate = true;
+        let build = |config: ResolverConfig, opts: ResolverOpts| {
+            TokioResolver::builder_with_config(config, TokioRuntimeProvider::default())
+                .with_options(opts)
+                .build()
+                .expect("DNS resolver")
+        };
+        let dnssec = build(config.clone(), validating);
+        let plain = build(config, opts);
+        let validates = dnssec::validates(&dnssec).await;
+        if !validates {
+            eprintln!(
+                "inbuxa-relay: the resolver doesn't validate DNSSEC, so DANE can't be checked"
+            );
+        }
+        Dns {
+            dnssec,
+            plain,
+            validates,
+            sts: Mutex::new(HashMap::new()),
+        }
+    }
+
+    fn resolvers(&self) -> Resolvers<'_> {
+        Resolvers {
+            dnssec: &self.dnssec,
+            plain: &self.plain,
+        }
+    }
+
+    async fn fetch_policy(&self, domain: &str, id: &str) -> Result<Policy, String> {
+        let url_host = format!("mta-sts.{domain}");
+        let body = super::https::get(&url_host, "/.well-known/mta-sts.txt", MTA_STS_MAX)
+            .await
+            .map_err(|e| format!("fetching the MTA-STS policy failed: {e}"))?;
+        let text = std::str::from_utf8(&body).map_err(|e| e.to_string())?;
+        Policy::parse(text, id.to_string())
+    }
+}
+
+fn fqdn(name: &str) -> String {
+    let name = name.trim_end_matches('.');
+    format!("{name}.")
+}
+
+fn lookup_error(err: mail_auth::Error) -> LookupError {
+    match err {
+        mail_auth::Error::Dns(mail_auth::DnsError::RecordNotFound(ResponseCode::NXDomain)) => {
+            LookupError::Permanent("the domain doesn't exist".into())
+        }
+        err => LookupError::Temporary(err.to_string()),
+    }
+}
+
+impl Resolve for Dns {
+    async fn mx(&self, domain: &str) -> Result<Mx, LookupError> {
+        let name = fqdn(domain);
+        let records = if self.validates {
+            dnssec::mx(&self.resolvers(), &name).await.map(|(r, _)| r)
+        } else {
+            // Without validation every answer is insecure anyway.
+            dnssec::mx(
+                &Resolvers {
+                    dnssec: &self.plain,
+                    plain: &self.plain,
+                },
+                &name,
+            )
+            .await
+            .map(|(r, _)| r)
+        }
+        .map_err(lookup_error)?;
+        let secure = self.validates && records.dnssec_status == DnssecStatus::Secure;
+        if records.rrset.is_empty() {
+            // RFC 5321 §5.1: no MX means the domain itself.
+            return Ok(Mx::Hosts {
+                hosts: vec![domain.trim_end_matches('.').to_lowercase()],
+                secure,
+            });
+        }
+        let mut hosts = Vec::new();
+        for mx in records.rrset.iter() {
+            for host in mx.exchanges.iter() {
+                if mx.preference == 0 && host.as_ref() == "." {
+                    return Ok(Mx::Null);
+                }
+                let host = host.trim_end_matches('.').to_string();
+                if !host.is_empty() && !hosts.contains(&host) {
+                    hosts.push(host);
+                }
+            }
+        }
+        Ok(Mx::Hosts { hosts, secure })
+    }
+
+    async fn addresses(
+        &self,
+        host: &str,
+        secure: bool,
+    ) -> Result<(Vec<IpAddr>, bool), LookupError> {
+        let name = fqdn(host);
+        let resolvers = if secure && self.validates {
+            self.resolvers()
+        } else {
+            Resolvers {
+                dnssec: &self.plain,
+                plain: &self.plain,
+            }
+        };
+        let v4 = dnssec::ipv4(&resolvers, &name).await;
+        let v6 = dnssec::ipv6(&resolvers, &name).await;
+        let mut ips = Vec::new();
+        let mut all_secure = true;
+        let mut last_err = None;
+        match v4 {
+            Ok((set, _)) => {
+                all_secure &= set.rrset.is_empty() || set.dnssec_status == DnssecStatus::Secure;
+                ips.extend(set.rrset.iter().copied().map(IpAddr::V4));
+            }
+            Err(e) => last_err = Some(e),
+        }
+        match v6 {
+            Ok((set, _)) => {
+                all_secure &= set.rrset.is_empty() || set.dnssec_status == DnssecStatus::Secure;
+                ips.extend(set.rrset.iter().copied().map(IpAddr::V6));
+            }
+            Err(e) => last_err = Some(e),
+        }
+        if ips.is_empty() {
+            return Err(match last_err {
+                Some(e) => lookup_error(e),
+                None => LookupError::Temporary(format!("{host} has no addresses")),
+            });
+        }
+        Ok((ips, secure && self.validates && all_secure))
+    }
+
+    async fn tlsa(&self, host: &str) -> Result<TlsaResult, LookupError> {
+        if !self.validates {
+            return Ok(TlsaResult::Missing);
+        }
+        match dnssec::tlsa(&self.resolvers(), &format!("_25._tcp.{}", fqdn(host))).await {
+            Ok((result, _)) => Ok(result),
+            Err(mail_auth::Error::Dns(mail_auth::DnsError::RecordNotFound(_))) => {
+                Ok(TlsaResult::Missing)
+            }
+            Err(e) => Err(LookupError::Temporary(e.to_string())),
+        }
+    }
+
+    async fn mta_sts(&self, domain: &str) -> Option<Arc<Policy>> {
+        let now = inbuxa_relay_proto::time::now();
+        let cached = self
+            .sts
+            .lock()
+            .expect("sts")
+            .get(domain)
+            .filter(|(_, until)| *until > now)
+            .map(|(p, _)| p.clone());
+        let record = match self
+            .plain
+            .txt_lookup(fqdn(&format!("_mta-sts.{domain}")))
+            .await
+        {
+            Ok(lookup) => lookup.answers().iter().find_map(|r| match &r.data {
+                mail_auth::hickory_resolver::proto::rr::RData::TXT(txt) => {
+                    let bytes: Vec<u8> = txt
+                        .txt_data
+                        .iter()
+                        .flat_map(|p| p.iter().copied())
+                        .collect();
+                    MtaSts::parse(&bytes).ok()
+                }
+                _ => None,
+            }),
+            // A failed lookup keeps a cached policy in force.
+            Err(_) => return cached,
+        };
+        let Some(record) = record else {
+            return cached;
+        };
+        if let Some(policy) = &cached
+            && policy.id == record.id
+        {
+            return cached;
+        }
+        // A new id whose policy can't be fetched keeps the cached one in
+        // force until it expires (RFC 8461 §5.1).
+        match tokio::time::timeout(MTA_STS_TIMEOUT, self.fetch_policy(domain, &record.id)).await {
+            Ok(Ok(policy)) => {
+                let max_age = if (3600..31_557_600).contains(&policy.max_age) {
+                    policy.max_age
+                } else {
+                    86_400
+                };
+                let policy = Arc::new(policy);
+                self.sts
+                    .lock()
+                    .expect("sts")
+                    .insert(domain.to_string(), (policy.clone(), now + max_age));
+                Some(policy)
+            }
+            _ => cached,
+        }
+    }
+}

crates/relay/src/deliver/tests.rs

@@ -0,0 +1,513 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! The delivery engine against a scripted MX on localhost, with DNS
+//! answers fixed by the test.
+
+use super::{
+    route::{LookupError, Mx, Resolve},
+    run,
+};
+use crate::{
+    app::App,
+    state::{Identity, Paths},
+};
+use inbuxa_outbound_tls::{
+    Mode, MxPattern, Policy, Tlsa, TlsaEntry, TlsaMatching, dnssec::TlsaResult,
+};
+use inbuxa_relay_proto::{
+    time,
+    token::JoinToken,
+    types::{Address, Delivered, Envelope, Outbound, RelaySettings},
+};
+use rustls_pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
+use sha2_shim::sha256;
+use std::{
+    collections::HashMap,
+    net::{IpAddr, Ipv4Addr},
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+use tokio::{
+    io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader},
+    net::TcpListener,
+};
+use tokio_rustls::TlsAcceptor;
+
+/// SHA-256, from the crypto the relay already links.
+mod sha2_shim {
+    pub fn sha256(data: &[u8]) -> Vec<u8> {
+        aws_lc_rs::digest::digest(&aws_lc_rs::digest::SHA256, data)
+            .as_ref()
+            .to_vec()
+    }
+}
+
+#[derive(Default)]
+struct Fixed {
+    port: u16,
+    mx: HashMap<String, Mx>,
+    secure: bool,
+    tlsa: Option<Arc<Tlsa>>,
+    sts: Option<Arc<Policy>>,
+}
+
+impl Resolve for Fixed {
+    async fn mx(&self, domain: &str) -> Result<Mx, LookupError> {
+        self.mx
+            .get(domain)
+            .cloned()
+            .ok_or_else(|| LookupError::Permanent("the domain doesn't exist".into()))
+    }
+
+    async fn addresses(&self, _: &str, _: bool) -> Result<(Vec<IpAddr>, bool), LookupError> {
+        Ok((vec![IpAddr::V4(Ipv4Addr::LOCALHOST)], self.secure))
+    }
+
+    async fn tlsa(&self, _: &str) -> Result<TlsaResult, LookupError> {
+        Ok(match &self.tlsa {
+            Some(t) => TlsaResult::Secure(t.clone()),
+            None => TlsaResult::Missing,
+        })
+    }
+
+    async fn mta_sts(&self, _: &str) -> Option<Arc<Policy>> {
+        self.sts.clone()
+    }
+
+    fn port(&self) -> u16 {
+        self.port
+    }
+}
+
+/// What the scripted MX saw.
+#[derive(Default)]
+struct Seen {
+    messages: Vec<(bool, Vec<u8>)>,
+}
+
+/// An MX on localhost: STARTTLS with a self-signed certificate for
+/// `mx.test`; `bob` is taken, `nobody` refused, `later` deferred.
+async fn scripted_mx() -> (u16, Arc<Mutex<Seen>>, Vec<u8>) {
+    let cert = rcgen::generate_simple_self_signed(vec!["mx.test".into()]).unwrap();
+    let der = cert.cert.der().to_vec();
+    let spki = x509_parser::parse_x509_certificate(&der)
+        .unwrap()
+        .1
+        .tbs_certificate
+        .subject_pki
+        .raw
+        .to_vec();
+    let config = rustls::ServerConfig::builder_with_provider(crate::link::tls::provider())
+        .with_safe_default_protocol_versions()
+        .unwrap()
+        .with_no_client_auth()
+        .with_single_cert(
+            vec![CertificateDer::from(der)],
+            PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der())),
+        )
+        .unwrap();
+    let acceptor = TlsAcceptor::from(Arc::new(config));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let seen = Arc::new(Mutex::new(Seen::default()));
+    let seen2 = seen.clone();
+    tokio::spawn(async move {
+        loop {
+            let (tcp, _) = listener.accept().await.unwrap();
+            let (acceptor, seen) = (acceptor.clone(), seen2.clone());
+            tokio::spawn(async move {
+                let mut conn = BufReader::new(tcp);
+                conn.get_mut()
+                    .write_all(b"220 mx.test ready\r\n")
+                    .await
+                    .unwrap();
+                if let Some(true) = talk(&mut conn, &seen, false).await {
+                    let tls = acceptor.accept(conn.into_inner()).await;
+                    if let Ok(tls) = tls {
+                        let mut conn = BufReader::new(tls);
+                        talk(&mut conn, &seen, true).await;
+                    }
+                }
+            });
+        }
+    });
+    (port, seen, spki)
+}
+
+/// Returns `Some(true)` when the client asked for STARTTLS.
+async fn talk<S: AsyncRead + AsyncWrite + Unpin>(
+    conn: &mut BufReader<S>,
+    seen: &Mutex<Seen>,
+    tls: bool,
+) -> Option<bool> {
+    let mut line = String::new();
+    loop {
+        line.clear();
+        if conn.read_line(&mut line).await.ok()? == 0 {
+            return None;
+        }
+        let upper = line.to_ascii_uppercase();
+        let reply: &[u8] = if upper.starts_with("EHLO") {
+            if tls {
+                b"250-mx.test\r\n250-SIZE 1000000\r\n250 8BITMIME\r\n"
+            } else {
+                b"250-mx.test\r\n250-STARTTLS\r\n250 SIZE 1000000\r\n"
+            }
+        } else if upper.starts_with("STARTTLS") {
+            conn.get_mut().write_all(b"220 go ahead\r\n").await.ok()?;
+            return Some(true);
+        } else if upper.starts_with("MAIL") {
+            b"250 2.1.0 ok\r\n"
+        } else if upper.starts_with("RCPT") {
+            if upper.contains("NOBODY@") {
+                b"550 5.1.1 no such user\r\n"
+            } else if upper.contains("LATER@") {
+                b"451 4.2.0 try later\r\n"
+            } else {
+                b"250 2.1.5 ok\r\n"
+            }
+        } else if upper.starts_with("DATA") {
+            conn.get_mut().write_all(b"354 go\r\n").await.ok()?;
+            let mut body = Vec::new();
+            loop {
+                let mut l = Vec::new();
+                conn.read_until(b'\n', &mut l).await.ok()?;
+                if l == b".\r\n" || l.is_empty() {
+                    break;
+                }
+                body.extend_from_slice(&l);
+            }
+            seen.lock().unwrap().messages.push((tls, body));
+            b"250 2.0.0 queued as X1\r\n"
+        } else if upper.starts_with("QUIT") {
+            conn.get_mut().write_all(b"221 bye\r\n").await.ok()?;
+            return None;
+        } else {
+            b"500 5.5.1 what\r\n"
+        };
+        conn.get_mut().write_all(reply).await.ok()?;
+    }
+}
+
+async fn relay(name: &str) -> (Arc<App>, std::path::PathBuf) {
+    let dir = std::env::temp_dir().join(format!(
+        "inbuxa-relay-deliver-{name}-{}",
+        std::process::id()
+    ));
+    let _ = std::fs::remove_dir_all(&dir);
+    let paths = Paths::new(&dir);
+    let token = JoinToken::issue("r1", [1; 32], time::now());
+    crate::join(&paths, &token.encode()).unwrap();
+    let identity = Identity::load(&paths).unwrap().unwrap();
+    let app = Arc::new(App::open(Paths::new(&dir), identity).unwrap());
+    app.settings.send_replace(Some(Arc::new(RelaySettings {
+        hostname: "relay.test".into(),
+        domains: vec!["example.com".into()],
+        sealing_key: Some([1; 32]),
+        recipient_key: Some([2; 32]),
+        outbound_retry: vec![60],
+        ..Default::default()
+    })));
+    (app, dir)
+}
+
+fn hand_over(app: &App, to: &[&str], body: &[u8]) -> String {
+    let blob = app.paths.blobs().join(crate::spool::new_id());
+    std::fs::write(&blob, body).unwrap();
+    let settings = app.settings().unwrap();
+    app.outbox
+        .lock()
+        .unwrap()
+        .create(
+            &blob,
+            "b",
+            Envelope {
+                mail_from: Address::new("[email protected]"),
+                rcpt_to: to.iter().map(|t| Address::new(*t)).collect(),
+            },
+            &settings,
+            false,
+            time::now(),
+        )
+        .unwrap()
+}
+
+/// Waits until the message has been tried at least once.
+async fn tried(app: &App, id: &str) -> Outbound {
+    for _ in 0..200 {
+        if let Some(out) = app.outbox.lock().unwrap().get(id)
+            && out.attempts > 0
+        {
+            return out;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    panic!("{id} was never tried");
+}
+
+fn events(app: &App, kind: &str) -> Vec<serde_json::Map<String, serde_json::Value>> {
+    let events = app.events.lock().unwrap();
+    events
+        .get(None)
+        .0
+        .into_iter()
+        .filter(|e| e.kind == kind)
+        .map(|e| e.data)
+        .collect()
+}
+
+fn hosts(port: u16) -> Fixed {
+    Fixed {
+        port,
+        mx: HashMap::from([
+            (
+                "example.org".to_string(),
+                Mx::Hosts {
+                    hosts: vec!["mx.test".into()],
+                    secure: false,
+                },
+            ),
+            ("null.example".to_string(), Mx::Null),
+        ]),
+        ..Default::default()
+    }
+}
+
+#[tokio::test]
+async fn delivers_with_opportunistic_tls_and_per_recipient_status() {
+    let (port, seen, _) = scripted_mx().await;
+    let (app, dir) = relay("plain").await;
+    let id = hand_over(
+        &app,
+        &[
+            "[email protected]",
+            "[email protected]",
+            "[email protected]",
+            "[email protected]",
+            "[email protected]",
+        ],
+        b"Subject: hi\r\n\r\n.dot line\r\nend",
+    );
+    tokio::spawn(run(app.clone(), Arc::new(hosts(port))));
+    app.outbound_wake.notify_one();
+    let out = tried(&app, &id).await;
+
+    let status = |rcpt: &str| out.delivery_status[rcpt].clone();
+    assert_eq!(status("[email protected]").delivered, Delivered::Yes);
+    assert_eq!(
+        status("[email protected]").smtp_reply,
+        "250 2.0.0 queued as X1"
+    );
+    // RL-T11: the remote server's own reply.
+    assert_eq!(status("[email protected]").delivered, Delivered::No);
+    assert_eq!(
+        status("[email protected]").smtp_reply,
+        "550 5.1.1 no such user"
+    );
+    assert_eq!(status("[email protected]").delivered, Delivered::Queued);
+    assert_eq!(
+        status("[email protected]").smtp_reply,
+        "451 4.2.0 try later"
+    );
+    assert_eq!(status("[email protected]").delivered, Delivered::No);
+    assert!(
+        status("[email protected]")
+            .smtp_reply
+            .starts_with("556 5.1.10")
+    );
+    assert_eq!(status("[email protected]").delivered, Delivered::No);
+    assert!(
+        out.next_attempt_at.is_some(),
+        "one recipient is still queued"
+    );
+
+    let messages = &seen.lock().unwrap().messages;
+    assert_eq!(messages.len(), 1);
+    let (tls, body) = &messages[0];
+    assert!(tls, "STARTTLS was offered, so it was used");
+    assert_eq!(body, b"Subject: hi\r\n\r\n..dot line\r\nend\r\n");
+
+    let attempts = events(&app, "outbound.attempt");
+    assert_eq!(attempts[0]["delivered"], 1);
+    assert_eq!(attempts[0]["refused"], 1);
+    assert_eq!(attempts[0]["deferred"], 1);
+    assert!(attempts[0]["tls"]["version"].is_string());
+    let tls = events(&app, "outbound.tls");
+    assert_eq!(tls[0]["policyType"], "no-policy-found");
+    assert_eq!(tls[0]["result"], "success");
+
+    // Only the relay may give up on a message mid-send; afterwards the
+    // downstream can clear it.
+    assert_eq!(
+        app.outbox.lock().unwrap().destroy(&id),
+        crate::outbound::Destroyed::Yes
+    );
+    std::fs::remove_dir_all(dir).unwrap();
+}
+
+fn tlsa(data: Vec<u8>) -> Arc<Tlsa> {
+    Arc::new(Tlsa {
+        entries: vec![TlsaEntry {
+            is_end_entity: true,
+            is_spki: true,
+            matching: TlsaMatching::Sha256,
+            data,
+        }],
+        has_end_entities: true,
+        has_intermediates: false,
+    })
+}
+
+/// RL-T12, first half: DANE against the MX's real key, then a wrong one.
+#[tokio::test]
+async fn dane_is_enforced() {
+    let (port, seen, spki) = scripted_mx().await;
+
+    let (app, dir) = relay("dane-ok").await;
+    let id = hand_over(&app, &["[email protected]"], b"Subject: dane\r\n\r\nhi\r\n");
+    let resolve = Fixed {
+        secure: true,
+        tlsa: Some(tlsa(sha256(&spki))),
+        mx: HashMap::from([(
+            "example.org".to_string(),
+            Mx::Hosts {
+                hosts: vec!["mx.test".into()],
+                secure: true,
+            },
+        )]),
+        ..hosts(port)
+    };
+    tokio::spawn(run(app.clone(), Arc::new(resolve)));
+    app.outbound_wake.notify_one();
+    let out = tried(&app, &id).await;
+    assert_eq!(
+        out.delivery_status["[email protected]"].delivered,
+        Delivered::Yes
+    );
+    let tls = events(&app, "outbound.tls");
+    assert_eq!(tls[0]["policyType"], "tlsa");
+    assert_eq!(tls[0]["result"], "success");
+    assert!(
+        tls[0]["policyStrings"][0]
+            .as_str()
+            .unwrap()
+            .starts_with("3 1 1 ")
+    );
+    std::fs::remove_dir_all(dir).unwrap();
+
+    let (app, dir) = relay("dane-bad").await;
+    let id = hand_over(&app, &["[email protected]"], b"Subject: dane\r\n\r\nhi\r\n");
+    let resolve = Fixed {
+        secure: true,
+        tlsa: Some(tlsa(vec![0; 32])),
+        mx: HashMap::from([(
+            "example.org".to_string(),
+            Mx::Hosts {
+                hosts: vec!["mx.test".into()],
+                secure: true,
+            },
+        )]),
+        ..hosts(port)
+    };
+    tokio::spawn(run(app.clone(), Arc::new(resolve)));
+    app.outbound_wake.notify_one();
+    let out = tried(&app, &id).await;
+    let status = &out.delivery_status["[email protected]"];
+    assert_eq!(status.delivered, Delivered::Queued);
+    assert!(
+        status.smtp_reply.starts_with("451 4.4.1 "),
+        "{}",
+        status.smtp_reply
+    );
+    let tls = events(&app, "outbound.tls");
+    assert_eq!(tls[0]["policyType"], "tlsa");
+    assert_eq!(tls[0]["result"], "validation-failure");
+    assert_eq!(
+        seen.lock().unwrap().messages.len(),
+        1,
+        "only the first went"
+    );
+    std::fs::remove_dir_all(dir).unwrap();
+}
+
+/// RL-T12, second half: MTA-STS enforce with a certificate no CA signed,
+/// and with an MX host the policy doesn't list.
+#[tokio::test]
+async fn mta_sts_enforce_is_enforced() {
+    let (port, seen, _) = scripted_mx().await;
+    for (name, pattern) in [
+        ("sts-cert", MxPattern::Equals("mx.test".into())),
+        ("sts-mx", MxPattern::Equals("other.test".into())),
+    ] {
+        let (app, dir) = relay(name).await;
+        let id = hand_over(&app, &["[email protected]"], b"Subject: sts\r\n\r\nhi\r\n");
+        let resolve = Fixed {
+            sts: Some(Arc::new(Policy {
+                id: "1".into(),
+                mode: Mode::Enforce,
+                mx: vec![pattern].into_boxed_slice(),
+                max_age: 86_400,
+            })),
+            ..hosts(port)
+        };
+        tokio::spawn(run(app.clone(), Arc::new(resolve)));
+        app.outbound_wake.notify_one();
+        let out = tried(&app, &id).await;
+        let status = &out.delivery_status["[email protected]"];
+        assert_eq!(status.delivered, Delivered::Queued, "{name}");
+        let tls = events(&app, "outbound.tls");
+        assert_eq!(tls[0]["policyType"], "sts", "{name}");
+        assert_ne!(tls[0]["result"], "success", "{name}");
+        assert!(
+            tls[0]["policyStrings"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .any(|l| l == "mode: enforce"),
+            "{name}"
+        );
+        std::fs::remove_dir_all(dir).unwrap();
+    }
+    assert!(seen.lock().unwrap().messages.is_empty());
+}
+
+#[tokio::test]
+async fn expiry_refuses_what_is_still_queued() {
+    let (port, _, _) = scripted_mx().await;
+    let (app, dir) = relay("expiry").await;
+    let id = hand_over(&app, &["[email protected]"], b"Subject: x\r\n\r\nx\r\n");
+    tokio::spawn(run(app.clone(), Arc::new(hosts(port))));
+    app.outbound_wake.notify_one();
+    let out = tried(&app, &id).await;
+    assert_eq!(
+        out.delivery_status["[email protected]"].delivered,
+        Delivered::Queued
+    );
+    let gone = app
+        .outbox
+        .lock()
+        .unwrap()
+        .expire(time::now() + 4 * 86_400)
+        .unwrap();
+    assert_eq!(gone, vec![(id.clone(), 1)]);
+    let out = app.outbox.lock().unwrap().get(&id).unwrap();
+    let status = &out.delivery_status["[email protected]"];
+    assert_eq!(status.delivered, Delivered::No);
+    assert_eq!(
+        status.smtp_reply, "451 4.2.0 try later",
+        "the last real reply"
+    );
+    assert!(out.next_attempt_at.is_none());
+    assert_eq!(
+        app.outbox
+            .lock()
+            .unwrap()
+            .sweep_finished(time::now() + 12 * 86_400, super::FINISHED_KEEP),
+        1
+    );
+    std::fs::remove_dir_all(dir).unwrap();
+}

crates/relay/src/deliver/tls.rs

@@ -0,0 +1,285 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! STARTTLS for outbound mail: DANE (RFC 7672) where the destination's
+//! TLSA records are signed, WebPKI checks where its MTA-STS policy
+//! enforces (RFC 8461), and otherwise encryption without a certificate
+//! check (spec "Outbound").
+
+use inbuxa_outbound_tls::{Tlsa, dane};
+use inbuxa_relay_proto::types::TlsInfo;
+use rustls::{
+    CertificateError, ClientConfig, DigitallySignedStruct, Error, SignatureScheme,
+    client::{
+        WebPkiServerVerifier,
+        danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
+    },
+};
+use rustls_pki_types::{CertificateDer, ServerName, UnixTime};
+use std::{
+    sync::{Arc, OnceLock},
+    time::Duration,
+};
+use tokio::net::TcpStream;
+use tokio_rustls::{TlsConnector, client::TlsStream};
+
+/// What a delivery to one MX host asks of TLS.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TlsMode {
+    /// In the clear: an opportunistic session whose TLS didn't set up.
+    None,
+    /// STARTTLS when offered, certificate unchecked.
+    Opportunistic,
+    /// STARTTLS required (`outboundTls: required`), certificate unchecked.
+    Required,
+    /// The host's zone is signed but its TLSA records are unusable: RFC
+    /// 7672 §2.2 still asks for encryption.
+    DaneEncrypt,
+    /// The chain must match these TLSA records.
+    Dane(Arc<Tlsa>),
+    /// MTA-STS enforce: a WebPKI chain valid for the MX host's name.
+    Verified,
+}
+
+impl TlsMode {
+    pub fn is_required(&self) -> bool {
+        !matches!(self, TlsMode::None | TlsMode::Opportunistic)
+    }
+
+    /// The TLS-RPT policy type (RFC 8460 §4.4).
+    pub fn policy_type(&self) -> &'static str {
+        match self {
+            TlsMode::Dane(_) | TlsMode::DaneEncrypt => "tlsa",
+            TlsMode::Verified => "sts",
+            _ => "no-policy-found",
+        }
+    }
+}
+
+/// A TLS failure, with its TLS-RPT result type (RFC 8460 §4.3).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TlsFailure {
+    pub reason: String,
+    pub result_type: &'static str,
+    /// The handshake itself failed, as opposed to a policy check after it.
+    pub handshake: bool,
+}
+
+impl TlsFailure {
+    pub fn starttls_not_supported() -> Self {
+        TlsFailure {
+            reason: "STARTTLS isn't offered".into(),
+            result_type: "starttls-not-supported",
+            handshake: false,
+        }
+    }
+}
+
+pub struct TlsOutcome {
+    pub stream: TlsStream<TcpStream>,
+    pub info: TlsInfo,
+}
+
+pub async fn handshake(
+    tcp: TcpStream,
+    mx: &str,
+    domain: &str,
+    mode: &TlsMode,
+    timeout: Duration,
+) -> Result<TlsOutcome, TlsFailure> {
+    let failure = |reason: String, result_type| TlsFailure {
+        reason,
+        result_type,
+        handshake: true,
+    };
+    let name = ServerName::try_from(mx.to_string())
+        .map_err(|_| failure(format!("{mx} isn't a host name"), "validation-failure"))?;
+    let config = match mode {
+        TlsMode::Verified => verified(),
+        _ => unverified(),
+    };
+    let stream = tokio::time::timeout(timeout, TlsConnector::from(config).connect(name, tcp))
+        .await
+        .map_err(|_| failure("the handshake timed out".into(), "validation-failure"))?
+        .map_err(|e| {
+            let result_type = match e.get_ref().and_then(|inner| inner.downcast_ref::<Error>()) {
+                Some(Error::InvalidCertificate(CertificateError::NotValidForName))
+                | Some(Error::InvalidCertificate(CertificateError::NotValidForNameContext {
+                    ..
+                })) => "certificate-host-mismatch",
+                Some(Error::InvalidCertificate(
+                    CertificateError::Expired | CertificateError::ExpiredContext { .. },
+                )) => "certificate-expired",
+                Some(Error::InvalidCertificate(CertificateError::UnknownIssuer)) => {
+                    "certificate-not-trusted"
+                }
+                _ => "validation-failure",
+            };
+            failure(e.to_string(), result_type)
+        })?;
+    let (_, conn) = stream.get_ref();
+    if let TlsMode::Dane(tlsa) = mode {
+        let certificates = conn.peer_certificates();
+        if let Err(e) = dane::verify(tlsa, &[mx, domain], certificates) {
+            return Err(TlsFailure {
+                reason: e.description().into(),
+                result_type: "validation-failure",
+                handshake: false,
+            });
+        }
+    }
+    let info = TlsInfo {
+        version: conn
+            .protocol_version()
+            .map_or("unknown".into(), |v| format!("{v:?}")),
+        cipher: conn
+            .negotiated_cipher_suite()
+            .map_or("unknown".into(), |c| format!("{:?}", c.suite())),
+    };
+    Ok(TlsOutcome { stream, info })
+}
+
+/// WebPKI against the system's roots: the container image carries them.
+pub(super) fn verified() -> Arc<ClientConfig> {
+    static CONFIG: OnceLock<Arc<ClientConfig>> = OnceLock::new();
+    CONFIG
+        .get_or_init(|| {
+            let provider = crate::link::tls::provider();
+            let mut roots = rustls::RootCertStore::empty();
+            let found = rustls_native_certs::load_native_certs();
+            let (added, _) = roots.add_parsable_certificates(found.certs);
+            if added == 0 {
+                eprintln!(
+                    "inbuxa-relay: no CA roots found; MTA-STS destinations can't be delivered to"
+                );
+            }
+            let verifier =
+                WebPkiServerVerifier::builder_with_provider(Arc::new(roots), provider.clone())
+                    .build();
+            let config = ClientConfig::builder_with_provider(provider)
+                .with_safe_default_protocol_versions()
+                .expect("default TLS versions");
+            Arc::new(match verifier {
+                Ok(verifier) => config.with_webpki_verifier(verifier).with_no_client_auth(),
+                // No roots: nothing verifies, so nothing is delivered.
+                Err(_) => config
+                    .dangerous()
+                    .with_custom_certificate_verifier(Arc::new(Refuse))
+                    .with_no_client_auth(),
+            })
+        })
+        .clone()
+}
+
+/// Any certificate: opportunistic TLS, and DANE, which checks the chain
+/// itself after the handshake.
+fn unverified() -> Arc<ClientConfig> {
+    static CONFIG: OnceLock<Arc<ClientConfig>> = OnceLock::new();
+    CONFIG
+        .get_or_init(|| {
+            Arc::new(
+                ClientConfig::builder_with_provider(crate::link::tls::provider())
+                    .with_safe_default_protocol_versions()
+                    .expect("default TLS versions")
+                    .dangerous()
+                    .with_custom_certificate_verifier(Arc::new(AnyCertificate))
+                    .with_no_client_auth(),
+            )
+        })
+        .clone()
+}
+
+#[derive(Debug)]
+struct AnyCertificate;
+
+impl ServerCertVerifier for AnyCertificate {
+    fn verify_server_cert(
+        &self,
+        _: &CertificateDer<'_>,
+        _: &[CertificateDer<'_>],
+        _: &ServerName<'_>,
+        _: &[u8],
+        _: UnixTime,
+    ) -> Result<ServerCertVerified, Error> {
+        Ok(ServerCertVerified::assertion())
+    }
+
+    // The signatures still have to be right: the server must hold the key
+    // of the certificate it sent, or DANE would check a stolen chain.
+    fn verify_tls12_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, Error> {
+        rustls::crypto::verify_tls12_signature(
+            message,
+            cert,
+            dss,
+            &crate::link::tls::provider().signature_verification_algorithms,
+        )
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, Error> {
+        rustls::crypto::verify_tls13_signature(
+            message,
+            cert,
+            dss,
+            &crate::link::tls::provider().signature_verification_algorithms,
+        )
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
+        crate::link::tls::provider()
+            .signature_verification_algorithms
+            .supported_schemes()
+    }
+}
+
+#[derive(Debug)]
+struct Refuse;
+
+impl ServerCertVerifier for Refuse {
+    fn verify_server_cert(
+        &self,
+        _: &CertificateDer<'_>,
+        _: &[CertificateDer<'_>],
+        _: &ServerName<'_>,
+        _: &[u8],
+        _: UnixTime,
+    ) -> Result<ServerCertVerified, Error> {
+        Err(Error::InvalidCertificate(CertificateError::UnknownIssuer))
+    }
+
+    fn verify_tls12_signature(
+        &self,
+        _: &[u8],
+        _: &CertificateDer<'_>,
+        _: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, Error> {
+        Err(Error::InvalidCertificate(CertificateError::UnknownIssuer))
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        _: &[u8],
+        _: &CertificateDer<'_>,
+        _: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, Error> {
+        Err(Error::InvalidCertificate(CertificateError::UnknownIssuer))
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
+        crate::link::tls::provider()
+            .signature_verification_algorithms
+            .supported_schemes()
+    }
+}

crates/relay/src/link/jmap.rs

@@ -10,7 +10,7 @@
 //! them. Before the join completes, only `Relay/join`, then `Relay/get` and
 //! `Relay/set`, answer; everything else is `forbidden`.
 
-use crate::{app::App, link::tls::cert_pin, state::JoinStage};
+use crate::{app::App, link::tls::cert_pin, outbound::Destroyed, state::JoinStage};
 use inbuxa_relay_proto::{
     CAPABILITY, CORE_CAPABILITY, b64,
     jmap::{Invocation, MethodError, Request, Response, method},
@@ -867,10 +867,17 @@ fn outbound_set(app: &App, args: &Map<String, Value>) -> MethodResult {
         );
     }
     for id in string_list(args, "destroy")?.unwrap_or_default() {
-        if outbox.destroy(&id) {
-            destroyed.push(id);
-        } else {
-            not_destroyed.insert(id, set_error("notFound", "no such message"));
+        match outbox.destroy(&id) {
+            Destroyed::Yes => destroyed.push(id),
+            Destroyed::NotFound => {
+                not_destroyed.insert(id, set_error("notFound", "no such message"));
+            }
+            Destroyed::InFlight => {
+                not_destroyed.insert(
+                    id,
+                    set_error("forbidden", "a delivery attempt is under way; try again"),
+                );
+            }
         }
     }
     let new_state = outbox.state();
@@ -879,6 +886,9 @@ fn outbound_set(app: &App, args: &Map<String, Value>) -> MethodResult {
         app.notify(type_name::OUTBOUND, new_state.clone());
         app.relay_changed();
     }
+    if !created.is_empty() {
+        app.outbound_wake.notify_one();
+    }
     Ok(json!({
         "accountId": ACCOUNT_ID,
         "oldState": old_state,

crates/relay/src/main.rs

@@ -13,6 +13,7 @@
 
 mod app;
 mod changelog;
+mod deliver;
 mod events;
 mod link;
 mod outbound;
@@ -188,6 +189,11 @@ fn run(paths: &Paths, link_listen: SocketAddr) -> Result<(), String> {
         let link = tokio::spawn(link::serve(app.clone(), listener));
         tokio::spawn(smtp::serve(app.clone()));
         tokio::spawn(sweeper(app.clone()));
+        let outbound = app.clone();
+        tokio::spawn(async move {
+            let dns = Arc::new(deliver::route::Dns::new().await);
+            deliver::run(outbound, dns).await
+        });
         tokio::select! {
             result = link => match result {
                 Ok(Err(e)) => Err(format!("the link stopped: {e}")),

crates/relay/src/outbound.rs

@@ -18,7 +18,7 @@ use crate::{
 };
 use inbuxa_relay_proto::{
     time,
-    types::{Address, DeliveryStatus, Envelope, Outbound, RelaySettings},
+    types::{Address, Delivered, DeliveryStatus, Envelope, Outbound, RelaySettings},
 };
 use serde::{Deserialize, Serialize};
 use std::{
@@ -45,6 +45,28 @@ pub struct Meta {
     pub next_attempt_at: u64,
     pub attempts: u32,
     pub expires_at: u64,
+    /// When the last recipient got a final status.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub finished_at: Option<u64>,
+}
+
+impl Meta {
+    pub fn queued(&self) -> impl Iterator<Item = &Address> {
+        self.envelope.rcpt_to.iter().filter(|r| {
+            self.delivery_status
+                .get(&r.email)
+                .is_some_and(|s| s.delivered == Delivered::Queued)
+        })
+    }
+}
+
+/// What `destroy` did.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Destroyed {
+    Yes,
+    NotFound,
+    /// A delivery attempt is under way; it can't be pulled back mid-send.
+    InFlight,
 }
 
 struct Entry {
@@ -72,6 +94,8 @@ pub struct Outbox {
     entries: BTreeMap<String, Entry>,
     log: Changelog,
     bytes: u64,
+    /// Messages with a delivery attempt under way.
+    attempting: HashSet<String>,
 }
 
 pub fn new_id() -> String {
@@ -94,6 +118,7 @@ impl Outbox {
             entries: BTreeMap::new(),
             log: Changelog::default(),
             bytes: 0,
+            attempting: HashSet::new(),
         };
         let mut messages = Vec::new();
         for entry in fs::read_dir(dir)? {
@@ -202,6 +227,7 @@ impl Outbox {
             next_attempt_at: now,
             attempts: 0,
             expires_at: now + u64::from(settings.outbound_expiry),
+            finished_at: None,
         };
         self.commit(blob, &meta)
             .map_err(|e| refuse("serverFail", format!("could not spool the message: {e}")))?;
@@ -235,10 +261,7 @@ impl Outbox {
     /// The JMAP object for one message.
     pub fn get(&self, id: &str) -> Option<Outbound> {
         let m = &self.entries.get(id)?.meta;
-        let queued = m
-            .delivery_status
-            .values()
-            .any(|s| s.delivered == inbuxa_relay_proto::types::Delivered::Queued);
+        let queued = m.queued().next().is_some();
         Some(Outbound {
             id: m.id.clone(),
             created_at: time::utc_date(m.created_at),
@@ -268,16 +291,138 @@ impl Outbox {
     }
 
     /// Cancels a queued message, or clears a finished one.
-    pub fn destroy(&mut self, id: &str) -> bool {
+    pub fn destroy(&mut self, id: &str) -> Destroyed {
+        if self.attempting.contains(id) {
+            return Destroyed::InFlight;
+        }
         let Some(e) = self.entries.remove(id) else {
-            return false;
+            return Destroyed::NotFound;
         };
         self.bytes = self.bytes.saturating_sub(e.meta.size);
         let _ = fs::remove_file(self.meta_path(id));
         let _ = fs::remove_file(self.eml_path(id));
         let _ = sync_dir(&self.dir);
         self.log.destroyed(id);
-        true
+        Destroyed::Yes
+    }
+
+    /// Up to `limit` messages due for an attempt, oldest first, now marked
+    /// as being attempted.
+    pub fn take_due(&mut self, now: u64, limit: usize) -> Vec<Meta> {
+        let mut due: Vec<&Meta> = self
+            .entries
+            .values()
+            .map(|e| &e.meta)
+            .filter(|m| {
+                m.next_attempt_at <= now
+                    && m.expires_at > now
+                    && !self.attempting.contains(&m.id)
+                    && m.queued().next().is_some()
+            })
+            .collect();
+        due.sort_by_key(|m| (m.next_attempt_at, m.created_at));
+        let due: Vec<Meta> = due.into_iter().take(limit).cloned().collect();
+        for m in &due {
+            self.attempting.insert(m.id.clone());
+        }
+        due
+    }
+
+    /// An attempt ended without trying: it can be taken again.
+    pub fn release(&mut self, id: &str) {
+        self.attempting.remove(id);
+    }
+
+    /// Records an attempt: each recipient's new status, then the next try
+    /// from `outbound_retry`, the last interval repeating. False if the
+    /// message is gone.
+    pub fn record(
+        &mut self,
+        id: &str,
+        statuses: Vec<(String, DeliveryStatus)>,
+        retry: &[u32],
+        now: u64,
+    ) -> io::Result<bool> {
+        self.attempting.remove(id);
+        let Some(e) = self.entries.get_mut(id) else {
+            return Ok(false);
+        };
+        let mut meta = e.meta.clone();
+        for (rcpt, status) in statuses {
+            if let Some(current) = meta.delivery_status.get_mut(&rcpt)
+                && current.delivered == Delivered::Queued
+            {
+                *current = status;
+            }
+        }
+        meta.attempts += 1;
+        let step = (meta.attempts as usize - 1).min(retry.len().saturating_sub(1));
+        meta.next_attempt_at = now + u64::from(retry.get(step).copied().unwrap_or(3600));
+        if meta.queued().next().is_none() {
+            meta.finished_at = Some(now);
+        }
+        self.save(meta)?;
+        Ok(true)
+    }
+
+    /// Gives up on recipients still queued past `expiresAt`: they are
+    /// refused, with the last reply seen. Returns the messages and how
+    /// many recipients each gave up.
+    pub fn expire(&mut self, now: u64) -> io::Result<Vec<(String, usize)>> {
+        let old: Vec<Meta> = self
+            .entries
+            .values()
+            .map(|e| &e.meta)
+            .filter(|m| {
+                m.expires_at <= now
+                    && !self.attempting.contains(&m.id)
+                    && m.queued().next().is_some()
+            })
+            .cloned()
+            .collect();
+        let mut out = Vec::new();
+        for mut meta in old {
+            let mut count = 0;
+            for status in meta.delivery_status.values_mut() {
+                if status.delivered == Delivered::Queued {
+                    status.delivered = Delivered::No;
+                    if status.smtp_reply.is_empty() {
+                        status.smtp_reply = "421 4.4.7 Delivery time expired".into();
+                    }
+                    count += 1;
+                }
+            }
+            meta.finished_at = Some(now);
+            out.push((meta.id.clone(), count));
+            self.save(meta)?;
+        }
+        Ok(out)
+    }
+
+    /// Deletes finished messages the downstream never cleared, `keep`
+    /// seconds after they finished. Returns how many.
+    pub fn sweep_finished(&mut self, now: u64, keep: u64) -> usize {
+        let old: Vec<String> = self
+            .entries
+            .values()
+            .filter(|e| e.meta.finished_at.is_some_and(|at| at + keep <= now))
+            .map(|e| e.meta.id.clone())
+            .collect();
+        old.iter()
+            .filter(|id| self.destroy(id) == Destroyed::Yes)
+            .count()
+    }
+
+    /// Writes changed properties, the commit point for a status.
+    fn save(&mut self, meta: Meta) -> io::Result<()> {
+        let json = serde_json::to_vec(&meta).map_err(io::Error::other)?;
+        write_atomic(&self.meta_path(&meta.id), &json)?;
+        let modseq = self.log.bump();
+        if let Some(e) = self.entries.get_mut(&meta.id) {
+            e.meta = meta;
+            e.modseq = modseq;
+        }
+        Ok(())
     }
 }
 
@@ -364,7 +509,6 @@ fn check_parameters(address: &Address, field: &str) -> Result<(), Refusal> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use inbuxa_relay_proto::types::Delivered;
 
     fn settings() -> RelaySettings {
         RelaySettings {
@@ -503,8 +647,8 @@ mod tests {
         assert_eq!(fs::read_dir(&out).unwrap().count(), 2);
 
         let s1 = outbox.state();
-        assert!(outbox.destroy(&id));
-        assert!(!outbox.destroy(&id));
+        assert_eq!(outbox.destroy(&id), Destroyed::Yes);
+        assert_eq!(outbox.destroy(&id), Destroyed::NotFound);
         assert_eq!(outbox.changes(&s1, 10).unwrap().destroyed, vec![id]);
         assert_eq!(fs::read_dir(&out).unwrap().count(), 0);
         fs::remove_dir_all(d).unwrap();

docs/spec/features/relay.md

@@ -250,7 +250,7 @@ with the server's own `inbuxa:` objects. In particular, the relay's
 | `heldMaxAge` | downstream | How long held inbound mail waits for the downstream. Default 30 days, maximum 90 |
 | `spoolMaxBytes` | downstream | Spool size cap |
 | `outboundRetry`, `outboundExpiry` | downstream | Retry schedule and give-up time. Defaults match the server's (three days) |
-| `outboundTls` | downstream | Opportunistic TLS default; DANE and MTA-STS always enforced where published |
+| `outboundTls` | downstream | Where neither DANE nor MTA-STS applies: `opportunistic` (the default; STARTTLS when offered) or `required` (no delivery without STARTTLS). The certificate isn't checked either way. DANE and MTA-STS are always enforced where published |
 | `smtpCertificate` | downstream | PEM chain and private key. `/get` never returns the key |
 | `sealingKey` | downstream | The downstream's public sealing key |
 | `linkAllowFrom` | downstream | Addresses and CIDR ranges that may dial the link port. Default empty: anyone (RL-33) |
@@ -450,6 +450,31 @@ When no link is up, or after RL-13:
   server's real reply. The relay never writes a message of its own.
 - TLS results for each destination are sent as `RelayEvent`s. The downstream
   builds TLS-RPT reports from them, as it does from its own deliveries.
+- How a domain is tried: its MX hosts best first (no MX means the domain
+  itself; a null MX, RFC 7505, refuses every recipient with `556 5.1.10`),
+  at most five hosts and two addresses each. With the MX answer and the
+  host's addresses both DNSSEC-signed, signed TLSA records make it a DANE
+  delivery, which takes precedence over MTA-STS. A signed zone whose TLSA
+  records are unusable still requires encryption (RFC 7672 §2.2). Bogus
+  TLSA records skip the host. Otherwise an MTA-STS `enforce` policy skips
+  hosts it doesn't list and requires a WebPKI certificate valid for the
+  host; `testing` only reports. Opportunistic TLS that fails to set up is
+  tried once more in the clear. A host that can't be reached, or fails its
+  TLS policy, passes the attempt to the next.
+- Each recipient's reply is the remote server's own: RCPT's for a refusal
+  or a deferral, the final reply after DATA for the rest. When no host
+  could take the attempt, the recipients stay queued with the reason, as a
+  `4.4.x` reply. The next try follows `outboundRetry`, the last interval
+  repeating. At `expiresAt` what is still queued becomes `delivered: no`,
+  keeping the last reply seen. A finished message the downstream never
+  clears is deleted after seven days. `destroy` is refused (`forbidden`)
+  while an attempt is under way.
+- Events: `outbound.attempt` per session that reached MAIL FROM (`id`,
+  `domain`, `mx`, `ip`, `tls`, and counts delivered, refused and
+  deferred); `outbound.tls` per session where TLS applied or failed
+  (`domain`, `mx`, `ip`, `policyType` as TLS-RPT names it, `policyStrings`,
+  `result` as a TLS-RPT result type or `success`, and `reason`);
+  `outbound.expired` (`id`, `recipients`).
 - **RL-20.** If no relay can be reached, the downstream keeps the message in
   its own queue and retries. With several relays, it hands messages to the
   linked ones in turn, skipping any marked `inboundOnly`.
@@ -771,9 +796,13 @@ In progress. Done so far:
   once, and `relay.link-up` when it is back (RL-T23). Both reach the log,
   webhooks and OpenTelemetry.
 - `Outbound` on the relay: `Outbound/set` create and destroy, `/get` and
-  `/changes`, with push, in `spool/out/` (RL-19, RL-T10). The relay doesn't
-  deliver it yet, and the downstream doesn't hand anything over yet; both
-  are the rest of step 4.
+  `/changes`, with push, in `spool/out/` (RL-19, RL-T10).
+- Outbound delivery on the relay: MX, DANE, MTA-STS and `outboundTls`, per
+  recipient status from the remote server's replies, retries, expiry, and
+  `outbound.*` events (RL-T12 tested against a scripted MX with fixed DNS
+  answers). The release binary went from 5.4 MB to 8.1 MB with it
+  (glibc, measured 2026-10-30). The downstream doesn't hand anything over
+  yet: that is the rest of step 4.
 - `crates/outbound-tls` (RL-30): the TLSA and MTA-STS types, the DANE
   certificate check, MTA-STS policy parsing and matching, and the DNSSEC
   lookups (with the bogus-answer recheck), moved out of `crates/smtp` and