diff --git a/crates/features/src/relay/link.rs b/crates/features/src/relay/link.rs index bcace8df..9a62e26c 100644 --- a/crates/features/src/relay/link.rs +++ b/crates/features/src/relay/link.rs @@ -9,7 +9,9 @@ //! the settings and recipient set a relay is sent, and each node's view of //! each relay's link, under `E` `s` + relay id + node id. -use super::{DownstreamKeys, FEATURE, Json, KIND_STATUS, Relay, get_versioned, replace}; +use super::{ + DownstreamKeys, FEATURE, Json, KIND_INGESTED, KIND_STATUS, Relay, get_versioned, replace, +}; use inbuxa_relay_proto::{ pin::{Pin, pins_match}, recipient::{self, RecipientDomain, RecipientHash}, @@ -319,6 +321,101 @@ pub(super) fn clear_statuses(batch: &mut BatchBuilder, relay_id: u32, nodes: &[N } } +// ---- Messages taken --------------------------------------------------------- + +/// The longest `Inbound` id kept: the relay's are 25 characters. +pub const INBOUND_ID_MAX: usize = 64; + +/// Whether `id` can be an `Inbound` id: never anything but letters and +/// digits. +pub fn is_inbound_id(id: &str) -> bool { + !id.is_empty() && id.len() <= INBOUND_ID_MAX && id.bytes().all(|b| b.is_ascii_alphanumeric()) +} + +/// A relay's message, taken at `at`. +#[derive(Debug, Clone, SerdeSerialize, SerdeDeserialize)] +struct Taken { + id: String, + at: u64, +} + +fn ingested_class(relay_id: u32, inbound_id: &[u8]) -> ValueClass { + let mut key = Vec::with_capacity(6 + inbound_id.len()); + key.push(FEATURE); + key.push(KIND_INGESTED); + key.extend_from_slice(&relay_id.to_be_bytes()); + key.extend_from_slice(inbound_id); + ValueClass::Any(AnyClass { + subspace: SUBSPACE_INBUXA, + key, + }) +} + +/// RL-16: whether a relay's message was taken already. +pub async fn was_ingested(data: &Store, relay_id: u32, inbound_id: &str) -> trc::Result { + data.get_value::>(ValueKey::from(ingested_class( + relay_id, + inbound_id.as_bytes(), + ))) + .await + .caused_by(trc::location!()) + .map(|value| value.is_some()) +} + +/// RL-16: records a relay's message as taken, before the relay is told so. +pub async fn record_ingested( + data: &Store, + relay_id: u32, + inbound_id: &str, + now: u64, +) -> trc::Result<()> { + let taken = Taken { + id: inbound_id.to_string(), + at: now, + }; + let mut batch = BatchBuilder::new(); + batch.set( + ingested_class(relay_id, inbound_id.as_bytes()), + Json(&taken).serialize()?, + ); + data.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + Ok(()) +} + +/// Forgets the messages from a relay taken before `before`: RL-16 keeps +/// them for `heldMaxAge` and a day. How many went. +pub async fn forget_ingested(data: &Store, relay_id: u32, before: u64) -> trc::Result { + let mut old = Vec::new(); + data.iterate( + IterateParams::new( + ValueKey::from(ingested_class(relay_id, &[])), + ValueKey::from(ingested_class(relay_id, &[u8::MAX; INBOUND_ID_MAX + 1])), + ), + |_, value| { + if let Ok(Json(taken)) = Json::::deserialize(value) + && taken.at < before + { + old.push(taken.id); + } + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + for chunk in old.chunks(500) { + let mut batch = BatchBuilder::new(); + for id in chunk { + batch.clear(ingested_class(relay_id, id.as_bytes())); + } + data.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + Ok(old.len()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/features/src/relay/mod.rs b/crates/features/src/relay/mod.rs index 48e14e00..47b946a6 100644 --- a/crates/features/src/relay/mod.rs +++ b/crates/features/src/relay/mod.rs @@ -51,6 +51,7 @@ const FEATURE: u8 = b'E'; 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 CREATE_ATTEMPTS: usize = 5; const MAX_NAME: usize = 100; @@ -569,6 +570,7 @@ pub async fn delete(data: &Store, id: u32) -> trc::Result<()> { data.write(batch.build_all()) .await .caused_by(trc::location!())?; + link::forget_ingested(data, id, u64::MAX).await?; Ok(()) } diff --git a/crates/registry/src/schema/enums.rs b/crates/registry/src/schema/enums.rs index 60020ad7..8d333fc7 100644 --- a/crates/registry/src/schema/enums.rs +++ b/crates/registry/src/schema/enums.rs @@ -840,6 +840,8 @@ pub enum ExpressionVariable { UrlOriginal = 89, Value = 90, ValueLower = 91, + // inbuxa: RL-23, the relay a message came through + Relay = 92, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)] @@ -3616,6 +3618,7 @@ pub static MTA_CONNECTION_VARIABLE: &[ExpressionVariable] = &[ ExpressionVariable::IsTls, ExpressionVariable::Asn, ExpressionVariable::Country, + ExpressionVariable::Relay, ]; pub static MTA_EHLO_VARIABLE: &[ExpressionVariable] = &[ @@ -3629,6 +3632,7 @@ pub static MTA_EHLO_VARIABLE: &[ExpressionVariable] = &[ ExpressionVariable::HeloDomain, ExpressionVariable::Asn, ExpressionVariable::Country, + ExpressionVariable::Relay, ]; pub static MTA_MAIL_FROM_VARIABLE: &[ExpressionVariable] = &[ @@ -3644,6 +3648,7 @@ pub static MTA_MAIL_FROM_VARIABLE: &[ExpressionVariable] = &[ ExpressionVariable::AuthenticatedAs, ExpressionVariable::Asn, ExpressionVariable::Country, + ExpressionVariable::Relay, ]; pub static MTA_QUEUE_HOST_VARIABLE: &[ExpressionVariable] = &[ @@ -3720,6 +3725,7 @@ pub static MTA_RCPT_TO_VARIABLE: &[ExpressionVariable] = &[ ExpressionVariable::HeloDomain, ExpressionVariable::Asn, ExpressionVariable::Country, + ExpressionVariable::Relay, ]; pub static MTA_RCPT_VARIABLE: &[ExpressionVariable] = &[ExpressionVariable::Rcpt]; diff --git a/crates/registry/src/schema/enums_impl.rs b/crates/registry/src/schema/enums_impl.rs index c74c4fb1..35e36eda 100644 --- a/crates/registry/src/schema/enums_impl.rs +++ b/crates/registry/src/schema/enums_impl.rs @@ -3775,6 +3775,7 @@ impl EnumImpl for ExpressionVariable { b"url_original" => ExpressionVariable::UrlOriginal, b"value" => ExpressionVariable::Value, b"value_lower" => ExpressionVariable::ValueLower, + b"relay" => ExpressionVariable::Relay, } .copied() } @@ -3873,6 +3874,7 @@ impl EnumImpl for ExpressionVariable { ExpressionVariable::UrlOriginal => "url_original", ExpressionVariable::Value => "value", ExpressionVariable::ValueLower => "value_lower", + ExpressionVariable::Relay => "relay", } } @@ -3974,11 +3976,12 @@ impl EnumImpl for ExpressionVariable { 89 => Some(ExpressionVariable::UrlOriginal), 90 => Some(ExpressionVariable::Value), 91 => Some(ExpressionVariable::ValueLower), + 92 => Some(ExpressionVariable::Relay), _ => None, } } - const COUNT: usize = 92; + const COUNT: usize = 93; } impl serde::Serialize for ExpressionVariable { diff --git a/crates/relay/src/spool.rs b/crates/relay/src/spool.rs index e2fbd2ae..783d9336 100644 --- a/crates/relay/src/spool.rs +++ b/crates/relay/src/spool.rs @@ -61,11 +61,13 @@ struct Entry { } impl Entry { + /// A message waiting for a verdict is pending even while leased: the + /// downstream has to know to answer it. fn state(&self, now: u64) -> InboundState { - if self.lease.as_ref().is_some_and(|l| l.until > now) { - InboundState::Claimed - } else if self.waiter.is_some() { + if self.waiter.is_some() { InboundState::Pending + } else if self.lease.as_ref().is_some_and(|l| l.until > now) { + InboundState::Claimed } else { InboundState::Held } @@ -267,13 +269,15 @@ impl Spool { /// The JMAP object for one message. pub fn get(&self, id: &str, now: u64) -> Option { let e = self.entries.get(id)?; - let state = e.state(now); Some(Inbound { id: e.meta.id.clone(), received_at: time::utc_date(e.meta.received_at), - state, - claimed_until: (state == InboundState::Claimed) - .then(|| time::utc_date(e.lease.as_ref().map_or(0, |l| l.until))), + state: e.state(now), + claimed_until: e + .lease + .as_ref() + .filter(|l| l.until > now) + .map(|l| time::utc_date(l.until)), envelope: e.meta.envelope.clone(), session: e.meta.session.clone(), size: e.meta.size, diff --git a/crates/relay/src/tests_e2e.rs b/crates/relay/src/tests_e2e.rs index 2273aa8c..73bd684a 100644 --- a/crates/relay/src/tests_e2e.rs +++ b/crates/relay/src/tests_e2e.rs @@ -589,6 +589,10 @@ async fn relay_end_to_end() { let got = ws.ok("Inbound/get", json!({ "ids": [live_id] })).await; assert_eq!(got["list"][0]["state"], "pending"); ws.ok("Inbound/claim", json!({})).await; + // Claimed, it still waits for its verdict + let got = ws.ok("Inbound/get", json!({ "ids": [live_id] })).await; + assert_eq!(got["list"][0]["state"], "pending"); + assert!(got["list"][0]["claimedUntil"].is_string()); let verdict = ws .ok( "Inbound/set", diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index 727d36f5..5f78fb4c 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/crates/services/src/inbuxa_relay_link.rs @@ -14,10 +14,19 @@ //! relay's settings and recipient set the same as this server's: in full //! when the link comes up, then as changes, every few seconds. //! -//! Push is asked for `Relay` only. Asking for `Inbound` would put the relay -//! in live mode, which waits for verdicts this node doesn't give yet. +//! Push is asked for `Relay` and `Inbound`, which puts the relay in live +//! mode (RL-12). Every message the relay takes is claimed, unsealed, and +//! judged by this server's own inbound pipeline with the relay's session +//! (RL-23): a live one gets this server's verdict before the sender is +//! answered, a held one is delivered, or reported to a sender who passed +//! SPF or DMARC (RL-17, RL-18). A message taken is recorded before the +//! relay is told, so one offered again isn't delivered twice (RL-16). -use common::{BuildServer, Inner, Server}; +use common::{ + BuildServer, Inner, Server, + config::server::{DEFAULT_TLS_TIMEOUT, ServerProtocol}, + network::{ServerInstance, TcpAcceptor, limiter::ConcurrencyLimiter}, +}; use futures::{SinkExt, StreamExt}; use http_body_util::{BodyExt, Full}; use hyper::body::Bytes; @@ -28,7 +37,8 @@ use inbuxa_features::{ self as relays, DownstreamKeys, JoinState, Relay, link::{ JoinRecorded, LinkState, LocalDomain, NODE_CERT_RENEW_AFTER, NodeCertificate, NodeLink, - RecipientPlan, issue_node_certificate, pushed_settings, record_join, set_status, + RecipientPlan, forget_ingested, is_inbound_id, issue_node_certificate, pushed_settings, + record_ingested, record_join, set_status, was_ingested, }, }, }; @@ -37,8 +47,11 @@ use inbuxa_relay_proto::{ jmap::{ClientFrame, Invocation, Request, ServerFrame, Session, method}, pin::{Pin, pins_match, spki_pin}, recipient::{RecipientDomain, RecipientHash}, - token, - types::{Capability, RELAY_SINGLETON, RelaySettings, RelayStatus}, + seal, token, + types::{ + Address, Capability, Inbound, InboundState, RELAY_SINGLETON, RelaySettings, RelayStatus, + Reply, Verdict, + }, }; use registry::schema::{ prelude::ObjectType, @@ -50,16 +63,25 @@ use rustls::{ }; use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}; use serde_json::{Value, json}; +use smtp::{ + core::Relayed, + inbound::relayed::{RelayedMessage, RelayedOutcome, ingest_relayed}, +}; use std::{ - collections::{BTreeMap, BTreeSet, HashMap}, + collections::{BTreeMap, BTreeSet, HashMap, HashSet}, sync::{Arc, Mutex}, time::Duration, }; use store::{registry::RegistryQuery, write::now}; -use tokio::{net::TcpStream, sync::watch, task::JoinHandle}; +use tokio::{ + net::TcpStream, + sync::{mpsc, watch}, + task::JoinHandle, +}; use tokio_rustls::{TlsConnector, client::TlsStream}; use tokio_tungstenite::{WebSocketStream, tungstenite::Message}; use types::id::Id; +use utils::snowflake::SnowflakeIdGenerator; /// How often the list of relays is read again. const SCAN: Duration = Duration::from_secs(10); @@ -79,16 +101,40 @@ const MAX_DELTA: usize = 2_000; const ACCOUNT_ID: &str = "relay"; /// Any name: the relay is known by its key, never its name (RL-6). const SERVER_NAME: &str = "relay.invalid"; +/// Messages one link takes at once. +const TAKE_MAX: usize = 8; +/// How long a claimed message is this node's (RL-15). +const LEASE_SECONDS: u64 = 300; +/// How often messages taken long ago are forgotten (RL-16). +const FORGET_EVERY: u64 = 3_600; pub fn spawn_relay_links(inner: Arc) { tokio::spawn(async move { let certs = Arc::new(NodeCertificates::default()); + // Relayed mail's sessions: port 25's, but no listener's + let instance = Arc::new(ServerInstance { + id: "relay".to_string(), + protocol: ServerProtocol::Smtp, + acceptor: TcpAcceptor::Plain, + limiter: ConcurrencyLimiter::new(100), + tls_timeout: DEFAULT_TLS_TIMEOUT, + shutdown_rx: watch::channel(false).1, + proxy_networks: vec![], + span_id_gen: Arc::new(SnowflakeIdGenerator::new()), + }); let mut links: HashMap = HashMap::new(); + let mut forgot_at = 0; loop { let server = inner.build_server(); - if let Err(err) = reconcile(&inner, &server, &certs, &mut links).await { + if let Err(err) = reconcile(&inner, &server, &certs, &instance, &mut links).await { trc::error!(err.details("Failed to start relay links")); } + if now() >= forgot_at + FORGET_EVERY { + forgot_at = now(); + if let Err(err) = forget_taken(&server).await { + trc::error!(err.details("Failed to forget relayed messages taken")); + } + } let woken = tokio::select! { _ = tokio::time::sleep(SCAN) => false, _ = relays::CHANGED.notified() => true, @@ -115,6 +161,7 @@ async fn reconcile( inner: &Arc, server: &Server, certs: &Arc, + instance: &Arc, links: &mut HashMap, ) -> trc::Result<()> { let data = server.store(); @@ -145,7 +192,13 @@ async fn reconcile( continue; } let (wake, woken) = watch::channel(0); - let handle = tokio::spawn(run_link(inner.clone(), id, certs.clone(), woken)); + let handle = tokio::spawn(run_link( + inner.clone(), + id, + certs.clone(), + instance.clone(), + woken, + )); links.insert( id, Running { @@ -158,6 +211,16 @@ async fn reconcile( Ok(()) } +/// RL-16: a relay's messages are remembered for its `heldMaxAge` and a day. +async fn forget_taken(server: &Server) -> trc::Result<()> { + let data = server.store(); + for relay in relays::all(data).await? { + let keep = (u64::from(relay.settings.held_max_age) + 1) * 86_400; + forget_ingested(data, relay.id, now().saturating_sub(keep)).await?; + } + Ok(()) +} + /// This node's link certificate, made again once it is old. #[derive(Default)] struct NodeCertificates(Mutex>>); @@ -210,6 +273,7 @@ async fn run_link( inner: Arc, relay_id: u32, certs: Arc, + instance: Arc, mut woken: watch::Receiver, ) { let mut backoff = BACKOFF_MIN; @@ -247,7 +311,16 @@ async fn run_link( "The join token ran out before the relay joined. Start a new join.", )), JoinState::Joined => { - session(&server, &relay, &keys, &cert, &mut woken, &mut status).await + session( + &server, + &instance, + &relay, + &keys, + &cert, + &mut woken, + &mut status, + ) + .await } }, Err(err) => Err(err), @@ -651,11 +724,12 @@ async fn join(data: &store::Store, relay: &Relay, cert: &NodeCertificate) -> Res // ---- The linked session --------------------------------------------------- -/// A WebSocket to the relay, with the `Relay` changes it pushed meanwhile. +/// A WebSocket to the relay, with the changes it pushed meanwhile. struct Ws { stream: WebSocketStream>, next_id: u64, relay_changed: bool, + inbound_changed: bool, } impl Ws { @@ -685,6 +759,7 @@ impl Ws { stream, next_id: 0, relay_changed: false, + inbound_changed: false, }) } @@ -716,10 +791,11 @@ impl Ws { } fn note(&mut self, frame: &ServerFrame) { - if let ServerFrame::StateChange { changed, .. } = frame - && changed.values().any(|types| types.contains_key("Relay")) - { - self.relay_changed = true; + if let ServerFrame::StateChange { changed, .. } = frame { + for types in changed.values() { + self.relay_changed |= types.contains_key("Relay"); + self.inbound_changed |= types.contains_key("Inbound"); + } } } @@ -801,6 +877,7 @@ struct Sent { /// turned off, moved or joined again, or this node stops. async fn session( server: &Server, + instance: &Arc, relay: &Relay, keys: &DownstreamKeys, cert: &NodeCertificate, @@ -809,11 +886,12 @@ async fn session( ) -> Result<(), LinkError> { let data = server.store(); let pin = relay.join.relay_pin; - let dialer = Dialer::new(relay, cert, pin)?; + let dialer = Arc::new(Dialer::new(relay, cert, pin)?); dialer.check_session().await?; let mut ws = Ws::open(&dialer).await?; + // Inbound too: live mode (RL-12) ws.send(&ClientFrame::WebSocketPushEnable { - data_types: Some(vec!["Relay".into()]), + data_types: Some(vec!["Relay".into(), "Inbound".into()]), push_state: None, }) .await?; @@ -823,6 +901,19 @@ async fn session( status.set(data, relay.id, LinkState::Linked, None).await; refresh_status(&mut ws, data, &relay, &sent, status).await?; + let (answers_tx, mut answers) = mpsc::unbounded_channel(); + let mut taker = Taker { + server: server.clone(), + instance: instance.clone(), + dialer: dialer.clone(), + relay_id: relay.id, + relay_name: relay.name.clone(), + secret: keys.sealing_secret, + taking: HashSet::new(), + answers: answers_tx, + }; + taker.take(&mut ws).await?; + let mut tick = tokio::time::interval(SYNC); tick.reset(); loop { @@ -833,6 +924,14 @@ async fn session( if ws.relay_changed { refresh_status(&mut ws, data, &relay, &sent, status).await?; } + if ws.inbound_changed { + taker.take(&mut ws).await?; + } + continue; + } + Some(answer) = answers.recv() => { + taker.answer(&mut ws, answer).await?; + taker.take(&mut ws).await?; continue; } _ = tick.tick() => {} @@ -856,6 +955,300 @@ async fn session( } sync(server, &relay, keys, &dialer, &mut ws, &mut sent).await?; refresh_status(&mut ws, data, &relay, &sent, status).await?; + taker.relay_name = relay.name.clone(); + taker.take(&mut ws).await?; + } +} + +// ---- Inbound mail ----------------------------------------------------------- + +/// Takes the relay's messages for this node: a few at a time, each judged in +/// its own task, each answered on the link when judged. +struct Taker { + server: Server, + instance: Arc, + dialer: Arc, + relay_id: u32, + relay_name: String, + secret: [u8; 32], + /// Claimed and still being judged. + taking: HashSet, + answers: mpsc::UnboundedSender, +} + +/// What a judged message's answer to the relay is. +struct Answer { + id: String, + action: Action, +} + +#[derive(Debug, PartialEq, Eq)] +enum Action { + /// Taken: the relay deletes it, and a live sender hears 250. + Destroy, + /// A live message refused: the relay gives the sender this reply. + Verdict(Verdict), + /// Not now: the lease runs out and it is offered again (RL-16). + Leave, +} + +impl Taker { + /// Claims what the relay holds, up to [`TAKE_MAX`] at once. + async fn take(&mut self, ws: &mut Ws) -> Result<(), LinkError> { + ws.inbound_changed = false; + let room = TAKE_MAX.saturating_sub(self.taking.len()); + if room == 0 { + return Ok(()); + } + let claimed = ws + .ok( + method::INBOUND_CLAIM, + json!({"limit": room, "leaseSeconds": LEASE_SECONDS}), + ) + .await?; + let ids: Vec<&str> = claimed["ids"] + .as_array() + .into_iter() + .flatten() + .filter_map(Value::as_str) + .collect(); + if ids.is_empty() { + return Ok(()); + } + let got = ws.ok(method::INBOUND_GET, json!({"ids": ids})).await?; + for inbound in got["list"].as_array().into_iter().flatten() { + let Ok(inbound) = serde_json::from_value::(inbound.clone()) else { + continue; + }; + if !is_inbound_id(&inbound.id) + || !is_inbound_id(&inbound.blob_id) + || !self.taking.insert(inbound.id.clone()) + { + continue; + } + let job = Job { + server: self.server.clone(), + instance: self.instance.clone(), + dialer: self.dialer.clone(), + relay_id: self.relay_id, + relay_name: self.relay_name.clone(), + secret: self.secret, + }; + let answers = self.answers.clone(); + tokio::spawn(async move { + let id = inbound.id.clone(); + let action = job.judge(inbound).await; + let _ = answers.send(Answer { id, action }); + }); + } + Ok(()) + } + + /// Tells the relay what became of one message. + async fn answer(&mut self, ws: &mut Ws, answer: Answer) -> Result<(), LinkError> { + self.taking.remove(&answer.id); + match answer.action { + Action::Destroy => { + ws.ok(method::INBOUND_SET, json!({"destroy": [answer.id]})) + .await?; + } + Action::Verdict(verdict) => { + // Refused as `notUpdated` if it stopped waiting meanwhile: + // then it is held, comes back when the lease runs out, and + // is judged as held + ws.ok( + method::INBOUND_SET, + json!({"update": {answer.id: {"verdict": verdict}}}), + ) + .await?; + } + Action::Leave => {} + } + Ok(()) + } +} + +/// Everything one message's judging needs, apart from the link. +struct Job { + server: Server, + instance: Arc, + dialer: Arc, + relay_id: u32, + relay_name: String, + secret: [u8; 32], +} + +impl Job { + async fn judge(self, inbound: Inbound) -> Action { + let data = self.server.store().clone(); + // RL-16: offered again after this server took it + match was_ingested(&data, self.relay_id, &inbound.id).await { + Ok(true) => return Action::Destroy, + Ok(false) => {} + Err(err) => { + trc::error!(err.details("Failed to read the relayed messages taken")); + return Action::Leave; + } + } + let live = inbound.state == InboundState::Pending; + + let message = match self.fetch(&inbound).await { + Ok(message) => message, + Err(reason) => { + trc::error!( + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(reason) + .details("Failed to take a relayed message") + .id(inbound.id.clone()) + ); + return Action::Leave; + } + }; + + let input = relayed_message(&self.relay_name, &inbound, !live, message); + let outcome = ingest_relayed(self.server.clone(), self.instance.clone(), input).await; + let action = action_for(&outcome, live); + if action == Action::Destroy + && let Err(err) = record_ingested(&data, self.relay_id, &inbound.id, now()).await + { + trc::error!(err.details("Failed to record a relayed message as taken")); + } + action + } + + /// The message, downloaded and unsealed (RL-21). + async fn fetch(&self, inbound: &Inbound) -> Result, String> { + let (status, sealed) = self + .dialer + .http( + "GET", + &format!("/jmap/download/relay/{}/msg", inbound.blob_id), + "application/octet-stream", + Vec::new(), + ) + .await + .map_err(|err| err.message)?; + if status != 200 { + return Err(format!( + "The relay answered {status} for a held message's download." + )); + } + seal::open(&self.secret, inbound.id.as_bytes(), &sealed) + .map_err(|err| format!("A relayed message didn't unseal: {err}")) + } +} + +/// The relay's session, as this server's pipeline takes it (RL-23). +fn relayed_message(relay: &str, inbound: &Inbound, held: bool, message: Vec) -> RelayedMessage { + let address = |address: &Address| { + ( + address.email.clone(), + address + .parameters + .iter() + .flatten() + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + ) + }; + let session = &inbound.session; + RelayedMessage { + relayed: Relayed { + relay: relay.to_string(), + held, + tls: session + .tls + .as_ref() + .map(|tls| (tls.version.clone(), tls.cipher.clone())), + dmarc_pass: false, + }, + local_ip: session.local_ip, + remote_ip: session.remote_ip, + remote_port: session.remote_port, + helo: session.helo.clone(), + mail_from: address(&inbound.envelope.mail_from), + rcpt_to: inbound.envelope.rcpt_to.iter().map(address).collect(), + message, + } +} + +/// What the relay is told, from what this server answered. +fn action_for(outcome: &RelayedOutcome, live: bool) -> Action { + match outcome { + RelayedOutcome::Accepted { .. } => Action::Destroy, + RelayedOutcome::Refused { reply, .. } if live => Action::Verdict(verdict_for(reply)), + // Held and refused for good: reported to the sender if it may be + // (RL-18), and gone + RelayedOutcome::Refused { reply, .. } if reply.starts_with('5') => Action::Destroy, + // Held and refused for now: tried again after the lease + RelayedOutcome::Refused { .. } => Action::Leave, + } +} + +/// A live message's verdict: this server's reply, as the relay gives it. +fn verdict_for(reply: &str) -> Verdict { + let mut parts = reply.splitn(3, ' '); + let code = parts.next().and_then(|code| code.parse::().ok()); + let class = code.map_or(4, |code| code / 100); + let (enhanced, text) = match parts.next() { + Some(esc) + if esc.split('.').count() == 3 + && esc.split('.').all(|p| p.parse::().is_ok()) + && esc.starts_with(char::from(b'0' + class as u8)) => + { + ( + esc.to_string(), + parts.next().unwrap_or_default().to_string(), + ) + } + Some(word) => ( + format!("{class}.0.0"), + [word, parts.next().unwrap_or_default()].join(" "), + ), + None => (format!("{class}.0.0"), String::new()), + }; + // Not a reply at all: its words, as a temporary failure + let text = if code.is_none() { + reply.to_string() + } else { + text + }; + let text: String = text + .trim() + .chars() + .filter(|c| !c.is_control()) + .take(400) + .collect(); + let text = if text.is_empty() { + "Refused".to_string() + } else { + text + }; + let verdict = match (code, class) { + (Some(code), 5) => Verdict::Reject(Reply { + code, + enhanced, + text, + }), + (Some(code), 4) => Verdict::Tempfail(Reply { + code, + enhanced, + text, + }), + _ => Verdict::Tempfail(Reply { + code: 451, + enhanced: "4.3.0".into(), + text, + }), + }; + if verdict.is_valid() { + verdict + } else { + Verdict::Tempfail(Reply { + code: 451, + enhanced: "4.3.0".into(), + text: "Try again later".into(), + }) } } @@ -1060,3 +1453,70 @@ async fn local_recipients(server: &Server) -> trc::Result<(Vec, Vec } Ok((local, addresses)) } + +#[cfg(test)] +mod tests { + use super::*; + + fn reply(code: u16, enhanced: &str, text: &str) -> Reply { + Reply { + code, + enhanced: enhanced.into(), + text: text.into(), + } + } + + #[test] + fn verdicts() { + assert_eq!( + verdict_for("550 5.7.1 Message rejected due to excessive spam score."), + Verdict::Reject(reply( + 550, + "5.7.1", + "Message rejected due to excessive spam score." + )) + ); + assert_eq!( + verdict_for("450 4.4.6 Too many Received headers."), + Verdict::Tempfail(reply(450, "4.4.6", "Too many Received headers.")) + ); + // No enhanced code, or one of the wrong class: the class's own + assert_eq!( + verdict_for("554 no thanks"), + Verdict::Reject(reply(554, "5.0.0", "no thanks")) + ); + assert_eq!( + verdict_for("451 5.7.1 odd"), + Verdict::Tempfail(reply(451, "4.0.0", "5.7.1 odd")) + ); + // Anything else is a temporary failure, never a 250 + assert_eq!( + verdict_for("garbage"), + Verdict::Tempfail(reply(451, "4.3.0", "garbage")) + ); + assert!(verdict_for("550 5.1.1").is_valid()); + } + + #[test] + fn actions() { + let accepted = RelayedOutcome::Accepted { + refused: vec![("joe@example.com".into(), "550 5.1.1 No".into())], + }; + let refused = |reply: &str| RelayedOutcome::Refused { + reply: reply.into(), + refused: vec![], + }; + assert_eq!(action_for(&accepted, true), Action::Destroy); + assert_eq!(action_for(&accepted, false), Action::Destroy); + assert!(matches!( + action_for(&refused("550 5.7.1 Spam"), true), + Action::Verdict(Verdict::Reject(_)) + )); + // Held: refused for good is reported and gone, for now is retried + assert_eq!(action_for(&refused("550 5.1.1 No"), false), Action::Destroy); + assert_eq!( + action_for(&refused("451 4.3.0 Later"), false), + Action::Leave + ); + } +} diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index 45145327..ef409303 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -109,6 +109,23 @@ pub struct SessionData { // to, and recipients rules added, by rule name pub journal_marks: Vec, pub journal_added: Vec<(String, String)>, + // inbuxa: RL-23: a message a relay took on port 25, judged here as if + // this server had taken the session + pub relayed: Option, +} + +/// inbuxa: RL-23: what a relay saw of the session it took a message in. +/// Only the relay link service sets it, for a pinned relay. +#[derive(Debug, Clone, Default)] +pub struct Relayed { + /// The relay's name, for `relay` in expressions. + pub relay: String, + /// The relay already said 250 (RL-17). + pub held: bool, + /// TLS version and cipher between the sender and the relay. + pub tls: Option<(String, String)>, + /// DMARC passed, once DATA checked it (RL-18). + pub dmarc_pass: bool, } /// inbuxa: a DATA refusal by DLP rules: blocked, or a warning the sender @@ -199,6 +216,7 @@ impl SessionData { journal_marks: Vec::new(), journal_added: Vec::new(), access_token: None, + relayed: None, } } } @@ -328,6 +346,7 @@ impl SessionData { journal_marks: Vec::new(), journal_added: Vec::new(), access_token: None, + relayed: None, } } } diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index c78f258d..9ac61587 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -428,6 +428,10 @@ impl Session { let dmarc_result = dmarc_output.result(); let pass = dmarc_result == DmarcResult::Pass; + // inbuxa: RL-18, a held message's failures bounce only then + if let Some(relayed) = self.data.relayed.as_mut() { + relayed.dmarc_pass = pass; + } let strict = dmarc.is_strict(); let is_temp_fail = matches!(dmarc_result, DmarcResult::TempError(_)); let rejected = strict @@ -604,6 +608,25 @@ impl Session { | rcpt_spam_flag(scores.spam_percentage(user_score)); } } + // inbuxa: RL-17: held mail was already accepted, so it is + // never dropped or refused for spam: it goes to Junk + SpamFilterAction::Discard | SpamFilterAction::Reject + if self.data.relayed.as_ref().is_some_and(|r| r.held) => + { + trc::event!( + Spam(SpamEvent::Classify), + SpanId = self.data.session_id, + QueueId = message_id, + Result = "junk", + Reason = "Held by a relay, over the reject threshold: filed as Junk.", + ); + headers.extend_from_slice(b"X-Inbuxa-Relay: held, over reject threshold\r\n"); + for recipient in self.data.rcpt_to.iter_mut() { + recipient.flags = (recipient.flags & !RCPT_SPAM_MASK) | rcpt_spam_flag(100); + } + let scores = &self.server.core.spam.scores; + spam_result = Some((scores.reject_threshold.max(scores.spam_threshold), 100)); + } SpamFilterAction::Discard => { trc::event!( Spam(SpamEvent::Classify), @@ -1184,8 +1207,13 @@ impl Session { headers.extend_from_slice(b")"); } headers.extend_from_slice(b")\r\n\t"); - if self.stream.is_tls() { - let (version, cipher) = self.stream.tls_version_and_cipher(); + // inbuxa: RL-23: a relayed message's TLS is the relay's with the sender + let tls = match self.data.relayed.as_ref().and_then(|r| r.tls.as_ref()) { + Some((version, cipher)) => Some((version.clone().into(), cipher.clone().into())), + None if self.stream.is_tls() => Some(self.stream.tls_version_and_cipher()), + None => None, + }; + if let Some((version, cipher)) = tls { headers.extend_from_slice(b"(using "); headers.extend_from_slice(version.as_bytes()); headers.extend_from_slice(b" with cipher "); @@ -1195,7 +1223,7 @@ impl Session { headers.extend_from_slice(b"by "); headers.extend_from_slice(self.hostname.as_bytes()); headers.extend_from_slice(concat!(" (", types::brand!(), " SMTP) with ").as_bytes()); - headers.extend_from_slice(match (self.stream.is_tls(), !self.is_authenticated()) { + headers.extend_from_slice(match (self.is_tls(), !self.is_authenticated()) { (true, true) => b"ESMTPS", (true, false) => b"ESMTPSA", (false, true) => b"ESMTP", diff --git a/crates/smtp/src/inbound/mod.rs b/crates/smtp/src/inbound/mod.rs index 78d54470..ccb4d7b7 100644 --- a/crates/smtp/src/inbound/mod.rs +++ b/crates/smtp/src/inbound/mod.rs @@ -18,6 +18,7 @@ pub mod mail; pub mod mailflow; // inbuxa: DLP and mail flow rules pub mod milter; pub mod rcpt; +pub mod relayed; // inbuxa: RL-23, mail an inbuxa Relay took pub mod session; pub mod spam; pub mod spawn; diff --git a/crates/smtp/src/inbound/relayed.rs b/crates/smtp/src/inbound/relayed.rs new file mode 100644 index 00000000..1fdcfce1 --- /dev/null +++ b/crates/smtp/src/inbound/relayed.rs @@ -0,0 +1,489 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! inbuxa: RL-23: a message an inbuxa Relay took on port 25, judged here as +//! if this server had taken the SMTP session itself. +//! +//! The relay's session is played again against this server's own handlers: +//! connect, EHLO, MAIL FROM, every RCPT TO, then DATA, with the original +//! sender's address, port and HELO. SPF, reverse IP, DMARC, DNSBL, the spam +//! filter, the AI classifier, Sieve and the directory all see the real +//! facts, and the answer is the one this server would have given. +//! +//! Only the relay link service calls this, for a relay whose key it pinned. +//! Mail arriving on this server's own port 25 is never treated as relayed. + +use crate::{ + core::{Relayed, Session, SessionData, SessionParameters, State}, + queue::{ + Error, ErrorDetails, MessageSource, Recipient, Status, UnexpectedResponse, dsn::SendDsn, + spool::SmtpSpool, + }, +}; +use common::{ + Server, + network::{ServerInstance, stream::NullIo}, +}; +use mail_auth::SpfResult; +use smtp_proto::{ + MAIL_RET_FULL, MAIL_RET_HDRS, MailFrom, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, + RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, RcptTo, Request, Response, + request::receiver::RequestReceiver, +}; +use std::{borrow::Cow, net::IpAddr, sync::Arc, time::Duration}; +use store::write::{BatchBuilder, BlobLink, BlobOp, now}; +use types::blob_hash::BlobHash; + +/// An address and its SMTP parameters, as the relay passed them on. +pub type EnvelopeAddress = (String, Vec<(String, Option)>); + +/// The session a relay took a message in, and the message. +pub struct RelayedMessage { + pub relayed: Relayed, + pub local_ip: IpAddr, + pub remote_ip: IpAddr, + pub remote_port: u16, + pub helo: String, + /// The return path and its MAIL FROM parameters. + pub mail_from: EnvelopeAddress, + /// Each recipient and its RCPT TO parameters. + pub rcpt_to: Vec, + pub message: Vec, +} + +/// What this server answered. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RelayedOutcome { + /// Queued for every recipient but those in `refused`, each with its + /// reply. + Accepted { refused: Vec<(String, String)> }, + /// Refused as a whole, with this reply. `refused` has the recipients + /// refused one by one before it, if any. + Refused { + reply: String, + refused: Vec<(String, String)>, + }, +} + +/// How long a held message's headers stay stored for its DSN. +const DSN_HEADERS_FOR: u64 = 600; + +/// Runs `input` through this server's inbound pipeline. A held message's +/// refusals are reported to its sender only when the sender passed SPF or +/// DMARC here (RL-18); a live one's whole refusal is the relay's to give. +pub async fn ingest_relayed( + server: Server, + instance: Arc, + input: RelayedMessage, +) -> RelayedOutcome { + let session_id = instance.span_id_gen.generate(); + let asn_geo_data = server.lookup_asn_country(input.remote_ip).await; + // Port 25: the relay took it on port 25, and this server's settings + // tell port 25 from submission by it + let mut data = SessionData::new( + input.local_ip, + 25, + input.remote_ip, + input.remote_port, + asn_geo_data, + session_id, + ); + let held = input.relayed.held; + data.relayed = Some(input.relayed); + let mut session = Session { + hostname: String::new(), + state: State::None, + instance, + server, + stream: NullIo::default(), + data, + params: SessionParameters::default(), + }; + + let mut lines = Lines::default(); + let outcome = Box::pin(session.play( + input.helo, + input.mail_from, + input.rcpt_to, + input.message, + &mut lines, + )) + .await; + if held { + let refused = match &outcome { + RelayedOutcome::Accepted { refused } => refused.clone(), + RelayedOutcome::Refused { reply, refused } if reply.starts_with('5') => { + // Every recipient that got as far as RCPT, refused at once + let mut all = refused.clone(); + all.extend( + session + .data + .rcpt_to + .iter() + .map(|rcpt| (rcpt.address.clone(), reply.clone())), + ); + all + } + RelayedOutcome::Refused { .. } => Vec::new(), + }; + if !refused.is_empty() { + session.bounce_held(&refused, &lines).await; + } + } + outcome +} + +/// The MAIL FROM and RCPT TO commands as parsed, for a DSN. +#[derive(Default)] +struct Lines { + mail_from: Option>, + rcpt_to: Vec>, +} + +impl Session { + async fn play( + &mut self, + helo: String, + mail_from: EnvelopeAddress, + rcpt_to: Vec, + message: Vec, + lines: &mut Lines, + ) -> RelayedOutcome { + let mut refused = Vec::new(); + let refuse = |reply: String, refused: Vec<(String, String)>| RelayedOutcome::Refused { + reply, + refused, + }; + + // Connect: rate limits, then the connect stage's scripts and filters + if !self.is_allowed().await { + return refuse( + "421 4.7.0 Too many connections, try again later.".into(), + refused, + ); + } + if !self.init_conn().await { + let reply = self + .has_failed() + .unwrap_or_else(|| "421 4.3.0 Try again later.".into()); + return refuse(reply, refused); + } + self.stream.tx_buf.clear(); + + // EHLO + let _ = self.handle_ehlo(Cow::Owned(helo), true).await; + if let Some(reply) = self.has_failed() { + return refuse(reply, refused); + } + + // MAIL FROM + let line = command("MAIL FROM", &mail_from); + let mut receiver = RequestReceiver::default(); + let Ok(Request::Mail { from }) = receiver.ingest(&mut line.iter()) else { + return refuse("501 5.5.4 Invalid MAIL FROM.".into(), refused); + }; + lines.mail_from = Some(from.clone().into_owned()); + let _ = self.handle_mail_from(from).await; + if let Some(reply) = self.has_failed() { + return refuse(reply, refused); + } + + // RCPT TO: the relay already checked each recipient (RL-1); this + // server checks again, by its own directory and rules + self.params.rcpt_errors_wait = Duration::ZERO; + for (address, parameters) in rcpt_to { + let line = command("RCPT TO", &(address.clone(), parameters)); + let mut receiver = RequestReceiver::default(); + let Ok(Request::Rcpt { mut to }) = receiver.ingest(&mut line.iter()) else { + refused.push((address, "501 5.5.4 Invalid RCPT TO.".into())); + continue; + }; + if !self.params.rcpt_dsn { + // The relay offers DSN to everyone; this server may not + to.flags &= !(RCPT_NOTIFY_DELAY + | RCPT_NOTIFY_FAILURE + | RCPT_NOTIFY_NEVER + | RCPT_NOTIFY_SUCCESS); + to.orcpt = None; + } + lines.rcpt_to.push(to.clone().into_owned()); + let _ = self.handle_rcpt_to(to).await; + if let Some(reply) = self.has_failed() { + refused.push((address, reply)); + } + } + if self.data.rcpt_to.is_empty() { + // The reply the sender would have been given: a temporary one + // if any recipient had one, so it tries again + let reply = refused + .iter() + .map(|(_, reply)| reply) + .find(|reply| reply.starts_with('4')) + .or(refused.first().map(|(_, reply)| reply)) + .cloned() + .unwrap_or_else(|| "554 5.5.1 No valid recipients.".into()); + return refuse(reply, refused); + } + + // DATA + if let Some(reply) = self.can_send_data().await { + return refuse(reply_text(reply), refused); + } + if message.len() >= self.params.max_message_size { + return refuse("552 5.3.4 Message too big for system.".into(), refused); + } + self.data.message = message; + let reply = self.queue_message().await; + let reply = reply_text(&reply); + if reply.starts_with('2') { + RelayedOutcome::Accepted { refused } + } else { + refuse(reply, refused) + } + } + + /// RL-18: a DSN for recipients refused after the relay had said 250, + /// only when the sender passed SPF or DMARC here, so a forged sender + /// gets none and the relay is no backscatter source. + async fn bounce_held(&mut self, refused: &[(String, String)], lines: &Lines) { + let Some(mail_from) = self.data.mail_from.as_ref() else { + return; + }; + let spf_pass = self + .data + .spf_mail_from + .as_ref() + .is_some_and(|spf| spf.result() == SpfResult::Pass); + let dmarc_pass = self.data.relayed.as_ref().is_some_and(|r| r.dmarc_pass); + if mail_from.address_lcase.is_empty() || !(spf_pass || dmarc_pass) { + trc::event!( + Delivery(trc::DeliveryEvent::DoubleBounce), + SpanId = self.data.session_id, + From = mail_from.address_lcase.clone(), + Reason = "Held by a relay and refused here; the sender passed neither SPF nor DMARC, so no DSN.", + ); + return; + } + + let mut wrapper = self.server.new_message( + mail_from.address.as_str(), + MessageSource::Unauthenticated { dmarc_pass }, + self.data.session_id, + ); + if let Some(from) = &lines.mail_from { + wrapper.message.env_id = from.env_id.as_deref().map(Into::into); + wrapper.message.flags |= from.flags & (MAIL_RET_FULL | MAIL_RET_HDRS); + } + wrapper.message.received_from_ip = self.data.remote_ip; + wrapper.message.received_via_port = 25; + for (address, reply) in refused { + let line = lines + .rcpt_to + .iter() + .find(|to| to.address.eq_ignore_ascii_case(address)); + let flags = line.map_or(0, |to| to.flags); + let flags = if flags + & (RCPT_NOTIFY_DELAY + | RCPT_NOTIFY_FAILURE + | RCPT_NOTIFY_NEVER + | RCPT_NOTIFY_SUCCESS) + == 0 + { + RCPT_NOTIFY_FAILURE + } else { + flags + }; + let mut recipient = Recipient::new(address).with_flags(flags); + recipient.orcpt = line.and_then(|to| to.orcpt.as_deref()).map(Into::into); + recipient.status = Status::PermanentFailure(ErrorDetails { + entity: self.hostname.as_str().into(), + details: Error::UnexpectedResponse(UnexpectedResponse { + command: format!("RCPT TO:<{address}>").into_boxed_str(), + response: parse_response(reply), + }), + }); + wrapper.message.recipients.push(recipient); + } + + // The DSN quotes the message's headers from the blob store + let message = &self.data.message; + let headers = &message[..header_end(message)]; + wrapper.message.blob_hash = BlobHash::generate(headers); + wrapper.message.size = message.len() as u64; + let mut batch = BatchBuilder::new(); + batch.set( + BlobOp::Link { + hash: wrapper.message.blob_hash.clone(), + to: BlobLink::Temporary { + until: now() + DSN_HEADERS_FOR, + }, + }, + vec![], + ); + let stored = match self.server.store().write(batch.build_all()).await { + Ok(_) => self + .server + .blob_store() + .put_blob( + wrapper.message.blob_hash.as_slice(), + headers, + self.server.core.email.compression, + ) + .await + .map(|_| ()), + Err(err) => Err(err), + }; + if let Err(err) = stored { + trc::error!( + err.details("Failed to store a relayed message's headers for its DSN.") + .span_id(self.data.session_id) + .caused_by(trc::location!()) + ); + return; + } + self.server.send_dsn(&mut wrapper).await; + } +} + +/// `MAIL FROM:
KEY=value...` or the RCPT TO equivalent. Anything +/// that could end the line or start another parameter is dropped. +fn command(verb: &str, (address, parameters): &EnvelopeAddress) -> Vec { + let clean = |text: &str| -> String { + text.chars() + .filter(|c| !c.is_control() && !matches!(c, ' ' | '<' | '>')) + .collect() + }; + let mut line = format!("{verb}:<{}>", clean(address)); + for (key, value) in parameters { + let key: String = clean(key).replace('=', ""); + if key.is_empty() { + continue; + } + line.push(' '); + line.push_str(&key); + if let Some(value) = value { + line.push('='); + line.push_str(&clean(value)); + } + } + line.push_str("\r\n"); + line.into_bytes() +} + +/// A reply as text, its last line if it has several. +fn reply_text(reply: &[u8]) -> String { + let text = String::from_utf8_lossy(reply); + text.trim_end() + .rsplit("\r\n") + .next() + .unwrap_or_default() + .chars() + .take(512) + .collect::() + .trim() + .to_string() +} + +/// `550 5.1.1 text` as a response. +fn parse_response(reply: &str) -> Response> { + let mut parts = reply.splitn(3, ' '); + let code = parts.next().and_then(|c| c.parse().ok()).unwrap_or(550); + let (esc, message) = match parts.next() { + Some(esc) if enhanced(esc).is_some() => ( + enhanced(esc).unwrap_or([5, 0, 0]), + parts.next().unwrap_or_default(), + ), + other => ( + [(code / 100) as u8, 0, 0], + reply + .split_once(' ') + .map(|(_, rest)| rest) + .filter(|_| other.is_some()) + .unwrap_or_default(), + ), + }; + Response { + code, + esc, + message: message.into(), + } +} + +fn enhanced(text: &str) -> Option<[u8; 3]> { + let mut parts = text.split('.').map(|p| p.parse::().ok()); + let esc = [parts.next()??, parts.next()??, parts.next()??]; + parts.next().is_none().then_some(esc) +} + +/// Where the headers end: after the blank line, or the whole message. +fn header_end(message: &[u8]) -> usize { + message + .windows(4) + .position(|w| w == b"\r\n\r\n") + .map_or(message.len(), |pos| pos + 4) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn builds_commands() { + let line = command( + "RCPT TO", + &( + "jane@example.com".into(), + vec![ + ("NOTIFY".into(), Some("FAILURE,DELAY".into())), + ("ORCPT".into(), Some("rfc822;jane@example.com".into())), + ], + ), + ); + assert_eq!( + line, + b"RCPT TO: NOTIFY=FAILURE,DELAY ORCPT=rfc822;jane@example.com\r\n" + ); + let mut receiver = RequestReceiver::default(); + let Ok(Request::Rcpt { to }) = receiver.ingest(&mut line.iter()) else { + panic!("not parsed"); + }; + assert_eq!(to.address, "jane@example.com"); + assert_ne!(to.flags & RCPT_NOTIFY_FAILURE, 0); + + // Nothing can end the line or add a parameter + let line = command( + "MAIL FROM", + &( + "a@b.c> AUTH=x\r\nRCPT TO: BODYX=8BITMIME\r\n" + ); + } + + #[test] + fn reads_replies() { + assert_eq!( + reply_text(b"550-5.7.1 First\r\n550 5.7.1 Message rejected.\r\n"), + "550 5.7.1 Message rejected." + ); + let response = parse_response("550 5.1.1 Mailbox unavailable"); + assert_eq!( + (response.code, response.esc, response.message.as_ref()), + (550, [5, 1, 1], "Mailbox unavailable") + ); + let response = parse_response("451 try later"); + assert_eq!( + (response.code, response.esc, response.message.as_ref()), + (451, [4, 0, 0], "try later") + ); + assert_eq!(header_end(b"A: b\r\n\r\nbody"), 8); + assert_eq!(header_end(b"A: b"), 4); + } +} diff --git a/crates/smtp/src/inbound/session.rs b/crates/smtp/src/inbound/session.rs index 11394997..d8ad4f11 100644 --- a/crates/smtp/src/inbound/session.rs +++ b/crates/smtp/src/inbound/session.rs @@ -29,6 +29,12 @@ use crate::core::{Session, State}; use super::auth::{AUTH_BEARER, SaslToken}; impl Session { + /// inbuxa: TLS on this session, or between the sender and the relay + /// that took it (RL-23). + pub fn is_tls(&self) -> bool { + self.stream.is_tls() || self.data.relayed.as_ref().is_some_and(|r| r.tls.is_some()) + } + pub async fn ingest(&mut self, bytes: &[u8]) -> Result { let mut iter = bytes.iter(); let mut state = std::mem::replace(&mut self.state, State::None); @@ -623,7 +629,7 @@ impl ResolveVariable for Session { ExpressionVariable::RemotePort => self.data.remote_port.into(), ExpressionVariable::LocalIp => self.data.local_ip_str.as_str().into(), ExpressionVariable::LocalPort => self.data.local_port.into(), - ExpressionVariable::IsTls => self.stream.is_tls().into(), + ExpressionVariable::IsTls => self.is_tls().into(), ExpressionVariable::Priority => self.data.priority.to_compact_string().into(), ExpressionVariable::Protocol => self.instance.protocol.as_str().into(), ExpressionVariable::Asn => self @@ -642,6 +648,13 @@ impl ResolveVariable for Session { .map(|c| c.as_str()) .unwrap_or_default() .into(), + ExpressionVariable::Relay => self + .data + .relayed + .as_ref() + .map(|r| r.relay.as_str()) + .unwrap_or_default() + .into(), _ => expr::Variable::default(), } } diff --git a/crates/smtp/src/inbound/spam.rs b/crates/smtp/src/inbound/spam.rs index 94e85e5e..ab1fcd36 100644 --- a/crates/smtp/src/inbound/spam.rs +++ b/crates/smtp/src/inbound/spam.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::core::Session; @@ -70,7 +72,7 @@ impl Session { authenticated_as: self.data.authenticated_as.as_ref().map(|a| a.name()), asn: self.data.asn_geo_data.asn.as_ref().map(|a| a.id), country: self.data.asn_geo_data.country.as_ref().map(|c| c.as_str()), - is_tls: self.stream.is_tls(), + is_tls: self.is_tls(), env_from: self .data .mail_from diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index be66a11c..a522edf8 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -227,6 +227,12 @@ so a reboot without the link comes back with the same settings (RL-22). stop someone holding the relay's key from testing guessed addresses offline; nothing on the relay could, since the relay has to check real ones. +- A domain whose recipients an external directory (LDAP, SQL, OIDC) + answers for can't be listed. Its recipient set holds the accounts the + downstream has already seen from that directory, and nothing more: no + catch-all, which would break RL-1. Anyone else there is refused at the + relay until they first sign in (decided 2026-10-29). A live lookup over + the link may come later. - Deltas use `Recipient/set` (create and destroy by hash). A full refresh uses the custom `Recipient/replace {blobId}`: a sorted file of 32-byte hashes, swapped in atomically. The relay's `Recipient` state string lets @@ -238,8 +244,8 @@ so a reboot without the link comes back with the same settings (RL-22). |---|---| | `id` | Stable for the message's whole life. The downstream deduplicates on it (RL-16) | | `receivedAt` | When the relay accepted it (or got the final dot, if pending) | -| `state` | `pending` (live, awaiting a verdict), `held`, or `claimed` | -| `claimedUntil` | End of the current lease, when claimed | +| `state` | `pending` (live, awaiting a verdict, claimed or not), `held`, or `claimed` (held and leased) | +| `claimedUntil` | End of the current lease, when leased | | `envelope` | RFC 8621 `Envelope`: `mailFrom` and `rcptTo`, with parameters | | `session` | `{remoteIp, remotePort, localIp, helo, tls: {version, cipher} or null, ptr}` | | `size` | Bytes | @@ -675,8 +681,17 @@ In progress. Done so far: recipient set (whole, then deltas), and reports each node's link in the relay's `links` (PR #253). Node link certificates last 30 days and are renewed after 20. -- Not done yet: ingest, live verdicts, `inbuxa:RelayMessage`, relay events - in the trace, and the link-down alert. +- Ingest and live verdicts: the link service asks for `Inbound` push, so + the relay is live while a node is linked. Each message is claimed, + unsealed, and played through the downstream's own SMTP handlers with + the relay's session on port 25 (RL-23), with `relay` in expressions. + Live, the downstream's reply is the verdict; held, a refusal for good is + reported only to a sender who passed SPF or DMARC (RL-18), one for now + is retried after the lease, and spam over the reject threshold goes to + Junk (RL-17). Every message taken is recorded until `heldMaxAge` and a + day have passed (RL-16). A pending message stays `pending` when claimed. +- Not done yet: `inbuxa:RelayMessage`, relay events in the trace, and the + link-down alert. Order: diff --git a/resources/schema/schema.json.gz b/resources/schema/schema.json.gz index 916d0991..d1e2a536 100644 Binary files a/resources/schema/schema.json.gz and b/resources/schema/schema.json.gz differ diff --git a/resources/schema/schema.json.sha256 b/resources/schema/schema.json.sha256 index f426104e..1c23e561 100644 --- a/resources/schema/schema.json.sha256 +++ b/resources/schema/schema.json.sha256 @@ -1 +1 @@ -Hc6Z_iXa_4ma9Pch8CLDiLJ7nuLZC_hz5_jqi4Dp2x4 \ No newline at end of file +B5ARrqKFTX10Ib2fIAsGacr8yxF4NyI3IwmJEPFiTP8 \ No newline at end of file diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index de8c63b4..2330e114 100644 --- a/tests/src/system/relay.rs +++ b/tests/src/system/relay.rs @@ -13,7 +13,10 @@ //! //! `relay_link_tests` runs a real `inbuxa-relay` (build it first with //! `cargo build -p inbuxa-relay`): the server joins it, sets it up, and its -//! port 25 takes mail only for this server's real recipients. +//! port 25 takes mail only for this server's real recipients. Live, a +//! message is delivered before the sender hears 250, and this server's +//! refusal is the sender's reply; held while the link is off, it is +//! delivered once the link is back, exactly once. use crate::utils::{ account::Account, @@ -450,6 +453,102 @@ async fn rcpt_replies(port: u16, recipients: &[&str]) -> Vec { replies.split_off(2) } +/// Sends one message through the relay; the reply to its final dot. +async fn send_mail(port: u16, rcpt: &str, message: &str) -> String { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + let stream = tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .unwrap(); + let (read, mut write) = stream.into_split(); + let mut read = BufReader::new(read); + async fn reply(read: &mut BufReader) -> String { + let mut all = String::new(); + loop { + let mut line = String::new(); + read.read_line(&mut line).await.unwrap(); + all.push_str(&line); + if line.len() < 4 || line.as_bytes()[3] != b'-' { + return all; + } + } + } + assert!(reply(&mut read).await.starts_with("220")); + for line in [ + "EHLO sender.example".to_string(), + "MAIL FROM:".to_string(), + format!("RCPT TO:<{rcpt}>"), + "DATA".to_string(), + ] { + write + .write_all(format!("{line}\r\n").as_bytes()) + .await + .unwrap(); + let answer = reply(&mut read).await; + assert!( + answer.starts_with('2') || answer.starts_with('3'), + "{line}: {answer}" + ); + } + let body = message.replace('\n', "\r\n"); + write + .write_all(format!("{body}\r\n.\r\n").as_bytes()) + .await + .unwrap(); + let answer = reply(&mut read).await; + let _ = write.write_all(b"QUIT\r\n").await; + answer +} + +/// The subjects of every message in `account`'s mailboxes. +async fn subjects(account: &Account) -> Vec { + let response = account + .jmap_request( + &["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"], + json!([ + ["Email/query", {"accountId": account.id_string()}, "q"], + ["Email/get", { + "accountId": account.id_string(), + "#ids": {"resultOf": "q", "name": "Email/query", "path": "/ids"}, + "properties": ["subject"] + }, "g"] + ]), + ) + .await; + response + .0 + .pointer("/methodResponses/1/1/list") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|email| email["subject"].as_str().map(str::to_string)) + .collect() +} + +/// Waits until `account` has `count` messages titled `subject`. +async fn wait_for_mail(account: &Account, subject: &str, count: usize) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let found = subjects(account).await; + let n = found.iter().filter(|s| *s == subject).count(); + if n == count { + return; + } + assert!( + n < count && std::time::Instant::now() < deadline, + "{count} of {subject:?}: {found:?}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } +} + +fn message(subject: &str, extra_headers: &str) -> String { + format!( + "{extra_headers}From: Someone \nTo: jane@example.com\n\ + Subject: {subject}\nMessage-ID: <{}@sender.example>\n\nHello from outside.", + subject.replace(' ', "-") + ) +} + pub async fn link(test: &mut TestServer) { println!("Running inbuxa Relay link tests..."); let admin = test.account("admin@example.com"); @@ -459,7 +558,7 @@ pub async fn link(test: &mut TestServer) { "{} is missing: run cargo build -p inbuxa-relay", binary.display() ); - admin + let jane = admin .create_user_account( "jane@example.com", "jane-secret-4417", @@ -525,7 +624,7 @@ pub async fn link(test: &mut TestServer) { .as_array() .into_iter() .flatten() - .any(|link| link["state"] == "linked" && link["relay"]["mode"] == "held") + .any(|link| link["state"] == "linked" && link["relay"]["mode"] == "live") }) .await; assert!(relay["linkKeyPin"].as_str().is_some(), "{relay}"); @@ -556,6 +655,27 @@ pub async fn link(test: &mut TestServer) { "not served: {replies:?}" ); + // Live (RL-12): delivered here before the sender hears 250 + let reply = send_mail(smtp_port, "jane@example.com", &message("Live one", "")).await; + assert!(reply.starts_with("250"), "{reply}"); + wait_for_mail(&jane, "Live one", 1).await; + + // Live: this server's refusal is the sender's reply, and the relay + // keeps nothing + let loop_headers = + "Received: from a.example by b.example; Thu, 29 Oct 2026 10:00:00 +0000\n".repeat(60); + let reply = send_mail( + smtp_port, + "jane@example.com", + &message("Looped one", &loop_headers), + ) + .await; + assert!(reply.starts_with("450 4.4.6"), "{reply}"); + wait_for(admin, &id, "the relay kept nothing", 30, |relay| { + relay["links"][0]["relay"]["spool"]["inbound"] == 0 + }) + .await; + // A new account reaches the relay as a change admin .create_user_account("joe@example.com", "joe-secret-5529", "Joe", &[], vec![]) @@ -606,6 +726,31 @@ pub async fn link(test: &mut TestServer) { }) .await; + // Held while the link is off: the relay says 250 itself, and the + // message is delivered once the link is back, once + let reply = send_mail(smtp_port, "jane@example.com", &message("Held one", "")).await; + assert!(reply.starts_with("250"), "{reply}"); + call( + admin, + "inbuxa:Relay/set", + json!({"update": {&id: {"enabled": true}}}), + ) + .await; + wait_for_mail(&jane, "Held one", 1).await; + wait_for(admin, &id, "the held message was taken", 30, |relay| { + relay["links"][0]["state"] == "linked" + && relay["links"][0]["relay"]["spool"]["inbound"] == 0 + }) + .await; + assert_eq!( + subjects(&jane) + .await + .iter() + .filter(|s| *s == "Held one" || *s == "Live one") + .count(), + 2 + ); + call(admin, "inbuxa:Relay/set", json!({"destroy": [&id]})).await; }