diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py new file mode 100644 index 000000000..5917d5839 --- /dev/null +++ b/tests/test_workflow_registry_audit.py @@ -0,0 +1,426 @@ +"""Regression tests for the read-only workflow registry audit tool.""" + +from __future__ import annotations + +import traceback +from dataclasses import dataclass + +import aiohttp +import pytest + +from workflow_registry_audit import ( + GitHubReadClient, + WorkflowRegistryAuditError, + audit_repository_workflows, +) + + +PROTECTED_SHA = "d0a4b30be1f46536e352443309f3a35533156767" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" +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 _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": TREE_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/commits/{PROTECTED_SHA}", _commit_payload()), + _JsonRoute( + f"/git/trees/{TREE_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 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) + ] + second_page = [_workflow(101, ".github/workflows/ci.yml")] + client = _FakeClient( + [ + _JsonRoute(f"/git/commits/{PROTECTED_SHA}", _commit_payload()), + _JsonRoute( + f"/git/trees/{TREE_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}, + ), + _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/commits/{PROTECTED_SHA}", _commit_payload()), + _JsonRoute( + f"/git/trees/{TREE_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/commits/{PROTECTED_SHA}", _commit_payload()), + _JsonRoute( + f"/git/trees/{TREE_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/commits/{PROTECTED_SHA}", _commit_payload()), + _JsonRoute( + f"/git/trees/{TREE_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/commits/{PROTECTED_SHA}", _commit_payload()), + _JsonRoute( + f"/git/trees/{TREE_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_authenticated_reads_use_fixed_aiohttp_origin_and_path_only( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """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 _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 + + async def __aexit__(self, *_args: object) -> None: + return None + + class _Session: + def __init__( + self, + *, + base_url: str, + headers: dict[str, str], + timeout: aiohttp.ClientTimeout, + ) -> None: + observed_sessions.append((base_url, float(timeout.total), headers)) + + async def __aenter__(self) -> "_Session": + return self + + async def __aexit__(self, *_args: object) -> None: + closed.append(True) + + 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_sessions == [ + ( + "https://api.github.com", + 15.0, + { + "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_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.""" + session_attempted = False + + class _Session: + def __init__(self, *_args: object, **_kwargs: object) -> None: + nonlocal session_attempted + session_attempted = True + + 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 session_attempted is False + + +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 _Session: + def __init__( + self, + *, + 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" + 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) -> 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: + 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") 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"} + ] 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", + ) 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..fcfac0b70 --- /dev/null +++ b/tests/test_workflow_registry_audit_exact_types.py @@ -0,0 +1,109 @@ +"""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 _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.""" + + 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 + + +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"), + [ + ("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 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", + ] 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 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") 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..8f5d9cb22 --- /dev/null +++ b/tests/test_workflow_registry_audit_ref_movement.py @@ -0,0 +1,139 @@ +"""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" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" +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, *, 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]: + """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 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: + """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, + ) + + +@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" + 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( + [ + _Route("/git/ref/heads/main", _ref_payload(PROTECTED_SHA)), + _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("/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, + ) 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 == [] 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..06cfd7d70 --- /dev/null +++ b/tests/test_workflow_registry_audit_response_bound.py @@ -0,0 +1,133 @@ +"""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") + + +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") diff --git a/tests/test_workflow_registry_audit_stability.py b/tests/test_workflow_registry_audit_stability.py new file mode 100644 index 000000000..8ed8f50a2 --- /dev/null +++ b/tests/test_workflow_registry_audit_stability.py @@ -0,0 +1,101 @@ +"""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" +TREE_SHA = "61e02626f1184dede4990f06704574e878012336" +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/commits/{PROTECTED_SHA}", + {"sha": PROTECTED_SHA, "tree": {"sha": TREE_SHA}}, + ), + _JsonRoute( + f"/git/trees/{TREE_SHA}?recursive=1", + { + "sha": TREE_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, + ) 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 diff --git a/tests/test_workflow_registry_audit_timeout.py b/tests/test_workflow_registry_audit_timeout.py new file mode 100644 index 000000000..0204418b0 --- /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, -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) diff --git a/workflow_registry_audit.py b/workflow_registry_audit.py new file mode 100644 index 000000000..88667189f --- /dev/null +++ b/workflow_registry_audit.py @@ -0,0 +1,602 @@ +"""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 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 + +import argparse +import asyncio +import json +import math +import os +import re +import sys +from datetime import datetime, timezone +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/" +_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 +_MAX_REGISTRY_WORKFLOWS = 10_000 +_MAX_RESPONSE_BYTES = 16 * 1024 * 1024 +_RESPONSE_CHUNK_BYTES = 64 * 1024 + + +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, + timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS, + ) -> None: + """Configure fixed-origin read-only API access with a finite timeout.""" + 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 + + def get_json(self, path: str) -> dict[str, object]: + """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)) + + async def _get_json(self, path: str) -> dict[str, object]: + """Perform one fixed-origin aiohttp GET with redirects and body growth bounded.""" + timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) + try: + 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: + 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" + ) + raw = await self._read_bounded_response(response) + payload = json.loads(raw.decode("utf-8")) + except WorkflowRegistryAuditError: + raise + except ( + aiohttp.ClientError, + asyncio.TimeoutError, + UnicodeDecodeError, + json.JSONDecodeError, + RecursionError, + ): + raise WorkflowRegistryAuditError("GitHub workflow audit read failed") from None + if type(payload) is not dict: + 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 type(declared_value) is int 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 = { + "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_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, + protected_sha: str, + client: _JsonClient, + captured_at: str | None = None, +) -> dict[str, object]: + """Return an exact-SHA read-only registry/source classification receipt. + + 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. + """ + _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=normalized_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: + 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": path, + "state": record["state"], + "source_kind": source_kind, + "source_present": source_present, + } + classified_records.append(classified) + if ( + record["state"] == "active" + and source_kind == "repository" + and source_present is False + ): + active_absent.append( + { + "workflow_id": record["workflow_id"], + "path": 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": normalized_sha, + "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_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 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 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 ( + 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() + + +def _read_protected_workflow_paths( + *, + repository_full_name: str, + protected_sha: str, + client: _JsonClient, +) -> set[str]: + """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 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 type(commit_tree) is not dict: + raise WorkflowRegistryAuditError("protected commit response is invalid") + tree_sha = commit_tree.get("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 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 type(tree) is not list: + raise WorkflowRegistryAuditError("protected tree response is invalid") + + paths: set[str] = set() + for entry in tree: + if type(entry) is not dict: + raise WorkflowRegistryAuditError("protected tree response is invalid") + path = entry.get("path") + entry_type = entry.get("type") + 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) + return paths + + +def _read_registry( + *, + repository_full_name: str, + client: _JsonClient, +) -> tuple[list[dict[str, object]], int, int]: + """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 + 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}" + ) + 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: + raise WorkflowRegistryAuditError( + "workflow registry exceeds supported workflow limit" + ) + 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 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") + + 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 _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 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 ( + type(workflow_id) is not int + or workflow_id <= 0 + 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") + 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} + + +def _require_nonnegative_int(value: object) -> int: + """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) + or any(component in {".", ".."} for component in components) + ): + 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 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("/") + 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") + + +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 a stable protected branch head." + ) + ) + parser.add_argument("--repository", required=True, help="GitHub owner/name repository") + parser.add_argument( + "--protected-sha", + required=True, + 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", + type=float, + default=_DEFAULT_TIMEOUT_SECONDS, + help="Finite timeout for each read-only GitHub API request", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + """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_live_protected_ref_workflows( + repository_full_name=args.repository, + protected_ref=args.protected_ref, + expected_protected_sha=args.protected_sha, + client=client, + ) + except WorkflowRegistryAuditError as exc: + print(f"workflow_registry_audit: {exc}", file=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())