diff --git a/.github/workflows/hourly-pr-governance.yml b/.github/workflows/hourly-pr-governance.yml index b41e327d6..a0489f206 100644 --- a/.github/workflows/hourly-pr-governance.yml +++ b/.github/workflows/hourly-pr-governance.yml @@ -8,11 +8,14 @@ on: branches: [main] paths: - .github/workflows/hourly-pr-governance.yml + - scripts/audit_workflow_registry.py - scripts/build_pr_queue_governance.py - tests/test_hourly_pr_governance_workflow.py - tests/test_pr_queue_governance.py + - tests/test_workflow_registry_audit.py permissions: + actions: read contents: read pull-requests: read @@ -133,6 +136,13 @@ jobs: sleep "$delay" attempt=$((attempt + 1)) done + - name: Build live workflow registry drift evidence + env: + GH_TOKEN: ${{ github.token }} + run: | + python scripts/audit_workflow_registry.py \ + --repo ContextualWisdomLab/fast-mlsirm \ + --out hourly-pr-queue-governance/workflow_registry_audit.json - name: Publish hourly governance evidence if: always() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a @@ -141,5 +151,6 @@ jobs: path: | hourly-pr-queue-governance/pr_queue_governance_manifest.json hourly-pr-queue-governance/pr_queue_governance_report.html + hourly-pr-queue-governance/workflow_registry_audit.json if-no-files-found: error retention-days: 30 diff --git a/scripts/audit_workflow_registry.py b/scripts/audit_workflow_registry.py new file mode 100644 index 000000000..55b057221 --- /dev/null +++ b/scripts/audit_workflow_registry.py @@ -0,0 +1,473 @@ +#!/usr/bin/env python +"""Audit GitHub Actions workflow-registry drift without mutating repository state.""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import time +from collections import Counter, defaultdict +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Callable +from urllib.parse import quote + + +_RETRYABLE_HTTP_RE = re.compile(r"\bHTTP (?:502|503|504)\b", re.IGNORECASE) +_REPO_SLUG_RE = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$") +_MAX_ATTEMPTS = 3 +_RETRY_SLEEP_SECONDS = 0.5 +_GH_TIMEOUT_SECONDS = 20 +_WORKFLOW_PREFIX = ".github/workflows/" +_DYNAMIC_PREFIX = "dynamic/" + + +@dataclass(frozen=True) +class GitHubApiError(RuntimeError): + """Fail-closed GitHub CLI API error with bounded diagnostic evidence.""" + + endpoint: str + returncode: int + stderr: str + + def __str__(self) -> str: + return ( + f"GitHub API request failed for {self.endpoint}: " + f"exit {self.returncode}: {self.stderr}" + ) + + +FetchJson = Callable[[str], Any] + + +def _repo_slug(value: str) -> str: + """Validate one unambiguous GitHub ``owner/name`` repository slug.""" + if _REPO_SLUG_RE.fullmatch(value) is None: + raise argparse.ArgumentTypeError("repo must be in owner/name form") + return value + + +def _integer_argument(value: Any, *, name: str) -> int: + """Return an exact integer or normalize invalid input to ``ValueError``.""" + if type(value) is not int: + raise ValueError(f"{name} must be an integer") + return value + + +def _utc_timestamp(value: datetime | None = None) -> str: + """Return one RFC-3339 UTC observation timestamp.""" + observed = value or datetime.now(UTC) + if observed.tzinfo is None: + observed = observed.replace(tzinfo=UTC) + return observed.astimezone(UTC).isoformat().replace("+00:00", "Z") + + +def _run_gh_api( + endpoint: str, + *, + max_attempts: int = _MAX_ATTEMPTS, + retry_sleep_seconds: float = _RETRY_SLEEP_SECONDS, +) -> Any: + """Fetch one GitHub REST payload, retrying bounded transient failures.""" + attempts = max(1, _integer_argument(max_attempts, name="max_attempts")) + last_error: GitHubApiError | None = None + for attempt in range(1, attempts + 1): + try: + completed = subprocess.run( + ["gh", "api", endpoint], + capture_output=True, + text=True, + timeout=_GH_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as exc: + last_error = GitHubApiError( + endpoint=endpoint, + returncode=124, + stderr="GitHub API request timed out", + ) + if attempt >= attempts: + raise last_error from exc + if retry_sleep_seconds > 0: + time.sleep(retry_sleep_seconds) + continue + if completed.returncode == 0: + try: + return json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise GitHubApiError( + endpoint=endpoint, + returncode=completed.returncode, + stderr="GitHub API returned invalid JSON", + ) from exc + + stderr = completed.stderr.strip() + last_error = GitHubApiError( + endpoint=endpoint, + returncode=completed.returncode, + stderr=stderr, + ) + if attempt >= attempts or _RETRYABLE_HTTP_RE.search(stderr) is None: + break + if retry_sleep_seconds > 0: + time.sleep(retry_sleep_seconds) + + if last_error is None: + raise GitHubApiError(endpoint=endpoint, returncode=1, stderr="unknown failure") + raise last_error + + +def _require_mapping(payload: Any, *, context: str) -> dict[str, Any]: + """Return a mapping payload or fail closed.""" + if not isinstance(payload, dict): + raise RuntimeError(f"{context} payload is not an object") + return payload + + +def _default_branch(fetch_json: FetchJson, repo: str) -> str: + """Resolve the repository default branch from live repository metadata.""" + payload = _require_mapping(fetch_json(f"repos/{repo}"), context="repository") + branch = payload.get("default_branch") + if not isinstance(branch, str) or not branch.strip(): + raise RuntimeError("repository default_branch is missing") + return branch.strip() + + +def _branch_sha(fetch_json: FetchJson, repo: str, branch: str) -> str: + """Resolve an exact branch SHA through the Git ref API.""" + encoded = quote(branch, safe="") + payload = _require_mapping( + fetch_json(f"repos/{repo}/git/ref/heads/{encoded}"), + context="branch ref", + ) + obj = payload.get("object") + if not isinstance(obj, dict): + raise RuntimeError("branch ref object is missing") + sha = obj.get("sha") + if not isinstance(sha, str) or not re.fullmatch(r"[0-9a-fA-F]{40}", sha): + raise RuntimeError("branch ref SHA is invalid") + return sha.lower() + + +def _workflow_paths(fetch_json: FetchJson, repo: str, sha: str) -> set[str]: + """Return exact workflow source paths from a complete commit tree.""" + payload = _require_mapping( + fetch_json(f"repos/{repo}/git/trees/{sha}?recursive=1"), + context="git tree", + ) + if payload.get("truncated") is True: + raise RuntimeError("default-branch tree is truncated") + tree = payload.get("tree") + if not isinstance(tree, list): + raise RuntimeError("git tree entries are missing") + + paths: set[str] = set() + for entry in tree: + if not isinstance(entry, dict) or entry.get("type") != "blob": + continue + path = entry.get("path") + if ( + isinstance(path, str) + and path.startswith(_WORKFLOW_PREFIX) + and path.endswith((".yml", ".yaml")) + ): + paths.add(path) + return paths + + +def collect_workflow_registry( + fetch_json: FetchJson, + repo: str, + *, + per_page: int = 100, +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Collect every Actions workflow-registry page with pagination receipts.""" + page_size = _integer_argument(per_page, name="per_page") + if page_size <= 0 or page_size > 100: + raise ValueError("per_page must be between 1 and 100") + + workflows: list[dict[str, Any]] = [] + seen_workflow_ids: set[int] = set() + receipts: list[dict[str, int]] = [] + expected_total: int | None = None + page = 1 + + while expected_total is None or len(workflows) < expected_total: + endpoint = f"repos/{repo}/actions/workflows?per_page={page_size}&page={page}" + payload = _require_mapping(fetch_json(endpoint), context="workflow registry") + total = payload.get("total_count") + batch = payload.get("workflows") + if type(total) is not int or total < 0: + raise RuntimeError("workflow registry total_count is invalid") + if not isinstance(batch, list) or any( + not isinstance(item, dict) for item in batch + ): + raise RuntimeError("workflow registry page is malformed") + + if expected_total is None: + expected_total = total + elif total != expected_total: + raise RuntimeError("workflow registry total_count changed during pagination") + + receipts.append({"page": page, "count": len(batch)}) + if not batch and len(workflows) < expected_total: + raise RuntimeError("partial workflow registry pagination") + + for workflow in batch: + workflow_id = workflow.get("id") + if type(workflow_id) is int: + if workflow_id in seen_workflow_ids: + raise RuntimeError("workflow registry contains duplicate workflow id") + seen_workflow_ids.add(workflow_id) + workflows.extend(batch) + + if len(workflows) > expected_total: + raise RuntimeError("workflow registry returned more records than total_count") + page += 1 + + if expected_total is None: + raise RuntimeError("workflow registry pagination produced no pages") + return workflows, { + "per_page": page_size, + "total_count": expected_total, + "received_count": len(workflows), + "pages": receipts, + "complete": len(workflows) == expected_total, + } + + +def _identity_conflicts( + workflows: list[dict[str, Any]], +) -> tuple[set[int], set[str]]: + """Find reused IDs and duplicate active repository paths.""" + id_paths: dict[int, set[str]] = defaultdict(set) + active_path_ids: dict[str, set[int]] = defaultdict(set) + + for workflow in workflows: + workflow_id = workflow.get("id") + path = workflow.get("path") + state = workflow.get("state") + if type(workflow_id) is not int or not isinstance(path, str): + continue + id_paths[workflow_id].add(path) + if state == "active" and path.startswith(_WORKFLOW_PREFIX): + active_path_ids[path].add(workflow_id) + + conflicting_ids = { + workflow_id for workflow_id, paths in id_paths.items() if len(paths) > 1 + } + conflicting_paths = { + path for path, workflow_ids in active_path_ids.items() if len(workflow_ids) > 1 + } + return conflicting_ids, conflicting_paths + + +def classify_workflows( + workflows: list[dict[str, Any]], + *, + present_paths: set[str], + default_branch_sha: str, + observed_at: datetime | None = None, +) -> list[dict[str, Any]]: + """Classify registry identities using exact paths and integrity checks.""" + timestamp = _utc_timestamp(observed_at) + conflicting_ids, conflicting_paths = _identity_conflicts(workflows) + records: list[dict[str, Any]] = [] + + for workflow in workflows: + workflow_id = workflow.get("id") + path = workflow.get("path") + state = workflow.get("state") + name = workflow.get("name") + integrity_conflict = ( + type(workflow_id) is int and workflow_id in conflicting_ids + ) or (isinstance(path, str) and path in conflicting_paths) + + if integrity_conflict: + classification = "unresolved" + elif not isinstance(path, str) or not path: + classification = "unresolved" + elif state == "active": + if path.startswith(_DYNAMIC_PREFIX): + classification = "dynamic" + elif path.startswith(_WORKFLOW_PREFIX): + classification = "present" if path in present_paths else "orphan" + else: + classification = "unresolved" + elif isinstance(state, str) and state.startswith("disabled_"): + classification = "disabled" + else: + classification = "unresolved" + + records.append( + { + "id": workflow_id, + "name": name, + "path": path, + "state": state, + "classification": classification, + "integrity_conflict": integrity_conflict, + "default_branch_sha": default_branch_sha, + "observed_at": timestamp, + } + ) + + return records + + +def _summary(records: list[dict[str, Any]]) -> dict[str, int]: + """Count registry classifications with explicit active-orphan naming.""" + counts = Counter( + str(record.get("classification", "unresolved")) for record in records + ) + return { + "total": len(records), + "present": counts["present"], + "active_orphan": counts["orphan"], + "disabled": counts["disabled"], + "dynamic": counts["dynamic"], + "unresolved": counts["unresolved"], + } + + +def _audit_workflow_registry_snapshot( + fetch_json: FetchJson, + repo: str, + branch: str, + *, + observed_at: datetime | None = None, +) -> dict[str, Any]: + """Build one exact branch-bound registry snapshot and detect movement.""" + timestamp = _utc_timestamp(observed_at) + start_sha = _branch_sha(fetch_json, repo, branch) + present_paths = _workflow_paths(fetch_json, repo, start_sha) + workflows, pagination = collect_workflow_registry(fetch_json, repo) + records = classify_workflows( + workflows, + present_paths=present_paths, + default_branch_sha=start_sha, + observed_at=observed_at, + ) + end_sha = _branch_sha(fetch_json, repo, branch) + end_branch = _default_branch(fetch_json, repo) + branch_stable = end_branch == branch + sha_stable = end_sha == start_sha + stable = branch_stable and sha_stable + + errors: list[str] = [] + if not branch_stable: + errors.append("default branch changed during workflow registry audit") + elif not sha_stable: + errors.append("default branch moved during workflow registry audit") + if any(record["classification"] == "unresolved" for record in records): + errors.append("workflow registry contains unresolved identities") + + return { + "schema_version": 1, + "status": "ok" if stable and not errors else "failed", + "repository": repo, + "default_branch": branch, + "default_branch_sha": start_sha, + "end_default_branch": end_branch, + "end_default_branch_sha": end_sha, + "snapshot_stable": stable, + "observed_at": timestamp, + "workflow_tree": { + "count": len(present_paths), + "paths": sorted(present_paths), + }, + "pagination": pagination, + "summary": _summary(records), + "workflows": records, + "errors": errors, + } + + +def audit_workflow_registry( + fetch_json: FetchJson, + repo: str, + *, + observed_at: datetime | None = None, + max_snapshot_attempts: int = 1, +) -> dict[str, Any]: + """Build a SHA-bound audit, retrying moving or renamed defaults when requested.""" + attempts = max( + 1, + _integer_argument(max_snapshot_attempts, name="max_snapshot_attempts"), + ) + branch = _default_branch(fetch_json, repo) + receipts: list[dict[str, Any]] = [] + audit: dict[str, Any] | None = None + + for attempt in range(1, attempts + 1): + audit = _audit_workflow_registry_snapshot( + fetch_json, + repo, + branch, + observed_at=observed_at, + ) + receipts.append( + { + "attempt": attempt, + "start_sha": audit["default_branch_sha"], + "end_sha": audit["end_default_branch_sha"], + "stable": audit["snapshot_stable"], + } + ) + audit["snapshot_attempts"] = list(receipts) + if audit["snapshot_stable"]: + return audit + branch = audit["end_default_branch"] + + if audit is None: + raise RuntimeError("workflow registry audit produced no snapshot") + return audit + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Audit GitHub Actions registry drift against protected default branch." + ) + parser.add_argument( + "--repo", + required=True, + type=_repo_slug, + help="Repository in owner/name form.", + ) + parser.add_argument("--out", required=True, type=Path, help="Output JSON path.") + return parser.parse_args() + + +def main() -> int: + """Run the live read-only workflow-registry audit.""" + args = _parse_args() + try: + audit = audit_workflow_registry( + _run_gh_api, + args.repo, + max_snapshot_attempts=_MAX_ATTEMPTS, + ) + except (GitHubApiError, RuntimeError, ValueError) as exc: + audit = { + "schema_version": 1, + "status": "failed", + "repository": args.repo, + "observed_at": _utc_timestamp(), + "errors": [str(exc)], + } + exit_code = 2 + else: + exit_code = 0 if audit["status"] == "ok" else 2 + + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text( + json.dumps(audit, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py new file mode 100644 index 000000000..5616f1cb2 --- /dev/null +++ b/tests/test_workflow_registry_audit.py @@ -0,0 +1,289 @@ +"""Adversarial tests for read-only GitHub Actions workflow-registry auditing.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from scripts.audit_workflow_registry import ( + GitHubApiError, + audit_workflow_registry, + classify_workflows, + collect_workflow_registry, +) + + +REPO = "ContextualWisdomLab/fast-mlsirm" +SHA = "a" * 40 +OBSERVED = datetime(2026, 8, 12, 15, 0, tzinfo=UTC) + + +class FakeApi: + """Deterministic endpoint fixture for registry-audit tests.""" + + def __init__(self, responses): + self.responses = { + endpoint: list(values) if isinstance(values, list) else [values] + for endpoint, values in responses.items() + } + self.calls: list[str] = [] + + def __call__(self, endpoint: str): + self.calls.append(endpoint) + values = self.responses.get(endpoint) + if not values: + raise AssertionError(f"unexpected endpoint: {endpoint}") + value = values.pop(0) + if isinstance(value, Exception): + raise value + return value + + +def _repo_payload(): + return {"default_branch": "main"} + + +def _ref_payload(sha: str = SHA): + return {"object": {"sha": sha}} + + +def _tree_payload(*paths: str): + return { + "truncated": False, + "tree": [{"path": path, "type": "blob"} for path in paths], + } + + +def _workflow(workflow_id: int, path: str, *, state: str = "active", name: str = "wf"): + return {"id": workflow_id, "path": path, "state": state, "name": name} + + +def test_registry_pagination_collects_every_page_and_receipt(): + first = [_workflow(i, f".github/workflows/w{i}.yml") for i in range(100)] + second = [ + _workflow(101, ".github/workflows/final.yml"), + _workflow(102, ".github/workflows/last.yml"), + ] + api = FakeApi( + { + f"repos/{REPO}/actions/workflows?per_page=100&page=1": { + "total_count": 102, + "workflows": first, + }, + f"repos/{REPO}/actions/workflows?per_page=100&page=2": { + "total_count": 102, + "workflows": second, + }, + } + ) + + workflows, pagination = collect_workflow_registry(api, REPO) + + assert len(workflows) == 102 + assert pagination["complete"] is True + assert pagination["total_count"] == 102 + assert pagination["pages"] == [ + {"page": 1, "count": 100}, + {"page": 2, "count": 2}, + ] + + +def test_registry_pagination_fails_closed_on_partial_empty_page(): + first = [_workflow(i, f".github/workflows/w{i}.yml") for i in range(100)] + api = FakeApi( + { + f"repos/{REPO}/actions/workflows?per_page=100&page=1": { + "total_count": 102, + "workflows": first, + }, + f"repos/{REPO}/actions/workflows?per_page=100&page=2": { + "total_count": 102, + "workflows": [], + }, + } + ) + + with pytest.raises(RuntimeError, match="partial workflow registry pagination"): + collect_workflow_registry(api, REPO) + + +def test_registry_pagination_rejects_duplicate_ids_even_when_payload_matches(): + """Page drift cannot hide missing identities behind duplicate registry IDs.""" + first = [_workflow(i, f".github/workflows/w{i}.yml") for i in range(100)] + second = [first[-2], first[-1]] + api = FakeApi( + { + f"repos/{REPO}/actions/workflows?per_page=100&page=1": { + "total_count": 102, + "workflows": first, + }, + f"repos/{REPO}/actions/workflows?per_page=100&page=2": { + "total_count": 102, + "workflows": second, + }, + } + ) + + with pytest.raises(RuntimeError, match="duplicate workflow id"): + collect_workflow_registry(api, REPO) + + +def test_classification_uses_exact_paths_not_name_heuristics(): + present_snapshot = ".github/workflows/dev-legitimate-snapshot.yml" + workflows = [ + _workflow(1, present_snapshot, name="Development Snapshot"), + _workflow(2, ".github/workflows/Dev-Legitimate-Snapshot.yml"), + _workflow(3, ".github%2Fworkflows%2Fencoded.yml"), + _workflow(4, "dynamic/github-code-scanning/codeql", name="CodeQL"), + _workflow(5, ".github/workflows/old.yml", state="disabled_manually"), + ] + + records = classify_workflows( + workflows, + present_paths={present_snapshot}, + default_branch_sha=SHA, + observed_at=OBSERVED, + ) + by_id = {record["id"]: record for record in records} + + assert by_id[1]["classification"] == "present" + assert by_id[2]["classification"] == "orphan" + assert by_id[3]["classification"] == "unresolved" + assert by_id[4]["classification"] == "dynamic" + assert by_id[5]["classification"] == "disabled" + + +def test_classification_does_not_treat_unknown_state_as_disabled(): + """Only explicit disabled registry states are accepted as disabled evidence.""" + records = classify_workflows( + [_workflow(9, ".github/workflows/unknown.yml", state="mystery_state")], + present_paths=set(), + default_branch_sha=SHA, + observed_at=OBSERVED, + ) + + assert records[0]["classification"] == "unresolved" + + +def test_classification_marks_reused_id_and_duplicate_active_path_unresolved(): + workflows = [ + _workflow(7, ".github/workflows/a.yml"), + _workflow(7, ".github/workflows/b.yml"), + _workflow(8, ".github/workflows/a.yml"), + ] + + records = classify_workflows( + workflows, + present_paths={".github/workflows/a.yml", ".github/workflows/b.yml"}, + default_branch_sha=SHA, + observed_at=OBSERVED, + ) + + assert {record["classification"] for record in records} == {"unresolved"} + assert all(record["integrity_conflict"] for record in records) + + +def test_audit_binds_tree_and_registry_to_unchanged_default_branch_sha(): + present = ".github/workflows/ci.yml" + orphan = ".github/workflows/old-repair.yml" + api = FakeApi( + { + f"repos/{REPO}": [_repo_payload(), _repo_payload()], + f"repos/{REPO}/git/ref/heads/main": [_ref_payload(), _ref_payload()], + f"repos/{REPO}/git/trees/{SHA}?recursive=1": _tree_payload(present), + f"repos/{REPO}/actions/workflows?per_page=100&page=1": { + "total_count": 2, + "workflows": [_workflow(1, present), _workflow(2, orphan)], + }, + } + ) + + audit = audit_workflow_registry(api, REPO, observed_at=OBSERVED) + + assert audit["status"] == "ok" + assert audit["default_branch"] == "main" + assert audit["end_default_branch"] == "main" + assert audit["default_branch_sha"] == SHA + assert audit["snapshot_stable"] is True + assert audit["summary"]["active_orphan"] == 1 + assert audit["pagination"]["complete"] is True + assert {record["default_branch_sha"] for record in audit["workflows"]} == {SHA} + assert {record["observed_at"] for record in audit["workflows"]} == { + OBSERVED.isoformat().replace("+00:00", "Z") + } + + +def test_audit_fails_closed_when_default_branch_moves_mid_snapshot(): + moved = "b" * 40 + api = FakeApi( + { + f"repos/{REPO}": [_repo_payload(), _repo_payload()], + f"repos/{REPO}/git/ref/heads/main": [_ref_payload(), _ref_payload(moved)], + f"repos/{REPO}/git/trees/{SHA}?recursive=1": _tree_payload( + ".github/workflows/ci.yml" + ), + f"repos/{REPO}/actions/workflows?per_page=100&page=1": { + "total_count": 1, + "workflows": [_workflow(1, ".github/workflows/ci.yml")], + }, + } + ) + + audit = audit_workflow_registry(api, REPO, observed_at=OBSERVED) + + assert audit["status"] == "failed" + assert audit["snapshot_stable"] is False + assert audit["end_default_branch"] == "main" + assert audit["end_default_branch_sha"] == moved + assert audit["errors"] == ["default branch moved during workflow registry audit"] + + +def test_audit_rejects_truncated_git_tree_instead_of_inventing_orphans(): + api = FakeApi( + { + f"repos/{REPO}": _repo_payload(), + f"repos/{REPO}/git/ref/heads/main": _ref_payload(), + f"repos/{REPO}/git/trees/{SHA}?recursive=1": { + "truncated": True, + "tree": [], + }, + } + ) + + with pytest.raises(RuntimeError, match="default-branch tree is truncated"): + audit_workflow_registry(api, REPO, observed_at=OBSERVED) + + +@pytest.mark.parametrize("status", [403, 404, 500]) +def test_audit_surfaces_permission_missing_and_server_failures(status): + error = GitHubApiError( + endpoint=f"repos/{REPO}", + returncode=1, + stderr=f"HTTP {status}: synthetic failure", + ) + api = FakeApi({f"repos/{REPO}": error}) + + with pytest.raises(GitHubApiError, match=f"HTTP {status}"): + audit_workflow_registry(api, REPO, observed_at=OBSERVED) + + +def test_hourly_governance_publishes_workflow_registry_audit_read_only(): + """The hourly evidence loop observes registry drift without mutation authority.""" + workflow = ( + Path(__file__).parents[1] + / ".github" + / "workflows" + / "hourly-pr-governance.yml" + ).read_text(encoding="utf-8") + + assert "actions: read" in workflow + assert "actions: write" not in workflow + assert "contents: write" not in workflow + assert "pull-requests: write" not in workflow + assert "python scripts/audit_workflow_registry.py" in workflow + assert "--repo ContextualWisdomLab/fast-mlsirm" in workflow + assert "workflow_registry_audit.json" in workflow + assert "scripts/audit_workflow_registry.py" in workflow + assert "tests/test_workflow_registry_audit.py" in workflow diff --git a/tests/test_workflow_registry_audit_freshness.py b/tests/test_workflow_registry_audit_freshness.py new file mode 100644 index 000000000..f33b6fe97 --- /dev/null +++ b/tests/test_workflow_registry_audit_freshness.py @@ -0,0 +1,141 @@ +"""Fresh-state and retry contracts for the workflow-registry audit.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +from datetime import UTC, datetime + +import pytest + +from scripts.audit_workflow_registry import ( + _repo_slug, + _run_gh_api, + audit_workflow_registry, +) + + +REPO = "ContextualWisdomLab/fast-mlsirm" +SHA_A = "a" * 40 +SHA_B = "b" * 40 +OBSERVED = datetime(2026, 8, 13, 9, 0, tzinfo=UTC) + + +class FakeApi: + """Return queued deterministic responses for exact REST endpoints.""" + + def __init__(self, responses): + self.responses = { + endpoint: list(values) if isinstance(values, list) else [values] + for endpoint, values in responses.items() + } + self.calls: list[str] = [] + + def __call__(self, endpoint: str): + self.calls.append(endpoint) + values = self.responses.get(endpoint) + if not values: + raise AssertionError(f"unexpected endpoint: {endpoint}") + return values.pop(0) + + +def _ref_payload(sha: str): + return {"object": {"sha": sha}} + + +def _tree_payload(path: str): + return {"truncated": False, "tree": [{"path": path, "type": "blob"}]} + + +def _registry_payload(path: str): + return { + "total_count": 1, + "workflows": [{"id": 1, "path": path, "state": "active", "name": "CI"}], + } + + +def test_audit_retries_whole_snapshot_after_default_branch_movement(): + path = ".github/workflows/ci.yml" + endpoint = f"repos/{REPO}/actions/workflows?per_page=100&page=1" + repo_endpoint = f"repos/{REPO}" + api = FakeApi( + { + repo_endpoint: [ + {"default_branch": "main"}, + {"default_branch": "main"}, + {"default_branch": "main"}, + ], + f"repos/{REPO}/git/ref/heads/main": [ + _ref_payload(SHA_A), + _ref_payload(SHA_B), + _ref_payload(SHA_B), + _ref_payload(SHA_B), + ], + f"repos/{REPO}/git/trees/{SHA_A}?recursive=1": _tree_payload(path), + f"repos/{REPO}/git/trees/{SHA_B}?recursive=1": _tree_payload(path), + endpoint: [_registry_payload(path), _registry_payload(path)], + } + ) + + audit = audit_workflow_registry( + api, + REPO, + observed_at=OBSERVED, + max_snapshot_attempts=2, + ) + + assert audit["status"] == "ok" + assert audit["snapshot_stable"] is True + assert audit["default_branch"] == "main" + assert audit["end_default_branch"] == "main" + assert audit["default_branch_sha"] == SHA_B + assert audit["snapshot_attempts"] == [ + {"attempt": 1, "start_sha": SHA_A, "end_sha": SHA_B, "stable": False}, + {"attempt": 2, "start_sha": SHA_B, "end_sha": SHA_B, "stable": True}, + ] + assert api.calls.count(repo_endpoint) == 3 + assert api.calls.count(endpoint) == 2 + + +def test_run_gh_api_retries_timeout_before_succeeding(monkeypatch): + calls = 0 + + def fake_run(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise subprocess.TimeoutExpired(cmd=args[0], timeout=20) + return subprocess.CompletedProcess( + args=args[0], + returncode=0, + stdout=json.dumps({"ok": True}), + stderr="", + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert _run_gh_api("repos/example/project", max_attempts=2, retry_sleep_seconds=0) == { + "ok": True + } + assert calls == 2 + + +@pytest.mark.parametrize( + "value", + [ + "owner/repo/extra", + "owner/repo?ref=main", + "owner/repo#fragment", + "/repo", + "owner/", + "owner repo/project", + ], +) +def test_repo_slug_rejects_ambiguous_or_path_injecting_values(value): + with pytest.raises(argparse.ArgumentTypeError, match="owner/name"): + _repo_slug(value) + + +def test_repo_slug_accepts_standard_owner_name(): + assert _repo_slug("ContextualWisdomLab/fast-mlsirm") == REPO diff --git a/tests/test_workflow_registry_review_regressions.py b/tests/test_workflow_registry_review_regressions.py new file mode 100644 index 000000000..ec91750a3 --- /dev/null +++ b/tests/test_workflow_registry_review_regressions.py @@ -0,0 +1,91 @@ +"""Regression tests for workflow-registry review findings.""" + +from __future__ import annotations + +import pytest + +from scripts.audit_workflow_registry import ( + _run_gh_api, + audit_workflow_registry, + collect_workflow_registry, +) + +REPO = "ContextualWisdomLab/fast-mlsirm" +SHA = "a" * 40 + + +class FakeApi: + def __init__(self, responses): + self.responses = { + endpoint: list(values) if isinstance(values, list) else [values] + for endpoint, values in responses.items() + } + + def __call__(self, endpoint: str): + values = self.responses.get(endpoint) + if not values: + raise AssertionError(f"unexpected endpoint: {endpoint}") + return values.pop(0) + + +def _ref_payload(sha: str = SHA): + return {"object": {"sha": sha}} + + +def _tree_payload(path: str): + return {"truncated": False, "tree": [{"path": path, "type": "blob"}]} + + +def _registry_payload(path: str): + return { + "total_count": 1, + "workflows": [{"id": 1, "path": path, "state": "active", "name": "CI"}], + } + + +@pytest.mark.parametrize("value", [None, 1.5, float("inf"), object()]) +def test_run_gh_api_rejects_invalid_max_attempts(value): + with pytest.raises(ValueError, match="max_attempts must be an integer"): + _run_gh_api("repos/example/project", max_attempts=value, retry_sleep_seconds=0) + + +@pytest.mark.parametrize("value", [None, 1.5, float("inf"), object()]) +def test_collect_registry_rejects_invalid_per_page(value): + def should_not_fetch(_endpoint: str): + raise AssertionError("invalid pagination controls must fail before REST access") + + with pytest.raises(ValueError, match="per_page must be an integer"): + collect_workflow_registry(should_not_fetch, REPO, per_page=value) + + +@pytest.mark.parametrize("value", [None, 1.5, float("inf"), object()]) +def test_audit_rejects_invalid_snapshot_attempts(value): + def should_not_fetch(_endpoint: str): + raise AssertionError("invalid snapshot controls must fail before REST access") + + with pytest.raises(ValueError, match="max_snapshot_attempts must be an integer"): + audit_workflow_registry(should_not_fetch, REPO, max_snapshot_attempts=value) + + +def test_audit_rejects_default_branch_rename_with_unchanged_old_ref(): + path = ".github/workflows/ci.yml" + registry_endpoint = f"repos/{REPO}/actions/workflows?per_page=100&page=1" + api = FakeApi( + { + f"repos/{REPO}": [ + {"default_branch": "main"}, + {"default_branch": "release"}, + ], + f"repos/{REPO}/git/ref/heads/main": [_ref_payload(), _ref_payload()], + f"repos/{REPO}/git/trees/{SHA}?recursive=1": _tree_payload(path), + registry_endpoint: _registry_payload(path), + } + ) + + audit = audit_workflow_registry(api, REPO) + + assert audit["status"] == "failed" + assert audit["snapshot_stable"] is False + assert audit["default_branch"] == "main" + assert audit["end_default_branch"] == "release" + assert "default branch changed during workflow registry audit" in audit["errors"] diff --git a/tests/test_workflow_registry_runtime_guards.py b/tests/test_workflow_registry_runtime_guards.py new file mode 100644 index 000000000..b06cd06bd --- /dev/null +++ b/tests/test_workflow_registry_runtime_guards.py @@ -0,0 +1,18 @@ +"""Regression guards for workflow-registry runtime integrity checks.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +def test_workflow_registry_runtime_integrity_does_not_depend_on_assert() -> None: + """Runtime fail-closed guards must survive ``python -O`` optimization.""" + source_path = ( + Path(__file__).parents[1] / "scripts" / "audit_workflow_registry.py" + ) + tree = ast.parse(source_path.read_text(encoding="utf-8")) + + assert not any(isinstance(node, ast.Assert) for node in ast.walk(tree)), ( + "workflow-registry runtime integrity must not depend on assert statements" + )