Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/hive-matrix-mcp/src/client.rs

569 lines
22 KiB
Rust

//! matrix-sdk `Client` setup for the daemon: take the per-agent access
//! token [`crate::credential`] resolved, probe `whoami` to recover the
//! agent's matrix `user_id` + `device_id`, restore the matrix-sdk
//! session, return the Client ready to start sync.
//!
//! 🩸 The token arrives as a [`Token`] and never as a path this module
//! reads — where it came from is [`crate::credential`]'s question, and the
//! only thing said about it here is its [`crate::credential::Origin`].
//!
//! No OAuth dance / cross-signing setup (in contrast to damocles-daemon's
//! ccc.de connection): for the swarm's tuwunel the swarm already minted the
//! token + user/device as its appservice and handed us the bearer. matrix-sdk's `restore_session` with a constructed
//! `MatrixSession` skips the login flow entirely.
//!
//! E2EE is enabled via `with_encryption_settings`. Crypto keys are persisted
//! in the sqlite store under `state_dir` (survives container restarts, lost on
//! `--purge`).
//!
//! Cross-signing is bootstrapped best-effort after session restore
//! (`ensure_cross_signing`) so the agent's device stops showing as
//! "unverified" in other users' Matrix clients: the in-hive tuwunel accepts
//! the initial device-signing-key upload on the bearer token alone (no UIAA
//! re-auth). A homeserver that DID require UIAA just fails the call, which is
//! swallowed — the session stays exactly as unverified as before.
//!
//! Server-side key backup is turned on or recovered by `ensure_recovery`, with
//! the passphrase `swarm-controller` mints per account, so a crypto store lost
//! to a purge or a token reset gets its history back.
use std::path::Path;
use anyhow::{Context, Result, anyhow, bail};
use matrix_sdk::{
Client, SessionMeta, SessionTokens,
authentication::matrix::MatrixSession,
encryption::{BackupDownloadStrategy, EncryptionSettings},
ruma::{OwnedDeviceId, OwnedUserId},
};
use swarm_matrix_client::WhoamiSession;
use tokio::fs;
use crate::credential::{Passphrase, Token};
/// Sentinel returned when `build_and_restore` detects that the token is
/// permanently invalid (`M_UNKNOWN_TOKEN`). The token has already been
/// discarded ([`Token::discard_stale`]). Retrying the same token at once
/// won't help — the account needs a new one, which the daemon's periodic
/// retry of a down secondary picks up.
///
/// Distinct from the general `anyhow::Error` path so callers can use
/// `err.downcast_ref::<PermanentBringUpError>()` to distinguish "retry
/// won't help" from a transient network/DNS/5xx failure.
#[derive(Debug)]
pub struct PermanentBringUpError(pub String);
impl std::fmt::Display for PermanentBringUpError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for PermanentBringUpError {}
/// Build + restore a matrix-sdk `Client` for the per-agent bearer
/// `token`. The Client points at `homeserver`, persists its sqlite cache
/// under `state_dir`, and is ready for sync once returned.
///
/// Steps:
/// 1. A `swarm-matrix-client` whoami call with
/// the bearer — this gives us back the matrix `user_id` +
/// `device_id` (the registration response had them but only the
/// token was kept; whoami is the cheapest recovery path
/// and avoids matrix-sdk's circular requirement of needing a
/// session to call whoami).
/// 2. Build the real Client with the sqlite store + `restore_session`
/// using a synthetic `MatrixSession`.
///
/// # Errors
///
/// Returns an error if the `whoami` request fails, or if the matrix-sdk
/// client fails to build or restore the session.
pub async fn build_and_restore(
homeserver: &str,
token: &Token,
state_dir: &Path,
is_primary: bool,
) -> Result<Client> {
let Some((user_id, device_id)) = whoami(homeserver, token.expose()).await? else {
// Homeserver rejected our token — stale session after a homeserver
// state wipe or token expiry. Discard it so we don't loop on it
// (which for a store-held credential is a log line and nothing
// else — see `Token::discard_stale`). The rest of the handling
// depends on whether this is the primary (hive-internal) account
// or a secondary one, because a single bad secondary token must
// NOT take down the whole daemon (and with it every healthy
// account).
token.discard_stale().await;
if is_primary {
tracing::warn!(
credential = %token.origin(),
"matrix token rejected (M_UNKNOWN_TOKEN); discarding stale token"
);
// Primary: also drop the matrix-sdk sqlite state keyed to the
// now-invalid session, then exit 0; the swarm's backfill
// re-mints the token and systemd restarts us once it
// exists (store re-check timer, or the path watcher). Exit 0 (not
// Err) keeps systemd's Restart=on-failure from looping; the
// call site is before any tasks are spawned so there's
// nothing to clean up.
if let Err(re) = fs::remove_dir_all(state_dir).await {
tracing::warn!(
path = %state_dir.display(),
err = %re,
"failed to remove sdk state dir; next startup may fail with stale state"
);
}
std::process::exit(0);
}
// Secondary: a file token is gone, a store token is untouched; the
// daemon tries the account again later and reads the token afresh.
// The sdk state stays in place in case the operator re-provisions a
// token for the same device.
// Return a PermanentBringUpError so the caller can distinguish
// "retrying now won't help" from a transient network/DNS failure.
// Every try of a down secondary lands here, so the warning is the
// caller's, once per outage.
tracing::debug!(
credential = %token.origin(),
"matrix token rejected (M_UNKNOWN_TOKEN); discarded stale token"
);
return Err(anyhow::Error::new(PermanentBringUpError(
"matrix token rejected (M_UNKNOWN_TOKEN); discarded stale token, leaving account down"
.into(),
)));
};
fs::create_dir_all(state_dir)
.await
.with_context(|| format!("mkdir matrix state dir {}", state_dir.display()))?;
let client = Client::builder()
.homeserver_url(homeserver)
.sqlite_store(state_dir, None)
.with_encryption_settings(EncryptionSettings {
backup_download_strategy: BackupDownloadStrategy::OneShot,
..EncryptionSettings::default()
})
.build()
.await
.with_context(|| format!("build matrix client for {homeserver}"))?;
let session = MatrixSession {
meta: SessionMeta { user_id, device_id },
tokens: SessionTokens {
access_token: token.expose().to_owned(),
refresh_token: None,
},
};
client
.restore_session(session)
.await
.context("restore matrix session")?;
tracing::info!(
user = %client.user_id().map(ToString::to_string).unwrap_or_default(),
device = %client.device_id().map(ToString::to_string).unwrap_or_default(),
"matrix session restored"
);
Ok(client)
}
/// Whoami probe — used at startup to recover the `user_id` + `device_id` the
/// registration response carried but hive-c0re didn't persist alongside the
/// access token. Cheaper than teaching the hive-c0re side to persist them,
/// plus matches what a fresh deployment with a hand-rolled token
/// (`HIVE_MATRIX_TOKEN_FILE` pointing somewhere unexpected) needs anyway.
///
/// `None` is the homeserver's `M_UNKNOWN_TOKEN` verdict on the token.
async fn whoami(homeserver: &str, token: &str) -> Result<Option<(OwnedUserId, OwnedDeviceId)>> {
let http = swarm_matrix_client::client()?;
let (user_id, device_id) =
match swarm_matrix_client::whoami_session(&http, homeserver, token).await? {
WhoamiSession::UnknownToken => return Ok(None),
WhoamiSession::Session { user_id, device_id } => (user_id, device_id),
};
let parsed: OwnedUserId = user_id
.parse()
.with_context(|| format!("invalid user_id in whoami: {user_id}"))?;
let device_id = device_id.ok_or_else(|| anyhow!("whoami response missing device_id"))?;
Ok(Some((parsed, device_id.into())))
}
/// Best-effort: bootstrap this account's cross-signing identity so its
/// device stops showing as "unverified" in other users' Matrix clients.
///
/// `bootstrap_cross_signing_if_needed(None)` is idempotent — it queries the
/// account's existing keys and only uploads the master / self-signing /
/// user-signing key set when no cross-signing identity exists yet. That
/// upload also self-signs the current device, which is what actually clears
/// the unverified warning. On a restart with the identity already present
/// it's a no-op.
///
/// The `None` is the UIAA auth: the in-hive tuwunel accepts the initial
/// `device_signing/upload` on the bearer token alone (no interactive
/// re-auth). If a homeserver required UIAA the call returns a `UiaaResponse`
/// error we can't satisfy with a bare bearer token — so this is best-effort:
/// the failure is logged and swallowed, leaving the session no worse off than
/// before. Never breaks account bring-up.
///
/// This only signs the *current* device against a freshly-bootstrapped
/// identity. A device under an already existing identity is a no-op here; it
/// is signed by [`ensure_recovery`]'s `recover`, which imports the
/// self-signing key from secret storage, and stays unverified for an account
/// with no recovery passphrase.
pub async fn ensure_cross_signing(client: &Client, account: &str) {
match client
.encryption()
.bootstrap_cross_signing_if_needed(None)
.await
{
Ok(()) => {
tracing::info!(
account,
"matrix cross-signing ensured (device self-verified)"
);
}
Err(e) => {
tracing::warn!(
account,
error = %format!("{e:#}"),
"matrix cross-signing bootstrap failed; session stays unverified (non-fatal)"
);
}
}
}
/// What [`ensure_recovery`] did.
enum Backup {
/// This device was already connected to the account's backup.
AlreadyOn,
/// The server had no backup; one was created, with secret storage that
/// `passphrase` opens.
Enabled,
/// The server had a backup; the keys secret storage holds for it, and
/// for cross-signing if it holds those, were imported.
Restored,
}
/// Best-effort: turn on server-side key backup for this account, or recover
/// from the backup it already has, with the account's recovery `passphrase`.
///
/// Run after [`ensure_cross_signing`], so a newly bootstrapped identity's
/// private keys are local when `enable` exports them into secret storage, and
/// an existing identity is left for `recover` to import. After a recover the
/// room keys download in the background (`BackupDownloadStrategy::OneShot`).
///
/// Every failure is logged and swallowed, as in [`ensure_cross_signing`]: the
/// account comes up as it would have without backup. That includes secret
/// storage this passphrase does not open, which is never replaced: with a
/// backup, `recover` fails; without one, this refuses.
pub async fn ensure_recovery(client: &Client, account: &str, passphrase: &Passphrase) {
match recover_or_enable(client, passphrase.expose()).await {
Ok(Backup::AlreadyOn) => {
tracing::debug!(account, "matrix key backup already on for this device");
}
Ok(Backup::Enabled) => tracing::info!(
account,
passphrase = %passphrase.path(),
"matrix recovery enabled: server-side key backup created"
),
Ok(Backup::Restored) => tracing::info!(
account,
passphrase = %passphrase.path(),
"matrix recovery: recovered from the server-side key backup; room keys download in the background"
),
Err(e) => tracing::warn!(
account,
passphrase = %passphrase.path(),
error = %format!("{e:#}"),
"matrix recovery failed; key backup stays off (non-fatal)"
),
}
}
/// The homeserver and device calls [`recover_or_enable`] decides from, so a
/// test can stand in for matrix-sdk.
trait KeyBackup {
/// Wait for the e2ee tasks a restored client starts in the background.
async fn settle(&self);
/// Whether this device is already connected to the account's backup.
async fn on_this_device(&self) -> bool;
/// Whether the homeserver holds a key backup version.
async fn backup_on_server(&self) -> Result<bool>;
/// Whether the account has a default secret-storage key.
async fn secret_storage_on_server(&self) -> Result<bool>;
/// Open secret storage with `passphrase` and import what it holds.
async fn recover(&self, passphrase: &str) -> Result<()>;
/// Create a backup and new secret storage that `passphrase` opens.
async fn enable(&self, passphrase: &str) -> Result<()>;
}
impl KeyBackup for Client {
async fn settle(&self) {
self.encryption().wait_for_e2ee_initialization_tasks().await;
}
async fn on_this_device(&self) -> bool {
self.encryption().backups().are_enabled().await
}
async fn backup_on_server(&self) -> Result<bool> {
Ok(self.encryption().backups().fetch_exists_on_server().await?)
}
async fn secret_storage_on_server(&self) -> Result<bool> {
Ok(self.encryption().secret_storage().is_enabled().await?)
}
async fn recover(&self, passphrase: &str) -> Result<()> {
Ok(self.encryption().recovery().recover(passphrase).await?)
}
async fn enable(&self, passphrase: &str) -> Result<()> {
// The recovery key `enable` returns is not kept: the passphrase opens
// the same secret storage.
self.encryption()
.recovery()
.enable()
.with_passphrase(passphrase)
.await?;
Ok(())
}
}
/// The body of [`ensure_recovery`], with its outcome returned rather than
/// logged.
///
/// ⚠️ `enable` makes a new secret-storage key and sets it as the account's
/// default, whatever default was there. With no backup but secret storage
/// present, this refuses rather than enable: that default key is someone
/// else's, and replacing it locks their recovery key out.
async fn recover_or_enable(account: &impl KeyBackup, passphrase: &str) -> Result<Backup> {
// A device restored from its own store reconnects to the backup in a
// background task; asking before it finishes says no.
account.settle().await;
if account.on_this_device().await {
return Ok(Backup::AlreadyOn);
}
if account
.backup_on_server()
.await
.context("asking the homeserver for the account's key backup")?
{
account
.recover(passphrase)
.await
.context("recovering from the key backup")?;
return Ok(Backup::Restored);
}
if account
.secret_storage_on_server()
.await
.context("asking the homeserver for the account's secret storage")?
{
bail!(
"the account has secret storage but no key backup; not enabling, \
which would replace a default secret-storage key this daemon did not create"
);
}
account
.enable(passphrase)
.await
.context("enabling key backup")?;
Ok(Backup::Enabled)
}
/// Best-effort: set this account's matrix avatar from the rasterized icon
/// PNG. The path comes from `HIVE_ICON_PNG` (a nix-built derivation the
/// harness forwards into the daemon env; absent when no icon resolves).
/// Idempotent via a per-account `avatar-icon-hash` file in the account's
/// sdk `state_dir` — a re-upload only happens when the icon bytes change,
/// so daemon restarts don't re-spam the homeserver.
///
/// This replaces the old `matrix-avatar-sync` systemd curl oneshot: the
/// daemon already holds an authenticated `Client` pointed at the correct
/// homeserver, so it uploads over the live connection — no hardcoded URL,
/// no token re-read, no token-file globbing. Any failure is logged and
/// swallowed: avatar trouble must never break account bring-up or sync.
pub async fn sync_avatar(client: &Client, state_dir: &Path, account: &str) {
let Ok(png_path) = std::env::var("HIVE_ICON_PNG") else {
return; // no icon forwarded → nothing to sync
};
let bytes = match fs::read(&png_path).await {
Ok(b) => b,
Err(e) => {
tracing::warn!(account, path = %png_path, error = %e, "matrix avatar: icon PNG unreadable; skipping");
return;
}
};
// FNV-1a content hash — enough to answer "did the icon change?" and
// deterministic across Rust/std versions (unlike `DefaultHasher`, whose
// output may change between toolchains, spuriously mismatching the
// persisted hash and re-uploading the same avatar). No crypto strength
// needed here, so no dependency on a hashing crate.
let hash = {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for &b in &bytes {
h ^= u64::from(b);
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
format!("{h:016x}")
};
let hash_file = state_dir.join("avatar-icon-hash");
if let Ok(prev) = fs::read_to_string(&hash_file).await
&& prev.trim() == hash
{
tracing::debug!(account, "matrix avatar: icon unchanged; skipping upload");
return;
}
match client
.account()
.upload_avatar(&mime::IMAGE_PNG, bytes)
.await
{
Ok(mxc) => {
tracing::info!(account, mxc = %mxc, "matrix avatar set");
if let Err(e) = fs::write(&hash_file, &hash).await {
tracing::warn!(account, error = %e, "matrix avatar: set ok but hash write failed (re-uploads next start)");
}
}
Err(e) => {
tracing::warn!(account, error = %format!("{e:#}"), "matrix avatar: upload failed; skipping (non-fatal)");
}
}
}
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use super::*;
/// An account as the homeserver and this device describe it, recording
/// which of the two writes [`recover_or_enable`] made.
#[derive(Default)]
struct Fake {
on_device: bool,
backup: bool,
secret_storage: bool,
calls: Mutex<Vec<&'static str>>,
}
impl KeyBackup for Fake {
async fn settle(&self) {}
async fn on_this_device(&self) -> bool {
self.on_device
}
async fn backup_on_server(&self) -> Result<bool> {
Ok(self.backup)
}
async fn secret_storage_on_server(&self) -> Result<bool> {
Ok(self.secret_storage)
}
async fn recover(&self, _: &str) -> Result<()> {
self.calls.lock().unwrap().push("recover");
Ok(())
}
async fn enable(&self, _: &str) -> Result<()> {
self.calls.lock().unwrap().push("enable");
Ok(())
}
}
fn calls(fake: &Fake) -> Vec<&'static str> {
fake.calls.lock().unwrap().clone()
}
#[tokio::test]
async fn secret_storage_without_a_backup_is_not_enabled_over() {
// `enable` would replace the account's default secret-storage key,
// which another client's recovery key opens.
let fake = Fake {
secret_storage: true,
..Fake::default()
};
assert!(recover_or_enable(&fake, "p").await.is_err());
assert!(calls(&fake).is_empty(), "{:?}", calls(&fake));
}
#[tokio::test]
async fn an_account_with_nothing_on_the_server_is_enabled() {
// Control for the test above: the guard is the only thing stopping
// `enable` there.
let fake = Fake::default();
assert!(matches!(
recover_or_enable(&fake, "p").await.unwrap(),
Backup::Enabled
));
assert_eq!(calls(&fake), ["enable"]);
}
#[tokio::test]
async fn a_backup_on_the_server_is_recovered_not_enabled() {
let fake = Fake {
backup: true,
secret_storage: true,
..Fake::default()
};
assert!(matches!(
recover_or_enable(&fake, "p").await.unwrap(),
Backup::Restored
));
assert_eq!(calls(&fake), ["recover"]);
}
#[tokio::test]
async fn a_device_already_on_the_backup_writes_nothing() {
let fake = Fake {
on_device: true,
backup: true,
secret_storage: true,
..Fake::default()
};
assert!(matches!(
recover_or_enable(&fake, "p").await.unwrap(),
Backup::AlreadyOn
));
assert!(calls(&fake).is_empty());
}
/// Serve `status` and `body` for every request on loopback; returns the
/// base URL.
async fn stub_homeserver(status: u16, body: &'static str) -> String {
let app = axum::Router::new().fallback(move || async move {
(
axum::http::StatusCode::from_u16(status).expect("a status"),
[(axum::http::header::CONTENT_TYPE, "application/json")],
body,
)
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind");
let base = format!("http://{}", listener.local_addr().expect("local addr"));
tokio::spawn(async move { axum::serve(listener, app).await });
base
}
#[tokio::test]
async fn a_whoami_refusal_does_not_quote_the_response_body() {
let base =
stub_homeserver(502, r#"{"errcode":"M_UNKNOWN","error":"free-form-marker"}"#).await;
let e = whoami(&base, "tok").await.expect_err("refused");
let message = format!("{e:#}");
assert!(message.contains("502"), "{message}");
assert!(!message.contains("free-form-marker"), "{message}");
}
#[tokio::test]
async fn whoami_reads_both_ids_and_reports_an_unknown_token_as_none() {
let base = stub_homeserver(200, r#"{"user_id":"@a:x","device_id":"DEV"}"#).await;
let (user, device) = whoami(&base, "tok").await.expect("live").expect("ids");
assert_eq!((user.as_str(), device.as_str()), ("@a:x", "DEV"));
let base = stub_homeserver(401, r#"{"errcode":"M_UNKNOWN_TOKEN"}"#).await;
assert!(whoami(&base, "tok").await.expect("a verdict").is_none());
}
}