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