From e5edf9fdb016a421bb905fcd785a6f5295c8f5ee Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 21:22:01 +0900 Subject: [PATCH 01/15] test(ci): reproduce orphaned workflow registry drift --- tests/test_workflow_registry_audit.py | 205 ++++++++++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 tests/test_workflow_registry_audit.py diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py new file mode 100644 index 0000000..4714596 --- /dev/null +++ b/tests/test_workflow_registry_audit.py @@ -0,0 +1,205 @@ +"""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.""" + + 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()) + + +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 From 4db904e4d963690b085a7e7c3236e465d782bc0f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 21:27:56 +0900 Subject: [PATCH 02/15] fix(ci): add bounded read-only workflow registry audit --- scripts/ci/audit_workflow_registry.py | 476 ++++++++++++++++++++++++++ 1 file changed, 476 insertions(+) create mode 100644 scripts/ci/audit_workflow_registry.py diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py new file mode 100644 index 0000000..fe5050f --- /dev/null +++ b/scripts/ci/audit_workflow_registry.py @@ -0,0 +1,476 @@ +"""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 json +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}$") + + +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_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) or 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") + try: + with opener(request, timeout=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: + raise AuditError(f"GitHub API returned HTTP {exc.code}") from None + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise AuditError("GitHub API request failed") 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 = workflow.get("state") + if not isinstance(state, str) or not state: + raise AuditError("workflow state is invalid") + 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 _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_paths(repository: str, token: str | None) -> set[str]: + """Collect current workflow paths still owned by bounded open pull requests.""" + reserved: set[str] = 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") + 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) + ): + reserved.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") + if len(pulls) < 100: + return reserved + raise AuditError("pull request pagination exceeded the safety bound") + + +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)) + + registry_pages = collect_registry_pages( + lambda page: request_json( + _api_url(repository, "actions/workflows", per_page=100, page=page), + token=token, + ) + ) + active_pr_paths = _collect_open_pr_workflow_paths(repository, token) + + 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=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()) From f16c96a0751b3dc0ca4c458819fa28e6c251cab2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 23:21:57 +0900 Subject: [PATCH 03/15] fix(ci): remove unused request exception binding --- scripts/ci/audit_workflow_registry.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index fe5050f..be0c868 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -135,7 +135,7 @@ def request_json( raise except urllib.error.HTTPError as exc: raise AuditError(f"GitHub API returned HTTP {exc.code}") from None - except (urllib.error.URLError, TimeoutError, OSError) as exc: + except (urllib.error.URLError, TimeoutError, OSError): raise AuditError("GitHub API request failed") from None if len(payload) > MAX_RESPONSE_BYTES: raise AuditError("GitHub API response exceeds the audit safety bound") From de3f8427378035fb91539428f6d92c30ccfb7b52 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 02:47:18 +0900 Subject: [PATCH 04/15] test(ci): fix HTTP response fixture status binding --- tests/test_workflow_registry_audit.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 4714596..be44bd5 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -174,7 +174,8 @@ def test_request_json_fails_closed_on_permission_or_transport_http_errors(status class Response: """Minimal context-managed HTTP response fixture.""" - status = status + def __init__(self) -> None: + self.status = status def __enter__(self): return self From fb105144f130047d233bfcaa2a8a4ebcf176732f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 04:33:00 +0900 Subject: [PATCH 05/15] test(ci): reject non-finite audit timeouts --- tests/test_workflow_registry_audit.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index be44bd5..3299f77 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -190,6 +190,22 @@ def read(self, limit: int) -> bytes: 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() From 3d90f914fc75e3737d783209df709849038a1c06 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 04:36:03 +0900 Subject: [PATCH 06/15] fix(ci): canonicalize finite audit timeout --- scripts/ci/audit_workflow_registry.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index be0c868..0a608bc 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -13,6 +13,7 @@ import argparse import json +import math import os import re import sys @@ -115,7 +116,13 @@ def request_json( ``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) or timeout <= 0: + 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", @@ -126,7 +133,7 @@ def request_json( headers["Authorization"] = f"Bearer {token}" request = urllib.request.Request(url, headers=headers, method="GET") try: - with opener(request, timeout=timeout) as response: + with opener(request, timeout=canonical_timeout) as response: status = _request_status(response) if status != 200: raise AuditError(f"GitHub API returned HTTP {status}") @@ -473,4 +480,4 @@ def main(argv: list[str] | None = None) -> int: if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file From e98d0c27c94ee22b8d45a849d0a80eb38c8e38db Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 04:40:37 +0900 Subject: [PATCH 07/15] test(ci): bind workflow reservations to PR heads --- tests/test_workflow_registry_audit.py | 72 +++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 3299f77..979b24b 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -220,3 +220,75 @@ def fetch_page(page: int) -> dict[str, object]: 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, + ) From 1db137df638b0af7a506704bc7cf4badbcb92cff Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 04:43:15 +0900 Subject: [PATCH 08/15] fix(ci): bind workflow reservations to exact PR heads --- scripts/ci/audit_workflow_registry.py | 46 ++++++++++++++++++++------- 1 file changed, 35 insertions(+), 11 deletions(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index 0a608bc..c45d904 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -1,11 +1,11 @@ """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 +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 +The script is intentionally read-only. It never disables a workflow, updates a branch, comments on a pull request, or requests additional credentials. """ @@ -212,7 +212,7 @@ def build_audit( 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 + request. GitHub-owned ``dynamic/...`` identities are deliberately separated from repository workflow lifecycle state. """ expected_sha = _require_sha(expected_default_sha, label="expected default branch") @@ -313,9 +313,13 @@ def _api_url(repository: str, suffix: str, **query: object) -> str: return f"{base}?{encoded_query}" if encoded_query else base -def _collect_open_pr_workflow_paths(repository: str, token: str | None) -> set[str]: - """Collect current workflow paths still owned by bounded open pull requests.""" - reserved: set[str] = set() +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), @@ -329,6 +333,14 @@ def _collect_open_pr_workflow_paths(repository: str, token: str | None) -> set[s 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( @@ -352,17 +364,25 @@ def _collect_open_pr_workflow_paths(repository: str, token: str | None) -> set[s and status != "removed" and filename.startswith(WORKFLOW_PREFIX) ): - reserved.add(_require_workflow_path(filename)) + 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 reserved + 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, @@ -414,7 +434,11 @@ def audit_repository( token=token, ) ) - active_pr_paths = _collect_open_pr_workflow_paths(repository, token) + 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_branch = request_json( _api_url(repository, f"branches/{encoded_branch}"), @@ -480,4 +504,4 @@ def main(argv: list[str] | None = None) -> int: if __name__ == "__main__": - raise SystemExit(main()) \ No newline at end of file + raise SystemExit(main()) From ef85cd004deffdf21c91872db4d2f6e559db1854 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 04:48:00 +0900 Subject: [PATCH 09/15] test(ci): reject changing workflow registry snapshots --- tests/test_workflow_registry_audit.py | 34 +++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 979b24b..658ed7d 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -292,3 +292,37 @@ def fake_request_json(url: str, *, token: str | None = None): 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, + ) From eb04a7463de5998d244600255488f5c0e9e70102 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 04:50:23 +0900 Subject: [PATCH 10/15] fix(ci): bind audit to a stable workflow registry --- scripts/ci/audit_workflow_registry.py | 50 +++++++++++++++++++++++++-- 1 file changed, 48 insertions(+), 2 deletions(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index c45d904..1baa61e 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -305,6 +305,36 @@ def build_audit( } +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 = record.get("state") + if not isinstance(state, str) or not state: + raise AuditError("workflow registry snapshot is malformed") + 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) @@ -428,17 +458,33 @@ def audit_repository( if item_type == "file": present_paths.add(_require_workflow_path(path)) - registry_pages = collect_registry_pages( + 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}"), @@ -454,7 +500,7 @@ def audit_repository( "+00:00", "Z" ) return build_audit( - registry_pages=registry_pages, + registry_pages=final_registry_pages, present_paths=present_paths, active_pr_paths=active_pr_paths, expected_default_sha=expected_sha, From 06f5fbf3a96f77bae9ba2e8bd674813773a761e9 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 12:03:20 +0900 Subject: [PATCH 11/15] test(ci): reproduce partial workflow registry transfer --- ...test_workflow_registry_partial_transfer.py | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 tests/test_workflow_registry_partial_transfer.py 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 From d936687d94ec26911854ceac8cb87aba1e173909 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 12:09:56 +0900 Subject: [PATCH 12/15] fix(ci): fail closed on partial workflow registry transfer --- scripts/ci/audit_workflow_registry.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index 1baa61e..e4f8414 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -12,6 +12,7 @@ from __future__ import annotations import argparse +import http.client import json import math import os @@ -142,7 +143,12 @@ def request_json( raise except urllib.error.HTTPError as exc: raise AuditError(f"GitHub API returned HTTP {exc.code}") from None - except (urllib.error.URLError, TimeoutError, OSError): + except ( + http.client.IncompleteRead, + urllib.error.URLError, + TimeoutError, + OSError, + ): raise AuditError("GitHub API request failed") from None if len(payload) > MAX_RESPONSE_BYTES: raise AuditError("GitHub API response exceeds the audit safety bound") @@ -550,4 +556,4 @@ def main(argv: list[str] | None = None) -> int: if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file From 7d3edee15b184c308b7fa77670a239218989c57e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 12:13:57 +0900 Subject: [PATCH 13/15] fix(ci): erase partial-transfer exception provenance --- scripts/ci/audit_workflow_registry.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index e4f8414..382ab08 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -133,6 +133,7 @@ def request_json( 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) @@ -142,14 +143,16 @@ def request_json( except AuditError: raise except urllib.error.HTTPError as exc: - raise AuditError(f"GitHub API returned HTTP {exc.code}") from None + failure_message = f"GitHub API returned HTTP {exc.code}" except ( http.client.IncompleteRead, urllib.error.URLError, TimeoutError, OSError, ): - raise AuditError("GitHub API request failed") from None + 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: From ee0a630d7f94f8f61d7a7c56e6e4b95799de3ff4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 12:22:06 +0900 Subject: [PATCH 14/15] test(ci): reject unknown workflow lifecycle states --- .../test_workflow_registry_state_integrity.py | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) create mode 100644 tests/test_workflow_registry_state_integrity.py 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", + ) From 9ac4bc2de5bdba2e92be184e46a42ef55627bd77 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 12:26:43 +0900 Subject: [PATCH 15/15] fix(ci): reject unknown workflow lifecycle states --- scripts/ci/audit_workflow_registry.py | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/scripts/ci/audit_workflow_registry.py b/scripts/ci/audit_workflow_registry.py index 382ab08..18fe684 100644 --- a/scripts/ci/audit_workflow_registry.py +++ b/scripts/ci/audit_workflow_registry.py @@ -36,6 +36,15 @@ 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): @@ -86,6 +95,13 @@ def _require_workflow_id(value: object) -> int: 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: @@ -271,9 +287,7 @@ def build_audit( raise AuditError("workflow registry item is malformed") workflow_id = _require_workflow_id(workflow.get("id")) path = _require_workflow_path(workflow.get("path")) - state = workflow.get("state") - if not isinstance(state, str) or not state: - raise AuditError("workflow state is invalid") + state = _require_workflow_state(workflow.get("state")) prior = seen_ids.get(workflow_id) identity = (path, state) if prior is not None: @@ -337,9 +351,7 @@ def _workflow_registry_snapshot( raise AuditError("workflow registry snapshot is malformed") workflow_id = _require_workflow_id(record.get("workflow_id")) path = _require_workflow_path(record.get("path")) - state = record.get("state") - if not isinstance(state, str) or not state: - raise AuditError("workflow registry snapshot is malformed") + state = _require_workflow_state(record.get("state")) snapshot.append((workflow_id, path, state)) return tuple(snapshot)