Relay: signed upgrades (RL-29), and older relays shown (RL-28) #270
No files matched your search
@@ -57,6 +57,10 @@ pub trait Sender: Send + Sync {
|
||||
/// One method on the relay's API, on its account: the answer's
|
||||
/// arguments, or why there are none (the relay's error included).
|
||||
fn call(&self, method: &'static str, args: serde_json::Value) -> Call<'_>;
|
||||
|
||||
/// Uploads a release binary and calls `Relay/upgrade` with it and its
|
||||
/// signature file (RL-29): the answer's arguments.
|
||||
fn upgrade(&self, binary: Vec<u8>, signature: String) -> Call<'_>;
|
||||
}
|
||||
|
||||
/// A relay this node is linked to.
|
||||
@@ -266,6 +270,10 @@ mod tests {
|
||||
fn call(&self, _: &'static str, _: serde_json::Value) -> Call<'_> {
|
||||
Box::pin(async { Err("no".into()) })
|
||||
}
|
||||
|
||||
fn upgrade(&self, _: Vec<u8>, _: String) -> Call<'_> {
|
||||
Box::pin(async { Err("no".into()) })
|
||||
}
|
||||
}
|
||||
|
||||
fn relay(id: u32, send_through: bool, inbound_only: bool) -> Relay {
|
||||
|
||||
@@ -50,6 +50,12 @@ pub enum RelayProperty {
|
||||
LinkKeyPin,
|
||||
/// Update only: `true` starts a new join, with a new token.
|
||||
Rejoin,
|
||||
/// Update only: `{blobId, signature}`, a signed release the relay
|
||||
/// installs (RL-29).
|
||||
Upgrade,
|
||||
/// Whether a linked relay runs an older release than this server
|
||||
/// (RL-28); null before any link has said.
|
||||
OlderThanServer,
|
||||
/// In a create or rejoin response only.
|
||||
JoinToken,
|
||||
/// Each node's link to the relay, and what the relay last said.
|
||||
@@ -94,6 +100,8 @@ impl Property for RelayProperty {
|
||||
RelayProperty::JoinedAt => "joinedAt",
|
||||
RelayProperty::LinkKeyPin => "linkKeyPin",
|
||||
RelayProperty::Rejoin => "rejoin",
|
||||
RelayProperty::Upgrade => "upgrade",
|
||||
RelayProperty::OlderThanServer => "olderThanServer",
|
||||
RelayProperty::JoinToken => "joinToken",
|
||||
RelayProperty::Links => "links",
|
||||
RelayProperty::CreatedBy => "createdBy",
|
||||
@@ -126,6 +134,8 @@ impl RelayProperty {
|
||||
b"joinedAt" => RelayProperty::JoinedAt,
|
||||
b"linkKeyPin" => RelayProperty::LinkKeyPin,
|
||||
b"rejoin" => RelayProperty::Rejoin,
|
||||
b"upgrade" => RelayProperty::Upgrade,
|
||||
b"olderThanServer" => RelayProperty::OlderThanServer,
|
||||
b"joinToken" => RelayProperty::JoinToken,
|
||||
b"links" => RelayProperty::Links,
|
||||
b"createdBy" => RelayProperty::CreatedBy,
|
||||
|
||||
@@ -8,14 +8,21 @@
|
||||
//! inbuxa Relays this server or cluster links to. A relay serves the whole
|
||||
//! downstream, so nobody in a tenant reaches them. A create returns the
|
||||
//! join token, once; so does an update with `rejoin: true`, which starts
|
||||
//! over with a relay that was reset. The request layer records every
|
||||
//! change in the audit log, with the private key and the token left out.
|
||||
//! over with a relay that was reset. An update with `upgrade` hands the
|
||||
//! relay a signed release over this node's link (RL-29). The request layer
|
||||
//! records every change in the audit log, with the private key and the
|
||||
//! token left out.
|
||||
|
||||
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,
|
||||
};
|
||||
use inbuxa_relay_proto::{b64, types::RelaySettings};
|
||||
use inbuxa_relay_proto::{
|
||||
b64,
|
||||
release::{self, Version},
|
||||
types::RelaySettings,
|
||||
};
|
||||
use jmap_proto::{
|
||||
error::set::SetError,
|
||||
method::{
|
||||
@@ -28,9 +35,9 @@ use jmap_proto::{
|
||||
};
|
||||
use jmap_tools::{Key, Map, Property, Value};
|
||||
use registry::schema::prelude::MASKED_PASSWORD;
|
||||
use std::borrow::Cow;
|
||||
use std::{borrow::Cow, str::FromStr};
|
||||
use store::write::now;
|
||||
use types::id::Id;
|
||||
use types::{blob::BlobId, id::Id};
|
||||
|
||||
type RValue = Value<'static, P, RelayValue>;
|
||||
type JsonMap = serde_json::Map<String, serde_json::Value>;
|
||||
@@ -58,11 +65,15 @@ const ALL: &[P] = &[
|
||||
P::JoinedAt,
|
||||
P::LinkKeyPin,
|
||||
P::Links,
|
||||
P::OlderThanServer,
|
||||
P::CreatedBy,
|
||||
P::CreatedAt,
|
||||
P::UpdatedAt,
|
||||
];
|
||||
|
||||
/// This server's release, which its relays are compared with (RL-28).
|
||||
const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
|
||||
|
||||
/// Properties the server sets; a client that sends them is refused.
|
||||
const SERVER_SET: &[P] = &[
|
||||
P::Id,
|
||||
@@ -74,6 +85,7 @@ const SERVER_SET: &[P] = &[
|
||||
P::LinkKeyPin,
|
||||
P::JoinToken,
|
||||
P::Links,
|
||||
P::OlderThanServer,
|
||||
P::CreatedBy,
|
||||
P::CreatedAt,
|
||||
P::UpdatedAt,
|
||||
@@ -146,6 +158,16 @@ fn links_value(links: &[NodeLink]) -> RValue {
|
||||
json_to_value(serde_json::Value::Array(list))
|
||||
}
|
||||
|
||||
/// Whether any link last heard an older release than this server's.
|
||||
fn older_than_server(links: &[NodeLink]) -> Option<bool> {
|
||||
let server = Version::parse(SERVER_VERSION)?;
|
||||
let versions: Vec<Version> = links
|
||||
.iter()
|
||||
.filter_map(|link| Version::parse(&link.relay.as_ref()?.relay_version))
|
||||
.collect();
|
||||
(!versions.is_empty()).then(|| versions.iter().any(|version| *version < server))
|
||||
}
|
||||
|
||||
fn to_value(
|
||||
relay: &StoredRelay,
|
||||
links: &[NodeLink],
|
||||
@@ -184,6 +206,7 @@ fn to_value(
|
||||
P::JoinedAt => relay.join.joined_at.map_or(Value::Null, date),
|
||||
P::LinkKeyPin => text(relay.join.relay_pin.map(|pin| b64::encode(&pin)).as_deref()),
|
||||
P::Links => links_value(links),
|
||||
P::OlderThanServer => older_than_server(links).map_or(Value::Null, Value::Bool),
|
||||
P::CreatedBy => text(Some(&relay.created_by)),
|
||||
P::CreatedAt => date(relay.created_at),
|
||||
P::UpdatedAt => date(relay.updated_at),
|
||||
@@ -201,6 +224,13 @@ struct Sent {
|
||||
properties: JsonMap,
|
||||
settings: JsonMap,
|
||||
rejoin: bool,
|
||||
upgrade: Option<Upgrade>,
|
||||
}
|
||||
|
||||
/// A release to hand the relay: an upload, and its signature file's text.
|
||||
struct Upgrade {
|
||||
blob_id: String,
|
||||
signature: String,
|
||||
}
|
||||
|
||||
fn client_json(value: Value<'_, P, RelayValue>) -> Result<Sent, SetError<P>> {
|
||||
@@ -220,6 +250,25 @@ fn client_json(value: Value<'_, P, RelayValue>) -> Result<Sent, SetError<P>> {
|
||||
.with_description("true, to start a new join."));
|
||||
}
|
||||
},
|
||||
Key::Property(P::Upgrade) => {
|
||||
let json: serde_json::Value = value.into();
|
||||
match (json["blobId"].as_str(), json["signature"].as_str()) {
|
||||
(Some(blob_id), Some(signature)) => {
|
||||
sent.upgrade = Some(Upgrade {
|
||||
blob_id: blob_id.to_string(),
|
||||
signature: signature.to_string(),
|
||||
})
|
||||
}
|
||||
_ => {
|
||||
return Err(SetError::invalid_properties()
|
||||
.with_property(P::Upgrade)
|
||||
.with_description(
|
||||
"{blobId, signature}: the release binary, uploaded, \
|
||||
and its signature file's text.",
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
Key::Property(p) => {
|
||||
sent.properties
|
||||
.insert(p.to_cow().into_owned(), value.into());
|
||||
@@ -344,6 +393,56 @@ fn apply(current: &StoredRelay, sent: &Sent) -> Result<StoredRelay, SetError<P>>
|
||||
})
|
||||
}
|
||||
|
||||
/// Hands a relay a signed release over this node's link (RL-29), once the
|
||||
/// release checks out here. The relay checks it again, with its own target
|
||||
/// and version, installs it and restarts.
|
||||
async fn upgrade_relay(
|
||||
server: &Server,
|
||||
access_token: &AccessToken,
|
||||
id: Id,
|
||||
upgrade: &Upgrade,
|
||||
) -> trc::Result<Result<(), SetError<P>>> {
|
||||
let refused = |text: String| {
|
||||
SetError::invalid_properties()
|
||||
.with_property(P::Upgrade)
|
||||
.with_description(text)
|
||||
};
|
||||
let Some(relay) = (match relay_id(id) {
|
||||
Some(relay_id) => relays::get(server.store(), relay_id).await?,
|
||||
None => None,
|
||||
}) else {
|
||||
return Ok(Err(SetError::not_found()));
|
||||
};
|
||||
let binary = match BlobId::from_str(&upgrade.blob_id) {
|
||||
Ok(blob_id) => server.blob_download(&blob_id, access_token).await?,
|
||||
Err(_) => None,
|
||||
};
|
||||
let Some(binary) = binary else {
|
||||
return Ok(Err(refused(format!(
|
||||
"No blob {} that you can read.",
|
||||
upgrade.blob_id
|
||||
))));
|
||||
};
|
||||
if let Err(err) = release::verify(&binary, &upgrade.signature, release::RELEASE_KEYS) {
|
||||
return Ok(Err(refused(format!("{err}."))));
|
||||
}
|
||||
let Some(linked) = server.inner.data.relay_senders.linked(relay.id) else {
|
||||
return Ok(Err(SetError::forbidden().with_description(format!(
|
||||
"This node has no link to {} now; try again once it is linked.",
|
||||
relay.name
|
||||
))));
|
||||
};
|
||||
match linked
|
||||
.sender
|
||||
.upgrade(binary, upgrade.signature.clone())
|
||||
.await
|
||||
{
|
||||
// The relay's own relay.upgraded event follows over the link
|
||||
Ok(_) => Ok(Ok(())),
|
||||
Err(why) => Ok(Err(refused(why))),
|
||||
}
|
||||
}
|
||||
|
||||
/// `inbuxa:Relay/get`: every relay, oldest first.
|
||||
pub async fn get(
|
||||
server: &Server,
|
||||
@@ -376,7 +475,7 @@ pub async fn get(
|
||||
}
|
||||
};
|
||||
for relay in wanted {
|
||||
let links = if properties.contains(&P::Links) {
|
||||
let links = if properties.contains(&P::Links) || properties.contains(&P::OlderThanServer) {
|
||||
relays::link::statuses(data, relay.id).await?
|
||||
} else {
|
||||
Vec::new()
|
||||
@@ -412,6 +511,11 @@ pub async fn set(
|
||||
.with_property(P::Rejoin)
|
||||
.with_description("A new relay gets its join token anyway."));
|
||||
}
|
||||
if sent.upgrade.is_some() {
|
||||
return Err(SetError::invalid_properties()
|
||||
.with_property(P::Upgrade)
|
||||
.with_description("A relay is upgraded once it has joined."));
|
||||
}
|
||||
if sent
|
||||
.properties
|
||||
.get("smtpPrivateKey")
|
||||
@@ -460,6 +564,21 @@ pub async fn set(
|
||||
continue;
|
||||
}
|
||||
};
|
||||
if let Some(upgrade) = &sent.upgrade {
|
||||
let outcome = if sent.rejoin || !sent.properties.is_empty() || !sent.settings.is_empty()
|
||||
{
|
||||
Err(SetError::invalid_properties()
|
||||
.with_property(P::Upgrade)
|
||||
.with_description("Send an upgrade on its own."))
|
||||
} else {
|
||||
upgrade_relay(server, access_token, id, upgrade).await?
|
||||
};
|
||||
match outcome {
|
||||
Ok(()) => response.updated.append(id, None),
|
||||
Err(error) => response.not_updated.append(id, error),
|
||||
}
|
||||
continue;
|
||||
}
|
||||
let mut outcome = None;
|
||||
// A node may record the join while this runs: start over from the
|
||||
// relay as it is then, a few times
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
pub mod jmap;
|
||||
pub mod pin;
|
||||
pub mod recipient;
|
||||
pub mod release;
|
||||
pub mod seal;
|
||||
pub mod time;
|
||||
pub mod token;
|
||||
|
||||
@@ -0,0 +1,349 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs LLC
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! Signed relay releases (RL-29). Each release binary comes with a
|
||||
//! signature file, `inbuxa-relay-<target>.sig`:
|
||||
//!
|
||||
//! ```text
|
||||
//! inbuxa-relay release 1
|
||||
//! version 27.1.0
|
||||
//! target x86_64-linux-musl
|
||||
//! sha256 <the binary's SHA-256, 64 hex digits>
|
||||
//! signature <Ed25519 over the four lines above, newlines included; base64url>
|
||||
//! ```
|
||||
//!
|
||||
//! The signature is by the inbuxa release key, whose public half is
|
||||
//! compiled in here. The server checks it before handing a release over;
|
||||
//! the relay checks it again, with its own target and version.
|
||||
|
||||
use crate::b64;
|
||||
use aws_lc_rs::{
|
||||
digest,
|
||||
signature::{ED25519, UnparsedPublicKey},
|
||||
};
|
||||
use std::{cmp::Ordering, fmt};
|
||||
|
||||
/// The inbuxa release keys' public halves (raw Ed25519). A build without
|
||||
/// one refuses every upgrade.
|
||||
pub const RELEASE_KEYS: &[[u8; 32]] = &[];
|
||||
|
||||
const HEADER: &str = "inbuxa-relay release 1";
|
||||
|
||||
/// What a signature file says of its binary.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Release {
|
||||
pub version: String,
|
||||
pub target: String,
|
||||
pub sha256: [u8; 32],
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum ReleaseError {
|
||||
/// The signature file isn't one.
|
||||
Unreadable(&'static str),
|
||||
NoKey,
|
||||
BadSignature,
|
||||
/// The binary isn't the one the file was signed for.
|
||||
OtherBinary,
|
||||
}
|
||||
|
||||
impl fmt::Display for ReleaseError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
ReleaseError::Unreadable(why) => write!(f, "The signature file is unreadable: {why}"),
|
||||
ReleaseError::NoKey => f.write_str("This build has no inbuxa release key"),
|
||||
ReleaseError::BadSignature => {
|
||||
f.write_str("The signature isn't by the inbuxa release key")
|
||||
}
|
||||
ReleaseError::OtherBinary => f.write_str("The signature is for another binary"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The target this build runs on, as signature files name it:
|
||||
/// `<arch>-<os>-<libc>`, like `x86_64-linux-musl`.
|
||||
pub fn target() -> String {
|
||||
let libc = if cfg!(target_env = "musl") {
|
||||
"musl"
|
||||
} else if cfg!(target_env = "gnu") {
|
||||
"gnu"
|
||||
} else {
|
||||
"other"
|
||||
};
|
||||
format!("{}-{}-{libc}", std::env::consts::ARCH, std::env::consts::OS)
|
||||
}
|
||||
|
||||
fn hex(bytes: &[u8]) -> String {
|
||||
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
||||
}
|
||||
|
||||
fn parse_sha256(text: &str) -> Option<[u8; 32]> {
|
||||
if text.len() != 64 {
|
||||
return None;
|
||||
}
|
||||
let mut out = [0u8; 32];
|
||||
for (i, byte) in out.iter_mut().enumerate() {
|
||||
*byte = u8::from_str_radix(text.get(i * 2..i * 2 + 2)?, 16).ok()?;
|
||||
}
|
||||
Some(out)
|
||||
}
|
||||
|
||||
/// The text a release key signs for `release`.
|
||||
pub fn signed_text(release: &Release) -> String {
|
||||
format!(
|
||||
"{HEADER}\nversion {}\ntarget {}\nsha256 {}\n",
|
||||
release.version,
|
||||
release.target,
|
||||
hex(&release.sha256)
|
||||
)
|
||||
}
|
||||
|
||||
/// Reads a signature file and checks its signature against `keys`, but not
|
||||
/// the binary: what it says, if a key signed it.
|
||||
pub fn verify_signature(signature_file: &str, keys: &[[u8; 32]]) -> Result<Release, ReleaseError> {
|
||||
let text = signature_file.replace("\r\n", "\n");
|
||||
let mut lines = text.lines();
|
||||
if lines.next() != Some(HEADER) {
|
||||
return Err(ReleaseError::Unreadable("it doesn't start as one"));
|
||||
}
|
||||
let mut field = |name: &'static str| {
|
||||
lines
|
||||
.next()
|
||||
.and_then(|line| line.strip_prefix(name))
|
||||
.and_then(|rest| rest.strip_prefix(' '))
|
||||
.map(str::trim)
|
||||
.filter(|value| !value.is_empty())
|
||||
.ok_or(ReleaseError::Unreadable(name))
|
||||
};
|
||||
let version = field("version")?.to_string();
|
||||
let target = field("target")?.to_string();
|
||||
let sha256 = parse_sha256(field("sha256")?).ok_or(ReleaseError::Unreadable("sha256"))?;
|
||||
let signature = b64::decode(field("signature")?.trim_end_matches('='))
|
||||
.filter(|sig| sig.len() == 64)
|
||||
.ok_or(ReleaseError::Unreadable("signature"))?;
|
||||
if Version::parse(&version).is_none() {
|
||||
return Err(ReleaseError::Unreadable("version"));
|
||||
}
|
||||
if lines.any(|line| !line.trim().is_empty()) {
|
||||
return Err(ReleaseError::Unreadable("it goes on past the signature"));
|
||||
}
|
||||
let release = Release {
|
||||
version,
|
||||
target,
|
||||
sha256,
|
||||
};
|
||||
if keys.is_empty() {
|
||||
return Err(ReleaseError::NoKey);
|
||||
}
|
||||
let signed = signed_text(&release);
|
||||
if keys.iter().any(|key| {
|
||||
UnparsedPublicKey::new(&ED25519, key)
|
||||
.verify(signed.as_bytes(), &signature)
|
||||
.is_ok()
|
||||
}) {
|
||||
Ok(release)
|
||||
} else {
|
||||
Err(ReleaseError::BadSignature)
|
||||
}
|
||||
}
|
||||
|
||||
/// Checks a release binary against its signature file: what it is, if a
|
||||
/// key signed it and it is that binary.
|
||||
pub fn verify(
|
||||
binary: &[u8],
|
||||
signature_file: &str,
|
||||
keys: &[[u8; 32]],
|
||||
) -> Result<Release, ReleaseError> {
|
||||
let release = verify_signature(signature_file, keys)?;
|
||||
if digest::digest(&digest::SHA256, binary).as_ref() != release.sha256 {
|
||||
return Err(ReleaseError::OtherBinary);
|
||||
}
|
||||
Ok(release)
|
||||
}
|
||||
|
||||
/// A release version, `MAJOR.MINOR.PATCH[-PRE][+BUILD]`, ordered as SemVer
|
||||
/// §11 orders them: a pre-release comes before its release.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Version {
|
||||
core: [u64; 3],
|
||||
pre: Vec<String>,
|
||||
}
|
||||
|
||||
impl Version {
|
||||
pub fn parse(text: &str) -> Option<Version> {
|
||||
let text = text.trim();
|
||||
let text = text.split_once('+').map_or(text, |(v, _)| v);
|
||||
let (core, pre) = match text.split_once('-') {
|
||||
Some((core, pre)) => (core, Some(pre)),
|
||||
None => (text, None),
|
||||
};
|
||||
let mut parts = core.split('.');
|
||||
let mut number = || parts.next()?.parse::<u64>().ok();
|
||||
let core = [number()?, number()?, number()?];
|
||||
if parts.next().is_some() {
|
||||
return None;
|
||||
}
|
||||
let pre = match pre {
|
||||
Some(pre) => {
|
||||
let ids: Vec<String> = pre.split('.').map(str::to_string).collect();
|
||||
if ids.iter().any(|id| {
|
||||
id.is_empty() || !id.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-')
|
||||
}) {
|
||||
return None;
|
||||
}
|
||||
ids
|
||||
}
|
||||
None => Vec::new(),
|
||||
};
|
||||
Some(Version { core, pre })
|
||||
}
|
||||
}
|
||||
|
||||
impl Ord for Version {
|
||||
fn cmp(&self, other: &Self) -> Ordering {
|
||||
self.core
|
||||
.cmp(&other.core)
|
||||
.then_with(|| match (self.pre.is_empty(), other.pre.is_empty()) {
|
||||
(true, true) => Ordering::Equal,
|
||||
(true, false) => Ordering::Greater,
|
||||
(false, true) => Ordering::Less,
|
||||
(false, false) => {
|
||||
for (a, b) in self.pre.iter().zip(&other.pre) {
|
||||
let order = match (a.parse::<u64>(), b.parse::<u64>()) {
|
||||
(Ok(a), Ok(b)) => a.cmp(&b),
|
||||
(Ok(_), Err(_)) => Ordering::Less,
|
||||
(Err(_), Ok(_)) => Ordering::Greater,
|
||||
(Err(_), Err(_)) => a.cmp(b),
|
||||
};
|
||||
if order != Ordering::Equal {
|
||||
return order;
|
||||
}
|
||||
}
|
||||
self.pre.len().cmp(&other.pre.len())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl PartialOrd for Version {
|
||||
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
|
||||
Some(self.cmp(other))
|
||||
}
|
||||
}
|
||||
|
||||
/// How `a` compares with `b`, if both are versions.
|
||||
pub fn compare_versions(a: &str, b: &str) -> Option<Ordering> {
|
||||
Some(Version::parse(a)?.cmp(&Version::parse(b)?))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aws_lc_rs::signature::{Ed25519KeyPair, KeyPair};
|
||||
|
||||
/// A key pair for tests, and its public half.
|
||||
pub fn test_key() -> (Ed25519KeyPair, [u8; 32]) {
|
||||
let pair = Ed25519KeyPair::from_seed_unchecked(&[7u8; 32]).unwrap();
|
||||
let public = pair.public_key().as_ref().try_into().unwrap();
|
||||
(pair, public)
|
||||
}
|
||||
|
||||
pub fn sign(pair: &Ed25519KeyPair, binary: &[u8], version: &str, target: &str) -> String {
|
||||
let release = Release {
|
||||
version: version.into(),
|
||||
target: target.into(),
|
||||
sha256: digest::digest(&digest::SHA256, binary)
|
||||
.as_ref()
|
||||
.try_into()
|
||||
.unwrap(),
|
||||
};
|
||||
let signed = signed_text(&release);
|
||||
format!(
|
||||
"{signed}signature {}\n",
|
||||
b64::encode(pair.sign(signed.as_bytes()).as_ref())
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn signed_releases() {
|
||||
let (pair, key) = test_key();
|
||||
let binary = b"a relay";
|
||||
let file = sign(&pair, binary, "27.1.0", "x86_64-linux-musl");
|
||||
let release = verify(binary, &file, &[key]).unwrap();
|
||||
assert_eq!(release.version, "27.1.0");
|
||||
assert_eq!(release.target, "x86_64-linux-musl");
|
||||
|
||||
assert_eq!(
|
||||
verify(b"another", &file, &[key]),
|
||||
Err(ReleaseError::OtherBinary)
|
||||
);
|
||||
assert_eq!(verify(binary, &file, &[]), Err(ReleaseError::NoKey));
|
||||
let other = Ed25519KeyPair::from_seed_unchecked(&[8u8; 32]).unwrap();
|
||||
let other_key: [u8; 32] = other.public_key().as_ref().try_into().unwrap();
|
||||
assert_eq!(
|
||||
verify(binary, &file, &[other_key]),
|
||||
Err(ReleaseError::BadSignature)
|
||||
);
|
||||
// Any changed line breaks the signature
|
||||
let tampered = file.replace("version 27.1.0", "version 99.0.0");
|
||||
assert_eq!(
|
||||
verify(binary, &tampered, &[key]),
|
||||
Err(ReleaseError::BadSignature)
|
||||
);
|
||||
assert!(matches!(
|
||||
verify(binary, &format!("{file}more\n"), &[key]),
|
||||
Err(ReleaseError::Unreadable(_))
|
||||
));
|
||||
assert!(matches!(
|
||||
verify(binary, "not a signature file", &[key]),
|
||||
Err(ReleaseError::Unreadable(_))
|
||||
));
|
||||
// CRLF line ends, as a download on another system may have
|
||||
assert!(verify(binary, &file.replace('\n', "\r\n"), &[key]).is_ok());
|
||||
}
|
||||
|
||||
/// A file made by tools/release/sign-relay.sh with openssl, from a key
|
||||
/// made for this test only: the script and this code agree.
|
||||
#[test]
|
||||
fn openssl_signature() {
|
||||
let key: [u8; 32] = b64::decode("VNaWnerWh2UXOrUMtSWPOKJHtigsUdIWUfzHmXbdgIU")
|
||||
.unwrap()
|
||||
.try_into()
|
||||
.unwrap();
|
||||
let file = "inbuxa-relay release 1\n\
|
||||
version 27.1.0-rc.5\n\
|
||||
target x86_64-linux-musl\n\
|
||||
sha256 ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad\n\
|
||||
signature SIG\n";
|
||||
let file = file.replace("SIG", OPENSSL_SIGNATURE);
|
||||
let release = verify(b"abc", &file, &[key]).unwrap();
|
||||
assert_eq!(release.version, "27.1.0-rc.5");
|
||||
}
|
||||
|
||||
const OPENSSL_SIGNATURE: &str =
|
||||
"zRj90ByVN18fmW-GOZ9cvGBK1ZsYYoNvowGsbSBYcslJhgIgmKrYHgRgq1CSIy4G-EzR5uexCzAUoa3FEcluAA";
|
||||
|
||||
#[test]
|
||||
fn versions_in_order() {
|
||||
let less = |a: &str, b: &str| compare_versions(a, b) == Some(Ordering::Less);
|
||||
assert!(less("27.1.0-rc.4", "27.1.0-rc.5"));
|
||||
assert!(less("27.1.0-rc.9", "27.1.0-rc.10"));
|
||||
assert!(less("27.1.0-rc.5", "27.1.0"));
|
||||
assert!(less("27.1.0", "27.1.1"));
|
||||
assert!(less("27.1.9", "27.2.0"));
|
||||
assert!(less("27.2.0", "28.0.0"));
|
||||
assert!(less("1.0.0-alpha", "1.0.0-alpha.1"));
|
||||
assert!(less("1.0.0-alpha.1", "1.0.0-alpha.beta"));
|
||||
assert!(less("1.0.0-beta.11", "1.0.0-rc.1"));
|
||||
assert_eq!(
|
||||
compare_versions("27.1.0+abc", "27.1.0"),
|
||||
Some(Ordering::Equal)
|
||||
);
|
||||
assert_eq!(compare_versions("27.1", "27.1.0"), None);
|
||||
assert_eq!(compare_versions("27.1.0-", "27.1.0"), None);
|
||||
assert!(target().contains(std::env::consts::ARCH));
|
||||
}
|
||||
}
|
||||
+13
-1
@@ -22,7 +22,7 @@ use std::{
|
||||
net::IpAddr,
|
||||
sync::{
|
||||
Arc, Mutex, RwLock,
|
||||
atomic::{AtomicU32, AtomicU64, Ordering},
|
||||
atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
|
||||
},
|
||||
};
|
||||
use tokio::sync::{Notify, broadcast, watch};
|
||||
@@ -57,6 +57,14 @@ pub struct App {
|
||||
pub last_error: Mutex<Option<String>>,
|
||||
pub limiter: Mutex<Limiter>,
|
||||
pub self_upgrade: bool,
|
||||
/// The keys a release must be signed by (RL-29).
|
||||
pub release_keys: Vec<[u8; 32]>,
|
||||
/// The binary an upgrade replaces.
|
||||
pub exe: Option<std::path::PathBuf>,
|
||||
/// An upgrade is installing, or installed and waiting to restart.
|
||||
pub upgrading: AtomicBool,
|
||||
/// Wakes the run loop to exit for the supervisor's restart.
|
||||
pub restart: Notify,
|
||||
}
|
||||
|
||||
impl App {
|
||||
@@ -91,6 +99,10 @@ impl App {
|
||||
last_error: Mutex::new(None),
|
||||
limiter: Mutex::new(Limiter::default()),
|
||||
self_upgrade: std::env::var_os("INBUXA_RELAY_CONTAINER").is_none(),
|
||||
release_keys: inbuxa_relay_proto::release::RELEASE_KEYS.to_vec(),
|
||||
exe: crate::upgrade::current_exe(),
|
||||
upgrading: AtomicBool::new(false),
|
||||
restart: Notify::new(),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -105,6 +105,7 @@ fn call_one(app: &App, caller: &Caller, call: &Invocation) -> MethodResult {
|
||||
method::RELAY_JOIN => relay_join(app, args),
|
||||
method::RELAY_GET => relay_get(app, args),
|
||||
method::RELAY_SET => relay_set(app, 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),
|
||||
method::RECIPIENT_SET => recipient_set(app, args),
|
||||
@@ -304,6 +305,81 @@ fn relay_set(app: &App, args: &Map<String, Value>) -> MethodResult {
|
||||
}))
|
||||
}
|
||||
|
||||
/// `Relay/upgrade` (RL-29): `{blobId, signature}`, the release binary as
|
||||
/// an upload and its signature file's text. Answers once the new binary
|
||||
/// is in place; the relay then exits for its supervisor to restart it.
|
||||
fn relay_upgrade(app: &App, args: &Map<String, Value>) -> MethodResult {
|
||||
use crate::upgrade::{self, Refused};
|
||||
use std::sync::atomic::Ordering;
|
||||
if !app.self_upgrade {
|
||||
return Err(err(
|
||||
MethodError::Forbidden,
|
||||
"this relay runs in a container image: update the image instead",
|
||||
));
|
||||
}
|
||||
let blob_id = args
|
||||
.get("blobId")
|
||||
.and_then(Value::as_str)
|
||||
.filter(|id| crate::link::is_blob_id(id))
|
||||
.ok_or_else(|| err(MethodError::InvalidArguments, "blobId: an uploaded blob"))?;
|
||||
let signature = args
|
||||
.get("signature")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| {
|
||||
err(
|
||||
MethodError::InvalidArguments,
|
||||
"signature: the release's signature file",
|
||||
)
|
||||
})?;
|
||||
let exe = app
|
||||
.exe
|
||||
.clone()
|
||||
.ok_or_else(|| err(MethodError::ServerFail, "the relay can't find its binary"))?;
|
||||
if app.upgrading.swap(true, Ordering::SeqCst) {
|
||||
return Err(err(
|
||||
MethodError::Forbidden,
|
||||
"an upgrade is already on its way",
|
||||
));
|
||||
}
|
||||
let path = app.paths.blobs().join(blob_id);
|
||||
let result = std::fs::read(&path)
|
||||
.map_err(|_| Refused::Invalid("no such blob".into()))
|
||||
.and_then(|binary| {
|
||||
let release =
|
||||
upgrade::check(&binary, signature, &app.release_keys, crate::app::VERSION)?;
|
||||
upgrade::install(&exe, &binary, &release.version)?;
|
||||
Ok(release)
|
||||
});
|
||||
let _ = std::fs::remove_file(&path);
|
||||
let release = match result {
|
||||
Ok(release) => release,
|
||||
Err(refused) => {
|
||||
app.upgrading.store(false, Ordering::SeqCst);
|
||||
let (kind, text) = match refused {
|
||||
Refused::Invalid(text) => (MethodError::InvalidArguments, text),
|
||||
Refused::Failed(text) => (MethodError::ServerFail, text),
|
||||
};
|
||||
app.record(
|
||||
"relay.upgrade-refused",
|
||||
crate::events::data([("reason", json!(text))]),
|
||||
);
|
||||
return Err(err(kind, text));
|
||||
}
|
||||
};
|
||||
app.record(
|
||||
"relay.upgraded",
|
||||
crate::events::data([
|
||||
("from", json!(crate::app::VERSION)),
|
||||
("to", json!(release.version)),
|
||||
]),
|
||||
);
|
||||
app.restart.notify_one();
|
||||
Ok(json!({
|
||||
"accountId": ACCOUNT_ID,
|
||||
"relayVersion": release.version,
|
||||
}))
|
||||
}
|
||||
|
||||
fn apply_settings(app: &App, patch: &Value) -> Result<(), Value> {
|
||||
let Value::Object(patch) = patch else {
|
||||
return Err(set_error("invalidPatch", "a patch is an object"));
|
||||
|
||||
@@ -23,6 +23,7 @@ mod spool;
|
||||
mod state;
|
||||
#[cfg(test)]
|
||||
mod tests_e2e;
|
||||
mod upgrade;
|
||||
|
||||
use app::App;
|
||||
use inbuxa_relay_proto::{time, token::JoinToken};
|
||||
@@ -112,7 +113,11 @@ fn main() -> ExitCode {
|
||||
Some(token) => join(&paths, token),
|
||||
None => Err("join needs the token from inbuxa-admin".into()),
|
||||
},
|
||||
"run" => run(&paths, args.link_listen),
|
||||
"run" => match run(&paths, args.link_listen) {
|
||||
Ok(Stopped::Upgraded) => return ExitCode::from(upgrade::RESTART_EXIT),
|
||||
Ok(Stopped::Asked) => Ok(()),
|
||||
Err(e) => Err(e),
|
||||
},
|
||||
"status" => status(&paths),
|
||||
"reset" => reset(&paths, args.discard_held),
|
||||
other => Err(format!("unknown command {other}")),
|
||||
@@ -154,7 +159,15 @@ fn join(paths: &Paths, token_text: &str) -> Result<(), String> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run(paths: &Paths, link_listen: SocketAddr) -> Result<(), String> {
|
||||
/// Why `run` ended, when it ended well.
|
||||
enum Stopped {
|
||||
/// A signal asked it to.
|
||||
Asked,
|
||||
/// A new binary is in place, for the supervisor to start (RL-29).
|
||||
Upgraded,
|
||||
}
|
||||
|
||||
fn run(paths: &Paths, link_listen: SocketAddr) -> Result<Stopped, String> {
|
||||
paths.create_dirs().map_err(|e| e.to_string())?;
|
||||
let _lock = RunLock::acquire(paths).map_err(|e| e.to_string())?;
|
||||
let identity = match Identity::load(paths).map_err(|e| e.to_string())? {
|
||||
@@ -211,7 +224,14 @@ fn run(paths: &Paths, link_listen: SocketAddr) -> Result<(), String> {
|
||||
_ = shutdown() => {
|
||||
app.write_status();
|
||||
eprintln!("inbuxa-relay: stopping");
|
||||
Ok(())
|
||||
Ok(Stopped::Asked)
|
||||
}
|
||||
_ = app.restart.notified() => {
|
||||
// Time for the answer to reach the downstream
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
app.write_status();
|
||||
eprintln!("inbuxa-relay: upgraded; exiting for the new binary to start");
|
||||
Ok(Stopped::Upgraded)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -343,7 +343,13 @@ async fn relay_end_to_end() {
|
||||
assert!(crate::join(&paths, &token.encode()).is_err(), "joined once");
|
||||
|
||||
let identity = Identity::load(&paths).unwrap().unwrap();
|
||||
let app = Arc::new(App::open(Paths::new(&dir), identity).unwrap());
|
||||
let mut app = App::open(Paths::new(&dir), identity).unwrap();
|
||||
let (release_key, release_public) = crate::upgrade::tests::test_key();
|
||||
app.release_keys = vec![release_public];
|
||||
let exe = dir.join("inbuxa-relay");
|
||||
std::fs::write(&exe, crate::upgrade::tests::script(crate::app::VERSION)).unwrap();
|
||||
app.exe = Some(exe.clone());
|
||||
let app = Arc::new(app);
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let link_addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(link::serve(app.clone(), listener));
|
||||
@@ -674,6 +680,62 @@ async fn relay_end_to_end() {
|
||||
assert_eq!(gone["destroyed"], json!([out_id]));
|
||||
assert!(!eml.exists());
|
||||
|
||||
// RL-T16 / RL-29: a signed release, newer, for this target, goes in;
|
||||
// nothing else does.
|
||||
let release = crate::upgrade::tests::script("99.0.0");
|
||||
let target = inbuxa_relay_proto::release::target();
|
||||
let sign = |version: &str, target: &str| {
|
||||
crate::upgrade::tests::sign(&release_key, &release, version, target)
|
||||
};
|
||||
let mut upgrade = async |signature: String| {
|
||||
let (status, blob) = down
|
||||
.http("POST", "/jmap/upload/relay", release.clone())
|
||||
.await;
|
||||
assert_eq!(status, 201);
|
||||
let blob: Value = serde_json::from_slice(&blob).unwrap();
|
||||
ws.call(
|
||||
"Relay/upgrade",
|
||||
json!({ "blobId": blob["blobId"], "signature": signature }),
|
||||
)
|
||||
.await
|
||||
};
|
||||
let forged = crate::upgrade::tests::sign(
|
||||
&aws_lc_rs::signature::Ed25519KeyPair::from_seed_unchecked(&[9u8; 32]).unwrap(),
|
||||
&release,
|
||||
"99.0.0",
|
||||
&target,
|
||||
);
|
||||
for signature in [
|
||||
forged,
|
||||
sign("27.0.0", &target),
|
||||
sign("99.0.0", "riscv64-none-other"),
|
||||
"not a signature file".to_string(),
|
||||
] {
|
||||
let (name, args) = upgrade(signature).await;
|
||||
assert_eq!(
|
||||
(name.as_str(), args["type"].as_str()),
|
||||
("error", Some("invalidArguments")),
|
||||
"{args}"
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
std::fs::read(&exe).unwrap(),
|
||||
crate::upgrade::tests::script(crate::app::VERSION)
|
||||
);
|
||||
let (name, args) = upgrade(sign("99.0.0", &target)).await;
|
||||
assert_eq!(name, "Relay/upgrade", "{args}");
|
||||
assert_eq!(args["relayVersion"], "99.0.0");
|
||||
assert_eq!(std::fs::read(&exe).unwrap(), release);
|
||||
tokio::time::timeout(Duration::from_secs(1), app.restart.notified())
|
||||
.await
|
||||
.expect("the run loop is told to exit");
|
||||
let (name, args) = upgrade(sign("99.0.0", &target)).await;
|
||||
assert_eq!(
|
||||
(name.as_str(), args["type"].as_str()),
|
||||
("error", Some("forbidden")),
|
||||
"one upgrade at a time"
|
||||
);
|
||||
|
||||
// Events went up the link.
|
||||
let events = ws.ok("RelayEvent/get", json!({})).await;
|
||||
let kinds: Vec<&str> = events["list"]
|
||||
@@ -687,6 +749,8 @@ async fn relay_end_to_end() {
|
||||
"inbound.accepted",
|
||||
"inbound.refused",
|
||||
"smtp.session",
|
||||
"relay.upgrade-refused",
|
||||
"relay.upgraded",
|
||||
] {
|
||||
assert!(kinds.contains(&kind), "{kind} in {kinds:?}");
|
||||
}
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs LLC
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! `Relay/upgrade` (RL-29): a release binary, checked against its signature
|
||||
//! by the inbuxa release key, replaces this one. The relay refuses a
|
||||
//! downgrade, or a binary for another target; runs the new one once, to see
|
||||
//! that it starts here and is the version it says; keeps the old one as
|
||||
//! `inbuxa-relay.previous`; and exits with [`RESTART_EXIT`] for its
|
||||
//! supervisor to start the new one.
|
||||
|
||||
use inbuxa_relay_proto::release::{self, Release, Version};
|
||||
use std::{
|
||||
fs,
|
||||
io::Write,
|
||||
path::{Path, PathBuf},
|
||||
process::{Command, Stdio},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
/// The exit status after an upgrade: EX_TEMPFAIL, which a supervisor set
|
||||
/// to restart on failure restarts too.
|
||||
pub const RESTART_EXIT: u8 = 75;
|
||||
|
||||
/// How long the new binary has to answer `--version`.
|
||||
const TRY_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Why an upgrade was refused.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum Refused {
|
||||
/// The release won't do here: `invalidArguments`.
|
||||
Invalid(String),
|
||||
/// It would, but couldn't be put in place: `serverFail`.
|
||||
Failed(String),
|
||||
}
|
||||
|
||||
/// Checks `binary` against `signature_file`, for this build at `current`.
|
||||
pub fn check(
|
||||
binary: &[u8],
|
||||
signature_file: &str,
|
||||
keys: &[[u8; 32]],
|
||||
current: &str,
|
||||
) -> Result<Release, Refused> {
|
||||
let release = release::verify(binary, signature_file, keys)
|
||||
.map_err(|err| Refused::Invalid(err.to_string()))?;
|
||||
let target = release::target();
|
||||
if release.target != target {
|
||||
return Err(Refused::Invalid(format!(
|
||||
"The release is for {}; this relay runs on {target}",
|
||||
release.target
|
||||
)));
|
||||
}
|
||||
let newer = match (Version::parse(&release.version), Version::parse(current)) {
|
||||
(Some(new), Some(current)) => new > current,
|
||||
_ => false,
|
||||
};
|
||||
if !newer {
|
||||
return Err(Refused::Invalid(format!(
|
||||
"The release is {}; this relay runs {current}, and won't go back or stay",
|
||||
release.version
|
||||
)));
|
||||
}
|
||||
Ok(release)
|
||||
}
|
||||
|
||||
/// Puts `binary` in place of `exe`, once it has run here as `version`.
|
||||
pub fn install(exe: &Path, binary: &[u8], version: &str) -> Result<(), Refused> {
|
||||
let failed = |what: &str, err: std::io::Error| Refused::Failed(format!("{what}: {err}"));
|
||||
let dir = exe
|
||||
.parent()
|
||||
.ok_or_else(|| Refused::Failed("The relay's binary has no directory".into()))?;
|
||||
let name = exe
|
||||
.file_name()
|
||||
.and_then(|name| name.to_str())
|
||||
.unwrap_or("inbuxa-relay");
|
||||
let new = dir.join(format!(".{name}.new"));
|
||||
let previous = dir.join(format!("{name}.previous"));
|
||||
let result = (|| {
|
||||
{
|
||||
let mut file = open_executable(&new).map_err(|e| failed("Couldn't write it", e))?;
|
||||
file.write_all(binary)
|
||||
.and_then(|()| file.sync_all())
|
||||
.map_err(|e| failed("Couldn't write it", e))?;
|
||||
}
|
||||
try_version(&new, version)?;
|
||||
fs::copy(exe, &previous).map_err(|e| failed("Couldn't keep the old binary", e))?;
|
||||
fs::rename(&new, exe).map_err(|e| failed("Couldn't put it in place", e))?;
|
||||
if let Ok(dir) = fs::File::open(dir) {
|
||||
let _ = dir.sync_all();
|
||||
}
|
||||
Ok(())
|
||||
})();
|
||||
if result.is_err() {
|
||||
let _ = fs::remove_file(&new);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn open_executable(path: &Path) -> std::io::Result<fs::File> {
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
let _ = fs::remove_file(path);
|
||||
fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.mode(0o755)
|
||||
.open(path)
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
fn open_executable(path: &Path) -> std::io::Result<fs::File> {
|
||||
let _ = fs::remove_file(path);
|
||||
fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(path)
|
||||
}
|
||||
|
||||
/// Runs `path --version`: it must say it is `version`, in time.
|
||||
fn try_version(path: &Path, version: &str) -> Result<(), Refused> {
|
||||
let mut child = Command::new(path)
|
||||
.arg("--version")
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::null())
|
||||
.spawn()
|
||||
.map_err(|err| Refused::Failed(format!("The new binary doesn't start here: {err}")))?;
|
||||
let started = Instant::now();
|
||||
loop {
|
||||
match child.try_wait() {
|
||||
Ok(Some(_)) => break,
|
||||
Ok(None) if started.elapsed() < TRY_TIMEOUT => {
|
||||
std::thread::sleep(Duration::from_millis(20))
|
||||
}
|
||||
_ => {
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
return Err(Refused::Failed(
|
||||
"The new binary didn't answer --version".into(),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
let output = child
|
||||
.wait_with_output()
|
||||
.map_err(|err| Refused::Failed(format!("The new binary failed: {err}")))?;
|
||||
let said = String::from_utf8_lossy(&output.stdout);
|
||||
let expected = format!("inbuxa-relay {version}");
|
||||
if !output.status.success() || said.trim() != expected {
|
||||
return Err(Refused::Failed(format!(
|
||||
"The new binary says {:?}, not {expected:?}",
|
||||
said.trim()
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// This process's binary, to replace.
|
||||
pub fn current_exe() -> Option<PathBuf> {
|
||||
std::env::current_exe()
|
||||
.ok()
|
||||
.and_then(|exe| exe.canonicalize().ok())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) mod tests {
|
||||
use super::*;
|
||||
use aws_lc_rs::{
|
||||
digest,
|
||||
signature::{Ed25519KeyPair, KeyPair},
|
||||
};
|
||||
use inbuxa_relay_proto::{b64, release::signed_text};
|
||||
|
||||
pub fn test_key() -> (Ed25519KeyPair, [u8; 32]) {
|
||||
let pair = Ed25519KeyPair::from_seed_unchecked(&[3u8; 32]).unwrap();
|
||||
let public = pair.public_key().as_ref().try_into().unwrap();
|
||||
(pair, public)
|
||||
}
|
||||
|
||||
pub fn sign(pair: &Ed25519KeyPair, binary: &[u8], version: &str, target: &str) -> String {
|
||||
let signed = signed_text(&Release {
|
||||
version: version.into(),
|
||||
target: target.into(),
|
||||
sha256: digest::digest(&digest::SHA256, binary)
|
||||
.as_ref()
|
||||
.try_into()
|
||||
.unwrap(),
|
||||
});
|
||||
format!(
|
||||
"{signed}signature {}\n",
|
||||
b64::encode(pair.sign(signed.as_bytes()).as_ref())
|
||||
)
|
||||
}
|
||||
|
||||
/// A stand-in release: a script that says it is `version`.
|
||||
pub fn script(version: &str) -> Vec<u8> {
|
||||
format!("#!/bin/sh\necho 'inbuxa-relay {version}'\n").into_bytes()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn checks_before_installing() {
|
||||
let (pair, key) = test_key();
|
||||
let binary = script("27.2.0");
|
||||
let target = release::target();
|
||||
let good = sign(&pair, &binary, "27.2.0", &target);
|
||||
assert!(check(&binary, &good, &[key], "27.1.0").is_ok());
|
||||
|
||||
// RL-T16: a bad signature, or an older version, is refused
|
||||
let other = Ed25519KeyPair::from_seed_unchecked(&[4u8; 32]).unwrap();
|
||||
let forged = sign(&other, &binary, "27.2.0", &target);
|
||||
assert!(matches!(
|
||||
check(&binary, &forged, &[key], "27.1.0"),
|
||||
Err(Refused::Invalid(_))
|
||||
));
|
||||
let older = sign(&pair, &binary, "27.0.3", &target);
|
||||
assert!(matches!(
|
||||
check(&binary, &older, &[key], "27.1.0"),
|
||||
Err(Refused::Invalid(_))
|
||||
));
|
||||
assert!(matches!(
|
||||
check(&binary, &good, &[key], "27.2.0"),
|
||||
Err(Refused::Invalid(_))
|
||||
));
|
||||
let elsewhere = sign(&pair, &binary, "27.2.0", "riscv64-linux-musl");
|
||||
assert!(matches!(
|
||||
check(&binary, &elsewhere, &[key], "27.1.0"),
|
||||
Err(Refused::Invalid(_))
|
||||
));
|
||||
assert!(matches!(
|
||||
check(&binary, &good, &[], "27.1.0"),
|
||||
Err(Refused::Invalid(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn installs_what_runs() {
|
||||
let dir = std::env::temp_dir().join(format!("inbuxa-relay-upgrade-{}", std::process::id()));
|
||||
let _ = fs::remove_dir_all(&dir);
|
||||
fs::create_dir_all(&dir).unwrap();
|
||||
let exe = dir.join("inbuxa-relay");
|
||||
fs::write(&exe, script("27.1.0")).unwrap();
|
||||
|
||||
// One that says another version is refused, and nothing changes
|
||||
assert!(matches!(
|
||||
install(&exe, &script("27.3.0"), "27.2.0"),
|
||||
Err(Refused::Failed(_))
|
||||
));
|
||||
assert_eq!(fs::read(&exe).unwrap(), script("27.1.0"));
|
||||
assert!(!dir.join(".inbuxa-relay.new").exists());
|
||||
assert!(!dir.join("inbuxa-relay.previous").exists());
|
||||
|
||||
install(&exe, &script("27.2.0"), "27.2.0").unwrap();
|
||||
assert_eq!(fs::read(&exe).unwrap(), script("27.2.0"));
|
||||
assert_eq!(
|
||||
fs::read(dir.join("inbuxa-relay.previous")).unwrap(),
|
||||
script("27.1.0")
|
||||
);
|
||||
let _ = fs::remove_dir_all(&dir);
|
||||
}
|
||||
}
|
||||
@@ -1250,6 +1250,21 @@ impl Sender for LinkSender {
|
||||
Ok(Value::Object(answer.1))
|
||||
})
|
||||
}
|
||||
|
||||
fn upgrade(&self, binary: Vec<u8>, signature: String) -> Call<'_> {
|
||||
Box::pin(async move {
|
||||
let blob_id = self
|
||||
.dialer
|
||||
.upload(binary)
|
||||
.await
|
||||
.map_err(|err| err.message)?;
|
||||
self.call(
|
||||
method::RELAY_UPGRADE,
|
||||
json!({"blobId": blob_id, "signature": signature}),
|
||||
)
|
||||
.await
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// A relay's `notCreated` for a message, as this server's refusal.
|
||||
|
||||
@@ -255,7 +255,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}`; `linkCaPins` lists the CAs it trusts (RL-34) |
|
||||
| `status` | relay | `{startedAt, linkClients, mode, spool: {inbound, outbound, bytes}, recipientState, certificateNotAfter, linkCaPins, lastError, selfUpgrade}`; `linkCaPins` lists the CAs it trusts (RL-34); `selfUpgrade` is false in a container image (RL-29) |
|
||||
|
||||
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).
|
||||
@@ -329,7 +329,7 @@ Standard RFC 8620 methods, plus five of its own:
|
||||
|---|---|
|
||||
| `Relay/get`, `Relay/set` | Read status; push settings |
|
||||
| `Relay/join` | Join proof (see "Joining and reset"). Answers only on an unjoined relay |
|
||||
| `Relay/upgrade` | `{blobId, signature}`: install a signed release (RL-29) |
|
||||
| `Relay/upgrade` | `{blobId, signature}`: install a signed release, the binary as an upload and its signature file's text (RL-29) |
|
||||
| `RecipientDomain/get`, `/set`; `Recipient/set`, `/changes` | Recipient sync |
|
||||
| `Recipient/replace` | Full recipient refresh from a blob |
|
||||
| `Inbound/changes`, `/get`, `/query` | See what is waiting |
|
||||
@@ -507,7 +507,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), 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 |
|
||||
| `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: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`,
|
||||
@@ -920,7 +920,40 @@ In progress. Done so far:
|
||||
a certificate for its name alone, isn't among this server's own
|
||||
certificates, shows `certificateNotAfter`, isn't ordered again, and
|
||||
its order record goes with it when it's removed.
|
||||
- Still to come in step 5: upgrades (RL-29).
|
||||
- Upgrades (RL-29). A release binary comes with a signature file,
|
||||
`inbuxa-relay-<target>.sig`: four lines, `inbuxa-relay release 1`,
|
||||
`version <Cargo version>`, `target <arch>-<os>-<libc>` (like
|
||||
`x86_64-linux-musl`) and `sha256 <hex>`, then `signature <base64url>`,
|
||||
an Ed25519 signature over those four lines.
|
||||
`tools/release/sign-relay.sh` makes it with openssl. The release keys'
|
||||
public halves are compiled into `crates/relay-proto/src/release.rs`;
|
||||
a build without one refuses every upgrade.
|
||||
- An administrator uploads the binary and sends `upgrade` with its blob
|
||||
and the signature file's text. This server checks the signature first,
|
||||
then uploads the binary over its link and calls `Relay/upgrade`; with
|
||||
no link from this node to the relay, the update is refused. The relay
|
||||
refuses it in a container image (`selfUpgrade: false`), while another
|
||||
upgrade is on its way, when the signature fails, when the target isn't
|
||||
its own, and when the version isn't newer than its own. Otherwise it
|
||||
writes the new binary beside its own, runs it with `--version`, which
|
||||
must name the signed version within 10 seconds, keeps the old one as
|
||||
`inbuxa-relay.previous`, and moves the new one into place. It answers,
|
||||
records `relay.upgraded` (or `relay.upgrade-refused`, with the reason),
|
||||
and exits with status 75 (EX_TEMPFAIL) for its supervisor to start the
|
||||
new binary: one set to restart on failure restarts it too.
|
||||
- `olderThanServer` compares each link's last `relayVersion` with this
|
||||
server's version, by SemVer precedence, so a release candidate comes
|
||||
before its release.
|
||||
- Tested: over the relay's real link, a forged signature, an older
|
||||
version, another target and an unreadable file are refused and leave
|
||||
the binary alone (RL-T16); a good release goes in, the old one is kept,
|
||||
the run loop is told to exit, and a second upgrade waits. A binary
|
||||
that says another version is never put in place. The signing script's
|
||||
output checks out against the Rust code. On this server, an upgrade
|
||||
with anything else, an unknown blob, or a signature by no release key
|
||||
is refused before any relay sees it.
|
||||
- Not yet: the release key itself, and signing in the release workflow,
|
||||
come with packaging (step 6). Until then no build accepts an upgrade.
|
||||
|
||||
Order:
|
||||
|
||||
|
||||
@@ -283,6 +283,70 @@ pub async fn test(test: &mut TestServer) {
|
||||
assert_eq!(again.ca_spki, first.ca_spki);
|
||||
assert_ne!(again.secret, first.secret);
|
||||
assert_eq!(get(admin, &id).await["joinState"], "waiting");
|
||||
assert_eq!(
|
||||
get(admin, &id).await["olderThanServer"],
|
||||
Value::Null,
|
||||
"no link has said"
|
||||
);
|
||||
|
||||
// RL-29: a release is checked here before it goes to a relay. This
|
||||
// build has no release key, or one that didn't sign this
|
||||
let release_blob = admin
|
||||
.jmap_client()
|
||||
.await
|
||||
.upload(
|
||||
None,
|
||||
b"#!/bin/sh\necho 'inbuxa-relay 99.0.0'\n".to_vec(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.take_blob_id();
|
||||
let signature = format!(
|
||||
"inbuxa-relay release 1\nversion 99.0.0\ntarget x86_64-linux-musl\n\
|
||||
sha256 {}\nsignature {}\n",
|
||||
"0".repeat(64),
|
||||
"A".repeat(86)
|
||||
);
|
||||
let upgrade = json!({"blobId": release_blob, "signature": signature});
|
||||
let mut created_upgraded = relay("x");
|
||||
created_upgraded["upgrade"] = upgrade.clone();
|
||||
let (_, response) = call(
|
||||
admin,
|
||||
"inbuxa:Relay/set",
|
||||
json!({"create": {"u": created_upgraded}}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
response["notCreated"]["u"]["properties"][0], "upgrade",
|
||||
"{response}"
|
||||
);
|
||||
for (patch, says) in [
|
||||
(json!({"upgrade": "latest"}), "{blobId, signature}"),
|
||||
(
|
||||
json!({"upgrade": upgrade.clone(), "name": "Edge"}),
|
||||
"on its own",
|
||||
),
|
||||
(
|
||||
json!({"upgrade": {"blobId": "nope", "signature": signature}}),
|
||||
"No blob",
|
||||
),
|
||||
(
|
||||
json!({"upgrade": {"blobId": release_blob, "signature": "garbage"}}),
|
||||
"unreadable",
|
||||
),
|
||||
(json!({"upgrade": upgrade}), "release key"),
|
||||
] {
|
||||
let (_, response) = call(admin, "inbuxa:Relay/set", json!({"update": {&id: patch}})).await;
|
||||
let refused = &response["notUpdated"][&id];
|
||||
assert_eq!(refused["properties"][0], "upgrade", "{response}");
|
||||
assert!(
|
||||
refused["description"]
|
||||
.as_str()
|
||||
.is_some_and(|d| d.contains(says)),
|
||||
"{says}: {response}"
|
||||
);
|
||||
}
|
||||
|
||||
// Nobody in a tenant reaches relays
|
||||
let tenant = admin
|
||||
@@ -810,6 +874,8 @@ pub async fn link(test: &mut TestServer) {
|
||||
let recipients = link["recipients"].as_u64().unwrap_or_default();
|
||||
assert!(recipients >= 3, "{relay}");
|
||||
assert_eq!(link["lastError"], Value::Null, "{relay}");
|
||||
// The relay is this build's: not older (RL-28)
|
||||
assert_eq!(relay["olderThanServer"], false, "{relay}");
|
||||
|
||||
// Real recipients only (RL-1)
|
||||
let replies = rcpt_replies(
|
||||
|
||||
Executable
+45
@@ -0,0 +1,45 @@
|
||||
#!/usr/bin/env bash
|
||||
# SPDX-FileCopyrightText: 2026 Coffey Labs LLC
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
#
|
||||
# Sign an inbuxa-relay release binary (relay spec, RL-29):
|
||||
#
|
||||
# tools/release/sign-relay.sh KEY.pem BINARY VERSION TARGET
|
||||
#
|
||||
# KEY.pem is the inbuxa release key (Ed25519, PKCS#8 PEM, as
|
||||
# `openssl genpkey -algorithm ed25519` makes it). VERSION is the Cargo
|
||||
# version the binary's `--version` prints (27.1.0-rc.5), TARGET its target
|
||||
# (x86_64-linux-musl). Writes BINARY.sig, which the relay checks against the
|
||||
# key's public half compiled into crates/relay-proto/src/release.rs.
|
||||
#
|
||||
# tools/release/sign-relay.sh --public KEY.pem
|
||||
#
|
||||
# prints that public half, base64url, for RELEASE_KEYS.
|
||||
#
|
||||
# openssl 3 and coreutils only, so it runs in a plain release image.
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
b64url() { basenc --base64url -w0 | tr -d '='; }
|
||||
|
||||
if [ "${1:-}" = "--public" ] && [ $# -eq 2 ]; then
|
||||
# The raw key is the last 32 bytes of the DER public key
|
||||
openssl pkey -in "$2" -pubout -outform DER | tail -c 32 | b64url
|
||||
echo
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [ $# -ne 4 ]; then
|
||||
sed -n '5,19p' "$0" | sed 's/^# \{0,1\}//'
|
||||
exit 2
|
||||
fi
|
||||
|
||||
key=$1 binary=$2 version=$3 target=$4
|
||||
sha=$(sha256sum "$binary" | cut -d' ' -f1)
|
||||
signed=$(mktemp)
|
||||
trap 'rm -f "$signed"' EXIT
|
||||
printf 'inbuxa-relay release 1\nversion %s\ntarget %s\nsha256 %s\n' \
|
||||
"$version" "$target" "$sha" > "$signed"
|
||||
signature=$(openssl pkeyutl -sign -inkey "$key" -rawin -in "$signed" | b64url)
|
||||
{ cat "$signed"; printf 'signature %s\n' "$signature"; } > "$binary.sig"
|
||||
echo "$binary.sig"
|
||||
Reference in new issue
Block a user