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]]

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"] }

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<u8>,
+    pub key_der: Vec<u8>,
+    pub ca_der: Vec<u8>,
+    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<NodeCertificate> {
+    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<Vec<u8>> {
+    use base64::Engine;
+    base64::engine::general_purpose::STANDARD
+        .decode(
+            pem.lines()
+                .filter(|line| !line.starts_with("-----"))
+                .collect::<String>(),
+        )
+        .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<JoinRecorded> {
+    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<String>,
+    /// Every address is accepted: a catch-all address, or relaying allowed.
+    pub catch_all: bool,
+    pub subaddress_separator: Option<String>,
+}
+
+/// 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<String, RecipientDomain>,
+    pub addresses: BTreeSet<String>,
+}
+
+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<String> = 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<RecipientHash> {
+        let mut hashes: Vec<RecipientHash> = 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<u64>,
+    pub last_error: Option<String>,
+    /// What the relay last said of itself.
+    pub relay: Option<RelayStatus>,
+    /// How many recipients this node last sent.
+    pub recipients: Option<u64>,
+    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<Vec<NodeLink>> {
+    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::<NodeLink>::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<LocalDomain> {
+        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![
+            "[email protected]".to_string(),
+            "[email protected]".into(),
+            "[email protected]".into(),
+            "[email protected]".into(),
+        ];
+        let all = RecipientPlan::new(&relay(&[]), &local(), &addresses);
+        assert_eq!(
+            all.domains.keys().collect::<Vec<_>>(),
+            ["example.com", "example.net", "other.org"]
+        );
+        assert!(all.domains["other.org"].catch_all);
+        assert_eq!(all.addresses.len(), 3);
+        assert!(all.addresses.contains("[email protected]"));
+
+        // A relay naming its domains serves only those
+        let one = RecipientPlan::new(&relay(&["Example.COM."]), &local(), &addresses);
+        assert_eq!(one.domains.keys().collect::<Vec<_>>(), ["example.com"]);
+        assert_eq!(
+            one.addresses.iter().collect::<Vec<_>>(),
+            ["[email protected]"]
+        );
+
+        // The hashes are what the relay looks up
+        let hashes = one.hashes(&[9; 32]);
+        assert_eq!(hashes, vec![recipient::hash(&[9; 32], "[email protected]")]);
+        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
+        );
+    }
+}

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<tokio::sync::Notify> =
+    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<b
     }
 }
 
+/// Removes a relay, and every node's record of its link.
 pub async fn delete(data: &Store, id: u32) -> 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!())?;

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,

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::<Vec<_>>();
+    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)
 }

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]
 

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

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:
 

resources/privacy/catalog.toml

@@ -128,6 +128,7 @@ settings = ["network"]
 smtpCertificate = ["network"]
 smtpPrivateKey = ["credential"]
 joinToken = ["credential"]
+links = ["network"]
 createdBy = ["identifier"]
 
 [object."inbuxa:Journal"]

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<String> {
+    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<tokio::net::tcp::OwnedReadHalf>) -> 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:<[email protected]>".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("[email protected]");
+    let binary = relay_binary();
+    assert!(
+        binary.exists(),
+        "{} is missing: run cargo build -p inbuxa-relay",
+        binary.display()
+    );
+    admin
+        .create_user_account(
+            "[email protected]",
+            "jane-secret-4417",
+            "Jane",
+            &["[email protected]"],
+            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,
+        &[
+            "[email protected]",
+            "[email protected]",
+            "[email protected]",
+            "[email protected]",
+            "[email protected]",
+        ],
+    )
+    .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("[email protected]", "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, &["[email protected]"]).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, &["[email protected]", "[email protected]"]).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("[email protected]").await;
+    test.insert_account(admin);
+    self::link(&mut test).await;
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}