From a3b178c41d99640ee2aebe0578c78a2f9ff838a3 Mon Sep 17 00:00:00 2001 From: AadarshLN Date: Fri, 14 Aug 2026 20:39:45 -0400 Subject: [PATCH] feat(reproducibility): verify Witness attestations for RE-02.01 hermeticity Replace the CI-text grep for "witness run" with real verification: fetch Witness/runtime-trace attestation artifacts from the repo's latest successful CI run via gh, cryptographically verify the DSSE envelope against the repo's GitHub Actions OIDC identity via Sigstore, and only treat an empty verified network log as a strong PASS signal. A verified attestation that does record network activity is fed into the violation list. Add a verify_witness_attestations config toggle and distinguish gh-not-authenticated from no-artifacts-found in evidence. Co-Authored-By: Claude Sonnet 5 --- .../src/darnit_reproducibility/handlers.py | 133 ++++-- .../witness_attestation.py | 331 +++++++++++++ tests/darnit_reproducibility/test_handlers.py | 258 ++++++++-- .../test_witness_attestation.py | 441 ++++++++++++++++++ 4 files changed, 1092 insertions(+), 71 deletions(-) create mode 100644 packages/darnit-reproducibility/src/darnit_reproducibility/witness_attestation.py create mode 100644 tests/darnit_reproducibility/test_witness_attestation.py diff --git a/packages/darnit-reproducibility/src/darnit_reproducibility/handlers.py b/packages/darnit-reproducibility/src/darnit_reproducibility/handlers.py index 6f74d394..43dc69c4 100644 --- a/packages/darnit-reproducibility/src/darnit_reproducibility/handlers.py +++ b/packages/darnit-reproducibility/src/darnit_reproducibility/handlers.py @@ -11,6 +11,8 @@ from darnit.core.logging import get_logger from darnit.sieve.handler_registry import HandlerContext, HandlerResult, HandlerResultStatus +from .witness_attestation import WitnessCheckResult, check_witness_attestation + logger = get_logger("darnit_reproducibility.handlers") @@ -329,19 +331,73 @@ def _iter_container_files(path: Path) -> list[Path]: return results[:_FILE_SCAN_LIMIT] -def _detect_strong_hermeticity_signal(path: Path, ci_files: list[Path]) -> str | None: - """Return a description of a build-system-enforced hermeticity signal, or None. + +# Bazel network sandbox flags — current name, negated shorthand, and the +# deprecated pre-rename name (still honored by Bazel as an alias). +_BAZEL_NETWORK_BLOCK_FLAGS: tuple[str, ...] = ( + "--sandbox_default_allow_network=false", + "--nosandbox_default_allow_network", + "--experimental_sandbox_default_allow_network=false", +) + + +def _maybe_check_witness_attestation( + ctx: HandlerContext, + config: dict[str, Any], +) -> WitnessCheckResult: + """Call check_witness_attestation() unless disabled via config. + + ``verify_witness_attestations = false`` in the TOML pass config opts out + of the network round-trip entirely — for air-gapped audits or + environments without a usable `gh` login for the audited repo. There is + deliberately no automatic "skip if CI text doesn't mention witness" + heuristic: that would reintroduce exactly the kind of unreliable text-only + guess this real verification replaced (e.g. a reusable/composite workflow + can produce a valid attestation without the calling repo's own CI files + ever spelling out "witness"). + """ + if not config.get("verify_witness_attestations", True): + return WitnessCheckResult(attempted=False, detail="witness attestation verification disabled via config") + + return check_witness_attestation(ctx) + + +def _detect_strong_hermeticity_signal( + path: Path, + ci_files: list[Path], + dependency_results: dict[str, Any], + ctx: HandlerContext, + config: dict[str, Any], +) -> tuple[str | None, WitnessCheckResult]: + """Return (signal description, witness check result). Signal is None if no + build-system-enforced hermeticity guarantee was found. Checks in priority order: - 1. Witness runtime attestation — records what the build actually did at runtime - 2. Nix flake used in CI — fixed-output derivations run network-isolated by default + 1. Witness runtime attestation — a Sigstore-verified DSSE envelope from the + repo's latest CI run, bound to its GitHub Actions OIDC identity, asserting + no network access occurred during the build (see witness_attestation.py). + Merely mentioning "witness run" in CI text is NOT sufficient — that only + proves the tool ran, not what it observed, so text mentions alone are no + longer treated as a strong signal. + 2. Nix flake used in CI — fixed-output derivations run network-isolated by default. + Gated on RE-01.02 (BuildEnvDeclared) having PASSED: a bare flake.nix that isn't + the project's confirmed, declared build environment isn't a strong signal on its + own. If RE-01.02 hasn't run (e.g. this control invoked standalone), the signal is + withheld rather than assumed — conservative-by-default. 3. Bazel with explicit network sandbox — Bazel allows network by default, so the - blocking flag must be present to count as a strong signal + blocking flag must be present to count as a strong signal. Checked in both CI + files (where "bazel" must also appear, to avoid matching an unrelated tool that + happens to share a flag name) and .bazelrc (the canonical place to set it, where + the file itself is the bazel signal). Comments are stripped before matching (same as ``_scan_line``) so a commented-out reference (e.g. ``# TODO: add witness run``) can't be mistaken for the real thing. """ + witness_result = _maybe_check_witness_attestation(ctx, config) + if witness_result.verified and witness_result.network_clean is True: + return f"Witness attestation verified — {witness_result.detail}", witness_result + ci_content: dict[str, str] = {} for f in ci_files: try: @@ -350,35 +406,37 @@ def _detect_strong_hermeticity_signal(path: Path, ci_files: list[Path]) -> str | continue ci_content[f.name] = "\n".join(_strip_comment(line) for line in raw.splitlines()) - witness_hits = sorted( - name - for name, content in ci_content.items() - if "witness run" in content or "testifysec/witness" in content or "in-toto/witness" in content - ) - if witness_hits: - return f"Witness runtime attestation in CI ({', '.join(witness_hits)})" - - if (path / "flake.nix").exists(): + if (path / "flake.nix").exists() and dependency_results.get("RE-01.02") == "PASS": nix_hits = sorted( name for name, content in ci_content.items() if any(cmd in content for cmd in ("nix build", "nix develop", "nix run", "nix flake")) ) if nix_hits: - return f"Nix flake build in CI ({', '.join(nix_hits)})" + return f"Nix flake build in CI ({', '.join(nix_hits)})", witness_result has_bazel = any((path / f).exists() for f in ("WORKSPACE", "WORKSPACE.bazel", "MODULE.bazel", "BUILD.bazel")) if has_bazel: - bazel_hits = sorted( + bazel_hits: list[str] = [ name for name, content in ci_content.items() - if "bazel" in content.lower() - and ("--sandbox_default_allow_network=false" in content or "--sandbox_network=block" in content) - ) + if "bazel" in content.lower() and any(flag in content for flag in _BAZEL_NETWORK_BLOCK_FLAGS) + ] + + bazelrc = path / ".bazelrc" + if bazelrc.exists(): + try: + raw = bazelrc.read_text(encoding="utf-8") + except Exception: + raw = "" + bazelrc_content = "\n".join(_strip_comment(line) for line in raw.splitlines()) + if any(flag in bazelrc_content for flag in _BAZEL_NETWORK_BLOCK_FLAGS): + bazel_hits.append(".bazelrc") + if bazel_hits: - return f"Bazel with network sandbox in CI ({', '.join(bazel_hits)})" + return f"Bazel with network sandbox ({', '.join(sorted(bazel_hits))})", witness_result - return None + return None, witness_result def repro_hermetic_build_handler( @@ -394,15 +452,30 @@ def repro_hermetic_build_handler( inside Dockerfiles are DEFERRED — building the image environment is fine; fetching application dependencies at build time is not. + v0.3: Adds a Sigstore-verified Witness/runtime-trace attestation check + (see witness_attestation.py) — fetches attestation artifacts from the + repo's latest successful CI run via the `gh` CLI, cryptographically + verifies them against the repo's GitHub Actions OIDC identity, and only + treats an empty, verified network log as a strong PASS signal. A verified + attestation that *does* record network activity is fed into the violation + list — stronger evidence than the CI-text grep below it. Requires the + `gh` CLI and `darnit-core[attestation]`; any missing prerequisite (no gh, + not authenticated, no matching CI run/artifact, sigstore not installed, + verification failure) degrades to a specific "no attestation evidence" + reason in ``evidence["strong_signal"]``/the witness result's ``detail``, + never to failing the audit. Set ``verify_witness_attestations = false`` in + the TOML pass config to skip the network round-trip entirely (air-gapped + audits, or repos with no usable `gh` login). + Result semantics (conservative-by-default): - - PASS: strong hermeticity signal (Witness, Nix flake CI, Bazel sandbox) - - FAIL: suspicious live network-fetch pattern in any scanned file + - PASS: strong hermeticity signal (verified Witness attestation, + Nix flake CI, Bazel sandbox) + - FAIL: suspicious live network-fetch pattern in any scanned file, + or a verified Witness attestation that recorded network activity - INCONCLUSIVE: files scanned, no violations, no strong signal (grep absence ≠ proof of hermeticity) - INCONCLUSIVE (confidence 0): no CI or build files found at all - - Roadmap: https://github.com/kusari-oss/darnit/issues/227 """ path = Path(ctx.local_path) @@ -428,7 +501,9 @@ def repro_hermetic_build_handler( }, ) - strong_signal = _detect_strong_hermeticity_signal(path, all_ci_files) + strong_signal, witness_result = _detect_strong_hermeticity_signal( + path, all_ci_files, ctx.dependency_results, ctx, config + ) if strong_signal: return HandlerResult( status=HandlerResultStatus.PASS, @@ -447,6 +522,12 @@ def repro_hermetic_build_handler( deferred: list[str] = [] files_scanned: list[str] = [] + # A verified Witness attestation that positively recorded network activity + # is stronger evidence than the grep heuristic below — surface it as a + # violation on its own rather than waiting for a matching CI-text pattern. + if witness_result.verified and witness_result.network_clean is False: + violations.append(f"witness attestation ({witness_result.evidence.get('artifact', '?')}): {witness_result.detail}") + for f in all_files: is_dockerfile = f in container_file_set try: diff --git a/packages/darnit-reproducibility/src/darnit_reproducibility/witness_attestation.py b/packages/darnit-reproducibility/src/darnit_reproducibility/witness_attestation.py new file mode 100644 index 00000000..84169f80 --- /dev/null +++ b/packages/darnit-reproducibility/src/darnit_reproducibility/witness_attestation.py @@ -0,0 +1,331 @@ +"""Verified Witness/in-toto runtime attestation check for RE-02.01. + +Unlike every other check in this plugin (pure local filesystem inspection), +this module reaches out to GitHub to fetch the latest CI run's attestation +artifacts and cryptographically verifies them before trusting anything they +claim. A JSON file that merely *says* "no network access" is not evidence — +only a Sigstore-verified DSSE envelope bound to the repo's GitHub Actions +OIDC identity is. + +Two predicate shapes are recognized: + +- Witness's own ``attestation-collection/v0.1``, whose nested ``command-run`` + attestation records ``processes[].cmdline``/``program`` (and opened file + digests) but has **no dedicated network field** — Witness's built-in tracer + does not observe sockets. For this shape we can only fall back to scanning + process command lines for the same suspicious substrings used by the grep + heuristic in ``handlers.py``, which is not an authoritative "no network + access" claim, only a negative-evidence hint. +- The newer, monitor-agnostic ``runtime-trace/v0.1`` predicate, which *does* + define a top-level ``network`` array. An empty array is treated as an + authoritative "no network access" claim; a non-empty one is authoritative + evidence of network access. (The spec still defers the internal shape of + each event to ``monitor.type``, so we only rely on emptiness, not content.) + +Every failure mode here (no ``gh`` CLI, no auth, no matching run/artifact, +``sigstore`` not installed, verification failure, unrecognized predicate) +degrades to "no attestation evidence" rather than raising — this must never +be able to fail an audit outright. +""" + +from __future__ import annotations + +import base64 +import json +import subprocess +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from darnit.core.logging import get_logger +from darnit.sieve.handler_registry import HandlerContext + +logger = get_logger("darnit_reproducibility.witness_attestation") + +try: + from sigstore.models import Bundle + from sigstore.verify import Verifier + from sigstore.verify.policy import AllOf, GitHubWorkflowRepository, OIDCIssuer + + SIGSTORE_VERIFY_AVAILABLE = True +except ImportError: + SIGSTORE_VERIFY_AVAILABLE = False + +GITHUB_ACTIONS_OIDC_ISSUER = "https://token.actions.githubusercontent.com" + +_WITNESS_COLLECTION_TYPE = "https://witness.dev/attestation-collection/v0.1" +_RUNTIME_TRACE_TYPE = "https://in-toto.io/attestation/runtime-trace/v0.1" + +# Same substrings as handlers._SUSPICIOUS_PATTERNS, duplicated deliberately: +# this is a best-effort fallback over attacker-influenced process cmdlines +# from a *different* data source (Witness command-run), not the CI-file scan. +_SUSPICIOUS_CMDLINE_PATTERNS: tuple[str, ...] = ( + "curl ", + "wget ", + "pip install ", + "npm install", + "yarn install", + "apt-get install", + "brew install", +) + +_GH_TIMEOUT_SECONDS = 60 +_MAX_ARTIFACT_FILES = 5 + +# Substrings gh prints to stderr on an auth failure — used to tell "you're not +# logged in" apart from "nothing found", which otherwise look identical (both +# are just "no candidate files") to the caller. +_AUTH_ERROR_HINTS: tuple[str, ...] = ( + "gh auth login", + "not logged into", + "authentication", + "bad credentials", + "401", + "requires authentication", +) + + +@dataclass +class WitnessCheckResult: + """Outcome of attempting to verify a Witness/runtime-trace attestation.""" + + attempted: bool + verified: bool = False + network_clean: bool | None = None # True/False = authoritative; None = no authoritative signal + detail: str = "" + evidence: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class _GhOutcome: + """Result of a single `gh` invocation, with a human-readable reason on failure.""" + + proc: subprocess.CompletedProcess[str] | None + reason: str | None = None + + +def _run_gh(args: list[str]) -> _GhOutcome: + try: + proc = subprocess.run( + ["gh", *args], + capture_output=True, + text=True, + timeout=_GH_TIMEOUT_SECONDS, + ) + except FileNotFoundError: + return _GhOutcome(None, "gh CLI not found in PATH") + except subprocess.TimeoutExpired: + return _GhOutcome(None, f"gh command timed out after {_GH_TIMEOUT_SECONDS}s") + + if proc.returncode == 0: + return _GhOutcome(proc) + + stderr_lower = proc.stderr.lower() + if any(hint in stderr_lower for hint in _AUTH_ERROR_HINTS): + return _GhOutcome(None, "gh is not authenticated for this repository (run `gh auth login`)") + return _GhOutcome(None, f"gh exited {proc.returncode}: {proc.stderr.strip()[:200]}") + + +def _latest_successful_run_id(owner: str, repo: str, branch: str) -> tuple[str | None, str | None]: + """Returns (run_id, failure_reason) — exactly one is None.""" + outcome = _run_gh( + [ + "run", + "list", + "--repo", + f"{owner}/{repo}", + "--branch", + branch, + "--status", + "success", + "--limit", + "1", + "--json", + "databaseId", + ] + ) + if outcome.proc is None: + return None, outcome.reason + if not outcome.proc.stdout: + return None, "gh returned no output for the run list query" + try: + rows = json.loads(outcome.proc.stdout) + except json.JSONDecodeError: + return None, "gh returned unparseable output for the run list query" + if not rows: + return None, f"no successful CI run found on branch '{branch}'" + run_id = rows[0].get("databaseId") + if not run_id: + return None, "latest successful run has no databaseId" + return str(run_id), None + + +def _download_candidate_artifacts(owner: str, repo: str, run_id: str, dest: Path) -> tuple[list[Path], str | None]: + """Returns (files, failure_reason) — files is empty iff failure_reason is set.""" + outcome = _run_gh( + [ + "run", + "download", + run_id, + "--repo", + f"{owner}/{repo}", + "--pattern", + "*witness*", + "--dir", + str(dest), + ] + ) + if outcome.proc is None: + return [], outcome.reason + found = sorted(dest.rglob("*.json"))[:_MAX_ARTIFACT_FILES] + if not found: + return [], f"run {run_id} has no artifacts matching '*witness*'" + return found, None + + +def _fetch_candidate_files(ctx: HandlerContext, scratch_dir: Path) -> tuple[list[Path], str | None]: + """Best-effort fetch of Witness attestation artifacts from the latest CI run. + + Returns (files, failure_reason) — files is empty iff failure_reason is set. + """ + if not ctx.owner or not ctx.repo: + return [], "repository owner/name not available in this context" + run_id, reason = _latest_successful_run_id(ctx.owner, ctx.repo, ctx.default_branch) + if not run_id: + return [], reason + return _download_candidate_artifacts(ctx.owner, ctx.repo, run_id, scratch_dir) + + +def _verify_bundle(raw_bytes: bytes, owner: str, repo: str) -> dict[str, Any] | None: + """Verify a Sigstore-bundled DSSE envelope against the repo's GitHub + Actions OIDC identity. Returns the decoded in-toto statement on success, + or None if verification is unavailable or fails. + """ + if not SIGSTORE_VERIFY_AVAILABLE: + return None + try: + bundle = Bundle.from_json(raw_bytes) + except Exception as exc: + logger.debug("not a Sigstore bundle: %s", exc) + return None + + policy = AllOf( + [ + OIDCIssuer(GITHUB_ACTIONS_OIDC_ISSUER), + GitHubWorkflowRepository(f"{owner}/{repo}"), + ] + ) + + try: + payload_type, payload_bytes = Verifier.production().verify_dsse(bundle, policy) + except Exception as exc: + logger.debug("Sigstore verification failed: %s", exc) + return None + + if "in-toto" not in payload_type: + return None + try: + return json.loads(payload_bytes) + except json.JSONDecodeError: + return None + + +def _decode_raw_dsse(raw_bytes: bytes) -> dict[str, Any] | None: + """Fallback for a bare (unsigned or non-Sigstore-bundled) DSSE envelope. + + Only used to populate evidence for debugging — never treated as verified. + """ + try: + envelope = json.loads(raw_bytes) + payload_b64 = envelope.get("payload") + if not payload_b64: + return None + return json.loads(base64.b64decode(payload_b64)) + except Exception: + return None + + +def _nested_attestations(statement: dict[str, Any]) -> list[dict[str, Any]]: + predicate = statement.get("predicate", {}) + if statement.get("predicateType") == _WITNESS_COLLECTION_TYPE: + return predicate.get("attestations", []) + return [{"type": statement.get("predicateType", ""), "attestation": predicate}] + + +def _check_network_cleanliness(statement: dict[str, Any]) -> tuple[bool | None, str]: + """Inspect a verified in-toto statement for network-access evidence. + + Returns ``(network_clean, detail)``: + - ``(True, ...)`` — authoritative: a runtime-trace ``network`` array was + present and empty. + - ``(False, ...)`` — authoritative: a non-empty ``network`` array, or a + command-run process cmdline matched a suspicious pattern. + - ``(None, ...)`` — no authoritative signal (command-run only, nothing + suspicious found — absence of evidence, not evidence of absence). + """ + for entry in _nested_attestations(statement): + entry_type = entry.get("type", "") + payload = entry.get("attestation", {}) + + if "runtime-trace" in entry_type or "network" in payload: + network_events = payload.get("monitorLog", {}).get("network", payload.get("network")) + if network_events is not None: + if len(network_events) == 0: + return True, "runtime-trace predicate recorded an empty network log" + return False, f"runtime-trace predicate recorded {len(network_events)} network event(s)" + + if "command-run" in entry_type or "commandrun" in entry_type: + for proc in payload.get("processes", []) or []: + haystack = f"{proc.get('program', '')} {proc.get('cmdline', '')}" + for pattern in _SUSPICIOUS_CMDLINE_PATTERNS: + if pattern in haystack: + return False, f"command-run process matched '{pattern.strip()}': {haystack.strip()[:120]}" + + return None, "no authoritative network signal in verified attestation" + + +def check_witness_attestation(ctx: HandlerContext) -> WitnessCheckResult: + """Fetch, verify, and inspect the latest CI run's Witness attestation. + + Returns a result with ``verified=False`` (and no PASS-worthy signal) for + any missing prerequisite. Never raises. + """ + if not SIGSTORE_VERIFY_AVAILABLE: + return WitnessCheckResult( + attempted=False, + detail="sigstore not installed — install darnit-core[attestation] to enable", + ) + + with tempfile.TemporaryDirectory(prefix="darnit-witness-") as tmp: + candidates, reason = _fetch_candidate_files(ctx, Path(tmp)) + if not candidates: + return WitnessCheckResult(attempted=True, detail=reason or "no Witness attestation artifacts found") + + checked_files: list[str] = [] + for f in candidates: + checked_files.append(f.name) + try: + raw_bytes = f.read_bytes() + except OSError as exc: + logger.debug("could not read %s: %s", f, exc) + continue + + statement = _verify_bundle(raw_bytes, ctx.owner, ctx.repo) + if statement is None: + continue # not verifiable — do not fall back to trusting unsigned content + + network_clean, detail = _check_network_cleanliness(statement) + return WitnessCheckResult( + attempted=True, + verified=True, + network_clean=network_clean, + detail=detail, + evidence={"artifact": f.name, "checked_files": checked_files}, + ) + + return WitnessCheckResult( + attempted=True, + detail="attestation artifact(s) found but none verified against the repo's GitHub Actions identity", + evidence={"checked_files": checked_files}, + ) diff --git a/tests/darnit_reproducibility/test_handlers.py b/tests/darnit_reproducibility/test_handlers.py index 30f92cd6..c620dd9a 100644 --- a/tests/darnit_reproducibility/test_handlers.py +++ b/tests/darnit_reproducibility/test_handlers.py @@ -2,6 +2,9 @@ from pathlib import Path +import pytest + +from darnit.sieve.handler_registry import HandlerContext, HandlerResultStatus from darnit_reproducibility.handlers import ( _detect_strong_hermeticity_signal, _iter_build_files, @@ -9,6 +12,7 @@ _iter_container_files, _iter_other_ci_files, _iter_workflow_files, + _maybe_check_witness_attestation, _scan_line, _strip_comment, repro_bit_for_bit_handler, @@ -17,11 +21,24 @@ repro_hermetic_build_handler, repro_provenance_exists_handler, ) +from darnit_reproducibility.witness_attestation import WitnessCheckResult + +# _detect_strong_hermeticity_signal and repro_hermetic_build_handler both call +# check_witness_attestation(), which shells out to `gh` and the network. Tests +# that aren't specifically exercising that path stub it out to a no-op result; +# witness-specific tests override it again with monkeypatch.setattr. +NO_WITNESS_EVIDENCE = WitnessCheckResult(attempted=False) -from darnit.sieve.handler_registry import HandlerContext, HandlerResultStatus + +@pytest.fixture(autouse=True) +def _stub_witness_attestation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "darnit_reproducibility.handlers.check_witness_attestation", + lambda ctx: NO_WITNESS_EVIDENCE, + ) -def make_ctx(tmp_path: Path) -> HandlerContext: +def make_ctx(tmp_path: Path, dependency_results: dict[str, str] | None = None) -> HandlerContext: return HandlerContext( local_path=str(tmp_path), owner="org", @@ -31,7 +48,7 @@ def make_ctx(tmp_path: Path) -> HandlerContext: project_context={}, gathered_evidence={}, shared_cache={}, - dependency_results={}, + dependency_results=dependency_results if dependency_results is not None else {}, ) @@ -197,77 +214,142 @@ def test_container_files_none_present(self, tmp_path: Path) -> None: class TestDetectStrongSignal: - """Unit tests for _detect_strong_hermeticity_signal.""" + """Unit tests for _detect_strong_hermeticity_signal. + + ``check_witness_attestation`` is stubbed to a no-op by the module-level + ``_stub_witness_attestation`` fixture unless a test overrides it below. + """ def test_returns_none_with_no_signals(self, tmp_path: Path) -> None: - result = _detect_strong_hermeticity_signal(tmp_path, []) - assert result is None + signal, _ = _detect_strong_hermeticity_signal(tmp_path, [], {}, make_ctx(tmp_path), {}) + assert signal is None - def test_witness_run_detected(self, tmp_path: Path) -> None: + def test_witness_mention_alone_is_not_a_signal(self, tmp_path: Path) -> None: + # Merely mentioning "witness run" in CI text proves the tool ran, not + # what it observed — only a verified attestation with a clean network + # log counts now (see test_verified_witness_attestation_is_a_signal). wf = tmp_path / "ci.yml" wf.write_text("- run: witness run -- make build\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is not None - assert "Witness" in result - - def test_testifysec_witness_action_detected(self, tmp_path: Path) -> None: - wf = tmp_path / "ci.yml" - wf.write_text("uses: testifysec/witness-run-action@v0.1\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is not None - assert "Witness" in result + signal, witness_result = _detect_strong_hermeticity_signal(tmp_path, [wf], {}, make_ctx(tmp_path), {}) + assert signal is None + assert witness_result.verified is False + + def test_verified_witness_attestation_is_a_signal(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + verified = WitnessCheckResult( + attempted=True, + verified=True, + network_clean=True, + detail="runtime-trace predicate recorded an empty network log", + ) + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: verified) + signal, witness_result = _detect_strong_hermeticity_signal(tmp_path, [], {}, make_ctx(tmp_path), {}) + assert signal is not None + assert "Witness" in signal + assert witness_result is verified + + def test_verified_witness_attestation_with_network_activity_is_not_a_pass_signal( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + dirty = WitnessCheckResult( + attempted=True, + verified=True, + network_clean=False, + detail="runtime-trace predicate recorded 2 network event(s)", + ) + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: dirty) + signal, witness_result = _detect_strong_hermeticity_signal(tmp_path, [], {}, make_ctx(tmp_path), {}) + assert signal is None + assert witness_result.network_clean is False + + def test_witness_check_disabled_via_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + verified = WitnessCheckResult(attempted=True, verified=True, network_clean=True, detail="clean") + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: verified) + signal, witness_result = _detect_strong_hermeticity_signal( + tmp_path, [], {}, make_ctx(tmp_path), {"verify_witness_attestations": False} + ) + # The mocked check_witness_attestation returns a verified/clean result, + # but the config toggle must prevent it from ever being called. + assert signal is None + assert witness_result.attempted is False + assert "disabled via config" in witness_result.detail def test_nix_flake_with_nix_build_in_ci(self, tmp_path: Path) -> None: (tmp_path / "flake.nix").write_text("{ outputs = {}; }") wf = tmp_path / "ci.yml" wf.write_text("- run: nix build .#default\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is not None - assert "Nix" in result + signal, _ = _detect_strong_hermeticity_signal( + tmp_path, [wf], {"RE-01.02": "PASS"}, make_ctx(tmp_path, {"RE-01.02": "PASS"}), {} + ) + assert signal is not None + assert "Nix" in signal def test_nix_flake_present_but_no_ci_usage(self, tmp_path: Path) -> None: (tmp_path / "flake.nix").write_text("{ outputs = {}; }") wf = tmp_path / "ci.yml" wf.write_text("- run: uv sync\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is None + signal, _ = _detect_strong_hermeticity_signal( + tmp_path, [wf], {"RE-01.02": "PASS"}, make_ctx(tmp_path, {"RE-01.02": "PASS"}), {} + ) + assert signal is None + + def test_nix_flake_not_gated_without_build_env_declared_pass(self, tmp_path: Path) -> None: + # flake.nix + CI usage looks right, but RE-01.02 (BuildEnvDeclared) never + # ran or didn't PASS — withhold the signal rather than assume. + (tmp_path / "flake.nix").write_text("{ outputs = {}; }") + wf = tmp_path / "ci.yml" + wf.write_text("- run: nix build .#default\n") + signal, _ = _detect_strong_hermeticity_signal(tmp_path, [wf], {}, make_ctx(tmp_path), {}) + assert signal is None + + def test_nix_flake_not_gated_when_build_env_declared_failed(self, tmp_path: Path) -> None: + (tmp_path / "flake.nix").write_text("{ outputs = {}; }") + wf = tmp_path / "ci.yml" + wf.write_text("- run: nix build .#default\n") + signal, _ = _detect_strong_hermeticity_signal( + tmp_path, [wf], {"RE-01.02": "FAIL"}, make_ctx(tmp_path, {"RE-01.02": "FAIL"}), {} + ) + assert signal is None def test_bazel_with_network_sandbox_flag(self, tmp_path: Path) -> None: (tmp_path / "MODULE.bazel").write_text("module(name = 'myproject')") wf = tmp_path / "ci.yml" wf.write_text("- run: bazel build //... --sandbox_default_allow_network=false\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is not None - assert "Bazel" in result + signal, _ = _detect_strong_hermeticity_signal(tmp_path, [wf], {}, make_ctx(tmp_path), {}) + assert signal is not None + assert "Bazel" in signal def test_bazel_workspace_without_sandbox_flag(self, tmp_path: Path) -> None: # Bazel allows network by default — workspace alone is not enough (tmp_path / "WORKSPACE").write_text("") wf = tmp_path / "ci.yml" wf.write_text("- run: bazel build //...\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is None - - def test_witness_takes_priority_over_nix(self, tmp_path: Path) -> None: + signal, _ = _detect_strong_hermeticity_signal(tmp_path, [wf], {}, make_ctx(tmp_path), {}) + assert signal is None + + def test_witness_takes_priority_over_nix(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # No CI file mentions "witness" at all here — the point of this test + # is that a verified attestation still wins even though there is no + # text-based hint that Witness is in use (e.g. it ran via a reusable + # workflow the caller's own CI files never name). + verified = WitnessCheckResult(attempted=True, verified=True, network_clean=True, detail="clean") + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: verified) (tmp_path / "flake.nix").write_text("{ outputs = {}; }") wf = tmp_path / "ci.yml" - wf.write_text("- run: witness run -- nix build .#default\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is not None - assert "Witness" in result - - def test_commented_witness_reference_is_not_a_signal(self, tmp_path: Path) -> None: - wf = tmp_path / "ci.yml" - wf.write_text("# TODO: add witness run someday\nsteps:\n - run: uv sync\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is None + wf.write_text("- run: nix build .#default\n") + signal, _ = _detect_strong_hermeticity_signal( + tmp_path, [wf], {"RE-01.02": "PASS"}, make_ctx(tmp_path, {"RE-01.02": "PASS"}), {} + ) + assert signal is not None + assert "Witness" in signal def test_commented_nix_reference_is_not_a_signal(self, tmp_path: Path) -> None: (tmp_path / "flake.nix").write_text("{ outputs = {}; }") wf = tmp_path / "ci.yml" wf.write_text("# TODO: nix build .#default someday\nsteps:\n - run: uv sync\n") - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is None + signal, _ = _detect_strong_hermeticity_signal( + tmp_path, [wf], {"RE-01.02": "PASS"}, make_ctx(tmp_path, {"RE-01.02": "PASS"}), {} + ) + assert signal is None def test_commented_bazel_sandbox_flag_is_not_a_signal(self, tmp_path: Path) -> None: (tmp_path / "MODULE.bazel").write_text("module(name = 'myproject')") @@ -277,8 +359,37 @@ def test_commented_bazel_sandbox_flag_is_not_a_signal(self, tmp_path: Path) -> N " # TODO: bazel build //... --sandbox_default_allow_network=false\n" " - run: bazel build //...\n" ) - result = _detect_strong_hermeticity_signal(tmp_path, [wf]) - assert result is None + signal, _ = _detect_strong_hermeticity_signal(tmp_path, [wf], {}, make_ctx(tmp_path), {}) + assert signal is None + + +class TestMaybeCheckWitnessAttestation: + """Unit tests for the config-toggle wrapper around check_witness_attestation().""" + + def test_disabled_via_config_short_circuits(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + called = False + + def spy(ctx: HandlerContext) -> WitnessCheckResult: + nonlocal called + called = True + return WitnessCheckResult(attempted=True, verified=True, network_clean=True) + + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", spy) + result = _maybe_check_witness_attestation(make_ctx(tmp_path), {"verify_witness_attestations": False}) + assert called is False + assert result.attempted is False + + def test_enabled_by_default(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + expected = WitnessCheckResult(attempted=True, verified=True, network_clean=True) + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: expected) + result = _maybe_check_witness_attestation(make_ctx(tmp_path), {}) + assert result is expected + + def test_explicitly_enabled_via_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + expected = WitnessCheckResult(attempted=True, verified=True, network_clean=True) + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: expected) + result = _maybe_check_witness_attestation(make_ctx(tmp_path), {"verify_witness_attestations": True}) + assert result is expected class TestRepoDepsPin: @@ -461,7 +572,23 @@ def test_inconclusive_inline_comment_with_violation(self, tmp_path: Path) -> Non # Strong signals → PASS # ------------------------------------------------------------------ - def test_pass_witness_in_workflow(self, tmp_path: Path) -> None: + def test_witness_mention_alone_does_not_pass(self, tmp_path: Path) -> None: + # Text-only mention of witness in CI is no longer sufficient for a + # PASS — see test_pass_verified_witness_attestation below. + wf_dir = tmp_path / ".github" / "workflows" + wf_dir.mkdir(parents=True) + (wf_dir / "ci.yml").write_text("steps:\n - uses: testifysec/witness-run-action@v0.1\n") + result = repro_hermetic_build_handler({}, make_ctx(tmp_path)) + assert result.status == HandlerResultStatus.INCONCLUSIVE + + def test_pass_verified_witness_attestation(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + verified = WitnessCheckResult( + attempted=True, + verified=True, + network_clean=True, + detail="runtime-trace predicate recorded an empty network log", + ) + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: verified) wf_dir = tmp_path / ".github" / "workflows" wf_dir.mkdir(parents=True) (wf_dir / "ci.yml").write_text("steps:\n - uses: testifysec/witness-run-action@v0.1\n") @@ -469,15 +596,56 @@ def test_pass_witness_in_workflow(self, tmp_path: Path) -> None: assert result.status == HandlerResultStatus.PASS assert "Witness" in result.message + def test_fail_verified_witness_attestation_with_network_activity( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + dirty = WitnessCheckResult( + attempted=True, + verified=True, + network_clean=False, + detail="runtime-trace predicate recorded 1 network event(s)", + evidence={"artifact": "witness-attestation.json"}, + ) + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: dirty) + wf_dir = tmp_path / ".github" / "workflows" + wf_dir.mkdir(parents=True) + (wf_dir / "ci.yml").write_text("steps:\n - run: uv sync\n") + result = repro_hermetic_build_handler({}, make_ctx(tmp_path)) + assert result.status == HandlerResultStatus.FAIL + assert any("witness attestation" in v for v in result.evidence["violations_found"]) + + def test_witness_verification_disabled_via_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # Even a mocked verified/clean attestation must not produce a PASS + # when the pass config opts out of the network round-trip. + verified = WitnessCheckResult(attempted=True, verified=True, network_clean=True, detail="clean") + monkeypatch.setattr("darnit_reproducibility.handlers.check_witness_attestation", lambda ctx: verified) + wf_dir = tmp_path / ".github" / "workflows" + wf_dir.mkdir(parents=True) + (wf_dir / "ci.yml").write_text("steps:\n - uses: testifysec/witness-run-action@v0.1\n") + result = repro_hermetic_build_handler({"verify_witness_attestations": False}, make_ctx(tmp_path)) + assert result.status == HandlerResultStatus.INCONCLUSIVE + def test_pass_nix_flake_build_in_ci(self, tmp_path: Path) -> None: (tmp_path / "flake.nix").write_text("{ outputs = {}; }") wf_dir = tmp_path / ".github" / "workflows" wf_dir.mkdir(parents=True) (wf_dir / "ci.yml").write_text("steps:\n - run: nix build .#default\n") - result = repro_hermetic_build_handler({}, make_ctx(tmp_path)) + ctx = make_ctx(tmp_path, dependency_results={"RE-01.02": "PASS"}) + result = repro_hermetic_build_handler({}, ctx) assert result.status == HandlerResultStatus.PASS assert "Nix" in result.message + def test_inconclusive_nix_flake_without_build_env_declared_pass(self, tmp_path: Path) -> None: + # Same repo shape as the PASS case above, but RE-01.02 (BuildEnvDeclared) + # never confirmed flake.nix as the declared build environment — the nix + # strong signal must be withheld, falling back to the grep heuristic. + (tmp_path / "flake.nix").write_text("{ outputs = {}; }") + wf_dir = tmp_path / ".github" / "workflows" + wf_dir.mkdir(parents=True) + (wf_dir / "ci.yml").write_text("steps:\n - run: nix build .#default\n") + result = repro_hermetic_build_handler({}, make_ctx(tmp_path)) + assert result.status == HandlerResultStatus.INCONCLUSIVE + def test_pass_bazel_with_sandbox_flag(self, tmp_path: Path) -> None: (tmp_path / "MODULE.bazel").write_text("module(name = 'myproject')") wf_dir = tmp_path / ".github" / "workflows" diff --git a/tests/darnit_reproducibility/test_witness_attestation.py b/tests/darnit_reproducibility/test_witness_attestation.py new file mode 100644 index 00000000..2e063153 --- /dev/null +++ b/tests/darnit_reproducibility/test_witness_attestation.py @@ -0,0 +1,441 @@ +"""Tests for the Witness/runtime-trace attestation verification helper. + +Sigstore-dependent tests (``_verify_bundle`` success/failure paths) are +skipped when ``sigstore`` isn't installed — install the `attestation` extra +(``uv sync --extra attestation``) to run them. +""" + +from __future__ import annotations + +import base64 +import json +import subprocess +from pathlib import Path +from typing import Any + +import pytest +from darnit.sieve.handler_registry import HandlerContext +from darnit_reproducibility import witness_attestation as wa + +needs_sigstore = pytest.mark.skipif( + not wa.SIGSTORE_VERIFY_AVAILABLE, + reason="sigstore not installed — run `uv sync --extra attestation`", +) + + +def make_ctx(owner: str = "org", repo: str = "repo", branch: str = "main") -> HandlerContext: + return HandlerContext(local_path=".", owner=owner, repo=repo, default_branch=branch) + + +def fake_proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(args=["gh"], returncode=returncode, stdout=stdout, stderr=stderr) + + +class TestRunGh: + def test_returns_outcome_with_proc_on_success(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa.subprocess, "run", lambda *a, **kw: fake_proc(stdout="ok")) + outcome = wa._run_gh(["run", "list"]) + assert outcome.proc is not None + assert outcome.proc.stdout == "ok" + assert outcome.reason is None + + def test_missing_binary_sets_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + def raise_not_found(*args: Any, **kwargs: Any) -> None: + raise FileNotFoundError("gh not found") + + monkeypatch.setattr(wa.subprocess, "run", raise_not_found) + outcome = wa._run_gh(["run", "list"]) + assert outcome.proc is None + assert "not found in PATH" in outcome.reason + + def test_timeout_sets_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + def raise_timeout(*args: Any, **kwargs: Any) -> None: + raise subprocess.TimeoutExpired(cmd="gh", timeout=60) + + monkeypatch.setattr(wa.subprocess, "run", raise_timeout) + outcome = wa._run_gh(["run", "list"]) + assert outcome.proc is None + assert "timed out" in outcome.reason + + def test_auth_failure_stderr_is_recognized(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + wa.subprocess, + "run", + lambda *a, **kw: fake_proc(returncode=1, stderr="To use GitHub CLI, please run `gh auth login`."), + ) + outcome = wa._run_gh(["run", "list"]) + assert outcome.proc is None + assert "not authenticated" in outcome.reason + + def test_other_failure_includes_stderr(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + wa.subprocess, "run", lambda *a, **kw: fake_proc(returncode=1, stderr="repository not found") + ) + outcome = wa._run_gh(["run", "list"]) + assert outcome.proc is None + assert "gh exited 1" in outcome.reason + assert "repository not found" in outcome.reason + + +class TestLatestSuccessfulRunId: + def test_gh_unavailable_propagates_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(None, "gh CLI not found in PATH")) + run_id, reason = wa._latest_successful_run_id("org", "repo", "main") + assert run_id is None + assert reason == "gh CLI not found in PATH" + + def test_auth_failure_propagates_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + wa, + "_run_gh", + lambda args: wa._GhOutcome(None, "gh is not authenticated for this repository (run `gh auth login`)"), + ) + run_id, reason = wa._latest_successful_run_id("org", "repo", "main") + assert run_id is None + assert "not authenticated" in reason + + def test_empty_stdout_returns_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(stdout=""))) + run_id, reason = wa._latest_successful_run_id("org", "repo", "main") + assert run_id is None + assert "no output" in reason + + def test_invalid_json_returns_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(stdout="not json"))) + run_id, reason = wa._latest_successful_run_id("org", "repo", "main") + assert run_id is None + assert "unparseable" in reason + + def test_empty_list_returns_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(stdout="[]"))) + run_id, reason = wa._latest_successful_run_id("org", "repo", "main") + assert run_id is None + assert "no successful CI run" in reason + + def test_valid_run_returns_id_as_string(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(stdout=json.dumps([{"databaseId": 123456}]))) + ) + run_id, reason = wa._latest_successful_run_id("org", "repo", "main") + assert run_id == "123456" + assert reason is None + + def test_requests_the_right_repo_and_branch(self, monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, list[str]] = {} + + def spy(args: list[str]) -> wa._GhOutcome: + captured["args"] = args + return wa._GhOutcome(fake_proc(stdout=json.dumps([{"databaseId": 1}]))) + + monkeypatch.setattr(wa, "_run_gh", spy) + wa._latest_successful_run_id("kusari-oss", "darnit", "main") + assert "kusari-oss/darnit" in captured["args"] + assert "main" in captured["args"] + assert "success" in captured["args"] + + +class TestDownloadCandidateArtifacts: + def test_gh_failure_propagates_reason(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(None, "gh CLI not found in PATH")) + files, reason = wa._download_candidate_artifacts("org", "repo", "123", tmp_path) + assert files == [] + assert reason == "gh CLI not found in PATH" + + def test_finds_downloaded_json_files(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # `gh run download` would have written these as a side effect; the + # mock only needs to report success and leave them in place. + nested = tmp_path / "witness-attestation" + nested.mkdir() + (nested / "attestation.json").write_text("{}") + (nested / "readme.txt").write_text("not json") + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(returncode=0))) + + files, reason = wa._download_candidate_artifacts("org", "repo", "123", tmp_path) + assert [f.name for f in files] == ["attestation.json"] + assert reason is None + + def test_no_matching_artifacts_returns_reason(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # gh succeeded but nothing matched the *witness* pattern. + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(returncode=0))) + files, reason = wa._download_candidate_artifacts("org", "repo", "123", tmp_path) + assert files == [] + assert "no artifacts matching" in reason + + def test_caps_at_max_artifact_files(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + for i in range(wa._MAX_ARTIFACT_FILES + 3): + (tmp_path / f"attestation-{i}.json").write_text("{}") + monkeypatch.setattr(wa, "_run_gh", lambda args: wa._GhOutcome(fake_proc(returncode=0))) + + files, _ = wa._download_candidate_artifacts("org", "repo", "123", tmp_path) + assert len(files) == wa._MAX_ARTIFACT_FILES + + +class TestFetchCandidateFiles: + def test_missing_owner_returns_reason(self, tmp_path: Path) -> None: + ctx = make_ctx(owner="") + files, reason = wa._fetch_candidate_files(ctx, tmp_path) + assert files == [] + assert "owner/name not available" in reason + + def test_missing_repo_returns_reason(self, tmp_path: Path) -> None: + ctx = make_ctx(repo="") + files, reason = wa._fetch_candidate_files(ctx, tmp_path) + assert files == [] + assert "owner/name not available" in reason + + def test_no_run_found_propagates_reason(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_latest_successful_run_id", lambda owner, repo, branch: (None, "no successful CI run found on branch 'main'")) + files, reason = wa._fetch_candidate_files(make_ctx(), tmp_path) + assert files == [] + assert "no successful CI run" in reason + + def test_delegates_to_download_with_run_id(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "_latest_successful_run_id", lambda owner, repo, branch: ("999", None)) + captured: dict[str, Any] = {} + + def spy(owner: str, repo: str, run_id: str, dest: Path) -> tuple[list[Path], str | None]: + captured.update(owner=owner, repo=repo, run_id=run_id) + return [dest / "attestation.json"], None + + monkeypatch.setattr(wa, "_download_candidate_artifacts", spy) + files, reason = wa._fetch_candidate_files(make_ctx(owner="org", repo="repo"), tmp_path) + assert captured == {"owner": "org", "repo": "repo", "run_id": "999"} + assert files == [tmp_path / "attestation.json"] + assert reason is None + + +class TestNestedAttestations: + def test_witness_collection_unwraps_attestations_array(self) -> None: + statement = { + "predicateType": wa._WITNESS_COLLECTION_TYPE, + "predicate": {"attestations": [{"type": "command-run", "attestation": {"processes": []}}]}, + } + result = wa._nested_attestations(statement) + assert result == [{"type": "command-run", "attestation": {"processes": []}}] + + def test_non_collection_type_wraps_predicate_directly(self) -> None: + statement = { + "predicateType": wa._RUNTIME_TRACE_TYPE, + "predicate": {"network": []}, + } + result = wa._nested_attestations(statement) + assert result == [{"type": wa._RUNTIME_TRACE_TYPE, "attestation": {"network": []}}] + + +class TestCheckNetworkCleanliness: + def test_empty_runtime_trace_network_array_is_clean(self) -> None: + statement = {"predicateType": wa._RUNTIME_TRACE_TYPE, "predicate": {"network": []}} + clean, detail = wa._check_network_cleanliness(statement) + assert clean is True + assert "empty network log" in detail + + def test_nonempty_runtime_trace_network_array_is_dirty(self) -> None: + statement = { + "predicateType": wa._RUNTIME_TRACE_TYPE, + "predicate": {"network": [{"host": "evil.example.com"}]}, + } + clean, detail = wa._check_network_cleanliness(statement) + assert clean is False + assert "1 network event" in detail + + def test_network_under_monitor_log_is_recognized(self) -> None: + statement = { + "predicateType": wa._RUNTIME_TRACE_TYPE, + "predicate": {"monitorLog": {"network": []}}, + } + clean, _ = wa._check_network_cleanliness(statement) + assert clean is True + + def test_command_run_with_suspicious_cmdline_is_dirty(self) -> None: + statement = { + "predicateType": wa._WITNESS_COLLECTION_TYPE, + "predicate": { + "attestations": [ + { + "type": "https://witness.dev/attestations/command-run/v0.1", + "attestation": {"processes": [{"program": "/usr/bin/curl", "cmdline": "curl https://x"}]}, + } + ] + }, + } + clean, detail = wa._check_network_cleanliness(statement) + assert clean is False + assert "curl" in detail + + def test_command_run_with_clean_processes_has_no_authoritative_signal(self) -> None: + statement = { + "predicateType": wa._WITNESS_COLLECTION_TYPE, + "predicate": { + "attestations": [ + { + "type": "https://witness.dev/attestations/command-run/v0.1", + "attestation": {"processes": [{"program": "/usr/bin/make", "cmdline": "make build"}]}, + } + ] + }, + } + clean, detail = wa._check_network_cleanliness(statement) + assert clean is None + assert "no authoritative" in detail + + def test_no_recognized_attestations_has_no_authoritative_signal(self) -> None: + statement = {"predicateType": "https://example.com/something-else/v1", "predicate": {}} + clean, _ = wa._check_network_cleanliness(statement) + assert clean is None + + def test_witness_takes_runtime_trace_over_command_run_when_both_present(self) -> None: + # A collection could in principle carry both a command-run entry and a + # runtime-trace entry; the authoritative network signal must win even + # if it's not first in the list. + statement = { + "predicateType": wa._WITNESS_COLLECTION_TYPE, + "predicate": { + "attestations": [ + { + "type": "https://witness.dev/attestations/command-run/v0.1", + "attestation": {"processes": [{"program": "/usr/bin/make", "cmdline": "make build"}]}, + }, + {"type": wa._RUNTIME_TRACE_TYPE, "attestation": {"network": []}}, + ] + }, + } + clean, _ = wa._check_network_cleanliness(statement) + assert clean is True + + +class TestDecodeRawDsse: + def test_valid_envelope_decodes_payload(self) -> None: + inner = {"predicateType": "x", "predicate": {}} + payload_b64 = base64.b64encode(json.dumps(inner).encode()).decode() + envelope = json.dumps({"payload": payload_b64, "payloadType": "application/vnd.in-toto+json"}).encode() + assert wa._decode_raw_dsse(envelope) == inner + + def test_missing_payload_returns_none(self) -> None: + assert wa._decode_raw_dsse(json.dumps({}).encode()) is None + + def test_invalid_json_returns_none(self) -> None: + assert wa._decode_raw_dsse(b"not json") is None + + def test_invalid_base64_returns_none(self) -> None: + envelope = json.dumps({"payload": "not-valid-base64!!!"}).encode() + assert wa._decode_raw_dsse(envelope) is None + + +class TestVerifyBundle: + def test_sigstore_unavailable_returns_none(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "SIGSTORE_VERIFY_AVAILABLE", False) + assert wa._verify_bundle(b"{}", "org", "repo") is None + + @needs_sigstore + def test_invalid_bundle_bytes_returns_none(self) -> None: + assert wa._verify_bundle(b"not a sigstore bundle", "org", "repo") is None + + @needs_sigstore + def test_verification_failure_returns_none(self, monkeypatch: pytest.MonkeyPatch) -> None: + class FakeBundle: + @staticmethod + def from_json(raw: bytes) -> FakeBundle: + return FakeBundle() + + class FakeVerifier: + def verify_dsse(self, bundle: Any, policy: Any) -> tuple[str, bytes]: + raise RuntimeError("boom") + + monkeypatch.setattr(wa, "Bundle", FakeBundle) + monkeypatch.setattr(wa, "Verifier", type("V", (), {"production": staticmethod(lambda: FakeVerifier())})) + assert wa._verify_bundle(b"{}", "org", "repo") is None + + @needs_sigstore + def test_non_intoto_payload_type_returns_none(self, monkeypatch: pytest.MonkeyPatch) -> None: + class FakeBundle: + @staticmethod + def from_json(raw: bytes) -> FakeBundle: + return FakeBundle() + + class FakeVerifier: + def verify_dsse(self, bundle: Any, policy: Any) -> tuple[str, bytes]: + return "application/octet-stream", b"{}" + + monkeypatch.setattr(wa, "Bundle", FakeBundle) + monkeypatch.setattr(wa, "Verifier", type("V", (), {"production": staticmethod(lambda: FakeVerifier())})) + assert wa._verify_bundle(b"{}", "org", "repo") is None + + @needs_sigstore + def test_successful_verification_returns_statement(self, monkeypatch: pytest.MonkeyPatch) -> None: + inner_statement = {"predicateType": wa._RUNTIME_TRACE_TYPE, "predicate": {"network": []}} + + class FakeBundle: + @staticmethod + def from_json(raw: bytes) -> FakeBundle: + return FakeBundle() + + class FakeVerifier: + def verify_dsse(self, bundle: Any, policy: Any) -> tuple[str, bytes]: + return "application/vnd.in-toto+json", json.dumps(inner_statement).encode() + + monkeypatch.setattr(wa, "Bundle", FakeBundle) + monkeypatch.setattr(wa, "Verifier", type("V", (), {"production": staticmethod(lambda: FakeVerifier())})) + result = wa._verify_bundle(b"{}", "org", "repo") + assert result == inner_statement + + +class TestCheckWitnessAttestation: + def test_sigstore_unavailable_short_circuits(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "SIGSTORE_VERIFY_AVAILABLE", False) + result = wa.check_witness_attestation(make_ctx()) + assert result.attempted is False + assert result.verified is False + + def test_no_candidates_found_surfaces_specific_reason(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "SIGSTORE_VERIFY_AVAILABLE", True) + monkeypatch.setattr( + wa, + "_fetch_candidate_files", + lambda ctx, scratch_dir: ([], "gh is not authenticated for this repository (run `gh auth login`)"), + ) + result = wa.check_witness_attestation(make_ctx()) + assert result.attempted is True + assert result.verified is False + assert "not authenticated" in result.detail + + def test_no_candidates_found_falls_back_to_generic_detail(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wa, "SIGSTORE_VERIFY_AVAILABLE", True) + monkeypatch.setattr(wa, "_fetch_candidate_files", lambda ctx, scratch_dir: ([], None)) + result = wa.check_witness_attestation(make_ctx()) + assert result.attempted is True + assert result.verified is False + assert "no Witness attestation artifacts" in result.detail + + def test_candidates_found_but_none_verify(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + f = tmp_path / "attestation.json" + f.write_text("{}") + monkeypatch.setattr(wa, "SIGSTORE_VERIFY_AVAILABLE", True) + monkeypatch.setattr(wa, "_fetch_candidate_files", lambda ctx, scratch_dir: ([f], None)) + monkeypatch.setattr(wa, "_verify_bundle", lambda raw, owner, repo: None) + result = wa.check_witness_attestation(make_ctx()) + assert result.attempted is True + assert result.verified is False + assert result.evidence["checked_files"] == ["attestation.json"] + + def test_verified_clean_attestation_short_circuits_remaining_candidates( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + first = tmp_path / "a.json" + first.write_text("{}") + second = tmp_path / "b.json" + second.write_text("{}") + monkeypatch.setattr(wa, "SIGSTORE_VERIFY_AVAILABLE", True) + monkeypatch.setattr(wa, "_fetch_candidate_files", lambda ctx, scratch_dir: ([first, second], None)) + + statement = {"predicateType": wa._RUNTIME_TRACE_TYPE, "predicate": {"network": []}} + + def fake_verify(raw: bytes, owner: str, repo: str) -> dict[str, Any]: + return statement + + monkeypatch.setattr(wa, "_verify_bundle", fake_verify) + result = wa.check_witness_attestation(make_ctx()) + assert result.verified is True + assert result.network_clean is True + assert result.evidence["artifact"] == "a.json" + # only the first candidate's bytes should have been read/verified + assert result.evidence["checked_files"] == ["a.json"]