hyperhive/swarm-controller/src/read_policy.rs

//! Provisioning a hive's access to the swarm's secret store: the policy its
//! certificate carries, and the cert-auth role that hands it that policy.
//!
//! The controller writes a credential; the hive fetches it with its own
//! certificate. Without the role the certificate authenticates to nothing, and
//! without the policy the token it gets may read nothing — either way the read
//! half of a delivery answers 403.
//!
//! A pass runs at startup, and repeats until it reaches the store. The two
//! start together, so the store being the slower one is the ordinary case
//! rather than an edge one, and a pass that gave up there provisions nobody
//! until something restarts this daemon. Per-hive failures are different: they
//! are reported and skipped — one hive that cannot be provisioned should not
//! take down a daemon that serves everything else, and a store that answers is
//! answering the same way next time, so the next start is the retry for those.
//!
//! The document names the hive it is written for: one stanza is the same
//! everywhere, the other is scoped to the reader's own name
//! ([`swarm_secret_client::policy`] explains the asymmetry). The same rule is
//! also written once for all hives, [`policy::SHARED_HIVE_POLICY`], with the
//! reader's own name taken from its login's alias on the cert mount instead of
//! written into the text; a hive's role attaches both. The hive list is loaded
//! once because a config change means a redeploy.
//!
//! Every agent's policy is rewritten by the same kind of pass
//! ([`ensure_agent_policies`]), so a change to `policy::render_agent` reaches
//! agents minted before it. Its roster is the store's `hive-agent-*` cert-auth
//! roles.

use std::{future::Future, time::Duration};

use anyhow::{Context, Result};
use backon::{BackoffBuilder as _, ConstantBuilder};
use swarm_secret_client::{Error, SecretStore, client::DEFAULT_CERT_MOUNT, policy};

/// File holding the authority hives are issued from, as
/// `swarm-controller.nix` names it.
pub const ENV_HIVE_CLIENT_CA: &str = "SWARM_CONTROLLER_HIVE_CLIENT_CA_FILE";

/// How long to wait before re-attempting a pass that never reached the store.
///
/// `nix/host-modules/swarm-bao.nix`'s bootstrap unit retries the other side of
/// this same race on the same cadence: the two halves of one race should not
/// disagree about how long a wait is worth.
const RETRY_INTERVAL: Duration = Duration::from_secs(30);

/// How many times [`RETRY_INTERVAL`] is waited before giving up. A store that
/// has not appeared within [`RETRY_WINDOW`] is absent rather than slow, and a
/// daemon logging into the void forever tells an operator nothing.
const RETRY_ATTEMPTS: u32 = 2_880;

/// What [`RETRY_ATTEMPTS`] at [`RETRY_INTERVAL`] come to. A named third
/// constant because it is the claim the other two make together, and either
/// one of them can be edited alone.
const RETRY_WINDOW: Duration = Duration::from_hours(24);

/// A pass that never got as far as the store, carrying what to say about it.
///
/// Returned rather than logged where it happens: the first attempt deserves a
/// warning and the two-hundredth is noise, and only the caller knows which one
/// this is.
struct Unreachable(String);

/// Give every hive in `hives` the access its certificate needs.
///
/// Returns once the first pass is done, so a store that is already up leaves
/// nothing deferred. When that pass cannot reach the store, it repeats in the
/// background for up to [`RETRY_WINDOW`].
pub async fn ensure_hive_access(hives: Vec<String>) {
    if let Err(Unreachable(reason)) = provision(&hives).await {
        tracing::warn!(
            reason,
            retry_in = ?RETRY_INTERVAL,
            "hive provisioning could not reach the swarm secret store; retrying in the background"
        );
        spawn_retry(hives);
    }
}

/// Repeat [`provision`] until it reaches the store, or [`RETRY_WINDOW`] passes.
fn spawn_retry(hives: Vec<String>) {
    tokio::spawn(async move {
        let reached = retry_until_ok(RETRY_INTERVAL, RETRY_ATTEMPTS, move || {
            let hives = hives.clone();
            async move {
                let outcome = provision(&hives).await;
                if let Err(Unreachable(reason)) = &outcome {
                    tracing::debug!(reason, "hive provisioning: store still unreachable");
                }
                outcome
            }
        })
        .await;
        if let Some(attempt) = reached {
            tracing::info!(attempt, "hive provisioning reached the store");
        } else {
            tracing::warn!(
                attempts = RETRY_ATTEMPTS,
                window = ?RETRY_WINDOW,
                "giving up on hive provisioning; no hive can log in to the store until this daemon is restarted"
            );
        }
    });
}

