hyperhive/hive-agent/src/swarm_queue.rs

//! This agent's swarm-queue credentials, resolved once at boot.
//!
//! Three of the four coordinates arrive as environment variables the meta
//! flake renders into the harness unit. The fourth — the OIDC client id —
//! arrives as a *file*, delivered beside the secret as one systemd
//! credential pair (`nix/agent-modules/queue.nix`). That is the whole reason
//! this module exists rather than a bare [`QueueConfig::from_env`] call: the
//! id rides with the secret so a reader never has to spell `hive-<name>-agent`
//! a second time, and `from_env` wants it as a value.
//!
//! Reading the file and assigning the variable would put the same rule back
//! in `from_env`'s hands, but `std::env::set_var` is unsound in a process that
//! has already spawned threads, and this one has. So the rule is restated
//! here over the inputs this consumer actually has, and [`decide`] is the
//! single place it lives.
//!
//! Beside all four sits a fifth coordinate, resolved by
//! [`decide_agent_secret`] and not part of their group. The four are the
//! *hive's* — one OIDC client shared by every container on it — so at the
//! queue's auth callout they say which hive is connecting and never which
//! agent. The fifth is this agent's own, minted per agent at swarm level and
//! fetched by the container itself (`nix/agent-modules/queue-identity.nix`).
//!
//! When it is present, [`client`] connects with it first, and the agent
//! publishes on its own hive-free subjects. When it is absent, or the queue
//! refuses it, the connect falls back to the hive's shared client and the
//! hive-scoped subjects. [`Connection::presented`] says which, so a publisher
//! builds the subject that credential is granted.

use std::path::{Path, PathBuf};
use std::sync::OnceLock;

use anyhow::Context as _;
use tokio::sync::OnceCell;

use swarm_queue_client::QueueConfig;

/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s
/// on purpose: an agent authenticates as its own client, not as its hive.
const ENV_PREFIX: &str = "HIVE_AGENT";

/// Resolved once at boot, so the publisher that connects to the queue reads
/// one answer rather than re-deriving it per call.
static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new();

/// Where to present this agent's own credential, resolved once at boot.
static AGENT: OnceLock<Option<AgentPath>> = OnceLock::new();

/// The one connection every publisher in this process shares. Separate from
/// [`CONFIG`] because resolving the coordinates is synchronous boot work and
/// connecting is not — see [`client`].
static CLIENT: OnceCell<Option<Connection>> = OnceCell::const_new();

/// What connecting with this agent's own credential needs.
#[derive(Debug, PartialEq, Eq)]
struct AgentPath {
    url: String,
    ca_file: Option<PathBuf>,
    /// The fetched secret. A path; the bytes are read at connect.
    secret_file: PathBuf,
}

/// Which credential the shared connection presented.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Presented {
    /// This agent's own, granted `<prefix>.<agent>`.
    Agent,
    /// The hive's shared client, granted `<prefix>.<hive>.>` for the hive the
    /// client id names.
    Hive { client_id: String },
}

/// The shared queue connection and the credential it was made with.
#[derive(Clone)]
pub struct Connection {
    pub client: async_nats::Client,
    pub presented: Presented,
}

/// The four variables the harness unit sets, before the client-id file is
/// read. Collected into a struct so [`decide`] is pure over them and the
/// process env is touched in exactly one place.
struct QueueEnv {
    nats_url: Option<String>,
    token_endpoint: Option<String>,
    client_id_file: Option<String>,
    client_secret_file: Option<String>,
    /// Independent of the all-or-none group below, exactly as in
    /// `QueueConfig::from_env`: a swarm behind a publicly-trusted
    /// certificate needs no extra anchor. Nothing in this tree sets it for
    /// an agent — a container already trusts the swarm root, which
    /// `hive_c0re::meta` embeds at build time — so it is here for a
    /// deployment that needs a different one, not for ours.
    ca_file: Option<String>,
    /// Where `nix/agent-modules/queue-identity.nix` fetched this agent's own
    /// per-agent secret to. Outside the all-or-none group above because it is
    /// governed by a different switch entirely — that unit is generated by
    /// the agent having a *store* address, not by its hive having queue
    /// coordinates — so an agent can legally have this and none of the four,
    /// or the four and not this.
    agent_secret_file: Option<String>,
}

