hyperhive/swarm-controller/src/forge/agent_token.rs

//! Each agent's own forge access token: minted here with the admin API, stored
//! at `swarm/agents/<agent>/forge-token`, and pulled from there by the agent
//! container itself (`nix/agent-modules/forge-token.nix`).
//!
//! One token per agent, under the fixed name [`AGENT_TOKEN_NAME`]. Forgejo
//! refuses a second token with a name the user already has, so a rotation is a
//! delete then a create, and there is never more than one of these per agent.
//! The `hyperhive-<unix-seconds>` tokens `hive-c0re` used to mint are never
//! touched here: the name cannot match them. [`spawn`]'s pass deletes them
//! separately, via [`super::legacy_tokens`].
//!
//! Two callers insert the same `MintAgentForgeToken` job node: agent creation,
//! and [`spawn`]'s pass at start and every [`RECONCILE_INTERVAL`] over every
//! agent that holds a store identity. The decision is pure ([`classify`] and
//! [`plan`]), so the tests pin it; the IO on either side only reads or acts.

use std::collections::BTreeSet;
use std::sync::Arc;

use anyhow::{Context, Result, bail};
use forgejo_api::structs::{AccessToken, CreateAccessTokenOption};
use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError};
use reqwest::StatusCode;
use swarm_secret_client::{client::DEFAULT_CERT_MOUNT, forge, policy};

use super::Client;

/// The forge's name for every agent token this module mints.
///
/// Deliberately not `hyperhive-…`: that prefix is what `hive-c0re` named each
/// of its re-mints, and [`super::legacy_tokens`] deletes those.
pub const AGENT_TOKEN_NAME: &str = "swarm-agent";

/// The scopes an agent token carries. Byte-identical to the `TOKEN_SCOPES`
/// `hive-c0re` minted with, so the move from hive to swarm changes nothing an
/// agent can do. Narrowing it is a separate decision. A test here pins the
/// literal.
pub const AGENT_TOKEN_SCOPES: &str = "read:user,write:user,read:notification,write:notification,write:repository,write:issue,write:organization,write:misc";

/// How often [`spawn`] re-checks every agent's token.
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);

/// Why a token has to be (re)minted.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MintReason {
    /// The store holds nothing for this agent.
    NotStored,
    /// The forge lists no [`AGENT_TOKEN_NAME`] token for this agent.
    NotOnForge,
    /// The forge's token is not the one the store holds.
    LastEightMismatch,
    /// The forge's token carries other scopes than [`AGENT_TOKEN_SCOPES`].
    ScopeMismatch,
}

/// What to do about one agent's token.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Decision {
    /// The stored token is the forge's token, with the right scopes.
    Keep,
    /// Delete any [`AGENT_TOKEN_NAME`] token, create a new one, store it.
    Mint(MintReason),
    /// The forge lists more than one [`AGENT_TOKEN_NAME`] token. Forgejo
    /// refuses a duplicate name, so this is a forge in a state nothing here
    /// can reason about; a delete by name would 422 on it anyway.
    Conflict,
}

/// A scope list reduced to what Forgejo would store for it.
///
/// Forgejo normalises a new token's scopes before storing them
/// (`models/auth/access_token_scope.go`, `Normalize` → `toScope`): a `read:X`
/// beside `write:X` is dropped, because the write bit implies the read one.
/// The admin list then returns the stored form. Comparing
/// [`AGENT_TOKEN_SCOPES`] to that list as written would never match, and every
/// pass would rotate every agent's token.
fn normalized_scopes<'a>(scopes: impl IntoIterator<Item = &'a str>) -> BTreeSet<String> {
    let all: BTreeSet<&str> = scopes.into_iter().map(str::trim).collect();
    all.iter()
        .filter(|s| {
            s.strip_prefix("read:")
                .is_none_or(|area| !all.contains(format!("write:{area}").as_str()))
        })
        .map(|s| (*s).to_owned())
        .collect()
}

/// The last eight characters of `token`, as Forgejo reports them for a token
/// it holds (`token_last_eight`). `None` for a token shorter than that, which
/// no Forgejo token is.
fn last_eight(token: &str) -> Option<&str> {
    token.len().checked_sub(8).and_then(|at| token.get(at..))
}

