hyperhive/hive-c0re/src/matrix.rs

1147 lines
46 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

//! Optional matrix-tuwunel wiring: shared registration token (host) +
//! per-agent UIAA registration → `<agent-state>/matrix-token`. No-op
//! when the `hive-matrix` container isn't running, so operators who
//! haven't flipped `hyperhive.matrix.enable = true` pay nothing.
//!
//! See `docs/matrix.md::Provisioning flow (registration token)` for
//! the full UIAA round-trip, token-file shape, and host/container
//! bind-mount layout.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use reqwest::StatusCode;
use crate::coordinator::Coordinator;
/// nspawn container name for the matrix homeserver — mirrors
/// `hive-forge` and matches the bare-name allow-list the lifecycle
/// scanner skips over.
const MATRIX_CONTAINER: &str = "hive-matrix";
/// Local-host URL of the tuwunel client-server API. Shares the host
/// netns so `localhost:<port>` resolves both from the daemon and from
/// inside any sub-agent container.
const MATRIX_HTTP: &str = "http://localhost:8008";
/// Host path of the matrix registration token. Must match
/// `hyperhive.matrix.registrationTokenFile` in `nix/modules/hive-matrix.nix`
/// (same path is bind-mounted read-only into the tuwunel container so
/// the homeserver can read it via `registration_token_file`).
const REGISTER_TOKEN_PATH: &str = "/var/lib/hyperhive/matrix-register-token";
/// Length (bytes) of the random registration token. 32 raw bytes ⇒
/// 64-char hex string; comfortable for a long-lived shared secret.
const REGISTER_TOKEN_BYTES: usize = 32;
/// HTTP timeout for registration round-trips. UIAA is two POSTs; even
/// the slow path should finish well inside this budget.
const HTTP_TIMEOUT_SECS: u64 = 10;
/// Length (bytes) of the throwaway per-agent matrix password. Random
/// 32-byte hex — agents never log in with the password (they
/// authenticate by `access_token`), so it's protocol overhead. We
/// store it nowhere.
const PASSWORD_BYTES: usize = 32;
/// Matrix localpart for the hive system admin account. Registered
/// before any agent account in [`ensure_all`] so it becomes the first
/// user on the homeserver — Conduit/tuwunel grants admin rights to the
/// first registered user automatically. Not an agent; has no state dir.
pub const HIVE_ADMIN_LOCALPART: &str = "hive";
/// Host path for the hive admin matrix access token. Outside every
/// purgeable path — not deleted by `destroy --purge` on any agent.
#[must_use]
pub fn admin_token_path() -> PathBuf {
PathBuf::from("/var/lib/hyperhive/matrix-admin-token")
}
/// Token file inside the agent's bind-mounted state dir (visible as
/// `/state/matrix-token` from inside the container).
fn token_path(name: &str) -> PathBuf {
Coordinator::agent_notes_dir(name).join("matrix-token")
}
/// Password file for the agent's matrix account. Stored OUTSIDE the
/// purgeable `agent_state_root` tree so it survives `destroy --purge`
/// and allows re-login recovery when the same agent name is re-spawned.
///
/// Path: `/var/lib/hyperhive/matrix-creds/<name>-password`
///
/// The token file lives inside the agent's bind-mounted state dir (under
/// `agent_notes_dir`) so the agent container can read it; the password
/// file is host-side only (agents never log in by password — they use
/// the access token exclusively) and belongs with other hive-c0re
/// credential state, not inside the purgeable per-agent tree.
fn password_path(name: &str) -> PathBuf {
PathBuf::from("/var/lib/hyperhive/matrix-creds").join(format!("{name}-password"))
}
/// Legacy password path (inside the old purgeable `agent_notes_dir`).
/// Used only during the one-time migration in [`ensure_user_for`] to
/// move credentials from old deployments to the new location. Safe to
/// call after `destroy --purge` — the path will simply not exist and
/// the migration is a no-op.
fn legacy_password_path(name: &str) -> PathBuf {
Coordinator::agent_notes_dir(name).join("matrix-password")
}
/// Host path where the hive Matrix Space room ID is persisted.
/// Outside every purgeable path — not deleted by `destroy --purge`.
#[must_use]
pub fn hive_space_room_id_path() -> PathBuf {
PathBuf::from("/var/lib/hyperhive/matrix-space-room-id")
}
/// Probe whether `hive-matrix` exists as a nixos-container. Cheap —
/// `nixos-container list` is just a directory scan in /etc. Same shape
/// as `forge::is_present` — routed through hive-priv since
/// `nixos-container` needs root and hive-c0re runs unprivileged.
pub async fn is_present() -> bool {
let Ok(stdout) = crate::priv_client::list_containers().await else {
return false;
};
stdout.lines().any(|l| l.trim() == MATRIX_CONTAINER)
}
/// Read `n` cryptographic-quality bytes from `/dev/urandom` and return
/// them hex-encoded. Avoids pulling a workspace `rand` dep just for
/// 32 bytes of randomness; the kernel's CSPRNG is more than enough for
/// a long-lived shared secret on the same host.
fn random_hex(n: usize) -> Result<String> {
use std::io::Read;
let mut buf = vec![0_u8; n];
let mut f = std::fs::File::open("/dev/urandom").context("open /dev/urandom")?;
f.read_exact(&mut buf).context("read /dev/urandom")?;
let mut hex = String::with_capacity(n * 2);
for b in &buf {
use std::fmt::Write as _;
write!(hex, "{b:02x}").ok();
}
Ok(hex)
}
/// Ensure the registration token file exists; returns its contents.
/// Generates a fresh 32-byte hex token on first call (mode 0600,
/// root-only), then re-reads on subsequent calls. The hive-matrix
/// nixos module bind-mounts this file read-only into the tuwunel
/// container so the homeserver can authenticate registration requests
/// against the same secret hive-c0re holds.
pub fn ensure_register_token() -> Result<String> {
use std::os::unix::fs::PermissionsExt;
let path = Path::new(REGISTER_TOKEN_PATH);
if let Ok(existing) = std::fs::read_to_string(path) {
let trimmed = existing.trim().to_owned();
if !trimmed.is_empty() {
return Ok(trimmed);
}
}
let token = random_hex(REGISTER_TOKEN_BYTES)?;
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).ok();
}
std::fs::write(path, format!("{token}\n"))
.with_context(|| format!("write registration token to {}", path.display()))?;
let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));
tracing::info!(path = %path.display(), "matrix: generated registration token");
Ok(token)
}
/// Build the localpart of a matrix user id for `agent`. Matrix
/// usernames are 1-255 chars from the `[a-z0-9._=-/]` alphabet; agent
/// names already conform (hyperhive enforces a strict subset), so no
/// escaping is needed at the boundary.
fn user_localpart(agent: &str) -> &str {
agent
}
/// Send a single registration POST and parse the response. Returns
/// `Ok((status, body))` on any successful HTTP round-trip (including
/// the expected 401 from the first UIAA leg); errors only on transport
/// failure. Body is the parsed JSON value — UIAA passes session +
/// flow state via JSON, never via headers.
async fn register_post(
client: &reqwest::Client,
body: &serde_json::Value,
) -> Result<(StatusCode, serde_json::Value)> {
let url = format!("{MATRIX_HTTP}/_matrix/client/v3/register");
let resp = client
.post(&url)
.json(body)
.send()
.await
.context("matrix: POST /register")?;
let status = resp.status();
let json = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse /register response")?;
Ok((status, json))
}
/// Generate a throwaway random password for matrix UIAA registration.
/// `PASSWORD_BYTES` raw bytes ⇒ 64-char hex string. Agents authenticate
/// by `access_token` so the password is protocol overhead we never
/// persist; the operator path in `hivectl` lets the caller supply a
/// real password instead so they can log into a matrix web client
/// (`m.login.password`).
pub fn random_password() -> Result<String> {
random_hex(PASSWORD_BYTES)
}
/// Run the matrix-spec UIAA flow to register `agent` with the given
/// `password` and return the resulting access token. Two round-trips:
/// first POST elicits the 401 + session id, second POST supplies the
/// registration token in the `auth` block. If the homeserver returns
/// 200 on the first POST (no flow stages required — `allow_registration`
/// with no token), we take the access token directly.
///
/// Caller picks the password: agents use [`random_password`] (throwaway
/// — they auth by `access_token`), operators on the `hivectl` path
/// supply their own so they can log into matrix web clients.
async fn register_user(
client: &reqwest::Client,
agent: &str,
register_token: &str,
password: &str,
) -> Result<String> {
let localpart = user_localpart(agent);
let initial = serde_json::json!({
"username": localpart,
"password": password,
// device_id stays stable across re-runs of ensure_user_for so
// a re-mint doesn't strand orphan devices in tuwunel.
"device_id": format!("hyperhive-{agent}"),
"initial_device_display_name": format!("hyperhive ({agent})"),
"inhibit_login": false,
});
let (status, body) = register_post(client, &initial).await?;
// Some servers accept the first POST when allow_registration is
// open. Take the access_token + we're done.
if status.is_success() {
return extract_access_token(&body);
}
if status != StatusCode::UNAUTHORIZED {
anyhow::bail!("matrix: /register first leg HTTP {status}, body: {body}");
}
let session = body["session"]
.as_str()
.with_context(|| format!("matrix: missing UIAA session in 401, body: {body}"))?;
let authed = serde_json::json!({
"username": localpart,
"password": password,
"device_id": format!("hyperhive-{agent}"),
"initial_device_display_name": format!("hyperhive ({agent})"),
"inhibit_login": false,
"auth": {
"type": "m.login.registration_token",
"token": register_token,
"session": session,
},
});
let (status, body) = register_post(client, &authed).await?;
if !status.is_success() {
anyhow::bail!("matrix: /register auth leg HTTP {status}, body: {body}");
}
extract_access_token(&body)
}
/// Pull `access_token` out of a successful /register response.
fn extract_access_token(body: &serde_json::Value) -> Result<String> {
body["access_token"]
.as_str()
.map(str::to_owned)
.with_context(|| format!("matrix: missing access_token in response: {body}"))
}
/// Login with `m.login.password` and return the access token. Fallback
/// for when registration fails with `M_USER_IN_USE` — the account
/// already exists in the homeserver but the token file was lost. Fails
/// if the stored password no longer matches (e.g. homeserver wiped),
/// in which case manual recovery via `hivectl matrix create-user` is
/// required.
async fn login_user(client: &reqwest::Client, agent: &str, password: &str) -> Result<String> {
let url = format!("{MATRIX_HTTP}/_matrix/client/v3/login");
let body = serde_json::json!({
"type": "m.login.password",
"identifier": {
"type": "m.id.user",
"user": user_localpart(agent),
},
"password": password,
"device_id": format!("hyperhive-{agent}"),
"initial_device_display_name": format!("hyperhive ({agent})"),
});
let resp = client
.post(&url)
.json(&body)
.send()
.await
.context("matrix: POST /login")?;
let status = resp.status();
let json = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse /login response")?;
if !status.is_success() {
anyhow::bail!("matrix: /login HTTP {status} for agent {agent}, body: {json}");
}
extract_access_token(&json)
}
/// Auto-recovery helper: reset a user's matrix password via the admin API
/// when the locally stored password is missing. Requires a valid hive admin
/// token at [`admin_token_path()`]. Returns the new password (already
/// persisted to [`password_path(name)`]) on success.
///
/// Called by [`ensure_user_for`] when registration returns `M_USER_IN_USE`
/// but the password file is absent — covers the case where agent state dirs
/// were wiped but the homeserver still has the accounts.
async fn auto_reset_password(client: &reqwest::Client, name: &str) -> anyhow::Result<String> {
let admin_token = read_admin_token()
.context("matrix: admin token unavailable for auto-recovery; provision hive admin first")?;
let server_name = discover_server_name(client)
.await
.context("matrix: discover_server_name for auto-recovery")?;
let effective_password =
reset_user_password(client, &admin_token, name, &server_name)
.await
.with_context(|| {
format!("matrix: admin-room password reset for {name} (auto-recovery)")
})?;
tracing::info!(%name, "matrix: auto-recovered password via admin-room reset");
Ok(effective_password)
}
// ---------------------------------------------------------------------------
// Admin-room fallback for password reset
// ---------------------------------------------------------------------------
/// Percent-encode a matrix room ID for use in a URL path segment.
/// Only `:` needs encoding; `!` and alphanumerics are path-safe.
fn encode_room_id_for_url(room_id: &str) -> String {
room_id.replace(':', "%3A")
}
/// Look up the room ID for the `#admins:<server>` alias.
async fn discover_admin_room_id(
client: &reqwest::Client,
admin_token: &str,
server_name: &str,
) -> Result<String> {
// #admins:server → %23admins%3A<server>
let encoded_alias = format!("%23admins%3A{server_name}");
let url = format!("{MATRIX_HTTP}/_matrix/client/v3/directory/room/{encoded_alias}");
let resp = client
.get(&url)
.bearer_auth(admin_token)
.send()
.await
.context("matrix: GET admin room alias")?;
let status = resp.status();
let json = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse admin room alias response")?;
if !status.is_success() {
anyhow::bail!("matrix: admin room alias lookup failed: HTTP {status}, body: {json}");
}
json["room_id"]
.as_str()
.map(|s| s.to_owned())
.ok_or_else(|| {
anyhow::anyhow!("matrix: admin room alias response missing room_id: {json}")
})
}
/// Try to parse the new password from an admin-room bot response.
/// Conduwuit/tuwunel responds with a message like:
/// "Done: Password of user @user:server has been reset. The new password is: <password>"
///
/// Uses [`str::to_ascii_lowercase`] for case folding — unlike `to_lowercase`,
/// ASCII lowercasing is guaranteed to produce a same-byte-length string, so the
/// byte offset from `find` is always a valid index into the original `bot_message`
/// and we never slice at a non-char boundary.
fn extract_new_password(bot_message: &str) -> Option<String> {
// ASCII lowercase: same byte length as the original, so positions from
// `lower.find(marker)` are valid byte indices into `bot_message`.
let lower = bot_message.to_ascii_lowercase();
for marker in &[
"new password is: ",
"new password is:",
"password is: ",
"password is:",
] {
if let Some(pos) = lower.find(marker) {
let rest = &bot_message[pos + marker.len()..];
let rest = rest.trim_start();
let end = rest
.find(|c: char| c.is_whitespace() || c == '\n')
.unwrap_or(rest.len());
let pw = rest[..end].trim();
if !pw.is_empty() {
return Some(pw.to_owned());
}
}
}
None
}
#[cfg(test)]
mod extract_new_password_tests {
use super::extract_new_password;
#[test]
fn tuwunel_style_response() {
let msg = "Done: Password of user @atlas:pr1ma.darkest.space has been reset. The new password is: abc123XYZ!";
assert_eq!(extract_new_password(msg).as_deref(), Some("abc123XYZ!"));
}
#[test]
fn case_insensitive_marker() {
let msg = "Password reset complete. New Password Is: S3cr3tP@ss";
assert_eq!(extract_new_password(msg).as_deref(), Some("S3cr3tP@ss"));
}
#[test]
fn marker_without_trailing_space() {
let msg = "new password is:hunter2";
assert_eq!(extract_new_password(msg).as_deref(), Some("hunter2"));
}
#[test]
fn shorter_marker_variant() {
let msg = "Your password is: Tr0ub4dor&3";
assert_eq!(extract_new_password(msg).as_deref(), Some("Tr0ub4dor&3"));
}
#[test]
fn no_match_returns_none() {
let msg = "Command not recognised. Please try again.";
assert_eq!(extract_new_password(msg), None);
}
#[test]
fn empty_after_marker_returns_none() {
let msg = "new password is: ";
assert_eq!(extract_new_password(msg), None);
}
#[test]
fn password_stops_at_whitespace() {
let msg = "New password is: abc123 (save it now)";
assert_eq!(extract_new_password(msg).as_deref(), Some("abc123"));
}
}
/// Send a command to the Matrix admin room and poll for a bot response.
///
/// 1. Anchors at the current end-token (`dir=b&limit=1`).
/// 2. Sends `command` as an `m.text` message from the @hive admin account.
/// 3. Polls up to 15 × 1 s for a bot message that satisfies `check`.
/// 4. Calls `check(body)` on each non-self message; returns `Ok(T)` on the
/// first `Some(T)`.
///
/// Generic over `T` so both password-returning and `()` callers share the loop.
async fn admin_room_send_and_poll<T>(
client: &reqwest::Client,
admin_token: &str,
server_name: &str,
room_url: &str,
command: &str,
check: impl Fn(&str) -> Option<T>,
) -> Result<T> {
// Pagination anchor — fetch the current tail so we only read events
// that arrive *after* our command, not old history.
let msgs_back_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/messages?dir=b&limit=1"
);
let msgs_back = client
.get(&msgs_back_url)
.bearer_auth(admin_token)
.send()
.await
.context("matrix: admin room messages (back)")?
.json::<serde_json::Value>()
.await
.context("matrix: parse admin room messages response")?;
let mut from_token = msgs_back["end"]
.as_str()
.ok_or_else(|| {
anyhow::anyhow!(
"matrix: admin room messages response missing 'end' token: {msgs_back}"
)
})?
.to_owned();
// Send the command.
let txn_id = random_hex(8)?;
let send_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/send/m.room.message/{txn_id}"
);
let send_resp = client
.put(&send_url)
.bearer_auth(admin_token)
.json(&serde_json::json!({"msgtype": "m.text", "body": command}))
.send()
.await
.context("matrix: PUT admin room message")?;
if !send_resp.status().is_success() {
let body = send_resp.json::<serde_json::Value>().await.unwrap_or_default();
anyhow::bail!("matrix: admin room send failed: {body}");
}
// Poll for bot response (up to 15 × 1 s). tuwunel can take several
// seconds to process admin-room commands during startup.
let own_user_id = format!("@{HIVE_ADMIN_LOCALPART}:{server_name}");
for _ in 0..15_u8 {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
let from_encoded = from_token.replace(':', "%3A").replace('+', "%2B");
let poll_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/messages?dir=f&from={from_encoded}&limit=10"
);
let poll_json = client
.get(&poll_url)
.bearer_auth(admin_token)
.send()
.await
.context("matrix: admin room poll")?
.json::<serde_json::Value>()
.await
.context("matrix: parse admin room poll response")?;
if let Some(events) = poll_json["chunk"].as_array() {
for event in events {
if event["type"].as_str() != Some("m.room.message") {
continue;
}
if event["sender"].as_str() == Some(own_user_id.as_str()) {
continue;
}
let body = event["content"]["body"].as_str().unwrap_or("");
if let Some(result) = check(body) {
return Ok(result);
}
}
}
if let Some(end) = poll_json["end"].as_str() {
from_token = end.to_owned();
}
}
anyhow::bail!(
"matrix: admin room command timed out after 15 seconds. \
Command: '{command}'. No matching bot response received."
)
}
/// Reset a user's password via the Matrix admin room (`#admins:<server>`).
/// Sends `!admin users reset-password @<localpart>:<server>` as @hive, polls for the bot's
/// response containing the new password.
///
/// Returns the new password; caller is responsible for persisting it.
async fn admin_room_reset_password(
client: &reqwest::Client,
admin_token: &str,
server_name: &str,
localpart: &str,
) -> Result<String> {
let room_id = discover_admin_room_id(client, admin_token, server_name).await?;
let room_url = encode_room_id_for_url(&room_id);
let command = format!("!admin users reset-password @{localpart}:{server_name}");
admin_room_send_and_poll(client, admin_token, server_name, &room_url, &command, extract_new_password)
.await
.with_context(|| {
format!(
"matrix: admin room reset-password for @{localpart}:{server_name}: \
no password response received within 15 seconds. \
Verify the admin room accepts '!admin users reset-password @user:server' commands."
)
})
}
/// Ensure `name` has a matrix user + token file on the local
/// homeserver. Skips provisioning entirely if the token file already
/// exists (treating a present token as proof the account is good).
/// To force re-provisioning, delete the token file.
///
/// When registration fails with `M_USER_IN_USE` (account exists in the
/// homeserver but the token file was deleted) this falls back to
/// `m.login.password` using the persisted `matrix-password` file. If
/// that file is also missing, recovery requires manual intervention:
/// `hivectl matrix create-user <name> --password <pw>`.
///
/// `client` is shared across the sweep so we build one reqwest
/// connection pool for all agents rather than one per call.
pub async fn ensure_user_for(
client: &reqwest::Client,
name: &str,
register_token: &str,
) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
let path = token_path(name);
if path.exists()
&& let Ok(existing) = std::fs::read_to_string(&path)
&& !existing.trim().is_empty()
{
tracing::debug!(%name, "matrix: token already present");
return Ok(());
}
// One-time migration: move the password from the old location inside
// agent_notes_dir (purgeable) to the new location outside it.
let new_pw_path = password_path(name);
let old_pw_path = legacy_password_path(name);
if !new_pw_path.exists() && old_pw_path.exists() {
if let Some(parent) = new_pw_path.parent() {
std::fs::create_dir_all(parent).ok();
}
if let Err(e) = std::fs::rename(&old_pw_path, &new_pw_path) {
// Rename across filesystems or read-only src — copy + delete.
if let Ok(content) = std::fs::read(&old_pw_path) {
if std::fs::write(&new_pw_path, &content).is_ok() {
let _ = std::fs::remove_file(&old_pw_path);
tracing::info!(%name, "matrix: migrated password file to non-purgeable location");
}
} else {
tracing::warn!(%name, rename_error = ?e, "matrix: password migration failed — could not read old path (old path stays)");
}
} else {
tracing::info!(%name, "matrix: migrated password file to non-purgeable location");
}
}
let password = random_password()?;
let access_token = match register_user(client, name, register_token, &password).await {
Ok(token) => {
// Successful registration — persist the password so we can
// fall back to login if the token file is deleted later.
let pw_path = password_path(name);
if let Some(parent) = pw_path.parent() {
std::fs::create_dir_all(parent).ok();
}
if let Err(e) = std::fs::write(&pw_path, format!("{password}\n")) {
tracing::warn!(%name, error = ?e, "matrix: failed to persist password (token still saved)");
} else {
let _ = std::fs::set_permissions(&pw_path, std::fs::Permissions::from_mode(0o600));
}
token
}
Err(reg_err) if reg_err.to_string().contains("M_USER_IN_USE") => {
// Account already exists — try to re-login with the stored password.
tracing::info!(%name, "matrix: user already exists, attempting login with stored password");
let pw_path = password_path(name);
let stored = match std::fs::read_to_string(&pw_path)
.ok()
.map(|s| s.trim().to_owned())
.filter(|s| !s.is_empty())
{
Some(pw) => pw,
None => {
// Password file missing — attempt auto-recovery via admin API.
// This covers the case where agent state dirs were wiped but the
// homeserver still has the accounts. Requires the hive admin
// token at /var/lib/hyperhive/matrix-admin-token.
tracing::info!(
%name,
"matrix: stored password missing, attempting admin-API auto-recovery"
);
match auto_reset_password(client, name).await {
Ok(new_pw) => new_pw,
Err(e) => {
anyhow::bail!(
"matrix: user {name} already exists but password is missing \
and admin auto-recovery failed ({e:#}) — run:\n\
hivectl matrix reset-password {name}\n\
hivectl matrix create-user {name}"
)
}
}
}
};
login_user(client, name, &stored).await.with_context(|| {
format!(
"matrix: login fallback for {name} failed — if homeserver was wiped, delete \
the matrix-password file and retry"
)
})?
}
Err(other) => return Err(other),
};
// Write the token via hive-priv (root helper): hive-c0re runs as the
// unprivileged `hive-core` user and cannot write to agent-owned state
// directories directly. hive-priv writes the file 0600 and chowns it
// to the agent user so it is readable from inside the container.
crate::priv_client::write_agent_matrix_token(name, &access_token)
.await
.with_context(|| format!("matrix: write matrix-token for {name} via hive-priv"))?;
tracing::info!(%name, "matrix: provisioned access token");
Ok(())
}
/// Register a matrix account for `name` with the supplied `password`
/// and return the freshly-minted access token. Unlike [`ensure_user_for`],
/// the token is **not** persisted to disk — the caller is responsible
/// for storing it. Used by `hivectl matrix create-user` for human
/// (non-agent) accounts so we don't create stray
/// `/var/lib/hyperhive/agents/<name>/` directories for users that
/// aren't agents. For operator accounts the caller passes a real
/// password so the operator can `m.login.password` into matrix web
/// clients afterwards; for headless agent re-provisioning the caller
/// can pass [`random_password`] to keep the existing throwaway
/// behaviour.
///
/// **Not idempotent** (unlike [`forge::provision_user_token`]): the
/// matrix UIAA `/register` endpoint returns `M_USER_IN_USE` (HTTP 400)
/// on second call for the same localpart. Callers re-running this for
/// a known-existing matrix user should expect a hard error from this
/// fn and route to a password-reset path instead.
pub async fn provision_user_token(
client: &reqwest::Client,
name: &str,
register_token: &str,
password: &str,
) -> Result<String> {
register_user(client, name, register_token, password).await
}
/// Per-agent matrix sync: ensure the agent has a matrix account + token.
/// All operations are idempotent; failures are logged as warnings but
/// don't abort the caller.
pub async fn sync_agent(client: &reqwest::Client, name: &str, register_token: &str) {
if let Err(e) = ensure_user_for(client, name, register_token).await {
tracing::warn!(%name, error = ?e, "matrix: ensure_user failed");
}
}
/// Standalone per-agent sync that handles its own setup: checks if
/// hive-matrix is present, reads the registration token, and builds
/// an HTTP client before delegating to [`sync_agent`]. Mirrors the
/// setup in [`ensure_all`] so the rebuild path and the startup sweep
/// stay equivalent. No-op when the matrix container is absent.
pub async fn sync_agent_standalone(name: &str) {
if !is_present().await {
return;
}
let register_token = match ensure_register_token() {
Ok(t) => t,
Err(e) => {
tracing::warn!(%name, error = ?e, "matrix: ensure_register_token failed");
return;
}
};
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.build()
{
Ok(c) => c,
Err(e) => {
tracing::warn!(%name, error = ?e, "matrix: build HTTP client failed");
return;
}
};
sync_agent(&client, name, &register_token).await;
}
/// Ensure the hive system admin matrix user exists and its token is
/// persisted at [`admin_token_path()`]. Must be called BEFORE
/// [`ensure_all`]'s agent loop so this account is the first to register
/// and becomes the homeserver admin automatically (Conduit/tuwunel:
/// first registered user = admin).
///
/// Idempotent — skips when the token file already exists and is
/// non-empty. Does NOT promote the account via API (that requires
/// admin rights which this fn bootstraps); on a fresh homeserver the
/// first-registered rule fires automatically; on an existing homeserver
/// the operator must promote the account once via `hivectl matrix
/// `hivectl matrix promote-user hive` or the conduit admin room.
pub async fn ensure_admin_user(client: &reqwest::Client, register_token: &str) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
let path = admin_token_path();
if path.exists()
&& let Ok(existing) = std::fs::read_to_string(&path)
&& !existing.trim().is_empty()
{
tracing::debug!("matrix: hive admin token already present");
return Ok(());
}
let password = random_password()?;
let access_token = match register_user(client, HIVE_ADMIN_LOCALPART, register_token, &password)
.await
{
Ok(token) => {
let pw_path = password_path(HIVE_ADMIN_LOCALPART);
if let Some(parent) = pw_path.parent() {
std::fs::create_dir_all(parent).ok();
}
if let Err(e) = std::fs::write(&pw_path, format!("{password}\n")) {
tracing::warn!(error = ?e, "matrix: failed to persist hive admin password");
} else {
let _ = std::fs::set_permissions(&pw_path, std::fs::Permissions::from_mode(0o600));
}
token
}
Err(reg_err) if reg_err.to_string().contains("M_USER_IN_USE") => {
tracing::info!("matrix: hive admin user already exists, re-logging in");
let pw_path = password_path(HIVE_ADMIN_LOCALPART);
let stored = std::fs::read_to_string(&pw_path)
.ok()
.map(|s| s.trim().to_owned())
.filter(|s| !s.is_empty())
.with_context(|| {
format!(
"matrix: hive admin user exists but password missing at {}\
manual recovery: reset password via admin API or conduit admin room",
pw_path.display()
)
})?;
login_user(client, HIVE_ADMIN_LOCALPART, &stored).await?
}
Err(other) => return Err(other),
};
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).ok();
}
std::fs::write(&path, format!("{access_token}\n"))
.with_context(|| format!("matrix: write hive admin token to {}", path.display()))?;
let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
tracing::info!(path = %path.display(), "matrix: provisioned hive admin token");
Ok(())
}
/// Promote a user to homeserver admin via the Matrix admin room
/// (`#admins:<server>`). Sends `!admin users make-user-admin @<localpart>:<server>` as
/// @hive, polls for the bot's "Done:" success response.
///
/// tuwunel 1.6.x does not implement `/_synapse/admin/v2/users`; all admin
/// operations go through the admin room.
pub async fn promote_user_to_admin(
client: &reqwest::Client,
admin_token: &str,
localpart: &str,
server_name: &str,
) -> Result<()> {
let room_id = discover_admin_room_id(client, admin_token, server_name).await?;
let room_url = encode_room_id_for_url(&room_id);
let command = format!("!admin users make-user-admin @{localpart}:{server_name}");
admin_room_send_and_poll(client, admin_token, server_name, &room_url, &command, |body| {
let lower = body.to_ascii_lowercase();
if lower.starts_with("done") || lower.contains("made") && lower.contains("admin") {
Some(())
} else {
None
}
})
.await
.with_context(|| {
format!(
"matrix: admin room make-user-admin for @{localpart}:{server_name}: \
no success response within 15 seconds. \
Verify the admin room accepts '!admin users make-user-admin @user:server' commands."
)
})
}
/// Reset a user's password via the Matrix admin room (`#admins:<server>`).
///
/// Sends `!admin users reset-password @<localpart>:<server>` to the admin room as @hive,
/// polls for the bot's response containing the new password, and persists
/// it to the non-purgeable creds path so [`ensure_user_for`] can re-login
/// on the next provisioning sweep.
///
/// Returns the new password for use in subsequent `login_user` calls.
pub async fn reset_user_password(
client: &reqwest::Client,
admin_token: &str,
localpart: &str,
server_name: &str,
) -> Result<String> {
let pw = admin_room_reset_password(client, admin_token, server_name, localpart)
.await
.with_context(|| {
format!("matrix: admin-room password reset for @{localpart}:{server_name}")
})?;
persist_password(localpart, &pw);
Ok(pw)
}
/// Persist the matrix password for `localpart` to the non-purgeable creds path.
fn persist_password(localpart: &str, password: &str) {
use std::os::unix::fs::PermissionsExt;
let pw_path = password_path(localpart);
if let Some(parent) = pw_path.parent() {
std::fs::create_dir_all(parent).ok();
}
if let Err(e) = std::fs::write(&pw_path, format!("{password}\n")) {
tracing::warn!(%localpart, error = ?e, "matrix: failed to persist reset password");
} else {
let _ = std::fs::set_permissions(&pw_path, std::fs::Permissions::from_mode(0o600));
}
}
/// Discover the matrix `server_name` from the running homeserver via
/// `GET /_matrix/key/v2/server` (unauthenticated federation key endpoint).
/// The response JSON always includes `"server_name"` per the matrix spec.
pub async fn discover_server_name(client: &reqwest::Client) -> Result<String> {
let url = format!("{MATRIX_HTTP}/_matrix/key/v2/server");
let resp = client
.get(&url)
.send()
.await
.context("matrix: GET /_matrix/key/v2/server")?;
let status = resp.status();
let body = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse /_matrix/key/v2/server response")?;
if !status.is_success() {
anyhow::bail!("matrix: /_matrix/key/v2/server returned HTTP {status}, body: {body}");
}
body["server_name"]
.as_str()
.map(str::to_owned)
.with_context(|| {
format!("matrix: /_matrix/key/v2/server response missing server_name field: {body}")
})
}
/// Read the hive admin access token from disk. Returns an error if it
/// is absent — callers should gate their admin-API calls on this.
pub fn read_admin_token() -> Result<String> {
let path = admin_token_path();
std::fs::read_to_string(&path)
.ok()
.map(|s| s.trim().to_owned())
.filter(|s| !s.is_empty())
.with_context(|| {
format!(
"hive admin matrix token not found at {}\
ensure hive-c0re has started at least once with matrix enabled \
(it provisions the admin account on boot)",
path.display()
)
})
}
/// Create the hive Matrix Space room using the admin account and persist
/// its room ID to [`hive_space_room_id_path()`]. Idempotent — returns
/// the stored room ID immediately if the file already exists.
///
/// The Space is a private `m.space` room owned by `@hive`. All agents
/// are invited after their accounts are provisioned in [`ensure_all`].
///
/// # Errors
///
/// Returns an error if the Matrix homeserver is unreachable, the
/// `createRoom` call fails, or the room-ID file cannot be written.
pub async fn ensure_hive_space(client: &reqwest::Client, admin_token: &str) -> Result<String> {
use std::os::unix::fs::PermissionsExt;
let path = hive_space_room_id_path();
if let Ok(existing) = std::fs::read_to_string(&path) {
let trimmed = existing.trim().to_owned();
if !trimmed.is_empty() {
tracing::debug!(room_id = %trimmed, "matrix: hive space already provisioned");
return Ok(trimmed);
}
}
let url = format!("{MATRIX_HTTP}/_matrix/client/v3/createRoom");
let body = serde_json::json!({
"name": "hive",
"creation_content": { "type": "m.space" },
"preset": "private_chat",
"visibility": "private",
});
let resp = client
.post(&url)
.bearer_auth(admin_token)
.json(&body)
.send()
.await
.context("matrix: POST /createRoom (hive space)")?;
let status = resp.status();
let json = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse /createRoom response")?;
if !status.is_success() {
anyhow::bail!("matrix: createRoom HTTP {status}, body: {json}");
}
let room_id = json["room_id"]
.as_str()
.with_context(|| format!("matrix: createRoom missing room_id: {json}"))?
.to_owned();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).ok();
}
std::fs::write(&path, format!("{room_id}\n"))
.with_context(|| format!("matrix: write space room_id to {}", path.display()))?;
let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
tracing::info!(%room_id, "matrix: created hive space");
Ok(room_id)
}
/// Invite `@{localpart}:{server_name}` to `room_id` using the admin
/// account. Idempotent — treats already-member responses as success.
async fn invite_to_room(
client: &reqwest::Client,
admin_token: &str,
room_id: &str,
localpart: &str,
server_name: &str,
) -> Result<()> {
// `:` must be percent-encoded in the room-id path segment; `!` is
// permitted in URL path characters per RFC 3986.
let encoded_room_id = room_id.replace(':', "%3A");
let url = format!("{MATRIX_HTTP}/_matrix/client/v3/rooms/{encoded_room_id}/invite");
let user_id = format!("@{localpart}:{server_name}");
let resp = client
.post(&url)
.bearer_auth(admin_token)
.json(&serde_json::json!({ "user_id": user_id }))
.send()
.await
.with_context(|| format!("matrix: POST /rooms/.../invite for {user_id}"))?;
let status = resp.status();
if status.is_success() {
tracing::debug!(%user_id, %room_id, "matrix: invited to hive space");
return Ok(());
}
// 403 with M_FORBIDDEN or M_BAD_STATE typically means the user is
// already a member or has a pending invite — both are fine.
if status == StatusCode::FORBIDDEN {
let body = resp.json::<serde_json::Value>().await.unwrap_or_default();
let errcode = body["errcode"].as_str().unwrap_or("");
if errcode == "M_FORBIDDEN" || errcode == "M_BAD_STATE" {
tracing::debug!(%user_id, %room_id, %errcode, "matrix: invite skipped (already member/invited)");
return Ok(());
}
anyhow::bail!("matrix: invite {user_id} to {room_id}: HTTP {status}, body: {body}");
}
let body = resp.json::<serde_json::Value>().await.unwrap_or_default();
anyhow::bail!("matrix: invite {user_id} to {room_id}: HTTP {status}, body: {body}")
}
/// Sweep every existing container (manager + sub-agents) and ensure
/// each has a matrix user + token on the local homeserver. Called once
/// at hive-c0re startup, alongside `forge::ensure_all`. No-op when the
/// hive-matrix container isn't running. Per-step failures are logged
/// but don't abort the sweep.
pub async fn ensure_all() {
if !is_present().await {
tracing::debug!("matrix: hive-matrix container absent, skipping user sweep");
return;
}
let register_token = match ensure_register_token() {
Ok(t) => t,
Err(e) => {
tracing::warn!(error = ?e, "matrix: ensure_register_token failed");
return;
}
};
// One HTTP client for the whole sweep — connection pool is
// reused across agents.
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.build()
{
Ok(c) => c,
Err(e) => {
tracing::warn!(error = ?e, "matrix: build HTTP client failed; skipping sweep");
return;
}
};
// Provision hive admin user FIRST so it's the first registered
// account on a fresh homeserver (Conduit/tuwunel makes the first
// registered user admin automatically).
if let Err(e) = ensure_admin_user(&client, &register_token).await {
tracing::warn!(error = ?e, "matrix: ensure_admin_user failed");
}
let Ok(containers) = crate::lifecycle::list().await else {
tracing::warn!("matrix: nixos-container list failed; skipping user sweep");
return;
};
let mut agent_names: Vec<String> = Vec::new();
for c in &containers {
let Some(name) = c.strip_prefix(crate::lifecycle::AGENT_PREFIX) else {
continue;
};
sync_agent(&client, name, &register_token).await;
agent_names.push(name.to_owned());
}
// Provision the hive Space and invite all agents (+ the admin account).
// server_name is needed to form full Matrix user IDs for invites.
let admin_token = match read_admin_token() {
Ok(t) => t,
Err(e) => {
tracing::warn!(error = ?e, "matrix: skipping hive space provisioning (no admin token)");
return;
}
};
let room_id = match ensure_hive_space(&client, &admin_token).await {
Ok(id) => id,
Err(e) => {
tracing::warn!(error = ?e, "matrix: ensure_hive_space failed");
return;
}
};
let server_name = match discover_server_name(&client).await {
Ok(s) => s,
Err(e) => {
tracing::warn!(error = ?e, "matrix: discover_server_name failed; skipping space invites");
return;
}
};
// Invite @hive admin first, then all agents.
if let Err(e) = invite_to_room(
&client,
&admin_token,
&room_id,
HIVE_ADMIN_LOCALPART,
&server_name,
)
.await
{
tracing::warn!(error = ?e, "matrix: invite @hive to space failed");
}
for name in &agent_names {
if let Err(e) = invite_to_room(&client, &admin_token, &room_id, name, &server_name).await {
tracing::warn!(%name, error = ?e, "matrix: invite agent to space failed");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn random_hex_is_well_formed_and_correct_length() {
let h = random_hex(16).expect("/dev/urandom readable");
assert_eq!(h.len(), 32);
assert!(h.chars().all(|c| c.is_ascii_hexdigit()));
}
#[test]
fn random_hex_two_calls_differ() {
// Sanity check — not a statistical claim, just guards
// against ever accidentally returning a constant.
let a = random_hex(16).expect("/dev/urandom readable");
let b = random_hex(16).expect("/dev/urandom readable");
assert_ne!(a, b);
}
#[test]
fn extract_access_token_pulls_from_success_body() {
let body = serde_json::json!({
"user_id": "@alice:matrix.example.org",
"access_token": "syt_abc123",
"device_id": "ABC",
});
assert_eq!(extract_access_token(&body).unwrap(), "syt_abc123");
}
#[test]
fn extract_access_token_errors_on_missing_field() {
let body = serde_json::json!({"user_id": "@alice:matrix.example.org"});
let err = extract_access_token(&body).unwrap_err();
assert!(err.to_string().contains("missing access_token"));
}
}