open-standards/scripts/check-wai-release-quickstart.py

#!/usr/bin/env python3
"""check-wai-release-quickstart.py: the release runs the quickstart from each archive before it
attaches it, and the Windows twin keeps the script's steps.

wai-release.yml builds the quickstart archive on each platform and attaches it to the release.
Before it is attached, each archive runs the quickstart as a new user runs it, from the archive
alone: wai/quickstart/test/clean-machine.sh on macOS and Linux, its PowerShell twin
clean-machine.ps1 on Windows. Each step of the twin is a step of the script, with the same
`wai` command. This holds the three files to that:

- each of the jobs macos, linux and windows runs its script in a step's command: the script is
  a line's command (a comment, a step's name, or a line that only echoes its path is not a run
  of it). That step has no `if:` and no `continue-on-error:`, and neither has its job. Nothing
  follows the command on its line (no `|| true`, `|| :` or `; exit 0`), the line is in no block,
  and no line before it ends the step, but in the macos job, inside the block of
  `if [ "${{ matrix.triple }}" != <the host's triple> ]; then` alone: an archive cross-built for
  another triple, which that host cannot run (its `else` is not that block). After
  clean-machine.sh's line the step has no other, so its exit status is the step's; after
  clean-machine.ps1's comes `if ($LASTEXITCODE -ne 0) { throw ... }`, as PowerShell does not
  stop on a program's exit status. And the step comes before the job's first step that attaches
  anything to the release (one that runs wai/quickstart/publish.sh or uploads to a release's
  assets);
- a job on the macOS runner whose step runs node, npm or npx has an earlier step that puts
  /opt/homebrew/bin on its PATH, which launchd's PATH on that runner omits;
- clean-machine.sh's steps are unpack, contents, hello, verify, create, dev-check and two-tabs
  (the browser's), and clean-machine.ps1's are the same steps in the same order, the browser's
  aside, each running the same `wai` command (the command alone, up to the first `;`, `}`,
  `'`, `|`, `>` or `&&` after it);
- the unpack step of each checks the archive against the SHA-256 beside it ("$ARCHIVE.sha256",
  "$Archive.sha256") and fails when they differ, before it unpacks anything;
- the contents step of each runs wai-quantum, wai-ingest, wai-pack and wai-remote-host, and does
  not only look for them; the steps after it run wai, wai-relay and wai-room. In clean-machine.sh
  each line that runs one ends in `|| return 1` but the step's last, which has nothing after
  its command, so its status is the step's; in clean-machine.ps1, `throw` follows a run whose
  exit status is not the one wanted;
- each script's exit status is whether a step failed: clean-machine.sh's step() sets `failed`
  in the branch where its command failed, nothing else sets it but its declaration, no `trap`
  can change the exit status, and its last line is `[ -z "$failed" ]`; clean-machine.ps1's
  Step sets `$script:Failed` when its block throws or its program exits non-zero, nothing else
  sets it but its declaration, and its last two lines are `if ($Failed) { exit 1 }` and
  `exit 0`. Every other `exit` of either script, in a step's function or the script's body,
  gives a non-zero number: one with 0 or with none could end it with status 0 before a failed
  step decides it.

What it reads is the text of these lines, not what they do when run: a rule above is a way the
quickstart has been seen, or planted, to pass a failed run, and the selftest plants each.

    check-wai-release-quickstart.py [--root DIR]

Exit 0 when all of it holds, 1 with one line a finding otherwise. Standard library only. The
selftest (check-wai-release-quickstart-selftest.sh) breaks each rule in a copy and requires a
finding.
"""
import os
import re
import sys