impl QueueEnv {
    fn from_env() -> Self {
        let var = |suffix: &str| std::env::var(format!("{ENV_PREFIX}_{suffix}")).ok();
        Self {
            nats_url: var("NATS_URL"),
            token_endpoint: var("OIDC_TOKEN_ENDPOINT"),
            client_id_file: var("OIDC_CLIENT_ID_FILE"),
            client_secret_file: var("OIDC_CLIENT_SECRET_FILE"),
            ca_file: var("OIDC_CA_FILE"),
            agent_secret_file: var("QUEUE_AGENT_SECRET_FILE"),
        }
    }
}

/// What the environment plus the client-id file add up to.
enum Resolution {
    /// Everything is here; the agent can reach the queue.
    Configured(Box<QueueConfig>),
    /// No queue coordinates for this agent, and that is a legal state — carries why.
    Absent(&'static str),
    /// Some of the environment, not all of it. A deployment bug rather than
    /// an absent integration, so it is reported and then survived.
    Partial,
}

/// Read the client id out of the file the credential landed at.
///
/// `None` for a missing or empty file, which is the ordinary state of a hive
/// whose secret store has nothing published yet — the credential simply is
/// not there, and nspawn forwards nothing. Trailing newline stripped: the
/// reader unit writes one and an id with a newline in it authenticates as
/// nobody.
fn read_client_id(path: &Path) -> Option<String> {
    match std::fs::read_to_string(path) {
        Ok(raw) => {
            let id = raw.trim();
            (!id.is_empty()).then(|| id.to_owned())
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => {
            tracing::warn!(path = %path.display(), error = %e, "reading the queue client id failed");
            None
        }
    }
}

/// Decide whether this agent has its own per-agent queue secret, given the
/// path the fetch unit was told to write it to.
///
/// Three states collapse to two answers. No variable means no store address
/// for this container, so no fetch unit was generated at all. A variable
/// naming a file that is missing or empty means the unit ran and found
/// nothing minted — the ordinary state of an agent created before its swarm
/// knew to mint one, which that unit reports and survives. Only a non-empty
/// file is a credential.
///
/// The file is not read. Its *contents* are the secret and belong nowhere but
/// the moment of use; what a caller needs from here is whether there is one
/// and where, which `metadata` answers without opening it.
fn decide_agent_secret(path: Option<&str>) -> Option<PathBuf> {
    let path = PathBuf::from(path?);
    match std::fs::metadata(&path) {
        Ok(m) if m.len() > 0 => Some(path),
        Ok(_) => None,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => {
            tracing::warn!(
                path = %path.display(),
                error = %e,
                "checking for this agent's own queue credential failed"
            );
            None
        }
    }
}

/// Decide whether this agent can connect with its own credential: it needs the
/// queue's address and a fetched secret, and nothing of the hive's client.
fn decide_agent_path(env: &QueueEnv, secret_file: Option<PathBuf>) -> Option<AgentPath> {
    Some(AgentPath {
        url: env.nats_url.clone()?,
        ca_file: env.ca_file.as_ref().map(Into::into),
        secret_file: secret_file?,
    })
}

/// Decide what this agent's queue configuration is, given the environment and
/// whatever the client-id file held.
///
/// All four variables or none, for the reason `QueueConfig::from_env` gives:
/// a half-set environment produces a process that comes up fine and never
/// connects. The client id is graded separately from the four because its
/// file has a legal absence the variables do not — the unit names the path a
/// credential *would* arrive at whether or not one has been published yet.
fn decide(env: &QueueEnv, client_id: Option<String>) -> Resolution {
    match (
        env.nats_url.as_ref(),
        env.token_endpoint.as_ref(),
        env.client_id_file.as_ref(),
        env.client_secret_file.as_ref(),
    ) {
        (None, None, None, None) => Resolution::Absent("this hive has no swarm queue configured"),
        (Some(url), Some(token_endpoint), Some(_), Some(secret)) => match client_id {
            Some(client_id) => Resolution::Configured(Box::new(QueueConfig {
                url: url.clone(),
                token_endpoint: token_endpoint.clone(),
                client_id,
                client_secret_file: secret.into(),
                ca_file: env.ca_file.as_ref().map(Into::into),
            })),
            None => {
                Resolution::Absent("the queue credential has not been published to this hive yet")
            }
        },
        _ => Resolution::Partial,
    }
}

/// Resolve this agent's queue configuration and record it for later use.
///
/// Never fails: a harness that cannot reach the queue still serves its
/// operator, its web UI and its turn loop, so every outcome here is a log
/// line and not an exit.
pub fn init() {
    let env = QueueEnv::from_env();
    let client_id = env
        .client_id_file
        .as_deref()
        .map(Path::new)
        .and_then(read_client_id);
    let resolved = match decide(&env, client_id) {
        Resolution::Configured(cfg) => {
            // url + client id only. The secret is a path in this struct and
            // stays one: neither it nor its contents belong in a log.
            tracing::info!(url = %cfg.url, client_id = %cfg.client_id, "swarm queue configured");
            Some(*cfg)
        }
        Resolution::Absent(why) => {
            tracing::info!(why, "no swarm queue coordinates for this agent");
            None
        }
        Resolution::Partial => {
            tracing::error!(
                prefix = ENV_PREFIX,
                "swarm queue half-configured: {ENV_PREFIX}_NATS_URL, \
                 {ENV_PREFIX}_OIDC_TOKEN_ENDPOINT, {ENV_PREFIX}_OIDC_CLIENT_ID_FILE and \
                 {ENV_PREFIX}_OIDC_CLIENT_SECRET_FILE are set together or not at all — \
                 this agent will not connect"
            );
            None
        }
    };
    let _ = CONFIG.set(resolved);

    // Independent of everything above: this agent may hold its own secret on
    // a hive whose shared client is not published yet, or the shared client
    // and no secret of its own.
    let secret = decide_agent_secret(env.agent_secret_file.as_deref());
    if let Some(path) = &secret {
        // The path, never the bytes: the file holds the secret itself.
        tracing::info!(
            path = %path.display(),
            "this agent has its own swarm queue credential"
        );
    } else {
        tracing::info!(
            "no per-agent swarm queue credential; this agent is known to the queue \
             by its hive's shared client"
        );
    }
    let _ = AGENT.set(decide_agent_path(&env, secret));
}

/// Whether this agent has any credential to reach the queue with. `false`
/// before [`init`] has run.
pub fn configured() -> bool {
    config().is_some() || AGENT.get().is_some_and(Option::is_some)
}

/// What [`init`] resolved, or `None` when this agent has no queue.
///
/// Borrowed from the `OnceLock` rather than cloned: a caller wants the client
/// id to derive its subject from, and handing out an owned copy of a struct
/// holding a credential path invites it being stored somewhere with a
/// different lifetime than the one place that owns it.
///
/// `None` before [`init`] has run, which is the same answer as "no queue" and
/// deliberately not a panic — the ordering is a boot detail, and a harness
/// that reordered its boot should lose the queue, not die.
pub fn config() -> Option<&'static QueueConfig> {
    CONFIG.get()?.as_ref()
}

/// The shared queue connection, made on first call and memoized for the rest
/// of the process.
///
/// One connection per process, not per publisher: the agent authenticates as
/// one client, so a second `connect` would be a second token mint and a second
/// live connection for the same identity rather than a second credential.
///
/// `None` covers both "no queue coordinates" and "configured but the connect
/// failed" — a caller does nothing differently between them, since either way
/// there is nothing to publish onto. Connecting is lazy so that an agent on a
/// hive with no queue pays nothing at boot.
pub async fn client() -> Option<Connection> {
    CLIENT.get_or_init(connect_once).await.clone()
}

/// This agent's own credential first, then the hive's. Each path taken is
/// logged once, here.
async fn connect_once() -> Option<Connection> {
    if let Some(agent) = AGENT.get().and_then(Option::as_ref) {
        match connect_as_agent(agent).await {
            Ok(client) => {
                tracing::info!(
                    url = %agent.url,
                    "connected to the swarm queue with this agent's own credential"
                );
                return Some(Connection {
                    client,
                    presented: Presented::Agent,
                });
            }
            Err(e) => tracing::warn!(
                error = format!("{e:#}"),
                "connecting with this agent's own credential failed; falling back to \
                 its hive's shared client"
            ),
        }
    }
    let cfg = config()?;
    match swarm_queue_client::connect(cfg.clone()).await {
        Ok(client) => {
            tracing::info!(
                url = %cfg.url,
                client_id = %cfg.client_id,
                "connecting to the swarm queue with the hive's shared client; the \
                 client retries in the background until the queue accepts it"
            );
            Some(Connection {
                client,
                presented: Presented::Hive {
                    client_id: cfg.client_id.clone(),
                },
            })
        }
        Err(e) => {
            // `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose
            // `Display` ignores the alternate flag, so `{:#}` renders the
            // headline and drops the cause that says which half failed.
            tracing::warn!(
                error = %swarm_queue_client::chain(&e),
                "swarm queue connect failed; this agent publishes nothing upward"
            );
            None
        }
    }
}

/// Connect presenting this agent's own token. Any failure, a refusal
/// included, is returned for [`connect_once`] to fall back on.
async fn connect_as_agent(agent: &AgentPath) -> anyhow::Result<async_nats::Client> {
    let name = crate::identity::label();
    let secret = std::fs::read_to_string(&agent.secret_file)
        .with_context(|| format!("reading {}", agent.secret_file.display()))?;
    let token = swarm_queue_client::agent_token::format_agent_token(&name, secret.trim())?;
    swarm_queue_client::connect_with_token(&agent.url, agent.ca_file.as_deref(), token)
        .await
        .map_err(|e| anyhow::anyhow!(swarm_queue_client::chain(&e)))
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;

    use super::{
        AgentPath, QueueEnv, Resolution, decide, decide_agent_path, decide_agent_secret,
        read_client_id,
    };

    fn env(parts: [Option<&str>; 4]) -> QueueEnv {
        let [nats_url, token_endpoint, client_id_file, client_secret_file] = parts;
        QueueEnv {
            nats_url: nats_url.map(str::to_owned),
            token_endpoint: token_endpoint.map(str::to_owned),
            client_id_file: client_id_file.map(str::to_owned),
            client_secret_file: client_secret_file.map(str::to_owned),
            ca_file: None,
            agent_secret_file: None,
        }
    }

    fn full() -> QueueEnv {
        env([
            Some("nats://10.42.0.1:4222"),
            Some("https://auth.t.local/api/oidc/token"),
            Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
            Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
        ])
    }

    /// The reader unit writes the id with a trailing newline; an id carrying
    /// one is a client authelia has never heard of.
    #[test]
    fn a_client_id_file_is_read_without_its_newline() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("client_id");
        std::fs::write(&path, "hive-h1-agent\n").expect("write");
        assert_eq!(read_client_id(&path).as_deref(), Some("hive-h1-agent"));
    }

