//! `.flow/pipeline.kdl`: the M0 subset of spec §3 (docs/m0.md §1), parsed //! from KDL 2.0 and validated before anything runs. Every problem is //! reported with its line, and all of them at once. use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet}; use std::fmt; /// Where a pipeline lives in the state being built. pub const PIPELINE_PATH: &str = ".flow/pipeline.kdl"; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Pipeline { pub schema_version: u32, /// By name. pub sandboxes: BTreeMap, /// By name. pub jobs: BTreeMap, } #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum Runtime { /// Tier 0 (spec §4.1); refused in M0. Wasm, /// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups. Syscall, /// Tier 2 (spec §4.3), the default. Microvm, } impl Runtime { pub fn as_str(self) -> &'static str { match self { Runtime::Wasm => "wasm", Runtime::Syscall => "syscall", Runtime::Microvm => "microvm", } } } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum Network { Isolated, Loopback, Egress, } impl Network { pub fn as_str(self) -> &'static str { match self { Network::Isolated => "isolated", Network::Loopback => "loopback", Network::Egress => "egress", } } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Sandbox { pub name: String, pub runtime: Runtime, /// BLAKE3 of the rootfs image, hex. pub rootfs: String, pub memory_mib: u32, pub cpus: u32, pub network: Network, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Job { pub name: String, pub sandbox: String, /// Glob patterns over repository paths (spec §6.2). pub scope: Vec, /// Sorted. pub depends_on: Vec, pub env: BTreeMap, /// Executed as is: no shell, no expansion (spec §2.2 I2). pub exec: Vec, /// Ephemeral scratch mounts, sorted by target (spec §3.1 `mount`). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub mounts: Vec, } /// An ephemeral scratch mount: empty at the start of every job and gone /// after it, so it cannot carry anything between jobs (spec §2.2 I4). #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Mount { pub source: MountSource, /// An absolute path under `/src` or `/tmp`. pub target: String, pub size_mib: u32, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum MountSource { /// `ephemeral://tmpfs`: memory in the sandbox, counted against its memory-limit. Tmpfs, /// `ephemeral://disk`: a fresh sparse disk on the runner, for large build trees. Disk, } impl MountSource { pub fn as_str(self) -> &'static str { match self { MountSource::Tmpfs => "ephemeral://tmpfs", MountSource::Disk => "ephemeral://disk", } } } /// Mounts a job may declare. pub const MAX_MOUNTS: usize = 4; /// Sizes a mount may have: 16 MiB to 64 GiB. pub const MIN_MOUNT_MIB: u64 = 16; pub const MAX_MOUNT_MIB: u64 = 64 * 1024; /// Whether `t` is a clean absolute path strictly under `/src` or `/tmp`. fn mount_target_ok(t: &str) -> bool { let under = t.strip_prefix("/src/").or_else(|| t.strip_prefix("/tmp/")); under.is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != "..")) } /// One problem in a pipeline, at a 1-based line. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Problem { pub line: usize, pub message: String, } impl fmt::Display for Problem { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "line {}: {}", self.line, self.message) } } /// Every problem found, in line order. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] pub struct Problems(pub Vec); impl fmt::Display for Problems { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let lines: Vec = self.0.iter().map(|p| p.to_string()).collect(); f.write_str(&lines.join("\n")) } } const MIN_MEMORY_MIB: u64 = 64; const MAX_MEMORY_MIB: u64 = 16 * 1024; const MAX_CPUS: i128 = 16; /// Nodes of the full spec that M0 recognises and refuses. const LATER_TOP: &[&str] = &["publish", "secret"]; const LATER_JOB: &[&str] = &["output", "secret", "attest"]; struct Ctx<'a> { src: &'a str, problems: Vec, } impl Ctx<'_> { fn line(&self, span: SourceSpan) -> usize { let at = span.offset().min(self.src.len()); self.src[..at].bytes().filter(|b| *b == b'\n').count() + 1 } fn err(&mut self, span: SourceSpan, message: impl Into) { let line = self.line(span); self.problems.push(Problem { line, message: message.into() }); } /// The node's arguments as strings; properties and other types are errors. fn strings(&mut self, node: &KdlNode) -> Vec { let mut out = Vec::new(); for e in node.entries() { if let Some(k) = e.name() { self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value())); continue; } match e.value() { KdlValue::String(s) => out.push(s.clone()), other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))), } } out } /// Exactly one string argument. fn one_string(&mut self, node: &KdlNode) -> Option { let v = self.strings(node); if v.len() != 1 { self.err(node.span(), format!("`{}` takes one string", node.name().value())); return None; } v.into_iter().next() } fn one_integer(&mut self, node: &KdlNode) -> Option { match node.entries() { [e] if e.name().is_none() => match e.value() { KdlValue::Integer(n) => Some(*n), other => { self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other))); None } }, _ => { self.err(node.span(), format!("`{}` takes one integer", node.name().value())); None } } } fn no_children(&mut self, node: &KdlNode) { if node.children().is_some() { self.err(node.span(), format!("`{}` takes no block", node.name().value())); } } fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool { let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)); if !ok { self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -")); } ok } } use kdl::{KdlDocument, KdlNode, KdlValue}; use miette::SourceSpan; fn kind_of(v: &KdlValue) -> &'static str { match v { KdlValue::String(_) => "a string", KdlValue::Integer(_) => "an integer", KdlValue::Float(_) => "a number", KdlValue::Bool(_) => "a boolean", KdlValue::Null => "null", } } /// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB. fn parse_memory(s: &str) -> Option { let (num, unit) = s.split_at(s.find(|c: char| !c.is_ascii_digit())?); let n: u64 = num.parse().ok()?; let mib = match unit { "KiB" => n.is_multiple_of(1024).then_some(n / 1024)?, "MiB" => n, "GiB" => n.checked_mul(1024)?, _ => return None, }; Some(mib) } impl Pipeline { /// Parse and validate a pipeline document. pub fn parse(src: &str) -> Result { let doc = match KdlDocument::parse_v2(src) { Ok(d) => d, Err(e) => { let mut ctx = Ctx { src, problems: Vec::new() }; for d in &e.diagnostics { let msg = d.message.clone().or_else(|| d.label.clone()).unwrap_or_else(|| "not valid KDL 2.0".into()); ctx.err(d.span, msg); } if ctx.problems.is_empty() { ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() }); } return Err(Problems(ctx.problems)); } }; let mut ctx = Ctx { src, problems: Vec::new() }; let mut schema = None; let mut sandboxes = BTreeMap::new(); let mut jobs: BTreeMap = BTreeMap::new(); for node in doc.nodes() { match node.name().value() { "schema-version" => { ctx.no_children(node); match ctx.one_integer(node) { Some(1) => schema = Some(1), Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")), None => {} } } "sandbox" => { if let Some(sb) = parse_sandbox(&mut ctx, node) { if sandboxes.contains_key(&sb.name) { ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name)); } else { sandboxes.insert(sb.name.clone(), sb); } } } "job" => { if let Some(job) = parse_job(&mut ctx, node) { if jobs.contains_key(&job.name) { ctx.err(node.span(), format!("job {:?} is declared twice", job.name)); } else { jobs.insert(job.name.clone(), (job, node.span())); } } } n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")), n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, sandbox or job")), } } if schema.is_none() && !ctx.problems.iter().any(|p| p.message.contains("schema-version")) { ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() }); } if jobs.is_empty() && ctx.problems.is_empty() { ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() }); } // References and cycles. for (job, span) in jobs.values() { if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) { ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox)); } for d in &job.depends_on { if !jobs.contains_key(d) { ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d)); } else if d == &job.name { ctx.err(*span, format!("job {:?} depends on itself", job.name)); } } } let graph: BTreeMap<&str, Vec<&str>> = jobs.values().map(|(j, _)| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect(); if let Some(cycle) = find_cycle(&graph) { let span = jobs[cycle[0]].1; ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → "))); } if !ctx.problems.is_empty() { ctx.problems.sort_by_key(|p| p.line); return Err(Problems(ctx.problems)); } Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(|(k, (j, _))| (k, j)).collect() }) } /// Jobs in an order where each comes after what it depends on (ties by name). pub fn order(&self) -> Vec<&Job> { let mut done: BTreeSet<&str> = BTreeSet::new(); let mut out = Vec::with_capacity(self.jobs.len()); while out.len() < self.jobs.len() { let before = out.len(); for j in self.jobs.values() { if !done.contains(j.name.as_str()) && j.depends_on.iter().all(|d| done.contains(d.as_str())) { out.push(j); } } for j in &out[before..] { done.insert(j.name.as_str()); } if out.len() == before { break; // validated pipelines have no cycles } } out } /// Each job's depth: 0 without dependencies, else one more than its deepest dependency. pub fn depths(&self) -> BTreeMap<&str, usize> { let mut depth: BTreeMap<&str, usize> = BTreeMap::new(); for j in self.order() { let d = j.depends_on.iter().map(|x| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0); depth.insert(j.name.as_str(), d); } depth } } fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option> { #[derive(Clone, Copy, PartialEq)] enum Mark { Visiting, Done, } fn visit<'a>( n: &'a str, graph: &BTreeMap<&'a str, Vec<&'a str>>, marks: &mut BTreeMap<&'a str, Mark>, stack: &mut Vec<&'a str>, ) -> Option> { match marks.get(n) { Some(Mark::Done) => return None, Some(Mark::Visiting) => { let at = stack.iter().position(|s| *s == n).unwrap_or(0); let mut c = stack[at..].to_vec(); c.push(n); return Some(c); } None => {} } marks.insert(n, Mark::Visiting); stack.push(n); for d in graph.get(n).into_iter().flatten() { if graph.contains_key(d) && *d != n { if let Some(c) = visit(d, graph, marks, stack) { return Some(c); } } } stack.pop(); marks.insert(n, Mark::Done); None } let mut marks = BTreeMap::new(); for n in graph.keys() { let mut stack = Vec::new(); if let Some(c) = visit(n, graph, &mut marks, &mut stack) { return Some(c); } } None } fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] { match node.children() { Some(d) => d.nodes(), None => { ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value())); &[] } } } fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option { let name = ctx.one_string(node)?; let named = ctx.name_ok(node.span(), "sandbox", &name); let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) = (Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated)); let mut seen = BTreeSet::new(); for c in block(ctx, node) { let key = c.name().value(); if !seen.insert(key.to_string()) { ctx.err(c.span(), format!("`{key}` is given twice")); } ctx.no_children(c); match key { "runtime" => { runtime = match ctx.one_string(c).as_deref() { Some("microvm") => Some(Runtime::Microvm), Some("syscall") | Some("gvisor") => Some(Runtime::Syscall), Some("wasm") => { ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\""); None } Some(other) => { ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\"")); None } None => None, } } "rootfs" => { rootfs = match ctx.one_string(c) { Some(r) => match r.strip_prefix("cas://blob/") { Some(h) if h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => { Some(h.to_string()) } _ => { ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits")); None } }, None => None, } } "memory-limit" => { memory = match ctx.one_string(c) { Some(m) => match parse_memory(&m) { Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32), Some(_) => { ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB")); None } None => { ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\"")); None } }, None => None, } } "cpu-limit" => { cpus = match ctx.one_integer(c) { Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32), Some(n) => { ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}")); None } None => None, } } "network" => { network = match ctx.one_string(c).as_deref() { Some("isolated") => Some(Network::Isolated), // Loopback only: the guest's `lo` comes up, and it still has // no device that reaches anything outside it. Some("loopback") => Some(Network::Loopback), Some("egress") => { ctx.err(c.span(), "network \"egress\" is not in M0; M0 has \"isolated\" and \"loopback\""); None } Some(other) => { ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\"")); None } None => None, } } other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")), } } if rootfs.is_none() && !seen.contains("rootfs") { ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs")); } if !named { return None; } Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network? }) } /// `mount { source "ephemeral://disk"; target "/src/target"; size "8GiB" }`. fn parse_mount(ctx: &mut Ctx, node: &KdlNode) -> Option { let (mut source, mut target, mut size) = (None, None, None); let mut seen = BTreeSet::new(); for c in block(ctx, node) { let key = c.name().value(); if !seen.insert(key.to_string()) { ctx.err(c.span(), format!("`{key}` is given twice")); continue; } ctx.no_children(c); match key { "source" => { source = match ctx.one_string(c).as_deref() { Some("ephemeral://tmpfs") => Some(MountSource::Tmpfs), Some("ephemeral://disk") => Some(MountSource::Disk), Some(other) => { ctx.err( c.span(), format!("mount source {other:?} is not in M0 yet; M0 mounts \"ephemeral://tmpfs\" or \"ephemeral://disk\""), ); None } None => None, } } "target" => { target = match ctx.one_string(c) { Some(t) if mount_target_ok(&t) => Some(t), Some(t) => { ctx.err( c.span(), format!("mount target {t:?} must be an absolute path under /src or /tmp, without . or .. or a trailing /"), ); None } None => None, } } "size" => { size = match ctx.one_string(c) { Some(m) => match parse_memory(&m) { Some(mib) if (MIN_MOUNT_MIB..=MAX_MOUNT_MIB).contains(&mib) => Some(mib as u32), Some(_) => { ctx.err(c.span(), format!("mount size {m:?} must be from 16MiB to 64GiB")); None } None => { ctx.err(c.span(), format!("mount size {m:?}: expected a whole number of KiB, MiB or GiB, as \"8GiB\"")); None } }, None => None, } } "readonly" | "include" => ctx.err(c.span(), format!("mount `{key}` is not in M0 yet")), other => ctx.err(c.span(), format!("unknown mount setting `{other}`")), } } for (name, missing) in [("source", source.is_none()), ("target", target.is_none()), ("size", size.is_none())] { if missing && !seen.contains(name) { ctx.err(node.span(), format!("a mount needs a {name}")); } } Some(Mount { source: source?, target: target?, size_mib: size? }) } fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option { let name = ctx.one_string(node)?; let named = ctx.name_ok(node.span(), "job", &name); let mut job = Job { name: name.clone(), sandbox: String::new(), scope: vec![], depends_on: vec![], env: BTreeMap::new(), exec: vec![], mounts: vec![], }; let mut seen = BTreeSet::new(); for c in block(ctx, node) { let key = c.name().value(); if key != "scope" && key != "depends-on" && key != "mount" && !seen.insert(key.to_string()) { ctx.err(c.span(), format!("`{key}` is given twice")); } match key { "sandbox" => { ctx.no_children(c); job.sandbox = ctx.one_string(c).unwrap_or_default(); } "scope" => { ctx.no_children(c); for g in ctx.strings(c) { match globset::Glob::new(&g) { Ok(_) => job.scope.push(g), Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")), } } } "depends-on" => { ctx.no_children(c); job.depends_on.extend(ctx.strings(c)); } "env" => { if !c.entries().is_empty() { ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines"); } for e in block(ctx, c) { let k = e.name().value().to_string(); let valid = !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_'); if !valid { ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _")); continue; } ctx.no_children(e); if let Some(v) = ctx.one_string(e) { if job.env.insert(k.clone(), v).is_some() { ctx.err(e.span(), format!("environment {k} is set twice")); } } } } "exec" => { ctx.no_children(c); job.exec = ctx.strings(c); if job.exec.is_empty() { ctx.err(c.span(), "`exec` needs a program and its arguments"); } else if job.exec.iter().any(|a| a.contains('\0')) { ctx.err(c.span(), "`exec` arguments cannot contain NUL"); } } "mount" => { if !c.entries().is_empty() { ctx.err(c.span(), "`mount` takes a block of source, target and size"); } if let Some(m) = parse_mount(ctx, c) { if job.mounts.len() == MAX_MOUNTS { ctx.err(c.span(), format!("a job has at most {MAX_MOUNTS} mounts")); } else if let Some(other) = job.mounts.iter().find(|o| { o.target == m.target || m.target.starts_with(&format!("{}/", o.target)) || o.target.starts_with(&format!("{}/", m.target)) }) { ctx.err(c.span(), format!("mount {:?} overlaps mount {:?}", m.target, other.target)); } else { job.mounts.push(m); } } } n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")), other => ctx.err(c.span(), format!("unknown job setting `{other}`")), } } job.mounts.sort_by(|a, b| a.target.cmp(&b.target)); if job.sandbox.is_empty() && !seen.contains("sandbox") { ctx.err(node.span(), format!("job {name:?} needs a sandbox")); } if job.exec.is_empty() && !seen.contains("exec") { ctx.err(node.span(), format!("job {name:?} needs an exec")); } job.depends_on.sort(); job.depends_on.dedup(); // Kept even when incomplete, so its references are checked too; the // problems above already make the pipeline invalid. named.then_some(job) }