OXYGENHUB
# 720eb8191059 oxygenflow / crates / flow-init / src / main.rs
16.4 KiB · Rust · last changed Secrets mount as files: secret blocks, the claim's secrets, flow-init writes them 1 month ago source blame history raw
1
//! PID 1 of a job's microVM (docs/m0.md §2). It mounts the read-only
2
//! rootfs and the workspace, runs the job's argv as an unprivileged user
3
//! with exactly the job's environment, frames the output on the serial
4
//! console for the host, and powers off. It never decides anything the
5
//! host has not: the job comes from `/job.json` in the initramfs the host
6
//! built, and the host enforces the time limit.
7
//!
8
//! Frames, one per line on the console: `\x1eO ` stdout, `\x1eE ` stderr,
9
//! `\x1eX <code>` the exit code (128 + signal when killed), `\x1eF <text>`
10
//! a failure of init itself.
11
12
use serde::Deserialize;
13
use std::ffi::CString;
14
use std::io::{Read, Write};
15
use std::os::fd::FromRawFd;
16
use std::sync::{Arc, Mutex};
17
18
#[derive(Deserialize)]
19
struct JobSpec {
20
    argv: Vec<String>,
21
    env: Vec<(String, String)>,
22
    #[serde(default)]
23
    mounts: Vec<MountSpec>,
24
    /// `network "loopback"`: bring up `lo` (the guest has no other device).
25
    #[serde(default)]
26
    loopback: bool,
27
    /// Secrets to write before the job starts (spec §9): each at `path`
28
    /// under `/run/secrets`, 0400 and the job's user's, on a tmpfs.
29
    #[serde(default)]
30
    secrets: Vec<SecretSpec>,
31
}
32
33
#[derive(serde::Deserialize)]
34
struct SecretSpec {
35
    path: String,
36
    value: String,
37
}
38
39
/// An ephemeral mount from the host: a tmpfs of `size_mib`, or the disk at
40
/// `device`, mounted at `target` under the new root.
41
#[derive(serde::Deserialize)]
42
struct MountSpec {
43
    kind: String,
44
    target: String,
45
    size_mib: u32,
46
    #[serde(default)]
47
    device: Option<String>,
48
}
49
50
const PATH: &str = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";
51
52
fn main() {
53
    let console = Arc::new(Mutex::new(std::io::stdout()));
54
    let code = match run(&console) {
55
        Ok(code) => code,
56
        Err(e) => {
57
            frame(&console, b'F', e.as_bytes());
58
            125
59
        }
60
    };
61
    frame(&console, b'X', code.to_string().as_bytes());
62
    unsafe {
63
        libc::sync();
64
        // Firecracker ends the VM when the guest reboots (reboot=k).
65
        libc::reboot(libc::LINUX_REBOOT_CMD_RESTART);
66
    }
67
}
68
69
fn frame(console: &Mutex<std::io::Stdout>, kind: u8, text: &[u8]) {
70
    let mut out = console.lock().unwrap_or_else(|p| p.into_inner());
71
    let mut line = Vec::with_capacity(text.len() + 4);
72
    line.extend_from_slice(&[0x1e, kind, b' ']);
73
    line.extend_from_slice(text);
74
    line.push(b'\n');
75
    let _ = out.write_all(&line);
76
    let _ = out.flush();
77
}
78
79
fn cstr(s: &str) -> Result<CString, String> {
80
    CString::new(s).map_err(|_| format!("{s:?} contains NUL"))
81
}
82
83
fn chown(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
84
    let p = cstr(path)?;
85
    if unsafe { libc::chown(p.as_ptr(), uid, gid) } != 0 {
86
        return Err(format!("chown {path}: {}", std::io::Error::last_os_error()));
87
    }
88
    Ok(())
89
}
90
91
/// Create `path` and its missing parents, each owned by the job's user.
92
fn make_dirs_owned(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
93
    let mut missing = Vec::new();
94
    let mut p = std::path::Path::new(path);
95
    while !p.exists() {
96
        missing.push(p.to_path_buf());
97
        p = match p.parent() {
98
            Some(parent) => parent,
99
            None => break,
100
        };
101
    }
102
    for d in missing.iter().rev() {
103
        std::fs::create_dir(d).map_err(|e| format!("mkdir {}: {e}", d.display()))?;
104
        chown(&d.to_string_lossy(), uid, gid)?;
105
    }
106
    Ok(())
107
}
108
109
/// The file `program` names: itself when it holds a `/`, otherwise the first
110
/// executable regular file of that name in `path`'s directories.
111
fn resolve_program(program: &str, path: &str) -> Option<String> {
112
    use std::os::unix::fs::PermissionsExt;
113
    if program.is_empty() {
114
        return None;
115
    }
116
    if program.contains('/') {
117
        return Some(program.to_string());
118
    }
119
    path.split(':')
120
        .filter(|d| !d.is_empty())
121
        .map(|d| format!("{d}/{program}"))
122
        .find(|p| std::fs::metadata(p).is_ok_and(|m| m.is_file() && m.permissions().mode() & 0o111 != 0))
123
}
124
125
/// A job's environment: `PATH` and `HOME=/tmp` by default, then the
126
/// image's own (`/etc/flow/env`, written by `flow rootfs import --env`),
127
/// then the job's (its sandbox's and its own, already merged by the host).
128
/// Later ones win, so a pipeline can always set `PATH`.
129
fn job_env(image: Option<&str>, job: &[(String, String)]) -> std::collections::BTreeMap<String, String> {
130
    let mut env = std::collections::BTreeMap::new();
131
    env.insert("PATH".to_string(), PATH.to_string());
132
    env.insert("HOME".to_string(), "/tmp".to_string());
133
    for line in image.unwrap_or("").lines() {
134
        if let Some((k, v)) = line.split_once('=') {
135
            if !k.is_empty() {
136
                env.insert(k.to_string(), v.to_string());
137
            }
138
        }
139
    }
140
    for (k, v) in job {
141
        env.insert(k.clone(), v.clone());
142
    }
143
    env
144
}
145
146
/// The links every Linux system has under /dev and devtmpfs does not
147
/// make: `/dev/fd` (bash's process substitution, `<(…)`) and the standard
148
/// streams, all through the job's own /proc.
149
fn dev_links(dev: &str) -> Result<(), String> {
150
    for (name, target) in DEV_LINKS {
151
        let link = format!("{dev}/{name}");
152
        if std::fs::symlink_metadata(&link).is_err() {
153
            std::os::unix::fs::symlink(target, &link).map_err(|e| format!("link {link}: {e}"))?;
154
        }
155
    }
156
    Ok(())
157
}
158
159
const DEV_LINKS: [(&str, &str); 4] =
160
    [("fd", "/proc/self/fd"), ("stdin", "/proc/self/fd/0"), ("stdout", "/proc/self/fd/1"), ("stderr", "/proc/self/fd/2")];
