vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

keep the matrix appservice registration in bao, with the host file as the last copy

Browse source
This commit is contained in:
atlas 2026-10-27 13:52:17 +01:00
commit d5c5862654
4 changed files with 1014 additions and 273 deletions
Download patch file Download diff file Expand all files Collapse all files

15
swarm-matrix-ctl/README.md
View file

@ -13,13 +13,18 @@ crate.
### `appservice render`
Mints the swarm appservice registration's tokens when absent and renders the
registration tuwunel loads. Runs before the homeserver and needs no network.
Renders the swarm appservice registration tuwunel loads from the tokens in the
state dir, minting them only on a host that has none. Runs before the
homeserver and needs no network: the tokens on disk are the last copy `sync`
took from the store.
### `appservice publish`
### `appservice sync`
Writes the rendered `as_token` to the swarm secret store for `swarm-controller`,
when the store's copy differs.
The store holds the registration's tokens. This writes them into the state dir
when they differ, restarts the homeserver whenever the registration is
newer than its last start, and publishes the `as_token` for `swarm-controller`. A store holding none is seeded from the
state dir, create-only, so nothing rotates. It runs beside the homeserver, so a
store outage leaves the homeserver running on the last copy.
Both are configured entirely by the `MATRIX_APPSERVICE_*` environment the units
set — no flags. A systemd `Environment=` block is what a nix module can render; a

790
swarm-matrix-ctl/src/appservice.rs
View file

