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