The syscall tier: jobs without KVM ea56c56eaf90

//! What a job's sandbox needs from inside, in either tier: its

//! environment, its program found in its own `PATH`, the usual `/dev`

//! links and `lo`. Compiled into `flow-init` (tier 2, the microVM's PID 1)

//! and into `flow` itself (tier 1, the syscall sandbox), so both start a

//! job the same way.

pub const PATH: &str = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";

/// The file `program` names: itself when it holds a `/`, otherwise the first

/// executable regular file of that name in `path`'s directories.

pub fn resolve_program(program: &str, path: &str) -> Option<String> {

    use std::os::unix::fs::PermissionsExt;

    if program.is_empty() {

        return None;

    }

    if program.contains('/') {

        return Some(program.to_string());

    }

    path.split(':')

        .filter(|d| !d.is_empty())

        .map(|d| format!("{d}/{program}"))

        .find(|p| std::fs::metadata(p).is_ok_and(|m| m.is_file() && m.permissions().mode() & 0o111 != 0))

}

/// A job's environment: `PATH` and `HOME=/tmp` by default, then the

/// image's own (`/etc/flow/env`, written by `flow rootfs import --env`),

/// then the job's (its sandbox's and its own, already merged by the host).

/// Later ones win, so a pipeline can always set `PATH`.

pub fn job_env(image: Option<&str>, job: &[(String, String)]) -> std::collections::BTreeMap<String, String> {

    let mut env = std::collections::BTreeMap::new();

    env.insert("PATH".to_string(), PATH.to_string());

    env.insert("HOME".to_string(), "/tmp".to_string());

    for line in image.unwrap_or("").lines() {

        if let Some((k, v)) = line.split_once('=') {

            if !k.is_empty() {

                env.insert(k.to_string(), v.to_string());

            }

        }

    }

    for (k, v) in job {

        env.insert(k.clone(), v.clone());

    }

    env

}

/// The links every Linux system has under /dev and devtmpfs does not

/// make: `/dev/fd` (bash's process substitution, `<(…)`) and the standard

/// streams, all through the job's own /proc.

pub fn dev_links(dev: &str) -> Result<(), String> {

    for (name, target) in DEV_LINKS {

        let link = format!("{dev}/{name}");

        if std::fs::symlink_metadata(&link).is_err() {

            std::os::unix::fs::symlink(target, &link).map_err(|e| format!("link {link}: {e}"))?;

        }

    }

    Ok(())

}

const DEV_LINKS: [(&str, &str); 4] =

    [("fd", "/proc/self/fd"), ("stdin", "/proc/self/fd/0"), ("stdout", "/proc/self/fd/1"), ("stderr", "/proc/self/fd/2")];

/// Set `lo` up; the kernel gives it 127.0.0.1 and ::1 itself.

pub fn loopback_up() -> Result<(), String> {

    let fd = unsafe { libc::socket(libc::AF_INET, libc::SOCK_DGRAM | libc::SOCK_CLOEXEC, 0) };

    if fd < 0 {

        return Err(format!("socket for lo: {}", std::io::Error::last_os_error()));

    }

    let mut req: libc::ifreq = unsafe { std::mem::zeroed() };

    for (i, b) in b"lo".iter().enumerate() {

        req.ifr_name[i] = *b as libc::c_char;

    }

    let ok = unsafe {

        libc::ioctl(fd, libc::SIOCGIFFLAGS, &mut req) == 0 && {

            req.ifr_ifru.ifru_flags |= libc::IFF_UP as libc::c_short;

            libc::ioctl(fd, libc::SIOCSIFFLAGS, &req) == 0

        }

    };

    let err = std::io::Error::last_os_error();

    unsafe { libc::close(fd) };

    if !ok {

        return Err(format!("bring up lo: {err}"));

    }

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::{dev_links, job_env, resolve_program};

    #[test]

    fn the_image_environment_sits_under_the_jobs() {

        let image = "PATH=/opt/llvm-mos/bin:/usr/bin:/bin\nX16EMU=/opt/x16emu\nnot a line\n";

        let env = job_env(Some(image), &[("X16EMU".into(), "/job".into()), ("LANG".into(), "C".into())]);

        assert_eq!(env["PATH"], "/opt/llvm-mos/bin:/usr/bin:/bin", "the image's PATH replaces the default");

        assert_eq!(env["X16EMU"], "/job", "the job's value wins");

        assert_eq!(env["HOME"], "/tmp");

        assert_eq!(env["LANG"], "C");

        assert_eq!(job_env(None, &[])["PATH"], super::PATH);

    }

    #[test]

    fn dev_gets_the_usual_links_once() {

        let dir = std::env::temp_dir().join(format!("flow-init-dev-{}", std::process::id()));

        let _ = std::fs::remove_dir_all(&dir);

        std::fs::create_dir_all(&dir).unwrap();

        let d = dir.to_str().unwrap();

        dev_links(d).unwrap();

        dev_links(d).unwrap();

        assert_eq!(std::fs::read_link(dir.join("fd")).unwrap(), std::path::Path::new("/proc/self/fd"));

        assert_eq!(std::fs::read_link(dir.join("stdout")).unwrap(), std::path::Path::new("/proc/self/fd/1"));

        // Through the host's /proc, the link works as in a job.

        assert!(dir.join("fd").join("0").exists() || !std::path::Path::new("/proc/self/fd/0").exists());

        std::fs::remove_dir_all(&dir).unwrap();

    }

    use std::os::unix::fs::PermissionsExt;

    #[test]

    fn programs_are_found_in_the_jobs_path() {

        let dir = std::env::temp_dir().join(format!("flow-init-test-{}", std::process::id()));

        let (a, b) = (dir.join("a"), dir.join("b"));

        std::fs::create_dir_all(&a).unwrap();

        std::fs::create_dir_all(&b).unwrap();

        std::fs::write(a.join("tool"), "not executable").unwrap();

        std::fs::write(b.join("tool"), "#!/bin/sh\n").unwrap();

        std::fs::set_permissions(b.join("tool"), std::fs::Permissions::from_mode(0o755)).unwrap();

        std::fs::create_dir_all(a.join("dir")).unwrap();

        let path = format!("{}::{}", a.display(), b.display());

        assert_eq!(resolve_program("tool", &path), Some(format!("{}/tool", b.display())), "skips a file that is not executable");

        assert_eq!(resolve_program("dir", &path), None, "and directories");

        assert_eq!(resolve_program("missing", &path), None);

        assert_eq!(resolve_program("./x", &path).as_deref(), Some("./x"), "a path is used as given");

        assert_eq!(resolve_program("", &path), None);

        std::fs::remove_dir_all(&dir).unwrap();

    }

}

//! Tier 1 (spec §4.2, docs/m0.md §3): the syscall sandbox, for machines

