mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

pull swarm closures through the nix daemon: substituters in the fragment, merged root-only netrc

Browse source
This commit is contained in:
atlas 2026-11-08 10:29:46 +01:00 • committed by mara
commit ef68baa00f
1 changed files with 582 additions and 382 deletions
Download patch file Download diff file Expand all files Collapse all files

343
hive-c0re/src/cache_pull.rs
View file

@ -1,36 +1,35 @@
//! Pull the swarm's closure into the store with `nix copy --from`.
//! Pull the swarm's closure into the store through the host's nix daemon.
//!
//! hive-c0re is not root and does not need to be. The nix client does the HTTP
//! fetch itself, with a netrc only hive-core can read ([`NETRC`]), and hands
//! the NARs to the host's nix daemon, which stores a path from an unprivileged
//! client only if a key in its `trusted-public-keys` signed it. The root unit
//! `hive-priv-cache-facts` puts the swarm's key there through a nix.conf
//! fragment ([`CONF_FRAGMENT`]) that names no substituter and no netrc, so the
//! daemon itself never contacts the cache and no other client of it can fetch
//! from the cache or send the token. hive-priv then applies the toplevel only
//! if the swarm's key signed it.
//! hive-c0re is not root and does not need to be: `nix-store --realise` of a
//! store path is a request to the daemon, which substitutes each path of the
//! closure from the swarm cache or from its other substituters (cache.nixos.org),
//! whichever has it. The swarm cache leaves out the paths cache.nixos.org
//! serves, so a client that fetched from the cache alone could not complete a
//! closure. The root unit `hive-priv-cache-facts` writes a nix.conf fragment
//! ([`CONF_FRAGMENT`]) that gives the daemon the cache's address, the swarm's
//! key and the netrc, which only root reads. The daemon stores a path only if a
//! key in its `trusted-public-keys` signed it, and hive-priv then applies the
//! toplevel only if the swarm's key signed it.
//!
//! The cache's address is the host's configuration, handed to the unit in
//! [`URL_ENV`]. [`pull`] first asks hive-priv to run the facts unit when the
//! netrc or the key is not on the host yet, which is how the swarm's copy of the
//! facts reaches the host without waiting for the unit's timer.
//! [`pull`] first asks hive-priv to run the facts unit when the fragment is not
//! on the host yet, which is how the swarm's copy of the facts reaches the host
//! without waiting for the unit's timer.
use std::time::Duration;
use anyhow::Result;
use hive_priv_sock::valid_closure_path;
/// The pull token as a netrc, `root:hive-core 0640`.
pub const NETRC: &str = "/run/hyperhive-nix/netrc";
/// The nix.conf fragment `hive-priv-cache-facts` writes (root-owned, readable).
pub const CONF_FRAGMENT: &str = "/run/hyperhive-nix/cache.conf";
/// The unit environment variable naming the cache: `https://<domain>/<cache>`.
const URL_ENV: &str = "HIVE_C0RE_CACHE_URL";
/// The fragment's line that gives the daemon the swarm's key.
const KEY_SETTING: &str = "extra-trusted-public-keys";
/// The fragment's lines the daemon needs to substitute from the swarm cache and
/// to trust what it gets.
const REQUIRED_SETTINGS: [&str; 3] = [
"extra-substituters",
"extra-trusted-public-keys",
"netrc-file",
];
/// Wall-clock bound on one pull. A stalled transfer is bounded by nix's own
/// `stalled-download-timeout`; this covers the rest.
@ -39,12 +38,12 @@ const PULL_TIMEOUT: Duration = Duration::from_mins(30);
/// Why a closure was not pulled.
#[derive(Debug, PartialEq, Eq)]
pub enum PullError {
/// This host has no cache facts: no address, no netrc for hive-core, or no
/// key in the fragment. Waiting for the facts unit fixes the last two.
/// This host has no cache facts: no fragment, or one that lacks a setting
/// the daemon needs. Waiting for the facts unit fixes it.
FactsMissing(String),
/// The daemon refused the path: no signature by a key it trusts.
SignatureRefused(String),
/// The cache does not have the path.
/// No substituter has the path.
NotFound(String),
/// Anything else nix said, or could not do.
Failed(String),
@ -63,16 +62,19 @@ impl std::fmt::Display for PullError {
f,
"the nix daemon refused the closure, which carries no signature by a key this host trusts: {last}"
),
Self::NotFound(last) => write!(f, "the swarm cache does not have the closure: {last}"),
Self::NotFound(last) => write!(
f,
"no substituter of the nix daemon has the closure: {last}"
),
Self::Failed(last) => f.write_str(last),
Self::TimedOut => write!(f, "nix copy timed out after {}s", PULL_TIMEOUT.as_secs()),
Self::TimedOut => write!(f, "nix-store timed out after {}s", PULL_TIMEOUT.as_secs()),
}
}
}
impl std::error::Error for PullError {}
/// What a `nix copy` run came to.
/// What a `nix-store` run came to.
pub enum Run {
Exited { success: bool, stderr: String },
TimedOut,
@ -80,63 +82,53 @@ pub enum Run {
/// What [`pull`] does on the host, so a test can stand in for it.
pub trait Effects: Sync {
/// The cache's address, when the host names one.
fn cache_url(&self) -> Option<String>;
/// Whether hive-core can read the netrc and the fragment gives the daemon
/// the swarm's key.
async fn facts_present(&self) -> bool;
/// The fragment's text; `None` when it cannot be read.
async fn fragment(&self) -> Option<String>;
/// Run the facts unit through hive-priv and wait for it.
async fn refresh(&self) -> Result<()>;
/// Run `nix` with `args`, bounded by [`PULL_TIMEOUT`].
async fn nix(&self, args: Vec<String>) -> std::io::Result<Run>;
/// Run `nix-store` with `args`, bounded by [`PULL_TIMEOUT`].
async fn nix_store(&self, args: Vec<String>) -> std::io::Result<Run>;
}
/// Whether `fragment` has a non-empty `extra-trusted-public-keys` line.
/// The name of the first setting in [`REQUIRED_SETTINGS`] that `fragment` gives
/// no value, or `None` when it gives all of them.
#[must_use]
pub fn gives_key(fragment: &str) -> bool {
fragment.lines().any(|line| {
line.split_once('=')
.is_some_and(|(name, value)| name.trim() == KEY_SETTING && !value.trim().is_empty())
pub fn missing_setting(fragment: &str) -> Option<&'static str> {
REQUIRED_SETTINGS.into_iter().find(|wanted| {
!fragment.lines().any(|line| {
line.split_once('=')
.is_some_and(|(name, value)| name.trim() == *wanted && !value.trim().is_empty())
})
})
}
/// The `nix` arguments that fetch `store_path` from the cache at `url`.
/// The `nix-store` arguments that fetch `store_path` and nothing else.
///
/// - `copy --from <url>` makes the client fetch. The daemon is never asked to
/// contact the cache, so it needs no trust in the address.
/// - `--option netrc-file` is the client's own setting: it reads the file as
/// hive-core. The daemon ignores the setting from an unprivileged client.
/// - No `--no-check-sigs`: the daemon checks every path's signature against its
/// trusted keys. Nothing builds: `copy` only imports.
/// - `--realise <path>` asks the daemon to substitute the path and every path
/// it references that is not yet valid, from the substituters the daemon is
/// configured with.
/// - `--max-jobs 0` forbids a local build, so a path no substituter has fails.
/// - No `substituters`, `trusted-public-keys`, `require-sigs` or `netrc-file`:
/// the daemon ignores them from an unprivileged client and applies its own.
#[must_use]
pub fn pull_args(store_path: &str, url: &str, netrc: &str) -> Vec<String> {
[
"--extra-experimental-features",
"nix-command",
"copy",
"--from",
url,
"--option",
"netrc-file",
netrc,
store_path,
]
.map(str::to_owned)
.into()
pub fn pull_args(store_path: &str) -> Vec<String> {
["--realise", store_path, "--max-jobs", "0"]
.map(str::to_owned)
.into()
}
/// Name nix's refusal from its stderr. A refusal for want of a signature is
/// told apart from a path the cache lacks and from a cache that is down.
/// told apart from a path no substituter has and from a cache that is down.
fn classify(stderr: &str) -> PullError {
let last = stderr
.lines()
.rev()
.find(|l| !l.trim().is_empty())
.unwrap_or("nix copy failed with no message")
.unwrap_or("nix-store failed with no message")
.trim()
.to_owned();
let lower = stderr.to_lowercase();
if lower.contains("signature") {
if lower.contains("signature") || lower.contains("not signed by any of the keys") {
PullError::SignatureRefused(last)
} else if lower.contains("don't know how to build")
|| lower.contains("no substituter that can build")
@ -150,9 +142,9 @@ fn classify(stderr: &str) -> PullError {
/// Pull `store_path` into the store.
///
/// # Errors
/// [`PullError`]: no cache facts on this host (after one refresh), the daemon
/// refusing the signature, the cache lacking the path, nix failing otherwise,
/// or a timeout.
/// [`PullError`]: no cache facts on this host (after one refresh), nix
/// refusing the signature, no substituter having the path, nix failing
/// otherwise, or a timeout.
pub async fn pull(store_path: &str) -> Result<(), PullError> {
pull_with(&Real, store_path).await
}
@ -163,14 +155,11 @@ async fn pull_with(fx: &impl Effects, store_path: &str) -> Result<(), PullError>
"{store_path:?} is not a store path"
)));
}
let url = fx
.cache_url()
.ok_or_else(|| PullError::FactsMissing("this host names no cache".to_owned()))?;
ensure_facts(fx).await?;
match fx
.nix(pull_args(store_path, &url, NETRC))
.nix_store(pull_args(store_path))
.await
.map_err(|e| PullError::Failed(format!("running nix: {e}")))?
.map_err(|e| PullError::Failed(format!("running nix-store: {e}")))?
{
Run::Exited { success: true, .. } => Ok(()),
Run::Exited { stderr, .. } => Err(classify(&stderr)),
@ -178,44 +167,44 @@ async fn pull_with(fx: &impl Effects, store_path: &str) -> Result<(), PullError>
}
}
/// What the fragment lacks, or `None` when the daemon has what it needs.
async fn facts_lacking(fx: &impl Effects) -> Option<String> {
match fx.fragment().await {
None => Some("no fragment".to_owned()),
Some(text) => missing_setting(&text).map(|name| format!("the fragment has no {name}")),
}
}
/// The facts are on the host, after one refresh when they are not.
async fn ensure_facts(fx: &impl Effects) -> Result<(), PullError> {
if fx.facts_present().await {
if facts_lacking(fx).await.is_none() {
return Ok(());
}
let refreshed = fx.refresh().await;
if fx.facts_present().await {
return Ok(());
match facts_lacking(fx).await {
None => Ok(()),
Some(lacking) => Err(PullError::FactsMissing(match refreshed {
Ok(()) => format!("the facts unit ran and {lacking}"),
Err(e) => format!("refreshing them failed: {e:#}"),
})),
}
Err(PullError::FactsMissing(match refreshed {
Ok(()) => "the facts unit ran and the swarm has given this hive none".to_owned(),
Err(e) => format!("refreshing them failed: {e:#}"),
}))
}
/// The host: the environment and files on disk, hive-priv for the refresh,
/// `nix` from the system profile.
/// The host: the fragment on disk, hive-priv for the refresh, `nix-store` from
/// the system profile.
struct Real;
impl Effects for Real {
fn cache_url(&self) -> Option<String> {
std::env::var(URL_ENV)
.ok()
.map(|u| u.trim().to_owned())
.filter(|u| u.starts_with("https://") && !u.contains(char::is_whitespace))
}
async fn facts_present(&self) -> bool {
let fragment = tokio::fs::read_to_string(CONF_FRAGMENT).await;
fragment.is_ok_and(|f| gives_key(&f)) && tokio::fs::File::open(NETRC).await.is_ok()
async fn fragment(&self) -> Option<String> {
tokio::fs::read_to_string(CONF_FRAGMENT).await.ok()
}
async fn refresh(&self) -> Result<()> {
crate::priv_client::refresh_cache_facts().await
}
async fn nix(&self, args: Vec<String>) -> std::io::Result<Run> {
let mut cmd = tokio::process::Command::new("nix");
async fn nix_store(&self, args: Vec<String>) -> std::io::Result<Run> {
let mut cmd = tokio::process::Command::new("nix-store");
// Explicit, so the store is the daemon's whatever the environment.
cmd.args(args)
.env("NIX_REMOTE", "daemon")
@ -242,25 +231,23 @@ mod tests {
use super::*;
const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
const URL: &str = "https://cache.example/swarm";
const FRAGMENT: &str = "extra-substituters = https://cache.example/swarm\nextra-trusted-public-keys = swarm:AAAA\nnetrc-file = /run/hyperhive-priv/attic/netrc";
/// A host that answers from a script: whether the facts are present now
/// (and after a refresh), what `nix` comes to, and a log of calls.
/// A host that answers from a script: the fragment now (and after a
/// refresh), what `nix-store` comes to, and a log of calls.
struct Fake {
url: Option<&'static str>,
present: Mutex<bool>,
present_after_refresh: bool,
fragment: Mutex<Option<&'static str>>,
fragment_after_refresh: Option<&'static str>,
refresh_fails: bool,
nix: Mutex<Option<Run>>,
calls: Mutex<Vec<String>>,
}
impl Fake {
fn new(present: bool, nix: Run) -> Self {
fn new(fragment: Option<&'static str>, nix: Run) -> Self {
Self {
url: Some(URL),
present: Mutex::new(present),
present_after_refresh: true,
fragment: Mutex::new(fragment),
fragment_after_refresh: Some(FRAGMENT),
refresh_fails: false,
nix: Mutex::new(Some(nix)),
calls: Mutex::new(Vec::new()),
@ -273,13 +260,9 @@ mod tests {
}
impl Effects for Fake {
fn cache_url(&self) -> Option<String> {
self.url.map(str::to_owned)
}
async fn facts_present(&self) -> bool {
self.calls.lock().unwrap().push("facts".to_owned());
*self.present.lock().unwrap()
async fn fragment(&self) -> Option<String> {
self.calls.lock().unwrap().push("fragment".to_owned());
self.fragment.lock().unwrap().map(str::to_owned)
}
async fn refresh(&self) -> Result<()> {
@ -287,15 +270,15 @@ mod tests {
if self.refresh_fails {
anyhow::bail!("hive-priv unreachable");
}
*self.present.lock().unwrap() = self.present_after_refresh;
*self.fragment.lock().unwrap() = self.fragment_after_refresh;
Ok(())
}
async fn nix(&self, args: Vec<String>) -> std::io::Result<Run> {
async fn nix_store(&self, args: Vec<String>) -> std::io::Result<Run> {
self.calls
.lock()
.unwrap()
.push(format!("nix {}", args.join(" ")));
.push(format!("nix-store {}", args.join(" ")));
Ok(self.nix.lock().unwrap().take().expect("one run"))
}
}
@ -307,75 +290,55 @@ mod tests {
}
}
fn after(args: &[String], flag: &str) -> Option<String> {
args.iter()
.position(|a| a == flag)
.and_then(|i| args.get(i + 1))
.cloned()
}
/// The client fetches: `copy --from <url>`, with its own netrc as a client
/// setting. Nothing asks the daemon to contact the cache or to skip the
/// signature check, and nothing can build.
/// The daemon substitutes: `--realise <path>`, no source named, no client
/// setting, nothing that widens trust, and nothing can build.
#[test]
fn the_command_copies_from_the_cache_with_the_clients_netrc_and_checks_signatures() {
let args = pull_args(PATH, URL, NETRC);
assert_eq!(args.last().map(String::as_str), Some(PATH));
let at = args.iter().position(|a| a == "copy").expect("verb");
assert_eq!(args[at + 1], "--from");
assert_eq!(args[at + 2], URL);
assert_eq!(after(&args, "--option").as_deref(), Some("netrc-file"));
fn the_command_realises_the_path_through_the_daemon_without_naming_a_source() {
assert_eq!(
args.windows(3)
.find(|w| w[0] == "--option" && w[1] == "netrc-file")
.map(|w| w[2].as_str()),
Some(NETRC)
pull_args(PATH),
["--realise", PATH, "--max-jobs", "0"].map(str::to_owned)
);
// Exactly one setting, and none that widens trust.
assert_eq!(
args.iter().filter(|a| *a == "--option").count(),
1,
"{args:?}"
);
for forbidden in [
"--no-check-sigs",
"--substitute-on-destination",
"--to",
"substituters",
"trusted-public-keys",
"require-sigs",
] {
assert!(
!args.iter().any(|a| a == forbidden),
"{forbidden}: {args:?}"
);
}
assert!(!args.iter().any(|a| a.starts_with("extra-")), "{args:?}");
}
#[test]
fn the_fragment_gives_the_key_when_it_has_a_value() {
assert!(gives_key("extra-trusted-public-keys = swarm:AAAA"));
for bad in [
"",
"extra-trusted-public-keys =",
"extra-trusted-public-keys",
"trusted-public-keys = swarm:AAAA",
"extra-trusted-substituters = https://cache.example/swarm",
fn the_fragment_is_complete_when_every_setting_has_a_value() {
assert_eq!(missing_setting(FRAGMENT), None);
for (bad, want) in [
("", "extra-substituters"),
(
"extra-substituters =\nextra-trusted-public-keys = k\nnetrc-file = n",
"extra-substituters",
),
(
"extra-substituters = u\nextra-trusted-public-keys\nnetrc-file = n",
"extra-trusted-public-keys",
),
(
"extra-substituters = u\nextra-trusted-public-keys = k",
"netrc-file",
),
(
"extra-trusted-substituters = u\nextra-trusted-public-keys = k\nnetrc-file = n",
"extra-substituters",
),
(
"extra-substituters = u\ntrusted-public-keys = k\nnetrc-file = n",
"extra-trusted-public-keys",
),
] {
assert!(!gives_key(bad), "{bad:?}");
assert_eq!(missing_setting(bad), Some(want), "{bad:?}");
}
}
#[tokio::test]
async fn with_the_facts_in_place_it_pulls_without_a_refresh() {
let fx = Fake::new(true, exited(true, ""));
let fx = Fake::new(Some(FRAGMENT), exited(true, ""));
pull_with(&fx, PATH).await.expect("pulled");
assert_eq!(
fx.calls(),
[
"facts".to_owned(),
format!("nix {}", pull_args(PATH, URL, NETRC).join(" "))
"fragment".to_owned(),
format!("nix-store {}", pull_args(PATH).join(" "))
]
);
}
@ -383,30 +346,37 @@ mod tests {
/// Missing facts are refreshed once, and the pull then goes ahead.
#[tokio::test]
async fn missing_facts_are_refreshed_before_the_pull() {
let fx = Fake::new(false, exited(true, ""));
let fx = Fake::new(None, exited(true, ""));
pull_with(&fx, PATH).await.expect("pulled");
assert_eq!(
fx.calls(),
[
"facts".to_owned(),
"fragment".to_owned(),
"refresh".to_owned(),
"facts".to_owned(),
format!("nix {}", pull_args(PATH, URL, NETRC).join(" "))
"fragment".to_owned(),
format!("nix-store {}", pull_args(PATH).join(" "))
]
);
}
/// Facts still absent after the refresh, whether the unit ran or hive-priv
/// could not start it, stop the pull before nix is asked.
/// could not start it, stop the pull before nix is asked. A fragment that
/// lacks a setting counts as absent.
#[tokio::test]
async fn facts_still_missing_after_the_refresh_are_reported_and_nix_is_not_run() {
for (case, refresh_fails, want) in [
("unit ran", false, "none"),
("refresh failed", true, "hive-priv unreachable"),
for (case, fragment, refresh_fails, want) in [
("no fragment", None, false, "no fragment"),
(
"no substituter",
Some("extra-trusted-public-keys = k\nnetrc-file = n"),
false,
"has no extra-substituters",
),
("refresh failed", None, true, "hive-priv unreachable"),
] {
let mut fx = Fake::new(false, exited(true, ""));
let mut fx = Fake::new(None, exited(true, ""));
fx.refresh_fails = refresh_fails;
fx.present_after_refresh = false;
fx.fragment_after_refresh = fragment;
let err = pull_with(&fx, PATH).await.expect_err(case);
let PullError::FactsMissing(why) = &err else {
panic!("{case}: {err:?}");
@ -416,20 +386,13 @@ mod tests {
format!("{err}").contains("a later deploy retries"),
"{case}"
);
assert!(!fx.calls().iter().any(|c| c.starts_with("nix")), "{case}");
assert!(
!fx.calls().iter().any(|c| c.starts_with("nix-store")),
"{case}"
);
}
}
/// A host that names no cache has nothing to refresh toward.
#[tokio::test]
async fn a_host_that_names_no_cache_does_not_pull_or_refresh() {
let mut fx = Fake::new(true, exited(true, ""));
fx.url = None;
let err = pull_with(&fx, PATH).await.expect_err("no cache");
assert!(matches!(err, PullError::FactsMissing(_)), "{err:?}");
assert!(fx.calls().is_empty(), "{:?}", fx.calls());
}
#[test]
fn nixs_refusals_are_told_apart() {
for (stderr, want) in [
@ -472,7 +435,7 @@ mod tests {
#[tokio::test]
async fn a_failed_run_is_mapped_and_a_timeout_is_a_timeout() {
let fx = Fake::new(
true,
Some(FRAGMENT),
exited(
false,
"error: cannot add path 'x' because it lacks a signature by a trusted key",
@ -482,7 +445,7 @@ mod tests {
pull_with(&fx, PATH).await,
Err(PullError::SignatureRefused(_))
));
let fx = Fake::new(true, Run::TimedOut);
let fx = Fake::new(Some(FRAGMENT), Run::TimedOut);
assert_eq!(pull_with(&fx, PATH).await, Err(PullError::TimedOut));
}
@ -490,7 +453,7 @@ mod tests {
#[tokio::test]
async fn only_a_store_path_is_pulled() {
for bad in ["", "main", "/etc/passwd", "--help", "/nix/store/x"] {
let fx = Fake::new(true, exited(true, ""));
let fx = Fake::new(Some(FRAGMENT), exited(true, ""));
assert!(
matches!(pull_with(&fx, bad).await, Err(PullError::Failed(_))),
"{bad}"