diff --git a/CHANGELOG.d/1324-python-command-injection-detector.md b/CHANGELOG.d/1324-python-command-injection-detector.md new file mode 100644 index 000000000..51b58a23a --- /dev/null +++ b/CHANGELOG.d/1324-python-command-injection-detector.md @@ -0,0 +1,9 @@ +# Python shell-call detection is syntax-aware + +- Detect direct or module-qualified `os.system(...)` calls and `subprocess` + calls whose `shell` keyword is not statically false. +- Ignore comments, string literals, assignments, `shell=False`, `shell=0`, + and `shell=None` so non-executable text does not create CRITICAL findings. +- Preserve the legacy regex only as a fail-closed fallback for syntax-invalid + Python and retain exact source offsets across UTF-8 text, Python newline + boundaries, form feeds, and Unicode line separators. diff --git a/CHANGELOG.d/929-workflow-lifecycle-detection.md b/CHANGELOG.d/929-workflow-lifecycle-detection.md deleted file mode 100644 index 2bb5c58a7..000000000 --- a/CHANGELOG.d/929-workflow-lifecycle-detection.md +++ /dev/null @@ -1,14 +0,0 @@ -- Add read-only, exact-default-SHA GitHub Actions registry reconciliation and - structural detection for mutable-branch writers and self-modifying workflows. -- Avoid a false alarm for protected-branch publishers that push with `HEAD:`. -- Preserve both detections when YAML quotes `contents: write` or - `permissions: write-all` scalars. -- Detect quoted keys, flow-style write permissions, and continued mutable - pushes, simple event-derived variable indirection, and continued workflow - mutation without treating comments, echoed commands, run-body permission - strings, or ambiguous protected-branch refs as executable mutable writes. -- Keep live inventory functional from installed packages, revalidate workflow - registry identity and state, bound workflow pagination, and atomically - replace success or failure evidence so stale artifacts cannot contradict the - current result. Reject aliased artifact paths and bracket both registry reads - with the exact default-branch SHA. diff --git a/DOCUMENTATION.md b/DOCUMENTATION.md index 4d37bfc12..3f9bb5c4a 100644 --- a/DOCUMENTATION.md +++ b/DOCUMENTATION.md @@ -16,7 +16,6 @@ AppGuardrail has extensive rule, scanner, report, release, issue, and scheduler | Architecture decisions | [`docs/adr/README.md`](docs/adr/README.md) | | Security reporting | [`SECURITY.md`](SECURITY.md) | | Release automation | [`docs/release-automation.md`](docs/release-automation.md) | -| GitHub workflow lifecycle detection | [`docs/workflow-lifecycle-detection.md`](docs/workflow-lifecycle-detection.md) | | Productization roadmap | [`docs/product/2026-07-02-2b-krw-sale-readiness-plan.md`](docs/product/2026-07-02-2b-krw-sale-readiness-plan.md) | | Agent development rules | [`AGENTS.md`](AGENTS.md) | | Agent context | [`CLAUDE.md`](CLAUDE.md) | @@ -31,4 +30,4 @@ AppGuardrail has extensive rule, scanner, report, release, issue, and scheduler - **external-engine** β€” capability delegated to an optional scanner such as Semgrep/Trivy/Bandit/ZAP rather than AppGuardrail's lightweight built-in matcher. - **evidence-only** β€” information visible in reports/history but not yet executable as an AppGuardrail detector. -Critical current distinction: PR #911's no-exclusions issue-to-detector registry and executable obligation coverage remain **active-PR**, not protected-branch behavior. Stored-webhook SSRF prevention from PR #924 and bounded built-in detection from PR #910 are **implemented-main** as separate controls; the packaged rule covers its tested Python `set_webhook` persistence patterns and is not a universal SSRF taint-analysis claim. +Critical current distinction: PR #911's no-exclusions issue-to-detector registry and executable obligation coverage remain **active-PR**, not protected-branch behavior. Stored-webhook SSRF prevention from PR #924 and bounded built-in detection from PR #910 are **implemented-main** as separate controls; the packaged rule covers its tested Python `set_webhook` persistence patterns and is not a universal SSRF taint-analysis claim. \ No newline at end of file diff --git a/appguardrail_core/workflow_lifecycle.py b/appguardrail_core/workflow_lifecycle.py deleted file mode 100644 index ad56db0cc..000000000 --- a/appguardrail_core/workflow_lifecycle.py +++ /dev/null @@ -1,919 +0,0 @@ -"""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 os -import re -import sys -import tempfile -from collections.abc import Callable, Iterable, Mapping, Sequence -from contextlib import suppress -from datetime import UTC, datetime -from pathlib import Path -from typing import Any, Protocol -from urllib.error import HTTPError -from urllib.parse import quote -from urllib.request import HTTPRedirectHandler, Request, build_opener - -SCHEMA_VERSION = "1" -CAPABILITY = "workflow_lifecycle_inventory" -MAX_PAYLOAD_BYTES = 1_048_576 -PER_PAGE_DEFAULT = 100 -MAX_PAGES = 100 -GITHUB_API = "https://api.github.com" -USER_AGENT = "appguardrail-workflow-lifecycle" -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.""" - - -class _NoRedirect(HTTPRedirectHandler): - """Prevent an authenticated GitHub request from leaving its fixed origin.""" - - def redirect_request(self, req, fp, code, msg, headers, newurl): - """Reject redirects by declining to construct a follow-up request.""" - del req, fp, code, msg, headers, newurl - - -class _GitHubReadOnlyClient: - """Packaged, fixed-origin GitHub REST transport for live inventory.""" - - def __init__(self, token: str) -> None: - """Bind one token to the public GitHub API without mutation methods.""" - self._token = token - self._opener = build_opener(_NoRedirect) - - def request(self, path: str, *, method: str = "GET", payload: Any = None) -> Any: - """Return one decoded GET response from the fixed GitHub API origin.""" - if method != "GET" or payload is not None: - raise InventoryError("workflow lifecycle transport is read-only") - if not path.startswith("/") or path.startswith("//"): - raise InventoryError("GitHub API path must be origin-relative") - request = Request( - f"{GITHUB_API}{path}", - method="GET", - headers={ - "Accept": "application/vnd.github+json", - "Authorization": f"Bearer {self._token}", - "User-Agent": USER_AGENT, - "X-GitHub-Api-Version": "2022-11-28", - }, - ) - try: - with self._opener.open(request, timeout=30) as response: - body = response.read() - except HTTPError as exc: - raise RuntimeError(f"GitHub GET failed: {exc.code}") from exc - if not body: - return None - return json.loads(body.decode("utf-8")) - - -class _ReadOnlyGitHubAdapter: - """Enforce GET-only inventory at a second transport boundary.""" - - def __init__(self, client: Any) -> None: - """Wrap one fixed-origin AppGuardrail GitHub client.""" - self._client = client - - def request(self, path: str, *, method: str = "GET", payload: Any = None) -> Any: - """Permit only unactionable GETs through the shared GitHub transport.""" - if method != "GET" or payload is not None: - raise InventoryError("workflow lifecycle transport is read-only") - return self._client.request(path, method="GET") - - -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 |failed: )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 - - def list_repositories() -> list[Mapping[str, Any]]: - found: 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") - found.extend(batch) - if len(batch) < 100: - return found - page += 1 - if page > MAX_PAGES: - raise InventoryError("repository pagination exceeded limit") - - repositories = list_repositories() - 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" - ) - - def identities(items: list[Mapping[str, Any]]) -> set[tuple[object, ...]]: - try: - return { - ( - item.get("id"), - item.get("full_name"), - item.get("default_branch"), - item.get("archived"), - ) - for item in items - } - except TypeError as exc: - raise InventoryError("repository identity is malformed") from exc - - baseline = identities(repositories) - second = list_repositories() - if ( - len(baseline) != len(repositories) - or len(second) != len(repositories) - or identities(second) != baseline - ): - raise InventoryError("repository inventory changed during pagination") - - def list_workflows( - full_name: str, repository_name: str - ) -> tuple[list[dict[str, Any]], frozenset[tuple[object, object, object]]]: - pages: list[dict[str, Any]] = [] - page = 1 - consumed = 0 - while True: - if page > MAX_PAGES: - raise InventoryError( - f"{repository_name} workflow pagination exceeded limit" - ) - path = ( - f"/repos/{full_name}/actions/workflows?per_page=100&page={page}" - ) - body = _live_get(client, path, receipts) - if not isinstance(body, dict) or not isinstance( - body.get("workflows"), list - ): - raise InventoryError( - f"{repository_name} workflow inventory is malformed" - ) - total = body.get("total_count") - if not isinstance(total, int) or total < 0: - raise InventoryError( - f"{repository_name} workflow total_count is malformed" - ) - consumed += len(body["workflows"]) - has_next = consumed < total - pages.append({**body, "_link_next": has_next}) - if not has_next: - break - if not body["workflows"]: - raise InventoryError( - f"{repository_name} workflow pagination is truncated" - ) - page += 1 - workflows, _ = collect_workflow_pages(pages) - try: - workflow_identities = frozenset( - (item.get("id"), item.get("path"), item.get("state")) - for item in workflows - ) - except TypeError as exc: - raise InventoryError( - f"{repository_name} workflow identity is malformed" - ) from exc - return pages, workflow_identities - - 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, workflow_identities = list_workflows(full_name, name) - _, final_workflow_identities = list_workflows(full_name, name) - if final_workflow_identities != workflow_identities: - raise InventoryError(f"{name} workflow registry changed during scan") - 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, - } - ) - final = list_repositories() - if len(final) != len(repositories) or identities(final) != baseline: - raise InventoryError("repository inventory changed during scan") - 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: - _write_atomic_text(output, text) - return text - - -def _write_atomic_text(output: Path, text: str) -> None: - """Replace one evidence file only after its complete content is durable.""" - descriptor, temporary_name = tempfile.mkstemp( - dir=output.parent, prefix=f".{output.name}.", suffix=".tmp" - ) - temporary = Path(temporary_name) - try: - with os.fdopen(descriptor, "w", encoding="utf-8") as stream: - stream.write(text) - stream.flush() - os.fsync(stream.fileno()) - os.replace(temporary, output) - except Exception: - with suppress(OSError): - os.close(descriptor) - with suppress(OSError): - temporary.unlink(missing_ok=True) - raise - - -def write_failure_evidence( - receipt_output: str | None, - failure_output: str | None, - ledger_output: str | None, - receipts: list[dict[str, Any]], - error: Exception, -) -> None: - """Keep available evidence when a live collection or output write fails.""" - failure = { - "capability": CAPABILITY, - "status": "failed", - "error_type": type(error).__name__, - } - for path, content in ( - (receipt_output, receipts), - (failure_output, failure), - (ledger_output, failure), - ): - if path: - try: - _write_atomic_text( - Path(path), - json.dumps(content, indent=2, sort_keys=True) + "\n", - ) - except OSError: - # Best-effort preservation must not mask the failure already in flight. - pass - - -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.live: - evidence_paths = [ - Path(value) - for value in (args.output, args.receipt_output, args.failure_output) - if value - ] - resolved_paths = [path.resolve() for path in evidence_paths] - aliased = len(set(resolved_paths)) != len(resolved_paths) - if not aliased: - for index, left in enumerate(evidence_paths): - for right in evidence_paths[index + 1 :]: - if left.exists() and right.exists() and left.samefile(right): - aliased = True - break - if aliased: - break - if aliased: - print("ERROR: live evidence paths must be distinct", file=sys.stderr) - return 2 - 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: - token = os.environ.get("GITHUB_TOKEN", "") - if not token: - raise InventoryError("GITHUB_TOKEN is required for live inventory") - client = _ReadOnlyGitHubAdapter(_GitHubReadOnlyClient(token)) - 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 - ) - 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: - write_failure_evidence( - args.receipt_output, - args.failure_output, - args.output, - live_receipts, - exc, - ) - print(f"ERROR: {exc}", file=sys.stderr) - return 2 - if args.live: - try: - _write_atomic_text( - Path(args.receipt_output), - json.dumps(receipts, indent=2, sort_keys=True) + "\n", - ) - except OSError as exc: - write_failure_evidence( - None, args.failure_output, args.output, live_receipts, exc - ) - print(f"ERROR: unable to write receipts: {exc}", file=sys.stderr) - return 2 - try: - text = write_ledger(ledger, Path(args.output) if args.output else None) - except (FileNotFoundError, OSError) as exc: - if args.live: - write_failure_evidence( - None, args.failure_output, args.output, live_receipts, exc - ) - print(f"ERROR: unable to write ledger: {exc}", file=sys.stderr) - return 2 - if args.live and args.failure_output: - try: - Path(args.failure_output).unlink(missing_ok=True) - except OSError as exc: - write_failure_evidence(None, None, args.output, live_receipts, exc) - print(f"ERROR: unable to clear stale failure evidence: {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/docs/workflow-lifecycle-detection.md b/docs/workflow-lifecycle-detection.md deleted file mode 100644 index b730c2cc5..000000000 --- a/docs/workflow-lifecycle-detection.md +++ /dev/null @@ -1,54 +0,0 @@ -# GitHub Actions workflow lifecycle detection - -AppGuardrail compares the live GitHub Actions registry with the recursive tree -of one exact default-branch commit. The implementation in -`appguardrail_core.workflow_lifecycle` reuses the reviewed central contract from -`ContextualWisdomLab/.github#945` and PR `ContextualWisdomLab/.github#1026` -(`d17ff23c555580d6e7789049818a452686ae0cba`). It does not disable workflows, -delete source, write branches, or create issues. - -Run a live read-only organization inventory with a GitHub App installation -token that can prove complete organization visibility: - -```bash -GITHUB_TOKEN=... python -m appguardrail_core.workflow_lifecycle \ - --live \ - --output workflow-lifecycle-ledger.json \ - --receipt-output workflow-lifecycle-receipts.json \ - --failure-output workflow-lifecycle-failure.json \ - --fail-on-orphan-active -``` - -The installed package includes the fixed-origin, GET-only GitHub transport -used by this command; it does not depend on repository-local CI scripts. - -The detector paginates repositories and workflows, binds every repository to -an unchanged 40-character default-branch SHA, validates the full recursive -tree, re-reads workflow IDs, paths, and states before accepting the snapshot, -then rechecks the default-branch SHA, and records content hashes for API -receipts. Ledger, receipt, and failure paths must be distinct. Their files are -atomically replaced; a failed live run invalidates an existing clean ledger and -a successful run removes stale failure evidence. It separates -`present_active`, `present_disabled`, `orphan_active`, `orphan_disabled`, -`dynamic_owned`, and `unresolved`. Permission loss, 404 ambiguity, exhausted -5xx retry, malformed data, incomplete pagination, reused workflow IDs, and -branch movement fail closed rather than becoming clean evidence. - -The packaged source scanner separately reports two structural writer shapes: - -- `github-actions-mutable-branch-writer` detects a write token combined with a - push to an event-derived mutable branch. An explicit protected-branch target - does not trigger this rule; -- `github-actions-self-modifying-writer` detects a write-capable workflow that - edits or deletes `.github/workflows/**`. - -Names such as `once`, `apply`, or `repair` are not findings by themselves. A -supported live one-shot file is `present_active`; GitHub-owned `dynamic/**` -records remain `dynamic_owned`. Exact-ID disablement is an operator decision -only after a fresh complete inventory proves the file absent, no active job -owns the identity, and the lifecycle owner approves the action. Production -schedule, security, reusable, and release workflows must never be disabled by -name heuristics. - -An orphan count alone does not explain a queue incident. Queue causality needs -separate run, concurrency, runner, and occupancy evidence. diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py index 0d853d644..a3d0e2643 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -40,6 +40,7 @@ """ import argparse +import ast import fnmatch import functools import importlib.resources as resources # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2 @@ -311,6 +312,80 @@ def _console_print(*values, **kwargs) -> None: # Scan patterns # --------------------------------------------------------------------------- +_PYTHON_COMMAND_INJECTION_PATTERN = re.compile( + r"(?i)(?:os\.system\s*\(|subprocess\.(?:Popen|run|call|check_call|check_output)\s*\([^)]*shell\s*=\s*(?:True|1))" +) +_EMPTY_PATTERN = re.compile("") +_SUBPROCESS_SHELL_CALLS = frozenset( + {"Popen", "run", "call", "check_call", "check_output"} +) + + +def _iter_python_command_injection_matches(content: str): + """Yield match-compatible positions for statically identifiable shell calls. + + Valid Python is parsed so comments and string literals do not become + deploy-blocking findings. Syntax-invalid input falls back to the legacy + regular expression instead of silently dropping previously detectable + evidence. + """ + try: + tree = ast.parse(content) + except (SyntaxError, ValueError, RecursionError, MemoryError): + yield from _PYTHON_COMMAND_INJECTION_PATTERN.finditer(content) + return + + line_starts = [0] + line_starts.extend( + match.end() for match in re.finditer(r"\r\n|\r|\n", content) + ) + + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute): + continue + owner = node.func.value + owner_name = ( + owner.id + if isinstance(owner, ast.Name) + else owner.attr if isinstance(owner, ast.Attribute) else None + ) + + is_os_system = owner_name == "os" and node.func.attr == "system" + is_subprocess_shell = ( + owner_name == "subprocess" + and node.func.attr in _SUBPROCESS_SHELL_CALLS + and any( + keyword.arg == "shell" + and not ( + (isinstance(keyword.value, ast.Constant) and keyword.value.value in (False, 0, None, "", b"")) + or (isinstance(keyword.value, ast.Tuple) and not keyword.value.elts) + or (isinstance(keyword.value, ast.List) and not keyword.value.elts) + or (isinstance(keyword.value, ast.Dict) and not keyword.value.keys) + or (isinstance(keyword.value, ast.Set) and not keyword.value.elts) + ) + for keyword in node.keywords + ) + ) + if not (is_os_system or is_subprocess_shell): + continue + + line_start = line_starts[node.lineno - 1] + line_end = ( + line_starts[node.lineno] + if node.lineno < len(line_starts) + else len(content) + ) + line = content[line_start:line_end] + column = len( + line.encode("utf-8")[: node.col_offset].decode("utf-8", "ignore") + ) + match = _EMPTY_PATTERN.match( + content, line_start + column + ) + if match is not None: + yield match + + SCAN_RULES = [ { "id": "python-insecure-deserialization", @@ -453,11 +528,10 @@ def _console_print(*values, **kwargs) -> None: }, { "id": "python-command-injection", - "pattern": re.compile( - r"(?i)(?:os\.system|subprocess\.(?:Popen|run|call|check_call|check_output))\s*\([^)]*shell\s*=\s*True" - ), + "pattern": _PYTHON_COMMAND_INJECTION_PATTERN, + "finder": _iter_python_command_injection_matches, "severity": "CRITICAL", - "message": "Potential Command Injection detected: shell=True used in Python subprocess/os command. [OWASP A03:2021 - Injection]", + "message": "Potential shell command execution detected: os.system() or subprocess with a non-false shell argument. Verify that no untrusted data reaches the command. [OWASP A03:2021 - Injection]", "extensions": [".py"], }, { @@ -2155,7 +2229,7 @@ def _get_applicable_rules(ext: str): rule["id"], rule["severity"], rule["message"], - rule["pattern"].finditer, + rule["finder"] if "finder" in rule else rule["pattern"].finditer, tuple(rule.get("include_paths") or ()), tuple(rule.get("exclude_paths") or ()), tuple(rule.get("required_substrings") or ()), diff --git a/scanner/rules/github_actions.yml b/scanner/rules/github_actions.yml index c405808d3..643f8f235 100644 --- a/scanner/rules/github_actions.yml +++ b/scanner/rules/github_actions.yml @@ -1,44 +1,4 @@ rules: - # A write token plus an explicit push back to contributor-controlled branch - # state is materially different from a reviewed release publisher. Match the - # structure, not temporary-looking names such as `once` or `repair`. - - id: github-actions-mutable-branch-writer - patterns: - - pattern-regex: '\A(?=[\s\S]*^[ \t]*(?:permissions|"permissions"|''permissions'')[ \t]*:)(?=[\s\S]*^[ \t]*(?:(?:contents|"contents"|''contents'')[ \t]*:[ \t]*(?:write|"write"|''write'')|(?:permissions|"permissions"|''permissions'')[ \t]*:[ \t]*(?:(?:write-all|"write-all"|''write-all'')|\{[^}\n]*(?:contents|"contents"|''contents'')[ \t]*:[ \t]*(?:write|"write"|''write'')[^}\n]*\}))[ \t]*(?:#.*)?$)(?:(?=[\s\S]*^[ \t]*(?![^#\n]*\b(?:echo|printf)\b[^#\n]*\bgit[ \t]+push\b)[^#\n]*\bgit[ \t]+push\b[^#\n]*(?:\\\n[ \t]*[^#\n]*)*github\.(?:head_ref|event\.pull_request\.head\.ref))|(?=[\s\S]*^[ \t]*(?P[A-Za-z_][A-Za-z0-9_]*)[ \t]*=[^#\n]*github\.(?:head_ref|event\.pull_request\.head\.ref)[^#\n]*$)(?=[\s\S]*^[ \t]*(?![^#\n]*\b(?:echo|printf)\b[^#\n]*\bgit[ \t]+push\b)[^#\n]*\bgit[ \t]+push\b[^#\n]*(?:\\\n[ \t]*[^#\n]*)*HEAD:[ \t]*(?:\$(?P=mutable_var)|\$\{(?P=mutable_var)\})))[\s\S]*\Z' - message: | - This workflow combines repository write authority with an explicit push - to mutable event-derived branch state. Contributor-controlled workflow - execution must not publish changes back into the branch under review. - Move the repair through a normal reviewed pull request or constrain a - separately reviewed publisher to an exact trusted protected-branch or - release contract. [OWASP A08:2021 - Software and Data Integrity Failures] - severity: HIGH - languages: [generic] - prefilter: ["write", "git", "push"] - paths: - include: - - .github/workflows/*.yml - - .github/workflows/*.yaml - - # Deleting or editing workflow definitions from the workflow itself leaves - # a separately persistent Actions registry identity and evades normal review. - - id: github-actions-self-modifying-writer - patterns: - - pattern-regex: '\A(?=[\s\S]*^[ \t]*(?:permissions|"permissions"|''permissions'')[ \t]*:)(?=[\s\S]*^[ \t]*(?:(?:contents|"contents"|''contents'')[ \t]*:[ \t]*(?:write|"write"|''write'')|(?:permissions|"permissions"|''permissions'')[ \t]*:[ \t]*(?:(?:write-all|"write-all"|''write-all'')|\{[^}\n]*(?:contents|"contents"|''contents'')[ \t]*:[ \t]*(?:write|"write"|''write'')[^}\n]*\}))[ \t]*(?:#.*)?$)(?=[\s\S]*^[ \t]*(?![^#\n]*\b(?:echo|printf)\b[^#\n]*(?:\bgit[ \t]+rm\b|\brm\b|\b(?:sed|perl)\b))[^#\n]*(?:\bgit[ \t]+rm\b|\brm(?:[ \t]+-[A-Za-z]+)?\b|\b(?:sed|perl)[ \t]+(?:-[A-Za-z]*i[A-Za-z]*|--in-place)\b)[^#\n]*(?:\\\n[ \t]*[^#\n]*)*[ \t"'']\.github/workflows/)(?=[\s\S]*^[ \t]*(?![^#\n]*\b(?:echo|printf)\b[^#\n]*\bgit[ \t]+commit\b)[^#\n]*\bgit[ \t]+commit\b)(?=[\s\S]*^[ \t]*(?![^#\n]*\b(?:echo|printf)\b[^#\n]*\bgit[ \t]+push\b)[^#\n]*\bgit[ \t]+push\b)[\s\S]*\Z' - message: | - This write-capable workflow edits or deletes a workflow definition from - its own execution path. Source deletion does not retire the persistent - Actions registry identity. Use a normal reviewed branch and keep exact-ID - registry disablement in a separate owner-approved operator step. - [OWASP A08:2021 - Software and Data Integrity Failures] - severity: HIGH - languages: [generic] - prefilter: ["write", ".github/workflows/", "commit", "git", "push"] - paths: - include: - - .github/workflows/*.yml - - .github/workflows/*.yaml - # A repository-local SARIF upload workflow needs a pull-request entry point # unless PR analysis is intentionally delegated to a centrally managed # required workflow. Delegated workflows can opt out explicitly with the diff --git a/tests/test_github_actions_writer_rules.py b/tests/test_github_actions_writer_rules.py deleted file mode 100644 index 2d6d0080b..000000000 --- a/tests/test_github_actions_writer_rules.py +++ /dev/null @@ -1,366 +0,0 @@ -"""Structural scanner coverage for write-capable GitHub workflows.""" - -from __future__ import annotations - -import pytest - -from scanner.cli.appguardrail import SCAN_RULES, _scan_file - - -def _matches(rule_id: str, workflow: str) -> bool: - matches = [rule for rule in SCAN_RULES if rule["id"] == rule_id] - assert len(matches) == 1 - return bool(matches[0]["pattern"].search(workflow)) - - -def test_mutable_contributor_branch_writer_is_reported() -> None: - workflow = """ -on: pull_request_target -permissions: - contents: write -jobs: - repair: - steps: - - run: git push origin HEAD:${{ github.event.pull_request.head.ref }} -""" - assert _matches("github-actions-mutable-branch-writer", workflow) - - -def test_self_deleting_workflow_is_reported_despite_benign_name() -> None: - workflow = """ -name: Documentation helper -on: workflow_dispatch -permissions: - contents: write -jobs: - docs: - steps: - - run: | - git rm .github/workflows/docs-helper.yml - git commit -m cleanup - git push origin HEAD -""" - assert _matches("github-actions-self-modifying-writer", workflow) - - -def test_write_all_and_spaced_push_reach_scanner_level_mutable_branch_rule( - tmp_path, -) -> None: - """The prefilter must admit write-all and shell-valid push whitespace.""" - workflow = tmp_path / ".github" / "workflows" / "repair.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: pull_request_target -permissions: write-all -jobs: - repair: - steps: - - run: git push origin HEAD:${{ github.event.pull_request.head.ref }} -""", - encoding="utf-8", - ) - findings = _scan_file(workflow, tmp_path) - assert [finding["rule_id"] for finding in findings] == [ - "github-actions-mutable-branch-writer" - ] - - -@pytest.mark.parametrize( - "permission", - [ - 'permissions:\n contents: "write"', - '"permissions":\n "contents": \'write\'', - "permissions: 'write-all'", - "permissions: {contents: write}", - ], -) -def test_quoted_write_permission_reaches_mutable_branch_rule( - tmp_path, permission: str -) -> None: - """Quoted YAML scalars retain the same repository write authority.""" - workflow = tmp_path / ".github" / "workflows" / "repair.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - f""" -on: pull_request_target -{permission} -jobs: - repair: - steps: - - run: git push origin HEAD:${{{{ github.event.pull_request.head.ref }}}} -""", - encoding="utf-8", - ) - findings = _scan_file(workflow, tmp_path) - assert "github-actions-mutable-branch-writer" in { - finding["rule_id"] for finding in findings - } - - -@pytest.mark.parametrize( - "permission", - [ - 'permissions:\n contents: "write"', - '"permissions":\n "contents": \'write\'', - "permissions: 'write-all'", - "permissions: {contents: write}", - ], -) -def test_quoted_write_permission_reaches_self_modifying_rule( - tmp_path, permission: str -) -> None: - """Quoted authority must not hide a persistent workflow mutation.""" - workflow = tmp_path / ".github" / "workflows" / "self-edit.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - f""" -on: workflow_dispatch -{permission} -jobs: - retire: - steps: - - run: | - git rm .github/workflows/self-edit.yml - git commit -m retire - git push origin HEAD:feature -""", - encoding="utf-8", - ) - findings = _scan_file(workflow, tmp_path) - assert "github-actions-self-modifying-writer" in { - finding["rule_id"] for finding in findings - } - - -def test_comment_only_writer_commands_are_not_reported(tmp_path) -> None: - """YAML and shell comments are not executable repository mutations.""" - workflow = tmp_path / ".github" / "workflows" / "comments.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: workflow_dispatch -permissions: - contents: write -jobs: - inspect: - steps: - - run: | - echo read-only - # git push origin HEAD:${{ github.head_ref }} - # git rm .github/workflows/comments.yml - # git commit -m retire - # git push origin HEAD:feature -""", - encoding="utf-8", - ) - rule_ids = {finding["rule_id"] for finding in _scan_file(workflow, tmp_path)} - assert "github-actions-mutable-branch-writer" not in rule_ids - assert "github-actions-self-modifying-writer" not in rule_ids - - -def test_continued_mutable_push_is_reported(tmp_path) -> None: - """A shell line continuation cannot hide an event-derived push target.""" - workflow = tmp_path / ".github" / "workflows" / "continued.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: pull_request_target -permissions: {contents: write} -jobs: - repair: - steps: - - run: | - git push origin \\ - HEAD:${{ github.head_ref }} -""", - encoding="utf-8", - ) - rule_ids = {finding["rule_id"] for finding in _scan_file(workflow, tmp_path)} - assert "github-actions-mutable-branch-writer" in rule_ids - - -def test_indirect_mutable_push_is_reported(tmp_path) -> None: - """A shell variable cannot hide an event-derived push target.""" - workflow = tmp_path / ".github" / "workflows" / "indirect.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: pull_request_target -permissions: - contents: write -jobs: - repair: - steps: - - run: | - target=${{ github.head_ref }} - git push origin HEAD:$target -""", - encoding="utf-8", - ) - rule_ids = {finding["rule_id"] for finding in _scan_file(workflow, tmp_path)} - assert "github-actions-mutable-branch-writer" in rule_ids - - -def test_continued_workflow_mutation_is_reported(tmp_path) -> None: - """A continued workflow path remains a persistent self-modification.""" - workflow = tmp_path / ".github" / "workflows" / "continued-self.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: workflow_dispatch -permissions: {contents: write} -jobs: - retire: - steps: - - run: | - git rm \\ - .github/workflows/continued-self.yml - git commit -m retire - git push origin HEAD:feature -""", - encoding="utf-8", - ) - rule_ids = {finding["rule_id"] for finding in _scan_file(workflow, tmp_path)} - assert "github-actions-self-modifying-writer" in rule_ids - - -def test_non_executable_writer_text_and_protected_ref_are_not_reported( - tmp_path, -) -> None: - """Strings, run-body permission text, and protected pushes stay benign.""" - templates = { - "echo.yml": """ -on: workflow_dispatch -permissions: {contents: write} -jobs: - inspect: - steps: - - run: echo "git push origin HEAD:${{ github.head_ref }}" -""", - "run-permission.yml": """ -on: workflow_dispatch -jobs: - inspect: - steps: - - run: | - contents: write - git push origin HEAD:${{ github.head_ref }} -""", - "protected.yml": """ -on: - push: - branches: [develop] -permissions: - contents: write -jobs: - publish: - steps: - - run: git push origin HEAD:${{ github.ref_name }} -""", - } - for name, text in templates.items(): - workflow = tmp_path / ".github" / "workflows" / name - workflow.parent.mkdir(parents=True, exist_ok=True) - workflow.write_text(text, encoding="utf-8") - rule_ids = { - finding["rule_id"] for finding in _scan_file(workflow, tmp_path) - } - assert "github-actions-mutable-branch-writer" not in rule_ids, name - assert "github-actions-self-modifying-writer" not in rule_ids, name - - -def test_variable_whitespace_commit_reaches_scanner_level_self_writer_rule( - tmp_path, -) -> None: - """The prefilter must match shell-valid whitespace between git and commit.""" - workflow = tmp_path / ".github" / "workflows" / "self-edit.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: workflow_dispatch -permissions: write-all -jobs: - retire: - steps: - - run: | - git rm .github/workflows/self-edit.yml - git commit -m retire - git push origin HEAD:feature -""", - encoding="utf-8", - ) - findings = _scan_file(workflow, tmp_path) - assert "github-actions-self-modifying-writer" in { - finding["rule_id"] for finding in findings - } - - -def test_protected_branch_publisher_is_not_a_mutable_branch_writer(tmp_path) -> None: - """An explicit protected-branch target must not trigger the contributor rule.""" - workflow = tmp_path / ".github" / "workflows" / "publish.yml" - workflow.parent.mkdir(parents=True) - workflow.write_text( - """ -on: - push: - branches: [develop] -permissions: - contents: write -jobs: - publish: - steps: - - run: git push origin HEAD:develop -""", - encoding="utf-8", - ) - findings = _scan_file(workflow, tmp_path) - assert "github-actions-mutable-branch-writer" not in { - finding["rule_id"] for finding in findings - } - - -def test_non_persistent_workflow_file_operations_are_not_reported() -> None: - """Read-only and uncommitted workspace cleanup are not repository writes.""" - templates = ( - "sed -n '1,20p' .github/workflows/ci.yml", - "git rm .github/workflows/copied.yml", - "rm -rf snapshot/.github/workflows/", - ) - for command in templates: - workflow = f""" -on: workflow_dispatch -permissions: - contents: write -jobs: - inspect: - steps: - - run: {command} -""" - assert not _matches("github-actions-self-modifying-writer", workflow) - - -def test_read_only_diagnostic_and_protected_release_are_not_reported() -> None: - diagnostic = """ -on: pull_request -permissions: - contents: read -jobs: - inspect: - steps: - - run: git status -""" - release = """ -on: - push: - tags: ['v*'] -permissions: - contents: write -jobs: - publish: - steps: - - run: gh release create "${{ github.ref_name }}" -""" - for text in (diagnostic, release): - assert not _matches("github-actions-mutable-branch-writer", text) - assert not _matches("github-actions-self-modifying-writer", text) diff --git a/tests/test_injection_rules.py b/tests/test_injection_rules.py index 99e0f38bb..943b98243 100644 --- a/tests/test_injection_rules.py +++ b/tests/test_injection_rules.py @@ -1,9 +1,9 @@ """Coverage tests for the injection + Anthropic-key detection rules.""" -from scanner.cli.appguardrail import SCAN_RULES +import scanner.cli.appguardrail as appguardrail _BY_ID = {} -for _r in SCAN_RULES: +for _r in appguardrail.SCAN_RULES: _BY_ID.setdefault(_r["id"], _r) @@ -34,3 +34,74 @@ def test_hardcoded_anthropic_api_key(): assert r["pattern"].search("key = 'sk-ant-api03-AbCdEf0123456789xyzXYZ_-abc'") assert not r["pattern"].search("key = 'sk-ant-'") # too short assert not r["pattern"].search("token = 'sk-live-notananthropickey'") + + +def _python_command_matches(source): + rule = _rule("python-command-injection") + return list(rule["finder"](source)) + + +def test_python_command_injection(): + rule = _rule("python-command-injection") + assert rule["severity"] == "CRITICAL" + + source = """os.system(user_input) +subprocess.run(build_command(user_input), shell=True) +subprocess.Popen(command, shell = 1) +subprocess.call(command, shell=enabled) +wrapper.os.system(user_input) +wrapper.subprocess.run(command, shell=True) +""" + matches = _python_command_matches(source) + assert len(matches) == 6 + assert [source.count("\n", 0, match.start()) + 1 for match in matches] == [ + 1, + 2, + 3, + 4, + 5, + 6, + ] + + +def test_python_command_injection_ignores_non_calls_comments_and_strings(): + source = """# os.system(user_input) +warning = "subprocess.run(command, shell=True)" +os.system = replacement +subprocess.run(["ls", "-l"]) +subprocess.Popen(["ls", "-l"], shell=False) +subprocess.call(command, shell=0) +subprocess.check_output(command, shell=None) +subprocess.run(command, shell="") +subprocess.Popen(command, shell=()) +subprocess.call(command, shell=b"") +""" + assert _python_command_matches(source) == [] + + +def test_python_command_injection_preserves_non_ascii_source_offset(): + source = "label = 'ν•œκΈ€'; os.system(command)\n" + match = _python_command_matches(source)[0] + assert source[match.start() :].startswith("os.system") + + +def test_python_command_injection_uses_python_newline_boundaries(): + for separator in ("\f", "\u2028"): + source = f"label = 'text{separator}'\nos.system(command)\n" + match = _python_command_matches(source)[0] + assert source[match.start() :].startswith("os.system") + + +def test_python_command_injection_falls_back_for_invalid_python(): + matches = _python_command_matches("if (\n os.system(user_input)\n") + assert len(matches) == 1 + +from unittest.mock import patch + +def test_python_command_injection_falls_back_when_ast_resource_limits_fail(): + """AST resource-limit failures retain conservative regex detection.""" + source = "os.system(user_input)" + for parse_error in (RecursionError("deep AST"), MemoryError("large AST")): + with patch.object(appguardrail.ast, "parse", side_effect=parse_error): + matches = _python_command_matches(source) + assert len(matches) == 1 diff --git a/tests/test_workflow_lifecycle.py b/tests/test_workflow_lifecycle.py deleted file mode 100644 index 37df54d60..000000000 --- a/tests/test_workflow_lifecycle.py +++ /dev/null @@ -1,589 +0,0 @@ -"""Adversarial contracts for read-only GitHub workflow lifecycle evidence.""" - -from __future__ import annotations - -import builtins -import json -import os -import runpy -import sys - -import pytest - -from appguardrail_core import workflow_lifecycle as lifecycle - -SHA = "a" * 40 -SHA_B = "b" * 40 - - -def _workflow(workflow_id: int, path: str, state: str = "active") -> dict: - return { - "id": workflow_id, - "name": path.rsplit("/", 1)[-1], - "path": path, - "state": state, - } - - -def _record( - *, workflows: list[dict], tree: list[str], pages: list[dict] | None = None -) -> dict: - return { - "name": "appguardrail", - "archived": False, - "default_branch": "develop", - "default_branch_sha": SHA, - "default_branch_sha_after": SHA, - "tree_paths": tree, - "workflow_pages": pages - or [ - {"total_count": len(workflows), "workflows": workflows, "_link_next": False} - ], - } - - -def test_complete_inventory_separates_all_control_plane_classes() -> None: - """Present, orphan, disabled, dynamic, and unresolved remain distinct.""" - workflows = [ - _workflow(1, ".github/workflows/ci.yml"), - _workflow(2, ".github/workflows/old.yml"), - _workflow(3, ".github/workflows/disabled.yml", "disabled_manually"), - _workflow(4, "dynamic/pages/pages-build-deployment"), - _workflow(5, ".GitHub/workflows/case.yml"), - ] - result = lifecycle.inventory_repository( - _record(workflows=workflows, tree=[".github/workflows/ci.yml"]) - ) - assert [item["classification"] for item in result["records"]] == [ - "present_active", - "orphan_active", - "orphan_disabled", - "dynamic_owned", - "unresolved", - ] - assert result["records"][1]["workflow_id"] == 2 - assert result["records"][1]["default_branch_sha"] == SHA - - -def test_pagination_requires_every_page_and_unique_workflow_ids() -> None: - """Missing pages and reused registry IDs fail closed.""" - first = { - "total_count": 2, - "workflows": [_workflow(7, ".github/workflows/a.yml")], - "_link_next": True, - } - with pytest.raises(lifecycle.InventoryError, match="truncated"): - lifecycle.collect_workflow_pages([first], per_page=1) - second = { - "total_count": 2, - "workflows": [_workflow(7, ".github/workflows/renamed.yml")], - "_link_next": False, - } - with pytest.raises(lifecycle.InventoryError, match="reused workflow id 7"): - lifecycle.collect_workflow_pages([first, second], per_page=1) - - -@pytest.mark.parametrize("status", [403, 404, 500, 503]) -def test_permission_missing_and_server_responses_never_become_clean( - status: int, -) -> None: - """Visibility loss and upstream failure are unresolved errors, not absence.""" - with pytest.raises(lifecycle.InventoryError): - lifecycle.interpret_status(status, resource="workflow inventory") - - -def test_server_error_retries_once_but_permission_error_does_not() -> None: - """Only 5xx receives the central contract's one bounded retry.""" - calls: list[str] = [] - - def recover(url: str): - calls.append(url) - return (503, {}, {}) if len(calls) == 1 else (200, {"ok": True}, {}) - - assert lifecycle.fetch_with_one_retry(recover, "https://api.github.com/test")[ - 0 - ] == {"ok": True} - assert len(calls) == 2 - - calls.clear() - - def forbidden(url: str): - calls.append(url) - return 403, {}, {} - - with pytest.raises(lifecycle.InventoryError, match="permission"): - lifecycle.fetch_with_one_retry(forbidden, "https://api.github.com/test") - assert len(calls) == 1 - - -def test_live_collection_retries_shared_client_5xx_shape_once() -> None: - """The production collector recognizes AppGuardrail's shared client error.""" - - class Client: - calls = 0 - - def request(self, _path: str): - self.calls += 1 - if self.calls == 1: - raise RuntimeError("GitHub API GET /orgs failed: 503 unavailable") - return {"ok": True} - - receipts: list[dict] = [] - assert lifecycle._live_get(Client(), "/orgs", receipts) == {"ok": True} - assert len(receipts) == 1 - - -def test_live_get_redacts_non_server_transport_failure() -> None: - """Permission details do not escape the bounded live error.""" - - class Client: - def request(self, _path: str): - raise RuntimeError("private permission detail") - - with pytest.raises(lifecycle.InventoryError, match="RuntimeError") as error: - lifecycle._live_get(Client(), "/private", []) - assert "private permission detail" not in str(error.value) - - -def test_read_only_adapter_rejects_every_mutating_shape() -> None: - """The shared transport cannot be used for a workflow mutation.""" - - class Client: - def request(self, path: str, *, method: str = "GET", payload=None): - assert payload is None - return {"method": method, "path": path} - - adapter = lifecycle._ReadOnlyGitHubAdapter(Client()) - assert adapter.request("/repos/example") == { - "method": "GET", - "path": "/repos/example", - } - with pytest.raises(lifecycle.InventoryError, match="read-only"): - adapter.request("/repos/example/actions/workflows/1/disable", method="PUT") - with pytest.raises(lifecycle.InventoryError, match="read-only"): - adapter.request("/repos/example", payload={}) - - -def test_packaged_client_is_fixed_origin_get_only(monkeypatch) -> None: - """The installed transport cannot redirect or accept absolute paths.""" - - class Response: - def __enter__(self): - return self - - def __exit__(self, *_args): - return False - - def read(self) -> bytes: - return b'{"ok": true}' - - class Opener: - def open(self, request, timeout: int): - assert request.full_url == "https://api.github.com/orgs/example" - assert request.get_method() == "GET" - assert timeout == 30 - return Response() - - client = lifecycle._GitHubReadOnlyClient("test-token") - monkeypatch.setattr(client, "_opener", Opener()) - assert client.request("/orgs/example") == {"ok": True} - assert lifecycle._ReadOnlyGitHubAdapter(client).request("/orgs/example") == { - "ok": True - } - with pytest.raises(lifecycle.InventoryError, match="read-only"): - client.request("/orgs/example", method="POST") - with pytest.raises(lifecycle.InventoryError, match="origin-relative"): - client.request("//attacker.example/path") - assert lifecycle._NoRedirect().redirect_request(None, None, 302, "", {}, "") is None - - -def test_packaged_client_bounds_empty_and_http_error_responses(monkeypatch) -> None: - """Empty success and terminal HTTP failure have deterministic outcomes.""" - - class EmptyResponse: - def __enter__(self): - return self - - def __exit__(self, *_args): - return False - - def read(self) -> bytes: - return b"" - - class EmptyOpener: - def open(self, _request, timeout: int): - assert timeout == 30 - return EmptyResponse() - - client = lifecycle._GitHubReadOnlyClient("test-token") - monkeypatch.setattr(client, "_opener", EmptyOpener()) - assert client.request("/empty") is None - - class ErrorOpener: - def open(self, request, timeout: int): - raise lifecycle.HTTPError(request.full_url, 403, "forbidden", {}, None) - - monkeypatch.setattr(client, "_opener", ErrorOpener()) - with pytest.raises(RuntimeError, match="failed: 403"): - client.request("/forbidden") - - -def test_live_main_writes_receipts_and_fail_closed_evidence( - tmp_path, monkeypatch, capsys -) -> None: - """The live CLI persists both successful and partial collection receipts.""" - monkeypatch.setenv("GITHUB_TOKEN", "test-token") - receipt_path = tmp_path / "receipts.json" - failure_path = tmp_path / "failure.json" - output_path = tmp_path / "ledger.json" - ledger = { - "counts": {"orphan_active": 0}, - "records": [], - } - - def collect(_client, *, receipts): - receipts.append({"method": "GET", "path": "/ok", "sha256": "a" * 64}) - return {"organization": "ContextualWisdomLab"}, receipts - - monkeypatch.setattr(lifecycle, "collect_live_organization", collect) - monkeypatch.setattr(lifecycle, "inventory_organization", lambda _payload: ledger) - assert ( - lifecycle.main( - [ - "--live", - "--output", - str(output_path), - "--receipt-output", - str(receipt_path), - ] - ) - == 0 - ) - assert json.loads(receipt_path.read_text())[0]["path"] == "/ok" - - def fail(_client, *, receipts): - receipts.append({"method": "GET", "path": "/partial", "sha256": "b" * 64}) - raise lifecycle.InventoryError("incomplete") - - monkeypatch.setattr(lifecycle, "collect_live_organization", fail) - assert ( - lifecycle.main( - [ - "--live", - "--output", - str(output_path), - "--receipt-output", - str(receipt_path), - "--failure-output", - str(failure_path), - ] - ) - == 2 - ) - assert json.loads(receipt_path.read_text())[0]["path"] == "/partial" - assert json.loads(failure_path.read_text())["status"] == "failed" - assert json.loads(output_path.read_text())["status"] == "failed" - - missing_path = tmp_path / "missing" / "evidence.json" - assert ( - lifecycle.main( - [ - "--live", - "--receipt-output", - str(missing_path), - "--failure-output", - str(failure_path), - ] - ) - == 2 - ) - assert json.loads(failure_path.read_text())["error_type"] == "InventoryError" - - monkeypatch.setattr(lifecycle, "collect_live_organization", collect) - assert ( - lifecycle.main( - [ - "--live", - "--receipt-output", - str(missing_path), - "--failure-output", - str(failure_path), - ] - ) - == 2 - ) - assert json.loads(failure_path.read_text())["error_type"] == "FileNotFoundError" - assert "unable to write receipts" in capsys.readouterr().err - assert ( - lifecycle.main( - [ - "--live", - "--receipt-output", - str(receipt_path), - "--failure-output", - str(failure_path), - "--output", - str(missing_path), - ] - ) - == 2 - ) - assert json.loads(failure_path.read_text())["error_type"] == "FileNotFoundError" - assert "unable to write ledger" in capsys.readouterr().err - - -def test_success_removes_stale_live_failure_evidence(tmp_path, monkeypatch) -> None: - """A successful live run cannot coexist with an earlier failure artifact.""" - monkeypatch.setenv("GITHUB_TOKEN", "test-token") - monkeypatch.setattr( - lifecycle, - "collect_live_organization", - lambda _client, *, receipts: ( - {"organization": "ContextualWisdomLab"}, - receipts, - ), - ) - monkeypatch.setattr( - lifecycle, - "inventory_organization", - lambda _payload: {"counts": {"orphan_active": 0}, "records": []}, - ) - failure_path = tmp_path / "failure.json" - failure_path.write_text('{"status":"failed"}\n', encoding="utf-8") - assert ( - lifecycle.main( - [ - "--live", - "--output", - str(tmp_path / "ledger.json"), - "--receipt-output", - str(tmp_path / "receipts.json"), - "--failure-output", - str(failure_path), - ] - ) - == 0 - ) - assert not failure_path.exists() - - -def test_atomic_evidence_write_preserves_previous_file_on_replace_failure( - tmp_path, monkeypatch -) -> None: - """A failed replacement cannot expose a partial evidence document.""" - output = tmp_path / "ledger.json" - output.write_text("previous\n", encoding="utf-8") - - def fail_replace(_source, _target) -> None: - raise OSError("replace failed") - - monkeypatch.setattr(lifecycle.os, "replace", fail_replace) - with pytest.raises(OSError, match="replace failed"): - lifecycle._write_atomic_text(output, "replacement\n") - assert output.read_text(encoding="utf-8") == "previous\n" - assert list(tmp_path.glob(".ledger.json.*.tmp")) == [] - - -def test_stale_failure_cleanup_failure_invalidates_ledger(tmp_path, monkeypatch) -> None: - """An undeletable failure artifact prevents a contradictory success result.""" - monkeypatch.setenv("GITHUB_TOKEN", "test-token") - monkeypatch.setattr( - lifecycle, - "collect_live_organization", - lambda _client, *, receipts: ( - {"organization": "ContextualWisdomLab"}, - receipts, - ), - ) - monkeypatch.setattr( - lifecycle, - "inventory_organization", - lambda _payload: {"counts": {"orphan_active": 0}, "records": []}, - ) - failure_path = tmp_path / "failure.json" - failure_path.write_text('{"status":"failed"}\n', encoding="utf-8") - original_unlink = lifecycle.Path.unlink - - def fail_failure_unlink(path, *args, **kwargs): - if path == failure_path: - raise OSError("unlink failed") - return original_unlink(path, *args, **kwargs) - - monkeypatch.setattr(lifecycle.Path, "unlink", fail_failure_unlink) - output = tmp_path / "ledger.json" - assert ( - lifecycle.main( - [ - "--live", - "--output", - str(output), - "--receipt-output", - str(tmp_path / "receipts.json"), - "--failure-output", - str(failure_path), - ] - ) - == 2 - ) - assert json.loads(output.read_text(encoding="utf-8"))["status"] == "failed" - - -@pytest.mark.parametrize( - "arguments", - [ - [ - "--output", - "evidence.json", - "--receipt-output", - "evidence.json", - "--failure-output", - "failure.json", - ], - [ - "--output", - "evidence.json", - "--receipt-output", - "receipts.json", - "--failure-output", - "evidence.json", - ], - [ - "--output", - "ledger.json", - "--receipt-output", - "evidence.json", - "--failure-output", - "evidence.json", - ], - ], -) -def test_live_main_rejects_aliased_evidence_paths( - tmp_path, monkeypatch, arguments -) -> None: - """Success, receipt, and failure artifacts require distinct identities.""" - monkeypatch.chdir(tmp_path) - monkeypatch.setenv("GITHUB_TOKEN", "test-token") - assert lifecycle.main(["--live", *arguments]) == 2 - assert not (tmp_path / "evidence.json").exists() - - -def test_live_main_rejects_hardlinked_evidence_paths(tmp_path, monkeypatch) -> None: - """Different spellings of one existing inode are still one artifact.""" - output = tmp_path / "ledger.json" - receipt = tmp_path / "receipts.json" - output.write_text("existing\n", encoding="utf-8") - os.link(output, receipt) - monkeypatch.setenv("GITHUB_TOKEN", "test-token") - assert ( - lifecycle.main( - [ - "--live", - "--output", - str(output), - "--receipt-output", - str(receipt), - ] - ) - == 2 - ) - assert output.read_text(encoding="utf-8") == "existing\n" - - -def test_live_main_requires_a_token(tmp_path, monkeypatch) -> None: - """Missing credentials fail before any GitHub request.""" - monkeypatch.delenv("GITHUB_TOKEN", raising=False) - assert lifecycle.main(["--live"]) == 2 - assert ( - lifecycle.main(["--live", "--receipt-output", str(tmp_path / "receipts.json")]) - == 2 - ) - - -def test_live_main_uses_packaged_client_without_repository_scripts( - tmp_path, monkeypatch -) -> None: - """An installed core keeps documented live inventory functional.""" - original_import = builtins.__import__ - - def without_repository_client(name, *args, **kwargs): - if name in { - "scripts.ci.commercial_readiness_loop", - "commercial_readiness_loop", - }: - raise ModuleNotFoundError(name) - return original_import(name, *args, **kwargs) - - monkeypatch.setattr(builtins, "__import__", without_repository_client) - - class GitHub: - def __init__(self, token: str): - assert token == "test-token" - - monkeypatch.setattr(lifecycle, "_GitHubReadOnlyClient", GitHub, raising=False) - monkeypatch.setenv("GITHUB_TOKEN", "test-token") - monkeypatch.setattr( - lifecycle, - "collect_live_organization", - lambda _client, *, receipts: ( - {"organization": "ContextualWisdomLab"}, - receipts, - ), - ) - monkeypatch.setattr( - lifecycle, - "inventory_organization", - lambda _payload: {"counts": {"orphan_active": 0}, "records": []}, - ) - receipts = tmp_path / "receipts.json" - assert ( - lifecycle.main( - [ - "--live", - "--receipt-output", - str(receipts), - ] - ) - == 0 - ) - assert json.loads(receipts.read_text()) == [] - - -def test_branch_movement_case_encoding_and_malformed_paths_fail_closed() -> None: - """A moved head or ambiguous registry path cannot prove an orphan.""" - with pytest.raises(lifecycle.InventoryError, match="moved"): - lifecycle.assert_default_branch_bound(SHA, SHA_B) - with pytest.raises(lifecycle.InventoryError, match="percent-encoded"): - lifecycle.decode_registry_path(".github/workflows/%6f%6ece.yml") - assert not lifecycle.is_repository_workflow_path(".GitHub/workflows/once.yml") - - -def test_supported_once_file_is_present_not_a_name_heuristic() -> None: - """A live reviewed one-shot name stays present_active.""" - path = ".github/workflows/apply-reviewed-once.yml" - result = lifecycle.inventory_repository( - _record(workflows=[_workflow(9, path)], tree=[path]) - ) - assert result["records"][0]["classification"] == "present_active" - - -def test_inventory_api_has_no_registry_mutation_primitive() -> None: - """The package exposes detection only, never exact-ID disablement.""" - assert not hasattr(lifecycle, "disable_workflow") - assert not hasattr(lifecycle, "delete_workflow") - - -@pytest.mark.filterwarnings("ignore:.*found in sys.modules.*:RuntimeWarning") -def test_module_entrypoint_executes_read_only_fixture(tmp_path, monkeypatch) -> None: - """The packaged module entry point exits successfully for clean evidence.""" - payload = { - "organization": "ContextualWisdomLab", - "observed_at": "2026-09-23T00:00:00Z", - "repository_inventory_complete": True, - "repositories": [_record(workflows=[], tree=[])], - } - source = tmp_path / "inventory.json" - source.write_text(json.dumps(payload), encoding="utf-8") - monkeypatch.setattr(sys, "argv", ["workflow_lifecycle", "--payload", str(source)]) - with pytest.raises(SystemExit) as stopped: - runpy.run_module("appguardrail_core.workflow_lifecycle", run_name="__main__") - assert stopped.value.code == 0 diff --git a/tests/test_workflow_lifecycle_central_contract.py b/tests/test_workflow_lifecycle_central_contract.py deleted file mode 100644 index 2ef7ca799..000000000 --- a/tests/test_workflow_lifecycle_central_contract.py +++ /dev/null @@ -1,1309 +0,0 @@ -"""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 appguardrail_core import workflow_lifecycle as inventory - -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 - - -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 - assert client.calls[-3:] == [ - f"/repos/{repo}/actions/workflows?per_page=100&page=1", - f"/repos/{repo}/commits/main", - "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1", - ] - - -def test_live_collection_rejects_concurrent_workflow_registry_change() -> None: - """A stable Git commit cannot mask a concurrent registry state change.""" - repo = "ContextualWisdomLab/appguardrail" - workflow_path = f"/repos/{repo}/actions/workflows?per_page=100&page=1" - 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"}], - }, - workflow_path: { - "total_count": 1, - "workflows": [_workflow(1, ".github/workflows/ci.yml")], - }, - } - - class ChangingWorkflowClient(_LiveClient): - workflow_reads = 0 - - def request( - self, path: str, *, method: str = "GET", payload: Any = None - ) -> Any: - response = super().request(path, method=method, payload=payload) - if path == workflow_path: - self.workflow_reads += 1 - if self.workflow_reads == 2: - return { - "total_count": 1, - "workflows": [ - _workflow( - 1, - ".github/workflows/ci.yml", - state="disabled_manually", - ) - ], - } - return response - - with pytest.raises(inventory.InventoryError, match="workflow registry changed"): - inventory.collect_live_organization(ChangingWorkflowClient(responses)) - - -def test_live_collection_bounds_workflow_pages(monkeypatch) -> None: - """An endless workflow inventory cannot exceed the shared page bound.""" - monkeypatch.setattr(inventory, "MAX_PAGES", 1) - 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": [], - }, - f"/repos/{repo}/actions/workflows?per_page=100&page=1": { - "total_count": 101, - "workflows": [ - _workflow(index, f".github/workflows/{index}.yml") - for index in range(100) - ], - }, - } - with pytest.raises(inventory.InventoryError, match="workflow pagination exceeded"): - inventory.collect_live_organization(_LiveClient(responses)) - - -@pytest.mark.parametrize("changed_listing", [2, 3]) -def test_live_collection_rejects_same_count_repository_change( - changed_listing: int, -) -> None: - """A repository replacement cannot hide behind an unchanged organization count.""" - repo = "ContextualWisdomLab/appguardrail" - list_path = ( - "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" - ) - first = { - "id": 1, - "name": "appguardrail", - "full_name": repo, - "archived": False, - "default_branch": "main", - } - replacement = { - **first, - "id": 2, - "name": "replacement", - "full_name": "ContextualWisdomLab/replacement", - } - - class ChangingClient(_LiveClient): - listings = 0 - - def request( - self, path: str, *, method: str = "GET", payload: Any = None - ) -> Any: - if path == list_path: - self.listings += 1 - return [replacement if self.listings == changed_listing else first] - return super().request(path, method=method, payload=payload) - - responses = { - "/orgs/ContextualWisdomLab": {"public_repos": 1, "total_private_repos": 0}, - 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": [], - }, - } - with pytest.raises(inventory.InventoryError, match="repository inventory changed"): - inventory.collect_live_organization(ChangingClient(responses)) - - -def test_live_collection_bounds_repository_pages() -> None: - """An endless full-page response cannot keep the fleet collector running.""" - - class FullPageClient: - def request( - self, path: str, *, method: str = "GET", payload: Any = None - ) -> Any: - assert method == "GET" and payload is None - assert path.startswith("/orgs/ContextualWisdomLab/repos?") - return [{"full_name": f"ContextualWisdomLab/repo-{i}"} for i in range(100)] - - with pytest.raises(inventory.InventoryError, match="pagination exceeded limit"): - inventory.collect_live_organization(FullPageClient()) - - -def test_live_collection_rejects_unhashable_repository_identity() -> None: - """Malformed repository metadata cannot become a stable fleet identity.""" - org_path = ( - "/orgs/ContextualWisdomLab/repos?type=all&sort=full_name&per_page=100&page=1" - ) - client = _LiveClient( - { - org_path: [ - {"id": [], "full_name": "ContextualWisdomLab/example", "archived": True} - ], - "/orgs/ContextualWisdomLab": {"public_repos": 1, "total_private_repos": 0}, - } - ) - with pytest.raises( - inventory.InventoryError, match="repository identity is malformed" - ): - inventory.collect_live_organization(client) - - -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_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"), - ( - { - "workflows": [ - { - "id": 1, - "path": ".github/workflows/ci.yml", - "state": [], - } - ], - "total_count": 1, - }, - "workflow identity", - ), - ]: - with pytest.raises(inventory.InventoryError, match=message): - inventory.collect_live_organization(client_for(**{workflow_path: response})) - - -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_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)