WORKFLOW = ".github/workflows/wai-release.yml"
SH = "wai/quickstart/test/clean-machine.sh"
PS = "wai/quickstart/test/clean-machine.ps1"
# The job, and the script it runs on its archive.
JOBS = [("macos", "clean-machine.sh"), ("linux", "clean-machine.sh"), ("windows", "clean-machine.ps1")]
# The command that runs each script: what a line of the step's command starts with.
COMMANDS = {
    "clean-machine.sh": "bash wai/quickstart/test/clean-machine.sh",
    "clean-machine.ps1": "pwsh -NoProfile -File wai\\quickstart\\test\\clean-machine.ps1",
}
# The one block in which the macos job's step may end before it runs the script: an archive
# cross-built for another triple, whose binaries the host cannot run.
CROSS_BUILT = 'if [ "${{ matrix.triple }}" != "$(rustc -vV | sed -n \'s/^host: //p\')" ]; then'
# Where node is on the macOS runner, which launchd's PATH omits.
HOMEBREW = "/opt/homebrew/bin"
# The binaries the contents step runs; the steps after it run wai, wai-relay and wai-room.
CONTENTS_RUNS = ["wai-quantum", "wai-ingest", "wai-pack", "wai-remote-host"]
SH_STEPS = ["unpack", "contents", "hello", "verify", "create", "dev-check", "two-tabs"]
BROWSER_STEP = "two-tabs"
# An `exit` a script's line runs, and its status (group 1, None when it gives none).
EXIT = r"(?:^|[;&|{(]\s*|\b(?:then|else|do)\s+)exit\b(?:\s+([^\s;}|&)]+))?"


def indent(line):
    return len(line) - len(line.lstrip(" "))


def is_comment(line):
    return line.strip().startswith("#")


def jobs(text):
    """{job name: its lines}, the jobs under the top-level `jobs:` key."""
    out = {}
    lines = text.splitlines()
    try:
        start = next(i for i, l in enumerate(lines) if l.rstrip() == "jobs:")
    except StopIteration:
        return out
    name = None
    for l in lines[start + 1:]:
        if l.strip() and not is_comment(l) and indent(l) == 0:
            break
        m = re.match(r"^  ([A-Za-z0-9_-]+):\s*$", l)
        if m:
            name = m.group(1)
            out[name] = []
        elif name is not None:
            out[name].append(l)
    return out


def job_keys(lines):
    """The job's own keys (indented four spaces), comments aside."""
    return [re.match(r"^\s*([A-Za-z0-9_-]+):", l).group(1) for l in lines if indent(l) == 4 and not is_comment(l) and re.match(r"^\s*([A-Za-z0-9_-]+):", l)]


def steps(lines):
    """The job's steps, in order: each {"keys": [its keys], "name": …, "run": [its command's
    lines, comments aside]}. A step is a list item of `steps:`; its keys are the item's first
    key and those at the indent of the item's content."""
    out = []
    in_steps = False
    item_indent = None
    cur = None
    run_indent = None
    for l in lines:
        if not l.strip():
            continue
        if indent(l) == 4 and not is_comment(l):
            in_steps = l.strip() == "steps:"
            cur = None
            continue
        if not in_steps:
            continue
        s = l.lstrip(" ")
        if s.startswith("- ") and not is_comment(l) and (item_indent is None or indent(l) == item_indent):
            item_indent = indent(l)
            cur = {"keys": [], "name": "", "run": []}
            out.append(cur)
            run_indent = None
            l = " " * (item_indent + 2) + s[2:]
        if cur is None:
            continue
        key_indent = item_indent + 2
        if run_indent is not None:
            if indent(l) > key_indent:
                if not is_comment(l):
                    cur["run"].append(l.strip())
                continue
            run_indent = None
        if is_comment(l) or indent(l) != key_indent:
            continue
        m = re.match(r"^\s*([A-Za-z0-9_-]+):\s*(.*)$", l)
        if not m:
            continue
        key, value = m.group(1), m.group(2).strip()
        cur["keys"].append(key)
        if key == "name":
            cur["name"] = value.strip("'\"")
        elif key == "run":
            if value in ("|", ">", "|-", ">-", "|+", ">+"):
                run_indent = key_indent
            elif value and not value.startswith("#"):
                cur["run"].append(value)
    return out


def command(line):
    """A line of a step's command, without a comment after it."""
    return line.split(" #")[0].strip()


