OXYGENHUB
# ec394f0c115e oxygenflow / crates / flow-core / src / pipeline.rs
62.7 KiB · Rust · last changed Sealed approvals: approval { sealed } and the oxygen pin 20 days ago source blame history raw
1
//! `.flow/pipeline.kdl`: the M0 subset of spec §3 (docs/m0.md §1), parsed
2
//! from KDL 2.0 and validated before anything runs. Every problem is
3
//! reported with its line, and all of them at once.
4
5
use serde::{Deserialize, Serialize};
6
use std::collections::{BTreeMap, BTreeSet};
7
use std::fmt;
8
9
/// Where a pipeline lives in the state being built.
10
pub const PIPELINE_PATH: &str = ".flow/pipeline.kdl";
11
12
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
13
pub struct Pipeline {
14
    pub schema_version: u32,
15
    /// By name.
16
    pub sandboxes: BTreeMap<String, Sandbox>,
17
    /// By name.
18
    pub jobs: BTreeMap<String, Job>,
19
    /// When the pipeline's jobs run, unless a job says (oxygenflow#24);
20
    /// absent: on every push and from "Run pipeline".
21
    #[serde(default, skip_serializing_if = "Option::is_none")]
22
    pub on: Option<On>,
23
}
24
25
/// What starts a job (`on { push "main"; tag "v*"; manual; schedule "0 3 * * *" }`).
26
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
27
pub struct On {
28
    /// Bookmark globs a push runs it for; `Some(vec![])`: any bookmark.
29
    #[serde(default, skip_serializing_if = "Option::is_none")]
30
    pub push: Option<Vec<String>>,
31
    /// Tag globs a pushed tag runs it for; `Some(vec![])`: any tag.
32
    #[serde(default, skip_serializing_if = "Option::is_none")]
33
    pub tag: Option<Vec<String>>,
34
    /// From "Run pipeline" (and `oh run start`).
35
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
36
    pub manual: bool,
37
    /// Five-field cron lines, on the hub's clock, for the default bookmark.
38
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
39
    pub schedule: Vec<String>,
40
}
41
42
/// Why a run started, for [`On::matches`].
43
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
44
pub enum Event<'a> {
45
    /// A push moved this bookmark.
46
    Push(&'a str),
47
    /// This tag was pushed.
48
    Tag(&'a str),
49
    /// "Run pipeline", `oh run start`.
50
    Manual,
51
    /// A `schedule` line came due.
52
    Schedule,
53
}
54
55
impl On {
56
    /// What a job without any `on` runs for: every push, and by hand.
57
    pub fn default_on() -> On {
58
        On { push: Some(vec![]), tag: None, manual: true, schedule: vec![] }
59
    }
60
61
    pub fn matches(&self, event: Event<'_>) -> bool {
62
        let any = |globs: &Option<Vec<String>>, name: &str| match globs {
63
            None => false,
64
            Some(g) if g.is_empty() => true,
65
            Some(g) => g.iter().any(|p| globset::Glob::new(p).is_ok_and(|p| p.compile_matcher().is_match(name))),
66
        };
67
        match event {
68
            Event::Push(b) => any(&self.push, b),
69
            Event::Tag(t) => any(&self.tag, t),
70
            Event::Manual => self.manual,
71
            Event::Schedule => !self.schedule.is_empty(),
72
        }
73
    }
74
}
75
76
impl Pipeline {
77
    /// When `job` runs: its own `on`, else the pipeline's, else the default.
78
    pub fn on_of(&self, job: &Job) -> On {
79
        job.on.clone().or_else(|| self.on.clone()).unwrap_or_else(On::default_on)
80
    }
81
}
82
83
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
84
#[serde(rename_all = "lowercase")]
85
pub enum Runtime {
86
    /// Tier 0 (spec §4.1); refused in M0.
87
    Wasm,
88
    /// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups.
89
    Syscall,
90
    /// Tier 2 (spec §4.3), the default.
91
    Microvm,
92
}
93
94
impl Runtime {
95
    pub fn as_str(self) -> &'static str {
96
        match self {
97
            Runtime::Wasm => "wasm",
98
            Runtime::Syscall => "syscall",
99
            Runtime::Microvm => "microvm",
100
        }
101
    }
102
}
103
104
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
105
#[serde(rename_all = "lowercase")]
106
pub enum Network {
107
    Isolated,
108
    Loopback,
109
    Egress,
110
}
111
112
impl Network {
113
    pub fn as_str(self) -> &'static str {
114
        match self {
115
            Network::Isolated => "isolated",
116
            Network::Loopback => "loopback",
117
            Network::Egress => "egress",
118
        }
119
    }
120
}
121
122
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
123
pub struct Sandbox {
124
    pub name: String,
125
    pub runtime: Runtime,
126
    /// BLAKE3 of the rootfs image, hex.
127
    pub rootfs: String,
128
    pub memory_mib: u32,
129
    pub cpus: u32,
130
    pub network: Network,
131
    /// Environment every job in this sandbox starts from; a job's own `env`
132
    /// overrides it.
133
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
134
    pub env: BTreeMap<String, String>,
135
    /// Labels a runner must have to run jobs here (`runs-on`), sorted.
136
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
137
    pub runs_on: Vec<String>,
138
    /// The hosts a `network "egress"` sandbox may reach (oxygenflow#55):
139
    /// names, or `*.` and a domain for any name under it; sorted. Its jobs
140
    /// reach them through the runner's proxy, and nothing else.
141
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
142
    pub egress_allow: Vec<String>,
143
    /// How long each job here may run, in seconds (`timeout "2h"`), when
144
    /// the job does not say; the hub caps it at the site's maximum.
145
    #[serde(default, skip_serializing_if = "Option::is_none")]
146
    pub timeout_s: Option<u64>,
147
    /// The architecture the jobs need (`arch "riscv64"`, spec §11.3);
148
    /// without it, the rootfs image's.
149
    #[serde(default, skip_serializing_if = "Option::is_none")]
150
    pub arch: Option<String>,
151
}
152
153
/// An architecture name as runners report it: `x86_64`, `aarch64`, `riscv64`.
154
pub fn arch_ok(a: &str) -> bool {
155
    (2..=16).contains(&a.len()) && a.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
156
}
157
158
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
159
pub struct Job {
160
    pub name: String,
161
    pub sandbox: String,
162
    /// Glob patterns over repository paths (spec §6.2).
163
    pub scope: Vec<String>,
164
    /// Sorted.
165
    pub depends_on: Vec<String>,
166
    pub env: BTreeMap<String, String>,
167
    /// Executed as is: no shell, no expansion (spec §2.2 I2).
168
    pub exec: Vec<String>,
169
    /// Ephemeral scratch mounts, sorted by target (spec §3.1 `mount`).
170
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
171
    pub mounts: Vec<Mount>,
172
    /// Files the job produces, sorted by name (spec §5.2 `output`).
173
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
174
    pub outputs: Vec<Output>,
175
    /// Artifacts of jobs this one depends on, put in its workspace before it
176
    /// starts (oxygenflow#58), sorted by path. The runner checks each against
177
    /// the producing job's receipt; this job's receipt lists them as inputs.
178
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
179
    pub inputs: Vec<Input>,
180
    /// Wait for a person before running (oxygenflow#57): `approval { by
181
    /// "maintain"; others-only #true; expires "24h" }`.
182
    #[serde(default, skip_serializing_if = "Option::is_none")]
183
    pub approval: Option<Approval>,
184
    /// The environment the job deploys to (oxygenflow#59), e.g. `prod`: the
185
    /// hub records what each environment runs, and the job's identity token
186
    /// names it once the job has been approved.
187
    #[serde(default, skip_serializing_if = "Option::is_none")]
188
    pub environment: Option<String>,
189
    /// Secrets the job mounts, sorted by name (spec §9 `secret`): the hub
190
    /// supplies each by name, the runner mounts it as a file.
191
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
192
    pub secrets: Vec<Secret>,
193
    /// Identities the job asks the hub for (oxygenhub#103): an OIDC token
194
    /// for an audience, mounted as a file like a secret.
195
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
196
    pub identities: Vec<Identity>,
197
    /// Labels a runner must have to take this job (`runs-on`), sorted;
198
    /// with its sandbox's, what the hub matches against a runner's.
199
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
200
    pub runs_on: Vec<String>,
201
    /// How long the job may run, in seconds (`timeout "90m"`); its
202
    /// sandbox's when absent, else the site's default (oxygenflow#23).
203
    #[serde(default, skip_serializing_if = "Option::is_none")]
204
    pub timeout_s: Option<u64>,
205
    /// When it runs, over the pipeline's `on` (oxygenflow#24).
206
    #[serde(default, skip_serializing_if = "Option::is_none")]
207
    pub on: Option<On>,
208
}
209
210
impl Job {
211
    /// The job's environment on top of its sandbox's: what it runs with.
212
    pub fn in_sandbox(&self, sandbox: &Sandbox) -> Job {
213
        let mut job = self.clone();
214
        for (k, v) in &sandbox.env {
215
            job.env.entry(k.clone()).or_insert_with(|| v.clone());
216
        }
217
        job.runs_on.extend(sandbox.runs_on.iter().cloned());
218
        job.runs_on.sort();
219
        job.runs_on.dedup();
220
        job.timeout_s = job.timeout_s.or(sandbox.timeout_s);
221
        job
222
    }
223
224
    /// The job with variables that tell it what it builds (`FLOW_STATE`,
225
    /// `FLOW_BOOKMARK` …) added to its environment; a variable the pipeline
226
    /// sets itself keeps the pipeline's value.
227
    pub fn with_flow_env(&self, vars: &[(String, String)]) -> Job {
228
        let mut job = self.clone();
229
        for (k, v) in vars {
230
            job.env.entry(k.clone()).or_insert_with(|| v.clone());
231
        }
232
        job
233
    }
234
}
235
236
/// An ephemeral scratch mount: empty at the start of every job and gone
237
/// after it, so it cannot carry anything between jobs (spec §2.2 I4) --
238
/// unless it names a `cache` (spec §5.4, #28): then a disk kept from an
239
/// earlier passing job may start it, and the receipt says which.
240
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
241
pub struct Mount {
242
    pub source: MountSource,
243
    /// An absolute path under `/src` or `/tmp`.
244
    pub target: String,
245
    pub size_mib: u32,
246
    /// `cache "name"`: keep this disk after a passing job and start later
247
    /// jobs from it (disk mounts only).
248
    #[serde(default, skip_serializing_if = "Option::is_none")]
249
    pub cache: Option<String>,
250
    /// `key "Cargo.lock" ...`: files whose content picks the cache entry.
251
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
252
    pub key: Vec<String>,
253
}
254
255
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
256
#[serde(rename_all = "lowercase")]
257
pub enum MountSource {
258
    /// `ephemeral://tmpfs`: memory in the sandbox, counted against its memory-limit.
259
    Tmpfs,
260
    /// `ephemeral://disk`: a fresh sparse disk on the runner, for large build trees.
261
    Disk,
262
}
263
264
impl MountSource {
265
    pub fn as_str(self) -> &'static str {
266
        match self {
267
            MountSource::Tmpfs => "ephemeral://tmpfs",
268
            MountSource::Disk => "ephemeral://disk",
269
        }
270
    }