//! without hardware virtualisation. The job runs in new user, mount, PID,

//! network, IPC and UTS namespaces, on the rootfs image's tree (unpacked

//! once per hash, read-only) with the workspace directory at `/src` and a

//! tmpfs at `/tmp`. Landlock limits writes to `/src`, `/tmp` and `/dev`; a

//! seccomp filter refuses sockets other than `AF_UNIX` and the calls that

//! reach past the sandbox (`ptrace`, `mount`, `bpf`, `perf_event_open`,

//! keys, modules, io_uring); and the sandbox's CPU and memory limits apply

//! through a cgroup v2 group of the runner's own delegated cgroup. Without

//! that delegation the tier is not offered, so no job runs unbounded.

//!

//! The runner starts `flow __sandbox <spec>` (this same binary, through

//! `/proc/self/exe`) for each job. That helper joins the job's cgroup,

//! makes the namespaces and forks the sandbox's PID 1, which builds the new

//! root, pivots into it and runs the job as an unprivileged user, the way

//! `flow-init` does in a microVM. Records, receipts and artifacts are the

//! same as tier 2's.

use crate::guest::{dev_links, job_env, loopback_up, resolve_program, PATH};

use crate::home::Home;

use crate::microvm::{Outcome, Sink};

use anyhow::{bail, Context, Result};

use flow_core::api::{Record, SecretInput};

use flow_core::{Environment, Job, MountSource, Network, Sandbox};

use serde::{Deserialize, Serialize};

use std::ffi::CString;

use std::io::{BufRead, BufReader, Read};

use std::path::{Path, PathBuf};

use std::sync::{mpsc, OnceLock};

use std::time::{Duration, Instant};

/// The user and group a job runs as inside the sandbox; the runner's own

/// on the host.

const JOB_ID: u32 = 1000;

/// The receipt environment for a job in this tier: the sandbox is this

/// binary, so its hash stands where tier 2 has Firecracker's and

/// flow-init's, with the host kernel's release.

pub fn environment(sandbox: &Sandbox, job: &Job) -> Environment {

    Environment {

        runtime: "syscall".into(),

        parts: vec![

            ("flow".into(), self_hash().to_string()),

            ("kernel".into(), kernel_release()),

            ("memory_mib".into(), sandbox.memory_mib.to_string()),

            ("cpus".into(), sandbox.cpus.to_string()),

        ]

        .into_iter()

        .chain(job.mounts.iter().map(|m| (format!("mount {}", m.target), format!("{} {} MiB", m.source.as_str(), m.size_mib))))

        .collect(),

        rootfs: sandbox.rootfs.clone(),

        argv: job.exec.clone(),

        env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),

    }

}

fn self_hash() -> &'static str {

    static HASH: OnceLock<String> = OnceLock::new();

    HASH.get_or_init(|| crate::home::hash_file(Path::new("/proc/self/exe")).unwrap_or_else(|_| "unknown".into()))

}

fn kernel_release() -> String {

    std::fs::read_to_string("/proc/sys/kernel/osrelease").map(|s| s.trim().to_string()).unwrap_or_else(|_| "unknown".into())

}

/// Whether this machine can run the tier now: a cgroup this runner may

/// manage, and namespaces, Landlock and seccomp that work for this binary

/// (Ubuntu's AppArmor can deny unprivileged user namespaces to programs

/// without a profile). `Err` says what is missing.

pub fn available() -> std::result::Result<(), String> {

    Cgroups::delegated()?;

    let out = std::process::Command::new("/proc/self/exe")

        .arg("__sandbox-probe")

        .stdin(std::process::Stdio::null())

        .output()

        .map_err(|e| format!("cannot start the probe: {e}"))?;

    if out.status.success() {

        Ok(())

    } else {

        Err(String::from_utf8_lossy(&out.stderr).trim().to_string())

    }

}

/// `flow __sandbox-probe`: try each kernel feature the sandbox uses, in a

/// throwaway process, and say which one failed.

pub fn probe_main() -> i32 {

    let r = (|| -> std::result::Result<(), String> {

        let (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) };

        enter_namespaces(libc::CLONE_NEWUSER | libc::CLONE_NEWNS, uid, gid)

            .map_err(|e| format!("user namespaces are not allowed for this program ({e}); on Ubuntu, kernel.apparmor_restrict_unprivileged_userns=1 denies them"))?;

        mount("none", "/", "", libc::MS_REC | libc::MS_PRIVATE, "")

            .and_then(|_| mount("tmpfs", "/tmp", "tmpfs", libc::MS_NOSUID | libc::MS_NODEV, ""))

            .map_err(|e| {

                format!(

                    "user namespaces give no mount rights here ({e}); on Ubuntu, kernel.apparmor_restrict_unprivileged_userns=1 does this"

                )

            })?;

        let abi = landlock_abi();

        if abi < 1 {

            return Err("Landlock is not enabled in this kernel (it must be in the lsm= list)".into());

        }

        no_new_privs()?;

        seccomp(&[]).map_err(|e| format!("seccomp filters are not available: {e}"))

    })();

    match r {

        Ok(()) => 0,

        Err(e) => {

            eprintln!("{e}");

            1

        }

    }

}

/// What the helper reads from the spec file the runner wrote (0600, in

/// the job's scratch directory, which the job never sees); it removes it

/// once read.

#[derive(Serialize, Deserialize)]

struct Spec {

    /// The image's unpacked tree.

    rootfs: PathBuf,

    /// An empty directory to build the new root on.

    root: PathBuf,

    workspace: PathBuf,

    /// The job's cgroup, joined before anything else.

    cgroup: PathBuf,

    argv: Vec<String>,

    env: Vec<(String, String)>,

    mounts: Vec<MountSpec>,

    loopback: bool,

    secrets: Vec<(String, String)>,

}

#[derive(Serialize, Deserialize)]

struct MountSpec {

    target: String,

    size_mib: u32,

    /// A host directory bound at `target` (`ephemeral://disk`); a tmpfs

    /// of `size_mib` when there is none.

    dir: Option<PathBuf>,

}

/// Run `job` with `workspace` (a directory) as `/src`. `scratch` is a

/// directory this call may fill; `name` names the job's cgroup.

#[allow(clippy::too_many_arguments)]

