OXYGENHUB
# 720eb8191059 oxygenflow / crates / flow-core / src / pipeline.rs
41.4 KiB · Rust · last changed A job's identity: identity blocks name an audience and a mount path 1 month 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
}
20
21
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
22
#[serde(rename_all = "lowercase")]
23
pub enum Runtime {
24
    /// Tier 0 (spec §4.1); refused in M0.
25
    Wasm,
26
    /// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups.
27
    Syscall,
28
    /// Tier 2 (spec §4.3), the default.
29
    Microvm,
30
}
31
32
impl Runtime {
33
    pub fn as_str(self) -> &'static str {
34
        match self {
35
            Runtime::Wasm => "wasm",
36
            Runtime::Syscall => "syscall",
37
            Runtime::Microvm => "microvm",
38
        }
39
    }
40
}
41
42
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
43
#[serde(rename_all = "lowercase")]
44
pub enum Network {
45
    Isolated,
46
    Loopback,
47
    Egress,
48
}
49
50
impl Network {
51
    pub fn as_str(self) -> &'static str {
52
        match self {
53
            Network::Isolated => "isolated",
54
            Network::Loopback => "loopback",
55
            Network::Egress => "egress",
56
        }
57
    }
58
}
59
60
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
61
pub struct Sandbox {
62
    pub name: String,
63
    pub runtime: Runtime,
64
    /// BLAKE3 of the rootfs image, hex.
65
    pub rootfs: String,
66
    pub memory_mib: u32,
67
    pub cpus: u32,
68
    pub network: Network,
69
    /// Environment every job in this sandbox starts from; a job's own `env`
70
    /// overrides it.
71
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
72
    pub env: BTreeMap<String, String>,
73
    /// Labels a runner must have to run jobs here (`runs-on`), sorted.
74
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
75
    pub runs_on: Vec<String>,
76
    /// How long each job here may run, in seconds (`timeout "2h"`), when
77
    /// the job does not say; the hub caps it at the site's maximum.
78
    #[serde(default, skip_serializing_if = "Option::is_none")]
79
    pub timeout_s: Option<u64>,
80
}
81
82
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
83
pub struct Job {
84
    pub name: String,
85
    pub sandbox: String,
86
    /// Glob patterns over repository paths (spec §6.2).
87
    pub scope: Vec<String>,
88
    /// Sorted.
89
    pub depends_on: Vec<String>,
90
    pub env: BTreeMap<String, String>,
91
    /// Executed as is: no shell, no expansion (spec §2.2 I2).
92
    pub exec: Vec<String>,
93
    /// Ephemeral scratch mounts, sorted by target (spec §3.1 `mount`).
94
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
95
    pub mounts: Vec<Mount>,
96
    /// Files the job produces, sorted by name (spec §5.2 `output`).
97
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
98
    pub outputs: Vec<Output>,
99
    /// Secrets the job mounts, sorted by name (spec §9 `secret`): the hub
100
    /// supplies each by name, the runner mounts it as a file.
101
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
102
    pub secrets: Vec<Secret>,
103
    /// Identities the job asks the hub for (oxygenhub#103): an OIDC token
104
    /// for an audience, mounted as a file like a secret.
105
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
106
    pub identities: Vec<Identity>,
107
    /// Labels a runner must have to take this job (`runs-on`), sorted;
108
    /// with its sandbox's, what the hub matches against a runner's.
109
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
110
    pub runs_on: Vec<String>,
111
    /// How long the job may run, in seconds (`timeout "90m"`); its
112
    /// sandbox's when absent, else the site's default (oxygenflow#23).
113
    #[serde(default, skip_serializing_if = "Option::is_none")]
114
    pub timeout_s: Option<u64>,
115
}
116
117
impl Job {
118
    /// The job's environment on top of its sandbox's: what it runs with.
119
    pub fn in_sandbox(&self, sandbox: &Sandbox) -> Job {
120
        let mut job = self.clone();
121
        for (k, v) in &sandbox.env {
122
            job.env.entry(k.clone()).or_insert_with(|| v.clone());
123
        }
124
        job.runs_on.extend(sandbox.runs_on.iter().cloned());
125
        job.runs_on.sort();
126
        job.runs_on.dedup();
127
        job.timeout_s = job.timeout_s.or(sandbox.timeout_s);
128
        job
129
    }
130
131
    /// The job with variables that tell it what it builds (`FLOW_STATE`,
132
    /// `FLOW_BOOKMARK` …) added to its environment; a variable the pipeline
133
    /// sets itself keeps the pipeline's value.
134
    pub fn with_flow_env(&self, vars: &[(String, String)]) -> Job {
135
        let mut job = self.clone();
136
        for (k, v) in vars {
137
            job.env.entry(k.clone()).or_insert_with(|| v.clone());
138
        }
139
        job
140
    }
141
}
142
143
/// An ephemeral scratch mount: empty at the start of every job and gone
144
/// after it, so it cannot carry anything between jobs (spec §2.2 I4).
145
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
146
pub struct Mount {
147
    pub source: MountSource,
148
    /// An absolute path under `/src` or `/tmp`.
149
    pub target: String,
150
    pub size_mib: u32,
151
}
152
153
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
154
#[serde(rename_all = "lowercase")]
155
pub enum MountSource {
156
    /// `ephemeral://tmpfs`: memory in the sandbox, counted against its memory-limit.
157
    Tmpfs,
158
    /// `ephemeral://disk`: a fresh sparse disk on the runner, for large build trees.
159
    Disk,
160
}
161
162
impl MountSource {
163
    pub fn as_str(self) -> &'static str {
164
        match self {
165
            MountSource::Tmpfs => "ephemeral://tmpfs",
166
            MountSource::Disk => "ephemeral://disk",
167
        }
168
    }
