hyperhive/swarm-nats-auth/src/main.rs

//! 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_util::StreamExt;

mod introspect;
mod policy;
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,

    /// Account an admitted client is placed in. Must name an entry in the
    /// server's own `accounts` block — in server-config mode the account is
    /// resolved by *name*, so a value the server does not know is a grant it
    /// refuses. The module passes its `clientAccount`; the default mirrors it.
    #[arg(long, default_value = "APP")]
    account: String,

    /// Path to this responder's own OIDC client secret.
    #[arg(long)]
    client_secret_file: PathBuf,

    /// Client-id prefix that marks a hive. `swarm-authelia.nix` mints one
    /// machine client per roster entry as `hive-<name>`, while the KV key is
    /// the bare `<name>` — this is the contract between the two, declared
    /// rather than inferred from the shape of an id.
    #[arg(long, default_value = "hive-")]
    hive_client_prefix: String,

    /// Client ids allowed to read every hive's status. Repeatable. The
    /// default is the swarm controller, which is the only reader that exists.
    #[arg(long = "reader-client", default_values_t = [String::from("swarm-controller")])]
    reader_clients: Vec<String>,

    /// Additional subjects a hive may publish to, with `{hive}` standing for
    /// its own name. Repeatable, empty by default.
    ///
    /// The extension point for a second stream published by the same
    /// `hive-<name>` identity — lifecycle notices, say. Without it, adding one
    /// means changing this responder; with it, a deployment says so and the
    /// subject still lands inside that hive's own namespace.
    #[arg(long = "hive-publish-subject")]
    hive_publish_subjects: Vec<String>,
}

/// 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")),
        )
        // This is a systemd-managed daemon — stdout always goes to journald,
        // never a human terminal, and journald doesn't strip ANSI escapes:
        // they land in victorialogs as raw byte-array spam otherwise.
        .with_ansi(false)
        .init();
    let args = Args::parse();

    let client_secret = read_secret(&args.client_secret_file)?;
    // The bucket is NOT a flag. `swarm_queue_client::status`'s own docs say
    // why: reader and writer must name the same bucket, and an option is a
    // way for two deployments to disagree about which one that is. This
    // responder is the third end that names it, so it takes the same
    // constant rather than a copy of the literal.
    // Fails the process rather than warning: a policy that cannot express a
    // per-hive namespace is not a policy this responder should run with, and
    // the queue's fail-closed state (no responder) is a legible outage where a
    // silently over-broad grant is not.
    let policy = policy::Policy::new(
        args.hive_client_prefix.clone(),
        swarm_queue_client::status::BUCKET.to_owned(),
        args.reader_clients.clone(),
        args.hive_publish_subjects.clone(),
    )?;
    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.
        //
        // The caller is an identity or nothing — see `introspect`'s module
        // docs. There is no "admitted, identity unknown" branch to write here
        // because there is no such value to receive.
        let caller = match &req.connect_opts.auth_token {
            Some(token) => introspect::identify_caller(
                &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");
                None
            }),
            None => None,
        };
        // Admission said who; the policy says what. A caller the `IdP`
        // vouches for but no rule matches is denied — see `policy`'s module
        // docs for why that is deny and not "connect with nothing".
        let permissions = caller.as_deref().and_then(|id| policy.permissions(id));
        if let (Some(id), None) = (caller.as_deref(), permissions.as_ref()) {
            // Loud, and the one case an operator has to be able to find: a
            // valid credential refused by our own policy. The alternative is
            // a client that authenticates fine and mysteriously cannot work.
            tracing::warn!(
                caller = %id,
                "authenticated client matches no policy rule; denying"
            );
        }
        // The client id is an identifier, not a credential, and it is the
        // only thing tying a connection in this log to a hive.
        tracing::info!(
            user_nkey = %req.user_nkey,
            server_id = %req.server_id.id,
            granted = permissions.is_some(),
            caller = caller.as_deref().unwrap_or("-"),
            "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 = match &permissions {
            Some(permissions) => respond::grant(
                &issuer,
                &args.account,
                &req.server_id.id,
                &req.user_nkey,
                permissions,
            ),
            None => 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")
}