hyperhive/swarm-queue-client/src/lib.rs

//! Connecting to the swarm message queue as an authenticated client.
//!
//! Shared by every process that needs the queue — the swarm controller reads
//! hive status out of it, a hive publishes its own status into it — because
//! the *connect* is identical for all of them and only the use differs.
//! Duplicating it per binary would put credential handling in two places, and
//! a token-refresh fix would then have to be found twice.
//!
//! The queue admits every non-responder client through `auth_callout`: a
//! client presents a token at CONNECT, the callout responder introspects it
//! against authelia and mints a user JWT if it is good. So each participant is
//! an ordinary client that needs an identity of its own — the controller is
//! not a hive, and the per-hive clients issued from the roster are not its to
//! use.
//!
//! Two things about that shape drive everything here:
//!
//! - **A token expires.** Authelia issues `client_credentials` access tokens
//!   with `expires_in: 3599`. Authentication happens at CONNECT, so a
//!   long-lived connection is fine — but a *reconnect* an hour later needs a
//!   token that was minted an hour later.
//! - **`async-nats` re-runs an auth callback per connection attempt** (it is
//!   handed that attempt's nonce). So the refresh belongs in the callback and
//!   not in a timer: there is no window in which the client holds a token it
//!   minted for a previous connection.
//!
//! The alternative — mint once, pass a static `auth_token`, own the reconnect
//! loop — fails in the way this subsystem exists to prevent: the controller
//! keeps serving, its status data quietly stops updating, and nothing says so
//! until someone reads a dashboard.

use std::path::PathBuf;

