OXYGENHUB
# 2814f4213e7c oxygenflow / crates / flow-init / src / main.rs
6.9 KiB · Rust · last changed OxygenFlow M0: pipeline parser and canonical job graph, receipts, the runner API types, flow-init, the Firecracker tier, flow check, run, rootfs import and runner 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
}
23
24
const PATH: &str = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";
25
26
fn main() {
27
    let console = Arc::new(Mutex::new(std::io::stdout()));
28
    let code = match run(&console) {
29
        Ok(code) => code,
30
        Err(e) => {
31
            frame(&console, b'F', e.as_bytes());
32
            125
33
        }
34
    };
35
    frame(&console, b'X', code.to_string().as_bytes());
36
    unsafe {
37
        libc::sync();
38
        // Firecracker ends the VM when the guest reboots (reboot=k).
39
        libc::reboot(libc::LINUX_REBOOT_CMD_RESTART);
40
    }
41
}
42
43
fn frame(console: &Mutex<std::io::Stdout>, kind: u8, text: &[u8]) {
44
    let mut out = console.lock().unwrap_or_else(|p| p.into_inner());
45
    let mut line = Vec::with_capacity(text.len() + 4);
46
    line.extend_from_slice(&[0x1e, kind, b' ']);
47
    line.extend_from_slice(text);
48
    line.push(b'\n');
49
    let _ = out.write_all(&line);
50
    let _ = out.flush();
51
}
52
53
fn cstr(s: &str) -> Result<CString, String> {
54
    CString::new(s).map_err(|_| format!("{s:?} contains NUL"))
55
}
56
57
fn mount(src: &str, target: &str, fstype: &str, flags: libc::c_ulong, data: &str) -> Result<(), String> {
58
    let (s, t, f, d) = (cstr(src)?, cstr(target)?, cstr(fstype)?, cstr(data)?);
59
    let rc = unsafe { libc::mount(s.as_ptr(), t.as_ptr(), f.as_ptr(), flags, d.as_ptr() as *const libc::c_void) };
60
    if rc != 0 {
61
        return Err(format!("mount {src} on {target}: {}", std::io::Error::last_os_error()));
62
    }
63
    Ok(())
64
}
65
66
fn run(console: &Arc<Mutex<std::io::Stdout>>) -> Result<i32, String> {
67
    let job: JobSpec = serde_json::from_slice(&std::fs::read("/job.json").map_err(|e| format!("read /job.json: {e}"))?)
68
        .map_err(|e| format!("parse /job.json: {e}"))?;
69
    if job.argv.is_empty() {
70
        return Err("empty argv".into());
71
    }
72
    let nosuid = libc::MS_NOSUID | libc::MS_NODEV;
73
    std::fs::create_dir_all("/dev").ok();
74
    mount("devtmpfs", "/dev", "devtmpfs", libc::MS_NOSUID, "")?;
75
    std::fs::create_dir_all("/newroot").map_err(|e| e.to_string())?;
76
    mount("/dev/vda", "/newroot", "ext4", libc::MS_RDONLY | libc::MS_NODEV, "")?;
77
    mount("/dev/vdb", "/newroot/src", "ext4", nosuid, "")?;
78
    mount("tmpfs", "/newroot/tmp", "tmpfs", nosuid, "mode=1777")?;
79
    mount("proc", "/newroot/proc", "proc", nosuid | libc::MS_NOEXEC, "")?;
80
    mount("devtmpfs", "/newroot/dev", "devtmpfs", libc::MS_NOSUID, "")?;
81
    if std::path::Path::new("/newroot/run").is_dir() {
82
        mount("tmpfs", "/newroot/run", "tmpfs", nosuid, "mode=755")?;
83
    }
84
    // The job runs as whoever owns the workspace (the host's builder), never root.
85
    let (uid, gid) = {
86
        let mut st: libc::stat = unsafe { std::mem::zeroed() };
87
        let p = cstr("/newroot/src")?;
88
        if unsafe { libc::stat(p.as_ptr(), &mut st) } != 0 {
89
            return Err("stat /src".into());
90
        }
91
        if st.st_uid == 0 {
92
            (1000, 1000)
93
        } else {
94
            (st.st_uid, st.st_gid)
95
        }
96
    };
97
    let root = cstr("/newroot")?;
98
    if unsafe { libc::chroot(root.as_ptr()) } != 0 || std::env::set_current_dir("/src").is_err() {
99
        return Err(format!("enter the rootfs: {}", std::io::Error::last_os_error()));
100
    }
101
102
    let argv: Vec<CString> = job.argv.iter().map(|a| cstr(a)).collect::<Result<_, _>>()?;
103
    let mut env: Vec<CString> = vec![cstr(&format!("PATH={PATH}"))?, cstr("HOME=/tmp")?];
104
    for (k, v) in &job.env {
105
        env.push(cstr(&format!("{k}={v}"))?);
106
    }
107
    let mut out = [0; 2];
108
    let mut err = [0; 2];
109
    if unsafe { libc::pipe2(out.as_mut_ptr(), libc::O_CLOEXEC) } != 0 || unsafe { libc::pipe2(err.as_mut_ptr(), libc::O_CLOEXEC) } != 0 {
110
        return Err("pipe".into());
111
    }
112
    let pid = unsafe { libc::fork() };
113
    if pid < 0 {
114
        return Err("fork".into());
115
    }
116
    if pid == 0 {
117
        // Child: no way back to root, then the program itself (execvpe: no shell).
118
        unsafe {
119
            let null = libc::open(c"/dev/null".as_ptr(), libc::O_RDONLY);
120
            libc::dup2(null, 0);
121
            libc::dup2(out[1], 1);
122
            libc::dup2(err[1], 2);
123
            libc::setsid();
124
            if libc::setgroups(0, std::ptr::null()) != 0 || libc::setgid(gid) != 0 || libc::setuid(uid) != 0 {
125
                libc::_exit(126);
126
            }
127
            let mut a: Vec<*const libc::c_char> = argv.iter().map(|c| c.as_ptr()).collect();
128
            a.push(std::ptr::null());
129
            let mut e: Vec<*const libc::c_char> = env.iter().map(|c| c.as_ptr()).collect();
130
            e.push(std::ptr::null());
131
            libc::execvpe(a[0], a.as_ptr(), e.as_ptr());
132
            let msg = format!("flow: cannot run {:?}: {}\n", job.argv[0], std::io::Error::last_os_error());
133
            libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());
134
            libc::_exit(127);
135
        }
136
    }