271
}
272
273
/// A file a job produces (spec §5.2): read from the job's disks when it
274
/// ends, kept in the repository as a blob, and named by its receipt.
275
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
276
pub struct Output {
277
    /// The artifact's name, unique in the job.
278
    pub name: String,
279
    /// A clean absolute path under `/src`, not on a tmpfs mount.
280
    pub path: String,
281
    /// `when "always"`: kept from a failing job too, such as a test report
282
    /// (oxygenflow#37). From a failing job it is never in the receipt.
283
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
284
    pub always: bool,
285
    /// An artifact is one file; a report is a directory with a
286
    /// `report.toml`, kept file by file with a manifest (oxygenhub#127).
287
    #[serde(default, skip_serializing_if = "OutputKind::is_artifact")]
288
    pub kind: OutputKind,
289
}
290
291
/// An artifact a job takes from a job it depends on (oxygenflow#58):
292
/// `input { artifact "tfplan" from "plan"; path "/src/tfplan" }`.
293
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
294
pub struct Input {
295
    /// The producing job's artifact.
296
    pub artifact: String,
297
    /// The producing job: one this job depends on.
298
    pub from: String,
299
    /// Where it is put: a clean absolute path under `/src`.
300
    pub path: String,
301
}
302
303
/// Who must approve a job before it runs, and for how long it waits
304
/// (oxygenflow#57).
305
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
306
pub struct Approval {
307
    /// The least role that approves: `write`, `maintain` (the default) or `admin`.
308
    pub by: String,
309
    /// Not the person whose push started the run.
310
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
311
    pub others_only: bool,
312
    /// How long it waits once the jobs it depends on are done, in seconds.
313
    pub expires_s: u64,
314
    /// The approval must be signed with the approver's own key
315
    /// (`oxy seal --flow`; oxygenhub#235). Defaults to on for a job that
316
    /// names an environment.
317
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
318
    pub sealed: bool,
319
}
320
321
/// The longest a job waits for approval: 30 days.
322
const MAX_APPROVAL_S: u64 = 30 * 24 * 3600;
323
324
/// The most inputs a job takes.
325
const MAX_INPUTS: usize = 8;
326
327
/// What an [`Output`] is.
328
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
329
#[serde(rename_all = "lowercase")]
330
pub enum OutputKind {
331
    #[default]
332
    Artifact,
333
    Report,
334
}
335
336
impl OutputKind {
337
    pub fn is_artifact(&self) -> bool {
338
        *self == OutputKind::Artifact
339
    }
340
341
    pub fn word(&self) -> &'static str {
342
        match self {
343
            OutputKind::Artifact => "artifact",
344
            OutputKind::Report => "report",
345
        }
346
    }
347
}
348
349
/// A secret the job reads as a file (spec §9): never in the environment.
350
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
351
pub struct Secret {
352
    /// Which secret, as the hub names it; unique in the job.
353
    pub name: String,
354
    /// A clean absolute path under `/run/secrets/`, unique in the job;
355
    /// the file is 0400, the job's user's, on an in-memory mount.
356
    pub mount_path: String,
357
}
358
359
/// An OIDC token the hub mints for the job (oxygenhub#103), naming the
360
/// repository, bookmark, run and job, for `audience`; read as a file.
361
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
362
pub struct Identity {
363
    /// Who the token is for (`sts.amazonaws.com`, a Vault role's audience, …).
364
    pub audience: String,
365
    /// A clean absolute path under `/run/secrets/`, unique among the
366
    /// job's secrets and identities.
367
    pub mount_path: String,
368
}
369
370
/// A secret's mount path: under `/run/secrets/`, no `.` or `..`.
371
pub fn secret_path_ok(p: &str) -> bool {
372
    p.strip_prefix("/run/secrets/").is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != ".."))
373
}
374
375
/// Artifacts a job may declare.
376
pub const MAX_OUTPUTS: usize = 16;
377
378
/// Whether `p` is a clean absolute path strictly under `/src`, with nothing
379
/// a debugfs command line would have to quote.
380
fn output_path_ok(p: &str) -> bool {
381
    p.strip_prefix("/src/").is_some_and(|rest| {
382
        !rest.is_empty()
383
            && rest.split('/').all(|c| !c.is_empty() && c != "." && c != "..")
384
            && p.bytes().all(|b| b.is_ascii_graphic() && b != b'"' && b != b'\\')
385
    })
386
}
387
388
/// Mounts a job may declare.
389
pub const MAX_MOUNTS: usize = 4;
390
/// The most secrets a job may mount.
391
pub const MAX_SECRETS: usize = 8;
392
/// The most identities a job may ask for.
393
pub const MAX_IDENTITIES: usize = 4;
394
/// Sizes a mount may have: 16 MiB to 64 GiB.
395
pub const MIN_MOUNT_MIB: u64 = 16;
396
pub const MAX_MOUNT_MIB: u64 = 64 * 1024;
397
398
/// Whether `t` is a clean absolute path strictly under `/src` or `/tmp`.
399
fn mount_target_ok(t: &str) -> bool {
400
    let under = t.strip_prefix("/src/").or_else(|| t.strip_prefix("/tmp/"));
401
    under.is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != ".."))