def script_line(step, script):
    """The place in the step's command of the line whose command is `script`, or None: a line
    that echoes its path, or that runs something else first, does not run it."""
    want = COMMANDS[script]
    return next((i for i, l in enumerate(step["run"]) if command(l) == want or command(l).startswith(want + " ")), None)


def block_change(line, powershell):
    """How many blocks the line opens, less those it closes."""
    n = line.count("{") - line.count("}") + line.count("(") - line.count(")")
    if not powershell:
        n += len(re.findall(r"(?:^|[;&(]\s*)(?:if|for|while|until|case)\s", line))
        n -= len(re.findall(r"(?:^|;\s*)(?:fi|done|esac)\b", line))
    return n


def ends_the_step(line):
    """Whether the line, reached, ends the step as a success."""
    return re.search(r"(?:^|[;&|{]\s*|\b(?:then|else|do)\s+)(?:exit|return)(?:\s+0)?\s*(?:;|$)", line) is not None


def failure_ignored(step, script, at, may_skip_cross_built):
    """What would let the step pass though the line `at`, which runs `script`, failed or never
    ran: a phrase for each. With `may_skip_cross_built`, a line inside the block of CROSS_BUILT
    (its `then` branch) may end the step."""
    run = [command(l) for l in step["run"]]
    powershell = script.endswith(".ps1")
    out = []
    rest = run[at][len(COMMANDS[script]):].strip()
    if re.search(r"[|;&]", rest):
        out.append(f"its line goes on after it ({rest})")
    depth = 0
    opener = None  # the line that opened the outermost block the line is in, or its else
    for l in run[:at]:
        if depth == 1 and re.match(r"(?:else|elif)\b", l):
            opener = l
        if ends_the_step(l):
            if depth == 0:
                out.append(f"`{l}` ends the step before it")
            elif not (may_skip_cross_built and opener == CROSS_BUILT):
                out.append(f"`{l}` ends the step before it, under `{opener}`, which is not the test for an archive cross-built for another triple")
        before = depth
        depth += block_change(l, powershell)
        if before == 0 and depth > 0:
            opener = l
    if depth:
        out.append("it runs inside a block")
    after = run[at + 1:]
    if powershell:
        if not after or not re.match(r"if \(\$LASTEXITCODE -ne 0\) \{\s*throw\b", after[0]):
            out.append("the line after it is not `if ($LASTEXITCODE -ne 0) { throw ... }`, and PowerShell does not stop on a program's exit status")
    elif after:
        out.append(f"`{after[0]}` follows it, so the step's exit status is not its own")
    return out


def attaches(step):
    """Whether the step attaches something to the release."""
    return any("wai/quickstart/publish.sh" in l or re.search(r"/assets\b", l) for l in step["run"])


def on_macos(lines):
    """Whether the job runs on the macOS runner, itself or as a row of its matrix."""
    return any(re.search(r"\b(?:runs-on|runner):\s*macos\b", l) for l in lines if not is_comment(l))


def runs_node(step):
    """Whether the step's command runs node, npm or npx."""
    return any(re.search(r"(?:^|[\s(;&|])(?:node|npm|npx)\s", command(l)) for l in step["run"])


def puts_node_on_path(step):
    """Whether the step adds Homebrew's directory to the job's PATH."""
    return any(HOMEBREW in l and "GITHUB_PATH" in l for l in step["run"])