169
}
170
171
/// A file a job produces (spec §5.2): read from the job's disks when it
172
/// ends, kept in the repository as a blob, and named by its receipt.
173
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
174
pub struct Output {
175
    /// The artifact's name, unique in the job.
176
    pub name: String,
177
    /// A clean absolute path under `/src`, not on a tmpfs mount.
178
    pub path: String,
179
}
180
181
/// A secret the job reads as a file (spec §9): never in the environment.
182
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
183
pub struct Secret {
184
    /// Which secret, as the hub names it; unique in the job.
185
    pub name: String,
186
    /// A clean absolute path under `/run/secrets/`, unique in the job;
187
    /// the file is 0400, the job's user's, on an in-memory mount.
188
    pub mount_path: String,
189
}
190
191
/// An OIDC token the hub mints for the job (oxygenhub#103), naming the
192
/// repository, bookmark, run and job, for `audience`; read as a file.
193
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
194
pub struct Identity {
195
    /// Who the token is for (`sts.amazonaws.com`, a Vault role's audience, …).
196
    pub audience: String,
197
    /// A clean absolute path under `/run/secrets/`, unique among the
198
    /// job's secrets and identities.
199
    pub mount_path: String,
200
}
201
202
/// A secret's mount path: under `/run/secrets/`, no `.` or `..`.
203
pub fn secret_path_ok(p: &str) -> bool {
204
    p.strip_prefix("/run/secrets/").is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != ".."))
205
}
206
207
/// Artifacts a job may declare.
208
pub const MAX_OUTPUTS: usize = 16;
209
210
/// Whether `p` is a clean absolute path strictly under `/src`, with nothing
211
/// a debugfs command line would have to quote.
212
fn output_path_ok(p: &str) -> bool {
213
    p.strip_prefix("/src/").is_some_and(|rest| {
214
        !rest.is_empty()
215
            && rest.split('/').all(|c| !c.is_empty() && c != "." && c != "..")
216
            && p.bytes().all(|b| b.is_ascii_graphic() && b != b'"' && b != b'\\')
217
    })
218
}
219
220
/// Mounts a job may declare.
221
pub const MAX_MOUNTS: usize = 4;
222
/// The most secrets a job may mount.
223
pub const MAX_SECRETS: usize = 8;
224
/// The most identities a job may ask for.
225
pub const MAX_IDENTITIES: usize = 4;
226
/// Sizes a mount may have: 16 MiB to 64 GiB.
227
pub const MIN_MOUNT_MIB: u64 = 16;
228
pub const MAX_MOUNT_MIB: u64 = 64 * 1024;
229
230
/// Whether `t` is a clean absolute path strictly under `/src` or `/tmp`.
231
fn mount_target_ok(t: &str) -> bool {
232
    let under = t.strip_prefix("/src/").or_else(|| t.strip_prefix("/tmp/"));
233
    under.is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != ".."))
234
}
235
236
/// One problem in a pipeline, at a 1-based line.
237
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
238
pub struct Problem {
239
    pub line: usize,
240
    pub message: String,
241
}
242
243
impl fmt::Display for Problem {
244
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
245
        write!(f, "line {}: {}", self.line, self.message)
246
    }
247
}
248
249
/// Every problem found, in line order.
250
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
251
pub struct Problems(pub Vec<Problem>);
252
253
impl fmt::Display for Problems {
254
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
255
        let lines: Vec<String> = self.0.iter().map(|p| p.to_string()).collect();
256
        f.write_str(&lines.join("\n"))
257
    }
258
}
259
260
const MIN_MEMORY_MIB: u64 = 64;
261
const MAX_MEMORY_MIB: u64 = 16 * 1024;
262
const MAX_CPUS: i128 = 16;
263
264
/// Nodes of the full spec that M0 recognises and refuses.
265
const LATER_TOP: &[&str] = &["publish", "secret"];
266
const LATER_JOB: &[&str] = &["attest"];
267
268
struct Ctx<'a> {
269
    src: &'a str,
270
    problems: Vec<Problem>,
271
}
272
273
impl Ctx<'_> {
274
    fn line(&self, span: SourceSpan) -> usize {
275
        let at = span.offset().min(self.src.len());
276
        self.src[..at].bytes().filter(|b| *b == b'\n').count() + 1
277
    }
278
279
    fn err(&mut self, span: SourceSpan, message: impl Into<String>) {
280
        let line = self.line(span);
281
        self.problems.push(Problem { line, message: message.into() });
282
    }
283
284
    /// The node's arguments as strings; properties and other types are errors.
