home.rs 80.6%

145 of 180 lines run, 24 of 64 functions · run #215 on main · run missed

//! `$FLOW_HOME` (default `~/.local/share/flow`): the pinned parts of each
//! isolation tier, rootfs images by hash, scratch space and the runner's
//! configuration, all described by `flow.toml`.
//!
//! ```toml
//! [microvm]
//! firecracker = { path = "bin/firecracker", blake3 = "…" }
//! kernel = { path = "kernel/vmlinux", blake3 = "…" }
//! init = { path = "bin/flow-init", blake3 = "…" }
//!
//! [runner]                       # only for `flow runner`
//! hub = "https://hub.example:7343"
//! token = "…"                    # a runner token from the hub's [flow] runner_tokens
//! ca = "/path/to/hub-ca.pem"     # when the hub's certificate is self-signed
//! name = "earth-1"
//! ```
//!
//! A runner signs its receipts with a composite key kept beside `flow.toml`
//! (`runner.key`, `runner.identity`), made on first start.

use anyhow::{bail, Context, Result};
use serde::Deserialize;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

#[derive(Clone, Debug, Deserialize)]
pub struct Pinned {
    pub path: PathBuf,
    pub blake3: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct MicrovmConfig {
    pub firecracker: Pinned,
    pub kernel: Pinned,
    pub init: Pinned,
}

/// Where `flow runner register` keeps the runner's hub and credential.
pub const RUNNER_TOML: &str = "runner.toml";

/// The hub's CA certificate as `flow runner register --ca` keeps it (#98).
pub const CA_PEM: &str = "hub-ca.pem";

#[derive(Clone, Debug, Deserialize)]
pub struct RunnerConfig {
    pub hub: String,
    pub token: String,
    pub ca: Option<PathBuf>,
    pub name: Option<String>,
    /// Jobs run at once (default 1).
    #[serde(default)]
    pub max_jobs: Option<usize>,
    /// Labels the runner offers, e.g. `gpu`, `big-disk`.
    #[serde(default)]
    pub labels: Vec<String>,
    /// Memory (MiB) and vCPUs the jobs running at once may use together.
    #[serde(default)]
    pub memory_mib: Option<u32>,
    #[serde(default)]
    pub cpus: Option<u32>,
    /// Sharing images with other runners (docs/jetstream.md §8.1): absent,
    /// the runner neither serves nor asks peers.
    #[serde(default)]
    pub swarm: Option<SwarmConfig>,
    /// The build cache's size limit in GiB (#28; default 64).
    #[serde(default)]
    pub cache_gb: Option<u64>,
    /// The share of its disk, in percent, a kept build cache may fill
    /// (oxygenflow#72; default 60): over it, the files the job did not use
    /// are pruned, oldest first, and a cache still over it is not kept.
    #[serde(default)]
    pub cache_prune_percent: Option<u8>,
    /// Let microVM jobs reach the hosts their sandbox allows, through this
    /// runner (oxygenflow#55); off unless set. The runner then offers the
    /// `egress` label, which such jobs need.
    #[serde(default)]
    pub egress: bool,
    /// Sign each job's receipt with a fresh key the hub certifies for that
    /// job, just before signing (keyless, oxygenhub#241), rather than with
    /// the runner's own key; off unless set. When the hub cannot certify,
    /// the runner's key signs, and the job's log says why.
    #[serde(default)]
    pub keyless: bool,
    /// Serve this runner's metrics at this address, e.g. `127.0.0.1:9464`
    /// (oxygenflow#35); off unless set.
    #[serde(default)]
    pub metrics: Option<String>,
    /// A bearer token a scrape of the metrics must send.
    #[serde(default)]
    pub metrics_token: Option<String>,
    /// `[runner.update]`: whether this runner takes updates (#85).
    #[serde(default)]
    pub update: Option<UpdateConfig>,
}

/// `[runner.update]` in flow.toml (docs/jetstream.md §8.2): the runner's
/// last word on updates, whatever the hub's policy says.
#[derive(Clone, Debug, Deserialize)]
pub struct UpdateConfig {
    /// `false` refuses every update, so a site can hold a machine back.
    #[serde(default = "yes")]
    pub allow: bool,
}

fn yes() -> bool {
    true
}

impl RunnerConfig {
    /// The labels offered to the hub: the configured ones, and `egress`
    /// when egress is on.
    pub fn offered_labels(&self) -> Vec<String> {
        let mut l = self.labels.clone();
        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {
            l.push(flow_core::EGRESS_LABEL.to_string());
        }
        l
    }
}

/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.
#[derive(Clone, Debug, Deserialize)]
pub struct SwarmConfig {
    /// Where peers reach this runner, e.g. `0.0.0.0:7401`.
    pub listen: String,
    /// The address peers are given (default: `listen`).
    #[serde(default)]
    pub advertise: Option<String>,
    /// Peers are only ever in the same site.
    #[serde(default)]
    pub site: String,
    /// The most this runner sends to peers, MiB a second (default 50).
    #[serde(default)]
    pub upload_mib_s: Option<u32>,
}

#[derive(Clone, Debug, Default, Deserialize)]
pub struct Config {
    pub microvm: Option<MicrovmConfig>,
    pub runner: Option<RunnerConfig>,
}

pub struct Home {
    pub root: PathBuf,
    pub config: Config,
    /// Files already verified, by path, with the (len, mtime) they had then.
    verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
}

impl Home {
    pub fn locate(explicit: Option<&Path>) -> Result<Home> {
        let root = match explicit {
            Some(p) => p.to_path_buf(),
            None => match std::env::var_os("FLOW_HOME") {
                Some(p) => PathBuf::from(p),
                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
            },
        };
        let file = root.join("flow.toml");
        let mut config: Config = if file.is_file() {
            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;
            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?
        } else {
            Config::default()
        };
        // `flow runner register` writes the runner's hub and credential to
        // runner.toml, which stands in for a [runner] section flow.toml lacks.
        let registered = root.join(RUNNER_TOML);
        if config.runner.is_none() && registered.is_file() {
            let text = std::fs::read_to_string(&registered).with_context(|| format!("read {}", registered.display()))?;
            let mut runner: RunnerConfig = toml::from_str(&text).with_context(|| format!("parse {}", registered.display()))?;
            // A relative CA there is the home's (#98): register keeps its
            // copy there, and the service may start in any directory.
            runner.ca = runner.ca.map(|ca| if ca.is_absolute() { ca } else { root.join(ca) });
            config.runner = Some(runner);
        }
        for d in ["cas", "work", "repos"] {
            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;
        }
        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
    }