pub fn run(

    home: &Home,

    sandbox: &Sandbox,

    job: &Job,

    secrets: &[SecretInput],

    workspace: &Path,

    scratch: &Path,

    timeout: Duration,

    sink: &mut dyn Sink,

    name: &str,

) -> Result<Outcome> {

    let cgroups = Cgroups::delegated().map_err(|e| anyhow::anyhow!("the syscall tier cannot run here: {e}"))?;

    let image = home.rootfs(&sandbox.rootfs);

    if !image.is_file() {

        bail!("rootfs {} is not on this runner (import it with `flow rootfs import`)", sandbox.rootfs);

    }

    let image = home.verified(&image, &sandbox.rootfs)?;

    let rootfs = unpacked(home, &sandbox.rootfs, &image)?;

    let environment = environment(sandbox, job);

    let mut mounts = Vec::new();

    for (i, m) in job.mounts.iter().enumerate() {

        let dir = match m.source {

            MountSource::Tmpfs => None,

            MountSource::Disk => {

                let d = disk_dir(scratch, job.mounts.iter().take(i).filter(|m| m.source == MountSource::Disk).count());

                std::fs::create_dir_all(&d)?;

                Some(d)

            }

        };

        mounts.push(MountSpec { target: m.target.clone(), size_mib: m.size_mib, dir });

    }

    let root = scratch.join("root");

    std::fs::create_dir_all(&root)?;

    let group = cgroups.job(name, sandbox.memory_mib, sandbox.cpus)?;

    let spec = Spec {

        rootfs,

        root,

        workspace: workspace.to_path_buf(),

        cgroup: group.dir.clone(),

        argv: job.exec.clone(),

        env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),

        mounts,

        loopback: sandbox.network == Network::Loopback,

        secrets: secrets.iter().map(|s| (s.path.clone(), s.value.clone())).collect(),

    };

    let spec_path = scratch.join("sandbox.json");

    {

        use std::io::Write;

        use std::os::unix::fs::OpenOptionsExt;

        let mut f = std::fs::OpenOptions::new().write(true).create_new(true).mode(0o600).open(&spec_path)?;

        f.write_all(&serde_json::to_vec(&spec)?)?;

    }

    let started = Instant::now();

    let mut child = std::process::Command::new("/proc/self/exe")

        .arg("__sandbox")

        .arg(&spec_path)

        .env_clear()

        .current_dir(scratch)

        .stdin(std::process::Stdio::null())

        .stdout(std::process::Stdio::piped())

        .stderr(std::process::Stdio::piped())

        .spawn()

        .context("start the sandbox")?;

    let (tx, rx) = mpsc::channel::<(bool, Vec<u8>)>();

    let out = child.stdout.take().unwrap();

    let err = child.stderr.take().unwrap();

    let tx2 = tx.clone();

    let readers = [std::thread::spawn(move || lines(out, true, tx)), std::thread::spawn(move || lines(err, false, tx2))];

    let mut failed_to_start = false;

    let mut timed_out = false;

    let mut cancelled = false;

    let deadline = started + timeout;

    let t = |s: &Instant| s.elapsed().as_millis() as u64;

    let stop = |child: &mut std::process::Child| {

        let _ = child.kill();

        group.kill();

    };

    loop {

        match rx.recv_timeout(Duration::from_millis(200)) {

            Ok((stdout, line)) => {

                let text = String::from_utf8_lossy(&line).into_owned();

                let rec = |stream: &str, text: &str| Record { t: t(&started), stream: stream.into(), text: text.into() };

                match text.strip_prefix(FAILED) {

                    Some(why) if !stdout => {

                        failed_to_start = true;

                        sink.record(rec("flow", &format!("the sandbox failed to start the job: {why}")))

                    }

                    _ => sink.record(rec(if stdout { "out" } else { "err" }, &text)),

                }

            }

            Err(mpsc::RecvTimeoutError::Timeout) => {}

            Err(mpsc::RecvTimeoutError::Disconnected) => break,

        }

        if !timed_out && !cancelled && sink.cancelled() {

            cancelled = true;

            stop(&mut child);

            sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });

        }

        if Instant::now() >= deadline && !timed_out && !cancelled {

            timed_out = true;

            stop(&mut child);

            sink.record(Record {

                t: t(&started),

                stream: "flow".into(),

                text: format!("stopped at the time limit of {} s", timeout.as_secs()),

            });

        }

    }

    let status = child.wait()?;

    for r in readers {

        let _ = r.join();

    }

    if group.out_of_memory() {

        sink.record(Record {

            t: t(&started),

            stream: "flow".into(),

            text: format!("the job ran out of memory: the sandbox allows {} MiB", sandbox.memory_mib),

        });

    }

    let exit_code = status.code();

    if exit_code.is_none() && !timed_out && !cancelled && !failed_to_start {

        sink.record(Record {

            t: t(&started),

            stream: "flow".into(),

            text: format!("the sandbox ended without the job's exit code ({status})"),

        });

    }

    drop(group);

    let _ = std::fs::remove_file(&spec_path);

    Ok(Outcome {

        exit_code: if timed_out || cancelled { None } else { exit_code },

        duration_ms: t(&started),

        environment,

        timed_out,

        cancelled,

    })

}

/// Where the `n`th `ephemeral://disk` mount of a job keeps its files.

pub fn disk_dir(scratch: &Path, n: usize) -> PathBuf {

    scratch.join(format!("mount-{n}"))

}

/// How the helper says the job never started, on its stderr.

const FAILED: &str = "\u{1e}F ";

fn lines(r: impl Read, stdout: bool, tx: mpsc::Sender<(bool, Vec<u8>)>) {

    let mut r = BufReader::new(r);

    let mut buf = Vec::new();

    loop {

        buf.clear();

        // A long line is cut at 16 KiB, as flow-init cuts it.

        match (&mut r).take(16 * 1024).read_until(b'\n', &mut buf) {

            Ok(0) | Err(_) => break,

            Ok(_) => {

                if buf.last() == Some(&b'\n') {

                    buf.pop();

                }

                if tx.send((stdout, buf.clone())).is_err() {

                    break;

                }

            }

        }

    }

}

/// The image's tree, unpacked from its ext4 file once per hash with

/// debugfs (no mount rights needed). Files belong to the runner; set-id

/// bits and device nodes are dropped, which a sandboxed job cannot use

/// anyway.

fn unpacked(home: &Home, hash: &str, image: &Path) -> Result<PathBuf> {

    let trees = home.root.join("trees");

    let dir = trees.join(hash);

    if dir.is_dir() {

        return Ok(dir);

    }

    std::fs::create_dir_all(&trees)?;

    let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.subsec_nanos()).unwrap_or(0);

    let tmp = trees.join(format!(".{hash}.{}.{nanos}", std::process::id()));

    std::fs::create_dir_all(&tmp)?;

    let out = std::process::Command::new("debugfs")

        .arg("-R")

        .arg(format!("rdump / \"{}\"", tmp.display()))

        .arg(image)

        .output()

        .context("run debugfs (e2fsprogs) to unpack the rootfs")?;

    let empty = std::fs::read_dir(&tmp).map(|mut d| d.next().is_none()).unwrap_or(true);

    if !out.status.success() || empty {

        let _ = crate::workspace::remove_tree(&tmp);

        bail!("unpacking rootfs {}: {}", &hash[..12.min(hash.len())], String::from_utf8_lossy(&out.stderr).trim());

    }

    match std::fs::rename(&tmp, &dir) {

        Ok(()) => Ok(dir),

        // Another job unpacked it first.

        Err(_) if dir.is_dir() => {

            let _ = crate::workspace::remove_tree(&tmp);

            Ok(dir)

        }

        Err(e) => Err(e).with_context(|| format!("keep the unpacked rootfs at {}", dir.display())),

    }

}

