hyperhive/hive-priv/src/closure.rs

//! `RealiseClosure`: download the closure the swarm built for an agent from
//! the swarm's binary cache into the store, without ever building.
//!
//! What this accepts is decided by root-owned sources alone: the cache's
//! address is the host's configuration, handed to the unit in
//! [`CACHE_URL_ENV`], and the cache's public key and the pull token's netrc are
//! files a root unit keeps in [`ROOT_FACTS_DIR`]. Nothing in the request, and
//! nothing hive-c0re (unprivileged) writes, chooses a cache or a key. They go to
//! nix's daemon as per-call settings, which the daemon applies for a trusted
//! client such as this one (root), so the host's nix configuration is untouched.

use std::os::unix::fs::MetadataExt as _;
use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{SIGNATURE_REFUSED, valid_closure_path};

use crate::container::{BoundedRun, run_bounded};
use crate::shared::validate_agent_name;

/// The unit's environment variable naming the cache: `https://<domain>/<cache>`
/// from the host's `services.hyperhive.swarm.attic` options.
pub(crate) const CACHE_URL_ENV: &str = "HIVE_PRIV_CACHE_URL";

/// Where the root unit `hive-priv-cache-facts` keeps `public-key` and `netrc`.
/// Not `/run/hyperhive/attic`, which hive-c0re writes.
const ROOT_FACTS_DIR: &str = "/run/hyperhive-priv/attic";

/// Wall-clock bound on one download. A stalled transfer is bounded by nix's
/// own `stalled-download-timeout`; this covers the rest.
const REALISE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(30);

/// The cache this host trusts.
#[derive(Debug, PartialEq, Eq)]
struct Cache {
    substituter: String,
    public_key: String,
    netrc: PathBuf,
}

/// Whether `path` is a regular file or directory owned by `owner` that nobody
/// else can read or write.
fn private_to(path: &Path, owner: u32, directory: bool) -> Result<()> {
    let meta = std::fs::symlink_metadata(path)
        .with_context(|| format!("{} is missing", path.display()))?;
    let kind_ok = if directory {
        meta.is_dir()
    } else {
        meta.is_file()
    };
    if !kind_ok {
        bail!(
            "{} is not a plain {}",
            path.display(),
            if directory { "directory" } else { "file" }
        );
    }
    if meta.uid() != owner {
        bail!("{} is not owned by uid {owner}", path.display());
    }
    if meta.mode() & 0o077 != 0 {
        bail!("{} is readable or writable by others", path.display());
    }
    Ok(())
}

impl Cache {
    /// The cache at `url`, trusting the key in `dir`.
    ///
    /// # Errors
    /// No `url`, one that is not a plain `https` address (the token would go
    /// out in clear), a `dir` or file that is missing, not owned by `owner` or
    /// open to others, or a key that is not `<name>:<key>`.
    fn read(url: Option<&str>, dir: &Path, owner: u32) -> Result<Self> {
        let url = url
            .map(str::trim)
            .filter(|u| !u.is_empty())
            .context("this host names no cache")?;
        if !url.starts_with("https://") || url.contains(char::is_whitespace) {
            bail!("the cache address is not an https URL");
        }
        private_to(dir, owner, true)?;
        let key_file = dir.join("public-key");
        private_to(&key_file, owner, false)?;
        let netrc = dir.join("netrc");
        private_to(&netrc, owner, false)?;
        let key = std::fs::read_to_string(&key_file)
            .with_context(|| format!("reading {}", key_file.display()))?;
        let key = key.trim();
        if key.contains(char::is_whitespace)
            || key
                .split_once(':')
                .is_none_or(|(name, key)| name.is_empty() || key.is_empty())
        {
            bail!("the cache's public key is not <name>:<key>");
        }
        Ok(Self {
            substituter: url.to_owned(),
            public_key: key.to_owned(),
            netrc,
        })
    }
}

