swarm UI: log an agent into claude from its card; the credentials go to bao and the agent picks them up

A new swarm service, `swarm-claude-login`, in its own container on the host
that sets `deploy.claude-login.enable` (defaulted on by `allSwarmServices`
where the host has the minter's store leaf, like the review classifier).

- swarm-ui: an agent card showing "login needed" gets a "log in" button.
  Its dialog starts the login, shows the URL claude printed, and takes the
  pasted code.
- swarm-controller: `POST/DELETE /api/agents/{name}/claude-login` and
  `POST /api/agents/{name}/claude-login/code`, behind the swarm UI's
  authelia-gated `/api/`, relay the steps to the minter as requests on
  `$SWARM.claude-login.request`.
- swarm-claude-login: runs `claude auth login` over piped stdio with HOME
  and CLAUDE_CONFIG_DIR in a per-login scratch dir, scrapes the first
  https URL, writes the code to stdin, and stores the `.credentials.json`
  at `swarm/agents/<agent>/claude-credentials` with a KV v2 write, a new
  version over any earlier one. The scratch dir goes on every exit path; a
  login waits 15 minutes for its code.
- hive-agent: while parked in `wait_for_login`, reads that leaf every 30 s
  under the agent's own cert and installs it only when its KV version is
  newer than the one recorded in `<harness>/claude-credentials-version`.
  An equal or older version is never put back, because claude rotates the
  refresh token and the stored copy is stale after the first refresh.
- swarm-secret-client: `claude_login` paths and `SecretStore::read_versioned`.
- Credentials: a `claude-login` store leaf minted by glue-bao-tls, a
  `swarm-claude-login` policy (create/update on
  `swarm/agents/+/claude-credentials`, read on its own queue client
  secret), an authelia machine client, and a `--claude-login-client`
  grant in the queue's auth-callout responder. The unit runs as a static
  system user.

The hive's own web-UI login is unchanged.

Refs #5268

swarm-claude-login/src/main.rs

@@ -0,0 +1,192 @@
+//! The claude login minter.
+//!
+//! Connects to the swarm queue as its own OIDC client, takes an operator's
+//! login steps from swarm-controller on
+//! [`swarm_queue_client::claude_login::SUBJECT`], runs `claude auth login`
+//! for the agent named ([`login`]), and writes the credentials it produces to
+//! that agent's store leaf, a new version over any stored before. The agent's
+//! harness takes a version newer than the one it last installed.
+//!
+//! The store identity comes from the `BAO_*` environment. The queue client
+//! secret is read at start and again when authelia refuses it; the store is
+//! logged in to afresh for each write.
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{Context, Result, bail};
+use futures_util::StreamExt as _;
+use swarm_queue_client::claude_login::{Reply, Request, SESSION_DEADLINE, STEP_DEADLINE, SUBJECT};
+use swarm_queue_client::stored_secret::StoredSecret;
+use swarm_queue_client::{ClientSecret, QueueConfig};
+use swarm_secret_client::claude_login::{Credentials, client_secret_path, credentials_path};
+
+mod login;
+
+/// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
+/// writes it, matching the leaf's subject.
+const CERT_ROLE: &str = "swarm-claude-login";
+
+/// The environment prefix `swarm-claude-login.nix` sets.
+const PREFIX: &str = "SWARM_CLAUDE_LOGIN";
+
+/// Seconds to wait after each failed read of the queue client secret. A
+/// longer outage fails the start, and the unit's `Restart=` tries again.
+const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30];
+
+/// The agent's store leaf, written under this service's certificate.
+struct AgentLeaf;
+
+impl login::CredentialSink for AgentLeaf {
+    async fn put(&self, agent: &str, credentials: &Credentials) -> Result<()> {
+        let path = credentials_path(agent)?;
+        let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
+            .await
+            .context("logging in to the secret store")?;
+        // `write`, not `create`: a relogin replaces what an earlier one
+        // stored.
+        store
+            .write(&path, credentials)
+            .await
+            .with_context(|| format!("writing {path}"))
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    hive_log::init();
+    swarm_queue_client::install_crypto_provider();
+    let var = |name: &str| {
+        let full = format!("{PREFIX}_{name}");
+        std::env::var(&full).with_context(|| format!("{full} is unset"))
+    };
+    let client_id = var("OIDC_CLIENT_ID")?;
+    let logins = Arc::new(login::Logins::new(
+        PathBuf::from(var("CLAUDE")?),
+        std::env::temp_dir(),
+        STEP_DEADLINE,
+    ));
+    let secret = client_secret_with_retry(&client_id).await?;
+    let reread_id = client_id.clone();
+    let secret = StoredSecret::new(client_id.clone(), secret, move || {
+        let client_id = reread_id.clone();
+        async move { client_secret(&client_id).await.map_err(Into::into) }
+    });
+    let queue = swarm_queue_client::connect(QueueConfig {
+        url: var("NATS_URL")?,
+        token_endpoint: var("OIDC_TOKEN_ENDPOINT")?,
+        client_id,
+        client_secret: ClientSecret::Stored(secret),
+        ca_file: std::env::var(format!("{PREFIX}_OIDC_CA_FILE"))
+            .ok()
+            .map(PathBuf::from),
+    })
+    .await?;
+    // No queue group: a running login lives in this process's memory, so
+    // every step for it has to reach this process.
+    let mut requests = queue
+        .subscribe(SUBJECT)
+        .await
+        .with_context(|| format!("subscribing to {SUBJECT}"))?;
+    tracing::info!(subject = SUBJECT, "claude login minter: awaiting logins");
+
+    while let Some(msg) = requests.next().await {
+        let Some(reply_to) = msg.reply.clone() else {
+            tracing::warn!("a login step arrived with no reply subject; dropping it");
+            continue;
+        };
+        let queue = queue.clone();
+        let logins = Arc::clone(&logins);
+        tokio::spawn(async move {
+            let reply = handle(&msg.payload, &logins).await;
+            match serde_json::to_vec(&reply) {
+                Ok(body) => {
+                    if let Err(e) = queue.publish(reply_to, body.into()).await {
+                        tracing::warn!(error = %e, "publishing a login step's reply failed");
+                    }
+                }
+                Err(e) => tracing::warn!(error = %e, "encoding a login step's reply failed"),
+            }
+        });
+    }
+    bail!("the subscription to {SUBJECT} ended")
+}
+
+/// Decode one step, run it and log it. Never logs the code or the URL.
+async fn handle(payload: &[u8], logins: &Arc<login::Logins>) -> Reply {
+    let request: Request = match serde_json::from_slice(payload) {
+        Ok(r) => r,
+        Err(e) => {
+            tracing::warn!(error = %e, "an undecodable login step; answering with the error");
+            return Reply::Failed {
+                error: format!("the minter could not decode the request: {e}"),
+            };
+        }
+    };
+    let (agent, reply) = match request {
+        Request::Start { agent, operator } => {
+            tracing::info!(%agent, %operator, "starting a claude login");
+            let (reply, id) = logins.start(&agent).await;
+            if let Some(id) = id {
+                let logins = Arc::clone(logins);
+                let expiring = agent.clone();
+                tokio::spawn(async move {
+                    tokio::time::sleep(SESSION_DEADLINE).await;
+                    logins.expire(&expiring, id).await;
+                });
+            }
+            (agent, reply)
+        }
+        Request::Code { agent, code } => {
+            let reply = logins.code(&agent, &code, &AgentLeaf).await;
+            (agent, reply)
+        }
+        Request::Cancel { agent } => {
+            let reply = logins.cancel(&agent).await;
+            (agent, reply)
+        }
+    };
+    match &reply {
+        Reply::Started { .. } => tracing::info!(%agent, "claude login waiting for its code"),
+        Reply::Stored => tracing::info!(%agent, "claude credentials stored"),
+        Reply::Cancelled => tracing::info!(%agent, "claude login cancelled"),
+        Reply::Failed { error } => tracing::warn!(%agent, %error, "claude login step failed"),
+    }
+    reply
+}
+
+/// The queue client secret, read from the store, retried through a store that
+/// is restarting alongside this service.
+async fn client_secret_with_retry(client_id: &str) -> Result<String> {
+    for delay in RETRY_DELAYS_S.map(Duration::from_secs) {
+        match client_secret(client_id).await {
+            Ok(secret) => return Ok(secret),
+            Err(e) => {
+                tracing::warn!(error = %format!("{e:#}"), retry_in = ?delay, "reading the queue client secret failed");
+                tokio::time::sleep(delay).await;
+            }
+        }
+    }
+    client_secret(client_id).await
+}
+
+async fn client_secret(client_id: &str) -> Result<String> {
+    use swarm_secret_client::builder::ClientSecret;
+
+    let path = client_secret_path(client_id)?;
+    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
+        .await
+        .context("logging in to the secret store")?;
+    let stored: Option<ClientSecret> = store
+        .read_optional(&path)
+        .await
+        .with_context(|| format!("reading {path}"))?;
+    match stored {
+        Some(s) if !s.value.trim().is_empty() => Ok(s.value.trim().to_owned()),
+        _ => bail!(
+            "nothing at {path}: swarm-secret-publish, on the host that runs authelia, has not \
+             published this service's client secret"
+        ),
+    }
+}