    pub fn resolve(&self, p: &Path) -> PathBuf {
        if p.is_absolute() {
            p.to_path_buf()
        } else {
            self.root.join(p)
        }
    }

    /// Where a rootfs image with this hash lives.
    pub fn rootfs(&self, hash: &str) -> PathBuf {
        self.root.join("cas").join(format!("{hash}.ext4"))
    }

    /// An image's manifest (`cas/<hash>.json`): what `flow rootfs import`
    /// found out about it.
    pub fn image_manifest(&self, hash: &str) -> PathBuf {
        self.root.join("cas").join(format!("{hash}.json"))
    }

    /// The architecture an image is for: as its manifest records, else
    /// read from the image's own shell (spec §11.2) and recorded, so an
    /// image fetched or imported without one is read once and a runner
    /// declines what it cannot run instead of failing it. `None` when the
    /// image is not here or names no architecture.
    pub fn image_arch(&self, hash: &str) -> Option<String> {
        let path = self.image_manifest(hash);
        let mut v: serde_json::Value =
            std::fs::read_to_string(&path).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_else(|| serde_json::json!({}));
        if let Some(a) = v["arch"].as_str() {
            return Some(a.to_string());
        }
        let image = self.rootfs(hash);
        if v.get("arch_read_ms").is_some() || !image.is_file() {
            return None;
        }
        let arch = jet::rootfs::arch_of_file(&image);
        let o = v.as_object_mut()?;
        if let Some(a) = arch {
            o.insert("arch".into(), a.into());
        }
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
        o.insert("arch_read_ms".into(), now.into());
        let tmp = path.with_extension("json.part");
        if std::fs::write(&tmp, v.to_string()).is_ok() {
            let _ = std::fs::rename(&tmp, &path);
        }
        arch.map(String::from)
    }

    /// Every image with a known architecture (spec §11.2).
    pub fn image_archs(&self) -> std::collections::BTreeMap<String, String> {
        self.images().into_iter().filter_map(|h| self.image_arch(&h).map(|a| (h, a))).collect()
    }

