diff --git a/AGENTS.md b/AGENTS.md index aea39123..e6b9f51d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -152,9 +152,8 @@ Swarm-level: bearer on the metrics store's query API. - **`swarm-logs/`** — agent CLI: a LogsQL query in, log lines out. - **`swarm-metrics/`** — agent CLI: a MetricsQL query in, the store's JSON out. -- **`swarm-matrix-client/`** — client-server API calls against the swarm - homeserver, shared by `swarm-controller`, `swarm-matrix-ctl`, `hive-c0re` - and `hive-matrix-mcp`. +- **`swarm-matrix-client/`** — appservice calls against the swarm + homeserver, shared by `swarm-controller` and `swarm-matrix-ctl`. - **`swarm-matrix-ctl/`** — the one binary run inside `containers.hive-matrix`; new in-container jobs are verbs here, not crates. diff --git a/Cargo.lock b/Cargo.lock index 00b94f85..b6674478 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1796,7 +1796,6 @@ dependencies = [ "serde_json", "sha2 0.11.0", "strum", - "swarm-matrix-client", "swarm-queue-client", "swarm-secret-client", "tempfile", @@ -1946,7 +1945,6 @@ dependencies = [ "schemars", "serde", "serde_json", - "swarm-matrix-client", "swarm-secret-client", "tokio", "tracing", diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 578a267c..a5fe5e1b 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -61,7 +61,6 @@ swarm-queue-client = { workspace = true, features = [ "agent-permissions", "agent-archetype", ] } -swarm-matrix-client.workspace = true swarm-secret-client.workspace = true tokio.workspace = true tokio-stream.workspace = true diff --git a/hive-c0re/src/matrix.rs b/hive-c0re/src/matrix.rs index 50928906..7f4846ac 100644 --- a/hive-c0re/src/matrix.rs +++ b/hive-c0re/src/matrix.rs @@ -10,6 +10,7 @@ use std::path::PathBuf; use anyhow::{Context, Result}; +use reqwest::StatusCode; /// Client-server API base this daemon provisions against, from /// `HIVE_MATRIX_API_URL` (set by the hyperhive NixOS module from @@ -45,6 +46,8 @@ fn matrix_base() -> Result<&'static str> { this path should have been gated on matrix::is_present()", ) } +/// HTTP timeout for the sweep's homeserver round-trips. +const HTTP_TIMEOUT_SECS: u64 = 10; /// Matrix localpart this hive acts as. Not an agent; has no state dir. /// @@ -127,6 +130,12 @@ pub fn is_present() -> bool { matrix_http().is_some() } +/// 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") +} + /// Bring the hive's `@hive-:` sender token at [`sender_token_path()`] in /// line with the swarm secret store. /// @@ -309,43 +318,57 @@ async fn find_joined_room( want_space: bool, name: &str, ) -> Option { - let rooms = swarm_matrix_client::joined_rooms(client, base, sender_token) + let joined_url = format!("{base}/_matrix/client/v3/joined_rooms"); + let joined: serde_json::Value = client + .get(&joined_url) + .bearer_auth(sender_token) + .send() + .await + .ok()? + .json() .await .ok()?; - for room_id in rooms { + let rooms = joined["joined_rooms"].as_array()?; + for room in rooms { + let Some(room_id) = room.as_str() else { + continue; + }; + let encoded = encode_room_id_for_url(room_id); // m.room.create `type` says whether this is a Space. - let is_space = keyless_state(client, base, sender_token, &room_id, "m.room.create") + let create_url = format!("{base}/_matrix/client/v3/rooms/{encoded}/state/m.room.create/"); + let is_space = match client + .get(&create_url) + .bearer_auth(sender_token) + .send() .await - .is_some_and(|c| c["type"].as_str() == Some("m.space")); + { + Ok(r) if r.status().is_success() => r + .json::() + .await + .ok() + .is_some_and(|c| c["type"].as_str() == Some("m.space")), + _ => false, + }; if is_space != want_space { continue; } // …and named `name` (m.room.name `name`). - let name_matches = keyless_state(client, base, sender_token, &room_id, "m.room.name") - .await - .is_some_and(|n| n["name"].as_str() == Some(name)); + let name_url = format!("{base}/_matrix/client/v3/rooms/{encoded}/state/m.room.name/"); + let name_matches = match client.get(&name_url).bearer_auth(sender_token).send().await { + Ok(r) if r.status().is_success() => r + .json::() + .await + .ok() + .is_some_and(|n| n["name"].as_str() == Some(name)), + _ => false, + }; if name_matches { - return Some(room_id); + return Some(room_id.to_owned()); } } None } -/// The content of a state event with an empty state key, or `None` when the -/// room has none or the read failed. -async fn keyless_state( - client: &reqwest::Client, - base: &str, - sender_token: &str, - room_id: &str, - event_type: &str, -) -> Option { - swarm_matrix_client::room_state(client, base, sender_token, room_id, event_type, "") - .await - .ok() - .flatten() -} - /// Scan the admin account's joined rooms for the canonical hive Space: the /// `m.space` whose name is [`HIVE_SPACE_NAME`]. Returns the first match /// (deterministic per the homeserver's joined-rooms order) so a lost @@ -394,15 +417,32 @@ pub async fn ensure_hive_space(client: &reqwest::Client, sender_token: &str) -> } // 3. Create the space (plain hardcoded name, no alias). + let url = format!("{base}/_matrix/client/v3/createRoom"); let body = serde_json::json!({ "name": HIVE_SPACE_NAME, "creation_content": { "type": "m.space" }, "preset": "private_chat", "visibility": "private", }); - let room_id = swarm_matrix_client::create_room(client, base, sender_token, &body) + let resp = client + .post(&url) + .bearer_auth(sender_token) + .json(&body) + .send() .await - .context("matrix: create hive space")?; + .context("matrix: POST /createRoom (hive space)")?; + let status = resp.status(); + let json = resp + .json::() + .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(); persist_space_room_id(&room_id)?; tracing::info!(%room_id, "matrix: created hive space"); Ok(room_id) @@ -445,40 +485,43 @@ fn state_needs_write(current: Option<&serde_json::Value>, desired: &serde_json:: } /// Read the room's current content for one state event. `None` on any -/// failure — absent state, transport error, a refusal, or an unparseable body -/// are all "we don't know", and [`state_needs_write`] turns that into a write. +/// failure — absent state, transport error, or an unparseable body are +/// all "we don't know", and [`state_needs_write`] turns that into a write. /// -/// A failure other than absent state is logged at `warn`: **a read that breaks -/// while the write still works** makes the guard degrade to writing every -/// time, and the sweep resumes waking the hive with nothing else to show for -/// it. An absent state event (404) is the expected first-setup case. +/// One failure is louder than the rest, and it is the only one that hides: +/// **a non-404 HTTP failure means the read broke while the write still +/// works**, so the guard degrades to writing every time and the sweep +/// resumes waking the hive with nothing else to show for it. A transport +/// error needs no warning of its own — the PUT immediately after it fails +/// too, loudly — and a 404 is the expected first-setup case. async fn current_room_state( client: &reqwest::Client, - base: &str, sender_token: &str, - room_id: &str, - event_type: &str, - state_key: &str, + url: &str, ) -> Option { - match swarm_matrix_client::room_state( - client, - base, - sender_token, - room_id, - event_type, - state_key, - ) - .await - { - Ok(content) => content, + let resp = match client.get(url).bearer_auth(sender_token).send().await { + Ok(resp) => resp, Err(e) => { + tracing::debug!(error = ?e, url, "matrix: state read unreachable; writing"); + return None; + } + }; + let status = resp.status(); + if !status.is_success() { + if status != reqwest::StatusCode::NOT_FOUND { tracing::warn!( - error = %format!("{e:#}"), - %room_id, - event_type, + %status, + url, "matrix: state read failed while writes still work — the \ skip-if-unchanged guard is off and every sweep will re-emit" ); + } + return None; + } + match resp.json::().await { + Ok(body) => Some(body), + Err(e) => { + tracing::debug!(error = ?e, url, "matrix: state read body unparseable; writing"); None } } @@ -506,8 +549,11 @@ async fn set_room_state( content: &serde_json::Value, ) -> Result<()> { let base = matrix_base()?; - let current = - current_room_state(client, base, sender_token, room_id, event_type, state_key).await; + let encoded_room = encode_room_id_for_url(room_id); + let encoded_key = encode_room_id_for_url(state_key); + let url = + format!("{base}/_matrix/client/v3/rooms/{encoded_room}/state/{event_type}/{encoded_key}"); + let current = current_room_state(client, sender_token, &url).await; if !state_needs_write(current.as_ref(), content) { tracing::debug!( %room_id, @@ -516,17 +562,19 @@ async fn set_room_state( ); return Ok(()); } - swarm_matrix_client::put_state( - client, - base, - sender_token, - room_id, - event_type, - state_key, - content, - ) - .await - .context("matrix: set room state") + let resp = client + .put(&url) + .bearer_auth(sender_token) + .json(content) + .send() + .await + .with_context(|| format!("matrix: PUT state {event_type} into {room_id}"))?; + let status = resp.status(); + if status.is_success() { + return Ok(()); + } + let body = resp.json::().await.unwrap_or_default(); + anyhow::bail!("matrix: set state {event_type} in {room_id}: HTTP {status}, body: {body}") } /// Create (or recover) the default hive chat room and wire it as a child of @@ -574,6 +622,7 @@ pub async fn ensure_hive_chat_room( } else { // `initial_state` is applied after the preset-derived state, so the // restricted join rule overrides private_chat's invite-only default. + let url = format!("{base}/_matrix/client/v3/createRoom"); let body = serde_json::json!({ "name": HIVE_CHAT_ROOM_NAME, "topic": HIVE_CHAT_ROOM_TOPIC, @@ -604,9 +653,25 @@ pub async fn ensure_hive_chat_room( } ] }); - let id = swarm_matrix_client::create_room(client, base, sender_token, &body) + let resp = client + .post(&url) + .bearer_auth(sender_token) + .json(&body) + .send() .await - .context("matrix: create hive chat room")?; + .context("matrix: POST /createRoom (hive chat room)")?; + let status = resp.status(); + let json = resp + .json::() + .await + .context("matrix: parse /createRoom response (chat room)")?; + if !status.is_success() { + anyhow::bail!("matrix: createRoom (chat) HTTP {status}, body: {json}"); + } + let id = json["room_id"] + .as_str() + .with_context(|| format!("matrix: createRoom (chat) missing room_id: {json}"))? + .to_owned(); persist_chat_room_id(&id)?; tracing::info!(room_id = %id, "matrix: created hive chat room"); id @@ -654,30 +719,36 @@ async fn invite_to_room( async fn room_membership( client: &reqwest::Client, sender_token: &str, - room_id: &str, + encoded_room_id: &str, user_id: &str, ) -> Option { let base = matrix_http()?; - // A refused or failed read is "unknown": the caller falls through to the - // invite attempt. - let member = swarm_matrix_client::room_state( - client, - base, - sender_token, - room_id, - "m.room.member", - user_id, - ) - .await - .ok()??; - member["membership"].as_str().map(ToOwned::to_owned) + // `:` must be percent-encoded in both the room-id and user-id path + // segments; `@` and `!` are permitted path characters per RFC 3986. + let encoded_user = user_id.replace(':', "%3A"); + let url = format!( + "{base}/_matrix/client/v3/rooms/{encoded_room_id}/state/m.room.member/{encoded_user}" + ); + let resp = client + .get(&url) + .bearer_auth(sender_token) + .send() + .await + .ok()?; + if !resp.status().is_success() { + // 404 = no membership event yet; anything else we treat as "unknown" + // and let the caller fall through to the invite attempt. + return None; + } + let body = resp.json::().await.ok()?; + body["membership"].as_str().map(ToOwned::to_owned) } /// Invite a fully-qualified Matrix user id (`@user:server`) to `room_id` /// using the sender token. Idempotent: a user who is already a member or /// already has a pending invite is left untouched (no fresh invite is sent, /// so they are not re-notified). A refused invite is an error unless the -/// user turns out to be invited or joined anyway — see [`forbidden_invite`]. +/// user turns out to be invited or joined anyway — see [`refused_invite`]. async fn invite_user_id( client: &reqwest::Client, sender_token: &str, @@ -685,47 +756,61 @@ async fn invite_user_id( user_id: &str, ) -> Result<()> { let base = matrix_base()?; + // `:` 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"); // Skip the invite entirely when the user is already invited or joined. // Re-POSTing an invite to a pending member re-sends the invite event, // which re-notifies the agent on every provisioning sweep. - if let Some(membership) = room_membership(client, sender_token, room_id, user_id).await + if let Some(membership) = room_membership(client, sender_token, &encoded_room_id, user_id).await && matches!(membership.as_str(), "invite" | "join") { tracing::debug!(%user_id, %room_id, %membership, "matrix: invite skipped (already a member/invited)"); return Ok(()); } - match swarm_matrix_client::invite(client, base, sender_token, room_id, user_id) + let url = format!("{base}/_matrix/client/v3/rooms/{encoded_room_id}/invite"); + let resp = client + .post(&url) + .bearer_auth(sender_token) + .json(&serde_json::json!({ "user_id": user_id })) + .send() .await - .context("matrix: invite")? - { - swarm_matrix_client::Invited::Sent => { - tracing::debug!(%user_id, %room_id, "matrix: invited to room"); - Ok(()) - } - swarm_matrix_client::Invited::Forbidden => { - // The pre-check misses a member when its read failed or a - // concurrent invite landed after it, so a refusal re-reads the - // membership. - let membership = room_membership(client, sender_token, room_id, user_id).await; - forbidden_invite(membership.as_deref(), user_id, room_id) - } + .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 room"); + return Ok(()); } + // The pre-check misses a member when its read failed or a concurrent + // invite landed after it, so a refusal re-reads the membership. + let membership = if status == StatusCode::FORBIDDEN { + room_membership(client, sender_token, &encoded_room_id, user_id).await + } else { + None + }; + refused_invite(resp, membership.as_deref(), user_id, room_id).await } -/// Outcome of an invite the homeserver refused with 403. That is success only -/// when `membership`, re-read after the refusal, shows the user invited or -/// joined: `M_FORBIDDEN` covers both "already in the room" and real refusals -/// (banned target, sender lacks power), so the errcode can't tell them apart. -fn forbidden_invite(membership: Option<&str>, user_id: &str, room_id: &str) -> Result<()> { - if matches!(membership, Some("invite" | "join")) { +/// Outcome of an invite POST the homeserver did not accept. A 403 is success +/// only when `membership`, re-read after the refusal, shows the user invited +/// or joined: `M_FORBIDDEN` covers both "already in the room" and real +/// refusals (banned target, sender lacks power), so the errcode can't tell +/// them apart. +async fn refused_invite( + resp: reqwest::Response, + membership: Option<&str>, + user_id: &str, + room_id: &str, +) -> Result<()> { + let status = resp.status(); + if status == StatusCode::FORBIDDEN && matches!(membership, Some("invite" | "join")) { tracing::debug!(%user_id, %room_id, "matrix: invite refused, user already member/invited"); return Ok(()); } - anyhow::bail!( - "matrix: invite {user_id} to {room_id}: HTTP 403 and the user is not invited or joined" - ) + let body = resp.json::().await.unwrap_or_default(); + anyhow::bail!("matrix: invite {user_id} to {room_id}: HTTP {status}, body: {body}") } /// Invite an arbitrary Matrix user to the hive Space (default) or an @@ -756,9 +841,7 @@ pub async fn invite_user( // Resolve the room: explicit override (id or #alias) wins; otherwise // the persisted hive Space. let room_id = match room_override { - Some(r) if r.starts_with('#') => { - swarm_matrix_client::resolve_alias(client, matrix_base()?, sender_token, r).await? - } + Some(r) if r.starts_with('#') => resolve_room_alias(client, sender_token, r).await?, Some(r) if r.starts_with('!') => r.to_owned(), Some(r) => anyhow::bail!( "matrix: --room {r:?} is neither a room id nor an alias; \ @@ -776,6 +859,32 @@ pub async fn invite_user( Ok(room_id) } +/// Resolve a `#alias:server` to its room id via the directory API. +async fn resolve_room_alias( + client: &reqwest::Client, + sender_token: &str, + alias: &str, +) -> Result { + let base = matrix_base()?; + let encoded = alias.replace('#', "%23").replace(':', "%3A"); + let url = format!("{base}/_matrix/client/v3/directory/room/{encoded}"); + let resp = client + .get(&url) + .bearer_auth(sender_token) + .send() + .await + .with_context(|| format!("matrix: GET directory for {alias}"))?; + let status = resp.status(); + let json = resp.json::().await.unwrap_or_default(); + if !status.is_success() { + anyhow::bail!("matrix: resolve alias {alias}: HTTP {status}, body: {json}"); + } + json["room_id"] + .as_str() + .map(str::to_owned) + .ok_or_else(|| anyhow::anyhow!("matrix: alias {alias} response missing room_id: {json}")) +} + /// Make sure the hive's own `@hive-:` account exists, then provision /// the hive Space and chat room and invite every agent container to both. /// Agents' own accounts are not created here: `swarm-controller` mints them @@ -796,7 +905,10 @@ pub async fn ensure_all() -> bool { } let mut ok = true; // One HTTP client for the whole sweep. - let client = match swarm_matrix_client::client() { + 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"); @@ -993,57 +1105,52 @@ mod tests { assert!(state_needs_write(None, &desired)); } - /// A banned target or a sender without power gets a 403; with no - /// membership behind it, the invite did not happen. - #[test] - fn a_refused_invite_without_membership_is_an_error() { + /// A homeserver response built in memory, so no client (and no TLS + /// roots) is needed to drive the response-handling halves. + fn response(status: u16, body: &'static str) -> reqwest::Response { + axum::http::Response::builder() + .status(status) + .body(body) + .expect("valid mock response") + .into() + } + + const FORBIDDEN_BODY: &str = r#"{"errcode":"M_FORBIDDEN","error":"refused"}"#; + + /// A banned target or a sender without power gets exactly this 403; + /// with no membership behind it, the invite did not happen. + #[tokio::test] + async fn a_refused_invite_without_membership_is_an_error() { for membership in [None, Some("ban"), Some("leave")] { - let outcome = forbidden_invite(membership, "@a:x", "!r:x"); + let outcome = + refused_invite(response(403, FORBIDDEN_BODY), membership, "@a:x", "!r:x").await; assert!(outcome.is_err(), "membership {membership:?} must not pass"); } } - #[test] - fn a_refused_invite_for_a_member_is_success() { + #[tokio::test] + async fn a_refused_invite_for_a_member_is_success() { for membership in ["invite", "join"] { - forbidden_invite(Some(membership), "@a:x", "!r:x") - .expect("already invited/joined is the goal state"); + refused_invite( + response(403, FORBIDDEN_BODY), + Some(membership), + "@a:x", + "!r:x", + ) + .await + .expect("already invited/joined is the goal state"); } } - /// The homeserver's reply body reaches no error: `invite_user_id` is - /// driven against a stub whose 500 and 403 bodies carry a marker. + /// Membership only excuses a 403; any other failure stays a failure. #[tokio::test] - async fn an_invite_refusal_error_does_not_quote_the_response_body() { - const BODY: &str = r#"{"errcode":"M_UNKNOWN","error":"free-form-marker"}"#; - for status in [500_u16, 403] { - 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")); - let server = tokio::spawn(async move { - axum::serve(listener, app).await.expect("serve"); - }); - let client = swarm_matrix_client::client().expect("client"); - let outcome = swarm_matrix_client::invite(&client, &base, "tok", "!r:x", "@a:x").await; - server.abort(); - let message = match (status, outcome) { - (403, Ok(swarm_matrix_client::Invited::Forbidden)) => { - format!("{:#}", forbidden_invite(None, "@a:x", "!r:x").unwrap_err()) - } - (500, Err(e)) => format!("{e:#}"), - (status, other) => panic!("HTTP {status}: unexpected {other:?}"), - }; - assert!(message.contains(&status.to_string()), "{message}"); - assert!(!message.contains("free-form-marker"), "{message}"); + async fn a_non_403_invite_failure_or_garbage_body_is_an_error() { + for (status, body) in [(500, "not json"), (403, "not json"), (429, "{}")] { + let outcome = refused_invite(response(status, body), None, "@a:x", "!r:x").await; + assert!(outcome.is_err(), "HTTP {status} {body:?} must fail"); } + let outcome = refused_invite(response(500, "{}"), Some("join"), "@a:x", "!r:x").await; + assert!(outcome.is_err(), "a 500 is not excused by membership"); } /// Serve a fake homeserver on loopback: `!space` is an `m.space` and diff --git a/hive-matrix-mcp/Cargo.toml b/hive-matrix-mcp/Cargo.toml index 31f3a9b3..9857502d 100644 --- a/hive-matrix-mcp/Cargo.toml +++ b/hive-matrix-mcp/Cargo.toml @@ -21,7 +21,6 @@ rmcp.workspace = true schemars.workspace = true serde.workspace = true serde_json.workspace = true -swarm-matrix-client.workspace = true swarm-secret-client.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index f128c9b6..5b478fda 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -36,7 +36,7 @@ use matrix_sdk::{ encryption::{BackupDownloadStrategy, EncryptionSettings}, ruma::{OwnedDeviceId, OwnedUserId}, }; -use swarm_matrix_client::WhoamiSession; +use serde::Deserialize; use tokio::fs; use crate::credential::{Passphrase, Token}; @@ -61,12 +61,22 @@ impl std::fmt::Display for PermanentBringUpError { impl std::error::Error for PermanentBringUpError {} +/// Subset of the `/_matrix/client/v3/account/whoami` response we care +/// about. matrix-spec field names; `device_id` is optional per spec +/// (servers MAY omit it for legacy bearer scopes) but tuwunel always +/// returns it. +#[derive(Debug, Deserialize)] +struct WhoamiResponse { + user_id: String, + device_id: Option, +} + /// 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 +/// 1. Plain reqwest GET to `/_matrix/client/v3/account/whoami` 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 @@ -85,53 +95,59 @@ pub async fn build_and_restore( state_dir: &Path, is_primary: bool, ) -> Result { - 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" + let (user_id, device_id) = match whoami(homeserver, token.expose()).await { + Ok(ids) => ids, + Err(e) => { + let msg = format!("{e:#}"); + if msg.contains("M_UNKNOWN_TOKEN") { + // 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(), + ))); } - std::process::exit(0); + return Err(e); } - // 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) @@ -166,25 +182,38 @@ pub async fn build_and_restore( 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> { - 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 +/// Bare-reqwest 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. +async fn whoami(homeserver: &str, token: &str) -> Result<(OwnedUserId, OwnedDeviceId)> { + let url = format!("{homeserver}/_matrix/client/v3/account/whoami"); + let resp = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build() + .context("build whoami reqwest client")? + .get(&url) + .bearer_auth(token) + .send() + .await + .with_context(|| format!("GET {url}"))?; + let status = resp.status(); + if !status.is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(anyhow!("whoami GET {url} → HTTP {status}, body: {body}")); + } + let body: WhoamiResponse = resp.json().await.context("parse whoami response")?; + let user_id: OwnedUserId = body + .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()))) + .with_context(|| format!("invalid user_id in whoami: {}", body.user_id))?; + let device_id_raw = body + .device_id + .ok_or_else(|| anyhow!("whoami response missing device_id"))?; + let device_id: OwnedDeviceId = device_id_raw.into(); + Ok((user_id, device_id)) } /// Best-effort: bootstrap this account's cross-signing identity so its @@ -529,41 +558,4 @@ mod tests { )); 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()); - } } diff --git a/swarm-matrix-client/README.md b/swarm-matrix-client/README.md index cbb97454..d7ce5ecd 100644 --- a/swarm-matrix-client/README.md +++ b/swarm-matrix-client/README.md @@ -1,11 +1,11 @@ # swarm-matrix-client -The client-server API calls against a Matrix homeserver, shared by the binaries +The appservice calls against the swarm's homeserver, shared by the two binaries that make them: `swarm-matrix-ctl` (inside the matrix container, for a hive's sender account) and `swarm-controller` (for each agent's account, with the swarm's own appservice token). -Three appservice calls: register an account as the appservice, log +Three calls, all client-server API: register an account as the appservice, log in to an existing one as the appservice, and ask a token who it is. Both mints pin the device id `hyperhive-`, so a second login **replaces** that device's token instead of adding a device. @@ -15,12 +15,6 @@ device's token instead of adding a device. the operator names. Neither call is an appservice call, so neither carries the appservice token. -`hive-c0re` provisions the hive Space and chat room with the room calls -(`joined_rooms`, `room_state`, `put_state`, `create_room`, `invite`, -`resolve_alias`), and `hive-matrix-mcp` recovers a session's user and device id -with `whoami_session`. Each takes the caller's `reqwest::Client`, so the caller -sets its own timeout; `client()` is the 10 second default. - ## 🩸 A secret is a path, never a value Every error here is built from the response's status and its `errcode`, never diff --git a/swarm-matrix-client/src/lib.rs b/swarm-matrix-client/src/lib.rs index 5a573a0e..ecc3ff68 100644 --- a/swarm-matrix-client/src/lib.rs +++ b/swarm-matrix-client/src/lib.rs @@ -1,13 +1,9 @@ -//! The calls against a Matrix homeserver's client-server API shared by the -//! swarm and the hive: the appservice calls (register, log in, ask a token who -//! it is), the room calls `swarm-controller`'s admin-room commands and -//! `hive-c0re`'s Space and chat-room provisioning need (resolve an alias, list -//! members and joined rooms, read and set room state, create a room, invite, send -//! a message, check an account exists), and the password login and logout -//! `swarm-controller` runs for an operator-linked account on any homeserver. -//! Shared by `swarm-matrix-ctl`, `swarm-controller`, `hive-c0re` and -//! `hive-matrix-mcp`. See the README for why both appservice mints go through -//! the same device id. +//! The appservice calls against the swarm's homeserver: register, log in, ask +//! a token who it is, and the room calls `swarm-controller`'s admin-room +//! commands need (resolve an alias, list members, check an account exists, +//! send a message). Also the password login and logout `swarm-controller` runs +//! for an operator-linked account on any homeserver. Shared by `swarm-matrix-ctl` and `swarm-controller`; +//! see the README for why both mint through the same device id. //! //! 🩸 **Every error in this module is built from the response's `status` and //! its `errcode`, never its body.** A successful `/register` or `/login` body @@ -433,237 +429,6 @@ pub async fn send_text( }) } -/// What the homeserver says a token is, with the device it belongs to. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum WhoamiSession { - /// A live token: the user id it authenticates as, and its device id, which - /// the spec lets a homeserver omit. - Session { - user_id: String, - device_id: Option, - }, - /// `M_UNKNOWN_TOKEN`, as for [`Whoami::UnknownToken`]. - UnknownToken, -} - -/// [`whoami`] for a caller that restores a session and so needs the device id -/// too. -/// -/// # Errors -/// As [`whoami`]. -pub async fn whoami_session( - client: &reqwest::Client, - base: &str, - token: &str, -) -> Result { - let (status, json) = get( - client, - &format!("{base}/_matrix/client/v3/account/whoami"), - token, - ) - .await - .context("GET /account/whoami")?; - whoami_session_verdict(status, &json) -} - -/// [`whoami_session`]'s reading of a response, apart from the IO. -fn whoami_session_verdict( - status: reqwest::StatusCode, - json: &serde_json::Value, -) -> Result { - match whoami_verdict(status, json)? { - Whoami::User(user_id) => Ok(WhoamiSession::Session { - user_id, - device_id: json["device_id"].as_str().map(str::to_owned), - }), - Whoami::UnknownToken => Ok(WhoamiSession::UnknownToken), - } -} - -/// The ids of the rooms `token`'s user has joined. -/// -/// # Errors -/// When the request cannot be sent, the response will not decode, the -/// homeserver refuses, or the answer carries no `joined_rooms`. -pub async fn joined_rooms( - client: &reqwest::Client, - base: &str, - token: &str, -) -> Result> { - let (status, json) = get( - client, - &format!("{base}/_matrix/client/v3/joined_rooms"), - token, - ) - .await - .context("GET /joined_rooms")?; - if !status.is_success() { - bail!( - "the homeserver refused to list joined rooms: {}", - why(status, &json) - ); - } - Ok(json["joined_rooms"] - .as_array() - .context("the homeserver's answer carried no `joined_rooms`")? - .iter() - .filter_map(|room| room.as_str().map(str::to_owned)) - .collect()) -} - -/// The content of `room_id`'s `event_type` state event under `state_key` -/// (empty for most), or `None` when the room has none. -/// -/// # Errors -/// When the request cannot be sent, the response will not decode, or the -/// homeserver refuses with anything other than 404. A 404 is the expected -/// answer for a state event never set, so it is `None`; every other refusal -/// says nothing about the event, so it is an error. -pub async fn room_state( - client: &reqwest::Client, - base: &str, - token: &str, - room_id: &str, - event_type: &str, - state_key: &str, -) -> Result> { - let (status, json) = get( - client, - &state_url(base, room_id, event_type, state_key), - token, - ) - .await - .with_context(|| format!("GET {event_type} state of {room_id}"))?; - if status.is_success() { - return Ok(Some(json)); - } - if status == reqwest::StatusCode::NOT_FOUND { - return Ok(None); - } - bail!( - "the homeserver refused to read {event_type} of {room_id}: {}", - why(status, &json) - ) -} - -/// Set `room_id`'s `event_type` state event under `state_key` to `content`. -/// -/// # Errors -/// When the request cannot be sent, the response will not decode, or the -/// homeserver refuses. -pub async fn put_state( - client: &reqwest::Client, - base: &str, - token: &str, - room_id: &str, - event_type: &str, - state_key: &str, - content: &serde_json::Value, -) -> Result<()> { - let (status, json) = exchange( - client - .put(state_url(base, room_id, event_type, state_key)) - .bearer_auth(token) - .json(content), - ) - .await - .with_context(|| format!("PUT {event_type} state into {room_id}"))?; - if !status.is_success() { - bail!( - "the homeserver refused to set {event_type} in {room_id}: {}", - why(status, &json) - ); - } - Ok(()) -} - -/// The state-event URL for `room_id`, `event_type` and `state_key`. -fn state_url(base: &str, room_id: &str, event_type: &str, state_key: &str) -> String { - format!( - "{base}/_matrix/client/v3/rooms/{}/state/{event_type}/{}", - path_segment(room_id), - path_segment(state_key) - ) -} - -/// Create a room from the `createRoom` request `body` and return its id. -/// -/// # Errors -/// When the request cannot be sent, the response will not decode, the -/// homeserver refuses, or the answer carries no `room_id`. -pub async fn create_room( - client: &reqwest::Client, - base: &str, - token: &str, - body: &serde_json::Value, -) -> Result { - let (status, json) = post( - client, - &format!("{base}/_matrix/client/v3/createRoom"), - token, - body, - ) - .await - .context("POST /createRoom")?; - if !status.is_success() { - bail!( - "the homeserver refused to create the room: {}", - why(status, &json) - ); - } - json["room_id"] - .as_str() - .map(str::to_owned) - .context("the homeserver's createRoom answer carried no `room_id`") -} - -/// What the homeserver did with an invite. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Invited { - /// The invite went out. - Sent, - /// HTTP 403. `M_FORBIDDEN` covers both "already in the room" and a real - /// refusal (banned target, sender lacks power), so the caller has to - /// tell them apart by the target's membership. - Forbidden, -} - -/// Invite `user_id` to `room_id`. -/// -/// # Errors -/// When the request cannot be sent, the response will not decode, or the -/// homeserver refuses with anything other than 403, which is -/// [`Invited::Forbidden`]. -pub async fn invite( - client: &reqwest::Client, - base: &str, - token: &str, - room_id: &str, - user_id: &str, -) -> Result { - let (status, json) = post( - client, - &format!( - "{base}/_matrix/client/v3/rooms/{}/invite", - path_segment(room_id) - ), - token, - &serde_json::json!({ "user_id": user_id }), - ) - .await - .with_context(|| format!("POST /rooms/{room_id}/invite for {user_id}"))?; - if status.is_success() { - return Ok(Invited::Sent); - } - if status == reqwest::StatusCode::FORBIDDEN { - return Ok(Invited::Forbidden); - } - bail!( - "the homeserver refused to invite {user_id} to {room_id}: {}", - why(status, &json) - ) -} - /// `value` percent-encoded for one URL path segment: room ids, aliases and /// user ids carry `!`, `#`, `@` and `:`, and `#` would otherwise end the path. fn path_segment(value: &str) -> String { @@ -1117,153 +882,4 @@ mod tests { assert!(format!("{e:#}").contains(&status.to_string()), "{e:#}"); } } - - /// A homeserver on a loopback port that answers every request with - /// `status` and `body`. - async fn any_stub(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("a loopback port"); - let addr = listener.local_addr().expect("a bound address"); - tokio::spawn(async move { axum::serve(listener, app).await }); - format!("http://{addr}") - } - - const LEAKY: &str = - r#"{"errcode":"M_UNKNOWN","error":"free-form text","access_token":"syt_leak"}"#; - - /// A refusal names the status and errcode and quotes nothing else of the - /// body. - fn assert_refusal_hides_body(e: &anyhow::Error) { - let message = format!("{e:#}"); - assert!(message.contains("500"), "{message}"); - assert!(message.contains("M_UNKNOWN"), "{message}"); - assert!(!message.contains("free-form"), "{message}"); - assert!(!message.contains("syt_leak"), "{message}"); - } - - #[tokio::test] - async fn every_room_call_refuses_without_quoting_the_body() { - let base = any_stub(500, LEAKY).await; - let c = client().expect("a client"); - let body = serde_json::json!({}); - assert_refusal_hides_body(&joined_rooms(&c, &base, "t").await.expect_err("refused")); - assert_refusal_hides_body( - &room_state(&c, &base, "t", "!r:x", "m.room.name", "") - .await - .expect_err("refused"), - ); - assert_refusal_hides_body( - &put_state(&c, &base, "t", "!r:x", "m.room.name", "", &body) - .await - .expect_err("refused"), - ); - assert_refusal_hides_body( - &create_room(&c, &base, "t", &body) - .await - .expect_err("refused"), - ); - assert_refusal_hides_body( - &invite(&c, &base, "t", "!r:x", "@a:x") - .await - .expect_err("refused"), - ); - // Control: the same body behind a 403 is the one verdict `invite` returns. - let forbidden = any_stub(403, LEAKY).await; - assert_eq!( - invite(&c, &forbidden, "t", "!r:x", "@a:x") - .await - .expect("a verdict"), - Invited::Forbidden - ); - } - - #[tokio::test] - async fn room_state_reads_content_and_treats_404_as_absent() { - let c = client().expect("a client"); - let base = any_stub(200, r#"{"name":"shared"}"#).await; - assert_eq!( - room_state(&c, &base, "t", "!r:x", "m.room.name", "") - .await - .expect("content"), - Some(serde_json::json!({"name":"shared"})) - ); - let base = any_stub(404, r#"{"errcode":"M_NOT_FOUND"}"#).await; - assert_eq!( - room_state(&c, &base, "t", "!r:x", "m.room.name", "") - .await - .expect("absent"), - None - ); - } - - #[tokio::test] - async fn joined_rooms_and_create_room_read_their_fields() { - let c = client().expect("a client"); - let base = any_stub(200, r#"{"joined_rooms":["!a:x","!b:x"]}"#).await; - assert_eq!( - joined_rooms(&c, &base, "t").await.expect("rooms"), - ["!a:x", "!b:x"] - ); - assert!( - create_room(&c, &base, "t", &serde_json::json!({})) - .await - .is_err() - ); - let base = any_stub(200, r#"{"room_id":"!new:x"}"#).await; - assert_eq!( - create_room(&c, &base, "t", &serde_json::json!({})) - .await - .expect("room"), - "!new:x" - ); - assert!(joined_rooms(&c, &base, "t").await.is_err()); - } - - #[test] - fn a_state_url_encodes_room_and_key_and_keeps_an_empty_key_slash() { - assert_eq!( - state_url("http://h", "!r:x", "m.room.create", ""), - "http://h/_matrix/client/v3/rooms/%21r%3Ax/state/m.room.create/" - ); - assert_eq!( - state_url("http://h", "!r:x", "m.space.child", "!c:x"), - "http://h/_matrix/client/v3/rooms/%21r%3Ax/state/m.space.child/%21c%3Ax" - ); - } - - #[test] - fn whoami_session_carries_the_device_id_when_the_homeserver_sends_one() { - let ok = reqwest::StatusCode::OK; - assert_eq!( - whoami_session_verdict(ok, &serde_json::json!({"user_id":"@a:x","device_id":"D"})) - .expect("live"), - WhoamiSession::Session { - user_id: "@a:x".to_owned(), - device_id: Some("D".to_owned()) - } - ); - assert_eq!( - whoami_session_verdict(ok, &serde_json::json!({"user_id":"@a:x"})).expect("live"), - WhoamiSession::Session { - user_id: "@a:x".to_owned(), - device_id: None - } - ); - assert_eq!( - whoami_session_verdict( - reqwest::StatusCode::UNAUTHORIZED, - &serde_json::json!({"errcode":"M_UNKNOWN_TOKEN"}) - ) - .expect("a verdict"), - WhoamiSession::UnknownToken - ); - } }