diff --git a/crates/features/src/relay/link.rs b/crates/features/src/relay/link.rs index 601e9621..bdb5b7f2 100644 --- a/crates/features/src/relay/link.rs +++ b/crates/features/src/relay/link.rs @@ -38,6 +38,8 @@ pub struct NodeCertificate { pub key_der: Vec, pub ca_der: Vec, 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> 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 { + 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 { + 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, 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(¤t)), + Some(LinkCa::Current) + ); + assert_eq!( + link_ca(&joined, &replacing, std::slice::from_ref(¤t)), + 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)); + } } diff --git a/crates/features/src/relay/mod.rs b/crates/features/src/relay/mod.rs index 20ef599f..fab07d45 100644 --- a/crates/features/src/relay/mod.rs +++ b/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, +} + +/// 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 { + 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 { + 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, #[serde(skip_serializing_if = "Option::is_none")] pub joined_at: Option, + /// 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, } #[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 { + 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 { 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); @@ -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 { + let Json(keys) = Json::::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 { + keys_or_new(data, now).await?; + loop { + let Some(VersionedKeys(keys, version)) = data + .get_value::(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 { let Json(relay) = Json::::deserialize(bytes)?; diff --git a/crates/jmap-proto/src/object/inbuxa_relay.rs b/crates/jmap-proto/src/object/inbuxa_relay.rs index a5df6c81..c6589c88 100644 --- a/crates/jmap-proto/src/object/inbuxa_relay.rs +++ b/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, diff --git a/crates/jmap/src/inbuxa/relay.rs b/crates/jmap/src/inbuxa/relay.rs index 07b66e54..fa3fff1c 100644 --- a/crates/jmap/src/inbuxa/relay.rs +++ b/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, + replace_link_ca: Option, } /// 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> { .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() { diff --git a/crates/relay-proto/src/types.rs b/crates/relay-proto/src/types.rs index a44c8247..3884dd7c 100644 --- a/crates/relay-proto/src/types.rs +++ b/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, + /// 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, /// The STARTTLS certificate for port 25. #[serde(skip_serializing_if = "Option::is_none")] pub smtp_certificate: Option, @@ -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, + /// 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, } /// Live while a link client has push on for `Inbound` (spec "Live mode"). diff --git a/crates/relay/src/app.rs b/crates/relay/src/app.rs index 51c980b9..8fdfb59c 100644 --- a/crates/relay/src/app.rs +++ b/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) -> Vec { + 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>>, + /// 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>, } 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) { + 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(), } } diff --git a/crates/relay/src/link/jmap.rs b/crates/relay/src/link/jmap.rs index b698ea11..8ba987af 100644 --- a/crates/relay/src/link/jmap.rs +++ b/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); @@ -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) -> 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) -> MethodResult { +fn relay_set(app: &App, caller: &Caller, args: &Map) -> 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) -> 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) -> 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"); diff --git a/crates/relay/src/link/mod.rs b/crates/relay/src/link/mod.rs index a0a11d43..58241fc1 100644 --- a/crates/relay/src/link/mod.rs +++ b/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, listener: TcpListener) -> std::io::Result<()> { let config = { @@ -118,7 +137,7 @@ pub async fn serve(app: Arc, 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, 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, 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, diff --git a/crates/relay/src/link/tls.rs b/crates/relay/src/link/tls.rs index 2ab22518..afed6bde 100644 --- a/crates/relay/src/link/tls.rs +++ b/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 { 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>, ) -> Result { 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>, algorithms: WebPkiSupportedAlgorithms, } @@ -96,12 +112,10 @@ impl ClientCertVerifier for PinnedCaVerifier { intermediates: &[CertificateDer<'_>], now: UnixTime, ) -> Result { - 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()); } } diff --git a/crates/relay/src/tests_e2e.rs b/crates/relay/src/tests_e2e.rs index 8d8db001..a729b4ce 100644 --- a/crates/relay/src/tests_e2e.rs +++ b/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(); } diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index 0f9ee951..c26203c3 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/crates/services/src/inbuxa_relay_link.rs @@ -184,10 +184,20 @@ async fn reconcile( links: &mut HashMap, ) -> 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, diff --git a/crates/trc/src/event/enums.rs b/crates/trc/src/event/enums.rs index 3c48b0b5..ae188bf4 100644 --- a/crates/trc/src/event/enums.rs +++ b/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 diff --git a/crates/trc/src/event/enums_impl.rs b/crates/trc/src/event/enums_impl.rs index 18cfa2e1..8c3c6e65 100644 --- a/crates/trc/src/event/enums_impl.rs +++ b/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), diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index a6b6133e..f6468c31 100644 --- a/docs/spec/features/relay.md +++ b/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 diff --git a/resources/schema/schema.json.gz b/resources/schema/schema.json.gz index 82cf0cc1..a70d5df1 100644 Binary files a/resources/schema/schema.json.gz and b/resources/schema/schema.json.gz differ diff --git a/resources/schema/schema.json.sha256 b/resources/schema/schema.json.sha256 index d476303e..664edecc 100644 --- a/resources/schema/schema.json.sha256 +++ b/resources/schema/schema.json.sha256 @@ -1 +1 @@ -au254ZOQm1gGMzM7Qp3nUep_vmcV_jOrKB7fraBxdNc \ No newline at end of file +bxhi7hGulmWKjZiUh2eaxDH78ZzRFaijGqkp16KrLXc diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 2b59d554..9331a352 100644 --- a/tests/src/system/relay.rs +++ b/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");