//! One agent's own identity at the swarm's secret store: issued by the store,
//! published here, granted here — and, before the job node reports success,
//! **used** here.
//!
//! The swarm obtains the agent's certificate so that no hive ever needs the
//! capability to obtain one; the hive only carries it down. The controller is
//! the swarm-level service that asks because it already logs in to the store,
//! and its grant covers exactly the calls made here (`swarm-bao.nix`'s
//! `controllerPolicyText`: `update` on the agent PKI role's `issue` path, and
//! `create/update` on `secret/data/swarm/agents/*`, on
//! `sys/policies/acl/hive-*`, and on `auth/cert/certs/hive-*`).
//!
//! **Two credentials, deliberately unrelated.** The certificate reaches the
//! store; the queue secret identifies the agent to the swarm queue. Both sit
//! under `swarm/agents/<agent>`, so neither costs a grant — but the second is
//! not derived from the first, so either renews without reference to the other.
//!
//! Four separate strings have to agree before an agent can authenticate: the
//! policy's name, the cert-auth role's name, the certificate's common name,
//! and the authority the role pins. That is the kind of agreement that holds
//! in review and fails in production — a mismatch is a 403 naming none of the
//! four — so [`mint_and_verify`] does not finish on a write. See
//! [`read_back_as_agent`].
//!
//! The authority is the store's own agent CA, generated inside its agent PKI
//! mount; its key never leaves the store. It signs no host leaf, and no host
//! role pins it, so an agent's certificate satisfies only that agent's role.
//! [`crate::agent_renewal`] re-issues a live agent's leaf once it is past half
//! its validity (the role's `ttl`), by queueing the same node as creation.

use anyhow::{Context, Result, bail};
use swarm_secret_client::{
    SecretStore,
    client::{DEFAULT_CERT_MOUNT, Settings},
    mtls, policy, queue,
};

/// The PKI mount agent leaves are issued from, as `swarm-controller.nix` sets
/// it from `deploy.bao.agentPkiMountPath`.
pub const ENV_AGENT_PKI_MOUNT: &str = "SWARM_CONTROLLER_AGENT_PKI_MOUNT";

/// The role on [`ENV_AGENT_PKI_MOUNT`] agent leaves are issued through, as
/// `swarm-controller.nix` sets it from `deploy.bao.agentPkiRoleName`.
pub const ENV_AGENT_PKI_ROLE: &str = "SWARM_CONTROLLER_AGENT_PKI_ROLE";

/// Where agent leaves are issued: `(mount, role)`, read from `get`.
///
/// # Errors
/// Naming the first of the two variables that is unset or empty.
fn agent_pki(get: impl Fn(&str) -> Option<String>) -> Result<(String, String)> {
    let required = |var: &str| -> Result<String> {
        get(var)
            .filter(|v| !v.is_empty())
            .with_context(|| format!("{var} is unset or empty, so no agent leaf can be issued"))
    };
    Ok((
        required(ENV_AGENT_PKI_MOUNT)?,
        required(ENV_AGENT_PKI_ROLE)?,
    ))
}

/// What the cert-auth role for `agent` is written from.
struct RoleInputs<'a> {
    /// Role name, policy name and the common name the role matches: one string.
    name: String,
    /// The authority the role pins: the one that signed this very leaf.
    ca: &'a str,
    /// The policy document the role attaches.
    policy: String,
}

fn role_inputs<'a>(agent: &str, credential: &'a mtls::Credential) -> Result<RoleInputs<'a>> {
    Ok(RoleInputs {
        name: policy::agent_object_name(agent)?,
        ca: &credential.ca,
        policy: policy::render_agent(agent)?,
    })
}

/// How many bytes of kernel randomness a queue secret is before encoding.
///
/// Thirty-two because the secret is a bearer token compared for equality and
/// nothing else — there is no work factor and no rate limit behind it, so the
/// only defence is that guessing is not worth attempting. Encoded it is 43
/// characters.
const QUEUE_SECRET_BYTES: usize = 32;

/// Generate a queue secret: [`QUEUE_SECRET_BYTES`] from the kernel's CSPRNG,
/// base64url without padding.
///
/// The alphabet matters and is the reason for `URL_SAFE_NO_PAD` rather than
/// the standard engine: the secret is destined to be carried in a token the
/// verifying end splits on `.`, so it must not be able to contain one. This
/// alphabet is `[A-Za-z0-9_-]`, and `=` padding is dropped as well so the
/// value survives anything that treats it as a word.
///
/// # Errors
/// When the kernel will not supply randomness. Bubbled rather than panicked
/// on: the caller is a job node that reports a named failure, and a secret
/// from a degraded source is worse than no secret.
pub(crate) fn generate_queue_secret() -> Result<String> {
    let mut bytes = [0u8; QUEUE_SECRET_BYTES];
    getrandom::fill(&mut bytes).context("drawing a queue secret from the kernel's CSPRNG")?;
    Ok(base64::Engine::encode(
        &base64::engine::general_purpose::URL_SAFE_NO_PAD,
        bytes,
    ))
}

