From a55d55e494f9e910d86637f0188677c885f204e0 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 27 Oct 2026 17:32:20 +0100 Subject: [PATCH] read the cache facts on the hive and let hives read the closure bucket hive-c0re reads its cache facts from the store every five minutes and writes /run/hyperhive/attic/{netrc,substituter,public-key}, each 0600. it accepts only an https url, so the pull token never travels in clear. hives get a read grant on the agent-closures bucket, and swarm-queue-client a reader for it. part 2 of 3: nothing uses the files or the reader until the probe lands. --- docs/swarm/builder.md | 8 +- docs/swarm/credentials-reference.md | 2 + hive-c0re/Cargo.toml | 1 + hive-c0re/src/attic/mod.rs | 4 + hive-c0re/src/attic/reader.rs | 276 +++++++++++++++++++++++ hive-c0re/src/main.rs | 1 + hive-c0re/src/paths.rs | 5 + hive-c0re/src/swarm_status.rs | 3 + swarm-nats-auth/src/policy.rs | 39 ++++ swarm-queue-client/Cargo.toml | 3 + swarm-queue-client/src/agent_closures.rs | 81 +++++++ swarm-queue-client/src/lib.rs | 5 +- 12 files changed, 425 insertions(+), 3 deletions(-) create mode 100644 hive-c0re/src/attic/mod.rs create mode 100644 hive-c0re/src/attic/reader.rs diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 0bbbad4e..2af56184 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -358,7 +358,7 @@ the swarm at an existing tenant: `secret/swarm/services/attic/pull-token`, with read rights on that cache. 3. Seed the three facts hives pull with, each as field `value`: `secret/swarm/services/attic/url` (`https://` and the host, no trailing - slash), `secret/swarm/services/attic/cache` (the cache's name) and + slash; hives refuse `http://`), `secret/swarm/services/attic/cache` (the cache's name) and `secret/swarm/services/attic/public-key` (`:`, from the cache's page or `attic cache info`). The controller copies the leaves that exist. @@ -383,6 +383,12 @@ leaves (`pull-token`, `url`, `cache`, `public-key`) to own prefix. It writes a copy only when the value differs, skips a leaf the store doesn't hold, and leaves a copy whose source disappeared. +`hive-c0re` reads its four leaves every five minutes under the hive's own store +identity and writes `/run/hyperhive/attic/netrc`, `substituter` and +`public-key` (tmpfs, each `0600`). It accepts only an `https` URL, since the +token goes out in a header and in the netrc; a hive that finds another scheme +writes nothing. No deploy reads the files yet: deploys still build locally. + ## Credentials | What | Where | Who reads it | diff --git a/docs/swarm/credentials-reference.md b/docs/swarm/credentials-reference.md index 747fbda0..42f85c79 100644 --- a/docs/swarm/credentials-reference.md +++ b/docs/swarm/credentials-reference.md @@ -1,3 +1,5 @@ +| this hive's cache pull facts | `swarm-controller`, into the secret store at `swarm/hives//attic/{pull-token,url,cache,public-key}`, copied every five minutes from `swarm/services/attic/` | `hive-c0re` reads them under the hive's own store identity and writes `/run/hyperhive/attic/netrc` (the pull token), `substituter` and `public-key` (tmpfs, each `0600` `hive-core`). Nothing uses the files yet. | + # Credentials: the per-secret reference Every credential in the swarm's secret store, in one table: who mints it, who diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 54d8572e..26b49e35 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -61,6 +61,7 @@ swarm-queue-client = { workspace = true, features = [ "agent-limits", "agent-permissions", "agent-archetype", + "agent-closures", ] } swarm-matrix-client.workspace = true swarm-secret-client = { workspace = true, features = ["otel"] } diff --git a/hive-c0re/src/attic/mod.rs b/hive-c0re/src/attic/mod.rs new file mode 100644 index 00000000..7bcb8238 --- /dev/null +++ b/hive-c0re/src/attic/mod.rs @@ -0,0 +1,4 @@ +//! The swarm's binary cache, as this hive sees it: the pull credentials it is +//! given ([`reader`]). + +pub mod reader; diff --git a/hive-c0re/src/attic/reader.rs b/hive-c0re/src/attic/reader.rs new file mode 100644 index 00000000..e352239e --- /dev/null +++ b/hive-c0re/src/attic/reader.rs @@ -0,0 +1,276 @@ +//! What this hive pulls from the swarm's binary cache with: four facts +//! `swarm-controller` copies to `swarm/hives//attic/*` +//! ([`swarm_secret_client::attic`]), read here under the hive's own identity. +//! +//! [`Facts::publish`] puts them in `/run/hyperhive/attic/` as a netrc, the +//! substituter URL and the public key, where nix and the root helper read them. + +use std::path::Path; +use std::time::Duration; + +use anyhow::{Context as _, Result, bail}; +use swarm_secret_client::attic::{Fact, Leaf, hive_path}; + +/// The pull token, the cache's address and its signing key. +/// +/// No `Debug` derive: `token` reads from the cache. +pub struct Facts { + token: String, + url: String, + cache: String, + public_key: String, +} + +impl Facts { + /// The facts as stored, checked: each is one token without whitespace, + /// since they land in a netrc line and a URL. + /// + /// # Errors + /// A value that is empty, holds whitespace, or is not what its leaf is + /// (a URL with an `https` scheme and a host; a public key `:`). + /// An `http` URL is refused because the token would travel in clear. + pub fn parse(token: &str, url: &str, cache: &str, public_key: &str) -> Result { + let [token, url, cache, public_key] = [token, url, cache, public_key].map(str::trim); + for (what, v) in [ + ("pull token", token), + ("url", url), + ("cache", cache), + ("public key", public_key), + ] { + if v.is_empty() || v.contains(char::is_whitespace) { + bail!("the stored {what} is empty or holds whitespace"); + } + } + let parsed = url::Url::parse(url).context("the stored url does not parse")?; + if parsed.scheme() != "https" || parsed.host_str().is_none() { + bail!("the stored url is not an https address"); + } + if public_key + .split_once(':') + .is_none_or(|(n, k)| n.is_empty() || k.is_empty()) + { + bail!("the stored public key is not :"); + } + Ok(Self { + token: token.to_owned(), + url: url.trim_end_matches('/').to_owned(), + cache: cache.to_owned(), + public_key: public_key.to_owned(), + }) + } + + /// The URL nix names the cache by: the server, then the cache. + pub fn substituter(&self) -> String { + format!("{}/{}", self.url, self.cache) + } + + /// The netrc line the server's host authenticates with. The token is the + /// password; the server ignores the login. + fn netrc(&self) -> String { + let host = url::Url::parse(&self.url) + .ok() + .and_then(|u| u.host_str().map(str::to_owned)) + .unwrap_or_default(); + format!("machine {host} password {}\n", self.token) + } + + /// Write the three files into `dir`: `netrc`, `substituter` and + /// `public-key`. Each is `0600`, and one that already holds its content is + /// left alone. + /// + /// # Errors + /// A file that cannot be written. The ones before it are written. + pub fn publish(&self, dir: &Path) -> Result<()> { + for (name, content) in [ + ("netrc", self.netrc()), + ("substituter", format!("{}\n", self.substituter())), + ("public-key", format!("{}\n", self.public_key)), + ] { + let path = dir.join(name); + if std::fs::read(&path).is_ok_and(|have| have == content.as_bytes()) { + continue; + } + crate::secret_file::write_secret_0600(&path, content.as_bytes())?; + } + Ok(()) + } +} + +/// How often [`follow`] reads the facts. +pub const READ_INTERVAL: Duration = Duration::from_mins(5); + +/// Read the facts and put them in [`crate::paths::ATTIC_DIR`]. The facts, or +/// `None` when the swarm has none for this hive. +/// +/// # Errors +/// As [`stored`]. A file that can't be written is logged, not an error. +pub async fn sync() -> Result> { + let facts = stored().await?; + if let Some(facts) = &facts + && let Err(e) = facts.publish(Path::new(crate::paths::ATTIC_DIR)) + { + tracing::warn!(error = %format!("{e:#}"), "attic: could not write the cache files"); + } + Ok(facts) +} + +/// [`sync`] now and every [`READ_INTERVAL`] until shutdown. A failed pass is +/// logged and leaves the files as they are. +pub async fn follow(mut shutdown: tokio::sync::watch::Receiver) { + loop { + if let Err(e) = sync().await { + tracing::warn!(error = %format!("{e:#}"), "attic: reading the cache facts failed"); + } + tokio::select! { + () = tokio::time::sleep(READ_INTERVAL) => {} + _ = shutdown.changed() => return, + } + } +} + +/// This hive's facts from the store. +/// +/// `Ok(None)` when the swarm has none for it: no cache configured, the +/// controller has not copied them yet (or is too old to), or one of the four is +/// missing or unusable. The last is logged; the others are the ordinary states. +/// +/// # Errors +/// The store cannot be reached or refuses a read, which says nothing about +/// whether the facts exist. +pub async fn stored() -> Result> { + let Some(hive) = std::env::var("HYPERHIVE_HIVE_NAME") + .ok() + .filter(|h| !h.is_empty()) + else { + return Ok(None); + }; + let store = swarm_secret_client::SecretStore::from_env(&hive) + .await + .context("logging in to the swarm secret store")?; + let mut values = Vec::with_capacity(Leaf::ALL.len()); + for leaf in Leaf::ALL { + let path = hive_path(&hive, leaf)?; + let fact = store + .read_optional::(&path) + .await + .with_context(|| format!("reading {path}"))?; + let Some(fact) = fact else { + tracing::debug!(%path, "attic: no such fact in the store"); + return Ok(None); + }; + values.push(fact.value); + } + let [token, url, cache, key] = values.as_slice() else { + return Ok(None); + }; + match Facts::parse(token, url, cache, key) { + Ok(facts) => Ok(Some(facts)), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "attic: the stored cache facts are unusable"); + Ok(None) + } + } +} + +#[cfg(test)] +mod tests { + use std::os::unix::fs::PermissionsExt as _; + + use super::*; + + fn facts() -> Facts { + Facts::parse( + "tok-secret", + "https://cache.example.org/", + "swarm", + "swarm:AAAA", + ) + .expect("valid") + } + + fn mode(path: &Path) -> u32 { + std::fs::metadata(path).expect("stat").permissions().mode() & 0o777 + } + + #[test] + fn the_files_hold_the_facts_and_every_one_is_0600() { + let dir = tempfile::tempdir().expect("tempdir"); + let out = dir.path().join("attic"); + facts().publish(&out).expect("published"); + let read = |n: &str| std::fs::read_to_string(out.join(n)).expect(n); + assert_eq!( + read("netrc"), + "machine cache.example.org password tok-secret\n" + ); + assert_eq!(read("substituter"), "https://cache.example.org/swarm\n"); + assert_eq!(read("public-key"), "swarm:AAAA\n"); + for name in ["netrc", "substituter", "public-key"] { + assert_eq!(mode(&out.join(name)), 0o600, "{name}"); + } + } + + #[test] + fn a_rotated_token_replaces_the_netrc_and_a_current_file_is_left_alone() { + let dir = tempfile::tempdir().expect("tempdir"); + facts().publish(dir.path()).expect("first"); + let ino = |n: &str| { + std::os::unix::fs::MetadataExt::ino( + &std::fs::metadata(dir.path().join(n)).expect("stat"), + ) + }; + let (netrc, substituter) = (ino("netrc"), ino("substituter")); + let rotated = Facts::parse( + "tok-new", + "https://cache.example.org", + "swarm", + "swarm:AAAA", + ) + .expect("valid"); + rotated.publish(dir.path()).expect("second"); + assert_ne!(ino("netrc"), netrc, "the changed file is rewritten"); + assert_eq!(ino("substituter"), substituter, "the unchanged one is not"); + assert_eq!( + std::fs::read_to_string(dir.path().join("netrc")).expect("read"), + "machine cache.example.org password tok-new\n" + ); + } + + /// The token is sent as a header and written to a netrc, so a URL that + /// isn't https is refused whatever the host, loopback included. + #[test] + fn an_http_url_is_refused_and_an_https_one_is_not() { + for url in [ + "http://cache.example", + "http://127.0.0.1:8910", + "HTTP://cache.example", + ] { + let err = Facts::parse("tok", url, "swarm", "swarm:AAAA") + .err() + .expect(url); + assert!(err.to_string().contains("https"), "{err:#}"); + } + assert!(Facts::parse("tok", "https://cache.example", "swarm", "swarm:AAAA").is_ok()); + } + + #[test] + fn a_fact_that_would_break_a_netrc_or_a_url_is_refused() { + let ok = ("tok", "https://c.example", "swarm", "swarm:AAAA"); + for (token, url, cache, key) in [ + ("to k", ok.1, ok.2, ok.3), + ("tok\npassword x", ok.1, ok.2, ok.3), + ("", ok.1, ok.2, ok.3), + (ok.0, "ftp://c.example", ok.2, ok.3), + (ok.0, "http://c.example", ok.2, ok.3), + (ok.0, "not a url", ok.2, ok.3), + (ok.0, ok.1, "sw arm", ok.3), + (ok.0, ok.1, ok.2, "nokeyname"), + ] { + assert!( + Facts::parse(token, url, cache, key).is_err(), + "{token:?} {url:?}" + ); + } + // The control: the same call accepts good values. + assert!(Facts::parse(ok.0, ok.1, ok.2, ok.3).is_ok()); + } +} diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 8a6cef52..11b0b615 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -16,6 +16,7 @@ use clap::{Parser, Subcommand}; mod actions; mod agent_config; mod agent_set; +mod attic; mod container_view; mod coordinator; mod dashboard; diff --git a/hive-c0re/src/paths.rs b/hive-c0re/src/paths.rs index 4b039be2..732ef9fd 100644 --- a/hive-c0re/src/paths.rs +++ b/hive-c0re/src/paths.rs @@ -231,6 +231,11 @@ pub fn gateway_agents_conf() -> PathBuf { // `nix/host-modules/lib/hive-forge-git.nix` — must match. pub const FORGE_HIVE_TOKEN: &str = "/run/hyperhive/forge-hive-token"; +/// `attic/` — the swarm cache's pull credentials for nix: `netrc`, `substituter` +/// and `public-key`, each `0600`, copied from the swarm secret store. Tmpfs: +/// re-published on every boot. +pub const ATTIC_DIR: &str = "/run/hyperhive/attic"; + /// `/run/hyperhive` — the runtime root (host admin socket + per-agent dirs). #[must_use] pub fn runtime_root() -> PathBuf { diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 07d3881e..1321c06b 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -109,6 +109,9 @@ pub fn spawn( shutdown.clone(), )); + // The cache facts this hive pulls with, read into /run/hyperhive/attic. + tokio::spawn(crate::attic::reader::follow(shutdown.clone())); + // The permissions the swarm grants, and the import of the meta repo's. tokio::spawn(crate::workers::swarm_permissions::follow( client.clone(), diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 5070c748..33d85dc8 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -617,6 +617,13 @@ impl Policy { "$JS.API.DIRECT.GET.{}.>", swarm_queue_client::agent_archetype::STREAM )); + // The latest closure the swarm built for each agent + // (`swarm_wire::agent_closures`), read the same way and every key for + // the same reason. + subjects.push(format!( + "$JS.API.DIRECT.GET.{}.>", + swarm_wire::agent_closures::STREAM + )); // The one-shot import of the grants this hive's meta repo held, sent // as a request on this hive's own subject: the controller takes the // hive from the subject, so one hive cannot import for another. @@ -980,6 +987,38 @@ mod tests { ))); } + /// A hive reads any agent's closure record and can change none, and the + /// grant reaches no other bucket. + #[test] + fn a_hive_may_read_every_agents_closure_and_write_none() { + use swarm_wire::agent_closures::{STREAM, key_subject}; + + let p = policy() + .permissions("hive-alpha") + .expect("a hive is admitted"); + let may = |subject: &str| p.publish.iter().any(|g| covers(g, subject)); + + for agent in ["atlas", "iris"] { + let read = format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)); + assert!(may(&read), "cannot read {agent}"); + } + // A neighbouring bucket stays closed. + assert!(!may(&swarm_wire::agent_schedules::direct_get_subject( + "atlas" + ))); + for forbidden in [ + key_subject("atlas"), + format!("$JS.API.STREAM.CREATE.{STREAM}"), + format!("$JS.API.STREAM.UPDATE.{STREAM}"), + format!("$JS.API.STREAM.DELETE.{STREAM}"), + format!("$JS.API.STREAM.PURGE.{STREAM}"), + format!("$JS.API.STREAM.MSG.DELETE.{STREAM}"), + format!("$JS.API.CONSUMER.CREATE.{STREAM}"), + ] { + assert!(!may(&forbidden), "hive alpha may publish {forbidden}"); + } + } + /// A hive reads any agent's permissions and can change none of them; an /// agent, with the template `swarm-nats.nix` passes, reads its own key and /// no other, and writes none. diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index fc598c9f..6538afba 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -59,6 +59,9 @@ agent-permissions = ["dep:hive-types", "async-nats/jetstream"] # `agent_archetype::read`, for hive-c0re, which reads the archetype of each # agent it provisions or rebuilds. Explicit for the same reason as `notices`. agent-archetype = ["async-nats/jetstream"] +# `agent_closures::read`, for hive-c0re, which checks the latest closure of each +# agent it runs can be pulled. Explicit for the same reason as `notices`. +agent-closures = ["async-nats/jetstream"] # `agents_md_initial::read_own`, for the agent harness, which reads its own # initial `AGENTS.md` text with one direct get. Explicit for the same reason as # `notices`. diff --git a/swarm-queue-client/src/agent_closures.rs b/swarm-queue-client/src/agent_closures.rs index 5c8f6810..d800578d 100644 --- a/swarm-queue-client/src/agent_closures.rs +++ b/swarm-queue-client/src/agent_closures.rs @@ -17,3 +17,84 @@ pub async fn open_or_create( ) .await } + +/// What reading one agent's key found. +#[cfg(feature = "agent-closures")] +#[derive(Debug)] +pub enum Read { + /// The bucket answered: the agent's latest build, or `None` for an absent + /// or deleted key. + Current(Option), + /// Nothing answers for the bucket's stream: no build has been recorded + /// yet, or `JetStream` is not serving it right now. + NoBucket, +} + +/// Read `agent`'s latest closure record with one direct get. +/// +/// # Errors +/// +/// A failed request or an undecodable value. A request the queue refused for +/// permissions reaches the client as a timeout, not as a refusal. +#[cfg(feature = "agent-closures")] +pub async fn read( + client: &async_nats::Client, + agent: &str, +) -> Result> { + use async_nats::jetstream::stream::DirectGetErrorKind; + use swarm_wire::agent_closures::{STREAM, key_subject}; + + let js = async_nats::jetstream::new(client.clone()); + let stream = js.get_stream_no_info(STREAM).await?; + let message = match stream.direct_get_last_for_subject(key_subject(agent)).await { + Ok(message) => message, + Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => { + return Ok(Read::Current(None)); + } + Err(e) + if matches!( + e.kind(), + DirectGetErrorKind::ErrorResponse(status, _) + if status == async_nats::StatusCode::NO_RESPONDERS + ) => + { + return Ok(Read::NoBucket); + } + Err(e) => return Err(e.into()), + }; + Ok(Read::Current(decode( + message + .headers + .get("KV-Operation") + .map(async_nats::HeaderValue::as_str), + &message.payload, + )?)) +} + +/// One stored message as a record: a KV delete or purge is stored as an empty +/// marker carrying the `KV-Operation` header, and reads as no record. +#[cfg(feature = "agent-closures")] +fn decode( + operation: Option<&str>, + payload: &[u8], +) -> serde_json::Result> { + if matches!(operation, Some("DEL" | "PURGE")) { + return Ok(None); + } + serde_json::from_slice(payload).map(Some) +} + +#[cfg(all(test, feature = "agent-closures"))] +mod tests { + use super::decode; + + #[test] + fn a_stored_record_reads_and_a_delete_is_no_record() { + let payload = br#"{"meta_rev":"abc","store_path":"/nix/store/x","system":"x86_64-linux","built_unix":7,"cached":true}"#; + let record = decode(None, payload).expect("decodes").expect("a record"); + assert_eq!(record.meta_rev, "abc"); + assert!(record.cached); + assert!(decode(Some("DEL"), b"").expect("decodes").is_none()); + assert!(decode(Some("PURGE"), b"").expect("decodes").is_none()); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 3cbec38b..357c57eb 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -251,8 +251,9 @@ pub mod operator_profiles; pub mod agent_archetype; /// Opening the bucket swarm-controller records each agent's latest built -/// system closure in. The names and the stored type are -/// `swarm_wire::agent_closures`; writing needs `kv`. +/// system closure in, read by the hive that runs the agent. The names and the +/// stored type are `swarm_wire::agent_closures`; writing needs `kv` and reading +/// needs `agent-closures`. pub mod agent_closures; /// The bucket swarm-controller keeps the evaluated agent archetypes in — one