//! `.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, /// When the pipeline's jobs run, unless a job says (oxygenflow#24); /// absent: on every push and from "Run pipeline". #[serde(default, skip_serializing_if = "Option::is_none")] pub on: Option, } /// What starts a job (`on { push "main"; tag "v*"; manual; schedule "0 3 * * *" }`). #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct On { /// Bookmark globs a push runs it for; `Some(vec![])`: any bookmark. #[serde(default, skip_serializing_if = "Option::is_none")] pub push: Option>, /// Tag globs a pushed tag runs it for; `Some(vec![])`: any tag. #[serde(default, skip_serializing_if = "Option::is_none")] pub tag: Option>, /// From "Run pipeline" (and `oh run start`). #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub manual: bool, /// Five-field cron lines, on the hub's clock, for the default bookmark. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub schedule: Vec, } /// Why a run started, for [`On::matches`]. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Event<'a> { /// A push moved this bookmark. Push(&'a str), /// This tag was pushed. Tag(&'a str), /// "Run pipeline", `oh run start`. Manual, /// A rerun of a scheduled run: any job with a schedule. Schedule, /// The hub's clock at a minute (UTC): the jobs with a line due then. ScheduleAt(When), } /// A minute of the hub's clock, UTC, as cron reads it. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct When { pub minute: u32, pub hour: u32, pub day: u32, pub month: u32, /// 0 is Sunday. pub weekday: u32, } impl On { /// What a job without any `on` runs for: every push, and by hand. pub fn default_on() -> On { On { push: Some(vec![]), tag: None, manual: true, schedule: vec![] } } pub fn matches(&self, event: Event<'_>) -> bool { let any = |globs: &Option>, name: &str| match globs { None => false, Some(g) if g.is_empty() => true, Some(g) => g.iter().any(|p| globset::Glob::new(p).is_ok_and(|p| p.compile_matcher().is_match(name))), }; match event { Event::Push(b) => any(&self.push, b), Event::Tag(t) => any(&self.tag, t), Event::Manual => self.manual, Event::Schedule => !self.schedule.is_empty(), Event::ScheduleAt(w) => self .schedule .iter() .filter_map(|l| crate::cron::Cron::parse(l)) .any(|c| c.matches(w.minute, w.hour, w.day, w.month, w.weekday)), } } } impl Pipeline { /// When `job` runs: its own `on`, else the pipeline's, else the default. pub fn on_of(&self, job: &Job) -> On { job.on.clone().or_else(|| self.on.clone()).unwrap_or_else(On::default_on) } } #[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, /// Environment every job in this sandbox starts from; a job's own `env` /// overrides it. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub env: BTreeMap, /// Labels a runner must have to run jobs here (`runs-on`), sorted. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub runs_on: Vec, /// The hosts a `network "egress"` sandbox may reach (oxygenflow#55): /// names, or `*.` and a domain for any name under it; sorted. Its jobs /// reach them through the runner's proxy, and nothing else. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub egress_allow: Vec, /// How long each job here may run, in seconds (`timeout "2h"`), when /// the job does not say; the hub caps it at the site's maximum. #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout_s: Option, /// The architecture the jobs need (`arch "riscv64"`, spec §11.3); /// without it, the rootfs image's. #[serde(default, skip_serializing_if = "Option::is_none")] pub arch: Option, } /// An architecture name as runners report it: `x86_64`, `aarch64`, `riscv64`. pub fn arch_ok(a: &str) -> bool { (2..=16).contains(&a.len()) && a.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_') } #[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, /// Files the job produces, sorted by name (spec §5.2 `output`). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub outputs: Vec, /// Artifacts of jobs this one depends on, put in its workspace before it /// starts (oxygenflow#58), sorted by path. The runner checks each against /// the producing job's receipt; this job's receipt lists them as inputs. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub inputs: Vec, /// Wait for a person before running (oxygenflow#57): `approval { by /// "maintain"; others-only #true; expires "24h" }`. #[serde(default, skip_serializing_if = "Option::is_none")] pub approval: Option, /// The environment the job deploys to (oxygenflow#59), e.g. `prod`: the /// hub records what each environment runs, and the job's identity token /// names it once the job has been approved. #[serde(default, skip_serializing_if = "Option::is_none")] pub environment: Option, /// Secrets the job mounts, sorted by name (spec §9 `secret`): the hub /// supplies each by name, the runner mounts it as a file. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub secrets: Vec, /// Identities the job asks the hub for (oxygenhub#103): an OIDC token /// for an audience, mounted as a file like a secret. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub identities: Vec, /// Labels a runner must have to take this job (`runs-on`), sorted; /// with its sandbox's, what the hub matches against a runner's. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub runs_on: Vec, /// How long the job may run, in seconds (`timeout "90m"`); its /// sandbox's when absent, else the site's default (oxygenflow#23). #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout_s: Option, /// When it runs, over the pipeline's `on` (oxygenflow#24). #[serde(default, skip_serializing_if = "Option::is_none")] pub on: Option, /// `low` or `normal` (`priority "low"`, or the pipeline's): a pipeline /// may lower its jobs below the repository's level, never raise them /// (oxygenhub#244); higher levels are the hub's to give. #[serde(default, skip_serializing_if = "Option::is_none")] pub priority: Option, } impl Job { /// The job's environment on top of its sandbox's: what it runs with. pub fn in_sandbox(&self, sandbox: &Sandbox) -> Job { let mut job = self.clone(); for (k, v) in &sandbox.env { job.env.entry(k.clone()).or_insert_with(|| v.clone()); } job.runs_on.extend(sandbox.runs_on.iter().cloned()); job.runs_on.sort(); job.runs_on.dedup(); job.timeout_s = job.timeout_s.or(sandbox.timeout_s); job } /// The job with variables that tell it what it builds (`FLOW_STATE`, /// `FLOW_BOOKMARK` …) added to its environment; a variable the pipeline /// sets itself keeps the pipeline's value. pub fn with_flow_env(&self, vars: &[(String, String)]) -> Job { let mut job = self.clone(); for (k, v) in vars { job.env.entry(k.clone()).or_insert_with(|| v.clone()); } job } } /// 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) -- /// unless it names a `cache` (spec §5.4, #28): then a disk kept from an /// earlier passing job may start it, and the receipt says which. #[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, /// `cache "name"`: keep this disk after a passing job and start later /// jobs from it (disk mounts only). #[serde(default, skip_serializing_if = "Option::is_none")] pub cache: Option, /// `key "Cargo.lock" ...`: files whose content picks the cache entry. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub key: Vec, } #[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", } } } /// A file a job produces (spec §5.2): read from the job's disks when it /// ends, kept in the repository as a blob, and named by its receipt. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Output { /// The artifact's name, unique in the job. pub name: String, /// A clean absolute path under `/src`, not on a tmpfs mount. pub path: String, /// `when "always"`: kept from a failing job too, such as a test report /// (oxygenflow#37). From a failing job it is never in the receipt. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub always: bool, /// An artifact is one file; a report is a directory with a /// `report.toml`, kept file by file with a manifest (oxygenhub#127). #[serde(default, skip_serializing_if = "OutputKind::is_artifact")] pub kind: OutputKind, } /// An artifact a job takes from a job it depends on (oxygenflow#58): /// `input { artifact "tfplan" from "plan"; path "/src/tfplan" }`. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Input { /// The producing job's artifact. pub artifact: String, /// The producing job: one this job depends on. pub from: String, /// Where it is put: a clean absolute path under `/src`. pub path: String, } /// Who must approve a job before it runs, and for how long it waits /// (oxygenflow#57). #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Approval { /// The least role that approves: `write`, `maintain` (the default) or `admin`. pub by: String, /// Not the person whose push started the run. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub others_only: bool, /// How long it waits once the jobs it depends on are done, in seconds. pub expires_s: u64, /// The approval must be signed with the approver's own key /// (`oxy seal --flow`; oxygenhub#235). Defaults to on for a job that /// names an environment. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub sealed: bool, } /// The longest a job waits for approval: 30 days. const MAX_APPROVAL_S: u64 = 30 * 24 * 3600; /// The most inputs a job takes. const MAX_INPUTS: usize = 8; /// What an [`Output`] is. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum OutputKind { #[default] Artifact, Report, } impl OutputKind { pub fn is_artifact(&self) -> bool { *self == OutputKind::Artifact } pub fn word(&self) -> &'static str { match self { OutputKind::Artifact => "artifact", OutputKind::Report => "report", } } } /// A secret the job reads as a file (spec §9): never in the environment. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Secret { /// Which secret, as the hub names it; unique in the job. pub name: String, /// A clean absolute path under `/run/secrets/`, unique in the job; /// the file is 0400, the job's user's, on an in-memory mount. pub mount_path: String, } /// An OIDC token the hub mints for the job (oxygenhub#103), naming the /// repository, bookmark, run and job, for `audience`; read as a file. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Identity { /// Who the token is for (`sts.amazonaws.com`, a Vault role's audience, …). pub audience: String, /// A clean absolute path under `/run/secrets/`, unique among the /// job's secrets and identities. pub mount_path: String, } /// A secret's mount path: under `/run/secrets/`, no `.` or `..`. pub fn secret_path_ok(p: &str) -> bool { p.strip_prefix("/run/secrets/").is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != "..")) } /// Artifacts a job may declare. pub const MAX_OUTPUTS: usize = 16; /// Whether `p` is a clean absolute path strictly under `/src`, with nothing /// a debugfs command line would have to quote. fn output_path_ok(p: &str) -> bool { p.strip_prefix("/src/").is_some_and(|rest| { !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != "..") && p.bytes().all(|b| b.is_ascii_graphic() && b != b'"' && b != b'\\') }) } /// Mounts a job may declare. pub const MAX_MOUNTS: usize = 4; /// The most secrets a job may mount. pub const MAX_SECRETS: usize = 8; /// The most identities a job may ask for. pub const MAX_IDENTITIES: 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] = &["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) } /// The line of the first escape KDL 2.0 does not have but shells and C /// do (`\0NN`, `\e`, `\xNN`), as someone colouring output writes them. fn foreign_escape_line(src: &str) -> Option { src.lines() .position(|l| { let b = l.as_bytes(); b.windows(2).enumerate().any(|(i, w)| { w[0] == b'\\' && (i == 0 || b[i - 1] != b'\\') && (w[1] == b'e' || w[1] == b'x' || (w[1] == b'0' && b.get(i + 2).is_some_and(u8::is_ascii_digit))) }) }) .map(|i| i + 1) } 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() }); } if let Some(line) = foreign_escape_line(src) { ctx.problems.push(Problem { line, message: "KDL strings have no \\033, \\e or \\x escapes: write ESC (for ANSI colours) as \\u{1b}".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(); let mut on = None; let mut priority = None; for node in doc.nodes() { match node.name().value() { "priority" => { if priority.is_some() { ctx.err(node.span(), "`priority` is given twice"); } priority = parse_priority(&mut ctx, node); } "on" => { if on.is_some() { ctx.err(node.span(), "`on` is given twice"); } on = Some(parse_on(&mut ctx, node)); } "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, on, priority, 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)); } } // Inputs come from a job this one depends on, which declares // that artifact (oxygenflow#58). for i in &job.inputs { let Some((producer, _)) = jobs.get(&i.from) else { ctx.err(*span, format!("job {:?} takes an input from {:?}, which is not declared", job.name, i.from)); continue; }; if !job.depends_on.contains(&i.from) { ctx.err(*span, format!("job {:?} takes {:?} from {:?}: add depends-on \"{}\"", job.name, i.artifact, i.from, i.from)); } match producer.outputs.iter().find(|o| o.name == i.artifact) { Some(o) if o.kind == OutputKind::Artifact => {} Some(_) => ctx.err(*span, format!("{:?} of job {:?} is a report; inputs are artifacts", i.artifact, i.from)), None => ctx.err(*span, format!("job {:?} declares no artifact {:?}", i.from, i.artifact)), } } } 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)); } // The pipeline's priority is each job's that does not say. let jobs = jobs .into_iter() .map(|(k, (mut j, _))| { if j.priority.is_none() { j.priority.clone_from(&priority); } (k, j) }) .collect(); Ok(Pipeline { schema_version: 1, sandboxes, jobs, on }) } /// 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(); let mut env = BTreeMap::new(); let mut runs_on = Vec::new(); let mut timeout_s = None; let mut arch: Option = None; let mut egress_allow = Vec::new(); for c in block(ctx, node) { let key = c.name().value(); if key != "runs-on" && !seen.insert(key.to_string()) { ctx.err(c.span(), format!("`{key}` is given twice")); } if key == "env" { parse_env(ctx, c, &mut env); continue; } if key == "runs-on" { ctx.no_children(c); parse_labels(ctx, c, &mut runs_on); continue; } if key != "network" { 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, } } "arch" => { arch = match ctx.one_string(c) { Some(a) if arch_ok(&a) => Some(a), Some(a) => { ctx.err(c.span(), format!("arch {a:?} is not an architecture name such as x86_64, aarch64 or riscv64")); 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" => { let kind = ctx.one_string(c); if kind.as_deref() != Some("egress") { ctx.no_children(c); } network = match kind.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), // Allow-listed egress (oxygenflow#55): still no network // device; a proxy in the guest relays to the runner, which // lets through only the hosts named here. Some("egress") => { egress_allow = parse_egress(ctx, c); Some(Network::Egress) } Some(other) => { ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\"")); None } None => None, } } "timeout" => timeout_s = parse_timeout(ctx, c), 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 network == Some(Network::Egress) { if runtime == Some(Runtime::Syscall) { ctx.err(node.span(), format!("sandbox {name:?}: network \"egress\" needs runtime \"microvm\"")); } // Only a runner that allows egress offers this label. if !runs_on.iter().any(|l| l == EGRESS_LABEL) { runs_on.push(EGRESS_LABEL.to_string()); runs_on.sort(); } } if !named { return None; } Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network?, env, runs_on, egress_allow, timeout_s, arch, }) } /// The label a runner offers when it allows egress, and an egress sandbox /// needs (oxygenflow#55). pub const EGRESS_LABEL: &str = "egress"; /// The most hosts one sandbox may name. const MAX_EGRESS_HOSTS: usize = 32; /// Whether `pattern` is a host name, or `*.` and a domain: lower-case /// letters, digits, `-` and dots, at least two labels. pub fn egress_pattern_ok(pattern: &str) -> bool { let name = pattern.strip_prefix("*.").unwrap_or(pattern); let labels: Vec<&str> = name.split('.').collect(); name.len() <= 253 && labels.len() >= 2 && labels.iter().all(|l| { !l.is_empty() && l.len() <= 63 && !l.starts_with('-') && !l.ends_with('-') && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') }) } /// Whether `host` is one `patterns` allow: a name exactly, or a name under /// a `*.` domain (not the domain itself). Case and a final dot are ignored. pub fn egress_allows(patterns: &[String], host: &str) -> bool { let host = host.trim_end_matches('.').to_ascii_lowercase(); patterns.iter().any(|p| match p.strip_prefix("*.") { Some(domain) => host.len() > domain.len() + 1 && host.ends_with(domain) && host[..host.len() - domain.len()].ends_with('.'), None => host == *p, }) } /// `network "egress" { allow "*.googleapis.com" "registry.opentofu.org" }`. fn parse_egress(ctx: &mut Ctx, node: &KdlNode) -> Vec { let mut out = Vec::new(); for c in block(ctx, node) { ctx.no_children(c); if c.name().value() != "allow" { ctx.err(c.span(), format!("unknown egress setting `{}`; egress takes `allow \"host\"...`", c.name().value())); continue; } for h in ctx.strings(c) { if egress_pattern_ok(&h) { out.push(h); } else { ctx.err( c.span(), format!("allow {h:?}: expected a lower-case host name such as \"registry.opentofu.org\", or \"*.\" and a domain"), ); } } } out.sort(); out.dedup(); if out.is_empty() { ctx.err(node.span(), "network \"egress\" needs at least one `allow \"host\"`"); } else if out.len() > MAX_EGRESS_HOSTS { ctx.err(node.span(), format!("network \"egress\" may allow at most {MAX_EGRESS_HOSTS} hosts")); } out } /// An `on` block: `push` and `tag` with globs (none: any), `manual`, /// `schedule` with five-field cron lines. fn parse_on(ctx: &mut Ctx, node: &KdlNode) -> On { let mut on = On::default(); if !node.entries().is_empty() { ctx.err(node.span(), "`on` takes a block: on { push \"main\"; tag \"v*\"; manual }"); } let globs = |ctx: &mut Ctx, c: &KdlNode| -> Vec { let g = ctx.strings(c); for p in &g { if let Err(e) = globset::Glob::new(p) { ctx.err(c.span(), format!("{p:?}: {e}")); } } g }; for c in block(ctx, node) { ctx.no_children(c); match c.name().value() { "push" => on.push.get_or_insert_with(Vec::new).extend(globs(ctx, c)), "tag" => on.tag.get_or_insert_with(Vec::new).extend(globs(ctx, c)), "manual" => on.manual = true, "schedule" => { for line in ctx.strings(c) { if crate::cron::Cron::parse(&line).is_some() { on.schedule.push(line); } else { ctx.err( c.span(), format!("schedule {line:?}: expected five cron fields (minute hour day month weekday), as \"0 3 * * *\""), ); } } } other => ctx.err(c.span(), format!("unknown `on` event `{other}`; expected push, tag, manual or schedule")), } } if on.push.is_none() && on.tag.is_none() && !on.manual && on.schedule.is_empty() { ctx.err(node.span(), "`on` names no event, so nothing would run it"); } on } /// The longest a pipeline may ask a job to run. pub const MAX_TIMEOUT_S: u64 = 24 * 3600; /// A duration: whole hours, minutes and seconds in that order, each /// optional, as "90m", "2h", "1h30m" or "45s". pub fn parse_duration(s: &str) -> Option { let (mut total, mut num, mut last) = (0u64, String::new(), 0u8); for ch in s.chars() { if ch.is_ascii_digit() { num.push(ch); continue; } let (unit, mult) = match ch { 'h' => (1, 3600), 'm' => (2, 60), 's' => (3, 1), _ => return None, }; // Each unit once, largest first, each with its number. if num.is_empty() || unit <= last { return None; } last = unit; total = total.checked_add(num.parse::().ok()?.checked_mul(mult)?)?; num.clear(); } (num.is_empty() && total > 0).then_some(total) } /// A `timeout` line: one duration, from a minute to a day. /// `priority "low"`: a pipeline may lower its jobs, never raise them /// (oxygenhub#244). fn parse_priority(ctx: &mut Ctx, c: &KdlNode) -> Option { ctx.no_children(c); let p = ctx.one_string(c)?; match p.as_str() { "low" | "normal" => Some(p), "high" | "urgent" => { ctx.err( c.span(), format!("priority {p:?}: a pipeline may lower its jobs (`low`, `normal`), not raise them; higher levels are set on the hub, and a person may expedite a run"), ); None } _ => { ctx.err(c.span(), format!("priority {p:?}: expected `low` or `normal`")); None } } } fn parse_timeout(ctx: &mut Ctx, c: &KdlNode) -> Option { let t = ctx.one_string(c)?; match parse_duration(&t) { Some(secs) if (60..=MAX_TIMEOUT_S).contains(&secs) => Some(secs), Some(_) => { ctx.err(c.span(), format!("timeout {t:?} must be from 1m to 24h")); None } None => { ctx.err(c.span(), format!("timeout {t:?}: expected hours, minutes and seconds, as \"90m\" or \"1h30m\"")); None } } } /// The labels of a `runs-on` line: lower-case letters, digits, `-`, `_` /// and `.`, at most 64 bytes each; into `into`, sorted, once each. fn parse_labels(ctx: &mut Ctx, c: &KdlNode, into: &mut Vec) { let labels = ctx.strings(c); if labels.is_empty() { ctx.err(c.span(), "`runs-on` needs one or more labels, as runs-on \"riscv64\""); } for l in labels { let ok = !l.is_empty() && l.len() <= 64 && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'-' | b'_' | b'.')); if ok { into.push(l); } else { ctx.err(c.span(), format!("runs-on {l:?}: a label is lower-case letters, digits, -, _ and ., at most 64")); } } into.sort(); into.dedup(); } /// An `env` block of `NAME "value"` lines, into `env`. fn parse_env(ctx: &mut Ctx, c: &KdlNode, env: &mut BTreeMap) { 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 env.insert(k.clone(), v).is_some() { ctx.err(e.span(), format!("environment {k} is set twice")); } } } } /// `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 cache, mut keys) = (None, Vec::new()); 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, } } "cache" => { cache = match ctx.one_string(c) { Some(n) if !n.is_empty() && n.len() <= 64 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)) => { Some(n) } Some(n) => { ctx.err(c.span(), format!("cache name {n:?}: letters, digits, '.', '_' and '-', at most 64")); None } None => None, } } "key" => { keys = ctx.strings(c); if keys.is_empty() { ctx.err(c.span(), "`key` names one or more files, as \"Cargo.lock\""); } for k in &keys { if k.starts_with('/') || k.split('/').any(|p| p.is_empty() || p == "." || p == "..") { ctx.err(c.span(), format!("key file {k:?} must be a path in the repository, without . or ..")); } } } "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}")); } } if source == Some(MountSource::Tmpfs) && cache.is_some() { ctx.err(node.span(), "only an ephemeral://disk mount can be a cache"); } if !keys.is_empty() && cache.is_none() { ctx.err(node.span(), "`key` goes with `cache`"); } Some(Mount { source: source?, target: target?, size_mib: size?, cache, key: keys }) } /// `output { artifact "name" { path "/src/target/release/tool" } }`. /// `secret "deploy-token" { mount-path "/run/secrets/token" }` (spec §9.1). /// The hub is the vault: the secret of that name set for the repository, /// its organisation or the site. `vault-ref` is not in this slice. fn parse_secret(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) { let Some(name) = ctx.one_string(node) else { ctx.err(node.span(), "`secret` takes a name and a block with mount-path"); return; }; if !ctx.name_ok(node.span(), "secret", &name) { return; } let mut path = None; for c in block(ctx, node) { ctx.no_children(c); match c.name().value() { "mount-path" => match ctx.one_string(c) { Some(p) if secret_path_ok(&p) => path = Some(p), Some(p) => { ctx.err(c.span(), format!("secret mount-path {p:?} must be an absolute path under /run/secrets/, without . or ..")) } None => {} }, "vault-ref" => ctx.err(c.span(), "`vault-ref` is not in M0; the hub supplies the secret of that name"), other => ctx.err(c.span(), format!("unknown secret setting `{other}`")), } } let Some(mount_path) = path else { ctx.err(node.span(), format!("secret {name:?} needs a mount-path")); return; }; if job.secrets.iter().any(|s| s.name == name) { ctx.err(node.span(), format!("secret {name:?} is given twice")); } else if job.secrets.iter().any(|s| s.mount_path == mount_path) || job.identities.iter().any(|i| i.mount_path == mount_path) { ctx.err(node.span(), format!("two files mount at {mount_path:?}")); } else if job.secrets.len() == MAX_SECRETS { ctx.err(node.span(), format!("a job has at most {MAX_SECRETS} secrets")); } else { job.secrets.push(Secret { name, mount_path }); } } /// `identity "sts.amazonaws.com" { mount-path "/run/secrets/aws-token" }`: /// the hub mints an OIDC token for that audience and the runner mounts it /// as a file (oxygenhub#103). fn parse_identity(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) { let Some(audience) = ctx.one_string(node) else { ctx.err(node.span(), "`identity` takes an audience and a block with mount-path"); return; }; let aud_ok = !audience.is_empty() && audience.len() <= 255 && !audience.chars().any(|c| c.is_whitespace() || c.is_control()); if !aud_ok { ctx.err(node.span(), format!("identity audience {audience:?} must be 1 to 255 characters without spaces")); return; } let mut path = None; for c in block(ctx, node) { ctx.no_children(c); match c.name().value() { "mount-path" => match ctx.one_string(c) { Some(p) if secret_path_ok(&p) => path = Some(p), Some(p) => { ctx.err(c.span(), format!("identity mount-path {p:?} must be an absolute path under /run/secrets/, without . or ..")) } None => {} }, other => ctx.err(c.span(), format!("unknown identity setting `{other}`")), } } let Some(mount_path) = path else { ctx.err(node.span(), format!("identity {audience:?} needs a mount-path")); return; }; if job.identities.iter().any(|i| i.audience == audience) { ctx.err(node.span(), format!("identity {audience:?} is given twice")); } else if job.identities.iter().any(|i| i.mount_path == mount_path) || job.secrets.iter().any(|s| s.mount_path == mount_path) { ctx.err(node.span(), format!("two files mount at {mount_path:?}")); } else if job.identities.len() == MAX_IDENTITIES { ctx.err(node.span(), format!("a job has at most {MAX_IDENTITIES} identities")); } else { job.identities.push(Identity { audience, mount_path }); } } fn parse_output(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) { if !node.entries().is_empty() { ctx.err(node.span(), "`output` takes a block of artifacts and reports"); } for a in block(ctx, node) { let kind = match a.name().value() { "artifact" => OutputKind::Artifact, "report" => OutputKind::Report, other => { ctx.err( a.span(), format!("unknown output `{other}`; outputs are `artifact \"name\" {{ path \"/src/file\" }}` and `report \"name\" {{ path \"/src/dir\" }}`"), ); continue; } }; let word = kind.word(); let Some(name) = ctx.one_string(a) else { continue }; if !ctx.name_ok(a.span(), word, &name) { continue; } let mut path = None; let mut always = false; for c in block(ctx, a) { ctx.no_children(c); match c.name().value() { "path" => match ctx.one_string(c) { Some(p) if output_path_ok(&p) => path = Some(p), Some(p) => ctx .err(c.span(), format!("{word} path {p:?} must be an absolute path under /src, without . or .., quotes or spaces")), None => {} }, "format" if kind == OutputKind::Artifact => { if ctx.one_string(c).as_deref() != Some("blob") { ctx.err(c.span(), "artifact format is \"blob\""); } } "when" => match ctx.one_string(c).as_deref() { Some("passed") => always = false, Some("always") => always = true, Some(_) => ctx.err(c.span(), format!("{word} `when` is \"passed\" (the default) or \"always\"")), None => {} }, other => ctx.err(c.span(), format!("unknown {word} setting `{other}`")), } } let Some(path) = path else { ctx.err(a.span(), format!("{word} {name:?} needs a path")); continue; }; if job.outputs.iter().any(|o| o.name == name) { ctx.err(a.span(), format!("output {name:?} is given twice (artifacts and reports share names)")); } else if job.outputs.len() == MAX_OUTPUTS { ctx.err(a.span(), format!("a job has at most {MAX_OUTPUTS} artifacts and reports")); } else { job.outputs.push(Output { name, path, always, kind }); } } } /// `approval { by "maintain"; others-only #true; expires "24h" }`, each /// optional: `maintain`, anyone with the role, 24 hours. /// The approval, and whether it said `sealed` itself (else the job's /// environment decides). fn parse_approval(ctx: &mut Ctx, node: &KdlNode) -> Option<(Approval, bool)> { if !node.entries().is_empty() { ctx.err(node.span(), "`approval` takes a block: approval { by \"maintain\"; others-only #true; expires \"24h\" }"); } let mut a = Approval { by: "maintain".into(), others_only: false, expires_s: 24 * 3600, sealed: false }; let mut sealed_said = false; let Some(children) = node.children() else { return Some((a, false)) }; for c in children.nodes() { ctx.no_children(c); match c.name().value() { "by" => match ctx.one_string(c).as_deref() { Some(r @ ("write" | "maintain" | "admin")) => a.by = r.to_string(), Some(r) => ctx.err(c.span(), format!("approval by {r:?}: expected \"write\", \"maintain\" or \"admin\"")), None => {} }, "others-only" => match c.entries() { [e] if e.name().is_none() => match e.value() { KdlValue::Bool(b) => a.others_only = *b, _ => ctx.err(c.span(), "`others-only` takes #true or #false"), }, _ => ctx.err(c.span(), "`others-only` takes #true or #false"), }, "sealed" => match c.entries() { [e] if e.name().is_none() => match e.value() { KdlValue::Bool(b) => { a.sealed = *b; sealed_said = true; } _ => ctx.err(c.span(), "`sealed` takes #true or #false"), }, _ => ctx.err(c.span(), "`sealed` takes #true or #false"), }, "expires" => { let Some(t) = ctx.one_string(c) else { continue }; match parse_duration(&t) { Some(secs) if (60..=MAX_APPROVAL_S).contains(&secs) => a.expires_s = secs, Some(_) => ctx.err(c.span(), format!("expires {t:?} must be from 1m to 720h")), None => ctx.err(c.span(), format!("expires {t:?}: expected hours, minutes and seconds, as \"24h\"")), } } other => { ctx.err(c.span(), format!("unknown approval setting `{other}`; approval takes `by`, `others-only`, `expires` and `sealed`")) } } } Some((a, sealed_said)) } /// `input { artifact "NAME" from "JOB"; path "/src/FILE" }`. fn parse_input(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) { if !node.entries().is_empty() { ctx.err(node.span(), "`input` takes a block: input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\" }"); } let (mut artifact, mut from, mut path) = (None, None, None); for c in block(ctx, node) { ctx.no_children(c); match c.name().value() { "artifact" => match ctx.strings(c).as_slice() { [name, word, producer] if word == "from" => { if ctx.name_ok(c.span(), "artifact", name) && ctx.name_ok(c.span(), "job", producer) { artifact = Some(name.clone()); from = Some(producer.clone()); } } _ => ctx.err(c.span(), "`artifact` names the artifact and its job: artifact \"tfplan\" from \"plan\""), }, "path" => match ctx.one_string(c) { Some(p) if output_path_ok(&p) => path = Some(p), Some(p) => { ctx.err(c.span(), format!("input path {p:?} must be an absolute path under /src, without . or .., quotes or spaces")) } None => {} }, other => ctx.err(c.span(), format!("unknown input setting `{other}`; an input has `artifact` and `path`")), } } match (artifact, from, path) { (Some(artifact), Some(from), Some(path)) => { if job.inputs.iter().any(|i| i.path == path) { ctx.err(node.span(), format!("two inputs are put at {path:?}")); } else if job.inputs.len() == MAX_INPUTS { ctx.err(node.span(), format!("a job takes at most {MAX_INPUTS} inputs")); } else { job.inputs.push(Input { artifact, from, path }); } } (a, _, p) => { if a.is_some() && p.is_none() { ctx.err(node.span(), "an input needs a `path`"); } else if a.is_none() && p.is_some() { ctx.err(node.span(), "an input needs `artifact \"NAME\" from \"JOB\"`"); } } } } 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 approval: Option<(Approval, bool)> = None; let mut job = Job { name: name.clone(), sandbox: String::new(), scope: vec![], depends_on: vec![], env: BTreeMap::new(), exec: vec![], mounts: vec![], secrets: vec![], identities: vec![], outputs: vec![], inputs: vec![], approval: None, environment: None, runs_on: vec![], timeout_s: None, on: None, priority: None, }; let mut seen = BTreeSet::new(); let mut outputs_at = None; for c in block(ctx, node) { let key = c.name().value(); if key != "scope" && key != "depends-on" && key != "mount" && key != "secret" && key != "identity" && key != "runs-on" && !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)); } "runs-on" => { ctx.no_children(c); parse_labels(ctx, c, &mut job.runs_on); } "timeout" => { ctx.no_children(c); job.timeout_s = parse_timeout(ctx, c); } "priority" => job.priority = parse_priority(ctx, c), "on" => job.on = Some(parse_on(ctx, c)), "env" => parse_env(ctx, c, &mut job.env), "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); } } } "output" => { outputs_at = Some(c.span()); parse_output(ctx, c, &mut job); } "input" => parse_input(ctx, c, &mut job), "approval" => approval = parse_approval(ctx, c), "environment" => { ctx.no_children(c); if let Some(e) = ctx.one_string(c) { let ok = !e.is_empty() && e.len() <= 32 && e.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-'); if ok { job.environment = Some(e); } else { ctx.err(c.span(), format!("environment {e:?}: 1 to 32 of a-z 0-9 -, as \"prod\"")); } } } "secret" => parse_secret(ctx, c, &mut job), "identity" => parse_identity(ctx, c, &mut job), 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}`")), } } // A job for an environment takes a sealed approval unless it says not. job.approval = approval.map(|(mut a, said)| { if !said { a.sealed = job.environment.is_some(); } a }); job.mounts.sort_by(|a, b| a.target.cmp(&b.target)); job.outputs.sort_by(|a, b| a.name.cmp(&b.name)); job.inputs.sort_by(|a, b| a.path.cmp(&b.path)); job.secrets.sort_by(|a, b| a.name.cmp(&b.name)); job.identities.sort_by(|a, b| a.audience.cmp(&b.audience)); // A tmpfs is gone when the job ends; nothing on it can be kept. for o in &job.outputs { if let Some(m) = job.mounts.iter().find(|m| m.source == MountSource::Tmpfs && o.path.starts_with(&format!("{}/", m.target))) { ctx.err( outputs_at.unwrap_or(node.span()), format!( "artifact {:?} is on the tmpfs mount {:?}, which is gone when the job ends; use ephemeral://disk", o.name, m.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) }