402
}
403
404
/// One problem in a pipeline, at a 1-based line.
405
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
406
pub struct Problem {
407
    pub line: usize,
408
    pub message: String,
409
}
410
411
impl fmt::Display for Problem {
412
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
413
        write!(f, "line {}: {}", self.line, self.message)
414
    }
415
}
416
417
/// Every problem found, in line order.
418
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
419
pub struct Problems(pub Vec<Problem>);
420
421
impl fmt::Display for Problems {
422
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
423
        let lines: Vec<String> = self.0.iter().map(|p| p.to_string()).collect();
424
        f.write_str(&lines.join("\n"))
425
    }
426
}
427
428
const MIN_MEMORY_MIB: u64 = 64;
429
const MAX_MEMORY_MIB: u64 = 16 * 1024;
430
const MAX_CPUS: i128 = 16;
431
432
/// Nodes of the full spec that M0 recognises and refuses.
433
const LATER_TOP: &[&str] = &["publish", "secret"];
434
const LATER_JOB: &[&str] = &["attest"];
435
436
struct Ctx<'a> {
437
    src: &'a str,
438
    problems: Vec<Problem>,
439
}
440
441
impl Ctx<'_> {
442
    fn line(&self, span: SourceSpan) -> usize {
443
        let at = span.offset().min(self.src.len());
444
        self.src[..at].bytes().filter(|b| *b == b'\n').count() + 1
445
    }
446
447
    fn err(&mut self, span: SourceSpan, message: impl Into<String>) {
448
        let line = self.line(span);
449
        self.problems.push(Problem { line, message: message.into() });
450
    }
451
452
    /// The node's arguments as strings; properties and other types are errors.
453
    fn strings(&mut self, node: &KdlNode) -> Vec<String> {
454
        let mut out = Vec::new();
455
        for e in node.entries() {
456
            if let Some(k) = e.name() {
457
                self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value()));
458
                continue;
459
            }
460
            match e.value() {
461
                KdlValue::String(s) => out.push(s.clone()),
462
                other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))),
463
            }
464
        }
465
        out
466
    }
467
468
    /// Exactly one string argument.
469
    fn one_string(&mut self, node: &KdlNode) -> Option<String> {
470
        let v = self.strings(node);
471
        if v.len() != 1 {
472
            self.err(node.span(), format!("`{}` takes one string", node.name().value()));
473
            return None;
474
        }
475
        v.into_iter().next()
476
    }
477
478
    fn one_integer(&mut self, node: &KdlNode) -> Option<i128> {
479
        match node.entries() {
480
            [e] if e.name().is_none() => match e.value() {
481
                KdlValue::Integer(n) => Some(*n),
482
                other => {
483
                    self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other)));
484
                    None
485
                }
486
            },
487
            _ => {
488
                self.err(node.span(), format!("`{}` takes one integer", node.name().value()));
489
                None
490
            }
491
        }
492
    }
493
494
    fn no_children(&mut self, node: &KdlNode) {
495
        if node.children().is_some() {
496
            self.err(node.span(), format!("`{}` takes no block", node.name().value()));
497
        }
498
    }
499
500
    fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool {
501
        let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b));
502
        if !ok {
503
            self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -"));
504
        }
505
        ok
506
    }
507
}
508
509
use kdl::{KdlDocument, KdlNode, KdlValue};
510
use miette::SourceSpan;
511
512
fn kind_of(v: &KdlValue) -> &'static str {
513
    match v {
514
        KdlValue::String(_) => "a string",
515
        KdlValue::Integer(_) => "an integer",
516
        KdlValue::Float(_) => "a number",
517
        KdlValue::Bool(_) => "a boolean",
518
        KdlValue::Null => "null",
519
    }
520
}
521
522
/// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB.
523
fn parse_memory(s: &str) -> Option<u64> {
524
    let (num, unit) = s.split_at(s.find(|c: char| !c.is_ascii_digit())?);
525
    let n: u64 = num.parse().ok()?;
526
    let mib = match unit {
527
        "KiB" => n.is_multiple_of(1024).then_some(n / 1024)?,
528
        "MiB" => n,
529
        "GiB" => n.checked_mul(1024)?,
530
        _ => return None,
531
    };
532
    Some(mib)
533
}
534
535
/// The line of the first escape KDL 2.0 does not have but shells and C
536
/// do (`\0NN`, `\e`, `\xNN`), as someone colouring output writes them.
537
fn foreign_escape_line(src: &str) -> Option<usize> {
538
    src.lines()
539
        .position(|l| {
540
            let b = l.as_bytes();
541
            b.windows(2).enumerate().any(|(i, w)| {
542
                w[0] == b'\\'
543
                    && (i == 0 || b[i - 1] != b'\\')
544
                    && (w[1] == b'e' || w[1] == b'x' || (w[1] == b'0' && b.get(i + 2).is_some_and(u8::is_ascii_digit)))
545
            })
546
        })
547
        .map(|i| i + 1)
548
}
549
550
impl Pipeline {
551
    /// Parse and validate a pipeline document.
552
    pub fn parse(src: &str) -> Result<Pipeline, Problems> {
553
        let doc = match KdlDocument::parse_v2(src) {
554
            Ok(d) => d,
555
            Err(e) => {
556
                let mut ctx = Ctx { src, problems: Vec::new() };
557
                for d in &e.diagnostics {
558
                    let msg = d.message.clone().or_else(|| d.label.clone()).unwrap_or_else(|| "not valid KDL 2.0".into());
559
                    ctx.err(d.span, msg);
560
                }
561
                if ctx.problems.is_empty() {
562
                    ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() });
563
                }
564
                if let Some(line) = foreign_escape_line(src) {
565
                    ctx.problems.push(Problem {
566
                        line,
567
                        message: "KDL strings have no \\033, \\e or \\x escapes: write ESC (for ANSI colours) as \\u{1b}".into(),
568
                    });
569
                }
570
                return Err(Problems(ctx.problems));
571
            }
572
        };
573
        let mut ctx = Ctx { src, problems: Vec::new() };
574
        let mut schema = None;
575
        let mut sandboxes = BTreeMap::new();
576
        let mut jobs: BTreeMap<String, (Job, SourceSpan)> = BTreeMap::new();
577
        let mut on = None;
578
        for node in doc.nodes() {
579
            match node.name().value() {
580
                "on" => {
581
                    if on.is_some() {
582
                        ctx.err(node.span(), "`on` is given twice");
583
                    }
584
                    on = Some(parse_on(&mut ctx, node));
585
                }
586
                "schema-version" => {
587
                    ctx.no_children(node);
588
                    match ctx.one_integer(node) {
589
                        Some(1) => schema = Some(1),
590
                        Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")),
591
                        None => {}
592
                    }
593
                }
594
                "sandbox" => {
595
                    if let Some(sb) = parse_sandbox(&mut ctx, node) {
596
                        if sandboxes.contains_key(&sb.name) {
597
                            ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name));
598
                        } else {
599
                            sandboxes.insert(sb.name.clone(), sb);
600
                        }
601
                    }
602
                }
603
                "job" => {
604
                    if let Some(job) = parse_job(&mut ctx, node) {
605
                        if jobs.contains_key(&job.name) {
606
                            ctx.err(node.span(), format!("job {:?} is declared twice", job.name));
607
                        } else {
608
                            jobs.insert(job.name.clone(), (job, node.span()));
609
                        }
610
                    }
611
                }
612
                n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")),
613
                n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, on, sandbox or job")),
614
            }
615
        }
616
        if schema.is_none() && !ctx.problems.iter().any(|p| p.message.contains("schema-version")) {
617
            ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() });
618
        }
619
        if jobs.is_empty() && ctx.problems.is_empty() {
620
            ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() });
621
        }
622
        // References and cycles.
623
        for (job, span) in jobs.values() {
624
            if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) {
625
                ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox));
626
            }
627
            for d in &job.depends_on {
628
                if !jobs.contains_key(d) {
629
                    ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d));
630
                } else if d == &job.name {
631
                    ctx.err(*span, format!("job {:?} depends on itself", job.name));
632
                }
