OXYGENHUB
# 0d663e91c598 oxygenflow / crates / flow / src / home.rs
12.9 KiB · Rust · last changed flow runner register keeps the CA in the Flow home by absolute path, and a relative ca in runner.toml is read against the home, so a service finds it from any directory (closes #98) 9 days ago source blame history raw
e0918b88 Andy McCall 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
//! ```
47c1d64d Andy McCall ↶ 17
//!
18
//! A runner signs its receipts with a composite key kept beside `flow.toml`
19
//! (`runner.key`, `runner.identity`), made on first start.
e0918b88 Andy McCall 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
f98a8f78 Andy McCall ↶ 40
/// Where `flow runner register` keeps the runner's hub and credential.
41
pub const RUNNER_TOML: &str = "runner.toml";
42
0d663e91 Andy McCall ↶ 43
/// The hub's CA certificate as `flow runner register --ca` keeps it (#98).
44
pub const CA_PEM: &str = "hub-ca.pem";
45
e0918b88 Andy McCall 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>,
bb643c25 Andy McCall ↶ 52
    /// Jobs run at once (default 1).
53
    #[serde(default)]
54
    pub max_jobs: Option<usize>,
f98a8f78 Andy McCall ↶ 55
    /// Labels the runner offers, e.g. `gpu`, `big-disk`.
56
    #[serde(default)]
57
    pub labels: Vec<String>,
bb643c25 Andy McCall ↶ 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>,
b14aeaae Andy McCall ↶ 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>,
d60aac6c Andy McCall ↶ 67
    /// The build cache's size limit in GiB (#28; default 64).
68
    #[serde(default)]
69
    pub cache_gb: Option<u64>,
077a6524 Andy McCall ↶ 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>,
ea81796b Andy McCall ↶ 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,
35f7a849 Andy McCall ↶ 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,
4be5e4cd Andy McCall ↶ 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>,
85275e1e Andy McCall ↶ 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
fn yes() -> bool {
108
    true
ea81796b Andy McCall ↶ 109
}
110
111
impl RunnerConfig {
112
    /// The labels offered to the hub: the configured ones, and `egress`
113
    /// when egress is on.
114
    pub fn offered_labels(&self) -> Vec<String> {
115
        let mut l = self.labels.clone();
116
        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {
117
            l.push(flow_core::EGRESS_LABEL.to_string());
118
        }
119
        l
120
    }
b14aeaae Andy McCall ↶ 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>,
e0918b88 Andy McCall 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
    pub fn locate(explicit: Option<&Path>) -> Result<Home> {
154
        let root = match explicit {
155
            Some(p) => p.to_path_buf(),
156
            None => match std::env::var_os("FLOW_HOME") {
157
                Some(p) => PathBuf::from(p),
158
                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
159
            },
160
        };
161
        let file = root.join("flow.toml");
f98a8f78 Andy McCall ↶ 162
        let mut config: Config = if file.is_file() {
e0918b88 Andy McCall 163
            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;
164
            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?
165
        } else {
166
            Config::default()
167
        };
f98a8f78 Andy McCall ↶ 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
        let registered = root.join(RUNNER_TOML);
171
        if config.runner.is_none() && registered.is_file() {
172
            let text = std::fs::read_to_string(&registered).with_context(|| format!("read {}", registered.display()))?;
0d663e91 Andy McCall ↶ 173
            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
            runner.ca = runner.ca.map(|ca| if ca.is_absolute() { ca } else { root.join(ca) });
177
            config.runner = Some(runner);
f98a8f78 Andy McCall ↶ 178
        }
e0918b88 Andy McCall 179
        for d in ["cas", "work", "repos"] {
180
            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;
181
        }
182
        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
183
    }
184
185
    pub fn resolve(&self, p: &Path) -> PathBuf {
186
        if p.is_absolute() {
187
            p.to_path_buf()
188
        } else {
189
            self.root.join(p)
190
        }
191
    }
192
193
    /// Where a rootfs image with this hash lives.
194
    pub fn rootfs(&self, hash: &str) -> PathBuf {
195
        self.root.join("cas").join(format!("{hash}.ext4"))
196
    }
197
1abadd4a Andy McCall ↶ 198
    /// An image's manifest (`cas/<hash>.json`): what `flow rootfs import`
199
    /// found out about it.
200
    pub fn image_manifest(&self, hash: &str) -> PathBuf {
201
        self.root.join("cas").join(format!("{hash}.json"))
202
    }
203
204
    /// The architecture an image was imported for, when its manifest says.
205
    pub fn image_arch(&self, hash: &str) -> Option<String> {
206
        let text = std::fs::read_to_string(self.image_manifest(hash)).ok()?;
207
        let v: serde_json::Value = serde_json::from_str(&text).ok()?;
208
        v["arch"].as_str().map(String::from)
209
    }
210
211
    /// Every image with a known architecture (spec §11.2).
212
    pub fn image_archs(&self) -> std::collections::BTreeMap<String, String> {
213
        self.images().into_iter().filter_map(|h| self.image_arch(&h).map(|a| (h, a))).collect()
214
    }
215
8c0a4e3a Andy McCall ↶ 216
    /// The rootfs images this home holds: the hashes of `cas/*.ext4`.
217
    pub fn images(&self) -> Vec<String> {
218
        let Ok(dir) = std::fs::read_dir(self.root.join("cas")) else {
219
            return Vec::new();
220
        };
221
        let mut out: Vec<String> = dir
222
            .filter_map(|e| {
223
                let name = e.ok()?.file_name().into_string().ok()?;
224
                let hash = name.strip_suffix(".ext4")?;
225
                (hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit())).then(|| hash.to_string())
226
            })
227
            .collect();
228
        out.sort();
229
        out
230
    }
231
e0918b88 Andy McCall 232
    /// A pinned file, checked against its BLAKE3 before use (again only
233
    /// when its size or modification time changed since).
234
    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
235
        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;
236
        let stamp = (meta.len(), meta.modified()?);
237
        if self.verified.lock().unwrap().get(path) == Some(&stamp) {
238
            return Ok(path.to_path_buf());
239
        }
240
        let got = hash_file(path)?;
241
        if got != blake3_hex {
242
            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
243
        }
244
        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
245
        Ok(path.to_path_buf())
246
    }
