OXYGENHUB
# 9ff1fb9f2865 oxygenflow / crates / flow / src / home.rs
11.0 KiB · Rust · last changed Prune build caches before they are kept 16 days 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
/// Where `flow runner register` keeps the runner's hub and credential.
41
pub const RUNNER_TOML: &str = "runner.toml";
42
43
#[derive(Clone, Debug, Deserialize)]
44
pub struct RunnerConfig {
45
    pub hub: String,
46
    pub token: String,
47
    pub ca: Option<PathBuf>,
48
    pub name: Option<String>,
49
    /// Jobs run at once (default 1).
50
    #[serde(default)]
51
    pub max_jobs: Option<usize>,
52
    /// Labels the runner offers, e.g. `gpu`, `big-disk`.
53
    #[serde(default)]
54
    pub labels: Vec<String>,
55
    /// Memory (MiB) and vCPUs the jobs running at once may use together.
56
    #[serde(default)]
57
    pub memory_mib: Option<u32>,
58
    #[serde(default)]
59
    pub cpus: Option<u32>,
60
    /// Sharing images with other runners (docs/jetstream.md §8.1): absent,
61
    /// the runner neither serves nor asks peers.
62
    #[serde(default)]
63
    pub swarm: Option<SwarmConfig>,
64
    /// The build cache's size limit in GiB (#28; default 64).
65
    #[serde(default)]
66
    pub cache_gb: Option<u64>,
67
    /// The share of its disk, in percent, a kept build cache may fill
68
    /// (oxygenflow#72; default 60): over it, the files the job did not use
69
    /// are pruned, oldest first, and a cache still over it is not kept.
70
    #[serde(default)]
71
    pub cache_prune_percent: Option<u8>,
72
    /// Let microVM jobs reach the hosts their sandbox allows, through this
73
    /// runner (oxygenflow#55); off unless set. The runner then offers the
74
    /// `egress` label, which such jobs need.
75
    #[serde(default)]
76
    pub egress: bool,
77
    /// Sign each job's receipt with a fresh key the hub certifies for that
78
    /// job, just before signing (keyless, oxygenhub#241), rather than with
79
    /// the runner's own key; off unless set. When the hub cannot certify,
80
    /// the runner's key signs, and the job's log says why.
81
    #[serde(default)]
82
    pub keyless: bool,
83
}
84
85
impl RunnerConfig {
86
    /// The labels offered to the hub: the configured ones, and `egress`
87
    /// when egress is on.
88
    pub fn offered_labels(&self) -> Vec<String> {
89
        let mut l = self.labels.clone();
90
        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {
91
            l.push(flow_core::EGRESS_LABEL.to_string());
92
        }
93
        l
94
    }
95
}
96
97
/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.
98
#[derive(Clone, Debug, Deserialize)]
99
pub struct SwarmConfig {
100
    /// Where peers reach this runner, e.g. `0.0.0.0:7401`.
101
    pub listen: String,
102
    /// The address peers are given (default: `listen`).
103
    #[serde(default)]
104
    pub advertise: Option<String>,
105
    /// Peers are only ever in the same site.
106
    #[serde(default)]
107
    pub site: String,
108
    /// The most this runner sends to peers, MiB a second (default 50).
109
    #[serde(default)]
110
    pub upload_mib_s: Option<u32>,
111
}
112
113
#[derive(Clone, Debug, Default, Deserialize)]
114
pub struct Config {
115
    pub microvm: Option<MicrovmConfig>,
116
    pub runner: Option<RunnerConfig>,
117
}
118
119
pub struct Home {
120
    pub root: PathBuf,
121
    pub config: Config,
122
    /// Files already verified, by path, with the (len, mtime) they had then.
123
    verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
124
}
125
126
impl Home {
127
    pub fn locate(explicit: Option<&Path>) -> Result<Home> {
128
        let root = match explicit {
129
            Some(p) => p.to_path_buf(),
130
            None => match std::env::var_os("FLOW_HOME") {
131
                Some(p) => PathBuf::from(p),
132
                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
133
            },
134
        };
135
        let file = root.join("flow.toml");
136
        let mut config: Config = if file.is_file() {
137
            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;
138
            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?
139
        } else {
140
            Config::default()
141
        };
142
        // `flow runner register` writes the runner's hub and credential to
143
        // runner.toml, which stands in for a [runner] section flow.toml lacks.
144
        let registered = root.join(RUNNER_TOML);
145
        if config.runner.is_none() && registered.is_file() {
146
            let text = std::fs::read_to_string(&registered).with_context(|| format!("read {}", registered.display()))?;
147
            config.runner = Some(toml::from_str(&text).with_context(|| format!("parse {}", registered.display()))?);
148
        }
149
        for d in ["cas", "work", "repos"] {
150
            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;
151
        }
152
        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
153
    }
154
155
    pub fn resolve(&self, p: &Path) -> PathBuf {
156
        if p.is_absolute() {
157
            p.to_path_buf()
158
        } else {
159
            self.root.join(p)
160
        }
161
    }
162
163
    /// Where a rootfs image with this hash lives.
164
    pub fn rootfs(&self, hash: &str) -> PathBuf {
165
        self.root.join("cas").join(format!("{hash}.ext4"))
166
    }
167
168
    /// An image's manifest (`cas/<hash>.json`): what `flow rootfs import`
169
    /// found out about it.
170
    pub fn image_manifest(&self, hash: &str) -> PathBuf {
171
        self.root.join("cas").join(format!("{hash}.json"))
172
    }
173
174
    /// The architecture an image was imported for, when its manifest says.
175
    pub fn image_arch(&self, hash: &str) -> Option<String> {
176
        let text = std::fs::read_to_string(self.image_manifest(hash)).ok()?;
177
        let v: serde_json::Value = serde_json::from_str(&text).ok()?;
178
        v["arch"].as_str().map(String::from)
179
    }
180
181
    /// Every image with a known architecture (spec §11.2).
182
    pub fn image_archs(&self) -> std::collections::BTreeMap<String, String> {
183
        self.images().into_iter().filter_map(|h| self.image_arch(&h).map(|a| (h, a))).collect()
184
    }
185
186
    /// The rootfs images this home holds: the hashes of `cas/*.ext4`.
187
    pub fn images(&self) -> Vec<String> {
188
        let Ok(dir) = std::fs::read_dir(self.root.join("cas")) else {
189
            return Vec::new();
190
        };
191
        let mut out: Vec<String> = dir
192
            .filter_map(|e| {
193
                let name = e.ok()?.file_name().into_string().ok()?;
194
                let hash = name.strip_suffix(".ext4")?;
195
                (hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit())).then(|| hash.to_string())
196
            })
197
            .collect();
198
        out.sort();
199
        out
200
    }
