//! 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(),
    }
}

pub(crate) 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 {
    crate::facts::kernel().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();
    };
    // The sandbox helper's peak resident memory (oxygenflow#41): the job's
    // own processes are its children, so this is a floor, not the whole.
    let mut peak_memory_mib = None;
    loop {
        if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
            peak_memory_mib = Some(m);
        }
        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,
        peak_memory_mib,
        pruned: Vec::new(),
    })
}

/// 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.
pub(crate) const FAILED: &str = "\u{1e}F ";

pub(crate) 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,
];

/// Classic BPF opcodes (linux/bpf_common.h), fixed by the kernel's ABI;
/// spelled out because libc does not export them for every target
/// (Android, where the runner is built for the WASM tier only).
const BPF_LD: u16 = 0x00;
const BPF_W: u16 = 0x00;
const BPF_ABS: u16 = 0x20;
const BPF_JMP: u16 = 0x05;
const BPF_JEQ: u16 = 0x10;
const BPF_K: u16 = 0x00;
const BPF_RET: u16 = 0x06;

/// 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 = BPF_LD | BPF_W | BPF_ABS;
    const JEQ: u16 = BPF_JMP | BPF_JEQ | BPF_K;
    const RET: u16 = BPF_RET | BPF_K;
    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(())
}

#[cfg(all(test, not(target_os = "android")))]
mod tests {
    /// The opcodes written out above are the ABI's, as glibc's libc has them.
    #[test]
    fn bpf_opcodes_are_the_kernels() {
        use super::*;
        assert_eq!(
            [BPF_LD, BPF_W, BPF_ABS, BPF_JMP, BPF_JEQ, BPF_K, BPF_RET],
            [libc::BPF_LD, libc::BPF_W, libc::BPF_ABS, libc::BPF_JMP, libc::BPF_JEQ, libc::BPF_K, libc::BPF_RET].map(|v| v as u16)
        );
    }
}