/// Run `attempt` every `interval` until it succeeds, at most `attempts` times.
/// Answers with the attempt that succeeded, or `None` if none did.
///
/// The cadence and the bound are parameters rather than the constants above so
/// that a test can drive this loop without waiting out a real day.
async fn retry_until_ok<F, Fut>(interval: Duration, attempts: u32, mut attempt: F) -> Option<u32>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<(), Unreachable>>,
{
    // One delay per attempt, waited before it: the pass the caller already ran
    // before deciding to retry at all is the attempt that needs no wait.
    let delays = ConstantBuilder::default()
        .with_delay(interval)
        .with_max_times(attempts as usize)
        .build();
    for (n, delay) in (1..).zip(delays) {
        tokio::time::sleep(delay).await;
        if attempt().await.is_ok() {
            return Some(n);
        }
    }
    None
}

/// One pass: log in once, write every hive's policy, then every hive's role.
///
/// Policy first, because the role names the policy — the other order leaves a
/// window in which the role points at nothing.
///
/// # Errors
/// [`Unreachable`] for the states a later attempt can resolve: a store that is
/// not up, and an authority file that is not placed yet. A deployment that is
/// configured for neither is `Ok` — there is nothing to wait for.
async fn provision(hives: &[String]) -> Result<(), Unreachable> {
    if hives.is_empty() {
        return Ok(());
    }
    let store = match crate::store::connect().await {
        Ok(store) => store,
        // An unset `BAO_*` variable is a deployment that runs no store, not
        // one whose store is late: asking again changes nothing.
        Err(Error::MissingEnv(var)) => {
            tracing::info!(
                var,
                "no secret store configured; hive access is not managed here"
            );
            return Ok(());
        }
        Err(e) => return Err(Unreachable(e.to_string())),
    };
    for hive in hives {
        if let Err(e) = write_policy_for(&store, hive).await {
            tracing::warn!(hive, error = %format!("{e:#}"), "writing this hive's read policy failed");
        }
    }
    let shared = write_shared_policy(&store).await;

    let Some(path) = std::env::var_os(ENV_HIVE_CLIENT_CA) else {
        tracing::info!("no hive certificate authority configured; hive roles are not managed here");
        return Ok(());
    };
    let path = path.to_string_lossy().into_owned();
    // Retryable rather than fatal: the unit that places this file and this
    // daemon start together, so "not there yet" is a moment, not a verdict.
    let ca = std::fs::read_to_string(&path)
        .map_err(|e| Unreachable(format!("reading the hive authority {path}: {e}")))?;
    for hive in hives {
        if let Err(e) = write_role(&store, hive, &ca, shared).await {
            tracing::warn!(hive, error = %format!("{e:#}"), "creating this hive's store role failed");
        }
    }
    Ok(())
}

/// One hive's policy: its own name, its own document.
async fn write_policy_for(store: &SecretStore, hive: &str) -> Result<()> {
    let name = policy::hive_object_name(hive)?;
    store
        .write_policy(&name, &policy::render(hive)?)
        .await
        .with_context(|| format!("writing the read policy {name}"))?;
    tracing::info!(hive, policy = %name, "hive read policy in place");
    Ok(())
}

/// Write [`policy::SHARED_HIVE_POLICY`], rendered with the cert mount's
/// accessor. Answers whether it is in place: a hive's role attaches it only
/// then.
///
/// Best effort. A hive's own policy grants it the same paths, so a store that
/// refuses the accessor read (the controller's policy may not carry it yet) or
/// the write costs the shared policy and nothing else.
async fn write_shared_policy(store: &SecretStore) -> bool {
    let accessor = match store.auth_mount_accessor(DEFAULT_CERT_MOUNT).await {
        Ok(accessor) => accessor,
        Err(e) => {
            tracing::warn!(
                error = %e,
                "the shared hive policy is not managed: the cert auth mount's accessor could not be read; hives keep their own policy"
            );
            return false;
        }
    };
    let document = match policy::render_shared_hives(&accessor) {
        Ok(document) => document,
        Err(e) => {
            tracing::warn!(error = %e, "the shared hive policy is not managed: the cert auth mount's accessor is not usable in a policy; hives keep their own policy");
            return false;
        }
    };
    match store
        .write_policy(policy::SHARED_HIVE_POLICY, &document)
        .await
    {
        Ok(()) => {
            tracing::info!(
                policy = policy::SHARED_HIVE_POLICY,
                "shared hive policy in place"
            );
            true
        }
        Err(e) => {
            tracing::warn!(
                policy = policy::SHARED_HIVE_POLICY,
                error = %e,
                "writing the shared hive policy failed; hives keep their own policy"
            );
            false
        }
    }
}