633
            }
634
            // Inputs come from a job this one depends on, which declares
635
            // that artifact (oxygenflow#58).
636
            for i in &job.inputs {
637
                let Some((producer, _)) = jobs.get(&i.from) else {
638
                    ctx.err(*span, format!("job {:?} takes an input from {:?}, which is not declared", job.name, i.from));
639
                    continue;
640
                };
641
                if !job.depends_on.contains(&i.from) {
642
                    ctx.err(*span, format!("job {:?} takes {:?} from {:?}: add depends-on \"{}\"", job.name, i.artifact, i.from, i.from));
643
                }
644
                match producer.outputs.iter().find(|o| o.name == i.artifact) {
645
                    Some(o) if o.kind == OutputKind::Artifact => {}
646
                    Some(_) => ctx.err(*span, format!("{:?} of job {:?} is a report; inputs are artifacts", i.artifact, i.from)),
647
                    None => ctx.err(*span, format!("job {:?} declares no artifact {:?}", i.from, i.artifact)),
648
                }
649
            }
650
        }
651
        let graph: BTreeMap<&str, Vec<&str>> =
652
            jobs.values().map(|(j, _)| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect();
653
        if let Some(cycle) = find_cycle(&graph) {
654
            let span = jobs[cycle[0]].1;
655
            ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → ")));
656
        }
657
        if !ctx.problems.is_empty() {
658
            ctx.problems.sort_by_key(|p| p.line);
659
            return Err(Problems(ctx.problems));
660
        }
661
        Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(|(k, (j, _))| (k, j)).collect(), on })
662
    }
663
664
    /// Jobs in an order where each comes after what it depends on (ties by name).
665
    pub fn order(&self) -> Vec<&Job> {
666
        let mut done: BTreeSet<&str> = BTreeSet::new();
667
        let mut out = Vec::with_capacity(self.jobs.len());
668
        while out.len() < self.jobs.len() {
669
            let before = out.len();
670
            for j in self.jobs.values() {
671
                if !done.contains(j.name.as_str()) && j.depends_on.iter().all(|d| done.contains(d.as_str())) {
672
                    out.push(j);
673
                }
674
            }
675
            for j in &out[before..] {
676
                done.insert(j.name.as_str());
677
            }
678
            if out.len() == before {
679
                break; // validated pipelines have no cycles
680
            }
681
        }
682
        out
683
    }
684
685
    /// Each job's depth: 0 without dependencies, else one more than its deepest dependency.
686
    pub fn depths(&self) -> BTreeMap<&str, usize> {
687
        let mut depth: BTreeMap<&str, usize> = BTreeMap::new();
688
        for j in self.order() {
689
            let d = j.depends_on.iter().map(|x| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0);
690
            depth.insert(j.name.as_str(), d);
691
        }
692
        depth
693
    }
694
}
695
696
fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option<Vec<&'a str>> {
697
    #[derive(Clone, Copy, PartialEq)]
698
    enum Mark {
699
        Visiting,
700
        Done,
701
    }
702
    fn visit<'a>(
703
        n: &'a str,
704
        graph: &BTreeMap<&'a str, Vec<&'a str>>,
705
        marks: &mut BTreeMap<&'a str, Mark>,
706
        stack: &mut Vec<&'a str>,
707
    ) -> Option<Vec<&'a str>> {
708
        match marks.get(n) {
709
            Some(Mark::Done) => return None,
710
            Some(Mark::Visiting) => {
711
                let at = stack.iter().position(|s| *s == n).unwrap_or(0);
712
                let mut c = stack[at..].to_vec();
713
                c.push(n);
714
                return Some(c);
715
            }
716
            None => {}
717
        }
718
        marks.insert(n, Mark::Visiting);
719
        stack.push(n);
720
        for d in graph.get(n).into_iter().flatten() {
721
            if graph.contains_key(d) && *d != n {
722
                if let Some(c) = visit(d, graph, marks, stack) {
723
                    return Some(c);
724
                }
725
            }
726
        }
727
        stack.pop();
728
        marks.insert(n, Mark::Done);
729
        None
730
    }
731
    let mut marks = BTreeMap::new();
732
    for n in graph.keys() {
733
        let mut stack = Vec::new();
734
        if let Some(c) = visit(n, graph, &mut marks, &mut stack) {
735
            return Some(c);
736
        }
737
    }
738
    None
739
}
740
741
fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] {
742
    match node.children() {
743
        Some(d) => d.nodes(),
744
        None => {
745
            ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value()));
746
            &[]
747
        }
748
    }
749
}
750
751
fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option<Sandbox> {
752
    let name = ctx.one_string(node)?;
753
    let named = ctx.name_ok(node.span(), "sandbox", &name);
754
    let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) =
755
        (Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated));
756
    let mut seen = BTreeSet::new();
757
    let mut env = BTreeMap::new();
758
    let mut runs_on = Vec::new();
759
    let mut timeout_s = None;
760
    let mut arch: Option<String> = None;
761
    let mut egress_allow = Vec::new();
762
    for c in block(ctx, node) {
763
        let key = c.name().value();
764
        if key != "runs-on" && !seen.insert(key.to_string()) {
765
            ctx.err(c.span(), format!("`{key}` is given twice"));
766
        }
767
        if key == "env" {
768
            parse_env(ctx, c, &mut env);
769
            continue;
770
        }
771
        if key == "runs-on" {
772
            ctx.no_children(c);
773
            parse_labels(ctx, c, &mut runs_on);
774
            continue;
775
        }
776
        if key != "network" {
777
            ctx.no_children(c);
778
        }
779
        match key {
780
            "runtime" => {
781
                runtime = match ctx.one_string(c).as_deref() {
782
                    Some("microvm") => Some(Runtime::Microvm),
783
                    Some("syscall") | Some("gvisor") => Some(Runtime::Syscall),
784
                    Some("wasm") => {
785
                        ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\"");
786
                        None
787
                    }
788
                    Some(other) => {
789
                        ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\""));
790
                        None
791
                    }
792
                    None => None,
793
                }
794
            }
795
            "rootfs" => {
796
                rootfs = match ctx.one_string(c) {
797
                    Some(r) => match r.strip_prefix("cas://blob/") {
798
                        Some(h) if h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => {
799
                            Some(h.to_string())
800
                        }
801
                        _ => {
802
                            ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits"));
803
                            None
804
                        }
805
                    },
806
                    None => None,
807
                }
808
            }
809
            "arch" => {
810
                arch = match ctx.one_string(c) {
811
                    Some(a) if arch_ok(&a) => Some(a),
812
                    Some(a) => {
813
                        ctx.err(c.span(), format!("arch {a:?} is not an architecture name such as x86_64, aarch64 or riscv64"));
814
                        None
815
                    }
816
                    None => None,
817
                }
818
            }
819
            "memory-limit" => {
820
                memory = match ctx.one_string(c) {
821
                    Some(m) => match parse_memory(&m) {
822
                        Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32),
823
                        Some(_) => {
824
                            ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB"));
825
                            None
826
                        }
827
                        None => {
828
                            ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\""));
829
                            None
830
                        }
831
                    },
832
                    None => None,
833
                }
834
            }
835
            "cpu-limit" => {
836
                cpus = match ctx.one_integer(c) {
837
                    Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32),
838
                    Some(n) => {
839
                        ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}"));
840
                        None
841
                    }
842
                    None => None,
843
                }
844
            }
845
            "network" => {
846
                let kind = ctx.one_string(c);
847
                if kind.as_deref() != Some("egress") {
848
                    ctx.no_children(c);
849
                }
850
                network = match kind.as_deref() {
851
                    Some("isolated") => Some(Network::Isolated),
852
                    // Loopback only: the guest's `lo` comes up, and it still has
853
                    // no device that reaches anything outside it.
854
                    Some("loopback") => Some(Network::Loopback),
855
                    // Allow-listed egress (oxygenflow#55): still no network
856
                    // device; a proxy in the guest relays to the runner, which
857
                    // lets through only the hosts named here.
858
                    Some("egress") => {
859
                        egress_allow = parse_egress(ctx, c);
860
                        Some(Network::Egress)
861
                    }
862
                    Some(other) => {
863
                        ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
864
                        None
865
                    }
866
                    None => None,
867
                }
868
            }