285
    fn strings(&mut self, node: &KdlNode) -> Vec<String> {
286
        let mut out = Vec::new();
287
        for e in node.entries() {
288
            if let Some(k) = e.name() {
289
                self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value()));
290
                continue;
291
            }
292
            match e.value() {
293
                KdlValue::String(s) => out.push(s.clone()),
294
                other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))),
295
            }
296
        }
297
        out
298
    }
299
300
    /// Exactly one string argument.
301
    fn one_string(&mut self, node: &KdlNode) -> Option<String> {
302
        let v = self.strings(node);
303
        if v.len() != 1 {
304
            self.err(node.span(), format!("`{}` takes one string", node.name().value()));
305
            return None;
306
        }
307
        v.into_iter().next()
308
    }
309
310
    fn one_integer(&mut self, node: &KdlNode) -> Option<i128> {
311
        match node.entries() {
312
            [e] if e.name().is_none() => match e.value() {
313
                KdlValue::Integer(n) => Some(*n),
314
                other => {
315
                    self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other)));
316
                    None
317
                }
318
            },
319
            _ => {
320
                self.err(node.span(), format!("`{}` takes one integer", node.name().value()));
321
                None
322
            }
323
        }
324
    }
325
326
    fn no_children(&mut self, node: &KdlNode) {
327
        if node.children().is_some() {
328
            self.err(node.span(), format!("`{}` takes no block", node.name().value()));
329
        }
330
    }
331
332
    fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool {
333
        let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b));
334
        if !ok {
335
            self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -"));
336
        }
337
        ok
338
    }
339
}
340
341
use kdl::{KdlDocument, KdlNode, KdlValue};
342
use miette::SourceSpan;
343
344
fn kind_of(v: &KdlValue) -> &'static str {
345
    match v {
346
        KdlValue::String(_) => "a string",
347
        KdlValue::Integer(_) => "an integer",
348
        KdlValue::Float(_) => "a number",
349
        KdlValue::Bool(_) => "a boolean",
350
        KdlValue::Null => "null",
351
    }
352
}
353
354
/// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB.
355
fn parse_memory(s: &str) -> Option<u64> {
356
    let (num, unit) = s.split_at(s.find(|c: char| !c.is_ascii_digit())?);
357
    let n: u64 = num.parse().ok()?;
358
    let mib = match unit {
359
        "KiB" => n.is_multiple_of(1024).then_some(n / 1024)?,
360
        "MiB" => n,
361
        "GiB" => n.checked_mul(1024)?,
362
        _ => return None,
363
    };
364
    Some(mib)
365
}
366
367
/// The line of the first escape KDL 2.0 does not have but shells and C
368
/// do (`\0NN`, `\e`, `\xNN`), as someone colouring output writes them.
369
fn foreign_escape_line(src: &str) -> Option<usize> {
370
    src.lines()
371
        .position(|l| {
372
            let b = l.as_bytes();
373
            b.windows(2).enumerate().any(|(i, w)| {
374
                w[0] == b'\\'
375
                    && (i == 0 || b[i - 1] != b'\\')
376
                    && (w[1] == b'e' || w[1] == b'x' || (w[1] == b'0' && b.get(i + 2).is_some_and(u8::is_ascii_digit)))
377
            })
378
        })
379
        .map(|i| i + 1)
380
}
381
382
impl Pipeline {
383
    /// Parse and validate a pipeline document.
384
    pub fn parse(src: &str) -> Result<Pipeline, Problems> {
385
        let doc = match KdlDocument::parse_v2(src) {
386
            Ok(d) => d,
387
            Err(e) => {
388
                let mut ctx = Ctx { src, problems: Vec::new() };
389
                for d in &e.diagnostics {
390
                    let msg = d.message.clone().or_else(|| d.label.clone()).unwrap_or_else(|| "not valid KDL 2.0".into());
391
                    ctx.err(d.span, msg);
392
                }
393
                if ctx.problems.is_empty() {
394
                    ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() });
395
                }
396
                if let Some(line) = foreign_escape_line(src) {
397
                    ctx.problems.push(Problem {
398
                        line,
399
                        message: "KDL strings have no \\033, \\e or \\x escapes: write ESC (for ANSI colours) as \\u{1b}".into(),
400
                    });
401
                }
402
                return Err(Problems(ctx.problems));
403
            }
404
        };
405
        let mut ctx = Ctx { src, problems: Vec::new() };
406
        let mut schema = None;
407
        let mut sandboxes = BTreeMap::new();
408
        let mut jobs: BTreeMap<String, (Job, SourceSpan)> = BTreeMap::new();
409
        for node in doc.nodes() {
410
            match node.name().value() {
411
                "schema-version" => {
412
                    ctx.no_children(node);
413
                    match ctx.one_integer(node) {
414
                        Some(1) => schema = Some(1),
415
                        Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")),
416
                        None => {}
417
                    }
418
                }
419
                "sandbox" => {
420
                    if let Some(sb) = parse_sandbox(&mut ctx, node) {
421
                        if sandboxes.contains_key(&sb.name) {
422
                            ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name));
423
                        } else {
424
                            sandboxes.insert(sb.name.clone(), sb);
425
                        }
426
                    }
427
                }