// ---- cgroups ----

/// The runner's own cgroup v2 group, when it may manage it (a systemd

/// service with `Delegate=yes`, or a user service). The runner moves

/// itself into a `flow-main` leaf so the group can hold job groups with

/// the cpu and memory controllers on.

#[derive(Clone)]

pub struct Cgroups {

    base: PathBuf,

}

/// The leaf the runner process moves itself into.

const MAIN_LEAF: &str = "flow-main";

impl Cgroups {

    pub fn delegated() -> std::result::Result<Cgroups, String> {

        static ONCE: OnceLock<std::result::Result<Cgroups, String>> = OnceLock::new();

        ONCE.get_or_init(Self::set_up).clone()

    }

    fn set_up() -> std::result::Result<Cgroups, String> {

        let own = std::fs::read_to_string("/proc/self/cgroup").map_err(|e| format!("read /proc/self/cgroup: {e}"))?;

        let rel = own

            .lines()

            .find_map(|l| l.strip_prefix("0::"))

            .ok_or("this machine does not use cgroup v2 (the unified hierarchy)")?

            .trim()

            .to_string();

        let mut base = Path::new("/sys/fs/cgroup").join(rel.trim_start_matches('/'));

        if base.file_name().is_some_and(|n| n == MAIN_LEAF) {

            base.pop();

        }

        let controllers = std::fs::read_to_string(base.join("cgroup.controllers")).unwrap_or_default();

        let has = |c: &str| controllers.split_whitespace().any(|x| x == c);

        let delegate_hint = "run the runner as a systemd service with Delegate=yes, or as a user service; for `flow run`, use `systemd-run --user --scope -p Delegate=yes flow run …`";

        if !has("cpu") || !has("memory") {

            return Err(format!("the cgroup {} has no cpu and memory controllers to give jobs; {delegate_hint}", base.display()));

        }

        let writable =

            |p: &Path| CString::new(p.as_os_str().as_encoded_bytes()).is_ok_and(|c| unsafe { libc::access(c.as_ptr(), libc::W_OK) } == 0);

        if !writable(&base) || !writable(&base.join("cgroup.subtree_control")) || !writable(&base.join("cgroup.procs")) {

            return Err(format!("this runner may not manage its cgroup {}; {delegate_hint}", base.display()));

        }

        // Only this process may be in the group: anything else would block

        // the controllers, and nothing is changed for a shared group.

        let me = std::process::id().to_string();

        let others = std::fs::read_to_string(base.join("cgroup.procs")).unwrap_or_default();

        if others.lines().any(|p| p.trim() != me) {

            return Err(format!("the cgroup {} holds other processes; {delegate_hint}", base.display()));

        }

        let leaf = base.join(MAIN_LEAF);

        if !leaf.is_dir() {

            std::fs::create_dir(&leaf).map_err(|e| format!("make {}: {e}", leaf.display()))?;

        }

        std::fs::write(leaf.join("cgroup.procs"), "0").map_err(|e| format!("move into {}: {e}", leaf.display()))?;

        let want: Vec<&str> = ["cpu", "memory", "pids"].into_iter().filter(|c| has(c)).collect();

        let enable = want.iter().map(|c| format!("+{c}")).collect::<Vec<_>>().join(" ");

        std::fs::write(base.join("cgroup.subtree_control"), &enable)

            .map_err(|e| format!("enable {enable} in {}: {e} (another process may still be in it; {delegate_hint})", base.display()))?;

        Ok(Cgroups { base })

    }

    /// A fresh group for one job, with its limits set.

    pub fn job(&self, name: &str, memory_mib: u32, cpus: u32) -> Result<JobGroup> {

        let dir = self.base.join(name);

        if dir.is_dir() {

            // Left by a runner that stopped mid-job.

            drop(JobGroup { dir: dir.clone() });

        }

        std::fs::create_dir(&dir).with_context(|| format!("make the job's cgroup {}", dir.display()))?;

        let group = JobGroup { dir };

        let set = |file: &str, value: String| {

            std::fs::write(group.dir.join(file), &value).with_context(|| format!("set {file} to {value} in {}", group.dir.display()))

        };

        set("memory.max", (u64::from(memory_mib) << 20).to_string())?;

        if group.dir.join("memory.swap.max").exists() {

            set("memory.swap.max", "0".into())?;

        }

        set("cpu.max", format!("{} 100000", u64::from(cpus.max(1)) * 100_000))?;

        Ok(group)

    }

}

/// One job's cgroup; dropping it kills what is left in it and removes it.

pub struct JobGroup {

    dir: PathBuf,

}

impl JobGroup {

    fn kill(&self) {

        let _ = std::fs::write(self.dir.join("cgroup.kill"), "1");

    }

    fn out_of_memory(&self) -> bool {

        std::fs::read_to_string(self.dir.join("memory.events"))

            .ok()

            .and_then(|s| s.lines().find_map(|l| l.strip_prefix("oom_kill ")).and_then(|n| n.trim().parse::<u64>().ok()))

            .is_some_and(|n| n > 0)

    }

}

impl Drop for JobGroup {

    fn drop(&mut self) {

        self.kill();

        for _ in 0..100 {

            if std::fs::remove_dir(&self.dir).is_ok() || !self.dir.exists() {

                return;

            }

            std::thread::sleep(Duration::from_millis(50));

        }

        eprintln!("flow: could not remove the cgroup {}", self.dir.display());

    }

}

// ---- the helper: `flow __sandbox <spec>` ----

/// The helper's entry point; returns the job's exit code (125 when the

/// sandbox could not start it, after a `FAILED` line on stderr).

pub fn sandbox_main(spec_path: &Path) -> i32 {

    match helper(spec_path) {

        Ok(code) => code,

        Err(e) => {

            eprintln!("{FAILED}{e}");

            125

        }

    }

}