161
162
/// Set `lo` up; the kernel gives it 127.0.0.1 and ::1 itself.
163
fn loopback_up() -> Result<(), String> {
164
    let fd = unsafe { libc::socket(libc::AF_INET, libc::SOCK_DGRAM | libc::SOCK_CLOEXEC, 0) };
165
    if fd < 0 {
166
        return Err(format!("socket for lo: {}", std::io::Error::last_os_error()));
167
    }
168
    let mut req: libc::ifreq = unsafe { std::mem::zeroed() };
169
    for (i, b) in b"lo".iter().enumerate() {
170
        req.ifr_name[i] = *b as libc::c_char;
171
    }
172
    let ok = unsafe {
173
        libc::ioctl(fd, libc::SIOCGIFFLAGS, &mut req) == 0 && {
174
            req.ifr_ifru.ifru_flags |= libc::IFF_UP as libc::c_short;
175
            libc::ioctl(fd, libc::SIOCSIFFLAGS, &req) == 0
176
        }
177
    };
178
    let err = std::io::Error::last_os_error();
179
    unsafe { libc::close(fd) };
180
    if !ok {
181
        return Err(format!("bring up lo: {err}"));
182
    }
183
    Ok(())
184
}
185
186
fn mount(src: &str, target: &str, fstype: &str, flags: libc::c_ulong, data: &str) -> Result<(), String> {
187
    let (s, t, f, d) = (cstr(src)?, cstr(target)?, cstr(fstype)?, cstr(data)?);
188
    let rc = unsafe { libc::mount(s.as_ptr(), t.as_ptr(), f.as_ptr(), flags, d.as_ptr() as *const libc::c_void) };
189
    if rc != 0 {
190
        return Err(format!("mount {src} on {target}: {}", std::io::Error::last_os_error()));
191
    }
192
    Ok(())
193
}
194
195
fn run(console: &Arc<Mutex<std::io::Stdout>>) -> Result<i32, String> {
196
    let job: JobSpec = serde_json::from_slice(&std::fs::read("/job.json").map_err(|e| format!("read /job.json: {e}"))?)
197
        .map_err(|e| format!("parse /job.json: {e}"))?;
198
    if job.argv.is_empty() {
199
        return Err("empty argv".into());
200
    }
201
    let nosuid = libc::MS_NOSUID | libc::MS_NODEV;
202
    std::fs::create_dir_all("/dev").ok();
203
    mount("devtmpfs", "/dev", "devtmpfs", libc::MS_NOSUID, "")?;
204
    std::fs::create_dir_all("/newroot").map_err(|e| e.to_string())?;
205
    mount("/dev/vda", "/newroot", "ext4", libc::MS_RDONLY | libc::MS_NODEV, "")?;
206
    mount("/dev/vdb", "/newroot/src", "ext4", nosuid, "")?;
207
    mount("tmpfs", "/newroot/tmp", "tmpfs", nosuid, "mode=1777")?;
208
    mount("proc", "/newroot/proc", "proc", nosuid | libc::MS_NOEXEC, "")?;
209
    mount("devtmpfs", "/newroot/dev", "devtmpfs", libc::MS_NOSUID, "")?;
210
    dev_links("/newroot/dev")?;
211
    if std::path::Path::new("/newroot/run").is_dir() {
212
        mount("tmpfs", "/newroot/run", "tmpfs", nosuid, "mode=755")?;
213
    }
214
    // The job runs as whoever owns the workspace (the host's builder), never root.
215
    let (uid, gid) = {
216
        let mut st: libc::stat = unsafe { std::mem::zeroed() };
217
        let p = cstr("/newroot/src")?;
218
        if unsafe { libc::stat(p.as_ptr(), &mut st) } != 0 {
219
            return Err("stat /src".into());
220
        }
221
        if st.st_uid == 0 {
222
            (1000, 1000)
223
        } else {
224
            (st.st_uid, st.st_gid)
225
        }
226
    };
227
    // Ephemeral mounts: mount points are made (and owned by the job's user)
228
    // under the writable /src or /tmp, then each is mounted and given to it.
229
    for m in &job.mounts {
230
        let target = format!("/newroot{}", m.target);
231
        make_dirs_owned(&target, uid, gid)?;
232
        match (m.kind.as_str(), &m.device) {
233
            ("tmpfs", _) => mount("tmpfs", &target, "tmpfs", nosuid, &format!("size={}m,uid={uid},gid={gid},mode=755", m.size_mib))?,
234
            ("disk", Some(dev)) => {
235
                mount(dev, &target, "ext4", nosuid, "")?;
236
                chown(&target, uid, gid)?;
237
            }
238
            (kind, _) => return Err(format!("unknown mount kind {kind}")),
239
        }
240
    }
241
    // Secrets: an in-memory mount of their own, only the job's user's,
242
    // each file 0400; the job file they came from is outside the new root.
243
    if !job.secrets.is_empty() {
244
        if !std::path::Path::new("/newroot/run").is_dir() {
245
            return Err("the image has no /run, where secrets are mounted".into());
246
        }
247
        let dir = "/newroot/run/secrets";
248
        std::fs::create_dir_all(dir).map_err(|e| format!("mkdir {dir}: {e}"))?;
249
        mount("tmpfs", dir, "tmpfs", nosuid | libc::MS_NOEXEC, &format!("size=1m,uid={uid},gid={gid},mode=700"))?;
250
        for s in &job.secrets {
251
            let Some(rel) = s.path.strip_prefix("/run/secrets/") else {
252
                return Err(format!("secret path {:?} is not under /run/secrets/", s.path));
253
            };
254
            let full = format!("{dir}/{rel}");
255
            if let Some(parent) = std::path::Path::new(&full).parent().and_then(|p| p.to_str()) {
256
                make_dirs_owned(parent, uid, gid)?;
257
            }
258
            std::fs::write(&full, s.value.as_bytes()).map_err(|e| format!("write {}: {e}", s.path))?;
259
            chown(&full, uid, gid)?;
260
            std::fs::set_permissions(&full, std::os::unix::fs::PermissionsExt::from_mode(0o400))
261
                .map_err(|e| format!("chmod {}: {e}", s.path))?;
262
        }
263
    }
264
    if job.loopback {
265
        loopback_up()?;
266
    }
267
    let root = cstr("/newroot")?;
268
    if unsafe { libc::chroot(root.as_ptr()) } != 0 || std::env::set_current_dir("/src").is_err() {
269
        return Err(format!("enter the rootfs: {}", std::io::Error::last_os_error()));
270
    }
271
272
    let argv: Vec<CString> = job.argv.iter().map(|a| cstr(a)).collect::<Result<_, _>>()?;
273
    // glibc's execvpe searches the caller's PATH, and PID 1 has none of its
274
    // own, so the program is found here, in the job's PATH, before forking.
275
    let vars = job_env(std::fs::read_to_string("/etc/flow/env").ok().as_deref(), &job.env);
276
    let path = vars.get("PATH").map(String::as_str).unwrap_or(PATH);
277
    let program = cstr(&resolve_program(&job.argv[0], path).ok_or_else(|| format!("cannot run {:?}: not found in {path}", job.argv[0]))?)?;
278
    let env: Vec<CString> = vars.iter().map(|(k, v)| cstr(&format!("{k}={v}"))).collect::<Result<_, _>>()?;
279
    let mut out = [0; 2];
280
    let mut err = [0; 2];
281
    if unsafe { libc::pipe2(out.as_mut_ptr(), libc::O_CLOEXEC) } != 0 || unsafe { libc::pipe2(err.as_mut_ptr(), libc::O_CLOEXEC) } != 0 {
282
        return Err("pipe".into());
283
    }
284
    let pid = unsafe { libc::fork() };
285
    if pid < 0 {
286
        return Err("fork".into());
287
    }
288
    if pid == 0 {
289
        // Child: no way back to root, then the program itself (execvpe: no shell).
290
        unsafe {
291
            let null = libc::open(c"/dev/null".as_ptr(), libc::O_RDONLY);
292
            libc::dup2(null, 0);
293
            libc::dup2(out[1], 1);
294
            libc::dup2(err[1], 2);
295
            libc::setsid();
296
            if libc::setgroups(0, std::ptr::null()) != 0 || libc::setgid(gid) != 0 || libc::setuid(uid) != 0 {
297
                libc::_exit(126);
298
            }
299
            let mut a: Vec<*const libc::c_char> = argv.iter().map(|c| c.as_ptr()).collect();
300
            a.push(std::ptr::null());
301
            let mut e: Vec<*const libc::c_char> = env.iter().map(|c| c.as_ptr()).collect();
302
            e.push(std::ptr::null());
303
            libc::execve(program.as_ptr(), a.as_ptr(), e.as_ptr());
304
            let msg = format!("flow: cannot run {:?}: {}\n", job.argv[0], std::io::Error::last_os_error());
305
            libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());
306
            libc::_exit(127);
307
        }
308
    }