137
    unsafe {
138
        libc::close(out[1]);
139
        libc::close(err[1]);
140
    }
141
    let readers: Vec<_> = [(out[0], b'O'), (err[0], b'E')]
142
        .into_iter()
143
        .map(|(fd, kind)| {
144
            let console = console.clone();
145
            std::thread::spawn(move || pump(unsafe { std::fs::File::from_raw_fd(fd) }, kind, &console))
146
        })
147
        .collect();
148
    let mut status = 0;
149
    // Reap everything (we are PID 1) until the job itself exits.
150
    loop {
151
        let r = unsafe { libc::waitpid(-1, &mut status, 0) };
152
        if r == pid || r < 0 {
153
            break;
154
        }
155
    }
156
    for r in readers {
157
        let _ = r.join();
158
    }
159
    Ok(if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) })
160
}
161
162
/// Frame a stream line by line; a long line is cut at 16 KiB.
163
fn pump(mut f: std::fs::File, kind: u8, console: &Mutex<std::io::Stdout>) {
164
    let mut buf = [0u8; 8192];
165
    let mut line: Vec<u8> = Vec::new();
166
    loop {
167
        let n = match f.read(&mut buf) {
168
            Ok(0) | Err(_) => break,
169
            Ok(n) => n,
170
        };
171
        for &b in &buf[..n] {
172
            if b == b'\n' || line.len() >= 16 * 1024 {
173
                frame(console, kind, &line);
174
                line.clear();
175
                if b == b'\n' {
176
                    continue;
177
                }
178
            }
179
            line.push(b);
180
        }
181
    }
182
    if !line.is_empty() {
183
        frame(console, kind, &line);
184
    }
185
}