fn helper(spec_path: &Path) -> std::result::Result<i32, String> {

    let spec: Spec = serde_json::from_slice(&std::fs::read(spec_path).map_err(|e| format!("read the spec: {e}"))?)

        .map_err(|e| format!("parse the spec: {e}"))?;

    let _ = std::fs::remove_file(spec_path);

    if spec.argv.is_empty() {

        return Err("empty argv".into());

    }

    std::fs::write(spec.cgroup.join("cgroup.procs"), "0").map_err(|e| format!("join the job's cgroup: {e}"))?;

    let (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) };

    enter_namespaces(

        libc::CLONE_NEWUSER | libc::CLONE_NEWNS | libc::CLONE_NEWPID | libc::CLONE_NEWNET | libc::CLONE_NEWIPC | libc::CLONE_NEWUTS,

        uid,

        gid,

    )?;

    // The first child is PID 1 of the new PID namespace; everything in it

    // dies with it, and it dies with this helper.

    let pid = unsafe { libc::fork() };

    if pid < 0 {

        return Err(format!("fork: {}", std::io::Error::last_os_error()));

    }

    if pid == 0 {

        let code = match init(&spec) {

            Ok(code) => code,

            Err(e) => {

                eprintln!("{FAILED}{e}");

                125

            }

        };

        unsafe { libc::_exit(code) };

    }

    Ok(wait_for(pid))

}

/// Wait for `pid`, reaping any other child on the way: its exit code, or

/// 128 + the signal that killed it.

fn wait_for(pid: libc::pid_t) -> i32 {

    let mut status = 0;

    loop {

        let r = unsafe { libc::waitpid(-1, &mut status, 0) };

        if r == pid {

            break;

        }

        if r < 0 && std::io::Error::last_os_error().raw_os_error() != Some(libc::EINTR) {

            return 125;

        }

    }

    if libc::WIFEXITED(status) {

        libc::WEXITSTATUS(status)

    } else {

        128 + libc::WTERMSIG(status)

    }

}

/// New namespaces of `flags` (which include a user namespace) where

/// [`JOB_ID`] is the caller's `uid` and `gid`.

fn enter_namespaces(flags: libc::c_int, uid: u32, gid: u32) -> std::result::Result<(), String> {

    if unsafe { libc::unshare(flags) } != 0 {

        return Err(format!("unshare: {}", std::io::Error::last_os_error()));

    }

    let write = |file: &str, body: String| std::fs::write(file, body).map_err(|e| format!("write {file}: {e}"));

    write("/proc/self/setgroups", "deny".into())?;

    write("/proc/self/uid_map", format!("{JOB_ID} {uid} 1\n"))?;

    write("/proc/self/gid_map", format!("{JOB_ID} {gid} 1\n"))

}

/// PID 1 of the sandbox: build the root, enter it, run the job, wait.

fn init(spec: &Spec) -> std::result::Result<i32, String> {

    // Dies with the helper; the job's cgroup is killed at the end as well.

    unsafe { libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) };

    let nosuid = libc::MS_NOSUID | libc::MS_NODEV;

    mount("none", "/", "", libc::MS_REC | libc::MS_PRIVATE, "")?;

    let root = spec.root.to_str().ok_or("the root path is not UTF-8")?;

    let at = |p: &str| format!("{root}{p}");

    bind(spec.rootfs.to_str().ok_or("the rootfs path is not UTF-8")?, root)?;

    for d in ["/src", "/tmp", "/proc", "/dev"] {

        if !Path::new(&at(d)).is_dir() {

            return Err(format!("the image has no {d} directory"));

        }

    }

    remount_bind(root, libc::MS_RDONLY | nosuid)?;

    bind(spec.workspace.to_str().ok_or("the workspace path is not UTF-8")?, &at("/src"))?;

    mount("tmpfs", &at("/tmp"), "tmpfs", nosuid, "mode=1777")?;

    mount("proc", &at("/proc"), "proc", nosuid | libc::MS_NOEXEC, "")?;

    // A /dev of the harmless devices, bound from the host's.

    mount("tmpfs", &at("/dev"), "tmpfs", libc::MS_NOSUID | libc::MS_NOEXEC, "mode=755")?;

    for d in ["null", "zero", "full", "random", "urandom"] {

        let target = at(&format!("/dev/{d}"));

        std::fs::File::create(&target).map_err(|e| format!("make {target}: {e}"))?;

        bind(&format!("/dev/{d}"), &target)?;

    }

    dev_links(&at("/dev"))?;

    std::fs::create_dir(at("/dev/shm")).map_err(|e| format!("make /dev/shm: {e}"))?;

    mount("tmpfs", &at("/dev/shm"), "tmpfs", nosuid, "mode=1777")?;

    if Path::new(&at("/run")).is_dir() {

        mount("tmpfs", &at("/run"), "tmpfs", nosuid, "mode=755")?;

    }

    // Ephemeral mounts: made under the writable /src or /tmp.

    for m in &spec.mounts {

        let target = at(&m.target);

        std::fs::create_dir_all(&target).map_err(|e| format!("make {}: {e}", m.target))?;

        match &m.dir {

            Some(dir) => bind(dir.to_str().ok_or("a mount path is not UTF-8")?, &target)?,

            None => mount("tmpfs", &target, "tmpfs", nosuid, &format!("size={}m,mode=755", m.size_mib))?,

        }

    }

    // Secrets: an in-memory mount of their own, each file 0400.

    if !spec.secrets.is_empty() {

        if !Path::new(&at("/run")).is_dir() {

            return Err("the image has no /run, where secrets are mounted".into());

        }

        let dir = at("/run/secrets");

        std::fs::create_dir_all(&dir).map_err(|e| format!("mkdir {dir}: {e}"))?;

        mount("tmpfs", &dir, "tmpfs", nosuid | libc::MS_NOEXEC, "size=1m,mode=700")?;

        for (path, value) in &spec.secrets {

            let Some(rel) = path.strip_prefix("/run/secrets/") else {

                return Err(format!("secret path {path:?} is not under /run/secrets/"));

            };

            let full = format!("{dir}/{rel}");

            if let Some(parent) = Path::new(&full).parent() {

                std::fs::create_dir_all(parent).map_err(|e| format!("mkdir for {path}: {e}"))?;

            }

            std::fs::write(&full, value.as_bytes()).map_err(|e| format!("write {path}: {e}"))?;

            std::fs::set_permissions(&full, std::os::unix::fs::PermissionsExt::from_mode(0o400))

                .map_err(|e| format!("chmod {path}: {e}"))?;

        }

    }

    let host = b"flow";

    unsafe { libc::sethostname(host.as_ptr() as *const libc::c_char, host.len()) };

    if spec.loopback {

        loopback_up()?;

    }

    // Into the new root; the old one is detached, so the host's files, the

    // spec and the scratch directory are out of reach.

    std::env::set_current_dir(root).map_err(|e| format!("cd {root}: {e}"))?;

    let dot = c".";

    if unsafe { libc::syscall(libc::SYS_pivot_root, dot.as_ptr(), dot.as_ptr()) } != 0 {

        return Err(format!("pivot_root: {}", std::io::Error::last_os_error()));

    }

    if unsafe { libc::umount2(dot.as_ptr(), libc::MNT_DETACH) } != 0 {

        return Err(format!("detach the old root: {}", std::io::Error::last_os_error()));

    }

    std::env::set_current_dir("/src").map_err(|e| format!("cd /src: {e}"))?;

    let vars = job_env(std::fs::read_to_string("/etc/flow/env").ok().as_deref(), &spec.env);

    let path = vars.get("PATH").map(String::as_str).unwrap_or(PATH);

    let program =

        cstr(&resolve_program(&spec.argv[0], path).ok_or_else(|| format!("cannot run {:?}: not found in {path}", spec.argv[0]))?)?;

    let argv: Vec<CString> = spec.argv.iter().map(|a| cstr(a)).collect::<Result<_, _>>()?;

    let env: Vec<CString> = vars.iter().map(|(k, v)| cstr(&format!("{k}={v}"))).collect::<Result<_, _>>()?;

    let mut writable: Vec<String> = ["/src", "/tmp", "/dev"].into_iter().map(String::from).collect();

    writable.extend(spec.mounts.iter().map(|m| m.target.clone()));

    let pid = unsafe { libc::fork() };

    if pid < 0 {

        return Err(format!("fork: {}", std::io::Error::last_os_error()));

    }

    if pid == 0 {

        // The job: no new privileges, then the walls, then the program.

        let confined = (|| {

            unsafe {

                let null = libc::open(c"/dev/null".as_ptr(), libc::O_RDONLY);

                libc::dup2(null, 0);

                libc::setsid();

            }

            no_new_privs()?;

            landlock(&writable)?;

            seccomp(DENIED)

        })();

        unsafe {

            if let Err(e) = confined {

                let msg = format!("{FAILED}{e}\n");

                libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());

                libc::_exit(125);

            }

            let mut a: Vec<*const libc::c_char> = argv.iter().map(|c| c.as_ptr()).collect();

            a.push(std::ptr::null());

            let mut e: Vec<*const libc::c_char> = env.iter().map(|c| c.as_ptr()).collect();

            e.push(std::ptr::null());

            libc::execve(program.as_ptr(), a.as_ptr(), e.as_ptr());

            let msg = format!("flow: cannot run {:?}: {}\n", spec.argv[0], std::io::Error::last_os_error());

            libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());

            libc::_exit(127);

        }

    }

    Ok(wait_for(pid))

}