428
                "job" => {
429
                    if let Some(job) = parse_job(&mut ctx, node) {
430
                        if jobs.contains_key(&job.name) {
431
                            ctx.err(node.span(), format!("job {:?} is declared twice", job.name));
432
                        } else {
433
                            jobs.insert(job.name.clone(), (job, node.span()));
434
                        }
435
                    }
436
                }
437
                n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")),
438
                n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, sandbox or job")),
439
            }
440
        }
441
        if schema.is_none() && !ctx.problems.iter().any(|p| p.message.contains("schema-version")) {
442
            ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() });
443
        }
444
        if jobs.is_empty() && ctx.problems.is_empty() {
445
            ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() });
446
        }
447
        // References and cycles.
448
        for (job, span) in jobs.values() {
449
            if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) {
450
                ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox));
451
            }
452
            for d in &job.depends_on {
453
                if !jobs.contains_key(d) {
454
                    ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d));
455
                } else if d == &job.name {
456
                    ctx.err(*span, format!("job {:?} depends on itself", job.name));
457
                }
458
            }
459
        }
460
        let graph: BTreeMap<&str, Vec<&str>> =
461
            jobs.values().map(|(j, _)| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect();
462
        if let Some(cycle) = find_cycle(&graph) {
463
            let span = jobs[cycle[0]].1;
464
            ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → ")));
465
        }
466
        if !ctx.problems.is_empty() {
467
            ctx.problems.sort_by_key(|p| p.line);
468
            return Err(Problems(ctx.problems));
469
        }
470
        Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(|(k, (j, _))| (k, j)).collect() })
471
    }
472
473
    /// Jobs in an order where each comes after what it depends on (ties by name).
474
    pub fn order(&self) -> Vec<&Job> {
475
        let mut done: BTreeSet<&str> = BTreeSet::new();
476
        let mut out = Vec::with_capacity(self.jobs.len());
477
        while out.len() < self.jobs.len() {
478
            let before = out.len();
479
            for j in self.jobs.values() {
480
                if !done.contains(j.name.as_str()) && j.depends_on.iter().all(|d| done.contains(d.as_str())) {
481
                    out.push(j);
482
                }
483
            }
484
            for j in &out[before..] {
485
                done.insert(j.name.as_str());
486
            }
487
            if out.len() == before {
488
                break; // validated pipelines have no cycles
489
            }
490
        }
491
        out
492
    }
493
494
    /// Each job's depth: 0 without dependencies, else one more than its deepest dependency.
495
    pub fn depths(&self) -> BTreeMap<&str, usize> {
496
        let mut depth: BTreeMap<&str, usize> = BTreeMap::new();
497
        for j in self.order() {
498
            let d = j.depends_on.iter().map(|x| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0);
499
            depth.insert(j.name.as_str(), d);
500
        }
501
        depth
502
    }
503
}
504
505
fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option<Vec<&'a str>> {
506
    #[derive(Clone, Copy, PartialEq)]
507
    enum Mark {
508
        Visiting,
509
        Done,
510
    }
511
    fn visit<'a>(
512
        n: &'a str,
513
        graph: &BTreeMap<&'a str, Vec<&'a str>>,
514
        marks: &mut BTreeMap<&'a str, Mark>,
515
        stack: &mut Vec<&'a str>,
516
    ) -> Option<Vec<&'a str>> {
517
        match marks.get(n) {
518
            Some(Mark::Done) => return None,
519
            Some(Mark::Visiting) => {
520
                let at = stack.iter().position(|s| *s == n).unwrap_or(0);
521
                let mut c = stack[at..].to_vec();
522
                c.push(n);
523
                return Some(c);
524
            }
525
            None => {}
526
        }
527
        marks.insert(n, Mark::Visiting);
528
        stack.push(n);
529
        for d in graph.get(n).into_iter().flatten() {
530
            if graph.contains_key(d) && *d != n {
531
                if let Some(c) = visit(d, graph, marks, stack) {
532
                    return Some(c);
533
                }
534
            }
535
        }
536
        stack.pop();
537
        marks.insert(n, Mark::Done);
538
        None
539
    }
540
    let mut marks = BTreeMap::new();
541
    for n in graph.keys() {
542
        let mut stack = Vec::new();
543
        if let Some(c) = visit(n, graph, &mut marks, &mut stack) {
544
            return Some(c);
545
        }
546
    }
547
    None
548
}
549
550
fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] {
551
    match node.children() {
552
        Some(d) => d.nodes(),
553
        None => {
554
            ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value()));
555
            &[]
556
        }
557
    }
558
}
559
560
fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option<Sandbox> {
561
    let name = ctx.one_string(node)?;
562
    let named = ctx.name_ok(node.span(), "sandbox", &name);
563
    let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) =
564
        (Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated));
565
    let mut seen = BTreeSet::new();
566
    let mut env = BTreeMap::new();
567
    let mut runs_on = Vec::new();
568
    let mut timeout_s = None;
