From 21d16fc150f52f866cb26597f96d27559671902c Mon Sep 17 00:00:00 2001 From: John Coffey Date: Fri, 30 Oct 2026 06:33:14 -0700 Subject: [PATCH] relay: the Outbound queue Outbound/set create takes an uploaded message and its envelope, refuses a sender outside the relay's domains (RL-19, RL-T10), and spools it in spool/out/ with a queued status for each recipient. Outbound/get and /changes report it, with push, and destroy cancels or clears one. The inbound spool's change log is shared with it. Delivery is the next step. --- crates/relay-proto/src/jmap.rs | 3 + crates/relay-proto/src/types.rs | 88 +++++ crates/relay/src/app.rs | 20 +- crates/relay/src/changelog.rs | 103 ++++++ crates/relay/src/link/jmap.rs | 151 ++++++++- crates/relay/src/main.rs | 9 +- crates/relay/src/outbound.rs | 550 ++++++++++++++++++++++++++++++++ crates/relay/src/spool.rs | 79 +---- crates/relay/src/state.rs | 11 +- crates/relay/src/tests_e2e.rs | 52 +++ docs/spec/features/relay.md | 17 +- 11 files changed, 1007 insertions(+), 76 deletions(-) create mode 100644 crates/relay/src/changelog.rs create mode 100644 crates/relay/src/outbound.rs diff --git a/crates/relay-proto/src/jmap.rs b/crates/relay-proto/src/jmap.rs index 1d8bc2e7..091d611a 100644 --- a/crates/relay-proto/src/jmap.rs +++ b/crates/relay-proto/src/jmap.rs @@ -28,6 +28,9 @@ pub mod method { pub const INBOUND_QUERY: &str = "Inbound/query"; pub const INBOUND_CLAIM: &str = "Inbound/claim"; pub const INBOUND_SET: &str = "Inbound/set"; + pub const OUTBOUND_GET: &str = "Outbound/get"; + pub const OUTBOUND_CHANGES: &str = "Outbound/changes"; + pub const OUTBOUND_SET: &str = "Outbound/set"; pub const RELAY_EVENT_GET: &str = "RelayEvent/get"; pub const RELAY_EVENT_CHANGES: &str = "RelayEvent/changes"; pub const RELAY_EVENT_SET: &str = "RelayEvent/set"; diff --git a/crates/relay-proto/src/types.rs b/crates/relay-proto/src/types.rs index 4a796a10..7952ce52 100644 --- a/crates/relay-proto/src/types.rs +++ b/crates/relay-proto/src/types.rs @@ -349,6 +349,70 @@ impl Verdict { } } +/// A message the downstream handed to the relay to send (RL-19). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Outbound { + pub id: String, + pub created_at: String, + /// The upload the message was made from, as the downstream signed it. + pub blob_id: String, + pub envelope: Envelope, + pub size: u64, + /// Per recipient, by address. + pub delivery_status: BTreeMap, + /// The next try, while any recipient is still queued. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_attempt_at: Option, + pub attempts: u32, + pub expires_at: String, +} + +/// What `Outbound/set` create takes. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct NewOutbound { + pub blob_id: String, + pub envelope: Envelope, +} + +/// RFC 8621 `DeliveryStatus`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DeliveryStatus { + /// The remote server's last reply, empty before the first try. + pub smtp_reply: String, + pub delivered: Delivered, + /// The relay never learns this. + pub displayed: Displayed, +} + +impl DeliveryStatus { + pub fn queued() -> Self { + DeliveryStatus { + smtp_reply: String::new(), + delivered: Delivered::Queued, + displayed: Displayed::Unknown, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum Delivered { + Queued, + Yes, + No, + Unknown, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum Displayed { + Unknown, + Yes, +} + /// A relay event, for the downstream's history and tracer. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -407,6 +471,30 @@ mod tests { assert!(s.serves("EXAMPLE.com")); } + #[test] + fn outbound_wire_form() { + let status = serde_json::to_value(DeliveryStatus::queued()).unwrap(); + assert_eq!( + status, + json!({"smtpReply": "", "delivered": "queued", "displayed": "unknown"}) + ); + let new: NewOutbound = serde_json::from_value(json!({ + "blobId": "b1", + "envelope": {"mailFrom": {"email": "a@example.com"}, "rcptTo": [{"email": "b@example.org"}]}, + })) + .unwrap(); + assert_eq!(new.envelope.rcpt_to.len(), 1); + assert!( + serde_json::from_value::(json!({ + "blobId": "b1", + "envelope": {"mailFrom": {"email": ""}, "rcptTo": []}, + "deliveryStatus": {}, + })) + .is_err(), + "the relay sets the status" + ); + } + #[test] fn certificate_key_stays_out_of_debug() { let cert = SmtpCertificate { diff --git a/crates/relay/src/app.rs b/crates/relay/src/app.rs index 9514ba2d..00e539f7 100644 --- a/crates/relay/src/app.rs +++ b/crates/relay/src/app.rs @@ -8,6 +8,7 @@ use crate::{ events::Events, + outbound::Outbox, recipients::Recipients, spool::Spool, state::{self, Identity, Paths}, @@ -43,6 +44,7 @@ pub struct App { settings_state: AtomicU64, pub recipients: RwLock, pub spool: Mutex, + pub outbox: Mutex, pub events: Mutex, pub push: broadcast::Sender, /// Link connections, and those with push on for `Inbound`. @@ -60,6 +62,7 @@ impl App { let settings = state::load_settings(&paths)?.map(Arc::new); let recipients = Recipients::load(&paths)?; let spool = Spool::open(&paths.spool_in())?; + let outbox = Outbox::open(&paths.spool_out())?; let events = Events::open(paths.events())?; // Uploads don't survive a restart. if let Ok(dir) = std::fs::read_dir(paths.blobs()) { @@ -75,6 +78,7 @@ impl App { settings_state: AtomicU64::new(1), recipients: RwLock::new(recipients), spool: Mutex::new(spool), + outbox: Mutex::new(outbox), events: Mutex::new(events), push, link_clients: AtomicU32::new(0), @@ -145,10 +149,14 @@ impl App { } pub fn status(&self) -> RelayStatus { - let (inbound, bytes) = { + let (inbound, inbound_bytes) = { let spool = self.spool.lock().expect("spool"); (spool.count(), spool.bytes()) }; + let (outbound, outbound_bytes) = { + let outbox = self.outbox.lock().expect("outbox"); + (outbox.count(), outbox.bytes()) + }; RelayStatus { started_at: time::utc_date(self.started_at), relay_version: VERSION.to_string(), @@ -156,8 +164,8 @@ impl App { mode: self.mode(), spool: SpoolStatus { inbound, - outbound: 0, - bytes, + outbound, + bytes: inbound_bytes + outbound_bytes, }, recipient_state: self.recipients.read().expect("recipients").state(), certificate_not_after: None, @@ -166,9 +174,11 @@ impl App { } } - /// Whether the spool is too full for new mail: past 90% of the cap. + /// Whether the spool is too full for new mail, either way: past 90% of + /// the cap. pub fn spool_full(&self, settings: &RelaySettings) -> bool { - let bytes = self.spool.lock().expect("spool").bytes(); + let bytes = + self.spool.lock().expect("spool").bytes() + self.outbox.lock().expect("outbox").bytes(); bytes.saturating_mul(10) >= settings.spool_max_bytes.saturating_mul(9) || disk_nearly_full(&self.paths.spool_in()) } diff --git a/crates/relay/src/changelog.rs b/crates/relay/src/changelog.rs new file mode 100644 index 00000000..6d4b89e1 --- /dev/null +++ b/crates/relay/src/changelog.rs @@ -0,0 +1,103 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! State strings and `/changes` for the spools: a counter, and the ids +//! destroyed lately. Runtime only; a restart starts every client over. + +use std::collections::VecDeque; + +/// How many destroyed ids `/changes` can still report. +const TOMBSTONES: usize = 10_000; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Changes { + pub old_state: String, + pub new_state: String, + pub created: Vec, + pub updated: Vec, + pub destroyed: Vec, + pub has_more: bool, +} + +pub struct Changelog { + modseq: u64, + tombstones: VecDeque<(u64, String)>, + /// `/changes` can't go back past this. + floor: u64, +} + +impl Default for Changelog { + fn default() -> Self { + Changelog { + modseq: 1, + tombstones: VecDeque::new(), + floor: 1, + } + } +} + +impl Changelog { + pub fn state(&self) -> String { + self.modseq.to_string() + } + + /// The next modseq, for an object created or changed. + pub fn bump(&mut self) -> u64 { + self.modseq += 1; + self.modseq + } + + pub fn destroyed(&mut self, id: &str) { + let modseq = self.bump(); + self.tombstones.push_back((modseq, id.to_string())); + while self.tombstones.len() > TOMBSTONES { + if let Some((seq, _)) = self.tombstones.pop_front() { + self.floor = seq; + } + } + } + + /// What changed since `since`, given each live object's id and its + /// created and last-changed modseqs. `None` when the client must start + /// over: an unknown state, or more than `max` changes. + pub fn changes<'a>( + &self, + since: &str, + max: usize, + objects: impl Iterator, + ) -> Option { + let since: u64 = since.parse().ok()?; + if since < self.floor || since > self.modseq { + return None; + } + let mut created = Vec::new(); + let mut updated = Vec::new(); + for (id, created_at, modseq) in objects { + if created_at > since { + created.push(id.clone()); + } else if modseq > since { + updated.push(id.clone()); + } + } + let destroyed: Vec = self + .tombstones + .iter() + .filter(|(seq, _)| *seq > since) + .map(|(_, id)| id.clone()) + .collect(); + if created.len() + updated.len() + destroyed.len() > max { + return None; + } + Some(Changes { + old_state: since.to_string(), + new_state: self.state(), + created, + updated, + destroyed, + has_more: false, + }) + } +} diff --git a/crates/relay/src/link/jmap.rs b/crates/relay/src/link/jmap.rs index ed498e30..f9470be2 100644 --- a/crates/relay/src/link/jmap.rs +++ b/crates/relay/src/link/jmap.rs @@ -16,7 +16,7 @@ use inbuxa_relay_proto::{ jmap::{Invocation, MethodError, Request, Response, method}, recipient::{RecipientDomain, RecipientHash}, time, token, - types::{RELAY_SINGLETON, RelaySettings, Verdict, type_name}, + types::{NewOutbound, RELAY_SINGLETON, RelaySettings, Verdict, type_name}, }; use rustls_pki_types::{CertificateDer, pem::PemObject}; use serde_json::{Map, Value, json}; @@ -115,6 +115,9 @@ fn call_one(app: &App, caller: &Caller, call: &Invocation) -> MethodResult { method::INBOUND_CHANGES => inbound_changes(app, args), method::INBOUND_CLAIM => inbound_claim(app, caller, args), method::INBOUND_SET => inbound_set(app, args), + method::OUTBOUND_GET => outbound_get(app, args), + method::OUTBOUND_CHANGES => outbound_changes(app, args), + method::OUTBOUND_SET => outbound_set(app, args), method::RELAY_EVENT_GET => event_get(app, args), method::RELAY_EVENT_CHANGES => event_changes(app, args), method::RELAY_EVENT_SET => event_set(app, args), @@ -747,6 +750,148 @@ fn inbound_set(app: &App, args: &Map) -> MethodResult { })) } +// ---- Outbound ------------------------------------------------------------- + +fn outbound_get(app: &App, args: &Map) -> MethodResult { + let ids = string_list(args, "ids")?; + let outbox = app.outbox.lock().expect("outbox"); + let ids = ids.unwrap_or_else(|| outbox.ids()); + let mut list = Vec::new(); + let mut not_found = Vec::new(); + for id in ids { + match outbox.get(&id) { + Some(outbound) => { + list.push(serde_json::to_value(outbound).expect("outbound serializes")) + } + None => not_found.push(id), + } + } + Ok(json!({ + "accountId": ACCOUNT_ID, + "state": outbox.state(), + "list": list, + "notFound": not_found, + })) +} + +fn outbound_changes(app: &App, args: &Map) -> MethodResult { + let since = args + .get("sinceState") + .and_then(Value::as_str) + .ok_or_else(|| err(MethodError::InvalidArguments, "sinceState is required"))?; + let max = args + .get("maxChanges") + .and_then(Value::as_u64) + .map_or(MAX_CHANGES, |m| (m as usize).clamp(1, MAX_CHANGES)); + let outbox = app.outbox.lock().expect("outbox"); + let changes = outbox + .changes(since, max) + .ok_or((MethodError::CannotCalculateChanges, None))?; + Ok(json!({ + "accountId": ACCOUNT_ID, + "oldState": changes.old_state, + "newState": changes.new_state, + "hasMoreChanges": changes.has_more, + "created": changes.created, + "updated": changes.updated, + "destroyed": changes.destroyed, + })) +} + +/// RL-19: the only way mail leaves through the relay. +fn outbound_set(app: &App, args: &Map) -> MethodResult { + let mut created = Map::new(); + let mut not_created = Map::new(); + let mut not_updated = Map::new(); + let mut destroyed = Vec::new(); + let mut not_destroyed = Map::new(); + let settings = app.settings(); + let create = object(args, "create")?; + let full = match (&settings, create) { + (Some(settings), Some(create)) if !create.is_empty() => app.spool_full(settings), + _ => false, + }; + let mut outbox = app.outbox.lock().expect("outbox"); + let old_state = outbox.state(); + for (key, value) in create.into_iter().flatten() { + let new = match serde_json::from_value::(value.clone()) { + Ok(new) if crate::link::is_blob_id(&new.blob_id) => new, + Ok(_) => { + not_created.insert( + key.clone(), + set_error("invalidProperties", "blobId: an uploaded blob"), + ); + continue; + } + Err(e) => { + not_created.insert(key.clone(), set_error("invalidProperties", e.to_string())); + continue; + } + }; + let Some(settings) = &settings else { + not_created.insert( + key.clone(), + set_error("forbidden", "the relay has no settings yet"), + ); + continue; + }; + let blob = app.paths.blobs().join(&new.blob_id); + match outbox.create( + &blob, + &new.blob_id, + new.envelope, + settings, + full, + time::now(), + ) { + Ok(id) => { + let object = outbox.get(&id).expect("just created"); + created.insert( + key.clone(), + serde_json::to_value(object).expect("outbound serializes"), + ); + } + Err(refusal) => { + not_created.insert(key.clone(), set_error(refusal.kind, refusal.description)); + } + } + } + for id in object(args, "update")? + .into_iter() + .flatten() + .map(|(id, _)| id) + { + not_updated.insert( + id.clone(), + set_error("forbidden", "only the relay changes an Outbound"), + ); + } + for id in string_list(args, "destroy")?.unwrap_or_default() { + if outbox.destroy(&id) { + destroyed.push(id); + } else { + not_destroyed.insert(id, set_error("notFound", "no such message")); + } + } + let new_state = outbox.state(); + drop(outbox); + if new_state != old_state { + app.notify(type_name::OUTBOUND, new_state.clone()); + app.relay_changed(); + } + Ok(json!({ + "accountId": ACCOUNT_ID, + "oldState": old_state, + "newState": new_state, + "created": created, + "notCreated": not_created, + "updated": {}, + "notUpdated": not_updated, + "destroyed": destroyed, + "notDestroyed": not_destroyed, + })) +} + // ---- RelayEvent ----------------------------------------------------------- fn event_get(app: &App, args: &Map) -> MethodResult { @@ -823,6 +968,10 @@ pub fn states(app: &App) -> Vec<(&'static str, String)> { (type_name::RECIPIENT, recipient.clone()), (type_name::RECIPIENT_DOMAIN, recipient), (type_name::INBOUND, app.spool.lock().expect("spool").state()), + ( + type_name::OUTBOUND, + app.outbox.lock().expect("outbox").state(), + ), ( type_name::RELAY_EVENT, app.events.lock().expect("events").state(), diff --git a/crates/relay/src/main.rs b/crates/relay/src/main.rs index 107418f6..ab0db699 100644 --- a/crates/relay/src/main.rs +++ b/crates/relay/src/main.rs @@ -12,8 +12,10 @@ //! command line can join, run, show status, and reset; nothing else. mod app; +mod changelog; mod events; mod link; +mod outbound; mod recipients; mod smtp; mod spool; @@ -286,10 +288,9 @@ fn status(paths: &Paths) -> Result<(), String> { println!("Running: yes, since {}", s.started_at); println!("Mode: {:?}", s.mode); println!("Linked: {} connection(s)", s.link_clients); - println!( - "Held mail: {} message(s), {} bytes", - s.spool.inbound, s.spool.bytes - ); + println!("Held mail: {} message(s)", s.spool.inbound); + println!("Outbound: {} message(s)", s.spool.outbound); + println!("Spool: {} bytes", s.spool.bytes); if let Some(e) = s.last_error { println!("Last error: {e}"); } diff --git a/crates/relay/src/outbound.rs b/crates/relay/src/outbound.rs new file mode 100644 index 00000000..260be180 --- /dev/null +++ b/crates/relay/src/outbound.rs @@ -0,0 +1,550 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! The outbound spool (spec "Outbound", "The spool", RL-19). +//! +//! Each message is two files in `spool/out/`: `.eml`, the message as +//! the downstream handed it over, moved in from its upload and fsynced +//! first; then `.json`, its properties and delivery status, written +//! atomically. The `.json` is the commit point, as in the inbound spool. +//! Outbound mail isn't sealed: the relay has to read it to send it. + +use crate::{ + changelog::{Changelog, Changes}, + state::{set_private, sync_dir, write_atomic}, +}; +use inbuxa_relay_proto::{ + time, + types::{Address, DeliveryStatus, Envelope, Outbound, RelaySettings}, +}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::{BTreeMap, HashSet}, + fs, io, + path::{Path, PathBuf}, +}; + +/// Longest address, as RFC 5321 ยง4.5.3.1 allows for a path. +const ADDRESS_MAX: usize = 256; +const PARAMETER_NAME_MAX: usize = 64; +const PARAMETER_VALUE_MAX: usize = 500; + +/// A message's stored properties. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Meta { + pub id: String, + pub created_at: u64, + pub blob_id: String, + pub envelope: Envelope, + pub size: u64, + pub delivery_status: BTreeMap, + pub next_attempt_at: u64, + pub attempts: u32, + pub expires_at: u64, +} + +struct Entry { + meta: Meta, + created: u64, + modseq: u64, +} + +/// Why a create was refused: a JMAP `SetError` type and a description. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Refusal { + pub kind: &'static str, + pub description: String, +} + +fn refuse(kind: &'static str, description: impl Into) -> Refusal { + Refusal { + kind, + description: description.into(), + } +} + +pub struct Outbox { + dir: PathBuf, + entries: BTreeMap, + log: Changelog, + bytes: u64, +} + +pub fn new_id() -> String { + let mut id = crate::spool::new_id(); + id.replace_range(0..1, "o"); + id +} + +/// Whether `id` is one of ours: never a path. +pub fn is_valid_id(id: &str) -> bool { + id.len() == 25 && id.starts_with('o') && id[1..].bytes().all(|b| b.is_ascii_hexdigit()) +} + +impl Outbox { + /// Opens the spool, deleting what was never committed. + pub fn open(dir: &Path) -> io::Result { + fs::create_dir_all(dir)?; + let mut outbox = Outbox { + dir: dir.to_path_buf(), + entries: BTreeMap::new(), + log: Changelog::default(), + bytes: 0, + }; + let mut messages = Vec::new(); + for entry in fs::read_dir(dir)? { + let entry = entry?; + let name = entry.file_name().to_string_lossy().into_owned(); + if let Some(id) = name.strip_suffix(".json") { + if !is_valid_id(id) { + continue; + } + let meta: Meta = match fs::read(entry.path()) + .ok() + .and_then(|b| serde_json::from_slice(&b).ok()) + { + Some(meta) => meta, + None => { + eprintln!("inbuxa-relay: unreadable outbound entry {id}, skipped"); + continue; + } + }; + if !outbox.eml_path(id).exists() { + let _ = fs::remove_file(entry.path()); + continue; + } + outbox.bytes += meta.size; + outbox.entries.insert( + id.to_string(), + Entry { + meta, + created: 1, + modseq: 1, + }, + ); + } else if name.ends_with(".eml") || name.ends_with(".tmp") { + messages.push((name, entry.path())); + } + } + for (name, path) in messages { + let id = name.split('.').next().unwrap_or(""); + if name.ends_with(".tmp") || !outbox.entries.contains_key(id) { + let _ = fs::remove_file(path); + } + } + Ok(outbox) + } + + pub fn eml_path(&self, id: &str) -> PathBuf { + self.dir.join(format!("{id}.eml")) + } + + fn meta_path(&self, id: &str) -> PathBuf { + self.dir.join(format!("{id}.json")) + } + + pub fn state(&self) -> String { + self.log.state() + } + + pub fn count(&self) -> u64 { + self.entries.len() as u64 + } + + pub fn bytes(&self) -> u64 { + self.bytes + } + + /// RL-19: takes the uploaded message at `blob` for `envelope`. The + /// upload is moved into the spool, so it can't be used twice. `full` + /// is whether the spool is past its limit. + pub fn create( + &mut self, + blob: &Path, + blob_id: &str, + envelope: Envelope, + settings: &RelaySettings, + full: bool, + now: u64, + ) -> Result { + check_envelope(&envelope, settings)?; + let size = fs::metadata(blob) + .map_err(|_| refuse("invalidProperties", "blobId: no such upload"))? + .len(); + if size == 0 { + return Err(refuse("invalidProperties", "the message is empty")); + } + if size > settings.max_message_size { + return Err(refuse( + "tooLarge", + format!("the message is over {} bytes", settings.max_message_size), + )); + } + if full { + return Err(refuse("overQuota", "the relay's spool is full")); + } + let id = new_id(); + let meta = Meta { + id: id.clone(), + created_at: now, + blob_id: blob_id.to_string(), + delivery_status: envelope + .rcpt_to + .iter() + .map(|r| (r.email.clone(), DeliveryStatus::queued())) + .collect(), + envelope, + size, + next_attempt_at: now, + attempts: 0, + expires_at: now + u64::from(settings.outbound_expiry), + }; + self.commit(blob, &meta) + .map_err(|e| refuse("serverFail", format!("could not spool the message: {e}")))?; + let modseq = self.log.bump(); + self.bytes += size; + self.entries.insert( + id.clone(), + Entry { + meta, + created: modseq, + modseq, + }, + ); + Ok(id) + } + + fn commit(&self, blob: &Path, meta: &Meta) -> io::Result<()> { + let eml = self.eml_path(&meta.id); + fs::File::open(blob)?.sync_all()?; + fs::rename(blob, &eml)?; + set_private(&eml, false)?; + sync_dir(&self.dir)?; + let json = serde_json::to_vec(meta).map_err(io::Error::other)?; + if let Err(e) = write_atomic(&self.meta_path(&meta.id), &json) { + let _ = fs::remove_file(&eml); + return Err(e); + } + Ok(()) + } + + /// The JMAP object for one message. + pub fn get(&self, id: &str) -> Option { + let m = &self.entries.get(id)?.meta; + let queued = m + .delivery_status + .values() + .any(|s| s.delivered == inbuxa_relay_proto::types::Delivered::Queued); + Some(Outbound { + id: m.id.clone(), + created_at: time::utc_date(m.created_at), + blob_id: m.blob_id.clone(), + envelope: m.envelope.clone(), + size: m.size, + delivery_status: m.delivery_status.clone(), + next_attempt_at: queued.then(|| time::utc_date(m.next_attempt_at)), + attempts: m.attempts, + expires_at: time::utc_date(m.expires_at), + }) + } + + /// Ids, oldest first. + pub fn ids(&self) -> Vec { + let mut ids: Vec<&Meta> = self.entries.values().map(|e| &e.meta).collect(); + ids.sort_by_key(|m| (m.created_at, m.id.clone())); + ids.into_iter().map(|m| m.id.clone()).collect() + } + + pub fn changes(&self, since: &str, max: usize) -> Option { + self.log.changes( + since, + max, + self.entries.iter().map(|(id, e)| (id, e.created, e.modseq)), + ) + } + + /// Cancels a queued message, or clears a finished one. + pub fn destroy(&mut self, id: &str) -> bool { + let Some(e) = self.entries.remove(id) else { + return false; + }; + self.bytes = self.bytes.saturating_sub(e.meta.size); + let _ = fs::remove_file(self.meta_path(id)); + let _ = fs::remove_file(self.eml_path(id)); + let _ = sync_dir(&self.dir); + self.log.destroyed(id); + true + } +} + +/// RL-19: a sender at one of the relay's domains, or none; real recipients; +/// SMTP parameters that can't break the command line they go on. +pub fn check_envelope(envelope: &Envelope, settings: &RelaySettings) -> Result<(), Refusal> { + let from = &envelope.mail_from; + if !from.email.is_empty() { + let domain = domain_of(&from.email) + .ok_or_else(|| refuse("invalidProperties", "mailFrom isn't an address"))?; + if !settings.serves(domain) { + return Err(refuse( + "forbiddenFrom", + "mailFrom isn't at one of the relay's domains", + )); + } + } + check_parameters(from, "mailFrom")?; + if envelope.rcpt_to.is_empty() { + return Err(refuse("invalidProperties", "rcptTo needs a recipient")); + } + if envelope.rcpt_to.len() > settings.max_recipients as usize { + return Err(refuse( + "tooLarge", + format!("at most {} recipients", settings.max_recipients), + )); + } + let mut seen = HashSet::new(); + for rcpt in &envelope.rcpt_to { + if domain_of(&rcpt.email).is_none() { + return Err(refuse( + "invalidProperties", + format!("rcptTo: {} isn't an address", rcpt.email), + )); + } + if !seen.insert(rcpt.email.to_lowercase()) { + return Err(refuse( + "invalidProperties", + format!("rcptTo: {} is listed twice", rcpt.email), + )); + } + check_parameters(rcpt, "rcptTo")?; + } + Ok(()) +} + +/// The domain of a plain address, if it is one. +fn domain_of(email: &str) -> Option<&str> { + if email.len() > ADDRESS_MAX + || email + .chars() + .any(|c| c.is_control() || c.is_whitespace() || matches!(c, '<' | '>')) + { + return None; + } + let (local, domain) = email.rsplit_once('@')?; + let quoted = local.len() > 1 && local.starts_with('"') && local.ends_with('"'); + let domain = domain.trim_end_matches('.'); + (!local.is_empty() && !domain.is_empty() && (quoted || !local.contains('@'))).then_some(domain) +} + +fn check_parameters(address: &Address, field: &str) -> Result<(), Refusal> { + for (name, value) in address.parameters.iter().flatten() { + let name_ok = !name.is_empty() + && name.len() <= PARAMETER_NAME_MAX + && name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-'); + let value_ok = value.as_ref().is_none_or(|v| { + !v.is_empty() + && v.len() <= PARAMETER_VALUE_MAX + && !v + .chars() + .any(|c| c.is_control() || c.is_whitespace() || c == '=') + }); + if !name_ok || !value_ok { + return Err(refuse( + "invalidProperties", + format!("{field}: the parameter {name} isn't valid"), + )); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use inbuxa_relay_proto::types::Delivered; + + fn settings() -> RelaySettings { + RelaySettings { + domains: vec!["example.com".into()], + max_recipients: 3, + max_message_size: 100, + ..Default::default() + } + } + + fn envelope(from: &str, to: &[&str]) -> Envelope { + Envelope { + mail_from: Address::new(from), + rcpt_to: to.iter().map(|t| Address::new(*t)).collect(), + } + } + + fn dir(name: &str) -> PathBuf { + let d = std::env::temp_dir().join(format!( + "inbuxa-relay-outbound-{name}-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&d); + d + } + + fn upload(dir: &Path, body: &[u8]) -> PathBuf { + fs::create_dir_all(dir).unwrap(); + let path = dir.join(crate::spool::new_id()); + fs::write(&path, body).unwrap(); + path + } + + #[test] + fn senders_only_from_the_relays_domains() { + let s = settings(); + assert!(check_envelope(&envelope("a@example.com", &["b@example.org"]), &s).is_ok()); + assert!(check_envelope(&envelope("a@EXAMPLE.com.", &["b@example.org"]), &s).is_ok()); + assert!( + check_envelope(&envelope("", &["b@example.org"]), &s).is_ok(), + "a bounce" + ); + for from in [ + "a@example.org", + "a@sub.example.com", + "a@example.com@evil.org", + ] { + assert_eq!( + check_envelope(&envelope(from, &["b@example.org"]), &s) + .unwrap_err() + .kind, + if from.contains("@evil") { + "invalidProperties" + } else { + "forbiddenFrom" + }, + "{from}" + ); + } + } + + #[test] + fn recipients_and_parameters_are_checked() { + let s = settings(); + let bad = [ + envelope("a@example.com", &[]), + envelope( + "a@example.com", + &["x@a.org", "y@a.org", "z@a.org", "w@a.org"], + ), + envelope("a@example.com", &["x@a.org", "X@a.org"]), + envelope("a@example.com", &["x@a.org\r\nRCPT TO:"]), + envelope("a@example.com", &["no-at-sign"]), + envelope("a@example.com", &[""]), + ]; + for e in &bad { + assert!(check_envelope(e, &s).is_err(), "{e:?}"); + } + let mut e = envelope("a@example.com", &["x@a.org"]); + e.rcpt_to[0].parameters = Some(BTreeMap::from([("NOTIFY".into(), Some("NEVER".into()))])); + e.mail_from.parameters = Some(BTreeMap::from([ + ("BODY".into(), Some("8BITMIME".into())), + ("SMTPUTF8".into(), None), + ])); + assert!(check_envelope(&e, &s).is_ok()); + e.rcpt_to[0].parameters = Some(BTreeMap::from([( + "NOTIFY".into(), + Some("NEVER AUTH=x".into()), + )])); + assert!(check_envelope(&e, &s).is_err()); + } + + #[test] + fn created_survives_restart_and_destroys() { + let d = dir("restart"); + let blobs = d.join("blobs"); + let out = d.join("out"); + let s = settings(); + let mut outbox = Outbox::open(&out).unwrap(); + let s0 = outbox.state(); + + let blob = upload(&blobs, b"Subject: hi\r\n\r\nhi\r\n"); + let id = outbox + .create( + &blob, + "b1", + envelope("a@example.com", &["b@example.org", "c@example.org"]), + &s, + false, + 100, + ) + .unwrap(); + assert!(!blob.exists(), "the upload moved into the spool"); + assert_eq!( + fs::read(outbox.eml_path(&id)).unwrap(), + b"Subject: hi\r\n\r\nhi\r\n" + ); + let got = outbox.get(&id).unwrap(); + assert_eq!(got.blob_id, "b1"); + assert_eq!(got.size, 19); + assert_eq!(got.delivery_status.len(), 2); + assert!( + got.delivery_status + .values() + .all(|s| s.delivered == Delivered::Queued) + ); + assert_eq!(got.next_attempt_at.as_deref(), Some(&*time::utc_date(100))); + assert_eq!(got.expires_at, time::utc_date(100 + 3 * 86_400)); + assert_eq!(outbox.changes(&s0, 10).unwrap().created, vec![id.clone()]); + + // A crash left an uncommitted message. + fs::write(out.join(format!("{}.eml", new_id())), b"junk").unwrap(); + let mut outbox = Outbox::open(&out).unwrap(); + assert_eq!(outbox.count(), 1); + assert_eq!(outbox.bytes(), 19); + assert_eq!(fs::read_dir(&out).unwrap().count(), 2); + + let s1 = outbox.state(); + assert!(outbox.destroy(&id)); + assert!(!outbox.destroy(&id)); + assert_eq!(outbox.changes(&s1, 10).unwrap().destroyed, vec![id]); + assert_eq!(fs::read_dir(&out).unwrap().count(), 0); + fs::remove_dir_all(d).unwrap(); + } + + #[test] + fn create_refusals() { + let d = dir("refusals"); + let blobs = d.join("blobs"); + let s = settings(); + let mut outbox = Outbox::open(&d.join("out")).unwrap(); + let ok = envelope("a@example.com", &["b@example.org"]); + + let big = upload(&blobs, &[b'x'; 101]); + let r = outbox.create(&big, "b", ok.clone(), &s, false, 0); + assert_eq!(r.unwrap_err().kind, "tooLarge"); + assert!(big.exists(), "a refused upload stays where it was"); + + let blob = upload(&blobs, b"x"); + let r = outbox.create(&blob, "b", ok.clone(), &s, true, 0); + assert_eq!(r.unwrap_err().kind, "overQuota"); + let r = outbox.create( + &blob, + "b", + envelope("a@example.org", &["b@x.org"]), + &s, + false, + 0, + ); + assert_eq!(r.unwrap_err().kind, "forbiddenFrom"); + let r = outbox.create(&blobs.join("gone"), "b", ok, &s, false, 0); + assert_eq!(r.unwrap_err().kind, "invalidProperties"); + assert_eq!(outbox.count(), 0); + fs::remove_dir_all(d).unwrap(); + } + + #[test] + fn ids_are_never_paths() { + assert!(is_valid_id(&new_id())); + assert!(!crate::spool::is_valid_id(&new_id())); + assert!(!is_valid_id("o../../etc/passwd")); + } +} diff --git a/crates/relay/src/spool.rs b/crates/relay/src/spool.rs index 783d9336..deb16ae3 100644 --- a/crates/relay/src/spool.rs +++ b/crates/relay/src/spool.rs @@ -15,7 +15,10 @@ //! Leases and live verdict waiters are runtime only. After a restart every //! message is held and unleased; the downstream deduplicates by id (RL-16). -use crate::state::{set_private, sync_dir, write_atomic}; +use crate::{ + changelog::Changelog, + state::{set_private, sync_dir, write_atomic}, +}; use inbuxa_relay_proto::{ seal::Sealer, time, @@ -23,14 +26,13 @@ use inbuxa_relay_proto::{ }; use serde::{Deserialize, Serialize}; use std::{ - collections::{BTreeMap, VecDeque}, + collections::BTreeMap, fs, io, path::{Path, PathBuf}, }; use tokio::{io::AsyncWriteExt, sync::oneshot}; -/// How many destroyed ids `/changes` can still report. -const TOMBSTONES: usize = 10_000; +pub use crate::changelog::Changes; /// A message's stored properties. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -74,16 +76,6 @@ impl Entry { } } -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Changes { - pub old_state: String, - pub new_state: String, - pub created: Vec, - pub updated: Vec, - pub destroyed: Vec, - pub has_more: bool, -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum VerdictError { NotFound, @@ -96,10 +88,7 @@ pub enum VerdictError { pub struct Spool { dir: PathBuf, entries: BTreeMap, - modseq: u64, - tombstones: VecDeque<(u64, String)>, - /// `/changes` can't go back past this. - floor: u64, + log: Changelog, bytes: u64, } @@ -129,9 +118,7 @@ impl Spool { let mut spool = Spool { dir: dir.to_path_buf(), entries: BTreeMap::new(), - modseq: 1, - tombstones: VecDeque::new(), - floor: 1, + log: Changelog::default(), bytes: 0, }; let mut msgs = Vec::new(); @@ -191,7 +178,7 @@ impl Spool { } pub fn state(&self) -> String { - self.modseq.to_string() + self.log.state() } pub fn count(&self) -> u64 { @@ -207,8 +194,7 @@ impl Spool { } fn bump(&mut self) -> u64 { - self.modseq += 1; - self.modseq + self.log.bump() } /// Starts writing a new message in the spool directory `dir`, sealed @@ -298,38 +284,11 @@ impl Spool { } pub fn changes(&self, since: &str, max: usize) -> Option { - let since: u64 = since.parse().ok()?; - if since < self.floor || since > self.modseq { - return None; - } - let mut created = Vec::new(); - let mut updated = Vec::new(); - for (id, e) in &self.entries { - if e.created > since { - created.push(id.clone()); - } else if e.modseq > since { - updated.push(id.clone()); - } - } - let destroyed: Vec = self - .tombstones - .iter() - .filter(|(seq, _)| *seq > since) - .map(|(_, id)| id.clone()) - .collect(); - let total = created.len() + updated.len() + destroyed.len(); - if total > max { - // Too much at once: the client starts over with /query. - return None; - } - Some(Changes { - old_state: since.to_string(), - new_state: self.state(), - created, - updated, - destroyed, - has_more: false, - }) + self.log.changes( + since, + max, + self.entries.iter().map(|(id, e)| (id, e.created, e.modseq)), + ) } /// RL-15: leases up to `limit` unleased messages to `conn`. @@ -438,13 +397,7 @@ impl Spool { let _ = fs::remove_file(self.meta_path(id)); let _ = fs::remove_file(self.msg_path(id)); let _ = sync_dir(&self.dir); - let modseq = self.bump(); - self.tombstones.push_back((modseq, id.to_string())); - while self.tombstones.len() > TOMBSTONES { - if let Some((seq, _)) = self.tombstones.pop_front() { - self.floor = seq; - } - } + self.log.destroyed(id); true } diff --git a/crates/relay/src/state.rs b/crates/relay/src/state.rs index 0e06379a..3b732a83 100644 --- a/crates/relay/src/state.rs +++ b/crates/relay/src/state.rs @@ -55,12 +55,21 @@ impl Paths { self.root.join("spool").join("in") } + pub fn spool_out(&self) -> PathBuf { + self.root.join("spool").join("out") + } + pub fn blobs(&self) -> PathBuf { self.root.join("blobs") } pub fn create_dirs(&self) -> io::Result<()> { - for dir in [self.root.clone(), self.spool_in(), self.blobs()] { + for dir in [ + self.root.clone(), + self.spool_in(), + self.spool_out(), + self.blobs(), + ] { fs::create_dir_all(&dir)?; set_private(&dir, true)?; } diff --git a/crates/relay/src/tests_e2e.rs b/crates/relay/src/tests_e2e.rs index 73bd684a..9acc8d00 100644 --- a/crates/relay/src/tests_e2e.rs +++ b/crates/relay/src/tests_e2e.rs @@ -622,6 +622,58 @@ async fn relay_end_to_end() { let got = ws.ok("Inbound/get", json!({ "ids": [held_id] })).await; assert_eq!(got["list"][0]["state"], "held"); + // Outbound: handed over as an upload (RL-19), never from a sender + // outside the relay's domains (RL-T10), and cancelled while queued. + let (status, body) = down + .http( + "POST", + "/jmap/upload/relay", + b"Subject: out\r\n\r\nhello\r\n".to_vec(), + ) + .await; + assert_eq!(status, 201); + let blob: Value = serde_json::from_slice(&body).unwrap(); + let envelope = |from: &str| json!({ "mailFrom": { "email": from }, "rcptTo": [{ "email": "bob@example.org" }] }); + let set = ws + .ok( + "Outbound/set", + json!({ "create": { + "a": { "blobId": blob["blobId"], "envelope": envelope("jane@example.com") }, + "b": { "blobId": blob["blobId"], "envelope": envelope("jane@example.org") }, + }}), + ) + .await; + assert_eq!(set["notCreated"]["b"]["type"], "forbiddenFrom", "{set}"); + let out_id = set["created"]["a"]["id"].as_str().unwrap().to_string(); + assert_eq!( + set["created"]["a"]["deliveryStatus"]["bob@example.org"]["delivered"], + "queued" + ); + ws.wait_change("Outbound").await; + let got = ws.ok("Outbound/get", json!({ "ids": [out_id] })).await; + assert_eq!(got["list"][0]["size"], 23); + let eml = app.paths.spool_out().join(format!("{out_id}.eml")); + assert!(eml.exists()); + let relay = ws.ok("Relay/get", json!({})).await; + assert_eq!(relay["list"][0]["status"]["spool"]["outbound"], 1); + let again = ws + .ok( + "Outbound/set", + json!({ + "create": { "c": { "blobId": blob["blobId"], "envelope": envelope("jane@example.com") }}, + "update": { out_id.clone(): { "attempts": 9 }}, + }), + ) + .await; + assert_eq!( + again["notCreated"]["c"]["type"], "invalidProperties", + "an upload is used once" + ); + assert_eq!(again["notUpdated"][&out_id]["type"], "forbidden"); + let gone = ws.ok("Outbound/set", json!({ "destroy": [out_id] })).await; + assert_eq!(gone["destroyed"], json!([out_id])); + assert!(!eml.exists()); + // Events went up the link. let events = ws.ok("RelayEvent/get", json!({})).await; let kinds: Vec<&str> = events["list"] diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index a6021f51..4b357273 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -306,8 +306,17 @@ so a reboot without the link comes back with the same settings (RL-22). | `id`, `createdAt` | | | `blobId` | The message as the downstream signed it | | `envelope` | RFC 8621 `Envelope` | -| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply`, `delivered` (`queued`, `yes`, `no`, `unknown`) | -| `nextAttemptAt`, `attempts`, `expiresAt` | Retry state | +| `size` | Bytes | +| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply` (empty before the first try), `delivered` (`queued`, `yes`, `no`, `unknown`) | +| `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 +update is allowed. The upload moves into the spool, so it can't be used +twice. Besides `forbiddenFrom` (RL-19), a create is refused as `tooLarge` +over `maxMessageSize` or `maxRecipients`, `overQuota` when the spool is +full, and `invalidProperties` for an empty `rcptTo`, a recipient listed +twice, or an address or SMTP parameter that couldn't go on a command line +as it is. **`RelayEvent`**: read-only, `{id, at, type, data}`. The relay's sessions, deliveries, rejections and errors, for the downstream's history and tracer. @@ -761,6 +770,10 @@ In progress. Done so far: link to a joined relay for an hour raises `relay.link-down` (a warning), once, and `relay.link-up` when it is back (RL-T23). Both reach the log, webhooks and OpenTelemetry. +- `Outbound` on the relay: `Outbound/set` create and destroy, `/get` and + `/changes`, with push, in `spool/out/` (RL-19, RL-T10). The relay doesn't + deliver it yet, and the downstream doesn't hand anything over yet; both + are the rest of step 4. - Not done yet: `inbuxa:RelayMessage`, which comes with inbuxa-admin's Relays page (step 5), once `Outbound` exists to show beside `Inbound`. -- 2.54.0