OXYGENHUB
# 47c1d64dec8a oxygenflow / crates / flow / src / home.rs
6.9 KiB · Rust · last changed Runners sign their receipts 1 month ago source blame history raw
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 
#[derive(Clone, Debug, Deserialize)]
41 
pub struct RunnerConfig {
42 
    pub hub: String,
43 
    pub token: String,
44 
    pub ca: Option<PathBuf>,
45 
    pub name: Option<String>,
46 
    /// Jobs run at once (default 1).
47 
    #[serde(default)]
48 
    pub max_jobs: Option<usize>,
49 
    /// Memory (MiB) and vCPUs the jobs running at once may use together.
50 
    #[serde(default)]
51 
    pub memory_mib: Option<u32>,
52 
    #[serde(default)]
53 
    pub cpus: Option<u32>,
54 
}
55
56 
#[derive(Clone, Debug, Default, Deserialize)]
57 
pub struct Config {
58 
    pub microvm: Option<MicrovmConfig>,
59 
    pub runner: Option<RunnerConfig>,
60 
}
61
62 
pub struct Home {
63 
    pub root: PathBuf,
64 
    pub config: Config,
65 
    /// Files already verified, by path, with the (len, mtime) they had then.
66 
    verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
67 
}
68
69 
impl Home {
70 
    pub fn locate(explicit: Option<&Path>) -> Result<Home> {
71 
        let root = match explicit {
72 
            Some(p) => p.to_path_buf(),
73 
            None => match std::env::var_os("FLOW_HOME") {
74 
                Some(p) => PathBuf::from(p),
75 
                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
76 
            },
77 
        };
78 
        let file = root.join("flow.toml");
79 
        let config = if file.is_file() {
80 
            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;
81 
            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?
82 
        } else {
83 
            Config::default()
84 
        };
85 
        for d in ["cas", "work", "repos"] {
86 
            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;
87 
        }
88 
        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
89 
    }
90
91 
    pub fn resolve(&self, p: &Path) -> PathBuf {
92 
        if p.is_absolute() {
93 
            p.to_path_buf()
94 
        } else {
95 
            self.root.join(p)
96 
        }
97 
    }
98
99 
    /// Where a rootfs image with this hash lives.
100 
    pub fn rootfs(&self, hash: &str) -> PathBuf {
101 
        self.root.join("cas").join(format!("{hash}.ext4"))
102 
    }
103
104 
    /// A pinned file, checked against its BLAKE3 before use (again only
105 
    /// when its size or modification time changed since).
106 
    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
107 
        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;
108 
        let stamp = (meta.len(), meta.modified()?);
109 
        if self.verified.lock().unwrap().get(path) == Some(&stamp) {
110 
            return Ok(path.to_path_buf());
111 
        }
112 
        let got = hash_file(path)?;
113 
        if got != blake3_hex {
114 
            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
115 
        }
116 
        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
117 
        Ok(path.to_path_buf())
118 
    }
119 
}
120
121 
pub fn hash_file(path: &Path) -> Result<String> {
122 
    let mut h = blake3::Hasher::new();
123 
    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;
124 
    Ok(h.finalize().to_hex().to_string())
125 
}
126
127 
/// The key a runner signs receipts with (Oxygen spec §14.6).
128 
pub struct RunnerKey {
129 
    pub identity: oxy_core::identity::Identity,
130 
    pub secret: oxy_core::identity::SecretKey,
131 
}
132
133 
impl Home {
134 
    /// The runner's key, made on first use as `runner.key` (owner-only)
135 
    /// and `runner.identity`, named `name`.
136 
    pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {
137 
        use oxy_core::identity::{Identity, SecretKey};
138 
        let key_path = self.root.join("runner.key");
139 
        let id_path = self.root.join("runner.identity");
140 
        if key_path.exists() {
141 
            let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;
142 
            let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;
143 
            if identity.hash() != key {
144 
                bail!("runner.identity does not belong to runner.key");
145 
            }
146 
            return Ok(RunnerKey { identity, secret });
147 
        }
148 
        let secret = SecretKey::generate();
149 
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
150 
        let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);
151 
        std::fs::write(&id_path, identity.encode())?;
152 
        let tmp = self.root.join("runner.key.tmp");
153 
        {
154 
            use std::io::Write;
155 
            use std::os::unix::fs::OpenOptionsExt;
156 
            let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;
157 
            f.write_all(&secret.encode(&identity.hash()))?;
158 
            f.sync_all()?;
159 
        }
160 
        std::fs::rename(&tmp, &key_path)?;
161 
        Ok(RunnerKey { identity, secret })
162 
    }
163 
}
164
165 
#[cfg(test)]
166 
mod tests {
167 
    use super::*;
168
169 
    #[test]
170 
    fn the_runner_key_is_made_once_and_kept_private() {
171 
        use std::os::unix::fs::PermissionsExt;
172 
        let dir = tempfile::tempdir().unwrap();
173 
        let home = Home::locate(Some(dir.path())).unwrap();
174 
        let a = home.runner_key("earth-1").unwrap();
175 
        assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");
176 
        let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();
177 
        assert_eq!(mode & 0o777, 0o600);
178 
        let b = home.runner_key("renamed").unwrap();
179 
        assert_eq!(a.identity, b.identity, "the same key after a restart");
180 
        let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));
181 
        oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();
182 
    }
183 
}