From 5486fc03339a17c85f46e86ca6818a28b97f855f Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 27 Oct 2026 17:33:57 +0100 Subject: [PATCH] report per agent whether a hive can pull its latest closure from the cache for each agent the hive looks up its record in agent-closures, fetches the closure's narinfo from the cache with the pull token and checks that a signature by the cache's public key covers it. the result is a log line and the gauge hyperhive.agent.closure_pullable, with a reason label when it is 0. nothing is downloaded and deploys still build locally. part 3 of 3: the probe needs the cache facts and the closure read grant from parts 1 and 2. --- Cargo.lock | 1 + Cargo.toml | 3 + docs/scheduler/observability.md | 19 +- docs/swarm/builder.md | 47 +++- hive-c0re/Cargo.toml | 1 + hive-c0re/src/attic/mod.rs | 5 +- hive-c0re/src/attic/narinfo.rs | 303 +++++++++++++++++++++ hive-c0re/src/attic/probe.rs | 396 ++++++++++++++++++++++++++++ hive-c0re/src/attic/reader.rs | 43 +-- hive-c0re/src/stats/otel_metrics.rs | 20 ++ hive-c0re/src/swarm_status.rs | 8 +- 11 files changed, 812 insertions(+), 34 deletions(-) create mode 100644 hive-c0re/src/attic/narinfo.rs create mode 100644 hive-c0re/src/attic/probe.rs diff --git a/Cargo.lock b/Cargo.lock index 3c5a9fad..e7d202fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1774,6 +1774,7 @@ dependencies = [ "clap", "clap-markdown", "clap_complete", + "ed25519-dalek", "forgejo-api", "futures-util", "hive-core-agent-sock", diff --git a/Cargo.toml b/Cargo.toml index cbea8d23..5ef450f6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -248,6 +248,9 @@ matrix-sdk = { version = "0.18", default-features = false, features = [ futures-util = "0.3" hmac = "0.13" sha2 = "0.11" +# Verifies the cache signature over a narinfo (hive-c0re). Already in the lock +# through nkeys. +ed25519-dalek = "2" # Constant-time comparison of a presented secret against the stored one # (`swarm-agent-verify`). Already in the tree through the TLS stack. subtle = "2.6" diff --git a/docs/scheduler/observability.md b/docs/scheduler/observability.md index c9e5b4a1..272d1558 100644 --- a/docs/scheduler/observability.md +++ b/docs/scheduler/observability.md @@ -334,15 +334,16 @@ exists, so off-the-shelf OTEL/Grafana container dashboards work. Resource `service.name = hyperhive-c0re`; hive-c0re tags each data point `container.name` (= the `h-` machine) and the hive `agent` label: -| Metric | Unit | Kind | Source | -| --------------------------------------- | ---- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `container.cpu.time` | `s` | counter | cumulative `cpu.stat` `usage_usec` → seconds | -| `container.memory.usage` | `By` | gauge | `memory.current` | -| `hyperhive.container.memory.limit` | `By` | gauge | `memory.max` (custom — semconv has no `.limit` metric; omitted when unlimited) | -| `hyperhive.container.memory.peak` | `By` | gauge | `memory.peak` (custom — no semconv metric; omitted if unavailable) | -| `hyperhive.container.storage.usage` | `By` | gauge | `state/` + `work/` + writable rootfs, measured by `hive-priv` every 30 min (custom — semconv only has `disk.io`; omitted while unknown: not measured yet, or the last measurement failed) | -| `hyperhive.container.cpu.percent` | `%` | gauge | host-normalised percent (custom — no `rate()` needed) | -| `hyperhive.agent.unrestricted_internet` | `1` | gauge | the agent's `unrestricted_internet` flag in the swarm permissions this hive last read: `1` on, `0` off; a flag never set reads `1`. Omitted while the stored permissions file is unreadable | +| Metric | Unit | Kind | Source | +| --------------------------------------- | ---- | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `container.cpu.time` | `s` | counter | cumulative `cpu.stat` `usage_usec` → seconds | +| `container.memory.usage` | `By` | gauge | `memory.current` | +| `hyperhive.container.memory.limit` | `By` | gauge | `memory.max` (custom — semconv has no `.limit` metric; omitted when unlimited) | +| `hyperhive.container.memory.peak` | `By` | gauge | `memory.peak` (custom — no semconv metric; omitted if unavailable) | +| `hyperhive.container.storage.usage` | `By` | gauge | `state/` + `work/` + writable rootfs, measured by `hive-priv` every 30 min (custom — semconv only has `disk.io`; omitted while unknown: not measured yet, or the last measurement failed) | +| `hyperhive.container.cpu.percent` | `%` | gauge | host-normalised percent (custom — no `rate()` needed) | +| `hyperhive.agent.unrestricted_internet` | `1` | gauge | the agent's `unrestricted_internet` flag in the swarm permissions this hive last read: `1` on, `0` off; a flag never set reads `1`. Omitted while the stored permissions file is unreadable | +| `hyperhive.agent.closure_pullable` | `1` | gauge | `1` when this hive can fetch the agent's latest closure from the swarm's binary cache and verify its signature, else `0` with a `reason` label (`no-record`, `rev-mismatch`, `not-cached`, `no-credentials`, `fetch-failed`, `bad-signature`; see [the builder docs](../swarm/builder.md#what-hives-do-with-the-cache)). Omitted before the first probe pass, every 5 minutes | The `hyperhive.`-prefixed metrics have no semconv equivalent (memory limit + peak, on-disk footprint, and an instantaneous cpu percent kept diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 2af56184..f1a0f542 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -249,8 +249,9 @@ KV bucket: the `internal/meta` commit, the closure's store path, the builder's platform and the time. Only the controller writes the bucket, and the builder holds no grant on it. A failed build writes nothing and leaves the previous key. The record's `cached` field says whether the builder pushed the closure to -[the binary cache](#binary-cache). Nothing reads the bucket yet. The closure stays -unrooted in the builder host's store. +[the binary cache](#binary-cache). Each hive reads its own agents' keys to +[report whether it can pull them](#what-hives-do-with-the-cache); no deploy +uses a record yet. The closure stays unrooted in the builder host's store. The controller gives the `builder` forge account read on `internal/meta` when it updates the flake, and again before each build, since the repo is private. A @@ -360,8 +361,8 @@ the swarm at an existing tenant: `secret/swarm/services/attic/url` (`https://` and the host, no trailing 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. + cache's page or `attic cache info`). Until all four exist, hives report + `no-credentials`. The builder checks the instance's certificate against the system roots and pins nothing, so a Let's Encrypt certificate needs no setup. The builder @@ -389,6 +390,44 @@ identity and writes `/run/hyperhive/attic/netrc`, `substituter` and 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. +The same pass checks each of the hive's agents. It reads the agent's record in +`agent-closures`, then fetches the closure's narinfo from +`//.narinfo` with the pull token as the basic-auth password. +It checks that a signature by the public key covers the narinfo, in Rust, with +no help from the host's nix daemon. It downloads nothing. The result is a log +line (at info when it changes) and the gauge +`hyperhive.agent.closure_pullable{agent}` (the `hive` label comes from the +collector, as for every `hyperhive.*` metric): `1` when the hive can pull the +latest closure, else `0` with a `reason` label. The first check that fails +decides the reason: + +| `reason` | Meaning | +| ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `no-credentials` | One of the four leaves is missing or unusable (a URL that isn't `https` is unusable). The swarm has no cache, the controller is too old to copy them, or you haven't seeded them yet. | +| `no-record` | `agent-closures` holds no build for the agent. | +| `rev-mismatch` | The record's build came from another `internal/meta` commit than its head, or the hive can't read the head. | +| `not-cached` | The record says the builder didn't push the closure, or the cache answers 404 for the narinfo. | +| `fetch-failed` | The request failed, or the cache answered with a status other than 200 or 404 (a refused token is a 401 or 403). | +| `bad-signature` | The narinfo describes another path, or no signature by the public key covers it. | + +A failed read of the store, the queue or the forge leaves the last result in +place and raises a warning after three failed passes. A hive reports nothing +before its first pass. + +Mixed versions: + +| controller | hive | what happens | +| ---------- | ---- | ----------------------------------------------------------------------------------- | +| new | new | the copies exist and the hive reports per agent | +| old | new | the controller copies nothing, so the hive reports `no-credentials` for every agent | +| new | old | the controller writes copies nobody reads | +| any | any | a swarm with no cache seeds no leaves, and every hive reports `no-credentials` | + +The signature check follows Nix's fingerprint format but hasn't run against a +live attic. A `bad-signature` on every agent in the first pass after a deploy +means check attic's fingerprint format; it says nothing about an attacker. A +`fetch-failed` on every agent points at the URL, the token or the network. + ## Credentials | What | Where | Who reads it | diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 26b49e35..b1cb8c29 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -48,6 +48,7 @@ listenfd = "1" petgraph.workspace = true hmac.workspace = true sha2.workspace = true +ed25519-dalek.workspace = true rusqlite.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/hive-c0re/src/attic/mod.rs b/hive-c0re/src/attic/mod.rs index 7bcb8238..c29214ec 100644 --- a/hive-c0re/src/attic/mod.rs +++ b/hive-c0re/src/attic/mod.rs @@ -1,4 +1,7 @@ //! The swarm's binary cache, as this hive sees it: the pull credentials it is -//! given ([`reader`]). +//! given ([`reader`]), the narinfo check ([`narinfo`]), and the probe that +//! reports per agent whether the latest closure could be pulled ([`probe`]). +pub mod narinfo; +pub mod probe; pub mod reader; diff --git a/hive-c0re/src/attic/narinfo.rs b/hive-c0re/src/attic/narinfo.rs new file mode 100644 index 00000000..e1c0fd55 --- /dev/null +++ b/hive-c0re/src/attic/narinfo.rs @@ -0,0 +1,303 @@ +//! A closure's narinfo in the binary cache: fetching it with the pull token, +//! and checking its signature against the cache's public key. +//! +//! The fingerprint is Nix's: `1;;;;`, signed with ed25519. A key is +//! `:`, a signature `:`. + +use std::time::Duration; + +use anyhow::{Context as _, Result, anyhow, bail}; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD; +use ed25519_dalek::{Signature, Verifier as _, VerifyingKey}; + +use super::reader::Facts; + +/// How long one narinfo request may take. +const FETCH_TIMEOUT: Duration = Duration::from_secs(20); + +/// A client for the cache: no redirects, since the token rides in a header. +/// +/// # Errors +/// The TLS roots cannot be loaded. +pub fn client() -> Result { + Ok(reqwest::Client::builder() + .timeout(FETCH_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .build()?) +} + +/// The narinfo of `store_path` from the cache `facts` name. +/// +/// `Ok(None)` is the cache saying it does not hold the path (a 404). +/// +/// # Errors +/// A request that failed, or any status but 200 and 404. The message names the +/// status, never the token. +pub async fn fetch( + client: &reqwest::Client, + facts: &Facts, + store_path: &str, +) -> Result> { + let hash = path_hash(store_path)?; + let url = format!("{}/{hash}.narinfo", facts.substituter()); + let response = client + .get(&url) + .basic_auth("hive", Some(facts.token())) + .send() + .await + .map_err(|e| anyhow!("GET {url}: {}", e.without_url()))?; + match response.status() { + reqwest::StatusCode::OK => { + Ok(Some(response.text().await.map_err(|e| { + anyhow!("reading {url}: {}", e.without_url()) + })?)) + } + reqwest::StatusCode::NOT_FOUND => Ok(None), + status => bail!("GET {url}: {status}"), + } +} + +/// The 32-character hash a narinfo is named by: `/nix/store/-`. +fn path_hash(store_path: &str) -> Result<&str> { + let base = store_path.rsplit('/').next().unwrap_or_default(); + match base.split_once('-') { + Some((hash, name)) if hash.len() == 32 && !name.is_empty() => Ok(hash), + _ => bail!("{store_path:?} is not a store path"), + } +} + +/// Whether `narinfo` describes `store_path` and carries a signature by +/// `public_key` over it. +/// +/// # Errors +/// A narinfo that is malformed, describes another path, or has no valid +/// signature by the key. +pub fn verify(narinfo: &str, store_path: &str, public_key: &str) -> Result<()> { + let field = |name: &str| { + narinfo + .lines() + .find_map(|l| l.strip_prefix(name).and_then(|r| r.strip_prefix(": "))) + }; + let path = field("StorePath").context("no StorePath")?; + if path != store_path { + bail!("the narinfo describes {path}, not {store_path}"); + } + let hash = nar_hash_nix32(field("NarHash").context("no NarHash")?)?; + let size: u64 = field("NarSize") + .context("no NarSize")? + .parse() + .context("NarSize is not a number")?; + let dir = store_path.rsplit_once('/').map_or("", |(d, _)| d); + let mut refs: Vec<&str> = field("References") + .unwrap_or_default() + .split_whitespace() + .collect(); + refs.sort_unstable(); + let refs: Vec = refs.iter().map(|r| format!("{dir}/{r}")).collect(); + let fingerprint = format!("1;{store_path};{hash};{size};{}", refs.join(",")); + + let (key_name, key_b64) = public_key + .split_once(':') + .context("the public key has no name")?; + let key = VerifyingKey::from_bytes( + &STANDARD + .decode(key_b64)? + .try_into() + .map_err(|_| anyhow!("the public key is not 32 bytes"))?, + )?; + for sig in narinfo.lines().filter_map(|l| l.strip_prefix("Sig: ")) { + let Some((name, b64)) = sig.split_once(':') else { + continue; + }; + if name != key_name { + continue; + } + let Ok(bytes) = STANDARD.decode(b64) else { + continue; + }; + let Ok(signature) = Signature::from_slice(&bytes) else { + continue; + }; + if key.verify(fingerprint.as_bytes(), &signature).is_ok() { + return Ok(()); + } + } + bail!("no signature by {key_name} verifies over the fingerprint") +} + +/// `sha256:` as Nix prints it: the hash in its base-32 form. A narinfo +/// carries it that way; hex is accepted too. +fn nar_hash_nix32(value: &str) -> Result { + let digest = value + .strip_prefix("sha256:") + .context("NarHash is not sha256")?; + match digest.len() { + 52 => Ok(value.to_owned()), + 64 => { + let bytes = (0..32) + .map(|i| u8::from_str_radix(&digest[2 * i..2 * i + 2], 16)) + .collect::, _>>() + .context("NarHash is not hex")?; + Ok(format!("sha256:{}", nix32(&bytes))) + } + _ => bail!("NarHash is in a form this check does not read"), + } +} + +/// Nix's base-32: its own alphabet, least significant bits first. +fn nix32(bytes: &[u8]) -> String { + const ALPHABET: &[u8; 32] = b"0123456789abcdfghijklmnpqrsvwxyz"; + let len = (bytes.len() * 8 - 1) / 5 + 1; + (0..len) + .rev() + .map(|n| { + let (i, j) = (n * 5 / 8, n * 5 % 8); + let lo = u16::from(bytes[i]) >> j; + let hi = bytes.get(i + 1).map_or(0, |b| u16::from(*b) << (8 - j)); + char::from(ALPHABET[usize::from((lo | hi) & 0x1f)]) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use ed25519_dalek::{Signer as _, SigningKey}; + + use super::*; + + const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas"; + const DEP: &str = "ffffffffffffffffffffffffffffffff-glibc"; + const HASH: &str = "sha256:0mdqa9w1p6cmli6976v4wi0sw9r4p5prkj7lzfd1877wk11c9c73"; + + fn key(seed: u8) -> SigningKey { + SigningKey::from_bytes(&[seed; 32]) + } + + fn public(name: &str, key: &SigningKey) -> String { + format!("{name}:{}", STANDARD.encode(key.verifying_key().to_bytes())) + } + + /// A narinfo signed by `key` under `name` over the fingerprint of `PATH`. + fn signed(name: &str, key: &SigningKey) -> String { + let fingerprint = format!("1;{PATH};{HASH};120;/nix/store/{DEP}"); + let sig = STANDARD.encode(key.sign(fingerprint.as_bytes()).to_bytes()); + format!( + "StorePath: {PATH}\nURL: nar/x.nar\nCompression: none\nNarHash: {HASH}\n\ + NarSize: 120\nReferences: {DEP}\nSig: {name}:{sig}\n" + ) + } + + #[test] + fn nix32_matches_nixs_own_encoding() { + let empty_sha256 = [ + 0xe3, 0xb0, 0xc4, 0x42, 0x98, 0xfc, 0x1c, 0x14, 0x9a, 0xfb, 0xf4, 0xc8, 0x99, 0x6f, + 0xb9, 0x24, 0x27, 0xae, 0x41, 0xe4, 0x64, 0x9b, 0x93, 0x4c, 0xa4, 0x95, 0x99, 0x1b, + 0x78, 0x52, 0xb8, 0x55, + ]; + assert_eq!( + format!("sha256:{}", nix32(&empty_sha256)), + HASH, + "from `nix-hash --to-base32`" + ); + let hex = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; + assert_eq!(nar_hash_nix32(&format!("sha256:{hex}")).expect("hex"), HASH); + } + + #[test] + fn a_signature_by_the_key_verifies() { + let k = key(7); + verify(&signed("swarm", &k), PATH, &public("swarm", &k)).expect("verifies"); + } + + #[test] + fn a_signature_by_another_key_or_over_other_content_does_not() { + let (ours, theirs) = (key(7), key(9)); + let err = verify(&signed("swarm", &theirs), PATH, &public("swarm", &ours)); + assert!(err.is_err(), "a stranger's signature under our key's name"); + // The key's name must match a Sig line's, whatever the bytes. + let err = verify(&signed("other", &ours), PATH, &public("swarm", &ours)); + assert!(err.is_err(), "a signature under another name"); + let tampered = signed("swarm", &ours).replace("NarSize: 120", "NarSize: 121"); + assert!(verify(&tampered, PATH, &public("swarm", &ours)).is_err()); + let unsigned = signed("swarm", &ours).replace("Sig: ", "Sig-ignored: "); + assert!(verify(&unsigned, PATH, &public("swarm", &ours)).is_err()); + } + + #[test] + fn a_narinfo_for_another_path_is_refused_even_when_signed() { + let k = key(7); + let other = "/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-other"; + assert!(verify(&signed("swarm", &k), other, &public("swarm", &k)).is_err()); + } + + #[test] + fn the_narinfo_is_named_by_the_paths_hash() { + assert_eq!( + path_hash(PATH).expect("a store path"), + "0123456789abcdfghijklmnpqrsvwxyz" + ); + assert!(path_hash("/nix/store/short-name").is_err()); + assert!(path_hash("nonsense").is_err()); + } + + /// A cache on loopback that answers a narinfo only to this basic-auth + /// password, 404 for another path, and 500 for a third. + async fn serve() -> (String, tokio::task::JoinHandle<()>) { + use axum::http::{HeaderMap, StatusCode}; + let app = axum::Router::new().route( + "/swarm/{file}", + axum::routing::get(|axum::extract::Path(file): axum::extract::Path, headers: HeaderMap| async move { + let expected = format!("Basic {}", STANDARD.encode("hive:pull-secret")); + let ok = headers.get("authorization").is_some_and(|v| *v == *expected); + match (ok, file.as_str()) { + (false, _) => (StatusCode::UNAUTHORIZED, String::new()), + (true, "0123456789abcdfghijklmnpqrsvwxyz.narinfo") => (StatusCode::OK, "narinfo-body".to_owned()), + (true, "ffffffffffffffffffffffffffffffff.narinfo") => (StatusCode::INTERNAL_SERVER_ERROR, String::new()), + (true, _) => (StatusCode::NOT_FOUND, String::new()), + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let base = format!("http://{}", listener.local_addr().expect("addr")); + ( + base, + tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") }), + ) + } + + #[tokio::test] + async fn the_fetch_sends_the_token_and_tells_missing_from_failed() { + let (base, server) = serve().await; + // No system roots in the build sandbox; the requests are plain HTTP. + let client = reqwest::Client::builder() + .tls_certs_only([]) + .build() + .expect("client"); + let with = |token: &str| Facts::unchecked_for_test(token, &base, "swarm", "swarm:AAAA"); + + let found = fetch(&client, &with("pull-secret"), PATH).await; + let missing = fetch( + &client, + &with("pull-secret"), + "/nix/store/eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee-x", + ) + .await; + let broken = fetch(&client, &with("pull-secret"), &format!("/nix/store/{DEP}")).await; + let refused = fetch(&client, &with("wrong"), PATH).await; + server.abort(); + + assert_eq!(found.expect("200").as_deref(), Some("narinfo-body")); + assert!(missing.expect("404 is not an error").is_none()); + assert!(broken.expect_err("500").to_string().contains("500")); + let refused = refused.expect_err("401"); + assert!(refused.to_string().contains("401")); + assert!( + !refused.to_string().contains("wrong"), + "the token stays out of the message" + ); + } +} diff --git a/hive-c0re/src/attic/probe.rs b/hive-c0re/src/attic/probe.rs new file mode 100644 index 00000000..5c39e76f --- /dev/null +++ b/hive-c0re/src/attic/probe.rs @@ -0,0 +1,396 @@ +//! Whether this hive could pull the latest closure of each of its agents from +//! the swarm's binary cache, and the answer as a log line and a gauge. +//! +//! Every [`PASS_INTERVAL`] each agent's record in `agent-closures` is looked up +//! and its narinfo fetched with the pull token and checked against the cache's +//! public key. Nothing is downloaded and nothing is deployed from it. +//! +//! The verdict is the first of these that fails, in this order: +//! no credentials (`no-credentials`), no record (`no-record`), a record built +//! from another `internal/meta` commit than the head (`rev-mismatch`), a record +//! the builder did not push (`not-cached`), the cache not holding the narinfo +//! (`not-cached`), a request that failed or was refused (`fetch-failed`), a +//! narinfo no signature of the key covers (`bad-signature`). + +use std::sync::Mutex; +use std::time::Duration; + +use anyhow::{Context as _, Result}; +use swarm_queue_client::agent_closures::{self, Read}; +use swarm_wire::agent_closures::AgentClosure; + +use super::narinfo; +use super::reader::{self, Facts}; +use crate::stats::sweep_health::{self, SweepHealth}; + +/// How often every agent is probed. +pub const PASS_INTERVAL: Duration = Duration::from_mins(5); + +/// At [`PASS_INTERVAL`] this is ~15 minutes of failing passes. +const FAILURES_BEFORE_WARNING: u32 = 3; + +/// Why a closure cannot be pulled; the gauge's `reason` label. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Reason { + NoRecord, + RevMismatch, + NotCached, + NoCredentials, + FetchFailed, + BadSignature, +} + +impl Reason { + /// The label value. + pub const fn label(self) -> &'static str { + match self { + Reason::NoRecord => "no-record", + Reason::RevMismatch => "rev-mismatch", + Reason::NotCached => "not-cached", + Reason::NoCredentials => "no-credentials", + Reason::FetchFailed => "fetch-failed", + Reason::BadSignature => "bad-signature", + } + } +} + +/// A closure that cannot be pulled, and what the log line says about it. +#[derive(Debug)] +pub struct Unpullable { + pub reason: Reason, + pub detail: String, +} + +fn no(reason: Reason, detail: impl Into) -> Unpullable { + Unpullable { + reason, + detail: detail.into(), + } +} + +/// The cache's narinfo for a store path, so a test can stand in for it. +pub(crate) trait Cache { + /// `None` when the cache does not hold the path. + async fn narinfo(&self, facts: &Facts, store_path: &str) -> Result>; +} + +struct Http(reqwest::Client); + +impl Cache for Http { + async fn narinfo(&self, facts: &Facts, store_path: &str) -> Result> { + narinfo::fetch(&self.0, facts, store_path).await + } +} + +/// Whether the closure `record` names can be pulled with `facts`. `head` is the +/// `internal/meta` commit a deploy would build from. +pub(crate) async fn judge( + facts: Option<&Facts>, + record: Option<&AgentClosure>, + head: Option<&str>, + cache: &impl Cache, +) -> Result<(), Unpullable> { + let facts = facts.ok_or_else(|| { + no( + Reason::NoCredentials, + "the swarm has given this hive no cache facts", + ) + })?; + let record = + record.ok_or_else(|| no(Reason::NoRecord, "no build is recorded for the agent"))?; + if head != Some(record.meta_rev.as_str()) { + return Err(no( + Reason::RevMismatch, + format!( + "built from {}, internal/meta is at {}", + record.meta_rev, + head.unwrap_or("nothing") + ), + )); + } + if !record.cached { + return Err(no( + Reason::NotCached, + "the builder did not push the closure", + )); + } + let text = match cache.narinfo(facts, &record.store_path).await { + Ok(Some(text)) => text, + Ok(None) => return Err(no(Reason::NotCached, "the cache does not hold the narinfo")), + Err(e) => return Err(no(Reason::FetchFailed, format!("{e:#}"))), + }; + narinfo::verify(&text, &record.store_path, facts.public_key()) + .map_err(|e| no(Reason::BadSignature, format!("{e:#}"))) +} + +/// The last pass's verdict per agent: `None` when it can be pulled. +static VERDICTS: Mutex)>> = Mutex::new(Vec::new()); + +/// The verdicts of the last pass, for the gauge. +pub fn verdicts() -> Vec<(String, Option)> { + VERDICTS.lock().map(|g| g.clone()).unwrap_or_default() +} + +/// Probe every agent now and every [`PASS_INTERVAL`] until shutdown. +pub async fn follow(client: async_nats::Client, mut shutdown: tokio::sync::watch::Receiver) { + let mut health = SweepHealth::new("attic_pull_probe", "warn", FAILURES_BEFORE_WARNING); + loop { + match pass(&client).await { + Ok(()) => health.record_ok(), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "attic: the pull probe failed; the last verdicts stand"); + let err = format!("{e:#}"); + health.record_err(|ctx| { + let age = ctx.since_last_ok.map_or_else( + || "no success this session".to_owned(), + |d| format!("last ok {} ago", sweep_health::fmt_age(d)), + ); + format!( + "checking whether agents' closures can be pulled is failing ({} consecutive, {age}) \ + — deploys are unaffected: {err}", + ctx.consecutive + ) + }); + } + } + tokio::select! { + () = tokio::time::sleep(PASS_INTERVAL) => {} + _ = shutdown.changed() => { + tracing::info!("attic: shutdown signal received"); + return; + } + } + } +} + +/// One pass: publish the facts, then judge every agent. +/// +/// # Errors +/// A read that failed (the store, the queue, the forge), which leaves the last +/// verdicts as they are: the failure says nothing about any agent. +async fn pass(client: &async_nats::Client) -> Result<()> { + swarm_queue_client::ensure_connected(client)?; + let facts = reader::sync().await?; + let agents = crate::lifecycle::agents_for_meta_listing() + .await + .context("enumerating this hive's agents")?; + let head = match &facts { + Some(_) => meta_head().await?, + None => None, + }; + let http = Http(narinfo::client()?); + let mut now = Vec::with_capacity(agents.len()); + for spec in agents { + let record = match &facts { + Some(_) => read_record(client, &spec.name).await?, + None => None, + }; + let verdict = judge(facts.as_ref(), record.as_ref(), head.as_deref(), &http).await; + now.push((spec.name, verdict.as_ref().err().map(|u| u.reason))); + report(&now[now.len() - 1], verdict.err().map(|u| u.detail)); + } + if let Ok(mut g) = VERDICTS.lock() { + *g = now; + } + Ok(()) +} + +async fn read_record(client: &async_nats::Client, agent: &str) -> Result> { + match agent_closures::read(client, agent) + .await + .map_err(|e| anyhow::anyhow!(e)) + .with_context(|| format!("reading the closure record of {agent}"))? + { + Read::Current(record) => Ok(record), + Read::NoBucket => Ok(None), + } +} + +/// The `internal/meta` head, or `None` when the swarm has not published it or +/// this hive has no forge token. +async fn meta_head() -> Result> { + use crate::meta_rev::MetaSource as _; + match crate::forge::internal_meta_source().await { + Some(source) => source + .meta_head() + .await + .context("reading internal/meta's head"), + None => Ok(None), + } +} + +/// The log line: at info when the verdict differs from the last pass, else +/// debug. +fn report((agent, verdict): &(String, Option), detail: Option) { + let changed = verdicts() + .iter() + .find(|(a, _)| a == agent) + .is_none_or(|(_, v)| v != verdict); + let reason = verdict.map_or("", Reason::label); + let pullable = verdict.is_none(); + let detail = detail.unwrap_or_default(); + if changed { + tracing::info!(%agent, pullable, reason, %detail, "attic: can this hive pull the agent's latest closure"); + } else { + tracing::debug!(%agent, pullable, reason, %detail, "attic: can this hive pull the agent's latest closure"); + } +} + +#[cfg(test)] +mod tests { + use base64::Engine as _; + use base64::engine::general_purpose::STANDARD; + use ed25519_dalek::{Signer as _, SigningKey}; + + use super::*; + + const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas"; + const HASH: &str = "sha256:0mdqa9w1p6cmli6976v4wi0sw9r4p5prkj7lzfd1877wk11c9c73"; + const REV: &str = "a1b2c3"; + + fn key() -> SigningKey { + SigningKey::from_bytes(&[7; 32]) + } + + fn facts() -> Facts { + let public = format!( + "swarm:{}", + STANDARD.encode(key().verifying_key().to_bytes()) + ); + Facts::parse("tok", "https://cache.example", "swarm", &public).expect("facts") + } + + fn record() -> AgentClosure { + AgentClosure { + meta_rev: REV.to_owned(), + store_path: PATH.to_owned(), + system: "x86_64-linux".to_owned(), + built_unix: 1, + cached: true, + } + } + + fn narinfo_signed_by(key: &SigningKey) -> String { + let sig = key.sign(format!("1;{PATH};{HASH};120;").as_bytes()); + format!( + "StorePath: {PATH}\nNarHash: {HASH}\nNarSize: 120\nReferences: \nSig: swarm:{}\n", + STANDARD.encode(sig.to_bytes()) + ) + } + + /// What the cache answers for the narinfo. + struct Answers(Result, &'static str>); + + impl Cache for Answers { + async fn narinfo(&self, _: &Facts, _: &str) -> Result> { + self.0.clone().map_err(|e| anyhow::anyhow!(e)) + } + } + + async fn reason( + facts: Option<&Facts>, + record: Option<&AgentClosure>, + head: Option<&str>, + cache: Answers, + ) -> Option { + judge(facts, record, head, &cache) + .await + .err() + .map(|u| u.reason) + } + + fn good() -> Answers { + Answers(Ok(Some(narinfo_signed_by(&key())))) + } + + #[tokio::test] + async fn a_signed_narinfo_for_the_head_build_is_pullable() { + assert_eq!( + reason(Some(&facts()), Some(&record()), Some(REV), good()).await, + None + ); + } + + #[tokio::test] + async fn no_facts_is_no_credentials_even_with_a_record() { + let r = reason(None, Some(&record()), Some(REV), good()).await; + assert_eq!(r, Some(Reason::NoCredentials)); + } + + #[tokio::test] + async fn no_record_is_no_record() { + let r = reason(Some(&facts()), None, Some(REV), good()).await; + assert_eq!(r, Some(Reason::NoRecord)); + } + + #[tokio::test] + async fn a_record_from_another_commit_is_a_rev_mismatch() { + let r = reason(Some(&facts()), Some(&record()), Some("other"), good()).await; + assert_eq!(r, Some(Reason::RevMismatch)); + let r = reason(Some(&facts()), Some(&record()), None, good()).await; + assert_eq!(r, Some(Reason::RevMismatch), "no head to compare with"); + } + + #[tokio::test] + async fn an_unpushed_record_or_a_missing_narinfo_is_not_cached() { + let unpushed = AgentClosure { + cached: false, + ..record() + }; + let r = reason(Some(&facts()), Some(&unpushed), Some(REV), good()).await; + assert_eq!(r, Some(Reason::NotCached)); + let r = reason( + Some(&facts()), + Some(&record()), + Some(REV), + Answers(Ok(None)), + ) + .await; + assert_eq!(r, Some(Reason::NotCached)); + } + + #[tokio::test] + async fn a_failed_request_is_fetch_failed() { + let r = reason( + Some(&facts()), + Some(&record()), + Some(REV), + Answers(Err("401")), + ) + .await; + assert_eq!(r, Some(Reason::FetchFailed)); + } + + #[tokio::test] + async fn a_narinfo_the_key_did_not_sign_is_a_bad_signature() { + let forged = Answers(Ok(Some(narinfo_signed_by(&SigningKey::from_bytes( + &[9; 32], + ))))); + let r = reason(Some(&facts()), Some(&record()), Some(REV), forged).await; + assert_eq!(r, Some(Reason::BadSignature)); + } + + #[test] + fn the_labels_are_the_documented_ones() { + let labels = [ + Reason::NoRecord, + Reason::RevMismatch, + Reason::NotCached, + Reason::NoCredentials, + Reason::FetchFailed, + Reason::BadSignature, + ] + .map(Reason::label); + assert_eq!( + labels, + [ + "no-record", + "rev-mismatch", + "not-cached", + "no-credentials", + "fetch-failed", + "bad-signature" + ] + ); + } +} diff --git a/hive-c0re/src/attic/reader.rs b/hive-c0re/src/attic/reader.rs index e352239e..6c64a73f 100644 --- a/hive-c0re/src/attic/reader.rs +++ b/hive-c0re/src/attic/reader.rs @@ -6,7 +6,6 @@ //! 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}; @@ -64,6 +63,31 @@ impl Facts { format!("{}/{}", self.url, self.cache) } + pub fn public_key(&self) -> &str { + &self.public_key + } + + pub fn token(&self) -> &str { + &self.token + } + + /// Facts for a test that reaches a server on loopback: no scheme check, + /// so the one place an `http` URL gets past [`Facts::parse`] is a test. + #[cfg(test)] + pub(crate) fn unchecked_for_test( + token: &str, + url: &str, + cache: &str, + public_key: &str, + ) -> Self { + Self { + token: token.to_owned(), + url: url.to_owned(), + cache: cache.to_owned(), + public_key: public_key.to_owned(), + } + } + /// The netrc line the server's host authenticates with. The token is the /// password; the server ignores the login. fn netrc(&self) -> String { @@ -96,9 +120,6 @@ impl Facts { } } -/// 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. /// @@ -114,20 +135,6 @@ pub async fn sync() -> Result> { 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 diff --git a/hive-c0re/src/stats/otel_metrics.rs b/hive-c0re/src/stats/otel_metrics.rs index aaa5df2c..cf387708 100644 --- a/hive-c0re/src/stats/otel_metrics.rs +++ b/hive-c0re/src/stats/otel_metrics.rs @@ -256,9 +256,29 @@ fn register_instruments(provider: &SdkMeterProvider, snapshot: Arc u64 { diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 1321c06b..71cdafa1 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -109,8 +109,12 @@ 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())); + // Whether each agent's latest closure could be pulled from the swarm's + // cache, on the same connection. + tokio::spawn(crate::attic::probe::follow( + client.clone(), + shutdown.clone(), + )); // The permissions the swarm grants, and the import of the meta repo's. tokio::spawn(crate::workers::swarm_permissions::follow(