/// Decide what to do about an agent's token, from what the store holds and
/// what the forge lists for the agent.
///
/// Every token not named [`AGENT_TOKEN_NAME`] is ignored, so the
/// `hyperhive-*` tokens `hive-c0re` left behind never affect the decision.
pub fn classify(stored: Option<&forge::Credential>, listed: &[AccessToken]) -> Decision {
    let ours: Vec<&AccessToken> = listed
        .iter()
        .filter(|t| t.name.as_deref() == Some(AGENT_TOKEN_NAME))
        .collect();
    let token = match ours[..] {
        [] => return Decision::Mint(MintReason::NotOnForge),
        [token] => token,
        _ => return Decision::Conflict,
    };
    let Some(stored) = stored else {
        return Decision::Mint(MintReason::NotStored);
    };
    if token.token_last_eight.as_deref() != last_eight(&stored.value) {
        return Decision::Mint(MintReason::LastEightMismatch);
    }
    let listed_scopes = normalized_scopes(token.scopes.iter().flatten().map(String::as_str));
    if listed_scopes != normalized_scopes(AGENT_TOKEN_SCOPES.split(',')) {
        return Decision::Mint(MintReason::ScopeMismatch);
    }
    Decision::Keep
}

/// What one pass found for one agent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Observed {
    /// Both reads worked, and this is what [`classify`] made of them.
    Decided(Decision),
    /// The forge has no user by this agent's name. Planned like a mint: the
    /// queued job creates the user first (`CreateForgeUser`, idempotent), so
    /// an agent that holds a store identity but was never given a forge
    /// account still ends up with both.
    NoForgeUser,
    /// A read failed. Nothing is known, so nothing is done.
    Unknown,
}

/// The agents a pass mints for: a [`Decision::Mint`], or an agent with no
/// forge user yet ([`Observed::NoForgeUser`]), whose job creates the user
/// before it mints.
///
/// [`Observed::Unknown`] in particular never mints: a forge or store outage
/// must not turn into a rotation of every agent's token.
pub fn plan(observed: &[(String, Observed)]) -> Vec<String> {
    observed
        .iter()
        .filter(|(_, o)| {
            matches!(
                o,
                Observed::Decided(Decision::Mint(_)) | Observed::NoForgeUser
            )
        })
        .map(|(agent, _)| agent.clone())
        .collect()
}

/// Whether a forge error is a 404, whichever of its two shapes it came in.
pub(super) fn is_not_found(e: &ForgejoError) -> bool {
    match e {
        ForgejoError::ApiError(api) => {
            matches!(api.error_kind(), ApiErrorKind::NotFound { .. })
                || matches!(api.error_kind(), ApiErrorKind::Other(s) if *s == StatusCode::NOT_FOUND)
        }
        ForgejoError::UnexpectedStatusCode(s) => *s == StatusCode::NOT_FOUND,
        _ => false,
    }
}

impl Client {
    /// Every access token the forge lists for `agent`, or `None` when the
    /// forge has no such user.
    pub(super) async fn list_agent_tokens(&self, agent: &str) -> Result<Option<Vec<AccessToken>>> {
        match self.api.admin_list_user_access_tokens(agent).all().await {
            Ok(tokens) => Ok(Some(tokens)),
            Err(e) if is_not_found(&e) => Ok(None),
            Err(e) => Err(e).with_context(|| format!("listing {agent}'s forge tokens")),
        }
    }