@ -1,38 +1,38 @@
//! `swarm-matrix-ctl appservice` — the **swarm's** own appservice
//! registration: minted here, loaded by the homeserver beside us, and published
//! to the one store path `swarm-controller` reads it from.
//! registration. The store holds it; this container keeps a copy the
//! homeserver starts from.
//!
//! Two verbs, because they have opposite failure rules:
//!
//! - [`render`] runs before tuwunel and touches nothing but this container's
//! state dir. tuwunel loads the file it writes through `LoadCredential`, and
//! a missing credential source fails the homeserver's start, so this half
//! must not be able to fail on a network.
//! - [`publish`] runs after it and needs the store. A sealed store delays it
//! must not be able to fail on a network. It renders from the tokens on
//! disk, which are the last copy [`sync`] took from the store, and mints
//! them only on a host that has none.
//! - [`sync`] runs beside the homeserver, never before it, and needs the store.
//! The store's tokens win: it writes the state dir only when they differ, and
//! restarts the homeserver whenever the registration is newer than the
//! homeserver's last start. A store holding none is seeded from
//! the state dir, create-only, so nothing rotates. A sealed store delays it
//! and nothing else.
//!
//! "Only once": the token file in the state dir is the record. [`render`]
//! mints only when it is absent and re-renders from it every time; [`publish`]
//! writes the store only when the store's copy differs.
//!
//! This registration's sender is promoted to homeserver admin at boot
//! (`nix/host-modules/hive-matrix.nix`), which is why its token goes to
//! `swarm_secret_client::matrix::swarm_appservice_token_path` — a path no
//! hive's policy reaches — and to nowhere else.
//! (`nix/host-modules/hive-matrix.nix`), which is why its `as_token` goes to
//! `swarm_secret_client::matrix::swarm_appservice_token_path` for
//! `swarm-controller` — a path no hive's policy reaches — and to nowhere else.
use std::io::Write as _;
use std::os::unix::fs::OpenOptionsExt as _;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use anyhow::{Context, Result, bail};
use swarm_secret_client::{
SecretStore,
client::{DEFAULT_CERT_MOUNT, Settings},
matrix,
matrix::{self, Registration},
};
use crate::registration;
/// This container's state dir for the registration and its two tokens.
const ENV_DIR: &str = "MATRIX_APPSERVICE_DIR";
/// The registration's `sender_localpart`: the account the homeserver creates
@ -40,8 +40,11 @@ const ENV_DIR: &str = "MATRIX_APPSERVICE_DIR";
const ENV_SENDER: &str = "MATRIX_APPSERVICE_SENDER";
/// The user namespace regex, rendered by nix.
const ENV_USER_REGEX: &str = "MATRIX_APPSERVICE_USER_REGEX";
/// Role on the store's `cert` auth mount that [`publish`] logs in with.
/// Role on the store's `cert` auth mount that [`sync`] logs in with.
const ENV_CERT_ROLE: &str = "MATRIX_APPSERVICE_CERT_ROLE";
/// The unit that loads the registration, which [`sync`] restarts when the
/// registration is newer than its last start.
const ENV_RESTART_UNIT: &str = "MATRIX_APPSERVICE_RESTART_UNIT";
/// The registration's `id`.
const ID: &str = "swarm";
@ -85,11 +88,8 @@ fn render_into(dir: &Path, sender: &str, regex: &str) -> Result<()> {
/// The token at `path`, minting and writing one first when there is none.
fn existing_or_minted(path: &Path) -> Result<String> {
match std::fs::read_to_string(path) {
Ok(t) if !t.trim().is_empty() => return Ok(t.trim().to_owned()),
Ok(_) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
if let Some(t) = read_token(path)? {
return Ok(t);
}
let token = swarm_matrix_client::random_hex(TOKEN_BYTES)?;
write_secret(path, &token)?;
@ -97,6 +97,16 @@ fn existing_or_minted(path: &Path) -> Result<String> {
Ok(token)
}
/// The trimmed token at `path`; `None` when the file is missing or blank.
fn read_token(path: &Path) -> Result<Option<String>> {
match std::fs::read_to_string(path) {
Ok(t) if !t.trim().is_empty() => Ok(Some(t.trim().to_owned())),
Ok(_) => Ok(None),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e).with_context(|| format!("reading {}", path.display())),
}
}
/// Write `contents` to `path` as a `0600` file, through a rename so a reader
/// never sees half of it.
fn write_secret(path: &Path, contents: &str) -> Result<()> {
@ -136,23 +146,74 @@ fn registration_yaml(sender: &str, regex: &str, as_token: &str, hs_token: &str)
)
}
/// Whether the store needs the local token written to it.
fn needs_publish(stored: Option<&matrix::Credential>, local: &str) -> bool {
stored.is_none_or(|c| c.value.trim() != local)
/// What [`sync`] needs of the store, so the decisions can be tested without
/// one.
trait Store {
/// The registration's tokens, or `None` when nothing is stored.
async fn registration(&self) -> Result<Option<Registration>>;
/// Store `r` only if nothing is stored yet.
async fn create_registration(&self, r: &Registration) -> Result<()>;
/// What `swarm-controller` reads.
async fn controller_token(&self) -> Result<Option<matrix::Credential>>;
async fn write_controller_token(&self, as_token: &str) -> Result<()>;
}
/// Write the rendered registration's `as_token` to the store, unless the store
/// already holds it.
/// The real store, logged in as matrix-ctl.
struct Bao(SecretStore);
impl Store for Bao {
async fn registration(&self) -> Result<Option<Registration>> {
let path = matrix::swarm_appservice_registration_path()?;
self.0
.read_optional(&path)
.await
.with_context(|| format!("reading {path}"))
}
async fn create_registration(&self, r: &Registration) -> Result<()> {
let path = matrix::swarm_appservice_registration_path()?;
self.0
.create(&path, r)
.await
.with_context(|| format!("creating {path}"))
}
async fn controller_token(&self) -> Result<Option<matrix::Credential>> {
let path = matrix::swarm_appservice_token_path()?;
self.0
.read_optional(&path)
.await
.with_context(|| format!("reading {path}"))
}
async fn write_controller_token(&self, as_token: &str) -> Result<()> {
let path = matrix::swarm_appservice_token_path()?;
self.0
.write(
&path,
&matrix::Credential {
value: as_token.to_owned(),
homeserver: None,
},
)
.await
.with_context(|| format!("writing {path}"))
}
}
/// Make the state dir hold the store's registration, and the controller's token
/// match it.
///
/// # Errors
/// If a variable is missing, the registration has not been rendered, or the
/// store refuses the login, the read or the write.
pub async fn publish() -> Result<()> {
/// If a variable is missing, the store refuses the login, a read or a write,
/// the state dir cannot be written, or the homeserver cannot be restarted.
pub async fn sync() -> Result<()> {
let get = |k: &str| std::env::var(k).ok();
let dir = PathBuf::from(required(&get, ENV_DIR)?);
let sender = required(&get, ENV_SENDER)?;
let regex = required(&get, ENV_USER_REGEX)?;
let cert_role = required(&get, ENV_CERT_ROLE)?;
let registration = dir.join(REGISTRATION);
let local = registration::as_token(&registration.to_string_lossy())?;
let unit = required(&get, ENV_RESTART_UNIT)?;
let settings = Settings::from_env().context("reading the store's BAO_* environment")?;
let store = SecretStore::connect(&settings, &cert_role, DEFAULT_CERT_MOUNT)
@ -160,31 +221,228 @@ pub async fn publish() -> Result<()> {
.with_context(|| {
format!("logging in to the swarm secret store as cert role {cert_role}")
})?;
let path = matrix::swarm_appservice_token_path()?;
let stored: Option<matrix::Credential> = store
.read_optional(&path)
.await
.with_context(|| format!("reading {path}"))?;
if !needs_publish(stored.as_ref(), &local) {
tracing::info!(%path, "the swarm appservice token is already published");
return Ok(());
sync_with(&Bao(store), &dir, &sender, &regex, mint, &Systemd(unit)).await
}
/// [`sync`] against an explicit store, directory and homeserver unit.
///
/// The store's registration wins. Where it holds none, the state dir's tokens
/// seed it, and only a host with neither mints a pair. The homeserver is
/// restarted whenever the registration was written after it last started, so
/// a run that failed between the write and the restart is finished by the next.
async fn sync_with(
store: &impl Store,
dir: &Path,
sender: &str,
regex: &str,
mint: impl FnOnce() -> Result<Registration>,
unit: &impl Homeserver,
) -> Result<()> {
let local = read_local(dir)?;
let current = stored_or_seeded(store, local.as_ref(), mint).await?;
let want = registration_yaml(sender, regex, &current.as_token, &current.hs_token);
let have = std::fs::read_to_string(dir.join(REGISTRATION)).ok();
if local.as_ref() != Some(&current) || have.as_deref() != Some(&want) {
write_secret(&dir.join(AS_TOKEN), &current.as_token)?;
write_secret(&dir.join(HS_TOKEN), &current.hs_token)?;
write_secret(&dir.join(REGISTRATION), &want)?;
tracing::info!("took the swarm appservice registration from the store");
}
restart_if_stale(unit, &dir.join(REGISTRATION))?;
if needs_publish(store.controller_token().await?.as_ref(), &current.as_token) {
store.write_controller_token(&current.as_token).await?;
tracing::info!("published the swarm appservice token for the controller");
}
Ok(())
}
/// The store's registration; where it holds none, `local` or a minted pair is
/// created there first. A refused create is a lost race or a denial, and the
/// read-back tells which: it finds the winner's value, or nothing.
async fn stored_or_seeded(
store: &impl Store,
local: Option<&Registration>,
mint: impl FnOnce() -> Result<Registration>,
) -> Result<Registration> {
if let Some(stored) = store.registration().await? {
return Ok(stored);
}
let seed = match local {
Some(l) => l.clone(),
None => mint()?,
};
if let Err(e) = store.create_registration(&seed).await {
tracing::warn!(error = %e, "could not create the swarm appservice registration; reading it back");
}
store
.write(
&path,
&matrix::Credential {
value: local,
homeserver: None,
},
)
.await
.with_context(|| format!("writing the swarm appservice token to {path}"))?;
tracing::info!(%path, "published the swarm appservice token");
Ok(())
.registration()
.await?
.context("the store holds no swarm appservice registration after creating one")
}
/// The tokens on disk, when both are there.
fn read_local(dir: &Path) -> Result<Option<Registration>> {
let as_token = read_token(&dir.join(AS_TOKEN))?;
let hs_token = read_token(&dir.join(HS_TOKEN))?;
Ok(as_token
.zip(hs_token)
.map(|(as_token, hs_token)| Registration { as_token, hs_token }))
}
fn mint() -> Result<Registration> {
Ok(Registration {
as_token: swarm_matrix_client::random_hex(TOKEN_BYTES)?,
hs_token: swarm_matrix_client::random_hex(TOKEN_BYTES)?,
})
}
/// Whether the store needs the local token written to it.
fn needs_publish(stored: Option<&matrix::Credential>, local: &str) -> bool {
stored.is_none_or(|c| c.value.trim() != local)
}
/// What [`sync`] needs of the unit that loads the registration.
trait Homeserver {
/// When the unit last became active, in microseconds since the epoch;
/// `None` when it never has.
fn started_us(&self) -> Result<Option<u64>>;
/// Restart it onto the registration on disk.
fn restart(&self) -> Result<()>;
}
/// The real unit, through `systemctl`.
struct Systemd(String);
impl Homeserver for Systemd {
fn started_us(&self) -> Result<Option<u64>> {
let out = std::process::Command::new("systemctl")
.args([
"show",
"--timestamp=us+utc",
"-p",
"ActiveEnterTimestamp",
"--value",
])
.arg(&self.0)
.output()
.context("running systemctl show")?;
if !out.status.success() {
bail!("systemctl show {} exited with {}", self.0, out.status);
}
parse_timestamp_us(String::from_utf8_lossy(&out.stdout).trim())
}
fn restart(&self) -> Result<()> {
restart_via(&self.0, systemctl)
}
}
/// Restart the homeserver when the registration on disk is newer than its last
/// start. Derived from the files and the unit on every run rather than from
/// what this run wrote, so a run that failed after writing is retried by the
/// next. A start time that cannot be read leaves the unit as it is: the next
/// run decides again.
fn restart_if_stale(unit: &impl Homeserver, registration: &Path) -> Result<()> {
let started = match unit.started_us() {
Ok(started) => started.unwrap_or(0),
Err(e) => {
tracing::warn!(error = %e, "cannot tell when the homeserver last started; leaving it as it is");
return Ok(());
}
};
let written = std::fs::metadata(registration)
.and_then(|m| m.modified())
.with_context(|| format!("reading the time of {}", registration.display()))?
.duration_since(std::time::UNIX_EPOCH)
.context("the registration is dated before the epoch")?
.as_micros();
if written <= u128::from(started) {
return Ok(());
}
tracing::info!("the registration is newer than the homeserver's start; restarting it");
unit.restart()
.context("restarting the homeserver onto the new registration")
}
/// `systemctl systemd-timestamp` output (`--timestamp=us+utc`, for example
/// `Tue 2026-10-27 12:00:00.123456 UTC`) as microseconds since the epoch. Empty
/// is `None`: a unit that never became active has no timestamp.
fn parse_timestamp_us(s: &str) -> Result<Option<u64>> {
if s.is_empty() {
return Ok(None);
}
let bad = || anyhow::anyhow!("unreadable systemd timestamp {s:?}");
let token = |sep: char| {
s.split_whitespace()
.find(|t| t.contains(sep))
.ok_or_else(bad)
};
let numbers = |text: &str, sep: char| -> Result<Vec<i64>> {
text.split(sep)
.map(|n| n.parse().map_err(|_| bad()))
.collect()
};
let time = token(':')?;
let (clock, fraction) = time.split_once('.').unwrap_or((time, ""));
let [year, month, day] = numbers(token('-')?, '-')?[..] else {
return Err(bad());
};
let [hour, minute, second] = numbers(clock, ':')?[..] else {
return Err(bad());
};
let micros: i64 = format!("{fraction:0<6}")[..6].parse().map_err(|_| bad())?;
// Days since 1970-01-01 of a proleptic Gregorian date.
let year = if month <= 2 { year - 1 } else { year };
let era = year.div_euclid(400);
let year_of_era = year - era * 400;
let day_of_year = (153 * ((month + 9) % 12) + 2) / 5 + day - 1;
let day_of_era = year_of_era * 365 + year_of_era / 4 - year_of_era / 100 + day_of_year;
let days = era * 146_097 + day_of_era - 719_468;
let seconds = days * 86_400 + hour * 3600 + minute * 60 + second;
u64::try_from(seconds * 1_000_000 + micros)
.map(Some)
.map_err(|_| bad())
}
/// Run `systemctl` with `args`, returning its exit status.
fn systemctl(args: &[&str]) -> Result<std::process::ExitStatus> {
std::process::Command::new("systemctl")
.args(args)
.status()
.with_context(|| format!("running systemctl {}", args.join(" ")))
}
/// Restart `unit` onto the new registration: a running one is restarted, and a
/// failed one — it may have failed for want of the file — is reset and started.
/// One stopped on purpose stays stopped. `--no-block`, so this run does not
/// wait on the homeserver's start. Every call but the `is-failed` probe must
/// exit zero.
fn restart_via(
unit: &str,
run: impl Fn(&[&str]) -> Result<std::process::ExitStatus>,
) -> Result<()> {
let must = |args: &[&str]| {
let status = run(args)?;
if status.success() {
Ok(())
} else {
bail!("systemctl {} exited with {status}", args.join(" "))
}
};
if run(&["is-failed", "--quiet", unit])?.success() {
must(&["reset-failed", unit])?;
must(&["start", "--no-block", unit])
} else {
must(&["try-restart", "--no-block", unit])
}
}
#[cfg(test)]
mod tests {
use std::cell::{Cell, RefCell};
use super::*;
const REGEX: &str = "^@[a-z0-9._=/-]+:t\\.local$";
@ -202,24 +460,152 @@ mod tests {
dir
}
fn reg(tag: &str) -> Registration {
Registration {
as_token: format!("as-{tag}"),
hs_token: format!("hs-{tag}"),
}
}
/// A store in memory. `down` fails every call, as a sealed or unreachable
/// store does; `racer` is what another writer stores between this run's
/// read and its create.
#[derive(Default)]
struct Fake {
registration: RefCell<Option<Registration>>,
token: RefCell<Option<String>>,
down: bool,
racer: RefCell<Option<Registration>>,
creates: Cell<u32>,
token_writes: Cell<u32>,
}
impl Store for Fake {
async fn registration(&self) -> Result<Option<Registration>> {
if self.down {
bail!("the store is down");
}
Ok(self.registration.borrow().clone())
}
async fn create_registration(&self, r: &Registration) -> Result<()> {
self.creates.set(self.creates.get() + 1);
if let Some(winner) = self.racer.borrow_mut().take() {
*self.registration.borrow_mut() = Some(winner);
bail!("already created");
}
*self.registration.borrow_mut() = Some(r.clone());
Ok(())
}
async fn controller_token(&self) -> Result<Option<matrix::Credential>> {
Ok(self.token.borrow().clone().map(|value| matrix::Credential {
value,
homeserver: None,
}))
}
async fn write_controller_token(&self, as_token: &str) -> Result<()> {
self.token_writes.set(self.token_writes.get() + 1);
*self.token.borrow_mut() = Some(as_token.to_owned());
Ok(())
}
}
/// Write the state dir as `render` leaves it for `r`.
fn host_with(dir: &Path, r: &Registration) {
write_secret(&dir.join(AS_TOKEN), &r.as_token).expect("as");
write_secret(&dir.join(HS_TOKEN), &r.hs_token).expect("hs");
write_secret(
&dir.join(REGISTRATION),
&registration_yaml("swarm", REGEX, &r.as_token, &r.hs_token),
)
.expect("yaml");
}
fn files(dir: &Path) -> Vec<String> {
[AS_TOKEN, HS_TOKEN, REGISTRATION]
.iter()
.map(|f| std::fs::read_to_string(dir.join(f)).unwrap_or_default())
.collect()
}
/// The homeserver unit. `started` is when it last became active; a restart
/// that succeeds makes it later than every file.
struct Unit {
started: Cell<Option<u64>>,
fail_restart: Cell<bool>,
restarts: Cell<u32>,
}
impl Unit {
/// Last started at the epoch, before any file in a test.
fn long_running() -> Self {
Self::started(Some(1))
}
/// Last started after every file in a test.
fn just_started() -> Self {
Self::started(Some(u64::MAX))
}
fn started(at: Option<u64>) -> Self {
Self {
started: Cell::new(at),
fail_restart: Cell::new(false),
restarts: Cell::new(0),
}
}
}
impl Homeserver for Unit {
fn started_us(&self) -> Result<Option<u64>> {
Ok(self.started.get())
}
fn restart(&self) -> Result<()> {
self.restarts.set(self.restarts.get() + 1);
if self.fail_restart.get() {
bail!("systemctl refused");
}
self.started.set(Some(u64::MAX));
Ok(())
}
}
/// Run one sync against `unit`, returning how often it minted.
async fn run(store: &Fake, dir: &Path, unit: &Unit) -> (Result<()>, u32) {
let mints = Cell::new(0);
let out = sync_with(
store,
dir,
"swarm",
REGEX,
|| {
mints.set(mints.get() + 1);
Ok(reg("minted"))
},
unit,
)
.await;
(out, mints.get())
}
#[test]
fn the_rendered_registration_carries_the_token_it_minted() {
let dir = scratch();
render_into(&dir, "swarm", REGEX).expect("renders");
let token = registration::as_token(&dir.join(REGISTRATION).to_string_lossy())
.expect("the as_token line parses");
let yaml = std::fs::read_to_string(dir.join(REGISTRATION)).expect("rendered");
let token = std::fs::read_to_string(dir.join(AS_TOKEN)).expect("minted");
assert_eq!(token.len(), TOKEN_BYTES * 2);
assert_eq!(
std::fs::read_to_string(dir.join(AS_TOKEN)).expect("minted"),
token
);
assert!(yaml.contains(&format!("\nas_token: {token}\n")), "{yaml}");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn a_second_render_keeps_the_token() {
// "Only once": a re-mint on every boot would hand the controller a
// token the homeserver no longer loads until publish catches up.
// token the homeserver no longer loads until sync catches up.
let dir = scratch();
render_into(&dir, "swarm", REGEX).expect("first");
let first = std::fs::read_to_string(dir.join(REGISTRATION)).expect("rendered");
@ -254,8 +640,288 @@ mod tests {
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn seeding_from_the_host_leaves_every_file_unchanged() {
// A host that has run for months with the store empty: the store takes
// its value, and nothing on the host moves, so nothing rotates.
let dir = scratch();
let on_disk = reg("old");
host_with(&dir, &on_disk);
let before = files(&dir);
let store = Fake::default();
let unit = Unit::just_started();
let (out, mints) = run(&store, &dir, &unit).await;
let restarts = unit.restarts.get();
out.expect("syncs");
assert_eq!(*store.registration.borrow(), Some(on_disk));
assert_eq!(files(&dir), before);
assert_eq!(mints, 0, "a host that has the value mints nothing");
assert_eq!(restarts, 0, "an unchanged host does not restart");
assert_eq!(store.token.borrow().as_deref(), Some("as-old"));
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn the_store_wins_when_both_hold_a_value() {
let dir = scratch();
host_with(&dir, &reg("disk"));
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
*store.token.borrow_mut() = Some("as-disk".to_owned());
let unit = Unit::long_running();
let (out, _) = run(&store, &dir, &unit).await;
let restarts = unit.restarts.get();
out.expect("syncs");
let want = registration_yaml("swarm", REGEX, "as-store", "hs-store");
assert_eq!(
files(&dir),
vec!["as-store".to_owned(), "hs-store".to_owned(), want]
);
assert_eq!(*store.registration.borrow(), Some(reg("store")));
assert_eq!(store.creates.get(), 0, "a held value is never rewritten");
assert_eq!(restarts, 1);
assert_eq!(store.token.borrow().as_deref(), Some("as-store"));
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_host_without_the_files_takes_them_from_the_store() {
let dir = scratch();
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
let unit = Unit::long_running();
let (out, mints) = run(&store, &dir, &unit).await;
let restarts = unit.restarts.get();
out.expect("syncs");
assert_eq!(read_local(&dir).expect("reads"), Some(reg("store")));
assert!(dir.join(REGISTRATION).exists());
assert_eq!((restarts, mints), (1, 0));
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_fresh_host_mints_into_the_store() {
let dir = scratch();
let store = Fake::default();
let unit = Unit::long_running();
let (out, mints) = run(&store, &dir, &unit).await;
let restarts = unit.restarts.get();
out.expect("syncs");
assert_eq!(mints, 1);
assert_eq!(*store.registration.borrow(), Some(reg("minted")));
assert_eq!(read_local(&dir).expect("reads"), Some(reg("minted")));
assert_eq!(restarts, 1);
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_lost_race_adopts_the_winners_registration() {
let dir = scratch();
host_with(&dir, &reg("disk"));
let store = Fake::default();
*store.racer.borrow_mut() = Some(reg("winner"));
let unit = Unit::long_running();
let (out, _) = run(&store, &dir, &unit).await;
let restarts = unit.restarts.get();
out.expect("syncs");
assert_eq!(read_local(&dir).expect("reads"), Some(reg("winner")));
assert_eq!(restarts, 1);
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_store_that_is_down_leaves_the_host_as_it_is() {
let dir = scratch();
host_with(&dir, &reg("disk"));
let before = files(&dir);
let store = Fake {
down: true,
..Fake::default()
};
let unit = Unit::just_started();
let (out, mints) = run(&store, &dir, &unit).await;
let restarts = unit.restarts.get();
assert!(out.is_err(), "the unit reports the outage");
assert_eq!(files(&dir), before, "the homeserver keeps the last copy");
assert_eq!((restarts, mints), (0, 0));
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn the_start_path_renders_the_copy_sync_took_from_the_store() {
// `render` has no store: after an outage it renders what the last sync
// wrote, without minting over it.
let dir = scratch();
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
run(&store, &dir, &Unit::long_running())
.await
.0
.expect("syncs");
let synced = files(&dir);
render_into(&dir, "swarm", REGEX).expect("renders");
assert_eq!(files(&dir), synced);
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_second_sync_restarts_nothing() {
let dir = scratch();
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
let unit = Unit::long_running();
run(&store, &dir, &unit).await.0.expect("first");
let first = unit.restarts.get();
let (out, _) = run(&store, &dir, &unit).await;
let second = unit.restarts.get() - first;
out.expect("syncs");
assert_eq!((first, second), (1, 0));
assert_eq!(
store.token_writes.get(),
1,
"the controller's copy is written once"
);
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_restart_that_failed_is_retried_by_the_next_run() {
// The files are written and the restart is refused. The next run finds
// nothing to write, and still restarts: the decision comes from the
// files' age against the unit's start, not from this run's writes.
let dir = scratch();
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
let unit = Unit::long_running();
unit.fail_restart.set(true);
let (first, _) = run(&store, &dir, &unit).await;
assert!(first.is_err(), "the refused restart fails the run");
let written = files(&dir);
unit.fail_restart.set(false);
let (second, _) = run(&store, &dir, &unit).await;
second.expect("the retry succeeds");
assert_eq!(files(&dir), written, "nothing was left to write");
assert_eq!(unit.restarts.get(), 2);
assert_eq!(read_local(&dir).expect("reads"), Some(reg("store")));
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_unit_that_started_after_the_files_is_left_alone() {
// The control for the retry above: the same files, a unit that started
// after them, and no restart.
let dir = scratch();
host_with(&dir, &reg("store"));
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
let unit = Unit::just_started();
run(&store, &dir, &unit).await.0.expect("syncs");
assert_eq!(unit.restarts.get(), 0);
std::fs::remove_dir_all(&dir).ok();
}
#[tokio::test]
async fn a_start_time_that_cannot_be_read_restarts_nothing() {
struct Unreadable(Cell<u32>);
impl Homeserver for Unreadable {
fn started_us(&self) -> Result<Option<u64>> {
bail!("no such unit")
}
fn restart(&self) -> Result<()> {
self.0.set(self.0.get() + 1);
Ok(())
}
}
let dir = scratch();
let store = Fake::default();
*store.registration.borrow_mut() = Some(reg("store"));
let unit = Unreadable(Cell::new(0));
let out = sync_with(&store, &dir, "swarm", REGEX, || Ok(reg("minted")), &unit).await;
out.expect("the run goes on");
assert_eq!(unit.0.get(), 0);
assert_eq!(store.token.borrow().as_deref(), Some("as-store"));
std::fs::remove_dir_all(&dir).ok();
}
/// `restart_via` against a `systemctl` that exits with the given codes for the
/// verbs named (zero for the rest), returning the verbs it was called with.
fn restart_script(exits: &[(&'static str, i32)]) -> (Result<()>, Vec<String>) {
use std::os::unix::process::ExitStatusExt as _;
let calls = RefCell::new(Vec::new());
let out = restart_via("tuwunel.service", |args| {
calls.borrow_mut().push(args[0].to_owned());
let code = exits.iter().find(|(v, _)| *v == args[0]).map_or(0, |e| e.1);
Ok(std::process::ExitStatus::from_raw(code << 8))
});
(out, calls.into_inner())
}
#[test]
fn publish_writes_only_what_the_store_lacks() {
fn a_running_unit_is_restarted_and_a_refusal_fails_the_run() {
let (out, calls) = restart_script(&[("is-failed", 1)]);
out.expect("restarts");
assert_eq!(calls, ["is-failed", "try-restart"]);
let (out, _) = restart_script(&[("is-failed", 1), ("try-restart", 1)]);
assert!(out.is_err());
}
#[test]
fn a_failed_unit_is_reset_and_started_and_a_refusal_of_either_fails_the_run() {
let (out, calls) = restart_script(&[]);
out.expect("resets and starts");
assert_eq!(calls, ["is-failed", "reset-failed", "start"]);
let (out, calls) = restart_script(&[("reset-failed", 1)]);
assert!(out.is_err(), "a refused reset fails the run");
assert_eq!(calls, ["is-failed", "reset-failed"], "and nothing starts");
let (out, _) = restart_script(&[("start", 1)]);
assert!(out.is_err(), "a refused start fails the run");
}
#[test]
fn systemd_timestamps_read_as_microseconds_since_the_epoch() {
let at = |s| parse_timestamp_us(s).expect("parses");
assert_eq!(at("Thu 1970-01-01 00:00:01.000002 UTC"), Some(1_000_002));
assert_eq!(
at("Wed 2000-03-01 00:00:00.5 UTC"),
Some(951_868_800_500_000)
);
assert_eq!(
at("Tue 2026-10-27 12:00:00 UTC"),
Some(1_793_102_400_000_000)
);
assert_eq!(at(""), None, "a unit that never started has none");
parse_timestamp_us("n/a garbage").expect_err("not a timestamp");
}
#[test]
fn the_controllers_copy_is_written_only_when_it_differs() {
let same = matrix::Credential {
value: "aa".to_owned(),
homeserver: None,
@ -271,7 +937,13 @@ mod tests {
#[test]
fn every_variable_is_scoped_to_the_verb() {
for var in [ENV_DIR, ENV_SENDER, ENV_USER_REGEX, ENV_CERT_ROLE] {
for var in [
ENV_DIR,
ENV_SENDER,
ENV_USER_REGEX,
ENV_CERT_ROLE,
ENV_RESTART_UNIT,
] {
assert!(var.starts_with("MATRIX_APPSERVICE_"), "{var}");
}
}

22
swarm-matrix-ctl/src/main.rs
View file

@ -8,14 +8,13 @@
//! here is a verb below, not a new crate.
//!
//! [`appservice`] mints the **swarm's** own appservice registration and
//! publishes its token for `swarm-controller`.
//! keeps it in step with the store and publishes its token for `swarm-controller`.
//!
//! 🩸 **A secret is a path, never a value.** The only identifier any verb here
//! logs is the store path; see `swarm_matrix_client`'s module doc for the same rule
//! applied to error messages.
mod appservice;
mod registration;
use anyhow::Result;
use clap::{Parser, Subcommand};
@ -43,9 +42,10 @@ enum Appservice {
/// Mint the tokens when absent and render the registration tuwunel loads.
/// Local only: it runs before the homeserver and must not need a network.
Render,
/// Write the rendered `as_token` to the swarm secret store when the
/// store's copy differs.
Publish,
/// Make the state dir hold the store's registration, seeding the store
/// from it when the store holds none, and restart the homeserver when the
/// registration is newer than its last start. Runs beside the homeserver, never before it.
Sync,
}
#[tokio::main]
@ -59,7 +59,7 @@ async fn main() -> Result<()> {
match Cli::parse().command {
Command::Appservice(Appservice::Render) => appservice::render(),
Command::Appservice(Appservice::Publish) => appservice::publish().await,
Command::Appservice(Appservice::Sync) => appservice::sync().await,
}
}
@ -83,12 +83,8 @@ mod tests {
cli.command,
Command::Appservice(Appservice::Render)
));
let cli =
Cli::try_parse_from(["swarm-matrix-ctl", "appservice", "publish"]).expect("a verb");
assert!(matches!(
cli.command,
Command::Appservice(Appservice::Publish)
));
let cli = Cli::try_parse_from(["swarm-matrix-ctl", "appservice", "sync"]).expect("a verb");
assert!(matches!(cli.command, Command::Appservice(Appservice::Sync)));
Cli::try_parse_from(["swarm-matrix-ctl", "appservice"])
.expect_err("a sub-verb is required");
}
@ -99,7 +95,7 @@ mod tests {
.expect_err("only declared verbs are accepted");
}
/// A bare invocation must not silently do something. `appservice publish`
/// A bare invocation must not silently do something. `appservice sync`
/// writes a credential, so "no verb" defaulting to it would make a typo in
/// the unit write rather than fail.
#[test]

102
swarm-matrix-ctl/src/registration.rs
View file

@ -1,102 +0,0 @@
//! Reading the `as_token` out of the appservice registration the matrix
//! container already has.
//!
//! No new credential is delivered for this. The registration **is** the
//! `as_token`, and `appservice render` writes it into this container's own
//! state dir — so the file this module opens is one this process could already
//! read, and one the homeserver beside it reads too.
//!
//! Scanned line-by-line rather than parsed as YAML. The file has exactly one
//! renderer (`registration_yaml` in `appservice.rs`, a `format!` of
//! `as_token: <hex>`), so a parser would be a second, looser reading of a shape
//! this repo writes itself — and it would pull a YAML crate into a binary whose
//! only other input is JSON.
use anyhow::{Context, Result, bail};
/// The key the token is stored under, and the whole of the agreement with the
/// renderer.
const KEY: &str = "as_token:";
/// Read the registration at `path` and return its `as_token`.
///
/// # Errors
/// When the file cannot be read, or holds no `as_token` with a value — which is
/// what a registration rendered by something other than this repo looks like
/// from here.
pub fn as_token(path: &str) -> Result<String> {
let text = std::fs::read_to_string(path)
.with_context(|| format!("reading the appservice registration at {path}"))?;
// The path, not the file's contents: every line of it is either a secret or
// a shape this module already knows.
extract(&text).with_context(|| format!("no `as_token` in the registration at {path}"))
}
/// [`as_token`] over text already in hand, so the agreement with the renderer
/// can be tested without a file.
fn extract(text: &str) -> Result<String> {
for line in text.lines() {
if let Some(rest) = line.strip_prefix(KEY) {
let token = rest.trim();
if token.is_empty() {
bail!("the registration's `as_token` is empty");
}
return Ok(token.to_owned());
}
}
bail!("the registration carries no `as_token` line")
}
#[cfg(test)]
mod tests {
use super::*;
/// The registration in the shape `registration_yaml` renders it, tokens
/// last. Reproduced rather than rendered so this module's tests do not
/// depend on the renderer's.
const RENDERED: &str = "id: swarm\n\
url: null\n\
sender_localpart: swarm\n\
rate_limited: false\n\
namespaces:\n \
users:\n \
- exclusive: false\n \
regex: '@[a-z0-9._=/+-]+:example\\.test$'\n \
aliases: []\n \
rooms: []\n\
as_token: deadbeef\n\
hs_token: cafebabe\n";
#[test]
fn the_token_is_taken_from_the_registration_this_repo_renders() {
assert_eq!(
extract(RENDERED).expect("the rendered shape parses"),
"deadbeef"
);
}
#[test]
fn the_homeservers_own_token_is_not_mistaken_for_the_appservices() {
// `hs_token` authenticates the homeserver TO the appservice and is a
// different secret with a confusingly similar name; a substring search
// would find it inside neither, but a `contains("s_token")`-shaped one
// would. The control is that the line order in `RENDERED` puts
// `as_token` first, so this arm needs the reverse to mean anything.
let reversed = "hs_token: cafebabe\nas_token: deadbeef\n";
assert_eq!(
extract(reversed).expect("order does not matter"),
"deadbeef"
);
}
#[test]
fn a_registration_with_no_token_is_an_error_rather_than_an_empty_string() {
// An empty token authenticates nothing, and a homeserver answers a
// request carrying one with a 403 that names the account rather than
// the credential — so failing here is the only report an operator can
// act on.
for bad in ["id: swarm\n", "as_token:\n", "as_token: \n"] {
assert!(extract(bad).is_err(), "{bad:?} must not yield a token");
}
}
}