/// Give `agent` an identity at the store, and prove it works.
///
/// Five store calls' worth of agreement, then the login that checks it:
///
/// 1. have the agent PKI role issue a leaf whose common name is
///    [`policy::agent_object_name`]; the store generates its key;
/// 2. publish it at [`mtls::identity_path`], where the agent's hive collects
///    it under the hive's own certificate;
/// 3. publish a queue secret at [`queue::agent_queue_path`] — the agent's own
///    identity at the swarm queue, minted here so that the credential an agent
///    presents names *it* rather than its hive;
/// 4. write the ACL document [`policy::render_agent`] renders — read and list on
///    this one agent's paths and nothing else;
/// 5. write the cert-auth role that ties the three together, pinning the CA
///    the store named as this leaf's issuer.
///
/// Policy before role, for the reason `read_policy::provision` gives: the role
/// names the policy, so the other order leaves a window in which it points at
/// nothing.
///
/// ⚠️ **Step 3 is idempotent and steps 1–2 are not.** Re-running issues a
/// fresh leaf, picked up on the agent's next boot, but leaves an existing
/// queue secret alone: this is re-run against running agents, which hold that
/// secret in a live connection. Revoking one means deleting the path;
/// replacing one by age is [`crate::agent_renewal`]'s.
///
/// # Errors
/// Anything that stops one of those five steps, with the step named. A
/// failure here fails the job node and nothing else — the agent is still
/// created, without a store identity.
pub async fn mint_and_verify(agent: &str) -> Result<()> {
    let (mount, pki_role) = agent_pki(|k| std::env::var(k).ok())?;
    let name = policy::agent_object_name(agent)?;
    let path = mtls::identity_path(agent)?;
    let queue_path = queue::agent_queue_path(agent)?;

    let store = crate::store::connect()
        .await
        .context("logging in to the swarm secret store")?;
    let credential = store
        .issue_client_certificate(&mount, &pki_role, &name)
        .await
        .with_context(|| format!("issuing {name}'s certificate from {mount}/issue/{pki_role}"))?;
    store
        .write(&path, &credential)
        .await
        .with_context(|| format!("publishing the agent identity at {path}"))?;

    // Read before write, and `read_optional` rather than `read`, so that only
    // a genuine 404 leads to a new secret — see that method for why every
    // other failure has to stay a failure here.
    let existing: Option<queue::AgentCredential> = store
        .read_optional(&queue_path)
        .await
        .with_context(|| format!("checking whether {queue_path} already holds a credential"))?;
    // The secret and its mint time survive a re-run. An object naming a
    // different agent is corrected by rewriting the name around the *same*
    // `value`, which no live connection notices.
    let wanted = match &existing {
        Some(existing) => queue::AgentCredential {
            agent: agent.to_owned(),
            ..existing.clone()
        },
        None => crate::agent_renewal::fresh(agent, crate::agent_renewal::unix_now())?,
    };
    if existing.as_ref() == Some(&wanted) {
        tracing::info!(
            agent,
            %queue_path,
            "agent queue credential already published; left as it is"
        );
    } else {
        store
            .write(&queue_path, &wanted)
            .await
            .with_context(|| format!("publishing the agent queue credential at {queue_path}"))?;
        tracing::info!(
            agent,
            %queue_path,
            // Never "rotated": the secret is the same one, only the name
            // around it moved.
            corrected = existing.is_some(),
            "agent queue credential published"
        );
    }
    let queue_credential = wanted;

    let inputs = role_inputs(agent, &credential)?;
    store
        .write_policy(&inputs.name, &inputs.policy)
        .await
        .with_context(|| format!("writing the read policy {}", inputs.name))?;
    store
        .write_cert_role(
            DEFAULT_CERT_MOUNT,
            &inputs.name,
            inputs.ca,
            &inputs.name,
            &inputs.name,
        )
        .await
        .with_context(|| format!("writing the cert-auth role {}", inputs.name))?;
    tracing::info!(agent, %path, role = %name, "agent store identity published");

    read_back_as_agent(&credential, &name, &path, &queue_path, &queue_credential).await?;
    tracing::info!(
        agent,
        role = %name,
        "agent store identity verified: the issued leaf logged in and read both its own paths"
    );
    Ok(())
}

