check_action_drift.py

#!/usr/bin/env python3
"""Detect drift between SHA-pinned GitHub Actions and their upstream tags.

METHOD-Kit family G4. A determinism-pinned action is monitored, never silently
refreshed (METHOD section 8): this script *detects* drift and reports it; the
workflow turns a non-zero exit into an issue. It never edits workflow files.

Decidable rules (each has a matching red-floor assertion):

  R1 (form)        - a `uses:` ref pinned by a mutable tag (e.g. `@v4`) instead
                     of a 40-hex commit SHA. The pin is not immutable.
  R2 (content)     - a pinned SHA that no longer matches the commit the named
                     version tag resolves to upstream (tag re-pointed, or stale).
  R3 (traceability)- a `@<sha>` ref with no trailing `# vX.Y.Z` comment: we
                     cannot tell which human version to monitor.

Annotated vs lightweight tags: `git ls-remote <repo> <tag>` returns the tag
object for an *annotated* tag and an extra `<tag>^{}` line for the commit it
dereferences to. We always compare against the *commit*, so the resolver must
prefer the `^{}` line when present (else a false R2 on every annotated tag).

The network is injected (a resolver callable mapping (repo, version) -> commit
SHA) so the logic is tested deterministically, outside any agent loop. In CI
the workflow passes the real `git ls-remote`-backed resolver.
"""

from __future__ import annotations

import argparse
import re
import subprocess  # noqa: S404  (fixed-arg list form, no shell; see run_ls_remote)
import sys
from dataclasses import dataclass
from pathlib import Path

SHA40 = re.compile(r"^[0-9a-f]{40}$")

# `uses: owner/repo[/sub/path]@ref  [# comment]`
USES = re.compile(
    r"""^\s*-?\s*uses:\s*
        (?P<slug>[^@\s]+)        # owner/repo[/path]
        @(?P<ref>\S+)            # ref: a 40-hex SHA or a mutable tag
        (?:\s*\#\s*(?P<comment>.+?))?\s*$""",
    re.VERBOSE,
)

# A version token inside the trailing comment, e.g. `v7.0.0` or `v4`.
VERSION = re.compile(r"\bv\d+(?:\.\d+){0,2}\b")


@dataclass(frozen=True)
class Finding:
    rule: str  # "R1" | "R2" | "R3"
    file: str
    line_no: int
    slug: str
    detail: str

    def __str__(self) -> str:
        return f"{self.rule}  {self.file}:{self.line_no}  {self.slug}  {self.detail}"


def repo_of(slug: str) -> str:
    """owner/repo from a slug that may carry a sub-path (codeql-action/init)."""
    parts = slug.split("/")
    return "/".join(parts[:2])


def parse_uses(text: str, filename: str):
    """Yield (line_no, slug, ref, comment) for every `uses:` ref in `text`."""
    for i, raw in enumerate(text.splitlines(), start=1):
        m = USES.match(raw)
        if m:
            yield i, m.group("slug"), m.group("ref"), m.group("comment")


def analyze_file(text: str, filename: str, resolver) -> list[Finding]:
    """Apply R1/R2/R3 to one workflow file. `resolver(repo, version) -> sha|None`."""
    findings: list[Finding] = []
    for line_no, slug, ref, comment in parse_uses(text, filename):
        if not SHA40.match(ref):
            # R1: not a commit SHA -> mutable pin.
            findings.append(
                Finding(
                    "R1", filename, line_no, slug, f"mutable ref '@{ref}', expected a 40-hex SHA"
                )
            )
            continue
        version = None
        if comment:
            vm = VERSION.search(comment)
            if vm:
                version = vm.group(0)
        if version is None:
            # R3: pinned by SHA but no version comment to monitor.
            findings.append(
                Finding(
                    "R3",
                    filename,
                    line_no,
                    slug,
                    f"SHA pin '@{ref[:12]}...' has no '# vX.Y.Z' comment",
                )
            )
            continue
        upstream = resolver(repo_of(slug), version)
        if upstream is None:
            findings.append(
                Finding(
                    "R2",
                    filename,
                    line_no,
                    slug,
                    f"cannot resolve {version} upstream (deleted tag?)",
                )
            )
        elif upstream != ref:
            # R2: pinned SHA no longer matches what the tag points to upstream.
            findings.append(
                Finding(
                    "R2",
                    filename,
                    line_no,
                    slug,
                    f"{version} resolves to {upstream[:12]}... but pin is {ref[:12]}...",
                )
            )
    return findings


def run_ls_remote(repo: str, version: str) -> str | None:
    """Real resolver: the commit SHA that `version` points to upstream.

    Prefers the dereferenced `<tag>^{}` line (annotated tags) over the bare tag
    line, so an annotated tag yields its *commit*, not the tag object.
    """
    url = f"https://github.com/{repo}"
    proc = subprocess.run(  # noqa: S603  (fixed list, no shell, trusted argv)
        ["git", "ls-remote", url, f"refs/tags/{version}^{{}}", f"refs/tags/{version}"],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=30,
        check=False,
    )
    if proc.returncode != 0:
        return None
    bare: str | None = None
    for row in proc.stdout.splitlines():
        if not row.strip():
            continue
        sha, _, ref = row.partition("\t")
        if ref.endswith("^{}"):
            return sha  # dereferenced commit wins outright
        bare = sha
    return bare


def collect_files(paths: list[str]) -> list[Path]:
    out: list[Path] = []
    for p in paths:
        path = Path(p)
        if path.is_dir():
            out.extend(sorted(path.glob("*.yml")))
            out.extend(sorted(path.glob("*.yaml")))
        elif path.is_file():
            out.append(path)
    return out


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Detect SHA-pin drift in GitHub Actions workflows."
    )
    parser.add_argument(
        "paths",
        nargs="*",
        default=[".github/workflows"],
        help="workflow files or directories (default: .github/workflows)",
    )
    args = parser.parse_args(argv)

    files = collect_files(args.paths or [".github/workflows"])
    if not files:
        print("no workflow files found", file=sys.stderr)
        return 1

    all_findings: list[Finding] = []
    for f in files:
        text = f.read_text(encoding="utf-8")
        all_findings.extend(analyze_file(text, str(f), run_ls_remote))

    if not all_findings:
        print(f"OK: {len(files)} workflow file(s), all action pins immutable and current")
        return 0

    print(f"DRIFT: {len(all_findings)} finding(s)")
    for fnd in all_findings:
        print(f"  {fnd}")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())


# --- floor-coverage registry ---
CODES = frozenset(
    {
        "R1",
        "R2",
        "R3",
    }
)
FLOOR = "../tests/test_check_action_drift.py"