diff --git a/Cargo.lock b/Cargo.lock index 863b93aa..fa84a396 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3965,11 +3965,13 @@ dependencies = [ "serde_json", "sha2 0.11.1", "store", + "time", "tokio", "toml", "trc", "types", "utils", + "x509-parser", "xxhash-rust", "zip", ] @@ -7819,7 +7821,11 @@ dependencies = [ "futures", "groupware", "hkdf 0.13.0", + "http-body-util", + "hyper", + "hyper-util", "inbuxa-features", + "inbuxa-relay-proto", "jmap-tools", "jmap_proto", "mail-auth", @@ -7831,6 +7837,8 @@ dependencies = [ "psl", "registry", "reqwest 0.13.5", + "rustls", + "rustls-pki-types", "serde_json", "sha2 0.11.1", "smtp", @@ -7838,9 +7846,12 @@ dependencies = [ "spam-filter", "store", "tokio", + "tokio-rustls", + "tokio-tungstenite 0.30.0", "trc", "types", "utils", + "x509-parser", ] [[package]] diff --git a/crates/features/Cargo.toml b/crates/features/Cargo.toml index 1a9b1af1..a133c55b 100644 --- a/crates/features/Cargo.toml +++ b/crates/features/Cargo.toml @@ -34,7 +34,9 @@ chrono-tz = "0.10" # inbuxa: the downstream side of inbuxa Relay (relay spec) inbuxa-relay-proto = { path = "../relay-proto" } aws-lc-rs = { version = "1" } -rcgen = { version = "0.14", default-features = false, features = ["crypto", "pem", "aws_lc_rs"] } +rcgen = { version = "0.14", default-features = false, features = ["crypto", "pem", "aws_lc_rs", "x509-parser"] } +time = "0.3" [dev-dependencies] +x509-parser = "0.18" tokio = { version = "1.53", features = ["macros", "rt"] } diff --git a/crates/features/src/relay/link.rs b/crates/features/src/relay/link.rs new file mode 100644 index 00000000..bcace8df --- /dev/null +++ b/crates/features/src/relay/link.rs @@ -0,0 +1,424 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! What the link service needs that isn't the link itself (relay spec, +//! "The link service"): each node's link certificate, recording a join, +//! 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 inbuxa_relay_proto::{ + pin::{Pin, pins_match}, + recipient::{self, RecipientDomain, RecipientHash}, + types::{RelaySettings, RelayStatus, SmtpCertificate}, +}; +use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize}; +use std::collections::{BTreeMap, BTreeSet}; +use store::{ + Deserialize, IterateParams, SUBSPACE_INBUXA, Serialize, Store, ValueKey, + write::{AnyClass, BatchBuilder, ValueClass}, +}; +use trc::AddContext; + +/// A node's link certificate lasts this long (relay spec, "Identities")... +pub const NODE_CERT_LIFETIME: u64 = 30 * 86_400; +/// ...and is made again after this long. +pub const NODE_CERT_RENEW_AFTER: u64 = 20 * 86_400; + +const JOIN_ATTEMPTS: usize = 5; + +/// A node's link client certificate and key, as DER, issued by the link CA. +pub struct NodeCertificate { + pub cert_der: Vec, + pub key_der: Vec, + pub ca_der: Vec, + pub issued_at: u64, +} + +fn cert_error(err: rcgen::Error) -> trc::Error { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .into_err() + .details("Could not issue a relay link certificate") + .reason(err) +} + +/// A link client certificate for `node_id`, valid from an hour ago for +/// [`NODE_CERT_LIFETIME`]. Each node makes its own when it starts and +/// when the last one is old; it is never stored. +pub fn issue_node_certificate( + keys: &DownstreamKeys, + node_id: u64, + now: u64, +) -> trc::Result { + let ca_key = rcgen::KeyPair::from_pem(&keys.link_ca_key_pem).map_err(cert_error)?; + let issuer = + rcgen::Issuer::from_ca_cert_pem(&keys.link_ca_cert_pem, ca_key).map_err(cert_error)?; + let key = rcgen::KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256).map_err(cert_error)?; + let mut params = rcgen::CertificateParams::new(vec![format!("node-{node_id}.link.invalid")]) + .map_err(cert_error)?; + params + .distinguished_name + .push(rcgen::DnType::CommonName, format!("inbuxa node {node_id}")); + params.extended_key_usages = vec![rcgen::ExtendedKeyUsagePurpose::ClientAuth]; + params.key_usages = vec![rcgen::KeyUsagePurpose::DigitalSignature]; + let at = |seconds: u64| { + time::OffsetDateTime::from_unix_timestamp(seconds as i64) + .unwrap_or(time::OffsetDateTime::UNIX_EPOCH) + }; + params.not_before = at(now.saturating_sub(3600)); + params.not_after = at(now + NODE_CERT_LIFETIME); + let cert = params.signed_by(&key, &issuer).map_err(cert_error)?; + let ca_der = rustls_pem_der(&keys.link_ca_cert_pem).ok_or_else(|| { + trc::StoreEvent::DataCorruption + .into_err() + .details("The relay link CA certificate is unreadable") + })?; + Ok(NodeCertificate { + cert_der: cert.der().to_vec(), + key_der: key.serialize_der(), + ca_der, + issued_at: now, + }) +} + +/// The DER inside one PEM block. +fn rustls_pem_der(pem: &str) -> Option> { + use base64::Engine; + base64::engine::general_purpose::STANDARD + .decode( + pem.lines() + .filter(|line| !line.starts_with("-----")) + .collect::(), + ) + .ok() +} + +/// How a join came out, for the node that tried it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum JoinRecorded { + /// This node recorded the relay's pin. + Joined, + /// Another node already recorded this same pin. + AlreadyJoined, + /// The relay was given a new token, or joined with another key, meanwhile. + Superseded, +} + +/// Records that the relay holding `secret`'s token proved it, with its link +/// key's pin: from now on only that key is trusted (RL-7, RL-8). The token's +/// secret is deleted. +pub async fn record_join( + data: &Store, + relay_id: u32, + secret: &[u8; 32], + pin: Pin, + now: u64, +) -> trc::Result { + for _ in 0..JOIN_ATTEMPTS { + let Some((mut relay, version)) = get_versioned(data, relay_id).await? else { + return Ok(JoinRecorded::Superseded); + }; + if let Some(pinned) = relay.join.relay_pin { + return Ok(if pins_match(&pinned, &pin) { + JoinRecorded::AlreadyJoined + } else { + JoinRecorded::Superseded + }); + } + if relay + .join + .secret + .is_none_or(|stored| !pins_match(&stored, secret)) + { + return Ok(JoinRecorded::Superseded); + } + relay.join.secret = None; + relay.join.relay_pin = Some(pin); + relay.join.joined_at = Some(now); + if replace(data, &relay, version).await? { + return Ok(JoinRecorded::Joined); + } + } + Ok(JoinRecorded::Superseded) +} + +/// A domain this server has, as the recipient set needs it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LocalDomain { + /// The domain's name and its aliases, lowercased. + pub names: Vec, + /// Every address is accepted: a catch-all address, or relaying allowed. + pub catch_all: bool, + pub subaddress_separator: Option, +} + +/// Every domain the relay serves and every address it accepts, before +/// hashing. The same on every node, so every node sends the same set. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RecipientPlan { + pub domains: BTreeMap, + pub addresses: BTreeSet, +} + +impl RecipientPlan { + /// The plan for `relay`: its own domain list if it names one, else every + /// local domain. `addresses` are this server's, lowercased; those at a + /// domain the relay doesn't serve are left out. + pub fn new(relay: &Relay, local: &[LocalDomain], addresses: &[String]) -> Self { + let named: BTreeSet = relay + .settings + .domains + .iter() + .map(|domain| domain.trim_end_matches('.').to_ascii_lowercase()) + .collect(); + let mut domains = BTreeMap::new(); + for domain in local { + for name in &domain.names { + if named.is_empty() || named.contains(name) { + domains.insert( + name.clone(), + RecipientDomain { + domain: name.clone(), + catch_all: domain.catch_all, + subaddress_separator: domain.subaddress_separator.clone(), + }, + ); + } + } + } + let addresses = addresses + .iter() + .filter_map(|address| recipient::normalize(address, None)) + .filter(|address| { + address + .rsplit_once('@') + .is_some_and(|(_, domain)| domains.contains_key(domain)) + }) + .collect(); + RecipientPlan { domains, addresses } + } + + /// The hashes the relay holds, sorted, as `Recipient/replace` takes them. + pub fn hashes(&self, key: &[u8; 32]) -> Vec { + let mut hashes: Vec = self + .addresses + .iter() + .map(|address| recipient::hash(key, address)) + .collect(); + hashes.sort_unstable(); + hashes.dedup(); + hashes + } +} + +/// What the relay is sent with `Relay/set`: its settings as an +/// administrator made them, with the served domains, the keys and the +/// STARTTLS certificate added. +pub fn pushed_settings( + relay: &Relay, + keys: &DownstreamKeys, + plan: &RecipientPlan, +) -> RelaySettings { + let mut settings = relay.settings.clone(); + settings.domains = plan.domains.keys().cloned().collect(); + settings.sealing_key = Some(keys.sealing_public); + settings.recipient_key = Some(relay.recipient_key); + settings.smtp_certificate = match (&relay.smtp_certificate, &relay.smtp_private_key) { + (Some(chain_pem), Some(key_pem)) => Some(SmtpCertificate { + chain_pem: chain_pem.clone(), + key_pem: key_pem.clone(), + }), + _ => None, + }; + settings +} + +/// A node's link to a relay, as that node sees it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, SerdeSerialize, SerdeDeserialize)] +#[serde(rename_all = "camelCase")] +pub enum LinkState { + #[default] + Connecting, + /// Joining with the token. + Joining, + Linked, + Down, + /// The join can't go on: a wrong proof, or the token ran out. + JoinFailed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Default, SerdeSerialize, SerdeDeserialize)] +#[serde(rename_all = "camelCase", default)] +pub struct NodeLink { + pub node_id: u64, + pub state: LinkState, + /// When `state` began. + pub since: u64, + /// The last time the relay answered. + pub last_seen: Option, + pub last_error: Option, + /// What the relay last said of itself. + pub relay: Option, + /// How many recipients this node last sent. + pub recipients: Option, + pub updated_at: u64, +} + +fn status_class(relay_id: u32, node_id: u64) -> ValueClass { + let mut key = Vec::with_capacity(14); + key.push(FEATURE); + key.push(KIND_STATUS); + key.extend_from_slice(&relay_id.to_be_bytes()); + key.extend_from_slice(&node_id.to_be_bytes()); + ValueClass::Any(AnyClass { + subspace: SUBSPACE_INBUXA, + key, + }) +} + +pub async fn set_status(data: &Store, relay_id: u32, link: &NodeLink) -> trc::Result<()> { + let mut batch = BatchBuilder::new(); + batch.set( + status_class(relay_id, link.node_id), + Json(link).serialize()?, + ); + data.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + Ok(()) +} + +/// Every node's link to `relay_id`, by node. +pub async fn statuses(data: &Store, relay_id: u32) -> trc::Result> { + let mut links = Vec::new(); + data.iterate( + IterateParams::new( + ValueKey::from(status_class(relay_id, 0)), + ValueKey::from(status_class(relay_id, u64::MAX)), + ), + |_, value| { + if let Ok(Json(link)) = Json::::deserialize(value) { + links.push(link); + } + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + links.sort_by_key(|link| link.node_id); + Ok(links) +} + +/// Forgets every node's link to a relay that is gone. +pub(super) fn clear_statuses(batch: &mut BatchBuilder, relay_id: u32, nodes: &[NodeLink]) { + for link in nodes { + batch.clear(status_class(relay_id, link.node_id)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::relay::{Join, new_downstream_keys}; + use inbuxa_relay_proto::time; + + fn relay(domains: &[&str]) -> Relay { + Relay { + id: 1, + name: "Edge".into(), + description: String::new(), + address: "relay.example.net".into(), + link_port: 443, + enabled: true, + send_through: false, + inbound_only: false, + settings: RelaySettings { + hostname: "mx1.example.com".into(), + domains: domains.iter().map(|d| d.to_string()).collect(), + ..Default::default() + }, + smtp_certificate: None, + smtp_private_key: None, + recipient_key: [9; 32], + join: Join::default(), + created_by: String::new(), + created_at: 0, + updated_at: 0, + } + } + + fn local() -> Vec { + vec![ + LocalDomain { + names: vec!["example.com".into(), "example.net".into()], + catch_all: false, + subaddress_separator: Some("+".into()), + }, + LocalDomain { + names: vec!["other.org".into()], + catch_all: true, + subaddress_separator: None, + }, + ] + } + + #[test] + fn plans_recipients() { + let addresses = vec![ + "Jane@Example.com".to_string(), + "jane@example.net".into(), + "list@other.org".into(), + "stray@elsewhere.com".into(), + ]; + let all = RecipientPlan::new(&relay(&[]), &local(), &addresses); + assert_eq!( + all.domains.keys().collect::>(), + ["example.com", "example.net", "other.org"] + ); + assert!(all.domains["other.org"].catch_all); + assert_eq!(all.addresses.len(), 3); + assert!(all.addresses.contains("jane@example.com")); + + // A relay naming its domains serves only those + let one = RecipientPlan::new(&relay(&["Example.COM."]), &local(), &addresses); + assert_eq!(one.domains.keys().collect::>(), ["example.com"]); + assert_eq!( + one.addresses.iter().collect::>(), + ["jane@example.com"] + ); + + // The hashes are what the relay looks up + let hashes = one.hashes(&[9; 32]); + assert_eq!(hashes, vec![recipient::hash(&[9; 32], "jane@example.com")]); + let keys = new_downstream_keys(time::now()).unwrap(); + let settings = pushed_settings(&relay(&["example.com"]), &keys, &one); + assert_eq!(settings.domains, ["example.com"]); + assert_eq!(settings.sealing_key, Some(keys.sealing_public)); + assert_eq!(settings.recipient_key, Some([9; 32])); + assert!(settings.problems().is_empty(), "{:?}", settings.problems()); + } + + #[test] + fn issues_node_certificates() { + let now = time::now(); + let keys = new_downstream_keys(now).unwrap(); + let cert = issue_node_certificate(&keys, 7, now).unwrap(); + // Chained to the CA the relay pins + let (_, parsed) = x509_parser::parse_x509_certificate(&cert.cert_der).unwrap(); + let (_, ca) = x509_parser::parse_x509_certificate(&cert.ca_der).unwrap(); + assert_eq!(parsed.issuer(), ca.subject()); + assert!(parsed.verify_signature(Some(ca.public_key())).is_ok()); + assert_eq!( + inbuxa_relay_proto::pin::spki_pin(ca.tbs_certificate.subject_pki.raw), + keys.link_ca_pin + ); + let validity = parsed.validity(); + assert_eq!( + validity.not_after.timestamp() as u64, + now + NODE_CERT_LIFETIME + ); + } +} diff --git a/crates/features/src/relay/mod.rs b/crates/features/src/relay/mod.rs index 8e79028b..48e14e00 100644 --- a/crates/features/src/relay/mod.rs +++ b/crates/features/src/relay/mod.rs @@ -15,9 +15,13 @@ //! - `k`: the downstream's keys ([`DownstreamKeys`]), as JSON. One per //! server or cluster, made with the first relay. //! - `r` + relay id (u32): a relay ([`Relay`]), as JSON. +//! - `s` + relay id (u32) + node id (u64): that node's link to the relay +//! ([`link::NodeLink`]), as JSON. //! //! Numbers are big-endian. There are few relays, so they're read whole. +pub mod link; + use inbuxa_relay_proto::{ b64, pin::{Pin, spki_pin}, @@ -33,9 +37,20 @@ use store::{ use trc::AddContext; use types::id::Id; +/// Wakes the link service on this node when a relay is added, changed or +/// removed here. Other nodes see it at their next scan. +pub static CHANGED: std::sync::LazyLock = + std::sync::LazyLock::new(tokio::sync::Notify::new); + +/// Tells this node's link service that the relays changed. +pub fn changed() { + CHANGED.notify_one(); +} + const FEATURE: u8 = b'E'; const KIND_KEYS: u8 = b'k'; const KIND_RELAY: u8 = b'r'; +const KIND_STATUS: u8 = b's'; const CREATE_ATTEMPTS: usize = 5; const MAX_NAME: usize = 100; @@ -545,9 +560,12 @@ pub async fn replace(data: &Store, relay: &Relay, version: u64) -> trc::Result trc::Result<()> { + let nodes = link::statuses(data, id).await?; let mut batch = BatchBuilder::new(); batch.clear(class(KIND_RELAY, Some(id))); + link::clear_statuses(&mut batch, id, &nodes); data.write(batch.build_all()) .await .caused_by(trc::location!())?; diff --git a/crates/jmap-proto/src/object/inbuxa_relay.rs b/crates/jmap-proto/src/object/inbuxa_relay.rs index c7584249..52367c62 100644 --- a/crates/jmap-proto/src/object/inbuxa_relay.rs +++ b/crates/jmap-proto/src/object/inbuxa_relay.rs @@ -46,6 +46,8 @@ pub enum RelayProperty { Rejoin, /// In a create or rejoin response only. JoinToken, + /// Each node's link to the relay, and what the relay last said. + Links, CreatedBy, CreatedAt, UpdatedAt, @@ -84,6 +86,7 @@ impl Property for RelayProperty { RelayProperty::LinkKeyPin => "linkKeyPin", RelayProperty::Rejoin => "rejoin", RelayProperty::JoinToken => "joinToken", + RelayProperty::Links => "links", RelayProperty::CreatedBy => "createdBy", RelayProperty::CreatedAt => "createdAt", RelayProperty::UpdatedAt => "updatedAt", @@ -112,6 +115,7 @@ impl RelayProperty { b"linkKeyPin" => RelayProperty::LinkKeyPin, b"rejoin" => RelayProperty::Rejoin, b"joinToken" => RelayProperty::JoinToken, + b"links" => RelayProperty::Links, b"createdBy" => RelayProperty::CreatedBy, b"createdAt" => RelayProperty::CreatedAt, b"updatedAt" => RelayProperty::UpdatedAt, diff --git a/crates/jmap/src/inbuxa/relay.rs b/crates/jmap/src/inbuxa/relay.rs index 36a47750..64f303b0 100644 --- a/crates/jmap/src/inbuxa/relay.rs +++ b/crates/jmap/src/inbuxa/relay.rs @@ -12,7 +12,7 @@ //! change in the audit log, with the private key and the token left out. use common::{Server, auth::AccessToken}; -use inbuxa_features::relay::{self as relays, JoinState, Relay as StoredRelay}; +use inbuxa_features::relay::{self as relays, JoinState, Relay as StoredRelay, link::NodeLink}; use inbuxa_relay_proto::{b64, types::RelaySettings}; use jmap_proto::{ error::set::SetError, @@ -52,6 +52,7 @@ const ALL: &[P] = &[ P::JoinExpiresAt, P::JoinedAt, P::LinkKeyPin, + P::Links, P::CreatedBy, P::CreatedAt, P::UpdatedAt, @@ -65,6 +66,7 @@ const SERVER_SET: &[P] = &[ P::JoinedAt, P::LinkKeyPin, P::JoinToken, + P::Links, P::CreatedBy, P::CreatedAt, P::UpdatedAt, @@ -115,7 +117,24 @@ fn text(value: Option<&str>) -> RValue { value.map_or(Value::Null, |value| Value::Str(value.to_string().into())) } -fn to_value(relay: &StoredRelay, properties: &[P], now: u64) -> RValue { +/// Each node's link, with its times as dates. +fn links_value(links: &[NodeLink]) -> RValue { + let date = |at: u64| UTCDate::from_timestamp(at as i64).to_string(); + let list = links + .iter() + .map(|link| { + let mut json = serde_json::to_value(link).unwrap_or_default(); + json["nodeId"] = link.node_id.to_string().into(); + json["since"] = date(link.since).into(); + json["updatedAt"] = date(link.updated_at).into(); + json["lastSeen"] = link.last_seen.map(date).into(); + json + }) + .collect::>(); + json_to_value(serde_json::Value::Array(list)) +} + +fn to_value(relay: &StoredRelay, links: &[NodeLink], properties: &[P], now: u64) -> RValue { let state = relay.join.state(now); let mut out = Map::with_capacity(properties.len()); for property in properties { @@ -139,6 +158,7 @@ fn to_value(relay: &StoredRelay, properties: &[P], now: u64) -> RValue { P::JoinExpiresAt if state != JoinState::Joined => date(relay.join.expires_at), P::JoinedAt => relay.join.joined_at.map_or(Value::Null, date), P::LinkKeyPin => text(relay.join.relay_pin.map(|pin| b64::encode(&pin)).as_deref()), + P::Links => links_value(links), P::CreatedBy => text(Some(&relay.created_by)), P::CreatedAt => date(relay.created_at), P::UpdatedAt => date(relay.updated_at), @@ -314,23 +334,31 @@ pub async fn get( list: Vec::new(), not_found, }; - let all = relays::all(server.store()).await?; + let data = server.store(); + let all = relays::all(data).await?; let now = now(); - match ids { - None => { - response.list = all - .iter() - .map(|relay| to_value(relay, &properties, now)) - .collect() - } + let wanted: Vec<&StoredRelay> = match ids { + None => all.iter().collect(), Some(ids) => { + let mut wanted = Vec::with_capacity(ids.len()); for id in ids { match relay_id(id).and_then(|id| all.iter().find(|r| r.id == id)) { - Some(relay) => response.list.push(to_value(relay, &properties, now)), + Some(relay) => wanted.push(relay), None => response.push_not_found(id), } } + wanted } + }; + for relay in wanted { + let links = if properties.contains(&P::Links) { + relays::link::statuses(data, relay.id).await? + } else { + Vec::new() + }; + response + .list + .push(to_value(relay, &links, &properties, now)); } Ok(response) } @@ -460,5 +488,11 @@ pub async fn set( response.destroyed.push(id); } + if !response.created.is_empty() + || !response.updated.is_empty() + || !response.destroyed.is_empty() + { + relays::changed(); + } Ok(response) } diff --git a/crates/services/Cargo.toml b/crates/services/Cargo.toml index 17c52e64..9d4bfe0d 100644 --- a/crates/services/Cargo.toml +++ b/crates/services/Cargo.toml @@ -19,7 +19,7 @@ jmap-tools = { version = "0.1" } registry = { path = "../registry" } inbuxa-features = { path = "../features" } smtp-proto = { version = "0.2", features = ["rkyv", "serde"] } -tokio = { version = "1.53", features = ["rt"] } +tokio = { version = "1.53", features = ["rt", "net", "time", "sync", "macros"] } mail-parser = { version = "0.11", features = ["full_encoding", "rkyv"] } mail-builder = { version = "1.0" } aho-corasick = { version = "1.1" } @@ -37,6 +37,16 @@ psl = "2" # inbuxa: the deliverability check mail-auth = { version = "0.13" } futures = "0.3" +# inbuxa: links to inbuxa Relays (relay spec) +inbuxa-relay-proto = { path = "../relay-proto" } +rustls = { version = "0.23.43", default-features = false, features = ["std", "aws_lc_rs", "tls12"] } +rustls-pki-types = { version = "1" } +tokio-rustls = { version = "0.26", default-features = false, features = ["aws_lc_rs", "tls12"] } +tokio-tungstenite = "0.30" +hyper = { version = "1.11.1", features = ["client", "http1"] } +hyper-util = { version = "0.1.20", features = ["tokio"] } +http-body-util = "0.1" +x509-parser = "0.18" [dev-dependencies] diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs new file mode 100644 index 00000000..727d36f5 --- /dev/null +++ b/crates/services/src/inbuxa_relay_link.rs @@ -0,0 +1,1062 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! The relay link service (docs/spec/features/relay.md, "The link +//! service"): on every node, a link to every enabled inbuxa Relay. +//! +//! The node dials the relay; the relay never dials in. A relay waiting to +//! join gets `Relay/join` on a connection that can't check its key yet, and +//! its proof decides whether its key is pinned (RL-7, RL-8). After that the +//! node only talks to that key, over a WebSocket (RFC 8887), and keeps the +//! 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. + +use common::{BuildServer, Inner, Server}; +use futures::{SinkExt, StreamExt}; +use http_body_util::{BodyExt, Full}; +use hyper::body::Bytes; +use hyper_util::rt::TokioIo; +use inbuxa_features::{ + masked_email, + relay::{ + 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, + }, + }, +}; +use inbuxa_relay_proto::{ + CAPABILITY, CORE_CAPABILITY, PROTOCOL_VERSION, b64, + jmap::{ClientFrame, Invocation, Request, ServerFrame, Session, method}, + pin::{Pin, pins_match, spki_pin}, + recipient::{RecipientDomain, RecipientHash}, + token, + types::{Capability, RELAY_SINGLETON, RelaySettings, RelayStatus}, +}; +use registry::schema::{ + prelude::ObjectType, + structs::{Domain, SubAddressing}, +}; +use rustls::{ + DigitallySignedStruct, SignatureScheme, + client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}, +}; +use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}; +use serde_json::{Value, json}; +use std::{ + collections::{BTreeMap, BTreeSet, HashMap}, + sync::{Arc, Mutex}, + time::Duration, +}; +use store::{registry::RegistryQuery, write::now}; +use tokio::{net::TcpStream, sync::watch, task::JoinHandle}; +use tokio_rustls::{TlsConnector, client::TlsStream}; +use tokio_tungstenite::{WebSocketStream, tungstenite::Message}; +use types::id::Id; + +/// How often the list of relays is read again. +const SCAN: Duration = Duration::from_secs(10); +/// How often a linked relay is brought up to date, and asked how it is. +const SYNC: Duration = Duration::from_secs(15); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); +const CALL_TIMEOUT: Duration = Duration::from_secs(60); +const BACKOFF_MIN: Duration = Duration::from_secs(5); +const BACKOFF_MAX: Duration = Duration::from_secs(120); +/// After a join that can't go on, wait this long, or for a change. +const JOIN_FAILED_WAIT: Duration = Duration::from_secs(300); +/// Status is written when it changes, and at least this often. +const STATUS_EVERY: u64 = 60; +/// More recipient changes than this go as a full `Recipient/replace`. +const MAX_DELTA: usize = 2_000; +/// The relay's only account. +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"; + +pub fn spawn_relay_links(inner: Arc) { + tokio::spawn(async move { + let certs = Arc::new(NodeCertificates::default()); + let mut links: HashMap = HashMap::new(); + loop { + let server = inner.build_server(); + if let Err(err) = reconcile(&inner, &server, &certs, &mut links).await { + trc::error!(err.details("Failed to start relay links")); + } + let woken = tokio::select! { + _ = tokio::time::sleep(SCAN) => false, + _ = relays::CHANGED.notified() => true, + }; + if woken { + for link in links.values() { + link.wake.send_modify(|n| *n = n.wrapping_add(1)); + } + } + } + }); +} + +struct Running { + handle: JoinHandle<()>, + /// Where the link goes: a change starts it again. + target: (String, u16), + wake: watch::Sender, +} + +/// Starts a link for every enabled relay, and stops those for relays gone, +/// turned off, or moved. +async fn reconcile( + inner: &Arc, + server: &Server, + certs: &Arc, + links: &mut HashMap, +) -> trc::Result<()> { + let data = server.store(); + let wanted: HashMap = match relays::keys(data).await? { + Some(_) => relays::all(data) + .await? + .into_iter() + .filter(|relay| relay.enabled) + .map(|relay| (relay.id, (relay.address.clone(), relay.link_port))) + .collect(), + None => HashMap::new(), + }; + links.retain(|id, link| match wanted.get(id) { + Some(target) if *target == link.target && !link.handle.is_finished() => true, + // Moved: start over at the new address + Some(_) => { + link.handle.abort(); + false + } + // Gone or turned off: the link sees it, says so, and ends + None => { + link.wake.send_modify(|n| *n = n.wrapping_add(1)); + false + } + }); + for (id, target) in wanted { + if links.contains_key(&id) { + continue; + } + let (wake, woken) = watch::channel(0); + let handle = tokio::spawn(run_link(inner.clone(), id, certs.clone(), woken)); + links.insert( + id, + Running { + handle, + target, + wake, + }, + ); + } + Ok(()) +} + +/// This node's link certificate, made again once it is old. +#[derive(Default)] +struct NodeCertificates(Mutex>>); + +impl NodeCertificates { + fn get(&self, keys: &DownstreamKeys, node_id: u64) -> trc::Result> { + let now = now(); + let mut current = self.0.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(cert) = current.as_ref() + && cert.issued_at + NODE_CERT_RENEW_AFTER > now + { + return Ok(cert.clone()); + } + let cert = Arc::new(issue_node_certificate(keys, node_id, now)?); + *current = Some(cert.clone()); + Ok(cert) + } +} + +/// Why a link went down, and how the node shows it. +struct LinkError { + state: LinkState, + message: String, +} + +impl LinkError { + fn down(message: impl Into) -> Self { + LinkError { + state: LinkState::Down, + message: message.into(), + } + } + + fn join_failed(message: impl Into) -> Self { + LinkError { + state: LinkState::JoinFailed, + message: message.into(), + } + } +} + +impl From for LinkError { + fn from(err: trc::Error) -> Self { + LinkError::down(format!("This server failed: {err}")) + } +} + +/// One relay's link, for as long as the relay is enabled. +async fn run_link( + inner: Arc, + relay_id: u32, + certs: Arc, + mut woken: watch::Receiver, +) { + let mut backoff = BACKOFF_MIN; + let mut status = StatusWriter::new(inner.build_server().core.network.node_id); + loop { + let server = inner.build_server(); + let data = server.store().clone(); + let relay = match relays::get(&data, relay_id).await { + Ok(Some(relay)) if relay.enabled => relay, + Ok(Some(_)) => { + status + .set( + &data, + relay_id, + LinkState::Down, + Some("The relay is turned off.".into()), + ) + .await; + return; + } + Ok(None) => return, + Err(err) => { + trc::error!(err.details("Failed to read a relay")); + tokio::time::sleep(backoff).await; + continue; + } + }; + let outcome = match link_keys(&server, &certs).await { + Ok((keys, cert)) => match relay.join.state(now()) { + JoinState::Waiting => { + status.set(&data, relay_id, LinkState::Joining, None).await; + join(&data, &relay, &cert).await + } + JoinState::Expired => Err(LinkError::join_failed( + "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 + } + }, + Err(err) => Err(err), + }; + let wait = match outcome { + Ok(()) => { + backoff = BACKOFF_MIN; + Duration::ZERO + } + Err(err) => { + let wait = if err.state == LinkState::JoinFailed { + JOIN_FAILED_WAIT + } else { + backoff + }; + status + .set(&data, relay_id, err.state, Some(err.message)) + .await; + backoff = (backoff * 2).min(BACKOFF_MAX); + wait + } + }; + if !wait.is_zero() { + tokio::select! { + _ = tokio::time::sleep(wait) => {} + _ = woken.changed() => {} + } + } + } +} + +async fn link_keys( + server: &Server, + certs: &NodeCertificates, +) -> Result<(DownstreamKeys, Arc), LinkError> { + let keys = relays::keys(server.store()) + .await? + .ok_or_else(|| LinkError::down("This server has no relay keys yet."))?; + let cert = certs.get(&keys, server.core.network.node_id)?; + Ok((keys, cert)) +} + +/// Writes this node's view of a link when it changes, and at least every +/// [`STATUS_EVERY`] seconds while it holds. +struct StatusWriter { + current: NodeLink, + written_at: u64, +} + +impl StatusWriter { + fn new(node_id: u64) -> Self { + StatusWriter { + current: NodeLink { + node_id, + ..Default::default() + }, + written_at: 0, + } + } + + async fn set( + &mut self, + data: &store::Store, + relay_id: u32, + state: LinkState, + error: Option, + ) { + let now = now(); + if state != self.current.state { + self.current.since = now; + } + let changed = state != self.current.state || error != self.current.last_error; + self.current.state = state; + if error.is_some() || state == LinkState::Linked { + self.current.last_error = error; + } + self.write(data, relay_id, changed, now).await; + } + + async fn seen( + &mut self, + data: &store::Store, + relay_id: u32, + relay: Option, + recipients: u64, + ) { + let now = now(); + let changed = relay != self.current.relay || Some(recipients) != self.current.recipients; + self.current.last_seen = Some(now); + if relay.is_some() { + self.current.relay = relay; + } + self.current.recipients = Some(recipients); + self.write(data, relay_id, changed, now).await; + } + + async fn write(&mut self, data: &store::Store, relay_id: u32, changed: bool, now: u64) { + if !changed && now < self.written_at + STATUS_EVERY { + return; + } + self.current.updated_at = now; + match set_status(data, relay_id, &self.current).await { + Ok(()) => self.written_at = now, + Err(err) => { + trc::error!(err.details("Failed to record a relay link")); + } + } + } +} + +// ---- TLS ------------------------------------------------------------------ + +/// Checks the relay's certificate against its pinned key, or, while it +/// joins, takes any key and keeps it for the proof (RL-7). +#[derive(Debug)] +struct RelayKey { + pin: Option, + seen: Mutex>, + algorithms: rustls::crypto::WebPkiSupportedAlgorithms, +} + +impl ServerCertVerifier for RelayKey { + fn verify_server_cert( + &self, + end_entity: &CertificateDer<'_>, + _: &[CertificateDer<'_>], + _: &ServerName<'_>, + _: &[u8], + _: UnixTime, + ) -> Result { + let pin = x509_parser::parse_x509_certificate(end_entity) + .map(|(_, cert)| spki_pin(cert.tbs_certificate.subject_pki.raw)) + .map_err(|_| { + rustls::Error::InvalidCertificate(rustls::CertificateError::BadEncoding) + })?; + if let Some(pinned) = &self.pin + && !pins_match(pinned, &pin) + { + return Err(rustls::Error::InvalidCertificate( + rustls::CertificateError::ApplicationVerificationFailure, + )); + } + *self.seen.lock().unwrap_or_else(|e| e.into_inner()) = Some(pin); + Ok(ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _: &[u8], + _: &CertificateDer<'_>, + _: &DigitallySignedStruct, + ) -> Result { + // The link is TLS 1.3 only + Err(rustls::Error::PeerIncompatible( + rustls::PeerIncompatible::Tls12NotOffered, + )) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls13_signature(message, cert, dss, &self.algorithms) + } + + fn supported_verify_schemes(&self) -> Vec { + self.algorithms.supported_schemes() + } +} + +/// Dials one relay: TLS 1.3, this node's link certificate, and the relay's +/// key checked against `pin` when there is one. +struct Dialer { + address: String, + port: u16, + connector: TlsConnector, + key: Arc, +} + +impl Dialer { + fn new(relay: &Relay, cert: &NodeCertificate, pin: Option) -> Result { + let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider()); + let key = Arc::new(RelayKey { + pin, + seen: Mutex::new(None), + algorithms: provider.signature_verification_algorithms, + }); + let private_key = PrivateKeyDer::try_from(cert.key_der.clone()) + .map_err(|err| LinkError::down(format!("The node's link key is unreadable: {err}")))?; + let mut config = rustls::ClientConfig::builder_with_provider(provider) + .with_protocol_versions(&[&rustls::version::TLS13]) + .map_err(|err| LinkError::down(err.to_string()))? + .dangerous() + .with_custom_certificate_verifier(key.clone()) + .with_client_auth_cert( + vec![ + CertificateDer::from(cert.cert_der.clone()), + CertificateDer::from(cert.ca_der.clone()), + ], + private_key, + ) + .map_err(|err| LinkError::down(err.to_string()))?; + config.alpn_protocols = vec![b"http/1.1".to_vec()]; + Ok(Dialer { + address: relay.address.trim().to_string(), + port: relay.link_port, + connector: TlsConnector::from(Arc::new(config)), + key, + }) + } + + async fn connect(&self) -> Result, LinkError> { + let host = self.address.trim_start_matches('[').trim_end_matches(']'); + let connect = async { + let tcp = TcpStream::connect((host, self.port)).await?; + self.connector + .connect( + ServerName::try_from(SERVER_NAME).expect("a valid name"), + tcp, + ) + .await + }; + match tokio::time::timeout(CONNECT_TIMEOUT, connect).await { + Ok(Ok(tls)) => Ok(tls), + Ok(Err(err)) => Err(LinkError::down(match self.key.pin { + Some(_) if err.to_string().contains("ApplicationVerificationFailure") => { + "The relay answered with another key than the one pinned at join. \ + If it was reset, start a new join." + .to_string() + } + _ => format!("Could not reach the relay: {err}"), + })), + Err(_) => Err(LinkError::down("Could not reach the relay in time.")), + } + } + + fn seen_pin(&self) -> Option { + *self.key.seen.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// One HTTP/1.1 request on its own connection. + async fn http( + &self, + method: &str, + path: &str, + content_type: &str, + body: Vec, + ) -> Result<(u16, Vec), LinkError> { + let tls = self.connect().await?; + let failed = + |err: &dyn std::fmt::Display| LinkError::down(format!("The relay failed: {err}")); + let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(tls)) + .await + .map_err(|e| failed(&e))?; + tokio::spawn(conn); + let request = hyper::Request::builder() + .method(method) + .uri(path) + .header("Host", SERVER_NAME) + .header("Content-Type", content_type) + .body(Full::new(Bytes::from(body))) + .map_err(|e| failed(&e))?; + let exchange = async { + let response = sender.send_request(request).await.map_err(|e| failed(&e))?; + let status = response.status().as_u16(); + let body = response + .into_body() + .collect() + .await + .map_err(|e| failed(&e))? + .to_bytes() + .to_vec(); + Ok((status, body)) + }; + tokio::time::timeout(CALL_TIMEOUT, exchange) + .await + .map_err(|_| LinkError::down("The relay didn't answer in time."))? + } + + /// The relay's JMAP session, refused if its link protocol is one this + /// server doesn't speak (RL-28). + async fn check_session(&self) -> Result { + let (status, body) = self + .http("GET", "/jmap/session", "application/json", Vec::new()) + .await?; + if status != 200 { + return Err(LinkError::down(format!( + "The relay answered {status} for its session." + ))); + } + let capability = serde_json::from_slice::(&body) + .ok() + .and_then(|session| session.capabilities.get(CAPABILITY).cloned()) + .and_then(|value| serde_json::from_value::(value).ok()) + .ok_or_else(|| LinkError::down("The relay's session is unreadable."))?; + if capability.protocol_version != PROTOCOL_VERSION { + return Err(LinkError::down(format!( + "The relay speaks link protocol {} and this server speaks {}: \ + update whichever is older.", + capability.protocol_version, PROTOCOL_VERSION + ))); + } + Ok(capability) + } + + async fn api(&self, calls: Vec) -> Result, LinkError> { + let request = Request { + using: vec![CORE_CAPABILITY.into(), CAPABILITY.into()], + method_calls: calls, + created_ids: None, + }; + let body = serde_json::to_vec(&request).unwrap_or_default(); + let (status, body) = self + .http("POST", "/jmap/api", "application/json", body) + .await?; + if status != 200 { + return Err(LinkError::down(format!( + "The relay answered {status}: {}", + String::from_utf8_lossy(&body) + ))); + } + serde_json::from_slice::(&body) + .map(|response| response.method_responses) + .map_err(|_| LinkError::down("The relay's answer is unreadable.")) + } + + /// Uploads `body`; the blob's id. + async fn upload(&self, body: Vec) -> Result { + let (status, body) = self + .http( + "POST", + "/jmap/upload/relay", + "application/octet-stream", + body, + ) + .await?; + if status != 200 && status != 201 { + return Err(LinkError::down(format!( + "The relay refused an upload ({status})." + ))); + } + serde_json::from_slice::(&body) + .ok() + .and_then(|v| v["blobId"].as_str().map(str::to_string)) + .ok_or_else(|| LinkError::down("The relay's upload answer is unreadable.")) + } +} + +// ---- Joining -------------------------------------------------------------- + +/// Joins a relay waiting with its token (relay spec, "Joining and reset", +/// step 4): one call, `Relay/join`, on a connection whose key is checked by +/// the answer instead. +async fn join(data: &store::Store, relay: &Relay, cert: &NodeCertificate) -> Result<(), LinkError> { + let Some(secret) = relay.join.secret else { + return Ok(()); + }; + let dialer = Dialer::new(relay, cert, None)?; + let nonce = relays::random_key(); + let responses = dialer + .api(vec![Invocation::new( + method::RELAY_JOIN, + json!({"accountId": ACCOUNT_ID, "nonce": b64::encode(&nonce)}), + "j", + )]) + .await?; + let seen = dialer + .seen_pin() + .ok_or_else(|| LinkError::down("The relay showed no key."))?; + let answer = responses + .into_iter() + .next() + .ok_or_else(|| LinkError::down("The relay didn't answer the join."))?; + if answer.name() == "error" { + // Another node may have joined it a moment ago: read the relay again + return Err(LinkError::down(format!( + "The relay refused the join: {}", + answer.args()["description"].as_str().unwrap_or("no reason") + ))); + } + let proof = answer.args()["proof"] + .as_str() + .and_then(b64::decode) + .unwrap_or_default(); + let claimed = answer.args()["relayPin"].as_str().and_then(b64::decode); + if claimed.as_deref() != Some(&seen[..]) + || !token::verify_join_proof(&secret, &nonce, &seen, &proof) + { + return Err(LinkError::join_failed( + "The relay's answer didn't prove it holds this relay's token. \ + Reset the relay and start a new join.", + )); + } + // Whichever way it went, the relay is read again: joined by this node + // or another, or given a new token meanwhile + let _: JoinRecorded = record_join(data, relay.id, &secret, seen, now()).await?; + Ok(()) +} + +// ---- The linked session --------------------------------------------------- + +/// A WebSocket to the relay, with the `Relay` changes it pushed meanwhile. +struct Ws { + stream: WebSocketStream>, + next_id: u64, + relay_changed: bool, +} + +impl Ws { + async fn open(dialer: &Dialer) -> Result { + let tls = dialer.connect().await?; + let request = tokio_tungstenite::tungstenite::http::Request::builder() + .uri(format!("wss://{SERVER_NAME}/jmap/ws")) + .header("Host", SERVER_NAME) + .header("Connection", "Upgrade") + .header("Upgrade", "websocket") + .header("Sec-WebSocket-Version", "13") + .header( + "Sec-WebSocket-Key", + tokio_tungstenite::tungstenite::handshake::client::generate_key(), + ) + .header("Sec-WebSocket-Protocol", "jmap") + .body(()) + .map_err(|err| LinkError::down(err.to_string()))?; + let (stream, _) = tokio::time::timeout( + CONNECT_TIMEOUT, + tokio_tungstenite::client_async(request, tls), + ) + .await + .map_err(|_| LinkError::down("The relay's WebSocket didn't open in time."))? + .map_err(|err| LinkError::down(format!("The relay's WebSocket didn't open: {err}")))?; + Ok(Ws { + stream, + next_id: 0, + relay_changed: false, + }) + } + + async fn send(&mut self, frame: &ClientFrame) -> Result<(), LinkError> { + let text = serde_json::to_string(frame).unwrap_or_default(); + self.stream + .send(Message::Text(text.into())) + .await + .map_err(|err| LinkError::down(format!("The link dropped: {err}"))) + } + + /// The next frame from the relay, `None` once it has gone. + async fn frame(&mut self) -> Result { + loop { + match self.stream.next().await { + Some(Ok(Message::Text(text))) => { + if let Ok(frame) = serde_json::from_str::(&text) { + return Ok(frame); + } + } + Some(Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_))) => {} + Some(Ok(Message::Binary(_))) => {} + Some(Ok(Message::Close(_))) | None => { + return Err(LinkError::down("The relay closed the link.")); + } + Some(Err(err)) => return Err(LinkError::down(format!("The link dropped: {err}"))), + } + } + } + + fn note(&mut self, frame: &ServerFrame) { + if let ServerFrame::StateChange { changed, .. } = frame + && changed.values().any(|types| types.contains_key("Relay")) + { + self.relay_changed = true; + } + } + + /// Sends `calls` and waits for their answers. + async fn call(&mut self, calls: Vec) -> Result, LinkError> { + self.next_id += 1; + let id = format!("r{}", self.next_id); + self.send(&ClientFrame::Request { + id: Some(id.clone()), + request: Request { + using: vec![CORE_CAPABILITY.into(), CAPABILITY.into()], + method_calls: calls, + created_ids: None, + }, + }) + .await?; + let wait = async { + loop { + let frame = self.frame().await?; + match frame { + ServerFrame::Response { + request_id, + response, + } if request_id.as_deref() == Some(id.as_str()) => { + return Ok(response.method_responses); + } + ServerFrame::RequestError { + request_id, detail, .. + } if request_id.as_deref() == Some(id.as_str()) => { + return Err(LinkError::down(format!( + "The relay refused a request: {}", + detail.unwrap_or_default() + ))); + } + other => self.note(&other), + } + } + }; + tokio::time::timeout(CALL_TIMEOUT, wait) + .await + .map_err(|_| LinkError::down("The relay didn't answer in time."))? + } + + /// One call on the relay's account, whose answer must not be an error. + async fn ok(&mut self, name: &str, mut args: Value) -> Result { + if args.get("accountId").is_none() { + args["accountId"] = ACCOUNT_ID.into(); + } + let answer = self + .call(vec![Invocation::new(name, args, "c")]) + .await? + .into_iter() + .next() + .ok_or_else(|| LinkError::down(format!("The relay didn't answer {name}.")))?; + if answer.name() == "error" { + return Err(LinkError::down(format!( + "The relay refused {name}: {}", + answer.args()["description"] + .as_str() + .or(answer.args()["type"].as_str()) + .unwrap_or("no reason") + ))); + } + Ok(Value::Object(answer.1)) + } +} + +/// What this node last sent the relay on this link. +#[derive(Default)] +struct Sent { + /// The relay's domains were read, and its recipients sent in full. + started: bool, + settings: Option, + domains: BTreeMap, + hashes: BTreeSet, +} + +/// The relay linked: kept up to date until the link drops, the relay is +/// turned off, moved or joined again, or this node stops. +async fn session( + server: &Server, + relay: &Relay, + keys: &DownstreamKeys, + cert: &NodeCertificate, + woken: &mut watch::Receiver, + status: &mut StatusWriter, +) -> Result<(), LinkError> { + let data = server.store(); + let pin = relay.join.relay_pin; + let dialer = Dialer::new(relay, cert, pin)?; + dialer.check_session().await?; + let mut ws = Ws::open(&dialer).await?; + ws.send(&ClientFrame::WebSocketPushEnable { + data_types: Some(vec!["Relay".into()]), + push_state: None, + }) + .await?; + let mut sent = Sent::default(); + let mut relay = relay.clone(); + sync(server, &relay, keys, &dialer, &mut ws, &mut sent).await?; + status.set(data, relay.id, LinkState::Linked, None).await; + refresh_status(&mut ws, data, &relay, &sent, status).await?; + + let mut tick = tokio::time::interval(SYNC); + tick.reset(); + loop { + tokio::select! { + frame = ws.frame() => { + let frame = frame?; + ws.note(&frame); + if ws.relay_changed { + refresh_status(&mut ws, data, &relay, &sent, status).await?; + } + continue; + } + _ = tick.tick() => {} + changed = woken.changed() => { + if changed.is_err() { + return Ok(()); + } + } + } + // The relay as it is now: a new join, a move or off ends this link + match relays::get(data, relay.id).await? { + Some(now_relay) + if now_relay.enabled + && now_relay.join.relay_pin == pin + && now_relay.address == relay.address + && now_relay.link_port == relay.link_port => + { + relay = now_relay; + } + _ => return Ok(()), + } + sync(server, &relay, keys, &dialer, &mut ws, &mut sent).await?; + refresh_status(&mut ws, data, &relay, &sent, status).await?; + } +} + +/// Asks the relay how it is, and records it. +async fn refresh_status( + ws: &mut Ws, + data: &store::Store, + relay: &Relay, + sent: &Sent, + status: &mut StatusWriter, +) -> Result<(), LinkError> { + ws.relay_changed = false; + let answer = ws + .ok(method::RELAY_GET, json!({"ids": [RELAY_SINGLETON]})) + .await?; + let relay_status = answer["list"][0] + .get("status") + .cloned() + .and_then(|value| serde_json::from_value::(value).ok()); + status + .seen(data, relay.id, relay_status, sent.hashes.len() as u64) + .await; + Ok(()) +} + +/// Brings the relay's settings, domains and recipients level with this +/// server's, sending only what changed since the last time. +async fn sync( + server: &Server, + relay: &Relay, + keys: &DownstreamKeys, + dialer: &Dialer, + ws: &mut Ws, + sent: &mut Sent, +) -> Result<(), LinkError> { + let (local, addresses) = local_recipients(server).await?; + let plan = RecipientPlan::new(relay, &local, &addresses); + if plan.domains.is_empty() { + return Err(LinkError::down( + "The relay has no domain to serve: none of its domains is on this server.", + )); + } + + // Settings, whole, when they changed + let settings = pushed_settings(relay, keys, &plan); + if sent.settings.as_ref() != Some(&settings) { + let patch = serde_json::to_value(&settings).unwrap_or_default(); + let answer = ws + .ok( + method::RELAY_SET, + json!({"update": {RELAY_SINGLETON: patch}}), + ) + .await?; + if let Some(error) = answer["notUpdated"][RELAY_SINGLETON].as_object() { + return Err(LinkError::down(format!( + "The relay refused its settings: {}", + error + .get("description") + .and_then(Value::as_str) + .unwrap_or("no reason") + ))); + } + sent.settings = Some(settings); + } + + // Domains: on a new link, whatever the relay holds that isn't served + // goes; after that, only changes + if !sent.started { + let answer = ws.ok(method::RECIPIENT_DOMAIN_GET, json!({})).await?; + for domain in answer["list"].as_array().into_iter().flatten() { + if let Ok(domain) = serde_json::from_value::(domain.clone()) { + sent.domains.insert(domain.domain.clone(), domain); + } + } + } + let create: serde_json::Map = plan + .domains + .iter() + .filter(|(name, domain)| sent.domains.get(*name) != Some(*domain)) + .map(|(name, domain)| { + ( + name.clone(), + serde_json::to_value(domain).unwrap_or_default(), + ) + }) + .collect(); + let destroy: Vec<&String> = sent + .domains + .keys() + .filter(|name| !plan.domains.contains_key(*name)) + .collect(); + if !create.is_empty() || !destroy.is_empty() { + ws.ok( + method::RECIPIENT_DOMAIN_SET, + json!({"create": create, "destroy": destroy}), + ) + .await?; + sent.domains = plan.domains.clone(); + } + + // Recipients: in full on a new link or after many changes, else as + // changes + let hashes: BTreeSet = plan.hashes(&relay.recipient_key).into_iter().collect(); + let added: Vec<&RecipientHash> = hashes.difference(&sent.hashes).collect(); + let removed: Vec<&RecipientHash> = sent.hashes.difference(&hashes).collect(); + if !sent.started || added.len() + removed.len() > MAX_DELTA { + let mut file = Vec::with_capacity(hashes.len() * 32); + for hash in &hashes { + file.extend_from_slice(hash); + } + let blob_id = dialer.upload(file).await?; + ws.ok(method::RECIPIENT_REPLACE, json!({"blobId": blob_id})) + .await?; + } else if !added.is_empty() || !removed.is_empty() { + let create: serde_json::Map = added + .iter() + .enumerate() + .map(|(i, hash)| (format!("h{i}"), json!({"hash": b64::encode(*hash)}))) + .collect(); + let destroy: Vec = removed.iter().map(|hash| b64::encode(*hash)).collect(); + ws.ok( + method::RECIPIENT_SET, + json!({"create": create, "destroy": destroy}), + ) + .await?; + } + sent.hashes = hashes; + sent.started = true; + Ok(()) +} + +// ---- This server's recipients --------------------------------------------- + +/// This server's domains, and every address it takes mail for: accounts and +/// groups with their aliases, mailing lists, and live masked addresses. +/// Each address counts at every name of its domain. +/// +/// A domain whose recipients an external directory answers for has only +/// the addresses this server has seen from it. +async fn local_recipients(server: &Server) -> trc::Result<(Vec, Vec)> { + let registry = server.registry(); + let mut names: HashMap> = HashMap::new(); + let mut local = Vec::new(); + for id in registry + .query::>(RegistryQuery::new(ObjectType::Domain)) + .await? + { + let Some(domain) = registry.object::(id).await? else { + continue; + }; + if !domain.is_enabled { + continue; + } + let mut domain_names = vec![domain.name.trim_end_matches('.').to_ascii_lowercase()]; + domain_names.extend( + domain + .aliases + .iter() + .map(|alias| alias.trim_end_matches('.').to_ascii_lowercase()), + ); + local.push(LocalDomain { + names: domain_names.clone(), + catch_all: domain.catch_all_address.is_some() || domain.allow_relaying, + subaddress_separator: match domain.sub_addressing { + SubAddressing::Disabled => None, + SubAddressing::Enabled | SubAddressing::Custom(_) => Some("+".into()), + }, + }); + names.insert(id.document_id(), domain_names); + } + + let mut addresses = Vec::new(); + let mut add = |local_part: &str, domain_id: u32| { + for name in names.get(&domain_id).into_iter().flatten() { + addresses.push(format!("{}@{name}", local_part.to_lowercase())); + } + }; + for id in registry + .query::>(RegistryQuery::new(ObjectType::Account)) + .await? + { + if let Some(account) = server.try_account(id.document_id()).await? { + for address in account.addresses.iter() { + add(&address.local_part, address.domain_id); + } + } + } + for id in registry + .query::>(RegistryQuery::new(ObjectType::MailingList)) + .await? + { + if let Some(list) = server.try_list(id.document_id()).await? { + for address in list.addresses.iter() { + add(&address.local_part, address.domain_id); + } + } + } + for mask in masked_email::ops::all(server.store(), registry).await? { + if mask.accepts_mail() { + addresses.push(mask.object.email.to_lowercase()); + } + } + Ok((local, addresses)) +} diff --git a/crates/services/src/lib.rs b/crates/services/src/lib.rs index 485ec4d9..5004d130 100644 --- a/crates/services/src/lib.rs +++ b/crates/services/src/lib.rs @@ -26,6 +26,7 @@ pub mod broadcast; pub mod inbuxa_deliverability; // inbuxa: the deliverability check pub mod inbuxa_lock_expiry; pub mod inbuxa_log_retention; // inbuxa: personal-data catalog, D1 +pub mod inbuxa_relay_link; // inbuxa: links to inbuxa Relays pub mod inbuxa_scheduled_reports; // inbuxa: scheduled reports and the weekly digest pub mod state_manager; pub mod task_manager; @@ -78,6 +79,9 @@ impl SpawnServices for IpcReceivers { // inbuxa: deliverability spec, DL-14: each node checks itself daily inbuxa_deliverability::spawn_deliverability(inner.clone()); + // inbuxa: relay spec, "The link service": every node links to every relay + inbuxa_relay_link::spawn_relay_links(inner.clone()); + // inbuxa: scheduled-reports spec, RP-16: every node looks for due reports inbuxa_scheduled_reports::spawn_scheduled_reports(inner.clone()); diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index ca56e3f7..be66a11c 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -669,8 +669,14 @@ In progress. Done so far: hashing and sealing (PR #244). - `crates/relay`: the relay binary, inbound in held and live mode (PR #245). - On the downstream: `inbuxa:Relay` and its permissions, the link CA and - sealing key, and join tokens. Status, `inbuxa:RelayMessage` and the link - service are not done yet. + sealing key, and join tokens (PR #249). +- The downstream's link service: every node dials each enabled relay, + completes the join and pins its key, pushes settings, domains and the + 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. Order: diff --git a/resources/privacy/catalog.toml b/resources/privacy/catalog.toml index 92bdb1dd..36946ce9 100644 --- a/resources/privacy/catalog.toml +++ b/resources/privacy/catalog.toml @@ -128,6 +128,7 @@ settings = ["network"] smtpCertificate = ["network"] smtpPrivateKey = ["credential"] joinToken = ["credential"] +links = ["network"] createdBy = ["identifier"] [object."inbuxa:Journal"] diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 85e5af17..de8c63b4 100644 --- a/tests/src/system/relay.rs +++ b/tests/src/system/relay.rs @@ -10,6 +10,10 @@ //! and is kept when written back that way, a rejoin issues a new token, //! nobody in a tenant reaches relays, and the audit log never holds the //! key or a token. +//! +//! `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. use crate::utils::{ account::Account, @@ -352,3 +356,272 @@ pub async fn relay_tests() { test.temp_dir.delete(); } } + +/// The `inbuxa-relay` binary: `INBUXA_RELAY_BIN`, or the workspace's debug +/// build. +fn relay_binary() -> std::path::PathBuf { + std::env::var_os("INBUXA_RELAY_BIN") + .map(Into::into) + .unwrap_or_else(|| { + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../target/debug/inbuxa-relay") + }) +} + +fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +/// The relay process, stopped when the test ends. +struct RelayProcess(std::process::Child); + +impl Drop for RelayProcess { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// Polls `check` against the relay object until it's true, or panics. +async fn wait_for( + account: &Account, + id: &str, + what: &str, + seconds: u64, + check: impl Fn(&Value) -> bool, +) -> Value { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(seconds); + loop { + let relay = get(account, id).await; + if check(&relay) { + return relay; + } + if std::time::Instant::now() > deadline { + panic!("{what}: {relay}"); + } + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } +} + +/// The relay's replies to one SMTP transaction's recipients, after EHLO and +/// MAIL FROM. Port 25 opens a few seconds after the relay is set up. +async fn rcpt_replies(port: u16, recipients: &[&str]) -> Vec { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + let stream = loop { + match tokio::net::TcpStream::connect(("127.0.0.1", port)).await { + Ok(stream) => break stream, + Err(e) if std::time::Instant::now() > deadline => panic!("port {port}: {e}"), + Err(_) => tokio::time::sleep(std::time::Duration::from_millis(250)).await, + } + }; + 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")); + let mut replies = Vec::new(); + for line in [ + "EHLO client.example.org".to_string(), + "MAIL FROM:".into(), + ] + .into_iter() + .chain(recipients.iter().map(|r| format!("RCPT TO:<{r}>"))) + { + write + .write_all(format!("{line}\r\n").as_bytes()) + .await + .unwrap(); + replies.push(reply(&mut read).await); + } + let _ = write.write_all(b"QUIT\r\n").await; + replies.split_off(2) +} + +pub async fn link(test: &mut TestServer) { + println!("Running inbuxa Relay link tests..."); + let admin = test.account("admin@example.com"); + let binary = relay_binary(); + assert!( + binary.exists(), + "{} is missing: run cargo build -p inbuxa-relay", + binary.display() + ); + admin + .create_user_account( + "jane@example.com", + "jane-secret-4417", + "Jane", + &["jane.doe@example.com"], + vec![], + ) + .await; + let link_port = free_port(); + let smtp_port = free_port(); + + let (_, response) = call( + admin, + "inbuxa:Relay/set", + json!({"create": {"r": { + "name": "Edge", + "address": "127.0.0.1", + "linkPort": link_port, + "settings": { + "hostname": "mx1.example.com", + "smtpListen": [format!("127.0.0.1:{smtp_port}")], + "domains": ["example.com"] + } + }}}), + ) + .await; + let id = response["created"]["r"]["id"] + .as_str() + .unwrap_or_else(|| panic!("relay created: {response}")) + .to_string(); + let token = response["created"]["r"]["joinToken"] + .as_str() + .unwrap() + .to_string(); + + // The relay host: join with the token, then run + let state = test.temp_dir.path.join("relay-state"); + let joined = std::process::Command::new(&binary) + .args(["join", &token, "--state"]) + .arg(&state) + .output() + .unwrap(); + assert!(joined.status.success(), "{joined:?}"); + let _relay = RelayProcess( + std::process::Command::new(&binary) + .args([ + "run", + "--link-listen", + &format!("127.0.0.1:{link_port}"), + "--state", + ]) + .arg(&state) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .unwrap(), + ); + + // The server joins it, pins its key, and sets it up: port 25 opens + let relay = wait_for(admin, &id, "the relay joined and took mail", 60, |relay| { + relay["joinState"] == "joined" + && relay["links"] + .as_array() + .into_iter() + .flatten() + .any(|link| link["state"] == "linked" && link["relay"]["mode"] == "held") + }) + .await; + assert!(relay["linkKeyPin"].as_str().is_some(), "{relay}"); + let link = &relay["links"][0]; + // The administrator, jane and her alias, at least + let recipients = link["recipients"].as_u64().unwrap_or_default(); + assert!(recipients >= 3, "{relay}"); + assert_eq!(link["lastError"], Value::Null, "{relay}"); + + // Real recipients only (RL-1) + let replies = rcpt_replies( + smtp_port, + &[ + "jane@example.com", + "Jane.Doe@example.com", + "jane+news@example.com", + "nobody@example.com", + "jane@elsewhere.org", + ], + ) + .await; + assert!(replies[0].starts_with("250"), "{replies:?}"); + assert!(replies[1].starts_with("250"), "alias: {replies:?}"); + assert!(replies[2].starts_with("250"), "sub-address: {replies:?}"); + assert!(replies[3].starts_with("550 5.1.1"), "unknown: {replies:?}"); + assert!( + replies[4].starts_with("550 5.7.1"), + "not served: {replies:?}" + ); + + // A new account reaches the relay as a change + admin + .create_user_account("joe@example.com", "joe-secret-5529", "Joe", &[], vec![]) + .await; + wait_for( + admin, + &id, + "the new recipient reached the relay", + 45, + |relay| relay["links"][0]["recipients"].as_u64() == Some(recipients + 1), + ) + .await; + let replies = rcpt_replies(smtp_port, &["joe@example.com"]).await; + assert!(replies[0].starts_with("250"), "{replies:?}"); + + // A setting reaches it too + call( + admin, + "inbuxa:Relay/set", + json!({"update": {&id: {"settings/maxRecipients": 1}}}), + ) + .await; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let replies = rcpt_replies(smtp_port, &["jane@example.com", "joe@example.com"]).await; + if replies[1].starts_with("452") { + break; + } + assert!( + std::time::Instant::now() < deadline, + "maxRecipients reached the relay: {replies:?}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + + // Turned off, the link ends and says why + call( + admin, + "inbuxa:Relay/set", + json!({"update": {&id: {"enabled": false}}}), + ) + .await; + wait_for(admin, &id, "the link ended", 30, |relay| { + relay["links"][0]["state"] == "down" + && relay["links"][0]["lastError"] + .as_str() + .is_some_and(|e| e.contains("turned off")) + }) + .await; + + call(admin, "inbuxa:Relay/set", json!({"destroy": [&id]})).await; +} + +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn relay_link_tests() { + let mut test = TestServerBuilder::new("relay_link_tests") + .await + .with_default_listeners() + .await + .build() + .await; + let admin = test.create_admin_account("admin@example.com").await; + test.insert_account(admin); + self::link(&mut test).await; + if test.is_reset() { + test.temp_dir.delete(); + } +}