/// The `nix-store` arguments that fetch `store_path` and nothing else.
///
/// - `--max-jobs 0` forbids a local build, so a path the cache lacks fails.
/// - `substituters` and `trusted-public-keys` are replaced, not extended: the
///   swarm cache is the only source and its key the only one that counts.
/// - `require-sigs true` makes nix refuse a path no trusted key signed.
fn realise_args(store_path: &str, cache: &Cache) -> Vec<String> {
    let mut args: Vec<String> = ["--realise", store_path, "--max-jobs", "0"]
        .map(str::to_owned)
        .into();
    let netrc = cache.netrc.display().to_string();
    for (name, value) in [
        ("substituters", cache.substituter.as_str()),
        ("trusted-public-keys", cache.public_key.as_str()),
        ("require-sigs", "true"),
        ("netrc-file", netrc.as_str()),
    ] {
        args.extend(["--option".to_owned(), name.to_owned(), value.to_owned()]);
    }
    args
}

/// Why nix did not realise the path, from its last stderr line. A refusal for
/// want of a signature is named, so the caller can tell it from a cache that
/// is down.
fn reason(stderr: &str) -> String {
    let last = stderr.lines().last().unwrap_or("").trim();
    if last.to_lowercase().contains("signature") {
        format!("{SIGNATURE_REFUSED}: {last}")
    } else {
        last.to_owned()
    }
}

/// `RealiseClosure`.
///
/// # Errors
/// Bad arguments, or `not available: <reason>` for anything that left the path
/// out of the store: no cache or key on this host, nix failing or timing out.
pub(crate) async fn realise_closure(agent: &str, store_path: &str) -> Result<(String, String)> {
    let url = std::env::var(CACHE_URL_ENV).ok();
    let done = realise_with(
        url.as_deref(),
        Path::new(ROOT_FACTS_DIR),
        0,
        agent,
        store_path,
    )
    .await?;
    if let Ok(mut realised) = REALISED.lock() {
        remember(&mut realised, agent, store_path);
    }
    Ok(done)
}

/// The store path each agent's last successful `RealiseClosure` put in the
/// store. `CreateContainer` / `UpdateContainer` apply only that path for that
/// agent. hive-priv is a long-running process, so the entry outlives the
/// calls between the download and the apply; a restart in between empties it
/// and the apply is refused.
static REALISED: std::sync::Mutex<Vec<(String, String)>> = std::sync::Mutex::new(Vec::new());

fn remember(realised: &mut Vec<(String, String)>, agent: &str, store_path: &str) {
    realised.retain(|(a, _)| a != agent);
    realised.push((agent.to_owned(), store_path.to_owned()));
}

/// The path recorded for `agent`.
fn lookup<'a>(realised: &'a [(String, String)], agent: &str) -> Option<&'a str> {
    realised
        .iter()
        .find(|(a, _)| a == agent)
        .map(|(_, p)| p.as_str())
}

/// `toplevel` when it is a store path and the one `agent`'s last
/// `RealiseClosure` returned (`realised`); else the refusal, which is also
/// logged at error level with the agent and both paths.
///
/// The request comes from hive-c0re, which is not root. A signed path is not
/// enough: it could be another agent's closure or the host's own.
///
/// # Errors
/// `toplevel` is not a store path, or is not `realised`.
pub(crate) fn applicable<'a>(
    agent: &str,
    toplevel: &'a str,
    realised: Option<&str>,
) -> Result<&'a str> {
    if !valid_closure_path(toplevel) {
        bail!("{toplevel:?} is not a store path");
    }
    if realised == Some(toplevel) {
        return Ok(toplevel);
    }
    let realised = realised.unwrap_or("none");
    tracing::error!(
        %agent, requested = %toplevel, %realised,
        "refusing to apply a closure that is not the one realised for the agent"
    );
    bail!("{agent}: {toplevel} is not the closure realised for this agent ({realised})")
}

/// [`applicable`] against what `RealiseClosure` recorded for `agent`.
///
/// # Errors
/// As [`applicable`].
pub(crate) fn applicable_for<'a>(agent: &str, toplevel: &'a str) -> Result<&'a str> {
    let realised = REALISED
        .lock()
        .ok()
        .and_then(|r| lookup(&r, agent).map(str::to_owned));
    applicable(agent, toplevel, realised.as_deref())
}