    /// Both shapes of "no credential here": never delivered, or delivered
    /// empty. Neither is an id, and treating an empty string as one would
    /// authenticate as the anonymous client rather than failing.
    #[test]
    fn a_missing_or_empty_client_id_file_reads_as_no_id() {
        let dir = tempfile::tempdir().expect("tempdir");
        let missing = dir.path().join("client_id");
        assert_eq!(read_client_id(&missing), None);
        std::fs::write(&missing, "\n").expect("write");
        assert_eq!(read_client_id(&missing), None);
    }

    #[test]
    fn a_complete_environment_with_a_published_id_configures_the_queue() {
        let e = full();
        let Resolution::Configured(cfg) = decide(&e, Some("hive-h1-agent".to_owned())) else {
            panic!("expected a configured queue");
        };
        assert_eq!(cfg.url, "nats://10.42.0.1:4222");
        assert_eq!(cfg.client_id, "hive-h1-agent");
        assert!(
            cfg.client_secret_file.ends_with("hive-queue-agent-secret"),
            "the secret stays a path: {}",
            cfg.client_secret_file.display()
        );
    }

    /// A hive that has not been given its queue's coordinates at all. Silence
    /// here is correct, and it has to be distinguishable from the half-set
    /// case below — that distinction is the only thing that makes the error
    /// branch worth logging.
    #[test]
    fn an_empty_environment_is_no_queue_rather_than_an_error() {
        let e = env([None, None, None, None]);
        assert!(matches!(decide(&e, None), Resolution::Absent(_)));
    }

