diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py new file mode 100644 index 0000000..18fe684 --- /dev/null +++ b/scripts/ci/audit_workflow_registry.py @@ -0,0 +1,574 @@ +"""Audit GitHub Actions workflow identities without mutating repository state. + +GitHub keeps an Actions workflow identity independently from the workflow YAML +that originally created it. Removing a temporary workflow file therefore does +not prove that the corresponding registry identity was disabled. This module +builds a bounded, exact-revision audit that keeps those two authorities separate. + +The script is intentionally read-only. It never disables a workflow, updates a +branch, comments on a pull request, or requests additional credentials. +""" + +from __future__ import annotations + +import argparse +import http.client +import json +import math +import os +import re +import sys +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable, Iterable, Mapping +from datetime import datetime, timezone +from typing import Any + +API_ROOT = "https://api.github.com" +WORKFLOW_PREFIX = ".github/workflows/" +DYNAMIC_PREFIX = "dynamic/" +MAX_RESPONSE_BYTES = 8 * 1024 * 1024 +DEFAULT_TIMEOUT_SECONDS = 10.0 +DEFAULT_PER_PAGE = 100 +DEFAULT_MAX_PAGES = 100 +DEFAULT_MAX_PR_PAGES = 10 +DEFAULT_MAX_PR_FILE_PAGES = 10 +_REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +_SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$") +_WORKFLOW_STATES = frozenset( + { + "active", + "deleted", + "disabled_fork", + "disabled_inactivity", + "disabled_manually", + } +) + + +class AuditError(RuntimeError): + """Report an incomplete or internally inconsistent workflow audit.""" + + +def _require_sha(value: object, *, label: str) -> str: + """Return one exact lowercase Git commit SHA or fail closed.""" + if not isinstance(value, str) or _SHA_PATTERN.fullmatch(value) is None: + raise AuditError(f"{label} is not an exact commit SHA") + return value + + +def _require_repository(value: str) -> str: + """Validate the public ``owner/repository`` identifier used in API paths.""" + if _REPOSITORY_PATTERN.fullmatch(value) is None: + raise AuditError("repository identity is invalid") + return value + + +def _require_workflow_path(value: object) -> str: + """Validate one canonical Actions registry path before classification.""" + if not isinstance(value, str) or not value or value != value.strip(): + raise AuditError("workflow path is invalid") + if "\\" in value or "%" in value: + raise AuditError("workflow path is not canonical") + if value.startswith(DYNAMIC_PREFIX): + if any(segment in {"", ".", ".."} for segment in value.split("/")): + raise AuditError("workflow path is not canonical") + return value + if not value.startswith(WORKFLOW_PREFIX): + raise AuditError("workflow path is not canonical") + filename = value[len(WORKFLOW_PREFIX) :] + if ( + not filename + or "/" in filename + or filename in {".", ".."} + or not filename.endswith((".yml", ".yaml")) + ): + raise AuditError("workflow path is not canonical") + return value + + +def _require_workflow_id(value: object) -> int: + """Return a positive GitHub workflow ID without accepting booleans.""" + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise AuditError("workflow id is invalid") + return value + + +def _require_workflow_state(value: object) -> str: + """Return one exact documented GitHub workflow lifecycle state.""" + if type(value) is not str or value not in _WORKFLOW_STATES: + raise AuditError("workflow state is invalid") + return value + + +def _require_total_count(value: object) -> int: + """Return a non-negative registry total count.""" + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise AuditError("workflow registry total_count is invalid") + return value + + +def _request_status(response: object) -> int: + """Read an HTTP status from a normal urllib-style response.""" + status = getattr(response, "status", None) + if status is None: + getcode = getattr(response, "getcode", None) + if callable(getcode): + status = getcode() + if isinstance(status, bool) or not isinstance(status, int): + raise AuditError("GitHub API response has no valid HTTP status") + return status + + +def request_json( + url: str, + *, + token: str | None = None, + opener: Callable[..., Any] = urllib.request.urlopen, + timeout: float = DEFAULT_TIMEOUT_SECONDS, +) -> Any: + """Fetch one bounded GitHub JSON response and fail closed on API errors. + + ``token`` is used only as a bearer header for the current request and is + never copied into output or diagnostics. + """ + if not isinstance(timeout, (int, float)) or isinstance(timeout, bool): + raise AuditError("GitHub API timeout is invalid") + try: + canonical_timeout = float(timeout) + except (OverflowError, ValueError): + raise AuditError("GitHub API timeout is invalid") from None + if not math.isfinite(canonical_timeout) or canonical_timeout <= 0: + raise AuditError("GitHub API timeout is invalid") + headers = { + "Accept": "application/vnd.github+json", + "User-Agent": "EgressWeave-workflow-registry-audit", + "X-GitHub-Api-Version": "2022-11-28", + } + if token: + headers["Authorization"] = f"Bearer {token}" + request = urllib.request.Request(url, headers=headers, method="GET") + failure_message: str | None = None + try: + with opener(request, timeout=canonical_timeout) as response: + status = _request_status(response) + if status != 200: + raise AuditError(f"GitHub API returned HTTP {status}") + payload = response.read(MAX_RESPONSE_BYTES + 1) + except AuditError: + raise + except urllib.error.HTTPError as exc: + failure_message = f"GitHub API returned HTTP {exc.code}" + except ( + http.client.IncompleteRead, + urllib.error.URLError, + TimeoutError, + OSError, + ): + failure_message = "GitHub API request failed" + if failure_message is not None: + raise AuditError(failure_message) from None + if len(payload) > MAX_RESPONSE_BYTES: + raise AuditError("GitHub API response exceeds the audit safety bound") + try: + return json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + raise AuditError("GitHub API returned malformed JSON") from None + + +def collect_registry_pages( + fetch_page: Callable[[int], Mapping[str, object]], + *, + per_page: int = DEFAULT_PER_PAGE, + max_pages: int = DEFAULT_MAX_PAGES, +) -> list[dict[str, object]]: + """Collect the complete Actions workflow registry within explicit bounds.""" + if per_page <= 0 or per_page > 100 or max_pages <= 0: + raise AuditError("workflow registry pagination bounds are invalid") + pages: list[dict[str, object]] = [] + expected_total: int | None = None + collected = 0 + for page_number in range(1, max_pages + 1): + raw_page = fetch_page(page_number) + if not isinstance(raw_page, Mapping): + raise AuditError("workflow registry page is malformed") + total_count = _require_total_count(raw_page.get("total_count")) + workflows = raw_page.get("workflows") + if not isinstance(workflows, list) or len(workflows) > per_page: + raise AuditError("workflow registry page is malformed") + if expected_total is None: + expected_total = total_count + elif total_count != expected_total: + raise AuditError("workflow registry total_count changed during pagination") + pages.append( + { + "page": page_number, + "total_count": total_count, + "workflows": list(workflows), + } + ) + collected += len(workflows) + if collected == expected_total: + return pages + if collected > expected_total or not workflows: + raise AuditError("workflow registry pagination is incomplete") + raise AuditError("workflow registry pagination exceeded the safety bound") + + +def _validated_known_paths(paths: Iterable[str]) -> set[str]: + """Validate repository workflow paths supplied by tree or PR evidence.""" + validated: set[str] = set() + for path in paths: + validated.add(_require_workflow_path(path)) + return validated + + +def build_audit( + *, + registry_pages: Iterable[Mapping[str, object]], + present_paths: Iterable[str], + active_pr_paths: Iterable[str], + expected_default_sha: str, + observed_default_sha: str, + observed_at: str, +) -> dict[str, object]: + """Classify exact workflow identities against one immutable repository view. + + A repository workflow is an ``active_orphan`` only when its registry path is + active, absent from the protected tree, and not reserved by an open pull + request. GitHub-owned ``dynamic/...`` identities are deliberately separated + from repository workflow lifecycle state. + """ + expected_sha = _require_sha(expected_default_sha, label="expected default branch") + observed_sha = _require_sha(observed_default_sha, label="observed default branch") + if expected_sha != observed_sha: + raise AuditError("default branch moved during workflow registry audit") + if not isinstance(observed_at, str) or not observed_at: + raise AuditError("workflow registry observation time is invalid") + + present = _validated_known_paths(present_paths) + reserved = _validated_known_paths(active_pr_paths) + pages = list(registry_pages) + if not pages: + raise AuditError("workflow registry pagination is incomplete") + + expected_total: int | None = None + seen_ids: dict[int, tuple[str, str]] = {} + records: list[dict[str, object]] = [] + receipts: list[dict[str, int]] = [] + item_count = 0 + + for expected_page_number, raw_page in enumerate(pages, start=1): + if not isinstance(raw_page, Mapping): + raise AuditError("workflow registry page is malformed") + page_number = raw_page.get("page") + if page_number != expected_page_number: + raise AuditError("workflow registry pagination is inconsistent") + total_count = _require_total_count(raw_page.get("total_count")) + workflows = raw_page.get("workflows") + if not isinstance(workflows, list): + raise AuditError("workflow registry page is malformed") + if expected_total is None: + expected_total = total_count + elif total_count != expected_total: + raise AuditError("workflow registry total_count changed during pagination") + + receipts.append( + { + "page": expected_page_number, + "item_count": len(workflows), + "total_count": total_count, + } + ) + item_count += len(workflows) + for workflow in workflows: + if not isinstance(workflow, Mapping): + raise AuditError("workflow registry item is malformed") + workflow_id = _require_workflow_id(workflow.get("id")) + path = _require_workflow_path(workflow.get("path")) + state = _require_workflow_state(workflow.get("state")) + prior = seen_ids.get(workflow_id) + identity = (path, state) + if prior is not None: + raise AuditError(f"workflow id {workflow_id} appears more than once") + seen_ids[workflow_id] = identity + + if path.startswith(DYNAMIC_PREFIX): + classification = "github_dynamic_workflow" + elif path in present: + classification = "present_repository_workflow" + elif state == "active" and path in reserved: + classification = "active_pr_reserved" + elif state == "active": + classification = "active_orphan" + else: + classification = "disabled_absent" + records.append( + { + "workflow_id": workflow_id, + "path": path, + "state": state, + "classification": classification, + "default_branch_sha": observed_sha, + "observed_at": observed_at, + "registry_page": expected_page_number, + } + ) + + if expected_total is None or item_count != expected_total: + raise AuditError("workflow registry pagination is incomplete") + records.sort(key=lambda record: (int(record["workflow_id"]), str(record["path"]))) + return { + "format": "egressweave.workflow-registry-audit.v1", + "repository_default_sha": observed_sha, + "observed_at": observed_at, + "records": records, + "receipts": receipts, + } + + +def _workflow_registry_snapshot( + registry_pages: Iterable[Mapping[str, object]], + *, + default_sha: str, +) -> tuple[tuple[int, str, str], ...]: + """Return validated workflow ID/path/state facts, excluding presentation metadata.""" + audit = build_audit( + registry_pages=registry_pages, + present_paths=(), + active_pr_paths=(), + expected_default_sha=default_sha, + observed_default_sha=default_sha, + observed_at="registry-snapshot", + ) + records = audit.get("records") + if not isinstance(records, list): + raise AuditError("workflow registry snapshot is malformed") + snapshot: list[tuple[int, str, str]] = [] + for record in records: + if not isinstance(record, Mapping): + raise AuditError("workflow registry snapshot is malformed") + workflow_id = _require_workflow_id(record.get("workflow_id")) + path = _require_workflow_path(record.get("path")) + state = _require_workflow_state(record.get("state")) + snapshot.append((workflow_id, path, state)) + return tuple(snapshot) + + +def _api_url(repository: str, suffix: str, **query: object) -> str: + """Build one validated GitHub REST URL without embedding credentials.""" + repository = _require_repository(repository) + encoded_query = urllib.parse.urlencode(query) + base = f"{API_ROOT}/repos/{repository}/{suffix.lstrip('/')}" + return f"{base}?{encoded_query}" if encoded_query else base + + +def _collect_open_pr_workflow_snapshot( + repository: str, + token: str | None, +) -> tuple[tuple[int, str, tuple[str, ...]], ...]: + """Snapshot every open PR head and its current non-removed workflow paths.""" + reservations: list[tuple[int, str, tuple[str, ...]]] = [] + seen_numbers: set[int] = set() + for pr_page in range(1, DEFAULT_MAX_PR_PAGES + 1): + pulls = request_json( + _api_url(repository, "pulls", state="open", per_page=100, page=pr_page), + token=token, + ) + if not isinstance(pulls, list): + raise AuditError("pull request pagination is malformed") + for pull in pulls: + if not isinstance(pull, Mapping): + raise AuditError("pull request record is malformed") + number = pull.get("number") + if isinstance(number, bool) or not isinstance(number, int) or number <= 0: + raise AuditError("pull request number is invalid") + if number in seen_numbers: + raise AuditError(f"pull request {number} appears more than once") + seen_numbers.add(number) + head = pull.get("head") + if not isinstance(head, Mapping): + raise AuditError("pull request head metadata is malformed") + head_sha = _require_sha(head.get("sha"), label=f"pull request {number} head") + workflow_paths: set[str] = set() + exhausted_files = False + for file_page in range(1, DEFAULT_MAX_PR_FILE_PAGES + 1): + files = request_json( + _api_url( + repository, + f"pulls/{number}/files", + per_page=100, + page=file_page, + ), + token=token, + ) + if not isinstance(files, list): + raise AuditError("pull request file pagination is malformed") + for changed_file in files: + if not isinstance(changed_file, Mapping): + raise AuditError("pull request file record is malformed") + filename = changed_file.get("filename") + status = changed_file.get("status") + if ( + isinstance(filename, str) + and status != "removed" + and filename.startswith(WORKFLOW_PREFIX) + ): + workflow_paths.add(_require_workflow_path(filename)) + if len(files) < 100: + exhausted_files = True + break + if not exhausted_files: + raise AuditError("pull request file pagination exceeded the safety bound") + reservations.append((number, head_sha, tuple(sorted(workflow_paths)))) + if len(pulls) < 100: + return tuple(sorted(reservations)) + raise AuditError("pull request pagination exceeded the safety bound") + + +def _workflow_paths_from_pr_snapshot( + snapshot: Iterable[tuple[int, str, tuple[str, ...]]], +) -> set[str]: + """Return workflow paths reserved by an exact validated open-PR snapshot.""" + return {path for _number, _head_sha, paths in snapshot for path in paths} + + +def audit_repository( + repository: str, + expected_default_sha: str, + *, + token: str | None = None, + observed_at: str | None = None, +) -> dict[str, object]: + """Build one live read-only audit bound to an expected protected revision.""" + repository = _require_repository(repository) + expected_sha = _require_sha(expected_default_sha, label="expected default branch") + repository_data = request_json(_api_url(repository, ""), token=token) + if not isinstance(repository_data, Mapping): + raise AuditError("repository metadata is malformed") + default_branch = repository_data.get("default_branch") + if not isinstance(default_branch, str) or not default_branch: + raise AuditError("repository default branch is invalid") + encoded_branch = urllib.parse.quote(default_branch, safe="") + initial_branch = request_json( + _api_url(repository, f"branches/{encoded_branch}"), + token=token, + ) + if not isinstance(initial_branch, Mapping): + raise AuditError("default branch metadata is malformed") + initial_commit = initial_branch.get("commit") + if not isinstance(initial_commit, Mapping): + raise AuditError("default branch commit metadata is malformed") + initial_sha = _require_sha(initial_commit.get("sha"), label="observed default branch") + if initial_sha != expected_sha: + raise AuditError("default branch moved before workflow registry audit") + + contents = request_json( + _api_url(repository, "contents/.github/workflows", ref=expected_sha), + token=token, + ) + if not isinstance(contents, list): + raise AuditError("workflow source listing is malformed") + present_paths: set[str] = set() + for item in contents: + if not isinstance(item, Mapping): + raise AuditError("workflow source entry is malformed") + path = item.get("path") + item_type = item.get("type") + if item_type == "file": + present_paths.add(_require_workflow_path(path)) + + initial_registry_pages = collect_registry_pages( + lambda page: request_json( + _api_url(repository, "actions/workflows", per_page=100, page=page), + token=token, + ) + ) + initial_registry_snapshot = _workflow_registry_snapshot( + initial_registry_pages, + default_sha=expected_sha, + ) + initial_pr_snapshot = _collect_open_pr_workflow_snapshot(repository, token) + active_pr_paths = _workflow_paths_from_pr_snapshot(initial_pr_snapshot) + final_pr_snapshot = _collect_open_pr_workflow_snapshot(repository, token) + if final_pr_snapshot != initial_pr_snapshot: + raise AuditError("pull request workflow reservations changed during audit") + final_registry_pages = collect_registry_pages( + lambda page: request_json( + _api_url(repository, "actions/workflows", per_page=100, page=page), + token=token, + ) + ) + final_registry_snapshot = _workflow_registry_snapshot( + final_registry_pages, + default_sha=expected_sha, + ) + if final_registry_snapshot != initial_registry_snapshot: + raise AuditError("workflow registry changed during audit") + + final_branch = request_json( + _api_url(repository, f"branches/{encoded_branch}"), + token=token, + ) + if not isinstance(final_branch, Mapping): + raise AuditError("default branch metadata is malformed") + final_commit = final_branch.get("commit") + if not isinstance(final_commit, Mapping): + raise AuditError("default branch commit metadata is malformed") + final_sha = _require_sha(final_commit.get("sha"), label="observed default branch") + timestamp = observed_at or datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace( + "+00:00", "Z" + ) + return build_audit( + registry_pages=final_registry_pages, + present_paths=present_paths, + active_pr_paths=active_pr_paths, + expected_default_sha=expected_sha, + observed_default_sha=final_sha, + observed_at=timestamp, + ) + + +def _parse_args(argv: list[str] | None) -> argparse.Namespace: + """Parse the bounded operator interface for the read-only detector.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repository", required=True, help="GitHub repository as owner/name") + parser.add_argument( + "--expected-default-sha", + required=True, + help="Exact protected default-branch SHA to which the audit must remain bound", + ) + parser.add_argument( + "--token-env", + default="GITHUB_TOKEN", + help="Environment variable containing an optional read-only GitHub token", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + """Emit deterministic JSON and return nonzero while active orphans remain.""" + args = _parse_args(argv) + token = os.environ.get(args.token_env) if args.token_env else None + try: + audit = audit_repository( + args.repository, + args.expected_default_sha, + token=token, + ) + except AuditError as exc: + print(f"workflow registry audit failed: {exc}", file=sys.stderr) + return 3 + print(json.dumps(audit, sort_keys=True, separators=(",", ":"))) + records = audit.get("records") + if isinstance(records, list) and any( + isinstance(record, Mapping) and record.get("classification") == "active_orphan" + for record in records + ): + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py new file mode 100644 index 0000000..658ed7d --- /dev/null +++ b/tests/test_workflow_registry_audit.py @@ -0,0 +1,328 @@ +"""Tests for the read-only GitHub Actions workflow-registry audit.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +AUDITOR_PATH = REPOSITORY_ROOT / "scripts" / "ci" / "audit_workflow_registry.py" +SOURCE_SHA = "0123456789abcdef0123456789abcdef01234567" +OBSERVED_AT = "2026-08-12T12:00:00Z" + + +def _load_auditor(): + """Load the repository-only audit script from its exact path.""" + specification = importlib.util.spec_from_file_location( + "egressweave_audit_workflow_registry", + AUDITOR_PATH, + ) + assert specification is not None and specification.loader is not None + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +def _workflow(workflow_id: int, path: str, state: str = "active") -> dict[str, object]: + """Build one minimal workflow-registry fixture.""" + return {"id": workflow_id, "path": path, "state": state, "name": path} + + +def _page(*workflows: dict[str, object], total_count: int | None = None) -> dict[str, object]: + """Build one paginated Actions registry response fixture.""" + return { + "page": 1, + "total_count": len(workflows) if total_count is None else total_count, + "workflows": list(workflows), + } + + +def test_audit_distinguishes_present_orphan_disabled_and_dynamic_workflows() -> None: + """Never infer lifecycle state from a workflow display name alone.""" + auditor = _load_auditor() + result = auditor.build_audit( + registry_pages=[ + _page( + _workflow(1, ".github/workflows/ci.yml"), + _workflow(2, ".github/workflows/old-one-shot.yml"), + _workflow(3, ".github/workflows/removed.yml", "disabled_manually"), + _workflow(4, "dynamic/dependabot/update-graph"), + ) + ], + present_paths={".github/workflows/ci.yml"}, + active_pr_paths=set(), + expected_default_sha=SOURCE_SHA, + observed_default_sha=SOURCE_SHA, + observed_at=OBSERVED_AT, + ) + + by_id = {record["workflow_id"]: record for record in result["records"]} + assert by_id[1]["classification"] == "present_repository_workflow" + assert by_id[2]["classification"] == "active_orphan" + assert by_id[3]["classification"] == "disabled_absent" + assert by_id[4]["classification"] == "github_dynamic_workflow" + assert by_id[2] == { + "workflow_id": 2, + "path": ".github/workflows/old-one-shot.yml", + "state": "active", + "classification": "active_orphan", + "default_branch_sha": SOURCE_SHA, + "observed_at": OBSERVED_AT, + "registry_page": 1, + } + assert result["receipts"] == [{"page": 1, "item_count": 4, "total_count": 4}] + + +def test_audit_reserves_absent_workflow_owned_by_an_active_pr() -> None: + """Do not disable a bounded workflow whose source is still owned by an open PR.""" + auditor = _load_auditor() + result = auditor.build_audit( + registry_pages=[_page(_workflow(17, ".github/workflows/bounded-pr.yml"))], + present_paths=set(), + active_pr_paths={".github/workflows/bounded-pr.yml"}, + expected_default_sha=SOURCE_SHA, + observed_default_sha=SOURCE_SHA, + observed_at=OBSERVED_AT, + ) + assert result["records"][0]["classification"] == "active_pr_reserved" + + +def test_audit_fails_closed_when_default_branch_moves() -> None: + """Reject a registry/tree comparison assembled across different protected heads.""" + auditor = _load_auditor() + with pytest.raises(auditor.AuditError, match="default branch moved"): + auditor.build_audit( + registry_pages=[_page(_workflow(1, ".github/workflows/ci.yml"))], + present_paths={".github/workflows/ci.yml"}, + active_pr_paths=set(), + expected_default_sha=SOURCE_SHA, + observed_default_sha="f" * 40, + observed_at=OBSERVED_AT, + ) + + +def test_audit_rejects_reused_workflow_id_with_conflicting_path() -> None: + """Treat workflow-ID reuse or inconsistent pagination as integrity failure.""" + auditor = _load_auditor() + pages = [ + { + "page": 1, + "total_count": 2, + "workflows": [_workflow(9, ".github/workflows/a.yml")], + }, + { + "page": 2, + "total_count": 2, + "workflows": [_workflow(9, ".github/workflows/b.yml")], + }, + ] + with pytest.raises(auditor.AuditError, match="workflow id 9"): + auditor.build_audit( + registry_pages=pages, + present_paths=set(), + active_pr_paths=set(), + expected_default_sha=SOURCE_SHA, + observed_default_sha=SOURCE_SHA, + observed_at=OBSERVED_AT, + ) + + +@pytest.mark.parametrize( + "path", + [ + ".Github/workflows/old.yml", + ".github/workflows/%6fld.yml", + ".github\\workflows\\old.yml", + ".github/workflows/../old.yml", + ], +) +def test_audit_rejects_noncanonical_repository_workflow_paths(path: str) -> None: + """Reject ambiguous case, encoding, separators, and traversal before classification.""" + auditor = _load_auditor() + with pytest.raises(auditor.AuditError, match="workflow path"): + auditor.build_audit( + registry_pages=[_page(_workflow(1, path))], + present_paths=set(), + active_pr_paths=set(), + expected_default_sha=SOURCE_SHA, + observed_default_sha=SOURCE_SHA, + observed_at=OBSERVED_AT, + ) + + +def test_audit_rejects_truncated_registry_pagination() -> None: + """Require pagination receipts to account for the complete Actions registry.""" + auditor = _load_auditor() + with pytest.raises(auditor.AuditError, match="registry pagination"): + auditor.build_audit( + registry_pages=[_page(_workflow(1, ".github/workflows/ci.yml"), total_count=2)], + present_paths={".github/workflows/ci.yml"}, + active_pr_paths=set(), + expected_default_sha=SOURCE_SHA, + observed_default_sha=SOURCE_SHA, + observed_at=OBSERVED_AT, + ) + + +@pytest.mark.parametrize("status", [403, 404, 500]) +def test_request_json_fails_closed_on_permission_or_transport_http_errors(status: int) -> None: + """Do not turn API permission, disappearance, or server failure into a clean audit.""" + auditor = _load_auditor() + + class Response: + """Minimal context-managed HTTP response fixture.""" + + def __init__(self) -> None: + self.status = status + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + def read(self, limit: int) -> bytes: + return b'{"message":"unavailable"}' + + with pytest.raises(auditor.AuditError, match=f"HTTP {status}"): + auditor.request_json("https://api.github.invalid/example", opener=lambda request, timeout: Response()) + + +@pytest.mark.parametrize("timeout", [float("nan"), float("inf")]) +def test_request_json_rejects_non_finite_timeouts(timeout: float) -> None: + """Keep control-plane reads bounded by rejecting non-finite socket timeouts.""" + auditor = _load_auditor() + + def unexpected_open(*args, **kwargs): + pytest.fail("non-finite timeout reached the network opener") + + with pytest.raises(auditor.AuditError, match="timeout is invalid"): + auditor.request_json( + "https://api.github.invalid/example", + opener=unexpected_open, + timeout=timeout, + ) + + +def test_collect_registry_pages_is_bounded_and_records_exact_page_numbers() -> None: + """Collect every advertised workflow without silently truncating the registry.""" + auditor = _load_auditor() + responses = { + 1: {"total_count": 3, "workflows": [_workflow(1, ".github/workflows/a.yml"), _workflow(2, ".github/workflows/b.yml")]}, + 2: {"total_count": 3, "workflows": [_workflow(3, ".github/workflows/c.yml")]}, + } + + def fetch_page(page: int) -> dict[str, object]: + return responses[page] + + pages = auditor.collect_registry_pages(fetch_page, per_page=2, max_pages=2) + assert [page["page"] for page in pages] == [1, 2] + assert sum(len(page["workflows"]) for page in pages) == 3 + + +def test_collect_open_pr_workflow_snapshot_binds_paths_to_exact_heads(monkeypatch) -> None: + """Include every open PR head so a same-path head replacement is observable.""" + auditor = _load_auditor() + + def fake_request_json(url: str, *, token: str | None = None): + del token + if "/pulls?" in url: + return [ + {"number": 41, "head": {"sha": "a" * 40}}, + {"number": 42, "head": {"sha": "b" * 40}}, + ] + if "/pulls/41/files?" in url: + return [ + {"filename": ".github/workflows/new.yml", "status": "added"}, + {"filename": "README.md", "status": "modified"}, + ] + if "/pulls/42/files?" in url: + return [{"filename": ".github/workflows/removed.yml", "status": "removed"}] + raise AssertionError(f"unexpected GitHub API URL: {url}") + + monkeypatch.setattr(auditor, "request_json", fake_request_json) + assert auditor._collect_open_pr_workflow_snapshot("ContextualWisdomLab/EgressWeave", None) == ( + (41, "a" * 40, (".github/workflows/new.yml",)), + (42, "b" * 40, ()), + ) + + +def test_audit_repository_fails_closed_when_open_pr_snapshot_changes(monkeypatch) -> None: + """Never emit active-PR reservations assembled across different PR heads.""" + auditor = _load_auditor() + responses = iter( + [ + {"default_branch": "main"}, + {"commit": {"sha": SOURCE_SHA}}, + [], + {"commit": {"sha": SOURCE_SHA}}, + ] + ) + + def fake_request_json(url: str, *, token: str | None = None): + del url, token + return next(responses) + + monkeypatch.setattr(auditor, "request_json", fake_request_json) + monkeypatch.setattr(auditor, "collect_registry_pages", lambda fetch_page: [_page()]) + snapshots = iter( + [ + ((41, "a" * 40, (".github/workflows/new.yml",)),), + ((41, "b" * 40, (".github/workflows/new.yml",)),), + ] + ) + monkeypatch.setattr( + auditor, + "_collect_open_pr_workflow_snapshot", + lambda repository, token: next(snapshots), + raising=False, + ) + monkeypatch.setattr( + auditor, + "_collect_open_pr_workflow_paths", + lambda repository, token: {".github/workflows/new.yml"}, + raising=False, + ) + + with pytest.raises(auditor.AuditError, match="pull request workflow reservations changed"): + auditor.audit_repository( + "ContextualWisdomLab/EgressWeave", + SOURCE_SHA, + observed_at=OBSERVED_AT, + ) + + +def test_audit_repository_fails_closed_when_workflow_registry_changes(monkeypatch) -> None: + """Never classify workflow lifecycle state from a registry that changes mid-audit.""" + auditor = _load_auditor() + responses = iter( + [ + {"default_branch": "main"}, + {"commit": {"sha": SOURCE_SHA}}, + [], + {"commit": {"sha": SOURCE_SHA}}, + ] + ) + + def fake_request_json(url: str, *, token: str | None = None): + del url, token + return next(responses) + + registry_pages = iter( + [ + [_page(_workflow(17, ".github/workflows/old.yml", "active"))], + [_page(_workflow(17, ".github/workflows/old.yml", "disabled_manually"))], + ] + ) + monkeypatch.setattr(auditor, "request_json", fake_request_json) + monkeypatch.setattr(auditor, "collect_registry_pages", lambda fetch_page: next(registry_pages)) + monkeypatch.setattr(auditor, "_collect_open_pr_workflow_snapshot", lambda repository, token: ()) + + with pytest.raises(auditor.AuditError, match="workflow registry changed"): + auditor.audit_repository( + "ContextualWisdomLab/EgressWeave", + SOURCE_SHA, + observed_at=OBSERVED_AT, + ) diff --git a/tests/test_workflow_registry_partial_transfer.py b/tests/test_workflow_registry_partial_transfer.py new file mode 100644 index 0000000..3851eb9 --- /dev/null +++ b/tests/test_workflow_registry_partial_transfer.py @@ -0,0 +1,53 @@ +"""Fail-closed regression for partial GitHub API response transfers.""" + +from __future__ import annotations + +import http.client +import importlib.util +from pathlib import Path + +import pytest + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +AUDITOR_PATH = REPOSITORY_ROOT / "scripts" / "ci" / "audit_workflow_registry.py" + + +def _load_auditor(): + """Load the repository-only workflow-registry auditor from its exact path.""" + specification = importlib.util.spec_from_file_location( + "egressweave_partial_transfer_audit", + AUDITOR_PATH, + ) + assert specification is not None and specification.loader is not None + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +def test_request_json_masks_partial_transfer_failure() -> None: + """A truncated HTTP body must become one generic non-leaking audit failure.""" + auditor = _load_auditor() + + class Response: + """Context-managed response that aborts after returning partial bytes.""" + + status = 200 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + def read(self, limit: int) -> bytes: + del limit + raise http.client.IncompleteRead(b'{"partial":', 100) + + with pytest.raises(auditor.AuditError, match="GitHub API request failed") as captured: + auditor.request_json( + "https://api.github.invalid/example", + opener=lambda request, timeout: Response(), + ) + + assert captured.value.__cause__ is None + assert captured.value.__context__ is None diff --git a/tests/test_workflow_registry_state_integrity.py b/tests/test_workflow_registry_state_integrity.py new file mode 100644 index 0000000..6721b54 --- /dev/null +++ b/tests/test_workflow_registry_state_integrity.py @@ -0,0 +1,50 @@ +"""Fail-closed workflow-state regression for the registry auditor.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +AUDITOR_PATH = REPOSITORY_ROOT / "scripts" / "ci" / "audit_workflow_registry.py" +SOURCE_SHA = "0123456789abcdef0123456789abcdef01234567" + + +def _load_auditor(): + """Load the repository-only workflow-registry auditor from its exact path.""" + specification = importlib.util.spec_from_file_location( + "egressweave_workflow_state_audit", + AUDITOR_PATH, + ) + assert specification is not None and specification.loader is not None + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +def test_audit_rejects_unknown_workflow_state() -> None: + """Do not silently classify an unknown lifecycle state as safely disabled.""" + auditor = _load_auditor() + page = { + "page": 1, + "total_count": 1, + "workflows": [ + { + "id": 7, + "path": ".github/workflows/removed.yml", + "state": "provider_future_state", + } + ], + } + + with pytest.raises(auditor.AuditError, match="workflow state is invalid"): + auditor.build_audit( + registry_pages=[page], + present_paths=set(), + active_pr_paths=set(), + expected_default_sha=SOURCE_SHA, + observed_default_sha=SOURCE_SHA, + observed_at="2026-08-13T03:30:00Z", + )