    /// The rootfs images this home holds: the hashes of `cas/*.ext4`.
    pub fn images(&self) -> Vec<String> {
        let Ok(dir) = std::fs::read_dir(self.root.join("cas")) else {
            return Vec::new();
        };
        let mut out: Vec<String> = dir
            .filter_map(|e| {
                let name = e.ok()?.file_name().into_string().ok()?;
                let hash = name.strip_suffix(".ext4")?;
                (hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit())).then(|| hash.to_string())
            })
            .collect();
        out.sort();
        out
    }

    /// A pinned file, checked against its BLAKE3 before use (again only
    /// when its size or modification time changed since).
    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;
        let stamp = (meta.len(), meta.modified()?);
        if self.verified.lock().unwrap().get(path) == Some(&stamp) {
            return Ok(path.to_path_buf());
        }
        let mut got = hash_file(path)?;
        if got != blake3_hex {
            // The read may have come from bad cached pages rather than the
            // disk (oxygenflow#22: a RAM fault on Earth failed every job for
            // hours while the file was intact): drop the file's pages and
            // read it again from the disk before giving up.
            drop_cached(path);
            let again = hash_file(path)?;
            if again == blake3_hex {
                eprintln!(
                    "warning: {} hashed {got} from the page cache but {again} from the disk: the cached copy was bad (check this machine's memory)",
                    path.display()
                );
            }
            got = again;
        }
        if got != blake3_hex {
            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
        }
        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
        Ok(path.to_path_buf())
    }
}

/// Ask the kernel to forget a file's cached pages, so the next read comes
/// from the disk. Best effort: dirty pages and other platforms stay.
fn drop_cached(path: &Path) {
    #[cfg(target_os = "linux")]
    if let Ok(f) = std::fs::File::open(path) {
        use std::os::fd::AsRawFd;
        // SAFETY: a valid descriptor; the advice only affects caching.
        unsafe {
            libc::posix_fadvise(f.as_raw_fd(), 0, 0, libc::POSIX_FADV_DONTNEED);
        }
    }
    #[cfg(not(target_os = "linux"))]
    let _ = path;
}

pub fn hash_file(path: &Path) -> Result<String> {
    let mut h = blake3::Hasher::new();
    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;
    Ok(h.finalize().to_hex().to_string())
}

/// The key a runner signs receipts with (Oxygen spec §14.6).
pub struct RunnerKey {
    pub identity: oxy_core::identity::Identity,
    pub secret: oxy_core::identity::SecretKey,
}