def check_workflow(text, findings):
    js = jobs(text)
    for job, script in JOBS:
        if job not in js:
            findings.append(f"{WORKFLOW}: no job {job}, which runs {script} on the archive it attaches")
            continue
        where = f"{WORKFLOW}: the {job} job"
        for k in ("if", "continue-on-error"):
            if k in job_keys(js[job]):
                findings.append(f"{where} has `{k}:`, so the quickstart from its archive can be skipped or fail without failing it")
        st = steps(js[job])
        running = [i for i, s in enumerate(st) if script_line(s, script) is not None]
        if not running:
            findings.append(f"{where} runs {script} in no step's command: the archive it attaches is not run as a new user runs it")
            continue
        attach = next((i for i, s in enumerate(st) if attaches(s)), None)
        if attach is None:
            findings.append(f"{where} has no step that attaches its archive to the release (wai/quickstart/publish.sh, or an upload to the release's assets)")
        for i in running:
            s = st[i]
            what = f"{where}'s step {s['name']!r}, which runs {script},"
            for k in ("if", "continue-on-error"):
                if k in s["keys"]:
                    findings.append(f"{what} has `{k}:`: it can be skipped or fail without failing the job")
            for why in failure_ignored(s, script, script_line(s, script), job == "macos"):
                findings.append(f"{what} can pass though it fails: {why}")
            if attach is not None and i > attach:
                findings.append(f"{what} comes after {st[attach]['name']!r}, which attaches the archive: the archive is attached before it is run")
    for job, lines in js.items():
        if not on_macos(lines):
            continue
        st = steps(lines)
        first = next((i for i, s in enumerate(st) if runs_node(s)), None)
        if first is not None and not any(puts_node_on_path(s) for s in st[:first]):
            findings.append(f"{WORKFLOW}: the {job} job's step {st[first]['name']!r} runs node or npm, and no step before it puts {HOMEBREW} on the job's PATH, which launchd's PATH on the macOS runner omits")


def wai_command(rest):
    """The `wai` command in a step's text, alone: from `wai ` up to the first `;`, `}`, `'`,
    `|`, `>` or `&&` after it (what the step does with its output is not the command); None
    when the step runs no `wai` command."""
    i = rest.find("wai ")
    return re.split(r"[;}'|>]|&&", rest[i:])[0].strip() if i >= 0 else None


def sh_steps(text):
    """clean-machine.sh's steps: (name, the `wai` command it runs, or None)."""
    out = []
    for l in text.splitlines():
        t = l.strip()
        m = re.match(r"^(?:\[[^]]*\]\s*&&\s*)*step ([A-Za-z0-9-]+)(?: (.*))?$", t)
        if not m:
            continue
        out.append((m.group(1), wai_command(m.group(2) or "")))
    return out


def ps_steps(text):
    """clean-machine.ps1's steps: (name, the `wai` command its block runs, or None)."""
    out = []
    for l in text.splitlines():
        m = re.search(r"\bStep '([A-Za-z0-9-]+)' \{(.*)$", l.strip())
        if not m:
            continue
        out.append((m.group(1), wai_command(m.group(2))))
    return out


def block(text, start_re):
    """The lines of the block that starts on the line matching `start_re` and ends at the next
    line that is `}` alone at the start of a line."""
    lines = text.splitlines()
    for i, l in enumerate(lines):
        if re.search(start_re, l):
            for j in range(i + 1, len(lines)):
                if lines[j].rstrip() == "}":
                    return lines[i + 1:j]
            return lines[i + 1:]
    return None


def check_unpack(name, body, sums, compare, unpack, findings):
    """The unpack step's body reads the SHA-256 from `sums`, fails when it differs (`compare`),
    and does both before it unpacks (`unpack`)."""
    if body is None:
        findings.append(f"{name}: no unpack step")
        return
    code = [l.strip() for l in body if not l.strip().startswith("#")]
    reads = next((i for i, l in enumerate(code) if sums in l), None)
    fails = next((i for i, l in enumerate(code) if re.search(compare, l)), None)
    opens = next((i for i, l in enumerate(code) if re.search(unpack, l)), None)
    if reads is None:
        findings.append(f"{name}: the unpack step does not read the SHA-256 beside the archive ({sums})")
    if fails is None:
        findings.append(f"{name}: the unpack step does not fail when the archive's SHA-256 is not the one beside it")
    if opens is None:
        findings.append(f"{name}: the unpack step unpacks nothing")
    elif any(x is not None and x > opens for x in (reads, fails)):
        findings.append(f"{name}: the unpack step unpacks the archive before it checks its SHA-256")