569
    for c in block(ctx, node) {
570
        let key = c.name().value();
571
        if key != "runs-on" && !seen.insert(key.to_string()) {
572
            ctx.err(c.span(), format!("`{key}` is given twice"));
573
        }
574
        if key == "env" {
575
            parse_env(ctx, c, &mut env);
576
            continue;
577
        }
578
        if key == "runs-on" {
579
            ctx.no_children(c);
580
            parse_labels(ctx, c, &mut runs_on);
581
            continue;
582
        }
583
        ctx.no_children(c);
584
        match key {
585
            "runtime" => {
586
                runtime = match ctx.one_string(c).as_deref() {
587
                    Some("microvm") => Some(Runtime::Microvm),
588
                    Some("syscall") | Some("gvisor") => Some(Runtime::Syscall),
589
                    Some("wasm") => {
590
                        ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\"");
591
                        None
592
                    }
593
                    Some(other) => {
594
                        ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\""));
595
                        None
596
                    }
597
                    None => None,
598
                }
599
            }
600
            "rootfs" => {
601
                rootfs = match ctx.one_string(c) {
602
                    Some(r) => match r.strip_prefix("cas://blob/") {
603
                        Some(h) if h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => {
604
                            Some(h.to_string())
605
                        }
606
                        _ => {
607
                            ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits"));
608
                            None
609
                        }
610
                    },
611
                    None => None,
612
                }
613
            }
614
            "memory-limit" => {
615
                memory = match ctx.one_string(c) {
616
                    Some(m) => match parse_memory(&m) {
617
                        Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32),
618
                        Some(_) => {
619
                            ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB"));
620
                            None
621
                        }
622
                        None => {
623
                            ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\""));
624
                            None
625
                        }
626
                    },
627
                    None => None,
628
                }
629
            }
630
            "cpu-limit" => {
631
                cpus = match ctx.one_integer(c) {
632
                    Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32),
633
                    Some(n) => {
634
                        ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}"));
635
                        None
636
                    }
637
                    None => None,
638
                }
639
            }
640
            "network" => {
641
                network = match ctx.one_string(c).as_deref() {
642
                    Some("isolated") => Some(Network::Isolated),
643
                    // Loopback only: the guest's `lo` comes up, and it still has
644
                    // no device that reaches anything outside it.
645
                    Some("loopback") => Some(Network::Loopback),
646
                    Some("egress") => {
647
                        ctx.err(c.span(), "network \"egress\" is not in M0; M0 has \"isolated\" and \"loopback\"");
648
                        None
649
                    }
650
                    Some(other) => {
651
                        ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
652
                        None
653
                    }
654
                    None => None,
655
                }
656
            }
657
            "timeout" => timeout_s = parse_timeout(ctx, c),
658
            other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")),
659
        }
660
    }
661
    if rootfs.is_none() && !seen.contains("rootfs") {
662
        ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
663
    }
664
    if !named {
665
        return None;
666
    }
667
    Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network?, env, runs_on, timeout_s })
668
}
669
670
/// The longest a pipeline may ask a job to run.
671
pub const MAX_TIMEOUT_S: u64 = 24 * 3600;
672
673
/// A duration: whole hours, minutes and seconds in that order, each
674
/// optional, as "90m", "2h", "1h30m" or "45s".
675
pub fn parse_duration(s: &str) -> Option<u64> {
676
    let (mut total, mut num, mut last) = (0u64, String::new(), 0u8);
677
    for ch in s.chars() {
678
        if ch.is_ascii_digit() {
679
            num.push(ch);
680
            continue;
681
        }
682
        let (unit, mult) = match ch {
683
            'h' => (1, 3600),
684
            'm' => (2, 60),
685
            's' => (3, 1),
686
            _ => return None,
687
        };
688
        // Each unit once, largest first, each with its number.
689
        if num.is_empty() || unit <= last {
690
            return None;
691
        }
692
        last = unit;
693
        total = total.checked_add(num.parse::<u64>().ok()?.checked_mul(mult)?)?;
694
        num.clear();
695
    }
696
    (num.is_empty() && total > 0).then_some(total)
697
}
698
699
/// A `timeout` line: one duration, from a minute to a day.
700
fn parse_timeout(ctx: &mut Ctx, c: &KdlNode) -> Option<u64> {
701
    let t = ctx.one_string(c)?;
702
    match parse_duration(&t) {
703
        Some(secs) if (60..=MAX_TIMEOUT_S).contains(&secs) => Some(secs),
704
        Some(_) => {
705
            ctx.err(c.span(), format!("timeout {t:?} must be from 1m to 24h"));
706
            None
707
        }
708
        None => {
709
            ctx.err(c.span(), format!("timeout {t:?}: expected hours, minutes and seconds, as \"90m\" or \"1h30m\""));
710
            None
711
        }
712
    }
713
}
714
715
/// The labels of a `runs-on` line: lower-case letters, digits, `-`, `_`
716
/// and `.`, at most 64 bytes each; into `into`, sorted, once each.
717
fn parse_labels(ctx: &mut Ctx, c: &KdlNode, into: &mut Vec<String>) {
718
    let labels = ctx.strings(c);
719
    if labels.is_empty() {
720
        ctx.err(c.span(), "`runs-on` needs one or more labels, as runs-on \"riscv64\"");
721
    }
722
    for l in labels {
723
        let ok = !l.is_empty()
724
            && l.len() <= 64
725
            && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'-' | b'_' | b'.'));