fn cstr(s: &str) -> std::result::Result<CString, String> {

    CString::new(s).map_err(|_| format!("{s:?} contains NUL"))

}

fn mount(src: &str, target: &str, fstype: &str, flags: libc::c_ulong, data: &str) -> std::result::Result<(), String> {

    let (s, t, f, d) = (cstr(src)?, cstr(target)?, cstr(fstype)?, cstr(data)?);

    let fstype = if fstype.is_empty() { std::ptr::null() } else { f.as_ptr() };

    let data = if data.is_empty() { std::ptr::null() } else { d.as_ptr() as *const libc::c_void };

    if unsafe { libc::mount(s.as_ptr(), t.as_ptr(), fstype, flags, data) } != 0 {

        return Err(format!("mount {src} on {target}: {}", std::io::Error::last_os_error()));

    }

    Ok(())

}

fn bind(src: &str, target: &str) -> std::result::Result<(), String> {

    mount(src, target, "", libc::MS_BIND | libc::MS_REC, "")

}

/// Remount a bind mount with `flags` added. In a user namespace the

/// flags the mount already had are locked, so they are kept.

fn remount_bind(target: &str, flags: libc::c_ulong) -> std::result::Result<(), String> {

    let t = cstr(target)?;

    let mut st: libc::statvfs = unsafe { std::mem::zeroed() };

    if unsafe { libc::statvfs(t.as_ptr(), &mut st) } != 0 {

        return Err(format!("statvfs {target}: {}", std::io::Error::last_os_error()));

    }

    let kept = [

        (libc::ST_NOSUID, libc::MS_NOSUID),

        (libc::ST_NODEV, libc::MS_NODEV),

        (libc::ST_NOEXEC, libc::MS_NOEXEC),

        (libc::ST_NOATIME, libc::MS_NOATIME),

        (libc::ST_NODIRATIME, libc::MS_NODIRATIME),

        (libc::ST_RELATIME, libc::MS_RELATIME),

    ]

    .into_iter()

    .filter(|(st_flag, _)| st.f_flag & *st_flag != 0)

    .fold(0, |acc, (_, ms)| acc | ms);

    mount("none", target, "", libc::MS_BIND | libc::MS_REMOUNT | flags | kept, "")

}

fn no_new_privs() -> std::result::Result<(), String> {

    if unsafe { libc::prctl(libc::PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) } != 0 {

        return Err(format!("no_new_privs: {}", std::io::Error::last_os_error()));

    }

    Ok(())

}

// ---- Landlock ----

const LANDLOCK_CREATE_RULESET_VERSION: libc::c_uint = 1;

const LANDLOCK_RULE_PATH_BENEATH: libc::c_int = 1;

const ACCESS_EXECUTE: u64 = 1;

const ACCESS_READ_FILE: u64 = 1 << 2;

const ACCESS_READ_DIR: u64 = 1 << 3;

/// Every right of Landlock ABI 1 (execute .. make_sym).

const ACCESS_ABI1: u64 = (1 << 13) - 1;

const ACCESS_REFER: u64 = 1 << 13;

const ACCESS_TRUNCATE: u64 = 1 << 14;

#[repr(C)]

struct RulesetAttr {

    handled_access_fs: u64,

}

#[repr(C, packed)]

struct PathBeneath {

    allowed_access: u64,

    parent_fd: i32,

}

/// The kernel's Landlock ABI version; below 1 when there is none.

fn landlock_abi() -> i64 {

    unsafe { libc::syscall(libc::SYS_landlock_create_ruleset, std::ptr::null::<RulesetAttr>(), 0usize, LANDLOCK_CREATE_RULESET_VERSION) }

}

/// Read and execute everywhere; write only beneath `writable`.