201
202
    /// A pinned file, checked against its BLAKE3 before use (again only
203
    /// when its size or modification time changed since).
204
    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
205
        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;
206
        let stamp = (meta.len(), meta.modified()?);
207
        if self.verified.lock().unwrap().get(path) == Some(&stamp) {
208
            return Ok(path.to_path_buf());
209
        }
210
        let got = hash_file(path)?;
211
        if got != blake3_hex {
212
            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
213
        }
214
        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
215
        Ok(path.to_path_buf())
216
    }
217
}
218
219
pub fn hash_file(path: &Path) -> Result<String> {
220
    let mut h = blake3::Hasher::new();
221
    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;
222
    Ok(h.finalize().to_hex().to_string())
223
}
224
225
/// The key a runner signs receipts with (Oxygen spec §14.6).
226
pub struct RunnerKey {
227
    pub identity: oxy_core::identity::Identity,
228
    pub secret: oxy_core::identity::SecretKey,
229
}
230
231
impl Home {
232
    /// The runner's key, made on first use as `runner.key` (owner-only)
233
    /// and `runner.identity`, named `name`.
234
    pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {
235
        use oxy_core::identity::{Identity, SecretKey};
236
        let key_path = self.root.join("runner.key");
237
        let id_path = self.root.join("runner.identity");
238
        if key_path.exists() {
239
            let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;
240
            let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;
241
            if identity.hash() != key {
242
                bail!("runner.identity does not belong to runner.key");
243
            }
244
            return Ok(RunnerKey { identity, secret });
245
        }
246
        let secret = SecretKey::generate();
247
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
248
        let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);
249
        std::fs::write(&id_path, identity.encode())?;
250
        let tmp = self.root.join("runner.key.tmp");
251
        {
252
            use std::io::Write;
253
            use std::os::unix::fs::OpenOptionsExt;
254
            let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;
255
            f.write_all(&secret.encode(&identity.hash()))?;
256
            f.sync_all()?;
257
        }
258
        std::fs::rename(&tmp, &key_path)?;
259
        Ok(RunnerKey { identity, secret })
260
    }
261
}
262
263
#[cfg(test)]
264
mod tests {
265
    use super::*;
266
267
    #[test]
268
    fn the_runner_key_is_made_once_and_kept_private() {
269
        use std::os::unix::fs::PermissionsExt;
270
        let dir = tempfile::tempdir().unwrap();
271
        let home = Home::locate(Some(dir.path())).unwrap();
272
        let a = home.runner_key("earth-1").unwrap();
273
        assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");
274
        let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();
275
        assert_eq!(mode & 0o777, 0o600);
276
        let b = home.runner_key("renamed").unwrap();
277
        assert_eq!(a.identity, b.identity, "the same key after a restart");
278
        let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));
279
        oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();
280
    }
281
}