crates/flow/src/main.rs at e38bb032e38b · andymccall/oxygenflow

# e38bb032e38b oxygenflow / crates / flow / src / main.rs

22.3 KiB · Rust · last changed flow rootfs push and pull use the login oh already saved 24 days ago source blame history raw

//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
//! (docs/spec.md, docs/m0.md).

mod cpio;
/// The job-starting code `flow-init` uses in a microVM, shared with the
/// syscall tier.
#[path = "../../flow-init/src/job.rs"]
mod guest;
mod home;
mod images;
mod microvm;
mod outputs;
mod runner;
mod syscall;
mod wait;
mod workspace;

use anyhow::{bail, Context, Result};
use clap::{Parser, Subcommand};
use flow_core::api::Record;
use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
use home::Home;
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::time::Duration;

#[derive(Parser)]
#[command(name = "flow", version = env!("OXY_VERSION_LONG"), about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
struct Cli {
    /// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
    #[arg(long, global = true)]
    home: Option<PathBuf>,
    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum RunnerCmd {
    /// Register this runner with a hub using a one-time token from its Admin → Runners page
    Register {
        /// The hub, e.g. https://earth.example:7343
        hub: String,
        /// The registration token (read from stdin when not given)
        #[arg(long)]
        token: Option<String>,
        /// The runner's name (default: the host name)
        #[arg(long)]
        name: Option<String>,
        /// The hub's CA certificate, when self-signed
        #[arg(long)]
        ca: Option<PathBuf>,
        /// A label the runner offers (repeatable)
        #[arg(long)]
        label: Vec<String>,
    },
}

#[derive(Subcommand)]
enum Cmd {
    /// Validate a pipeline and print its jobs and hash
    Check {
        /// The pipeline file (default: .flow/pipeline.kdl)
        file: Option<PathBuf>,
    },
    /// Print a file's BLAKE3, for pinning in flow.toml
    Hash { file: PathBuf },
    /// Rootfs images for sandboxes
    Rootfs {
        #[command(subcommand)]
        cmd: RootfsCmd,
    },
    /// Run the pipeline on this directory's files, locally
    Run {
        /// Only these jobs (and what they depend on)
        jobs: Vec<String>,
        /// The directory to build (default: the current one)
        #[arg(long, default_value = ".")]
        dir: PathBuf,
        /// Minutes before a job is stopped
        #[arg(long, default_value_t = 60)]
        timeout: u64,
        /// A variable for every job, as the hub's settings would give it;
        /// the pipeline's own `env` wins
        #[arg(long = "var", value_name = "NAME=VALUE")]
        vars: Vec<String>,
        /// A secret a job's `secret` block names, as the hub would supply
        /// it; mounted as a file, never in the environment
        #[arg(long = "secret", value_name = "NAME=VALUE")]
        secrets: Vec<String>,
    },
    /// Claim and run jobs for the hub in flow.toml's [runner] (or the
    /// runner.toml `flow runner register` wrote)
    Runner {
        #[command(subcommand)]
        cmd: Option<RunnerCmd>,
    },
    /// Follow the checks of a pushed state on OxygenHub until they finish:
    /// exit 0 when all passed, 1 when one failed, 2 at the time limit.
    /// Uses $OXY_TOKEN and $OXY_CA as oxy does.
    Wait {
        /// The repository's URL on the hub (https://hub/owner/name)
        url: String,
        /// The state hash the push moved the bookmark to (`oxy push --json` prints it)
        state: String,
        /// Minutes to wait at most
        #[arg(long, default_value_t = 60)]
        timeout: u64,
    },
    /// Show which isolation tiers this machine can run jobs in, and why
    /// the others cannot
    Tiers,
    /// The syscall tier's sandbox helper, started by the runner for each job
    #[command(name = "__sandbox", hide = true)]
    Sandbox { spec: PathBuf },
    /// Check that this machine can run the syscall tier
    #[command(name = "__sandbox-probe", hide = true)]
    SandboxProbe,
}

#[derive(Subcommand)]
enum RootfsCmd {
    /// Make an ext4 rootfs image from a root file system tarball and store it by hash
    Import {
        tarball: PathBuf,
        /// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
        #[arg(long)]
        sha256: Option<String>,
        /// The image's own environment, NAME=value (repeatable): what a
        /// Dockerfile's ENV sets and `docker export` loses. Jobs start from
        /// it; their sandbox's and their own `env` override it
        #[arg(long = "env", value_name = "NAME=VALUE")]
        env: Vec<String>,
    },
    /// List the images this home holds
    List,
    /// Send an image to a hub, so its runners fetch it instead of having it
    /// copied to them: only the chunks the hub lacks travel. Signs in as
    /// `oh` is signed in to that hub, or with $OXY_TOKEN and $OXY_CA
    Push {
        /// The image's hash, as pipelines pin it (cas://blob/HASH)
        hash: String,
        /// The hub (default: $OH_HUB, else the runner's, from runner.toml or flow.toml)
        #[arg(long)]
        hub: Option<String>,
    },
    /// Fetch an image a hub holds into this home, as a runner does for a
    /// job whose image it lacks. Signs in as `oh` is, or with $OXY_TOKEN
    Pull {
        hash: String,
        #[arg(long)]
        hub: Option<String>,
    },
}

fn main() {
    let cli = Cli::parse();
    if let Err(e) = run(cli) {
        eprintln!("flow: {e:#}");
        std::process::exit(1);
    }
}

fn run(cli: Cli) -> Result<()> {
    match cli.cmd {
        Cmd::Sandbox { spec } => std::process::exit(syscall::sandbox_main(&spec)),
        Cmd::SandboxProbe => std::process::exit(syscall::probe_main()),
        Cmd::Check { file } => check(&file.unwrap_or_else(|| PathBuf::from(PIPELINE_PATH))),
        Cmd::Hash { file } => {
            println!("{}", home::hash_file(&file)?);
            Ok(())
        }
        Cmd::Rootfs { cmd } => {
            let home = Home::locate(cli.home.as_deref())?;
            match cmd {
                RootfsCmd::Import { tarball, sha256, env } => rootfs_import(&home, &tarball, sha256.as_deref(), &env),
                RootfsCmd::Push { hash, hub } => {
                    let client = image_client(&home, hub)?;
                    eprintln!("Sending image {} …", &hash[..12.min(hash.len())]);
                    let m = client.push(&home, &hash)?;
                    println!(
                        "Sent image {}: {} chunks the hub lacked, {} in {:.1} s",
                        &hash[..12],
                        m.chunks,
                        images::human(m.bytes),
                        m.elapsed.as_secs_f64()
                    );
                    Ok(())
                }
                RootfsCmd::Pull { hash, hub } => {
                    let client = image_client(&home, hub)?;
                    match client.fetch(&home, &hash)? {
                        Some(m) => {
                            println!(
                                "Fetched image {}: {} chunks, {} in {:.1} s",
                                &hash[..12],
                                m.chunks,
                                images::human(m.bytes),
                                m.elapsed.as_secs_f64()
                            );
                            Ok(())
                        }
                        None => anyhow::bail!("the hub does not hold image {}; send it with flow rootfs push", &hash[..12.min(hash.len())]),
                    }
                }
                RootfsCmd::List => {
                    for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
                        let size = e.metadata().map(|m| m.len()).unwrap_or(0);
                        if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
                            println!("cas://blob/{h}  {} MiB", size >> 20);
                        }
                    }
                    Ok(())
                }
            }
        }
        Cmd::Run { jobs, dir, timeout, vars, secrets } => {
            let home = Home::locate(cli.home.as_deref())?;
            let pairs = |what: &str, list: &[String]| {
                list.iter()
                    .map(|e| {
                        e.split_once('=')
                            .map(|(k, v)| (k.to_string(), v.to_string()))
                            .ok_or_else(|| anyhow::anyhow!("--{what} {e:?}: expected NAME=VALUE"))
                    })
                    .collect::<Result<Vec<_>>>()
            };
            let (vars, secrets) = (pairs("var", &vars)?, pairs("secret", &secrets)?);
            run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60), &vars, &secrets)
        }
        Cmd::Runner { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {
            let home = Home::locate(cli.home.as_deref())?;
            if std::env::var_os("OXY_HOME").is_none() {
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
            }
            let name = name.unwrap_or_else(|| {
                std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_else(|_| "flow".into())
            });
            let token = match token {
                Some(t) => t,
                None => {
                    eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");
                    let mut line = String::new();
                    std::io::stdin().read_line(&mut line)?;
                    line.trim().to_string()
                }
            };
            let got = runner::register(&home, &hub, &token, &name, ca, label)?;
            println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);
            Ok(())
        }
        Cmd::Runner { cmd: None } => {
            let home = Home::locate(cli.home.as_deref())?;
            // The runner's oxy state (identity, user config) stays in its home.
            if std::env::var_os("OXY_HOME").is_none() {
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
            }
            runner::serve(&home)
        }
        Cmd::Tiers => {
            let home = Home::locate(cli.home.as_deref())?;
            match runner::tiers(&home) {
                Ok((have, missing)) => {
                    for t in have {
                        println!("{t}: available");
                    }
                    for why in missing {
                        println!("{}", why.replacen(": ", ": unavailable: ", 1));
                    }
                }
                Err(why) => println!("{why}"),
            }
            Ok(())
        }
        Cmd::Wait { url, state, timeout } => {
            let token = std::env::var("OXY_TOKEN").ok();
            let ca = std::env::var_os("OXY_CA").map(PathBuf::from);
            let code =
                wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;
            std::process::exit(code);
        }
    }
}

