OXYGENHUB
# eb8e445f27b6 oxygenflow / crates / flow / src / home.rs
11.3 KiB · Rust · last changed A runner's own metrics (refs #35) 15 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
    /// Serve this runner's metrics at this address, e.g. `127.0.0.1:9464`
84
    /// (oxygenflow#35); off unless set.
85
    #[serde(default)]
86
    pub metrics: Option<String>,
87
    /// A bearer token a scrape of the metrics must send.
88
    #[serde(default)]
89
    pub metrics_token: Option<String>,
90
}
91
92
impl RunnerConfig {
93
    /// The labels offered to the hub: the configured ones, and `egress`
94
    /// when egress is on.
95
    pub fn offered_labels(&self) -> Vec<String> {
96
        let mut l = self.labels.clone();
97
        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {
98
            l.push(flow_core::EGRESS_LABEL.to_string());
99
        }
100
        l
101
    }
102
}
103
104
/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.
105
#[derive(Clone, Debug, Deserialize)]
106
pub struct SwarmConfig {
107
    /// Where peers reach this runner, e.g. `0.0.0.0:7401`.
108
    pub listen: String,
109
    /// The address peers are given (default: `listen`).
110
    #[serde(default)]
111
    pub advertise: Option<String>,
112
    /// Peers are only ever in the same site.
113
    #[serde(default)]
114
    pub site: String,
115
    /// The most this runner sends to peers, MiB a second (default 50).
116
    #[serde(default)]
117
    pub upload_mib_s: Option<u32>,
118
}
119
120
#[derive(Clone, Debug, Default, Deserialize)]
121
pub struct Config {
122
    pub microvm: Option<MicrovmConfig>,
123
    pub runner: Option<RunnerConfig>,
124
}
125
126
pub struct Home {
127
    pub root: PathBuf,
128
    pub config: Config,
129
    /// Files already verified, by path, with the (len, mtime) they had then.
130
    verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
131
}
132
133
impl Home {
134
    pub fn locate(explicit: Option<&Path>) -> Result<Home> {
135
        let root = match explicit {
136
            Some(p) => p.to_path_buf(),
137
            None => match std::env::var_os("FLOW_HOME") {
138
                Some(p) => PathBuf::from(p),
139
                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
140
            },
141
        };
142
        let file = root.join("flow.toml");
143
        let mut config: Config = if file.is_file() {
144
            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;
145
            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?
146
        } else {
147
            Config::default()
148
        };
149
        // `flow runner register` writes the runner's hub and credential to
150
        // runner.toml, which stands in for a [runner] section flow.toml lacks.
151
        let registered = root.join(RUNNER_TOML);
152
        if config.runner.is_none() && registered.is_file() {
153
            let text = std::fs::read_to_string(&registered).with_context(|| format!("read {}", registered.display()))?;
154
            config.runner = Some(toml::from_str(&text).with_context(|| format!("parse {}", registered.display()))?);
155
        }
156
        for d in ["cas", "work", "repos"] {
157
            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;
158
        }
159
        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
160
    }
161
162
    pub fn resolve(&self, p: &Path) -> PathBuf {
163
        if p.is_absolute() {
164
            p.to_path_buf()
165
        } else {
166
            self.root.join(p)
167
        }
168
    }
169
170
    /// Where a rootfs image with this hash lives.
171
    pub fn rootfs(&self, hash: &str) -> PathBuf {
172
        self.root.join("cas").join(format!("{hash}.ext4"))
173
    }
174
175
    /// An image's manifest (`cas/<hash>.json`): what `flow rootfs import`
176
    /// found out about it.
177
    pub fn image_manifest(&self, hash: &str) -> PathBuf {
178
        self.root.join("cas").join(format!("{hash}.json"))
179
    }
180
181
    /// The architecture an image was imported for, when its manifest says.
182
    pub fn image_arch(&self, hash: &str) -> Option<String> {
183
        let text = std::fs::read_to_string(self.image_manifest(hash)).ok()?;
184
        let v: serde_json::Value = serde_json::from_str(&text).ok()?;
185
        v["arch"].as_str().map(String::from)
186
    }
187
188
    /// Every image with a known architecture (spec §11.2).
189
    pub fn image_archs(&self) -> std::collections::BTreeMap<String, String> {
190
        self.images().into_iter().filter_map(|h| self.image_arch(&h).map(|a| (h, a))).collect()
191
    }
192
193
    /// The rootfs images this home holds: the hashes of `cas/*.ext4`.
194
    pub fn images(&self) -> Vec<String> {
195
        let Ok(dir) = std::fs::read_dir(self.root.join("cas")) else {
196
            return Vec::new();
197
        };
198
        let mut out: Vec<String> = dir
199
            .filter_map(|e| {
200
                let name = e.ok()?.file_name().into_string().ok()?;
201
                let hash = name.strip_suffix(".ext4")?;
202
                (hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit())).then(|| hash.to_string())
203
            })
204
            .collect();
205
        out.sort();
206
        out
207
    }
