feat(workflows): audit orphan registry identities read-only

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"}
+    ]

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",
+        )

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