From ff5295963ab3d4f4f3d324395941e9e1075df0e1 Mon Sep 17 00:00:00 2001 From: Pieter Date: Wed, 24 Jun 2026 18:59:46 +0200 Subject: [PATCH] various: publish signed closure narinfos for build outputs Co-authored-by: Amaan Qureshi --- Cargo.lock | 1 + Cargo.toml | 1 + crates/common/Cargo.toml | 69 +-- crates/common/src/lib.rs | 1 + crates/common/src/narinfo_signing.rs | 42 ++ crates/queue-runner/src/rpc/server.rs | 41 +- crates/queue-runner/src/runner_loop.rs | 101 ++++- crates/queue-runner/src/worker.rs | 580 ++++++++++++++++++++++++- crates/server/src/routes/cache.rs | 34 +- 9 files changed, 791 insertions(+), 79 deletions(-) create mode 100644 crates/common/src/narinfo_signing.rs diff --git a/Cargo.lock b/Cargo.lock index d4b20c6e..0eb81fa3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -692,6 +692,7 @@ dependencies = [ "clap", "color-eyre", "git2", + "harmonia-utils-signature", "hex", "lettre", "libc", diff --git a/Cargo.toml b/Cargo.toml index afa4d81e..eda6225f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -138,6 +138,7 @@ harmonia-store-nar-info = { git = "https://github.com/nix-community/harmo harmonia-store-path = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" } harmonia-store-path-info = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" } harmonia-utils-hash = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" } +harmonia-utils-signature = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" } # evix, in-process Nix evaluator (nix-eval-jobs replacement) driven as a library. evix = { version = "0.3.3", features = [ "flake" ] } diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml index 37708d71..9d5d710e 100644 --- a/crates/common/Cargo.toml +++ b/crates/common/Cargo.toml @@ -6,40 +6,41 @@ license.workspace = true repository.workspace = true [dependencies] -argon2.workspace = true -base64.workspace = true -chrono.workspace = true -circus-config.workspace = true -circus-logs.workspace = true -circus-migrations.workspace = true -circus-nix.workspace = true -circus-types = { workspace = true, features = [ "sqlx" ] } -clap.workspace = true -color-eyre.workspace = true -git2.workspace = true -hex.workspace = true -lettre.workspace = true -libc.workspace = true -nix.workspace = true -num_enum.workspace = true -password-hash.workspace = true -regex.workspace = true -reqwest.workspace = true -ring.workspace = true -rustls.workspace = true -serde.workspace = true -serde_json.workspace = true -sha2.workspace = true -shellexpand.workspace = true -sqlx.workspace = true -tempfile.workspace = true -thiserror.workspace = true -tokio.workspace = true -tracing.workspace = true -url.workspace = true -urlencoding.workspace = true -uuid.workspace = true -whoami.workspace = true +argon2.workspace = true +base64.workspace = true +chrono.workspace = true +circus-config.workspace = true +circus-logs.workspace = true +circus-migrations.workspace = true +circus-nix.workspace = true +circus-types = { workspace = true, features = [ "sqlx" ] } +clap.workspace = true +color-eyre.workspace = true +git2.workspace = true +harmonia-utils-signature.workspace = true +hex.workspace = true +lettre.workspace = true +libc.workspace = true +nix.workspace = true +num_enum.workspace = true +password-hash.workspace = true +regex.workspace = true +reqwest.workspace = true +ring.workspace = true +rustls.workspace = true +serde.workspace = true +serde_json.workspace = true +sha2.workspace = true +shellexpand.workspace = true +sqlx.workspace = true +tempfile.workspace = true +thiserror.workspace = true +tokio.workspace = true +tracing.workspace = true +url.workspace = true +urlencoding.workspace = true +uuid.workspace = true +whoami.workspace = true [dev-dependencies] toml.workspace = true diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index a38d3def..6400c468 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -10,6 +10,7 @@ pub mod log_storage; pub mod migrate; pub mod migrate_cli; pub mod models; +pub mod narinfo_signing; pub mod pg_notify; pub mod psi; pub mod repo; diff --git a/crates/common/src/narinfo_signing.rs b/crates/common/src/narinfo_signing.rs new file mode 100644 index 00000000..8d63ab45 --- /dev/null +++ b/crates/common/src/narinfo_signing.rs @@ -0,0 +1,42 @@ +//! Shared Nix narinfo fingerprint signing. + +use std::path::Path; + +use color_eyre::eyre::{Context as _, Result, eyre}; +use harmonia_utils_signature::SecretKey; + +/// Read a Nix-format `:` signing key from disk. +/// +/// # Errors +/// +/// Returns an error when the file cannot be read, or the key is malformed or +/// its public half does not match its seed. +pub async fn read_signing_key(path: &Path) -> Result { + let raw = tokio::fs::read_to_string(path) + .await + .with_context(|| format!("read signing key {}", path.display()))?; + raw + .trim() + .parse::() + .map_err(|e| eyre!("invalid signing key: {e:?}")) +} + +/// Sign the canonical Nix narinfo fingerprint, returning +/// `:`. References are signed as a sorted set to match +/// Nix's verification. +#[must_use] +pub fn sign_narinfo( + key: &SecretKey, + store_path: &str, + nar_hash: &str, + nar_size: i64, + references: &[String], +) -> String { + let mut sorted_refs = references.to_vec(); + sorted_refs.sort(); + let fingerprint = format!( + "1;{store_path};{nar_hash};{nar_size};{}", + sorted_refs.join(",") + ); + key.sign(fingerprint.as_bytes()).to_string() +} diff --git a/crates/queue-runner/src/rpc/server.rs b/crates/queue-runner/src/rpc/server.rs index 4c323f32..a1f34646 100644 --- a/crates/queue-runner/src/rpc/server.rs +++ b/crates/queue-runner/src/rpc/server.rs @@ -25,7 +25,7 @@ use circus_proto::{ result_sink, runner, }; -use color_eyre::eyre::{Context as _, bail, eyre}; +use color_eyre::eyre::{Context as _, bail}; use sha2::{Digest as _, Sha256}; use sqlx::PgPool; use subtle::ConstantTimeEq as _; @@ -897,11 +897,7 @@ impl runner::Server for RunnerImpl { } } -/// Sign a narinfo fingerprint with the Nix-format signing key on disk. -/// -/// Nix key files are one line: `:`. The secret -/// is a 64-byte concatenation of the Ed25519 seed and public key (the -/// canonical libsodium "secret key" layout). Output is +/// Sign a narinfo fingerprint with the on-disk Nix signing key, returning /// `:`. async fn sign_fingerprint( key_file: &std::path::Path, @@ -910,35 +906,10 @@ async fn sign_fingerprint( nar_size: i64, references: &[String], ) -> color_eyre::Result { - use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; - use ring::signature::Ed25519KeyPair; - let raw = tokio::fs::read_to_string(key_file) - .await - .with_context(|| format!("read signing key {}", key_file.display()))?; - let raw = raw.trim(); - let (name, secret_b64) = raw - .split_once(':') - .ok_or_else(|| eyre!("signing key not in `name:base64` form"))?; - let secret = B64 - .decode(secret_b64) - .with_context(|| "signing key base64 decode")?; - if secret.len() != 64 { - bail!("signing key has {} bytes, expected 64", secret.len()); - } - // libsodium layout is `seed (32) || public key (32)`. ring wants the - // seed alone. - let key = Ed25519KeyPair::from_seed_unchecked(&secret[..32]) - .map_err(|e| eyre!("ring rejected key seed: {e}"))?; - // Nix builds the fingerprint from a sorted StorePathSet and re-sorts the - // references when verifying so the signed order must match. - let mut sorted_refs = references.to_vec(); - sorted_refs.sort(); - let fingerprint = format!( - "1;{store_path};{nar_hash};{nar_size};{}", - sorted_refs.join(",") - ); - let sig = key.sign(fingerprint.as_bytes()); - Ok(format!("{name}:{}", B64.encode(sig.as_ref()))) + let key = circus_common::narinfo_signing::read_signing_key(key_file).await?; + Ok(circus_common::narinfo_signing::sign_narinfo( + &key, store_path, nar_hash, nar_size, references, + )) } /// Map a compression algorithm name to the conventional NAR file extension. diff --git a/crates/queue-runner/src/runner_loop.rs b/crates/queue-runner/src/runner_loop.rs index e6af55d5..67c96abc 100644 --- a/crates/queue-runner/src/runner_loop.rs +++ b/crates/queue-runner/src/runner_loop.rs @@ -2,7 +2,7 @@ use std::{sync::Arc, time::Duration}; use circus_common::{ error::Result as CiResult, - models::{BuildStatus, JobsetState}, + models::{Build, BuildStatus, JobsetState}, repo, }; use circus_config::HotConfig; @@ -96,6 +96,87 @@ async fn mark_build_done( .map(|_| ()) } +async fn publish_succeeded_output_closure( + pool: &PgPool, + worker_pool: &WorkerPool, + build_id: Uuid, + output_paths: &[String], +) { + if output_paths.is_empty() { + return; + } + let project_id = match repo::builds::get(pool, build_id).await { + Ok(build) => { + if let Some((project, _)) = get_project_for_build(pool, &build).await { + Some(project.id) + } else { + tracing::warn!( + build_id = %build_id, + "could not resolve project for closure narinfo publication" + ); + None + } + }, + Err(e) => { + tracing::warn!( + build_id = %build_id, + "failed to reload build for closure narinfo publication: {e}" + ); + None + }, + }; + worker_pool + .persist_closure_narinfos(build_id, output_paths, project_id) + .await; +} + +async fn completed_build_output_paths( + pool: &PgPool, + build: &Build, +) -> Vec { + let mut output_paths = match repo::build_outputs::list_for_build( + pool, build.id, + ) + .await + { + Ok(outputs) => { + outputs + .into_iter() + .filter_map(|output| output.path) + .collect::>() + }, + Err(e) => { + tracing::warn!( + build_id = %build.id, + "failed to load completed build outputs for closure narinfo publication: {e}" + ); + Vec::new() + }, + }; + + if output_paths.is_empty() + && let Some(outputs) = + build.outputs.as_ref().and_then(|value| value.as_object()) + { + output_paths.extend( + outputs + .values() + .filter_map(serde_json::Value::as_str) + .map(ToOwned::to_owned), + ); + } + + if output_paths.is_empty() + && let Some(output_path) = &build.build_output_path + { + output_paths.push(output_path.clone()); + } + + output_paths.sort(); + output_paths.dedup(); + output_paths +} + /// Main queue runner loop. Polls for pending builds and dispatches them to /// workers. /// @@ -256,6 +337,16 @@ pub async fn run( .await { tracing::warn!(build_id = %build.id, "Failed to complete dedup build: {e}"); + } else { + let output_paths = + completed_build_output_paths(&pool, &existing).await; + publish_succeeded_output_closure( + &pool, + &worker_pool, + build.id, + &output_paths, + ) + .await; } continue; }, @@ -303,6 +394,14 @@ pub async fn run( .await { tracing::warn!(build_id = %build.id, "Failed to complete FOD build: {e}"); + } else { + publish_succeeded_output_closure( + &pool, + &worker_pool, + build.id, + std::slice::from_ref(&output_path), + ) + .await; } continue; } diff --git a/crates/queue-runner/src/worker.rs b/crates/queue-runner/src/worker.rs index 60f71192..d3c16700 100644 --- a/crates/queue-runner/src/worker.rs +++ b/crates/queue-runner/src/worker.rs @@ -18,6 +18,7 @@ use circus_common::{ metric_names, metric_units, }, + narinfo_signing::{read_signing_key, sign_narinfo}, repo, }; use circus_config::{ @@ -54,6 +55,16 @@ use crate::{ rpc::AgentPool, }; +#[derive(Debug, Clone)] +struct ClosurePathInfo { + store_path: String, + nar_hash: String, + nar_size: i64, + references: Vec, + deriver: Option, + ca: Option, +} + pub type ActiveBuilds = Arc>; pub struct WorkerPool { @@ -165,6 +176,22 @@ impl WorkerPool { &self.active_builds } + pub async fn persist_closure_narinfos( + &self, + build_id: Uuid, + output_paths: &[String], + project_id: Option, + ) { + persist_closure_narinfos( + &self.pool, + build_id, + output_paths, + &self.signing_config, + project_id, + ) + .await; + } + #[tracing::instrument(skip(self, build), fields(build_id = %build.id, job = %build.job_name))] pub fn dispatch(&self, build: Build) { if self.drain_token.is_cancelled() { @@ -285,12 +312,111 @@ async fn get_path_info(output_path: &str) -> Option<(String, i64)> { let parsed: serde_json::Value = serde_json::from_str(&stdout).ok()?; let entry = first_path_info_entry(&parsed)?; - let nar_hash = entry.get("narHash")?.as_str()?.to_string(); + let nar_hash = canonical_nix_sha256_hash(entry.get("narHash")?.as_str()?)?; let nar_size = entry.get("narSize")?.as_i64()?; Some((nar_hash, nar_size)) } +async fn get_recursive_path_infos_with_nix( + nix: &Path, + output_paths: &[String], +) -> Option> { + if output_paths.is_empty() { + return Some(Vec::new()); + } + + let output = Command::new(nix) + .args(["path-info", "--json", "--recursive"]) + .args(output_paths) + .output() + .await + .ok()?; + + if !output.status.success() { + tracing::warn!( + stderr = %String::from_utf8_lossy(&output.stderr), + "nix path-info --recursive failed while recording cache closure" + ); + return None; + } + + let parsed: serde_json::Value = + serde_json::from_slice(&output.stdout).ok()?; + Some(parse_recursive_path_infos(&parsed)) +} + +fn parse_recursive_path_infos( + parsed: &serde_json::Value, +) -> Vec { + match parsed { + serde_json::Value::Array(entries) => { + entries + .iter() + .filter_map(|entry| parse_closure_path_info(None, entry)) + .collect() + }, + serde_json::Value::Object(entries) => { + entries + .iter() + .filter_map(|(path, entry)| parse_closure_path_info(Some(path), entry)) + .collect() + }, + _ => Vec::new(), + } +} + +fn parse_closure_path_info( + path_key: Option<&String>, + entry: &serde_json::Value, +) -> Option { + let store_path = entry + .get("path") + .and_then(serde_json::Value::as_str) + .or_else(|| path_key.map(String::as_str))? + .to_string(); + let raw_nar_hash = entry.get("narHash")?.as_str()?; + let Some(nar_hash) = canonical_nix_sha256_hash(raw_nar_hash) else { + tracing::warn!( + store_path = %store_path, + nar_hash = %raw_nar_hash, + "skipping closure path info with unsupported nar hash format" + ); + return None; + }; + let nar_size = entry.get("narSize")?.as_i64()?; + let references = entry + .get("references") + .and_then(serde_json::Value::as_array) + .map(|refs| { + refs + .iter() + .filter_map(serde_json::Value::as_str) + .map(ToOwned::to_owned) + .collect() + }) + .unwrap_or_default(); + let deriver = entry + .get("deriver") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned); + let ca = entry + .get("ca") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned); + + Some(ClosurePathInfo { + store_path, + nar_hash, + nar_size, + references, + deriver, + ca, + }) +} + fn first_path_info_entry( parsed: &serde_json::Value, ) -> Option<&serde_json::Value> { @@ -301,6 +427,194 @@ fn first_path_info_entry( } } +fn store_hash_part(store_path: &str) -> Option<&str> { + let name = store_path.rsplit('/').next()?; + let (hash, _) = name.split_once('-')?; + circus_nix::NixHash::is_valid(hash).then_some(hash) +} + +fn nar_hash_key_segment(nar_hash: &str) -> Option<&str> { + nar_hash + .strip_prefix("sha256:") + .filter(|hash| !hash.is_empty()) +} + +fn canonical_nix_sha256_hash(text: &str) -> Option { + if let Some(sri) = text.strip_prefix("sha256-") { + let mut padded = sri.to_owned(); + while padded.len() % 4 != 0 { + padded.push('='); + } + let bytes = { + use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; + B64.decode(padded).ok()? + }; + return canonical_sha256_bytes(&bytes); + } + + let rest = text.strip_prefix("sha256:")?; + if rest.len() == 52 && rest.bytes().all(is_nix_base32_byte) { + return Some(text.to_owned()); + } + if rest.len() == 64 && rest.bytes().all(|b| b.is_ascii_hexdigit()) { + let bytes = hex::decode(rest).ok()?; + return canonical_sha256_bytes(&bytes); + } + None +} + +fn canonical_sha256_bytes(bytes: &[u8]) -> Option { + if bytes.len() != 32 { + return None; + } + Some(format!("sha256:{}", encode_nix_base32_sha256(bytes))) +} + +const fn is_nix_base32_byte(byte: u8) -> bool { + matches!( + byte, + b'0' + ..=b'9' + | b'a' + | b'b' + | b'c' + | b'd' + | b'f' + | b'g' + | b'h' + | b'i' + | b'j' + | b'k' + | b'l' + | b'm' + | b'n' + | b'p' + | b'q' + | b'r' + | b's' + | b'v' + | b'w' + | b'x' + | b'y' + | b'z' + ) +} + +fn encode_nix_base32_sha256(bytes: &[u8]) -> String { + const ALPHABET: &[u8; 32] = b"0123456789abcdfghijklmnpqrsvwxyz"; + let len = 52; + let mut out = String::with_capacity(len); + for pos in 0..len { + let n = len - 1 - pos; + let bit = n * 5; + let byte = bit / 8; + let offset = bit % 8; + let mut value = u16::from(bytes[byte]) >> offset; + if byte + 1 < bytes.len() { + value |= u16::from(bytes[byte + 1]) << (8 - offset); + } + out.push(ALPHABET[(value & 0x1F) as usize] as char); + } + out +} + +fn nar_url_for_path(info: &ClosurePathInfo) -> Option { + let output_hash = store_hash_part(&info.store_path)?; + let nar_hash = nar_hash_key_segment(&info.nar_hash)?; + Some(format!("nar/{nar_hash}.nar?hash={output_hash}")) +} + +async fn persist_closure_narinfos( + pool: &PgPool, + build_id: Uuid, + output_paths: &[String], + signing_config: &SigningConfig, + project_id: Option, +) { + persist_closure_narinfos_with_nix( + Path::new("nix"), + pool, + build_id, + output_paths, + signing_config, + project_id, + ) + .await; +} + +async fn persist_closure_narinfos_with_nix( + nix: &Path, + pool: &PgPool, + build_id: Uuid, + output_paths: &[String], + signing_config: &SigningConfig, + project_id: Option, +) { + let key_file = match &signing_config.key_file { + Some(kf) if signing_config.enabled && kf.exists() => kf, + _ => return, + }; + let signing_key = match read_signing_key(key_file).await { + Ok(key) => key, + Err(e) => { + tracing::warn!( + key_file = %key_file.display(), + "failed to read closure narinfo signing key: {e}" + ); + return; + }, + }; + + let Some(infos) = get_recursive_path_infos_with_nix(nix, output_paths).await + else { + return; + }; + + for info in infos { + let Some(url) = nar_url_for_path(&info) else { + tracing::warn!( + store_path = %info.store_path, + nar_hash = %info.nar_hash, + "skipping closure narinfo with unsupported hash format" + ); + continue; + }; + + let sig = sign_narinfo( + &signing_key, + &info.store_path, + &info.nar_hash, + info.nar_size, + &info.references, + ); + + if let Err(e) = + repo::narinfo_cache::upsert(pool, repo::narinfo_cache::UpsertNarInfo { + store_path: &info.store_path, + nar_hash: &info.nar_hash, + nar_size: info.nar_size, + file_hash: Some(&info.nar_hash), + file_size: Some(info.nar_size), + compression: "none", + url: &url, + deriver: info.deriver.as_deref(), + references: &info.references, + sig: Some(&sig), + ca: info.ca.as_deref(), + build_id: Some(build_id), + project_id, + }) + .await + { + tracing::warn!( + build_id = %build_id, + store_path = %info.store_path, + "failed to persist closure narinfo: {e}" + ); + } + } +} + fn nix_args_for_build( base_args: &[String], interval_rebuild: bool, @@ -1242,6 +1556,15 @@ async fn run_build(ctx: BuildContext, build: &Build) -> color_eyre::Result<()> { tracing::warn!(build_id = %build.id, "Failed to mark build as signed: {e}"); } + persist_closure_narinfos( + pool, + build.id, + &build_result.output_paths, + signing_config, + project_context.as_ref().map(|(project, _)| project.id), + ) + .await; + // Push to external binary cache if configured let upload_failed_paths = if cache_upload_config.enabled && !build_result.cache_upload_handled @@ -1420,6 +1743,261 @@ mod tests { use super::*; + #[test] + fn test_canonical_nix_sha256_hash_accepts_common_formats() { + let bytes = [7u8; 32]; + let nix32 = encode_nix_base32_sha256(&bytes); + let expected = format!("sha256:{nix32}"); + + assert_eq!( + canonical_nix_sha256_hash(&format!("sha256:{nix32}")).as_deref(), + Some(expected.as_str()) + ); + assert_eq!( + canonical_nix_sha256_hash(&format!("sha256:{}", hex::encode(bytes))) + .as_deref(), + Some(expected.as_str()) + ); + assert_eq!( + canonical_nix_sha256_hash(&format!("sha256-{}", { + use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; + B64.encode(bytes) + })) + .as_deref(), + Some(expected.as_str()) + ); + } + + #[test] + fn test_parse_recursive_path_infos_canonicalizes_sri_nar_hashes() { + let bytes = [11u8; 32]; + let sri_hash = { + use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; + format!("sha256-{}", B64.encode(bytes)) + }; + let expected_hash = format!("sha256:{}", encode_nix_base32_sha256(&bytes)); + let parsed = serde_json::json!({ + "/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-linux-6.18.33-valve2": { + "narHash": sri_hash, + "narSize": 1234, + "references": [ + "/nix/store/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-glibc" + ], + "deriver": "/nix/store/cccccccccccccccccccccccccccccccc-linux.drv", + "ca": null + } + }); + + let infos = parse_recursive_path_infos(&parsed); + + assert_eq!(infos.len(), 1); + assert_eq!( + infos[0].store_path, + "/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-linux-6.18.33-valve2" + ); + assert_eq!(infos[0].nar_hash, expected_hash); + } + + #[tokio::test] + async fn test_persist_closure_narinfos_records_dependency_closure() { + let Ok(url) = std::env::var("TEST_DATABASE_URL") else { + return; + }; + + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&url) + .await + .expect("failed to connect"); + + sqlx::migrate!("../migrations/migrations") + .run(&pool) + .await + .expect("migration failed"); + + let project = repo::projects::create(&pool, circus_common::CreateProject { + name: format!("closure-cache-{}", Uuid::new_v4().simple()), + description: None, + repository_url: "https://github.com/test/closure-cache".to_string(), + cache_enabled: true, + cache_url: None, + cache_upstreams: BinaryCacheUpstreams::default(), + }) + .await + .expect("create project"); + let jobset = repo::jobsets::create(&pool, circus_common::CreateJobset { + project_id: project.id, + name: "main".to_string(), + nix_expression: "packages".to_string(), + enabled: None, + flake_mode: None, + check_interval: None, + trigger_mode: None, + branch: None, + branch_pattern: None, + tag_pattern: None, + scheduling_shares: None, + state: None, + keep_nr: None, + }) + .await + .expect("create jobset"); + let evaluation = + repo::evaluations::create(&pool, circus_common::CreateEvaluation { + jobset_id: jobset.id, + commit_hash: Uuid::new_v4().simple().to_string(), + pr_number: None, + pr_head_branch: None, + pr_base_branch: None, + pr_action: None, + }) + .await + .expect("create evaluation"); + let build = repo::builds::create(&pool, circus_common::CreateBuild { + evaluation_id: evaluation.id, + job_name: "closure-cache-job".to_string(), + drv_path: format!( + "/nix/store/{}-closure-cache-job.drv", + "33333333333333333333333333333333" + ), + system: Some("x86_64-linux".to_string()), + outputs: None, + is_aggregate: None, + constituents: None, + is_fod: None, + fod_hash: None, + ..Default::default() + }) + .await + .expect("create build"); + + let temp_dir = std::env::temp_dir() + .join(format!("circus-closure-cache-{}", Uuid::new_v4().simple())); + tokio::fs::create_dir_all(&temp_dir) + .await + .expect("create temp dir"); + let key_file = temp_dir.join("signing.key"); + let fake_nix = temp_dir.join("nix"); + + // A real Ed25519 key (seed + matching public half); the signer now + // verifies the public key against the seed, so a placeholder is rejected. + let signing_key = "circus-test-1:\ + OlzHrxDxaOpPjkL5uNXF77Xq4VRiz6Zy0LqlK6GCNqRX90gxFy2HSr/\ + hxqdpc2VMU2UIlDOAEBv842MCsbPfgQ==" + .to_string(); + tokio::fs::write(&key_file, signing_key) + .await + .expect("write signing key"); + + let output_store_hash = "11111111111111111111111111111111"; + let dep_store_hash = "22222222222222222222222222222222"; + let output_path = format!("/nix/store/{output_store_hash}-output"); + let dep_path = format!("/nix/store/{dep_store_hash}-dependency"); + let output_nar_bytes = [21u8; 32]; + let dep_nar_bytes = [22u8; 32]; + let output_sri = { + use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; + format!("sha256-{}", B64.encode(output_nar_bytes)) + }; + let dep_sri = { + use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; + format!("sha256-{}", B64.encode(dep_nar_bytes)) + }; + let output_nar_hash = + format!("sha256:{}", encode_nix_base32_sha256(&output_nar_bytes)); + let dep_nar_hash = + format!("sha256:{}", encode_nix_base32_sha256(&dep_nar_bytes)); + let path_info = serde_json::json!({ + output_path.clone(): { + "narHash": output_sri, + "narSize": 123, + "references": [dep_path.clone()], + "deriver": "/nix/store/44444444444444444444444444444444-output.drv", + "ca": null + }, + dep_path.clone(): { + "narHash": dep_sri, + "narSize": 45, + "references": [], + "deriver": serde_json::Value::Null, + "ca": null + } + }); + tokio::fs::write( + &fake_nix, + format!("#!/bin/sh\ncat <<'JSON'\n{path_info}\nJSON\n"), + ) + .await + .expect("write fake nix"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + + let mut permissions = std::fs::metadata(&fake_nix) + .expect("stat fake nix") + .permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&fake_nix, permissions).expect("chmod fake nix"); + } + + persist_closure_narinfos_with_nix( + &fake_nix, + &pool, + build.id, + std::slice::from_ref(&output_path), + &SigningConfig { + enabled: true, + key_file: Some(key_file), + }, + Some(project.id), + ) + .await; + + let output_row = repo::narinfo_cache::get(&pool, &output_path) + .await + .expect("output narinfo row"); + let dep_row = repo::narinfo_cache::get(&pool, &dep_path) + .await + .expect("dependency narinfo row"); + + assert_eq!(output_row.build_id, Some(build.id)); + assert_eq!(output_row.project_id, Some(project.id)); + assert_eq!(output_row.nar_hash, output_nar_hash); + assert_eq!( + output_row.url, + format!( + "nar/{}.nar?hash={output_store_hash}", + output_nar_hash + .strip_prefix("sha256:") + .expect("canonical nar hash should have sha256 prefix") + ) + ); + assert_eq!(output_row.references, vec![dep_path.clone()]); + assert!(output_row.sig.as_deref().is_some_and(|sig| { + sig.starts_with("circus-test-1:") && sig.len() > "circus-test-1:".len() + })); + + // The transitive dependency is cached and signed too, not just the output. + assert_eq!(dep_row.nar_hash, dep_nar_hash); + assert!(dep_row.sig.as_deref().is_some_and(|sig| { + sig.starts_with("circus-test-1:") && sig.len() > "circus-test-1:".len() + })); + + let cleanup_paths = [output_path, dep_path]; + let _ = sqlx::query("DELETE FROM narinfo_cache WHERE store_path = ANY($1)") + .bind(&cleanup_paths[..]) + .execute(&pool) + .await; + let _ = repo::builds::delete(&pool, build.id).await; + let _ = sqlx::query("DELETE FROM evaluations WHERE id = $1") + .bind(evaluation.id) + .execute(&pool) + .await; + let _ = repo::jobsets::delete(&pool, jobset.id).await; + let _ = repo::projects::delete(&pool, project.id).await; + let _ = tokio::fs::remove_dir_all(&temp_dir).await; + } + #[test] fn test_cache_args_use_project_cache_and_upstreams() { let cache_config = CacheConfig { diff --git a/crates/server/src/routes/cache.rs b/crates/server/src/routes/cache.rs index e23a36f4..12e9bb0a 100644 --- a/crates/server/src/routes/cache.rs +++ b/crates/server/src/routes/cache.rs @@ -223,6 +223,23 @@ async fn has_circus_build_product( .map_err(|e| ApiError(circus_common::CiError::Database(e))) } +async fn has_signed_persisted_narinfo( + pool: &PgPool, + store_path: &str, + project_id: Option, +) -> Result { + sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM narinfo_cache WHERE store_path = $1 AND sig \ + IS NOT NULL AND btrim(sig) != '' AND ($2::uuid IS NULL OR project_id = \ + $2))", + ) + .bind(store_path) + .bind(project_id) + .fetch_one(pool) + .await + .map_err(|e| ApiError(circus_common::CiError::Database(e))) +} + /// As [`has_circus_build_product`], but additionally requires that the build /// was signed by Circus. async fn has_circus_signed_build_product( @@ -301,15 +318,12 @@ async fn is_servable_harmonia_path( info: &ValidPathInfo, scope: CacheScope, ) -> Result { - // The unauthenticated cache may only rebroadcast paths Circus built, never - // arbitrary store paths. Content addressing makes a path self-verifying - // (integrity) but not confidential, so it is no license to serve; the - // boundary is provenance. + // The unauthenticated cache only rebroadcasts paths Circus built, never + // arbitrary store paths. let store_path = info.info.store_dir.display(&info.path).to_string(); - // A dispatched build's own .drv: agents substitute it from this cache to - // start the build. Derivations are content-addressed, so this must be - // checked before the generic CA branch below, which only covers build - // outputs and their direct inputs, never the derivation file itself. + + // A dispatched build's own .drv, which agents substitute from this cache to + // start the build. if PathBuf::from(&store_path) .extension() .is_some_and(|ext| ext.eq_ignore_ascii_case("drv")) @@ -317,6 +331,10 @@ async fn is_servable_harmonia_path( { return Ok(true); } + if has_signed_persisted_narinfo(pool, &store_path, scope.project_id()).await? + { + return Ok(true); + } if info.info.ca.is_some() { // Serve when Circus built it or when it is a direct input of an evaluated // derivation Circus may dispatch to an agent.