    /// Make sure `agent`'s stored token is a live [`AGENT_TOKEN_NAME`] token
    /// with [`AGENT_TOKEN_SCOPES`], and mint one if it is not. The whole job
    /// of the `MintAgentForgeToken` node.
    ///
    /// A rotation deletes the old token first, then creates, stores, and uses
    /// the new one: it logs in with it and asks the forge who it is, so the
    /// node does not report success on a write nobody has read. A crash
    /// between any two steps leaves a state the next run classifies as
    /// [`Decision::Mint`] again.
    ///
    /// # Errors
    /// When the store or the forge refuses a step, when the agent has no forge
    /// user, or when the new token authenticates as someone else.
    pub async fn ensure_agent_forge_token(&self, agent: &str) -> Result<()> {
        let path = forge::agent_token_path(agent)?;
        let store = crate::store::connect()
            .await
            .context("logging in to the swarm secret store")?;
        let stored: Option<forge::Credential> = store
            .read_optional(&path)
            .await
            .with_context(|| format!("reading {path}"))?;
        let Some(listed) = self.list_agent_tokens(agent).await? else {
            bail!(
                "the forge has no user {agent:?} even after this job's create_forge_user \
                 node, so there is nothing to mint a token for"
            );
        };
        let reason = match classify(stored.as_ref(), &listed) {
            Decision::Keep => {
                tracing::debug!(agent, "agent forge token is current; left as it is");
                return Ok(());
            }
            Decision::Conflict => bail!(
                "the forge lists more than one {AGENT_TOKEN_NAME:?} token for {agent:?}; \
                 delete them by id and re-run"
            ),
            Decision::Mint(reason) => reason,
        };

        if listed
            .iter()
            .any(|t| t.name.as_deref() == Some(AGENT_TOKEN_NAME))
        {
            match self
                .api
                .admin_delete_user_access_token(agent, AGENT_TOKEN_NAME)
                .await
            {
                Ok(()) => {}
                Err(e) if is_not_found(&e) => {}
                Err(e) => {
                    return Err(e)
                        .with_context(|| format!("deleting {agent}'s {AGENT_TOKEN_NAME} token"));
                }
            }
        }
        let created = self
            .api
            .admin_create_user_access_token(
                agent,
                CreateAccessTokenOption {
                    name: AGENT_TOKEN_NAME.to_owned(),
                    repositories: None,
                    scopes: Some(AGENT_TOKEN_SCOPES.split(',').map(str::to_owned).collect()),
                },
            )
            .await
            .with_context(|| format!("creating {agent}'s {AGENT_TOKEN_NAME} token"))?;
        let Some(value) = created.sha1.filter(|v| !v.is_empty()) else {
            bail!("the forge created {agent}'s token but returned no value for it");
        };
        let credential = forge::Credential {
            value,
            name: AGENT_TOKEN_NAME.to_owned(),
        };
        store
            .write(&path, &credential)
            .await
            .with_context(|| format!("storing {agent}'s forge token at {path}"))?;

        let as_agent = Forgejo::new(Auth::Token(&credential.value), self.url.clone())
            .context("building a forge client with the new token")?;
        let me = as_agent
            .user_get_current()
            .await
            .with_context(|| format!("using {agent}'s new forge token"))?;
        if me.login.as_deref() != Some(agent) {
            bail!("{agent}'s new forge token authenticates as someone else");
        }
        tracing::info!(agent, ?reason, %path, "agent forge token minted and stored");
        Ok(())
    }

    /// What the forge and the store say about `agent`'s token.
    pub(super) async fn observe_agent(
        &self,
        store: &swarm_secret_client::SecretStore,
        agent: &str,
    ) -> Observed {
        let stored = match forge::agent_token_path(agent) {
            Ok(path) => store.read_optional::<forge::Credential>(&path).await,
            Err(e) => Err(e),
        };
        let stored = match stored {
            Ok(stored) => stored,
            Err(e) => {
                tracing::warn!(agent, error = %e, "agent forge token: store read failed");
                return Observed::Unknown;
            }
        };
        match self.list_agent_tokens(agent).await {
            Ok(Some(listed)) => Observed::Decided(classify(stored.as_ref(), &listed)),
            Ok(None) => Observed::NoForgeUser,
            Err(e) => {
                tracing::warn!(agent, error = %format!("{e:#}"), "agent forge token: forge read failed");
                Observed::Unknown
            }
        }
    }

    /// One pass: every agent holding a store identity, observed.
    async fn observe_all(&self) -> Result<Vec<(String, Observed)>> {
        let store = crate::store::connect()
            .await
            .context("logging in to the swarm secret store")?;
        let roles = store
            .list_cert_roles(DEFAULT_CERT_MOUNT)
            .await
            .context("listing the store's cert-auth roles")?;
        let mut observed = Vec::new();
        for agent in policy::agents_from_role_names(&roles) {
            let o = self.observe_agent(&store, &agent).await;
            if o == Observed::NoForgeUser {
                tracing::info!(
                    agent,
                    "agent forge token: no forge user; creating one first"
                );
            }
            observed.push((agent, o));
        }
        Ok(observed)
    }
}

