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