247
}
248
249
pub fn hash_file(path: &Path) -> Result<String> {
250
    let mut h = blake3::Hasher::new();
251
    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;
252
    Ok(h.finalize().to_hex().to_string())
47c1d64d Andy McCall ↶ 253
}
254
255
/// The key a runner signs receipts with (Oxygen spec §14.6).
256
pub struct RunnerKey {
257
    pub identity: oxy_core::identity::Identity,
258
    pub secret: oxy_core::identity::SecretKey,
259
}
260
261
impl Home {
262
    /// The runner's key, made on first use as `runner.key` (owner-only)
263
    /// and `runner.identity`, named `name`.
264
    pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {
265
        use oxy_core::identity::{Identity, SecretKey};
266
        let key_path = self.root.join("runner.key");
267
        let id_path = self.root.join("runner.identity");
268
        if key_path.exists() {
269
            let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;
270
            let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;
271
            if identity.hash() != key {
272
                bail!("runner.identity does not belong to runner.key");
273
            }
274
            return Ok(RunnerKey { identity, secret });
275
        }
276
        let secret = SecretKey::generate();
277
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
278
        let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);
279
        std::fs::write(&id_path, identity.encode())?;
280
        let tmp = self.root.join("runner.key.tmp");
281
        {
282
            use std::io::Write;
3554afef Andy McCall ↶ 283
            let mut f = crate::portable::private(std::fs::OpenOptions::new().write(true).create(true).truncate(true)).open(&tmp)?;
47c1d64d Andy McCall ↶ 284
            f.write_all(&secret.encode(&identity.hash()))?;
285
            f.sync_all()?;
286
        }
287
        std::fs::rename(&tmp, &key_path)?;
288
        Ok(RunnerKey { identity, secret })
289
    }
290
}
291
292
#[cfg(test)]
293
mod tests {
294
    use super::*;
295
296
    #[test]
297
    fn the_runner_key_is_made_once_and_kept_private() {
298
        use std::os::unix::fs::PermissionsExt;
299
        let dir = tempfile::tempdir().unwrap();
300
        let home = Home::locate(Some(dir.path())).unwrap();
301
        let a = home.runner_key("earth-1").unwrap();
302
        assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");
303
        let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();
304
        assert_eq!(mode & 0o777, 0o600);
305
        let b = home.runner_key("renamed").unwrap();
306
        assert_eq!(a.identity, b.identity, "the same key after a restart");
307
        let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));
308
        oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();
0d663e91 Andy McCall ↶ 309
    }
310
311
    /// #98: runner.toml's relative CA is the home's, wherever flow starts.
312
    #[test]
313
    fn a_registered_runners_relative_ca_is_in_its_home() {
314
        let dir = tempfile::tempdir().unwrap();
315
        std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"hub-ca.pem\"\n").unwrap();
316
        let home = Home::locate(Some(dir.path())).unwrap();
317
        assert_eq!(home.config.runner.unwrap().ca, Some(dir.path().join("hub-ca.pem")));
318
        std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"/etc/ssl/hub.pem\"\n").unwrap();
319
        let home = Home::locate(Some(dir.path())).unwrap();
320
        assert_eq!(home.config.runner.unwrap().ca, Some(PathBuf::from("/etc/ssl/hub.pem")));
47c1d64d Andy McCall ↶ 321
    }
e0918b88 Andy McCall 322
}