crates/features/src/relay/link.rs

@@ -38,6 +38,8 @@ pub struct NodeCertificate {
     pub key_der: Vec<u8>,
     pub ca_der: Vec<u8>,
     pub issued_at: u64,
+    /// The link CA that issued it: a new one means a new certificate.
+    pub ca_pin: Pin,
 }
 
 fn cert_error(err: rcgen::Error) -> trc::Error {
@@ -83,6 +85,7 @@ pub fn issue_node_certificate(
         key_der: key.serialize_der(),
         ca_der,
         issued_at: now,
+        ca_pin: keys.link_ca_pin,
     })
 }
 
@@ -228,6 +231,8 @@ pub fn pushed_settings(
     settings.domains = plan.domains.keys().cloned().collect();
     settings.sealing_key = Some(keys.sealing_public);
     settings.recipient_key = Some(relay.recipient_key);
+    // While the link CA is replaced, the relay trusts the next one too (RL-34)
+    settings.next_link_ca_pin = keys.next_link_ca.as_ref().map(|next| next.pin);
     settings.smtp_certificate = match (&relay.smtp_certificate, &relay.smtp_private_key) {
         (Some(chain_pem), Some(key_pem)) => Some(SmtpCertificate {
             chain_pem: chain_pem.clone(),
@@ -314,6 +319,84 @@ pub async fn statuses(data: &Store, relay_id: u32) -> trc::Result<Vec<NodeLink>>
     Ok(links)
 }
 
+/// Which link CA a relay trusts, as it last said (RL-34).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LinkCa {
+    /// This server's.
+    Current,
+    /// This server's, and the one replacing it.
+    Both,
+    /// Only one this server replaced: it needs a new join.
+    Old,
+}
+
+impl LinkCa {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            LinkCa::Current => "current",
+            LinkCa::Both => "both",
+            LinkCa::Old => "old",
+        }
+    }
+}
+
+/// What a relay's links last said of the CAs it trusts, the newest report
+/// first; `None` before any has. A relay still joining has said nothing,
+/// but a token that named a CA since replaced can't complete.
+pub fn link_ca(relay: &Relay, keys: &DownstreamKeys, links: &[NodeLink]) -> Option<LinkCa> {
+    if relay.join.relay_pin.is_none() {
+        return relay
+            .join
+            .ca_pin
+            .filter(|pin| !pins_match(pin, &keys.link_ca_pin))
+            .map(|_| LinkCa::Old);
+    }
+    let pins = links
+        .iter()
+        .filter_map(|link| Some((link.updated_at, &link.relay.as_ref()?.link_ca_pins)))
+        .filter(|(_, pins)| !pins.is_empty())
+        .max_by_key(|(at, _)| *at)?
+        .1;
+    let trusts = |pin: &Pin| {
+        pins.iter()
+            .any(|p| *p == inbuxa_relay_proto::b64::encode(pin))
+    };
+    Some(if !trusts(&keys.link_ca_pin) {
+        LinkCa::Old
+    } else if keys
+        .next_link_ca
+        .as_ref()
+        .is_some_and(|next| trusts(&next.pin))
+    {
+        LinkCa::Both
+    } else {
+        LinkCa::Current
+    })
+}
+
+/// Whether a link CA replacement can finish: every enabled relay that has
+/// joined says it trusts the next CA too. A relay that never says so holds
+/// it until an administrator finishes it anyway (RL-34).
+pub async fn link_ca_ready(
+    data: &Store,
+    keys: &DownstreamKeys,
+    relays: &[Relay],
+) -> trc::Result<bool> {
+    if keys.next_link_ca.is_none() {
+        return Ok(false);
+    }
+    for relay in relays
+        .iter()
+        .filter(|r| r.enabled && r.join.relay_pin.is_some())
+    {
+        let links = statuses(data, relay.id).await?;
+        if link_ca(relay, keys, &links) != Some(LinkCa::Both) {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
 /// 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 {
@@ -519,4 +602,57 @@ mod tests {
             now + NODE_CERT_LIFETIME
         );
     }
+
+    /// RL-34: which CA a relay trusts, from what its links last said.
+    #[test]
+    fn link_ca_states() {
+        let keys = new_downstream_keys(time::now()).unwrap();
+        let mut replacing = keys.clone();
+        replacing.next_link_ca = Some(super::super::NextLinkCa {
+            key_pem: String::new(),
+            cert_pem: String::new(),
+            pin: [7; 32],
+            started_at: 1,
+        });
+        let pin = |pin: &Pin| inbuxa_relay_proto::b64::encode(pin);
+        let said = |pins: Vec<String>, at: u64| NodeLink {
+            relay: Some(RelayStatus {
+                link_ca_pins: pins,
+                ..Default::default()
+            }),
+            updated_at: at,
+            ..Default::default()
+        };
+        let mut joined = relay(&[]);
+        joined.join.relay_pin = Some([1; 32]);
+        // Nothing said yet, or by a relay from before RL-34
+        assert_eq!(link_ca(&joined, &keys, &[]), None);
+        assert_eq!(link_ca(&joined, &keys, &[said(vec![], 5)]), None);
+        let current = said(vec![pin(&keys.link_ca_pin)], 5);
+        assert_eq!(
+            link_ca(&joined, &keys, std::slice::from_ref(&current)),
+            Some(LinkCa::Current)
+        );
+        assert_eq!(
+            link_ca(&joined, &replacing, std::slice::from_ref(&current)),
+            Some(LinkCa::Current)
+        );
+        let both = said(vec![pin(&keys.link_ca_pin), pin(&[7; 32])], 9);
+        // The newest report wins
+        assert_eq!(
+            link_ca(&joined, &replacing, &[current.clone(), both.clone()]),
+            Some(LinkCa::Both)
+        );
+        // Once finished, a relay that never took the new CA is left on the old
+        let finished = replacing.finished().unwrap();
+        assert_eq!(finished.link_ca_pin, [7; 32]);
+        assert!(finished.next_link_ca.is_none());
+        assert_eq!(link_ca(&joined, &finished, &[current]), Some(LinkCa::Old));
+        assert_eq!(link_ca(&joined, &finished, &[both]), Some(LinkCa::Current));
+        // A join whose token named the old CA can't complete
+        let mut joining = relay(&[]);
+        joining.start_join(&keys, time::now());
+        assert_eq!(link_ca(&joining, &keys, &[]), None);
+        assert_eq!(link_ca(&joining, &finished, &[]), Some(LinkCa::Old));
+    }
 }

crates/features/src/relay/mod.rs

@@ -87,6 +87,55 @@ pub struct DownstreamKeys {
     #[serde(with = "b64")]
     pub sealing_public: [u8; 32],
     pub created_at: u64,
+    /// The CA replacing the link CA, while a replacement is under way
+    /// (RL-34).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_link_ca: Option<NextLinkCa>,
+}
+
+/// A link CA on its way in: pushed to the relays as `nextLinkCaPin` until
+/// every joined relay trusts it, then the link CA.
+#[derive(Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NextLinkCa {
+    pub key_pem: String,
+    pub cert_pem: String,
+    #[serde(with = "b64")]
+    pub pin: Pin,
+    pub started_at: u64,
+}
+
+/// What an administrator asks of a link CA replacement (RL-34).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LinkCaChange {
+    Start,
+    Finish,
+    Cancel,
+}
+
+impl LinkCaChange {
+    pub fn parse(text: &str) -> Option<Self> {
+        match text {
+            "start" => Some(LinkCaChange::Start),
+            "finish" => Some(LinkCaChange::Finish),
+            "cancel" => Some(LinkCaChange::Cancel),
+            _ => None,
+        }
+    }
+}
+
+impl DownstreamKeys {
+    /// The keys with the next link CA as the link CA, if there is one.
+    pub fn finished(&self) -> Option<DownstreamKeys> {
+        let next = self.next_link_ca.as_ref()?;
+        Some(DownstreamKeys {
+            link_ca_key_pem: next.key_pem.clone(),
+            link_ca_cert_pem: next.cert_pem.clone(),
+            link_ca_pin: next.pin,
+            next_link_ca: None,
+            ..self.clone()
+        })
+    }
 }
 
 impl std::fmt::Debug for DownstreamKeys {
@@ -111,6 +160,10 @@ pub struct Join {
     pub relay_pin: Option<Pin>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub joined_at: Option<u64>,
+    /// The link CA the token named. A join started before a replacement
+    /// finished can't complete after it (RL-34).
+    #[serde(with = "b64::option", skip_serializing_if = "Option::is_none")]
+    pub ca_pin: Option<Pin>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize)]
@@ -270,6 +323,7 @@ impl Relay {
         if settings.smtp_certificate.is_some()
             || settings.sealing_key.is_some()
             || settings.recipient_key.is_some()
+            || settings.next_link_ca_pin.is_some()
         {
             return invalid(
                 "settings",
@@ -353,6 +407,7 @@ impl Relay {
             expires_at: token.expires_at,
             relay_pin: None,
             joined_at: None,
+            ca_pin: Some(keys.link_ca_pin),
         };
         token.encode()
     }
@@ -370,6 +425,21 @@ pub fn random_key() -> [u8; 32] {
 
 /// New downstream keys: a P-256 link CA, and an X25519 sealing key pair.
 pub fn new_downstream_keys(now: u64) -> trc::Result<DownstreamKeys> {
+    let (key_pem, cert_pem, pin) = new_link_ca()?;
+    let (sealing_secret, sealing_public) = seal::generate_keypair();
+    Ok(DownstreamKeys {
+        link_ca_pin: pin,
+        link_ca_key_pem: key_pem,
+        link_ca_cert_pem: cert_pem,
+        sealing_secret: *sealing_secret,
+        sealing_public,
+        created_at: now,
+        next_link_ca: None,
+    })
+}
+
+/// A new P-256 link CA: its key, its certificate, and its key's pin.
+fn new_link_ca() -> trc::Result<(String, String, Pin)> {
     let failed = |err: rcgen::Error| {
         trc::EventType::Server(trc::ServerEvent::ThreadError)
             .into_err()
@@ -391,15 +461,11 @@ pub fn new_downstream_keys(now: u64) -> trc::Result<DownstreamKeys> {
     params.not_before = rcgen::date_time_ymd(2026, 1, 1);
     params.not_after = rcgen::date_time_ymd(2126, 1, 1);
     let cert = params.self_signed(&key).map_err(failed)?;
-    let (sealing_secret, sealing_public) = seal::generate_keypair();
-    Ok(DownstreamKeys {
-        link_ca_pin: spki_pin(&rcgen::PublicKeyData::subject_public_key_info(&key)),
-        link_ca_key_pem: key.serialize_pem(),
-        link_ca_cert_pem: cert.pem(),
-        sealing_secret: *sealing_secret,
-        sealing_public,
-        created_at: now,
-    })
+    Ok((
+        key.serialize_pem(),
+        cert.pem(),
+        spki_pin(&rcgen::PublicKeyData::subject_public_key_info(&key)),
+    ))
 }
 
 struct Json<T>(T);
@@ -540,6 +606,71 @@ pub async fn create(data: &Store, relay: &Relay, now: u64) -> trc::Result<(Relay
 /// stored bytes.
 struct Versioned(Relay, u64);
 
+struct VersionedKeys(DownstreamKeys, u64);
+
+impl Deserialize for VersionedKeys {
+    fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
+        let Json(keys) = Json::<DownstreamKeys>::deserialize(bytes)?;
+        Ok(VersionedKeys(keys, xxhash_rust::xxh3::xxh3_64(bytes)))
+    }
+}
+
+/// Starts, finishes or cancels a replacement of the link CA (RL-34), and
+/// says whether anything changed: starting one already under way, or
+/// finishing or cancelling none, changes nothing. Finishing makes the next
+/// CA the link CA; each node then takes a certificate from it, and the
+/// relays that trust it make it their only one.
+pub async fn change_link_ca(data: &Store, change: LinkCaChange, now: u64) -> trc::Result<bool> {
+    keys_or_new(data, now).await?;
+    loop {
+        let Some(VersionedKeys(keys, version)) = data
+            .get_value::<VersionedKeys>(ValueKey::from(class(KIND_KEYS, None)))
+            .await
+            .caused_by(trc::location!())?
+        else {
+            return Ok(false);
+        };
+        let next = match change {
+            LinkCaChange::Start if keys.next_link_ca.is_none() => {
+                let (key_pem, cert_pem, pin) = new_link_ca()?;
+                DownstreamKeys {
+                    next_link_ca: Some(NextLinkCa {
+                        key_pem,
+                        cert_pem,
+                        pin,
+                        started_at: now,
+                    }),
+                    ..keys
+                }
+            }
+            LinkCaChange::Finish => match keys.finished() {
+                Some(next) => next,
+                None => return Ok(false),
+            },
+            LinkCaChange::Cancel if keys.next_link_ca.is_some() => DownstreamKeys {
+                next_link_ca: None,
+                ..keys
+            },
+            _ => return Ok(false),
+        };
+        let mut batch = BatchBuilder::new();
+        batch.assert_value(class(KIND_KEYS, None), AssertValue::Hash(version));
+        batch.set(class(KIND_KEYS, None), Json(&next).serialize()?);
+        match data.write(batch.build_all()).await {
+            Ok(_) => {
+                changed();
+                return Ok(true);
+            }
+            Err(err)
+                if matches!(
+                    err.as_ref(),
+                    trc::EventType::Store(trc::StoreEvent::AssertValueFailed)
+                ) => {}
+            Err(err) => return Err(err.caused_by(trc::location!())),
+        }
+    }
+}
+
 impl Deserialize for Versioned {
     fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
         let Json(relay) = Json::<Relay>::deserialize(bytes)?;

crates/jmap-proto/src/object/inbuxa_relay.rs

@@ -56,6 +56,15 @@ pub enum RelayProperty {
     /// Whether a linked relay runs an older release than this server
     /// (RL-28); null before any link has said.
     OlderThanServer,
+    /// Update only, sent on its own: `start`, `finish` or `cancel` a
+    /// replacement of the link CA every relay shares (RL-34).
+    ReplaceLinkCa,
+    /// Which link CA the relay last said it trusts: `current`, `both`
+    /// (the current and the next, during a replacement) or `old` (one this
+    /// server replaced: it needs a new join); null before it has said.
+    LinkCa,
+    /// When a replacement of the link CA began, while one is under way.
+    LinkCaReplacing,
     /// In a create or rejoin response only.
     JoinToken,
     /// Each node's link to the relay, and what the relay last said.
@@ -102,6 +111,9 @@ impl Property for RelayProperty {
             RelayProperty::Rejoin => "rejoin",
             RelayProperty::Upgrade => "upgrade",
             RelayProperty::OlderThanServer => "olderThanServer",
+            RelayProperty::ReplaceLinkCa => "replaceLinkCa",
+            RelayProperty::LinkCa => "linkCa",
+            RelayProperty::LinkCaReplacing => "linkCaReplacing",
             RelayProperty::JoinToken => "joinToken",
             RelayProperty::Links => "links",
             RelayProperty::CreatedBy => "createdBy",
@@ -136,6 +148,9 @@ impl RelayProperty {
             b"rejoin" => RelayProperty::Rejoin,
             b"upgrade" => RelayProperty::Upgrade,
             b"olderThanServer" => RelayProperty::OlderThanServer,
+            b"replaceLinkCa" => RelayProperty::ReplaceLinkCa,
+            b"linkCa" => RelayProperty::LinkCa,
+            b"linkCaReplacing" => RelayProperty::LinkCaReplacing,
             b"joinToken" => RelayProperty::JoinToken,
             b"links" => RelayProperty::Links,
             b"createdBy" => RelayProperty::CreatedBy,

crates/jmap/src/inbuxa/relay.rs

@@ -16,7 +16,8 @@
 use crate::blob::download::BlobDownload;
 use common::{Server, auth::AccessToken, network::acme::relay::relay_certificate_not_after};
 use inbuxa_features::relay::{
-    self as relays, JoinState, Relay as StoredRelay, certificate::CertificateOrder, link::NodeLink,
+    self as relays, DownstreamKeys, JoinState, Relay as StoredRelay, certificate::CertificateOrder,
+    link::NodeLink,
 };
 use inbuxa_relay_proto::{
     b64,
@@ -66,6 +67,8 @@ const ALL: &[P] = &[
     P::LinkKeyPin,
     P::Links,
     P::OlderThanServer,
+    P::LinkCa,
+    P::LinkCaReplacing,
     P::CreatedBy,
     P::CreatedAt,
     P::UpdatedAt,
@@ -86,6 +89,8 @@ const SERVER_SET: &[P] = &[
     P::JoinToken,
     P::Links,
     P::OlderThanServer,
+    P::LinkCa,
+    P::LinkCaReplacing,
     P::CreatedBy,
     P::CreatedAt,
     P::UpdatedAt,
@@ -172,6 +177,7 @@ fn to_value(
     relay: &StoredRelay,
     links: &[NodeLink],
     order: Option<&CertificateOrder>,
+    keys: Option<&DownstreamKeys>,
     properties: &[P],
     now: u64,
 ) -> RValue {
@@ -207,6 +213,13 @@ fn to_value(
             P::LinkKeyPin => text(relay.join.relay_pin.map(|pin| b64::encode(&pin)).as_deref()),
             P::Links => links_value(links),
             P::OlderThanServer => older_than_server(links).map_or(Value::Null, Value::Bool),
+            P::LinkCa => text(
+                keys.and_then(|keys| relays::link::link_ca(relay, keys, links))
+                    .map(|ca| ca.as_str()),
+            ),
+            P::LinkCaReplacing => keys
+                .and_then(|keys| keys.next_link_ca.as_ref())
+                .map_or(Value::Null, |next| date(next.started_at)),
             P::CreatedBy => text(Some(&relay.created_by)),
             P::CreatedAt => date(relay.created_at),
             P::UpdatedAt => date(relay.updated_at),
@@ -225,6 +238,7 @@ struct Sent {
     settings: JsonMap,
     rejoin: bool,
     upgrade: Option<Upgrade>,
+    replace_link_ca: Option<relays::LinkCaChange>,
 }
 
 /// A release to hand the relay: an upload, and its signature file's text.
@@ -250,6 +264,21 @@ fn client_json(value: Value<'_, P, RelayValue>) -> Result<Sent, SetError<P>> {
                         .with_description("true, to start a new join."));
                 }
             },
+            Key::Property(P::ReplaceLinkCa) => {
+                match value
+                    .as_str()
+                    .and_then(|text| relays::LinkCaChange::parse(&text))
+                {
+                    Some(change) => sent.replace_link_ca = Some(change),
+                    None => {
+                        return Err(SetError::invalid_properties()
+                            .with_property(P::ReplaceLinkCa)
+                            .with_description(
+                                "start, finish or cancel a replacement of the link CA.",
+                            ));
+                    }
+                }
+            }
             Key::Property(P::Upgrade) => {
                 let json: serde_json::Value = value.into();
                 match (json["blobId"].as_str(), json["signature"].as_str()) {
@@ -460,6 +489,7 @@ pub async fn get(
     };
     let data = server.store();
     let all = relays::all(data).await?;
+    let keys = relays::keys(data).await?;
     let now = now();
     let wanted: Vec<&StoredRelay> = match ids {
         None => all.iter().collect(),
@@ -475,7 +505,10 @@ pub async fn get(
         }
     };
     for relay in wanted {
-        let links = if properties.contains(&P::Links) || properties.contains(&P::OlderThanServer) {
+        let links = if properties.contains(&P::Links)
+            || properties.contains(&P::OlderThanServer)
+            || properties.contains(&P::LinkCa)
+        {
             relays::link::statuses(data, relay.id).await?
         } else {
             Vec::new()
@@ -485,9 +518,14 @@ pub async fn get(
         } else {
             None
         };
-        response
-            .list
-            .push(to_value(relay, &links, order.as_ref(), &properties, now));
+        response.list.push(to_value(
+            relay,
+            &links,
+            order.as_ref(),
+            keys.as_ref(),
+            &properties,
+            now,
+        ));
     }
     Ok(response)
 }
@@ -516,6 +554,11 @@ pub async fn set(
                     .with_property(P::Upgrade)
                     .with_description("A relay is upgraded once it has joined."));
             }
+            if sent.replace_link_ca.is_some() {
+                return Err(SetError::invalid_properties()
+                    .with_property(P::ReplaceLinkCa)
+                    .with_description("The link CA is replaced on an existing relay."));
+            }
             if sent
                 .properties
                 .get("smtpPrivateKey")
@@ -564,6 +607,38 @@ pub async fn set(
                 continue;
             }
         };
+        // The link CA every relay shares (RL-34): asked of any one of them
+        if let Some(change) = sent.replace_link_ca {
+            let outcome = if sent.rejoin
+                || sent.upgrade.is_some()
+                || !sent.properties.is_empty()
+                || !sent.settings.is_empty()
+            {
+                Err(SetError::invalid_properties()
+                    .with_property(P::ReplaceLinkCa)
+                    .with_description("Send replaceLinkCa on its own."))
+            } else if relay_id(id).is_none()
+                || relays::get(data, relay_id(id).unwrap_or_default())
+                    .await?
+                    .is_none()
+            {
+                Err(SetError::not_found())
+            } else if relays::change_link_ca(data, change, now()).await? {
+                Ok(())
+            } else {
+                Err(SetError::invalid_properties()
+                    .with_property(P::ReplaceLinkCa)
+                    .with_description(match change {
+                        relays::LinkCaChange::Start => "A replacement is already under way.",
+                        _ => "No replacement is under way.",
+                    }))
+            };
+            match outcome {
+                Ok(()) => response.updated.append(id, None),
+                Err(error) => response.not_updated.append(id, error),
+            }
+            continue;
+        }
         if let Some(upgrade) = &sent.upgrade {
             let outcome = if sent.rejoin || !sent.properties.is_empty() || !sent.settings.is_empty()
             {

crates/relay-proto/src/types.rs

@@ -70,6 +70,10 @@ pub struct RelaySettings {
     /// Addresses and CIDR ranges that may dial the link port; empty for
     /// anyone (RL-33).
     pub link_allow_from: Vec<String>,
+    /// A second link CA to trust while the downstream replaces its own
+    /// (RL-34). Pushed as null by a link on that CA, it becomes the only one.
+    #[serde(with = "b64::option", skip_serializing_if = "Option::is_none")]
+    pub next_link_ca_pin: Option<crate::pin::Pin>,
     /// The STARTTLS certificate for port 25.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub smtp_certificate: Option<SmtpCertificate>,
@@ -98,6 +102,7 @@ impl Default for RelaySettings {
             outbound_expiry: 3 * 86_400,
             outbound_tls: OutboundTls::default(),
             link_allow_from: Vec::new(),
+            next_link_ca_pin: None,
             smtp_certificate: None,
             sealing_key: None,
             recipient_key: None,
@@ -239,6 +244,10 @@ pub struct RelayStatus {
     /// it: what `linkAllowFrom` has to let in (RL-33).
     #[serde(skip_serializing_if = "Option::is_none")]
     pub link_from: Option<String>,
+    /// The link CAs it trusts, base64url pins: its own first, then the one
+    /// the downstream is moving to, if any (RL-34).
+    #[serde(default)]
+    pub link_ca_pins: Vec<String>,
 }
 
 /// Live while a link client has push on for `Inbound` (spec "Live mode").

crates/relay/src/app.rs

@@ -15,6 +15,8 @@ use crate::{
 };
 use inbuxa_relay_proto::{
     allow::AllowList,
+    b64,
+    pin::Pin,
     time,
     types::{Mode, RelaySettings, RelayStatus, SpoolStatus, type_name},
 };
@@ -30,6 +32,12 @@ use tokio::sync::{Notify, broadcast, watch};
 
 pub const VERSION: &str = env!("CARGO_PKG_VERSION");
 
+fn trusted_cas(ca: Pin, next: Option<Pin>) -> Vec<Pin> {
+    std::iter::once(ca)
+        .chain(next.filter(|next| *next != ca))
+        .collect()
+}
+
 /// A data type's new state, for push.
 #[derive(Debug, Clone)]
 pub struct Changed {
@@ -69,6 +77,9 @@ pub struct App {
     /// `run --link-allow-any`: the `linkAllowFrom` to ignore, until the
     /// downstream sets another (RL-33).
     pub link_allow_any: Mutex<Option<Vec<String>>>,
+    /// The link CAs trusted: the relay's own, then the downstream's next
+    /// while it replaces it (RL-34). A link on a CA dropped from it ends.
+    pub link_ca: watch::Sender<Vec<Pin>>,
 }
 
 impl App {
@@ -85,6 +96,10 @@ impl App {
             }
         }
         let (push, _) = broadcast::channel(256);
+        let link_ca = trusted_cas(
+            identity.ca_pin,
+            settings.as_ref().and_then(|s| s.next_link_ca_pin),
+        );
         Ok(App {
             paths,
             identity: Mutex::new(identity),
@@ -108,6 +123,7 @@ impl App {
             upgrading: AtomicBool::new(false),
             restart: Notify::new(),
             link_allow_any: Mutex::new(None),
+            link_ca: watch::Sender::new(link_ca),
         })
     }
 
@@ -144,6 +160,17 @@ impl App {
         }
     }
 
+    /// Trusts `ca` and `next` for the link from now on, ending every link
+    /// on a CA no longer among them.
+    pub fn set_link_ca(&self, ca: Pin, next: Option<Pin>) {
+        let trusted = trusted_cas(ca, next);
+        self.link_ca.send_if_modified(|current| {
+            let changed = *current != trusted;
+            *current = trusted;
+            changed
+        });
+    }
+
     pub fn conn_id(&self) -> u64 {
         self.next_conn.fetch_add(1, Ordering::Relaxed)
     }
@@ -230,6 +257,12 @@ impl App {
             last_error: self.last_error.lock().expect("last_error").clone(),
             self_upgrade: self.self_upgrade,
             link_from: None,
+            link_ca_pins: self
+                .link_ca
+                .borrow()
+                .iter()
+                .map(|pin| b64::encode(pin))
+                .collect(),
         }
     }
 

crates/relay/src/link/jmap.rs

@@ -14,6 +14,7 @@ use crate::{app::App, link::tls::cert_pin, outbound::Destroyed, state::JoinStage
 use inbuxa_relay_proto::{
     CAPABILITY, CORE_CAPABILITY, b64,
     jmap::{Invocation, MethodError, Request, Response, method},
+    pin::pins_match,
     recipient::{RecipientDomain, RecipientHash},
     time, token,
     types::{NewOutbound, RELAY_SINGLETON, RelaySettings, RelayStatus, Verdict, type_name},
@@ -42,6 +43,8 @@ pub struct RequestError {
 pub struct Caller {
     pub conn: u64,
     pub peer: std::net::IpAddr,
+    /// The link CA its certificate is from (RL-34).
+    pub ca: inbuxa_relay_proto::pin::Pin,
 }
 
 type Failure = (MethodError, Option<String>);
@@ -106,7 +109,7 @@ fn call_one(app: &App, caller: &Caller, call: &Invocation) -> MethodResult {
     match call.name() {
         method::RELAY_JOIN => relay_join(app, args),
         method::RELAY_GET => relay_get(app, caller, args),
-        method::RELAY_SET => relay_set(app, args),
+        method::RELAY_SET => relay_set(app, caller, args),
         method::RELAY_UPGRADE => relay_upgrade(app, args),
         method::RECIPIENT_DOMAIN_GET => recipient_domain_get(app),
         method::RECIPIENT_DOMAIN_SET => recipient_domain_set(app, args),
@@ -266,7 +269,7 @@ fn relay_get(app: &App, caller: &Caller, args: &Map<String, Value>) -> MethodRes
 /// Read-only on `Relay`: what only the relay sets.
 const RELAY_READ_ONLY: &[&str] = &["id", "status", "relayVersion"];
 
-fn relay_set(app: &App, args: &Map<String, Value>) -> MethodResult {
+fn relay_set(app: &App, caller: &Caller, args: &Map<String, Value>) -> MethodResult {
     let old_state = app.relay_state();
     if let Some(if_in) = args.get("ifInState").and_then(Value::as_str)
         && if_in != old_state
@@ -292,7 +295,7 @@ fn relay_set(app: &App, args: &Map<String, Value>) -> MethodResult {
                 );
                 continue;
             }
-            match apply_settings(app, patch) {
+            match apply_settings(app, caller, patch) {
                 Ok(()) => {
                     updated.insert(id.clone(), Value::Null);
                 }
@@ -386,7 +389,7 @@ fn relay_upgrade(app: &App, args: &Map<String, Value>) -> MethodResult {
     }))
 }
 
-fn apply_settings(app: &App, patch: &Value) -> Result<(), Value> {
+fn apply_settings(app: &App, caller: &Caller, patch: &Value) -> Result<(), Value> {
     let Value::Object(patch) = patch else {
         return Err(set_error("invalidPatch", "a patch is an object"));
     };
@@ -435,10 +438,52 @@ fn apply_settings(app: &App, patch: &Value) -> Result<(), Value> {
             format!("smtpCertificate: {e}"),
         ));
     }
+    // The link CA (RL-34). A link on the downstream's next CA that pushes
+    // it as null makes it the relay's only one; nothing else may take it
+    // from under that link.
+    let ca = app.identity.lock().expect("identity").ca_pin;
+    if settings
+        .next_link_ca_pin
+        .is_some_and(|next| pins_match(&next, &ca))
+    {
+        return Err(set_error(
+            "invalidProperties",
+            "nextLinkCaPin: the relay trusts that CA already",
+        ));
+    }
+    let on_next = !pins_match(&caller.ca, &ca)
+        && current
+            .next_link_ca_pin
+            .is_some_and(|next| pins_match(&next, &caller.ca));
+    let promote = match settings.next_link_ca_pin {
+        None => on_next,
+        Some(next) if on_next && !pins_match(&next, &caller.ca) => {
+            return Err(set_error(
+                "invalidProperties",
+                "nextLinkCaPin: a link on the next CA can only make it the only one",
+            ));
+        }
+        Some(_) => false,
+    };
+    if promote {
+        // Saved first: if the relay stops before the settings are, it
+        // trusts the new CA twice, not the old one alone
+        let mut identity = app.identity.lock().expect("identity");
+        identity.ca_pin = caller.ca;
+        identity
+            .save(&app.paths)
+            .map_err(|e| set_error("serverFail", e.to_string()))?;
+    }
     crate::state::save_settings(&app.paths, &settings)
         .map_err(|e| set_error("serverFail", e.to_string()))?;
     app.link_allow_from_changed(&settings.link_allow_from);
+    let next = settings.next_link_ca_pin;
     app.settings.send_replace(Some(Arc::new(settings)));
+    // Links on a CA dropped here end now
+    app.set_link_ca(if promote { caller.ca } else { ca }, next);
+    if promote {
+        app.record("relay.link-ca-replaced", Map::new());
+    }
 
     // The first Relay/set after the proof completes the join.
     let mut identity = app.identity.lock().expect("identity");

crates/relay/src/link/mod.rs

@@ -29,6 +29,7 @@ use inbuxa_relay_proto::{
         self as proto, Account, ClientFrame, ServerFrame, Session, WEBSOCKET_CAPABILITY,
         websocket_capability,
     },
+    pin::{Pin, pins_match},
     types::{Capability, type_name},
 };
 use std::{
@@ -111,6 +112,24 @@ impl Handshakes {
     }
 }
 
+/// Resolves once `ca` is no longer trusted for the link (RL-34): a link on
+/// it then ends.
+async fn untrusted(app: &App, ca: Pin) {
+    let mut trusted = app.link_ca.subscribe();
+    loop {
+        if !trusted
+            .borrow_and_update()
+            .iter()
+            .any(|pin| pins_match(pin, &ca))
+        {
+            return;
+        }
+        if trusted.changed().await.is_err() {
+            return std::future::pending().await;
+        }
+    }
+}
+
 /// Runs the link listener until the process ends.
 pub async fn serve(app: Arc<App>, listener: TcpListener) -> std::io::Result<()> {
     let config = {
@@ -118,7 +137,7 @@ pub async fn serve(app: Arc<App>, listener: TcpListener) -> std::io::Result<()>
         tls::server_config(
             &identity.link_key_pem,
             &identity.link_cert_pem,
-            identity.ca_pin,
+            app.link_ca.subscribe(),
         )
         .map_err(std::io::Error::other)?
     };
@@ -154,10 +173,18 @@ pub async fn serve(app: Arc<App>, listener: TcpListener) -> std::io::Result<()>
                     // peer gets a TLS alert and nothing more.
                     _ => return,
                 };
+            // Which of the trusted CAs it came in on (RL-34)
+            let Some(ca) = tls.get_ref().1.peer_certificates().and_then(|chain| {
+                tls::client_ca(chain.get(1..).unwrap_or_default(), &app.link_ca.borrow())
+                    .map(|(_, pin)| pin)
+            }) else {
+                return;
+            };
             let conn = app.conn_id();
             let caller = jmap::Caller {
                 conn,
                 peer: peer.ip().to_canonical(),
+                ca,
             };
             let service = {
                 let app = app.clone();
@@ -166,12 +193,15 @@ pub async fn serve(app: Arc<App>, listener: TcpListener) -> std::io::Result<()>
                     async move { Ok::<_, std::convert::Infallible>(route(app, caller, req).await) }
                 })
             };
-            let _ = hyper::server::conn::http1::Builder::new()
+            let connection = hyper::server::conn::http1::Builder::new()
                 .timer(hyper_util::rt::TokioTimer::new())
                 .header_read_timeout(Duration::from_secs(30))
                 .serve_connection(TokioIo::new(tls), service)
-                .with_upgrades()
-                .await;
+                .with_upgrades();
+            tokio::select! {
+                _ = connection => {}
+                _ = untrusted(&app, ca) => {}
+            }
             let released = app.spool.lock().expect("spool").release_conn(conn);
             if released {
                 app.inbound_changed();
@@ -438,8 +468,11 @@ where
         live: false,
     };
     let idle = Duration::from_secs(120);
+    let cut = untrusted(&app, caller.ca);
+    tokio::pin!(cut);
     loop {
         tokio::select! {
+            _ = &mut cut => break,
             frame = tokio::time::timeout(idle, stream.next()) => {
                 let message = match frame {
                     Ok(Some(Ok(message))) => message,

crates/relay/src/link/tls.rs

@@ -17,6 +17,7 @@ use rustls::{
 };
 use rustls_pki_types::{CertificateDer, PrivateKeyDer, UnixTime, pem::PemObject};
 use std::sync::Arc;
+use tokio::sync::watch;
 
 pub fn provider() -> Arc<CryptoProvider> {
     Arc::new(rustls::crypto::aws_lc_rs::default_provider())
@@ -47,7 +48,7 @@ pub fn new_link_identity(relay_id: &str) -> Result<(String, String), rcgen::Erro
 pub fn server_config(
     key_pem: &str,
     cert_pem: &str,
-    ca_pin: Pin,
+    ca_pins: watch::Receiver<Vec<Pin>>,
 ) -> Result<rustls::ServerConfig, String> {
     let provider = provider();
     let cert = CertificateDer::from_pem_slice(cert_pem.as_bytes())
@@ -55,7 +56,7 @@ pub fn server_config(
     let key =
         PrivateKeyDer::from_pem_slice(key_pem.as_bytes()).map_err(|e| format!("link key: {e}"))?;
     let verifier = Arc::new(PinnedCaVerifier {
-        ca_pin,
+        ca_pins,
         algorithms: provider.signature_verification_algorithms,
     });
     let mut config = rustls::ServerConfig::builder_with_provider(provider)
@@ -68,12 +69,27 @@ pub fn server_config(
     Ok(config)
 }
 
+/// The pinned CA among the certificates a client sent with its own, if one
+/// is there.
+pub fn client_ca<'a>(
+    intermediates: &'a [CertificateDer<'a>],
+    pins: &[Pin],
+) -> Option<(&'a CertificateDer<'a>, Pin)> {
+    intermediates.iter().find_map(|c| {
+        let pin = cert_pin(c)?;
+        pins.iter()
+            .any(|trusted| pins_match(&pin, trusted))
+            .then_some((c, pin))
+    })
+}
+
 /// Accepts a client certificate only if one of the certificates the client
-/// sent has the pinned CA key and signed the client's own certificate,
-/// which must be valid now and for client authentication.
+/// sent has a pinned CA key and signed the client's own certificate,
+/// which must be valid now and for client authentication. The pins are the
+/// relay's CA and, while the downstream replaces it, the next one (RL-34).
 #[derive(Debug)]
 pub struct PinnedCaVerifier {
-    ca_pin: Pin,
+    ca_pins: watch::Receiver<Vec<Pin>>,
     algorithms: WebPkiSupportedAlgorithms,
 }
 
@@ -96,12 +112,10 @@ impl ClientCertVerifier for PinnedCaVerifier {
         intermediates: &[CertificateDer<'_>],
         now: UnixTime,
     ) -> Result<ClientCertVerified, Error> {
-        let ca = intermediates
-            .iter()
-            .find(|c| cert_pin(c).is_some_and(|pin| pins_match(&pin, &self.ca_pin)))
-            .ok_or(Error::InvalidCertificate(
-                rustls::CertificateError::UnknownIssuer,
-            ))?;
+        let pins = self.ca_pins.borrow().clone();
+        let (ca, _) = client_ca(intermediates, &pins).ok_or(Error::InvalidCertificate(
+            rustls::CertificateError::UnknownIssuer,
+        ))?;
         let anchor = webpki::anchor_from_trusted_cert(ca)
             .map_err(|_| Error::InvalidCertificate(rustls::CertificateError::BadEncoding))?;
         let ee = webpki::EndEntityCert::try_from(end_entity)
@@ -187,7 +201,7 @@ pub mod tests {
 
     fn verifier(pin: Pin) -> PinnedCaVerifier {
         PinnedCaVerifier {
-            ca_pin: pin,
+            ca_pins: watch::channel(vec![pin]).1,
             algorithms: provider().signature_verification_algorithms,
         }
     }
@@ -222,6 +236,6 @@ pub mod tests {
         let (key, cert) = new_link_identity("r1").unwrap();
         let der = CertificateDer::from_pem_slice(cert.as_bytes()).unwrap();
         assert!(cert_pin(&der).is_some());
-        assert!(server_config(&key, &cert, [1; 32]).is_ok());
+        assert!(server_config(&key, &cert, watch::channel(vec![[1; 32]]).1).is_ok());
     }
 }

crates/relay/src/tests_e2e.rs

@@ -797,6 +797,69 @@ async fn relay_end_to_end() {
         assert!(kinds.contains(&kind), "{kind} in {kinds:?}");
     }
 
+    // RL-T26: a new link CA without a new join. The downstream pushes the
+    // next CA's pin, and the relay trusts both...
+    let next_ca = test_ca(true);
+    let next_pin = cert_pin(&next_ca.ca_der).unwrap();
+    let pins = |relay: &Value| relay["list"][0]["status"]["linkCaPins"].clone();
+    let set_next = async |ws: &mut Ws, pin: Value| {
+        let set = ws
+            .ok(
+                "Relay/set",
+                json!({ "update": { "singleton": { "nextLinkCaPin": pin }}}),
+            )
+            .await;
+        set["updated"] == json!({ "singleton": null })
+    };
+    assert!(
+        !set_next(&mut ws, json!(b64::encode(&ca_pin))).await,
+        "its own CA"
+    );
+    assert!(set_next(&mut ws, json!(b64::encode(&next_pin))).await);
+    let relay = ws.ok("Relay/get", json!({})).await;
+    assert_eq!(
+        pins(&relay),
+        json!([b64::encode(&ca_pin), b64::encode(&next_pin)])
+    );
+    let next_down = Downstream::new(&next_ca, link_addr);
+    let mut next_ws = next_down.ws().await.unwrap();
+    // ...a link on the next CA can't point it elsewhere...
+    assert!(!set_next(&mut next_ws, json!(b64::encode(&[5u8; 32]))).await);
+    // ...and pushing it as null from there makes it the only one
+    assert!(set_next(&mut next_ws, Value::Null).await);
+    let relay = next_ws.ok("Relay/get", json!({})).await;
+    assert_eq!(pins(&relay), json!([b64::encode(&next_pin)]));
+    assert_eq!(Identity::load(&paths).unwrap().unwrap().ca_pin, next_pin);
+    // A link on the old CA ends at once, and can't come back
+    let ended = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            match ws.ws.next().await {
+                None | Some(Err(_)) | Some(Ok(Message::Close(_))) => return,
+                Some(Ok(_)) => {}
+            }
+        }
+    })
+    .await;
+    assert!(ended.is_ok(), "the old CA's link was cut");
+    // TLS 1.3 tells the client of a refused certificate on its first read
+    assert!(down.ws().await.is_err());
+    assert!(next_down.ws().await.is_ok());
+    let events = next_ws.ok("RelayEvent/get", json!({})).await;
+    assert!(
+        events["list"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|e| e["type"] == "relay.link-ca-replaced"),
+        "{events}"
+    );
+    // A link on the current CA can start and stop a replacement
+    assert!(set_next(&mut next_ws, json!(b64::encode(&[6u8; 32]))).await);
+    assert!(set_next(&mut next_ws, Value::Null).await);
+    let relay = next_ws.ok("Relay/get", json!({})).await;
+    assert_eq!(pins(&relay), json!([b64::encode(&next_pin)]));
+
     drop(ws);
+    drop(next_ws);
     std::fs::remove_dir_all(dir).unwrap();
 }

crates/services/src/inbuxa_relay_link.rs

@@ -184,10 +184,20 @@ async fn reconcile(
     links: &mut HashMap<u32, Running>,
 ) -> trc::Result<()> {
     let data = server.store();
-    let all = match relays::keys(data).await? {
-        Some(_) => relays::all(data).await?,
-        None => Vec::new(),
+    let (keys, all) = match relays::keys(data).await? {
+        Some(keys) => (Some(keys), relays::all(data).await?),
+        None => (None, Vec::new()),
     };
+    // A link CA replacement every joined relay is ready for (RL-34)
+    if let Some(keys) = &keys
+        && relays::link::link_ca_ready(data, keys, &all).await?
+        && relays::change_link_ca(data, relays::LinkCaChange::Finish, now()).await?
+    {
+        trc::event!(
+            Relay(trc::RelayEvent::LinkCaReplaced),
+            Details = "Every relay trusts the new link CA: it is the only one now."
+        );
+    }
     // Which relays outbound mail goes through (RL-20, RL-24)
     server.inner.data.relay_senders.set_relays(&all);
     // The domains' MX and SPF records, when the relays in them changed (RL-25)
@@ -259,6 +269,7 @@ impl NodeCertificates {
         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
+            && cert.ca_pin == keys.link_ca_pin
         {
             return Ok(cert.clone());
         }
@@ -987,7 +998,8 @@ async fn session(
     .await?;
     let mut sent = Sent::default();
     let mut relay = relay.clone();
-    sync(server, &relay, keys, &dialer, &mut ws, &mut sent).await?;
+    let mut keys = keys.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?;
 
@@ -1065,7 +1077,13 @@ async fn session(
             }
             _ => return Ok(()),
         }
-        sync(server, &relay, keys, &dialer, &mut ws, &mut sent).await?;
+        // A new link CA: a link on a certificate from the old one starts
+        // again with one from the new (RL-34). A next CA, or none, is pushed.
+        match relays::keys(data).await? {
+            Some(now_keys) if now_keys.link_ca_pin == cert.ca_pin => keys = now_keys,
+            _ => return Ok(()),
+        }
+        sync(server, &relay, &keys, &dialer, &mut ws, &mut sent).await?;
         refresh_status(&mut ws, data, &relay, &sent, status).await?;
         taker.relay_name = relay.name.clone();
         taker.take(&mut ws).await?;
@@ -1791,7 +1809,14 @@ async fn sync(
     // 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 mut patch = serde_json::to_value(&settings).unwrap_or_default();
+        // The relay merges what it is sent: a setting left out stays as it
+        // was, so one this server has cleared goes as null
+        if let Some(patch) = patch.as_object_mut() {
+            for cleared in ["smtpCertificate", "nextLinkCaPin"] {
+                patch.entry(cleared).or_insert(Value::Null);
+            }
+        }
         let answer = ws
             .ok(
                 method::RELAY_SET,

crates/trc/src/event/enums.rs

@@ -13,10 +13,10 @@
 // security.legacy-protocols-changed (LP-8); 644 to 646 are the cluster
 // coordinator's connection events; 647 and 648 are the audit log's
 // (audit-hold-lock spec, AU-3, AU-8); 649 is cluster.node-id-reassigned;
-// 650 to 658 are inbuxa Relay's (relay spec, RL-23, RL-T23).
+// 650 to 659 are inbuxa Relay's (relay spec, RL-23, RL-T23).
 // TOTAL_EVENT_COUNT sizes the collector's per-event tables, so it is one
 // more than the highest id; the test below holds it to the variant list
-pub const TOTAL_EVENT_COUNT: usize = 659;
+pub const TOTAL_EVENT_COUNT: usize = 660;
 pub const TOTAL_METRIC_COUNT: usize = 369;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -659,6 +659,8 @@ pub enum RelayEvent {
     Event = 656,
     LinkDown = 657,
     LinkUp = 658,
+    /// Every joined relay trusts the new link CA: it is the only one now (RL-34).
+    LinkCaReplaced = 659,
 }
 
 // inbuxa: SCIM-54: every write an identity provider makes

crates/trc/src/event/enums_impl.rs

@@ -48,6 +48,7 @@ impl EventType {
             b"relay.event" => EventType::Relay(RelayEvent::Event),
             b"relay.link-down" => EventType::Relay(RelayEvent::LinkDown),
             b"relay.link-up" => EventType::Relay(RelayEvent::LinkUp),
+            b"relay.link-ca-replaced" => EventType::Relay(RelayEvent::LinkCaReplaced),
             // inbuxa: SCIM-54
             b"scim.resource-created" => EventType::Scim(ScimEvent::ResourceCreated),
             b"scim.resource-updated" => EventType::Scim(ScimEvent::ResourceUpdated),
@@ -719,6 +720,7 @@ impl EventType {
             EventType::Relay(RelayEvent::Event) => "relay.event",
             EventType::Relay(RelayEvent::LinkDown) => "relay.link-down",
             EventType::Relay(RelayEvent::LinkUp) => "relay.link-up",
+            EventType::Relay(RelayEvent::LinkCaReplaced) => "relay.link-ca-replaced",
             // inbuxa: SCIM-54
             EventType::Scim(ScimEvent::ResourceCreated) => "scim.resource-created",
             EventType::Scim(ScimEvent::ResourceUpdated) => "scim.resource-updated",
@@ -1531,6 +1533,7 @@ impl EventType {
             EventType::Relay(RelayEvent::Event) => 656,
             EventType::Relay(RelayEvent::LinkDown) => 657,
             EventType::Relay(RelayEvent::LinkUp) => 658,
+            EventType::Relay(RelayEvent::LinkCaReplaced) => 659,
             // inbuxa: SCIM-54
             EventType::Scim(ScimEvent::ResourceCreated) => 637,
             EventType::Scim(ScimEvent::ResourceUpdated) => 638,
@@ -2201,6 +2204,7 @@ impl EventType {
             656 => Some(EventType::Relay(RelayEvent::Event)),
             657 => Some(EventType::Relay(RelayEvent::LinkDown)),
             658 => Some(EventType::Relay(RelayEvent::LinkUp)),
+            659 => Some(EventType::Relay(RelayEvent::LinkCaReplaced)),
             // inbuxa: SCIM-54
             637 => Some(EventType::Scim(ScimEvent::ResourceCreated)),
             638 => Some(EventType::Scim(ScimEvent::ResourceUpdated)),
@@ -3193,6 +3197,7 @@ impl EventType {
             EventType::Relay(RelayEvent::Event) => Level::Info,
             EventType::Relay(RelayEvent::LinkDown) => Level::Warn,
             EventType::Relay(RelayEvent::LinkUp) => Level::Info,
+            EventType::Relay(RelayEvent::LinkCaReplaced) => Level::Info,
             // inbuxa: SCIM-54
             EventType::Scim(ScimEvent::ResourceCreated) => Level::Info,
             EventType::Scim(ScimEvent::ResourceUpdated) => Level::Info,
@@ -3294,6 +3299,7 @@ impl EventType {
             EventType::Relay(RelayEvent::Event) => "Relay event",
             EventType::Relay(RelayEvent::LinkDown) => "Relay link down for an hour",
             EventType::Relay(RelayEvent::LinkUp) => "Relay link back",
+            EventType::Relay(RelayEvent::LinkCaReplaced) => "Relay link CA replaced",
             // inbuxa: SCIM-54
             EventType::Scim(ScimEvent::ResourceCreated) => "SCIM resource created",
             EventType::Scim(ScimEvent::ResourceUpdated) => "SCIM resource updated",
@@ -4409,6 +4415,7 @@ impl EventType {
             EventType::Relay(RelayEvent::Event),
             EventType::Relay(RelayEvent::LinkDown),
             EventType::Relay(RelayEvent::LinkUp),
+            EventType::Relay(RelayEvent::LinkCaReplaced),
             // inbuxa: SCIM-54
             EventType::Scim(ScimEvent::ResourceCreated),
             EventType::Scim(ScimEvent::ResourceUpdated),

docs/spec/features/relay.md

@@ -170,6 +170,30 @@ the protocol the rest of the suite already speaks.
   CA. From then on a certificate from the old one fails the handshake.
   A relay that missed the change, because it was down meanwhile, still
   has the old pin and needs a new join; inbuxa-admin lists it.
+
+  How it goes:
+  - An update of any `inbuxa:Relay` with `replaceLinkCa`, sent on its
+    own, acts on the CA they all share: `start` makes the next CA,
+    `finish` makes it the link CA without waiting, and `cancel` drops it.
+    One that changes nothing (a second `start`, a `finish` with none
+    under way) is refused. Every relay shows `linkCaReplacing`, when the
+    replacement began, and its own `linkCa`: `current`, `both`, or `old`
+    (only a CA this server replaced, from what its links last said, or a
+    join whose token named one: it needs a new join), null before it has
+    said.
+  - The link service finishes it by itself once every enabled relay that
+    has joined says `both`, and records `relay.link-ca-replaced`. A
+    relay that never does holds it until an administrator finishes it.
+  - Each node then takes a certificate from the new CA and links again
+    with it. Its settings push carries `nextLinkCaPin: null`; from a link
+    on the next CA, that makes the next CA the relay's only one: the
+    relay saves it, ends every link on the old CA, and records
+    `relay.link-ca-replaced` too. From a link on the current CA, null
+    just drops the next one (a cancel). A link on the next CA can't name
+    another. Settings a server has cleared go to the relay as null, as a
+    setting left out of a push stays as it was.
+  - A join started before the replacement finished names the old CA in
+    its token and can't complete: start it again.
 - **RL-35. The downstream's relay keys, wrapped.** The link CA key, the
   sealing key and each relay's recipient key are stored encrypted under a
   wrapping key the store doesn't hold: a file every node reads, named by
@@ -261,7 +285,7 @@ with the server's own `inbuxa:` objects. In particular, the relay's
 | `sealingKey` | downstream | The downstream's public sealing key |
 | `linkAllowFrom` | downstream | Addresses and CIDR ranges that may dial the link port. Default empty: anyone (RL-33) |
 | `nextLinkCaPin` | downstream | A second link CA to trust during a replacement (RL-34). Default null |
-| `status` | relay | `{startedAt, linkClients, mode, spool: {inbound, outbound, bytes}, recipientState, certificateNotAfter, linkCaPins, lastError, selfUpgrade, linkFrom}`; `linkCaPins` lists the CAs it trusts (RL-34); `selfUpgrade` is false in a container image (RL-29); `linkFrom` is the address the asking link connection comes from (RL-33) |
+| `status` | relay | `{startedAt, linkClients, mode, spool: {inbound, outbound, bytes}, recipientState, certificateNotAfter, linkCaPins, lastError, selfUpgrade, linkFrom}`; `linkCaPins` lists the CAs it trusts, its own first (RL-34); `selfUpgrade` is false in a container image (RL-29); `linkFrom` is the address the asking link connection comes from (RL-33) |
 
 Settings are cached in the state directory as the downstream last set them,
 so a reboot without the link comes back with the same settings (RL-22).
@@ -513,7 +537,7 @@ When no link is up, or after RL-13:
 
 | Object | What it holds |
 |---|---|
-| `inbuxa:Relay` | One per relay: `name`, `address`, `linkPort`, `enabled`, the settings pushed to the relay, `sendThrough` (route outbound mail through relays), `inboundOnly` (takes mail in but sends none, e.g. while its IP isn't warmed up or isn't in SPF yet), `mxPreference` (its MX preference, 10 by default, RL-25), `certificateNotAfter` and `certificateError` (when its certificate runs out, and why the last order failed, RL-27), `olderThanServer` (a link last heard an older release than this server's, RL-28), and read-only status mirrored from the relay: link state per node, version, queue counts, `lastSeen`. Create, or an update with `rejoin: true`, returns `joinToken` once. An update with `upgrade: {blobId, signature}`, sent on its own, hands the relay a release over the node's link (RL-29) |
+| `inbuxa:Relay` | One per relay: `name`, `address`, `linkPort`, `enabled`, the settings pushed to the relay, `sendThrough` (route outbound mail through relays), `inboundOnly` (takes mail in but sends none, e.g. while its IP isn't warmed up or isn't in SPF yet), `mxPreference` (its MX preference, 10 by default, RL-25), `certificateNotAfter` and `certificateError` (when its certificate runs out, and why the last order failed, RL-27), `olderThanServer` (a link last heard an older release than this server's, RL-28), `linkCa` and `linkCaReplacing` (RL-34), and read-only status mirrored from the relay: link state per node, version, queue counts, `lastSeen`. Create, or an update with `rejoin: true`, returns `joinToken` once. An update with `upgrade: {blobId, signature}`, sent on its own, hands the relay a release over the node's link (RL-29). An update with `replaceLinkCa`, sent on its own, starts, finishes or cancels a replacement of the shared link CA (RL-34) |
 | `inbuxa:RelayMessage` | Read-only view of a relay's `Inbound` and `Outbound` queues, fetched over the link when asked. Never message content. `/get`, `/query` (filter `relayId`, `direction`; newest first) and `/set` destroy only: discards held mail, or cancels an outbound message, whose recipients fail here with no DSN as a delivery cancelled in the queue does. Each id is the relay's id and a hash of the relay's own message id, so nothing is stored |
 
 Permissions: `sysRelayGet`, `sysRelayUpdate`, `sysRelayQueueGet`,
@@ -829,7 +853,8 @@ In progress. Done so far:
   `relay.event` for any other type), then acknowledges it. A node with no
   link to a joined relay for an hour raises `relay.link-down` (a warning),
   once, and `relay.link-up` when it is back (RL-T23). Both reach the log,
-  webhooks and OpenTelemetry.
+  webhooks and OpenTelemetry. `relay.link-ca-replaced` (659) records a
+  finished link CA replacement (RL-34).
 - `Outbound` on the relay: `Outbound/set` create and destroy, `/get` and
   `/changes`, with push, in `spool/out/` (RL-19, RL-T10).
 - Outbound delivery on the relay: MX, DANE, MTA-STS and `outboundTls`, per

resources/schema/schema.json.sha256

@@ -1 +1 @@
-au254ZOQm1gGMzM7Qp3nUep_vmcV_jOrKB7fraBxdNc
+bxhi7hGulmWKjZiUh2eaxDH78ZzRFaijGqkp16KrLXc

tests/src/system/relay.rs

@@ -266,6 +266,19 @@ pub async fn test(test: &mut TestServer) {
             .is_some_and(|d| d.contains("relay.example is not an address or CIDR range")),
         "{response}"
     );
+    // The next link CA is the server's to push (RL-34)
+    let (_, response) = call(
+        admin,
+        "inbuxa:Relay/set",
+        json!({"update": {&id: {"settings/nextLinkCaPin": "K28RckuyO0ozszKIofsQT6Ka3nR2k76IuCMLtio1hEU"}}}),
+    )
+    .await;
+    assert_eq!(
+        response["notUpdated"][&id]["properties"][0], "settings",
+        "{response}"
+    );
+    // A relay still joining has said nothing of its CA
+    assert_eq!(get(admin, &id).await["linkCa"], Value::Null);
     let (_, response) = call(
         admin,
         "inbuxa:Relay/set",
@@ -1396,6 +1409,50 @@ pub async fn link(test: &mut TestServer) {
         "one DSN, for the refusal"
     );
 
+    // RL-T26: a new link CA, without a new join. Sent alone, and only
+    // when it changes something.
+    let replace = async |change: Value, extra: Value| {
+        let mut update = json!({"replaceLinkCa": change});
+        if let Value::Object(extra) = extra {
+            update.as_object_mut().unwrap().extend(extra);
+        }
+        let (_, response) = call(admin, "inbuxa:Relay/set", json!({"update": {&id: update}})).await;
+        response["updated"].get(&id).is_some()
+    };
+    assert!(!replace(json!("start"), json!({"description": "and this"})).await);
+    assert!(!replace(json!("now"), Value::Null).await);
+    assert!(
+        !replace(json!("cancel"), Value::Null).await,
+        "none to cancel"
+    );
+    let relay = get(admin, &id).await;
+    assert_eq!(relay["linkCa"], "current", "{relay}");
+    assert_eq!(relay["linkCaReplacing"], Value::Null, "{relay}");
+    let old_pin = relay["links"][0]["relay"]["linkCaPins"][0].clone();
+    assert!(old_pin.is_string(), "{relay}");
+    assert!(replace(json!("start"), Value::Null).await);
+    assert!(!replace(json!("start"), Value::Null).await, "one at a time");
+    assert!(get(admin, &id).await["linkCaReplacing"].is_string());
+    // The relay trusts both, so the server finishes it: every node takes a
+    // certificate from the new CA, and the relay makes it its only one
+    let relay = wait_for(admin, &id, "the relay took the new link CA", 90, |relay| {
+        relay["linkCaReplacing"].is_null()
+            && relay["linkCa"] == "current"
+            && relay["links"].as_array().into_iter().flatten().any(|link| {
+                link["state"] == "linked"
+                    && link["relay"]["linkCaPins"]
+                        .as_array()
+                        .is_some_and(|pins| pins.len() == 1 && pins[0] != old_pin)
+            })
+    })
+    .await;
+    let new_pin = relay["links"][0]["relay"]["linkCaPins"][0].clone();
+    let identity: Value =
+        serde_json::from_str(&std::fs::read_to_string(state.join("identity.json")).unwrap())
+            .unwrap();
+    assert_eq!(identity["caPin"], new_pin, "the relay keeps the new CA");
+    assert_eq!(relay["joinState"], "joined", "without a new join");
+
     // The relay's events were taken as this server's relay.* trace events,
     // and acknowledged, so the relay forgot them
     let events = state.join("events.jsonl");