869
            "timeout" => timeout_s = parse_timeout(ctx, c),
870
            other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")),
871
        }
872
    }
873
    if rootfs.is_none() && !seen.contains("rootfs") {
874
        ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
875
    }
876
    if network == Some(Network::Egress) {
877
        if runtime == Some(Runtime::Syscall) {
878
            ctx.err(node.span(), format!("sandbox {name:?}: network \"egress\" needs runtime \"microvm\""));
879
        }
880
        // Only a runner that allows egress offers this label.
881
        if !runs_on.iter().any(|l| l == EGRESS_LABEL) {
882
            runs_on.push(EGRESS_LABEL.to_string());
883
            runs_on.sort();
884
        }
885
    }
886
    if !named {
887
        return None;
888
    }
889
    Some(Sandbox {
890
        name,
891
        runtime: runtime?,
892
        rootfs: rootfs?,
893
        memory_mib: memory?,
894
        cpus: cpus?,
895
        network: network?,
896
        env,
897
        runs_on,
898
        egress_allow,
899
        timeout_s,
900
        arch,
901
    })
902
}
903
904
/// The label a runner offers when it allows egress, and an egress sandbox
905
/// needs (oxygenflow#55).
906
pub const EGRESS_LABEL: &str = "egress";
907
908
/// The most hosts one sandbox may name.
909
const MAX_EGRESS_HOSTS: usize = 32;
910
911
/// Whether `pattern` is a host name, or `*.` and a domain: lower-case
912
/// letters, digits, `-` and dots, at least two labels.
913
pub fn egress_pattern_ok(pattern: &str) -> bool {
914
    let name = pattern.strip_prefix("*.").unwrap_or(pattern);
915
    let labels: Vec<&str> = name.split('.').collect();
916
    name.len() <= 253
917
        && labels.len() >= 2
918
        && labels.iter().all(|l| {
919
            !l.is_empty()
920
                && l.len() <= 63
921
                && !l.starts_with('-')
922
                && !l.ends_with('-')
923
                && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
924
        })
925
}
926
927
/// Whether `host` is one `patterns` allow: a name exactly, or a name under
928
/// a `*.` domain (not the domain itself). Case and a final dot are ignored.
929
pub fn egress_allows(patterns: &[String], host: &str) -> bool {
930
    let host = host.trim_end_matches('.').to_ascii_lowercase();
931
    patterns.iter().any(|p| match p.strip_prefix("*.") {
932
        Some(domain) => host.len() > domain.len() + 1 && host.ends_with(domain) && host[..host.len() - domain.len()].ends_with('.'),
933
        None => host == *p,
934
    })
935
}
936
937
/// `network "egress" { allow "*.googleapis.com" "registry.opentofu.org" }`.
938
fn parse_egress(ctx: &mut Ctx, node: &KdlNode) -> Vec<String> {
939
    let mut out = Vec::new();
940
    for c in block(ctx, node) {
941
        ctx.no_children(c);
942
        if c.name().value() != "allow" {
943
            ctx.err(c.span(), format!("unknown egress setting `{}`; egress takes `allow \"host\"...`", c.name().value()));
944
            continue;
945
        }
946
        for h in ctx.strings(c) {
947
            if egress_pattern_ok(&h) {
948
                out.push(h);
949
            } else {
950
                ctx.err(
951
                    c.span(),
952
                    format!("allow {h:?}: expected a lower-case host name such as \"registry.opentofu.org\", or \"*.\" and a domain"),
953
                );
954
            }
955
        }
956
    }
957
    out.sort();
958
    out.dedup();
959
    if out.is_empty() {
960
        ctx.err(node.span(), "network \"egress\" needs at least one `allow \"host\"`");
961
    } else if out.len() > MAX_EGRESS_HOSTS {
962
        ctx.err(node.span(), format!("network \"egress\" may allow at most {MAX_EGRESS_HOSTS} hosts"));
963
    }
964
    out
965
}
966
967
/// An `on` block: `push` and `tag` with globs (none: any), `manual`,
968
/// `schedule` with five-field cron lines.
969
fn parse_on(ctx: &mut Ctx, node: &KdlNode) -> On {
970
    let mut on = On::default();
971
    if !node.entries().is_empty() {
972
        ctx.err(node.span(), "`on` takes a block: on { push \"main\"; tag \"v*\"; manual }");
973
    }
974
    let globs = |ctx: &mut Ctx, c: &KdlNode| -> Vec<String> {
975
        let g = ctx.strings(c);
976
        for p in &g {
977
            if let Err(e) = globset::Glob::new(p) {
978
                ctx.err(c.span(), format!("{p:?}: {e}"));
979
            }
980
        }
981
        g
982
    };
983
    for c in block(ctx, node) {
984
        ctx.no_children(c);
985
        match c.name().value() {
986
            "push" => on.push.get_or_insert_with(Vec::new).extend(globs(ctx, c)),
987
            "tag" => on.tag.get_or_insert_with(Vec::new).extend(globs(ctx, c)),
988
            "manual" => on.manual = true,
989
            "schedule" => {
990
                for line in ctx.strings(c) {
991
                    if crate::cron::Cron::parse(&line).is_some() {
992
                        on.schedule.push(line);
993
                    } else {
994
                        ctx.err(
995
                            c.span(),
996
                            format!("schedule {line:?}: expected five cron fields (minute hour day month weekday), as \"0 3 * * *\""),
997
                        );
998
                    }
999
                }
1000
            }
1001
            other => ctx.err(c.span(), format!("unknown `on` event `{other}`; expected push, tag, manual or schedule")),
1002
        }
1003
    }
1004
    if on.push.is_none() && on.tag.is_none() && !on.manual && on.schedule.is_empty() {
1005
        ctx.err(node.span(), "`on` names no event, so nothing would run it");
1006
    }
1007
    on
1008
}
1009
1010
/// The longest a pipeline may ask a job to run.
1011
pub const MAX_TIMEOUT_S: u64 = 24 * 3600;
1012
1013
/// A duration: whole hours, minutes and seconds in that order, each
1014
/// optional, as "90m", "2h", "1h30m" or "45s".
1015
pub fn parse_duration(s: &str) -> Option<u64> {
1016
    let (mut total, mut num, mut last) = (0u64, String::new(), 0u8);
1017
    for ch in s.chars() {
1018
        if ch.is_ascii_digit() {
1019
            num.push(ch);
1020
            continue;
1021
        }
1022
        let (unit, mult) = match ch {
1023
            'h' => (1, 3600),
1024
            'm' => (2, 60),
1025
            's' => (3, 1),
1026
            _ => return None,
1027
        };
1028
        // Each unit once, largest first, each with its number.
1029
        if num.is_empty() || unit <= last {
1030
            return None;
1031
        }
1032
        last = unit;
1033
        total = total.checked_add(num.parse::<u64>().ok()?.checked_mul(mult)?)?;
1034
        num.clear();
1035
    }
1036
    (num.is_empty() && total > 0).then_some(total)
1037
}
1038
1039
/// A `timeout` line: one duration, from a minute to a day.
1040
fn parse_timeout(ctx: &mut Ctx, c: &KdlNode) -> Option<u64> {
1041
    let t = ctx.one_string(c)?;
1042
    match parse_duration(&t) {
1043
        Some(secs) if (60..=MAX_TIMEOUT_S).contains(&secs) => Some(secs),
1044
        Some(_) => {
1045
            ctx.err(c.span(), format!("timeout {t:?} must be from 1m to 24h"));
1046
            None
1047
        }
1048
        None => {
1049
            ctx.err(c.span(), format!("timeout {t:?}: expected hours, minutes and seconds, as \"90m\" or \"1h30m\""));
1050
            None
1051
        }
1052
    }
1053
}
1054
1055
/// The labels of a `runs-on` line: lower-case letters, digits, `-`, `_`
1056
/// and `.`, at most 64 bytes each; into `into`, sorted, once each.
1057
fn parse_labels(ctx: &mut Ctx, c: &KdlNode, into: &mut Vec<String>) {
1058
    let labels = ctx.strings(c);
1059
    if labels.is_empty() {
1060
        ctx.err(c.span(), "`runs-on` needs one or more labels, as runs-on \"riscv64\"");
1061
    }
1062
    for l in labels {
1063
        let ok = !l.is_empty()
1064
            && l.len() <= 64
1065
            && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'-' | b'_' | b'.'));