/// [`realise_closure`] with the cache's address and the key's directory and
/// owner named, so a test can stand in for the host.
async fn realise_with(
    url: Option<&str>,
    dir: &Path,
    owner: u32,
    agent: &str,
    store_path: &str,
) -> Result<(String, String)> {
    validate_agent_name(agent)?;
    if !valid_closure_path(store_path) {
        bail!("{store_path:?} is not a store path");
    }
    let outcome = async {
        let cache = Cache::read(url, dir, owner)?;
        let mut cmd = tokio::process::Command::new("nix-store");
        cmd.args(realise_args(store_path, &cache));
        match run_bounded(cmd, REALISE_TIMEOUT)
            .await
            .context("running nix-store")?
        {
            BoundedRun::Exited { status, .. } if status.success() => Ok(()),
            BoundedRun::Exited { stderr, .. } => bail!("{}", reason(&stderr)),
            BoundedRun::TimedOut => bail!(
                "nix-store timed out after {}s; killed it",
                REALISE_TIMEOUT.as_secs()
            ),
        }
    }
    .await;
    match outcome {
        Ok(()) => {
            tracing::info!(%agent, %store_path, "substituted the agent's closure");
            Ok(("substituted".to_owned(), String::new()))
        }
        Err(e) => {
            let reason = format!("{e:#}");
            if reason.contains(SIGNATURE_REFUSED) {
                tracing::error!(%agent, %store_path, %reason, "the cache's closure was refused");
            }
            bail!("not available: {reason}")
        }
    }
}

#[cfg(test)]
mod tests {
    use std::os::unix::fs::PermissionsExt as _;

    use super::*;
    use crate::shared::test_support::scratch;

    const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
    const URL: &str = "https://cache.example/swarm";

    fn cache() -> Cache {
        Cache {
            substituter: URL.to_owned(),
            public_key: "swarm:AAAA".to_owned(),
            netrc: PathBuf::from("/run/hyperhive-priv/attic/netrc"),
        }
    }

    const OTHER: &str = "/nix/store/zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-nixos-system-iris";

    #[test]
    fn the_path_realised_for_the_agent_is_applied() {
        assert_eq!(applicable("atlas", PATH, Some(PATH)).unwrap(), PATH);
    }

    /// Another agent's realised path, a valid-shaped path never realised, and
    /// a restart that emptied the record are all refused, naming the agent
    /// and both paths.
    #[test]
    fn any_other_path_is_refused() {
        for (case, realised) in [
            ("another path was realised for the agent", Some(OTHER)),
            ("nothing was realised for the agent", None),
        ] {
            let err = applicable("atlas", PATH, realised).unwrap_err();
            let msg = format!("{err:#}");
            assert!(
                msg.contains("atlas")
                    && msg.contains(PATH)
                    && msg.contains(realised.unwrap_or("none")),
                "{case}: {msg}"
            );
        }
    }

    #[test]
    fn only_a_store_path_is_applied_even_when_it_was_recorded() {
        for bad in [
            "",
            "main",
            "/etc/passwd",
            "/nix/store/x",
            &format!("{PATH}/bin"),
        ] {
            let err = applicable("atlas", bad, Some(bad)).unwrap_err();
            assert!(format!("{err:#}").contains("not a store path"), "{bad}");
        }
    }

    /// Each agent keeps its own entry, and a newer download replaces the
    /// older one.
    #[test]
    fn a_record_is_per_agent_and_the_latest_wins() {
        let mut realised = Vec::new();
        remember(&mut realised, "atlas", PATH);
        remember(&mut realised, "iris", OTHER);
        assert_eq!(lookup(&realised, "atlas"), Some(PATH));
        assert_eq!(lookup(&realised, "iris"), Some(OTHER));
        remember(&mut realised, "atlas", OTHER);
        assert_eq!(lookup(&realised, "atlas"), Some(OTHER));
        assert_eq!(lookup(&realised, "nobody"), None);
        assert_eq!(realised.len(), 2);
    }