/// The policies `hive`'s role attaches: its own, and the shared one when that
/// is in place.
fn attached_policies(hive: &str, shared: bool) -> Result<Vec<String>, Error> {
    let mut attached = vec![policy::hive_object_name(hive)?];
    if shared {
        attached.push(policy::SHARED_HIVE_POLICY.to_owned());
    }
    Ok(attached)
}

/// The role for one hive: its own certificate's subject, its own policy, and
/// the shared one when `shared`.
async fn write_role(store: &SecretStore, hive: &str, ca: &str, shared: bool) -> Result<()> {
    let name = policy::hive_object_name(hive)?;
    let attached = attached_policies(hive, shared)?;
    store
        // The subject is the hive's name: `glue-bao-tls.nix` mints a hive's
        // client leaf with its name as the CN, and cert auth matches on that.
        .write_cert_role(
            DEFAULT_CERT_MOUNT,
            &name,
            ca,
            hive,
            &attached.iter().map(String::as_str).collect::<Vec<_>>(),
        )
        .await
        .with_context(|| format!("writing the cert-auth role {name}"))?;
    tracing::info!(hive, role = %name, "hive store role in place");
    Ok(())
}

/// Rewrite every agent's policy from [`policy::render_agent`], with the retry
/// [`ensure_hive_access`] has and for the same reason.
///
/// Returns once the first pass is done. When that pass cannot reach the store,
/// it repeats in the background for up to [`RETRY_WINDOW`].
pub async fn ensure_agent_policies() {
    if let Err(Unreachable(reason)) = rewrite_agent_policies().await {
        tracing::warn!(
            reason,
            retry_in = ?RETRY_INTERVAL,
            "agent policy rewrite could not reach the swarm secret store; retrying in the background"
        );
        tokio::spawn(async {
            let reached = retry_until_ok(RETRY_INTERVAL, RETRY_ATTEMPTS, || async {
                let outcome = rewrite_agent_policies().await;
                if let Err(Unreachable(reason)) = &outcome {
                    tracing::debug!(reason, "agent policy rewrite: store still unreachable");
                }
                outcome
            })
            .await;
            if let Some(attempt) = reached {
                tracing::info!(attempt, "agent policy rewrite reached the store");
            } else {
                tracing::warn!(
                    attempts = RETRY_ATTEMPTS,
                    window = ?RETRY_WINDOW,
                    "giving up on the agent policy rewrite; agents keep the policy they were minted with until this daemon is restarted"
                );
            }
        });
    }
}

/// One pass: log in once, list the agents, write the shared agent policy and
/// each one's own, and give each its identity entity.
///
/// Policies and entities only. An agent's cert-auth role and certificate are
/// `agent_identity::mint_and_verify`'s, and that reissues the certificate, so
/// it is not what a pass on every start may call.
///
/// # Errors
/// [`Unreachable`] when the store cannot be reached or will not list its
/// roles: either way no agent was written, and a later attempt may get
/// further. A deployment with no store configured is `Ok`.
async fn rewrite_agent_policies() -> Result<(), Unreachable> {
    let store = match crate::store::connect().await {
        Ok(store) => store,
        Err(Error::MissingEnv(var)) => {
            tracing::info!(
                var,
                "no secret store configured; agent policies are not managed here"
            );
            return Ok(());
        }
        Err(e) => return Err(Unreachable(e.to_string())),
    };
    let roles = store
        .list_cert_roles(DEFAULT_CERT_MOUNT)
        .await
        .map_err(|e| Unreachable(format!("listing the store's cert-auth roles: {e}")))?;
    // First, so the cert-auth role that names it never points at nothing.
    match store
        .write_policy(policy::SHARED_AGENT_POLICY, &policy::render_shared_agent())
        .await
    {
        Ok(()) => tracing::info!(
            policy = policy::SHARED_AGENT_POLICY,
            "shared agent policy in place"
        ),
        Err(e) => tracing::warn!(
            policy = policy::SHARED_AGENT_POLICY,
            error = %e,
            "writing the shared agent policy failed"
        ),
    }
    let policies = agent_policies(&roles);
    let mut written = 0_usize;
    for (name, document) in &policies {
        match store.write_policy(name, document).await {
            Ok(()) => written += 1,
            Err(e) => {
                tracing::warn!(policy = %name, error = %e, "rewriting this agent's policy failed");
            }
        }
    }
    tracing::info!(written, agents = policies.len(), "agent policies rewritten");
    crate::agent_entity::ensure_agents(&store, &policy::agents_from_role_names(&roles)).await;
    Ok(())
}