1066
        if ok {
1067
            into.push(l);
1068
        } else {
1069
            ctx.err(c.span(), format!("runs-on {l:?}: a label is lower-case letters, digits, -, _ and ., at most 64"));
1070
        }
1071
    }
1072
    into.sort();
1073
    into.dedup();
1074
}
1075
1076
/// An `env` block of `NAME "value"` lines, into `env`.
1077
fn parse_env(ctx: &mut Ctx, c: &KdlNode, env: &mut BTreeMap<String, String>) {
1078
    if !c.entries().is_empty() {
1079
        ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines");
1080
    }
1081
    for e in block(ctx, c) {
1082
        let k = e.name().value().to_string();
1083
        let valid =
1084
            !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
1085
        if !valid {
1086
            ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _"));
1087
            continue;
1088
        }
1089
        ctx.no_children(e);
1090
        if let Some(v) = ctx.one_string(e) {
1091
            if env.insert(k.clone(), v).is_some() {
1092
                ctx.err(e.span(), format!("environment {k} is set twice"));
1093
            }
1094
        }
1095
    }
1096
}
1097
1098
/// `mount { source "ephemeral://disk"; target "/src/target"; size "8GiB" }`.
1099
fn parse_mount(ctx: &mut Ctx, node: &KdlNode) -> Option<Mount> {
1100
    let (mut source, mut target, mut size) = (None, None, None);
1101
    let (mut cache, mut keys) = (None, Vec::new());
1102
    let mut seen = BTreeSet::new();
1103
    for c in block(ctx, node) {
1104
        let key = c.name().value();
1105
        if !seen.insert(key.to_string()) {
1106
            ctx.err(c.span(), format!("`{key}` is given twice"));
1107
            continue;
1108
        }
1109
        ctx.no_children(c);
1110
        match key {
1111
            "source" => {
1112
                source = match ctx.one_string(c).as_deref() {
1113
                    Some("ephemeral://tmpfs") => Some(MountSource::Tmpfs),
1114
                    Some("ephemeral://disk") => Some(MountSource::Disk),
1115
                    Some(other) => {
1116
                        ctx.err(
1117
                            c.span(),
1118
                            format!("mount source {other:?} is not in M0 yet; M0 mounts \"ephemeral://tmpfs\" or \"ephemeral://disk\""),
1119
                        );
1120
                        None
1121
                    }
1122
                    None => None,
1123
                }
1124
            }
1125
            "target" => {
1126
                target = match ctx.one_string(c) {
1127
                    Some(t) if mount_target_ok(&t) => Some(t),
1128
                    Some(t) => {
1129
                        ctx.err(
1130
                            c.span(),
1131
                            format!("mount target {t:?} must be an absolute path under /src or /tmp, without . or .. or a trailing /"),
1132
                        );
1133
                        None
1134
                    }
1135
                    None => None,
1136
                }
1137
            }
1138
            "size" => {
1139
                size = match ctx.one_string(c) {
1140
                    Some(m) => match parse_memory(&m) {
1141
                        Some(mib) if (MIN_MOUNT_MIB..=MAX_MOUNT_MIB).contains(&mib) => Some(mib as u32),
1142
                        Some(_) => {
1143
                            ctx.err(c.span(), format!("mount size {m:?} must be from 16MiB to 64GiB"));
1144
                            None
1145
                        }
1146
                        None => {
1147
                            ctx.err(c.span(), format!("mount size {m:?}: expected a whole number of KiB, MiB or GiB, as \"8GiB\""));
1148
                            None
1149
                        }
1150
                    },
1151
                    None => None,
1152
                }
1153
            }
1154
            "cache" => {
1155
                cache = match ctx.one_string(c) {
1156
                    Some(n) if !n.is_empty() && n.len() <= 64 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)) => {
1157
                        Some(n)
1158
                    }
1159
                    Some(n) => {
1160
                        ctx.err(c.span(), format!("cache name {n:?}: letters, digits, '.', '_' and '-', at most 64"));
1161
                        None
1162
                    }
1163
                    None => None,
1164
                }
1165
            }
1166
            "key" => {
1167
                keys = ctx.strings(c);
1168
                if keys.is_empty() {
1169
                    ctx.err(c.span(), "`key` names one or more files, as \"Cargo.lock\"");
1170
                }
1171
                for k in &keys {
1172
                    if k.starts_with('/') || k.split('/').any(|p| p.is_empty() || p == "." || p == "..") {
1173
                        ctx.err(c.span(), format!("key file {k:?} must be a path in the repository, without . or .."));
1174
                    }
1175
                }
1176
            }
1177
            "readonly" | "include" => ctx.err(c.span(), format!("mount `{key}` is not in M0 yet")),
1178
            other => ctx.err(c.span(), format!("unknown mount setting `{other}`")),
1179
        }
1180
    }
1181
    for (name, missing) in [("source", source.is_none()), ("target", target.is_none()), ("size", size.is_none())] {
1182
        if missing && !seen.contains(name) {
1183
            ctx.err(node.span(), format!("a mount needs a {name}"));
1184
        }
1185
    }
1186
    if source == Some(MountSource::Tmpfs) && cache.is_some() {
1187
        ctx.err(node.span(), "only an ephemeral://disk mount can be a cache");
1188
    }
1189
    if !keys.is_empty() && cache.is_none() {
1190
        ctx.err(node.span(), "`key` goes with `cache`");
1191
    }
1192
    Some(Mount { source: source?, target: target?, size_mib: size?, cache, key: keys })
