workflow_runners

  1#!/usr/bin/env python3
  2"""Only the GPU publish workflow may run on the self-hosted runner. MIP-0065 §5.2.
  3
  4A public repo's self-hosted runner executes whatever a pull request asks it to, so this fails when
  5any `runs-on:` or reusable-workflow `runner:` outside marola-sea-publish.yml names `self-hosted`
  6or is an expression (a variable or matrix entry can be pointed at the desktop), or when that
  7workflow's `on:` gains a trigger a pull request can reach: pull_request(_target) or workflow_call.
  8
  9    scripts/workflow_runners.py [.github/workflows]
 10    scripts/workflow_runners.py --self-test
 11"""
 12
 13from __future__ import annotations
 14
 15import argparse
 16import re
 17import sys
 18from pathlib import Path
 19
 20GPU_WORKFLOW = "marola-sea-publish.yml"
 21RUNNER_KEY = re.compile(r"^(\s*)(?:-\s+)?(runs-on|runner):\s*(.*)$")
 22FORBIDDEN = re.compile(r"self-hosted|\$\{\{")
 23PR_TRIGGER = re.compile(r"\b(pull_request(?:_target)?|workflow_call)\b")
 24# A key or list item, so an input's `description:` that mentions pull_request is not a trigger.
 25PR_TRIGGER_KEY = re.compile(
 26    r"^\s+(?:-\s+)?['\"]?(pull_request(?:_target)?|workflow_call)['\"]?\s*(:|$)"
 27)
 28
 29
 30def _strip_comment(value: str) -> str:
 31    return value.split(" #", 1)[0].strip()
 32
 33
 34def runner_findings(name: str, text: str) -> list[str]:
 35    lines = text.splitlines()
 36    out = []
 37    for i, line in enumerate(lines):
 38        m = RUNNER_KEY.match(line)
 39        if not m:
 40            continue
 41        indent, value = len(m.group(1)), _strip_comment(m.group(3))
 42        # Block form (a list, or `group:`/`labels:`): every line indented under the key counts.
 43        j = i + 1
 44        while not m.group(3).strip() and j < len(lines):
 45            nxt = lines[j]
 46            if nxt.strip() and (len(nxt) - len(nxt.lstrip())) <= indent:
 47                break
 48            value += " " + _strip_comment(nxt)
 49            j += 1
 50        hit = FORBIDDEN.search(value)
 51        if hit:
 52            what = "an expression" if hit.group(0) == "${{" else hit.group(0)
 53            out.append(f"{name}:{i + 1} targets {what} — only {GPU_WORKFLOW} may be self-hosted")
 54    return out
 55
 56
 57def trigger_findings(name: str, text: str) -> list[str]:
 58    out = []
 59    in_on = False
 60    for i, line in enumerate(text.splitlines()):
 61        top = re.match(r"^['\"]?(\w+)['\"]?:\s*(.*)$", line)
 62        if top:
 63            in_on = top.group(1) in ("on", "true")  # YAML 1.1 reads a bare `on` as true
 64            hit = in_on and PR_TRIGGER.search(_strip_comment(top.group(2)))
 65        else:
 66            hit = in_on and PR_TRIGGER_KEY.match(line)
 67        if hit:
 68            out.append(f"{name}:{i + 1} adds {hit.group(1)} to the self-hosted workflow")
 69    return out
 70
 71
 72def scan(root: Path) -> list[str]:
 73    findings = []
 74    for f in sorted([*root.glob("*.yml"), *root.glob("*.yaml")]):
 75        text = f.read_text(encoding="utf-8")
 76        if f.name == GPU_WORKFLOW:
 77            findings += trigger_findings(f.name, text)
 78        else:
 79            findings += runner_findings(f.name, text)
 80    return findings
 81
 82
 83def check(root: Path) -> int:
 84    if not root.is_dir():
 85        print(f"workflow_runners: no such directory: {root}", file=sys.stderr)
 86        return 2
 87    findings = scan(root)
 88    for line in findings:
 89        print(f"workflow_runners: {line}", file=sys.stderr)
 90    if findings:
 91        return 1
 92    print(f"workflow_runners: only {GPU_WORKFLOW} is self-hosted")
 93    return 0
 94
 95
 96CLEAN_CI = """name: CI
 97on:
 98  push:
 99  pull_request:
100jobs:
101  changes:
102    # the desktop runner is self-hosted; this job is not
103    runs-on: ubuntu-latest
104    steps: []
105"""
106GPU = """name: marola-sea publish
107on:
108  workflow_dispatch:
109    inputs:
110      preset:
111        description: "not a pull_request"
112jobs:
113  publish:
114    runs-on: [self-hosted, marola-sea]
115"""
116
117
118def self_test() -> int:
119    import tempfile
120
121    fails = 0
122
123    def ok(got, want, label):
124        nonlocal fails
125        if got == want:
126            print(f"  ok   {label}")
127        else:
128            fails += 1
129            print(f"  FAIL {label} — got {got!r}, want {want!r}")
130
131    def tree(ci: str, gpu: str = GPU) -> list[str]:
132        with tempfile.TemporaryDirectory() as tmp:
133            (Path(tmp) / "ci.yml").write_text(ci, encoding="utf-8")
134            (Path(tmp) / GPU_WORKFLOW).write_text(gpu, encoding="utf-8")
135            return scan(Path(tmp))
136
137    ok(tree(CLEAN_CI), [], "a clean tree has no findings, comments and the GPU job included")
138    stray = tree(CLEAN_CI.replace("ubuntu-latest", "self-hosted"))
139    ok(len(stray), 1, "a stray self-hosted in ci.yml is caught")
140    ok(stray[0].startswith("ci.yml:8 targets self-hosted"), True, "and named by file and line")
141    ok(
142        len(tree(CLEAN_CI.replace("ubuntu-latest", "${{ vars.CI_RUNNER || 'ubuntu-latest' }}"))),
143        1,
144        "a CI_RUNNER fallback is caught even when it falls back to a hosted label",
145    )
146    ok(
147        len(tree(CLEAN_CI.replace("ubuntu-latest", "${{ matrix.os }}"))),
148        1,
149        "so is any other expression, a matrix entry included",
150    )
151    ok(
152        len(tree(CLEAN_CI.replace("ubuntu-latest", "\n      - linux\n\n      - self-hosted"))),
153        1,
154        "the block-list form is read to its end, past a blank line",
155    )
156    ok(
157        len(
158            tree(
159                CLEAN_CI.replace(
160                    "ubuntu-latest", "\n      group: desktop\n      labels: self-hosted"
161                )
162            )
163        ),
164        1,
165        "and so is the group/labels form",
166    )
167    reusable = "jobs:\n  ping:\n    uses: o/r/.github/workflows/p.yml@main\n    with:\n      runner: self-hosted\n"
168    ok(len(tree(reusable)), 1, "a reusable workflow's runner: input counts as runs-on")
169    ok(
170        len(
171            tree(
172                CLEAN_CI,
173                GPU.replace("  workflow_dispatch:", "  pull_request:\n  workflow_dispatch:"),
174            )
175        ),
176        1,
177        "pull_request added to the GPU workflow is caught",
178    )
179    ok(
180        len(
181            tree(
182                CLEAN_CI,
183                GPU.replace(
184                    "on:\n  workflow_dispatch:", "on: [workflow_dispatch, pull_request_target]\nx:"
185                ),
186            )
187        ),
188        1,
189        "and so is the inline form, pull_request_target included",
190    )
191    ok(
192        len(
193            tree(
194                CLEAN_CI,
195                GPU.replace("  workflow_dispatch:", "  workflow_call:\n  workflow_dispatch:"),
196            )
197        ),
198        1,
199        "workflow_call on the GPU workflow is caught: a PR workflow could `uses:` it",
200    )
201
202    if fails:
203        print(f"workflow_runners self-test: {fails} failure(s)", file=sys.stderr)
204        return 1
205    print("workflow_runners self-test: ok")
206    return 0
207
208
209def main(argv: list[str] | None = None) -> int:
210    ap = argparse.ArgumentParser(
211        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
212    )
213    ap.add_argument("root", nargs="?", type=Path, default=Path(".github/workflows"))
214    ap.add_argument("--self-test", action="store_true")
215    args = ap.parse_args(argv)
216    return self_test() if args.self_test else check(args.root)
217
218
219if __name__ == "__main__":
220    sys.exit(main())