    /// The credential's own absence. The unit names the path unconditionally
    /// once the hive has a queue, so this is the state of every agent on a
    /// swarm whose publisher has not run — legal, and not the error branch.
    #[test]
    fn a_complete_environment_with_no_published_id_is_no_queue() {
        let e = full();
        assert!(matches!(decide(&e, None), Resolution::Absent(_)));
    }

    /// Each single-variable omission, because the failure a partial set
    /// produces is a harness that looks healthy and publishes nothing.
    #[test]
    fn any_missing_variable_is_a_partial_configuration() {
        for drop in 0..4 {
            let mut parts = [
                Some("nats://10.42.0.1:4222"),
                Some("https://auth.t.local/api/oidc/token"),
                Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
                Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
            ];
            parts[drop] = None;
            let e = env(parts);
            assert!(
                matches!(
                    decide(&e, Some("hive-h1-agent".to_owned())),
                    Resolution::Partial
                ),
                "dropping variable {drop} must report a partial configuration"
            );
        }
    }

    /// The whole point of the fetch: a non-empty file is this agent's own
    /// credential, and the answer is the path rather than what is in it.
    #[test]
    fn a_fetched_secret_resolves_to_its_path() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("secret");
        std::fs::write(&path, "s3cr3t").expect("write");
        assert_eq!(
            decide_agent_secret(path.to_str()).as_deref(),
            Some(path.as_path())
        );
    }

    /// The rollout state, and the one this must not confuse with a
    /// credential: the fetch unit ran, found nothing minted for this agent,
    /// and left no file. Treating that as a secret would have the harness
    /// present zero bytes to the queue.
    #[test]
    fn a_missing_or_empty_fetched_secret_is_no_credential() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("secret");
        assert_eq!(decide_agent_secret(path.to_str()), None);
        std::fs::write(&path, "").expect("write");
        assert_eq!(decide_agent_secret(path.to_str()), None);
    }

    /// No variable at all: this container was given no store address, so no
    /// fetch unit exists to have written anything.
    #[test]
    fn no_fetch_path_is_no_credential() {
        assert_eq!(decide_agent_secret(None), None);
    }

    /// The two credentials are resolved by separate switches, and this is the
    /// asymmetry that makes keeping them apart worth it: an agent whose hive
    /// has no queue can still hold its own swarm-minted secret, because that
    /// one is minted with no hive in the chain.
    #[test]
    fn the_per_agent_secret_is_independent_of_the_hive_coordinates() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("secret");
        std::fs::write(&path, "s3cr3t").expect("write");

        let e = env([None, None, None, None]);
        assert!(matches!(decide(&e, None), Resolution::Absent(_)));
        assert!(decide_agent_secret(path.to_str()).is_some());
    }

    /// The agent's own path needs the queue's address and its own secret, and
    /// not the hive's client id: that is what lets it connect on a hive whose
    /// shared client is not published.
    #[test]
    fn an_agent_with_its_own_secret_and_the_queue_address_connects_as_itself() {
        let secret = PathBuf::from("/run/queue-identity/secret");
        assert_eq!(
            decide_agent_path(&full(), Some(secret.clone())),
            Some(AgentPath {
                url: "nats://10.42.0.1:4222".to_owned(),
                ca_file: None,
                secret_file: secret.clone(),
            })
        );
        let url_only = env([Some("nats://10.42.0.1:4222"), None, None, None]);
        assert!(decide_agent_path(&url_only, Some(secret)).is_some());
    }

    #[test]
    fn without_its_own_secret_or_the_queue_address_an_agent_does_not_connect_as_itself() {
        assert_eq!(decide_agent_path(&full(), None), None);
        assert_eq!(
            decide_agent_path(
                &env([None, None, None, None]),
                Some(PathBuf::from("/run/queue-identity/secret"))
            ),
            None
        );
    }
}