From e7c776a258d13f73ebcf28ce35b4542c912be9c7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:03:42 +0900 Subject: [PATCH 01/47] test: define fail-closed workflow registry audit contract --- tests/test_workflow_registry_audit.py | 310 ++++++++++++++++++++++++++ 1 file changed, 310 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 000000000..4fb6be0a5 --- /dev/null +++ b/tests/test_workflow_registry_audit.py @@ -0,0 +1,310 @@ +"""Regression tests for the read-only workflow registry audit tool.""" + +from __future__ import annotations + +import io +import json +import traceback +from dataclasses import dataclass +from urllib.error import HTTPError + +import pytest + +from workflow_registry_audit import ( + GitHubReadClient, + WorkflowRegistryAuditError, + audit_repository_workflows, +) + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" +CAPTURED_AT = "2026-08-15T00:00:00Z" + + +@dataclass +class _JsonRoute: + """Describe one expected fake GitHub JSON response.""" + + suffix: str + payload: dict[str, object] + + +class _FakeClient: + """Serve deterministic GitHub API payloads while recording exact reads.""" + + def __init__(self, routes: list[_JsonRoute]) -> None: + self._routes = list(routes) + self.requested_paths: list[str] = [] + + def get_json(self, path: str) -> dict[str, object]: + """Return the next payload only when the exact expected path matches.""" + self.requested_paths.append(path) + if not self._routes: + raise AssertionError(f"unexpected GitHub read: {path}") + route = self._routes.pop(0) + assert path.endswith(route.suffix) + return route.payload + + +def _tree_payload(*paths: str, truncated: bool = False) -> dict[str, object]: + return { + "sha": PROTECTED_SHA, + "truncated": truncated, + "tree": [ + {"path": path, "type": "blob", "sha": f"blob-{index}"} + for index, path in enumerate(paths, start=1) + ], + } + + +def _workflow(workflow_id: int, path: str, state: str = "active") -> dict[str, object]: + return { + "id": workflow_id, + "path": path, + "state": state, + "name": f"workflow-{workflow_id}", + } + + +def test_exact_path_presence_drives_classification_not_workflow_name() -> None: + """One-shot-like names stay safe when their exact protected path exists.""" + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + _tree_payload( + ".github/workflows/ci.yml", + ".github/workflows/one-shot-legitimate.yml", + ), + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + { + "total_count": 4, + "workflows": [ + _workflow(1, ".github/workflows/ci.yml"), + _workflow(2, ".github/workflows/deleted-one-shot.yml"), + _workflow(3, ".github/workflows/one-shot-legitimate.yml"), + _workflow(4, ".github/workflows/CI.yml", state="disabled_manually"), + ], + }, + ), + ] + ) + + receipt = audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) + + assert receipt["protected_sha"] == PROTECTED_SHA + assert receipt["captured_at"] == CAPTURED_AT + assert receipt["pages_scanned"] == 1 + assert receipt["registry_total_count"] == 4 + assert receipt["active_absent_workflows"] == [ + { + "workflow_id": 2, + "path": ".github/workflows/deleted-one-shot.yml", + "state": "active", + } + ] + records = {item["workflow_id"]: item for item in receipt["workflow_records"]} + assert records[1]["source_present"] is True + assert records[3]["source_present"] is True + assert records[4]["source_present"] is False + + +def test_registry_pagination_is_complete_and_receipted() -> None: + """The detector reads every page required by the first stable total count.""" + first_page = [ + _workflow(index, f".github/workflows/old-{index}.yml", state="disabled_manually") + for index in range(1, 101) + ] + second_page = [_workflow(101, ".github/workflows/ci.yml")] + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + _tree_payload(".github/workflows/ci.yml"), + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + {"total_count": 101, "workflows": first_page}, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=2", + {"total_count": 101, "workflows": second_page}, + ), + ] + ) + + receipt = audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) + + assert receipt["pages_scanned"] == 2 + assert receipt["registry_total_count"] == 101 + assert len(receipt["workflow_records"]) == 101 + assert receipt["active_absent_workflows"] == [] + + +def test_incomplete_pagination_fails_closed() -> None: + """An empty page before the advertised total is not accepted as complete.""" + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + _tree_payload(".github/workflows/ci.yml"), + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + {"total_count": 101, "workflows": [_workflow(1, ".github/workflows/ci.yml")]}, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=2", + {"total_count": 101, "workflows": []}, + ), + ] + ) + + with pytest.raises(WorkflowRegistryAuditError, match="pagination is incomplete"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) + + +def test_registry_change_during_pagination_fails_closed() -> None: + """A moving registry cannot be presented as one coherent audit receipt.""" + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + _tree_payload(".github/workflows/ci.yml"), + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + { + "total_count": 101, + "workflows": [ + _workflow(index, f".github/workflows/old-{index}.yml", "disabled_manually") + for index in range(1, 101) + ], + }, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=2", + {"total_count": 102, "workflows": [_workflow(101, ".github/workflows/ci.yml")]}, + ), + ] + ) + + with pytest.raises(WorkflowRegistryAuditError, match="changed during audit"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) + + +def test_duplicate_workflow_id_fails_closed() -> None: + """A reused ID with ambiguous path/state cannot be silently normalized.""" + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + _tree_payload(".github/workflows/ci.yml"), + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + { + "total_count": 2, + "workflows": [ + _workflow(7, ".github/workflows/ci.yml"), + _workflow(7, ".github/workflows/old.yml"), + ], + }, + ), + ] + ) + + with pytest.raises(WorkflowRegistryAuditError, match="duplicate workflow id"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) + + +def test_truncated_protected_tree_fails_closed() -> None: + """A partial protected tree is not evidence that a workflow source is absent.""" + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + _tree_payload(".github/workflows/ci.yml", truncated=True), + ) + ] + ) + + with pytest.raises(WorkflowRegistryAuditError, match="protected tree is truncated"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) + + +def test_invalid_sha_is_rejected_before_any_github_read() -> None: + """A mutable branch name cannot masquerade as exact protected-head evidence.""" + client = _FakeClient([]) + + with pytest.raises(WorkflowRegistryAuditError, match="exact 40-hex protected SHA"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha="main", + client=client, + captured_at=CAPTURED_AT, + ) + + assert client.requested_paths == [] + + +def test_http_failure_does_not_retain_lower_layer_diagnostics(monkeypatch: pytest.MonkeyPatch) -> None: + """403/404/5xx-style transport details remain outside rendered evidence.""" + secret_sentinel = "SECRET_HTTP_DIAGNOSTIC_SHOULD_NOT_ESCAPE" + + def _raise_http_error(*_args: object, **_kwargs: object) -> object: + raise HTTPError( + "https://api.github.com/private?token=secret", + 503, + secret_sentinel, + hdrs=None, + fp=io.BytesIO(json.dumps({"message": secret_sentinel}).encode()), + ) + + monkeypatch.setattr("workflow_registry_audit.urlopen", _raise_http_error) + client = GitHubReadClient(token="token-value-that-must-not-render") + + try: + client.get_json(f"/repos/{REPOSITORY}/actions/workflows?per_page=100&page=1") + except WorkflowRegistryAuditError as exc: + rendered = "".join(traceback.format_exception(exc)) + assert str(exc) == "GitHub workflow audit read failed" + assert secret_sentinel not in rendered + assert "token-value-that-must-not-render" not in rendered + assert exc.__cause__ is None + assert exc.__suppress_context__ is True + else: + raise AssertionError("expected WorkflowRegistryAuditError") From 578b101d51cd3007de761cce4c9d18292d5a5804 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:05:08 +0900 Subject: [PATCH 02/47] feat: add fail-closed workflow registry audit --- workflow_registry_audit.py | 326 +++++++++++++++++++++++++++++++++++++ 1 file changed, 326 insertions(+) create mode 100644 workflow_registry_audit.py diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py new file mode 100644 index 000000000..95727d2c8 --- /dev/null +++ b/workflow_registry_audit.py @@ -0,0 +1,326 @@ +"""Read-only GitHub Actions registry audit for protected-source drift. + +The tool intentionally performs no workflow mutation. It binds one audit receipt to +an exact protected commit, reads the repository Actions registry with complete +pagination, and reports active workflow identities whose exact source path is +absent from that protected tree. Those records are candidates for separate +operator review, not automatic disable decisions. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +from datetime import datetime, timezone +from typing import Any, Protocol +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +_REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +_SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") +_WORKFLOW_PREFIX = ".github/workflows/" +_DEFAULT_API_URL = "https://api.github.com" +_DEFAULT_TIMEOUT_SECONDS = 15.0 +_PAGE_SIZE = 100 + + +class WorkflowRegistryAuditError(RuntimeError): + """Represent a bounded fail-closed workflow audit failure.""" + + +class _JsonClient(Protocol): + """Describe the minimal read-only JSON client used by the audit.""" + + def get_json(self, path: str) -> dict[str, object]: + """Return one decoded GitHub API object for ``path``.""" + + +class GitHubReadClient: + """Perform bounded read-only GitHub API requests without diagnostic leakage.""" + + def __init__( + self, + *, + token: str | None = None, + api_url: str = _DEFAULT_API_URL, + timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS, + ) -> None: + """Configure read-only API access without retaining response diagnostics.""" + if timeout_seconds <= 0: + raise WorkflowRegistryAuditError("GitHub audit timeout must be positive") + self._token = token + self._api_url = api_url.rstrip("/") + self._timeout_seconds = timeout_seconds + + def get_json(self, path: str) -> dict[str, object]: + """Read one JSON object and replace transport/parser failures with fixed evidence.""" + request = Request( + f"{self._api_url}{path}", + headers=self._headers(), + method="GET", + ) + try: + with urlopen(request, timeout=self._timeout_seconds) as response: + raw = response.read() + payload = json.loads(raw.decode("utf-8")) + except ( + HTTPError, + URLError, + TimeoutError, + OSError, + UnicodeDecodeError, + json.JSONDecodeError, + ): + raise WorkflowRegistryAuditError("GitHub workflow audit read failed") from None + if not isinstance(payload, dict): + raise WorkflowRegistryAuditError("GitHub workflow audit response is invalid") + return payload + + def _headers(self) -> dict[str, str]: + """Build fixed GitHub headers without exposing the token in output.""" + headers = { + "Accept": "application/vnd.github+json", + "User-Agent": "pg-llm-batch-workflow-registry-audit/1", + "X-GitHub-Api-Version": "2022-11-28", + } + if self._token: + headers["Authorization"] = f"Bearer {self._token}" + return headers + + +def audit_repository_workflows( + *, + repository_full_name: str, + protected_sha: str, + client: _JsonClient, + captured_at: str | None = None, +) -> dict[str, object]: + """Return an exact-SHA read-only registry/source classification receipt. + + Active identities absent from the exact protected tree are reported as + candidates only. The caller must separately prove whether a candidate has a + legitimate platform role before any future workflow-state mutation. + """ + _validate_repository(repository_full_name) + _validate_protected_sha(protected_sha) + + protected_paths = _read_protected_workflow_paths( + repository_full_name=repository_full_name, + protected_sha=protected_sha, + client=client, + ) + workflow_records, pages_scanned, registry_total_count = _read_registry( + repository_full_name=repository_full_name, + client=client, + ) + + classified_records: list[dict[str, object]] = [] + active_absent: list[dict[str, object]] = [] + for record in workflow_records: + source_present = record["path"] in protected_paths + classified = { + "workflow_id": record["workflow_id"], + "path": record["path"], + "state": record["state"], + "source_present": source_present, + } + classified_records.append(classified) + if record["state"] == "active" and not source_present: + active_absent.append( + { + "workflow_id": record["workflow_id"], + "path": record["path"], + "state": record["state"], + } + ) + + active_absent.sort(key=lambda item: (str(item["path"]), int(item["workflow_id"]))) + return { + "repository_full_name": repository_full_name, + "protected_sha": protected_sha.lower(), + "captured_at": captured_at or _utc_timestamp(), + "pages_scanned": pages_scanned, + "registry_total_count": registry_total_count, + "protected_workflow_paths": sorted(protected_paths), + "workflow_records": classified_records, + "active_absent_workflows": active_absent, + } + + +def _read_protected_workflow_paths( + *, + repository_full_name: str, + protected_sha: str, + client: _JsonClient, +) -> set[str]: + """Read exact protected-tree workflow blob paths or fail closed.""" + payload = client.get_json( + f"/repos/{repository_full_name}/git/trees/{protected_sha}?recursive=1" + ) + if payload.get("sha") != protected_sha: + raise WorkflowRegistryAuditError("protected tree SHA does not match requested SHA") + if payload.get("truncated") is not False: + raise WorkflowRegistryAuditError("protected tree is truncated") + tree = payload.get("tree") + if not isinstance(tree, list): + raise WorkflowRegistryAuditError("protected tree response is invalid") + + paths: set[str] = set() + for entry in tree: + if not isinstance(entry, dict): + raise WorkflowRegistryAuditError("protected tree response is invalid") + path = entry.get("path") + entry_type = entry.get("type") + if not isinstance(path, str) or not isinstance(entry_type, str): + raise WorkflowRegistryAuditError("protected tree response is invalid") + if entry_type == "blob" and path.startswith(_WORKFLOW_PREFIX): + paths.add(path) + return paths + + +def _read_registry( + *, + repository_full_name: str, + client: _JsonClient, +) -> tuple[list[dict[str, object]], int, int]: + """Read the complete stable workflow registry with explicit pagination evidence.""" + expected_total: int | None = None + page = 1 + pages_scanned = 0 + records: list[dict[str, object]] = [] + seen_ids: set[int] = set() + + while expected_total is None or len(records) < expected_total: + payload = client.get_json( + f"/repos/{repository_full_name}/actions/workflows" + f"?per_page={_PAGE_SIZE}&page={page}" + ) + pages_scanned += 1 + total_count = _require_nonnegative_int(payload.get("total_count")) + if expected_total is None: + expected_total = total_count + elif total_count != expected_total: + raise WorkflowRegistryAuditError("workflow registry changed during audit") + + workflows = payload.get("workflows") + if not isinstance(workflows, list): + raise WorkflowRegistryAuditError("workflow registry response is invalid") + if not workflows and len(records) < expected_total: + raise WorkflowRegistryAuditError("workflow registry pagination is incomplete") + + for raw_record in workflows: + record = _validate_workflow_record(raw_record) + workflow_id = int(record["workflow_id"]) + if workflow_id in seen_ids: + raise WorkflowRegistryAuditError("workflow registry contains duplicate workflow id") + seen_ids.add(workflow_id) + records.append(record) + + if len(records) > expected_total: + raise WorkflowRegistryAuditError("workflow registry changed during audit") + page += 1 + + if expected_total is None: + raise WorkflowRegistryAuditError("workflow registry response is invalid") + return records, pages_scanned, expected_total + + +def _validate_workflow_record(raw_record: object) -> dict[str, object]: + """Validate only identity/path/state fields needed for safe classification.""" + if not isinstance(raw_record, dict): + raise WorkflowRegistryAuditError("workflow registry record is invalid") + workflow_id = raw_record.get("id") + path = raw_record.get("path") + state = raw_record.get("state") + if ( + isinstance(workflow_id, bool) + or not isinstance(workflow_id, int) + or workflow_id <= 0 + or not isinstance(path, str) + or not isinstance(state, str) + or not state + ): + raise WorkflowRegistryAuditError("workflow registry record is invalid") + if not path.startswith(_WORKFLOW_PREFIX) or "\\" in path or ".." in path.split("/"): + raise WorkflowRegistryAuditError("workflow registry record path is invalid") + return {"workflow_id": workflow_id, "path": path, "state": state} + + +def _require_nonnegative_int(value: object) -> int: + """Require a non-boolean, non-negative integer API count.""" + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise WorkflowRegistryAuditError("workflow registry total count is invalid") + return value + + +def _validate_repository(repository_full_name: str) -> None: + """Reject malformed repository selectors before any network read.""" + if not _REPOSITORY_RE.fullmatch(repository_full_name): + raise WorkflowRegistryAuditError("repository must use owner/name syntax") + + +def _validate_protected_sha(protected_sha: str) -> None: + """Require immutable commit identity instead of a branch/ref name.""" + if not _SHA_RE.fullmatch(protected_sha): + raise WorkflowRegistryAuditError("protected_sha must be an exact 40-hex protected SHA") + + +def _utc_timestamp() -> str: + """Return a UTC RFC 3339 timestamp for the audit receipt.""" + return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") + + +def _parser() -> argparse.ArgumentParser: + """Build the command-line interface for an explicit exact-SHA audit.""" + parser = argparse.ArgumentParser( + description=( + "Read a repository Actions registry and classify active workflow identities " + "whose exact path is absent from an exact protected commit." + ) + ) + parser.add_argument("--repository", required=True, help="GitHub owner/name repository") + parser.add_argument( + "--protected-sha", + required=True, + help="Exact 40-hex protected commit SHA; mutable ref names are rejected", + ) + parser.add_argument( + "--api-url", + default=_DEFAULT_API_URL, + help="GitHub API base URL (default: https://api.github.com)", + ) + parser.add_argument( + "--timeout-seconds", + type=float, + default=_DEFAULT_TIMEOUT_SECONDS, + help="Finite timeout for each read-only API request", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + """Run the audit, print a JSON receipt, and signal active-absent candidates.""" + args = _parser().parse_args(argv) + try: + client = GitHubReadClient( + token=os.environ.get("GITHUB_TOKEN"), + api_url=args.api_url, + timeout_seconds=args.timeout_seconds, + ) + receipt = audit_repository_workflows( + repository_full_name=args.repository, + protected_sha=args.protected_sha, + client=client, + ) + except WorkflowRegistryAuditError as exc: + print(f"workflow_registry_audit: {exc}", file=os.sys.stderr) + return 1 + + print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) + return 2 if receipt["active_absent_workflows"] else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 59b6a9403722b5755df89fc8829c0c5cf70b312f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:07:20 +0900 Subject: [PATCH 03/47] fix: keep workflow audit token on fixed GitHub API --- workflow_registry_audit.py | 23 ++++++++--------------- 1 file changed, 8 insertions(+), 15 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 95727d2c8..bfa5095f6 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -14,14 +14,14 @@ import os import re from datetime import datetime, timezone -from typing import Any, Protocol +from typing import Protocol from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen _REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") _WORKFLOW_PREFIX = ".github/workflows/" -_DEFAULT_API_URL = "https://api.github.com" +_GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 @@ -44,20 +44,18 @@ def __init__( self, *, token: str | None = None, - api_url: str = _DEFAULT_API_URL, timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS, ) -> None: - """Configure read-only API access without retaining response diagnostics.""" + """Configure fixed-origin read-only API access with a finite timeout.""" if timeout_seconds <= 0: raise WorkflowRegistryAuditError("GitHub audit timeout must be positive") self._token = token - self._api_url = api_url.rstrip("/") self._timeout_seconds = timeout_seconds def get_json(self, path: str) -> dict[str, object]: """Read one JSON object and replace transport/parser failures with fixed evidence.""" request = Request( - f"{self._api_url}{path}", + f"{_GITHUB_API_URL}{path}", headers=self._headers(), method="GET", ) @@ -105,10 +103,11 @@ def audit_repository_workflows( """ _validate_repository(repository_full_name) _validate_protected_sha(protected_sha) + normalized_sha = protected_sha.lower() protected_paths = _read_protected_workflow_paths( repository_full_name=repository_full_name, - protected_sha=protected_sha, + protected_sha=normalized_sha, client=client, ) workflow_records, pages_scanned, registry_total_count = _read_registry( @@ -139,7 +138,7 @@ def audit_repository_workflows( active_absent.sort(key=lambda item: (str(item["path"]), int(item["workflow_id"]))) return { "repository_full_name": repository_full_name, - "protected_sha": protected_sha.lower(), + "protected_sha": normalized_sha, "captured_at": captured_at or _utc_timestamp(), "pages_scanned": pages_scanned, "registry_total_count": registry_total_count, @@ -286,16 +285,11 @@ def _parser() -> argparse.ArgumentParser: required=True, help="Exact 40-hex protected commit SHA; mutable ref names are rejected", ) - parser.add_argument( - "--api-url", - default=_DEFAULT_API_URL, - help="GitHub API base URL (default: https://api.github.com)", - ) parser.add_argument( "--timeout-seconds", type=float, default=_DEFAULT_TIMEOUT_SECONDS, - help="Finite timeout for each read-only API request", + help="Finite timeout for each read-only GitHub API request", ) return parser @@ -306,7 +300,6 @@ def main(argv: list[str] | None = None) -> int: try: client = GitHubReadClient( token=os.environ.get("GITHUB_TOKEN"), - api_url=args.api_url, timeout_seconds=args.timeout_seconds, ) receipt = audit_repository_workflows( From be28004e57e4a0eadeb4bba8b6ead51c814f6085 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:12:11 +0900 Subject: [PATCH 04/47] test: require protected-ref movement detection --- ...st_workflow_registry_audit_ref_movement.py | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) create mode 100644 tests/test_workflow_registry_audit_ref_movement.py diff --git a/tests/test_workflow_registry_audit_ref_movement.py b/tests/test_workflow_registry_audit_ref_movement.py new file mode 100644 index 000000000..7c576343a --- /dev/null +++ b/tests/test_workflow_registry_audit_ref_movement.py @@ -0,0 +1,84 @@ +"""Protected-ref movement regressions for the workflow registry audit.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest + +from workflow_registry_audit import ( + WorkflowRegistryAuditError, + audit_live_protected_ref_workflows, +) + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +MOVED_SHA = "f" * 40 +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +@dataclass +class _Route: + """Describe one exact fake GitHub read.""" + + suffix: str + payload: dict[str, object] + + +class _FakeClient: + """Serve exact ordered GitHub responses for ref movement tests.""" + + def __init__(self, routes: list[_Route]) -> None: + self._routes = list(routes) + + def get_json(self, path: str) -> dict[str, object]: + """Return the next expected response and reject unexpected reads.""" + if not self._routes: + raise AssertionError(f"unexpected GitHub read: {path}") + route = self._routes.pop(0) + assert path.endswith(route.suffix) + return route.payload + + +def _ref_payload(sha: str) -> dict[str, object]: + return {"ref": "refs/heads/main", "object": {"sha": sha, "type": "commit"}} + + +def _tree_payload() -> dict[str, object]: + return {"sha": PROTECTED_SHA, "truncated": False, "tree": []} + + +def test_initial_protected_ref_mismatch_fails_before_tree_or_registry_reads() -> None: + """A supplied stale protected SHA cannot be certified as live main evidence.""" + client = _FakeClient([_Route("/git/ref/heads/main", _ref_payload(MOVED_SHA))]) + + with pytest.raises(WorkflowRegistryAuditError, match="does not match expected SHA"): + audit_live_protected_ref_workflows( + repository_full_name=REPOSITORY, + protected_ref="main", + expected_protected_sha=PROTECTED_SHA, + client=client, + ) + + +def test_protected_ref_movement_during_registry_scan_fails_closed() -> None: + """A main-branch move during pagination invalidates the entire audit receipt.""" + client = _FakeClient( + [ + _Route("/git/ref/heads/main", _ref_payload(PROTECTED_SHA)), + _Route(f"/git/trees/{PROTECTED_SHA}?recursive=1", _tree_payload()), + _Route( + "/actions/workflows?per_page=100&page=1", + {"total_count": 0, "workflows": []}, + ), + _Route("/git/ref/heads/main", _ref_payload(MOVED_SHA)), + ] + ) + + with pytest.raises(WorkflowRegistryAuditError, match="moved during audit"): + audit_live_protected_ref_workflows( + repository_full_name=REPOSITORY, + protected_ref="main", + expected_protected_sha=PROTECTED_SHA, + client=client, + ) From f84b5c6be490fa7b6eb001377d8a9dcaae361076 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:13:10 +0900 Subject: [PATCH 05/47] fix: fail closed when protected ref moves --- workflow_registry_audit.py | 105 ++++++++++++++++++++++++++++++++++--- 1 file changed, 99 insertions(+), 6 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index bfa5095f6..5c90f45c0 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -16,10 +16,12 @@ from datetime import datetime, timezone from typing import Protocol from urllib.error import HTTPError, URLError +from urllib.parse import quote from urllib.request import Request, urlopen _REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") +_PROTECTED_REF_RE = re.compile(r"^[A-Za-z0-9._/-]+$") _WORKFLOW_PREFIX = ".github/workflows/" _GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 @@ -88,6 +90,53 @@ def _headers(self) -> dict[str, str]: return headers +def audit_live_protected_ref_workflows( + *, + repository_full_name: str, + protected_ref: str, + expected_protected_sha: str, + client: _JsonClient, + captured_at: str | None = None, +) -> dict[str, object]: + """Audit one live protected branch while proving its exact SHA stays unchanged. + + The caller supplies the independently resolved expected protected SHA. The + function verifies the live branch immediately before and after the exact-SHA + tree/registry audit. Any mismatch or observed movement invalidates the whole + receipt instead of silently certifying a stale protected head. + """ + _validate_repository(repository_full_name) + _validate_protected_ref(protected_ref) + _validate_protected_sha(expected_protected_sha) + normalized_sha = expected_protected_sha.lower() + + initial_sha = _read_protected_ref_sha( + repository_full_name=repository_full_name, + protected_ref=protected_ref, + client=client, + ) + if initial_sha != normalized_sha: + raise WorkflowRegistryAuditError("protected ref does not match expected SHA") + + receipt = audit_repository_workflows( + repository_full_name=repository_full_name, + protected_sha=normalized_sha, + client=client, + captured_at=captured_at, + ) + + final_sha = _read_protected_ref_sha( + repository_full_name=repository_full_name, + protected_ref=protected_ref, + client=client, + ) + if final_sha != normalized_sha: + raise WorkflowRegistryAuditError("protected ref moved during audit") + + receipt["protected_ref"] = protected_ref + return receipt + + def audit_repository_workflows( *, repository_full_name: str, @@ -99,7 +148,9 @@ def audit_repository_workflows( Active identities absent from the exact protected tree are reported as candidates only. The caller must separately prove whether a candidate has a - legitimate platform role before any future workflow-state mutation. + legitimate platform role before any future workflow-state mutation. Use + :func:`audit_live_protected_ref_workflows` when live-ref movement must also + invalidate the receipt. """ _validate_repository(repository_full_name) _validate_protected_sha(protected_sha) @@ -148,6 +199,29 @@ def audit_repository_workflows( } +def _read_protected_ref_sha( + *, + repository_full_name: str, + protected_ref: str, + client: _JsonClient, +) -> str: + """Resolve an exact protected branch head SHA from bounded GitHub metadata.""" + encoded_ref = quote(protected_ref, safe="") + payload = client.get_json( + f"/repos/{repository_full_name}/git/ref/heads/{encoded_ref}" + ) + if payload.get("ref") != f"refs/heads/{protected_ref}": + raise WorkflowRegistryAuditError("protected ref response is invalid") + ref_object = payload.get("object") + if not isinstance(ref_object, dict): + raise WorkflowRegistryAuditError("protected ref response is invalid") + sha = ref_object.get("sha") + object_type = ref_object.get("type") + if not isinstance(sha, str) or not _SHA_RE.fullmatch(sha) or object_type != "commit": + raise WorkflowRegistryAuditError("protected ref response is invalid") + return sha.lower() + + def _read_protected_workflow_paths( *, repository_full_name: str, @@ -266,6 +340,19 @@ def _validate_protected_sha(protected_sha: str) -> None: raise WorkflowRegistryAuditError("protected_sha must be an exact 40-hex protected SHA") +def _validate_protected_ref(protected_ref: str) -> None: + """Reject ambiguous or path-like branch selectors before any network read.""" + if ( + not _PROTECTED_REF_RE.fullmatch(protected_ref) + or protected_ref.startswith("/") + or protected_ref.endswith("/") + or "//" in protected_ref + or "@{" in protected_ref + or any(component in {"", ".", ".."} for component in protected_ref.split("/")) + ): + raise WorkflowRegistryAuditError("protected_ref must be a safe branch name") + + def _utc_timestamp() -> str: """Return a UTC RFC 3339 timestamp for the audit receipt.""" return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") @@ -276,14 +363,19 @@ def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "Read a repository Actions registry and classify active workflow identities " - "whose exact path is absent from an exact protected commit." + "whose exact path is absent from a stable protected branch head." ) ) parser.add_argument("--repository", required=True, help="GitHub owner/name repository") parser.add_argument( "--protected-sha", required=True, - help="Exact 40-hex protected commit SHA; mutable ref names are rejected", + help="Independently resolved exact 40-hex protected commit SHA", + ) + parser.add_argument( + "--protected-ref", + default="main", + help="Protected branch to verify before and after the audit (default: main)", ) parser.add_argument( "--timeout-seconds", @@ -295,16 +387,17 @@ def _parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: - """Run the audit, print a JSON receipt, and signal active-absent candidates.""" + """Run the live-ref audit, print JSON, and signal active-absent candidates.""" args = _parser().parse_args(argv) try: client = GitHubReadClient( token=os.environ.get("GITHUB_TOKEN"), timeout_seconds=args.timeout_seconds, ) - receipt = audit_repository_workflows( + receipt = audit_live_protected_ref_workflows( repository_full_name=args.repository, - protected_sha=args.protected_sha, + protected_ref=args.protected_ref, + expected_protected_sha=args.protected_sha, client=client, ) except WorkflowRegistryAuditError as exc: From 6f6eaa05ee148870121e4b94aa82866ccde82db4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:20:39 +0900 Subject: [PATCH 06/47] test: lock authenticated audit reads to GitHub API --- tests/test_workflow_registry_audit.py | 31 +++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 4fb6be0a5..5470d177e 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -281,6 +281,37 @@ def test_invalid_sha_is_rejected_before_any_github_read() -> None: assert client.requested_paths == [] +def test_authenticated_reads_are_pinned_to_github_api(monkeypatch: pytest.MonkeyPatch) -> None: + """A maintenance token can never be redirected to a caller-selected API origin.""" + observed_urls: list[str] = [] + observed_authorizations: list[str | None] = [] + + class _Response: + def __enter__(self) -> "_Response": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def read(self) -> bytes: + return b"{}" + + def _capture_request(request: object, *, timeout: float) -> _Response: + assert timeout == 15.0 + observed_urls.append(request.full_url) # type: ignore[attr-defined] + observed_authorizations.append(request.get_header("Authorization")) # type: ignore[attr-defined] + return _Response() + + monkeypatch.setattr("workflow_registry_audit.urlopen", _capture_request) + client = GitHubReadClient(token="bounded-test-token") + + assert client.get_json(f"/repos/{REPOSITORY}/actions/workflows") == {} + assert observed_urls == [ + f"https://api.github.com/repos/{REPOSITORY}/actions/workflows" + ] + assert observed_authorizations == ["Bearer bounded-test-token"] + + def test_http_failure_does_not_retain_lower_layer_diagnostics(monkeypatch: pytest.MonkeyPatch) -> None: """403/404/5xx-style transport details remain outside rendered evidence.""" secret_sentinel = "SECRET_HTTP_DIAGNOSTIC_SHOULD_NOT_ESCAPE" From a4f9dc7ba39f23e0a484fd76c18874b78d166c35 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:28:56 +0900 Subject: [PATCH 07/47] test(workflows): expose same-count pagination drift --- .../test_workflow_registry_audit_stability.py | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 tests/test_workflow_registry_audit_stability.py diff --git a/tests/test_workflow_registry_audit_stability.py b/tests/test_workflow_registry_audit_stability.py new file mode 100644 index 000000000..a358c3afe --- /dev/null +++ b/tests/test_workflow_registry_audit_stability.py @@ -0,0 +1,96 @@ +"""Adversarial stability regressions for workflow registry pagination.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest + +from workflow_registry_audit import WorkflowRegistryAuditError, audit_repository_workflows + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" +CAPTURED_AT = "2026-08-15T00:00:00Z" + + +@dataclass +class _JsonRoute: + """Describe one expected fake GitHub JSON response.""" + + suffix: str + payload: dict[str, object] + + +class _FakeClient: + """Serve deterministic GitHub API payloads while recording exact reads.""" + + def __init__(self, routes: list[_JsonRoute]) -> None: + self._routes = list(routes) + + def get_json(self, path: str) -> dict[str, object]: + """Return the next payload only for the exact expected request suffix.""" + if not self._routes: + raise AssertionError(f"unexpected GitHub read: {path}") + route = self._routes.pop(0) + assert path.endswith(route.suffix) + return route.payload + + +def _workflow(workflow_id: int, *, state: str = "disabled_manually") -> dict[str, object]: + """Return one deterministic workflow-registry record.""" + return { + "id": workflow_id, + "path": f".github/workflows/workflow-{workflow_id}.yml", + "state": state, + } + + +def test_same_count_page_shift_is_not_accepted_as_stable_registry() -> None: + """A same-cardinality delete/add between pages must invalidate the receipt. + + The first pagination pass can otherwise contain 1..100 from the old registry + plus 102 from the new registry: 101 unique rows with an unchanged total_count, + but not one coherent registry state. A stable verification pass observes the + actual new set 2..102 and must make the audit fail closed. + """ + first_page_before_shift = [_workflow(workflow_id) for workflow_id in range(1, 101)] + second_page_after_shift = [_workflow(102, state="active")] + stable_new_first_page = [_workflow(workflow_id) for workflow_id in range(2, 102)] + stable_new_second_page = [_workflow(102, state="active")] + client = _FakeClient( + [ + _JsonRoute( + f"/git/trees/{PROTECTED_SHA}?recursive=1", + { + "sha": PROTECTED_SHA, + "truncated": False, + "tree": [], + }, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + {"total_count": 101, "workflows": first_page_before_shift}, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=2", + {"total_count": 101, "workflows": second_page_after_shift}, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + {"total_count": 101, "workflows": stable_new_first_page}, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=2", + {"total_count": 101, "workflows": stable_new_second_page}, + ), + ] + ) + + with pytest.raises(WorkflowRegistryAuditError, match="changed during audit"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at=CAPTURED_AT, + ) From 8d0c6ede73776e5f38d0a1151d5f95f140eb52e4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:30:24 +0900 Subject: [PATCH 08/47] fix(workflows): verify multi-page registry stability --- workflow_registry_audit.py | 47 +++++++++++++++++++++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 5c90f45c0..ceb40f29e 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -258,7 +258,38 @@ def _read_registry( repository_full_name: str, client: _JsonClient, ) -> tuple[list[dict[str, object]], int, int]: - """Read the complete stable workflow registry with explicit pagination evidence.""" + """Read one coherent workflow registry or fail closed on pagination drift. + + A single multi-page traversal can combine rows from two different registry + states even when ``total_count`` remains unchanged. Therefore multi-page + receipts are accepted only when a second complete traversal observes the + same validated workflow identities, paths, and states. Single-page reads + already arrive as one response and do not need the extra pass. + """ + records, pages_scanned, expected_total = _read_registry_pass( + repository_full_name=repository_full_name, + client=client, + ) + if pages_scanned > 1: + verification_records, verification_pages, verification_total = _read_registry_pass( + repository_full_name=repository_full_name, + client=client, + ) + if ( + verification_total != expected_total + or verification_pages != pages_scanned + or _registry_signature(verification_records) != _registry_signature(records) + ): + raise WorkflowRegistryAuditError("workflow registry changed during audit") + return records, pages_scanned, expected_total + + +def _read_registry_pass( + *, + repository_full_name: str, + client: _JsonClient, +) -> tuple[list[dict[str, object]], int, int]: + """Read one complete pagination pass while enforcing stable cardinality.""" expected_total: int | None = None page = 1 pages_scanned = 0 @@ -300,6 +331,20 @@ def _read_registry( return records, pages_scanned, expected_total +def _registry_signature(records: list[dict[str, object]]) -> tuple[tuple[int, str, str], ...]: + """Return an order-independent identity/path/state signature for one pass.""" + return tuple( + sorted( + ( + int(record["workflow_id"]), + str(record["path"]), + str(record["state"]), + ) + for record in records + ) + ) + + def _validate_workflow_record(raw_record: object) -> dict[str, object]: """Validate only identity/path/state fields needed for safe classification.""" if not isinstance(raw_record, dict): From d1b8449662cbaac748a47d86fd7c7cedbdeff1b5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:31:10 +0900 Subject: [PATCH 09/47] test(workflows): verify stable multi-page registry pass --- tests/test_workflow_registry_audit.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 5470d177e..f57514482 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -118,7 +118,7 @@ def test_exact_path_presence_drives_classification_not_workflow_name() -> None: def test_registry_pagination_is_complete_and_receipted() -> None: - """The detector reads every page required by the first stable total count.""" + """The detector verifies a complete multi-page registry twice before receipt.""" first_page = [ _workflow(index, f".github/workflows/old-{index}.yml", state="disabled_manually") for index in range(1, 101) @@ -138,6 +138,14 @@ def test_registry_pagination_is_complete_and_receipted() -> None: "/actions/workflows?per_page=100&page=2", {"total_count": 101, "workflows": second_page}, ), + _JsonRoute( + "/actions/workflows?per_page=100&page=1", + {"total_count": 101, "workflows": first_page}, + ), + _JsonRoute( + "/actions/workflows?per_page=100&page=2", + {"total_count": 101, "workflows": second_page}, + ), ] ) From c33759e61f67f06579755e375048301894958b9a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:39:33 +0900 Subject: [PATCH 10/47] test(workflows): require fixed HTTPS transport boundary --- tests/test_workflow_registry_audit.py | 112 ++++++++++++++++++-------- 1 file changed, 79 insertions(+), 33 deletions(-) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index f57514482..8a12fa2af 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -2,11 +2,8 @@ from __future__ import annotations -import io -import json import traceback from dataclasses import dataclass -from urllib.error import HTTPError import pytest @@ -289,51 +286,100 @@ def test_invalid_sha_is_rejected_before_any_github_read() -> None: assert client.requested_paths == [] -def test_authenticated_reads_are_pinned_to_github_api(monkeypatch: pytest.MonkeyPatch) -> None: - """A maintenance token can never be redirected to a caller-selected API origin.""" - observed_urls: list[str] = [] - observed_authorizations: list[str | None] = [] +def test_authenticated_reads_use_fixed_https_origin_and_path_only( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Credentials stay on a fixed GitHub HTTPS connection with a path-only target.""" + observed_connections: list[tuple[str, float]] = [] + observed_requests: list[tuple[str, str, dict[str, str]]] = [] + closed: list[bool] = [] class _Response: - def __enter__(self) -> "_Response": - return self - - def __exit__(self, *_args: object) -> None: - return None + status = 200 def read(self) -> bytes: return b"{}" - def _capture_request(request: object, *, timeout: float) -> _Response: - assert timeout == 15.0 - observed_urls.append(request.full_url) # type: ignore[attr-defined] - observed_authorizations.append(request.get_header("Authorization")) # type: ignore[attr-defined] - return _Response() + class _Connection: + def __init__(self, host: str, *, timeout: float) -> None: + observed_connections.append((host, timeout)) - monkeypatch.setattr("workflow_registry_audit.urlopen", _capture_request) - client = GitHubReadClient(token="bounded-test-token") + def request( + self, + method: str, + path: str, + *, + headers: dict[str, str], + ) -> None: + observed_requests.append((method, path, headers)) + + def getresponse(self) -> _Response: + return _Response() - assert client.get_json(f"/repos/{REPOSITORY}/actions/workflows") == {} - assert observed_urls == [ - f"https://api.github.com/repos/{REPOSITORY}/actions/workflows" + def close(self) -> None: + closed.append(True) + + monkeypatch.setattr("workflow_registry_audit.HTTPSConnection", _Connection) + client = GitHubReadClient(token="bounded-test-token") + path = f"/repos/{REPOSITORY}/actions/workflows" + + assert client.get_json(path) == {} + assert observed_connections == [("api.github.com", 15.0)] + assert observed_requests == [ + ( + "GET", + path, + { + "Accept": "application/vnd.github+json", + "User-Agent": "pg-llm-batch-workflow-registry-audit/1", + "X-GitHub-Api-Version": "2022-11-28", + "Authorization": "Bearer bounded-test-token", + }, + ) ] - assert observed_authorizations == ["Bearer bounded-test-token"] + assert closed == [True] + + +def test_absolute_uri_is_rejected_before_transport(monkeypatch: pytest.MonkeyPatch) -> None: + """A caller cannot turn the request target into another URL or scheme.""" + connection_attempted = False + + class _Connection: + def __init__(self, *_args: object, **_kwargs: object) -> None: + nonlocal connection_attempted + connection_attempted = True + + monkeypatch.setattr("workflow_registry_audit.HTTPSConnection", _Connection) + client = GitHubReadClient(token="bounded-test-token") + + with pytest.raises(WorkflowRegistryAuditError, match="GitHub API path is invalid"): + client.get_json("https://attacker.invalid/repos/owner/repo") + + assert connection_attempted is False def test_http_failure_does_not_retain_lower_layer_diagnostics(monkeypatch: pytest.MonkeyPatch) -> None: - """403/404/5xx-style transport details remain outside rendered evidence.""" + """Transport details remain outside rendered evidence and exception chaining.""" secret_sentinel = "SECRET_HTTP_DIAGNOSTIC_SHOULD_NOT_ESCAPE" - def _raise_http_error(*_args: object, **_kwargs: object) -> object: - raise HTTPError( - "https://api.github.com/private?token=secret", - 503, - secret_sentinel, - hdrs=None, - fp=io.BytesIO(json.dumps({"message": secret_sentinel}).encode()), - ) + class _Connection: + def __init__(self, _host: str, *, timeout: float) -> None: + assert timeout == 15.0 + + def request( + self, + _method: str, + _path: str, + *, + headers: dict[str, str], + ) -> None: + assert headers["Authorization"] == "Bearer token-value-that-must-not-render" + raise OSError(secret_sentinel) + + def close(self) -> None: + return None - monkeypatch.setattr("workflow_registry_audit.urlopen", _raise_http_error) + monkeypatch.setattr("workflow_registry_audit.HTTPSConnection", _Connection) client = GitHubReadClient(token="token-value-that-must-not-render") try: From 580b757110a64261e450457236c47e58ffad219a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:41:30 +0900 Subject: [PATCH 11/47] fix(workflows): pin audit transport to HTTPS host --- workflow_registry_audit.py | 29 ++++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index ceb40f29e..6f122c988 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -14,16 +14,16 @@ import os import re from datetime import datetime, timezone +from http.client import HTTPException, HTTPSConnection from typing import Protocol -from urllib.error import HTTPError, URLError from urllib.parse import quote -from urllib.request import Request, urlopen _REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") _PROTECTED_REF_RE = re.compile(r"^[A-Za-z0-9._/-]+$") +_API_PATH_RE = re.compile(r"^/(?!/)[^\r\n]*$") _WORKFLOW_PREFIX = ".github/workflows/" -_GITHUB_API_URL = "https://api.github.com" +_GITHUB_API_HOST = "api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 @@ -55,25 +55,28 @@ def __init__( self._timeout_seconds = timeout_seconds def get_json(self, path: str) -> dict[str, object]: - """Read one JSON object and replace transport/parser failures with fixed evidence.""" - request = Request( - f"{_GITHUB_API_URL}{path}", - headers=self._headers(), - method="GET", - ) + """Read one path-only GitHub API object over a fixed HTTPS origin.""" + if not _API_PATH_RE.fullmatch(path): + raise WorkflowRegistryAuditError("GitHub API path is invalid") + + connection = HTTPSConnection(_GITHUB_API_HOST, timeout=self._timeout_seconds) try: - with urlopen(request, timeout=self._timeout_seconds) as response: - raw = response.read() + connection.request("GET", path, headers=self._headers()) + response = connection.getresponse() + if response.status // 100 != 2: + raise HTTPException("unexpected GitHub API response") + raw = response.read() payload = json.loads(raw.decode("utf-8")) except ( - HTTPError, - URLError, + HTTPException, TimeoutError, OSError, UnicodeDecodeError, json.JSONDecodeError, ): raise WorkflowRegistryAuditError("GitHub workflow audit read failed") from None + finally: + connection.close() if not isinstance(payload, dict): raise WorkflowRegistryAuditError("GitHub workflow audit response is invalid") return payload From 93ba3ba0781cd696f60cecec399b3f8b0934632e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:46:21 +0900 Subject: [PATCH 12/47] test(workflows): require fixed aiohttp origin boundary --- tests/test_workflow_registry_audit.py | 91 ++++++++++++++++----------- 1 file changed, 53 insertions(+), 38 deletions(-) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 8a12fa2af..0b6aa57ef 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -5,6 +5,7 @@ import traceback from dataclasses import dataclass +import aiohttp import pytest from workflow_registry_audit import ( @@ -286,49 +287,55 @@ def test_invalid_sha_is_rejected_before_any_github_read() -> None: assert client.requested_paths == [] -def test_authenticated_reads_use_fixed_https_origin_and_path_only( +def test_authenticated_reads_use_fixed_aiohttp_origin_and_path_only( monkeypatch: pytest.MonkeyPatch, ) -> None: - """Credentials stay on a fixed GitHub HTTPS connection with a path-only target.""" - observed_connections: list[tuple[str, float]] = [] - observed_requests: list[tuple[str, str, dict[str, str]]] = [] + """Credentials stay on one fixed GitHub origin with a path-only request target.""" + observed_sessions: list[tuple[str, float, dict[str, str]]] = [] + observed_requests: list[tuple[str, bool]] = [] closed: list[bool] = [] class _Response: status = 200 - def read(self) -> bytes: - return b"{}" + async def __aenter__(self) -> "_Response": + return self + + async def __aexit__(self, *_args: object) -> None: + return None - class _Connection: - def __init__(self, host: str, *, timeout: float) -> None: - observed_connections.append((host, timeout)) + async def read(self) -> bytes: + return b"{}" - def request( + class _Session: + def __init__( self, - method: str, - path: str, *, + base_url: str, headers: dict[str, str], + timeout: aiohttp.ClientTimeout, ) -> None: - observed_requests.append((method, path, headers)) + observed_sessions.append((base_url, float(timeout.total), headers)) - def getresponse(self) -> _Response: - return _Response() + async def __aenter__(self) -> "_Session": + return self - def close(self) -> None: + async def __aexit__(self, *_args: object) -> None: closed.append(True) - monkeypatch.setattr("workflow_registry_audit.HTTPSConnection", _Connection) + def get(self, path: str, *, allow_redirects: bool) -> _Response: + observed_requests.append((path, allow_redirects)) + return _Response() + + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) client = GitHubReadClient(token="bounded-test-token") path = f"/repos/{REPOSITORY}/actions/workflows" assert client.get_json(path) == {} - assert observed_connections == [("api.github.com", 15.0)] - assert observed_requests == [ + assert observed_sessions == [ ( - "GET", - path, + "https://api.github.com", + 15.0, { "Accept": "application/vnd.github+json", "User-Agent": "pg-llm-batch-workflow-registry-audit/1", @@ -337,49 +344,57 @@ def close(self) -> None: }, ) ] + assert observed_requests == [(path, False)] assert closed == [True] def test_absolute_uri_is_rejected_before_transport(monkeypatch: pytest.MonkeyPatch) -> None: """A caller cannot turn the request target into another URL or scheme.""" - connection_attempted = False + session_attempted = False - class _Connection: + class _Session: def __init__(self, *_args: object, **_kwargs: object) -> None: - nonlocal connection_attempted - connection_attempted = True + nonlocal session_attempted + session_attempted = True - monkeypatch.setattr("workflow_registry_audit.HTTPSConnection", _Connection) + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) client = GitHubReadClient(token="bounded-test-token") with pytest.raises(WorkflowRegistryAuditError, match="GitHub API path is invalid"): client.get_json("https://attacker.invalid/repos/owner/repo") - assert connection_attempted is False + assert session_attempted is False -def test_http_failure_does_not_retain_lower_layer_diagnostics(monkeypatch: pytest.MonkeyPatch) -> None: +def test_transport_failure_does_not_retain_lower_layer_diagnostics( + monkeypatch: pytest.MonkeyPatch, +) -> None: """Transport details remain outside rendered evidence and exception chaining.""" secret_sentinel = "SECRET_HTTP_DIAGNOSTIC_SHOULD_NOT_ESCAPE" - class _Connection: - def __init__(self, _host: str, *, timeout: float) -> None: - assert timeout == 15.0 - - def request( + class _Session: + def __init__( self, - _method: str, - _path: str, *, + base_url: str, headers: dict[str, str], + timeout: aiohttp.ClientTimeout, ) -> None: + assert base_url == "https://api.github.com" assert headers["Authorization"] == "Bearer token-value-that-must-not-render" - raise OSError(secret_sentinel) + assert timeout.total == 15.0 + + async def __aenter__(self) -> "_Session": + return self - def close(self) -> None: + async def __aexit__(self, *_args: object) -> None: return None - monkeypatch.setattr("workflow_registry_audit.HTTPSConnection", _Connection) + def get(self, _path: str, *, allow_redirects: bool) -> object: + assert allow_redirects is False + raise aiohttp.ClientConnectionError(secret_sentinel) + + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) client = GitHubReadClient(token="token-value-that-must-not-render") try: From ab948bac5128d410102ee602f1d650a7f9cccdb7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 06:47:43 +0900 Subject: [PATCH 13/47] fix(workflows): use fixed-origin aiohttp transport --- workflow_registry_audit.py | 46 ++++++++++++++++++++++++++------------ 1 file changed, 32 insertions(+), 14 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 6f122c988..74da79cbd 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -10,20 +10,22 @@ from __future__ import annotations import argparse +import asyncio import json import os import re from datetime import datetime, timezone -from http.client import HTTPException, HTTPSConnection from typing import Protocol from urllib.parse import quote +import aiohttp + _REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") _PROTECTED_REF_RE = re.compile(r"^[A-Za-z0-9._/-]+$") _API_PATH_RE = re.compile(r"^/(?!/)[^\r\n]*$") _WORKFLOW_PREFIX = ".github/workflows/" -_GITHUB_API_HOST = "api.github.com" +_GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 @@ -55,28 +57,44 @@ def __init__( self._timeout_seconds = timeout_seconds def get_json(self, path: str) -> dict[str, object]: - """Read one path-only GitHub API object over a fixed HTTPS origin.""" + """Read one path-only GitHub API object over a fixed verified-TLS origin.""" if not _API_PATH_RE.fullmatch(path): raise WorkflowRegistryAuditError("GitHub API path is invalid") + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: + raise WorkflowRegistryAuditError( + "GitHub workflow audit sync client cannot run inside an active event loop" + ) + return asyncio.run(self._get_json(path)) - connection = HTTPSConnection(_GITHUB_API_HOST, timeout=self._timeout_seconds) + async def _get_json(self, path: str) -> dict[str, object]: + """Perform one fixed-origin aiohttp GET with redirects disabled.""" + timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) try: - connection.request("GET", path, headers=self._headers()) - response = connection.getresponse() - if response.status // 100 != 2: - raise HTTPException("unexpected GitHub API response") - raw = response.read() + async with aiohttp.ClientSession( + base_url=_GITHUB_API_URL, + headers=self._headers(), + timeout=timeout, + ) as session: + async with session.get(path, allow_redirects=False) as response: + if response.status // 100 != 2: + raise WorkflowRegistryAuditError( + "GitHub workflow audit read failed" + ) + raw = await response.read() payload = json.loads(raw.decode("utf-8")) + except WorkflowRegistryAuditError: + raise except ( - HTTPException, - TimeoutError, - OSError, + aiohttp.ClientError, + asyncio.TimeoutError, UnicodeDecodeError, json.JSONDecodeError, ): raise WorkflowRegistryAuditError("GitHub workflow audit read failed") from None - finally: - connection.close() if not isinstance(payload, dict): raise WorkflowRegistryAuditError("GitHub workflow audit response is invalid") return payload From 3c6ae8ac228438f2e25393ccc8a51f01eaf3fb4a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:07:21 +0900 Subject: [PATCH 14/47] test(workflows): expose commit-tree identity mismatch --- ...flow_registry_audit_git_object_identity.py | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 tests/test_workflow_registry_audit_git_object_identity.py diff --git a/tests/test_workflow_registry_audit_git_object_identity.py b/tests/test_workflow_registry_audit_git_object_identity.py new file mode 100644 index 000000000..2a44a4943 --- /dev/null +++ b/tests/test_workflow_registry_audit_git_object_identity.py @@ -0,0 +1,81 @@ +"""Git-object identity regression for the workflow registry audit.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from workflow_registry_audit import audit_repository_workflows + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +@dataclass +class _Route: + """Describe one exact fake GitHub read.""" + + suffix: str + payload: dict[str, object] + + +class _FakeClient: + """Require the documented commit-to-tree lookup sequence.""" + + def __init__(self, routes: list[_Route]) -> None: + self._routes = list(routes) + self.requested_paths: list[str] = [] + + def get_json(self, path: str) -> dict[str, object]: + """Return only the next exact response and record every request.""" + self.requested_paths.append(path) + route = self._routes.pop(0) + assert path.endswith(route.suffix) + return route.payload + + +def test_protected_commit_is_resolved_to_its_tree_before_recursive_read() -> None: + """The audit must not treat a commit object SHA as a tree object SHA.""" + client = _FakeClient( + [ + _Route( + f"/git/commits/{PROTECTED_SHA}", + {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}}, + ), + _Route( + f"/git/trees/{TREE_SHA}?recursive=1", + { + "sha": TREE_SHA, + "truncated": False, + "tree": [{"path": ".github/workflows/ci.yml", "type": "blob"}], + }, + ), + _Route( + "/actions/workflows?per_page=100&page=1", + { + "total_count": 1, + "workflows": [ + { + "id": 1, + "path": ".github/workflows/ci.yml", + "state": "active", + } + ], + }, + ), + ] + ) + + receipt = audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at="2026-08-15T00:00:00Z", + ) + + assert receipt["active_absent_workflows"] == [] + assert client.requested_paths[:2] == [ + f"/repos/{REPOSITORY}/git/commits/{PROTECTED_SHA}", + f"/repos/{REPOSITORY}/git/trees/{TREE_SHA}?recursive=1", + ] From df2d4de81bf7368690e6df92d948673e961b9f99 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:08:40 +0900 Subject: [PATCH 15/47] fix(workflows): resolve commit tree before registry audit --- workflow_registry_audit.py | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 74da79cbd..6e1b8ddb8 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -14,6 +14,7 @@ import json import os import re +import sys from datetime import datetime, timezone from typing import Protocol from urllib.parse import quote @@ -249,12 +250,25 @@ def _read_protected_workflow_paths( protected_sha: str, client: _JsonClient, ) -> set[str]: - """Read exact protected-tree workflow blob paths or fail closed.""" + """Resolve a protected commit to its exact tree and return workflow blob paths.""" + commit_payload = client.get_json( + f"/repos/{repository_full_name}/git/commits/{protected_sha}" + ) + if commit_payload.get("sha") != protected_sha: + raise WorkflowRegistryAuditError("protected commit response is invalid") + commit_tree = commit_payload.get("tree") + if not isinstance(commit_tree, dict): + raise WorkflowRegistryAuditError("protected commit response is invalid") + tree_sha = commit_tree.get("sha") + if not isinstance(tree_sha, str) or not _SHA_RE.fullmatch(tree_sha): + raise WorkflowRegistryAuditError("protected commit response is invalid") + tree_sha = tree_sha.lower() + payload = client.get_json( - f"/repos/{repository_full_name}/git/trees/{protected_sha}?recursive=1" + f"/repos/{repository_full_name}/git/trees/{tree_sha}?recursive=1" ) - if payload.get("sha") != protected_sha: - raise WorkflowRegistryAuditError("protected tree SHA does not match requested SHA") + if payload.get("sha") != tree_sha: + raise WorkflowRegistryAuditError("protected tree SHA does not match commit tree SHA") if payload.get("truncated") is not False: raise WorkflowRegistryAuditError("protected tree is truncated") tree = payload.get("tree") @@ -467,7 +481,7 @@ def main(argv: list[str] | None = None) -> int: client=client, ) except WorkflowRegistryAuditError as exc: - print(f"workflow_registry_audit: {exc}", file=os.sys.stderr) + print(f"workflow_registry_audit: {exc}", file=sys.stderr) return 1 print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) From fabaad7cd5fd1f42c66c3c8ff18ae30d14652aef Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:09:32 +0900 Subject: [PATCH 16/47] test(workflows): separate protected commit and tree fixtures --- tests/test_workflow_registry_audit.py | 29 +++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 0b6aa57ef..9f8ad994d 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -16,6 +16,7 @@ PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" REPOSITORY = "ContextualWisdomLab/pg-llm-batch" CAPTURED_AT = "2026-08-15T00:00:00Z" @@ -45,9 +46,15 @@ def get_json(self, path: str) -> dict[str, object]: return route.payload +def _commit_payload() -> dict[str, object]: + """Return a commit response whose tree identity differs from the commit SHA.""" + return {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}} + + def _tree_payload(*paths: str, truncated: bool = False) -> dict[str, object]: + """Return the recursive tree object resolved from the protected commit.""" return { - "sha": PROTECTED_SHA, + "sha": TREE_SHA, "truncated": truncated, "tree": [ {"path": path, "type": "blob", "sha": f"blob-{index}"} @@ -69,8 +76,9 @@ def test_exact_path_presence_drives_classification_not_workflow_name() -> None: """One-shot-like names stay safe when their exact protected path exists.""" client = _FakeClient( [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload( ".github/workflows/ci.yml", ".github/workflows/one-shot-legitimate.yml", @@ -124,8 +132,9 @@ def test_registry_pagination_is_complete_and_receipted() -> None: second_page = [_workflow(101, ".github/workflows/ci.yml")] client = _FakeClient( [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload(".github/workflows/ci.yml"), ), _JsonRoute( @@ -164,8 +173,9 @@ def test_incomplete_pagination_fails_closed() -> None: """An empty page before the advertised total is not accepted as complete.""" client = _FakeClient( [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload(".github/workflows/ci.yml"), ), _JsonRoute( @@ -192,8 +202,9 @@ def test_registry_change_during_pagination_fails_closed() -> None: """A moving registry cannot be presented as one coherent audit receipt.""" client = _FakeClient( [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload(".github/workflows/ci.yml"), ), _JsonRoute( @@ -226,8 +237,9 @@ def test_duplicate_workflow_id_fails_closed() -> None: """A reused ID with ambiguous path/state cannot be silently normalized.""" client = _FakeClient( [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload(".github/workflows/ci.yml"), ), _JsonRoute( @@ -256,10 +268,11 @@ def test_truncated_protected_tree_fails_closed() -> None: """A partial protected tree is not evidence that a workflow source is absent.""" client = _FakeClient( [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload(".github/workflows/ci.yml", truncated=True), - ) + ), ] ) From 4dd097925be2631088050f7ce0a7e1de551db4ab Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:11:25 +0900 Subject: [PATCH 17/47] test(workflows): resolve protected commit tree in ref movement --- tests/test_workflow_registry_audit_ref_movement.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/tests/test_workflow_registry_audit_ref_movement.py b/tests/test_workflow_registry_audit_ref_movement.py index 7c576343a..16212c31e 100644 --- a/tests/test_workflow_registry_audit_ref_movement.py +++ b/tests/test_workflow_registry_audit_ref_movement.py @@ -13,6 +13,7 @@ PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" MOVED_SHA = "f" * 40 REPOSITORY = "ContextualWisdomLab/pg-llm-batch" @@ -44,8 +45,14 @@ def _ref_payload(sha: str) -> dict[str, object]: return {"ref": "refs/heads/main", "object": {"sha": sha, "type": "commit"}} +def _commit_payload() -> dict[str, object]: + """Return the protected commit with a distinct recursive-tree identity.""" + return {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}} + + def _tree_payload() -> dict[str, object]: - return {"sha": PROTECTED_SHA, "truncated": False, "tree": []} + """Return the tree object referenced by the protected commit.""" + return {"sha": TREE_SHA, "truncated": False, "tree": []} def test_initial_protected_ref_mismatch_fails_before_tree_or_registry_reads() -> None: @@ -66,7 +73,8 @@ def test_protected_ref_movement_during_registry_scan_fails_closed() -> None: client = _FakeClient( [ _Route("/git/ref/heads/main", _ref_payload(PROTECTED_SHA)), - _Route(f"/git/trees/{PROTECTED_SHA}?recursive=1", _tree_payload()), + _Route(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), + _Route(f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload()), _Route( "/actions/workflows?per_page=100&page=1", {"total_count": 0, "workflows": []}, From 1514b1620a2509ec438b343dbfdd594a9f726181 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:11:58 +0900 Subject: [PATCH 18/47] test(workflows): resolve commit tree in stability fixtures --- tests/test_workflow_registry_audit_stability.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/test_workflow_registry_audit_stability.py b/tests/test_workflow_registry_audit_stability.py index a358c3afe..8ed8f50a2 100644 --- a/tests/test_workflow_registry_audit_stability.py +++ b/tests/test_workflow_registry_audit_stability.py @@ -10,6 +10,7 @@ PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" REPOSITORY = "ContextualWisdomLab/pg-llm-batch" CAPTURED_AT = "2026-08-15T00:00:00Z" @@ -61,9 +62,13 @@ def test_same_count_page_shift_is_not_accepted_as_stable_registry() -> None: client = _FakeClient( [ _JsonRoute( - f"/git/trees/{PROTECTED_SHA}?recursive=1", + f"/git/commits/{PROTECTED_SHA}", + {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}}, + ), + _JsonRoute( + f"/git/trees/{TREE_SHA}?recursive=1", { - "sha": PROTECTED_SHA, + "sha": TREE_SHA, "truncated": False, "tree": [], }, From 1197fd4363db45b5b4c712211b1bdb30b5e2afc3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:15:31 +0900 Subject: [PATCH 19/47] test(workflows): expose rate-limit diagnostic ambiguity --- ...test_workflow_registry_audit_rate_limit.py | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 tests/test_workflow_registry_audit_rate_limit.py diff --git a/tests/test_workflow_registry_audit_rate_limit.py b/tests/test_workflow_registry_audit_rate_limit.py new file mode 100644 index 000000000..7ad6cc06f --- /dev/null +++ b/tests/test_workflow_registry_audit_rate_limit.py @@ -0,0 +1,107 @@ +"""Rate-limit classification regressions for the workflow registry audit.""" + +from __future__ import annotations + +import aiohttp +import pytest + +from workflow_registry_audit import GitHubReadClient, WorkflowRegistryAuditError + + +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +@pytest.mark.parametrize( + ("status", "headers"), + [ + (429, {}), + (403, {"X-RateLimit-Remaining": "0"}), + (403, {"Retry-After": "60"}), + ], +) +def test_rate_limit_responses_have_bounded_distinct_diagnostic( + monkeypatch: pytest.MonkeyPatch, + status: int, + headers: dict[str, str], +) -> None: + """Known GitHub rate-limit signals must not be conflated with generic failure.""" + + class _Response: + async def __aenter__(self) -> "_Response": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + async def read(self) -> bytes: + raise AssertionError("rate-limit response bodies must not be rendered or parsed") + + response = _Response() + response.status = status + response.headers = headers + + class _Session: + def __init__( + self, + *, + base_url: str, + headers: dict[str, str], + timeout: aiohttp.ClientTimeout, + ) -> None: + assert base_url == "https://api.github.com" + assert timeout.total == 15.0 + + async def __aenter__(self) -> "_Session": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + def get(self, _path: str, *, allow_redirects: bool) -> _Response: + assert allow_redirects is False + return response + + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) + client = GitHubReadClient(token="bounded-test-token") + + with pytest.raises(WorkflowRegistryAuditError, match="^GitHub workflow audit rate limited$"): + client.get_json(f"/repos/{REPOSITORY}/actions/workflows?per_page=100&page=1") + + +def test_ordinary_forbidden_response_remains_generic( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A 403 without a rate-limit signal must remain an authorization-agnostic failure.""" + + class _Response: + status = 403 + headers: dict[str, str] = {} + + async def __aenter__(self) -> "_Response": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + async def read(self) -> bytes: + raise AssertionError("non-success response bodies must not be rendered or parsed") + + class _Session: + def __init__(self, **_kwargs: object) -> None: + return None + + async def __aenter__(self) -> "_Session": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + def get(self, _path: str, *, allow_redirects: bool) -> _Response: + assert allow_redirects is False + return _Response() + + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) + client = GitHubReadClient(token="bounded-test-token") + + with pytest.raises(WorkflowRegistryAuditError, match="^GitHub workflow audit read failed$"): + client.get_json(f"/repos/{REPOSITORY}/actions/workflows?per_page=100&page=1") From 9d54970f83cee30616e4f776f580c9bb1be51069 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 07:17:56 +0900 Subject: [PATCH 20/47] fix(workflows): classify GitHub rate-limit failures --- workflow_registry_audit.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 6e1b8ddb8..8e508fb63 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -82,6 +82,15 @@ async def _get_json(self, path: str) -> dict[str, object]: ) as session: async with session.get(path, allow_redirects=False) as response: if response.status // 100 != 2: + remaining = response.headers.get("X-RateLimit-Remaining") + retry_after = response.headers.get("Retry-After") + if response.status == 429 or ( + response.status == 403 + and (remaining == "0" or retry_after is not None) + ): + raise WorkflowRegistryAuditError( + "GitHub workflow audit rate limited" + ) raise WorkflowRegistryAuditError( "GitHub workflow audit read failed" ) From a2920a5a4e59989630f2b1fa65fdb9d9e07334ec Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 08:04:31 +0900 Subject: [PATCH 21/47] test(workflows): prove slash ref path regression --- ...st_workflow_registry_audit_ref_movement.py | 35 +++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/tests/test_workflow_registry_audit_ref_movement.py b/tests/test_workflow_registry_audit_ref_movement.py index 16212c31e..43df204ac 100644 --- a/tests/test_workflow_registry_audit_ref_movement.py +++ b/tests/test_workflow_registry_audit_ref_movement.py @@ -41,8 +41,12 @@ def get_json(self, path: str) -> dict[str, object]: return route.payload -def _ref_payload(sha: str) -> dict[str, object]: - return {"ref": "refs/heads/main", "object": {"sha": sha, "type": "commit"}} +def _ref_payload(sha: str, *, protected_ref: str = "main") -> dict[str, object]: + """Return one exact branch-reference response for ``protected_ref``.""" + return { + "ref": f"refs/heads/{protected_ref}", + "object": {"sha": sha, "type": "commit"}, + } def _commit_payload() -> dict[str, object]: @@ -68,6 +72,33 @@ def test_initial_protected_ref_mismatch_fails_before_tree_or_registry_reads() -> ) +def test_slash_protected_ref_preserves_github_ref_path_shape() -> None: + """A valid slash-bearing branch uses GitHub's documented heads/ path.""" + protected_ref = "release/1.0" + ref_payload = _ref_payload(PROTECTED_SHA, protected_ref=protected_ref) + client = _FakeClient( + [ + _Route(f"/git/ref/heads/{protected_ref}", ref_payload), + _Route(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), + _Route(f"/git/trees/{TREE_SHA}?recursive=1", _tree_payload()), + _Route( + "/actions/workflows?per_page=100&page=1", + {"total_count": 0, "workflows": []}, + ), + _Route(f"/git/ref/heads/{protected_ref}", ref_payload), + ] + ) + + receipt = audit_live_protected_ref_workflows( + repository_full_name=REPOSITORY, + protected_ref=protected_ref, + expected_protected_sha=PROTECTED_SHA, + client=client, + ) + + assert receipt["protected_ref"] == protected_ref + + def test_protected_ref_movement_during_registry_scan_fails_closed() -> None: """A main-branch move during pagination invalidates the entire audit receipt.""" client = _FakeClient( From c2ca8b587af1008daddb01a7baebdf03f252fb47 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 08:06:17 +0900 Subject: [PATCH 22/47] fix(workflows): preserve slash refs in GitHub paths --- workflow_registry_audit.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 8e508fb63..79e2eb358 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -237,7 +237,7 @@ def _read_protected_ref_sha( client: _JsonClient, ) -> str: """Resolve an exact protected branch head SHA from bounded GitHub metadata.""" - encoded_ref = quote(protected_ref, safe="") + encoded_ref = quote(protected_ref, safe="/") payload = client.get_json( f"/repos/{repository_full_name}/git/ref/heads/{encoded_ref}" ) From 8a1215c2b7d60c913cc6de3a9c2f497c8ef7a4fa Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 08:09:14 +0900 Subject: [PATCH 23/47] test(workflows): reproduce dynamic registry identity gap --- tests/test_workflow_registry_audit_dynamic.py | 89 +++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 tests/test_workflow_registry_audit_dynamic.py diff --git a/tests/test_workflow_registry_audit_dynamic.py b/tests/test_workflow_registry_audit_dynamic.py new file mode 100644 index 000000000..15d03fad4 --- /dev/null +++ b/tests/test_workflow_registry_audit_dynamic.py @@ -0,0 +1,89 @@ +"""Live-registry regressions for platform-managed dynamic workflow identities.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from workflow_registry_audit import audit_repository_workflows + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +@dataclass +class _Route: + """Describe one exact fake GitHub read.""" + + suffix: str + payload: dict[str, object] + + +class _FakeClient: + """Serve exact ordered GitHub responses for dynamic-workflow tests.""" + + def __init__(self, routes: list[_Route]) -> None: + self._routes = list(routes) + + def get_json(self, path: str) -> dict[str, object]: + """Return the next expected response and reject unexpected reads.""" + if not self._routes: + raise AssertionError(f"unexpected GitHub read: {path}") + route = self._routes.pop(0) + assert path.endswith(route.suffix) + return route.payload + + +def test_dynamic_platform_workflow_is_receipted_but_never_an_orphan_candidate() -> None: + """GitHub-managed dynamic identities cannot be mistaken for deleted YAML.""" + repository_path = ".github/workflows/deleted-one-shot.yml" + dynamic_path = "dynamic/github-code-scanning/codeql" + client = _FakeClient( + [ + _Route( + f"/git/commits/{PROTECTED_SHA}", + {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}}, + ), + _Route( + f"/git/trees/{TREE_SHA}?recursive=1", + {"sha": TREE_SHA, "truncated": False, "tree": []}, + ), + _Route( + "/actions/workflows?per_page=100&page=1", + { + "total_count": 2, + "workflows": [ + {"id": 7, "path": repository_path, "state": "active"}, + {"id": 8, "path": dynamic_path, "state": "active"}, + ], + }, + ), + ] + ) + + receipt = audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at="2026-08-15T00:00:00Z", + ) + + records = {record["workflow_id"]: record for record in receipt["workflow_records"]} + assert records[7] == { + "workflow_id": 7, + "path": repository_path, + "state": "active", + "source_kind": "repository", + "source_present": False, + } + assert records[8] == { + "workflow_id": 8, + "path": dynamic_path, + "state": "active", + "source_kind": "platform_dynamic", + "source_present": None, + } + assert receipt["active_absent_workflows"] == [ + {"workflow_id": 7, "path": repository_path, "state": "active"} + ] From 7aa1ab913748f99afc302e3d2404148fb7406c0d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 08:10:34 +0900 Subject: [PATCH 24/47] fix(workflows): classify dynamic registry identities safely --- workflow_registry_audit.py | 47 ++++++++++++++++++++++++++++---------- 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 79e2eb358..308c4fddc 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -2,9 +2,10 @@ The tool intentionally performs no workflow mutation. It binds one audit receipt to an exact protected commit, reads the repository Actions registry with complete -pagination, and reports active workflow identities whose exact source path is -absent from that protected tree. Those records are candidates for separate -operator review, not automatic disable decisions. +pagination, and reports active repository-backed workflow identities whose exact +source path is absent from that protected tree. GitHub-managed ``dynamic/`` +identities remain receipted separately and are never treated as deleted-YAML +candidates. All reported candidates still require separate operator review. """ from __future__ import annotations @@ -26,6 +27,7 @@ _PROTECTED_REF_RE = re.compile(r"^[A-Za-z0-9._/-]+$") _API_PATH_RE = re.compile(r"^/(?!/)[^\r\n]*$") _WORKFLOW_PREFIX = ".github/workflows/" +_DYNAMIC_WORKFLOW_PREFIX = "dynamic/" _GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 @@ -177,9 +179,11 @@ def audit_repository_workflows( ) -> dict[str, object]: """Return an exact-SHA read-only registry/source classification receipt. - Active identities absent from the exact protected tree are reported as - candidates only. The caller must separately prove whether a candidate has a - legitimate platform role before any future workflow-state mutation. Use + Active repository-backed identities absent from the exact protected tree are + reported as candidates only. GitHub-managed dynamic identities are retained + in the receipt with indeterminate protected-source presence and never become + orphan candidates. The caller must separately prove whether any candidate has + a legitimate platform role before future workflow-state mutation. Use :func:`audit_live_protected_ref_workflows` when live-ref movement must also invalidate the receipt. """ @@ -200,19 +204,30 @@ def audit_repository_workflows( classified_records: list[dict[str, object]] = [] active_absent: list[dict[str, object]] = [] for record in workflow_records: - source_present = record["path"] in protected_paths + path = str(record["path"]) + source_kind = ( + "repository" if path.startswith(_WORKFLOW_PREFIX) else "platform_dynamic" + ) + source_present: bool | None = ( + path in protected_paths if source_kind == "repository" else None + ) classified = { "workflow_id": record["workflow_id"], - "path": record["path"], + "path": path, "state": record["state"], + "source_kind": source_kind, "source_present": source_present, } classified_records.append(classified) - if record["state"] == "active" and not source_present: + if ( + record["state"] == "active" + and source_kind == "repository" + and source_present is False + ): active_absent.append( { "workflow_id": record["workflow_id"], - "path": record["path"], + "path": path, "state": record["state"], } ) @@ -390,7 +405,7 @@ def _registry_signature(records: list[dict[str, object]]) -> tuple[tuple[int, st def _validate_workflow_record(raw_record: object) -> dict[str, object]: - """Validate only identity/path/state fields needed for safe classification.""" + """Validate identity/path/state for repository and GitHub-managed records.""" if not isinstance(raw_record, dict): raise WorkflowRegistryAuditError("workflow registry record is invalid") workflow_id = raw_record.get("id") @@ -405,7 +420,15 @@ def _validate_workflow_record(raw_record: object) -> dict[str, object]: or not state ): raise WorkflowRegistryAuditError("workflow registry record is invalid") - if not path.startswith(_WORKFLOW_PREFIX) or "\\" in path or ".." in path.split("/"): + components = path.split("/") + if ( + "\\" in path + or any(component in {"", ".", ".."} for component in components) + or not ( + path.startswith(_WORKFLOW_PREFIX) + or path.startswith(_DYNAMIC_WORKFLOW_PREFIX) + ) + ): raise WorkflowRegistryAuditError("workflow registry record path is invalid") return {"workflow_id": workflow_id, "path": path, "state": state} From 3cd6896b78dd37e32a8aedd18e3f282fab31009f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 08:19:01 +0900 Subject: [PATCH 25/47] test(workflows): reproduce protected-ref namespace confusion --- .../test_workflow_registry_audit_ref_movement.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/test_workflow_registry_audit_ref_movement.py b/tests/test_workflow_registry_audit_ref_movement.py index 43df204ac..8f5d9cb22 100644 --- a/tests/test_workflow_registry_audit_ref_movement.py +++ b/tests/test_workflow_registry_audit_ref_movement.py @@ -72,6 +72,22 @@ def test_initial_protected_ref_mismatch_fails_before_tree_or_registry_reads() -> ) +@pytest.mark.parametrize("protected_ref", ["refs/heads/main", "heads/main"]) +def test_namespace_like_protected_ref_is_rejected_before_github_reads( + protected_ref: str, +) -> None: + """Git ref namespace prefixes cannot masquerade as ordinary branch names.""" + client = _FakeClient([]) + + with pytest.raises(WorkflowRegistryAuditError, match="safe branch name"): + audit_live_protected_ref_workflows( + repository_full_name=REPOSITORY, + protected_ref=protected_ref, + expected_protected_sha=PROTECTED_SHA, + client=client, + ) + + def test_slash_protected_ref_preserves_github_ref_path_shape() -> None: """A valid slash-bearing branch uses GitHub's documented heads/ path.""" protected_ref = "release/1.0" From 5faf4be3a341f9af4a9abd52a153b1096a91ea10 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 08:20:55 +0900 Subject: [PATCH 26/47] fix(workflows): reject protected-ref namespace prefixes --- workflow_registry_audit.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 308c4fddc..0de6e5cb9 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -460,6 +460,8 @@ def _validate_protected_ref(protected_ref: str) -> None: or protected_ref.endswith("/") or "//" in protected_ref or "@{" in protected_ref + or protected_ref.startswith("refs/") + or protected_ref.startswith("heads/") or any(component in {"", ".", ".."} for component in protected_ref.split("/")) ): raise WorkflowRegistryAuditError("protected_ref must be a safe branch name") From 7618fc4b77a55e6e57fc6fb4d3e8cf31679218a0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 11:08:16 +0900 Subject: [PATCH 27/47] test(workflows): reject nonfinite audit timeouts --- tests/test_workflow_registry_audit_timeout.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 tests/test_workflow_registry_audit_timeout.py diff --git a/tests/test_workflow_registry_audit_timeout.py b/tests/test_workflow_registry_audit_timeout.py new file mode 100644 index 000000000..5dfc49938 --- /dev/null +++ b/tests/test_workflow_registry_audit_timeout.py @@ -0,0 +1,16 @@ +"""Timeout-boundary regressions for the read-only workflow registry audit.""" + +from __future__ import annotations + +import math + +import pytest + +from workflow_registry_audit import GitHubReadClient, WorkflowRegistryAuditError + + +@pytest.mark.parametrize("timeout_seconds", [math.nan, math.inf]) +def test_nonfinite_timeout_is_rejected_before_transport(timeout_seconds: float) -> None: + """NaN and infinity cannot disable the client's finite timeout contract.""" + with pytest.raises(WorkflowRegistryAuditError, match="positive finite"): + GitHubReadClient(timeout_seconds=timeout_seconds) From 2f6788c279337826ce29a310615f244849abe5e7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 11:09:07 +0900 Subject: [PATCH 28/47] fix(workflows): require finite audit timeout --- workflow_registry_audit.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 0de6e5cb9..9d1bb85cf 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -13,6 +13,7 @@ import argparse import asyncio import json +import math import os import re import sys @@ -54,8 +55,8 @@ def __init__( timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS, ) -> None: """Configure fixed-origin read-only API access with a finite timeout.""" - if timeout_seconds <= 0: - raise WorkflowRegistryAuditError("GitHub audit timeout must be positive") + if not math.isfinite(timeout_seconds) or timeout_seconds <= 0: + raise WorkflowRegistryAuditError("GitHub audit timeout must be positive finite") self._token = token self._timeout_seconds = timeout_seconds From 842610c56af2708e1dc48f2deeead4e750e41f7e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 12:22:58 +0900 Subject: [PATCH 29/47] test(security): reject repository dot segments before transport --- ...flow_registry_audit_repository_selector.py | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 tests/test_workflow_registry_audit_repository_selector.py diff --git a/tests/test_workflow_registry_audit_repository_selector.py b/tests/test_workflow_registry_audit_repository_selector.py new file mode 100644 index 000000000..8368d49a9 --- /dev/null +++ b/tests/test_workflow_registry_audit_repository_selector.py @@ -0,0 +1,51 @@ +"""Regression tests for bounded workflow-audit repository selectors.""" + +from __future__ import annotations + +import pytest + +from workflow_registry_audit import ( + WorkflowRegistryAuditError, + audit_repository_workflows, +) + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" + + +class _NoReadClient: + """Reject every GitHub read so validation ordering is explicit.""" + + def __init__(self) -> None: + self.requested_paths: list[str] = [] + + def get_json(self, path: str) -> dict[str, object]: + """Record and reject transport that input validation should prevent.""" + self.requested_paths.append(path) + raise AssertionError(f"unexpected GitHub read: {path}") + + +@pytest.mark.parametrize( + "repository_full_name", + [ + "../repo", + "./repo", + "owner/..", + "owner/.", + ], +) +def test_repository_dot_segments_fail_before_github_read( + repository_full_name: str, +) -> None: + """Dot segments must not escape the fixed ``/repos/owner/name`` API namespace.""" + client = _NoReadClient() + + with pytest.raises(WorkflowRegistryAuditError, match="owner/name syntax"): + audit_repository_workflows( + repository_full_name=repository_full_name, + protected_sha=PROTECTED_SHA, + client=client, + captured_at="2026-08-15T00:00:00Z", + ) + + assert client.requested_paths == [] From bdb0fb33a7ad1162b7e575cff97ca7a44f5bd261 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 12:26:50 +0900 Subject: [PATCH 30/47] fix(security): reject repository dot segments --- workflow_registry_audit.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 9d1bb85cf..1f43b5fc4 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -443,7 +443,11 @@ def _require_nonnegative_int(value: object) -> int: def _validate_repository(repository_full_name: str) -> None: """Reject malformed repository selectors before any network read.""" - if not _REPOSITORY_RE.fullmatch(repository_full_name): + components = repository_full_name.split("/") + if ( + not _REPOSITORY_RE.fullmatch(repository_full_name) + or any(component in {".", ".."} for component in components) + ): raise WorkflowRegistryAuditError("repository must use owner/name syntax") @@ -524,4 +528,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 a50d8032adb4962d374d0813823e882bf68452d5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 12:28:54 +0900 Subject: [PATCH 31/47] test(audit): cover negative infinite timeout --- tests/test_workflow_registry_audit_timeout.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_workflow_registry_audit_timeout.py b/tests/test_workflow_registry_audit_timeout.py index 5dfc49938..0204418b0 100644 --- a/tests/test_workflow_registry_audit_timeout.py +++ b/tests/test_workflow_registry_audit_timeout.py @@ -9,7 +9,7 @@ from workflow_registry_audit import GitHubReadClient, WorkflowRegistryAuditError -@pytest.mark.parametrize("timeout_seconds", [math.nan, math.inf]) +@pytest.mark.parametrize("timeout_seconds", [math.nan, math.inf, -math.inf]) def test_nonfinite_timeout_is_rejected_before_transport(timeout_seconds: float) -> None: """NaN and infinity cannot disable the client's finite timeout contract.""" with pytest.raises(WorkflowRegistryAuditError, match="positive finite"): From 5c7c3e271323f945a6a3c0b400587365b3b98476 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:40:01 +0900 Subject: [PATCH 32/47] test(workflows): prove GitHub audit response is unbounded --- ..._workflow_registry_audit_response_bound.py | 106 ++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 tests/test_workflow_registry_audit_response_bound.py diff --git a/tests/test_workflow_registry_audit_response_bound.py b/tests/test_workflow_registry_audit_response_bound.py new file mode 100644 index 000000000..2fa236121 --- /dev/null +++ b/tests/test_workflow_registry_audit_response_bound.py @@ -0,0 +1,106 @@ +"""Regression tests for bounded GitHub API response reads in the audit tool.""" + +from __future__ import annotations + +import traceback + +import aiohttp +import pytest + +from workflow_registry_audit import GitHubReadClient, WorkflowRegistryAuditError + + +class _Content: + """Expose deterministic response chunks through aiohttp's streaming shape.""" + + def __init__(self, *chunks: bytes) -> None: + self._chunks = chunks + + async def iter_chunked(self, _size: int): + """Yield the configured chunks without buffering them into one body first.""" + for chunk in self._chunks: + yield chunk + + +class _Response: + """Represent one successful GitHub response with controllable body metadata.""" + + status = 200 + headers: dict[str, str] = {} + + def __init__(self, body: bytes, *, content_length: int | None) -> None: + self._body = body + self.content_length = content_length + self.content = _Content(body) + + async def __aenter__(self) -> "_Response": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + async def read(self) -> bytes: + """Model the unsafe whole-body path that the production client must avoid.""" + return self._body + + +class _Session: + """Return exactly one prepared response from the fixed-origin client session.""" + + response: _Response + + def __init__( + self, + *, + base_url: str, + headers: dict[str, str], + timeout: aiohttp.ClientTimeout, + ) -> None: + assert base_url == "https://api.github.com" + assert headers["Accept"] == "application/vnd.github+json" + assert timeout.total == 15.0 + + async def __aenter__(self) -> "_Session": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + def get(self, _path: str, *, allow_redirects: bool) -> _Response: + assert allow_redirects is False + return self.response + + +def test_declared_oversize_response_fails_before_whole_body_read( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A declared oversized response must fail before materializing its body.""" + _Session.response = _Response(b"{}", content_length=32 * 1024 * 1024) + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) + client = GitHubReadClient(token="bounded-token") + + with pytest.raises(WorkflowRegistryAuditError, match="response exceeded byte limit"): + client.get_json("/repos/ContextualWisdomLab/pg-llm-batch/actions/workflows") + + +def test_chunked_oversize_response_fails_with_bounded_nonsecret_evidence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Chunked responses without Content-Length cannot bypass the body budget.""" + secret_sentinel = "SECRET_RESPONSE_BODY_MUST_NOT_ESCAPE" + body = ('{"value":"' + secret_sentinel + '"}').encode() + _Session.response = _Response(body, content_length=None) + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) + monkeypatch.setattr("workflow_registry_audit._MAX_RESPONSE_BYTES", 8, raising=False) + client = GitHubReadClient(token="bounded-token") + + try: + client.get_json("/repos/ContextualWisdomLab/pg-llm-batch/actions/workflows") + except WorkflowRegistryAuditError as exc: + rendered = "".join(traceback.format_exception(exc)) + assert str(exc) == "GitHub workflow audit response exceeded byte limit" + assert secret_sentinel not in rendered + assert "bounded-token" not in rendered + assert exc.__cause__ is None + else: + raise AssertionError("expected WorkflowRegistryAuditError") From 8770ad8526b4135e0aa19135f13866e34de1b5df Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:42:59 +0900 Subject: [PATCH 33/47] fix(workflows): bound GitHub audit response memory --- workflow_registry_audit.py | 54 +++++++++++++++++++++++++++++++++++--- 1 file changed, 51 insertions(+), 3 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 1f43b5fc4..ac9e69f6d 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -32,6 +32,8 @@ _GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 +_MAX_RESPONSE_BYTES = 16 * 1024 * 1024 +_RESPONSE_CHUNK_BYTES = 64 * 1024 class WorkflowRegistryAuditError(RuntimeError): @@ -75,7 +77,7 @@ def get_json(self, path: str) -> dict[str, object]: return asyncio.run(self._get_json(path)) async def _get_json(self, path: str) -> dict[str, object]: - """Perform one fixed-origin aiohttp GET with redirects disabled.""" + """Perform one fixed-origin aiohttp GET with redirects and body growth bounded.""" timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) try: async with aiohttp.ClientSession( @@ -97,7 +99,7 @@ async def _get_json(self, path: str) -> dict[str, object]: raise WorkflowRegistryAuditError( "GitHub workflow audit read failed" ) - raw = await response.read() + raw = await self._read_bounded_response(response) payload = json.loads(raw.decode("utf-8")) except WorkflowRegistryAuditError: raise @@ -112,6 +114,52 @@ async def _get_json(self, path: str) -> dict[str, object]: raise WorkflowRegistryAuditError("GitHub workflow audit response is invalid") return payload + async def _read_bounded_response(self, response: object) -> bytes: + """Stream one decoded HTTP body while enforcing a fixed memory budget. + + GitHub documents a 7 MB recursive-tree response ceiling. The 16 MiB + budget deliberately leaves protocol/envelope headroom while preventing + a malformed or unexpectedly large authenticated response from being + materialized without bound. The limit is also enforced for chunked + responses that omit ``Content-Length``. + """ + declared_value = getattr(response, "content_length", None) + declared_bytes = ( + declared_value + if isinstance(declared_value, int) + and not isinstance(declared_value, bool) + and declared_value >= 0 + else None + ) + if declared_bytes is not None and declared_bytes > _MAX_RESPONSE_BYTES: + raise WorkflowRegistryAuditError( + "GitHub workflow audit response exceeded byte limit" + ) + + stream = getattr(response, "content", None) + iterator = getattr(stream, "iter_chunked", None) + if not callable(iterator): + raise WorkflowRegistryAuditError( + "GitHub workflow audit response stream is invalid" + ) + + payload = bytearray() + async for chunk in iterator(_RESPONSE_CHUNK_BYTES): + if isinstance(chunk, memoryview): + chunk_bytes = chunk.tobytes() + elif isinstance(chunk, (bytes, bytearray)): + chunk_bytes = bytes(chunk) + else: + raise WorkflowRegistryAuditError( + "GitHub workflow audit response stream is invalid" + ) + if len(payload) + len(chunk_bytes) > _MAX_RESPONSE_BYTES: + raise WorkflowRegistryAuditError( + "GitHub workflow audit response exceeded byte limit" + ) + payload.extend(chunk_bytes) + return bytes(payload) + def _headers(self) -> dict[str, str]: """Build fixed GitHub headers without exposing the token in output.""" headers = { @@ -528,4 +576,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 c54978c27212ff5945ccf5a9ad502c2791a8c7c3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:44:27 +0900 Subject: [PATCH 34/47] test(workflows): exercise bounded aiohttp response stream --- tests/test_workflow_registry_audit.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py index 9f8ad994d..5917d5839 100644 --- a/tests/test_workflow_registry_audit.py +++ b/tests/test_workflow_registry_audit.py @@ -308,8 +308,14 @@ def test_authenticated_reads_use_fixed_aiohttp_origin_and_path_only( observed_requests: list[tuple[str, bool]] = [] closed: list[bool] = [] + class _Content: + async def iter_chunked(self, _size: int): + yield b"{}" + class _Response: status = 200 + content_length = 2 + content = _Content() async def __aenter__(self) -> "_Response": return self @@ -317,9 +323,6 @@ async def __aenter__(self) -> "_Response": async def __aexit__(self, *_args: object) -> None: return None - async def read(self) -> bytes: - return b"{}" - class _Session: def __init__( self, From 09fc0deecae11683d4eb35d86c1c240388d68f4d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:50:40 +0900 Subject: [PATCH 35/47] test(workflows): prove recursive JSON fails closed --- ..._workflow_registry_audit_response_bound.py | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/test_workflow_registry_audit_response_bound.py b/tests/test_workflow_registry_audit_response_bound.py index 2fa236121..06cfd7d70 100644 --- a/tests/test_workflow_registry_audit_response_bound.py +++ b/tests/test_workflow_registry_audit_response_bound.py @@ -104,3 +104,30 @@ def test_chunked_oversize_response_fails_with_bounded_nonsecret_evidence( assert exc.__cause__ is None else: raise AssertionError("expected WorkflowRegistryAuditError") + + +def test_recursive_json_decoder_failure_is_normalized_without_diagnostics( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A recursion-limited JSON decoder failure remains a fixed audit error.""" + secret_sentinel = "SECRET_RECURSION_DIAGNOSTIC_MUST_NOT_ESCAPE" + _Session.response = _Response(b"{}", content_length=2) + monkeypatch.setattr("workflow_registry_audit.aiohttp.ClientSession", _Session) + + def _raise_recursion(_content: str) -> object: + raise RecursionError(secret_sentinel) + + monkeypatch.setattr("workflow_registry_audit.json.loads", _raise_recursion) + client = GitHubReadClient(token="bounded-token") + + try: + client.get_json("/repos/ContextualWisdomLab/pg-llm-batch/actions/workflows") + except WorkflowRegistryAuditError as exc: + rendered = "".join(traceback.format_exception(exc)) + assert str(exc) == "GitHub workflow audit read failed" + assert secret_sentinel not in rendered + assert "bounded-token" not in rendered + assert exc.__cause__ is None + assert exc.__suppress_context__ is True + else: + raise AssertionError("expected WorkflowRegistryAuditError") From 5defb013a7bc66336e15c4f90b0727c644a33c62 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:54:02 +0900 Subject: [PATCH 36/47] fix(workflows): normalize recursive JSON failure --- workflow_registry_audit.py | 1 + 1 file changed, 1 insertion(+) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index ac9e69f6d..f5e498103 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -108,6 +108,7 @@ async def _get_json(self, path: str) -> dict[str, object]: asyncio.TimeoutError, UnicodeDecodeError, json.JSONDecodeError, + RecursionError, ): raise WorkflowRegistryAuditError("GitHub workflow audit read failed") from None if not isinstance(payload, dict): From 156f1344858001c356debfd17efb5cbaa5def476 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 22:58:47 +0900 Subject: [PATCH 37/47] test(workflows): bound registry pagination cardinality --- ...orkflow_registry_audit_pagination_bound.py | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 tests/test_workflow_registry_audit_pagination_bound.py diff --git a/tests/test_workflow_registry_audit_pagination_bound.py b/tests/test_workflow_registry_audit_pagination_bound.py new file mode 100644 index 000000000..57fb577f2 --- /dev/null +++ b/tests/test_workflow_registry_audit_pagination_bound.py @@ -0,0 +1,45 @@ +"""Regression test for globally bounded workflow-registry pagination.""" + +from __future__ import annotations + +import pytest + +from workflow_registry_audit import WorkflowRegistryAuditError, _read_registry + + +class _OversizedRegistryClient: + """Expose an implausibly large registry cardinality without real network work.""" + + def __init__(self) -> None: + self.calls = 0 + + def get_json(self, _path: str) -> dict[str, object]: + """Return one full page while claiming an excessive registry size.""" + self.calls += 1 + return { + "total_count": 10_001, + "workflows": [ + { + "id": workflow_id, + "path": f".github/workflows/workflow-{workflow_id}.yml", + "state": "active", + } + for workflow_id in range(1, 101) + ], + } + + +def test_excessive_registry_cardinality_fails_before_second_page() -> None: + """One audit must cap total API work instead of trusting unbounded cardinality.""" + client = _OversizedRegistryClient() + + with pytest.raises( + WorkflowRegistryAuditError, + match="workflow registry exceeds supported workflow limit", + ): + _read_registry( + repository_full_name="ContextualWisdomLab/pg-llm-batch", + client=client, + ) + + assert client.calls == 1 From 77732cd2b0d61683e749af70d87397f269384860 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 23:00:41 +0900 Subject: [PATCH 38/47] fix(workflows): bound registry audit cardinality --- workflow_registry_audit.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index f5e498103..151d49c93 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -32,6 +32,7 @@ _GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 +_MAX_REGISTRY_WORKFLOWS = 10_000 _MAX_RESPONSE_BYTES = 16 * 1024 * 1024 _RESPONSE_CHUNK_BYTES = 64 * 1024 @@ -412,6 +413,10 @@ def _read_registry_pass( ) pages_scanned += 1 total_count = _require_nonnegative_int(payload.get("total_count")) + if total_count > _MAX_REGISTRY_WORKFLOWS: + raise WorkflowRegistryAuditError( + "workflow registry exceeds supported workflow limit" + ) if expected_total is None: expected_total = total_count elif total_count != expected_total: @@ -577,4 +582,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 150b71dfccfa2cfeea1ce9bb9f3b2631a37b1479 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 00:14:58 +0900 Subject: [PATCH 39/47] test(workflows): reject unknown workflow registry state --- ...orkflow_registry_audit_state_validation.py | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 tests/test_workflow_registry_audit_state_validation.py diff --git a/tests/test_workflow_registry_audit_state_validation.py b/tests/test_workflow_registry_audit_state_validation.py new file mode 100644 index 000000000..dd5dc84d3 --- /dev/null +++ b/tests/test_workflow_registry_audit_state_validation.py @@ -0,0 +1,56 @@ +"""Regression tests for fail-closed GitHub workflow-state evidence.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from workflow_registry_audit import WorkflowRegistryAuditError, audit_repository_workflows + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" +REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +class _FakeClient: + """Return exact GitHub commit, tree, and registry payloads in sequence.""" + + def __init__(self) -> None: + self.paths: list[str] = [] + + def get_json(self, path: str) -> dict[str, Any]: + """Serve one deterministic response for each expected audit read.""" + self.paths.append(path) + if path.endswith(f"/git/commits/{PROTECTED_SHA}"): + return {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}} + if path.endswith(f"/git/trees/{TREE_SHA}?recursive=1"): + return {"sha": TREE_SHA, "truncated": False, "tree": []} + if path.endswith("/actions/workflows?per_page=100&page=1"): + return { + "total_count": 1, + "workflows": [ + { + "id": 17, + "path": ".github/workflows/removed.yml", + "state": "future_unknown_state", + } + ], + } + raise AssertionError(f"unexpected path: {path}") + + +def test_unknown_workflow_state_fails_closed_instead_of_suppressing_candidate() -> None: + """Unrecognized registry state must not silently become non-active evidence.""" + client = _FakeClient() + + with pytest.raises(WorkflowRegistryAuditError, match="record is invalid"): + audit_repository_workflows( + repository_full_name=REPOSITORY, + protected_sha=PROTECTED_SHA, + client=client, + captured_at="2026-08-15T00:00:00Z", + ) + + assert len(client.paths) == 3 From 6a48171d983116877a2b348ddb4ca3519dac69f2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 00:17:08 +0900 Subject: [PATCH 40/47] fix(workflows): reject unknown workflow registry states --- workflow_registry_audit.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 151d49c93..1c79c3825 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -29,6 +29,15 @@ _API_PATH_RE = re.compile(r"^/(?!/)[^\r\n]*$") _WORKFLOW_PREFIX = ".github/workflows/" _DYNAMIC_WORKFLOW_PREFIX = "dynamic/" +_WORKFLOW_STATES = frozenset( + { + "active", + "deleted", + "disabled_fork", + "disabled_inactivity", + "disabled_manually", + } +) _GITHUB_API_URL = "https://api.github.com" _DEFAULT_TIMEOUT_SECONDS = 15.0 _PAGE_SIZE = 100 @@ -471,8 +480,7 @@ def _validate_workflow_record(raw_record: object) -> dict[str, object]: or not isinstance(workflow_id, int) or workflow_id <= 0 or not isinstance(path, str) - or not isinstance(state, str) - or not state + or state not in _WORKFLOW_STATES ): raise WorkflowRegistryAuditError("workflow registry record is invalid") components = path.split("/") From ca65227da7bae24f703d46845b22b7ccae051f71 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 19:24:52 +0900 Subject: [PATCH 41/47] test(workflows): require exact registry primitive types --- ...est_workflow_registry_audit_exact_types.py | 85 +++++++++++++++++++ 1 file changed, 85 insertions(+) create mode 100644 tests/test_workflow_registry_audit_exact_types.py diff --git a/tests/test_workflow_registry_audit_exact_types.py b/tests/test_workflow_registry_audit_exact_types.py new file mode 100644 index 000000000..73f1c3f68 --- /dev/null +++ b/tests/test_workflow_registry_audit_exact_types.py @@ -0,0 +1,85 @@ +"""Hostile primitive-subclass regressions for workflow registry evidence.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from workflow_registry_audit import ( + WorkflowRegistryAuditError, + _validate_workflow_record, +) + +_SECRET_SENTINEL = "SECRET-SENTINEL hostile registry primitive" + + +class _HostileWorkflowId(int): + """Execute caller-controlled code when numeric range validation runs.""" + + def __gt__(self, _other: object) -> bool: + """Raise instead of supplying a trustworthy positive-ID comparison.""" + raise RuntimeError(_SECRET_SENTINEL) + + +class _HostileWorkflowPath(str): + """Execute caller-controlled code when path components are inspected.""" + + def split(self, *_args: object, **_kwargs: object) -> list[str]: + """Raise instead of supplying trustworthy path components.""" + raise RuntimeError(_SECRET_SENTINEL) + + +class _HostileWorkflowState(str): + """Execute caller-controlled code during finite-state membership checks.""" + + def __hash__(self) -> int: + """Raise instead of supplying a trustworthy state hash.""" + raise RuntimeError(_SECRET_SENTINEL) + + +@pytest.mark.parametrize( + "state", + [[], {}, {"active"}], +) +def test_unhashable_workflow_state_is_bounded_invalid_evidence(state: Any) -> None: + """Malformed state containers must not escape as raw ``TypeError``.""" + with pytest.raises(WorkflowRegistryAuditError, match="record is invalid") as caught: + _validate_workflow_record( + { + "id": 1, + "path": ".github/workflows/ci.yml", + "state": state, + } + ) + + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("id", _HostileWorkflowId(1)), + ("path", _HostileWorkflowPath(".github/workflows/ci.yml")), + ("state", _HostileWorkflowState("active")), + ], +) +def test_primitive_subclasses_are_rejected_before_custom_code( + field: str, + value: object, +) -> None: + """Registry evidence must use exact decoder primitive types only.""" + record: dict[str, object] = { + "id": 1, + "path": ".github/workflows/ci.yml", + "state": "active", + } + record[field] = value + + with pytest.raises(WorkflowRegistryAuditError, match="record is invalid") as caught: + _validate_workflow_record(record) + + assert _SECRET_SENTINEL not in str(caught.value) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None From 6140e45b585141ac511a2146baeac2946f5e4b24 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 19:28:02 +0900 Subject: [PATCH 42/47] test(workflows): reject hostile registry mappings --- ...est_workflow_registry_audit_exact_types.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/test_workflow_registry_audit_exact_types.py b/tests/test_workflow_registry_audit_exact_types.py index 73f1c3f68..fcfac0b70 100644 --- a/tests/test_workflow_registry_audit_exact_types.py +++ b/tests/test_workflow_registry_audit_exact_types.py @@ -14,6 +14,14 @@ _SECRET_SENTINEL = "SECRET-SENTINEL hostile registry primitive" +class _HostileWorkflowRecord(dict[str, object]): + """Execute caller-controlled code when record members are retrieved.""" + + def get(self, _key: str, _default: object = None) -> object: + """Raise instead of supplying trustworthy decoded JSON members.""" + raise RuntimeError(_SECRET_SENTINEL) + + class _HostileWorkflowId(int): """Execute caller-controlled code when numeric range validation runs.""" @@ -57,6 +65,22 @@ def test_unhashable_workflow_state_is_bounded_invalid_evidence(state: Any) -> No assert caught.value.__context__ is None +def test_record_subclass_is_rejected_before_member_access() -> None: + """Registry parsing must not execute caller-controlled mapping methods.""" + with pytest.raises(WorkflowRegistryAuditError, match="record is invalid") as caught: + _validate_workflow_record( + _HostileWorkflowRecord( + id=1, + path=".github/workflows/ci.yml", + state="active", + ) + ) + + assert _SECRET_SENTINEL not in str(caught.value) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + @pytest.mark.parametrize( ("field", "value"), [ From 124e5b3a4f893a9c7a10dcb26dd585e70b7473ff Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 19:31:05 +0900 Subject: [PATCH 43/47] fix(workflows): bound malformed registry primitive types --- workflow_registry_audit.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index 1c79c3825..b52b948fc 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -469,17 +469,17 @@ def _registry_signature(records: list[dict[str, object]]) -> tuple[tuple[int, st def _validate_workflow_record(raw_record: object) -> dict[str, object]: - """Validate identity/path/state for repository and GitHub-managed records.""" - if not isinstance(raw_record, dict): + """Validate exact decoder primitive types for one workflow registry record.""" + if type(raw_record) is not dict: raise WorkflowRegistryAuditError("workflow registry record is invalid") workflow_id = raw_record.get("id") path = raw_record.get("path") state = raw_record.get("state") if ( - isinstance(workflow_id, bool) - or not isinstance(workflow_id, int) + type(workflow_id) is not int or workflow_id <= 0 - or not isinstance(path, str) + or type(path) is not str + or type(state) is not str or state not in _WORKFLOW_STATES ): raise WorkflowRegistryAuditError("workflow registry record is invalid") From 7a71400ffad8ffbe33fbeae85f583686cad08359 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 21:38:19 +0900 Subject: [PATCH 44/47] test(workflows): reject hostile audit boundary subclasses --- ...flow_registry_audit_exact_payload_types.py | 164 ++++++++++++++++++ 1 file changed, 164 insertions(+) create mode 100644 tests/test_workflow_registry_audit_exact_payload_types.py diff --git a/tests/test_workflow_registry_audit_exact_payload_types.py b/tests/test_workflow_registry_audit_exact_payload_types.py new file mode 100644 index 000000000..4f5024ce6 --- /dev/null +++ b/tests/test_workflow_registry_audit_exact_payload_types.py @@ -0,0 +1,164 @@ +"""Regression tests for exact primitive workflow-audit trust boundaries.""" + +from __future__ import annotations + +import pytest + +from workflow_registry_audit import ( + WorkflowRegistryAuditError, + audit_live_protected_ref_workflows, + audit_repository_workflows, +) + + +_PROTECTED_SHA = "a" * 40 +_TREE_SHA = "b" * 40 +_REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +class _NoReadClient: + def get_json(self, _path: str) -> dict[str, object]: + raise AssertionError("invalid authority must fail before GitHub reads") + + +class _StaticClient: + def __init__(self, responses: dict[str, object]) -> None: + self._responses = responses + + def get_json(self, path: str) -> dict[str, object]: + return self._responses[path] # type: ignore[return-value] + + +def _valid_responses() -> dict[str, object]: + return { + f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}": { + "sha": _PROTECTED_SHA, + "tree": {"sha": _TREE_SHA}, + }, + f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1": { + "sha": _TREE_SHA, + "truncated": False, + "tree": [], + }, + f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + + +class _HostileRepository(str): + def split(self, *_args, **_kwargs): + raise AssertionError("hostile repository split executed") + + +class _HostileProtectedSha(str): + def lower(self): + raise AssertionError("hostile SHA lower executed") + + +class _HostileProtectedRef(str): + def startswith(self, *_args, **_kwargs): + raise AssertionError("hostile ref startswith executed") + + +class _HostileMapping(dict): + def get(self, *_args, **_kwargs): + raise AssertionError("hostile mapping get executed") + + +class _HostileInt(int): + def __lt__(self, _other): + raise AssertionError("hostile integer comparison executed") + + +class _HostileList(list): + def __len__(self): + raise AssertionError("hostile list length executed") + + +@pytest.mark.parametrize( + ("repository_full_name", "protected_sha"), + [ + (_HostileRepository(_REPOSITORY), _PROTECTED_SHA), + (_REPOSITORY, _HostileProtectedSha(_PROTECTED_SHA)), + ], +) +def test_audit_rejects_hostile_authority_subclasses_before_client_access( + repository_full_name, protected_sha +): + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=repository_full_name, + protected_sha=protected_sha, + client=_NoReadClient(), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_live_audit_rejects_hostile_ref_subclass_before_client_access(): + with pytest.raises(WorkflowRegistryAuditError): + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref=_HostileProtectedRef("main"), + expected_protected_sha=_PROTECTED_SHA, + client=_NoReadClient(), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_top_level_commit_mapping_without_member_access(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = _HostileMapping() + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_nested_commit_tree_without_member_access(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { + "sha": _PROTECTED_SHA, + "tree": _HostileMapping(), + } + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_registry_total_count_before_comparison(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = { + "total_count": _HostileInt(0), + "workflows": [], + } + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_registry_list_before_shape_operations(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = { + "total_count": 0, + "workflows": _HostileList(), + } + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) From b336272f076df7afb874c52303d695a371101f95 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 21:40:49 +0900 Subject: [PATCH 45/47] revert(workflows): restore pre-run audit branch after blocked fix --- ...flow_registry_audit_exact_payload_types.py | 164 ------------------ 1 file changed, 164 deletions(-) delete mode 100644 tests/test_workflow_registry_audit_exact_payload_types.py diff --git a/tests/test_workflow_registry_audit_exact_payload_types.py b/tests/test_workflow_registry_audit_exact_payload_types.py deleted file mode 100644 index 4f5024ce6..000000000 --- a/tests/test_workflow_registry_audit_exact_payload_types.py +++ /dev/null @@ -1,164 +0,0 @@ -"""Regression tests for exact primitive workflow-audit trust boundaries.""" - -from __future__ import annotations - -import pytest - -from workflow_registry_audit import ( - WorkflowRegistryAuditError, - audit_live_protected_ref_workflows, - audit_repository_workflows, -) - - -_PROTECTED_SHA = "a" * 40 -_TREE_SHA = "b" * 40 -_REPOSITORY = "ContextualWisdomLab/pg-llm-batch" - - -class _NoReadClient: - def get_json(self, _path: str) -> dict[str, object]: - raise AssertionError("invalid authority must fail before GitHub reads") - - -class _StaticClient: - def __init__(self, responses: dict[str, object]) -> None: - self._responses = responses - - def get_json(self, path: str) -> dict[str, object]: - return self._responses[path] # type: ignore[return-value] - - -def _valid_responses() -> dict[str, object]: - return { - f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}": { - "sha": _PROTECTED_SHA, - "tree": {"sha": _TREE_SHA}, - }, - f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1": { - "sha": _TREE_SHA, - "truncated": False, - "tree": [], - }, - f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1": { - "total_count": 0, - "workflows": [], - }, - } - - -class _HostileRepository(str): - def split(self, *_args, **_kwargs): - raise AssertionError("hostile repository split executed") - - -class _HostileProtectedSha(str): - def lower(self): - raise AssertionError("hostile SHA lower executed") - - -class _HostileProtectedRef(str): - def startswith(self, *_args, **_kwargs): - raise AssertionError("hostile ref startswith executed") - - -class _HostileMapping(dict): - def get(self, *_args, **_kwargs): - raise AssertionError("hostile mapping get executed") - - -class _HostileInt(int): - def __lt__(self, _other): - raise AssertionError("hostile integer comparison executed") - - -class _HostileList(list): - def __len__(self): - raise AssertionError("hostile list length executed") - - -@pytest.mark.parametrize( - ("repository_full_name", "protected_sha"), - [ - (_HostileRepository(_REPOSITORY), _PROTECTED_SHA), - (_REPOSITORY, _HostileProtectedSha(_PROTECTED_SHA)), - ], -) -def test_audit_rejects_hostile_authority_subclasses_before_client_access( - repository_full_name, protected_sha -): - with pytest.raises(WorkflowRegistryAuditError): - audit_repository_workflows( - repository_full_name=repository_full_name, - protected_sha=protected_sha, - client=_NoReadClient(), - captured_at="2026-08-16T00:00:00Z", - ) - - -def test_live_audit_rejects_hostile_ref_subclass_before_client_access(): - with pytest.raises(WorkflowRegistryAuditError): - audit_live_protected_ref_workflows( - repository_full_name=_REPOSITORY, - protected_ref=_HostileProtectedRef("main"), - expected_protected_sha=_PROTECTED_SHA, - client=_NoReadClient(), - captured_at="2026-08-16T00:00:00Z", - ) - - -def test_audit_rejects_hostile_top_level_commit_mapping_without_member_access(): - responses = _valid_responses() - responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = _HostileMapping() - with pytest.raises(WorkflowRegistryAuditError): - audit_repository_workflows( - repository_full_name=_REPOSITORY, - protected_sha=_PROTECTED_SHA, - client=_StaticClient(responses), - captured_at="2026-08-16T00:00:00Z", - ) - - -def test_audit_rejects_hostile_nested_commit_tree_without_member_access(): - responses = _valid_responses() - responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { - "sha": _PROTECTED_SHA, - "tree": _HostileMapping(), - } - with pytest.raises(WorkflowRegistryAuditError): - audit_repository_workflows( - repository_full_name=_REPOSITORY, - protected_sha=_PROTECTED_SHA, - client=_StaticClient(responses), - captured_at="2026-08-16T00:00:00Z", - ) - - -def test_audit_rejects_hostile_registry_total_count_before_comparison(): - responses = _valid_responses() - responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = { - "total_count": _HostileInt(0), - "workflows": [], - } - with pytest.raises(WorkflowRegistryAuditError): - audit_repository_workflows( - repository_full_name=_REPOSITORY, - protected_sha=_PROTECTED_SHA, - client=_StaticClient(responses), - captured_at="2026-08-16T00:00:00Z", - ) - - -def test_audit_rejects_hostile_registry_list_before_shape_operations(): - responses = _valid_responses() - responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = { - "total_count": 0, - "workflows": _HostileList(), - } - with pytest.raises(WorkflowRegistryAuditError): - audit_repository_workflows( - repository_full_name=_REPOSITORY, - protected_sha=_PROTECTED_SHA, - client=_StaticClient(responses), - captured_at="2026-08-16T00:00:00Z", - ) From 4e87d9e630afc9426cb45e081c776d3094fff43b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 23:30:34 +0900 Subject: [PATCH 46/47] test(workflows): reject hostile audit boundary subclasses --- ...flow_registry_audit_exact_payload_types.py | 164 ++++++++++++++++++ 1 file changed, 164 insertions(+) create mode 100644 tests/test_workflow_registry_audit_exact_payload_types.py diff --git a/tests/test_workflow_registry_audit_exact_payload_types.py b/tests/test_workflow_registry_audit_exact_payload_types.py new file mode 100644 index 000000000..4f5024ce6 --- /dev/null +++ b/tests/test_workflow_registry_audit_exact_payload_types.py @@ -0,0 +1,164 @@ +"""Regression tests for exact primitive workflow-audit trust boundaries.""" + +from __future__ import annotations + +import pytest + +from workflow_registry_audit import ( + WorkflowRegistryAuditError, + audit_live_protected_ref_workflows, + audit_repository_workflows, +) + + +_PROTECTED_SHA = "a" * 40 +_TREE_SHA = "b" * 40 +_REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +class _NoReadClient: + def get_json(self, _path: str) -> dict[str, object]: + raise AssertionError("invalid authority must fail before GitHub reads") + + +class _StaticClient: + def __init__(self, responses: dict[str, object]) -> None: + self._responses = responses + + def get_json(self, path: str) -> dict[str, object]: + return self._responses[path] # type: ignore[return-value] + + +def _valid_responses() -> dict[str, object]: + return { + f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}": { + "sha": _PROTECTED_SHA, + "tree": {"sha": _TREE_SHA}, + }, + f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1": { + "sha": _TREE_SHA, + "truncated": False, + "tree": [], + }, + f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + + +class _HostileRepository(str): + def split(self, *_args, **_kwargs): + raise AssertionError("hostile repository split executed") + + +class _HostileProtectedSha(str): + def lower(self): + raise AssertionError("hostile SHA lower executed") + + +class _HostileProtectedRef(str): + def startswith(self, *_args, **_kwargs): + raise AssertionError("hostile ref startswith executed") + + +class _HostileMapping(dict): + def get(self, *_args, **_kwargs): + raise AssertionError("hostile mapping get executed") + + +class _HostileInt(int): + def __lt__(self, _other): + raise AssertionError("hostile integer comparison executed") + + +class _HostileList(list): + def __len__(self): + raise AssertionError("hostile list length executed") + + +@pytest.mark.parametrize( + ("repository_full_name", "protected_sha"), + [ + (_HostileRepository(_REPOSITORY), _PROTECTED_SHA), + (_REPOSITORY, _HostileProtectedSha(_PROTECTED_SHA)), + ], +) +def test_audit_rejects_hostile_authority_subclasses_before_client_access( + repository_full_name, protected_sha +): + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=repository_full_name, + protected_sha=protected_sha, + client=_NoReadClient(), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_live_audit_rejects_hostile_ref_subclass_before_client_access(): + with pytest.raises(WorkflowRegistryAuditError): + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref=_HostileProtectedRef("main"), + expected_protected_sha=_PROTECTED_SHA, + client=_NoReadClient(), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_top_level_commit_mapping_without_member_access(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = _HostileMapping() + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_nested_commit_tree_without_member_access(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { + "sha": _PROTECTED_SHA, + "tree": _HostileMapping(), + } + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_registry_total_count_before_comparison(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = { + "total_count": _HostileInt(0), + "workflows": [], + } + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_audit_rejects_hostile_registry_list_before_shape_operations(): + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = { + "total_count": 0, + "workflows": _HostileList(), + } + with pytest.raises(WorkflowRegistryAuditError): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) From 1127c75653b7b4d79db1c4321741cd2149b5bbb6 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 23:33:04 +0900 Subject: [PATCH 47/47] fix(workflows): reject hostile audit boundary subclasses --- workflow_registry_audit.py | 47 +++++++++++++++++++++++--------------- 1 file changed, 28 insertions(+), 19 deletions(-) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py index b52b948fc..88667189f 100644 --- a/workflow_registry_audit.py +++ b/workflow_registry_audit.py @@ -121,7 +121,7 @@ async def _get_json(self, path: str) -> dict[str, object]: RecursionError, ): raise WorkflowRegistryAuditError("GitHub workflow audit read failed") from None - if not isinstance(payload, dict): + if type(payload) is not dict: raise WorkflowRegistryAuditError("GitHub workflow audit response is invalid") return payload @@ -137,9 +137,7 @@ async def _read_bounded_response(self, response: object) -> bytes: declared_value = getattr(response, "content_length", None) declared_bytes = ( declared_value - if isinstance(declared_value, int) - and not isinstance(declared_value, bool) - and declared_value >= 0 + if type(declared_value) is int and declared_value >= 0 else None ) if declared_bytes is not None and declared_bytes > _MAX_RESPONSE_BYTES: @@ -316,14 +314,19 @@ def _read_protected_ref_sha( payload = client.get_json( f"/repos/{repository_full_name}/git/ref/heads/{encoded_ref}" ) - if payload.get("ref") != f"refs/heads/{protected_ref}": + if type(payload) is not dict or payload.get("ref") != f"refs/heads/{protected_ref}": raise WorkflowRegistryAuditError("protected ref response is invalid") ref_object = payload.get("object") - if not isinstance(ref_object, dict): + if type(ref_object) is not dict: raise WorkflowRegistryAuditError("protected ref response is invalid") sha = ref_object.get("sha") object_type = ref_object.get("type") - if not isinstance(sha, str) or not _SHA_RE.fullmatch(sha) or object_type != "commit": + if ( + type(sha) is not str + or not _SHA_RE.fullmatch(sha) + or type(object_type) is not str + or object_type != "commit" + ): raise WorkflowRegistryAuditError("protected ref response is invalid") return sha.lower() @@ -338,34 +341,34 @@ def _read_protected_workflow_paths( commit_payload = client.get_json( f"/repos/{repository_full_name}/git/commits/{protected_sha}" ) - if commit_payload.get("sha") != protected_sha: + if type(commit_payload) is not dict or commit_payload.get("sha") != protected_sha: raise WorkflowRegistryAuditError("protected commit response is invalid") commit_tree = commit_payload.get("tree") - if not isinstance(commit_tree, dict): + if type(commit_tree) is not dict: raise WorkflowRegistryAuditError("protected commit response is invalid") tree_sha = commit_tree.get("sha") - if not isinstance(tree_sha, str) or not _SHA_RE.fullmatch(tree_sha): + if type(tree_sha) is not str or not _SHA_RE.fullmatch(tree_sha): raise WorkflowRegistryAuditError("protected commit response is invalid") tree_sha = tree_sha.lower() payload = client.get_json( f"/repos/{repository_full_name}/git/trees/{tree_sha}?recursive=1" ) - if payload.get("sha") != tree_sha: + if type(payload) is not dict or payload.get("sha") != tree_sha: raise WorkflowRegistryAuditError("protected tree SHA does not match commit tree SHA") if payload.get("truncated") is not False: raise WorkflowRegistryAuditError("protected tree is truncated") tree = payload.get("tree") - if not isinstance(tree, list): + if type(tree) is not list: raise WorkflowRegistryAuditError("protected tree response is invalid") paths: set[str] = set() for entry in tree: - if not isinstance(entry, dict): + if type(entry) is not dict: raise WorkflowRegistryAuditError("protected tree response is invalid") path = entry.get("path") entry_type = entry.get("type") - if not isinstance(path, str) or not isinstance(entry_type, str): + if type(path) is not str or type(entry_type) is not str: raise WorkflowRegistryAuditError("protected tree response is invalid") if entry_type == "blob" and path.startswith(_WORKFLOW_PREFIX): paths.add(path) @@ -420,6 +423,8 @@ def _read_registry_pass( f"/repos/{repository_full_name}/actions/workflows" f"?per_page={_PAGE_SIZE}&page={page}" ) + if type(payload) is not dict: + raise WorkflowRegistryAuditError("workflow registry response is invalid") pages_scanned += 1 total_count = _require_nonnegative_int(payload.get("total_count")) if total_count > _MAX_REGISTRY_WORKFLOWS: @@ -432,7 +437,7 @@ def _read_registry_pass( raise WorkflowRegistryAuditError("workflow registry changed during audit") workflows = payload.get("workflows") - if not isinstance(workflows, list): + if type(workflows) is not list: raise WorkflowRegistryAuditError("workflow registry response is invalid") if not workflows and len(records) < expected_total: raise WorkflowRegistryAuditError("workflow registry pagination is incomplete") @@ -497,14 +502,16 @@ def _validate_workflow_record(raw_record: object) -> dict[str, object]: def _require_nonnegative_int(value: object) -> int: - """Require a non-boolean, non-negative integer API count.""" - if isinstance(value, bool) or not isinstance(value, int) or value < 0: + """Require an exact non-negative integer API count.""" + if type(value) is not int or value < 0: raise WorkflowRegistryAuditError("workflow registry total count is invalid") return value def _validate_repository(repository_full_name: str) -> None: """Reject malformed repository selectors before any network read.""" + if type(repository_full_name) is not str: + raise WorkflowRegistryAuditError("repository must use owner/name syntax") components = repository_full_name.split("/") if ( not _REPOSITORY_RE.fullmatch(repository_full_name) @@ -515,12 +522,14 @@ def _validate_repository(repository_full_name: str) -> None: def _validate_protected_sha(protected_sha: str) -> None: """Require immutable commit identity instead of a branch/ref name.""" - if not _SHA_RE.fullmatch(protected_sha): + if type(protected_sha) is not str or not _SHA_RE.fullmatch(protected_sha): raise WorkflowRegistryAuditError("protected_sha must be an exact 40-hex protected SHA") def _validate_protected_ref(protected_ref: str) -> None: """Reject ambiguous or path-like branch selectors before any network read.""" + if type(protected_ref) is not str: + raise WorkflowRegistryAuditError("protected_ref must be a safe branch name") if ( not _PROTECTED_REF_RE.fullmatch(protected_ref) or protected_ref.startswith("/") @@ -590,4 +599,4 @@ def main(argv: list[str] | None = None) -> int: if __name__ == "__main__": - raise SystemExit(main()) \ No newline at end of file + raise SystemExit(main())