OXYGENHUB
# ea81796bb757 oxygenflow / crates / flow-init / src / main.rs
11.1 KiB · Rust · last changed Allow-listed egress over vsock: network "egress" { allow "host"... } 20 days 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
mod egress;
19
mod job;
20
use job::{dev_links, job_env, loopback_up, resolve_program, PATH};
21
22
#[derive(Deserialize)]
23
struct JobSpec {
24
    argv: Vec<String>,
25
    env: Vec<(String, String)>,
26
    #[serde(default)]
27
    mounts: Vec<MountSpec>,
28
    /// `network "loopback"`: bring up `lo` (the guest has no other device).
29
    #[serde(default)]
30
    loopback: bool,
31
    /// `network "egress"` (oxygenflow#55): `lo` comes up and a proxy on
32
    /// 127.0.0.1:3128 relays over vsock to the runner.
33
    #[serde(default)]
34
    egress: bool,
35
    /// Secrets to write before the job starts (spec §9): each at `path`
36
    /// under `/run/secrets`, 0400 and the job's user's, on a tmpfs.
37
    #[serde(default)]
38
    secrets: Vec<SecretSpec>,
39
}
40
41
#[derive(serde::Deserialize)]
42
struct SecretSpec {
43
    path: String,
44
    value: String,
45
}
46
47
/// An ephemeral mount from the host: a tmpfs of `size_mib`, or the disk at
48
/// `device`, mounted at `target` under the new root.
49
#[derive(serde::Deserialize)]
50
struct MountSpec {
51
    kind: String,
52
    target: String,
53
    size_mib: u32,
54
    #[serde(default)]
55
    device: Option<String>,
56
}
57
58
fn main() {
59
    let console = Arc::new(Mutex::new(std::io::stdout()));
60
    let code = match run(&console) {
61
        Ok(code) => code,
62
        Err(e) => {
63
            frame(&console, b'F', e.as_bytes());
64
            125
65
        }
66
    };
67
    frame(&console, b'X', code.to_string().as_bytes());
68
    unsafe {
69
        libc::sync();
70
        // Firecracker ends the VM when the guest reboots (reboot=k).
71
        libc::reboot(libc::LINUX_REBOOT_CMD_RESTART);
72
    }
73
}
74
75
fn frame(console: &Mutex<std::io::Stdout>, kind: u8, text: &[u8]) {
76
    let mut out = console.lock().unwrap_or_else(|p| p.into_inner());
77
    let mut line = Vec::with_capacity(text.len() + 4);
78
    line.extend_from_slice(&[0x1e, kind, b' ']);
79
    line.extend_from_slice(text);
80
    line.push(b'\n');
81
    let _ = out.write_all(&line);
82
    let _ = out.flush();
83
}
84
85
fn cstr(s: &str) -> Result<CString, String> {
86
    CString::new(s).map_err(|_| format!("{s:?} contains NUL"))
87
}
88
89
fn chown(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
90
    let p = cstr(path)?;
91
    if unsafe { libc::chown(p.as_ptr(), uid, gid) } != 0 {
92
        return Err(format!("chown {path}: {}", std::io::Error::last_os_error()));
93
    }
94
    Ok(())
95
}
96
97
/// Create `path` and its missing parents, each owned by the job's user.
98
fn make_dirs_owned(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {
99
    let mut missing = Vec::new();
100
    let mut p = std::path::Path::new(path);
101
    while !p.exists() {
102
        missing.push(p.to_path_buf());
103
        p = match p.parent() {
104
            Some(parent) => parent,
105
            None => break,
106
        };
107
    }
108
    for d in missing.iter().rev() {
109
        std::fs::create_dir(d).map_err(|e| format!("mkdir {}: {e}", d.display()))?;
110
        chown(&d.to_string_lossy(), uid, gid)?;
111
    }
112
    Ok(())
113
}
114
115
fn mount(src: &str, target: &str, fstype: &str, flags: libc::c_ulong, data: &str) -> Result<(), String> {
116
    let (s, t, f, d) = (cstr(src)?, cstr(target)?, cstr(fstype)?, cstr(data)?);
117
    let rc = unsafe { libc::mount(s.as_ptr(), t.as_ptr(), f.as_ptr(), flags, d.as_ptr() as *const libc::c_void) };
118
    if rc != 0 {
119
        return Err(format!("mount {src} on {target}: {}", std::io::Error::last_os_error()));
120
    }
121
    Ok(())
122
}
123
124
fn run(console: &Arc<Mutex<std::io::Stdout>>) -> Result<i32, String> {
125
    let job: JobSpec = serde_json::from_slice(&std::fs::read("/job.json").map_err(|e| format!("read /job.json: {e}"))?)
126
        .map_err(|e| format!("parse /job.json: {e}"))?;
127
    if job.argv.is_empty() {
128
        return Err("empty argv".into());
129
    }
130
    let nosuid = libc::MS_NOSUID | libc::MS_NODEV;
131
    std::fs::create_dir_all("/dev").ok();
132
    mount("devtmpfs", "/dev", "devtmpfs", libc::MS_NOSUID, "")?;
133
    std::fs::create_dir_all("/newroot").map_err(|e| e.to_string())?;
134
    mount("/dev/vda", "/newroot", "ext4", libc::MS_RDONLY | libc::MS_NODEV, "")?;
135
    mount("/dev/vdb", "/newroot/src", "ext4", nosuid, "")?;
136
    mount("tmpfs", "/newroot/tmp", "tmpfs", nosuid, "mode=1777")?;
137
    mount("proc", "/newroot/proc", "proc", nosuid | libc::MS_NOEXEC, "")?;
138
    mount("devtmpfs", "/newroot/dev", "devtmpfs", libc::MS_NOSUID, "")?;
139
    dev_links("/newroot/dev")?;
140
    if std::path::Path::new("/newroot/run").is_dir() {
141
        mount("tmpfs", "/newroot/run", "tmpfs", nosuid, "mode=755")?;
142
    }
143
    // The job runs as whoever owns the workspace (the host's builder), never root.
144
    let (uid, gid) = {
145
        let mut st: libc::stat = unsafe { std::mem::zeroed() };
146
        let p = cstr("/newroot/src")?;
147
        if unsafe { libc::stat(p.as_ptr(), &mut st) } != 0 {
148
            return Err("stat /src".into());
149
        }
150
        if st.st_uid == 0 {
151
            (1000, 1000)
152
        } else {
153
            (st.st_uid, st.st_gid)
154
        }
155
    };
156
    // Ephemeral mounts: mount points are made (and owned by the job's user)
157
    // under the writable /src or /tmp, then each is mounted and given to it.
158
    for m in &job.mounts {
159
        let target = format!("/newroot{}", m.target);
160
        make_dirs_owned(&target, uid, gid)?;
161
        match (m.kind.as_str(), &m.device) {
162
            ("tmpfs", _) => mount("tmpfs", &target, "tmpfs", nosuid, &format!("size={}m,uid={uid},gid={gid},mode=755", m.size_mib))?,
163
            ("disk", Some(dev)) => {
164
                mount(dev, &target, "ext4", nosuid, "")?;
165
                chown(&target, uid, gid)?;
166
            }
167
            (kind, _) => return Err(format!("unknown mount kind {kind}")),
168
        }
169
    }
170
    // Secrets: an in-memory mount of their own, only the job's user's,
171
    // each file 0400; the job file they came from is outside the new root.
172
    if !job.secrets.is_empty() {
173
        if !std::path::Path::new("/newroot/run").is_dir() {
174
            return Err("the image has no /run, where secrets are mounted".into());
175
        }
176
        let dir = "/newroot/run/secrets";
177
        std::fs::create_dir_all(dir).map_err(|e| format!("mkdir {dir}: {e}"))?;
178
        mount("tmpfs", dir, "tmpfs", nosuid | libc::MS_NOEXEC, &format!("size=1m,uid={uid},gid={gid},mode=700"))?;
179
        for s in &job.secrets {
180
            let Some(rel) = s.path.strip_prefix("/run/secrets/") else {
181
                return Err(format!("secret path {:?} is not under /run/secrets/", s.path));
182
            };
183
            let full = format!("{dir}/{rel}");
184
            if let Some(parent) = std::path::Path::new(&full).parent().and_then(|p| p.to_str()) {
185
                make_dirs_owned(parent, uid, gid)?;
186
            }
187
            std::fs::write(&full, s.value.as_bytes()).map_err(|e| format!("write {}: {e}", s.path))?;
188
            chown(&full, uid, gid)?;
189
            std::fs::set_permissions(&full, std::os::unix::fs::PermissionsExt::from_mode(0o400))
190
                .map_err(|e| format!("chmod {}: {e}", s.path))?;
191
        }
192
    }
193
    if job.loopback || job.egress {
194
        loopback_up()?;
195
    }
196
    if job.egress {
197
        egress::start()?;
198
    }
199
    let root = cstr("/newroot")?;
200
    if unsafe { libc::chroot(root.as_ptr()) } != 0 || std::env::set_current_dir("/src").is_err() {
201
        return Err(format!("enter the rootfs: {}", std::io::Error::last_os_error()));
202
    }
203
204
    let argv: Vec<CString> = job.argv.iter().map(|a| cstr(a)).collect::<Result<_, _>>()?;
205
    // glibc's execvpe searches the caller's PATH, and PID 1 has none of its
206
    // own, so the program is found here, in the job's PATH, before forking.
207
    let vars = job_env(std::fs::read_to_string("/etc/flow/env").ok().as_deref(), &job.env);
208
    let path = vars.get("PATH").map(String::as_str).unwrap_or(PATH);
209
    let program = cstr(&resolve_program(&job.argv[0], path).ok_or_else(|| format!("cannot run {:?}: not found in {path}", job.argv[0]))?)?;
210
    let env: Vec<CString> = vars.iter().map(|(k, v)| cstr(&format!("{k}={v}"))).collect::<Result<_, _>>()?;
211
    let mut out = [0; 2];
212
    let mut err = [0; 2];
213
    if unsafe { libc::pipe2(out.as_mut_ptr(), libc::O_CLOEXEC) } != 0 || unsafe { libc::pipe2(err.as_mut_ptr(), libc::O_CLOEXEC) } != 0 {
214
        return Err("pipe".into());
215
    }
216
    let pid = unsafe { libc::fork() };
217
    if pid < 0 {
218
        return Err("fork".into());
219
    }
220
    if pid == 0 {
221
        // Child: no way back to root, then the program itself (execvpe: no shell).
222
        unsafe {
223
            let null = libc::open(c"/dev/null".as_ptr(), libc::O_RDONLY);
224
            libc::dup2(null, 0);
225
            libc::dup2(out[1], 1);
226
            libc::dup2(err[1], 2);
227
            libc::setsid();
228
            if libc::setgroups(0, std::ptr::null()) != 0 || libc::setgid(gid) != 0 || libc::setuid(uid) != 0 {
229
                libc::_exit(126);
230
            }
231
            let mut a: Vec<*const libc::c_char> = argv.iter().map(|c| c.as_ptr()).collect();
232
            a.push(std::ptr::null());
233
            let mut e: Vec<*const libc::c_char> = env.iter().map(|c| c.as_ptr()).collect();
234
            e.push(std::ptr::null());
235
            libc::execve(program.as_ptr(), a.as_ptr(), e.as_ptr());
236
            let msg = format!("flow: cannot run {:?}: {}\n", job.argv[0], std::io::Error::last_os_error());
237
            libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());
238
            libc::_exit(127);
239
        }
240
    }
241
    unsafe {
242
        libc::close(out[1]);
243
        libc::close(err[1]);
244
    }
245
    let readers: Vec<_> = [(out[0], b'O'), (err[0], b'E')]
246
        .into_iter()
247
        .map(|(fd, kind)| {
248
            let console = console.clone();
249
            std::thread::spawn(move || pump(unsafe { std::fs::File::from_raw_fd(fd) }, kind, &console))
250
        })
251
        .collect();
252
    let mut status = 0;
253
    // Reap everything (we are PID 1) until the job itself exits.
254
    loop {
255
        let r = unsafe { libc::waitpid(-1, &mut status, 0) };
256
        if r == pid || r < 0 {
257
            break;
258
        }
259
    }
260
    for r in readers {
261
        let _ = r.join();
262
    }
263
    Ok(if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) })
264
}
265
266
/// Frame a stream line by line; a long line is cut at 16 KiB.
267
fn pump(mut f: std::fs::File, kind: u8, console: &Mutex<std::io::Stdout>) {
268
    let mut buf = [0u8; 8192];
269
    let mut line: Vec<u8> = Vec::new();
270
    loop {
271
        let n = match f.read(&mut buf) {
272
            Ok(0) | Err(_) => break,
273
            Ok(n) => n,
274
        };
275
        for &b in &buf[..n] {
276
            if b == b'\n' || line.len() >= 16 * 1024 {
277
                frame(console, kind, &line);
278
                line.clear();
279
                if b == b'\n' {
280
                    continue;
281
                }
282
            }
283
            line.push(b);
284
        }
285
    }
286
    if !line.is_empty() {
287
        frame(console, kind, &line);
288
    }
289
}