swarm-controller: CreateIdentity node, auth-bridge client, POST /api/agents

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",

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

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<Option<Self>> {
+        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<BridgeResponse> {
+        // `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()
+        );
+    }
+}

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
+/// Node payload for the swarm-level job graph. Named `Swarm*` rather than
-/// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource`
+/// the bare `NodeKind`/`Resource` `hive-c0re::job_queue::model` already
-/// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't
+/// uses, so a grep for either doesn't land on both crates.
-/// land on both crates. The *scheduler loop* below is real and running
+///
-/// (`spawn_jobq_worker`, mirroring `hive-c0re/src/job_queue/scheduler.rs`'s
+/// `CreateIdentity` is the first real variant — "the minimal shape for
-/// `run_worker`) — what's still missing is a real job to give it: no
+/// agent creation is creating the identity and wiring that in" (the design
-/// variant exists yet, so nothing is ever inserted into the graph and
+/// thread's own framing for why this landed before the forge-node work a
-/// `claim_next` always returns `None`. Giving this real variants (starting
+/// standalone `CreateRepo` variant would have started with). Its only
-/// with `CreateRepo`) is the next slice, landing together with
+/// effect is calling `swarm-controller::auth`, which calls
-/// `swarm-controller::forge`, the client those nodes will call. `WireNode`
+/// `swarm-authelia-bridge`; nothing forge- or deploy-shaped happens yet.
-/// 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.
 #[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
+/// variant turns into a real effect.
-/// (`match kind {}`) because the enum has no variants yet; the first
+///
-/// real arm (`CreateRepo`, calling `swarm-controller::forge`) lands
+/// `auth` is `None` on a host that runs a controller split from
-/// alongside that variant, not before.
+/// `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<SwarmNodeKind, SwarmResourceKind>,
+    auth: Option<std::sync::Arc<auth::AuthBridge>>,
 ) -> (
     hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
     hive_jobq::scheduler::Outcome,
 ) {
-    let _ = builder;
+    let outcome = match kind {
-    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
+/// ever inserted into just returns `None` every poll.
-/// harmless idle loop until the first real node kind exists.
+///
+/// `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<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
+    auth: Option<Arc<auth::AuthBridge>>,
 ) {
     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<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
 }
 
@@ -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<AppState>,
+    Json(req): Json<CreateAgentRequest>,
+) -> Result<Json<CreateAgentResponse>, (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

swarm-queue-client/src/lib.rs

@@ -321,6 +321,45 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedT
     })
 }
 
+/// Build the HTTP client used to reach `cfg.token_endpoint`, trusting
+/// `cfg.ca_file` when set. Shared by [`connect`]'s auth callback and by
+/// [`mint_token_for`] — anything presenting this identity's credentials to
+/// the token endpoint needs the swarm's own CA trusted the same way, so the
+/// trust-anchor logic lives here once rather than once per caller.
+fn build_http_client(cfg: &QueueConfig) -> Result<reqwest::Client, Error> {
+    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<String, Error> {
+    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<async_nats::Client, Error> {
-    // A timeout, because this client runs INSIDE the auth callback: a token
+    // Trusts `cfg.ca_file` when set (the swarm's own CA, when the token
-    // endpoint that accepts the connection and then never answers would hang
+    // endpoint is signed by it) — see `build_http_client`. A timeout too,
-    // the callback, and with it the connection attempt that invoked it, with
+    // because this client runs INSIDE the auth callback: a token endpoint
-    // no retry and nothing in the log to say why. Failing fast lets
+    // 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 http = build_http_client(&cfg)?;
-    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 url = cfg.url.clone();
 
     // Shared across every invocation of the callback below, which is the