def check_contents(name, body, runs, findings):
    """The contents step's body runs each of CONTENTS_RUNS; `runs` is the pattern of a line that
    runs one, `{b}` its name."""
    if body is None:
        findings.append(f"{name}: no contents step")
        return
    code = [l.strip() for l in body if not l.strip().startswith("#")]
    for b in CONTENTS_RUNS:
        if not any(re.search(runs.replace("{b}", re.escape(b)), l) for l in code):
            findings.append(f"{name}: the contents step does not run {b}: it only looks for it")


def code_lines(lines):
    """The lines that are code, stripped: neither blank nor a comment."""
    return [l.strip() for l in lines if l.strip() and not l.strip().startswith("#")]


def check_contents_status(sh, ps, findings):
    """A line of the contents step that runs a binary fails the step when the binary does."""
    body = block(sh, r"^contents\(\)\s*\{")
    if body is not None:
        code = code_lines(body)
        for i, l in enumerate(code):
            b = next((b for b in ["wai-quantum"] + CONTENTS_RUNS if re.match(rf"^(?:says_usage\s+)?{re.escape(b)}\s", l)), None)
            if b is None:
                continue
            if i == len(code) - 1:
                if re.search(r"[|;&]", l):
                    findings.append(f"{SH}: the contents step's last line, which runs {b}, goes on after it (`{l}`), so the step's status is not {b}'s")
            elif not re.match(r"^[^|;&]*\|\| return 1$", l):
                findings.append(f"{SH}: the contents step's line `{l}` does not end in `|| return 1`, so the step passes though {b} fails")
    body = block(ps, r"^\s*Step 'contents' \{")
    if body is not None:
        code = code_lines(body)
        for i, l in enumerate(code):
            b = next((b for b in ["wai-quantum"] + CONTENTS_RUNS if re.match(rf"^{re.escape(b)}\s", l)), None)
            if b is not None and not (i + 1 < len(code) and re.match(r"if \(\$LASTEXITCODE -ne 0\) \{\s*throw\b", code[i + 1])):
                findings.append(f"{PS}: the contents step's line `{l}` is not followed by `if ($LASTEXITCODE -ne 0) {{ throw ... }}`, so the step passes though {b} fails")
        at = next((i for i, l in enumerate(code) if re.match(r"if \(\$p\.ExitCode -ne \$u\.c\b", l)), None)
        if at is None or at + 1 >= len(code) or not code[at + 1].startswith("throw "):
            findings.append(f"{PS}: the contents step does not throw when a program it starts exits with another status than its own, or without its usage")