/// Everything that can go wrong reaching the swarm queue.
///
/// `thiserror` and not `anyhow` because this is a library: a caller gets a
/// type it can match on, and the binaries that consume it keep using
/// `anyhow` — `?` converts for free, so nothing downstream is more verbose
/// for it. The same split `hive-claude` uses.
///
/// The variants are the failures an operator acts on differently: a
/// half-configured environment is a deployment bug, a refused token is an
/// identity-provider config problem, an unreachable queue is a network
/// one. Collapsing them into one string would make that distinction a
/// matter of reading prose.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error(
        "swarm queue is half-configured: {prefix}_NATS_URL, \
         {prefix}_OIDC_TOKEN_ENDPOINT, {prefix}_OIDC_CLIENT_ID and \
         {prefix}_OIDC_CLIENT_SECRET_FILE must be set together or not at all"
    )]
    PartialConfig { prefix: String },

    #[error("reading the queue client secret from {path}")]
    ClientSecret {
        path: String,
        #[source]
        source: std::io::Error,
    },

    #[error("building the token-endpoint HTTP client")]
    HttpClient(#[source] reqwest::Error),

    #[error("requesting an access token from authelia")]
    TokenRequest(#[source] reqwest::Error),

    /// The body carries authelia's own `error_description`, and it is far
    /// more useful than the status alone: a wrong grant says
    /// `unauthorized_client`, a wrong secret says `invalid_client`, and
    /// those point at different config.
    #[error("authelia refused the token request ({status}): {body}")]
    TokenRefused {
        status: reqwest::StatusCode,
        body: String,
    },

    #[error("parsing authelia's token response")]
    TokenResponse(#[source] serde_json::Error),

    #[error("connecting to the swarm queue at {url}")]
    Connect {
        url: String,
        #[source]
        source: async_nats::ConnectError,
    },
}

/// Render an error and its source chain on one line.
///
/// The auth callback below hands `async-nats` a *string*, so a `Display`
/// that stopped at the top message would drop the cause — which is the
/// half that says why the mint failed. `anyhow`'s `{:#}` did this for
/// free; a library owes its callers the same detail without owing them
/// anyhow.
fn chain(error: &dyn std::error::Error) -> String {
    let mut rendered = error.to_string();
    let mut source = error.source();
    while let Some(cause) = source {
        rendered.push_str(": ");
        rendered.push_str(&cause.to_string());
        source = cause.source();
    }
    rendered
}

/// Only the one field this needs; authelia returns several.
#[derive(serde::Deserialize)]
struct TokenResponse {
    access_token: String,
}

/// Where the controller finds the queue and what it authenticates with.
///
/// Every field comes from an environment variable the NixOS module sets, the
/// same way `load_hives` takes the roster — a config change is a redeploy, and
/// this process reads no file it was not pointed at.
#[derive(Debug, Clone)]
pub struct QueueConfig {
    /// `nats://host:port` for the swarm queue.
    pub url: String,
    /// Authelia's token endpoint, e.g. `https://auth.<swarm>/api/oidc/token`.
    pub token_endpoint: String,
    /// The controller's own `OAuth2` client id.
    pub client_id: String,
    /// File holding the client secret's PLAINTEXT.
    ///
    /// A path and not a value: the secret is minted on the authelia host and
    /// read here, and putting it in the environment would publish it to
    /// anything that can read `/proc/<pid>/environ`.
    pub client_secret_file: PathBuf,
}

impl QueueConfig {
    /// Read the config from `<prefix>_NATS_URL`, `<prefix>_OIDC_TOKEN_ENDPOINT`,
    /// `<prefix>_OIDC_CLIENT_ID` and `<prefix>_OIDC_CLIENT_SECRET_FILE`, or
    /// `None` when the queue was not wired up for this deployment.
    ///
    /// The prefix is a parameter rather than a constant because the variables
    /// belong to the *consuming unit* — a NixOS module sets them alongside its
    /// other options, and two daemons sharing one name would be a worse
    /// coupling than passing four characters. What is shared is the RULE
    /// below, not the spelling.
    ///
    /// `None` rather than an error on purpose: a daemon serves its other
    /// surfaces on hosts where the queue is not enabled, and refusing to start
    /// there would trade a missing feature for a dead process. What must NOT
    /// happen is a *half* configuration silently behaving like an absent one —
    /// hence the explicit partial check below.
    pub fn from_env(prefix: &str) -> Result<Option<Self>, Error> {
        let url = std::env::var(format!("{prefix}_NATS_URL")).ok();
        let token_endpoint = std::env::var(format!("{prefix}_OIDC_TOKEN_ENDPOINT")).ok();
        let client_id = std::env::var(format!("{prefix}_OIDC_CLIENT_ID")).ok();
        let secret = std::env::var(format!("{prefix}_OIDC_CLIENT_SECRET_FILE")).ok();

        match (url, token_endpoint, client_id, secret) {
            (None, None, None, None) => Ok(None),
            (Some(url), Some(token_endpoint), Some(client_id), Some(secret)) => Ok(Some(Self {
                url,
                token_endpoint,
                client_id,
                client_secret_file: PathBuf::from(secret),
            })),
            // A partially-set environment is a deployment bug, and the failure
            // it would otherwise produce is the expensive kind: the process
            // comes up "fine", never connects, and the data it was supposed to
            // move silently stops moving. Naming the variables costs one line.
            _ => Err(Error::PartialConfig {
                prefix: prefix.to_owned(),
            }),
        }
    }
}

/// Mint a fresh access token for the controller's own client.
///
/// `client_credentials`, because there is no user here: the controller
/// authenticates as itself. Authelia refuses the `openid` scope for this grant
/// (a machine client receives an access token and never an id-token), so no
/// scope is requested.
async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String, Error> {
    // Read per call rather than caching: the file is small, and a cached
    // secret would survive a rotation that the operator believes took effect.
    let secret = tokio::fs::read_to_string(&cfg.client_secret_file)
        .await
        .map_err(|source| Error::ClientSecret {
            path: cfg.client_secret_file.display().to_string(),
            source,
        })?;

    let response = http
        .post(&cfg.token_endpoint)
        .form(&[
            ("grant_type", "client_credentials"),
            ("client_id", cfg.client_id.as_str()),
            ("client_secret", secret.trim()),
        ])
        .send()
        .await
        .map_err(Error::TokenRequest)?;

    // Why the body and not just the status: see `Error::TokenRefused`.
    let status = response.status();
    let body = response.text().await.unwrap_or_default();
    if !status.is_success() {
        return Err(Error::TokenRefused { status, body });
    }

    let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?;
    Ok(parsed.access_token)
}

/// Connect to the swarm queue, minting a token for each connection attempt.
pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
    // 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
    // `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 = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(10))
        .build()
        .map_err(Error::HttpClient)?;
    let url = cfg.url.clone();

    let client = async_nats::ConnectOptions::with_auth_callback(move |_nonce| {
        let http = http.clone();
        let cfg = cfg.clone();
        async move {
            let token = mint_token(&http, &cfg)
                .await
                // The callback's error type carries a string, so the source
                // chain would be lost; flatten it rather than dropping it.
                .map_err(|e| async_nats::AuthError::new(chain(&e)))?;
            let mut auth = async_nats::Auth::new();
            auth.token = Some(token);
            Ok(auth)
        }
    })
    // The controller and the queue are separate units on (possibly)
    // separate hosts, and nothing orders them. Without this, a queue that
    // comes up one second later leaves the controller permanently
    // queue-less until someone restarts it — a boot-order race that
    // presents as "status has been unavailable since Tuesday".
    //
    // It also composes with the callback above rather than fighting it:
    // each background attempt is a connection attempt, so each one mints
    // its own token instead of retrying a stale one.
    .retry_on_initial_connect()
    .connect(&url)
    .await
    .map_err(|source| Error::Connect {
        url: url.clone(),
        source,
    })?;

    Ok(client)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The all-unset case is the common one — most hosts do not run the queue.
    ///
    /// Uses a prefix no deployment sets, so it cannot pass vacuously by
    /// running inside a configured environment (the guard below covers the
    /// same ground, and both are cheap).
    #[test]
    fn an_absent_environment_is_not_an_error() {
        for k in [
            "SWARM_QUEUE_TEST_NATS_URL",
            "SWARM_QUEUE_TEST_OIDC_TOKEN_ENDPOINT",
            "SWARM_QUEUE_TEST_OIDC_CLIENT_ID",
            "SWARM_QUEUE_TEST_OIDC_CLIENT_SECRET_FILE",
        ] {
            assert!(std::env::var(k).is_err(), "{k} must be unset for this test");
        }
        assert!(
            QueueConfig::from_env("SWARM_QUEUE_TEST")
                .expect("absent is not an error")
                .is_none()
        );
    }

    /// The half-set case is the one the rule exists for: a deployment bug that
    /// would otherwise look exactly like "no queue configured".
    ///
    /// SAFETY: single-threaded mutation of a process env var under a prefix no
    /// other test or deployment uses; removed before returning.
    #[test]
    fn a_half_set_environment_is_a_hard_error() {
        unsafe {
            std::env::set_var("SWARM_QUEUE_HALF_NATS_URL", "nats://127.0.0.1:4222");
        }
        let err = QueueConfig::from_env("SWARM_QUEUE_HALF")
            .expect_err("a partial set must not read as absent");
        let msg = format!("{err}");
        assert!(
            msg.contains("SWARM_QUEUE_HALF_OIDC_CLIENT_ID"),
            "the error must name the missing variables, got: {msg}"
        );
        unsafe {
            std::env::remove_var("SWARM_QUEUE_HALF_NATS_URL");
        }
    }
}