feat(swarm): the auth-callout responder (#3112 slice 2)

Slice 1 shipped the NATS container with an auth_callout block and no
responder, which is the fail-closed state: the server answers
auth_required and admits nobody. This crate is what lets it say yes.

Connects as the callout-exempt user by nkey (never by name - the server
refuses to start if that entry carries a username), subscribes to
$SYS.REQ.USER.AUTH, validates the presented bearer token against
authelia's introspection endpoint, and replies with a signed NATS user
JWT. A denial is a signed response carrying an error, never silence: a
server that hears nothing cannot tell a refusing responder from a dead
one, so staying quiet would turn every rejection into a timeout and hide
an outage inside what looks like ordinary denials.

Everything that is not an explicit active:true denies - network error,
timeout, non-2xx, unparseable body, no token at all. Those are exactly
the conditions under which an attacker would most like this to fall
open. The introspection budget is held under the server's own 2s
auth_callout timeout by a test, since the two numbers live in different
languages in different files.

nats-jwt mints the user JWT. It cannot mint the authorization_response
wrapper - its claim enum is closed and its claims carry no aud, which
the response needs so a reply cannot be replayed at another server in
the cluster - so that half is hand-written, and a test builds a user
token both ways and requires the bytes to match. That is the only
honest basis for trusting the hand-written path on the shape the crate
does not model.

async-nats is taken with default-features off: the default set carries
jetstream, kv, object-store, websockets and service, none of which a
callout responder speaks.

swarm-nats-auth/src/main.rs

@@ -0,0 +1,179 @@
+//! Auth-callout responder for the swarm's NATS queue.
+//!
+//! `nix/host-modules/swarm-nats.nix` configures `nats-server` with an
+//! `auth_callout` block and no responder, which is the fail-closed state: the
+//! server answers `"auth_required":true` and admits nobody. This binary is what
+//! makes it able to say *yes*.
+//!
+//! It connects as the one callout-exempt user (by nkey, never by name — the
+//! server refuses to start if that entry carries a username), subscribes to
+//! `$SYS.REQ.USER.AUTH`, validates the presented bearer token against
+//! authelia's introspection endpoint, and answers with a NATS user JWT signed
+//! by the account key. A rejection is answered explicitly: silence is
+//! indistinguishable from the responder being down, and the queue is the
+//! swarm's control path.
+//!
+//! # Secrets
+//!
+//! Every credential is taken as a **path**, never a value. Two reasons, both
+//! previously learned the hard way here: a value in nix config is rendered into
+//! the world-readable store, and a value in `argv` is readable by anyone via
+//! `/proc/<pid>/cmdline`, which is `0444`. Paths are not secrets, so passing
+//! them as flags is fine.
+
+use std::path::PathBuf;
+
+use anyhow::Context;
+use clap::Parser;
+use futures::StreamExt;
+
+mod introspect;
+mod request;
+mod respond;
+
+/// Subject the NATS server publishes authorization requests on.
+const AUTH_SUBJECT: &str = "$SYS.REQ.USER.AUTH";
+
+#[derive(Debug, Parser)]
+#[command(
+    name = "swarm-nats-auth",
+    about = "Auth-callout responder for the swarm NATS queue"
+)]
+struct Args {
+    /// NATS server to connect to.
+    #[arg(long, default_value = "nats://127.0.0.1:4222")]
+    nats_url: String,
+
+    /// Path to the seed of the callout-exempt user this responder connects as.
+    /// Its public half is `services.hyperhive.swarm.nats.calloutUserPublicKey`.
+    #[arg(long)]
+    user_seed_file: PathBuf,
+
+    /// Path to the account signing seed used to sign issued user JWTs. Its
+    /// public half is `services.hyperhive.swarm.nats.calloutIssuerPublicKey`.
+    #[arg(long)]
+    issuer_seed_file: PathBuf,
+
+    /// Authelia's OIDC introspection endpoint.
+    #[arg(long)]
+    introspection_url: String,
+
+    /// `OAuth2` client id this responder introspects as. Must match
+    /// `services.hyperhive.swarm.nats.clientId`, whose default this mirrors.
+    #[arg(long, default_value = "swarm-nats")]
+    client_id: String,
+
+    /// Path to this responder's own OIDC client secret.
+    #[arg(long)]
+    client_secret_file: PathBuf,
+}
+
+/// Read a secret file and strip surrounding whitespace.
+///
+/// The trim matters: an `echo`-created seed file ends in a newline, and an
+/// nkey seed with a trailing byte is not a seed — it fails at parse with a
+/// message about encoding rather than about the file, which sends you looking
+/// in the wrong place. The value is never logged, and the error deliberately
+/// names only the path.
+fn read_secret(path: &std::path::Path) -> anyhow::Result<String> {
+    let raw = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
+    let trimmed = raw.trim();
+    if trimmed.is_empty() {
+        anyhow::bail!("{} is empty", path.display());
+    }
+    Ok(trimmed.to_owned())
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+    let args = Args::parse();
+
+    let client_secret = read_secret(&args.client_secret_file)?;
+    let http = reqwest::Client::new();
+    let issuer = nkeys::KeyPair::from_seed(&read_secret(&args.issuer_seed_file)?)
+        .context("parse the account signing seed")?;
+    let user_seed = read_secret(&args.user_seed_file)?;
+    let client = async_nats::ConnectOptions::with_nkey(user_seed)
+        .name("swarm-nats-auth")
+        .connect(&args.nats_url)
+        .await
+        .with_context(|| format!("connect to {}", args.nats_url))?;
+    let mut requests = client
+        .subscribe(AUTH_SUBJECT)
+        .await
+        .with_context(|| format!("subscribe to {AUTH_SUBJECT}"))?;
+    tracing::info!(
+        nats_url = %args.nats_url,
+        subject = AUTH_SUBJECT,
+        "swarm-nats-auth: connected, awaiting authorization requests"
+    );
+
+    while let Some(msg) = requests.next().await {
+        // Decode failures are logged and dropped, never propagated: this loop
+        // is the swarm's login path, and exiting on one malformed payload
+        // would let any client take authentication down for everyone.
+        let req = match request::decode(&msg.payload) {
+            Ok(req) => req,
+            Err(e) => {
+                tracing::warn!(error = ?e, "undecodable auth request, ignoring");
+                continue;
+            }
+        };
+        // No token is a denial, not an error: an anonymous connect is a
+        // normal thing for a client to attempt and an abnormal thing to
+        // grant. Introspection is only reached once something was presented.
+        let granted = match &req.connect_opts.auth_token {
+            Some(token) => introspect::is_active(
+                &http,
+                &args.introspection_url,
+                &args.client_id,
+                &client_secret,
+                token,
+            )
+            .await
+            // An introspection that could not be *made* is a denial too. The
+            // failure modes of an HTTP call are exactly the conditions under
+            // which an attacker would most like this to fall open.
+            .unwrap_or_else(|e| {
+                tracing::warn!(error = ?e, "introspection failed; denying");
+                false
+            }),
+            None => false,
+        };
+        tracing::info!(
+            user_nkey = %req.user_nkey,
+            server_id = %req.server_id.id,
+            granted,
+            "auth request"
+        );
+
+        // Always reply, including on a denial. A server that hears nothing
+        // cannot tell a refusing responder from a dead one, so silence turns
+        // every rejection into a 2s timeout and hides an outage inside what
+        // looks like ordinary denials.
+        let Some(reply_to) = msg.reply.clone() else {
+            tracing::warn!("auth request had no reply subject; dropping");
+            continue;
+        };
+        let token = if granted {
+            respond::grant(
+                &issuer,
+                &issuer.public_key(),
+                &req.server_id.id,
+                &req.user_nkey,
+            )
+        } else {
+            respond::deny(&issuer, &req.server_id.id, &req.user_nkey)
+        };
+        if let Err(e) = client.publish(reply_to, token.into()).await {
+            tracing::warn!(error = ?e, "failed to publish auth response");
+        }
+    }
+    anyhow::bail!("subscription to {AUTH_SUBJECT} ended")
+}