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");