/// Revoke `agent`'s queue credential: delete the path
/// [`mint_and_verify`]'s step 3 published, and everything ever written at it.
///
/// The undo of that one step and of no other. The leaf, the ACL document and
/// the cert-auth role that make up the rest of an agent's identity stay where
/// they are — they are what a *hive* uses to collect an agent's secrets, they
/// are minted afresh on every run of the mint, and tearing them down is not
/// what the queue credential outliving its holder is about.
///
/// **Deletes every version, not the newest one.** The mint rewrites this path
/// whenever the principal it names has to be corrected, so a soft delete would
/// leave the identical secret sitting in version history, readable at
/// `?version=N` by anything that can read the path at all — a value still
/// recoverable has not been revoked. See
/// [`SecretStore::delete_all_versions`][swarm_secret_client::SecretStore::delete_all_versions].
///
/// **Idempotent**: revoking an agent that never had a credential, or one
/// already revoked, succeeds and says so. A teardown that runs twice is
/// ordinary, and a second run that failed would be a worse fault than the one
/// this exists to fix.
///
/// # Errors
/// When the store cannot be reached or refuses the delete. The caller decides
/// what that costs — `set_agent_state` logs it and lets the destroy proceed,
/// since a credential that is still live is a smaller harm than an agent that
/// cannot be torn down.
pub async fn revoke_queue_credential(agent: &str) -> Result<()> {
    let queue_path = queue::agent_queue_path(agent)?;
    let store = crate::store::connect()
        .await
        .context("logging in to the swarm secret store")?;
    store
        .delete_all_versions(&queue_path)
        .await
        .with_context(|| format!("revoking the agent queue credential at {queue_path}"))?;
    // At `info` and unconditional: an unlogged revocation is indistinguishable
    // from a leak, and this line is the only record an operator has that the
    // credential stopped being usable. It cannot say whether one was there —
    // the controller's grant on these paths is write-only by design, so it
    // deletes blind.
    tracing::info!(
        agent,
        %queue_path,
        "agent queue credential revoked: every version of the path deleted"
    );
    Ok(())
}

