diff --git a/Cargo.lock b/Cargo.lock index 17cf3b9c..5eaa8456 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/outbound-tls/src/dnssec.rs b/crates/outbound-tls/src/dnssec.rs index 9516d588..e98c934f 100644 --- a/crates/outbound-tls/src/dnssec.rs +++ b/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, Option); +/// 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<'_>, diff --git a/crates/relay-proto/src/types.rs b/crates/relay-proto/src/types.rs index 7952ce52..05fa90ef 100644 --- a/crates/relay-proto/src/types.rs +++ b/crates/relay-proto/src/types.rs @@ -65,6 +65,8 @@ pub struct RelaySettings { pub outbound_retry: Vec, /// 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, @@ -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 { diff --git a/crates/relay/Cargo.toml b/crates/relay/Cargo.toml index 110518a5..11f6d0b3 100644 --- a/crates/relay/Cargo.toml +++ b/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" diff --git a/crates/relay/src/app.rs b/crates/relay/src/app.rs index 00e539f7..3318a9f7 100644 --- a/crates/relay/src/app.rs +++ b/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, pub spool: Mutex, pub outbox: Mutex, + /// Wakes the delivery engine when mail is handed over. + pub outbound_wake: Notify, pub events: Mutex, pub push: broadcast::Sender, /// 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); diff --git a/crates/relay/src/deliver/client.rs b/crates/relay/src/deliver/client.rs new file mode 100644 index 00000000..85d4bf32 --- /dev/null +++ b/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, +} + +fn host_failure(reason: impl Into) -> 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, + pub tls: Option, +} + +impl Session<'_> { + pub async fn run(&self) -> Result { + 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 { + 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( + &self, + conn: &mut BufReader, + ) -> Result { + 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( + &self, + mut conn: BufReader, + ehlo: &Ehlo, + tls: Option, + ) -> Result { + 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, + 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> { + 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( + conn: &mut BufReader, + 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( + conn: &mut BufReader, +) -> 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( + conn: &mut BufReader, + timeout: Duration, +) -> Result { + 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( + conn: &mut BufReader, +) -> Result { + let mut code = None; + let mut lines: Vec = 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::().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( + conn: &mut BufReader, + 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, +} + +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) { + 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) { + 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 { + 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("a@example.com"); + 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:\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: SIZE=10 BODY=8BITMIME SMTPUTF8 RET=HDRS\r\n" + ); + let mut rcpt = Address::new("b@example.org"); + rcpt.parameters = Some([("NOTIFY".to_string(), Some("NEVER".to_string()))].into()); + assert_eq!(rcpt_to(&rcpt, &bare), "RCPT TO:\r\n"); + assert_eq!( + rcpt_to(&rcpt, &full), + "RCPT TO: NOTIFY=NEVER\r\n" + ); + } +} diff --git a/crates/relay/src/deliver/https.rs b/crates/relay/src/deliver/https.rs new file mode 100644 index 00000000..4dc452fd --- /dev/null +++ b/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, 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::::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()) +} diff --git a/crates/relay/src/deliver/mod.rs b/crates/relay/src/deliver/mod.rs new file mode 100644 index 00000000..63311e1c --- /dev/null +++ b/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(app: Arc, resolve: Arc) { + 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(app: &App, resolve: &R, settings: &RelaySettings, meta: Meta) { + let path = app.outbox.lock().expect("outbox").eml_path(&meta.id); + let mut by_domain: BTreeMap> = 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 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, + mode: &TlsMode, + sts: Option<&Policy>, + failure: Option<&TlsFailure>, + ) { + let policy_strings: Vec = 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 { + 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() +} diff --git a/crates/relay/src/deliver/route.rs b/crates/relay/src/deliver/route.rs new file mode 100644 index 00000000..f83e3c75 --- /dev/null +++ b/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, + 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> + 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, bool), LookupError>> + Send; + + fn tlsa(&self, host: &str) -> impl Future> + Send; + + /// The domain's MTA-STS policy, if it publishes one. + fn mta_sts(&self, domain: &str) -> impl Future>> + 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, 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 { + 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 { + 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, 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 { + 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> { + 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 = 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, + } + } +} diff --git a/crates/relay/src/deliver/tests.rs b/crates/relay/src/deliver/tests.rs new file mode 100644 index 00000000..2115128f --- /dev/null +++ b/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 { + aws_lc_rs::digest::digest(&aws_lc_rs::digest::SHA256, data) + .as_ref() + .to_vec() + } +} + +#[derive(Default)] +struct Fixed { + port: u16, + mx: HashMap, + secure: bool, + tlsa: Option>, + sts: Option>, +} + +impl Resolve for Fixed { + async fn mx(&self, domain: &str) -> Result { + self.mx + .get(domain) + .cloned() + .ok_or_else(|| LookupError::Permanent("the domain doesn't exist".into())) + } + + async fn addresses(&self, _: &str, _: bool) -> Result<(Vec, bool), LookupError> { + Ok((vec![IpAddr::V4(Ipv4Addr::LOCALHOST)], self.secure)) + } + + async fn tlsa(&self, _: &str) -> Result { + Ok(match &self.tlsa { + Some(t) => TlsaResult::Secure(t.clone()), + None => TlsaResult::Missing, + }) + } + + async fn mta_sts(&self, _: &str) -> Option> { + self.sts.clone() + } + + fn port(&self) -> u16 { + self.port + } +} + +/// What the scripted MX saw. +#[derive(Default)] +struct Seen { + messages: Vec<(bool, Vec)>, +} + +/// 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>, Vec) { + 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( + conn: &mut BufReader, + seen: &Mutex, + tls: bool, +) -> Option { + 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, 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("jane@example.com"), + 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> { + 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, + &[ + "bob@example.org", + "nobody@example.org", + "later@example.org", + "x@null.example", + "y@missing.example", + ], + 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("bob@example.org").delivered, Delivered::Yes); + assert_eq!( + status("bob@example.org").smtp_reply, + "250 2.0.0 queued as X1" + ); + // RL-T11: the remote server's own reply. + assert_eq!(status("nobody@example.org").delivered, Delivered::No); + assert_eq!( + status("nobody@example.org").smtp_reply, + "550 5.1.1 no such user" + ); + assert_eq!(status("later@example.org").delivered, Delivered::Queued); + assert_eq!( + status("later@example.org").smtp_reply, + "451 4.2.0 try later" + ); + assert_eq!(status("x@null.example").delivered, Delivered::No); + assert!( + status("x@null.example") + .smtp_reply + .starts_with("556 5.1.10") + ); + assert_eq!(status("y@missing.example").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) -> Arc { + 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, &["bob@example.org"], 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["bob@example.org"].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, &["bob@example.org"], 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["bob@example.org"]; + 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, &["bob@example.org"], 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["bob@example.org"]; + 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, &["later@example.org"], 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["later@example.org"].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["later@example.org"]; + 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(); +} diff --git a/crates/relay/src/deliver/tls.rs b/crates/relay/src/deliver/tls.rs new file mode 100644 index 00000000..41b9fc1b --- /dev/null +++ b/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), + /// 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, + pub info: TlsInfo, +} + +pub async fn handshake( + tcp: TcpStream, + mx: &str, + domain: &str, + mode: &TlsMode, + timeout: Duration, +) -> Result { + 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::()) { + 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 { + static CONFIG: OnceLock> = 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 { + static CONFIG: OnceLock> = 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 { + 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 { + 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 { + rustls::crypto::verify_tls13_signature( + message, + cert, + dss, + &crate::link::tls::provider().signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + 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 { + Err(Error::InvalidCertificate(CertificateError::UnknownIssuer)) + } + + fn verify_tls12_signature( + &self, + _: &[u8], + _: &CertificateDer<'_>, + _: &DigitallySignedStruct, + ) -> Result { + Err(Error::InvalidCertificate(CertificateError::UnknownIssuer)) + } + + fn verify_tls13_signature( + &self, + _: &[u8], + _: &CertificateDer<'_>, + _: &DigitallySignedStruct, + ) -> Result { + Err(Error::InvalidCertificate(CertificateError::UnknownIssuer)) + } + + fn supported_verify_schemes(&self) -> Vec { + crate::link::tls::provider() + .signature_verification_algorithms + .supported_schemes() + } +} diff --git a/crates/relay/src/link/jmap.rs b/crates/relay/src/link/jmap.rs index f9470be2..178e3c6d 100644 --- a/crates/relay/src/link/jmap.rs +++ b/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) -> 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) -> 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, diff --git a/crates/relay/src/main.rs b/crates/relay/src/main.rs index ab0db699..f3673fcb 100644 --- a/crates/relay/src/main.rs +++ b/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}")), diff --git a/crates/relay/src/outbound.rs b/crates/relay/src/outbound.rs index 260be180..82cca2be 100644 --- a/crates/relay/src/outbound.rs +++ b/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, +} + +impl Meta { + pub fn queued(&self) -> impl Iterator { + 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, log: Changelog, bytes: u64, + /// Messages with a delivery attempt under way. + attempting: HashSet, } 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 { 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 { + 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 = 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 { + 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> { + let old: Vec = 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 = 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(); diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index fe765833..eb6fe1fc 100644 --- a/docs/spec/features/relay.md +++ b/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