1193
}
1194
1195
/// `output { artifact "name" { path "/src/target/release/tool" } }`.
1196
/// `secret "deploy-token" { mount-path "/run/secrets/token" }` (spec §9.1).
1197
/// The hub is the vault: the secret of that name set for the repository,
1198
/// its organisation or the site. `vault-ref` is not in this slice.
1199
fn parse_secret(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
1200
    let Some(name) = ctx.one_string(node) else {
1201
        ctx.err(node.span(), "`secret` takes a name and a block with mount-path");
1202
        return;
1203
    };
1204
    if !ctx.name_ok(node.span(), "secret", &name) {
1205
        return;
1206
    }
1207
    let mut path = None;
1208
    for c in block(ctx, node) {
1209
        ctx.no_children(c);
1210
        match c.name().value() {
1211
            "mount-path" => match ctx.one_string(c) {
1212
                Some(p) if secret_path_ok(&p) => path = Some(p),
1213
                Some(p) => {
1214
                    ctx.err(c.span(), format!("secret mount-path {p:?} must be an absolute path under /run/secrets/, without . or .."))
1215
                }
1216
                None => {}
1217
            },
1218
            "vault-ref" => ctx.err(c.span(), "`vault-ref` is not in M0; the hub supplies the secret of that name"),
1219
            other => ctx.err(c.span(), format!("unknown secret setting `{other}`")),
1220
        }
1221
    }
1222
    let Some(mount_path) = path else {
1223
        ctx.err(node.span(), format!("secret {name:?} needs a mount-path"));
1224
        return;
1225
    };
1226
    if job.secrets.iter().any(|s| s.name == name) {
1227
        ctx.err(node.span(), format!("secret {name:?} is given twice"));
1228
    } else if job.secrets.iter().any(|s| s.mount_path == mount_path) || job.identities.iter().any(|i| i.mount_path == mount_path) {
1229
        ctx.err(node.span(), format!("two files mount at {mount_path:?}"));
1230
    } else if job.secrets.len() == MAX_SECRETS {
1231
        ctx.err(node.span(), format!("a job has at most {MAX_SECRETS} secrets"));
1232
    } else {
1233
        job.secrets.push(Secret { name, mount_path });
1234
    }
1235
}
1236
1237
/// `identity "sts.amazonaws.com" { mount-path "/run/secrets/aws-token" }`:
1238
/// the hub mints an OIDC token for that audience and the runner mounts it
1239
/// as a file (oxygenhub#103).
1240
fn parse_identity(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
1241
    let Some(audience) = ctx.one_string(node) else {
1242
        ctx.err(node.span(), "`identity` takes an audience and a block with mount-path");
1243
        return;
1244
    };
1245
    let aud_ok = !audience.is_empty() && audience.len() <= 255 && !audience.chars().any(|c| c.is_whitespace() || c.is_control());
1246
    if !aud_ok {
1247
        ctx.err(node.span(), format!("identity audience {audience:?} must be 1 to 255 characters without spaces"));
1248
        return;
1249
    }
1250
    let mut path = None;
1251
    for c in block(ctx, node) {
1252
        ctx.no_children(c);
1253
        match c.name().value() {
1254
            "mount-path" => match ctx.one_string(c) {
1255
                Some(p) if secret_path_ok(&p) => path = Some(p),
1256
                Some(p) => {
1257
                    ctx.err(c.span(), format!("identity mount-path {p:?} must be an absolute path under /run/secrets/, without . or .."))
1258
                }
1259
                None => {}
1260
            },
1261
            other => ctx.err(c.span(), format!("unknown identity setting `{other}`")),
1262
        }
1263
    }
1264
    let Some(mount_path) = path else {
1265
        ctx.err(node.span(), format!("identity {audience:?} needs a mount-path"));
1266
        return;
1267
    };
1268
    if job.identities.iter().any(|i| i.audience == audience) {
1269
        ctx.err(node.span(), format!("identity {audience:?} is given twice"));
1270
    } else if job.identities.iter().any(|i| i.mount_path == mount_path) || job.secrets.iter().any(|s| s.mount_path == mount_path) {
1271
        ctx.err(node.span(), format!("two files mount at {mount_path:?}"));
1272
    } else if job.identities.len() == MAX_IDENTITIES {
1273
        ctx.err(node.span(), format!("a job has at most {MAX_IDENTITIES} identities"));
1274
    } else {
1275
        job.identities.push(Identity { audience, mount_path });
1276
    }
1277
}
1278
1279
fn parse_output(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
1280
    if !node.entries().is_empty() {
1281
        ctx.err(node.span(), "`output` takes a block of artifacts and reports");
1282
    }
1283
    for a in block(ctx, node) {
1284
        let kind = match a.name().value() {
1285
            "artifact" => OutputKind::Artifact,
1286
            "report" => OutputKind::Report,
1287
            other => {
1288
                ctx.err(
1289
                    a.span(),
1290
                    format!("unknown output `{other}`; outputs are `artifact \"name\" {{ path \"/src/file\" }}` and `report \"name\" {{ path \"/src/dir\" }}`"),
1291
                );
1292
                continue;
1293
            }
1294
        };
1295
        let word = kind.word();
1296
        let Some(name) = ctx.one_string(a) else { continue };
1297
        if !ctx.name_ok(a.span(), word, &name) {
1298
            continue;
1299
        }
1300
        let mut path = None;
1301
        let mut always = false;
1302
        for c in block(ctx, a) {
1303
            ctx.no_children(c);
1304
            match c.name().value() {
1305
                "path" => match ctx.one_string(c) {
1306
                    Some(p) if output_path_ok(&p) => path = Some(p),
1307
                    Some(p) => ctx
1308
                        .err(c.span(), format!("{word} path {p:?} must be an absolute path under /src, without . or .., quotes or spaces")),
1309
                    None => {}
1310
                },
1311
                "format" if kind == OutputKind::Artifact => {
1312
                    if ctx.one_string(c).as_deref() != Some("blob") {
1313
                        ctx.err(c.span(), "artifact format is \"blob\"");
1314
                    }
1315
                }
1316
                "when" => match ctx.one_string(c).as_deref() {
1317
                    Some("passed") => always = false,
1318
                    Some("always") => always = true,
1319
                    Some(_) => ctx.err(c.span(), format!("{word} `when` is \"passed\" (the default) or \"always\"")),
1320
                    None => {}
1321
                },
1322
                other => ctx.err(c.span(), format!("unknown {word} setting `{other}`")),
1323
            }
1324
        }
1325
        let Some(path) = path else {
1326
            ctx.err(a.span(), format!("{word} {name:?} needs a path"));
1327
            continue;
1328
        };
1329
        if job.outputs.iter().any(|o| o.name == name) {
1330
            ctx.err(a.span(), format!("output {name:?} is given twice (artifacts and reports share names)"));
1331
        } else if job.outputs.len() == MAX_OUTPUTS {
1332
            ctx.err(a.span(), format!("a job has at most {MAX_OUTPUTS} artifacts and reports"));
1333
        } else {
1334
            job.outputs.push(Output { name, path, always, kind });
1335
        }
1336
    }
1337
}
1338
1339
/// `approval { by "maintain"; others-only #true; expires "24h" }`, each
1340
/// optional: `maintain`, anyone with the role, 24 hours.
1341
/// The approval, and whether it said `sealed` itself (else the job's
1342
/// environment decides).
1343
fn parse_approval(ctx: &mut Ctx, node: &KdlNode) -> Option<(Approval, bool)> {
1344
    if !node.entries().is_empty() {
1345
        ctx.err(node.span(), "`approval` takes a block: approval { by \"maintain\"; others-only #true; expires \"24h\" }");
1346
    }
1347
    let mut a = Approval { by: "maintain".into(), others_only: false, expires_s: 24 * 3600, sealed: false };
1348
    let mut sealed_said = false;
1349
    let Some(children) = node.children() else { return Some((a, false)) };
1350
    for c in children.nodes() {
1351
        ctx.no_children(c);
1352
        match c.name().value() {
1353
            "by" => match ctx.one_string(c).as_deref() {
1354
                Some(r @ ("write" | "maintain" | "admin")) => a.by = r.to_string(),
1355
                Some(r) => ctx.err(c.span(), format!("approval by {r:?}: expected \"write\", \"maintain\" or \"admin\"")),
1356
                None => {}
1357
            },
1358
            "others-only" => match c.entries() {
1359
                [e] if e.name().is_none() => match e.value() {
1360
                    KdlValue::Bool(b) => a.others_only = *b,
1361
                    _ => ctx.err(c.span(), "`others-only` takes #true or #false"),
1362
                },
1363
                _ => ctx.err(c.span(), "`others-only` takes #true or #false"),
1364
            },
1365
            "sealed" => match c.entries() {
1366
                [e] if e.name().is_none() => match e.value() {
1367
                    KdlValue::Bool(b) => {
1368
                        a.sealed = *b;
1369
                        sealed_said = true;
1370
                    }
1371
                    _ => ctx.err(c.span(), "`sealed` takes #true or #false"),
1372
                },
1373
                _ => ctx.err(c.span(), "`sealed` takes #true or #false"),
1374
            },
1375
            "expires" => {
1376
                let Some(t) = ctx.one_string(c) else { continue };
1377
                match parse_duration(&t) {
1378
                    Some(secs) if (60..=MAX_APPROVAL_S).contains(&secs) => a.expires_s = secs,
1379
                    Some(_) => ctx.err(c.span(), format!("expires {t:?} must be from 1m to 720h")),
1380
                    None => ctx.err(c.span(), format!("expires {t:?}: expected hours, minutes and seconds, as \"24h\"")),
1381
                }
1382
            }
1383
            other => {
1384
                ctx.err(c.span(), format!("unknown approval setting `{other}`; approval takes `by`, `others-only`, `expires` and `sealed`"))
1385
            }
1386
        }
1387
    }
1388
    Some((a, sealed_said))
1389
}
1390
1391
/// `input { artifact "NAME" from "JOB"; path "/src/FILE" }`.
1392
fn parse_input(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
1393
    if !node.entries().is_empty() {
1394
        ctx.err(node.span(), "`input` takes a block: input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\" }");
1395
    }
1396
    let (mut artifact, mut from, mut path) = (None, None, None);
1397
    for c in block(ctx, node) {
1398
        ctx.no_children(c);
1399
        match c.name().value() {
1400
            "artifact" => match ctx.strings(c).as_slice() {
1401
                [name, word, producer] if word == "from" => {
1402
                    if ctx.name_ok(c.span(), "artifact", name) && ctx.name_ok(c.span(), "job", producer) {
1403
                        artifact = Some(name.clone());
1404
                        from = Some(producer.clone());
1405
                    }
1406
                }
1407
                _ => ctx.err(c.span(), "`artifact` names the artifact and its job: artifact \"tfplan\" from \"plan\""),
1408
            },
1409
            "path" => match ctx.one_string(c) {
1410
                Some(p) if output_path_ok(&p) => path = Some(p),
1411
                Some(p) => {
1412
                    ctx.err(c.span(), format!("input path {p:?} must be an absolute path under /src, without . or .., quotes or spaces"))
1413
                }
1414
                None => {}
1415
            },
1416
            other => ctx.err(c.span(), format!("unknown input setting `{other}`; an input has `artifact` and `path`")),
1417
        }
1418
    }
1419
    match (artifact, from, path) {
1420
        (Some(artifact), Some(from), Some(path)) => {
1421
            if job.inputs.iter().any(|i| i.path == path) {
1422
                ctx.err(node.span(), format!("two inputs are put at {path:?}"));
1423
            } else if job.inputs.len() == MAX_INPUTS {
1424
                ctx.err(node.span(), format!("a job takes at most {MAX_INPUTS} inputs"));
1425
            } else {
1426
                job.inputs.push(Input { artifact, from, path });
1427
            }
1428
        }
1429
        (a, _, p) => {
1430
            if a.is_some() && p.is_none() {
1431
                ctx.err(node.span(), "an input needs a `path`");
1432
            } else if a.is_none() && p.is_some() {
1433
                ctx.err(node.span(), "an input needs `artifact \"NAME\" from \"JOB\"`");
1434
            }
1435
        }
1436
    }
1437
}
1438
1439
fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option<Job> {
1440
    let name = ctx.one_string(node)?;
1441
    let named = ctx.name_ok(node.span(), "job", &name);
1442
    let mut approval: Option<(Approval, bool)> = None;
1443
    let mut job = Job {
1444
        name: name.clone(),
1445
        sandbox: String::new(),
1446
        scope: vec![],
1447
        depends_on: vec![],
1448
        env: BTreeMap::new(),
1449
        exec: vec![],
1450
        mounts: vec![],
1451
        secrets: vec![],
1452
        identities: vec![],
1453
        outputs: vec![],
1454
        inputs: vec![],
1455
        approval: None,
1456
        environment: None,
1457
        runs_on: vec![],
1458
        timeout_s: None,
1459
        on: None,
1460
    };
1461
    let mut seen = BTreeSet::new();
1462
    let mut outputs_at = None;
1463
    for c in block(ctx, node) {
1464
        let key = c.name().value();
1465
        if key != "scope"
1466
            && key != "depends-on"
1467
            && key != "mount"
1468
            && key != "secret"
1469
            && key != "identity"
1470
            && key != "runs-on"
1471
            && !seen.insert(key.to_string())
1472
        {
1473
            ctx.err(c.span(), format!("`{key}` is given twice"));
1474
        }
1475
        match key {
1476
            "sandbox" => {
1477
                ctx.no_children(c);
1478
                job.sandbox = ctx.one_string(c).unwrap_or_default();
1479
            }
1480
            "scope" => {
1481
                ctx.no_children(c);
1482
                for g in ctx.strings(c) {
1483
                    match globset::Glob::new(&g) {
1484
                        Ok(_) => job.scope.push(g),
1485
                        Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")),
1486
                    }
1487
                }
1488
            }
1489
            "depends-on" => {
1490
                ctx.no_children(c);
1491
                job.depends_on.extend(ctx.strings(c));
1492
            }
1493
            "runs-on" => {
1494
                ctx.no_children(c);
1495
                parse_labels(ctx, c, &mut job.runs_on);
1496
            }
1497
            "timeout" => {
1498
                ctx.no_children(c);
1499
                job.timeout_s = parse_timeout(ctx, c);
1500
            }
1501
            "on" => job.on = Some(parse_on(ctx, c)),
1502
            "env" => parse_env(ctx, c, &mut job.env),
1503
            "exec" => {
1504
                ctx.no_children(c);
1505
                job.exec = ctx.strings(c);
1506
                if job.exec.is_empty() {
1507
                    ctx.err(c.span(), "`exec` needs a program and its arguments");
1508
                } else if job.exec.iter().any(|a| a.contains('\0')) {
1509
                    ctx.err(c.span(), "`exec` arguments cannot contain NUL");
1510
                }
1511
            }
1512
            "mount" => {
1513
                if !c.entries().is_empty() {
1514
                    ctx.err(c.span(), "`mount` takes a block of source, target and size");
1515
                }
1516
                if let Some(m) = parse_mount(ctx, c) {
1517
                    if job.mounts.len() == MAX_MOUNTS {
1518
                        ctx.err(c.span(), format!("a job has at most {MAX_MOUNTS} mounts"));
1519
                    } else if let Some(other) = job.mounts.iter().find(|o| {
1520
                        o.target == m.target
1521
                            || m.target.starts_with(&format!("{}/", o.target))
1522
                            || o.target.starts_with(&format!("{}/", m.target))
1523
                    }) {
1524
                        ctx.err(c.span(), format!("mount {:?} overlaps mount {:?}", m.target, other.target));
1525
                    } else {
1526
                        job.mounts.push(m);
1527
                    }
1528
                }
1529
            }
1530
            "output" => {
1531
                outputs_at = Some(c.span());
1532
                parse_output(ctx, c, &mut job);
1533
            }
1534
            "input" => parse_input(ctx, c, &mut job),
1535
            "approval" => approval = parse_approval(ctx, c),
1536
            "environment" => {
1537
                ctx.no_children(c);
1538
                if let Some(e) = ctx.one_string(c) {
1539
                    let ok = !e.is_empty() && e.len() <= 32 && e.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-');
1540
                    if ok {
1541
                        job.environment = Some(e);
1542
                    } else {
1543
                        ctx.err(c.span(), format!("environment {e:?}: 1 to 32 of a-z 0-9 -, as \"prod\""));
1544
                    }
1545
                }
1546
            }
1547
            "secret" => parse_secret(ctx, c, &mut job),
1548
            "identity" => parse_identity(ctx, c, &mut job),
1549
            n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")),
1550
            other => ctx.err(c.span(), format!("unknown job setting `{other}`")),
1551
        }
1552
    }
1553
    // A job for an environment takes a sealed approval unless it says not.
1554
    job.approval = approval.map(|(mut a, said)| {
1555
        if !said {
1556
            a.sealed = job.environment.is_some();
1557
        }
1558
        a
1559
    });
1560
    job.mounts.sort_by(|a, b| a.target.cmp(&b.target));
1561
    job.outputs.sort_by(|a, b| a.name.cmp(&b.name));
1562
    job.inputs.sort_by(|a, b| a.path.cmp(&b.path));
1563
    job.secrets.sort_by(|a, b| a.name.cmp(&b.name));
1564
    job.identities.sort_by(|a, b| a.audience.cmp(&b.audience));
1565
    // A tmpfs is gone when the job ends; nothing on it can be kept.
1566
    for o in &job.outputs {
1567
        if let Some(m) = job.mounts.iter().find(|m| m.source == MountSource::Tmpfs && o.path.starts_with(&format!("{}/", m.target))) {
1568
            ctx.err(
1569
                outputs_at.unwrap_or(node.span()),
1570
                format!(
1571
                    "artifact {:?} is on the tmpfs mount {:?}, which is gone when the job ends; use ephemeral://disk",
1572
                    o.name, m.target
1573
                ),
1574
            );
1575
        }
1576
    }
1577
    if job.sandbox.is_empty() && !seen.contains("sandbox") {
1578
        ctx.err(node.span(), format!("job {name:?} needs a sandbox"));
1579
    }
1580
    if job.exec.is_empty() && !seen.contains("exec") {
1581
        ctx.err(node.span(), format!("job {name:?} needs an exec"));
1582
    }
1583
    job.depends_on.sort();
1584
    job.depends_on.dedup();
1585
    // Kept even when incomplete, so its references are checked too; the
1586
    // problems above already make the pipeline invalid.
1587
    named.then_some(job)
1588
}