mint per-agent proxy credentials and a per-hive squid self-check credential

swarm-controller/src/main.rs

@@ -60,6 +60,7 @@ mod meta_flake;
 mod notify_flows;
 mod notify_router_api;
 mod otel_http_client;
+mod proxy_credential;
 mod pulls;
 mod queue_identity;
 mod read_policy;

swarm-controller/src/proxy_credential.rs

@@ -0,0 +1,320 @@
+//! The credentials squid authenticates its clients with: one per agent, at
+//! `swarm_secret_client::proxy::agent_credential_path`, and one per hive for
+//! that hive's squid self-check, at `proxy::selfcheck_path`.
+//!
+//! Minted here because the agent's store policy is read-only and a hive's
+//! policy cannot create under another principal's prefix. [`spawn`]'s pass runs
+//! at start and every [`RECONCILE_INTERVAL`] over every agent that holds a store
+//! identity, so an agent that existed before this pass was added gets one on the
+//! next tick, with no restart.
+//!
+//! ⚠️ **Never replaced.** A path that holds anything is left alone, and the write
+//! is a `cas=0` create. Replacing a credential is a rotation: squid's users file
+//! is rendered from whatever the store holds, and a rotation has to reach every
+//! client that carries the old one in the same step.
+
+use anyhow::{Context, Result};
+use serde::de::IgnoredAny;
+use swarm_secret_client::{Error, SecretStore, client::DEFAULT_CERT_MOUNT, policy, proxy};
+
+/// How often [`spawn`] re-checks every agent and hive.
+const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
+
+/// What [`ensure`] did at one path.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Outcome {
+    /// The path already held a credential.
+    Present,
+    /// This pass wrote one.
+    Minted,
+    /// Another writer's create landed between this pass's read and its own.
+    LostRace,
+}
+
+/// The two store calls [`ensure`] makes, so a test can stand in for the store.
+pub(crate) trait CredentialStore {
+    /// Whether anything is stored at `path`, as [`SecretStore::read_optional`]
+    /// answers it. The value is decoded as [`IgnoredAny`] and never held.
+    fn exists(&self, path: &str) -> impl Future<Output = Result<bool, Error>> + Send;
+
+    /// As [`SecretStore::create`].
+    fn create(
+        &self,
+        path: &str,
+        value: &proxy::Credential,
+    ) -> impl Future<Output = Result<(), Error>> + Send;
+}
+
+impl CredentialStore for SecretStore {
+    async fn exists(&self, path: &str) -> Result<bool, Error> {
+        Ok(self.read_optional::<IgnoredAny>(path).await?.is_some())
+    }
+
+    async fn create(&self, path: &str, value: &proxy::Credential) -> Result<(), Error> {
+        SecretStore::create(self, path, value).await
+    }
+}
+
+/// Make sure `path` holds a credential, writing one only if it holds nothing.
+///
+/// A create that fails is read back: if the path now holds something, a
+/// concurrent writer won and that is the credential.
+///
+/// # Errors
+/// When the store refuses the read, or the create fails and the path is still
+/// empty.
+pub(crate) async fn ensure(store: &impl CredentialStore, path: &str) -> Result<Outcome> {
+    if store
+        .exists(path)
+        .await
+        .with_context(|| format!("reading {path}"))?
+    {
+        return Ok(Outcome::Present);
+    }
+    let credential = proxy::Credential {
+        value: crate::agent_identity::generate_queue_secret()?,
+    };
+    let Err(e) = store.create(path, &credential).await else {
+        return Ok(Outcome::Minted);
+    };
+    if store
+        .exists(path)
+        .await
+        .with_context(|| format!("reading {path} back after a failed create"))?
+    {
+        return Ok(Outcome::LostRace);
+    }
+    Err(anyhow::Error::new(e).context(format!("creating {path}")))
+}
+
+/// [`ensure`] at each of `paths`, which `what` names in the log. Returns how
+/// many it minted. A failure at one path is logged and the rest still run.
+async fn ensure_all(
+    store: &impl CredentialStore,
+    what: &str,
+    paths: Vec<(String, Result<String, Error>)>,
+) -> usize {
+    let mut minted = 0;
+    for (name, path) in paths {
+        let path = match path {
+            Ok(path) => path,
+            Err(e) => {
+                tracing::warn!(name, error = %e, "{what}: unusable name");
+                continue;
+            }
+        };
+        match ensure(store, &path).await {
+            Ok(Outcome::Present) => {}
+            Ok(Outcome::Minted) => {
+                minted += 1;
+                tracing::info!(name, %path, "{what} minted");
+            }
+            Ok(Outcome::LostRace) => {
+                tracing::info!(name, %path, "{what}: another writer created it first");
+            }
+            Err(e) => tracing::warn!(name, error = %format!("{e:#}"), "{what}: mint failed"),
+        }
+    }
+    minted
+}
+
+/// One pass: a credential for every agent in `agents` and a self-check
+/// credential for every hive in `hives`. Returns how many were minted.
+async fn mint_missing(store: &impl CredentialStore, agents: &[String], hives: &[String]) -> usize {
+    let for_agents = agents
+        .iter()
+        .map(|a| (a.clone(), proxy::agent_credential_path(a)))
+        .collect();
+    let for_hives = hives
+        .iter()
+        .map(|h| (h.clone(), proxy::selfcheck_path(h)))
+        .collect();
+    ensure_all(store, "agent proxy credential", for_agents).await
+        + ensure_all(store, "hive squid self-check credential", for_hives).await
+}
+
+/// The agents that hold a store identity: exactly the ones whose hive reads
+/// what a pass writes.
+async fn roster(store: &SecretStore) -> Result<Vec<String>> {
+    let roles = store
+        .list_cert_roles(DEFAULT_CERT_MOUNT)
+        .await
+        .context("listing the store's cert-auth roles")?;
+    Ok(policy::agents_from_role_names(&roles))
+}
+
+/// Run a pass now and every [`RECONCILE_INTERVAL`] after. A pass that fails is
+/// logged and retried on the next tick; it never stops the daemon.
+pub fn spawn(hives: Vec<String>) {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
+        loop {
+            ticker.tick().await;
+            let pass = async {
+                let store = crate::store::connect()
+                    .await
+                    .context("logging in to the swarm secret store")?;
+                let agents = roster(&store).await?;
+                anyhow::Ok((agents.len(), mint_missing(&store, &agents, &hives).await))
+            };
+            match pass.await {
+                Ok((agents, minted)) => {
+                    tracing::debug!(agents, minted, "proxy credentials: pass done");
+                }
+                Err(e) => tracing::warn!(
+                    error = %format!("{e:#}"),
+                    retry_in_s = RECONCILE_INTERVAL.as_secs(),
+                    "proxy credentials: pass failed; retrying next tick"
+                ),
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::sync::Mutex;
+
+    use super::*;
+
+    /// An in-memory store that refuses a create over a held path the way a
+    /// `cas=0` write does, and counts the creates it was asked for.
+    #[derive(Default)]
+    struct Fake {
+        held: Mutex<HashMap<String, String>>,
+        creates: Mutex<usize>,
+        /// A path whose create fails and leaves nothing behind.
+        broken: Option<String>,
+        /// Makes the next create find the path taken by someone else.
+        race: Option<String>,
+    }
+
+    impl Fake {
+        fn holding(path: &str, value: &str) -> Self {
+            let fake = Self::default();
+            fake.held
+                .lock()
+                .unwrap()
+                .insert(path.to_owned(), value.to_owned());
+            fake
+        }
+
+        fn value(&self, path: &str) -> Option<String> {
+            self.held.lock().unwrap().get(path).cloned()
+        }
+    }
+
+    fn refused() -> Error {
+        Error::MissingEnv("BAO_ADDR")
+    }
+
+    impl CredentialStore for Fake {
+        async fn exists(&self, path: &str) -> Result<bool, Error> {
+            Ok(self.held.lock().unwrap().contains_key(path))
+        }
+
+        async fn create(&self, path: &str, value: &proxy::Credential) -> Result<(), Error> {
+            *self.creates.lock().unwrap() += 1;
+            let mut held = self.held.lock().unwrap();
+            if self.broken.as_deref() == Some(path) {
+                return Err(refused());
+            }
+            if let Some(winner) = &self.race {
+                held.insert(path.to_owned(), winner.clone());
+                return Err(refused());
+            }
+            if held.contains_key(path) {
+                return Err(refused());
+            }
+            held.insert(path.to_owned(), value.value.clone());
+            Ok(())
+        }
+    }
+
+    const ATLAS: &str = "swarm/agents/atlas/proxy-credential";
+    const ARGUS: &str = "swarm/agents/argus/proxy-credential";
+    const H1: &str = "swarm/hives/h1/squid-selfcheck";
+
+    fn names(names: &[&str]) -> Vec<String> {
+        names.iter().map(|n| (*n).to_owned()).collect()
+    }
+
+    #[tokio::test]
+    async fn an_absent_credential_is_created() {
+        let store = Fake::default();
+        assert_eq!(ensure(&store, ATLAS).await.unwrap(), Outcome::Minted);
+        let value = store.value(ATLAS).expect("written");
+        // 32 bytes, base64url without padding.
+        assert_eq!(value.len(), 43, "{value}");
+        assert!(
+            value
+                .bytes()
+                .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+        );
+    }
+
+    #[tokio::test]
+    async fn a_present_credential_is_untouched_and_never_updated() {
+        let store = Fake::holding(ATLAS, "existing");
+        assert_eq!(ensure(&store, ATLAS).await.unwrap(), Outcome::Present);
+        assert_eq!(store.value(ATLAS).as_deref(), Some("existing"));
+        assert_eq!(*store.creates.lock().unwrap(), 0, "no write was attempted");
+    }
+
+    #[tokio::test]
+    async fn a_lost_race_keeps_the_winners_value() {
+        let store = Fake {
+            race: Some("winner".to_owned()),
+            ..Fake::default()
+        };
+        assert_eq!(ensure(&store, ATLAS).await.unwrap(), Outcome::LostRace);
+        assert_eq!(store.value(ATLAS).as_deref(), Some("winner"));
+    }
+
+    #[tokio::test]
+    async fn a_failed_create_with_nothing_behind_it_is_an_error() {
+        let store = Fake {
+            broken: Some(ATLAS.to_owned()),
+            ..Fake::default()
+        };
+        assert!(ensure(&store, ATLAS).await.is_err());
+        assert_eq!(store.value(ATLAS), None);
+    }
+
+    #[tokio::test]
+    async fn a_pass_mints_for_every_agent_and_hive_that_lacks_one() {
+        let store = Fake::holding(ARGUS, "existing");
+        let minted = mint_missing(&store, &names(&["atlas", "argus"]), &names(&["h1"])).await;
+        assert_eq!(minted, 2);
+        assert!(store.value(ATLAS).is_some());
+        assert!(store.value(H1).is_some());
+        assert_eq!(store.value(ARGUS).as_deref(), Some("existing"));
+        // Control: the second pass has nothing left to mint, and writes nothing.
+        let creates = *store.creates.lock().unwrap();
+        assert_eq!(
+            mint_missing(&store, &names(&["atlas", "argus"]), &names(&["h1"])).await,
+            0
+        );
+        assert_eq!(*store.creates.lock().unwrap(), creates);
+    }
+
+    #[tokio::test]
+    async fn one_agent_failing_does_not_stop_the_rest() {
+        let store = Fake {
+            broken: Some(ATLAS.to_owned()),
+            ..Fake::default()
+        };
+        let minted = mint_missing(&store, &names(&["atlas", "argus"]), &names(&["h1"])).await;
+        assert_eq!(minted, 2);
+        assert!(store.value(ARGUS).is_some());
+        assert!(store.value(H1).is_some());
+    }
+
+    #[tokio::test]
+    async fn an_unusable_agent_name_mints_nothing() {
+        let store = Fake::default();
+        assert_eq!(mint_missing(&store, &names(&["../argus"]), &[]).await, 0);
+        assert!(store.held.lock().unwrap().is_empty());
+    }
+}

swarm-controller/src/startup.rs

@@ -368,7 +368,7 @@ fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, hives: &[H
 /// The per-agent background passes: subagent and terminal streams, and
 /// schedule keys, and the permissions bucket and its import, and the
 /// archetypes agents are created from. Beside them, the
-/// copy of the cache facts to each hive.
+/// copy of the cache facts to each hive, and the proxy credentials squid checks.
 /// Lifted out of `main` for `clippy::too_many_lines`.
 pub(crate) fn spawn_per_agent_upkeep(
     status: Option<&Arc<status::StatusReader>>,
@@ -378,6 +378,7 @@ pub(crate) fn spawn_per_agent_upkeep(
     spawn_schedule_reconcile(status, hives);
     spawn_agent_permissions(status, hives);
     crate::attic_copy::spawn(hives.iter().map(|h| h.name.clone()).collect());
+    crate::proxy_credential::spawn(hives.iter().map(|h| h.name.clone()).collect());
     if let Some(status) = status {
         archetypes::spawn(status.queue_client());
     }