def check_exit(sh, ps, findings):
    """Each script's exit status is whether a step failed."""
    code = code_lines(sh.splitlines())
    if not code or code[-1] != '[ -z "$failed" ]':
        findings.append(f'{SH}: its last line is not `[ -z "$failed" ]`, so its exit status is not whether a step failed')
    body = code_lines(block(sh, r"^step\(\)\s*\{") or [])
    run = next((i for i, l in enumerate(body) if re.match(r'^if "\$@"[ ;>]', l)), None)
    other = next((i for i, l in enumerate(body) if l == "else"), None)
    sets = next((i for i, l in enumerate(body) if l == 'failed="$name"'), None)
    if run is None or other is None or sets is None or not run < other < sets:
        findings.append(f'{SH}: step() does not set `failed="$name"` in the branch where its command failed')
    for l in code:
        if re.match(r"trap\b", l):
            findings.append(f"{SH}: `{l}`: a trap can change the script's exit status")
        elif re.search(r"(?:^|[\s;&(])(?:failed=|unset\s+failed\b)", l) and l not in ('steps=() failed=""', 'failed="$name"'):
            findings.append(f"{SH}: `{l}` sets `failed` outside step(), so a failed step can leave the script's exit status 0")
    code = code_lines(ps.splitlines())
    if code[-2:] != ["if ($Failed) { exit 1 }", "exit 0"]:
        findings.append(f"{PS}: its last lines are not `if ($Failed) {{ exit 1 }}` and `exit 0`, so its exit status is not whether a step failed")
    body = code_lines(block(ps, r"^function Step\(") or [])
    status = "if ($LASTEXITCODE -ne 0) { $ok = $false }" in body
    caught = next((i for i, l in enumerate(body) if l.startswith("} catch {")), None)
    caught_ok = caught is not None and "$ok = $false" in body[caught:caught + 3]
    good = next((i for i, l in enumerate(body) if l == "if ($ok) {"), None)
    other = next((i for i, l in enumerate(body) if l == "} else {"), None)
    sets = next((i for i, l in enumerate(body) if l == "$script:Failed = $Name"), None)
    if not (status and caught_ok and good is not None and other is not None and sets is not None and good < other < sets):
        findings.append(f"{PS}: Step does not set `$script:Failed` when its block throws or its program exits non-zero")
    for l in code:
        if re.search(r"\$(?:script:|global:)?Failed\s*=(?!=)", l) and l not in ("$Failed = ''", "$script:Failed = $Name"):
            findings.append(f"{PS}: `{l}` sets `$Failed` outside Step, so a failed step can leave the script's exit status 0")
    # An exit before the end, in a step's function or the script's body, ends the script there:
    # with 0, or with no status (the last command's), a failed or unrun step passes.
    for name, lines in ((SH, code_lines(sh.splitlines())), (PS, code[:-1] if code[-1:] == ["exit 0"] else code)):
        for l in lines:
            for m in re.finditer(EXIT, l):
                if not re.fullmatch(r"[1-9][0-9]*", m.group(1) or ""):
                    findings.append(f"{name}: `{l}` can end the script with status 0 before a failed step decides it: each exit but the twin's last line has a non-zero number")


def check_scripts(sh, ps, findings):
    s, p = sh_steps(sh), ps_steps(ps)
    names = [n for n, _ in s]
    if names != SH_STEPS:
        findings.append(f"{SH}: its steps are {names}, not {SH_STEPS}")
    want = [x for x in s if x[0] != BROWSER_STEP]
    if p != want:
        findings.append(f"{PS}: its steps are {p}; the twin runs clean-machine.sh's, the browser's aside: {want}")
    check_unpack(SH, block(sh, r"^unpack\(\)\s*\{"), '"$ARCHIVE.sha256"', r'\[ "\$got" = "\$want" \] \|\| \{.*\breturn 1', r"\btar -x", findings)
    check_unpack(PS, block(ps, r"^Step 'unpack' \{"), '"$Archive.sha256"', r"if \(\$got -ne \$want\) \{\s*throw\b", r"\bExpand-Archive\b|\btar -x", findings)
    check_contents(SH, block(sh, r"^contents\(\)\s*\{"), r"^(?:says_usage\s+)?{b}\s", findings)
    check_contents(PS, block(ps, r"^\s*Step 'contents' \{"), r"^{b}\s|\bn = '{b}'", findings)
    check_contents_status(sh, ps, findings)
    check_exit(sh, ps, findings)


def main(argv):
    root = "."
    if argv[:1] == ["--root"] and len(argv) == 2:
        root = argv[1]
    elif argv:
        print("usage: check-wai-release-quickstart.py [--root DIR]", file=sys.stderr)
        return 2
    read = lambda p: open(os.path.join(root, p), encoding="utf-8").read()
    findings = []
    check_workflow(read(WORKFLOW), findings)
    check_scripts(read(SH), read(PS), findings)
    for f in findings:
        print(f"FAIL  {f}")
    if not findings:
        print(f"ok    {WORKFLOW}: the macos, linux and windows jobs run the quickstart from the archive before attaching it")
        print(f"ok    {PS} keeps {SH}'s steps, and both check the archive's SHA-256 before unpacking it and run what it holds")
    print(f"check-wai-release-quickstart: {len(findings)} finding(s)")
    return 1 if findings else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))