/// The hub image commands talk to: `--hub`, else $OH_HUB, else the
/// runner's. The token: $OXY_TOKEN, else the login `oh auth login` (or
/// `oxy auth login`) saved for that hub in Oxygen's credential store
/// (oxygenflow#44), else the runner's own.
fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
    let runner = home.config.runner.clone();
    let hub = hub
        .or_else(|| std::env::var("OH_HUB").ok().filter(|h| !h.is_empty()))
        .or_else(|| runner.as_ref().map(|r| r.hub.clone()))
        .context("name the hub with --hub https://HUB (or set OH_HUB)")?;
    let (saved_token, saved_ca) = oxy_core::credentials::Credentials::user().lookup(&hub);
    let token = std::env::var("OXY_TOKEN")
        .ok()
        .filter(|t| !t.is_empty())
        .or(saved_token)
        .or_else(|| runner.as_ref().map(|r| r.token.clone()))
        .with_context(|| format!("no token for {hub}: sign in with `oh auth login --hub {hub}`, or set OXY_TOKEN"))?;
    let ca = std::env::var_os("OXY_CA").map(PathBuf::from).or(saved_ca).or_else(|| runner.as_ref().and_then(|r| r.ca.clone()));
    images::Client::new(&hub, &token, ca.as_deref())
}