fn landlock(writable: &[String]) -> std::result::Result<(), String> {

    let abi = landlock_abi();

    if abi < 1 {

        return Err("Landlock is not available".into());

    }

    let mut all = ACCESS_ABI1;

    if abi >= 2 {

        all |= ACCESS_REFER;

    }

    if abi >= 3 {

        all |= ACCESS_TRUNCATE;

    }

    let attr = RulesetAttr { handled_access_fs: all };

    let ruleset = unsafe { libc::syscall(libc::SYS_landlock_create_ruleset, &attr, std::mem::size_of::<RulesetAttr>(), 0) };

    if ruleset < 0 {

        return Err(format!("landlock ruleset: {}", std::io::Error::last_os_error()));

    }

    let ruleset = ruleset as libc::c_int;

    let rule = |path: &str, access: u64| -> std::result::Result<(), String> {

        let p = cstr(path)?;

        let fd = unsafe { libc::open(p.as_ptr(), libc::O_PATH | libc::O_CLOEXEC) };

        if fd < 0 {

            return Err(format!("open {path} for Landlock: {}", std::io::Error::last_os_error()));

        }

        let beneath = PathBeneath { allowed_access: access, parent_fd: fd };

        let rc = unsafe { libc::syscall(libc::SYS_landlock_add_rule, ruleset, LANDLOCK_RULE_PATH_BENEATH, &beneath, 0) };

        unsafe { libc::close(fd) };

        if rc != 0 {

            return Err(format!("landlock rule for {path}: {}", std::io::Error::last_os_error()));

        }

        Ok(())

    };

    rule("/", ACCESS_EXECUTE | ACCESS_READ_FILE | ACCESS_READ_DIR)?;

    for w in writable {

        rule(w, all)?;

    }

    if unsafe { libc::syscall(libc::SYS_landlock_restrict_self, ruleset, 0) } != 0 {

        return Err(format!("landlock restrict: {}", std::io::Error::last_os_error()));

    }

    unsafe { libc::close(ruleset) };

    Ok(())

}

// ---- seccomp ----

#[cfg(target_arch = "x86_64")]

const AUDIT_ARCH: u32 = 0xC000_003E;

#[cfg(target_arch = "aarch64")]

const AUDIT_ARCH: u32 = 0xC000_00B7;

#[cfg(target_arch = "riscv64")]

const AUDIT_ARCH: u32 = 0xC000_00F3;

/// Calls a job never makes: they reach past the sandbox (tracing, mounts,

/// kernel programs and keys, modules, kexec, swap, reboot) or around the

/// socket check (io_uring).

const DENIED: &[libc::c_long] = &[

    libc::SYS_ptrace,

    libc::SYS_process_vm_readv,

    libc::SYS_process_vm_writev,

    libc::SYS_mount,

    libc::SYS_umount2,

    libc::SYS_pivot_root,

    libc::SYS_open_tree,

    libc::SYS_move_mount,

    libc::SYS_fsopen,

    libc::SYS_fsconfig,

    libc::SYS_fsmount,

    libc::SYS_fspick,

    libc::SYS_bpf,

    libc::SYS_perf_event_open,

    libc::SYS_keyctl,

    libc::SYS_add_key,

    libc::SYS_request_key,

    libc::SYS_init_module,

    libc::SYS_finit_module,

    libc::SYS_delete_module,

    libc::SYS_kexec_load,

    libc::SYS_swapon,

    libc::SYS_swapoff,

    libc::SYS_reboot,

    libc::SYS_io_uring_setup,

    libc::SYS_io_uring_enter,

    libc::SYS_io_uring_register,

];

/// Install a filter that fails each of `denied` with EPERM and `socket`

/// for any family but `AF_UNIX` with EAFNOSUPPORT (an empty `denied`

/// still checks sockets). Another architecture's calls kill the process.

fn seccomp(denied: &[libc::c_long]) -> std::result::Result<(), String> {

    const LD: u16 = (libc::BPF_LD | libc::BPF_W | libc::BPF_ABS) as u16;

    const JEQ: u16 = (libc::BPF_JMP | libc::BPF_JEQ | libc::BPF_K) as u16;

    const RET: u16 = (libc::BPF_RET | libc::BPF_K) as u16;

    let op = |code: u16, k: u32, jt: u8, jf: u8| libc::sock_filter { code, jt, jf, k };

    let errno = |e: libc::c_int| libc::SECCOMP_RET_ERRNO | (e as u32 & libc::SECCOMP_RET_DATA);

    // struct seccomp_data: nr at 0, arch at 4, args[0] at 16 (its low

    // half on these little-endian machines).

    let mut f = vec![op(LD, 4, 0, 0), op(JEQ, AUDIT_ARCH, 1, 0), op(RET, libc::SECCOMP_RET_KILL_PROCESS, 0, 0), op(LD, 0, 0, 0)];

    #[cfg(target_arch = "x86_64")]

    {

        // x32 calls carry this bit; there are none in a 64-bit job.

        const JGE: u16 = (libc::BPF_JMP | libc::BPF_JGE | libc::BPF_K) as u16;

        f.push(op(JGE, 0x4000_0000, 0, 1));

        f.push(op(RET, errno(libc::EPERM), 0, 0));

    }

    for nr in denied {

        f.push(op(JEQ, *nr as u32, 0, 1));

        f.push(op(RET, errno(libc::EPERM), 0, 0));

    }

    f.push(op(JEQ, libc::SYS_socket as u32, 0, 3));

    f.push(op(LD, 16, 0, 0));

    f.push(op(JEQ, libc::AF_UNIX as u32, 1, 0));

    f.push(op(RET, errno(libc::EAFNOSUPPORT), 0, 0));

    f.push(op(RET, libc::SECCOMP_RET_ALLOW, 0, 0));

    let prog = libc::sock_fprog { len: f.len() as u16, filter: f.as_mut_ptr() };

    if unsafe { libc::syscall(libc::SYS_seccomp, libc::SECCOMP_SET_MODE_FILTER, 0, &prog) } != 0 {

        return Err(format!("seccomp: {}", std::io::Error::last_os_error()));

    }

    Ok(())

}

//! Tier 1 conformance (docs/m0.md §3): a pipeline on a busybox rootfs in

//! the syscall sandbox. Runs where `flow tiers` offers the syscall tier

//! (user namespaces, Landlock, seccomp and a delegated cgroup: each `flow`

//! runs in a delegated systemd user scope when one can be made) and a static

//! busybox is at /usr/bin/busybox; skipped otherwise.

use std::path::Path;

use std::process::Command;

const APPLETS: &[&str] = &[

    "sh", "echo", "cat", "id", "touch", "wget", "tail", "cut", "tr", "true", "false", "ls", "grep", "wc", "mount", "hostname", "head",

    "sed",

];

/// `flow`, in a delegated cgroup scope of its own when systemd's user

/// manager can make one (the test itself shares cargo's).

fn flow(home: &Path) -> Command {

    static SCOPE: std::sync::OnceLock<bool> = std::sync::OnceLock::new();

    let scope = *SCOPE

        .get_or_init(|| Command::new("systemd-run").args(["--user", "--scope", "-q", "true"]).output().is_ok_and(|o| o.status.success()));

    let mut c = if scope {

        let mut c = Command::new("systemd-run");

        c.args(["--user", "--scope", "-q", "-p", "Delegate=yes", env!("CARGO_BIN_EXE_flow")]);

        c

    } else {

        Command::new(env!("CARGO_BIN_EXE_flow"))

    };

    c.arg("--home").arg(home);

    c

}

