swarm-nats-auth: verify an agent's own token against the store

An `auth_token` spelled `swarm-agent.<agent>.<secret>` is no longer sent
to introspection. The responder reads `swarm/agents/<agent>/queue` with
an identity of its own, checks that the stored object names the same
agent, compares the secret in constant time, and grants the subjects
`--agent-token-publish-subject` lists with `{agent}` expanded. Every
other outcome denies: a malformed token, no store identity, nothing
stored, a failed or slow lookup, a different secret. A token without
the prefix takes the OIDC path unchanged.

The journal's `auth request` line names such a caller `agent:<agent>`;
the hive-shared credential keeps `hive-<h>-agent`.

The new principal: a `swarm-nats-auth` cert-auth role and policy with
read on `secret/data/swarm/agents/+/queue` alone, a leaf signed by the
store's PKI glue, and `glue-nats-auth-bao-identity.nix` pairing the two.
The copy unit delivers the identity into the queue's container, and an
absent leaf is delivered empty so the responder still starts and only
agent tokens are refused.

The policy and role are written by `swarm-bao-nats-auth-policy`, logged in
as the bao granter: both names fall under its `swarm-*` globs, so the
deploy writes them with no operator step. module-eval counts it among the
granting units, so every generic granting-unit case covers it.

The secret compare uses `subtle`, already in the lock file through the
TLS stack; no workspace crate offered one directly.

swarm-nats-auth/src/agent_token.rs

