diff --git a/Cargo.lock b/Cargo.lock index 6571d5b5..c4c5b8ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4588,8 +4588,10 @@ dependencies = [ "futures-util", "hive-jobq", "hive-jobq-wire", + "reqwest 0.13.1", "serde", "serde_json", + "swarm-authelia-bridge-sock", "swarm-queue-client", "tokio", "tracing", diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index cb5f2f54..b94ebe9f 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -26,8 +26,12 @@ futures-util.workspace = true # to serve. hive-jobq.workspace = true hive-jobq-wire.workspace = true +# `auth`'s bridge client — same crate the bridge itself uses to define the +# request/response shape, so the two ends cannot drift. +reqwest.workspace = true serde.workspace = true serde_json.workspace = true +swarm-authelia-bridge-sock.workspace = true # The queue connect (token mint + auth callback + reconnect) is shared with # every other participant - a hive publishing its own status runs the same # code with a different client id. Two copies of credential handling is one diff --git a/swarm-controller/src/auth.rs b/swarm-controller/src/auth.rs new file mode 100644 index 00000000..fcc759ba --- /dev/null +++ b/swarm-controller/src/auth.rs @@ -0,0 +1,105 @@ +//! Client for `swarm-authelia-bridge` — the only writer of the swarm's +//! authelia users database (see that crate's README for why this daemon +//! cannot write it directly). +//! +//! Authenticated with THIS daemon's own queue OIDC identity +//! (`SWARM_CONTROLLER_OIDC_*`, the same one `swarm-queue-client` mints for +//! the queue connection) — "one identity per principal" means a second +//! op that needs to prove who this process is reuses the identity it +//! already has rather than provisioning a new one. A fresh token is +//! minted per call for the same reason the queue client mints one per +//! connection attempt: no window in which this process holds a token +//! that outlives its intended use. +//! +//! `None` when this deployment did not wire a bridge up — the bridge only +//! exists on hosts that also run `swarm-authelia`, so a controller split +//! from it simply has no identity-creation capability yet +//! (`SwarmNodeKind::CreateIdentity` fails such a job explicitly rather +//! than this module papering over the gap). + +use anyhow::{Context, Result}; +use swarm_authelia_bridge_sock::{BridgeRequest, BridgeResponse}; + +/// A configured connection to `swarm-authelia-bridge`. +#[derive(Clone)] +pub struct AuthBridge { + http: reqwest::Client, + base_url: String, + queue_cfg: swarm_queue_client::QueueConfig, +} + +impl AuthBridge { + /// Read `SWARM_CONTROLLER_AUTH_BRIDGE_URL`; `Ok(None)` when unset. + /// + /// The queue identity (`SWARM_CONTROLLER_OIDC_*`) is not optional once + /// the bridge URL is set: the nix module sets `queueEnv` unconditionally + /// for every controller (the queue is required, not just co-located + /// service), so a bridge URL with no queue identity to authenticate + /// with is a deployment bug, not an absent-feature case — hence the + /// hard error rather than a second `None`. + pub fn from_env() -> Result> { + let Ok(base_url) = std::env::var("SWARM_CONTROLLER_AUTH_BRIDGE_URL") else { + return Ok(None); + }; + let queue_cfg = swarm_queue_client::QueueConfig::from_env("SWARM_CONTROLLER") + .context("reading the queue OIDC identity the auth bridge authenticates with")? + .ok_or_else(|| { + anyhow::anyhow!( + "SWARM_CONTROLLER_AUTH_BRIDGE_URL is set but SWARM_CONTROLLER_OIDC_* is \ + not — the bridge is authenticated with this daemon's queue identity, so \ + that identity must exist first" + ) + })?; + Ok(Some(Self { + http: reqwest::Client::new(), + base_url, + queue_cfg, + })) + } + + /// Idempotently ensure `name` exists as an authelia subject. + pub async fn ensure_agent_identity(&self, name: &str) -> Result { + // `mint_token_for` builds its own HTTP client (trusting the queue's + // configured CA, if any) — deliberately not `self.http`, which is + // the bridge's own client and has nothing to do with authelia's + // token endpoint's trust anchors. + let token = swarm_queue_client::mint_token_for(&self.queue_cfg) + .await + .context("minting a bearer token for swarm-authelia-bridge")?; + + let response = self + .http + .post(format!("{}/requests", self.base_url)) + .bearer_auth(token) + .json(&BridgeRequest::EnsureAgentIdentity { + name: name.to_owned(), + }) + .send() + .await + .context("calling swarm-authelia-bridge")?; + + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + if !status.is_success() { + anyhow::bail!("swarm-authelia-bridge refused the request ({status}): {body}"); + } + + serde_json::from_str(&body).context("parsing swarm-authelia-bridge's response") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The common case: no bridge wired up for this deployment. + #[test] + fn absent_url_is_not_an_error() { + assert!(std::env::var("SWARM_CONTROLLER_AUTH_BRIDGE_URL").is_err()); + assert!( + AuthBridge::from_env() + .expect("absent is not an error") + .is_none() + ); + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 5be49dde..d65c1e03 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -32,31 +32,38 @@ use serde::{Deserialize, Serialize}; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; +mod auth; mod status; -/// Placeholder node payload for the swarm-level job graph — uninhabited on -/// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource` -/// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't -/// land on both crates. The *scheduler loop* below is real and running -/// (`spawn_jobq_worker`, mirroring `hive-c0re/src/job_queue/scheduler.rs`'s -/// `run_worker`) — what's still missing is a real job to give it: no -/// variant exists yet, so nothing is ever inserted into the graph and -/// `claim_next` always returns `None`. Giving this real variants (starting -/// with `CreateRepo`) is the next slice, landing together with -/// `swarm-controller::forge`, the client those nodes will call. `WireNode` -/// is trivially satisfiable on an empty enum (`match *self {}`), so the -/// wire machinery below is real and typechecked today, with nothing yet to -/// put in it. +/// Node payload for the swarm-level job graph. Named `Swarm*` rather than +/// the bare `NodeKind`/`Resource` `hive-c0re::job_queue::model` already +/// uses, so a grep for either doesn't land on both crates. +/// +/// `CreateIdentity` is the first real variant — "the minimal shape for +/// agent creation is creating the identity and wiring that in" (the design +/// thread's own framing for why this landed before the forge-node work a +/// standalone `CreateRepo` variant would have started with). Its only +/// effect is calling `swarm-controller::auth`, which calls +/// `swarm-authelia-bridge`; nothing forge- or deploy-shaped happens yet. #[derive(Clone, Debug)] -enum SwarmNodeKind {} +enum SwarmNodeKind { + /// Ensure `agent` exists as an authelia subject at the swarm level. + CreateIdentity { agent: String }, +} impl hive_jobq_wire::WireNode for SwarmNodeKind { fn label(&self) -> String { - match *self {} + match self { + SwarmNodeKind::CreateIdentity { .. } => "create_identity".to_owned(), + } } fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value { - match *self {} + match self { + SwarmNodeKind::CreateIdentity { agent } => { + serde_json::json!({ "agent": agent }) + } + } } } @@ -73,20 +80,33 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind { /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind` -/// variant turns into a real effect. Trivially exhaustive today -/// (`match kind {}`) because the enum has no variants yet; the first -/// real arm (`CreateRepo`, calling `swarm-controller::forge`) lands -/// alongside that variant, not before. +/// variant turns into a real effect. +/// +/// `auth` is `None` on a host that runs a controller split from +/// `swarm-authelia` (no bridge configured there) — `CreateIdentity` fails +/// explicitly in that case rather than this fn papering over a +/// misconfigured deployment. async fn run_swarm_node( _id: hive_jobq::NodeId, kind: SwarmNodeKind, builder: hive_jobq::builder::JobBuilder, + auth: Option>, ) -> ( hive_jobq::builder::JobBuilder, hive_jobq::scheduler::Outcome, ) { - let _ = builder; - match kind {} + let outcome = match kind { + SwarmNodeKind::CreateIdentity { agent } => match auth { + None => hive_jobq::scheduler::Outcome::Failed( + "no swarm-authelia-bridge is configured on this host".to_owned(), + ), + Some(bridge) => match bridge.ensure_agent_identity(&agent).await { + Ok(_) => hive_jobq::scheduler::Outcome::Done, + Err(e) => hive_jobq::scheduler::Outcome::Failed(format!("{e:#}")), + }, + }, + }; + (builder, outcome) } /// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/ @@ -102,14 +122,23 @@ async fn run_swarm_node( /// every in-flight HTTP request does. /// /// Cheap to run with an empty graph: `claim_next` on a graph nothing was -/// ever inserted into just returns `None` every poll, so this is a -/// harmless idle loop until the first real node kind exists. +/// ever inserted into just returns `None` every poll. +/// +/// `auth` is cloned per iteration (an `Arc` clone, not a reconnect) and +/// moved into the closure `claim_next` takes ownership of — `run_swarm_node` +/// needs its own owned copy since the claimed future may outlive this loop +/// iteration. fn spawn_jobq_worker( sched: Arc>>, + auth: Option>, ) { tokio::spawn(async move { loop { - let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, run_swarm_node); + let auth = auth.clone(); + let runner = + hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { + run_swarm_node(id, kind, builder, auth) + }); match runner { Some(runner) => { tokio::spawn(async move { @@ -174,6 +203,7 @@ fn socket_path() -> PathBuf { (name = "hives", description = "the swarm's hive directory"), (name = "links", description = "swarm service quick links"), (name = "jobq", description = "the swarm-level job graph"), + (name = "agents", description = "creating agent identities at swarm level"), ) )] struct ApiDoc; @@ -222,6 +252,13 @@ struct AppState { /// is synchronous (no `.await` while held). Always present, never /// gated on the swarm queue: this is process state, not something /// read over the network. + /// **Not** consulted by `POST /api/agents` — that endpoint only ever + /// queues the job (see [`create_agent`]); whether a bridge is + /// configured is `run_swarm_node`'s concern (it holds its own clone, + /// handed to it by `spawn_jobq_worker`), not this handler's. A request + /// still queues cleanly on a bridge-less host, then fails loud once + /// claimed — same "queue now, fail per-job" shape as an unreachable + /// swarm queue. jobq: Arc>>, } @@ -366,6 +403,62 @@ async fn get_hives_status( } } +/// Body of `POST /api/agents` — the agent name to create a swarm-level +/// identity for. No other fields: this endpoint is deliberately narrow — +/// "identity in authelia only, no forge or deploy yet" — so it asks for +/// nothing a `CreateIdentity` node doesn't use. +#[derive(Clone, Debug, Deserialize, ToSchema)] +struct CreateAgentRequest { + name: String, +} + +/// Where the queued job landed — a caller polls `/api/jobq/graph` (or +/// `?states=`) with this id to watch it settle, same as every other job +/// kind this daemon will ever queue. +#[derive(Clone, Debug, Serialize, ToSchema)] +struct CreateAgentResponse { + node_id: u64, +} + +/// Queue a `CreateIdentity` job for `name`. Returns as soon as the node is +/// inserted — **not** once the identity exists; `run_swarm_node` does that +/// work asynchronously off the scheduler loop already running +/// (`spawn_jobq_worker`), same as every other node kind. This is also the +/// first genuine non-test caller `SwarmNodeKind::CreateIdentity` has: the +/// node kind's `dead_code` bound was the whole reason this endpoint had to +/// land in the same change as the variant, not as a follow-up. +#[utoipa::path( + post, + path = "/api/agents", + request_body = CreateAgentRequest, + responses( + (status = 200, description = "job queued", body = CreateAgentResponse), + (status = 500, description = "the job could not be queued", body = String), + ), + tag = "agents" +)] +async fn create_agent( + State(state): State, + Json(req): Json, +) -> Result, (axum::http::StatusCode, String)> { + let mut sched = state + .jobq + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let ids = sched + .insert_job(None, |b| { + vec![ + b.node(SwarmNodeKind::CreateIdentity { agent: req.name }) + .guid(), + ] + }) + .map_err(|e| (axum::http::StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + let [id] = ids[..] else { + unreachable!("exactly one handle was asked for"); + }; + Ok(Json(CreateAgentResponse { node_id: id.get() })) +} + /// Query params for `GET /api/jobq/graph` — `?states=` narrows to root /// groups in the named states, same shape `hive_jobq_wire::parse_states` /// parses. @@ -501,11 +594,23 @@ async fn main() -> Result<()> { }, }; + // Same "not fatal, log and carry on" shape as the queue connect above: + // a controller with no bridge wired up still serves everything else, + // and `run_swarm_node` gives an honest per-job failure instead of this + // fn refusing to start. + let auth = match auth::AuthBridge::from_env() { + Ok(auth) => auth.map(Arc::new), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "swarm-authelia-bridge misconfigured; agent identity creation is off"); + None + } + }; + let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ))); - spawn_jobq_worker(Arc::clone(&jobq)); + spawn_jobq_worker(Arc::clone(&jobq), auth.clone()); let state = AppState { hives: Arc::new(load_hives()), @@ -521,6 +626,7 @@ async fn main() -> Result<()> { .routes(routes!(get_links)) .routes(routes!(get_jobq_graph)) .routes(routes!(get_jobq_rollup)) + .routes(routes!(create_agent)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from // the nix store (see the module doc comment above). `api` is diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index c95c6cc3..862d72a6 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -321,6 +321,45 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result Result { + let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10)); + if let Some(path) = &cfg.ca_file { + let pem = std::fs::read(path).map_err(|source| Error::CaFile { + path: path.display().to_string(), + source, + })?; + let cert = reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse { + path: path.display().to_string(), + source, + })?; + builder = builder.add_root_certificate(cert); + } + builder.build().map_err(Error::HttpClient) +} + +/// Mint a fresh token for `cfg`'s identity and hand back just the string — +/// no caching, a fresh HTTP client per call. +/// +/// Public because the queue connection is not the only thing this identity +/// authenticates: "one identity per principal" means a caller that already +/// holds a [`QueueConfig`] for its queue connection authenticates anywhere +/// else it needs to prove who it is from the exact same client, rather than +/// a second identity being provisioned per destination. No caching here +/// unlike [`connect`]'s callback: that one exists because `async-nats` reruns +/// its callback per reconnect *attempt*, a hot path this isn't — a caller +/// outside that loop (e.g. `swarm-controller::auth`'s bridge client) mints +/// per call, same as this crate did before the reconnect-storm fix added the +/// cache. +pub async fn mint_token_for(cfg: &QueueConfig) -> Result { + let http = build_http_client(cfg)?; + Ok(mint_token(&http, cfg).await?.token) +} + /// Fail fast unless the client is actually connected. /// /// **Call this before every `JetStream` request.** `retry_on_initial_connect` @@ -355,36 +394,14 @@ pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> { /// token expires, the controller keeps serving, its status data quietly stops /// updating, and nothing says so until someone reads a dashboard. pub async fn connect(cfg: QueueConfig) -> Result { - // A timeout, because this client runs INSIDE the auth callback: a token - // endpoint that accepts the connection and then never answers would hang - // the callback, and with it the connection attempt that invoked it, with - // no retry and nothing in the log to say why. Failing fast lets + // Trusts `cfg.ca_file` when set (the swarm's own CA, when the token + // endpoint is signed by it) — see `build_http_client`. A timeout too, + // because this client runs INSIDE the auth callback: a token endpoint + // that accepts the connection and then never answers would hang the + // callback, and with it the connection attempt that invoked it, with no + // retry and nothing in the log to say why. Failing fast lets // `async-nats` do what it already does well — back off and try again. - // 10s is generous for a form POST to a local IdP. - let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10)); - - // The swarm's own CA, when the token endpoint is signed by it. ADDED, not - // substituted: `add_root_certificate` extends the default set rather than - // replacing it, so a swarm can front authelia publicly and still have - // this work. - // - // Failing here rather than falling back to the platform roots is the - // point — an operator who named a CA file wants that anchor, and a - // silent fallback would turn their typo into `UnknownIssuer` five layers - // away, inside an auth callback, on a retry loop. - if let Some(path) = &cfg.ca_file { - let pem = std::fs::read(path).map_err(|source| Error::CaFile { - path: path.display().to_string(), - source, - })?; - let cert = reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse { - path: path.display().to_string(), - source, - })?; - builder = builder.add_root_certificate(cert); - } - - let http = builder.build().map_err(Error::HttpClient)?; + let http = build_http_client(&cfg)?; let url = cfg.url.clone(); // Shared across every invocation of the callback below, which is the