OXYGENHUB
# eb5c4db673e5 oxygenflow / crates / flow-init / src / main.rs
11.8 KiB · Rust · last changed Build image for Rust and Node jobs (images/rust: toolchain, rustfmt, clippy, Node 24, vendored crates, offline cargo); flow-init finds the program in the job's PATH (glibc's execvpe used PID 1's own); network "loopback" in M0 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
}
28
29
/// An ephemeral mount from the host: a tmpfs of `size_mib`, or the disk at
30
/// `device`, mounted at `target` under the new root.
31
#[derive(serde::Deserialize)]
32
struct MountSpec {
33
    kind: String,
34
    target: String,
35
    size_mib: u32,
36
    #[serde(default)]
37
    device: Option<String>,
38
}
39
40
const PATH: &str = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";
41
42
fn main() {
43
    let console = Arc::new(Mutex::new(std::io::stdout()));
44
    let code = match run(&console) {
45
        Ok(code) => code,
46
        Err(e) => {
47
            frame(&console, b'F', e.as_bytes());
48
            125
49
        }
50
    };
51
    frame(&console, b'X', code.to_string().as_bytes());
52
    unsafe {
53
        libc::sync();
54
        // Firecracker ends the VM when the guest reboots (reboot=k).
55
        libc::reboot(libc::LINUX_REBOOT_CMD_RESTART);
56
    }
57
}
58
59
fn frame(console: &Mutex<std::io::Stdout>, kind: u8, text: &[u8]) {
60
    let mut out = console.lock().unwrap_or_else(|p| p.into_inner());
61
    let mut line = Vec::with_capacity(text.len() + 4);
62
    line.extend_from_slice(&[0x1e, kind, b' ']);
63
    line.extend_from_slice(text);
64
    line.push(b'\n');
65
    let _ = out.write_all(&line);
66
    let _ = out.flush();
67
}
68
69
fn cstr(s: &str) -> Result<CString, String> {
70
    CString::new(s).map_err(|_| format!("{s:?} contains NUL"))
71
}
72
73
fn chown(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
74
    let p = cstr(path)?;
75
    if unsafe { libc::chown(p.as_ptr(), uid, gid) } != 0 {
76
        return Err(format!("chown {path}: {}", std::io::Error::last_os_error()));
77
    }
78
    Ok(())
79
}
80
81
/// Create `path` and its missing parents, each owned by the job's user.
82
fn make_dirs_owned(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
83
    let mut missing = Vec::new();
84
    let mut p = std::path::Path::new(path);
85
    while !p.exists() {
86
        missing.push(p.to_path_buf());
87
        p = match p.parent() {
88
            Some(parent) => parent,
89
            None => break,
90
        };
91
    }
92
    for d in missing.iter().rev() {
93
        std::fs::create_dir(d).map_err(|e| format!("mkdir {}: {e}", d.display()))?;
94
        chown(&d.to_string_lossy(), uid, gid)?;
95
    }
96
    Ok(())
97
}
98
99
/// The file `program` names: itself when it holds a `/`, otherwise the first
100
/// executable regular file of that name in `path`'s directories.
101
fn resolve_program(program: &str, path: &str) -> Option<String> {
102
    use std::os::unix::fs::PermissionsExt;
103
    if program.is_empty() {
104
        return None;
105
    }
106
    if program.contains('/') {
107
        return Some(program.to_string());
108
    }
109
    path.split(':')
110
        .filter(|d| !d.is_empty())
111
        .map(|d| format!("{d}/{program}"))
112
        .find(|p| std::fs::metadata(p).is_ok_and(|m| m.is_file() && m.permissions().mode() & 0o111 != 0))
113
}
114
115
/// Set `lo` up; the kernel gives it 127.0.0.1 and ::1 itself.
116
fn loopback_up() -> Result<(), String> {
117
    let fd = unsafe { libc::socket(libc::AF_INET, libc::SOCK_DGRAM | libc::SOCK_CLOEXEC, 0) };
118
    if fd < 0 {
119
        return Err(format!("socket for lo: {}", std::io::Error::last_os_error()));
120
    }
121
    let mut req: libc::ifreq = unsafe { std::mem::zeroed() };
122
    for (i, b) in b"lo".iter().enumerate() {
123
        req.ifr_name[i] = *b as libc::c_char;
124
    }
125
    let ok = unsafe {
126
        libc::ioctl(fd, libc::SIOCGIFFLAGS, &mut req) == 0 && {
127
            req.ifr_ifru.ifru_flags |= libc::IFF_UP as libc::c_short;
128
            libc::ioctl(fd, libc::SIOCSIFFLAGS, &req) == 0
129
        }
130
    };
131
    let err = std::io::Error::last_os_error();
132
    unsafe { libc::close(fd) };
133
    if !ok {
134
        return Err(format!("bring up lo: {err}"));
135
    }
136
    Ok(())
137
}
138
139
fn mount(src: &str, target: &str, fstype: &str, flags: libc::c_ulong, data: &str) -> Result<(), String> {
140
    let (s, t, f, d) = (cstr(src)?, cstr(target)?, cstr(fstype)?, cstr(data)?);
141
    let rc = unsafe { libc::mount(s.as_ptr(), t.as_ptr(), f.as_ptr(), flags, d.as_ptr() as *const libc::c_void) };
142
    if rc != 0 {
143
        return Err(format!("mount {src} on {target}: {}", std::io::Error::last_os_error()));
144
    }
145
    Ok(())
146
}
147
148
fn run(console: &Arc<Mutex<std::io::Stdout>>) -> Result<i32, String> {
149
    let job: JobSpec = serde_json::from_slice(&std::fs::read("/job.json").map_err(|e| format!("read /job.json: {e}"))?)
150
        .map_err(|e| format!("parse /job.json: {e}"))?;
151
    if job.argv.is_empty() {
152
        return Err("empty argv".into());
153
    }
154
    let nosuid = libc::MS_NOSUID | libc::MS_NODEV;
155
    std::fs::create_dir_all("/dev").ok();
156
    mount("devtmpfs", "/dev", "devtmpfs", libc::MS_NOSUID, "")?;
157
    std::fs::create_dir_all("/newroot").map_err(|e| e.to_string())?;
158
    mount("/dev/vda", "/newroot", "ext4", libc::MS_RDONLY | libc::MS_NODEV, "")?;
159
    mount("/dev/vdb", "/newroot/src", "ext4", nosuid, "")?;
160
    mount("tmpfs", "/newroot/tmp", "tmpfs", nosuid, "mode=1777")?;
161
    mount("proc", "/newroot/proc", "proc", nosuid | libc::MS_NOEXEC, "")?;
162
    mount("devtmpfs", "/newroot/dev", "devtmpfs", libc::MS_NOSUID, "")?;
163
    if std::path::Path::new("/newroot/run").is_dir() {
164
        mount("tmpfs", "/newroot/run", "tmpfs", nosuid, "mode=755")?;
165
    }
166
    // The job runs as whoever owns the workspace (the host's builder), never root.
167
    let (uid, gid) = {
168
        let mut st: libc::stat = unsafe { std::mem::zeroed() };
169
        let p = cstr("/newroot/src")?;
170
        if unsafe { libc::stat(p.as_ptr(), &mut st) } != 0 {
171
            return Err("stat /src".into());
172
        }
173
        if st.st_uid == 0 {
174
            (1000, 1000)
175
        } else {
176
            (st.st_uid, st.st_gid)
177
        }
178
    };
179
    // Ephemeral mounts: mount points are made (and owned by the job's user)
180
    // under the writable /src or /tmp, then each is mounted and given to it.
181
    for m in &job.mounts {
182
        let target = format!("/newroot{}", m.target);
183
        make_dirs_owned(&target, uid, gid)?;
184
        match (m.kind.as_str(), &m.device) {
185
            ("tmpfs", _) => mount("tmpfs", &target, "tmpfs", nosuid, &format!("size={}m,uid={uid},gid={gid},mode=755", m.size_mib))?,
186
            ("disk", Some(dev)) => {
187
                mount(dev, &target, "ext4", nosuid, "")?;
188
                chown(&target, uid, gid)?;
189
            }
190
            (kind, _) => return Err(format!("unknown mount kind {kind}")),
191
        }
192
    }
193
    if job.loopback {
194
        loopback_up()?;
195
    }
196
    let root = cstr("/newroot")?;
197
    if unsafe { libc::chroot(root.as_ptr()) } != 0 || std::env::set_current_dir("/src").is_err() {
198
        return Err(format!("enter the rootfs: {}", std::io::Error::last_os_error()));
199
    }
200
201
    let argv: Vec<CString> = job.argv.iter().map(|a| cstr(a)).collect::<Result<_, _>>()?;
202
    // glibc's execvpe searches the caller's PATH, and PID 1 has none of its
203
    // own, so the program is found here, in the job's PATH, before forking.
204
    let program = cstr(&resolve_program(&job.argv[0], PATH).ok_or_else(|| format!("cannot run {:?}: not found in {PATH}", job.argv[0]))?)?;
205
    let mut env: Vec<CString> = vec![cstr(&format!("PATH={PATH}"))?, cstr("HOME=/tmp")?];
206
    for (k, v) in &job.env {
207
        env.push(cstr(&format!("{k}={v}"))?);
208
    }
209
    let mut out = [0; 2];
210
    let mut err = [0; 2];
211
    if unsafe { libc::pipe2(out.as_mut_ptr(), libc::O_CLOEXEC) } != 0 || unsafe { libc::pipe2(err.as_mut_ptr(), libc::O_CLOEXEC) } != 0 {
212
        return Err("pipe".into());
213
    }
214
    let pid = unsafe { libc::fork() };
215
    if pid < 0 {
216
        return Err("fork".into());
217
    }
218
    if pid == 0 {
219
        // Child: no way back to root, then the program itself (execvpe: no shell).
220
        unsafe {
221
            let null = libc::open(c"/dev/null".as_ptr(), libc::O_RDONLY);
222
            libc::dup2(null, 0);
223
            libc::dup2(out[1], 1);
224
            libc::dup2(err[1], 2);
225
            libc::setsid();
226
            if libc::setgroups(0, std::ptr::null()) != 0 || libc::setgid(gid) != 0 || libc::setuid(uid) != 0 {
227
                libc::_exit(126);
228
            }
229
            let mut a: Vec<*const libc::c_char> = argv.iter().map(|c| c.as_ptr()).collect();
230
            a.push(std::ptr::null());
231
            let mut e: Vec<*const libc::c_char> = env.iter().map(|c| c.as_ptr()).collect();
232
            e.push(std::ptr::null());
233
            libc::execve(program.as_ptr(), a.as_ptr(), e.as_ptr());
234
            let msg = format!("flow: cannot run {:?}: {}\n", job.argv[0], std::io::Error::last_os_error());
235
            libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());
236
            libc::_exit(127);
237
        }
238
    }