/// The consumer of everything [`mint_and_verify`] wrote: log in **as the
/// agent**, with the leaf just issued, and read back both paths just
/// published.
///
/// The address and the store's CA come from this process's own `BAO_*`
/// environment; the *identity* deliberately does not — see
/// [`SecretStore::connect_with_identity`]. The freshly issued private key
/// never touches a filesystem.
///
/// Both paths, not just the certificate's, for the reason this function
/// exists at all: the queue credential is readable only because it sits inside
/// the stanza [`policy::render_agent`] already grants, and a policy that
/// drifted from that path would otherwise fail at the agent's first connection
/// — far from here, as a denial naming neither the policy nor the path.
///
/// # Errors
/// When the store refuses the login (the role, the authority or the common
/// name disagree), when either read is denied (the policy does not cover the
/// path, or the role attached the wrong policy), or when what comes back is
/// not what went in.
async fn read_back_as_agent(
    credential: &mtls::Credential,
    role: &str,
    path: &str,
    queue_path: &str,
    queue_credential: &queue::AgentCredential,
) -> Result<()> {
    let settings = Settings::from_env()
        .context("reading this daemon's own store settings for the read-back")?;

    // The concatenated blob `connect_with_identity` takes, built in memory.
    let mut identity = credential.cert.clone().into_bytes();
    identity.push(b'\n');
    identity.extend_from_slice(credential.key.as_bytes());

    let as_agent =
        SecretStore::connect_with_identity(&settings, &identity, role, DEFAULT_CERT_MOUNT)
            .await
            .with_context(|| {
                format!("logging in to the store as {role} with the leaf just issued")
            })?;
    let read_back: mtls::Credential = as_agent
        .read(path)
        .await
        .with_context(|| format!("reading {path} back under {role}'s own token"))?;

    // Compared, not merely decoded: a successful read of an object written by
    // some earlier run would otherwise pass this check while this run's leaf
    // was the one nobody could use. Nothing about either value is printed.
    if read_back != *credential {
        bail!("the store returned a different object at {path} than the one just published");
    }

    let read_back: queue::AgentCredential = as_agent
        .read(queue_path)
        .await
        .with_context(|| format!("reading {queue_path} back under {role}'s own token"))?;
    // The agent is compared as well as the secret: the verifying end refuses
    // an object naming a different agent than the path, so a mismatch here is
    // the same class of fault as an unreadable path.
    if read_back != *queue_credential {
        bail!("the store returned a different object at {queue_path} than the one just published");
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        ENV_AGENT_PKI_MOUNT, ENV_AGENT_PKI_ROLE, QUEUE_SECRET_BYTES, agent_pki,
        generate_queue_secret, role_inputs,
    };
    use swarm_secret_client::{mtls, policy};

    /// The alphabet claim the token format rests on: the secret is carried in
    /// a composite the verifying end splits on `.`, so a secret that could
    /// contain one would make that split ambiguous. Also the entropy claim —
    /// a generator that silently returned a short or constant value would pass
    /// every other test in this file.
    #[test]
    fn a_queue_secret_is_high_entropy_and_carries_no_separator() {
        let a = generate_queue_secret().expect("the kernel supplies randomness");
        let b = generate_queue_secret().expect("twice");
        assert_ne!(a, b, "two draws must not agree");

        // base64 without padding: one character per six bits, rounded up.
        assert_eq!(a.len(), (QUEUE_SECRET_BYTES * 8).div_ceil(6));
        assert!(
            a.chars()
                .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'),
            "{a}"
        );
        assert!(!a.contains('.'), "{a}");
        assert!(!a.contains('='), "{a}");
    }

    /// Revocation undoes step 3 of the mint and nothing else, and both ends
    /// spell the path through one function — the property that keeps a
    /// revocation from missing the object it is meant to remove.
    ///
    /// The second half is what stops this being a tautology: the agent's
    /// other published object, the mTLS leaf, is at a different path and is
    /// deliberately left alone. Revoking both would take away the identity a
    /// re-created agent is re-minted under, for a credential problem that is
    /// only about the queue.
    #[test]
    fn revocation_names_the_path_the_mint_published_and_not_the_leaf_beside_it() {
        use swarm_secret_client::{mtls, queue};

        assert_eq!(
            queue::agent_queue_path("atlas").expect("a plain name is legal"),
            "swarm/agents/atlas/queue"
        );
        assert_ne!(
            queue::agent_queue_path("atlas").expect("legal"),
            mtls::identity_path("atlas").expect("legal"),
        );
    }

    /// A name the store must never be asked to delete under. `revoke_queue_credential`
    /// builds its path with the same validating function the mint does, so a
    /// traversal is refused before a request is made rather than addressing
    /// some other principal's secret.
    #[test]
    fn a_traversal_never_becomes_a_revocation() {
        swarm_secret_client::queue::agent_queue_path("../pr1ma")
            .expect_err("a traversal is not a legal agent name");
    }

    fn both(k: &str) -> Option<String> {
        match k {
            ENV_AGENT_PKI_MOUNT => Some("pki-agents".to_owned()),
            ENV_AGENT_PKI_ROLE => Some("swarm-agent".to_owned()),
            _ => None,
        }
    }

    #[test]
    fn the_issue_path_comes_from_the_two_variables_the_module_sets() {
        assert_eq!(
            agent_pki(both).expect("both set"),
            ("pki-agents".to_owned(), "swarm-agent".to_owned())
        );
    }

    #[test]
    fn a_missing_or_empty_pki_variable_is_named() {
        for var in [ENV_AGENT_PKI_MOUNT, ENV_AGENT_PKI_ROLE] {
            let unset = agent_pki(|k| if k == var { None } else { both(k) })
                .expect_err("one variable is unset");
            assert!(format!("{unset:#}").contains(var), "{unset:#}");
            let empty = agent_pki(|k| {
                if k == var {
                    Some(String::new())
                } else {
                    both(k)
                }
            })
            .expect_err("one variable is empty");
            assert!(format!("{empty:#}").contains(var), "{empty:#}");
        }
    }

    /// Three of the four strings that must agree, checked where this process
    /// sets them: role, policy and matched CN are one name; the pinned CA is
    /// the issuer the store reported for this leaf; the policy is the agent's
    /// own single stanza.
    #[test]
    fn the_role_pins_the_leafs_own_issuer_under_the_agents_name() {
        let credential = mtls::Credential {
            cert: "LEAF".to_owned(),
            key: "KEY".to_owned(),
            ca: "AGENT-CA".to_owned(),
        };
        let inputs = role_inputs("atlas", &credential).expect("legal");
        assert_eq!(inputs.name, "hive-agent-atlas");
        assert_eq!(
            inputs.name,
            policy::agent_object_name("atlas").expect("legal"),
            "the CN the leaf is issued for"
        );
        assert_eq!(inputs.ca, "AGENT-CA");
        assert_eq!(inputs.policy, policy::render_agent("atlas").expect("legal"));
        assert!(!inputs.policy.contains("swarm/hives/"), "{}", inputs.policy);

        // The control: another agent's inputs differ in both name and grant.
        let other = role_inputs("argus", &credential).expect("legal");
        assert_ne!(other.name, inputs.name);
        assert!(!other.policy.contains("agents/atlas/"), "{}", other.policy);
    }
}