/// The policy to write for each agent among `roles`, a listing of cert-auth
/// roles: its name, which is the role's own, and its document.
///
/// A role that is not an agent's — a hive's `hive-<hive>`, a host's — gets
/// nothing: the same `write_policy` call with an agent's document would
/// narrow that principal to one agent's subtree.
fn agent_policies(roles: &[String]) -> Vec<(String, String)> {
    policy::agents_from_role_names(roles)
        .iter()
        .filter_map(|agent| {
            Some((
                policy::agent_object_name(agent).ok()?,
                policy::render_agent(agent).ok()?,
            ))
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use std::{cell::Cell, time::Duration};

    use super::{
        RETRY_ATTEMPTS, RETRY_INTERVAL, RETRY_WINDOW, Unreachable, agent_policies,
        attached_policies, policy, retry_until_ok,
    };

    #[test]
    fn the_retry_bound_is_the_window_it_claims() {
        assert_eq!(RETRY_INTERVAL * RETRY_ATTEMPTS, RETRY_WINDOW);
    }

    /// The property this issue is about: a pass that fails because the store is
    /// not up yet happens again.
    #[tokio::test]
    async fn a_pass_is_repeated_until_it_reaches_the_store() {
        let calls = Cell::new(0_u32);
        let reached = retry_until_ok(Duration::from_millis(1), 10, || {
            calls.set(calls.get() + 1);
            let still_down = calls.get() < 3;
            async move {
                if still_down {
                    Err(Unreachable("store not up".to_owned()))
                } else {
                    Ok(())
                }
            }
        })
        .await;
        assert_eq!(reached, Some(3));
        assert_eq!(calls.get(), 3, "the loop stops at the first success");
    }

    #[tokio::test]
    async fn the_loop_is_bounded_rather_than_endless() {
        let calls = Cell::new(0_u32);
        let reached = retry_until_ok(Duration::from_millis(1), 5, || {
            calls.set(calls.get() + 1);
            async { Err(Unreachable("store not up".to_owned())) }
        })
        .await;
        assert_eq!(reached, None, "a store that never appears is given up on");
        assert_eq!(calls.get(), 5, "one attempt per interval, and no more");
    }

    #[test]
    fn the_pass_writes_agents_policies_under_their_role_names_and_no_one_elses() {
        let roles: Vec<String> = [
            "hive-agent-atlas",
            "hive-pr1ma",
            "swarm-controller",
            "hive-agent-argus",
        ]
        .map(str::to_owned)
        .to_vec();
        let written = agent_policies(&roles);
        let names: Vec<&str> = written.iter().map(|(name, _)| name.as_str()).collect();
        // The role names an agent's policy by its own name, so a policy written
        // under any other name is one the role does not attach.
        assert_eq!(names, ["hive-agent-atlas", "hive-agent-argus"]);
        assert_eq!(
            written[0].1,
            policy::render_agent("atlas").expect("legal"),
            "the prefix is stripped before rendering, or the document names `agent-atlas`"
        );
        assert_eq!(written[1].1, policy::render_agent("argus").expect("legal"));
    }

    #[test]
    fn a_hives_role_attaches_its_own_policy_and_the_shared_one() {
        assert_eq!(
            attached_policies("pr1ma", true).expect("legal"),
            ["hive-pr1ma", policy::SHARED_HIVE_POLICY]
        );
    }

    #[test]
    fn without_the_shared_policy_the_role_keeps_the_hives_own() {
        assert_eq!(
            attached_policies("pr1ma", false).expect("legal"),
            ["hive-pr1ma"]
        );
    }

    #[test]
    fn the_shared_text_serves_every_hive_what_its_own_policy_does() {
        let shared = policy::render_shared_hives("auth_cert_0a1b2c3d").expect("legal accessor");
        let template = "{{identity.entity.aliases.auth_cert_0a1b2c3d.name}}";
        for hive in ["pr1ma", "beta", "gamma-2", "delta_3"] {
            assert_eq!(
                shared.replace(template, hive),
                policy::render(hive).expect("legal"),
                "{hive}"
            );
        }
    }
}