diff --git a/AGENTS.md b/AGENTS.md index 781cd36df..fb99277d7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,6 +41,20 @@ add CODEOWNERS-based merge gates until multiple independent maintainers exist. realistic tenant-isolation, migration, rollback, compatibility, and concurrency tests. +## Workflow registry audit + +- Keep the detector read-only. It may classify active identities that are + absent from an exact protected tree as candidates; it must never disable, + enable, edit, or rerun workflows. +- Derive audit identity from an independently resolved 40-hex protected SHA + plus a live protected-ref pre/post check. Registry rows, `dynamic/` paths, + and transport headers are not source authority. +- Require exact JSON-decoder types for containers and for `ref` / commit + `sha` / tree `sha` members before any equality test. +- Keep the production module under `pg_llm_batch`, the console script + `pg-llm-batch-workflow-audit`, and the operator/ADR/doctoring/CHANGELOG + contract synchronized. + ## Provider retry invariant Automatic provider retries are restricted to idempotent GET operations. The diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 3cfea1b03..58c9a6c6a 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -63,6 +63,19 @@ does not require either host and retains standalone operation. When embedded, tenant scope is a local control-plane identity and not model- or provider-returned data. +## Workflow registry audit + +`pg-llm-batch-workflow-audit` is a packaged read-only control-plane detector. +It compares the GitHub Actions registry to one exact protected commit and +reports active repository-backed identities that are absent from that tree as +`active_absent_workflows` candidates. Platform-managed `dynamic/` identities +are receipted and never treated as deleted YAML. The tool does not mutate +workflow state, select tenant scope, or perform provider I/O. + +Operators install the package and run the console script. A moving protected +ref, truncated tree, pagination drift, rate limit, or hostile identity-member +type fails closed. Candidate review remains a human control-plane decision. + ## Verification boundary Deterministic gates cover strict tenant validation, standalone compatibility, diff --git a/CHANGELOG.md b/CHANGELOG.md index 97ecdbf16..832525465 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Read-only workflow-registry audit (`pg-llm-batch-workflow-audit`) that binds one receipt to an exact protected commit, verifies the live protected ref before and after the read, classifies active repository-backed identities missing from that tree as `active_absent_workflows` candidates only, receipts GitHub-managed `dynamic/` identities without treating them as orphans, and fails closed on hostile identity-member types, truncated trees, pagination drift, and rate limits. The tool never disables or edits workflows. - Read-only exact-head release acceptance that builds wheel and source distribution artifacts twice from clean Git archives, proves byte-identical SHA-256 identity, records bounded canonical evidence, and keeps publication and attestation authority separate. - Optional bounded provider output/error-file lifetime controls for batch creation, with exact local validation before credential resolution and backward-compatible omission for provider-neutral callers. - Trusted tenant-scoped durable lifecycle identities for shared-table MSA deployments, including `TenantDurableBatchAPIClient`, tenant-qualified persistence and read helpers, transaction-local PostgreSQL context, forced default-deny row-level security, and explicit standalone compatibility. diff --git a/CLAUDE.md b/CLAUDE.md index d7843cd44..8d8d490d0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -27,3 +27,7 @@ - Maintain 100% production statement, branch, and public-docstring coverage. Add realistic migration, rollback, compatibility, security, and tenant-isolation tests before implementation changes. +- Keep the workflow-registry auditor read-only, packaged as + `pg_llm_batch.workflow_registry_audit` / `pg-llm-batch-workflow-audit`, + exact-type fail-closed on identity members, and documented in README, + architecture, ADR, doctoring, and CHANGELOG. diff --git a/README.md b/README.md index 9a699698b..d84277638 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,7 @@ than a second database-side network authority. | DDL subset | `pg_llm_batch/schema.sql` | | Readiness (`/healthz`) | `pg_llm_batch/health.py` | | CLI | `pg_llm_batch/cli.py` | +| Read-only workflow registry audit | `pg_llm_batch/workflow_registry_audit.py` | ## Requirements @@ -154,6 +155,27 @@ python -m pg_llm_batch health # prints the report, exit 0 ready / 1 not ready The Docker `HEALTHCHECK` and the compose `postgres` service both gate on the same `pg_llm_batch_health_check()` SQL function. +### Workflow registry audit + +Detect active GitHub Actions identities that are missing from an exact +protected source tree. The command is read-only: it never disables, edits, or +reruns workflows. Review `active_absent_workflows` yourself. GitHub-managed +`dynamic/` identities are listed but are not orphan candidates. + +```bash +export GITHUB_TOKEN="$(gh auth token)" # contents:read + actions:read +pg-llm-batch-workflow-audit \ + --repository ContextualWisdomLab/pg-llm-batch \ + --protected-ref main \ + --protected-sha "$(git rev-parse origin/main)" +``` + +Exit `0` means the protected ref stayed on the supplied SHA and no +repository-backed orphan candidates were found. Exit `2` prints a JSON receipt +with candidates for a separate review. Exit `1` is a fail-closed audit error. +See [`docs/doctoring/workflow-registry-audit.md`](docs/doctoring/workflow-registry-audit.md) +for token scope, rate-limit evidence, and recovery. + --- ## Durable lifecycle modes diff --git a/docs/adr/0016-workflow-registry-audit.md b/docs/adr/0016-workflow-registry-audit.md new file mode 100644 index 000000000..124a44aaf --- /dev/null +++ b/docs/adr/0016-workflow-registry-audit.md @@ -0,0 +1,84 @@ +# ADR 0016: Read-only exact-SHA workflow registry audit + +- **Status:** Accepted for the read-only detector slice +- **Date:** 2026-08-16 + +## Context + +GitHub Actions can retain active workflow identities after their repository +YAML is removed, or can register platform-managed `dynamic/` identities that +never had a protected-tree blob. Operators need a fail-closed way to list those +identities against one immutable protected commit. Automatic disable, rerun, or +YAML rewrite would change production control-plane state from a detector. + +GitHub's Git database and Actions registry are separate evidence sources. A +tree response `sha` is a tree object identity, not a commit SHA. A multi-page +registry read can splice two registry generations even when `total_count` is +unchanged. JSON subclasses can lie about equality if identity members are +compared before an exact-type check. + +NIST SP 800-218 (SSDF) and SLSA v1.0 treat build-system integrity as a +provenance control: verify the source that is supposed to produce automation +before changing that automation. in-toto's farm-to-table argument is the same +shape: bind an observation to an exact artifact identity rather than to a +moving name. + +## Decision + +`pg_llm_batch.workflow_registry_audit` is a packaged, installable, read-only +auditor. + +1. Transport is GET-only to `https://api.github.com`, path-only, no redirects, + finite timeout, default TLS verification, no automatic retry. +2. The CLI always verifies the live protected ref before and after the SHA + audit. Library callers may classify one SHA without live-ref checks. +3. Commit objects are resolved first; tree reads use `commit.tree.sha`. +4. Multi-page registries require a second identical pass. +5. Active repository-backed paths absent from the exact tree are candidates + only. `dynamic/` identities are receipted and never orphaned. +6. Exact decoder types are required for containers and for identity members + before any `==` / `!=`. +7. Exit codes are `0` (clean), `1` (failed closed), and `2` (candidates). +8. The tool never disables, enables, edits, or reruns workflows. + +## Consequences + +Operators can install `pg-llm-batch` and run +`pg-llm-batch-workflow-audit` without a special PYTHONPATH. Coverage, +docstring, and compile gates now see the production module because it lives +under `pg_llm_batch`. A buyer can distinguish a moving branch, a truncated +tree, a rate-limit, and a true orphan candidate. + +The detector does not replace branch protection, required reviews, or a human +decision to disable a workflow. False confidence from a stale PR body or from +comparing a tree SHA to a commit SHA is rejected. + +## Rollback + +Rollback is code-only: remove the package module, console script, checkout +shim, ADR, doctoring, changelog, and regression files. No database migration +or Actions mutation is required. Adding mutation to this tool requires a +separate reviewed control-plane contract. + +## References + +GitHub. (n.d.). *REST API endpoints for GitHub Actions workflows*. GitHub Docs. +Retrieved August 16, 2026, from +https://docs.github.com/en/rest/actions/workflows + +GitHub. (n.d.). *REST API endpoints for Git trees*. GitHub Docs. Retrieved +August 16, 2026, from +https://docs.github.com/en/rest/git/trees + +National Institute of Standards and Technology. (2022). *Secure software +development framework (SSDF) version 1.1: Recommendations for mitigating the +risk of software vulnerabilities* (NIST SP 800-218). +https://doi.org/10.6028/NIST.SP.800-218 + +The Linux Foundation. (2023). *Supply-chain Levels for Software Artifacts +(SLSA) v1.0*. https://slsa.dev/spec/v1.0/ + +Torres-Arias, S., Afzali, H., Kuppusamy, T. K., Curtmola, R., & Cappos, J. +(2019). in-toto: Providing farm-to-table guarantees for bits and bytes. In +*Proceedings of the 28th USENIX Security Symposium* (pp. 1393–1410). USENIX +Association. https://www.usenix.org/conference/usenixsecurity19/presentation/torres-arias diff --git a/docs/doctoring/workflow-registry-audit.md b/docs/doctoring/workflow-registry-audit.md new file mode 100644 index 000000000..d21491872 --- /dev/null +++ b/docs/doctoring/workflow-registry-audit.md @@ -0,0 +1,130 @@ +# Workflow registry audit doctoring + +## Purpose + +This record tells an operator how to detect GitHub Actions identities that are +active in a repository registry but absent from an exact protected source tree, +without disabling, editing, rerunning, or recreating any workflow. + +Use the installed command after `pip install pg-llm-batch` or an editable +checkout. Do not treat a missing checkout-root module as the supported +interface. The detector follows NIST SP 800-218 and SLSA v1.0 by binding the +receipt to an exact source commit before any operator reviews a candidate. + +```bash +export GITHUB_TOKEN="$(gh auth token)" # contents:read + actions:read is enough +pg-llm-batch-workflow-audit \ + --repository ContextualWisdomLab/pg-llm-batch \ + --protected-ref main \ + --protected-sha "$(git rev-parse origin/main)" +``` + +Equivalent module form: + +```bash +python -m pg_llm_batch.workflow_registry_audit \ + --repository ContextualWisdomLab/pg-llm-batch \ + --protected-ref main \ + --protected-sha "$(git rev-parse origin/main)" +``` + +## What to do with the result + +- Exit `0`: the live protected ref still matches the supplied SHA, the registry + was stable across a complete pagination pass, and no active + repository-backed identity is missing from that exact tree. Keep operating. +- Exit `2`: stdout is a JSON receipt whose `active_absent_workflows` lists + candidates only. Open a separate operator review. Do not disable those + workflows from this tool. +- Exit `1`: the audit failed closed. Read the single stderr line, then correct + the repository selector, SHA, ref, token scope, rate-limit budget, or + truncated-tree condition and run the same command again. + +GitHub-managed `dynamic/` identities appear in `workflow_records` with +`source_kind=platform_dynamic` and `source_present=null`. They are never +orphans. Do not delete platform-managed identities because they lack a +`.github/workflows/` blob. + +The library function `audit_repository_workflows` can classify one exact SHA +without live-ref pre/post checks. The CLI always uses +`audit_live_protected_ref_workflows` so a moving protected branch cannot +certify a stale head. + +## Trust and security boundary + +The client sends path-only GET requests to the fixed origin +`https://api.github.com`, disables redirects, and uses the default TLS +verifier. It never retries automatically. Non-success bodies are not read. +Rate-limit evidence is limited to HTTP 429 or HTTP 403 plus +`X-RateLimit-Remaining: 0` or `Retry-After`. + +JSON objects, arrays, strings, integers, and booleans are accepted only as +exact decoder built-in types. Identity members (`ref`, commit `sha`, tree +`sha`) are type-checked before equality. A hostile subclass cannot certify the +caller SHA while resolving a different tree. + +`GITHUB_TOKEN` is used only as an `Authorization` header. It is never copied +into stdout, stderr, receipts, or exception cause chains. + +This tool does not select tenant scope, touch PostgreSQL, or mutate Actions +state. It is a read-only control-plane detector for this repository and for +hosts that embed `pg-llm-batch` as a module. + +## Recovery and rollback + +No schema, secret, or workflow mutation is performed. Recovery is to rerun the +same read after correcting the failed input or GitHub condition. Rollback is +removing the console script, package module, ADR, and this record. Reintroducing +workflow disable/enable from this tool is not a routine rollback step. + +## Verification + +Permanent regression coverage requires that: + +- hostile `str` subclasses cannot satisfy commit, tree, or ref identity by + lying about equality; +- raising identity subclasses become `WorkflowRegistryAuditError` without + leaking custom exception text; +- `dynamic/` identities are receipted and never orphaned; +- unknown workflow states fail closed; +- multi-page registries are accepted only after a second identical pass; +- truncated trees, moving protected refs, and oversize responses fail closed; +- the console script `pg-llm-batch-workflow-audit` is declared in package + metadata; and +- README, architecture, ADR, changelog, and this doctoring record all tell the + operator to review candidates instead of disabling workflows. + +## References + +GitHub. (n.d.). *REST API endpoints for GitHub Actions workflows*. GitHub Docs. +Retrieved August 16, 2026, from +https://docs.github.com/en/rest/actions/workflows + +GitHub. (n.d.). *REST API endpoints for Git commits*. GitHub Docs. Retrieved +August 16, 2026, from +https://docs.github.com/en/rest/git/commits + +GitHub. (n.d.). *REST API endpoints for Git trees*. GitHub Docs. Retrieved +August 16, 2026, from +https://docs.github.com/en/rest/git/trees + +GitHub. (n.d.). *REST API endpoints for Git references*. GitHub Docs. Retrieved +August 16, 2026, from +https://docs.github.com/en/rest/git/refs + +GitHub. (n.d.). *Rate limits for the REST API*. GitHub Docs. Retrieved August +16, 2026, from +https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api + +National Institute of Standards and Technology. (2022). *Secure software +development framework (SSDF) version 1.1: Recommendations for mitigating the +risk of software vulnerabilities* (NIST SP 800-218). +https://doi.org/10.6028/NIST.SP.800-218 + +The Linux Foundation. (2023). *Supply-chain Levels for Software Artifacts +(SLSA) v1.0*. https://slsa.dev/spec/v1.0/ + +Torres-Arias, S., Afzali, H., Kuppusamy, T. K., Curtmola, R., & Cappos, J. +(2019). in-toto: Providing farm-to-table guarantees for bits and bytes. In +*Proceedings of the 28th USENIX Security Symposium* (pp. 1393–1410). USENIX +Association. https://www.usenix.org/conference/usenixsecurity19/presentation/torres-arias diff --git a/pg_llm_batch/workflow_registry_audit.py b/pg_llm_batch/workflow_registry_audit.py new file mode 100644 index 000000000..c96b51df9 --- /dev/null +++ b/pg_llm_batch/workflow_registry_audit.py @@ -0,0 +1,639 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) ContextualWisdomLab. +"""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]*$") +_CAPTURED_AT_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$") +_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_captured_at(captured_at) + _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_captured_at(captured_at) + _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: + raise WorkflowRegistryAuditError("protected ref response is invalid") + ref_name = payload.get("ref") + expected_ref = f"refs/heads/{protected_ref}" + if type(ref_name) is not str or ref_name != expected_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: + raise WorkflowRegistryAuditError("protected commit response is invalid") + commit_sha = commit_payload.get("sha") + if ( + type(commit_sha) is not str + or not _SHA_RE.fullmatch(commit_sha) + or commit_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: + raise WorkflowRegistryAuditError("protected tree SHA does not match commit tree SHA") + response_sha = payload.get("sha") + if ( + type(response_sha) is not str + or not _SHA_RE.fullmatch(response_sha) + or response_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 + + 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_captured_at(captured_at: str | None) -> None: + """Require a finite canonical UTC RFC 3339 receipt timestamp before reads.""" + message = "captured_at must be a canonical UTC RFC 3339 timestamp" + if captured_at is None: + return + if type(captured_at) is not str or not _CAPTURED_AT_RE.fullmatch(captured_at): + raise WorkflowRegistryAuditError(message) + try: + parsed = datetime.fromisoformat(captured_at[:-1] + "+00:00") + except ValueError: + raise WorkflowRegistryAuditError(message) from None + canonical = parsed.isoformat(timespec="seconds").replace("+00:00", "Z") + if canonical != captured_at: + raise WorkflowRegistryAuditError(message) + + +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__": # pragma: no cover + raise SystemExit(main()) \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index d1aba037c..8aceca534 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ dev = [ [project.scripts] pg-llm-batch = "pg_llm_batch.cli:main" +pg-llm-batch-workflow-audit = "pg_llm_batch.workflow_registry_audit:main" [project.urls] Homepage = "https://github.com/ContextualWisdomLab/pg-llm-batch" diff --git a/tests/test_workflow_registry_audit.py b/tests/test_workflow_registry_audit.py new file mode 100644 index 000000000..228b8a3fe --- /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 pg_llm_batch.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("pg_llm_batch.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("pg_llm_batch.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("pg_llm_batch.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_captured_at.py b/tests/test_workflow_registry_audit_captured_at.py new file mode 100644 index 000000000..bf6ce7f5b --- /dev/null +++ b/tests/test_workflow_registry_audit_captured_at.py @@ -0,0 +1,48 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Regression tests for bounded canonical workflow-audit timestamps.""" + +from __future__ import annotations + +import pytest + +from pg_llm_batch.workflow_registry_audit import ( + WorkflowRegistryAuditError, + audit_repository_workflows, +) + + +class _NoReadClient: + """Prove invalid receipt timestamps fail before any GitHub read.""" + + def get_json(self, path: str) -> dict[str, object]: + del path + raise AssertionError("invalid captured_at must fail before GitHub access") + + +@pytest.mark.parametrize( + "captured_at", + [ + "", + "not-a-time", + "2026-08-16 16:20:00Z", + "2026-08-16T16:20:00+00:00", + "2026-08-16T16:20:00.000Z", + "2026-8-16T16:20:00Z", + "2026-02-30T16:20:00Z", + "2026-08-16T24:00:00Z", + "2026-08-16T16:20:00z", + "2" * 10_000, + ], +) +def test_invalid_captured_at_fails_before_github_access(captured_at: str) -> None: + """Reject malformed, noncanonical, non-UTC, or unbounded receipt timestamps.""" + with pytest.raises( + WorkflowRegistryAuditError, + match="captured_at must be a canonical UTC RFC 3339 timestamp", + ): + audit_repository_workflows( + repository_full_name="ContextualWisdomLab/pg-llm-batch", + protected_sha="a" * 40, + client=_NoReadClient(), + captured_at=captured_at, + ) diff --git a/tests/test_workflow_registry_audit_cli.py b/tests/test_workflow_registry_audit_cli.py new file mode 100644 index 000000000..06f68836c --- /dev/null +++ b/tests/test_workflow_registry_audit_cli.py @@ -0,0 +1,154 @@ +# SPDX-License-Identifier: Apache-2.0 +"""CLI exit-code and packaging contracts for the workflow registry audit.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from pg_llm_batch.workflow_registry_audit import ( + WorkflowRegistryAuditError, + main, +) + +_PROTECTED_SHA = "a" * 40 +_REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +def test_main_prints_receipt_and_exits_zero_when_no_orphans(monkeypatch, capsys) -> None: + """A clean live-ref receipt is JSON on stdout and exit status 0.""" + receipt = { + "repository_full_name": _REPOSITORY, + "protected_sha": _PROTECTED_SHA, + "protected_ref": "main", + "active_absent_workflows": [], + } + + def fake_audit(**kwargs): + assert kwargs["repository_full_name"] == _REPOSITORY + assert kwargs["protected_ref"] == "main" + assert kwargs["expected_protected_sha"] == _PROTECTED_SHA + return receipt + + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.audit_live_protected_ref_workflows", + fake_audit, + ) + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + + assert main( + [ + "--repository", + _REPOSITORY, + "--protected-sha", + _PROTECTED_SHA, + ] + ) == 0 + printed = json.loads(capsys.readouterr().out) + assert printed["active_absent_workflows"] == [] + + +def test_main_exits_two_when_active_absent_candidates_exist(monkeypatch, capsys) -> None: + """Operators get exit 2 so automation can fail a job without mutating workflows.""" + receipt = { + "active_absent_workflows": [ + { + "workflow_id": 9, + "path": ".github/workflows/orphan.yml", + "state": "active", + } + ] + } + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.audit_live_protected_ref_workflows", + lambda **_kwargs: receipt, + ) + + assert main( + [ + "--repository", + _REPOSITORY, + "--protected-sha", + _PROTECTED_SHA, + "--protected-ref", + "release/1.0", + ] + ) == 2 + printed = json.loads(capsys.readouterr().out) + assert printed["active_absent_workflows"][0]["path"] == ".github/workflows/orphan.yml" + + +def test_main_exits_one_and_prints_fixed_stderr_on_audit_error(monkeypatch, capsys) -> None: + """Transport and classification failures stay body-free on stderr.""" + + def boom(**_kwargs): + raise WorkflowRegistryAuditError("protected ref moved during audit") + + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.audit_live_protected_ref_workflows", + boom, + ) + + assert main( + [ + "--repository", + _REPOSITORY, + "--protected-sha", + _PROTECTED_SHA, + ] + ) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err == "workflow_registry_audit: protected ref moved during audit\n" + + +def test_main_forwards_github_token_and_timeout_to_read_client(monkeypatch) -> None: + """The CLI reads GITHUB_TOKEN from the environment and never prints it.""" + seen: dict[str, object] = {} + + class _FakeClient: + def __init__(self, *, token, timeout_seconds): + seen["token"] = token + seen["timeout_seconds"] = timeout_seconds + + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.GitHubReadClient", + _FakeClient, + ) + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.audit_live_protected_ref_workflows", + lambda **kwargs: {"active_absent_workflows": []}, + ) + monkeypatch.setenv("GITHUB_TOKEN", "ghs_test_token_not_for_output") + + assert main( + [ + "--repository", + _REPOSITORY, + "--protected-sha", + _PROTECTED_SHA, + "--timeout-seconds", + "7.5", + ] + ) == 0 + assert seen["token"] == "ghs_test_token_not_for_output" + assert seen["timeout_seconds"] == 7.5 + + +def test_checkout_shim_reexports_packaged_public_api() -> None: + """A repository checkout can still import the historical module path.""" + import workflow_registry_audit as shim + from pg_llm_batch import workflow_registry_audit as packaged + + assert shim.main is packaged.main + assert shim.GitHubReadClient is packaged.GitHubReadClient + assert shim.audit_repository_workflows is packaged.audit_repository_workflows + + +def test_pyproject_declares_workflow_audit_console_script() -> None: + """Installed wheels must expose the operator command, not a checkout-only module.""" + config = Path("pyproject.toml").read_text(encoding="utf-8") + assert ( + 'pg-llm-batch-workflow-audit = "pg_llm_batch.workflow_registry_audit:main"' + in config + ) diff --git a/tests/test_workflow_registry_audit_coverage.py b/tests/test_workflow_registry_audit_coverage.py new file mode 100644 index 000000000..62edb2bf6 --- /dev/null +++ b/tests/test_workflow_registry_audit_coverage.py @@ -0,0 +1,437 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Remaining fail-closed branches for the packaged workflow registry audit.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from pg_llm_batch.workflow_registry_audit import ( + GitHubReadClient, + WorkflowRegistryAuditError, + _read_registry, + _validate_workflow_record, + audit_live_protected_ref_workflows, + audit_repository_workflows, +) + +_PROTECTED_SHA = "a" * 40 +_TREE_SHA = "b" * 40 +_REPOSITORY = "ContextualWisdomLab/pg-llm-batch" + + +class _StaticClient: + def __init__(self, responses: dict[str, object]) -> None: + self._responses = responses + + def get_json(self, path: str) -> object: + return self._responses[path] + + +def _commit_and_tree(*, tree_entries: list[object] | None = None) -> dict[str, object]: + return { + f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}": { + "sha": _PROTECTED_SHA, + "tree": {"sha": _TREE_SHA.upper()}, + }, + f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1": { + "sha": _TREE_SHA, + "truncated": False, + "tree": [] if tree_entries is None else tree_entries, + }, + f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + + +def test_sync_client_rejects_an_active_event_loop() -> None: + """Operators cannot hide the sync client inside another running loop.""" + + async def _invoke() -> None: + with pytest.raises(WorkflowRegistryAuditError, match="active event loop"): + GitHubReadClient().get_json(f"/repos/{_REPOSITORY}/actions/workflows") + + asyncio.run(_invoke()) + + +def test_decoded_non_object_json_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """A JSON array is not a GitHub object receipt.""" + + class _Content: + async def iter_chunked(self, _size: int): + yield b"[]" + + class _Response: + status = 200 + headers: dict[str, str] = {} + 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, **_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( + "pg_llm_batch.workflow_registry_audit.aiohttp.ClientSession", + _Session, + ) + with pytest.raises(WorkflowRegistryAuditError, match="response is invalid"): + GitHubReadClient().get_json(f"/repos/{_REPOSITORY}/actions/workflows") + + +def test_response_stream_must_expose_chunk_iterator(monkeypatch: pytest.MonkeyPatch) -> None: + """A response without iter_chunked cannot be materialized unsafely.""" + + class _Response: + status = 200 + headers: dict[str, str] = {} + content_length = 2 + content = object() + + async def __aenter__(self) -> "_Response": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + 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: + return _Response() + + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.aiohttp.ClientSession", + _Session, + ) + with pytest.raises(WorkflowRegistryAuditError, match="response stream is invalid"): + GitHubReadClient().get_json(f"/repos/{_REPOSITORY}/actions/workflows") + + +def test_memoryview_and_invalid_chunks_are_classified( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Exact byte chunks are accepted; non-buffer chunks fail closed.""" + + class _Content: + def __init__(self, chunks: list[object]) -> None: + self._chunks = chunks + + async def iter_chunked(self, _size: int): + for chunk in self._chunks: + yield chunk + + class _Response: + status = 200 + headers: dict[str, str] = {} + content_length = None + + def __init__(self, chunks: list[object]) -> None: + self.content = _Content(chunks) + + async def __aenter__(self) -> "_Response": + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + class _Session: + chunks: list[object] = [] + + 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: + return _Response(_Session.chunks) + + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.aiohttp.ClientSession", + _Session, + ) + _Session.chunks = [memoryview(b'{"ok":true}')] + assert GitHubReadClient().get_json(f"/repos/{_REPOSITORY}/actions/workflows") == { + "ok": True + } + + _Session.chunks = ["not-bytes"] + with pytest.raises(WorkflowRegistryAuditError, match="response stream is invalid"): + GitHubReadClient().get_json(f"/repos/{_REPOSITORY}/actions/workflows") + + +def test_unauthenticated_client_omits_authorization_header( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Checkout-local dry runs must not invent a bearer token.""" + seen: dict[str, object] = {} + + class _Content: + async def iter_chunked(self, _size: int): + yield b"{}" + + class _Response: + status = 200 + headers: dict[str, str] = {} + 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, *, headers: dict[str, str], **_kwargs: object) -> None: + seen["headers"] = headers + + 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: + return _Response() + + monkeypatch.setattr( + "pg_llm_batch.workflow_registry_audit.aiohttp.ClientSession", + _Session, + ) + GitHubReadClient().get_json(f"/repos/{_REPOSITORY}/actions/workflows") + assert "Authorization" not in seen["headers"] + + +def test_live_ref_rejects_non_object_and_non_commit_payloads() -> None: + """Protected-ref evidence must be an exact commit object.""" + with pytest.raises(WorkflowRegistryAuditError, match="protected ref response is invalid"): + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref="main", + expected_protected_sha=_PROTECTED_SHA, + client=_StaticClient( + {f"/repos/{_REPOSITORY}/git/ref/heads/main": ["not-an-object"]} + ), + captured_at="2026-08-16T00:00:00Z", + ) + + with pytest.raises(WorkflowRegistryAuditError, match="protected ref response is invalid"): + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref="main", + expected_protected_sha=_PROTECTED_SHA, + client=_StaticClient( + { + f"/repos/{_REPOSITORY}/git/ref/heads/main": { + "ref": "refs/heads/main", + "object": "commit", + } + } + ), + captured_at="2026-08-16T00:00:00Z", + ) + + with pytest.raises(WorkflowRegistryAuditError, match="protected ref response is invalid"): + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref="main", + expected_protected_sha=_PROTECTED_SHA, + client=_StaticClient( + { + f"/repos/{_REPOSITORY}/git/ref/heads/main": { + "ref": "refs/heads/main", + "object": {"sha": _PROTECTED_SHA, "type": "tag"}, + } + } + ), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_live_ref_lowercases_an_exact_commit_sha() -> None: + """GitHub may return mixed-case hex; the receipt stays lowercase.""" + responses = _commit_and_tree() + responses[f"/repos/{_REPOSITORY}/git/ref/heads/main"] = { + "ref": "refs/heads/main", + "object": {"sha": _PROTECTED_SHA.upper(), "type": "commit"}, + } + receipt = audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref="main", + expected_protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + assert receipt["protected_sha"] == _PROTECTED_SHA + assert receipt["protected_ref"] == "main" + + +def test_commit_tree_and_tree_container_shapes_fail_closed() -> None: + """Malformed commit/tree containers cannot become a false empty path set.""" + responses = _commit_and_tree() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { + "sha": _PROTECTED_SHA, + "tree": "not-an-object", + } + with pytest.raises(WorkflowRegistryAuditError, match="commit response is invalid"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + responses = _commit_and_tree() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { + "sha": _PROTECTED_SHA, + "tree": {"sha": "not-a-sha"}, + } + with pytest.raises(WorkflowRegistryAuditError, match="commit response is invalid"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + responses = _commit_and_tree() + responses[f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1"] = ["not-a-tree"] + with pytest.raises(WorkflowRegistryAuditError, match="tree SHA does not match"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + responses = _commit_and_tree() + responses[f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1"] = { + "sha": _TREE_SHA, + "truncated": False, + "tree": {"path": ".github/workflows/ci.yml"}, + } + with pytest.raises(WorkflowRegistryAuditError, match="tree response is invalid"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_tree_entries_must_be_exact_blob_records() -> None: + """Non-dict, non-string, and non-workflow entries stay fail-closed or ignored.""" + with pytest.raises(WorkflowRegistryAuditError, match="tree response is invalid"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(_commit_and_tree(tree_entries=["not-a-dict"])), + captured_at="2026-08-16T00:00:00Z", + ) + + with pytest.raises(WorkflowRegistryAuditError, match="tree response is invalid"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient( + _commit_and_tree(tree_entries=[{"path": 1, "type": "blob"}]) + ), + captured_at="2026-08-16T00:00:00Z", + ) + + receipt = audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient( + _commit_and_tree( + tree_entries=[ + {"path": "README.md", "type": "blob"}, + {"path": ".github/workflows", "type": "tree"}, + ] + ) + ), + captured_at="2026-08-16T00:00:00Z", + ) + assert receipt["protected_workflow_paths"] == [] + + +def test_registry_container_and_overflow_fail_closed() -> None: + """Registry pages must be objects whose row count cannot exceed total_count.""" + responses = _commit_and_tree() + responses[f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1"] = [ + "not-an-object" + ] + with pytest.raises(WorkflowRegistryAuditError, match="registry response is invalid"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + with pytest.raises(WorkflowRegistryAuditError, match="changed during audit"): + _read_registry( + repository_full_name=_REPOSITORY, + client=_StaticClient( + { + f"/repos/{_REPOSITORY}/actions/workflows?per_page=100&page=1": { + "total_count": 1, + "workflows": [ + { + "id": 1, + "path": ".github/workflows/ci.yml", + "state": "active", + }, + { + "id": 2, + "path": ".github/workflows/release.yml", + "state": "active", + }, + ], + } + } + ), + ) + + +def test_non_workflow_path_is_rejected_before_classification() -> None: + """Only repository workflow blobs and dynamic/ identities are valid paths.""" + with pytest.raises(WorkflowRegistryAuditError, match="path is invalid"): + _validate_workflow_record( + { + "id": 1, + "path": "scripts/ci.yml", + "state": "active", + } + ) diff --git a/tests/test_workflow_registry_audit_documentation.py b/tests/test_workflow_registry_audit_documentation.py new file mode 100644 index 000000000..af9255e92 --- /dev/null +++ b/tests/test_workflow_registry_audit_documentation.py @@ -0,0 +1,47 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Operator-document contracts for the read-only workflow registry audit.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +DOCTORING = ROOT / "docs/doctoring/workflow-registry-audit.md" +ADR = ROOT / "docs/adr/0016-workflow-registry-audit.md" +CHANGELOG = ROOT / "CHANGELOG.md" +README = ROOT / "README.md" +ARCHITECTURE = ROOT / "ARCHITECTURE.md" + + +def _normalized(path: Path) -> str: + """Read one authoritative document with layout-only whitespace collapsed.""" + return " ".join(path.read_text(encoding="utf-8").split()).lower() + + +def test_operator_docs_tell_the_next_safe_action() -> None: + """A buyer must see the installable command, exit codes, and no-mutation rule.""" + documents = { + "doctoring": _normalized(DOCTORING), + "adr": _normalized(ADR), + "changelog": _normalized(CHANGELOG), + "readme": _normalized(README), + "architecture": _normalized(ARCHITECTURE), + } + + for name, document in documents.items(): + assert "pg-llm-batch-workflow-audit" in document, name + assert "active_absent_workflows" in document or "candidates" in document, name + assert "dynamic/" in document or "platform-managed" in document, name + + doctoring = documents["doctoring"] + for token in ( + "exit `0`", + "exit `2`", + "exit `1`", + "do not disable", + "https://api.github.com", + "exact decoder", + "nist sp 800-218", + "slsa v1.0", + "torres-arias", + ): + assert token in doctoring, token diff --git a/tests/test_workflow_registry_audit_dynamic.py b/tests/test_workflow_registry_audit_dynamic.py new file mode 100644 index 000000000..27500a1db --- /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 pg_llm_batch.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..898a32a2e --- /dev/null +++ b/tests/test_workflow_registry_audit_exact_payload_types.py @@ -0,0 +1,317 @@ +"""Regression tests for exact primitive workflow-audit trust boundaries.""" + +from __future__ import annotations + +import pytest + +from pg_llm_batch.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") + + +_SECRET_IDENTITY = "SECRET-SENTINEL hostile identity member" + + +class _LyingIdentity(str): + def __eq__(self, _other: object) -> bool: + return True + + def __ne__(self, _other: object) -> bool: + return False + + +class _RaisingIdentity(str): + def __eq__(self, _other: object) -> bool: + raise RuntimeError(_SECRET_IDENTITY) + + def __ne__(self, _other: object) -> bool: + raise RuntimeError(_SECRET_IDENTITY) + + +@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", + ) + + +def test_audit_rejects_lying_commit_sha_before_tree_resolution() -> None: + """A sha subclass must not certify the caller SHA while resolving another tree.""" + planted_tree = "c" * 40 + planted_path = ".github/workflows/planted.yml" + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { + "sha": _LyingIdentity(planted_tree), + "tree": {"sha": planted_tree}, + } + responses[f"/repos/{_REPOSITORY}/git/trees/{planted_tree}?recursive=1"] = { + "sha": planted_tree, + "truncated": False, + "tree": [{"path": planted_path, "type": "blob"}], + } + + with pytest.raises(WorkflowRegistryAuditError, match="commit response is invalid"): + 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_raising_commit_sha_without_leaking_custom_code() -> None: + """Identity comparison must not execute subclass equality methods.""" + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/commits/{_PROTECTED_SHA}"] = { + "sha": _RaisingIdentity(_PROTECTED_SHA), + "tree": {"sha": _TREE_SHA}, + } + + with pytest.raises(WorkflowRegistryAuditError, match="commit response is invalid") as caught: + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + assert _SECRET_IDENTITY not in str(caught.value) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_audit_rejects_lying_tree_sha_before_path_collection() -> None: + """Tree object identity must be an exact decoder string before comparison.""" + planted_tree = "c" * 40 + responses = _valid_responses() + responses[f"/repos/{_REPOSITORY}/git/trees/{_TREE_SHA}?recursive=1"] = { + "sha": _LyingIdentity(planted_tree), + "truncated": False, + "tree": [{"path": ".github/workflows/planted.yml", "type": "blob"}], + } + + with pytest.raises(WorkflowRegistryAuditError, match="tree SHA does not match"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(responses), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_live_audit_rejects_lying_ref_member_before_equality() -> None: + """A ref subclass must not certify the requested branch by lying about equality.""" + + class _LyingRefClient: + def get_json(self, path: str) -> dict[str, object]: + if path == f"/repos/{_REPOSITORY}/git/ref/heads/main": + return { + "ref": _LyingIdentity("refs/heads/other"), + "object": {"sha": _PROTECTED_SHA, "type": "commit"}, + } + raise AssertionError(f"unexpected GitHub read: {path}") + + with pytest.raises(WorkflowRegistryAuditError, match="protected ref response is invalid"): + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref="main", + expected_protected_sha=_PROTECTED_SHA, + client=_LyingRefClient(), + captured_at="2026-08-16T00:00:00Z", + ) + + +def test_live_audit_rejects_raising_ref_member_without_leaking_custom_code() -> None: + """Protected-ref membership checks must not run subclass equality methods.""" + + class _RaisingRefClient: + def get_json(self, path: str) -> dict[str, object]: + if path == f"/repos/{_REPOSITORY}/git/ref/heads/main": + return { + "ref": _RaisingIdentity("refs/heads/main"), + "object": {"sha": _PROTECTED_SHA, "type": "commit"}, + } + raise AssertionError(f"unexpected GitHub read: {path}") + + with pytest.raises(WorkflowRegistryAuditError, match="protected ref response is invalid") as caught: + audit_live_protected_ref_workflows( + repository_full_name=_REPOSITORY, + protected_ref="main", + expected_protected_sha=_PROTECTED_SHA, + client=_RaisingRefClient(), + captured_at="2026-08-16T00:00:00Z", + ) + + assert _SECRET_IDENTITY not in str(caught.value) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_audit_generates_rfc3339_captured_at_when_omitted() -> None: + """Operators receive a UTC receipt timestamp when they do not supply one.""" + receipt = audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_StaticClient(_valid_responses()), + ) + assert type(receipt["captured_at"]) is str + assert receipt["captured_at"].endswith("Z") + + +def test_audit_rejects_non_string_captured_at_before_receipt() -> None: + """Receipt timestamps must be exact strings or omitted.""" + with pytest.raises(WorkflowRegistryAuditError, match="captured_at"): + audit_repository_workflows( + repository_full_name=_REPOSITORY, + protected_sha=_PROTECTED_SHA, + client=_NoReadClient(), + captured_at=20260816, # type: ignore[arg-type] + ) 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..5983093ec --- /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 pg_llm_batch.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..012c021fd --- /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 pg_llm_batch.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..dce96c420 --- /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 pg_llm_batch.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..6ee90e7f8 --- /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 pg_llm_batch.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("pg_llm_batch.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("pg_llm_batch.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..7baf1b8d5 --- /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 pg_llm_batch.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..b7805fe02 --- /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 pg_llm_batch.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..241858c29 --- /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 pg_llm_batch.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("pg_llm_batch.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("pg_llm_batch.workflow_registry_audit.aiohttp.ClientSession", _Session) + monkeypatch.setattr("pg_llm_batch.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("pg_llm_batch.workflow_registry_audit.aiohttp.ClientSession", _Session) + + def _raise_recursion(_content: str) -> object: + raise RecursionError(secret_sentinel) + + monkeypatch.setattr("pg_llm_batch.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..e30de0a94 --- /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 pg_llm_batch.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..9382bebac --- /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 pg_llm_batch.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..0f533216a --- /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 pg_llm_batch.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..30eed1eb9 --- /dev/null +++ b/workflow_registry_audit.py @@ -0,0 +1,31 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) ContextualWisdomLab. +"""Checkout-local entry point for the packaged workflow registry audit. + +Prefer the installed console script ``pg-llm-batch-workflow-audit`` or +``python -m pg_llm_batch.workflow_registry_audit`` after installation. This +module remains so a repository checkout can still run +``python workflow_registry_audit.py`` without changing PYTHONPATH. +""" + +from __future__ import annotations + +from pg_llm_batch.workflow_registry_audit import ( + GitHubReadClient, + WorkflowRegistryAuditError, + audit_live_protected_ref_workflows, + audit_repository_workflows, + main, +) + +__all__ = [ + "GitHubReadClient", + "WorkflowRegistryAuditError", + "audit_live_protected_ref_workflows", + "audit_repository_workflows", + "main", +] + + +if __name__ == "__main__": + raise SystemExit(main())