//! 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(),
                minted_at: None,
            },
        }
    }

    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(),
            "$KV.agent-icons.{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(),
                "$KV.agent-icons.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(_)
        ));
    }
}