fn load(file: &Path) -> Result<Pipeline> {
    let src = std::fs::read_to_string(file).with_context(|| format!("read {}", file.display()))?;
    Pipeline::parse(&src).map_err(|p| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
}

fn check(file: &Path) -> Result<()> {
    let p = load(file)?;
    let depths = p.depths();
    for j in p.order() {
        let sb = &p.sandboxes[&j.sandbox];
        let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
        println!("{}{}  {} [{}]{after}", "  ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
    }
    println!("pipeline {}", hex::encode(p.hash()));
    Ok(())
}

/// The image environment file flow-init reads (`/etc/flow/env`): one
/// `NAME=value` per line, names checked as a pipeline's are.
fn image_env_file(env: &[String]) -> Result<String> {
    let mut out = String::new();
    for e in env {
        let Some((k, v)) = e.split_once('=') else { bail!("--env {e:?}: expected NAME=value") };
        let ok = !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
        if !ok || v.contains('\n') || v.contains('\0') {
            bail!("--env {e:?}: a name is letters, digits and _, and a value is one line");
        }
        out.push_str(&format!("{k}={v}\n"));
    }
    Ok(out)
}

/// The architecture an ELF file was built for, by its header.
fn elf_arch(path: &Path) -> Option<&'static str> {
    let bytes = std::fs::read(path).ok()?;
    if bytes.len() < 20 || &bytes[..4] != b"\x7fELF" {
        return None;
    }
    let little = bytes[5] == 1;
    let machine = if little { u16::from_le_bytes([bytes[18], bytes[19]]) } else { u16::from_be_bytes([bytes[18], bytes[19]]) };
    Some(match machine {
        0x3E => "x86_64",
        0xB7 => "aarch64",
        0xF3 => "riscv64",
        0x03 => "x86",
        0x28 => "arm",
        0x15 => "powerpc64",
        0x16 => "s390x",
        _ => return None,
    })
}

/// The architecture of an unpacked rootfs: its shell's, or the first
/// program's found under bin.
fn tree_arch(tree: &Path) -> Option<&'static str> {
    for rel in ["bin/sh", "bin/busybox", "usr/bin/env", "bin/bash", "usr/bin/sh"] {
        let p = tree.join(rel);
        let p = std::fs::canonicalize(&p).unwrap_or(p);
        if let Some(a) = elf_arch(&p) {
            return Some(a);
        }
    }
    None
}

fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>, env: &[String]) -> Result<()> {
    let env_file = image_env_file(env)?;
    if let Some(want) = sha256 {
        let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
        let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
        if got != want.to_lowercase() {
            bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
        }
    }
    let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&tmp);
    let tree = tmp.join("tree");
    std::fs::create_dir_all(&tree)?;
    let result = (|| -> Result<String> {
        let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
        if !status.success() {
            bail!("tar could not unpack {}", tarball.display());
        }
        // The image's architecture must be this machine's: there is no
        // emulation tier (spec §11.2).
        let arch = tree_arch(&tree).unwrap_or("unknown");
        if arch != "unknown" && arch != std::env::consts::ARCH {
            bail!(
                "{} is a {arch} image; this runner is {}, and jobs here run no other architecture",
                tarball.display(),
                std::env::consts::ARCH
            );
        }
        // Mount points flow-init needs (docs/m0.md §2).
        for d in ["src", "tmp", "proc", "dev", "run"] {
            std::fs::create_dir_all(tree.join(d))?;
        }
        if !env_file.is_empty() {
            std::fs::create_dir_all(tree.join("etc/flow"))?;
            std::fs::write(tree.join("etc/flow/env"), &env_file)?;
        }
        let used: u64 = ignore::WalkBuilder::new(&tree)
            .hidden(false)
            .ignore(false)
            .git_ignore(false)
            .build()
            .flatten()
            .filter_map(|e| e.metadata().ok())
            .map(|m| m.len().max(4096))
            .sum();
        let mib = used * 13 / 10 / (1 << 20) + 64;
        let img = tmp.join("rootfs.ext4");
        let status = std::process::Command::new("mkfs.ext4")
            .args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
            .arg(&tree)
            .arg(&img)
            .arg(format!("{mib}M"))
            .status()
            .context("run mkfs.ext4")?;
        if !status.success() {
            bail!("mkfs.ext4 failed");
        }
        let hash = home::hash_file(&img)?;
        std::fs::rename(&img, home.rootfs(&hash))?;
        let manifest = serde_json::json!({ "arch": if arch == "unknown" { std::env::consts::ARCH } else { arch }, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0) });
        std::fs::write(home.image_manifest(&hash), manifest.to_string())?;
        Ok(hash)
    })();
    let _ = std::fs::remove_dir_all(&tmp);
    let hash = result?;
    println!("rootfs \"cas://blob/{hash}\"");
    Ok(())
}