impl Home {
    /// The runner's key, made on first use as `runner.key` (owner-only)
    /// and `runner.identity`, named `name`.
    pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {
        use oxy_core::identity::{Identity, SecretKey};
        let key_path = self.root.join("runner.key");
        let id_path = self.root.join("runner.identity");
        if key_path.exists() {
            let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;
            let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;
            if identity.hash() != key {
                bail!("runner.identity does not belong to runner.key");
            }
            return Ok(RunnerKey { identity, secret });
        }
        let secret = SecretKey::generate();
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
        let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);
        std::fs::write(&id_path, identity.encode())?;
        let tmp = self.root.join("runner.key.tmp");
        {
            use std::io::Write;
            let mut f = crate::portable::private(std::fs::OpenOptions::new().write(true).create(true).truncate(true)).open(&tmp)?;
            f.write_all(&secret.encode(&identity.hash()))?;
            f.sync_all()?;
        }
        std::fs::rename(&tmp, &key_path)?;
        Ok(RunnerKey { identity, secret })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn the_runner_key_is_made_once_and_kept_private() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().unwrap();
        let home = Home::locate(Some(dir.path())).unwrap();
        let a = home.runner_key("earth-1").unwrap();
        assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");
        let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();
        assert_eq!(mode & 0o777, 0o600);
        let b = home.runner_key("renamed").unwrap();
        assert_eq!(a.identity, b.identity, "the same key after a restart");
        let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));
        oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();
    }

    /// #98: runner.toml's relative CA is the home's, wherever flow starts.
    #[test]
    fn a_registered_runners_relative_ca_is_in_its_home() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"hub-ca.pem\"\n").unwrap();
        let home = Home::locate(Some(dir.path())).unwrap();
        assert_eq!(home.config.runner.unwrap().ca, Some(dir.path().join("hub-ca.pem")));
        std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"/etc/ssl/hub.pem\"\n").unwrap();
        let home = Home::locate(Some(dir.path())).unwrap();
        assert_eq!(home.config.runner.unwrap().ca, Some(PathBuf::from("/etc/ssl/hub.pem")));
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! `$FLOW_HOME` (default `~/.local/share/flow`): the pinned parts of each
2		//! isolation tier, rootfs images by hash, scratch space and the runner's
3		//! configuration, all described by `flow.toml`.
4		//!
5		//! ```toml
6		//! [microvm]
7		//! firecracker = { path = "bin/firecracker", blake3 = "…" }
8		//! kernel = { path = "kernel/vmlinux", blake3 = "…" }
9		//! init = { path = "bin/flow-init", blake3 = "…" }
10		//!
11		//! [runner] # only for `flow runner`
12		//! hub = "https://hub.example:7343"
13		//! token = "…" # a runner token from the hub's [flow] runner_tokens
14		//! ca = "/path/to/hub-ca.pem" # when the hub's certificate is self-signed
15		//! name = "earth-1"
16		//! ```
17		//!
18		//! A runner signs its receipts with a composite key kept beside `flow.toml`
19		//! (`runner.key`, `runner.identity`), made on first start.
20
21		use anyhow::{bail, Context, Result};
22		use serde::Deserialize;
23		use std::collections::HashMap;
24		use std::path::{Path, PathBuf};
25		use std::sync::Mutex;
26
27		#[derive(Clone, Debug, Deserialize)]
28		pub struct Pinned {
29		pub path: PathBuf,
30		pub blake3: String,
31		}
32
33		#[derive(Clone, Debug, Deserialize)]
34		pub struct MicrovmConfig {
35		pub firecracker: Pinned,
36		pub kernel: Pinned,
37		pub init: Pinned,
38		}
39
40		/// Where `flow runner register` keeps the runner's hub and credential.
41		pub const RUNNER_TOML: &str = "runner.toml";
42
43		/// The hub's CA certificate as `flow runner register --ca` keeps it (#98).
44		pub const CA_PEM: &str = "hub-ca.pem";
45
46		#[derive(Clone, Debug, Deserialize)]
47		pub struct RunnerConfig {
48		pub hub: String,
49		pub token: String,
50		pub ca: Option<PathBuf>,
51		pub name: Option<String>,
52		/// Jobs run at once (default 1).
53		#[serde(default)]
54		pub max_jobs: Option<usize>,
55		/// Labels the runner offers, e.g. `gpu`, `big-disk`.
56		#[serde(default)]
57		pub labels: Vec<String>,
58		/// Memory (MiB) and vCPUs the jobs running at once may use together.
59		#[serde(default)]
60		pub memory_mib: Option<u32>,
61		#[serde(default)]
62		pub cpus: Option<u32>,
63		/// Sharing images with other runners (docs/jetstream.md §8.1): absent,
64		/// the runner neither serves nor asks peers.
65		#[serde(default)]
66		pub swarm: Option<SwarmConfig>,
67		/// The build cache's size limit in GiB (#28; default 64).
68		#[serde(default)]
69		pub cache_gb: Option<u64>,
70		/// The share of its disk, in percent, a kept build cache may fill
71		/// (oxygenflow#72; default 60): over it, the files the job did not use
72		/// are pruned, oldest first, and a cache still over it is not kept.
73		#[serde(default)]
74		pub cache_prune_percent: Option<u8>,
75		/// Let microVM jobs reach the hosts their sandbox allows, through this
76		/// runner (oxygenflow#55); off unless set. The runner then offers the
77		/// `egress` label, which such jobs need.
78		#[serde(default)]
79		pub egress: bool,
80		/// Sign each job's receipt with a fresh key the hub certifies for that
81		/// job, just before signing (keyless, oxygenhub#241), rather than with
82		/// the runner's own key; off unless set. When the hub cannot certify,
83		/// the runner's key signs, and the job's log says why.
84		#[serde(default)]
85		pub keyless: bool,
86		/// Serve this runner's metrics at this address, e.g. `127.0.0.1:9464`
87		/// (oxygenflow#35); off unless set.
88		#[serde(default)]
89		pub metrics: Option<String>,
90		/// A bearer token a scrape of the metrics must send.
91		#[serde(default)]
92		pub metrics_token: Option<String>,
93		/// `[runner.update]`: whether this runner takes updates (#85).
94		#[serde(default)]
95		pub update: Option<UpdateConfig>,
96		}
97
98		/// `[runner.update]` in flow.toml (docs/jetstream.md §8.2): the runner's
99		/// last word on updates, whatever the hub's policy says.
100		#[derive(Clone, Debug, Deserialize)]
101		pub struct UpdateConfig {
102		/// `false` refuses every update, so a site can hold a machine back.
103		#[serde(default = "yes")]
104		pub allow: bool,
105		}
106
107	0×	fn yes() -> bool {
108	0×	true
109	0×	}
110
111		impl RunnerConfig {
112		/// The labels offered to the hub: the configured ones, and `egress`
113		/// when egress is on.
114	0×	pub fn offered_labels(&self) -> Vec<String> {
115	0×	let mut l = self.labels.clone();
116	0×	if self.egress && !l.iter().any(\|x\| x == flow_core::EGRESS_LABEL) {
117	0×	l.push(flow_core::EGRESS_LABEL.to_string());
118	0×	}
119	0×	l
120	0×	}
121		}
122
123		/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.
124		#[derive(Clone, Debug, Deserialize)]
125		pub struct SwarmConfig {
126		/// Where peers reach this runner, e.g. `0.0.0.0:7401`.
127		pub listen: String,
128		/// The address peers are given (default: `listen`).
129		#[serde(default)]
130		pub advertise: Option<String>,
131		/// Peers are only ever in the same site.
132		#[serde(default)]
133		pub site: String,
134		/// The most this runner sends to peers, MiB a second (default 50).
135		#[serde(default)]
136		pub upload_mib_s: Option<u32>,
137		}
138
139		#[derive(Clone, Debug, Default, Deserialize)]
140		pub struct Config {
141		pub microvm: Option<MicrovmConfig>,
142		pub runner: Option<RunnerConfig>,
143		}
144
145		pub struct Home {
146		pub root: PathBuf,
147		pub config: Config,
148		/// Files already verified, by path, with the (len, mtime) they had then.
149		verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
150		}
151
152		impl Home {
153	23×	pub fn locate(explicit: Option<&Path>) -> Result<Home> {
154	23×	let root = match explicit {
155	23×	Some(p) => p.to_path_buf(),
156	0×	None => match std::env::var_os("FLOW_HOME") {
157	0×	Some(p) => PathBuf::from(p),
158	0×	None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
159		},
160		};
161	23×	let file = root.join("flow.toml");
162	23×	let mut config: Config = if file.is_file() {
163	0×	let text = std::fs::read_to_string(&file).with_context(\|\| format!("read {}", file.display()))?;
164	0×	toml::from_str(&text).with_context(\|\| format!("parse {}", file.display()))?
165		} else {
166	23×	Config::default()
167		};
168		// `flow runner register` writes the runner's hub and credential to
169		// runner.toml, which stands in for a [runner] section flow.toml lacks.
170	23×	let registered = root.join(RUNNER_TOML);
171	23×	if config.runner.is_none() && registered.is_file() {
172	14×	let text = std::fs::read_to_string(&registered).with_context(\|\| format!("read {}", registered.display()))?;
173	14×	let mut runner: RunnerConfig = toml::from_str(&text).with_context(\|\| format!("parse {}", registered.display()))?;
174		// A relative CA there is the home's (#98): register keeps its
175		// copy there, and the service may start in any directory.
176	14×	runner.ca = runner.ca.map(\|ca\| if ca.is_absolute() { ca } else { root.join(ca) });
177	14×	config.runner = Some(runner);
178	9×	}
179	69×	for d in ["cas", "work", "repos"] {
180	69×	std::fs::create_dir_all(root.join(d)).with_context(\|\| format!("create {}", root.join(d).display()))?;
181		}
182	23×	Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
183	23×	}
184
185	0×	pub fn resolve(&self, p: &Path) -> PathBuf {
186	0×	if p.is_absolute() {
187	0×	p.to_path_buf()
188		} else {
189	0×	self.root.join(p)
190		}
191	0×	}
192
193		/// Where a rootfs image with this hash lives.
194	40×	pub fn rootfs(&self, hash: &str) -> PathBuf {
195	40×	self.root.join("cas").join(format!("{hash}.ext4"))
196	40×	}
197
198		/// An image's manifest (`cas/<hash>.json`): what `flow rootfs import`
199		/// found out about it.
200	16×	pub fn image_manifest(&self, hash: &str) -> PathBuf {
201	16×	self.root.join("cas").join(format!("{hash}.json"))
202	16×	}
203
204		/// The architecture an image is for: as its manifest records, else
205		/// read from the image's own shell (spec §11.2) and recorded, so an
206		/// image fetched or imported without one is read once and a runner
207		/// declines what it cannot run instead of failing it. `None` when the
208		/// image is not here or names no architecture.
209	10×	pub fn image_arch(&self, hash: &str) -> Option<String> {
210	10×	let path = self.image_manifest(hash);
211	10×	let mut v: serde_json::Value =
212	10×	std::fs::read_to_string(&path).ok().and_then(\|t\| serde_json::from_str(&t).ok()).unwrap_or_else(\|\| serde_json::json!({}));
213	10×	if let Some(a) = v["arch"].as_str() {
214	3×	return Some(a.to_string());
215	7×	}
216	7×	let image = self.rootfs(hash);
217	7×	if v.get("arch_read_ms").is_some() \|\| !image.is_file() {
218	1×	return None;
219	6×	}
220	6×	let arch = jet::rootfs::arch_of_file(&image);
221	6×	let o = v.as_object_mut()?;
222	6×	if let Some(a) = arch {
223	1×	o.insert("arch".into(), a.into());
224	5×	}
225	6×	let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0);
226	6×	o.insert("arch_read_ms".into(), now.into());
227	6×	let tmp = path.with_extension("json.part");
228	6×	if std::fs::write(&tmp, v.to_string()).is_ok() {
229	6×	let _ = std::fs::rename(&tmp, &path);
230	6×	}
231	6×	arch.map(String::from)
232	10×	}
233
234		/// Every image with a known architecture (spec §11.2).
235	1×	pub fn image_archs(&self) -> std::collections::BTreeMap<String, String> {
236	1×	self.images().into_iter().filter_map(\|h\| self.image_arch(&h).map(\|a\| (h, a))).collect()
237	1×	}
238
239		/// The rootfs images this home holds: the hashes of `cas/*.ext4`.
240	13×	pub fn images(&self) -> Vec<String> {
241	13×	let Ok(dir) = std::fs::read_dir(self.root.join("cas")) else {
242	0×	return Vec::new();
243		};
244	13×	let mut out: Vec<String> = dir
245	13×	.filter_map(\|e\| {
246	6×	let name = e.ok()?.file_name().into_string().ok()?;
247	6×	let hash = name.strip_suffix(".ext4")?;
248	320×	(hash.len() == 64 && hash.chars().all(\|c\| c.is_ascii_hexdigit())).then(\|\| hash.to_string())
249	6×	})
250	13×	.collect();
251	13×	out.sort();
252	13×	out
253	13×	}
254
255		/// A pinned file, checked against its BLAKE3 before use (again only
256		/// when its size or modification time changed since).
257	4×	pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
258	4×	let meta = std::fs::metadata(path).with_context(\|\| format!("{} is missing", path.display()))?;
259	4×	let stamp = (meta.len(), meta.modified()?);
260	4×	if self.verified.lock().unwrap().get(path) == Some(&stamp) {
261	1×	return Ok(path.to_path_buf());
262	3×	}
263	3×	let mut got = hash_file(path)?;
264	3×	if got != blake3_hex {
265		// The read may have come from bad cached pages rather than the
266		// disk (oxygenflow#22: a RAM fault on Earth failed every job for
267		// hours while the file was intact): drop the file's pages and
268		// read it again from the disk before giving up.
269	1×	drop_cached(path);
270	1×	let again = hash_file(path)?;
271	1×	if again == blake3_hex {
272	0×	eprintln!(
273	0×	"warning: {} hashed {got} from the page cache but {again} from the disk: the cached copy was bad (check this machine's memory)",
274	0×	path.display()
275	0×	);
276	1×	}
277	1×	got = again;
278	2×	}
279	3×	if got != blake3_hex {
280	1×	bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
281	2×	}
282	2×	self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
283	2×	Ok(path.to_path_buf())
284	4×	}
285		}
286
287		/// Ask the kernel to forget a file's cached pages, so the next read comes
288		/// from the disk. Best effort: dirty pages and other platforms stay.
289	1×	fn drop_cached(path: &Path) {
290		#[cfg(target_os = "linux")]
291	1×	if let Ok(f) = std::fs::File::open(path) {
292		use std::os::fd::AsRawFd;
293		// SAFETY: a valid descriptor; the advice only affects caching.
294	1×	unsafe {
295	1×	libc::posix_fadvise(f.as_raw_fd(), 0, 0, libc::POSIX_FADV_DONTNEED);
296	1×	}
297	0×	}
298		#[cfg(not(target_os = "linux"))]
299		let _ = path;
300	1×	}
301
302	21×	pub fn hash_file(path: &Path) -> Result<String> {
303	21×	let mut h = blake3::Hasher::new();
304	21×	h.update_reader(std::fs::File::open(path).with_context(\|\| format!("open {}", path.display()))?)?;
305	21×	Ok(h.finalize().to_hex().to_string())
306	21×	}
307
308		/// The key a runner signs receipts with (Oxygen spec §14.6).
309		pub struct RunnerKey {
310		pub identity: oxy_core::identity::Identity,
311		pub secret: oxy_core::identity::SecretKey,
312		}
313
314		impl Home {
315		/// The runner's key, made on first use as `runner.key` (owner-only)
316		/// and `runner.identity`, named `name`.
317	4×	pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {
318		use oxy_core::identity::{Identity, SecretKey};
319	4×	let key_path = self.root.join("runner.key");
320	4×	let id_path = self.root.join("runner.identity");
321	4×	if key_path.exists() {
322	2×	let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;
323	2×	let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;
324	2×	if identity.hash() != key {
325	0×	bail!("runner.identity does not belong to runner.key");
326	2×	}
327	2×	return Ok(RunnerKey { identity, secret });
328	2×	}
329	2×	let secret = SecretKey::generate();
330	2×	let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0);
331	2×	let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);
332	2×	std::fs::write(&id_path, identity.encode())?;
333	2×	let tmp = self.root.join("runner.key.tmp");
334		{
335		use std::io::Write;
336	2×	let mut f = crate::portable::private(std::fs::OpenOptions::new().write(true).create(true).truncate(true)).open(&tmp)?;
337	2×	f.write_all(&secret.encode(&identity.hash()))?;
338	2×	f.sync_all()?;
339		}
340	2×	std::fs::rename(&tmp, &key_path)?;
341	2×	Ok(RunnerKey { identity, secret })
342	4×	}
343		}
344
345		#[cfg(test)]
346		mod tests {
347		use super::*;
348
349		#[test]
350	1×	fn the_runner_key_is_made_once_and_kept_private() {
351		use std::os::unix::fs::PermissionsExt;
352	1×	let dir = tempfile::tempdir().unwrap();
353	1×	let home = Home::locate(Some(dir.path())).unwrap();
354	1×	let a = home.runner_key("earth-1").unwrap();
355	1×	assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");
356	1×	let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();
357	1×	assert_eq!(mode & 0o777, 0o600);
358	1×	let b = home.runner_key("renamed").unwrap();
359	1×	assert_eq!(a.identity, b.identity, "the same key after a restart");
360	1×	let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));
361	1×	oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();
362	1×	}
363
364		/// #98: runner.toml's relative CA is the home's, wherever flow starts.
365		#[test]
366	1×	fn a_registered_runners_relative_ca_is_in_its_home() {
367	1×	let dir = tempfile::tempdir().unwrap();
368	1×	std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"hub-ca.pem\"\n").unwrap();
369	1×	let home = Home::locate(Some(dir.path())).unwrap();
370	1×	assert_eq!(home.config.runner.unwrap().ca, Some(dir.path().join("hub-ca.pem")));
371	1×	std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"/etc/ssl/hub.pem\"\n").unwrap();
372	1×	let home = Home::locate(Some(dir.path())).unwrap();
373	1×	assert_eq!(home.config.runner.unwrap().ca, Some(PathBuf::from("/etc/ssl/hub.pem")));
374	1×	}
375		}

Coverage / crates/flow/src/home.rs 80.6%

Keyboard