239
    unsafe {
240
        libc::close(out[1]);
241
        libc::close(err[1]);
242
    }
243
    let readers: Vec<_> = [(out[0], b'O'), (err[0], b'E')]
244
        .into_iter()
245
        .map(|(fd, kind)| {
246
            let console = console.clone();
247
            std::thread::spawn(move || pump(unsafe { std::fs::File::from_raw_fd(fd) }, kind, &console))
248
        })
249
        .collect();
250
    let mut status = 0;
251
    // Reap everything (we are PID 1) until the job itself exits.
252
    loop {
253
        let r = unsafe { libc::waitpid(-1, &mut status, 0) };
254
        if r == pid || r < 0 {
255
            break;
256
        }
257
    }
258
    for r in readers {
259
        let _ = r.join();
260
    }
261
    Ok(if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) })
262
}
263
264
/// Frame a stream line by line; a long line is cut at 16 KiB.
265
fn pump(mut f: std::fs::File, kind: u8, console: &Mutex<std::io::Stdout>) {
266
    let mut buf = [0u8; 8192];
267
    let mut line: Vec<u8> = Vec::new();
268
    loop {
269
        let n = match f.read(&mut buf) {
270
            Ok(0) | Err(_) => break,
271
            Ok(n) => n,
272
        };
273
        for &b in &buf[..n] {
274
            if b == b'\n' || line.len() >= 16 * 1024 {
275
                frame(console, kind, &line);
276
                line.clear();
277
                if b == b'\n' {
278
                    continue;
279
                }
280
            }
281
            line.push(b);
282
        }
283
    }
284
    if !line.is_empty() {
285
        frame(console, kind, &line);
286
    }
287
}
288
289
#[cfg(test)]
290
mod tests {
291
    use super::resolve_program;
292
    use std::os::unix::fs::PermissionsExt;
293
294
    #[test]
295
    fn programs_are_found_in_the_jobs_path() {
296
        let dir = std::env::temp_dir().join(format!("flow-init-test-{}", std::process::id()));
297
        let (a, b) = (dir.join("a"), dir.join("b"));
298
        std::fs::create_dir_all(&a).unwrap();
299
        std::fs::create_dir_all(&b).unwrap();
300
        std::fs::write(a.join("tool"), "not executable").unwrap();
301
        std::fs::write(b.join("tool"), "#!/bin/sh\n").unwrap();
302
        std::fs::set_permissions(b.join("tool"), std::fs::Permissions::from_mode(0o755)).unwrap();
303
        std::fs::create_dir_all(a.join("dir")).unwrap();
304
        let path = format!("{}::{}", a.display(), b.display());
305
        assert_eq!(resolve_program("tool", &path), Some(format!("{}/tool", b.display())), "skips a file that is not executable");
306
        assert_eq!(resolve_program("dir", &path), None, "and directories");
307
        assert_eq!(resolve_program("missing", &path), None);
308
        assert_eq!(resolve_program("./x", &path).as_deref(), Some("./x"), "a path is used as given");
309
        assert_eq!(resolve_program("", &path), None);
310
        std::fs::remove_dir_all(&dir).unwrap();
311
    }
312
}