@@ -0,0 +1,382 @@
+//! Verifying an agent's own credential, presented as `auth_token`.
+//!
+//! The spelling is `swarm_queue_client::agent_token`'s, shared with the agent
+//! that presents it: a prefix, the agent's name, and its secret. The name is only a
+//! claim. It becomes an identity when the secret equals the one stored at
+//! `swarm/agents/<agent>/queue`, and nothing in this module grants on the name
+//! alone: every path that does not reach that comparison, and every lookup
+//! that fails, is a denial.
+//!
+//! A token without the prefix is not this module's: it goes to introspection
+//! unchanged.
+
+use std::future::Future;
+
+use anyhow::Context;
+use subtle::ConstantTimeEq;
+use swarm_queue_client::agent_token::{AgentToken, Malformed, parse_agent_token};
+use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings};
+use swarm_secret_client::queue::{self, AgentCredential};
+
+use crate::policy::{Permissions, Policy};
+
+/// What a presented `auth_token` is.
+pub enum Presented<'a> {
+    /// An agent's own credential, to be checked against the store.
+    Agent(AgentToken<'a>),
+    /// Carries the agent-token prefix but is not well-formed. Denied without a
+    /// lookup, and never handed to introspection: it holds a secret meant for
+    /// the store, not for the `IdP`.
+    Malformed(Malformed),
+    /// Anything else, which is an OIDC access token for introspection.
+    Bearer(&'a str),
+}
+
+/// Sort `token` onto the agent path or the introspection path.
+pub fn classify(token: &str) -> Presented<'_> {
+    match parse_agent_token(token) {
+        Some(Ok(agent)) => Presented::Agent(agent),
+        Some(Err(e)) => Presented::Malformed(e),
+        None => Presented::Bearer(token),
+    }
+}
+
+/// Where the stored credential for an agent comes from.
+pub trait CredentialSource {
+    /// The object at `agent`'s queue path, `None` when nothing is stored
+    /// there, or an error when the store could not answer.
+    fn lookup(
+        &self,
+        agent: &str,
+    ) -> impl Future<Output = anyhow::Result<Option<AgentCredential>>> + Send;
+}
+
+/// The swarm secret store, reached with this responder's own certificate.
+pub struct Store {
+    settings: Settings,
+    cert_role: String,
+}
+
+impl Store {
+    /// The store named by the `BAO_*` environment, or `None` when that is
+    /// unset: a responder with no store identity denies every agent token and
+    /// serves the OIDC path unchanged.
+    pub fn from_env(cert_role: String) -> Option<Self> {
+        match Settings::from_env() {
+            Ok(settings) => Some(Self {
+                settings,
+                cert_role,
+            }),
+            Err(e) => {
+                tracing::info!(reason = %e, "no secret store configured; agent tokens are denied");
+                None
+            }
+        }
+    }
+}
+
+impl CredentialSource for Store {
+    /// Logs in per lookup. An agent connects once per boot and per reconnect,
+    /// so a login each time costs little, and it leaves no token to expire
+    /// inside a long-lived process.
+    async fn lookup(&self, agent: &str) -> anyhow::Result<Option<AgentCredential>> {
+        let path = queue::agent_queue_path(agent)?;
+        let store = swarm_secret_client::SecretStore::connect(
+            &self.settings,
+            &self.cert_role,
+            DEFAULT_CERT_MOUNT,
+        )
+        .await
+        .context("logging in to the secret store")?;
+        store
+            .read_optional(&path)
+            .await
+            .with_context(|| format!("reading {path}"))
+    }
+}
+
+/// Whether `token`'s secret is the one stored for the agent it names.
+///
+/// The lookup is bounded by introspection's budget, under the server's
+/// `authorization.timeout`. The callout loop answers one request at a time, so
+/// a store that does not answer holds every request behind it, OIDC ones
+/// included, for at most that long, and this then denies.
+async fn verify(source: &impl CredentialSource, token: &AgentToken<'_>) -> bool {
+    let agent = token.agent;
+    let stored = match tokio::time::timeout(
+        crate::introspect::INTROSPECTION_TIMEOUT,
+        source.lookup(agent),
+    )
+    .await
+    {
+        Ok(Ok(Some(stored))) => stored,
+        Ok(Ok(None)) => {
+            tracing::warn!(
+                agent,
+                "no queue credential is stored for this agent; denying"
+            );
+            return false;
+        }
+        Ok(Err(e)) => {
+            tracing::warn!(
+                agent,
+                error = format!("{e:#}"),
+                "credential lookup failed; denying"
+            );
+            return false;
+        }
+        Err(_) => {
+            tracing::warn!(agent, "credential lookup timed out; denying");
+            return false;
+        }
+    };
+    if stored.agent != agent {
+        tracing::warn!(
+            agent,
+            stored_agent = %stored.agent,
+            "the stored credential names a different agent than its path; denying"
+        );
+        return false;
+    }
+    // Constant time, so the time to refuse says nothing about how much of the
+    // secret was right. A length mismatch returns early; the length is not
+    // secret, every minted secret has the same one.
+    stored
+        .value
+        .as_bytes()
+        .ct_eq(token.secret.as_bytes())
+        .into()
+}
+
+/// The grant for an agent token, or `None` for a denial.
+///
+/// `source` is `None` when this responder has no store identity, which denies.
+pub async fn authorize<S: CredentialSource>(
+    policy: &Policy,
+    source: Option<&S>,
+    token: &AgentToken<'_>,
+) -> Option<Permissions> {
+    let Some(source) = source else {
+        tracing::warn!(
+            agent = token.agent,
+            "agent token presented, but this responder has no secret store; denying"
+        );
+        return None;
+    };
+    if !verify(source, token).await {
+        return None;
+    }
+    let permissions = policy.agent_token_permissions(token.agent);
+    if permissions.is_none() {
+        tracing::warn!(
+            agent = token.agent,
+            "verified agent token, but no --agent-token-publish-subject is configured; denying"
+        );
+    }
+    permissions
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A store holding at most one credential, or failing outright.
+    enum Fake {
+        /// `credential` is stored at `path_agent`'s queue path.
+        Holds {
+            path_agent: &'static str,
+            credential: AgentCredential,
+        },
+        Fails,
+        /// A store that accepts the request and never answers.
+        Hangs,
+    }
+
+    impl CredentialSource for Fake {
+        async fn lookup(&self, agent: &str) -> anyhow::Result<Option<AgentCredential>> {
+            match self {
+                Self::Holds {
+                    path_agent,
+                    credential,
+                } => Ok((agent == *path_agent).then(|| credential.clone())),
+                Self::Fails => anyhow::bail!("store unreachable"),
+                Self::Hangs => std::future::pending().await,
+            }
+        }
+    }
+
+    fn stored(path_agent: &'static str, named: &str) -> Fake {
+        Fake::Holds {
+            path_agent,
+            credential: AgentCredential {
+                value: "Ab9_-zSECRET".to_owned(),
+                agent: named.to_owned(),
+            },
+        }
+    }
+
+    fn atlas() -> Fake {
+        stored("atlas", "atlas")
+    }
+
+    fn policy() -> Policy {
+        Policy::new(
+            "hive-".to_owned(),
+            "-agent".to_owned(),
+            "hive-status".to_owned(),
+            vec!["swarm-controller".to_owned()],
+            vec![],
+            vec!["$SWARM.term.{hive}.>".to_owned()],
+        )
+        .expect("valid")
+        .with_agent_token_subjects(vec![
+            "$SWARM.term.{agent}".to_owned(),
+            "$SWARM.agent-state.{agent}".to_owned(),
+        ])
+        .expect("valid")
+    }
+
+    async fn grant(source: Option<&Fake>, token: &str) -> Option<Permissions> {
+        let Presented::Agent(token) = classify(token) else {
+            panic!("{token:?} must classify as an agent token");
+        };
+        authorize(&policy(), source, &token).await
+    }
+
+    #[tokio::test]
+    async fn a_valid_agent_token_is_granted_exactly_its_own_subjects() {
+        let g = grant(Some(&atlas()), "swarm-agent.atlas.Ab9_-zSECRET")
+            .await
+            .expect("granted");
+        assert_eq!(
+            g.publish,
+            vec![
+                "$SWARM.term.atlas".to_owned(),
+                "$SWARM.agent-state.atlas".to_owned(),
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_wrong_secret_is_denied() {
+        assert!(
+            grant(Some(&atlas()), "swarm-agent.atlas.Ab9_-zSECREX")
+                .await
+                .is_none()
+        );
+        assert!(
+            grant(Some(&atlas()), "swarm-agent.atlas.Ab9_-zSECRE")
+                .await
+                .is_none()
+        );
+    }
+
+    /// The secret check is what stops one agent claiming another's name: the
+    /// right secret under someone else's name finds that agent's credential,
+    /// or none.
+    #[tokio::test]
+    async fn another_agents_name_with_this_agents_secret_is_denied() {
+        assert!(
+            grant(Some(&atlas()), "swarm-agent.argus.Ab9_-zSECRET")
+                .await
+                .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn an_agent_with_nothing_stored_is_denied() {
+        assert!(
+            grant(
+                Some(&stored("argus", "argus")),
+                "swarm-agent.atlas.Ab9_-zSECRET"
+            )
+            .await
+            .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn a_failed_lookup_is_denied() {
+        assert!(
+            grant(Some(&Fake::Fails), "swarm-agent.atlas.Ab9_-zSECRET")
+                .await
+                .is_none()
+        );
+    }
+
+    /// The callout loop answers one request at a time, so a store that never
+    /// answers must cost at most the bound, and then deny.
+    #[tokio::test]
+    async fn a_store_that_never_answers_is_denied_within_the_bound() {
+        let bound = crate::introspect::INTROSPECTION_TIMEOUT;
+        let started = std::time::Instant::now();
+        assert!(
+            grant(Some(&Fake::Hangs), "swarm-agent.atlas.Ab9_-zSECRET")
+                .await
+                .is_none()
+        );
+        let took = started.elapsed();
+        assert!(took >= bound, "denied before the bound: {took:?}");
+        assert!(
+            took < bound + std::time::Duration::from_millis(250),
+            "denied well after the bound: {took:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn no_store_is_denied() {
+        assert!(
+            grant(None, "swarm-agent.atlas.Ab9_-zSECRET")
+                .await
+                .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn a_stored_object_naming_another_agent_is_denied() {
+        assert!(
+            grant(
+                Some(&stored("argus", "atlas")),
+                "swarm-agent.argus.Ab9_-zSECRET"
+            )
+            .await
+            .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn a_verified_agent_with_no_subjects_configured_is_denied() {
+        let bare = Policy::new(
+            "hive-".to_owned(),
+            "-agent".to_owned(),
+            "hive-status".to_owned(),
+            vec![],
+            vec![],
+            vec![],
+        )
+        .expect("valid");
+        let Presented::Agent(token) = classify("swarm-agent.atlas.Ab9_-zSECRET") else {
+            panic!("an agent token");
+        };
+        assert!(authorize(&bare, Some(&atlas()), &token).await.is_none());
+    }
+
+    /// Everything without the prefix reaches introspection as it arrived.
+    #[test]
+    fn a_token_without_the_prefix_goes_to_introspection_unchanged() {
+        for token in ["authelia_at_abc.def", "atlas.Ab9_-zSECRET"] {
+            let Presented::Bearer(t) = classify(token) else {
+                panic!("{token:?} is not an agent token");
+            };
+            assert_eq!(t, token);
+        }
+    }
+
+    #[test]
+    fn a_malformed_agent_token_goes_nowhere() {
+        assert!(matches!(
+            classify("swarm-agent.atlas"),
+            Presented::Malformed(_)
+        ));
+    }
+}