rustd-resolved 0.2.1

#!/usr/bin/env python3
"""Reject accidental and obsolete GitHub Actions launchers.

The repository previously accumulated one-shot, self-mutating workflows that
were meant to delete themselves after landing a tested change. Invalid or
stranded launchers caused every push to show unrelated red workflow runs. Keep
only the permanent workflow fleet here so additions and removals are explicit.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

import yaml
from yaml.constructor import ConstructorError
from yaml.nodes import MappingNode

ROOT = Path(__file__).resolve().parents[1]
WORKFLOW_DIR = ROOT / ".github" / "workflows"

PERMANENT_WORKFLOWS = {
    "build-and-test.yml",
    "dnssd-live.yml",
    "llmnr-live.yml",
    "mdns-duplex.yml",
    "mdns-live.yml",
    "mdns-responder-live.yml",
    "pin-upstream-resolved.yml",
    "replacement-boot-proof.yml",
    "replacement-full-certification.yml",
    "replacement-readiness-certificate.yml",
    "replacement-security-gates.yml",
    "replacement-security-proof.yml",
    "replacement-upstream-test-75.yml",
    "replacement-upstream-test-89-mdns.yml",
    "reproducible-release.yml",
    "rustd-naming.yml",
    "upstream-surface-audit.yml",
    "verify-upstream-baseline.yml",
}

OBSOLETE_PREFIXES = ("finalize-", "fix-", "integrate-", "land-", "reconcile-")
TOP_LEVEL_KEYS = ("name", "on", "jobs")
MKOSI_COMMIT = "60ed8c964f8d98aa4b325f381c4b3bc6de91a0b7"
PINNED_ACTIONS = {
    "actions/cache": "0057852bfaa89a56745cba8c7296529d2fc39830",
    "actions/checkout": "11d5960a326750d5838078e36cf38b85af677262",
    "actions/upload-artifact": "ea165f8d65b6e75b540449e92b4886f43607fa02",
}
EXACT_SHA_WORKFLOWS = {
    "reproducible-release.yml",
    "replacement-boot-proof.yml",
    "replacement-full-certification.yml",
    "replacement-readiness-certificate.yml",
    "replacement-security-gates.yml",
    "replacement-security-proof.yml",
    "replacement-upstream-test-75.yml",
    "replacement-upstream-test-89-mdns.yml",
}


class UniqueKeyLoader(yaml.SafeLoader):
    pass


# PyYAML implements YAML 1.1 booleans, where the ordinary workflow key ``on``
# is parsed as True.  Keep scalar values intact so all security-sensitive keys
# are inspected by their actual normalized string value.
UniqueKeyLoader.yaml_implicit_resolvers = {
    key: [
        (tag, expression)
        for tag, expression in resolvers
        if tag != "tag:yaml.org,2002:bool"
    ]
    for key, resolvers in yaml.SafeLoader.yaml_implicit_resolvers.items()
}


def construct_unique_mapping(
    loader: UniqueKeyLoader, node: MappingNode, deep: bool = False
) -> dict[object, object]:
    if not isinstance(node, MappingNode):
        raise ConstructorError(None, None, "expected a mapping node", node.start_mark)
    mapping: dict[object, object] = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        try:
            duplicate = key in mapping
        except TypeError as error:
            raise ConstructorError(
                "while constructing a mapping",
                node.start_mark,
                "found an unhashable mapping key",
                key_node.start_mark,
            ) from error
        if duplicate:
            raise ConstructorError(
                "while constructing a mapping",
                node.start_mark,
                f"found duplicate key {key!r}",
                key_node.start_mark,
            )
        mapping[key] = loader.construct_object(value_node, deep=deep)
    return mapping


UniqueKeyLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    construct_unique_mapping,
)


def parse_workflow(text: str) -> dict[object, object]:
    try:
        value = yaml.load(text, Loader=UniqueKeyLoader)
    except yaml.YAMLError as error:
        raise ValueError(f"invalid or ambiguous YAML: {error}") from error
    if not isinstance(value, dict):
        raise ValueError("workflow YAML root is not a mapping")
    return value


def fail(message: str) -> None:
    print(f"workflow fleet check failed: {message}", file=sys.stderr)
    raise SystemExit(1)


def has_floating_nightly(text: str) -> bool:
    for match in re.finditer(
        r"(?<![-\w])nightly(?:-[A-Za-z0-9${}_]+)*",
        text,
    ):
        if re.fullmatch(r"nightly-[0-9]{4}-[0-9]{2}-[0-9]{2}", match.group()) is None:
            return True
    return False


def has_unpinned_mkosi(text: str) -> bool:
    lines = text.splitlines()
    step_starts = [
        index for index, line in enumerate(lines) if re.match(r"^\s*-\s+name\s*:", line)
    ]
    for index, line in enumerate(lines):
        if "github.com/systemd/mkosi" not in line:
            continue
        start = max((value for value in step_starts if value <= index), default=0)
        end = min((value for value in step_starts if value > index), default=len(lines))
        step = "\n".join(lines[start:end])
        logical_step = re.sub(r"\\\s*\n\s*", " ", step)
        mkosi_url = r"https://github\.com/systemd/mkosi(?:\.git)?"
        escaped_commit = re.escape(MKOSI_COMMIT)
        clone = re.search(rf"\bgit\s+clone\b[^\n]*{mkosi_url}", logical_step)
        fetch = re.search(
            rf"\bgit\s+-C\s+[^\n]*mkosi[^\n]*\sfetch\b[^\n]*{escaped_commit}",
            logical_step,
        )
        checkout = re.search(
            rf"\bgit\s+-C\s+[^\n]*mkosi[^\n]*\scheckout\b[^\n]*{escaped_commit}",
            logical_step,
        )
        if clone is None or fetch is None or checkout is None:
            return True
    return False


def mappings(value: object, seen: set[int] | None = None):  # noqa: ANN201
    if seen is None:
        seen = set()
    identity = id(value)
    if identity in seen:
        return
    if isinstance(value, dict):
        seen.add(identity)
        yield value
        for item in value.values():
            yield from mappings(item, seen)
    elif isinstance(value, list):
        seen.add(identity)
        for item in value:
            yield from mappings(item, seen)


def unpinned_actions(document: dict[object, object]) -> list[str]:
    failures: list[str] = []
    for mapping in mappings(document):
        if "uses" not in mapping:
            continue
        raw_value = mapping["uses"]
        if not isinstance(raw_value, str):
            failures.append("<non-string uses>")
            continue
        value = raw_value.strip()
        if value.startswith("./") and re.fullmatch(r"\./[^\s]+", value):
            continue
        if value.startswith("docker://"):
            if re.fullmatch(r"docker://[^@\s]+@sha256:[0-9a-f]{64}", value) is None:
                failures.append(value or "<empty uses>")
            continue
        remote = re.fullmatch(r"([^@\s]+)@([0-9a-f]{40})", value)
        if remote is None:
            failures.append(value or "<empty uses>")
            continue
        action, reference = remote.groups()
        expected = PINNED_ACTIONS.get(action)
        if expected is not None and reference != expected:
            failures.append(value)
    return failures


def immutable_image(value: str) -> bool:
    return re.fullmatch(r"[^@\s]+@sha256:[0-9a-f]{64}", value) is not None


def unpinned_images(document: dict[object, object]) -> list[str]:
    """Find mutable job container and service image references."""
    failures: list[str] = []
    jobs = document.get("jobs")
    if not isinstance(jobs, dict):
        return ["jobs mapping is missing"]
    for job_name, raw_job in jobs.items():
        if not isinstance(raw_job, dict):
            continue
        container = raw_job.get("container")
        if container is not None:
            if isinstance(container, str):
                image = container.strip()
            elif isinstance(container, dict) and isinstance(container.get("image"), str):
                image = container["image"].strip()
            else:
                failures.append(f"container for {job_name}: <invalid>")
                image = ""
            if image and not immutable_image(image):
                failures.append(f"container for {job_name}: {image}")
        services = raw_job.get("services")
        if services is None:
            continue
        if not isinstance(services, dict):
            failures.append(f"services for {job_name}: <invalid>")
            continue
        for service_name, raw_service in services.items():
            if isinstance(raw_service, dict) and isinstance(raw_service.get("image"), str):
                image = raw_service["image"].strip()
            else:
                failures.append(f"service {service_name}: <invalid>")
                continue
            if not immutable_image(image):
                failures.append(f"service {service_name}: {image or '<empty>'}")
    return failures


def main() -> None:
    assert has_floating_nightly("cargo +nightly test")
    assert has_floating_nightly("rustup toolchain install 'nightly'")
    assert has_floating_nightly("rustup toolchain install \\\n      \"nightly\"")
    assert has_floating_nightly("cargo +nightly-${TOOLCHAIN_DATE} test")
    assert has_floating_nightly("cargo +nightly-2025-02-15-extra test")
    assert not has_floating_nightly("cargo +nightly-2025-02-15 test")
    assert has_unpinned_mkosi("pipx install git+https://github.com/systemd/mkosi.git")
    assert has_unpinned_mkosi("git clone https://github.com/systemd/mkosi /tmp/mkosi")
    pinned_mkosi = (
        "- name: Install mkosi\n"
        "  run: |\n"
        "    git clone --no-checkout \\\n      https://github.com/systemd/mkosi.git /tmp/mkosi\n"
        f"    git -C /tmp/mkosi fetch origin {MKOSI_COMMIT}\n"
        f"    git -C /tmp/mkosi checkout --detach {MKOSI_COMMIT}\n"
    )
    assert not has_unpinned_mkosi(
        pinned_mkosi
    )
    assert has_unpinned_mkosi(
        pinned_mkosi
        + "- name: Mutable mkosi\n"
        + "  run: git clone https://github.com/systemd/mkosi.git /tmp/other\n"
    )
    def parsed(text: str) -> dict[object, object]:
        return parse_workflow("name: fixture\non: workflow_dispatch\njobs:\n" + text)

    assert unpinned_actions(parsed("  test:\n    uses: actions/checkout@v4\n")) == [
        "actions/checkout@v4"
    ]
    assert unpinned_actions(parsed("  test:\n    uses: owner/action\n")) == ["owner/action"]
    assert unpinned_actions(parsed("  test:\n    'uses': owner/action@main\n")) == [
        "owner/action@main"
    ]
    assert unpinned_actions(parsed("  test:\n    uses: owner/action@main\n")) == [
        "owner/action@main"
    ]
    assert unpinned_actions(parsed("  test:\n    uses: owner/action@" + "a" * 40 + "\n")) == []
    assert unpinned_actions(parsed("  test:\n    uses: owner/action@" + "A" * 40 + "\n")) == [
        "owner/action@" + "A" * 40
    ]
    assert unpinned_actions(parsed("  test:\n    uses: actions/cache@" + "a" * 40 + "\n")) == [
        "actions/cache@" + "a" * 40
    ]
    assert unpinned_actions(parsed("  test:\n    uses: ./local-action\n")) == []
    assert unpinned_actions(parsed("  test:\n    uses: docker://alpine:latest\n")) == [
        "docker://alpine:latest"
    ]
    assert unpinned_actions(
        parsed("  test:\n    uses: docker://alpine@sha256:" + "b" * 64 + "\n")
    ) == []
    assert unpinned_actions(
        parsed("  test:\n    uses: docker://alpine@sha256:" + "B" * 64 + "\n")
    ) == ["docker://alpine@sha256:" + "B" * 64]
    assert unpinned_actions(
        parsed(
            "  test:\n    uses: actions/checkout@"
            + PINNED_ACTIONS["actions/checkout"]
            + " # v4.4.0\n"
        )
    ) == []
    escaped_uses = '"u\\u0073es"'
    assert unpinned_actions(
        parsed(f"  test:\n    steps:\n      - {{{escaped_uses}: actions/checkout@v4}}\n")
    ) == ["actions/checkout@v4"]
    assert unpinned_actions(
        parsed("  test:\n    steps:\n      - !!str uses: actions/checkout@v4\n")
    ) == ["actions/checkout@v4"]
    assert unpinned_actions(
        parsed(
            "  test:\n"
            "    steps:\n"
            "      - ? uses\n"
            "        : actions/checkout@v4\n"
        )
    ) == ["actions/checkout@v4"]
    try:
        parse_workflow(
            "name: duplicate\non: workflow_dispatch\njobs:\n"
            "  test:\n    uses: owner/one@main\n    uses: owner/two@main\n"
        )
    except ValueError as error:
        assert "duplicate key 'uses'" in str(error)
    else:
        raise AssertionError("duplicate workflow keys were accepted")
    digest = "sha256:" + "c" * 64
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            "  test:\n"
            f"    container: ghcr.io/example/test@{digest}\n"
            "    services:\n"
            "      database:\n"
            f"        image: postgres@{digest}\n"
        )
    ) == []
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            "  test:\n    container: ubuntu:latest\n"
        )
    ) == ["container for test: ubuntu:latest"]
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            "  test:\n"
            "    container:\n"
            "      image: ubuntu\n"
            "    services:\n"
            "      database:\n"
            "        image: postgres:latest\n"
        )
    ) == ["container for test: ubuntu", "service database: postgres:latest"]
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            "  'quoted-job': &job\n"
            "    container: &container\n"
            "      image: ubuntu:latest\n"
        )
    ) == ["container for quoted-job: ubuntu:latest"]
    escaped_container = '"conta\\u0069ner"'
    escaped_image = '"im\\u0061ge"'
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            f"  test: {{{escaped_container}: {{{escaped_image}: ubuntu:latest}}}}\n"
        )
    ) == ["container for test: ubuntu:latest"]
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            "  test:\n"
            "    ? services\n"
            "    : database:\n"
            "        !!str image: postgres:latest\n"
        )
    ) == ["service database: postgres:latest"]
    assert unpinned_images(
        parse_workflow(
            "name: fixture\non: workflow_dispatch\njobs:\n"
            "  test:\n"
            "    steps:\n"
            "      - name: This is an ordinary action input\n"
            "        with:\n"
            "          image: mutable-but-not-a-runtime-container\n"
        )
    ) == []
    if not WORKFLOW_DIR.is_dir():
        fail(f"missing workflow directory: {WORKFLOW_DIR}")

    paths = sorted(
        path
        for path in WORKFLOW_DIR.iterdir()
        if path.is_file() and path.suffix in {".yml", ".yaml"}
    )
    actual = {path.name for path in paths}
    missing = sorted(PERMANENT_WORKFLOWS - actual)
    unexpected = sorted(actual - PERMANENT_WORKFLOWS)
    if missing:
        fail("missing permanent workflows: " + ", ".join(missing))
    if unexpected:
        fail("unexpected workflows: " + ", ".join(unexpected))

    obsolete = sorted(name for name in actual if name.startswith(OBSOLETE_PREFIXES))
    if obsolete:
        fail("obsolete integration launchers remain: " + ", ".join(obsolete))

    for path in paths:
        text = path.read_text(encoding="utf-8")
        if not text.strip():
            fail(f"empty workflow: {path.name}")
        if "\t" in text:
            fail(f"tab indentation in {path.name}")
        try:
            document = parse_workflow(text)
        except ValueError as error:
            fail(f"cannot parse {path.name}: {error}")
        for key in TOP_LEVEL_KEYS:
            if key not in document:
                fail(f"{path.name} is missing top-level {key!r}")
        if "git push origin HEAD:main" in text and path.name != "pin-upstream-resolved.yml":
            fail(f"self-mutating permanent workflow: {path.name}")
        if re.search(r"cargo build[^\n]*--release[^\n]*--all-features", text):
            fail(f"research features enabled in release artifact: {path.name}")
        if has_floating_nightly(text):
            fail(f"floating Rust nightly in {path.name}")
        if has_unpinned_mkosi(text):
            fail(f"unpinned mkosi source in {path.name}")
        action_failures = unpinned_actions(document)
        if action_failures:
            fail(
                f"unpinned external action in {path.name}: "
                + ", ".join(action_failures)
            )
        image_failures = unpinned_images(document)
        if image_failures:
            fail(
                f"unpinned container image in {path.name}: "
                + ", ".join(image_failures)
            )
        if path.name in EXACT_SHA_WORKFLOWS:
            if "source_sha:" not in text:
                fail(f"exact-SHA workflow has no source_sha input: {path.name}")
            if "run-name:" not in text or "inputs.source_sha || github.sha" not in text:
                fail(f"exact-SHA workflow has no bound run name: {path.name}")
            checkout_count = text.count(
                "uses: actions/checkout@" + PINNED_ACTIONS["actions/checkout"]
            )
            bound_checkout_count = text.count(
                "ref: ${{ inputs.source_sha || github.sha }}"
            )
            if checkout_count == 0 or checkout_count != bound_checkout_count:
                fail(f"exact-SHA checkout is incomplete: {path.name}")
            identity_count = text.count("name: Verify exact source identity")
            if identity_count != checkout_count:
                fail(f"exact-SHA identity check is incomplete: {path.name}")

    orchestrator = (WORKFLOW_DIR / "replacement-full-certification.yml").read_text(
        encoding="utf-8"
    )
    prerequisite_workflows = EXACT_SHA_WORKFLOWS - {
        "replacement-full-certification.yml"
    }
    for workflow in sorted(prerequisite_workflows):
        if workflow not in orchestrator:
            fail(f"full certification does not dispatch {workflow}")

    canonical_binary_workflows = {
        "replacement-boot-proof.yml",
        "replacement-readiness-certificate.yml",
        "replacement-upstream-test-75.yml",
        "replacement-upstream-test-89-mdns.yml",
        "reproducible-release.yml",
    }
    for name in sorted(canonical_binary_workflows):
        text = (WORKFLOW_DIR / name).read_text(encoding="utf-8")
        if "scripts/build-reproducible-release.sh" not in text:
            fail(f"certification workflow does not use the reproducible builder: {name}")
        if "rustup toolchain install 1.74.0" not in text:
            fail(f"certification workflow does not pin Rust 1.74.0: {name}")
        if name != "reproducible-release.yml" and "target/reproducible-release" not in text:
            fail(f"certification workflow does not consume canonical artifacts: {name}")

    print(f"workflow fleet check passed: {len(paths)} permanent workflows")


if __name__ == "__main__":
    main()