#[test]

fn a_pipeline_runs_confined_in_the_syscall_sandbox() {

    let busybox = Path::new("/usr/bin/busybox");

    let dir = tempfile::tempdir().unwrap();

    let home = dir.path().join("home");

    let tiers = flow(&home).arg("tiers").output().unwrap();

    let tiers = String::from_utf8_lossy(&tiers.stdout).into_owned();

    if !tiers.contains("syscall: available") || !busybox.is_file() {

        eprintln!("skipped: the syscall tier is not available here or there is no static busybox\n{tiers}");

        return;

    }

    // A rootfs of busybox and its applets.

    let tree = dir.path().join("tree");

    for d in ["bin", "src", "tmp", "proc", "dev", "etc", "run"] {

        std::fs::create_dir_all(tree.join(d)).unwrap();

    }

    std::fs::copy(busybox, tree.join("bin/busybox")).unwrap();

    for a in APPLETS {

        std::os::unix::fs::symlink("busybox", tree.join("bin").join(a)).unwrap();

    }

    let tarball = dir.path().join("rootfs.tar");

    assert!(Command::new("tar").arg("-cf").arg(&tarball).arg("-C").arg(&tree).arg(".").status().unwrap().success());

    let import = flow(&home).args(["rootfs", "import"]).arg(&tarball).output().unwrap();

    let said = String::from_utf8_lossy(&import.stdout).to_string() + &String::from_utf8_lossy(&import.stderr);

    assert!(import.status.success(), "{said}");

    let rootfs = said.split(|c: char| !c.is_ascii_hexdigit()).find(|w| w.len() == 64).expect("the import names the hash").to_string();

    let work = dir.path().join("work");

    std::fs::create_dir_all(work.join(".flow")).unwrap();

    std::fs::write(

        work.join(".flow/pipeline.kdl"),

        format!(

            r#"schema-version 1

sandbox "bb" {{

    runtime "syscall"

    rootfs "cas://blob/{rootfs}"

    memory-limit "256MiB"

}}

job "literal" {{

    sandbox "bb"

    exec "echo" "a; touch /src/pwned && $(id) | `id`"

}}

job "isolated" {{

    sandbox "bb"

    exec "sh" "-c" "cat /proc/net/dev | tail -n +3 | cut -d: -f1 | tr -d ' '; wget -q -T 2 -O /dev/null http://1.1.1.1 && echo NETWORK || echo no-network"

}}

job "readonly" {{

    sandbox "bb"

    exec "sh" "-c" "touch /etc/x 2>/dev/null && echo ROOTFS-WRITABLE; touch /run/x 2>/dev/null && echo RUN-WRITABLE; echo ok > /src/w && cat /src/w; id -u; hostname"

}}

job "walls" {{

    sandbox "bb"

    exec "sh" "-c" "mount -t tmpfs t /tmp 2>/dev/null && echo MOUNTED; ls /proc | grep -c '^[0-9]' | sed 's/^/procs /'; ls / | tr '\n' ' '; echo; echo done"

}}

job "greedy" {{

    sandbox "bb"

    exec "sh" "-c" "x=$(head -c 400000000 /dev/zero | tr '\\0' a); echo SURVIVED"

}}

job "fails" {{

    sandbox "bb"

    exec "false"

}}

job "after-failure" {{

    sandbox "bb"

    depends-on "fails"

    exec "echo" "SHOULD-NOT-RUN"

}}

"#

        ),

    )

    .unwrap();

    let out = flow(&home).arg("run").current_dir(&work).output().unwrap();

    let stdout = String::from_utf8_lossy(&out.stdout);

    let stderr = String::from_utf8_lossy(&out.stderr);

    let all = format!("{stdout}\n{stderr}");

    assert!(!out.status.success(), "a failed job fails the run: {all}");

    assert!(stdout.contains("   a; touch /src/pwned && $(id) | `id`\n"), "argv is literal: {all}");

    assert!(!work.join("pwned").exists());

    assert!(stdout.contains("   lo\n") && stdout.contains("no-network") && !stdout.contains("   NETWORK\n"), "only loopback: {all}");

    assert!(stdout.contains("   ok\n") && !stdout.contains("   ROOTFS-WRITABLE\n"), "rootfs read-only, workspace writable: {all}");

    assert!(stdout.contains("   1000\n") && !stdout.contains("   0\n"), "the job never runs as root: {all}");

    assert!(stdout.contains("   flow\n"), "its own host name: {all}");

    assert!(!stdout.contains("   MOUNTED\n") && stdout.contains("   done\n"), "no mounts: {all}");

    assert!(stderr.contains("Address family not supported"), "sockets other than AF_UNIX are refused: {all}");

    assert!(!work.join("w").exists(), "jobs write to their own copy: {all}");

    let procs: u32 = stdout.lines().find_map(|l| l.trim().strip_prefix("procs ")).and_then(|n| n.parse().ok()).expect(&all);

    assert!(procs < 10, "only the sandbox's processes are visible: {all}");

    assert!(!stdout.contains("home "), "none of the host's files: {all}");

    assert!(!stdout.contains("   RUN-WRITABLE\n"), "Landlock keeps a writable mount outside /src and /tmp read-only: {all}");

    assert!(stdout.contains("── greedy failed") && !stdout.contains("   SURVIVED\n"), "{all}");

    assert!(stderr.contains("the job ran out of memory: the sandbox allows 256 MiB"), "{all}");

    assert!(stdout.contains("── fails failed (exit 1)"), "{all}");

    assert!(stdout.contains("── after-failure skipped: fails failed") && !stdout.contains("   SHOULD-NOT-RUN\n"), "{all}");

}

- **The syscall tier (#36, docs/m0.md §3).** `runtime "syscall"` jobs run without KVM, so boards such as the OrangePi RV2 can take them. The sandbox uses user, mount, PID, network, IPC and UTS namespaces on the image's unpacked tree, read-only. Landlock limits writes to `/src` and `/tmp`, and a seccomp filter refuses network sockets, tracing, mounts, BPF, keys, modules and io_uring. A delegated cgroup v2 group enforces the CPU and memory limits, and a job killed for memory is told so. A runner offers the tier only where all of this works. `flow tiers` shows which tiers a machine can run and why the others cannot, and `flow run` runs syscall jobs too.

mod job;

use job::{dev_links, job_env, loopback_up, resolve_program, PATH};

/// The job-starting code `flow-init` uses in a microVM, shared with the

/// syscall tier.

#[path = "../../flow-init/src/job.rs"]

mod guest;

mod syscall;