/// Check every agent's token now and every [`RECONCILE_INTERVAL`] after, and
/// hand the agents that need one to `enqueue`, which inserts a
/// `MintAgentForgeToken` node for each. Also sweeps each `Keep`-decided
/// agent's legacy `hyperhive-<seconds>` tokens off the same observation
/// (`super::legacy_tokens::sweepable`), so a boot where the first tick's
/// store or forge read fails retries the sweep on the next tick instead of
/// leaving those tokens live until a restart.
///
/// The roster is the store's `hive-agent-*` cert-auth roles: exactly the
/// agents that can read what the node writes. The first tick fires
/// immediately (`tokio::time::interval`'s default). A pass that fails is
/// logged and retried on the next tick; it never stops the daemon.
pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + Send + 'static) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
        loop {
            ticker.tick().await;
            match client.observe_all().await {
                Ok(observed) => {
                    let legacy = super::legacy_tokens::sweepable(&observed);
                    client.sweep_legacy_tokens(&legacy).await;

                    let agents = plan(&observed);
                    if agents.is_empty() {
                        tracing::debug!(
                            checked = observed.len(),
                            "agent forge tokens: all current"
                        );
                    } else {
                        tracing::info!(
                            checked = observed.len(),
                            minting = agents.len(),
                            "agent forge tokens: queueing mints"
                        );
                        enqueue(agents);
                    }
                }
                Err(e) => tracing::warn!(
                    error = %format!("{e:#}"),
                    retry_in_s = RECONCILE_INTERVAL.as_secs(),
                    "agent forge tokens: pass failed; retrying next tick"
                ),
            }
        }
    });
}

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

    const VALUE: &str = "0123456789abcdef0123456789abcdef01234567";

    fn stored() -> forge::Credential {
        forge::Credential {
            value: VALUE.to_owned(),
            name: AGENT_TOKEN_NAME.to_owned(),
        }
    }

    fn token(name: &str, last_eight: &str, scopes: &[&str]) -> AccessToken {
        AccessToken {
            created_at: None,
            id: Some(1),
            name: Some(name.to_owned()),
            repositories: None,
            scopes: Some(scopes.iter().map(|s| (*s).to_owned()).collect()),
            sha1: None,
            token_last_eight: Some(last_eight.to_owned()),
        }
    }

    /// What Forgejo v16 stores and lists for [`AGENT_TOKEN_SCOPES`]: the
    /// `toScope` order, with each `read:X` folded into its `write:X`.
    const LISTED_SCOPES: [&str; 6] = [
        "write:misc",
        "write:notification",
        "write:organization",
        "write:issue",
        "write:repository",
        "write:user",
    ];

    fn ours() -> AccessToken {
        token(AGENT_TOKEN_NAME, "01234567", &LISTED_SCOPES)
    }

    #[test]
    fn the_scopes_are_hive_c0res_byte_for_byte() {
        assert_eq!(
            AGENT_TOKEN_SCOPES,
            "read:user,write:user,read:notification,write:notification,write:repository,write:issue,write:organization,write:misc"
        );
    }

    #[test]
    fn a_current_token_is_kept() {
        assert_eq!(classify(Some(&stored()), &[ours()]), Decision::Keep);
    }

    #[test]
    fn scopes_as_written_are_kept_too() {
        // A forge that listed the scopes un-normalised must not rotate either.
        let t = token(
            AGENT_TOKEN_NAME,
            "01234567",
            &AGENT_TOKEN_SCOPES.split(',').collect::<Vec<_>>(),
        );
        assert_eq!(classify(Some(&stored()), &[t]), Decision::Keep);
    }

    #[test]
    fn nothing_stored_mints() {
        assert_eq!(
            classify(None, &[ours()]),
            Decision::Mint(MintReason::NotStored)
        );
    }

    #[test]
    fn nothing_on_the_forge_mints() {
        assert_eq!(
            classify(Some(&stored()), &[]),
            Decision::Mint(MintReason::NotOnForge)
        );
        assert_eq!(classify(None, &[]), Decision::Mint(MintReason::NotOnForge));
    }

    #[test]
    fn a_different_last_eight_mints() {
        let t = token(AGENT_TOKEN_NAME, "ffffffff", &LISTED_SCOPES);
        assert_eq!(
            classify(Some(&stored()), &[t]),
            Decision::Mint(MintReason::LastEightMismatch)
        );
    }

    #[test]
    fn a_missing_scope_mints() {
        let t = token(AGENT_TOKEN_NAME, "01234567", &LISTED_SCOPES[1..]);
        assert_eq!(
            classify(Some(&stored()), &[t]),
            Decision::Mint(MintReason::ScopeMismatch)
        );
    }

    #[test]
    fn an_extra_scope_mints() {
        let mut scopes = LISTED_SCOPES.to_vec();
        scopes.push("write:admin");
        let t = token(AGENT_TOKEN_NAME, "01234567", &scopes);
        assert_eq!(
            classify(Some(&stored()), &[t]),
            Decision::Mint(MintReason::ScopeMismatch)
        );
    }

    #[test]
    fn scope_order_does_not_matter() {
        let mut scopes = LISTED_SCOPES.to_vec();
        scopes.reverse();
        let t = token(AGENT_TOKEN_NAME, "01234567", &scopes);
        assert_eq!(classify(Some(&stored()), &[t]), Decision::Keep);
    }

    #[test]
    fn legacy_tokens_are_ignored() {
        // A pile of `hyperhive-*` tokens, one of them even matching the
        // stored value's last eight, is still "no swarm token".
        let legacy = [
            token("hyperhive-1700000000", "01234567", &LISTED_SCOPES),
            token("hyperhive-1700000001", "aaaaaaaa", &LISTED_SCOPES),
        ];
        assert_eq!(
            classify(Some(&stored()), &legacy),
            Decision::Mint(MintReason::NotOnForge)
        );
        let mut with_ours = legacy.to_vec();
        with_ours.push(ours());
        assert_eq!(classify(Some(&stored()), &with_ours), Decision::Keep);
    }

    #[test]
    fn two_swarm_tokens_are_a_conflict_not_a_mint() {
        assert_eq!(
            classify(Some(&stored()), &[ours(), ours()]),
            Decision::Conflict
        );
    }

    #[test]
    fn last_eight_is_the_tail() {
        assert_eq!(last_eight(VALUE), Some("01234567"));
        assert_eq!(last_eight("short"), None);
    }

    #[test]
    fn a_mint_or_a_missing_user_is_planned() {
        let observed = [
            ("a".to_owned(), Observed::Decided(Decision::Keep)),
            (
                "b".to_owned(),
                Observed::Decided(Decision::Mint(MintReason::NotStored)),
            ),
            ("c".to_owned(), Observed::Unknown),
            ("d".to_owned(), Observed::NoForgeUser),
            ("e".to_owned(), Observed::Decided(Decision::Conflict)),
            (
                "f".to_owned(),
                Observed::Decided(Decision::Mint(MintReason::ScopeMismatch)),
            ),
        ];
        assert_eq!(plan(&observed), ["b", "d", "f"]);
    }

    /// An agent with a store identity and no forge user used to be skipped
    /// on every pass, forever. It is planned now; the job it gets creates
    /// the user before it mints (see `queue_forge_token_mints`).
    #[test]
    fn a_missing_forge_user_is_planned() {
        let observed = [("ruth".to_owned(), Observed::NoForgeUser)];
        assert_eq!(plan(&observed), ["ruth"]);
    }

    #[test]
    fn an_outage_plans_nothing() {
        let observed = [
            ("a".to_owned(), Observed::Unknown),
            ("b".to_owned(), Observed::Unknown),
        ];
        assert!(plan(&observed).is_empty());
    }

    #[test]
    fn both_404_shapes_are_not_found() {
        assert!(is_not_found(&ForgejoError::UnexpectedStatusCode(
            StatusCode::NOT_FOUND
        )));
        assert!(is_not_found(&ForgejoError::ApiError(
            ApiErrorKind::NotFound { errors: None }.into()
        )));
        assert!(!is_not_found(&ForgejoError::UnexpectedStatusCode(
            StatusCode::FORBIDDEN
        )));
    }
}