//! 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"))