From dd2554aa8272af8dab8765548e74dbe0e72a9581 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Fri, 30 Oct 2026 09:16:02 -0700 Subject: [PATCH] relay: hand outbound mail to relays and follow its status With sendThrough on, every non-local recipient is handed to a linked relay that may send, in turn, skipping inboundOnly ones; with none linked the mail waits in the queue (RL-20, RL-24). The queue marks the recipients as handed to the relay and shows its replies. The lowest linked node follows Outbound by push, records each final status so the queue sends the DSNs, and destroys the relay's copy. DeliveryStatus gains mx, the host a reply came from. The relay's test-mx feature sends all mail to one local port for the system test, which now covers hand-over, delivery, one DSN for a refusal (RL-T11), and a deferral shown in the queue. --- Cargo.lock | 1 + crates/common/src/config/inner.rs | 2 + crates/common/src/config/smtp/queue.rs | 4 + crates/common/src/lib.rs | 4 + crates/features/src/relay/mod.rs | 5 + crates/features/src/relay/outbound.rs | 305 ++++++++++++++++++ crates/relay-proto/src/types.rs | 4 + crates/relay/Cargo.toml | 5 + crates/relay/src/deliver/mod.rs | 7 +- crates/relay/src/deliver/route.rs | 33 ++ crates/relay/src/main.rs | 9 + crates/services/src/inbuxa_relay_link.rs | 297 ++++++++++++++++-- crates/smtp/Cargo.toml | 2 + crates/smtp/src/outbound/delivery.rs | 44 ++- crates/smtp/src/outbound/mod.rs | 9 + crates/smtp/src/outbound/relay.rs | 378 +++++++++++++++++++++++ crates/smtp/src/queue/mod.rs | 3 + docs/spec/features/relay.md | 34 +- tests/src/system/relay.rs | 202 +++++++++++- 19 files changed, 1318 insertions(+), 30 deletions(-) create mode 100644 crates/features/src/relay/outbound.rs create mode 100644 crates/smtp/src/outbound/relay.rs diff --git a/Cargo.lock b/Cargo.lock index 5eaa8456..07252fae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8162,6 +8162,7 @@ dependencies = [ "hashify", "inbuxa-features", "inbuxa-outbound-tls", + "inbuxa-relay-proto", "mail-auth", "mail-builder 1.0.0", "mail-parser", diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index 44f3da74..6e0e9a9e 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -100,6 +100,7 @@ impl Data { smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"), build_errors: Default::default(), audit: Default::default(), + relay_senders: Default::default(), asn_geo_data: Default::default(), } } @@ -246,6 +247,7 @@ impl Default for Data { smtp_connectors: TlsConnectors::try_new().unwrap(), build_errors: Default::default(), audit: Default::default(), + relay_senders: Default::default(), asn_geo_data: Default::default(), lookup_stores: Default::default(), } diff --git a/crates/common/src/config/smtp/queue.rs b/crates/common/src/config/smtp/queue.rs index 1dd4faa6..963d63cb 100644 --- a/crates/common/src/config/smtp/queue.rs +++ b/crates/common/src/config/smtp/queue.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use super::*; @@ -78,6 +80,8 @@ pub enum RoutingStrategy { Local, Mx(MxConfig), Relay(RelayConfig), + // inbuxa: handed to an inbuxa Relay over its link (relay spec, RL-24) + InbuxaRelay, } #[derive(Clone, Debug)] diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 4e6984a8..e44da706 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -194,6 +194,10 @@ pub struct Data { // inbuxa: the audit log's chain heads and recent-access marks (AU) pub audit: inbuxa_features::audit::AuditLog, + + // inbuxa: the relays outbound mail goes through, and this node's links + // to them (relay spec, RL-20, RL-24) + pub relay_senders: inbuxa_features::relay::outbound::Senders, } #[derive(Clone)] diff --git a/crates/features/src/relay/mod.rs b/crates/features/src/relay/mod.rs index 47b946a6..6c17f230 100644 --- a/crates/features/src/relay/mod.rs +++ b/crates/features/src/relay/mod.rs @@ -17,10 +17,14 @@ //! - `r` + relay id (u32): a relay ([`Relay`]), as JSON. //! - `s` + relay id (u32) + node id (u64): that node's link to the relay //! ([`link::NodeLink`]), as JSON. +//! - `i` + relay id (u32) + `Inbound` id: a message taken from the relay. +//! - `o` + relay id (u32) + `Outbound` id: a message handed to the relay +//! ([`outbound::Handed`]), as JSON. //! //! Numbers are big-endian. There are few relays, so they're read whole. pub mod link; +pub mod outbound; use inbuxa_relay_proto::{ b64, @@ -52,6 +56,7 @@ const KIND_KEYS: u8 = b'k'; const KIND_RELAY: u8 = b'r'; const KIND_STATUS: u8 = b's'; const KIND_INGESTED: u8 = b'i'; +const KIND_HANDED: u8 = b'o'; const CREATE_ATTEMPTS: usize = 5; const MAX_NAME: usize = 100; diff --git a/crates/features/src/relay/outbound.rs b/crates/features/src/relay/outbound.rs new file mode 100644 index 00000000..107ca389 --- /dev/null +++ b/crates/features/src/relay/outbound.rs @@ -0,0 +1,305 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Outbound mail through relays, the downstream's side (relay spec, +//! "Outbound routing", RL-20, RL-24): whether mail goes through relays, +//! which of them this node can hand a message to, and the messages handed +//! to each until it reports a final status. +//! +//! A message handed over is kept under `o` + relay id (u32) + the relay's +//! `Outbound` id ([`Handed`]), as JSON, in the shared store: on a cluster +//! the node that follows a relay's statuses needn't be the one that handed +//! the message over. + +use super::{ + FEATURE, Json, KIND_HANDED, Relay, + link::{INBOUND_ID_MAX, is_inbound_id}, +}; +use inbuxa_relay_proto::types::Envelope; +use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize}; +use std::{ + collections::{BTreeMap, HashSet}, + future::Future, + pin::Pin, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, +}; +use store::{ + Deserialize, IterateParams, SUBSPACE_INBUXA, Serialize, Store, ValueKey, + write::{AnyClass, BatchBuilder, ValueClass}, +}; +use trc::AddContext; + +/// Why a relay didn't take a message. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Refused { + /// No relay will: the message is too large, or its sender isn't at + /// one of the relays' domains. An SMTP reply, e.g. `552 5.3.4 ...`. + Permanent(String), + /// Not this relay, or not now: the link failed, or its spool is full. + Temporary(String), +} + +pub type HandOver<'a> = Pin> + Send + 'a>>; + +/// One relay link on this node, able to take a message: uploads it and +/// creates an `Outbound`, whose id it returns. +pub trait Sender: Send + Sync { + fn hand_over(&self, message: Vec, envelope: Envelope) -> HandOver<'_>; +} + +/// A relay this node is linked to. +pub struct Linked { + pub relay_id: u32, + pub name: String, + pub sender: Box, +} + +/// The relays mail goes through, as this node sees them. Read from the +/// registry by the link service, and on first use. +#[derive(Default)] +pub struct Senders { + state: Mutex, + next: AtomicUsize, +} + +#[derive(Default)] +struct State { + /// The relays were read at least once. + known: bool, + /// An enabled relay has `sendThrough` on (RL-24). + send_through: bool, + /// Enabled relays not marked `inboundOnly` (RL-20). + may_send: HashSet, + linked: BTreeMap>, +} + +impl Senders { + /// Takes the relays as they are in the registry now. + pub fn set_relays(&self, relays: &[Relay]) { + let mut state = self.state.lock().expect("senders"); + state.known = true; + state.send_through = relays.iter().any(|r| r.enabled && r.send_through); + state.may_send = relays + .iter() + .filter(|r| r.enabled && !r.inbound_only) + .map(|r| r.id) + .collect(); + } + + /// Whether the relays were read yet. + pub fn known(&self) -> bool { + self.state.lock().expect("senders").known + } + + /// Reads the relays, if nothing has yet. + pub async fn load(&self, data: &Store) -> trc::Result<()> { + if !self.known() { + let relays = super::all(data).await?; + self.set_relays(&relays); + } + Ok(()) + } + + /// Whether every non-local recipient goes through a relay (RL-24). With + /// none of them linked, mail waits in this server's queue (RL-20). + pub fn sending(&self) -> bool { + self.state.lock().expect("senders").send_through + } + + pub fn link_up(&self, linked: Arc) { + self.state + .lock() + .expect("senders") + .linked + .insert(linked.relay_id, linked); + } + + /// Forgets a link, unless a newer one to the same relay replaced it. + pub fn link_down(&self, linked: &Arc) { + let mut state = self.state.lock().expect("senders"); + if state + .linked + .get(&linked.relay_id) + .is_some_and(|current| Arc::ptr_eq(current, linked)) + { + state.linked.remove(&linked.relay_id); + } + } + + /// The relays to try for one message, in order: those this node is + /// linked to and that may send, each message starting with the next + /// one in turn (RL-20). + pub fn candidates(&self) -> Vec> { + let state = self.state.lock().expect("senders"); + let mut linked: Vec> = state + .linked + .values() + .filter(|l| state.may_send.contains(&l.relay_id)) + .cloned() + .collect(); + if !linked.is_empty() { + let first = self.next.fetch_add(1, Ordering::Relaxed) % linked.len(); + linked.rotate_left(first); + } + linked + } +} + +// ---- Messages handed over --------------------------------------------------- + +/// A message handed to a relay: which queued message, and which of its +/// recipients, as the envelope named them. +#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)] +#[serde(rename_all = "camelCase")] +pub struct Handed { + /// The relay's `Outbound` id. + pub id: String, + pub queue_id: u64, + pub queue_name: String, + pub recipients: Vec, + pub at: u64, +} + +fn handed_class(relay_id: u32, outbound_id: &[u8]) -> ValueClass { + let mut key = Vec::with_capacity(6 + outbound_id.len()); + key.push(FEATURE); + key.push(KIND_HANDED); + key.extend_from_slice(&relay_id.to_be_bytes()); + key.extend_from_slice(outbound_id); + ValueClass::Any(AnyClass { + subspace: SUBSPACE_INBUXA, + key, + }) +} + +pub async fn record_handed(data: &Store, relay_id: u32, handed: &Handed) -> trc::Result<()> { + if !is_inbound_id(&handed.id) { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("A relay's outbound id isn't one") + .ctx(trc::Key::Id, handed.id.clone())); + } + let mut batch = BatchBuilder::new(); + batch.set( + handed_class(relay_id, handed.id.as_bytes()), + Json(handed).serialize()?, + ); + data.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + Ok(()) +} + +/// The messages handed to a relay with no final status yet. +pub async fn handed(data: &Store, relay_id: u32) -> trc::Result> { + let mut handed = Vec::new(); + data.iterate( + IterateParams::new( + ValueKey::from(handed_class(relay_id, &[])), + ValueKey::from(handed_class(relay_id, &[u8::MAX; INBOUND_ID_MAX + 1])), + ), + |_, value| { + if let Ok(Json(one)) = Json::::deserialize(value) { + handed.push(one); + } + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + Ok(handed) +} + +pub async fn forget_handed(data: &Store, relay_id: u32, ids: &[String]) -> trc::Result<()> { + for chunk in ids.chunks(500) { + let mut batch = BatchBuilder::new(); + for id in chunk { + batch.clear(handed_class(relay_id, id.as_bytes())); + } + data.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + struct Nothing; + + impl Sender for Nothing { + fn hand_over(&self, _: Vec, _: Envelope) -> HandOver<'_> { + Box::pin(async { Err(Refused::Temporary("no".into())) }) + } + } + + fn relay(id: u32, send_through: bool, inbound_only: bool) -> Relay { + let mut relay: Relay = serde_json::from_value(serde_json::json!({ + "name": format!("r{id}"), + "address": "relay.example.com", + })) + .unwrap(); + relay.id = id; + relay.send_through = send_through; + relay.inbound_only = inbound_only; + relay + } + + fn linked(relay_id: u32) -> Arc { + Arc::new(Linked { + relay_id, + name: format!("r{relay_id}"), + sender: Box::new(Nothing), + }) + } + + #[test] + fn rotation_skips_inbound_only() { + let senders = Senders::default(); + assert!(!senders.known() && !senders.sending()); + // Off until a relay has sendThrough on + senders.set_relays(&[relay(1, false, false), relay(2, false, true)]); + assert!(senders.known() && !senders.sending()); + senders.set_relays(&[ + relay(1, true, false), + relay(2, false, true), + relay(3, false, false), + ]); + assert!(senders.sending()); + // Nothing linked: nothing to try, and the mail waits + assert!(senders.candidates().is_empty()); + for id in [1, 2, 3] { + senders.link_up(linked(id)); + } + let mut firsts = Vec::new(); + for _ in 0..4 { + let ids: Vec = senders.candidates().iter().map(|l| l.relay_id).collect(); + assert_eq!(ids.len(), 2, "{ids:?}"); + assert!(!ids.contains(&2), "inboundOnly: {ids:?}"); + firsts.push(ids[0]); + } + assert_eq!(firsts, [1, 3, 1, 3]); + } + + #[test] + fn an_old_link_leaves_a_new_one() { + let senders = Senders::default(); + senders.set_relays(&[relay(1, true, false)]); + let old = linked(1); + let new = linked(1); + senders.link_up(old.clone()); + senders.link_up(new.clone()); + senders.link_down(&old); + assert_eq!(senders.candidates().len(), 1); + senders.link_down(&new); + assert!(senders.candidates().is_empty()); + } +} diff --git a/crates/relay-proto/src/types.rs b/crates/relay-proto/src/types.rs index 05fa90ef..1bba80e4 100644 --- a/crates/relay-proto/src/types.rs +++ b/crates/relay-proto/src/types.rs @@ -401,6 +401,9 @@ pub struct DeliveryStatus { pub delivered: Delivered, /// The relay never learns this. pub displayed: Displayed, + /// The MX host the reply came from, when one answered. Not in RFC 8621. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mx: Option, } impl DeliveryStatus { @@ -409,6 +412,7 @@ impl DeliveryStatus { smtp_reply: String::new(), delivered: Delivered::Queued, displayed: Displayed::Unknown, + mx: None, } } } diff --git a/crates/relay/Cargo.toml b/crates/relay/Cargo.toml index 11f6d0b3..3e1c4265 100644 --- a/crates/relay/Cargo.toml +++ b/crates/relay/Cargo.toml @@ -32,6 +32,11 @@ futures-util = "0.3.34" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +[features] +# For the server's system tests only: `INBUXA_RELAY_TEST_MX=` sends +# every domain's mail to 127.0.0.1 on that port. Never in a release. +test-mx = [] + [lints] workspace = true diff --git a/crates/relay/src/deliver/mod.rs b/crates/relay/src/deliver/mod.rs index 63311e1c..d73b323e 100644 --- a/crates/relay/src/deliver/mod.rs +++ b/crates/relay/src/deliver/mod.rs @@ -146,11 +146,12 @@ struct Target<'a, R> { rcpts: &'a [&'a Address], } -fn status(delivered: Delivered, smtp_reply: String) -> DeliveryStatus { +fn status(delivered: Delivered, smtp_reply: String, mx: Option<&str>) -> DeliveryStatus { DeliveryStatus { smtp_reply, delivered, displayed: Displayed::Unknown, + mx: mx.map(str::to_string), } } @@ -158,7 +159,7 @@ 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()))) + .map(|r| (r.email.clone(), status(delivered, reply.clone(), None))) .collect() } @@ -288,7 +289,7 @@ impl Target<'_, R> { RcptOutcome::Deferred(r) => (Delivered::Queued, r, 2), }; counts[n] += 1; - (rcpt.email.clone(), status(delivered, reply)) + (rcpt.email.clone(), status(delivered, reply, Some(host))) }) .collect(); self.app.record( diff --git a/crates/relay/src/deliver/route.rs b/crates/relay/src/deliver/route.rs index f83e3c75..ee3effc6 100644 --- a/crates/relay/src/deliver/route.rs +++ b/crates/relay/src/deliver/route.rs @@ -301,3 +301,36 @@ impl Resolve for Dns { } } } + +/// The server's system tests: every domain is its own MX, at 127.0.0.1, +/// unsigned and without MTA-STS. +#[cfg(feature = "test-mx")] +pub struct TestMx { + pub port: u16, +} + +#[cfg(feature = "test-mx")] +impl Resolve for TestMx { + async fn mx(&self, domain: &str) -> Result { + Ok(Mx::Hosts { + hosts: vec![domain.to_string()], + secure: false, + }) + } + + async fn addresses(&self, _: &str, _: bool) -> Result<(Vec, bool), LookupError> { + Ok((vec![IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)], false)) + } + + async fn tlsa(&self, _: &str) -> Result { + Ok(TlsaResult::Missing) + } + + async fn mta_sts(&self, _: &str) -> Option> { + None + } + + fn port(&self) -> u16 { + self.port + } +} diff --git a/crates/relay/src/main.rs b/crates/relay/src/main.rs index f3673fcb..4c239175 100644 --- a/crates/relay/src/main.rs +++ b/crates/relay/src/main.rs @@ -191,6 +191,15 @@ fn run(paths: &Paths, link_listen: SocketAddr) -> Result<(), String> { tokio::spawn(sweeper(app.clone())); let outbound = app.clone(); tokio::spawn(async move { + #[cfg(feature = "test-mx")] + if let Some(port) = std::env::var("INBUXA_RELAY_TEST_MX") + .ok() + .and_then(|port| port.parse().ok()) + { + eprintln!("inbuxa-relay: a test build: all mail goes to 127.0.0.1:{port}"); + let test = Arc::new(deliver::route::TestMx { port }); + return deliver::run(outbound, test).await; + } let dns = Arc::new(deliver::route::Dns::new().await); deliver::run(outbound, dns).await }); diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index 5d545b8e..68fc4c9f 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/crates/services/src/inbuxa_relay_link.rs @@ -45,6 +45,7 @@ use inbuxa_features::{ RecipientPlan, forget_ingested, is_inbound_id, issue_node_certificate, pushed_settings, record_ingested, record_join, set_status, was_ingested, }, + outbound::{self, HandOver, Handed, Linked, Refused, Sender}, }, }; use inbuxa_relay_proto::{ @@ -54,8 +55,9 @@ use inbuxa_relay_proto::{ recipient::{RecipientDomain, RecipientHash}, seal, token, types::{ - Address, Capability, Inbound, InboundState, RELAY_SINGLETON, RelayEvent as ProtoRelayEvent, - RelaySettings, RelayStatus, Reply, Verdict, + Address, Capability, Delivered, DeliveryStatus, Envelope, Inbound, InboundState, + Outbound as ProtoOutbound, RELAY_SINGLETON, RelayEvent as ProtoRelayEvent, RelaySettings, + RelayStatus, Reply, Verdict, }, }; use registry::schema::{ @@ -71,6 +73,7 @@ use serde_json::{Value, json}; use smtp::{ core::Relayed, inbound::relayed::{RelayedMessage, RelayedOutcome, ingest_relayed}, + outbound::relay::{Applied, RelayedStatus, apply_relayed}, }; use std::{ collections::{BTreeMap, BTreeSet, HashMap, HashSet}, @@ -177,15 +180,17 @@ async fn reconcile( links: &mut HashMap, ) -> trc::Result<()> { let data = server.store(); - let wanted: HashMap = match relays::keys(data).await? { - Some(_) => relays::all(data) - .await? - .into_iter() - .filter(|relay| relay.enabled) - .map(|relay| (relay.id, (relay.address.clone(), relay.link_port))) - .collect(), - None => HashMap::new(), + let all = match relays::keys(data).await? { + Some(_) => relays::all(data).await?, + None => Vec::new(), }; + // Which relays outbound mail goes through (RL-20, RL-24) + server.inner.data.relay_senders.set_relays(&all); + let wanted: HashMap = all + .into_iter() + .filter(|relay| relay.enabled) + .map(|relay| (relay.id, (relay.address.clone(), relay.link_port))) + .collect(); links.retain(|id, link| match wanted.get(id) { Some(target) if *target == link.target && !link.handle.is_finished() => true, // Moved: start over at the new address @@ -780,6 +785,7 @@ struct Ws { relay_changed: bool, inbound_changed: bool, events_changed: bool, + outbound_changed: bool, } impl Ws { @@ -812,6 +818,7 @@ impl Ws { relay_changed: false, inbound_changed: false, events_changed: false, + outbound_changed: false, }) } @@ -861,6 +868,7 @@ impl Ws { self.relay_changed |= types.contains_key("Relay"); self.inbound_changed |= types.contains_key("Inbound"); self.events_changed |= types.contains_key("RelayEvent"); + self.outbound_changed |= types.contains_key("Outbound"); } } } @@ -957,7 +965,12 @@ async fn session( let mut ws = Ws::open(&dialer).await?; // Inbound too: live mode (RL-12) ws.send(&ClientFrame::WebSocketPushEnable { - data_types: Some(vec!["Relay".into(), "Inbound".into(), "RelayEvent".into()]), + data_types: Some(vec![ + "Relay".into(), + "Inbound".into(), + "RelayEvent".into(), + "Outbound".into(), + ]), push_state: None, }) .await?; @@ -981,6 +994,13 @@ async fn session( taker.take(&mut ws).await?; take_events(&mut ws, data, &relay, status.current.node_id).await?; + // Outbound mail can go through this link now (RL-24), until it drops + let _sending = Sending::up(server, &relay, dialer.clone()); + let mut follower = Follower::default(); + follower + .follow(&mut ws, server, &relay, status.current.node_id) + .await?; + let mut tick = tokio::time::interval(SYNC); tick.reset(); let mut ping = tokio::time::interval(PING_EVERY); @@ -1001,6 +1021,9 @@ async fn session( if ws.events_changed { take_events(&mut ws, data, &relay, status.current.node_id).await?; } + if ws.outbound_changed { + follower.follow(&mut ws, server, &relay, status.current.node_id).await?; + } continue; } Some(answer) = answers.recv() => { @@ -1036,6 +1059,9 @@ async fn session( taker.relay_name = relay.name.clone(); taker.take(&mut ws).await?; take_events(&mut ws, data, &relay, status.current.node_id).await?; + follower + .follow(&mut ws, server, &relay, status.current.node_id) + .await?; } } @@ -1050,16 +1076,7 @@ async fn take_events( node_id: u64, ) -> Result<(), LinkError> { ws.events_changed = false; - let first = relays::link::statuses(data, relay.id) - .await? - .iter() - .filter(|link| { - link.state == LinkState::Linked && link.updated_at + 3 * STATUS_EVERY >= now() - }) - .map(|link| link.node_id) - .min() - .is_none_or(|first| first >= node_id); - if !first { + if !first_linked(data, relay.id, node_id).await? { return Ok(()); } let answer = ws.ok(method::RELAY_EVENT_GET, json!({})).await?; @@ -1077,6 +1094,20 @@ async fn take_events( Ok(()) } +/// Whether this node is the lowest linked to the relay, which takes what +/// only one node should: events, and outbound statuses. +async fn first_linked(data: &store::Store, relay_id: u32, node_id: u64) -> trc::Result { + Ok(relays::link::statuses(data, relay_id) + .await? + .iter() + .filter(|link| { + link.state == LinkState::Linked && link.updated_at + 3 * STATUS_EVERY >= now() + }) + .map(|link| link.node_id) + .min() + .is_none_or(|first| first >= node_id)) +} + /// One relay event, as a `relay.*` trace event naming the relay. fn trace_event(relay: &Relay, event: &ProtoRelayEvent) { let id = relay.relay_id(); @@ -1140,6 +1171,230 @@ fn trace_event(relay: &Relay, event: &ProtoRelayEvent) { } } +// ---- Outbound mail ---------------------------------------------------------- + +/// Hands messages to the relay over this node's link (RL-24): the message +/// as an upload, then `Outbound/set` create. +struct LinkSender { + dialer: Arc, +} + +impl Sender for LinkSender { + fn hand_over(&self, message: Vec, envelope: Envelope) -> HandOver<'_> { + Box::pin(async move { + let blob_id = self + .dialer + .upload(message) + .await + .map_err(|err| Refused::Temporary(err.message))?; + let answer = self + .dialer + .api(vec![Invocation::new( + method::OUTBOUND_SET, + json!({ + "accountId": ACCOUNT_ID, + "create": {"m": {"blobId": blob_id, "envelope": envelope}}, + }), + "c", + )]) + .await + .map_err(|err| Refused::Temporary(err.message))? + .into_iter() + .next() + .ok_or_else(|| Refused::Temporary("The relay didn't answer.".into()))?; + let args = answer.args(); + if let Some(id) = args["created"]["m"]["id"].as_str() { + return Ok(id.to_string()); + } + Err(refusal(&args["notCreated"]["m"])) + }) + } +} + +/// A relay's `notCreated` for a message, as this server's refusal. +fn refusal(error: &Value) -> Refused { + let description = error["description"].as_str().unwrap_or("no reason given"); + match error["type"].as_str().unwrap_or_default() { + "tooLarge" => Refused::Permanent(format!("552 5.3.4 {description}")), + "forbiddenFrom" => Refused::Permanent(format!("550 5.7.1 {description}")), + "invalidProperties" => Refused::Permanent(format!("550 5.5.2 {description}")), + kind => Refused::Temporary(format!("{kind}: {description}")), + } +} + +/// This node's link, offered for outbound mail while it is up. +struct Sending { + inner: Arc, + linked: Arc, +} + +impl Sending { + fn up(server: &Server, relay: &Relay, dialer: Arc) -> Self { + let linked = Arc::new(Linked { + relay_id: relay.id, + name: relay.name.clone(), + sender: Box::new(LinkSender { dialer }), + }); + server.inner.data.relay_senders.link_up(linked.clone()); + Sending { + inner: server.inner.clone(), + linked, + } + } +} + +impl Drop for Sending { + fn drop(&mut self) { + self.inner.data.relay_senders.link_down(&self.linked); + } +} + +/// How many messages one `Outbound/get` asks for. +const FOLLOW_BATCH: usize = 200; + +/// Follows the messages handed to the relay, from any node, until each +/// has a final status (RL-24): the lowest linked node does, and records +/// the relay's replies in this server's queue, which sends the DSNs. +#[derive(Default)] +struct Follower { + /// Each message's statuses as last recorded, to skip ones unchanged. + recorded: HashMap>, +} + +impl Follower { + async fn follow( + &mut self, + ws: &mut Ws, + server: &Server, + relay: &Relay, + node_id: u64, + ) -> Result<(), LinkError> { + ws.outbound_changed = false; + let data = server.store(); + if !first_linked(data, relay.id, node_id).await? { + self.recorded.clear(); + return Ok(()); + } + let handed: HashMap = outbound::handed(data, relay.id) + .await? + .into_iter() + .map(|h| (h.id.clone(), h)) + .collect(); + self.recorded.retain(|id, _| handed.contains_key(id)); + let ids: Vec<&String> = handed.keys().collect(); + for batch in ids.chunks(FOLLOW_BATCH) { + let answer = ws.ok(method::OUTBOUND_GET, json!({"ids": batch})).await?; + let mut done = Vec::new(); + for found in answer["list"].as_array().into_iter().flatten() { + let Ok(found) = serde_json::from_value::(found.clone()) else { + continue; + }; + let Some(handed) = handed.get(&found.id) else { + continue; + }; + if self.recorded.get(&found.id) == Some(&found.delivery_status) { + continue; + } + let statuses = relayed_statuses(handed, &found); + let finished = statuses.iter().all(|(_, s)| { + matches!( + s, + RelayedStatus::Delivered { .. } | RelayedStatus::Refused { .. } + ) + }); + match apply_relayed( + server, + handed.queue_id, + &handed.queue_name, + &relay.name, + &statuses, + ) + .await + { + Applied::Done if finished => done.push(found.id.clone()), + Applied::Done => { + self.recorded + .insert(found.id.clone(), found.delivery_status.clone()); + } + // Cancelled here: the relay stops too + Applied::Gone => done.push(found.id.clone()), + Applied::Locked => {} + } + } + // Gone from the relay (a reset, or kept past seven days): the + // queue hands the recipients over again + for id in answer["notFound"].as_array().into_iter().flatten() { + let Some(handed) = id.as_str().and_then(|id| handed.get(id)) else { + continue; + }; + let lost: Vec<(String, RelayedStatus)> = handed + .recipients + .iter() + .map(|r| (r.clone(), RelayedStatus::Lost)) + .collect(); + match apply_relayed( + server, + handed.queue_id, + &handed.queue_name, + &relay.name, + &lost, + ) + .await + { + Applied::Done | Applied::Gone => { + outbound::forget_handed(data, relay.id, std::slice::from_ref(&handed.id)) + .await?; + } + Applied::Locked => {} + } + } + if !done.is_empty() { + let answer = ws + .ok(method::OUTBOUND_SET, json!({"destroy": done})) + .await?; + // One being tried right now is cleared next time + let busy: HashSet<&str> = answer["notDestroyed"] + .as_object() + .into_iter() + .flatten() + .filter(|(_, e)| e["type"] == "forbidden") + .map(|(id, _)| id.as_str()) + .collect(); + let forget: Vec = done + .into_iter() + .filter(|id| !busy.contains(id.as_str())) + .collect(); + for id in &forget { + self.recorded.remove(id); + } + outbound::forget_handed(data, relay.id, &forget).await?; + } + } + Ok(()) + } +} + +/// What the relay says of each recipient a message was handed over for. +fn relayed_statuses(handed: &Handed, found: &ProtoOutbound) -> Vec<(String, RelayedStatus)> { + handed + .recipients + .iter() + .filter_map(|address| { + let status = found.delivery_status.get(address)?; + let reply = status.smtp_reply.clone(); + let mx = status.mx.clone(); + Some(( + address.clone(), + match status.delivered { + Delivered::Yes | Delivered::Unknown => RelayedStatus::Delivered { reply, mx }, + Delivered::No => RelayedStatus::Refused { reply, mx }, + Delivered::Queued => RelayedStatus::Queued { reply, mx }, + }, + )) + }) + .collect() +} + // ---- Inbound mail ----------------------------------------------------------- /// Takes the relay's messages for this node: a few at a time, each judged in diff --git a/crates/smtp/Cargo.toml b/crates/smtp/Cargo.toml index 728df283..97abfe78 100644 --- a/crates/smtp/Cargo.toml +++ b/crates/smtp/Cargo.toml @@ -19,6 +19,8 @@ inbuxa-outbound-tls = { path = "../outbound-tls" } email = { path = "../email" } registry = { path = "../registry" } inbuxa-features = { path = "../features" } +# inbuxa: outbound through inbuxa Relay (relay spec, RL-24) +inbuxa-relay-proto = { path = "../relay-proto" } spam-filter = { path = "../spam-filter" } trc = { path = "../trc" } mail-auth = { version = "0.13", features = ["rkyv"] } diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index d70e3ceb..197e5968 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -15,13 +15,14 @@ use crate::outbound::dane::dnssec::{DnssecStatus, TlsaLookup, TlsaResult}; use crate::outbound::error::ClientError; use crate::outbound::lookup::{DnsLookup, SourceIp}; use crate::outbound::mta_sts::lookup::MtaStsLookup; +use crate::outbound::relay::RELAYED_RECHECK; use crate::outbound::{client::StartTlsResult, dane::verify::TlsaVerify}; use crate::queue::dsn::{DsnStatus, SendDsn}; use crate::queue::spool::{DSN_RETRY, SmtpSpool}; use crate::queue::throttle::IsAllowed; use crate::queue::{ Error, FROM_REPORT, HostResponse, MessageWrapper, Metadata, QueueEnvelope, QueuedMessage, - Status, + RCPT_RELAYED, Status, }; use crate::reporting::send::MtaReportSend; use crate::{queue::ErrorDetails, reporting::tls::TlsRptOptions}; @@ -237,6 +238,16 @@ impl QueuedMessage { } } } + // inbuxa: with a relay's sendThrough on, every non-local recipient + // goes through a relay (relay spec, RL-24). REQUIRETLS mail doesn't + // yet: the relay can't keep that promise for it. + static THROUGH_RELAY: RoutingStrategy = RoutingStrategy::InbuxaRelay; + let senders = &server.inner.data.relay_senders; + if let Err(err) = senders.load(server.store()).await { + trc::error!(err.details("Failed to read the relays").span_id(span_id)); + } + let through_relay = senders.sending() && message.message.flags & MAIL_REQUIRETLS == 0; + let mut relayed_waiting = Vec::new(); for (rcpt_idx, rcpt) in message.message.recipients.iter().enumerate() { if matches!( &rcpt.status, @@ -244,6 +255,11 @@ impl QueuedMessage { ) && rcpt.retry.due <= now_ && rcpt.queue == message.queue_name { + // inbuxa: waiting on a relay's status, not handed over again + if rcpt.flags & RCPT_RELAYED != 0 { + relayed_waiting.push(rcpt_idx); + continue; + } let envelope = QueueEnvelope::new(&message.message, rcpt); let route = server.get_route_or_default( &server @@ -252,6 +268,11 @@ impl QueuedMessage { .unwrap_or_else(|| "default".to_string()), message.span_id, ); + let route = if through_relay && !matches!(route, RoutingStrategy::Local) { + &THROUGH_RELAY + } else { + route + }; // Map RCPT headers let mut rcpt_headers = default_rcpt_header; @@ -313,6 +334,13 @@ impl QueuedMessage { .await; continue 'next_route; } + // inbuxa: handed to an inbuxa Relay (relay spec, RL-24) + RoutingStrategy::InbuxaRelay => { + message + .hand_to_relay(&rcpt_idxs, rcpt_headers, &mut delivery_results, &server) + .await; + continue 'next_route; + } RoutingStrategy::Mx(mx_config) => (Vec::with_capacity(0), Some(mx_config), true), RoutingStrategy::Relay(relay_config) => ( vec![NextHop::Relay(relay_config)], @@ -1520,8 +1548,22 @@ impl QueuedMessage { message.set_rcpt_rate_limit(rcpt_idx, retry_at); } } + // inbuxa: kept here until the relay reports a final status + DeliveryResult::Relayed { relay, rcpt_idx } => { + let rcpt = &mut message.message.recipients[rcpt_idx]; + rcpt.flags |= RCPT_RELAYED; + rcpt.retry.due = now() + RELAYED_RECHECK; + rcpt.status = Status::TemporaryFailure(ErrorDetails { + entity: relay.clone(), + details: Error::Io(format!("Handed to relay {relay}").into()), + }); + } } } + // inbuxa: still waiting on a relay: looked at again later + for rcpt_idx in relayed_waiting { + message.message.recipients[rcpt_idx].retry.due = now() + RELAYED_RECHECK; + } // Send Delivery Status Notifications let dsn_status = server.send_dsn(&mut message).await; diff --git a/crates/smtp/src/outbound/mod.rs b/crates/smtp/src/outbound/mod.rs index 1cff2b1e..bdfcee7b 100644 --- a/crates/smtp/src/outbound/mod.rs +++ b/crates/smtp/src/outbound/mod.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{ @@ -24,6 +26,8 @@ pub mod error; pub mod local; pub mod lookup; pub mod mta_sts; +// inbuxa: outbound through inbuxa Relay (relay spec, RL-24) +pub mod relay; pub mod session; pub(super) enum DeliveryResult { @@ -39,6 +43,11 @@ pub(super) enum DeliveryResult { rcpt_idxs: Vec, retry_at: u64, }, + // inbuxa: handed to the relay named, which reports what became of it + Relayed { + relay: Box, + rcpt_idx: usize, + }, } impl Status>, ErrorDetails> { diff --git a/crates/smtp/src/outbound/relay.rs b/crates/smtp/src/outbound/relay.rs new file mode 100644 index 00000000..e93221bd --- /dev/null +++ b/crates/smtp/src/outbound/relay.rs @@ -0,0 +1,378 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Outbound mail through an inbuxa Relay (relay spec, "Outbound routing", +//! RL-20, RL-24). +//! +//! With a relay's `sendThrough` on, every non-local recipient is handed to +//! one of the relays this node is linked to, in turn. The recipients stay +//! in this server's queue, marked [`RCPT_RELAYED`] and shown as handed to +//! the relay, until the link service reports the relay's final status for +//! them through [`apply_relayed`]. The usual DSNs follow from there. With +//! no relay linked, they wait in the queue and are retried as usual. + +use super::DeliveryResult; +use crate::queue::{ + Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_RELAYED, Status, UnexpectedResponse, + spool::SmtpSpool, +}; +use common::{Server, config::smtp::queue::QueueName, ipc::QueueEvent}; +use inbuxa_features::relay::outbound::{Handed, Refused, record_handed}; +use inbuxa_relay_proto::types::{Address, Envelope}; +use smtp_proto::{ + MAIL_BODY_8BITMIME, MAIL_RET_FULL, MAIL_RET_HDRS, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, + RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, Response, +}; +use std::collections::BTreeMap; +use store::write::now; +use trc::RelayEvent; +use types::id::Id; + +/// How long a recipient handed to a relay waits before the queue looks at +/// it again. The relay's status normally comes first. +pub const RELAYED_RECHECK: u64 = 3_600; + +impl MessageWrapper { + /// Hands one domain's recipients to the first relay that takes them. + pub(super) async fn hand_to_relay( + &self, + rcpt_idxs: &[usize], + rcpt_headers: Option<&[u8]>, + results: &mut Vec, + server: &Server, + ) { + let candidates = server.inner.data.relay_senders.candidates(); + if candidates.is_empty() { + let status = Status::TemporaryFailure(ErrorDetails { + entity: "relay".into(), + details: Error::ConnectionError("No relay is linked from this node.".into()), + }); + results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec())); + return; + } + let blob = match server + .get_message_blob(self.message.blob_hash.as_slice(), 0..usize::MAX) + .await + { + Ok(Some(blob)) => blob, + Ok(None) => { + let status = Status::PermanentFailure(ErrorDetails { + entity: "localhost".into(), + details: Error::Io("The message is gone from the blob store.".into()), + }); + results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec())); + return; + } + Err(err) => { + trc::error!( + err.details("Failed to read a message to hand to a relay") + .span_id(self.span_id) + ); + let status = Status::TemporaryFailure(ErrorDetails { + entity: "localhost".into(), + details: Error::Io("The message couldn't be read.".into()), + }); + results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec())); + return; + } + }; + let mut message = Vec::with_capacity(blob.len() + rcpt_headers.map_or(0, |h| h.len())); + if let Some(headers) = rcpt_headers { + message.extend_from_slice(headers); + } + message.extend_from_slice(&blob); + drop(blob); + let envelope = self.relay_envelope(rcpt_idxs); + let recipients: Vec = envelope.rcpt_to.iter().map(|r| r.email.clone()).collect(); + + let mut last = String::new(); + for relay in candidates { + match relay + .sender + .hand_over(message.clone(), envelope.clone()) + .await + { + Ok(outbound_id) => { + let handed = Handed { + id: outbound_id.clone(), + queue_id: self.queue_id, + queue_name: self.queue_name.as_str().to_string(), + recipients: recipients.clone(), + at: now(), + }; + if let Err(err) = record_handed(server.store(), relay.relay_id, &handed).await { + // The relay sends it anyway; only its status is lost + trc::error!( + err.details("Failed to record a message handed to a relay") + .span_id(self.span_id) + ); + } + trc::event!( + Relay(RelayEvent::Event), + SpanId = self.span_id, + Id = Id::from(relay.relay_id).to_string(), + Type = "outbound.handed-over", + Details = outbound_id, + To = recipients + .iter() + .map(|r| trc::Value::String(r.as_str().into())) + .collect::>(), + ); + for &rcpt_idx in rcpt_idxs { + results.push(DeliveryResult::Relayed { + relay: relay.name.as_str().into(), + rcpt_idx, + }); + } + return; + } + Err(Refused::Permanent(reply)) => { + let status = Status::PermanentFailure(ErrorDetails { + entity: relay.name.as_str().into(), + details: Error::UnexpectedResponse(UnexpectedResponse { + command: "".into(), + response: parse_reply(&reply), + }), + }); + results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec())); + return; + } + Err(Refused::Temporary(reason)) => { + last = format!("Relay {}: {reason}", relay.name); + } + } + } + let status = Status::TemporaryFailure(ErrorDetails { + entity: "relay".into(), + details: Error::ConnectionError(last.into()), + }); + results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec())); + } + + /// The envelope as the relay takes it: the DSN parameters this server + /// would send itself, and 8BITMIME (RFC 8621 `Address` parameters). + fn relay_envelope(&self, rcpt_idxs: &[usize]) -> Envelope { + let flags = self.message.flags; + let mut from = BTreeMap::new(); + if flags & MAIL_BODY_8BITMIME != 0 { + from.insert("BODY".to_string(), Some("8BITMIME".to_string())); + } + if flags & MAIL_RET_FULL != 0 { + from.insert("RET".to_string(), Some("FULL".to_string())); + } else if flags & MAIL_RET_HDRS != 0 { + from.insert("RET".to_string(), Some("HDRS".to_string())); + } + if let Some(env_id) = &self.message.env_id { + from.insert("ENVID".to_string(), Some(env_id.to_string())); + } + let rcpt_to = rcpt_idxs + .iter() + .map(|&idx| { + let rcpt = &self.message.recipients[idx]; + let mut notify = Vec::new(); + for (flag, name) in [ + (RCPT_NOTIFY_SUCCESS, "SUCCESS"), + (RCPT_NOTIFY_DELAY, "DELAY"), + (RCPT_NOTIFY_FAILURE, "FAILURE"), + ] { + if rcpt.flags & flag != 0 { + notify.push(name); + } + } + if notify.is_empty() && rcpt.flags & RCPT_NOTIFY_NEVER != 0 { + notify.push("NEVER"); + } + Address { + email: rcpt.address().to_string(), + parameters: (!notify.is_empty()) + .then(|| [("NOTIFY".to_string(), Some(notify.join(",")))].into()), + } + }) + .collect(); + Envelope { + mail_from: Address { + email: self.message.return_path.to_string(), + parameters: (!from.is_empty()).then_some(from), + }, + rcpt_to, + } + } +} + +/// `550 5.1.1 text` as a reply. A reply without an enhanced code gets its +/// class's. +pub fn parse_reply(reply: &str) -> Response> { + let reply = reply.trim(); + let (code, rest) = reply.split_once(' ').unwrap_or((reply, "")); + let code = code + .parse::() + .ok() + .filter(|c| (200..600).contains(c)) + .unwrap_or(451); + let class = (code / 100) as u8; + let mut esc = [class, 0, 0]; + let mut message = rest; + if let Some((first, after)) = rest.split_once(' ').or(Some((rest, ""))) + && let [a, b, c] = first.split('.').collect::>()[..] + && let (Ok(a), Ok(b), Ok(c)) = (a.parse::(), b.parse::(), c.parse::()) + && a == class + { + esc = [a, b, c]; + message = after; + } + Response { + code, + esc, + message: message.into(), + } +} + +/// What a relay says became of one recipient handed to it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RelayedStatus { + Delivered { + reply: String, + mx: Option, + }, + Refused { + reply: String, + mx: Option, + }, + /// Still queued at the relay, with its last reply, if any. + Queued { + reply: String, + mx: Option, + }, + /// The relay no longer has the message: hand it over again. + Lost, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Applied { + Done, + /// The queue is working on the message: try again later. + Locked, + /// The message is no longer queued here. + Gone, +} + +/// Records a relay's statuses for recipients it was handed, by address, +/// and wakes the queue to send any DSN they call for. +pub async fn apply_relayed( + server: &Server, + queue_id: u64, + queue_name: &str, + relay: &str, + statuses: &[(String, RelayedStatus)], +) -> Applied { + let Some(queue_name) = QueueName::new(queue_name) else { + return Applied::Gone; + }; + if !server.try_lock_event(queue_id, queue_name).await { + return Applied::Locked; + } + let Some(mut message) = server.read_message(queue_id, queue_name).await else { + server.unlock_event(queue_id, queue_name).await; + return Applied::Gone; + }; + message.span_id = server.inner.data.span_id_gen.generate(); + let now = now(); + let mut changed = false; + let mut wake = false; + for rcpt in message.message.recipients.iter_mut() { + // Expired here meanwhile: this server's word stands + if rcpt.flags & RCPT_RELAYED == 0 + || matches!( + rcpt.status, + Status::Completed(_) | Status::PermanentFailure(_) + ) + { + continue; + } + let Some((_, status)) = statuses + .iter() + .find(|(address, _)| address.eq_ignore_ascii_case(rcpt.address())) + else { + continue; + }; + let entity = |mx: &Option| -> Box { mx.as_deref().unwrap_or(relay).into() }; + let new = match status { + RelayedStatus::Delivered { reply, mx } => Status::Completed(HostResponse { + hostname: entity(mx), + response: parse_reply(reply), + }), + RelayedStatus::Refused { reply, mx } => Status::PermanentFailure(ErrorDetails { + entity: entity(mx), + details: Error::UnexpectedResponse(UnexpectedResponse { + command: "".into(), + response: parse_reply(reply), + }), + }), + RelayedStatus::Queued { reply, mx } if !reply.is_empty() => { + Status::TemporaryFailure(ErrorDetails { + entity: entity(mx), + details: Error::UnexpectedResponse(UnexpectedResponse { + command: "".into(), + response: parse_reply(reply), + }), + }) + } + RelayedStatus::Queued { .. } => continue, + RelayedStatus::Lost => Status::TemporaryFailure(ErrorDetails { + entity: relay.into(), + details: Error::Io("The relay no longer has the message.".into()), + }), + }; + if matches!( + status, + RelayedStatus::Delivered { .. } | RelayedStatus::Refused { .. } | RelayedStatus::Lost + ) { + rcpt.flags &= !RCPT_RELAYED; + rcpt.retry.due = now; + wake = true; + } + if rcpt.status != new { + rcpt.status = new; + changed = true; + } + } + if changed || wake { + message + .save_changes(server, None, wake.then_some(now)) + .await; + } + server.unlock_event(queue_id, queue_name).await; + if wake { + let _ = server.inner.ipc.queue_tx.send(QueueEvent::Refresh).await; + } + Applied::Done +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn replies() { + let r = parse_reply("550 5.1.1 No such user here"); + assert_eq!( + (r.code, r.esc, r.message.as_ref()), + (550, [5, 1, 1], "No such user here") + ); + let r = parse_reply("250 Ok"); + assert_eq!((r.code, r.esc, r.message.as_ref()), (250, [2, 0, 0], "Ok")); + // An enhanced code of another class isn't one + let r = parse_reply("451 5.1.1 odd"); + assert_eq!( + (r.code, r.esc, r.message.as_ref()), + (451, [4, 0, 0], "5.1.1 odd") + ); + let r = parse_reply(""); + assert_eq!((r.code, r.esc), (451, [4, 0, 0])); + let r = parse_reply("552 5.3.4"); + assert_eq!((r.code, r.esc, r.message.as_ref()), (552, [5, 3, 4], "")); + } +} diff --git a/crates/smtp/src/queue/mod.rs b/crates/smtp/src/queue/mod.rs index 4a89b14c..be0e5435 100644 --- a/crates/smtp/src/queue/mod.rs +++ b/crates/smtp/src/queue/mod.rs @@ -158,6 +158,9 @@ pub const FROM_REPORT: u64 = 1 << 36; pub const FROM_AUTOGENERATED: u64 = 1 << 37; pub const RCPT_DSN_SENT: u64 = 1 << 32; +// inbuxa: handed to an inbuxa Relay, which hasn't reported a final status +// (relay spec, RL-24) +pub const RCPT_RELAYED: u64 = 1 << 40; pub const RCPT_SPAM_SHIFT: u64 = 56; pub const RCPT_SPAM_MASK: u64 = 0xff << RCPT_SPAM_SHIFT; diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index eb6fe1fc..b8343c9b 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -307,7 +307,7 @@ so a reboot without the link comes back with the same settings (RL-22). | `blobId` | The message as the downstream signed it | | `envelope` | RFC 8621 `Envelope` | | `size` | Bytes | -| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply` (empty before the first try), `delivered` (`queued`, `yes`, `no`, `unknown`) | +| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply` (empty before the first try), `delivered` (`queued`, `yes`, `no`, `unknown`), and `mx`, the host the reply came from when one answered (not in RFC 8621) | | `nextAttemptAt`, `attempts`, `expiresAt` | Retry state. `nextAttemptAt` only while a recipient is queued | A create takes `blobId` and `envelope` only; the relay sets the rest, and no @@ -552,9 +552,25 @@ allowing the edge's IP and disabling SPF, DMARC and reverse IP for it. - **RL-24.** A new route type, `inbuxaRelay`, alongside the existing route types, names one or more `inbuxa:Relay`s. With `sendThrough` on, it is the route for every non-local recipient. +- `sendThrough` on any enabled relay sends every non-local recipient + through relays: the enabled ones not marked `inboundOnly`, that the node + delivering is linked to, each message starting with the next in turn. + A relay that refuses a message for now (its spool full, its link + failing) passes it to the next; one that refuses it for good + (`tooLarge`, `forbiddenFrom`) fails those recipients. With no relay + linked, the recipients wait in the queue and are retried as usual. - The downstream's queue keeps the message until a relay reports a final status. The item shows "handed to relay ‹name›", and then the remote - server's reply. + server's reply. A handed-over recipient isn't handed over again while it + waits, except when the relay no longer has the message (a reset). A + message cancelled from the downstream's queue is destroyed on the relay + too, unless an attempt is under way, and then at the next status. +- The relay's statuses are followed by one node, the lowest linked to that + relay, for every message handed to it from any node. A final status is + recorded in the downstream's queue, which sends any DSN as for its own + deliveries, and the message is then destroyed on the relay. +- Mail sent with REQUIRETLS (RFC 8689) isn't handed to a relay yet: it + takes the route it would without relays. ### DNS and certificates @@ -801,8 +817,18 @@ In progress. Done so far: 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. + (glibc, measured 2026-10-30). +- Outbound through relays on the downstream: `sendThrough` and + `inboundOnly` decide the route (RL-20, RL-24); each message is uploaded + and created with `Outbound/set` over the delivering node's link, its + recipients are marked in the queue as handed to the relay, and the link + service follows `Outbound` with push, records each reply, and destroys + the relay's copy once every recipient is final. Tested end to end with + the relay sending to a scripted MX (RL-T11): a test build of the relay, + with the `test-mx` feature, sends every domain to one local port. + `outboundTls` is set like any other setting, `settings/outboundTls`. +- Still to come in step 4: the named `inbuxaRelay` route, TLS-RPT reports + from the relay's `outbound.tls` events, and REQUIRETLS through relays. - `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 diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 49e8c064..ab91ce6e 100644 --- a/tests/src/system/relay.rs +++ b/tests/src/system/relay.rs @@ -33,6 +33,8 @@ use std::str::FromStr; const USING: &[&str] = &[ "urn:ietf:params:jmap:core", + "urn:ietf:params:jmap:mail", + "urn:ietf:params:jmap:submission", "urn:inbuxa:jmap", "urn:inbuxa:jmap:registry", ]; @@ -542,6 +544,99 @@ async fn wait_for_mail(account: &Account, subject: &str, count: usize) { } } +/// Sends a message from `sender` to `to` by JMAP submission. +async fn submit(sender: &Account, identity: &str, mailbox: &str, to: &[&str], subject: &str) { + let (_, response) = call( + sender, + "Email/set", + json!({"create": {"e": { + "mailboxIds": {mailbox: true}, + "from": [{"email": sender.name()}], + "to": to.iter().map(|a| json!({"email": a})).collect::>(), + "subject": subject, + "bodyValues": {"b": {"value": "Hello from the far side."}}, + "textBody": [{"partId": "b", "type": "text/plain"}] + }}}), + ) + .await; + let email = response["created"]["e"]["id"] + .as_str() + .unwrap_or_else(|| panic!("draft: {response}")) + .to_string(); + let (_, response) = call( + sender, + "EmailSubmission/set", + json!({"create": {"s": {"emailId": email, "identityId": identity}}}), + ) + .await; + assert!(response["created"].get("s").is_some(), "{response}"); +} + +/// The destination the relay sends to (`INBUXA_RELAY_TEST_MX`): takes +/// `bob`, refuses `nobody`, defers `later`. The messages it took. +async fn scripted_mx() -> (u16, std::sync::Arc>>) { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let taken = seen.clone(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let taken = taken.clone(); + tokio::spawn(async move { + let (read, mut write) = stream.into_split(); + let mut read = BufReader::new(read); + let _ = write.write_all(b"220 remote.test ready\r\n").await; + let mut accepted = 0; + loop { + let mut line = String::new(); + if read.read_line(&mut line).await.unwrap_or(0) == 0 { + return; + } + let upper = line.to_ascii_uppercase(); + let answer = if upper.starts_with("EHLO") { + "250-remote.test\r\n250 8BITMIME\r\n" + } else if upper.starts_with("MAIL") { + accepted = 0; + "250 2.1.0 Ok\r\n" + } else if upper.starts_with("RCPT") && upper.contains(" 0 { + let _ = write.write_all(b"354 Go ahead\r\n").await; + let mut message = String::new(); + loop { + let mut line = String::new(); + if read.read_line(&mut line).await.unwrap_or(0) == 0 { + return; + } + if line == ".\r\n" { + break; + } + message.push_str(&line); + } + taken.lock().unwrap().push(message); + "250 2.0.0 Queued as 1\r\n" + } else if upper.starts_with("DATA") { + "554 5.5.1 No valid recipients\r\n" + } else if upper.starts_with("QUIT") { + let _ = write.write_all(b"221 Bye\r\n").await; + return; + } else { + "250 Ok\r\n" + }; + let _ = write.write_all(answer.as_bytes()).await; + } + }); + } + }); + (port, seen) +} + fn message(subject: &str, extra_headers: &str) -> String { format!( "{extra_headers}From: Someone \nTo: jane@example.com\n\ @@ -556,7 +651,7 @@ pub async fn link(test: &mut TestServer) { let binary = relay_binary(); assert!( binary.exists(), - "{} is missing: run cargo build -p inbuxa-relay", + "{} is missing: run cargo build -p inbuxa-relay --features test-mx", binary.display() ); let jane = admin @@ -570,6 +665,7 @@ pub async fn link(test: &mut TestServer) { .await; let link_port = free_port(); let smtp_port = free_port(); + let (mx_port, mx_seen) = scripted_mx().await; let (_, response) = call( admin, @@ -612,6 +708,7 @@ pub async fn link(test: &mut TestServer) { "--state", ]) .arg(&state) + .env("INBUXA_RELAY_TEST_MX", mx_port.to_string()) .stdout(std::process::Stdio::null()) .stderr(std::process::Stdio::null()) .spawn() @@ -757,6 +854,109 @@ pub async fn link(test: &mut TestServer) { 2 ); + // Outbound through the relay (RL-24): handed over, sent by the relay, + // and each recipient's reply recorded here, with one DSN (RL-T11) + call( + admin, + "inbuxa:Relay/set", + json!({"update": {&id: {"sendThrough": true, "settings/maxRecipients": 10}}}), + ) + .await; + let (_, response) = call( + &jane, + "Identity/set", + json!({"create": {"i": {"name": "Jane", "email": "jane@example.com"}}}), + ) + .await; + let identity = response["created"]["i"]["id"] + .as_str() + .unwrap_or_else(|| panic!("{response}")) + .to_string(); + let (_, response) = call( + &jane, + "Mailbox/set", + json!({"create": {"m": {"name": "Outbound drafts"}}}), + ) + .await; + let drafts = response["created"]["m"]["id"].as_str().unwrap().to_string(); + // The relay has the new maxRecipients before anything is handed to it + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let replies = rcpt_replies(smtp_port, &["jane@example.com", "joe@example.com"]).await; + if replies[1].starts_with("250") { + break; + } + assert!( + std::time::Instant::now() < deadline, + "maxRecipients reached the relay: {replies:?}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + submit( + &jane, + &identity, + &drafts, + &["bob@remote.test", "nobody@remote.test"], + "Through the relay", + ) + .await; + wait_for_mail(&jane, "Failed to deliver message", 1).await; + let seen = mx_seen.lock().unwrap().clone(); + assert!( + seen.iter() + .any(|m| m.contains("Subject: Through the relay")), + "the relay sent it on: {seen:?}" + ); + // Gone from this server's queue, and from the relay's + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let (_, queued) = call(admin, "x:QueuedMessage/get", json!({"ids": null})).await; + let in_queue = queued.to_string().contains("nobody@remote.test"); + let at_relay = std::fs::read_dir(state.join("spool").join("out")) + .map(|dir| dir.count()) + .unwrap_or_default(); + if !in_queue && at_relay == 0 { + break; + } + assert!( + std::time::Instant::now() < deadline, + "queued here: {in_queue}, files at the relay: {at_relay}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + // A deferral stays queued here, with the remote server's reply + submit( + &jane, + &identity, + &drafts, + &["later@remote.test"], + "Deferred by the remote", + ) + .await; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let (_, queued) = call(admin, "x:QueuedMessage/get", json!({"ids": null})).await; + let text = queued.to_string(); + if text.contains("later@remote.test") && text.contains("Try again later") { + break; + } + assert!( + std::time::Instant::now() < deadline, + "the deferral is shown: {queued}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + assert_eq!( + subjects(&jane) + .await + .iter() + .filter(|s| s.starts_with("Failed") || s.starts_with("Warning")) + .count(), + 1, + "one DSN, for the refusal" + ); + // The relay's events were taken as this server's relay.* trace events, // and acknowledged, so the relay forgot them let events = state.join("events.jsonl"); -- 2.54.0