diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 3b5e5d24..6570a43d 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -34,6 +34,7 @@ use utoipa::ToSchema; use crate::webhook::DeliveryKind; pub mod agent_token; +pub mod legacy_tokens; pub mod objects; pub mod site_admin; diff --git a/swarm-controller/src/forge/agent_token.rs b/swarm-controller/src/forge/agent_token.rs index 5932bce8..b6d89ed1 100644 --- a/swarm-controller/src/forge/agent_token.rs +++ b/swarm-controller/src/forge/agent_token.rs @@ -27,7 +27,7 @@ use super::Client; /// The forge's name for every agent token this module mints. /// /// Deliberately not `hyperhive-…`: that prefix is what `hive-c0re` named each -/// of its re-mints, and a later sweep of those must never match this one. +/// of its re-mints, and [`super::legacy_tokens`] deletes those. pub const AGENT_TOKEN_NAME: &str = "swarm-agent"; /// The scopes an agent token carries. Byte-identical to the `TOKEN_SCOPES` @@ -167,7 +167,7 @@ pub(super) fn is_not_found(e: &ForgejoError) -> bool { impl Client { /// Every access token the forge lists for `agent`, or `None` when the /// forge has no such user. - async fn list_agent_tokens(&self, agent: &str) -> Result>> { + pub(super) async fn list_agent_tokens(&self, agent: &str) -> Result>> { match self.api.admin_list_user_access_tokens(agent).all().await { Ok(tokens) => Ok(Some(tokens)), Err(e) if is_not_found(&e) => Ok(None), @@ -270,7 +270,7 @@ impl Client { } /// What the forge and the store say about `agent`'s token. - async fn observe_agent( + pub(super) async fn observe_agent( &self, store: &swarm_secret_client::SecretStore, agent: &str, diff --git a/swarm-controller/src/forge/legacy_tokens.rs b/swarm-controller/src/forge/legacy_tokens.rs new file mode 100644 index 00000000..44a08527 --- /dev/null +++ b/swarm-controller/src/forge/legacy_tokens.rs @@ -0,0 +1,361 @@ +//! A startup sweep of the `hyperhive-` access tokens `hive-c0re` +//! minted for agents on every spawn and rebuild, each one live on the forge +//! with write scopes. +//! +//! Two filters decide what goes, and both must pass: +//! +//! - **Whose tokens.** Only agents on the store's `hive-agent-*` roster, the +//! one [`super::agent_token`]'s pass walks, and only those whose +//! [`AGENT_TOKEN_NAME`] token that pass would [`Decision::Keep`]: until the +//! replacement is stored and live, the agent may still be reading its last +//! `hyperhive-*` token from `/forge-token`. [`CORE_USER`] is refused +//! by name at every step. `hive-c0re` still names `core`'s live admin token +//! `hyperhive-`, so nothing about the token tells them apart. +//! - **Which tokens.** A name that is exactly `hyperhive-` followed by one or +//! more ASCII digits ([`is_legacy_token_name`]). [`AGENT_TOKEN_NAME`] cannot +//! match. +//! +//! One pass per daemon start. A failed read or delete is logged and skipped; +//! the next start picks up whatever is left, and a start with nothing left +//! deletes nothing. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use swarm_secret_client::{client::DEFAULT_CERT_MOUNT, policy}; + +use super::Client; +use super::agent_token::{AGENT_TOKEN_NAME, Decision, Observed}; + +/// The forge user `hive-c0re` runs as. Its admin token carries a legacy-shaped +/// name and is in use. +pub const CORE_USER: &str = "core"; + +/// The prefix `hive-c0re`'s `mint_token` put before the unix seconds. +const LEGACY_TOKEN_PREFIX: &str = "hyperhive-"; + +/// Whether `name` is exactly `hyperhive-`, the shape of every token +/// `hive-c0re` minted. +pub fn is_legacy_token_name(name: &str) -> bool { + name.strip_prefix(LEGACY_TOKEN_PREFIX) + .is_some_and(|secs| !secs.is_empty() && secs.bytes().all(|b| b.is_ascii_digit())) +} + +/// The agents a sweep may touch: those whose [`AGENT_TOKEN_NAME`] token is +/// current, never [`CORE_USER`]. +pub fn sweepable(observed: &[(String, Observed)]) -> Vec { + observed + .iter() + .filter(|(agent, o)| agent != CORE_USER && *o == Observed::Decided(Decision::Keep)) + .map(|(agent, _)| agent.clone()) + .collect() +} + +impl Client { + /// Delete every legacy token the forge lists for `user`, and return how + /// many went. A delete the forge refuses is logged and skipped. + async fn sweep_user(&self, user: &str) -> Result { + if user == CORE_USER { + return Ok(0); + } + let Some(listed) = self.list_agent_tokens(user).await? else { + return Ok(0); + }; + let mut removed = 0; + for name in listed + .iter() + .filter_map(|t| t.name.as_deref()) + .filter(|n| is_legacy_token_name(n)) + { + match self.api.admin_delete_user_access_token(user, name).await { + Ok(()) => removed += 1, + Err(e) if super::agent_token::is_not_found(&e) => {} + Err(e) => tracing::warn!( + agent = user, + token = name, + error = %e, + "legacy forge token sweep: delete failed; skipped" + ), + } + } + Ok(removed) + } + + /// Sweep each of `agents` and log what went, per agent and in total. + /// Returns the total. + async fn sweep_legacy_tokens(&self, agents: &[String]) -> usize { + let mut total = 0; + for agent in agents { + match self.sweep_user(agent).await { + Ok(removed) => { + tracing::info!(agent, removed, "legacy forge token sweep: agent done"); + total += removed; + } + Err(e) => tracing::warn!( + agent, + error = %format!("{e:#}"), + "legacy forge token sweep: listing failed; agent skipped" + ), + } + } + tracing::info!( + agents = agents.len(), + removed = total, + "legacy forge token sweep: done" + ); + total + } + + /// The roster agents [`sweepable`] allows, observed the way the mint pass + /// observes them. + async fn sweep_roster(&self) -> Result> { + let store = crate::store::connect() + .await + .context("logging in to the swarm secret store")?; + let roles = store + .list_cert_roles(DEFAULT_CERT_MOUNT) + .await + .context("listing the store's cert-auth roles")?; + let mut observed = Vec::new(); + for agent in policy::agents_from_role_names(&roles) { + let o = self.observe_agent(&store, &agent).await; + if o != Observed::Decided(Decision::Keep) { + tracing::info!( + agent, + ?o, + "legacy forge token sweep: {AGENT_TOKEN_NAME} token not current; agent skipped" + ); + } + observed.push((agent, o)); + } + Ok(sweepable(&observed)) + } +} + +/// Run one sweep in the background. A failure is logged and not retried. +pub fn spawn(client: Arc) { + tokio::spawn(async move { + match client.sweep_roster().await { + Ok(agents) => { + client.sweep_legacy_tokens(&agents).await; + } + Err(e) => tracing::warn!( + error = %format!("{e:#}"), + "legacy forge token sweep: roster unavailable; not run" + ), + } + }); +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + use std::sync::Mutex; + + use forgejo_api::{Auth, Forgejo}; + + use super::super::agent_token::MintReason; + use super::*; + + #[test] + fn only_hyperhive_digits_is_legacy() { + for name in ["hyperhive-1", "hyperhive-1700000000"] { + assert!(is_legacy_token_name(name), "{name}"); + } + for name in [ + AGENT_TOKEN_NAME, + "hyperhive", + "hyperhive-", + "hyperhive-abc", + "hyperhive-123abc", + "hyperhive-12 3", + "hyperhive-+123", + "xhyperhive-123", + "hyperhive-123-x", + "swarm-controller-boot-1700000000", + ] { + assert!(!is_legacy_token_name(name), "{name}"); + } + } + + #[test] + fn only_current_non_core_agents_are_sweepable() { + let observed = [ + ("a".to_owned(), Observed::Decided(Decision::Keep)), + (CORE_USER.to_owned(), Observed::Decided(Decision::Keep)), + ( + "b".to_owned(), + Observed::Decided(Decision::Mint(MintReason::NotOnForge)), + ), + ("c".to_owned(), Observed::Unknown), + ("d".to_owned(), Observed::NoForgeUser), + ("e".to_owned(), Observed::Decided(Decision::Conflict)), + ("f".to_owned(), Observed::Decided(Decision::Keep)), + ]; + assert_eq!(sweepable(&observed), ["a", "f"]); + } + + /// Each user's token names, as the stub forge holds them. + type Tokens = Arc>>>; + /// Every `DELETE` the stub forge was sent, as `(user, token)`. + type Deletes = Arc>>; + + /// A stand-in forge on a loopback port answering the admin token list and + /// delete from `tokens`. A delete of `fail` answers 500 and removes + /// nothing. + async fn stub_forge(tokens: Tokens, fail: (&'static str, &'static str)) -> (Client, Deletes) { + use axum::http::{Method, StatusCode, Uri, header::CONTENT_TYPE}; + + let deletes = Deletes::default(); + let recorded = deletes.clone(); + let app = axum::Router::new().fallback(move |method: Method, uri: Uri| { + let tokens = tokens.clone(); + let recorded = recorded.clone(); + async move { + let parts: Vec<&str> = uri.path().trim_matches('/').split('/').collect(); + let (status, body) = match (method, parts.as_slice()) { + (Method::GET, ["api", "v1", "admin", "users", user, "tokens"]) => { + match tokens.lock().unwrap().get(*user) { + Some(held) => { + let list: Vec = held + .iter() + .enumerate() + .map(|(i, name)| serde_json::json!({ "id": i, "name": name })) + .collect(); + (StatusCode::OK, serde_json::Value::from(list)) + } + None => ( + StatusCode::NOT_FOUND, + serde_json::json!({ "message": "user does not exist" }), + ), + } + } + (Method::DELETE, ["api", "v1", "admin", "users", user, "tokens", token]) => { + recorded + .lock() + .unwrap() + .push(((*user).to_owned(), (*token).to_owned())); + if (*user, *token) == fail { + (StatusCode::INTERNAL_SERVER_ERROR, serde_json::json!({})) + } else { + let mut all = tokens.lock().unwrap(); + let held = all.entry((*user).to_owned()).or_default(); + held.retain(|n| n != token); + (StatusCode::NO_CONTENT, serde_json::Value::Null) + } + } + _ => ( + StatusCode::NOT_FOUND, + serde_json::json!({ "message": "not stubbed" }), + ), + }; + let count = body.as_array().map_or(0, Vec::len).to_string(); + let body = if status == StatusCode::NO_CONTENT { + String::new() + } else { + body.to_string() + }; + ( + status, + [ + (CONTENT_TYPE.as_str(), "application/json".to_owned()), + ("x-total-count", count), + ], + body, + ) + } + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); + (Client { api, url }, deletes) + } + + const MIX: [&str; 5] = [ + "hyperhive-123", + "hyperhive-1700000000", + "hyperhive-abc", + "hyperhive", + AGENT_TOKEN_NAME, + ]; + + fn forge_with_core_and_two_agents() -> Tokens { + let held = |names: &[&str]| names.iter().map(|n| (*n).to_owned()).collect(); + Arc::new(Mutex::new(BTreeMap::from([ + (CORE_USER.to_owned(), held(&MIX)), + ("alice".to_owned(), held(&MIX)), + ("bob".to_owned(), held(&MIX)), + ]))) + } + + fn sorted(deletes: &Deletes) -> Vec<(String, String)> { + let mut d = deletes.lock().unwrap().clone(); + d.sort(); + d + } + + fn pair(user: &str, token: &str) -> (String, String) { + (user.to_owned(), token.to_owned()) + } + + /// Core is in the list handed to the sweep, as a stand-in for any path + /// that lets it through [`sweepable`]; its tokens must survive anyway. + #[tokio::test] + async fn only_agents_legacy_tokens_go_and_a_rerun_removes_nothing() { + let tokens = forge_with_core_and_two_agents(); + let (client, deletes) = stub_forge(tokens.clone(), ("", "")).await; + let users = [CORE_USER.to_owned(), "alice".to_owned(), "bob".to_owned()]; + + assert_eq!(client.sweep_legacy_tokens(&users).await, 4); + assert_eq!( + sorted(&deletes), + [ + pair("alice", "hyperhive-123"), + pair("alice", "hyperhive-1700000000"), + pair("bob", "hyperhive-123"), + pair("bob", "hyperhive-1700000000"), + ] + ); + let left = tokens.lock().unwrap().clone(); + assert_eq!(left[CORE_USER], MIX); + for agent in ["alice", "bob"] { + assert_eq!( + left[agent], + ["hyperhive-abc", "hyperhive", AGENT_TOKEN_NAME] + ); + } + + deletes.lock().unwrap().clear(); + assert_eq!(client.sweep_legacy_tokens(&users).await, 0); + assert!(deletes.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_refused_delete_moves_on_to_the_next() { + let tokens = forge_with_core_and_two_agents(); + let (client, deletes) = stub_forge(tokens.clone(), ("alice", "hyperhive-123")).await; + let users = ["alice".to_owned(), "bob".to_owned()]; + + assert_eq!(client.sweep_legacy_tokens(&users).await, 3); + assert_eq!(deletes.lock().unwrap().len(), 4); + assert_eq!( + tokens.lock().unwrap()["alice"], + [ + "hyperhive-123", + "hyperhive-abc", + "hyperhive", + AGENT_TOKEN_NAME + ] + ); + } + + #[tokio::test] + async fn a_user_the_forge_does_not_have_removes_nothing() { + let tokens = forge_with_core_and_two_agents(); + let (client, deletes) = stub_forge(tokens, ("", "")).await; + + assert_eq!(client.sweep_legacy_tokens(&["ruth".to_owned()]).await, 0); + assert!(deletes.lock().unwrap().is_empty()); + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index d6f0861c..ff89b29d 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -2159,8 +2159,8 @@ fn spawn_matrix_account_backfill( } /// Start the forge's periodic passes — the swarm-wide objects and agents' -/// tokens — when a forge is configured. Lifted out of `main` for -/// `clippy::too_many_lines`. +/// tokens — and the one-shot legacy token sweep, when a forge is configured. +/// Lifted out of `main` for `clippy::too_many_lines`. fn spawn_forge_workers( jobq: &Arc>>, forge_client: Option>, @@ -2169,6 +2169,7 @@ fn spawn_forge_workers( return; }; forge::objects::spawn(Arc::clone(&client)); + forge::legacy_tokens::spawn(Arc::clone(&client)); let sched = Arc::clone(jobq); forge::agent_token::spawn(client, move |agents| { if let Err(e) = queue_forge_token_mints(&sched, agents) {