swarm-controller: stop minting the per-hive matrix sender account and token nothing reads
This commit is contained in:
parent
cb5fb3c730
commit
c66832febf
6 changed files with 40 additions and 556 deletions
|
|
@ -4,8 +4,7 @@
|
|||
//! stored at `swarm/hives/<hive>/forge/token`, where hive-c0re reads it under
|
||||
//! the hive's own store identity.
|
||||
//!
|
||||
//! The roster is the controller's hive directory, as for
|
||||
//! `matrix_account::hive_sender`. The token decision is
|
||||
//! The roster is the controller's hive directory. The token decision is
|
||||
//! [`super::agent_token`]'s, under this module's name and scopes, so a stored
|
||||
//! token the forge still lists with the same last eight characters and scopes
|
||||
//! is kept, and a pass whose reads fail queues nothing.
|
||||
|
|
|
|||
|
|
@ -91,10 +91,6 @@ pub(crate) enum SwarmNodeKind {
|
|||
/// Carries no hive: the path it writes has no hive segment, so an agent
|
||||
/// that moves between hives keeps one matrix identity.
|
||||
MintAgentMatrixAccount { agent: String },
|
||||
/// Make sure `hive`'s sender account (`@hive-<hive>:`) holds a live token
|
||||
/// in the swarm store, where that hive's matrix sweep reads it. See
|
||||
/// `matrix_account::hive_sender`.
|
||||
MintHiveSenderToken { hive: String },
|
||||
/// Make sure `hive`'s read-only forge account (`hive-<hive>`) exists, is in
|
||||
/// the `hives` team, and holds a live token in the swarm store, where
|
||||
/// that hive's hive-c0re reads it. See `forge::hive_token`.
|
||||
|
|
@ -189,7 +185,6 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
|||
SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(),
|
||||
SwarmNodeKind::MintAgentForgeToken { .. } => "mint_agent_forge_token".to_owned(),
|
||||
SwarmNodeKind::MintAgentMatrixAccount { .. } => "mint_agent_matrix_account".to_owned(),
|
||||
SwarmNodeKind::MintHiveSenderToken { .. } => "mint_hive_sender_token".to_owned(),
|
||||
SwarmNodeKind::MintHiveForgeToken { .. } => "mint_hive_forge_token".to_owned(),
|
||||
SwarmNodeKind::MintClassifierForgeToken => "mint_classifier_forge_token".to_owned(),
|
||||
SwarmNodeKind::MintBuilderForgeToken => "mint_builder_forge_token".to_owned(),
|
||||
|
|
@ -234,9 +229,9 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
|||
SwarmNodeKind::SeedArchetypeConfig { agent, archetype } => {
|
||||
serde_json::json!({ "agent": agent, "archetype": archetype.name })
|
||||
}
|
||||
SwarmNodeKind::MintHiveSenderToken { hive }
|
||||
| SwarmNodeKind::MintHiveForgeToken { hive }
|
||||
| SwarmNodeKind::CopyAtticFacts { hive } => serde_json::json!({ "hive": hive }),
|
||||
SwarmNodeKind::MintHiveForgeToken { hive } | SwarmNodeKind::CopyAtticFacts { hive } => {
|
||||
serde_json::json!({ "hive": hive })
|
||||
}
|
||||
SwarmNodeKind::MintClassifierForgeToken
|
||||
| SwarmNodeKind::MintBuilderForgeToken
|
||||
| SwarmNodeKind::MintCiMetricsForgeToken
|
||||
|
|
@ -438,9 +433,6 @@ async fn run_node_work(
|
|||
SwarmNodeKind::MintAgentMatrixAccount { agent } => {
|
||||
mint_matrix_account(deps.matrix_homeserver.as_deref(), &agent).await
|
||||
}
|
||||
SwarmNodeKind::MintHiveSenderToken { hive } => {
|
||||
mint_hive_sender(deps.matrix_homeserver.as_deref(), &hive).await
|
||||
}
|
||||
SwarmNodeKind::MintHiveForgeToken { hive } => mint_hive_forge_token(forge, &hive).await,
|
||||
SwarmNodeKind::MintClassifierForgeToken => mint_classifier_forge_token(forge).await,
|
||||
SwarmNodeKind::MintBuilderForgeToken => mint_builder_forge_token(forge).await,
|
||||
|
|
@ -1119,23 +1111,6 @@ async fn mint_matrix_account(
|
|||
}
|
||||
}
|
||||
|
||||
/// The `MintHiveSenderToken` arm, lifted out for the same reason.
|
||||
async fn mint_hive_sender(homeserver: Option<&str>, hive: &str) -> hive_jobq::scheduler::Outcome {
|
||||
use hive_jobq::scheduler::Outcome;
|
||||
|
||||
let Some(base) = homeserver else {
|
||||
return Outcome::Failed(format!(
|
||||
"no matrix homeserver configured on this host ({} unset), so no hive \
|
||||
sender token can be minted",
|
||||
matrix_account::DEFAULT_HOMESERVER_ENV
|
||||
));
|
||||
};
|
||||
match matrix_account::hive_sender::ensure_hive_sender_token(base, hive).await {
|
||||
Ok(()) => Outcome::Done,
|
||||
Err(e) => Outcome::Failed(format!("{e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// The `SetAgentWanted` arm, lifted out so `run_node_work` stays under the line
|
||||
/// limit: it fails without a wanted-state writer.
|
||||
async fn set_agent_wanted(
|
||||
|
|
@ -1724,27 +1699,6 @@ pub(crate) fn queue_matrix_account_mints(
|
|||
Ok(ids)
|
||||
}
|
||||
|
||||
/// Insert one `MintHiveSenderToken` job per hive and return the nodes' ids.
|
||||
/// `matrix_account::hive_sender::spawn`'s periodic pass is the one caller.
|
||||
pub(crate) fn queue_hive_sender_mints(
|
||||
sched: &Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
|
||||
hives: Vec<String>,
|
||||
) -> Result<Vec<hive_jobq::NodeId>> {
|
||||
let mut sched = sched
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut ids = Vec::with_capacity(hives.len());
|
||||
for hive in hives {
|
||||
let queued = sched
|
||||
.insert_job(None, |b| {
|
||||
vec![b.node(SwarmNodeKind::MintHiveSenderToken { hive }).guid()]
|
||||
})
|
||||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||||
ids.extend(queued);
|
||||
}
|
||||
Ok(ids)
|
||||
}
|
||||
|
||||
/// Insert one `MintHiveForgeToken` job per hive and return the nodes' ids.
|
||||
/// `forge::hive_token::spawn`'s periodic pass is the one caller.
|
||||
pub(crate) fn queue_hive_forge_token_mints(
|
||||
|
|
@ -3970,37 +3924,6 @@ mod tests {
|
|||
assert_eq!(agents, ["a", "b"]);
|
||||
}
|
||||
|
||||
/// The hive sender-token pass's queueing: one mint node per hive, and
|
||||
/// the node names the hive rather than an agent.
|
||||
#[test]
|
||||
fn a_queued_hive_sender_mint_is_one_node_per_hive() {
|
||||
use hive_jobq_wire::WireNode as _;
|
||||
|
||||
let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
|
||||
hive_jobq::Graph::new(),
|
||||
hive_jobq::resources::ResourceTable::new(),
|
||||
));
|
||||
let ids =
|
||||
super::queue_hive_sender_mints(&sched, vec!["alpha".to_owned(), "beta".to_owned()])
|
||||
.expect("two jobs insert");
|
||||
assert_eq!(ids.len(), 2);
|
||||
let guard = sched
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut hives: Vec<String> = guard
|
||||
.graph()
|
||||
.nodes()
|
||||
.map(|n| {
|
||||
assert_eq!(n.payload.label(), "mint_hive_sender_token");
|
||||
let data = n.payload.data(n.id.get());
|
||||
assert!(data.get("agent").is_none(), "{data}");
|
||||
data["hive"].as_str().expect("hive is a string").to_owned()
|
||||
})
|
||||
.collect();
|
||||
hives.sort();
|
||||
assert_eq!(hives, ["alpha", "beta"]);
|
||||
}
|
||||
|
||||
/// The hive forge-token pass's queueing: one node per hive, naming the
|
||||
/// hive rather than an agent.
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -172,7 +172,7 @@ async fn main() -> Result<()> {
|
|||
hive_jobq::resources::ResourceTable::new(),
|
||||
)));
|
||||
let hives = load_hives();
|
||||
spawn_matrix_account_backfill(&jobq, deps.matrix_homeserver.clone(), &hives, auth.clone());
|
||||
spawn_matrix_account_backfill(&jobq, deps.matrix_homeserver.clone(), auth.clone());
|
||||
spawn_jobq_worker(Arc::clone(&jobq), deps);
|
||||
// Bound to a named variable, not `_` — dropping the provider stops its
|
||||
// `PeriodicReader`, so it must live as long as `main` does (which it
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@
|
|||
//! the swarm's homeserver, minted with the swarm's appservice token and stored
|
||||
//! where the agent's daemon reads it. See docs/4-implementation/swarm/credentials-reference.md for why
|
||||
//! `main` is the one name [`put_matrix_account`] refuses to write.
|
||||
//! [`hive_sender`] mints each **hive's** sender account the same way, sharing this module's helpers with [`agent_token`].
|
||||
//! [`operator_admin`] makes authelia's operators homeserver admins with the same
|
||||
//! appservice token.
|
||||
//! [`recovery`] mints the passphrase that opens each account's server-side key
|
||||
|
|
@ -38,7 +37,6 @@ use crate::routes::agents::swarm_hive;
|
|||
use crate::routes::{AppState, error_problem};
|
||||
|
||||
pub mod agent_token;
|
||||
pub mod hive_sender;
|
||||
pub mod operator_admin;
|
||||
pub mod recovery;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,195 +0,0 @@
|
|||
//! Each hive's sender account, `@hive-<hive>:`, on the swarm's homeserver:
|
||||
//! created here with the **swarm's** appservice token and stored at
|
||||
//! `swarm/hives/<hive>/matrix/sender-token`, where hive-c0re's matrix sweep
|
||||
//! reads it under the hive's own store identity.
|
||||
//!
|
||||
//! Runs for every hive in the directory, local or remote, and decides the
|
||||
//! same way [`super::agent_token`] does: a stored token that `whoami`
|
||||
//! confirms as `@hive-<hive>:` is kept, so this writes only when the path is
|
||||
//! empty or its token is dead.
|
||||
//!
|
||||
//! This is the only writer of that path: hive-c0re never mints, it takes
|
||||
//! whatever the store holds.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use swarm_matrix_client as homeserver;
|
||||
use swarm_secret_client::matrix;
|
||||
|
||||
use super::agent_token::{Decision, Observed, appservice_token, classify, mint_at, plan, probe_at};
|
||||
|
||||
/// How often [`spawn`] re-checks every hive's sender token.
|
||||
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
|
||||
|
||||
/// Decide what to do about `hive`'s sender token from what the stored one is.
|
||||
fn classify_hive(hive: &str, probe: &super::agent_token::Probe) -> Decision {
|
||||
classify(&matrix::hive_localpart(hive), probe)
|
||||
}
|
||||
|
||||
/// Make sure `hive`'s sender account holds a live token in the store,
|
||||
/// creating the account or logging in to it when it does not. The whole job
|
||||
/// of the `MintHiveSenderToken` node.
|
||||
///
|
||||
/// # Errors
|
||||
/// When the store or the homeserver refuses a step, the swarm appservice
|
||||
/// token has not been published, or the new token authenticates as someone
|
||||
/// else.
|
||||
pub async fn ensure_hive_sender_token(base: &str, hive: &str) -> Result<()> {
|
||||
let store = crate::store::connect()
|
||||
.await
|
||||
.context("logging in to the swarm secret store")?;
|
||||
let as_token = appservice_token(&store).await?;
|
||||
let http = homeserver::client()?;
|
||||
let path = matrix::sender_token_path(hive)?;
|
||||
let reason = match classify_hive(hive, &probe_at(&store, &http, base, &path).await?) {
|
||||
Decision::Keep => {
|
||||
tracing::debug!(hive, "hive sender token is current; left as it is");
|
||||
return Ok(());
|
||||
}
|
||||
Decision::Mint(reason) => reason,
|
||||
};
|
||||
let localpart = matrix::hive_localpart(hive);
|
||||
mint_at(&store, &http, base, &localpart, &as_token, &path).await?;
|
||||
tracing::info!(hive, ?reason, %path, "hive sender token minted and stored");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// One pass: every hive in `hives`, observed.
|
||||
///
|
||||
/// Fails as a whole when the swarm appservice token is not published, for
|
||||
/// the reason `agent_token`'s pass does.
|
||||
async fn observe_all(base: &str, hives: &[String]) -> Result<Vec<(String, Observed)>> {
|
||||
let store = crate::store::connect()
|
||||
.await
|
||||
.context("logging in to the swarm secret store")?;
|
||||
appservice_token(&store).await?;
|
||||
let http = homeserver::client()?;
|
||||
let mut observed = Vec::with_capacity(hives.len());
|
||||
for hive in hives {
|
||||
let o = match matrix::sender_token_path(hive) {
|
||||
Ok(path) => match probe_at(&store, &http, base, &path).await {
|
||||
Ok(p) => Observed::Decided(classify_hive(hive, &p)),
|
||||
Err(e) => {
|
||||
tracing::warn!(hive, error = %format!("{e:#}"), "hive sender token: read failed");
|
||||
Observed::Unknown
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
tracing::warn!(hive, error = %e, "hive sender token: the hive name forms no store path");
|
||||
Observed::Unknown
|
||||
}
|
||||
};
|
||||
observed.push((hive.clone(), o));
|
||||
}
|
||||
Ok(observed)
|
||||
}
|
||||
|
||||
/// Check every hive's sender token now and every [`RECONCILE_INTERVAL`]
|
||||
/// after, and hand the hives that need one to `enqueue`, which inserts a
|
||||
/// `MintHiveSenderToken` node for each.
|
||||
///
|
||||
/// The roster is the controller's hive directory. A pass that fails is
|
||||
/// logged and retried on the next tick; it never stops the daemon.
|
||||
pub fn spawn(base: Arc<str>, hives: Vec<String>, enqueue: impl Fn(Vec<String>) + Send + 'static) {
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
match observe_all(&base, &hives).await {
|
||||
Ok(observed) => {
|
||||
let minting = plan(&observed);
|
||||
if minting.is_empty() {
|
||||
tracing::debug!(
|
||||
checked = observed.len(),
|
||||
"hive sender tokens: all current"
|
||||
);
|
||||
} else {
|
||||
tracing::info!(
|
||||
checked = observed.len(),
|
||||
minting = minting.len(),
|
||||
"hive sender tokens: queueing mints"
|
||||
);
|
||||
enqueue(minting);
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
retry_in_s = RECONCILE_INTERVAL.as_secs(),
|
||||
"hive sender tokens: pass failed; retrying next tick"
|
||||
),
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::agent_token::{MintReason, Probe};
|
||||
use super::*;
|
||||
use swarm_matrix_client::Whoami;
|
||||
|
||||
fn user(id: &str) -> Probe {
|
||||
Probe::Whoami(Whoami::User(id.to_owned()))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_live_token_for_the_hive_account_is_kept() {
|
||||
assert_eq!(
|
||||
classify_hive("pr1ma", &user("@hive-pr1ma:t.local")),
|
||||
Decision::Keep
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nothing_stored_mints() {
|
||||
// A hive whose homeserver is remote: nothing has ever written its path.
|
||||
assert_eq!(
|
||||
classify_hive("pr1ma", &Probe::NotStored),
|
||||
Decision::Mint(MintReason::NotStored)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_dead_token_mints() {
|
||||
// A token the homeserver revoked, or a device a manual login replaced.
|
||||
assert_eq!(
|
||||
classify_hive("pr1ma", &Probe::Whoami(Whoami::UnknownToken)),
|
||||
Decision::Mint(MintReason::Revoked)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_token_for_another_account_mints() {
|
||||
// The bare hive name is an agent's localpart, not the hive's account,
|
||||
// and another hive's account is not this one's.
|
||||
for other in ["@pr1ma:t.local", "@hive-beta:t.local", "@hive:t.local"] {
|
||||
assert_eq!(
|
||||
classify_hive("pr1ma", &user(other)),
|
||||
Decision::Mint(MintReason::OtherUser),
|
||||
"{other}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_outage_plans_nothing() {
|
||||
// A mint replaces the token the hive is running on, so a pass that
|
||||
// could not read must not mint for every hive.
|
||||
let observed = [
|
||||
("alpha".to_owned(), Observed::Unknown),
|
||||
("beta".to_owned(), Observed::Unknown),
|
||||
];
|
||||
assert!(plan(&observed).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_published_path_is_the_one_the_hive_reads() {
|
||||
// hive-c0re's `stored_sender_token` resolves this path; the literal is
|
||||
// what the bao grant names.
|
||||
assert_eq!(
|
||||
matrix::sender_token_path("pr1ma").expect("a plain name is legal"),
|
||||
"swarm/hives/pr1ma/matrix/sender-token"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -21,8 +21,7 @@ use utoipa_axum::{router::OpenApiRouter, routes};
|
|||
use crate::jobq::{
|
||||
Late, SwarmNodeKind, SwarmResourceKind, queue_agent_renewals, queue_builder_forge_token_mint,
|
||||
queue_ci_metrics_forge_token_mint, queue_classifier_forge_token_mint, queue_docs_grants,
|
||||
queue_forge_token_mints, queue_hive_forge_token_mints, queue_hive_sender_mints,
|
||||
queue_matrix_account_mints,
|
||||
queue_forge_token_mints, queue_hive_forge_token_mints, queue_matrix_account_mints,
|
||||
};
|
||||
use crate::routes::hives::{HiveEntry, ServiceLink};
|
||||
use crate::routes::{self, AppState, agents, config_prs, forge_admin, hives, jobq_graph};
|
||||
|
|
@ -229,13 +228,12 @@ pub(crate) fn configured_matrix_homeserver() -> Option<Arc<str>> {
|
|||
}
|
||||
|
||||
/// Start the recovery passphrase pass, which needs only the store; the
|
||||
/// matrix-account backfill and the hive sender-token pass when a homeserver is
|
||||
/// configured; and the operator admin pass when the bridge is too. Lifted out
|
||||
/// of `main` for `clippy::too_many_lines`.
|
||||
/// matrix-account backfill when a homeserver is configured; and the operator
|
||||
/// admin pass when the bridge is too. Lifted out of `main` for
|
||||
/// `clippy::too_many_lines`.
|
||||
pub(crate) fn spawn_matrix_account_backfill(
|
||||
jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
||||
homeserver: Option<Arc<str>>,
|
||||
hives: &[HiveEntry],
|
||||
auth: Option<Arc<auth::AuthBridge>>,
|
||||
) {
|
||||
matrix_account::recovery::spawn();
|
||||
|
|
@ -246,18 +244,11 @@ pub(crate) fn spawn_matrix_account_backfill(
|
|||
matrix_account::operator_admin::spawn(bridge, Arc::clone(&base));
|
||||
}
|
||||
let sched = Arc::clone(jobq);
|
||||
matrix_account::agent_token::spawn(Arc::clone(&base), move |agents| {
|
||||
matrix_account::agent_token::spawn(base, move |agents| {
|
||||
if let Err(e) = queue_matrix_account_mints(&sched, agents) {
|
||||
tracing::warn!(error = %format!("{e:#}"), "agent matrix accounts: queueing failed");
|
||||
}
|
||||
});
|
||||
let sched = Arc::clone(jobq);
|
||||
let hive_names = hives.iter().map(|h| h.name.clone()).collect();
|
||||
matrix_account::hive_sender::spawn(base, hive_names, move |hives| {
|
||||
if let Err(e) = queue_hive_sender_mints(&sched, hives) {
|
||||
tracing::warn!(error = %format!("{e:#}"), "hive sender tokens: queueing failed");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Start the forge's periodic passes — the swarm-wide objects, the runner
|
||||
|
|
|
|||
Loading…
Reference in a new issue