fn run_local(
    home: &Home,
    dir: &Path,
    only: &[String],
    timeout: Duration,
    vars: &[(String, String)],
    secrets: &[(String, String)],
) -> Result<()> {
    let p = load(&dir.join(PIPELINE_PATH))?;
    // The named jobs and everything they depend on.
    let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
    for name in only {
        if !p.jobs.contains_key(name) {
            bail!("no job {name:?} in the pipeline");
        }
    }
    if !only.is_empty() {
        loop {
            let more: Vec<String> = wanted.iter().flat_map(|j| p.jobs[j].depends_on.clone()).filter(|d| !wanted.contains(d)).collect();
            if more.is_empty() {
                break;
            }
            wanted.extend(more);
        }
    }
    let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&scratch);
    std::fs::create_dir_all(scratch.join("tree"))?;
    workspace::copy_tree(dir, &scratch.join("tree"))?;
    let mut failed: BTreeSet<String> = BTreeSet::new();
    let mut worst = 0;
    for job in p.order() {
        if !only.is_empty() && !wanted.contains(&job.name) {
            continue;
        }
        if let Some(d) = job.depends_on.iter().find(|d| failed.contains(*d)) {
            println!("── {} skipped: {d} failed", job.name);
            failed.insert(job.name.clone());
            continue;
        }
        let sb = &p.sandboxes[&job.sandbox];
        if sb.runtime == Runtime::Wasm {
            bail!("job {}: runtime {} is not available in this build yet", job.name, sb.runtime.as_str());
        }
        println!("── {} ({})", job.name, job.exec.join(" "));
        let js = scratch.join(&job.name);
        std::fs::create_dir_all(&js)?;
        let job = &job
            .in_sandbox(sb)
            .with_flow_env(&[("FLOW_JOB".into(), job.name.clone()), ("FLOW_RUN".into(), "local".into())])
            .with_flow_env(vars);
        // Identities come from a hub: a local run has none.
        for i in &job.identities {
            println!("── {}: identity {:?} is not available locally; {} will be missing", job.name, i.audience, i.mount_path);
        }
        // Each secret the job declares must be given, as the hub would.
        let inputs = job
            .secrets
            .iter()
            .map(|s| {
                let value = secrets.iter().find(|(k, _)| *k == s.name).map(|(_, v)| v.clone());
                value
                    .map(|value| flow_core::api::SecretInput { name: s.name.clone(), path: s.mount_path.clone(), value })
                    .ok_or_else(|| anyhow::anyhow!("job {}: secret {:?} is not given; pass --secret {}=VALUE", job.name, s.name, s.name))
            })
            .collect::<Result<Vec<_>>>()?;
        // Each job starts from the same files: jobs do not share writes.
        let out = if sb.runtime == Runtime::Syscall {
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
            let group = format!("local-{}-{}", std::process::id(), js.file_name().unwrap_or_default().to_string_lossy());
            syscall::run(home, sb, job, &inputs, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r), &group)?
        } else {
            workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
            microvm::run(home, sb, job, &inputs, &js.join("workspace.ext4"), &js, timeout, &mut |r: Record| print_record(&r))?
        };
        let _ = workspace::remove_tree(&js);
        match out.exit_code {
            Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
            other => {
                println!(
                    "── {} failed ({}) in {} ms",
                    job.name,
                    other.map(|c| format!("exit {c}")).unwrap_or_else(|| "no exit code".into()),
                    out.duration_ms
                );
                failed.insert(job.name.clone());
                worst = 1;
            }
        }
    }
    let _ = workspace::remove_tree(&scratch);
    if worst != 0 {
        std::process::exit(worst);
    }
    Ok(())
}

fn print_record(r: &Record) {
    match r.stream.as_str() {
        "out" => println!("   {}", r.text),
        "err" => eprintln!("   {}", r.text),
        s => eprintln!("   [{s}] {}", r.text),
    }
}

#[cfg(test)]
mod tests {
    use super::image_env_file;