208
209
    /// A pinned file, checked against its BLAKE3 before use (again only
210
    /// when its size or modification time changed since).
211
    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
212
        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;
213
        let stamp = (meta.len(), meta.modified()?);
214
        if self.verified.lock().unwrap().get(path) == Some(&stamp) {
215
            return Ok(path.to_path_buf());
216
        }
217
        let got = hash_file(path)?;
218
        if got != blake3_hex {
219
            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
220
        }
221
        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
222
        Ok(path.to_path_buf())
223
    }
224
}
225
226
pub fn hash_file(path: &Path) -> Result<String> {
227
    let mut h = blake3::Hasher::new();
228
    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;
229
    Ok(h.finalize().to_hex().to_string())
230
}
231
232
/// The key a runner signs receipts with (Oxygen spec §14.6).
233
pub struct RunnerKey {
234
    pub identity: oxy_core::identity::Identity,
235
    pub secret: oxy_core::identity::SecretKey,
236
}
237
238
impl Home {
239
    /// The runner's key, made on first use as `runner.key` (owner-only)
240
    /// and `runner.identity`, named `name`.
241
    pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {
242
        use oxy_core::identity::{Identity, SecretKey};
243
        let key_path = self.root.join("runner.key");
244
        let id_path = self.root.join("runner.identity");
245
        if key_path.exists() {
246
            let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;
247
            let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;
248
            if identity.hash() != key {
249
                bail!("runner.identity does not belong to runner.key");
250
            }
251
            return Ok(RunnerKey { identity, secret });
252
        }
253
        let secret = SecretKey::generate();
254
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
255
        let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);
256
        std::fs::write(&id_path, identity.encode())?;
257
        let tmp = self.root.join("runner.key.tmp");
258
        {
259
            use std::io::Write;
260
            use std::os::unix::fs::OpenOptionsExt;
261
            let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;
262
            f.write_all(&secret.encode(&identity.hash()))?;
263
            f.sync_all()?;
264
        }
265
        std::fs::rename(&tmp, &key_path)?;
266
        Ok(RunnerKey { identity, secret })
267
    }
268
}
269
270
#[cfg(test)]
271
mod tests {
272
    use super::*;
273
274
    #[test]
275
    fn the_runner_key_is_made_once_and_kept_private() {
276
        use std::os::unix::fs::PermissionsExt;
277
        let dir = tempfile::tempdir().unwrap();
278
        let home = Home::locate(Some(dir.path())).unwrap();
279
        let a = home.runner_key("earth-1").unwrap();
280
        assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");
281
        let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();
282
        assert_eq!(mode & 0o777, 0o600);
283
        let b = home.runner_key("renamed").unwrap();
284
        assert_eq!(a.identity, b.identity, "the same key after a restart");
285
        let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));
286
        oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();
287
    }
288
}