Coverage / crates/flow/src/syscall.rs 3.0%

pub fn environment(sandbox: &Sandbox, job: &Job) -> 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 kernel_release() -> String {

    crate::facts::kernel().unwrap_or_else(|| "unknown".into())

}

    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(())

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

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!(

            })?;

        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(&[], families(false)).map_err(|e| format!("seccomp filters are not available: {e}"))

    match r {

        Ok(()) => 0,

        Err(e) => {

            eprintln!("{e}");

            1

}

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,

                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");

        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();

    };

    let mut peak_memory_mib = None;

        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);

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

        duration_ms: t(&started),

        environment,

        timed_out,

        cancelled,

        peak_memory_mib,

        pruned: Vec::new(),

}

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),

        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())),

}

            .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()));

        }

        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 })

    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)

    }

    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()))