    #[test]
    fn image_environments_are_checked_lines() {
        let f = image_env_file(&["PATH=/opt/bin:/usr/bin".into(), "X16EMU=/opt/x16emu".into(), "EMPTY=".into()]).unwrap();
        assert_eq!(f, "PATH=/opt/bin:/usr/bin\nX16EMU=/opt/x16emu\nEMPTY=\n");
        assert!(image_env_file(&["NOEQUALS".into()]).is_err());
        assert!(image_env_file(&["1X=y".into()]).is_err());
        assert!(image_env_file(&["X=two\nlines".into()]).is_err());
        assert_eq!(image_env_file(&[]).unwrap(), "");
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1	//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
2	//! (docs/spec.md, docs/m0.md).
3
4	mod cpio;
5	/// The job-starting code `flow-init` uses in a microVM, shared with the
6	/// syscall tier.
7	#[path = "../../flow-init/src/job.rs"]
8	mod guest;
9	mod home;
10	mod images;
11	mod microvm;
12	mod outputs;
13	mod runner;
14	mod syscall;
15	mod wait;
16	mod workspace;
17
18	use anyhow::{bail, Context, Result};
19	use clap::{Parser, Subcommand};
20	use flow_core::api::Record;
21	use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
22	use home::Home;
23	use std::collections::BTreeSet;
24	use std::path::{Path, PathBuf};
25	use std::time::Duration;
26
27	#[derive(Parser)]
28	#[command(name = "flow", version = env!("OXY_VERSION_LONG"), about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
29	struct Cli {
30	/// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
31	#[arg(long, global = true)]
32	home: Option<PathBuf>,
33	#[command(subcommand)]
34	cmd: Cmd,
35	}
36
37	#[derive(Subcommand)]
38	enum RunnerCmd {
39	/// Register this runner with a hub using a one-time token from its Admin → Runners page
40	Register {
41	/// The hub, e.g. https://earth.example:7343
42	hub: String,
43	/// The registration token (read from stdin when not given)
44	#[arg(long)]
45	token: Option<String>,
46	/// The runner's name (default: the host name)
47	#[arg(long)]
48	name: Option<String>,
49	/// The hub's CA certificate, when self-signed
50	#[arg(long)]
51	ca: Option<PathBuf>,
52	/// A label the runner offers (repeatable)
53	#[arg(long)]
54	label: Vec<String>,
55	},
56	}
57
58	#[derive(Subcommand)]
59	enum Cmd {
60	/// Validate a pipeline and print its jobs and hash
61	Check {
62	/// The pipeline file (default: .flow/pipeline.kdl)
63	file: Option<PathBuf>,
64	},
65	/// Print a file's BLAKE3, for pinning in flow.toml
66	Hash { file: PathBuf },
67	/// Rootfs images for sandboxes
68	Rootfs {
69	#[command(subcommand)]
70	cmd: RootfsCmd,
71	},
72	/// Run the pipeline on this directory's files, locally
73	Run {
74	/// Only these jobs (and what they depend on)
75	jobs: Vec<String>,
76	/// The directory to build (default: the current one)
77	#[arg(long, default_value = ".")]
78	dir: PathBuf,
79	/// Minutes before a job is stopped
80	#[arg(long, default_value_t = 60)]
81	timeout: u64,
82	/// A variable for every job, as the hub's settings would give it;
83	/// the pipeline's own `env` wins
84	#[arg(long = "var", value_name = "NAME=VALUE")]
85	vars: Vec<String>,
86	/// A secret a job's `secret` block names, as the hub would supply
87	/// it; mounted as a file, never in the environment
88	#[arg(long = "secret", value_name = "NAME=VALUE")]
89	secrets: Vec<String>,
90	},
91	/// Claim and run jobs for the hub in flow.toml's [runner] (or the
92	/// runner.toml `flow runner register` wrote)
93	Runner {
94	#[command(subcommand)]
95	cmd: Option<RunnerCmd>,
96	},
97	/// Follow the checks of a pushed state on OxygenHub until they finish:
98	/// exit 0 when all passed, 1 when one failed, 2 at the time limit.
99	/// Uses $OXY_TOKEN and $OXY_CA as oxy does.
100	Wait {
101	/// The repository's URL on the hub (https://hub/owner/name)
102	url: String,
103	/// The state hash the push moved the bookmark to (`oxy push --json` prints it)
104	state: String,
105	/// Minutes to wait at most
106	#[arg(long, default_value_t = 60)]
107	timeout: u64,
108	},
109	/// Show which isolation tiers this machine can run jobs in, and why
110	/// the others cannot
111	Tiers,
112	/// The syscall tier's sandbox helper, started by the runner for each job
113	#[command(name = "__sandbox", hide = true)]
114	Sandbox { spec: PathBuf },
115	/// Check that this machine can run the syscall tier
116	#[command(name = "__sandbox-probe", hide = true)]
117	SandboxProbe,
118	}
119
120	#[derive(Subcommand)]
121	enum RootfsCmd {
122	/// Make an ext4 rootfs image from a root file system tarball and store it by hash
123	Import {
124	tarball: PathBuf,
125	/// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
126	#[arg(long)]
127	sha256: Option<String>,
128	/// The image's own environment, NAME=value (repeatable): what a
129	/// Dockerfile's ENV sets and `docker export` loses. Jobs start from
130	/// it; their sandbox's and their own `env` override it
131	#[arg(long = "env", value_name = "NAME=VALUE")]
132	env: Vec<String>,
133	},
134	/// List the images this home holds
135	List,
136	/// Send an image to a hub, so its runners fetch it instead of having it
137	/// copied to them: only the chunks the hub lacks travel. Signs in as
138	/// `oh` is signed in to that hub, or with $OXY_TOKEN and $OXY_CA
139	Push {
140	/// The image's hash, as pipelines pin it (cas://blob/HASH)
141	hash: String,
142	/// The hub (default: $OH_HUB, else the runner's, from runner.toml or flow.toml)
143	#[arg(long)]
144	hub: Option<String>,
145	},
146	/// Fetch an image a hub holds into this home, as a runner does for a
147	/// job whose image it lacks. Signs in as `oh` is, or with $OXY_TOKEN
148	Pull {
149	hash: String,
150	#[arg(long)]
151	hub: Option<String>,
152	},
153	}
154
155	fn main() {
156	let cli = Cli::parse();
157	if let Err(e) = run(cli) {
158	eprintln!("flow: {e:#}");
159	std::process::exit(1);
160	}
161	}
162
163	fn run(cli: Cli) -> Result<()> {
164	match cli.cmd {
165	Cmd::Sandbox { spec } => std::process::exit(syscall::sandbox_main(&spec)),
166	Cmd::SandboxProbe => std::process::exit(syscall::probe_main()),
167	Cmd::Check { file } => check(&file.unwrap_or_else(\|\| PathBuf::from(PIPELINE_PATH))),
168	Cmd::Hash { file } => {
169	println!("{}", home::hash_file(&file)?);
170	Ok(())
171	}
172	Cmd::Rootfs { cmd } => {
173	let home = Home::locate(cli.home.as_deref())?;
174	match cmd {
175	RootfsCmd::Import { tarball, sha256, env } => rootfs_import(&home, &tarball, sha256.as_deref(), &env),
176	RootfsCmd::Push { hash, hub } => {
177	let client = image_client(&home, hub)?;
178	eprintln!("Sending image {} …", &hash[..12.min(hash.len())]);
179	let m = client.push(&home, &hash)?;
180	println!(
181	"Sent image {}: {} chunks the hub lacked, {} in {:.1} s",
182	&hash[..12],
183	m.chunks,
184	images::human(m.bytes),
185	m.elapsed.as_secs_f64()
186	);
187	Ok(())
188	}
189	RootfsCmd::Pull { hash, hub } => {
190	let client = image_client(&home, hub)?;
191	match client.fetch(&home, &hash)? {
192	Some(m) => {
193	println!(
194	"Fetched image {}: {} chunks, {} in {:.1} s",
195	&hash[..12],
196	m.chunks,
197	images::human(m.bytes),
198	m.elapsed.as_secs_f64()
199	);
200	Ok(())
201	}
202	None => anyhow::bail!("the hub does not hold image {}; send it with flow rootfs push", &hash[..12.min(hash.len())]),
203	}
204	}
205	RootfsCmd::List => {
206	for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
207	let size = e.metadata().map(\|m\| m.len()).unwrap_or(0);
208	if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
209	println!("cas://blob/{h} {} MiB", size >> 20);
210	}
211	}
212	Ok(())
213	}
214	}
215	}
216	Cmd::Run { jobs, dir, timeout, vars, secrets } => {
217	let home = Home::locate(cli.home.as_deref())?;
218	let pairs = \|what: &str, list: &[String]\| {
219	list.iter()
220	.map(\|e\| {
221	e.split_once('=')
222	.map(\|(k, v)\| (k.to_string(), v.to_string()))
223	.ok_or_else(\|\| anyhow::anyhow!("--{what} {e:?}: expected NAME=VALUE"))
224	})
225	.collect::<Result<Vec<_>>>()
226	};
227	let (vars, secrets) = (pairs("var", &vars)?, pairs("secret", &secrets)?);
228	run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60), &vars, &secrets)
229	}
230	Cmd::Runner { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {
231	let home = Home::locate(cli.home.as_deref())?;
232	if std::env::var_os("OXY_HOME").is_none() {
233	std::env::set_var("OXY_HOME", home.root.join("oxy"));
234	}
235	let name = name.unwrap_or_else(\|\| {
236	std::fs::read_to_string("/etc/hostname").map(\|h\| h.trim().to_string()).unwrap_or_else(\|_\| "flow".into())
237	});
238	let token = match token {
239	Some(t) => t,
240	None => {
241	eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");
242	let mut line = String::new();
243	std::io::stdin().read_line(&mut line)?;
244	line.trim().to_string()
245	}
246	};
247	let got = runner::register(&home, &hub, &token, &name, ca, label)?;
248	println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);
249	Ok(())
250	}
251	Cmd::Runner { cmd: None } => {
252	let home = Home::locate(cli.home.as_deref())?;
253	// The runner's oxy state (identity, user config) stays in its home.
254	if std::env::var_os("OXY_HOME").is_none() {
255	std::env::set_var("OXY_HOME", home.root.join("oxy"));
256	}
257	runner::serve(&home)
258	}
259	Cmd::Tiers => {
260	let home = Home::locate(cli.home.as_deref())?;
261	match runner::tiers(&home) {
262	Ok((have, missing)) => {
263	for t in have {
264	println!("{t}: available");
265	}
266	for why in missing {
267	println!("{}", why.replacen(": ", ": unavailable: ", 1));
268	}
269	}
270	Err(why) => println!("{why}"),
271	}
272	Ok(())
273	}
274	Cmd::Wait { url, state, timeout } => {
275	let token = std::env::var("OXY_TOKEN").ok();
276	let ca = std::env::var_os("OXY_CA").map(PathBuf::from);
277	let code =
278	wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;
279	std::process::exit(code);
280	}
281	}
282	}
283
284	/// The hub image commands talk to: `--hub`, else $OH_HUB, else the
285	/// runner's. The token: $OXY_TOKEN, else the login `oh auth login` (or
286	/// `oxy auth login`) saved for that hub in Oxygen's credential store
287	/// (oxygenflow#44), else the runner's own.
288	fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
289	let runner = home.config.runner.clone();
290	let hub = hub
291	.or_else(\|\| std::env::var("OH_HUB").ok().filter(\|h\| !h.is_empty()))
292	.or_else(\|\| runner.as_ref().map(\|r\| r.hub.clone()))
293	.context("name the hub with --hub https://HUB (or set OH_HUB)")?;
294	let (saved_token, saved_ca) = oxy_core::credentials::Credentials::user().lookup(&hub);
295	let token = std::env::var("OXY_TOKEN")
296	.ok()
297	.filter(\|t\| !t.is_empty())
298	.or(saved_token)
299	.or_else(\|\| runner.as_ref().map(\|r\| r.token.clone()))
300	.with_context(\|\| format!("no token for {hub}: sign in with `oh auth login --hub {hub}`, or set OXY_TOKEN"))?;
301	let ca = std::env::var_os("OXY_CA").map(PathBuf::from).or(saved_ca).or_else(\|\| runner.as_ref().and_then(\|r\| r.ca.clone()));
302	images::Client::new(&hub, &token, ca.as_deref())
303	}
304
305	fn load(file: &Path) -> Result<Pipeline> {
306	let src = std::fs::read_to_string(file).with_context(\|\| format!("read {}", file.display()))?;
307	Pipeline::parse(&src).map_err(\|p\| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
308	}
309
310	fn check(file: &Path) -> Result<()> {
311	let p = load(file)?;
312	let depths = p.depths();
313	for j in p.order() {
314	let sb = &p.sandboxes[&j.sandbox];
315	let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
316	println!("{}{} {} [{}]{after}", " ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
317	}
318	println!("pipeline {}", hex::encode(p.hash()));
319	Ok(())
320	}
321
322	/// The image environment file flow-init reads (`/etc/flow/env`): one
323	/// `NAME=value` per line, names checked as a pipeline's are.
324	fn image_env_file(env: &[String]) -> Result<String> {
325	let mut out = String::new();
326	for e in env {
327	let Some((k, v)) = e.split_once('=') else { bail!("--env {e:?}: expected NAME=value") };
328	let ok = !k.is_empty() && !k.starts_with(\|c: char\| c.is_ascii_digit()) && k.bytes().all(\|b\| b.is_ascii_alphanumeric() \|\| b == b'_');
329	if !ok \|\| v.contains('\n') \|\| v.contains('\0') {
330	bail!("--env {e:?}: a name is letters, digits and _, and a value is one line");
331	}
332	out.push_str(&format!("{k}={v}\n"));
333	}
334	Ok(out)
335	}
336
337	/// The architecture an ELF file was built for, by its header.
338	fn elf_arch(path: &Path) -> Option<&'static str> {
339	let bytes = std::fs::read(path).ok()?;
340	if bytes.len() < 20 \|\| &bytes[..4] != b"\x7fELF" {
341	return None;
342	}
343	let little = bytes[5] == 1;
344	let machine = if little { u16::from_le_bytes([bytes[18], bytes[19]]) } else { u16::from_be_bytes([bytes[18], bytes[19]]) };
345	Some(match machine {
346	0x3E => "x86_64",
347	0xB7 => "aarch64",
348	0xF3 => "riscv64",
349	0x03 => "x86",
350	0x28 => "arm",
351	0x15 => "powerpc64",
352	0x16 => "s390x",
353	_ => return None,
354	})
355	}
356
357	/// The architecture of an unpacked rootfs: its shell's, or the first
358	/// program's found under bin.
359	fn tree_arch(tree: &Path) -> Option<&'static str> {
360	for rel in ["bin/sh", "bin/busybox", "usr/bin/env", "bin/bash", "usr/bin/sh"] {
361	let p = tree.join(rel);
362	let p = std::fs::canonicalize(&p).unwrap_or(p);
363	if let Some(a) = elf_arch(&p) {
364	return Some(a);
365	}
366	}
367	None
368	}
369
370	fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>, env: &[String]) -> Result<()> {
371	let env_file = image_env_file(env)?;
372	if let Some(want) = sha256 {
373	let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
374	let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
375	if got != want.to_lowercase() {
376	bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
377	}
378	}
379	let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
380	let _ = std::fs::remove_dir_all(&tmp);
381	let tree = tmp.join("tree");
382	std::fs::create_dir_all(&tree)?;
383	let result = (\|\| -> Result<String> {
384	let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
385	if !status.success() {
386	bail!("tar could not unpack {}", tarball.display());
387	}
388	// The image's architecture must be this machine's: there is no
389	// emulation tier (spec §11.2).
390	let arch = tree_arch(&tree).unwrap_or("unknown");
391	if arch != "unknown" && arch != std::env::consts::ARCH {
392	bail!(
393	"{} is a {arch} image; this runner is {}, and jobs here run no other architecture",
394	tarball.display(),
395	std::env::consts::ARCH
396	);
397	}
398	// Mount points flow-init needs (docs/m0.md §2).
399	for d in ["src", "tmp", "proc", "dev", "run"] {
400	std::fs::create_dir_all(tree.join(d))?;
401	}
402	if !env_file.is_empty() {
403	std::fs::create_dir_all(tree.join("etc/flow"))?;
404	std::fs::write(tree.join("etc/flow/env"), &env_file)?;
405	}
406	let used: u64 = ignore::WalkBuilder::new(&tree)
407	.hidden(false)
408	.ignore(false)
409	.git_ignore(false)
410	.build()
411	.flatten()
412	.filter_map(\|e\| e.metadata().ok())
413	.map(\|m\| m.len().max(4096))
414	.sum();
415	let mib = used * 13 / 10 / (1 << 20) + 64;
416	let img = tmp.join("rootfs.ext4");
417	let status = std::process::Command::new("mkfs.ext4")
418	.args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
419	.arg(&tree)
420	.arg(&img)
421	.arg(format!("{mib}M"))
422	.status()
423	.context("run mkfs.ext4")?;
424	if !status.success() {
425	bail!("mkfs.ext4 failed");
426	}
427	let hash = home::hash_file(&img)?;
428	std::fs::rename(&img, home.rootfs(&hash))?;
429	let manifest = serde_json::json!({ "arch": if arch == "unknown" { std::env::consts::ARCH } else { arch }, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0) });
430	std::fs::write(home.image_manifest(&hash), manifest.to_string())?;
431	Ok(hash)
432	})();
433	let _ = std::fs::remove_dir_all(&tmp);
434	let hash = result?;
435	println!("rootfs \"cas://blob/{hash}\"");
436	Ok(())
437	}
438
439	fn run_local(
440	home: &Home,
441	dir: &Path,
442	only: &[String],
443	timeout: Duration,
444	vars: &[(String, String)],
445	secrets: &[(String, String)],
446	) -> Result<()> {
447	let p = load(&dir.join(PIPELINE_PATH))?;
448	// The named jobs and everything they depend on.
449	let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
450	for name in only {
451	if !p.jobs.contains_key(name) {
452	bail!("no job {name:?} in the pipeline");
453	}
454	}
455	if !only.is_empty() {
456	loop {
457	let more: Vec<String> = wanted.iter().flat_map(\|j\| p.jobs[j].depends_on.clone()).filter(\|d\| !wanted.contains(d)).collect();
458	if more.is_empty() {
459	break;
460	}
461	wanted.extend(more);
462	}
463	}
464	let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
465	let _ = std::fs::remove_dir_all(&scratch);
466	std::fs::create_dir_all(scratch.join("tree"))?;
467	workspace::copy_tree(dir, &scratch.join("tree"))?;
468	let mut failed: BTreeSet<String> = BTreeSet::new();
469	let mut worst = 0;
470	for job in p.order() {
471	if !only.is_empty() && !wanted.contains(&job.name) {
472	continue;
473	}
474	if let Some(d) = job.depends_on.iter().find(\|d\| failed.contains(*d)) {
475	println!("── {} skipped: {d} failed", job.name);
476	failed.insert(job.name.clone());
477	continue;
478	}
479	let sb = &p.sandboxes[&job.sandbox];
480	if sb.runtime == Runtime::Wasm {
481	bail!("job {}: runtime {} is not available in this build yet", job.name, sb.runtime.as_str());
482	}
483	println!("── {} ({})", job.name, job.exec.join(" "));
484	let js = scratch.join(&job.name);
485	std::fs::create_dir_all(&js)?;
486	let job = &job
487	.in_sandbox(sb)
488	.with_flow_env(&[("FLOW_JOB".into(), job.name.clone()), ("FLOW_RUN".into(), "local".into())])
489	.with_flow_env(vars);
490	// Identities come from a hub: a local run has none.
491	for i in &job.identities {
492	println!("── {}: identity {:?} is not available locally; {} will be missing", job.name, i.audience, i.mount_path);
493	}
494	// Each secret the job declares must be given, as the hub would.
495	let inputs = job
496	.secrets
497	.iter()
498	.map(\|s\| {
499	let value = secrets.iter().find(\|(k, _)\| *k == s.name).map(\|(_, v)\| v.clone());
500	value
501	.map(\|value\| flow_core::api::SecretInput { name: s.name.clone(), path: s.mount_path.clone(), value })
502	.ok_or_else(\|\| anyhow::anyhow!("job {}: secret {:?} is not given; pass --secret {}=VALUE", job.name, s.name, s.name))
503	})
504	.collect::<Result<Vec<_>>>()?;
505	// Each job starts from the same files: jobs do not share writes.
506	let out = if sb.runtime == Runtime::Syscall {
507	workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
508	let group = format!("local-{}-{}", std::process::id(), js.file_name().unwrap_or_default().to_string_lossy());
509	syscall::run(home, sb, job, &inputs, &js.join("tree"), &js, timeout, &mut \|r: Record\| print_record(&r), &group)?
510	} else {
511	workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
512	microvm::run(home, sb, job, &inputs, &js.join("workspace.ext4"), &js, timeout, &mut \|r: Record\| print_record(&r))?
513	};
514	let _ = workspace::remove_tree(&js);
515	match out.exit_code {
516	Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
517	other => {
518	println!(
519	"── {} failed ({}) in {} ms",
520	job.name,
521	other.map(\|c\| format!("exit {c}")).unwrap_or_else(\|\| "no exit code".into()),
522	out.duration_ms
523	);
524	failed.insert(job.name.clone());
525	worst = 1;
526	}
527	}
528	}
529	let _ = workspace::remove_tree(&scratch);
530	if worst != 0 {
531	std::process::exit(worst);
532	}
533	Ok(())
534	}
535
536	fn print_record(r: &Record) {
537	match r.stream.as_str() {
538	"out" => println!(" {}", r.text),
539	"err" => eprintln!(" {}", r.text),
540	s => eprintln!(" [{s}] {}", r.text),
541	}
542	}
543
544	#[cfg(test)]
545	mod tests {
546	use super::image_env_file;
547
548	#[test]
549	fn image_environments_are_checked_lines() {
550	let f = image_env_file(&["PATH=/opt/bin:/usr/bin".into(), "X16EMU=/opt/x16emu".into(), "EMPTY=".into()]).unwrap();
551	assert_eq!(f, "PATH=/opt/bin:/usr/bin\nX16EMU=/opt/x16emu\nEMPTY=\n");
552	assert!(image_env_file(&["NOEQUALS".into()]).is_err());
553	assert!(image_env_file(&["1X=y".into()]).is_err());
554	assert!(image_env_file(&["X=two\nlines".into()]).is_err());
555	assert_eq!(image_env_file(&[]).unwrap(), "");
556	}
557	}

Keyboard