    fn option<'a>(args: &'a [String], name: &str) -> Option<&'a str> {
        args.windows(3)
            .find(|w| w[0] == "--option" && w[1] == name)
            .map(|w| w[2].as_str())
    }

    #[test]
    fn the_command_never_builds() {
        let args = realise_args(PATH, &cache());
        assert_eq!(args[..2], ["--realise", PATH]);
        let jobs = args.iter().position(|a| a == "--max-jobs").expect("flag");
        assert_eq!(args[jobs + 1], "0");
    }

    #[test]
    fn only_the_swarm_cache_and_its_key_count_and_signatures_are_required() {
        let args = realise_args(PATH, &cache());
        assert_eq!(option(&args, "substituters"), Some(URL));
        assert_eq!(option(&args, "trusted-public-keys"), Some("swarm:AAAA"));
        assert_eq!(option(&args, "require-sigs"), Some("true"));
        assert_eq!(
            option(&args, "netrc-file"),
            Some("/run/hyperhive-priv/attic/netrc")
        );
        // The settings that would add to the defaults are not used.
        assert!(!args.iter().any(|a| a.starts_with("extra-")), "{args:?}");
    }

    /// The uid the test runs as, which stands in for root.
    fn me() -> u32 {
        std::fs::metadata(scratch()).expect("stat").uid()
    }

    fn put(path: &Path, text: &str, mode: u32) {
        std::fs::write(path, text).expect("write");
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode)).expect("chmod");
    }

    /// A facts directory as the root unit leaves it.
    fn facts(key: &str) -> PathBuf {
        let dir = scratch();
        std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).expect("chmod");
        put(&dir.join("public-key"), &format!("{key}\n"), 0o600);
        put(
            &dir.join("netrc"),
            "machine cache.example password t\n",
            0o600,
        );
        dir
    }

    #[test]
    fn the_cache_is_the_hosts_address_and_the_root_files_key() {
        let dir = facts("swarm:AAAA");
        let cache = Cache::read(Some(URL), &dir, me()).expect("read");
        assert_eq!(cache.substituter, URL);
        assert_eq!(cache.public_key, "swarm:AAAA");
        assert_eq!(cache.netrc, dir.join("netrc"));
    }

    /// A `substituter` or `public-key` file beside the facts, as hive-c0re
    /// writes them, decides nothing: the address is the argument and the key is
    /// the root file's.
    #[test]
    fn the_address_is_never_read_from_a_file_and_the_key_only_from_the_root_files() {
        let dir = facts("swarm:ROOT");
        put(
            &dir.join("substituter"),
            "https://evil.example/cache\n",
            0o600,
        );
        let cache = Cache::read(Some(URL), &dir, me()).expect("read");
        assert_eq!(cache.substituter, URL);
        assert_eq!(cache.public_key, "swarm:ROOT");
        // The directory hive-c0re writes is not the one this reads.
        assert!(!ROOT_FACTS_DIR.starts_with("/run/hyperhive/"));
        assert_eq!(ROOT_FACTS_DIR, "/run/hyperhive-priv/attic");
        // And without an address from the host there is no cache, whatever the
        // directory holds.
        for none in [None, Some(""), Some("  ")] {
            assert!(Cache::read(none, &dir, me()).is_err(), "{none:?}");
        }
    }

    #[test]
    fn a_file_that_is_not_the_roots_or_is_open_to_others_is_refused() {
        let me = me();
        // Control: the facts as the unit leaves them are accepted.
        assert!(Cache::read(Some(URL), &facts("swarm:AAAA"), me).is_ok());
        // Another owner than the one trusted.
        assert!(Cache::read(Some(URL), &facts("swarm:AAAA"), me + 1).is_err());
        // The key or the netrc readable by others, or the directory open.
        for (name, mode) in [("public-key", 0o644), ("netrc", 0o644), ("netrc", 0o660)] {
            let dir = facts("swarm:AAAA");
            std::fs::set_permissions(dir.join(name), std::fs::Permissions::from_mode(mode))
                .expect("chmod");
            assert!(Cache::read(Some(URL), &dir, me).is_err(), "{name} {mode:o}");
        }
        let dir = facts("swarm:AAAA");
        std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).expect("chmod");
        assert!(Cache::read(Some(URL), &dir, me).is_err());
        // A symlink in place of the key.
        let dir = facts("swarm:AAAA");
        std::fs::remove_file(dir.join("public-key")).expect("rm");
        std::os::unix::fs::symlink(dir.join("netrc"), dir.join("public-key")).expect("link");
        assert!(Cache::read(Some(URL), &dir, me).is_err());
    }

    #[test]
    fn missing_files_and_unusable_values_are_refused() {
        let me = me();
        for url in ["http://cache.example/swarm", "https://cache.example/a b"] {
            assert!(
                Cache::read(Some(url), &facts("swarm:AAAA"), me).is_err(),
                "{url}"
            );
        }
        for key in ["nokeyname", "swarm: AAAA", ":AAAA", "swarm:"] {
            assert!(Cache::read(Some(URL), &facts(key), me).is_err(), "{key:?}");
        }
        let dir = facts("swarm:AAAA");
        std::fs::remove_file(dir.join("netrc")).expect("rm");
        assert!(Cache::read(Some(URL), &dir, me).is_err());
        assert!(Cache::read(Some(URL), &dir.join("nowhere"), me).is_err());
    }

    /// Missing facts reach the caller as `not available`, which a rebuild
    /// reports as a closure that cannot be pulled.
    #[tokio::test]
    async fn without_the_root_files_the_answer_is_not_available() {
        let nowhere = scratch().join("nowhere");
        let err = realise_with(Some(URL), &nowhere, 0, "atlas", PATH)
            .await
            .expect_err("no facts");
        assert!(format!("{err:#}").starts_with("not available: "), "{err:#}");
        let err = realise_with(None, &facts("swarm:AAAA"), me(), "atlas", PATH)
            .await
            .expect_err("no address");
        assert!(format!("{err:#}").starts_with("not available: "), "{err:#}");
    }

    /// The request carries an agent and a path and nothing that chooses a
    /// cache: fields beyond those are dropped when it is read.
    #[test]
    fn the_request_cannot_carry_an_address_or_a_key() {
        let line = r#"{"op":"realise_closure","agent":"atlas","store_path":"/nix/store/x",
            "substituter":"https://evil.example/c","public_key":"evil:AAAA","netrc":"/tmp/x"}"#;
        let hive_priv_sock::PrivRequest::RealiseClosure { agent, store_path } =
            serde_json::from_str(line).expect("reads")
        else {
            panic!("not a RealiseClosure");
        };
        assert_eq!(
            (agent.as_str(), store_path.as_str()),
            ("atlas", "/nix/store/x")
        );
    }

    #[test]
    fn a_signature_refusal_is_named_and_another_failure_is_not() {
        let refused =
            reason("error: cannot add path 'x' because it lacks a signature by a trusted key");
        assert!(refused.starts_with(SIGNATURE_REFUSED), "{refused}");
        let down =
            reason("warning: x\nerror: unable to download 'https://c/x.narinfo': Couldn't connect");
        assert!(!down.contains(SIGNATURE_REFUSED), "{down}");
        assert!(down.contains("Couldn't connect"), "{down}");
    }

    #[tokio::test]
    async fn bad_arguments_are_refused_before_anything_runs() {
        for (agent, path) in [
            ("Bad Agent", PATH),
            ("", PATH),
            ("atlas", "/nix/store/x"),
            ("atlas", "/etc/passwd"),
            ("atlas", &format!("{PATH}/bin/sh")),
        ] {
            let err = realise_closure(agent, path).await.expect_err("refused");
            assert!(!format!("{err:#}").contains("not available"), "{err:#}");
        }
    }
}