309
    unsafe {
310
        libc::close(out[1]);
311
        libc::close(err[1]);
312
    }
313
    let readers: Vec<_> = [(out[0], b'O'), (err[0], b'E')]
314
        .into_iter()
315
        .map(|(fd, kind)| {
316
            let console = console.clone();
317
            std::thread::spawn(move || pump(unsafe { std::fs::File::from_raw_fd(fd) }, kind, &console))
318
        })
319
        .collect();
320
    let mut status = 0;
321
    // Reap everything (we are PID 1) until the job itself exits.
322
    loop {
323
        let r = unsafe { libc::waitpid(-1, &mut status, 0) };
324
        if r == pid || r < 0 {
325
            break;
326
        }
327
    }
328
    for r in readers {
329
        let _ = r.join();
330
    }
331
    Ok(if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) })
332
}
333
334
/// Frame a stream line by line; a long line is cut at 16 KiB.
335
fn pump(mut f: std::fs::File, kind: u8, console: &Mutex<std::io::Stdout>) {
336
    let mut buf = [0u8; 8192];
337
    let mut line: Vec<u8> = Vec::new();
338
    loop {
339
        let n = match f.read(&mut buf) {
340
            Ok(0) | Err(_) => break,
341
            Ok(n) => n,
342
        };
343
        for &b in &buf[..n] {
344
            if b == b'\n' || line.len() >= 16 * 1024 {
345
                frame(console, kind, &line);
346
                line.clear();
347
                if b == b'\n' {
348
                    continue;
349
                }
350
            }
351
            line.push(b);
352
        }
353
    }
354
    if !line.is_empty() {
355
        frame(console, kind, &line);
356
    }
357
}
358
359
#[cfg(test)]
360
mod tests {
361
    use super::{dev_links, job_env, resolve_program};
362
363
    #[test]
364
    fn the_image_environment_sits_under_the_jobs() {
365
        let image = "PATH=/opt/llvm-mos/bin:/usr/bin:/bin\nX16EMU=/opt/x16emu\nnot a line\n";
366
        let env = job_env(Some(image), &[("X16EMU".into(), "/job".into()), ("LANG".into(), "C".into())]);
367
        assert_eq!(env["PATH"], "/opt/llvm-mos/bin:/usr/bin:/bin", "the image's PATH replaces the default");
368
        assert_eq!(env["X16EMU"], "/job", "the job's value wins");
369
        assert_eq!(env["HOME"], "/tmp");
370
        assert_eq!(env["LANG"], "C");
371
        assert_eq!(job_env(None, &[])["PATH"], super::PATH);
372
    }
373
374
    #[test]
375
    fn dev_gets_the_usual_links_once() {
376
        let dir = std::env::temp_dir().join(format!("flow-init-dev-{}", std::process::id()));
377
        let _ = std::fs::remove_dir_all(&dir);
378
        std::fs::create_dir_all(&dir).unwrap();
379
        let d = dir.to_str().unwrap();
380
        dev_links(d).unwrap();
381
        dev_links(d).unwrap();
382
        assert_eq!(std::fs::read_link(dir.join("fd")).unwrap(), std::path::Path::new("/proc/self/fd"));
383
        assert_eq!(std::fs::read_link(dir.join("stdout")).unwrap(), std::path::Path::new("/proc/self/fd/1"));
384
        // Through the host's /proc, the link works as in a job.
385
        assert!(dir.join("fd").join("0").exists() || !std::path::Path::new("/proc/self/fd/0").exists());
386
        std::fs::remove_dir_all(&dir).unwrap();
387
    }
388
389
    use std::os::unix::fs::PermissionsExt;
390
391
    #[test]
392
    fn programs_are_found_in_the_jobs_path() {
393
        let dir = std::env::temp_dir().join(format!("flow-init-test-{}", std::process::id()));
394
        let (a, b) = (dir.join("a"), dir.join("b"));
395
        std::fs::create_dir_all(&a).unwrap();
396
        std::fs::create_dir_all(&b).unwrap();
397
        std::fs::write(a.join("tool"), "not executable").unwrap();
398
        std::fs::write(b.join("tool"), "#!/bin/sh\n").unwrap();
399
        std::fs::set_permissions(b.join("tool"), std::fs::Permissions::from_mode(0o755)).unwrap();
400
        std::fs::create_dir_all(a.join("dir")).unwrap();
401
        let path = format!("{}::{}", a.display(), b.display());
402
        assert_eq!(resolve_program("tool", &path), Some(format!("{}/tool", b.display())), "skips a file that is not executable");
403
        assert_eq!(resolve_program("dir", &path), None, "and directories");
404
        assert_eq!(resolve_program("missing", &path), None);
405
        assert_eq!(resolve_program("./x", &path).as_deref(), Some("./x"), "a path is used as given");
406
        assert_eq!(resolve_program("", &path), None);
407
        std::fs::remove_dir_all(&dir).unwrap();
408
    }
409
}