diff --git a/.github/workflows/workflow-lifecycle-inventory-quality-ci.yml b/.github/workflows/workflow-lifecycle-inventory-quality-ci.yml new file mode 100644 index 0000000000..366a3e9480 --- /dev/null +++ b/.github/workflows/workflow-lifecycle-inventory-quality-ci.yml @@ -0,0 +1,57 @@ +name: Workflow Lifecycle Inventory Quality CI + +on: + pull_request: + branches: [main] + paths: + - ".github/workflows/workflow-lifecycle-inventory.yml" + - ".github/workflows/workflow-lifecycle-inventory-quality-ci.yml" + - "scripts/ci/inventory_orphaned_workflows.py" + - "tests/test_inventory_orphaned_workflows.py" + - "tests/test_workflow_lifecycle_inventory_workflow.py" + - "docs/doctoring/orphaned-workflow-lifecycle.md" + - "ARCHITECTURE.md" + - "CHANGELOG.md" + +permissions: + contents: read + +jobs: + exact-head-quality: + runs-on: ubuntu-24.04 + timeout-minutes: 10 + steps: + - name: Checkout exact pull-request head + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + + - name: Install hash-verified test tools + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + python -m pip install --only-binary=:all: --require-hashes -r <(cat <<'EOF' + coverage==7.15.2 --hash=sha256:b9a6367e4aff723e8ee8190836836124284e8fcd4265e307c844010cfa074f3f + iniconfig==2.1.0 --hash=sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760 + packaging==26.2 --hash=sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e + pluggy==1.6.0 --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746 + pygments==2.20.0 --hash=sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176 + pytest==9.1.1 --hash=sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c + EOF + ) + + - name: Prove exact-head behavior and branch coverage + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + test "$(git rev-parse HEAD)" = "${{ github.event.pull_request.head.sha }}" + coverage run --branch -m pytest --import-mode=importlib \ + tests/test_inventory_orphaned_workflows.py \ + tests/test_workflow_lifecycle_inventory_workflow.py -q + coverage report --include='scripts/ci/inventory_orphaned_workflows.py' --show-missing --fail-under=100 + python -m compileall -q scripts/ci/inventory_orphaned_workflows.py tests/test_inventory_orphaned_workflows.py + git diff --exit-code diff --git a/.github/workflows/workflow-lifecycle-inventory.yml b/.github/workflows/workflow-lifecycle-inventory.yml new file mode 100644 index 0000000000..cd2cc15e27 --- /dev/null +++ b/.github/workflows/workflow-lifecycle-inventory.yml @@ -0,0 +1,77 @@ +name: Workflow Lifecycle Inventory + +on: + schedule: + - cron: "31 4 * * *" + +concurrency: + group: workflow-lifecycle-inventory + cancel-in-progress: false + +permissions: + contents: read + actions: read + +jobs: + inventory: + if: >- + github.repository == 'ContextualWisdomLab/.github' && + github.ref == format('refs/heads/{0}', github.event.repository.default_branch) + runs-on: ubuntu-24.04 + timeout-minutes: 30 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" + steps: + - name: Harden runner + uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.13.2 + with: + egress-policy: block + allowed-endpoints: >- + api.github.com:443 + github.com:443 + objects.githubusercontent.com:443 + release-assets.githubusercontent.com:443 + results-receiver.actions.githubusercontent.com:443 + *.actions.githubusercontent.com:443 + *.blob.core.windows.net:443 + + - name: Checkout exact trusted inventory source + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + ref: ${{ github.sha }} + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + + - name: Collect read-only organization evidence + env: + GH_TOKEN: ${{ secrets.PR_REVIEW_MERGE_TOKEN }} + shell: bash --noprofile --norc -e -o pipefail {0} + run: | + if [ -z "${GH_TOKEN:-}" ]; then + echo "::error::PR_REVIEW_MERGE_TOKEN is required for complete organization visibility." + exit 1 + fi + echo "::add-mask::$GH_TOKEN" + test "$(git rev-parse HEAD)" = "$GITHUB_SHA" + python scripts/ci/inventory_orphaned_workflows.py --live \ + --output "$RUNNER_TEMP/workflow-lifecycle-ledger.json" \ + --receipt-output "$RUNNER_TEMP/workflow-lifecycle-api-receipts.json" \ + --failure-output "$RUNNER_TEMP/workflow-lifecycle-failure.json" + python -m json.tool "$RUNNER_TEMP/workflow-lifecycle-ledger.json" >/dev/null + python -m json.tool "$RUNNER_TEMP/workflow-lifecycle-api-receipts.json" >/dev/null + + - name: Preserve immutable read-only evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: workflow-lifecycle-inventory-${{ github.run_id }}-${{ github.run_attempt }} + path: | + ${{ runner.temp }}/workflow-lifecycle-ledger.json + ${{ runner.temp }}/workflow-lifecycle-api-receipts.json + ${{ runner.temp }}/workflow-lifecycle-failure.json + if-no-files-found: error + retention-days: 30 diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e12f33542d..dc4400d0e6 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -142,6 +142,18 @@ Product callers stagger Clearfolio at minute 23, DiskSage at minute 37, and fast-mlsirm at minute 49. Each caller is read-only, dispatches at most one repair, and delegates all privileged logic to the same sealed scheduler. +## Workflow lifecycle inventory + +GitHub persists Actions registry identities independently of the protected +default-branch tree. `scripts/ci/inventory_orphaned_workflows.py` is a +read-only classifier and live collector: it paginates the organization and +registry, binds each advertised workflow to a revalidated default-branch SHA, +distinguishes repository YAML from GitHub-owned `dynamic/` identities, and +fail-closes on incomplete trees, pagination, permissions, or visibility. The +scheduled integration retains content-bound API receipts. Classification never +disables or recreates workflows; the operator primitive is separately reviewed +and accepts only an immutable orphan ledger record on an unchanged head. + ## Exact-artifact SBOM attestation ```mermaid diff --git a/CHANGELOG.md b/CHANGELOG.md index 46d599a320..e1239b59b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1255,6 +1255,7 @@ Semantic Versioning where the repository publishes a release. - Added a dedicated DiskSage hourly caller that invokes the same product-neutral RCA and remediation-feasibility scheduler with an exact repository target, one-dispatch budget, two-hour same-head retry floor, non-cancelling single-flight heartbeat, and explicit established scheduler credentials. - Added a dedicated fast-mlsirm hourly caller that preserves Rust-owned psychometric arithmetic while dispatching at most one exact-head, root-cause-driven repair with a two-hour same-head retry floor. - Added a dedicated Orgmetra hourly caller at minute 58 that targets protected `develop`, dispatches at most one exact-head repair, preserves a two-hour same-head retry floor and non-cancelling single-flight execution, and maps only the established scheduler credentials. +- Added read-only orphan-workflow lifecycle classification with exact default-branch binding, fail-closed visibility and pagination checks, explicit case-insensitive owner-issue routing for active and disabled orphans, and no registry mutation authority. ### Changed @@ -1281,6 +1282,18 @@ Semantic Versioning where the repository publishes a release. ### Fixed +- Require orphan-workflow lifecycle fixtures to prove complete organization + repository visibility before emitting an audit ledger, preventing partial + inventories from overstating fleet coverage. Live collection now verifies + authenticated organization-wide repository totals, retries one explicit + HTTP 5xx once, rejects every malformed Git tree member, preserves partial + read receipts and structured failure evidence, gates completeness before + classification, and isolates disable/issue writes in the operator module. +- Add a live, paginated organization workflow inventory with content-bound API + receipts, exact default-head revalidation, and a separately reviewed, + ledger-bound operator disable primitive. +- Schedule the central read-only workflow-lifecycle sweep on the protected + default branch and retain its exact ledger and API receipts for 30 days. - Prefer the job-scoped `github.token` when the central OpenCode dispatch publishes a commit status back to the same `.github` repository. The job's declared `statuses: write` permission now reaches the endpoint instead of an diff --git a/docs/doctoring/orphaned-workflow-lifecycle.md b/docs/doctoring/orphaned-workflow-lifecycle.md new file mode 100644 index 0000000000..e5ae70d947 --- /dev/null +++ b/docs/doctoring/orphaned-workflow-lifecycle.md @@ -0,0 +1,127 @@ +# Orphaned GitHub Actions workflow-lifecycle inventory + +검토 기준일: **2026-08-16** + +## Incident + +Live Actions inventories showed the same recurrence in multiple +ContextualWisdomLab repositories (ContextualWisdomLab/.github#945): + +- AppGuardrail advertised dozens of historical `apply-*`, `finalize-*`, + and `*-once.yml` identities as `state: active` while sampled default-branch + paths returned 404 (ContextualWisdomLab/appguardrail#929); +- Clearfolio retained `one-shot-*` and PR-specific repair identities after + the YAML had left the protected default branch (ContextualWisdomLab/clearfolio#423); +- DiskSage retained PR-specific finalizers in the same shape + (ContextualWisdomLab/disksage#191). + +Source deletion is not a complete workflow lifecycle. GitHub persists +registry records independently of the default-branch tree, so a buyer or +reviewer cannot treat "the YAML is gone" as "no writer remains enabled." + +## Decision + +1. The central `.github` repository owns a **read-only** inventory that + binds every advertised workflow identity to the exact protected + default-branch SHA observed at the start and re-read at the end. +2. Classification is evidence-based: `present_active`, `present_disabled`, + `orphan_active`, `orphan_disabled`, `dynamic_owned`, or `unresolved`. + A file named `once` is not alone proof of invalidity. A benign name + does not hide a missing source file. +3. Incomplete visibility (401/403/404), a 5xx after one retry, pagination + truncation, `total_count` drift, reused workflow IDs, percent-encoded + paths, and default-branch movement fail closed. +4. This scanner never disables, deletes, or recreates workflows. Disablement + remains a separately reviewed operator step after the ledger is + revalidated. +5. `NVIDIA_NIM_API_KEY` may exist elsewhere in the control plane. This + inventory never reads `COPILOT_GITHUB_TOKEN`. +6. CSAP and SOC 2 are design constraints (access visibility, change + management, evidence retention). This record is not a certification + claim. Operational identities (repository, workflow path, workflow ID) + are not masked as PII. +7. Confirmed repository owner routes are maintained as an explicit, + linkable registry from live fleet evidence. Repository slugs are matched + case-insensitively, and both `orphan_active` and `orphan_disabled` + classifications retain the route. The scanner does not infer issue + numbers, create issues, or convert an absent owner route into a passing + result. + +## Trust boundary + +The production CLI uses `--live` with the established central `GH_TOKEN` +transport. It paginates all visible repositories and workflows, rejects a +truncated recursive tree, and re-reads each default-branch head. A mandatory +API receipt file content-binds every read. Fixture input remains available for +deterministic tests. Neither boundary receives `secrets: inherit` or a guessed PAT. +GitHub-owned `dynamic/` identities are never treated as deleted repository +files. + +The ledger improves operational visibility into enabled control-plane writers; +that visibility gap is not itself CWE-200 sensitive-information exposure. +CWE-862 describes missing authorization when a registry mutation is performed +without a reviewed operator path. This increment closes the visibility gap and +refuses the mutation. + +## Operator contract + +For a live read-only sweep, run: + +```bash +python3 scripts/ci/inventory_orphaned_workflows.py --live \ + --output /tmp/workflow-lifecycle-ledger.json \ + --receipt-output /tmp/workflow-lifecycle-api-receipts.json \ + --failure-output /tmp/workflow-lifecycle-failure.json +``` + +The protected-default-branch integration is +`.github/workflows/workflow-lifecycle-inventory.yml`. Its scheduled runs have +read-only repository permissions, verify the checked-out SHA, and +retain completed API receipts plus either the immutable ledger or structured +failure evidence for 30 days. The live collector proves fleet completeness by +matching the paginated repository list to authenticated organization-wide +public/private totals; pagination alone is not accepted. It contains no disable endpoint; +operator mutation remains a later reviewed action. + +For fixture verification, feed a JSON payload with `organization`, `observed_at`, +`repository_inventory_complete: true`, and one object per visible non-archived +repository. The completeness flag is mandatory: a partial repository list must +fail closed instead of producing a ledger that overstates fleet coverage. Each +repository must include the +start and end default-branch SHAs, the exact tree paths at that SHA, and +complete workflow pages (`total_count`, `workflows`, and either `_link_next` +or a GitHub `Link` header). Archived repositories are skipped. + +```bash +python3 scripts/ci/inventory_orphaned_workflows.py \ + --payload schemas/examples/cwl-workflow-lifecycle-ledger-v1.example.json \ + --output /tmp/workflow-lifecycle-ledger.json +``` + +The scanner never exports a write primitive. The separately reviewed +`scripts/ci/workflow_lifecycle_operator.py` module's +`disable_confirmed_orphan` primitive accepts only an immutable `orphan_active` record and an identical +fresh head SHA, then addresses its exact numeric workflow ID. After a reviewed +operator pass, rerun the organization sweep and retain both receipt sets. +Known AppGuardrail, Clearfolio, and DiskSage owner routes bind the same live +evidence to their governance issues without heuristic issue creation. + +## Rollback + +Rollback removes the inventory script, focused tests, schema example, +architecture entry, changelog entry, and this doctoring record together. No +registry state is mutated, so rollback does not re-enable or disable workflows. + +## References + +GitHub. (2026). *REST API endpoints for workflows*. GitHub Docs. +https://docs.github.com/en/rest/actions/workflows + +GitHub. (2026). *Security hardening for GitHub Actions*. GitHub Docs. +https://docs.github.com/en/actions/security-guides/security-hardening-for-github-actions + +MITRE. (2026a). *CWE-200: Exposure of sensitive information to an unauthorized actor*. +https://cwe.mitre.org/data/definitions/200.html + +MITRE. (2026b). *CWE-862: Missing authorization*. +https://cwe.mitre.org/data/definitions/862.html diff --git a/schemas/examples/cwl-workflow-lifecycle-ledger-v1.example.json b/schemas/examples/cwl-workflow-lifecycle-ledger-v1.example.json new file mode 100644 index 0000000000..026254c1fe --- /dev/null +++ b/schemas/examples/cwl-workflow-lifecycle-ledger-v1.example.json @@ -0,0 +1,41 @@ +{ + "organization": "ContextualWisdomLab", + "observed_at": "2026-08-16T12:00:00Z", + "repository_inventory_complete": true, + "repositories": [ + { + "name": "appguardrail", + "archived": false, + "default_branch": "main", + "default_branch_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "default_branch_sha_after": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "tree_paths": [".github/workflows/ci.yml"], + "workflow_pages": [ + { + "total_count": 3, + "workflows": [ + { + "id": 11, + "name": "finalize-once", + "path": ".github/workflows/finalize-once.yml", + "state": "active" + }, + { + "id": 12, + "name": "ci", + "path": ".github/workflows/ci.yml", + "state": "active" + }, + { + "id": 13, + "name": "pages", + "path": "dynamic/pages/pages-build-deployment", + "state": "active" + } + ], + "_link_next": false + } + ] + } + ] +} diff --git a/scripts/ci/inventory_orphaned_workflows.py b/scripts/ci/inventory_orphaned_workflows.py new file mode 100755 index 0000000000..8932ad70dd --- /dev/null +++ b/scripts/ci/inventory_orphaned_workflows.py @@ -0,0 +1,721 @@ +#!/usr/bin/env python3 +"""Read-only GitHub Actions workflow-lifecycle inventory. + +The classifier answers whether an advertised workflow identity still has +source on the exact protected default-branch SHA. It never disables, +deletes, or recreates workflows and never reads ``COPILOT_GITHUB_TOKEN``. +CSAP and SOC 2 appear only as design constraints. + +See ContextualWisdomLab/.github#945. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from collections.abc import Callable, Iterable, Mapping, Sequence +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Protocol +from urllib.parse import quote + +SCHEMA_VERSION = "1" +CAPABILITY = "workflow_lifecycle_inventory" +MAX_PAYLOAD_BYTES = 1_048_576 +PER_PAGE_DEFAULT = 100 +HEX_SHA = re.compile(r"^[0-9a-f]{40}$") +HEX_SHA256 = re.compile(r"^[0-9a-f]{64}$") +REPO_SLUG = re.compile(r"^[A-Za-z0-9._-]+$") +REPO_WORKFLOW_NAME = re.compile(r"^[^/]+\.(yml|yaml)$") +DYNAMIC_PREFIXES = ("dynamic/",) +DISABLED_STATES = frozenset( + { + "disabled_manually", + "disabled_inactivity", + "disabled_fork", + "deleted", + } +) +ACTIVE_STATES = frozenset({"active"}) +CLASSIFICATIONS = ( + "present_active", + "present_disabled", + "orphan_active", + "orphan_disabled", + "dynamic_owned", + "unresolved", +) +KNOWN_OWNER_ISSUES = { + "appguardrail": "ContextualWisdomLab/appguardrail#929", + "bandscope": "ContextualWisdomLab/bandscope#847", + "clearfolio": "ContextualWisdomLab/clearfolio#423", + "codec-carver": "ContextualWisdomLab/codec-carver#401", + "contextual-orchestrator": "ContextualWisdomLab/contextual-orchestrator#122", + "DiagramWeave": "ContextualWisdomLab/DiagramWeave#27", + "disksage": "ContextualWisdomLab/disksage#191", + "EgressWeave": "ContextualWisdomLab/EgressWeave#202", + "fast-mlsirm": "ContextualWisdomLab/fast-mlsirm#809", + "four-pillars": "ContextualWisdomLab/four-pillars#33", + "inkspan": "ContextualWisdomLab/inkspan#278", + "keyverse": "ContextualWisdomLab/keyverse#99", + "naruon": "ContextualWisdomLab/naruon#1324", + "newsdom-api": "ContextualWisdomLab/newsdom-api#604", + "noema": "ContextualWisdomLab/noema#226", + "OriginWeave": "ContextualWisdomLab/OriginWeave#123", + "pg-erd-cloud": "ContextualWisdomLab/pg-erd-cloud#865", + "RankWeave": "ContextualWisdomLab/RankWeave#38", + "saju-caldav": "ContextualWisdomLab/saju-caldav#33", + "ThreadWeave": "ContextualWisdomLab/ThreadWeave#31", +} +FORBIDDEN_TOKENS = frozenset({"COPILOT_GITHUB_TOKEN"}) + + +class InventoryError(ValueError): + """Fail-closed defect in workflow-lifecycle evidence.""" + + +class GitHubTransport(Protocol): + """Minimal authenticated REST transport shared with central coordination.""" + + def request(self, path: str, *, method: str = "GET", payload: Any = None) -> Any: + """Return the decoded JSON response for one bounded request.""" + + +def _live_get( + client: GitHubTransport, path: str, receipts: list[dict[str, Any]] +) -> Any: + """Read one endpoint and append a content-bound API receipt.""" + try: + body = client.request(path) + except Exception as first_exc: + if not re.search(r"\(HTTP 5\d\d\)", str(first_exc)): + raise InventoryError( + f"live GitHub visibility failed for {path}: {type(first_exc).__name__}" + ) from first_exc + try: + body = client.request(path) + except Exception as retry_exc: + raise InventoryError( + f"live GitHub visibility failed for {path}: {type(retry_exc).__name__}" + ) from retry_exc + encoded = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() + receipts.append( + {"method": "GET", "path": path, "sha256": hashlib.sha256(encoded).hexdigest()} + ) + return body + + +def collect_live_organization( + client: GitHubTransport, + organization: str = "ContextualWisdomLab", + *, + receipts: list[dict[str, Any]] | None = None, +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Collect a complete live fleet payload with exact-head revalidation.""" + if organization != "ContextualWisdomLab": + raise InventoryError("organization must be ContextualWisdomLab") + receipts = [] if receipts is None else receipts + repositories: list[Mapping[str, Any]] = [] + page = 1 + while True: + path = f"/orgs/{organization}/repos?type=all&sort=full_name&per_page=100&page={page}" + batch = _live_get(client, path, receipts) + if not isinstance(batch, list) or any( + not isinstance(item, Mapping) for item in batch + ): + raise InventoryError("repository inventory response is malformed") + repositories.extend(batch) + if len(batch) < 100: + break + page += 1 + if not repositories: + raise InventoryError("live repository inventory is empty") + organization_body = _live_get(client, f"/orgs/{organization}", receipts) + if not isinstance(organization_body, Mapping): + raise InventoryError("organization visibility proof is malformed") + public_repos = organization_body.get("public_repos") + private_repos = organization_body.get("total_private_repos") + if ( + not isinstance(public_repos, int) + or public_repos < 0 + or not isinstance(private_repos, int) + or private_repos < 0 + ): + raise InventoryError("organization-wide visibility proof is unavailable") + visible_names = {item.get("full_name") for item in repositories} + if ( + len(visible_names) != len(repositories) + or len(repositories) != public_repos + private_repos + ): + raise InventoryError( + "organization-wide visibility proof does not match repository inventory" + ) + records: list[dict[str, Any]] = [] + for repository in repositories: + name = repository.get("name") + full_name = repository.get("full_name") + archived = repository.get("archived") + default_branch = repository.get("default_branch") + if not isinstance(name, str) or not isinstance(full_name, str): + raise InventoryError("repository identity is malformed") + if archived is True: + records.append({"name": name, "archived": True}) + continue + if ( + archived is not False + or not isinstance(default_branch, str) + or not default_branch + ): + raise InventoryError(f"{name} repository metadata is incomplete") + branch = quote(default_branch, safe="") + commit_path = f"/repos/{full_name}/commits/{branch}" + start = _live_get(client, commit_path, receipts) + start_sha = start.get("sha") if isinstance(start, Mapping) else None + if not is_exact_sha(start_sha): + raise InventoryError(f"{name} default-branch SHA is invalid") + tree_body = _live_get( + client, f"/repos/{full_name}/git/trees/{start_sha}?recursive=1", receipts + ) + if ( + not isinstance(tree_body, Mapping) + or tree_body.get("truncated") is not False + ): + raise InventoryError( + f"{name} default-branch tree is truncated or malformed" + ) + tree = tree_body.get("tree") + if not isinstance(tree, list): + raise InventoryError(f"{name} default-branch tree is malformed") + tree_paths: list[str] = [] + for item in tree: + if not isinstance(item, Mapping): + raise InventoryError( + f"{name} default-branch tree entry is not an object" + ) + item_type = item.get("type") + path = item.get("path") + if ( + item_type not in {"blob", "tree", "commit"} + or not isinstance(path, str) + or not path + ): + raise InventoryError(f"{name} default-branch tree entry is malformed") + if item_type == "blob": + tree_paths.append(path) + workflow_pages: list[dict[str, Any]] = [] + workflow_page = 1 + while True: + workflow_path = f"/repos/{full_name}/actions/workflows?per_page=100&page={workflow_page}" + body = _live_get(client, workflow_path, receipts) + if not isinstance(body, dict) or not isinstance( + body.get("workflows"), list + ): + raise InventoryError(f"{name} workflow inventory is malformed") + total = body.get("total_count") + if not isinstance(total, int) or total < 0: + raise InventoryError(f"{name} workflow total_count is malformed") + consumed = sum(len(item["workflows"]) for item in workflow_pages) + len( + body["workflows"] + ) + has_next = consumed < total + workflow_pages.append({**body, "_link_next": has_next}) + if not has_next: + break + if not body["workflows"]: + raise InventoryError(f"{name} workflow pagination is truncated") + workflow_page += 1 + end = _live_get(client, commit_path, receipts) + end_sha = end.get("sha") if isinstance(end, Mapping) else None + assert_default_branch_bound(start_sha, end_sha) + records.append( + { + "name": name, + "archived": False, + "default_branch": default_branch, + "default_branch_sha": start_sha, + "default_branch_sha_after": end_sha, + "tree_paths": tree_paths, + "workflow_pages": workflow_pages, + } + ) + return ( + { + "organization": organization, + "observed_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "repository_inventory_complete": True, + "repositories": records, + }, + receipts, + ) + + +def reject_forbidden_token(name: str) -> None: + """Refuse GitHub Copilot tokens and other forbidden credentials.""" + if name in FORBIDDEN_TOKENS: + raise InventoryError(f"{name} is forbidden for this inventory") + + +def refuse_registry_mutation(action: str) -> None: + """Keep disablement on a separately reviewed operator path.""" + raise InventoryError( + f"registry mutation {action!r} is out of scope for this scanner" + ) + + +def is_exact_sha(value: object) -> bool: + """Return whether *value* is a 40-character lowercase hex SHA.""" + return isinstance(value, str) and HEX_SHA.fullmatch(value) is not None + + +def parse_link_has_next(link_header: object) -> bool: + """Return whether a GitHub ``Link`` header advertises another page.""" + if link_header is None: + return False + if not isinstance(link_header, str) or not link_header.strip(): + raise InventoryError("Link header is malformed") + return 'rel="next"' in link_header + + +def page_link_header(page: Mapping[str, Any]) -> object: + """Return one case-insensitive HTTP ``Link`` field or fail closed.""" + values = [ + value + for key, value in page.items() + if isinstance(key, str) and key.casefold() == "link" + ] + if len(values) > 1: + raise InventoryError("workflow page has ambiguous Link headers") + return values[0] if values else None + + +def decode_registry_path(path: object) -> str: + """Validate a workflow path and reject traversal disguises.""" + if not isinstance(path, str) or not path or "\x00" in path: + raise InventoryError("workflow path is missing or NUL-bearing") + if "\\" in path: + raise InventoryError("workflow path must not contain backslashes") + if "%" in path: + raise InventoryError("workflow path must not be percent-encoded") + parts = [part for part in path.split("/") if part not in {"", "."}] + if ".." in parts: + raise InventoryError("path traversal is not a workflow identity") + return path + + +def is_dynamic_owned_path(path: str) -> bool: + """Return whether *path* is a GitHub-owned dynamic workflow identity.""" + return path.startswith(DYNAMIC_PREFIXES) + + +def is_repository_workflow_path(path: str) -> bool: + """Return whether *path* is an exact repository workflow file.""" + if not path.startswith(".github/workflows/"): + return False + name = path[len(".github/workflows/") :] + return REPO_WORKFLOW_NAME.fullmatch(name) is not None + + +def interpret_status(status: int, *, resource: str) -> None: + """Fail closed on incomplete visibility or transient upstream errors.""" + if status == 200: + return + if status in {401, 403}: + raise InventoryError(f"permission loss for {resource}") + if status == 404: + raise InventoryError(f"missing visibility for {resource}") + if status >= 500: + raise InventoryError(f"transient upstream error {status} for {resource}") + raise InventoryError(f"unexpected status {status} for {resource}") + + +def fetch_with_one_retry( + fetch: Callable[[str], tuple[int, Any, Mapping[str, str]]], + url: str, +) -> tuple[Any, Mapping[str, str]]: + """GET JSON once, retry a 5xx exactly once, then fail closed.""" + status, body, headers = fetch(url) + if status >= 500: + status, body, headers = fetch(url) + interpret_status(status, resource=url) + return body, headers + + +def classify_workflow( + *, + path: str, + state: object, + source_present: bool | None, +) -> str: + """Classify one registry identity against the bound default-branch tree.""" + if is_dynamic_owned_path(path): + return "dynamic_owned" + if not is_repository_workflow_path(path): + return "unresolved" + if source_present is None: + return "unresolved" + if state in ACTIVE_STATES: + return "present_active" if source_present else "orphan_active" + if state in DISABLED_STATES: + return "present_disabled" if source_present else "orphan_disabled" + return "unresolved" + + +def assert_unique_workflow_ids(workflows: Sequence[Mapping[str, Any]]) -> None: + """Fail closed when a page set reuses a workflow id.""" + seen: set[int] = set() + for workflow in workflows: + workflow_id = workflow.get("id") + if not isinstance(workflow_id, int) or workflow_id <= 0: + raise InventoryError("workflow id must be a positive integer") + if workflow_id in seen: + raise InventoryError(f"reused workflow id {workflow_id}") + seen.add(workflow_id) + + +def collect_workflow_pages( + pages: Iterable[Mapping[str, Any]], + *, + per_page: int = PER_PAGE_DEFAULT, +) -> tuple[list[dict[str, Any]], int]: + """Merge workflow pages and return the number actually consumed.""" + if per_page <= 0: + raise InventoryError("per_page must be positive") + collected: list[dict[str, Any]] = [] + expected_total: int | None = None + saw_page = False + consumed_pages = 0 + open_next = False + for page in pages: + saw_page = True + if not isinstance(page, Mapping): + raise InventoryError("workflow page is not an object") + consumed_pages += 1 + workflows = page.get("workflows") + total_count = page.get("total_count") + if not isinstance(workflows, list): + raise InventoryError("workflow page is missing a workflows array") + if not isinstance(total_count, int) or total_count < 0: + raise InventoryError("workflow page total_count is invalid") + if expected_total is None: + expected_total = total_count + elif total_count != expected_total: + raise InventoryError("workflow total_count drifted across pages") + if len(workflows) > per_page: + raise InventoryError("workflow page exceeds per_page") + for workflow in workflows: + if not isinstance(workflow, dict): + raise InventoryError("workflow record is not an object") + collected.extend(workflows) + link_next = page.get("_link_next") + if link_next is None: + open_next = parse_link_has_next(page_link_header(page)) + elif isinstance(link_next, bool): + open_next = link_next + else: + raise InventoryError("workflow page _link_next must be boolean") + if open_next and not workflows: + raise InventoryError("empty workflow page advertised a next link") + if not open_next: + break + else: + if open_next: + raise InventoryError("pagination truncated after last next link") + if not saw_page: + raise InventoryError("no workflow pages") + if expected_total is None or len(collected) != expected_total: + raise InventoryError("pagination truncated or incomplete") + assert_unique_workflow_ids(collected) + return collected, consumed_pages + + +def assert_default_branch_bound(start_sha: object, end_sha: object) -> str: + """Bind the inventory to one default-branch SHA or abort.""" + if not is_exact_sha(start_sha) or not is_exact_sha(end_sha): + raise InventoryError("default-branch SHA is not a 40-character hex digest") + if start_sha != end_sha: + raise InventoryError("default-branch SHA moved during inventory") + return str(start_sha) + + +def owner_issue_for(repository: str) -> str | None: + """Return the known owning issue for a fleet repository, if any.""" + repository_key = repository.casefold() + return next( + ( + issue + for name, issue in KNOWN_OWNER_ISSUES.items() + if name.casefold() == repository_key + ), + None, + ) + + +def inventory_repository(record: Mapping[str, Any]) -> dict[str, Any]: + """Inventory one non-archived repository against its bound SHA.""" + name = record.get("name") + if not isinstance(name, str) or REPO_SLUG.fullmatch(name) is None: + raise InventoryError("repository name is not a valid slug") + if record.get("archived") is True: + return { + "repository": name, + "skipped": "archived", + "records": [], + } + if record.get("archived") is not False: + raise InventoryError(f"{name} archived flag is not boolean") + sha = assert_default_branch_bound( + record.get("default_branch_sha"), + record.get("default_branch_sha_after"), + ) + tree_paths = record.get("tree_paths") + if not isinstance(tree_paths, list) or any( + not isinstance(path, str) for path in tree_paths + ): + raise InventoryError(f"{name} tree_paths must be a list of strings") + tree = set(tree_paths) + pages = record.get("workflow_pages") + if not isinstance(pages, list): + raise InventoryError(f"{name} workflow_pages must be a list") + workflows, page_count = collect_workflow_pages(pages) + records: list[dict[str, Any]] = [] + for workflow in workflows: + path = decode_registry_path(workflow.get("path")) + source_present: bool | None + if is_repository_workflow_path(path): + source_present = path in tree + elif is_dynamic_owned_path(path): + source_present = None + else: + source_present = None + classification = classify_workflow( + path=path, + state=workflow.get("state"), + source_present=source_present, + ) + item = { + "repository": name, + "workflow_id": workflow.get("id"), + "name": workflow.get("name"), + "path": path, + "state": workflow.get("state"), + "default_branch_sha": sha, + "classification": classification, + } + if classification in {"orphan_active", "orphan_disabled"}: + owner = owner_issue_for(name) + if owner is not None: + item["owner_issue"] = owner + records.append(item) + return { + "repository": name, + "default_branch": record.get("default_branch"), + "default_branch_sha": sha, + "page_count": page_count, + "workflow_count": len(records), + "records": records, + } + + +def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + """Build an object while rejecting duplicate JSON keys.""" + seen: set[str] = set() + result: dict[str, Any] = {} + for key, value in pairs: + if key in seen: + raise InventoryError(f"duplicate object key {key!r}") + seen.add(key) + result[key] = value + return result + + +def load_payload_bytes(raw: bytes) -> dict[str, Any]: + """Parse a bounded UTF-8 inventory payload and reject duplicate keys.""" + if not raw: + raise InventoryError("payload is empty") + if len(raw) > MAX_PAYLOAD_BYTES: + raise InventoryError("payload exceeds 1048576 bytes") + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as exc: + raise InventoryError("payload is not UTF-8") from exc + try: + payload = json.loads(text, object_pairs_hook=reject_duplicate_keys) + except json.JSONDecodeError as exc: + raise InventoryError(f"payload is not JSON: {exc}") from exc + if not isinstance(payload, dict): + raise InventoryError("payload must be a JSON object") + return payload + + +def inventory_organization(payload: Mapping[str, Any]) -> dict[str, Any]: + """Inventory every supplied repository and emit an immutable ledger.""" + organization = payload.get("organization") + if organization != "ContextualWisdomLab": + raise InventoryError("organization must be ContextualWisdomLab") + observed_at = payload.get("observed_at") + if not isinstance(observed_at, str) or not observed_at: + raise InventoryError("observed_at is required") + repositories = payload.get("repositories") + if not isinstance(repositories, list) or not repositories: + raise InventoryError("repositories must be a non-empty list") + if payload.get("repository_inventory_complete") is not True: + raise InventoryError( + "repository inventory is incomplete; caller must prove full visibility" + ) + inventories: list[dict[str, Any]] = [] + for record in repositories: + if not isinstance(record, Mapping): + raise InventoryError("repository record is not an object") + inventories.append(inventory_repository(record)) + records = [ + item for inventory in inventories for item in inventory.get("records", []) + ] + counts = {name: 0 for name in CLASSIFICATIONS} + for item in records: + classification = item["classification"] + if classification not in counts: + raise InventoryError(f"unknown classification {classification!r}") + counts[classification] += 1 + return { + "schema_version": SCHEMA_VERSION, + "capability": CAPABILITY, + "organization": organization, + "observed_at": observed_at, + "repository_inventory_complete": True, + "assurance_posture": { + "csap": "design_constraint", + "soc2": "design_constraint", + "certification_claim": False, + "operational_pii_mask": False, + }, + "counts": counts, + "repositories": inventories, + "records": records, + } + + +def write_ledger(ledger: Mapping[str, Any], output: Path | None) -> str: + """Serialize the ledger as UTF-8 JSON with a trailing newline.""" + text = json.dumps(ledger, indent=2, sort_keys=True) + "\n" + if output is not None: + output.write_text(text, encoding="utf-8") + return text + + +def main(argv: Sequence[str] | None = None) -> int: + """Load a fixture payload, emit a ledger, and optionally fail on orphans.""" + parser = argparse.ArgumentParser( + description="Classify GitHub Actions workflow registry identities." + ) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--payload", help="JSON inventory fixture") + source.add_argument( + "--live", action="store_true", help="collect the live organization inventory" + ) + parser.add_argument("--output", help="optional ledger output path") + parser.add_argument( + "--receipt-output", help="required API receipt output for --live" + ) + parser.add_argument( + "--failure-output", help="structured failure evidence for --live" + ) + parser.add_argument( + "--fail-on-orphan-active", + action="store_true", + help="exit 1 when any orphan_active identity is observed", + ) + parser.add_argument( + "--mutate", + help=argparse.SUPPRESS, + ) + args = parser.parse_args(argv) + if args.mutate: + try: + refuse_registry_mutation(args.mutate) + except InventoryError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + live_receipts: list[dict[str, Any]] = [] + try: + if args.live: + if not args.receipt_output: + raise InventoryError("--receipt-output is required for --live") + try: + from scripts.ci.organization_commercial_readiness_loop import ( # pylint: disable=import-outside-toplevel + GitHubClient, + ) + except ModuleNotFoundError: # pragma: no cover - direct-script smoke tested + from organization_commercial_readiness_loop import GitHubClient # type: ignore[no-redef] # pylint: disable=import-outside-toplevel # pragma: no cover + + try: + client = GitHubClient.from_environment() + except Exception as exc: + raise InventoryError( + f"live GitHub credential unavailable: {type(exc).__name__}" + ) from exc + payload, receipts = collect_live_organization( + client, receipts=live_receipts + ) + Path(args.receipt_output).write_text( + json.dumps(receipts, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + else: + raw = Path(args.payload).read_bytes() + payload = load_payload_bytes(raw) + ledger = inventory_organization(payload) + except FileNotFoundError as exc: + print(f"ERROR: payload not found: {exc}", file=sys.stderr) + return 2 + except OSError as exc: + print(f"ERROR: unable to read payload: {exc}", file=sys.stderr) + return 2 + except InventoryError as exc: + if args.live: + if args.receipt_output: + Path(args.receipt_output).write_text( + json.dumps(live_receipts, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + if args.failure_output: + Path(args.failure_output).write_text( + json.dumps( + { + "capability": CAPABILITY, + "status": "failed", + "error_type": type(exc).__name__, + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + try: + text = write_ledger(ledger, Path(args.output) if args.output else None) + except (FileNotFoundError, OSError) as exc: + print(f"ERROR: unable to write ledger: {exc}", file=sys.stderr) + return 2 + if args.output is None: + sys.stdout.write(text) + orphan_active = ledger["counts"]["orphan_active"] + if args.fail_on_orphan_active and orphan_active: + print( + f"FAIL: {orphan_active} orphan_active workflow identit(y/ies)", + file=sys.stderr, + ) + return 1 + print( + f"PASS: inventoried {len(ledger['records'])} identities " + f"({orphan_active} orphan_active)", + file=sys.stderr, + ) + return 0 + + +if __name__ == "__main__": # pragma: no cover - exercised through main() + raise SystemExit(main()) diff --git a/scripts/ci/test_strix_quick_gate.sh b/scripts/ci/test_strix_quick_gate.sh index b9b1c43de3..5e62bc3db6 100755 --- a/scripts/ci/test_strix_quick_gate.sh +++ b/scripts/ci/test_strix_quick_gate.sh @@ -1566,6 +1566,7 @@ assert_pr_review_merge_scheduler_uses_github_actions_bot_token() { assert_file_not_contains "$workflow_file" "github.event.pull_request.number == 240" "scheduler must not hard-code repository-specific PR bypasses" assert_file_contains "$workflow_file" "github.event_name == 'pull_request_target' && format('pr-{0}', github.event.pull_request.number)" "scheduler scopes pull_request_target concurrency to the active PR" assert_file_contains "$workflow_file" "github.event_name == 'schedule' && format('schedule-{0}', github.event.schedule)" "scheduler isolates repository-local recovery from PR runs" + assert_file_contains "$workflow_file" "github.event_name == 'repository_dispatch' && format('repo-dispatch-{0}', github.repository)" "scheduler keeps manual queue scans isolated per repository dispatch target" assert_file_contains "$workflow_file" "github.event_name == 'repository_dispatch' && github.event.client_payload.target_repository != '' && github.event.client_payload.pr_number != ''" "scheduler scopes targeted manual queue scans to the requested PR" assert_file_contains "$workflow_file" "cancel-in-progress: \${{ github.event_name == 'pull_request_target' || github.event_name == 'pull_request_review' || github.event_name == 'repository_dispatch' }}" "scheduler cancels stale PR/review/manual queue scans instead of accumulating merge/update attempts" assert_file_not_contains "$workflow_file" 'github.event.workflow_run' "scheduler does not poll required-check completion through follow-up workflow runs" diff --git a/scripts/ci/workflow_lifecycle_operator.py b/scripts/ci/workflow_lifecycle_operator.py new file mode 100644 index 0000000000..0e5dbabe19 --- /dev/null +++ b/scripts/ci/workflow_lifecycle_operator.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Reviewed write primitives for confirmed workflow-lifecycle findings.""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from scripts.ci.inventory_orphaned_workflows import ( + HEX_SHA256, + REPO_SLUG, + GitHubTransport, + InventoryError, + assert_default_branch_bound, + owner_issue_for, +) + + +def disable_confirmed_orphan( + client: GitHubTransport, + record: Mapping[str, Any], + *, + confirmed_head_sha: str, +) -> None: + """Disable one reviewed ledger orphan after a fresh exact-head check.""" + if record.get("classification") != "orphan_active": + raise InventoryError("operator may disable only a ledger orphan_active record") + repository = record.get("repository") + workflow_id = record.get("workflow_id") + ledger_sha = record.get("default_branch_sha") + if not isinstance(repository, str) or not isinstance(workflow_id, int): + raise InventoryError("operator record identity is malformed") + assert_default_branch_bound(ledger_sha, confirmed_head_sha) + try: + client.request( + f"/repos/ContextualWisdomLab/{repository}/actions/workflows/{workflow_id}/disable", + method="PUT", + ) + except Exception as exc: + raise InventoryError( + f"operator disable failed closed: {type(exc).__name__}" + ) from exc + + +def publish_owner_issue( + client: GitHubTransport, + record: Mapping[str, Any], + *, + ledger_sha256: str, +) -> str: + """Create or update the bounded owner issue for one confirmed orphan.""" + if ( + record.get("classification") != "orphan_active" + or HEX_SHA256.fullmatch(ledger_sha256) is None + ): + raise InventoryError( + "issue publication requires an orphan_active and ledger digest" + ) + repository = record.get("repository") + if not isinstance(repository, str) or REPO_SLUG.fullmatch(repository) is None: + raise InventoryError("issue publication repository is malformed") + issue = owner_issue_for(repository) + body = ( + "\n" + f"Exact workflow registry evidence: `{record.get('workflow_id')}` / " + f"`{record.get('path')}` at `{record.get('default_branch_sha')}`.\n" + f"Ledger SHA-256: `{ledger_sha256}`.\n" + ) + try: + if issue is not None: + number = issue.rsplit("#", 1)[1] + client.request( + f"/repos/ContextualWisdomLab/{repository}/issues/{number}/comments", + method="POST", + payload={"body": body}, + ) + return issue + created = client.request( + f"/repos/ContextualWisdomLab/{repository}/issues", + method="POST", + payload={ + "title": "Disable orphaned workflow registry identity", + "body": body, + }, + ) + except Exception as exc: + raise InventoryError( + f"owner issue publication failed closed: {type(exc).__name__}" + ) from exc + number = created.get("number") if isinstance(created, Mapping) else None + if not isinstance(number, int) or number <= 0: + raise InventoryError("owner issue creation returned no issue number") + return f"ContextualWisdomLab/{repository}#{number}" diff --git a/tests/test_inventory_orphaned_workflows.py b/tests/test_inventory_orphaned_workflows.py new file mode 100644 index 0000000000..ad9c2d0837 --- /dev/null +++ b/tests/test_inventory_orphaned_workflows.py @@ -0,0 +1,1329 @@ +"""Fail-closed contracts for the read-only workflow-lifecycle inventory.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from scripts.ci import inventory_orphaned_workflows as inventory +from scripts.ci import workflow_lifecycle_operator as operator + +SHA = "a" * 40 +SHA_B = "b" * 40 + + +def _workflow( + workflow_id: int, + path: str, + *, + state: str = "active", + name: str | None = None, +) -> dict[str, Any]: + """Return one GitHub Actions workflow registry record.""" + return { + "id": workflow_id, + "name": name or Path(path).name, + "path": path, + "state": state, + } + + +def _repo( + name: str, + workflows: list[dict[str, Any]], + tree_paths: list[str], + *, + archived: bool = False, + sha: str = SHA, + sha_after: str | None = None, + pages: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Return one repository inventory fixture.""" + return { + "name": name, + "archived": archived, + "default_branch": "main", + "default_branch_sha": sha, + "default_branch_sha_after": sha if sha_after is None else sha_after, + "tree_paths": tree_paths, + "workflow_pages": pages + if pages is not None + else [ + {"total_count": len(workflows), "workflows": workflows, "_link_next": False} + ], + } + + +def _payload(repositories: list[dict[str, Any]]) -> dict[str, Any]: + """Return an organization inventory fixture.""" + return { + "organization": "ContextualWisdomLab", + "observed_at": "2026-08-16T12:00:00Z", + "repository_inventory_complete": True, + "repositories": repositories, + } + + +def test_reject_forbidden_token_and_allow_unrelated_names() -> None: + """Copilot tokens are forbidden; unrelated credential names are ignored.""" + inventory.reject_forbidden_token("NVIDIA_NIM_API_KEY") + with pytest.raises(inventory.InventoryError, match="COPILOT_GITHUB_TOKEN"): + inventory.reject_forbidden_token("COPILOT_GITHUB_TOKEN") + + +def test_refuse_registry_mutation() -> None: + """Disablement stays on a separately reviewed operator path.""" + with pytest.raises(inventory.InventoryError, match="disable"): + inventory.refuse_registry_mutation("disable") + + +def test_is_exact_sha() -> None: + """Only a 40-character lowercase hex digest is a bound SHA.""" + assert inventory.is_exact_sha(SHA) is True + assert inventory.is_exact_sha("A" * 40) is False + assert inventory.is_exact_sha(1) is False + assert inventory.is_exact_sha("deadbeef") is False + + +def test_parse_link_has_next() -> None: + """Link pagination is boolean and fail-closed on malformed headers.""" + assert inventory.parse_link_has_next(None) is False + assert inventory.parse_link_has_next('; rel="next"') is True + assert ( + inventory.parse_link_has_next('; rel="prev"') is False + ) + with pytest.raises(inventory.InventoryError, match="malformed"): + inventory.parse_link_has_next("") + with pytest.raises(inventory.InventoryError, match="malformed"): + inventory.parse_link_has_next(1) + + +def test_decode_registry_path_rejects_encoding_and_traversal() -> None: + """Percent-encoding, NULs, backslashes, and `..` fail closed.""" + assert ( + inventory.decode_registry_path(".github/workflows/ci.yml") + == ".github/workflows/ci.yml" + ) + with pytest.raises(inventory.InventoryError, match=r"NUL|missing"): + inventory.decode_registry_path("") + with pytest.raises(inventory.InventoryError, match=r"NUL|missing"): + inventory.decode_registry_path(".github/workflows/ci.yml\x00") + with pytest.raises(inventory.InventoryError, match="backslash"): + inventory.decode_registry_path(".github\\workflows\\ci.yml") + with pytest.raises(inventory.InventoryError, match="percent-encoded"): + inventory.decode_registry_path(".github/workflows/%2e%2e/ci.yml") + with pytest.raises(inventory.InventoryError, match="traversal"): + inventory.decode_registry_path(".github/workflows/../../secret.yml") + assert ( + inventory.decode_registry_path(".github/workflows//./ci.yml") + == ".github/workflows//./ci.yml" + ) + + +def test_path_predicates() -> None: + """Dynamic GitHub identities stay distinct from repository YAML files.""" + assert inventory.is_dynamic_owned_path("dynamic/pages/pages-build-deployment") + assert inventory.is_repository_workflow_path(".github/workflows/ci.yml") + assert inventory.is_repository_workflow_path(".github/workflows/ci.yaml") + assert not inventory.is_repository_workflow_path(".GitHub/workflows/ci.yml") + assert not inventory.is_repository_workflow_path(".github/workflows/nested/ci.yml") + assert not inventory.is_repository_workflow_path(".github/workflows/ci.txt") + assert not inventory.is_dynamic_owned_path(".github/workflows/ci.yml") + + +def test_interpret_status_and_single_retry() -> None: + """Visibility loss and 5xx fail closed after at most one retry.""" + inventory.interpret_status(200, resource="workflows") + with pytest.raises(inventory.InventoryError, match="permission"): + inventory.interpret_status(401, resource="workflows") + with pytest.raises(inventory.InventoryError, match="permission"): + inventory.interpret_status(403, resource="workflows") + with pytest.raises(inventory.InventoryError, match="missing visibility"): + inventory.interpret_status(404, resource="workflows") + with pytest.raises(inventory.InventoryError, match="transient"): + inventory.interpret_status(503, resource="workflows") + with pytest.raises(inventory.InventoryError, match="unexpected"): + inventory.interpret_status(418, resource="workflows") + + calls = {"n": 0} + + def once_ok(url: str) -> tuple[int, dict[str, str], dict[str, str]]: + calls["n"] += 1 + return 200, {"ok": url}, {"link": ""} + + body, headers = inventory.fetch_with_one_retry(once_ok, "https://example/ok") + assert body == {"ok": "https://example/ok"} + assert headers == {"link": ""} + assert calls["n"] == 1 + + def recover(url: str) -> tuple[int, dict[str, str], dict[str, str]]: + calls["n"] += 1 + if calls["n"] == 2: + return 503, {}, {} + return 200, {"recovered": True}, {} + + calls["n"] = 1 + body, _headers = inventory.fetch_with_one_retry(recover, "https://example/retry") + assert body == {"recovered": True} + + def stay_down(_url: str) -> tuple[int, dict[str, str], dict[str, str]]: + return 500, {}, {} + + with pytest.raises(inventory.InventoryError, match="transient"): + inventory.fetch_with_one_retry(stay_down, "https://example/down") + + +def test_classify_workflow_matrix() -> None: + """Every advertised class is produced from path, state, and tree presence.""" + repo = ".github/workflows/ci.yml" + assert ( + inventory.classify_workflow( + path="dynamic/pages/pages-build-deployment", + state="active", + source_present=None, + ) + == "dynamic_owned" + ) + assert ( + inventory.classify_workflow( + path=".GitHub/workflows/ci.yml", + state="active", + source_present=None, + ) + == "unresolved" + ) + assert ( + inventory.classify_workflow(path=repo, state="active", source_present=None) + == "unresolved" + ) + assert ( + inventory.classify_workflow(path=repo, state="active", source_present=True) + == "present_active" + ) + assert ( + inventory.classify_workflow(path=repo, state="active", source_present=False) + == "orphan_active" + ) + assert ( + inventory.classify_workflow( + path=repo, state="disabled_manually", source_present=True + ) + == "present_disabled" + ) + assert ( + inventory.classify_workflow( + path=repo, state="disabled_inactivity", source_present=False + ) + == "orphan_disabled" + ) + assert ( + inventory.classify_workflow(path=repo, state="mystery", source_present=True) + == "unresolved" + ) + + +def test_collect_workflow_pages_fail_closed() -> None: + """Partial pagination, drift, reuse, and empty next-pages fail closed.""" + first = { + "total_count": 2, + "workflows": [_workflow(1, ".github/workflows/a.yml")], + "_link_next": True, + } + second = { + "total_count": 2, + "workflows": [_workflow(2, ".github/workflows/b.yml")], + "_link_next": False, + } + workflows, page_count = inventory.collect_workflow_pages( + [first, second], per_page=1 + ) + assert len(workflows) == 2 + assert page_count == 2 + + with pytest.raises(inventory.InventoryError, match="per_page"): + inventory.collect_workflow_pages([], per_page=0) + with pytest.raises(inventory.InventoryError, match="no workflow pages"): + inventory.collect_workflow_pages([]) + with pytest.raises(inventory.InventoryError, match="not an object"): + inventory.collect_workflow_pages([None]) # type: ignore[list-item] + with pytest.raises(inventory.InventoryError, match="workflows array"): + inventory.collect_workflow_pages([{"total_count": 0}]) + with pytest.raises(inventory.InventoryError, match="total_count"): + inventory.collect_workflow_pages([{"total_count": -1, "workflows": []}]) + with pytest.raises(inventory.InventoryError, match="drifted"): + inventory.collect_workflow_pages( + [ + { + "total_count": 2, + "workflows": [_workflow(1, ".github/workflows/a.yml")], + "_link_next": True, + }, + { + "total_count": 3, + "workflows": [_workflow(2, ".github/workflows/b.yml")], + "_link_next": False, + }, + ], + per_page=1, + ) + with pytest.raises(inventory.InventoryError, match="exceeds per_page"): + inventory.collect_workflow_pages( + [ + { + "total_count": 2, + "workflows": [ + _workflow(1, ".github/workflows/a.yml"), + _workflow(2, ".github/workflows/b.yml"), + ], + "_link_next": False, + } + ], + per_page=1, + ) + with pytest.raises(inventory.InventoryError, match="not an object"): + inventory.collect_workflow_pages( + [{"total_count": 1, "workflows": ["bad"], "_link_next": False}] + ) + with pytest.raises(inventory.InventoryError, match="truncated"): + inventory.collect_workflow_pages( + [ + { + "total_count": 2, + "workflows": [_workflow(1, ".github/workflows/a.yml")], + "_link_next": False, + } + ] + ) + with pytest.raises(inventory.InventoryError, match="empty workflow page"): + inventory.collect_workflow_pages( + [{"total_count": 0, "workflows": [], "_link_next": True}] + ) + with pytest.raises(inventory.InventoryError, match="truncated after last"): + inventory.collect_workflow_pages( + [ + { + "total_count": 1, + "workflows": [_workflow(1, ".github/workflows/a.yml")], + "_link_next": True, + } + ] + ) + with pytest.raises(inventory.InventoryError, match="_link_next"): + inventory.collect_workflow_pages( + [{"total_count": 0, "workflows": [], "_link_next": "yes"}] + ) + linked = { + "total_count": 0, + "workflows": [], + "link": '; rel="prev"', + } + assert inventory.collect_workflow_pages([linked]) == ([], 1) + with pytest.raises(inventory.InventoryError, match="reused workflow id"): + inventory.collect_workflow_pages( + [ + { + "total_count": 2, + "workflows": [ + _workflow(7, ".github/workflows/a.yml"), + _workflow(7, ".github/workflows/renamed.yml"), + ], + "_link_next": False, + } + ] + ) + with pytest.raises(inventory.InventoryError, match="positive integer"): + inventory.collect_workflow_pages( + [ + { + "total_count": 1, + "workflows": [{"id": "7", "path": ".github/workflows/a.yml"}], + "_link_next": False, + } + ] + ) + + +def test_inventory_records_only_consumed_pages() -> None: + """The audit ledger never counts unconsumed trailing page fixtures.""" + terminal = { + "total_count": 1, + "workflows": [_workflow(1, ".github/workflows/ci.yml")], + "_link_next": False, + } + trailing = { + "total_count": 1, + "workflows": [_workflow(2, ".github/workflows/unread.yml")], + "_link_next": False, + } + + result = inventory.inventory_repository( + _repo( + "naruon", + [], + [".github/workflows/ci.yml"], + pages=[terminal, trailing], + ) + ) + + assert result["page_count"] == 1 + assert result["workflow_count"] == 1 + assert result["records"][0]["workflow_id"] == 1 + + +def test_collect_workflow_pages_accepts_case_insensitive_link_header() -> None: + """GitHub's canonical ``Link`` header spelling drives pagination.""" + first = { + "total_count": 2, + "workflows": [_workflow(1, ".github/workflows/a.yml")], + "Link": '; rel="next"', + } + second = { + "total_count": 2, + "workflows": [_workflow(2, ".github/workflows/b.yml")], + "Link": '; rel="prev"', + } + + workflows, page_count = inventory.collect_workflow_pages( + [first, second], per_page=1 + ) + + assert [workflow["id"] for workflow in workflows] == [1, 2] + assert page_count == 2 + + ambiguous = dict(first, link='; rel="next"') + with pytest.raises(inventory.InventoryError, match="ambiguous Link"): + inventory.collect_workflow_pages([ambiguous], per_page=1) + + +def test_assert_default_branch_bound() -> None: + """A moved or malformed default-branch SHA aborts the inventory.""" + assert inventory.assert_default_branch_bound(SHA, SHA) == SHA + with pytest.raises(inventory.InventoryError, match="moved"): + inventory.assert_default_branch_bound(SHA, SHA_B) + with pytest.raises(inventory.InventoryError, match="hex digest"): + inventory.assert_default_branch_bound("main", "main") + + +def test_owner_issue_for_known_fleet() -> None: + """Known fleet orphans route to the existing owner issues.""" + assert ( + inventory.owner_issue_for("appguardrail") + == "ContextualWisdomLab/appguardrail#929" + ) + assert inventory.owner_issue_for("naruon") == "ContextualWisdomLab/naruon#1324" + assert inventory.owner_issue_for("APPGUARDRAIL") == ( + "ContextualWisdomLab/appguardrail#929" + ) + + +def test_disabled_orphan_routes_to_owner_issue() -> None: + """Disabled orphan evidence keeps its explicit owner route.""" + result = inventory.inventory_repository( + _repo( + "appguardrail", + [_workflow(14, ".github/workflows/finalize-once.yml", state="deleted")], + [], + ) + ) + assert result["records"][0]["classification"] == "orphan_disabled" + assert result["records"][0]["owner_issue"] == ( + "ContextualWisdomLab/appguardrail#929" + ) + + +def test_owner_issue_registry_covers_confirmed_fleet() -> None: + """Every confirmed fleet owner has an explicit, linkable issue route.""" + assert inventory.KNOWN_OWNER_ISSUES == { + "appguardrail": "ContextualWisdomLab/appguardrail#929", + "bandscope": "ContextualWisdomLab/bandscope#847", + "clearfolio": "ContextualWisdomLab/clearfolio#423", + "codec-carver": "ContextualWisdomLab/codec-carver#401", + "contextual-orchestrator": "ContextualWisdomLab/contextual-orchestrator#122", + "DiagramWeave": "ContextualWisdomLab/DiagramWeave#27", + "disksage": "ContextualWisdomLab/disksage#191", + "EgressWeave": "ContextualWisdomLab/EgressWeave#202", + "fast-mlsirm": "ContextualWisdomLab/fast-mlsirm#809", + "four-pillars": "ContextualWisdomLab/four-pillars#33", + "inkspan": "ContextualWisdomLab/inkspan#278", + "keyverse": "ContextualWisdomLab/keyverse#99", + "naruon": "ContextualWisdomLab/naruon#1324", + "newsdom-api": "ContextualWisdomLab/newsdom-api#604", + "noema": "ContextualWisdomLab/noema#226", + "OriginWeave": "ContextualWisdomLab/OriginWeave#123", + "pg-erd-cloud": "ContextualWisdomLab/pg-erd-cloud#865", + "RankWeave": "ContextualWisdomLab/RankWeave#38", + "saju-caldav": "ContextualWisdomLab/saju-caldav#33", + "ThreadWeave": "ContextualWisdomLab/ThreadWeave#31", + } + + +def test_inventory_repository_classifies_known_shapes() -> None: + """One-shot names, orphans, dynamic workflows, and archives stay honest.""" + result = inventory.inventory_repository( + _repo( + "clearfolio", + [ + _workflow(1, ".github/workflows/one-shot-cleanup.yml"), + _workflow(2, ".github/workflows/missing.yml"), + _workflow( + 3, + "dynamic/pages/pages-build-deployment", + name="pages", + ), + _workflow( + 4, + ".github/workflows/old.yml", + state="disabled_manually", + ), + ], + [".github/workflows/one-shot-cleanup.yml"], + ) + ) + classes = { + item["workflow_id"]: item["classification"] for item in result["records"] + } + assert classes[1] == "present_active" + assert classes[2] == "orphan_active" + assert classes[3] == "dynamic_owned" + assert classes[4] == "orphan_disabled" + orphan = next(item for item in result["records"] if item["workflow_id"] == 2) + assert orphan["owner_issue"] == "ContextualWisdomLab/clearfolio#423" + + skipped = inventory.inventory_repository(_repo("old", [], [], archived=True)) + assert skipped["skipped"] == "archived" + assert skipped["records"] == [] + + +def test_inventory_routes_inkspan_orphan_to_owner_issue() -> None: + """Inkspan orphan evidence reaches its confirmed central owner issue.""" + result = inventory.inventory_repository( + _repo( + "inkspan", + [_workflow(20, ".github/workflows/apply-preparse-envelope-limits.yml")], + [], + ) + ) + assert result["records"][0]["classification"] == "orphan_active" + assert result["records"][0]["owner_issue"] == "ContextualWisdomLab/inkspan#278" + + +def test_inventory_repository_rejects_malformed_records() -> None: + """Malformed repository fixtures fail closed before classification.""" + with pytest.raises(inventory.InventoryError, match="valid slug"): + inventory.inventory_repository(_repo("bad name", [], [])) + bad_flag = _repo("naruon", [], []) + bad_flag["archived"] = "no" + with pytest.raises(inventory.InventoryError, match="archived flag"): + inventory.inventory_repository(bad_flag) + bad_tree = _repo("naruon", [], []) + bad_tree["tree_paths"] = [1] + with pytest.raises(inventory.InventoryError, match="tree_paths"): + inventory.inventory_repository(bad_tree) + not_list = _repo("naruon", [], []) + not_list["tree_paths"] = ".github/workflows/ci.yml" + with pytest.raises(inventory.InventoryError, match="tree_paths"): + inventory.inventory_repository(not_list) + unnamed_orphan = inventory.inventory_repository( + _repo( + "unknown-repo", + [_workflow(8, ".github/workflows/missing.yml")], + [], + ) + ) + assert unnamed_orphan["records"][0]["classification"] == "orphan_active" + assert "owner_issue" not in unnamed_orphan["records"][0] + bad_pages = _repo("naruon", [], []) + bad_pages["workflow_pages"] = "pages" + with pytest.raises(inventory.InventoryError, match="workflow_pages"): + inventory.inventory_repository(bad_pages) + present_disabled = inventory.inventory_repository( + _repo( + "naruon", + [ + _workflow( + 9, + ".github/workflows/ci.yml", + state="disabled_fork", + ) + ], + [".github/workflows/ci.yml"], + ) + ) + assert present_disabled["records"][0]["classification"] == "present_disabled" + unresolved = inventory.inventory_repository( + _repo( + "naruon", + [_workflow(10, "not-a-workflow")], + [], + ) + ) + assert unresolved["records"][0]["classification"] == "unresolved" + + +def test_payload_loading_rejects_duplicates_and_bounds() -> None: + """Empty, oversized, non-UTF-8, non-object, and duplicate-key payloads fail.""" + with pytest.raises(inventory.InventoryError, match="empty"): + inventory.load_payload_bytes(b"") + with pytest.raises(inventory.InventoryError, match="exceeds"): + inventory.load_payload_bytes(b"{" + b"a" * (inventory.MAX_PAYLOAD_BYTES + 1)) + with pytest.raises(inventory.InventoryError, match="UTF-8"): + inventory.load_payload_bytes(b"\xff") + with pytest.raises(inventory.InventoryError, match="not JSON"): + inventory.load_payload_bytes(b"{") + with pytest.raises(inventory.InventoryError, match="JSON object"): + inventory.load_payload_bytes(b"[]") + with pytest.raises(inventory.InventoryError, match="duplicate object key"): + inventory.reject_duplicate_keys([("a", 1), ("a", 2)]) + payload = inventory.load_payload_bytes(b'{"organization":"ContextualWisdomLab"}') + assert payload["organization"] == "ContextualWisdomLab" + + +def test_inventory_organization_and_unknown_classification( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Organization ledgers count classes and reject unknown ones.""" + payload = _payload( + [ + _repo( + "disksage", + [_workflow(1, ".github/workflows/gone.yml")], + [], + ) + ] + ) + ledger = inventory.inventory_organization(payload) + assert ledger["schema_version"] == "1" + assert ledger["repository_inventory_complete"] is True + assert ledger["assurance_posture"]["certification_claim"] is False + assert ledger["assurance_posture"]["operational_pii_mask"] is False + assert ledger["counts"]["orphan_active"] == 1 + assert ledger["records"][0]["owner_issue"] == "ContextualWisdomLab/disksage#191" + + with pytest.raises(inventory.InventoryError, match="organization"): + inventory.inventory_organization({"organization": "other"}) + with pytest.raises(inventory.InventoryError, match="observed_at"): + inventory.inventory_organization( + {"organization": "ContextualWisdomLab", "observed_at": ""} + ) + with pytest.raises(inventory.InventoryError, match="non-empty"): + inventory.inventory_organization( + { + "organization": "ContextualWisdomLab", + "observed_at": "2026-08-16T12:00:00Z", + "repositories": [], + } + ) + with pytest.raises(inventory.InventoryError, match="not an object"): + inventory.inventory_organization( + { + "organization": "ContextualWisdomLab", + "observed_at": "2026-08-16T12:00:00Z", + "repository_inventory_complete": True, + "repositories": ["naruon"], + } + ) + + def lie(**_kwargs: object) -> str: + return "not-a-class" + + monkeypatch.setattr(inventory, "classify_workflow", lie) + with pytest.raises(inventory.InventoryError, match="unknown classification"): + inventory.inventory_organization( + _payload( + [ + _repo( + "naruon", + [_workflow(1, ".github/workflows/ci.yml")], + [".github/workflows/ci.yml"], + ) + ] + ) + ) + + +def test_inventory_requires_complete_repository_visibility() -> None: + """A partial organization repository list cannot produce an audit ledger.""" + payload = { + "organization": "ContextualWisdomLab", + "observed_at": "2026-08-16T12:00:00Z", + "repositories": [_repo("naruon", [], [])], + } + with pytest.raises(inventory.InventoryError, match="repository inventory"): + inventory.inventory_organization(payload) + + payload["repository_inventory_complete"] = False + with pytest.raises(inventory.InventoryError, match="repository inventory"): + inventory.inventory_organization(payload) + + +def test_write_ledger_and_main( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """The CLI writes a ledger, fails closed, and never mutates the registry.""" + payload = _payload( + [ + _repo( + "appguardrail", + [ + _workflow(1, ".github/workflows/apply-once.yml"), + _workflow(2, ".github/workflows/ci.yml"), + ], + [".github/workflows/ci.yml"], + ) + ] + ) + payload_path = tmp_path / "payload.json" + payload_path.write_text(json.dumps(payload), encoding="utf-8") + output = tmp_path / "ledger.json" + assert ( + inventory.main(["--payload", str(payload_path), "--output", str(output)]) == 0 + ) + ledger = json.loads(output.read_text(encoding="utf-8")) + assert ledger["counts"]["orphan_active"] == 1 + assert ledger["counts"]["present_active"] == 1 + err = capsys.readouterr().err + assert "PASS:" in err + + assert ( + inventory.main(["--payload", str(payload_path), "--fail-on-orphan-active"]) == 1 + ) + captured = capsys.readouterr() + assert "FAIL:" in captured.err + assert '"schema_version"' in captured.out + + assert inventory.main(["--payload", str(payload_path), "--mutate", "disable"]) == 2 + assert "registry mutation" in capsys.readouterr().err + + missing = tmp_path / "missing.json" + assert inventory.main(["--payload", str(missing)]) == 2 + assert "not found" in capsys.readouterr().err + + directory = tmp_path / "dir" + directory.mkdir() + assert inventory.main(["--payload", str(directory)]) == 2 + assert "unable to read payload" in capsys.readouterr().err + + payload_path.write_text("[]", encoding="utf-8") + assert inventory.main(["--payload", str(payload_path)]) == 2 + assert "JSON object" in capsys.readouterr().err + + +def test_main_reports_ledger_output_failures_separately( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """The CLI identifies an unwritable ledger path as an output failure.""" + payload_path = tmp_path / "payload.json" + payload_path.write_text( + json.dumps(_payload([_repo("naruon", [], [])])), encoding="utf-8" + ) + output = tmp_path / "missing" / "ledger.json" + + assert ( + inventory.main(["--payload", str(payload_path), "--output", str(output)]) == 2 + ) + error = capsys.readouterr().err + assert "unable to write ledger" in error + assert "unable to read payload" not in error + + +def test_example_fixture_classifies_without_masking_identities() -> None: + """The committed example ledger fixture is executable and unredacted.""" + raw = Path( + "schemas/examples/cwl-workflow-lifecycle-ledger-v1.example.json" + ).read_bytes() + ledger = inventory.inventory_organization(inventory.load_payload_bytes(raw)) + assert ledger["assurance_posture"]["operational_pii_mask"] is False + classes = {item["path"]: item["classification"] for item in ledger["records"]} + assert classes[".github/workflows/finalize-once.yml"] == "orphan_active" + assert classes[".github/workflows/ci.yml"] == "present_active" + assert classes["dynamic/pages/pages-build-deployment"] == "dynamic_owned" + + +class _LiveClient: + """Record deterministic live API calls for collector tests.""" + + def __init__(self, responses: dict[str, Any]) -> None: + self.responses = responses + self.calls: list[str] = [] + + def request(self, path: str, *, method: str = "GET", payload: Any = None) -> Any: + """Return one canned response and retain its exact request path.""" + if method in {"PUT", "POST"}: + self.calls.append(path) + return self.responses.get(path) + assert method == "GET" + assert payload is None + self.calls.append(path) + if path == "/orgs/ContextualWisdomLab" and path not in self.responses: + visible = sum( + len(value) + for key, value in self.responses.items() + if key.startswith("/orgs/ContextualWisdomLab/repos?") + and isinstance(value, list) + ) + return {"public_repos": visible, "total_private_repos": 0} + value = self.responses[path] + if path.endswith("/commits/main") and isinstance(value, list): + return value.pop(0) + return value + + +def test_collect_live_organization_paginates_and_rechecks_head() -> None: + """The live collector consumes every page and binds both head reads.""" + repo = "ContextualWisdomLab/appguardrail" + responses = { + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1": [ + { + "name": "appguardrail", + "full_name": repo, + "archived": False, + "default_branch": "main", + } + ], + f"/repos/{repo}/commits/main": [{"sha": SHA}, {"sha": SHA}], + f"/repos/{repo}/git/trees/{SHA}?recursive=1": { + "truncated": False, + "tree": [{"type": "blob", "path": ".github/workflows/ci.yml"}], + }, + f"/repos/{repo}/actions/workflows?per_page=100&page=1": { + "total_count": 2, + "workflows": [ + _workflow(1, ".github/workflows/ci.yml"), + _workflow(2, ".github/workflows/gone.yml"), + ], + }, + } + client = _LiveClient(responses) + payload, receipts = inventory.collect_live_organization(client) + ledger = inventory.inventory_organization(payload) + assert ledger["counts"]["orphan_active"] == 1 + assert payload["repository_inventory_complete"] is True + assert len(receipts) == len(client.calls) + assert client.calls.count(f"/repos/{repo}/commits/main") == 2 + + +def test_collect_live_organization_requires_org_wide_visibility_proof() -> None: + """A syntactically complete visible page cannot hide unselected repositories.""" + org_path = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" + ) + client = _LiveClient( + { + org_path: [ + { + "name": "public", + "full_name": "ContextualWisdomLab/public", + "archived": True, + } + ], + "/orgs/ContextualWisdomLab": {"public_repos": 1, "total_private_repos": 2}, + } + ) + with pytest.raises(inventory.InventoryError, match="visibility proof"): + inventory.collect_live_organization(client) + + +def test_live_get_retries_one_actual_http_5xx() -> None: + """The production exception transport retries only one explicit HTTP 5xx.""" + + class Once: + calls = 0 + + def request(self, _path: str, **_kwargs: Any) -> Any: + self.calls += 1 + if self.calls == 1: + raise RuntimeError("GitHub API failed: server unavailable (HTTP 503)") + return {"ok": True} + + receipts: list[dict[str, Any]] = [] + client = Once() + assert inventory._live_get(client, "/test", receipts) == {"ok": True} + assert client.calls == 2 + assert len(receipts) == 1 + + class Down: + def request(self, _path: str, **_kwargs: Any) -> Any: + raise RuntimeError("GitHub API failed (HTTP 502)") + + with pytest.raises(inventory.InventoryError, match="RuntimeError"): + inventory._live_get(Down(), "/test", []) + + +@pytest.mark.parametrize( + "proof", + ([], {}, {"public_repos": "1", "total_private_repos": 0}), +) +def test_live_collector_rejects_malformed_visibility_proof(proof: object) -> None: + """Only numeric organization-wide repository totals prove completeness.""" + org_path = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" + ) + with pytest.raises(inventory.InventoryError, match="visibility proof"): + inventory.collect_live_organization( + _LiveClient( + { + org_path: [ + { + "name": "old", + "full_name": "ContextualWisdomLab/old", + "archived": True, + } + ], + "/orgs/ContextualWisdomLab": proof, + } + ) + ) + + +@pytest.mark.parametrize( + "entry", + ( + "not-an-object", + {"path": "x"}, + {"type": "mystery", "path": "x"}, + {"type": "blob"}, + ), +) +def test_live_tree_rejects_every_malformed_entry(entry: object) -> None: + """Malformed tree members never become negative source evidence.""" + repo = "ContextualWisdomLab/naruon" + org_path = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" + ) + responses = { + org_path: [ + { + "name": "naruon", + "full_name": repo, + "archived": False, + "default_branch": "main", + } + ], + f"/repos/{repo}/commits/main": [{"sha": SHA}, {"sha": SHA}], + f"/repos/{repo}/git/trees/{SHA}?recursive=1": { + "truncated": False, + "tree": [entry], + }, + f"/repos/{repo}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + with pytest.raises(inventory.InventoryError, match="tree entry"): + inventory.collect_live_organization(_LiveClient(responses)) + + +def test_live_tree_accepts_non_blob_entries_without_source_paths() -> None: + """Valid tree and submodule entries are checked but never workflow sources.""" + repo = "ContextualWisdomLab/naruon" + org_path = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" + ) + responses = { + org_path: [ + { + "name": "naruon", + "full_name": repo, + "archived": False, + "default_branch": "main", + } + ], + f"/repos/{repo}/commits/main": [{"sha": SHA}, {"sha": SHA}], + f"/repos/{repo}/git/trees/{SHA}?recursive=1": { + "truncated": False, + "tree": [ + {"type": "tree", "path": ".github"}, + {"type": "commit", "path": "vendor"}, + ], + }, + f"/repos/{repo}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + payload, _ = inventory.collect_live_organization(_LiveClient(responses)) + assert payload["repositories"][0]["tree_paths"] == [] + + +def test_collect_live_organization_fails_on_truncated_tree_or_head_move() -> None: + """Incomplete trees and a moving default branch never produce a ledger.""" + repo = "ContextualWisdomLab/naruon" + base = { + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1": [ + { + "name": "naruon", + "full_name": repo, + "archived": False, + "default_branch": "main", + } + ], + f"/repos/{repo}/commits/main": [{"sha": SHA}, {"sha": SHA_B}], + f"/repos/{repo}/git/trees/{SHA}?recursive=1": {"truncated": False, "tree": []}, + f"/repos/{repo}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + with pytest.raises(inventory.InventoryError, match="moved"): + inventory.collect_live_organization(_LiveClient(base)) + base[f"/repos/{repo}/commits/main"] = [{"sha": SHA}, {"sha": SHA}] + base[f"/repos/{repo}/git/trees/{SHA}?recursive=1"] = {"truncated": True, "tree": []} + with pytest.raises(inventory.InventoryError, match="truncated"): + inventory.collect_live_organization(_LiveClient(base)) + + +def test_operator_disable_is_ledger_and_head_bound() -> None: + """The mutation path accepts only a reviewed orphan on its unchanged head.""" + record = { + "repository": "appguardrail", + "workflow_id": 9, + "classification": "orphan_active", + "default_branch_sha": SHA, + } + client = _LiveClient({}) + operator.disable_confirmed_orphan(client, record, confirmed_head_sha=SHA) + assert client.calls == [ + "/repos/ContextualWisdomLab/appguardrail/actions/workflows/9/disable" + ] + with pytest.raises(inventory.InventoryError, match="only"): + operator.disable_confirmed_orphan( + client, + {**record, "classification": "present_active"}, + confirmed_head_sha=SHA, + ) + with pytest.raises(inventory.InventoryError, match="moved"): + operator.disable_confirmed_orphan(client, record, confirmed_head_sha=SHA_B) + + +def test_owner_issue_update_and_create_are_explicit() -> None: + """Known owners receive an update; unknown owners receive one bounded issue.""" + known = { + "repository": "appguardrail", + "workflow_id": 9, + "path": ".github/workflows/gone.yml", + "classification": "orphan_active", + "default_branch_sha": SHA, + } + client = _LiveClient({}) + assert operator.publish_owner_issue(client, known, ledger_sha256="c" * 64).endswith( + "#929" + ) + assert client.calls[-1].endswith("/issues/929/comments") + unknown = {**known, "repository": "new-product"} + create_path = "/repos/ContextualWisdomLab/new-product/issues" + creator = _LiveClient({create_path: {"number": 41}}) + assert operator.publish_owner_issue( + creator, unknown, ledger_sha256="d" * 64 + ).endswith("#41") + with pytest.raises(inventory.InventoryError, match="ledger digest"): + operator.publish_owner_issue(client, known, ledger_sha256="bad") + + +def test_live_collector_rejects_every_incomplete_api_shape() -> None: + """All malformed live inventory boundaries fail closed.""" + org_path = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" + ) + with pytest.raises(inventory.InventoryError, match="organization"): + inventory.collect_live_organization(_LiveClient({}), "other") + for response, message in [({}, "repository inventory"), ([], "empty")]: + with pytest.raises(inventory.InventoryError, match=message): + inventory.collect_live_organization(_LiveClient({org_path: response})) + + repo = "ContextualWisdomLab/naruon" + valid_repo = { + "name": "naruon", + "full_name": repo, + "archived": False, + "default_branch": "main", + } + + def client_for(**updates: Any) -> _LiveClient: + responses: dict[str, Any] = { + org_path: [valid_repo], + f"/repos/{repo}/commits/main": [{"sha": SHA}, {"sha": SHA}], + f"/repos/{repo}/git/trees/{SHA}?recursive=1": { + "truncated": False, + "tree": [], + }, + f"/repos/{repo}/actions/workflows?per_page=100&page=1": { + "total_count": 0, + "workflows": [], + }, + } + responses.update(updates) + return _LiveClient(responses) + + malformed_repositories = [ + ( + {"name": 1, "full_name": repo, "archived": False, "default_branch": "main"}, + "identity", + ), + ( + { + "name": "naruon", + "full_name": repo, + "archived": "no", + "default_branch": "main", + }, + "metadata", + ), + ] + for malformed, message in malformed_repositories: + with pytest.raises(inventory.InventoryError, match=message): + inventory.collect_live_organization(_LiveClient({org_path: [malformed]})) + archived, _ = inventory.collect_live_organization( + _LiveClient( + { + org_path: [ + { + "name": "old", + "full_name": "ContextualWisdomLab/old", + "archived": True, + } + ] + } + ) + ) + assert archived["repositories"][0]["archived"] is True + with pytest.raises(inventory.InventoryError, match="SHA is invalid"): + inventory.collect_live_organization( + client_for(**{f"/repos/{repo}/commits/main": [{"sha": "bad"}]}) + ) + with pytest.raises(inventory.InventoryError, match="tree is malformed"): + inventory.collect_live_organization( + client_for( + **{ + f"/repos/{repo}/git/trees/{SHA}?recursive=1": { + "truncated": False, + "tree": {}, + } + } + ) + ) + with pytest.raises(inventory.InventoryError, match="tree entry"): + inventory.collect_live_organization( + client_for( + **{ + f"/repos/{repo}/git/trees/{SHA}?recursive=1": { + "truncated": False, + "tree": [{"type": "blob", "path": 1}], + } + } + ) + ) + workflow_path = f"/repos/{repo}/actions/workflows?per_page=100&page=1" + for response, message in [ + ([], "workflow inventory"), + ({"workflows": [], "total_count": "0"}, "total_count"), + ({"workflows": [], "total_count": 1}, "pagination"), + ]: + with pytest.raises(inventory.InventoryError, match=message): + inventory.collect_live_organization(client_for(**{workflow_path: response})) + + +def test_live_transport_and_mutation_failures_are_redacted_by_type() -> None: + """Transport and write exceptions become bounded fail-closed errors.""" + + class Broken: + def request(self, *_args: Any, **_kwargs: Any) -> Any: + raise RuntimeError("secret detail") + + with pytest.raises(inventory.InventoryError, match="RuntimeError"): + inventory.collect_live_organization(Broken()) + record = { + "repository": "appguardrail", + "workflow_id": 9, + "path": ".github/workflows/gone.yml", + "classification": "orphan_active", + "default_branch_sha": SHA, + } + with pytest.raises(inventory.InventoryError, match="disable failed"): + operator.disable_confirmed_orphan(Broken(), record, confirmed_head_sha=SHA) + with pytest.raises(inventory.InventoryError, match="identity"): + operator.disable_confirmed_orphan( + Broken(), {**record, "workflow_id": "9"}, confirmed_head_sha=SHA + ) + with pytest.raises(inventory.InventoryError, match="publication failed"): + operator.publish_owner_issue(Broken(), record, ledger_sha256="e" * 64) + with pytest.raises(inventory.InventoryError, match="repository"): + operator.publish_owner_issue( + Broken(), {**record, "repository": "bad name"}, ledger_sha256="e" * 64 + ) + bad_create = _LiveClient({"/repos/ContextualWisdomLab/new-product/issues": {}}) + with pytest.raises(inventory.InventoryError, match="no issue number"): + operator.publish_owner_issue( + bad_create, {**record, "repository": "new-product"}, ledger_sha256="e" * 64 + ) + + +def test_live_collector_consumes_second_repository_and_workflow_pages() -> None: + """Full 100-item pages force the next API page instead of truncating.""" + first_org = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" + ) + second_org = ( + "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=2" + ) + archived = [ + { + "name": f"repo-{index}", + "full_name": f"ContextualWisdomLab/repo-{index}", + "archived": True, + } + for index in range(100) + ] + payload, _ = inventory.collect_live_organization( + _LiveClient({first_org: archived, second_org: []}) + ) + assert len(payload["repositories"]) == 100 + + repo = "ContextualWisdomLab/naruon" + first = [ + _workflow(index + 1, f".github/workflows/{index}.yml") for index in range(100) + ] + responses = { + first_org: [ + { + "name": "naruon", + "full_name": repo, + "archived": False, + "default_branch": "main", + } + ], + f"/repos/{repo}/commits/main": [{"sha": SHA}, {"sha": SHA}], + f"/repos/{repo}/git/trees/{SHA}?recursive=1": {"truncated": False, "tree": []}, + f"/repos/{repo}/actions/workflows?per_page=100&page=1": { + "total_count": 101, + "workflows": first, + }, + f"/repos/{repo}/actions/workflows?per_page=100&page=2": { + "total_count": 101, + "workflows": [_workflow(101, ".github/workflows/last.yml")], + }, + } + payload, _ = inventory.collect_live_organization(_LiveClient(responses)) + assert len(payload["repositories"][0]["workflow_pages"]) == 2 + + +def test_live_main_requires_receipts_and_writes_outputs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Live CLI fails without receipts and writes both evidence files on success.""" + assert inventory.main(["--live"]) == 2 + assert "receipt-output" in capsys.readouterr().err + import scripts.ci.organization_commercial_readiness_loop as readiness + + monkeypatch.setattr( + readiness.GitHubClient, "from_environment", classmethod(lambda cls: object()) + ) + payload = _payload([_repo("naruon", [], [])]) + monkeypatch.setattr( + inventory, + "collect_live_organization", + lambda _client, **_kwargs: (payload, [{"method": "GET"}]), + ) + output = tmp_path / "ledger.json" + receipts = tmp_path / "receipts.json" + assert ( + inventory.main( + ["--live", "--output", str(output), "--receipt-output", str(receipts)] + ) + == 0 + ) + assert json.loads(receipts.read_text())[0]["method"] == "GET" + monkeypatch.setattr( + readiness.GitHubClient, + "from_environment", + classmethod(lambda cls: (_ for _ in ()).throw(RuntimeError())), + ) + assert inventory.main(["--live", "--receipt-output", str(receipts)]) == 2 + assert "credential unavailable" in capsys.readouterr().err + + +def test_live_main_preserves_partial_receipts_and_failure_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed sweep still leaves completed reads and a structured failure artifact.""" + import scripts.ci.organization_commercial_readiness_loop as readiness + + monkeypatch.setattr( + readiness.GitHubClient, "from_environment", classmethod(lambda cls: object()) + ) + + def fail( + _client: object, *, receipts: list[dict[str, Any]] + ) -> tuple[dict[str, Any], list[dict[str, Any]]]: + receipts.append({"method": "GET", "path": "/completed", "sha256": "a" * 64}) + raise inventory.InventoryError("visibility proof unavailable") + + monkeypatch.setattr(inventory, "collect_live_organization", fail) + receipts = tmp_path / "receipts.json" + failure = tmp_path / "failure.json" + assert ( + inventory.main( + [ + "--live", + "--receipt-output", + str(receipts), + "--failure-output", + str(failure), + ] + ) + == 2 + ) + assert json.loads(receipts.read_text())[0]["path"] == "/completed" + assert json.loads(failure.read_text())["status"] == "failed" + + +def test_incomplete_inventory_fails_before_repository_classification( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The fleet completeness gate precedes all repository processing.""" + payload = _payload([_repo("naruon", [], [])]) + payload["repository_inventory_complete"] = False + monkeypatch.setattr( + inventory, + "inventory_repository", + lambda _record: pytest.fail( + "incomplete fleet reached repository classification" + ), + ) + with pytest.raises(inventory.InventoryError, match="incomplete"): + inventory.inventory_organization(payload) + + +def test_known_fleet_fixture_routes_owner_issues() -> None: + """The three named fleet incidents remain routed, not heuristically deleted.""" + payload = _payload( + [ + _repo( + "appguardrail", + [_workflow(11, ".github/workflows/finalize-once.yml")], + [], + ), + _repo( + "clearfolio", + [_workflow(12, ".github/workflows/one-shot-repair.yml")], + [".github/workflows/one-shot-repair.yml"], + ), + _repo( + "disksage", + [_workflow(13, ".github/workflows/pr-123-finalizer.yml")], + [], + ), + ] + ) + ledger = inventory.inventory_organization(payload) + by_repo = {item["repository"]: item for item in ledger["records"]} + assert by_repo["appguardrail"]["classification"] == "orphan_active" + assert by_repo["appguardrail"]["owner_issue"] == ( + "ContextualWisdomLab/appguardrail#929" + ) + assert by_repo["clearfolio"]["classification"] == "present_active" + assert "owner_issue" not in by_repo["clearfolio"] + assert by_repo["disksage"]["classification"] == "orphan_active" + assert by_repo["disksage"]["owner_issue"] == "ContextualWisdomLab/disksage#191" diff --git a/tests/test_pr_review_fix_scheduler_direct_rca_regressions.py b/tests/test_pr_review_fix_scheduler_direct_rca_regressions.py index da634bcfe4..f8b330d28a 100644 --- a/tests/test_pr_review_fix_scheduler_direct_rca_regressions.py +++ b/tests/test_pr_review_fix_scheduler_direct_rca_regressions.py @@ -95,6 +95,18 @@ def test_scan_queue_control_plane_failure_does_not_trigger_rca() -> None: assert fix.needs_rca_repair(pr) == (False, ()) +@pytest.mark.parametrize("draft", [True, False]) +def test_conflicted_pr_requires_nondraft_authorization(monkeypatch: Any, draft: bool) -> None: + """A draft or otherwise unauthorized conflict cannot dispatch repair.""" + pr = make_pr(is_draft=draft) + pr["mergeStateStatus"] = "DIRTY" + monkeypatch.setattr(fix, "needs_conflict_resolution", lambda *_args, **_kwargs: (False, ())) + args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main", "--dry-run"]) + action, reasons = fix.inspect_pr("owner/repo", pr, args) + assert action == "skip" + assert reasons == (("draft PR",) if draft else ("merge conflict is not authorized for repair",)) + + @pytest.mark.parametrize( "workflow_name", ["OpenCode Review", "Required OpenCode Review", "OpenCode PR Review"], diff --git a/tests/test_pr_review_fix_scheduler_rest_workflow_identity.py b/tests/test_pr_review_fix_scheduler_rest_workflow_identity.py index 4e36544061..bfbc0d787e 100644 --- a/tests/test_pr_review_fix_scheduler_rest_workflow_identity.py +++ b/tests/test_pr_review_fix_scheduler_rest_workflow_identity.py @@ -10,6 +10,25 @@ from scripts.ci import pr_review_merge_scheduler as merge +def test_workflow_name_rest_paginates_and_propagates_real_errors(monkeypatch: Any) -> None: + """REST workflow identity consumes full pages and fails on non-permission errors.""" + calls = {"count": 0} + + def pages(_path: str) -> Any: + calls["count"] += 1 + if calls["count"] == 1: + return {"workflow_runs": [{"check_suite_id": index, "name": f"workflow-{index}"} for index in range(100)]} + return {"workflow_runs": [{"check_suite_id": 101, "name": "last"}, {"check_suite_id": None, "name": "ignored"}, {"check_suite_id": 102, "name": ""}]} + + monkeypatch.setattr(merge, "gh_api_json", pages) + names = merge.fetch_workflow_names_by_check_suite_rest("owner/repo", "a" * 40) + assert names[101] == "last" + assert calls["count"] == 2 + monkeypatch.setattr(merge, "gh_api_json", lambda _path: (_ for _ in ()).throw(RuntimeError("boom"))) + with pytest.raises(RuntimeError, match="boom"): + merge.fetch_workflow_names_by_check_suite_rest("owner/repo", "a" * 40) + + def test_rest_fallback_preserves_renamed_opencode_workflow_identity( monkeypatch: Any, ) -> None: diff --git a/tests/test_workflow_lifecycle_inventory_workflow.py b/tests/test_workflow_lifecycle_inventory_workflow.py new file mode 100644 index 0000000000..bbf6cfb18b --- /dev/null +++ b/tests/test_workflow_lifecycle_inventory_workflow.py @@ -0,0 +1,30 @@ +"""Prevent mutation or credential regression in the lifecycle inventory.""" + +from pathlib import Path + + +def test_lifecycle_inventory_workflow_is_read_only_and_exact_head() -> None: + """The integrated sweep publishes receipts without registry mutation.""" + text = Path(".github/workflows/workflow-lifecycle-inventory.yml").read_text() + assert "contents: read" in text + assert "actions: read" in text + assert "persist-credentials: false" in text + assert 'test "$(git rev-parse HEAD)" = "$GITHUB_SHA"' in text + assert "--live" in text + assert "--receipt-output" in text + assert "--failure-output" in text + assert "retention-days: 30" in text + assert "secrets: inherit" not in text + assert "COPILOT_GITHUB_TOKEN" not in text + assert "/disable" not in text + assert "--mutate" not in text + + +def test_read_only_scanner_exports_no_write_primitive() -> None: + """Disable and issue writes live only in the reviewed operator module.""" + scanner = Path("scripts/ci/inventory_orphaned_workflows.py").read_text() + operator = Path("scripts/ci/workflow_lifecycle_operator.py").read_text() + assert "def disable_confirmed_orphan" not in scanner + assert "def publish_owner_issue" not in scanner + assert "def disable_confirmed_orphan" in operator + assert "def publish_owner_issue" in operator