726
        if ok {
727
            into.push(l);
728
        } else {
729
            ctx.err(c.span(), format!("runs-on {l:?}: a label is lower-case letters, digits, -, _ and ., at most 64"));
730
        }
731
    }
732
    into.sort();
733
    into.dedup();
734
}
735
736
/// An `env` block of `NAME "value"` lines, into `env`.
737
fn parse_env(ctx: &mut Ctx, c: &KdlNode, env: &mut BTreeMap<String, String>) {
738
    if !c.entries().is_empty() {
739
        ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines");
740
    }
741
    for e in block(ctx, c) {
742
        let k = e.name().value().to_string();
743
        let valid =
744
            !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
745
        if !valid {
746
            ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _"));
747
            continue;
748
        }
749
        ctx.no_children(e);
750
        if let Some(v) = ctx.one_string(e) {
751
            if env.insert(k.clone(), v).is_some() {
752
                ctx.err(e.span(), format!("environment {k} is set twice"));
753
            }
754
        }
755
    }
756
}
757
758
/// `mount { source "ephemeral://disk"; target "/src/target"; size "8GiB" }`.
759
fn parse_mount(ctx: &mut Ctx, node: &KdlNode) -> Option<Mount> {
760
    let (mut source, mut target, mut size) = (None, None, None);
761
    let mut seen = BTreeSet::new();
762
    for c in block(ctx, node) {
763
        let key = c.name().value();
764
        if !seen.insert(key.to_string()) {
765
            ctx.err(c.span(), format!("`{key}` is given twice"));
766
            continue;
767
        }
768
        ctx.no_children(c);
769
        match key {
770
            "source" => {
771
                source = match ctx.one_string(c).as_deref() {
772
                    Some("ephemeral://tmpfs") => Some(MountSource::Tmpfs),
773
                    Some("ephemeral://disk") => Some(MountSource::Disk),
774
                    Some(other) => {
775
                        ctx.err(
776
                            c.span(),
777
                            format!("mount source {other:?} is not in M0 yet; M0 mounts \"ephemeral://tmpfs\" or \"ephemeral://disk\""),
778
                        );
779
                        None
780
                    }
781
                    None => None,
782
                }
783
            }
784
            "target" => {
785
                target = match ctx.one_string(c) {
786
                    Some(t) if mount_target_ok(&t) => Some(t),
787
                    Some(t) => {
788
                        ctx.err(
789
                            c.span(),
790
                            format!("mount target {t:?} must be an absolute path under /src or /tmp, without . or .. or a trailing /"),
791
                        );
792
                        None
793
                    }
794
                    None => None,
795
                }
796
            }
797
            "size" => {
798
                size = match ctx.one_string(c) {
799
                    Some(m) => match parse_memory(&m) {
800
                        Some(mib) if (MIN_MOUNT_MIB..=MAX_MOUNT_MIB).contains(&mib) => Some(mib as u32),
801
                        Some(_) => {
802
                            ctx.err(c.span(), format!("mount size {m:?} must be from 16MiB to 64GiB"));
803
                            None
804
                        }
805
                        None => {
806
                            ctx.err(c.span(), format!("mount size {m:?}: expected a whole number of KiB, MiB or GiB, as \"8GiB\""));
807
                            None
808
                        }
809
                    },
810
                    None => None,
811
                }
812
            }
813
            "readonly" | "include" => ctx.err(c.span(), format!("mount `{key}` is not in M0 yet")),
814
            other => ctx.err(c.span(), format!("unknown mount setting `{other}`")),
815
        }
816
    }
817
    for (name, missing) in [("source", source.is_none()), ("target", target.is_none()), ("size", size.is_none())] {
818
        if missing && !seen.contains(name) {
819
            ctx.err(node.span(), format!("a mount needs a {name}"));
820
        }
821
    }
822
    Some(Mount { source: source?, target: target?, size_mib: size? })
823
}
824
825
/// `output { artifact "name" { path "/src/target/release/tool" } }`.
826
/// `secret "deploy-token" { mount-path "/run/secrets/token" }` (spec §9.1).
827
/// The hub is the vault: the secret of that name set for the repository,
828
/// its organisation or the site. `vault-ref` is not in this slice.
829
fn parse_secret(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
830
    let Some(name) = ctx.one_string(node) else {
831
        ctx.err(node.span(), "`secret` takes a name and a block with mount-path");
832
        return;
833
    };
834
    if !ctx.name_ok(node.span(), "secret", &name) {
835
        return;
836
    }
837
    let mut path = None;
838
    for c in block(ctx, node) {
839
        ctx.no_children(c);
840
        match c.name().value() {
841
            "mount-path" => match ctx.one_string(c) {
842
                Some(p) if secret_path_ok(&p) => path = Some(p),
843
                Some(p) => {
844
                    ctx.err(c.span(), format!("secret mount-path {p:?} must be an absolute path under /run/secrets/, without . or .."))
845
                }
846
                None => {}
847
            },
848
            "vault-ref" => ctx.err(c.span(), "`vault-ref` is not in M0; the hub supplies the secret of that name"),
849
            other => ctx.err(c.span(), format!("unknown secret setting `{other}`")),
850
        }
851
    }
852
    let Some(mount_path) = path else {
853
        ctx.err(node.span(), format!("secret {name:?} needs a mount-path"));
854
        return;
855
    };
856
    if job.secrets.iter().any(|s| s.name == name) {
857
        ctx.err(node.span(), format!("secret {name:?} is given twice"));
858
    } else if job.secrets.iter().any(|s| s.mount_path == mount_path) || job.identities.iter().any(|i| i.mount_path == mount_path) {
859
        ctx.err(node.span(), format!("two files mount at {mount_path:?}"));
860
    } else if job.secrets.len() == MAX_SECRETS {
861
        ctx.err(node.span(), format!("a job has at most {MAX_SECRETS} secrets"));
862
    } else {
863
        job.secrets.push(Secret { name, mount_path });
864
    }
865
}
866
867
/// `identity "sts.amazonaws.com" { mount-path "/run/secrets/aws-token" }`:
868
/// the hub mints an OIDC token for that audience and the runner mounts it
869
/// as a file (oxygenhub#103).
870
fn parse_identity(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
871
    let Some(audience) = ctx.one_string(node) else {
872
        ctx.err(node.span(), "`identity` takes an audience and a block with mount-path");
873
        return;
874
    };
875
    let aud_ok = !audience.is_empty() && audience.len() <= 255 && !audience.chars().any(|c| c.is_whitespace() || c.is_control());
876
    if !aud_ok {
877
        ctx.err(node.span(), format!("identity audience {audience:?} must be 1 to 255 characters without spaces"));
878
        return;
879
    }
880
    let mut path = None;
881
    for c in block(ctx, node) {
882
        ctx.no_children(c);
883
        match c.name().value() {
884
            "mount-path" => match ctx.one_string(c) {
885
                Some(p) if secret_path_ok(&p) => path = Some(p),
886
                Some(p) => {
887
                    ctx.err(c.span(), format!("identity mount-path {p:?} must be an absolute path under /run/secrets/, without . or .."))
888
                }
889
                None => {}
890
            },
891
            other => ctx.err(c.span(), format!("unknown identity setting `{other}`")),
892
        }
893
    }
894
    let Some(mount_path) = path else {
895
        ctx.err(node.span(), format!("identity {audience:?} needs a mount-path"));
896
        return;
897
    };
898
    if job.identities.iter().any(|i| i.audience == audience) {
899
        ctx.err(node.span(), format!("identity {audience:?} is given twice"));
900
    } else if job.identities.iter().any(|i| i.mount_path == mount_path) || job.secrets.iter().any(|s| s.mount_path == mount_path) {
901
        ctx.err(node.span(), format!("two files mount at {mount_path:?}"));
902
    } else if job.identities.len() == MAX_IDENTITIES {
903
        ctx.err(node.span(), format!("a job has at most {MAX_IDENTITIES} identities"));
904
    } else {
905
        job.identities.push(Identity { audience, mount_path });
906
    }
907
}
908
909
fn parse_output(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
910
    if !node.entries().is_empty() {
911
        ctx.err(node.span(), "`output` takes a block of artifacts");
912
    }
913
    for a in block(ctx, node) {
914
        if a.name().value() != "artifact" {
915
            ctx.err(a.span(), format!("unknown output `{}`; outputs are `artifact \"name\" {{ path \"/src/…\" }}`", a.name().value()));
916
            continue;
917
        }
918
        let Some(name) = ctx.one_string(a) else { continue };
919
        if !ctx.name_ok(a.span(), "artifact", &name) {
920
            continue;
921
        }
922
        let mut path = None;
923
        for c in block(ctx, a) {
924
            ctx.no_children(c);
925
            match c.name().value() {
926
                "path" => match ctx.one_string(c) {
927
                    Some(p) if output_path_ok(&p) => path = Some(p),
928
                    Some(p) => ctx.err(
929
                        c.span(),
930
                        format!("artifact path {p:?} must be an absolute path under /src, without . or .., quotes or spaces"),
931
                    ),
932
                    None => {}
933
                },
934
                "format" => {
935
                    if ctx.one_string(c).as_deref() != Some("blob") {
936
                        ctx.err(c.span(), "artifact format is \"blob\"");
937
                    }
938
                }
939
                other => ctx.err(c.span(), format!("unknown artifact setting `{other}`")),
940
            }
941
        }
942
        let Some(path) = path else {
943
            ctx.err(a.span(), format!("artifact {name:?} needs a path"));
944
            continue;
945
        };
946
        if job.outputs.iter().any(|o| o.name == name) {
947
            ctx.err(a.span(), format!("artifact {name:?} is given twice"));
948
        } else if job.outputs.len() == MAX_OUTPUTS {
949
            ctx.err(a.span(), format!("a job has at most {MAX_OUTPUTS} artifacts"));
950
        } else {
951
            job.outputs.push(Output { name, path });
952
        }
953
    }
954
}
955
956
fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option<Job> {
957
    let name = ctx.one_string(node)?;
958
    let named = ctx.name_ok(node.span(), "job", &name);
959
    let mut job = Job {
960
        name: name.clone(),
961
        sandbox: String::new(),
962
        scope: vec![],
963
        depends_on: vec![],
964
        env: BTreeMap::new(),
965
        exec: vec![],
966
        mounts: vec![],
967
        secrets: vec![],
968
        identities: vec![],
969
        outputs: vec![],
970
        runs_on: vec![],
971
        timeout_s: None,
972
    };
973
    let mut seen = BTreeSet::new();
974
    let mut outputs_at = None;
975
    for c in block(ctx, node) {
976
        let key = c.name().value();
977
        if key != "scope"
978
            && key != "depends-on"
979
            && key != "mount"
980
            && key != "secret"
981
            && key != "identity"
982
            && key != "runs-on"
983
            && !seen.insert(key.to_string())
984
        {
985
            ctx.err(c.span(), format!("`{key}` is given twice"));
986
        }
987
        match key {
988
            "sandbox" => {
989
                ctx.no_children(c);
990
                job.sandbox = ctx.one_string(c).unwrap_or_default();
991
            }
992
            "scope" => {
993
                ctx.no_children(c);
994
                for g in ctx.strings(c) {
995
                    match globset::Glob::new(&g) {
996
                        Ok(_) => job.scope.push(g),
997
                        Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")),
998
                    }
999
                }
1000
            }
1001
            "depends-on" => {
1002
                ctx.no_children(c);
1003
                job.depends_on.extend(ctx.strings(c));
1004
            }
1005
            "runs-on" => {
1006
                ctx.no_children(c);
1007
                parse_labels(ctx, c, &mut job.runs_on);
1008
            }
1009
            "timeout" => {
1010
                ctx.no_children(c);
1011
                job.timeout_s = parse_timeout(ctx, c);
1012
            }
1013
            "env" => parse_env(ctx, c, &mut job.env),
1014
            "exec" => {
1015
                ctx.no_children(c);
1016
                job.exec = ctx.strings(c);
1017
                if job.exec.is_empty() {
1018
                    ctx.err(c.span(), "`exec` needs a program and its arguments");
1019
                } else if job.exec.iter().any(|a| a.contains('\0')) {
1020
                    ctx.err(c.span(), "`exec` arguments cannot contain NUL");
1021
                }
1022
            }
1023
            "mount" => {
1024
                if !c.entries().is_empty() {
1025
                    ctx.err(c.span(), "`mount` takes a block of source, target and size");
1026
                }
1027
                if let Some(m) = parse_mount(ctx, c) {
1028
                    if job.mounts.len() == MAX_MOUNTS {
1029
                        ctx.err(c.span(), format!("a job has at most {MAX_MOUNTS} mounts"));
1030
                    } else if let Some(other) = job.mounts.iter().find(|o| {
1031
                        o.target == m.target
1032
                            || m.target.starts_with(&format!("{}/", o.target))
1033
                            || o.target.starts_with(&format!("{}/", m.target))
1034
                    }) {
1035
                        ctx.err(c.span(), format!("mount {:?} overlaps mount {:?}", m.target, other.target));
1036
                    } else {
1037
                        job.mounts.push(m);
1038
                    }
1039
                }
1040
            }
1041
            "output" => {
1042
                outputs_at = Some(c.span());
1043
                parse_output(ctx, c, &mut job);
1044
            }
1045
            "secret" => parse_secret(ctx, c, &mut job),
1046
            "identity" => parse_identity(ctx, c, &mut job),
1047
            n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")),
1048
            other => ctx.err(c.span(), format!("unknown job setting `{other}`")),
1049
        }
1050
    }
1051
    job.mounts.sort_by(|a, b| a.target.cmp(&b.target));
1052
    job.outputs.sort_by(|a, b| a.name.cmp(&b.name));
1053
    job.secrets.sort_by(|a, b| a.name.cmp(&b.name));
1054
    job.identities.sort_by(|a, b| a.audience.cmp(&b.audience));
1055
    // A tmpfs is gone when the job ends; nothing on it can be kept.
1056
    for o in &job.outputs {
1057
        if let Some(m) = job.mounts.iter().find(|m| m.source == MountSource::Tmpfs && o.path.starts_with(&format!("{}/", m.target))) {
1058
            ctx.err(
1059
                outputs_at.unwrap_or(node.span()),
1060
                format!(
1061
                    "artifact {:?} is on the tmpfs mount {:?}, which is gone when the job ends; use ephemeral://disk",
1062
                    o.name, m.target
1063
                ),
1064
            );
1065
        }
1066
    }
1067
    if job.sandbox.is_empty() && !seen.contains("sandbox") {
1068
        ctx.err(node.span(), format!("job {name:?} needs a sandbox"));
1069
    }
1070
    if job.exec.is_empty() && !seen.contains("exec") {
1071
        ctx.err(node.span(), format!("job {name:?} needs an exec"));
1072
    }
1073
    job.depends_on.sort();
1074
    job.depends_on.dedup();
1075
    // Kept even when incomplete, so its references are checked too; the
1076
    // problems above already make the pipeline invalid.
1077
    named.then_some(job)
1078
}