From 6aad42e20a406aeb2817f60e79c7610dd0fee08f Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Thu, 23 Jul 2026 22:38:44 -0600 Subject: [PATCH] feat(release): non-destructive, mirror-bound, signed-packet make audit Replaces the GitHub-DB-refreshing make audit with a stable, non-destructive audit: verifies a minisign-signed freshness receipt against pinned sol trust (key ID + public-key SHA-256), materializes the RustSec advisory DB only from a locally-supplied git bundle at the receipt's signed commit, and runs the pinned cargo-deny advisory check fully offline against that isolated DB, emitting exactly one JSON success witness. Four explicit operator inputs (AUDIT_ADVISORY_BUNDLE/RECEIPT/PUBKEY/LOCATOR); signature is derived as the adjacent .minisig; locator is used only as cargo-deny's db-urls identity and never contacted; no GitHub fallback. New authority module scripts/advisory_mirror_audit.py reuses the release-advisory isolation, minisign, and redaction primitives via import; the release-candidate prepare_policy_run path, PolicyRun/ledger schemas, and the 24h/14d rules are unchanged (regression-covered). Recipe keeps the cargo-deny + minisign preflights, redirects preflight stdout to stderr, and suppresses command echo so make audit stdout is exactly the witness JSON and the private locator is never printed. Adds a deterministic 113-test suite (fake tools/packets, real-git bundle materialization), rewrites the audit recipe contract test, and reconciles the now-false release_advisory_policy docstring. Co-Authored-By: Claude Opus 4.8 (1M context) --- Makefile | 5 +- docs/PORTING.md | 36 +- scripts/advisory_mirror_audit.py | 1079 +++++++++++++++++++++++ scripts/release_advisory_policy.py | 4 +- tests/test_advisory_mirror_audit.py | 1237 +++++++++++++++++++++++++++ tests/test_rust_policy_baseline.py | 37 +- 6 files changed, 2380 insertions(+), 18 deletions(-) create mode 100644 scripts/advisory_mirror_audit.py create mode 100644 tests/test_advisory_mirror_audit.py diff --git a/Makefile b/Makefile index 70bd5c59e..806c302fa 100644 --- a/Makefile +++ b/Makefile @@ -173,9 +173,8 @@ check-rust-deny: audit: @$(REQUIRE_CARGO) - @python3 scripts/check_release_preflight.py cargo-deny - @cargo deny --manifest-path $(RUST_MANIFEST) fetch db || { echo "ERROR: RustSec advisory refresh failed; no current advisory result was produced. Restore network access and rerun 'make audit'." >&2; exit 1; } - cargo deny --manifest-path $(RUST_MANIFEST) --locked --offline check advisories + @python3 scripts/check_release_preflight.py cargo-deny >&2 + @python3 scripts/advisory_mirror_audit.py --bundle "$(AUDIT_ADVISORY_BUNDLE)" --receipt "$(AUDIT_ADVISORY_RECEIPT)" --pubkey "$(AUDIT_ADVISORY_PUBKEY)" --locator "$(AUDIT_ADVISORY_LOCATOR)" # Setup skill symlinks skills: diff --git a/docs/PORTING.md b/docs/PORTING.md index ce05ae801..297849808 100644 --- a/docs/PORTING.md +++ b/docs/PORTING.md @@ -43,10 +43,44 @@ target, document the blocker and stop the conversion before merging it. | Rust lint | `make check-rust-clippy` | GNU-host check | Runs the existing clippy `-D warnings` gate. | | Rust tests | `make check-rust-test` | GNU-host check | Runs workspace Rust tests on the GNU host. | | Rust dependency policy | `make check-rust-deny` | GNU-host check | Locked, offline bans/licenses/sources policy over the supported cargo-deny graph. | -| Rust advisories | `make audit` | GNU-host check | Refreshes the advisory DB, then performs a locked offline advisory check. | +| Rust advisories | `make audit` | GNU-host check | Verifies a signed advisory mirror packet, materializes its bundle locally, then performs a locked offline advisory check without refreshing or mutating the operator inputs. | | iOS canary | `make check-rust-ios` | iOS cross-target canary | Cross-target drift evidence for eligible library crates; explicitly excludes `solstone-core-indexer-store` because the native SQLite store is not yet in the iOS gate. | | Release candidate rail | `scripts/release.sh --candidate` / `scripts/release.sh --recover ` | Local readiness evidence | DESTRUCTIVE: `--candidate` is fresh construction; before policy or build work it deletes prior raw build/dist outputs and that version's stale payload/evidence. It binds candidate payload, ledger, and per-target install/smoke proofs, then reports canonical local readiness JSON. `--recover` is retained-byte-only, read-only validation; it preserves retained payload, ledger, and proofs and never rebuilds or refreshes. Proofs cover local candidate bytes and native smoke only; publication is temporarily locked out of this rail. | +### Signed Advisory Mirror Audit + +`make audit` requires four operator-provided inputs: `AUDIT_ADVISORY_BUNDLE` +for the local advisory bundle, `AUDIT_ADVISORY_RECEIPT` for the freshness +receipt, `AUDIT_ADVISORY_PUBKEY` for the approved minisign public key, and +`AUDIT_ADVISORY_LOCATOR` for the private mirror locator. The signature selector +is derived from the receipt path as `.minisig`; there is no separate +signature option. + +The audit is local-only. Git verifies and clones only the local bundle file, and +the locator is used only as cargo-deny's advisory database identity in the +offline check. It is never used as a clone source, fetched, pulled, or probed. +Use a placeholder such as `PRIVATE_MIRROR_LOCATOR` in notes and logs; do not +record a real private host, path, credential, or URL-derived token. + +The public trust pins are key ID `5FCC81CD3DE12315` and public-key SHA-256 +`c9fb713fe57791afbdebddde7b334e950ce1efcc167d49daf4cc1cbd930bb122`. The +receipt must be canonical JSON, its adjacent minisign signature must carry the +trusted comment for the same advisory commit and UTC time, and the receipt UTC +is the only freshness authority. + +On success, stdout is exactly one compact JSON object with these fields: +`product`, `advisory_cohort`, `synced_commit`, `receipt_utc`, `max_age`, +`checked_at`, `cargo_lock_sha256`, `cargo_deny_version`, and `verdict`. +The witness contains no paths, locators, credentials, or child process output. + +The audit is non-destructive: packet inputs, the source tree, ambient Cargo +state, and release candidate/evidence directories are not modified. The +bundle-cloned advisory database and cargo-deny config are owned temporary +materialization and are removed before the success witness is emitted. +If any gate fails, reacquire the signed packet from the controlled mirror +process, place the adjacent signature next to the receipt, verify the public key +pin, and rerun `make audit`. + ## Owner Timezone The Python owner-timezone fallback is effectively `identity.timezone` from diff --git a/scripts/advisory_mirror_audit.py b/scripts/advisory_mirror_audit.py new file mode 100644 index 000000000..6f3421031 --- /dev/null +++ b/scripts/advisory_mirror_audit.py @@ -0,0 +1,1079 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Signed-packet advisory mirror audit for ``make audit``.""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path +from urllib.parse import urlparse + +ROOT = Path(__file__).resolve().parent.parent +_SCRIPTS_DIR = Path(__file__).resolve().parent +for _path in (str(ROOT), str(_SCRIPTS_DIR)): + if _path not in sys.path: + sys.path.insert(0, _path) + +from scripts.check_rust_release_manifest import ( # noqa: E402 + CONTROL_RE, + RAW_ENV_RE, + SECRET_RE, + SHA256_RE, + Failure, + validate_public_evidence_text, +) +from scripts.release_advisory_policy import ( # noqa: E402 + ADVISORY_TABLE_RE, + ReleasePolicyError, + _assert_scanned_snapshot, + _cleanup_temp, + _combined_release_policy_error, + _count_advisories, + _default_temp_path_factory, + _failure, + _format_utc, + _parse_utc, + _scanned_advisory_db, + _toml_string, + _unlink_path, + _utc_now, + _validate_advisory_count, + _validate_source, + advisory_check_argv, + is_normalized_utc_timestamp, +) +from scripts.release_tool_pins import CARGO_DENY_VERSION # noqa: E402 +from scripts.transparency_signing import ( # noqa: E402 + DriverError, + LocalMinisignSigner, + TransparencySigner, + check_minisign_binary, +) + +Runner = Callable[..., subprocess.CompletedProcess[str]] +Clock = Callable[[], datetime] +TempPathFactory = Callable[[str], Path] +PathRemover = Callable[[Path], None] + +PINNED_KEY_ID = "5FCC81CD3DE12315" +PINNED_PUBKEY_SHA256 = ( + "c9fb713fe57791afbdebddde7b334e950ce1efcc167d49daf4cc1cbd930bb122" +) +ADVISORY_COHORT_ID = "sol-controlled-rustsec-mirror-v1" +TRUSTED_COMMENT_SCHEME = "solpbc-advisory-mirror-v1" +RECEIPT_MAX_AGE = 86400 +MAX_CLOCK_SKEW = timedelta(minutes=5) + +PRODUCT = "solstone-journal" +ADVISORY_LOCATOR_TERMINALS = frozenset({"advisory-db", "rustsec-advisory-db.git"}) +GIT_COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") +KEY_ID_RE = re.compile(r"^[0-9A-F]{16}$") +SCP_LOCATOR_RE = re.compile(r"(?:[^@:/]+@)?(?P[^:/]+):(?P[^:]+)") +ABSOLUTE_PATH_RE = re.compile( + r"(^|\s)(?:/[^ \t\r\n]+|~[^ \t\r\n]*|[A-Za-z]:[\\/][^ \t\r\n]*)" +) +EMAIL_RE = re.compile(r"(?i)[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}") +UUID_RE = re.compile( + r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b" +) +PRIVATE_HOST_RE = re.compile(r"(?i)\b(?:localhost|[A-Za-z0-9-]+\.local)\b") +IP_CANDIDATE_RE = re.compile( + r"\b(?:\d{1,3}\.){3}\d{1,3}\b|\b[0-9a-f:]*:[0-9a-f:]+\b", + re.IGNORECASE, +) +CONTROL_EXCEPT_NEWLINE_RE = re.compile(r"[\x00-\x09\x0b-\x1f\x7f]") + +AUDIT_REPAIR = "reacquire the signed advisory mirror packet and rerun make audit" +INPUT_REPAIR = ( + "set AUDIT_ADVISORY_BUNDLE, AUDIT_ADVISORY_RECEIPT, " + "AUDIT_ADVISORY_PUBKEY, and AUDIT_ADVISORY_LOCATOR" +) + + +@dataclass(frozen=True) +class ReceiptAuthority: + synced_commit: str + utc: str + max_age: int + canonical_bytes: bytes + trusted_comment: str + + +@dataclass(frozen=True) +class PublicKeyMinisignVerifier: + public_key: Path + minisign: str = "minisign" + + def check(self) -> None: + check_minisign_binary(self.minisign) + + def sign_file( + self, + message_path: Path, + signature_path: Path, + *, + trusted_comment: str, + ) -> None: + raise NotImplementedError("advisory mirror audit verifier cannot sign") + + def verify_file( + self, + message_path: Path, + signature_path: Path, + *, + expected_trusted_comment: str, + ) -> None: + # LocalMinisignSigner.verify_file never reads the secret key; check() only + # performs the minisign 0.12 binary preflight and never requires a secret key. + LocalMinisignSigner( + secret_key=Path("__verify_only_unused_secret__"), + public_key=self.public_key, + minisign=self.minisign, + ).verify_file( + message_path, + signature_path, + expected_trusted_comment=expected_trusted_comment, + ) + + def trusted_comment(self, signature_path: Path) -> str: + return LocalMinisignSigner( + secret_key=Path("__verify_only_unused_secret__"), + public_key=self.public_key, + minisign=self.minisign, + ).trusted_comment(signature_path) + + +def _failure_record( + error: str, + *, + expected: str, + actual: str, + repair: str = AUDIT_REPAIR, +) -> Failure: + return _failure(error, expected=expected, actual=actual, repair=repair) + + +def _raise_one( + error: str, + *, + expected: str, + actual: str, + repair: str = AUDIT_REPAIR, +) -> None: + raise ReleasePolicyError( + [_failure_record(error, expected=expected, actual=actual, repair=repair)] + ) + + +def _redact_child_output(text: str, *, secrets: set[str]) -> str: + redacted = text + for secret in sorted((item for item in secrets if item), key=len, reverse=True): + redacted = redacted.replace(secret, "") + redacted = RAW_ENV_RE.sub("", redacted) + redacted = SECRET_RE.sub("", redacted) + redacted = ABSOLUTE_PATH_RE.sub( + lambda match: f"{match.group(1)}", redacted + ) + redacted = EMAIL_RE.sub("", redacted) + redacted = UUID_RE.sub("", redacted) + redacted = PRIVATE_HOST_RE.sub("", redacted) + redacted = IP_CANDIDATE_RE.sub("", redacted) + redacted = CONTROL_EXCEPT_NEWLINE_RE.sub("?", redacted) + if validate_public_evidence_text("child-output", redacted): + return "" + return redacted + + +def _convert_driver_error(exc: DriverError, *, secrets: set[str]) -> ReleasePolicyError: + failures: list[Failure] = [] + for failure in exc.failures: + failures.append( + _failure_record( + failure.error, + expected=failure.expected, + actual=_redact_child_output(failure.actual, secrets=secrets), + repair=failure.repair, + ) + ) + return ReleasePolicyError(failures) + + +def _safe_failure(failure: Failure) -> Failure: + text = ( + f"ERROR: {failure.error}\n" + f"expected: {failure.expected}\n" + f"actual: {failure.actual}\n" + f"repair: {failure.repair}\n" + ) + if not validate_public_evidence_text("failure", text): + return failure + return _failure_record( + failure.error, + expected=failure.expected, + actual="redacted", + repair=failure.repair, + ) + + +def _format_failures(failures: Sequence[Failure]) -> None: + for failure in failures: + safe = _safe_failure(failure) + print(f"ERROR: {safe.error}", file=sys.stderr) + print(f" expected: {safe.expected}", file=sys.stderr) + print(f" actual: {safe.actual}", file=sys.stderr) + print(f" repair command: {safe.repair}", file=sys.stderr) + + +def _validate_regular_input(path: Path, *, label: str) -> list[Failure]: + if path.is_symlink(): + return [ + _failure_record( + f"advisory mirror input {label} is unsafe", + expected=f"{label} regular file, not a symlink", + actual="symlink", + repair=INPUT_REPAIR, + ) + ] + if not path.exists(): + return [ + _failure_record( + f"advisory mirror input {label} is missing", + expected=f"{label} existing regular file", + actual="missing", + repair=INPUT_REPAIR, + ) + ] + if not path.is_file(): + return [ + _failure_record( + f"advisory mirror input {label} is unsafe", + expected=f"{label} regular file", + actual="not a regular file", + repair=INPUT_REPAIR, + ) + ] + return [] + + +def _receipt_signature_path(receipt: Path) -> Path: + return receipt.parent / f"{receipt.name}.minisig" + + +def _validate_inputs( + *, + bundle: Path, + receipt: Path, + pubkey: Path, +) -> tuple[Path, list[Failure]]: + signature = _receipt_signature_path(receipt) + failures: list[Failure] = [] + failures.extend(_validate_regular_input(bundle, label="bundle")) + failures.extend(_validate_regular_input(receipt, label="receipt")) + failures.extend(_validate_regular_input(pubkey, label="pubkey")) + signature_failures = _validate_regular_input(signature, label="receipt signature") + for failure in signature_failures: + failures.append( + _failure_record( + "advisory mirror receipt signature is missing or unsafe", + expected=failure.expected, + actual=failure.actual, + repair="place freshness.json.minisig next to freshness.json", + ) + ) + return signature, failures + + +def _has_whitespace_or_control(value: str) -> bool: + return any(char.isspace() for char in value) or CONTROL_RE.search(value) is not None + + +def _terminal_name(locator: str) -> str | None: + parsed = urlparse(locator) + if parsed.scheme: + path = parsed.path + if not path: + return None + return Path(path).name + match = SCP_LOCATOR_RE.fullmatch(locator) + if match is None: + return None + path = match.group("path") + if not path: + return None + return Path(path).name + + +def validate_locator(locator: str) -> list[Failure]: + failures: list[Failure] = [] + if not isinstance(locator, str) or not locator: + return [ + _failure_record( + "advisory mirror locator is empty", + expected="private advisory mirror git locator", + actual="", + repair=INPUT_REPAIR, + ) + ] + if _has_whitespace_or_control(locator): + failures.append( + _failure_record( + "advisory mirror locator contains whitespace or control characters", + expected="single git locator without whitespace or controls", + actual="redacted", + repair=INPUT_REPAIR, + ) + ) + if locator.endswith("/"): + failures.append( + _failure_record( + "advisory mirror locator has a trailing slash", + expected="locator without trailing slash", + actual="trailing slash", + repair=INPUT_REPAIR, + ) + ) + if "?" in locator or "#" in locator: + failures.append( + _failure_record( + "advisory mirror locator contains query or fragment", + expected="locator without query or fragment", + actual="query or fragment", + repair=INPUT_REPAIR, + ) + ) + if failures: + return failures + + source_failures = _validate_source(ADVISORY_COHORT_ID, (locator,)) + if source_failures: + return [ + _failure_record( + failure.error, + expected=failure.expected, + actual=failure.actual, + repair=INPUT_REPAIR, + ) + for failure in source_failures + ] + + terminal = _terminal_name(locator) + if terminal not in ADVISORY_LOCATOR_TERMINALS: + failures.append( + _failure_record( + "advisory mirror locator terminal name is not allowed", + expected=", ".join(sorted(ADVISORY_LOCATOR_TERMINALS)), + actual=terminal or "", + repair=INPUT_REPAIR, + ) + ) + return failures + + +def _canonical_receipt_bytes(*, synced_commit: str, utc: str) -> bytes: + return ( + f'{{"max_age":86400,"synced_commit":"{synced_commit}","utc":"{utc}"}}\n' + ).encode("utf-8") + + +def _trusted_comment(*, synced_commit: str, utc: str) -> str: + return ( + f"{TRUSTED_COMMENT_SCHEME} synced_commit={synced_commit} " + f"utc={utc} max_age=86400" + ) + + +def _read_receipt_authority(receipt: Path) -> ReceiptAuthority: + raw = receipt.read_bytes() + try: + text = raw.decode("utf-8") + payload = json.loads(text) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + _raise_one( + "advisory mirror receipt is not valid canonical JSON", + expected="canonical UTF-8 JSON object", + actual=type(exc).__name__, + ) + if not isinstance(payload, dict): + _raise_one( + "advisory mirror receipt is not a JSON object", + expected="receipt JSON object", + actual=type(payload).__name__, + ) + if set(payload) != {"max_age", "synced_commit", "utc"}: + _raise_one( + "advisory mirror receipt key set is invalid", + expected="max_age, synced_commit, utc", + actual=", ".join(sorted(str(key) for key in payload)) or "", + ) + max_age = payload.get("max_age") + synced_commit = payload.get("synced_commit") + utc = payload.get("utc") + if type(max_age) is not int or max_age != RECEIPT_MAX_AGE: + _raise_one( + "advisory mirror receipt max_age is invalid", + expected=str(RECEIPT_MAX_AGE), + actual=repr(max_age), + ) + if ( + not isinstance(synced_commit, str) + or GIT_COMMIT_RE.fullmatch(synced_commit) is None + ): + _raise_one( + "advisory mirror receipt synced_commit is invalid", + expected="40 lowercase hexadecimal git commit", + actual="redacted", + ) + if not isinstance(utc, str) or not is_normalized_utc_timestamp(utc): + _raise_one( + "advisory mirror receipt utc is invalid", + expected="RFC3339 UTC timestamp normalized with Z", + actual="redacted", + ) + canonical = _canonical_receipt_bytes(synced_commit=synced_commit, utc=utc) + if canonical != raw: + _raise_one( + "advisory mirror receipt bytes are not canonical", + expected='{"max_age":86400,"synced_commit":"<40hex>","utc":""}\\n', + actual="non-canonical JSON", + ) + return ReceiptAuthority( + synced_commit=synced_commit, + utc=utc, + max_age=max_age, + canonical_bytes=canonical, + trusted_comment=_trusted_comment(synced_commit=synced_commit, utc=utc), + ) + + +def _validate_receipt_freshness(receipt: ReceiptAuthority, *, clock: Clock) -> datetime: + now = clock() + receipt_time = _parse_utc(receipt.utc, label="advisory mirror receipt utc") + if receipt_time - now > MAX_CLOCK_SKEW: + _raise_one( + "advisory mirror receipt utc is in the future", + expected="receipt utc no more than 5 minutes in the future", + actual=receipt.utc, + repair="check the system clock, then reacquire the signed advisory mirror packet", + ) + if now - receipt_time > timedelta(seconds=receipt.max_age): + _raise_one( + "advisory mirror receipt is stale", + expected="receipt utc within max_age", + actual=receipt.utc, + repair=AUDIT_REPAIR, + ) + return now + + +def _validate_pubkey_binding( + pubkey: Path, + *, + pinned_key_id: str, + pinned_pubkey_sha256: str, +) -> None: + if KEY_ID_RE.fullmatch(pinned_key_id) is None: + _raise_one( + "advisory mirror pinned key ID is invalid", + expected="16 uppercase hexadecimal characters", + actual="redacted", + ) + if SHA256_RE.fullmatch(pinned_pubkey_sha256) is None: + _raise_one( + "advisory mirror pinned public key SHA-256 is invalid", + expected="64 lowercase hexadecimal characters", + actual="redacted", + ) + raw = pubkey.read_bytes() + digest = hashlib.sha256(raw).hexdigest() + if digest != pinned_pubkey_sha256: + _raise_one( + "advisory mirror public key SHA-256 does not match the pin", + expected="pinned public key SHA-256", + actual="sha256 mismatch", + ) + lines = raw.decode("utf-8", errors="replace").splitlines() + if len(lines) < 2: + _raise_one( + "advisory mirror public key is malformed", + expected="minisign public key with base64 body on line 2", + actual=f"{len(lines)} lines", + ) + try: + decoded = base64.b64decode(lines[1], validate=True) + except ValueError: + _raise_one( + "advisory mirror public key body is not valid base64", + expected="base64 minisign public key body", + actual="invalid base64", + ) + if len(decoded) != 42 or decoded[:2] != b"Ed": + _raise_one( + "advisory mirror public key body is not an Ed25519 minisign key", + expected="42-byte minisign Ed public key blob", + actual="invalid public key blob", + ) + key_id = decoded[2:10][::-1].hex().upper() + if key_id != pinned_key_id: + _raise_one( + "advisory mirror public key ID does not match the pin", + expected="pinned minisign key ID", + actual="key ID mismatch", + ) + + +def _verify_signature( + *, + verifier: TransparencySigner, + receipt: Path, + signature: Path, + trusted_comment: str, + secrets: set[str], +) -> None: + try: + verifier.check() + verifier.verify_file( + receipt, + signature, + expected_trusted_comment=trusted_comment, + ) + except DriverError as exc: + raise _convert_driver_error(exc, secrets=secrets) from exc + + +def audit_config_bytes( + base_bytes: bytes, + *, + db_root: Path, + db_urls: Sequence[str], +) -> bytes: + try: + base_text = base_bytes.decode("utf-8") + except UnicodeDecodeError as exc: + raise ReleasePolicyError( + [ + _failure_record( + "core deny.toml is not UTF-8", + expected="UTF-8 TOML", + actual=str(exc), + repair="python3 scripts/check_rust_release_manifest.py", + ) + ] + ) from exc + if ADVISORY_TABLE_RE.search(base_text): + raise ReleasePolicyError( + [ + _failure_record( + "core deny.toml already defines advisories", + expected="core/deny.toml without [advisories]", + actual="[advisories] present", + repair="python3 scripts/check_rust_release_manifest.py", + ) + ] + ) + prefix = base_bytes if base_bytes.endswith(b"\n") else base_bytes + b"\n" + urls = ", ".join(_toml_string(url) for url in db_urls) + block = ( + f"\n[advisories]\ndb-path = {_toml_string(str(db_root))}\ndb-urls = [{urls}]\n" + ) + return prefix + block.encode("utf-8") + + +def _write_audit_config( + root: Path, + temp_root: Path, + *, + db_root: Path, + db_urls: Sequence[str], +) -> Path: + materialized = audit_config_bytes( + (root / "core" / "deny.toml").read_bytes(), + db_root=db_root, + db_urls=db_urls, + ) + temp_root.mkdir(parents=True, exist_ok=True) + path = temp_root / "deny.audit-advisories.toml" + path.write_bytes(materialized) + return path + + +def _run( + runner: Runner, + argv: Sequence[str], + *, + cwd: Path | None = None, + env: Mapping[str, str] | None = None, +) -> subprocess.CompletedProcess[str]: + kwargs: dict[str, object] = { + "capture_output": True, + "text": True, + "check": False, + } + if cwd is not None: + kwargs["cwd"] = cwd + if env is not None: + kwargs["env"] = dict(env) + return runner(list(argv), **kwargs) + + +def _cargo_env() -> dict[str, str]: + env = dict(os.environ) + env["CARGO_NET_OFFLINE"] = "true" + return env + + +def _looks_like_remote(value: str) -> bool: + parsed = urlparse(value) + return bool(parsed.scheme and parsed.scheme != "file") + + +def _assert_local_git_argv(argv: Sequence[str], *, bundle: Path) -> None: + command = list(argv) + forbidden = {"fetch", "pull", "ls-remote", "remote"} + if any(part in forbidden for part in command[1:]): + _raise_one( + "advisory mirror attempted a remote git operation", + expected="local git bundle operations only", + actual="forbidden git subcommand", + ) + if len(command) >= 2 and command[1] == "clone": + source = command[2] if len(command) > 2 else "" + if source != str(bundle) or _looks_like_remote(source): + _raise_one( + "advisory mirror attempted to clone a non-bundle source", + expected="git clone from the local advisory bundle", + actual="non-local clone source", + ) + + +def _run_git_checked( + runner: Runner, + argv: Sequence[str], + *, + bundle: Path, + error: str, + secrets: set[str], + cwd: Path | None = None, +) -> subprocess.CompletedProcess[str]: + _assert_local_git_argv(argv, bundle=bundle) + result = _run(runner, argv, cwd=cwd) + if result.returncode != 0: + _raise_one( + error, + expected="git exit 0", + actual=f"exit {result.returncode}", + ) + return result + + +def _run_cargo_deny_checked( + runner: Runner, + argv: Sequence[str], + *, + cwd: Path, + env: Mapping[str, str], + secrets: set[str], +) -> subprocess.CompletedProcess[str]: + result = _run(runner, argv, cwd=cwd, env=env) + if result.returncode != 0: + actual = ( + result.stderr.strip() + or result.stdout.strip() + or f"exit {result.returncode}" + ) + _raise_one( + "advisory mirror cargo-deny final check failed", + expected="cargo-deny advisory check exit 0", + actual=_redact_child_output(actual, secrets=secrets), + ) + return result + + +def _run_cargo_deny_unchecked( + runner: Runner, + argv: Sequence[str], + *, + cwd: Path, + env: Mapping[str, str], +) -> subprocess.CompletedProcess[str]: + return _run(runner, argv, cwd=cwd, env=env) + + +def _parse_bundle_heads(stdout: str, *, synced_commit: str) -> None: + lines = stdout.splitlines() + if len(lines) != 2: + _raise_one( + "advisory mirror bundle head set is invalid", + expected="exactly HEAD and refs/heads/main", + actual=f"{len(lines)} refs", + ) + observed: set[tuple[str, str]] = set() + for line in lines: + parts = line.split() + if len(parts) != 2: + _raise_one( + "advisory mirror bundle head line is malformed", + expected=" ", + actual="malformed line", + ) + commit, ref = parts + if GIT_COMMIT_RE.fullmatch(commit) is None: + _raise_one( + "advisory mirror bundle commit is invalid", + expected="40 lowercase hexadecimal git commit", + actual="redacted", + ) + observed.add((commit, ref)) + expected = { + (synced_commit, "HEAD"), + (synced_commit, "refs/heads/main"), + } + if observed != expected: + _raise_one( + "advisory mirror bundle head set does not match the signed receipt", + expected="HEAD and refs/heads/main at synced_commit", + actual="head set mismatch", + ) + + +def _assert_direct_child(path: Path, parent: Path) -> None: + if path.name in {"", ".", ".."}: + _raise_one( + "advisory mirror derived database path is invalid", + expected="safe direct child basename", + actual="invalid basename", + ) + if path.resolve(strict=False).parent != parent.resolve(strict=False): + _raise_one( + "advisory mirror derived database path escapes the temp parent", + expected="cargo-deny database path under temp parent", + actual="redacted", + ) + + +def _assert_final_scanned_snapshot(stderr: str, snapshot: Path) -> None: + try: + _assert_scanned_snapshot(stderr, snapshot) + except ReleasePolicyError as exc: + failures = [ + _failure_record( + failure.error, + expected="materialized advisory snapshot", + actual="redacted", + repair=AUDIT_REPAIR, + ) + for failure in exc.failures + ] + raise ReleasePolicyError(failures) from exc + + +def _cargo_deny_version( + cargo_deny: str, + *, + runner: Runner, + secrets: set[str], +) -> str: + result = _run(runner, [cargo_deny, "--version"]) + actual = ( + result.stdout.strip() or result.stderr.strip() or f"exit {result.returncode}" + ) + parts = actual.split() + if ( + result.returncode != 0 + or len(parts) < 2 + or parts[0] != "cargo-deny" + or parts[1] != CARGO_DENY_VERSION + ): + _raise_one( + "advisory mirror cargo-deny version is not pinned", + expected=CARGO_DENY_VERSION, + actual=_redact_child_output(actual, secrets=secrets), + repair=f"cargo install cargo-deny@{CARGO_DENY_VERSION} --locked --force", + ) + return parts[1] + + +def _cargo_lock_sha256(root: Path) -> str: + try: + return hashlib.sha256((root / "core" / "Cargo.lock").read_bytes()).hexdigest() + except OSError as exc: + raise ReleasePolicyError( + [ + _failure_record( + "advisory mirror cargo lock could not be read", + expected="core/Cargo.lock readable", + actual=type(exc).__name__, + repair="restore core/Cargo.lock and rerun make audit", + ) + ] + ) from exc + + +def _success_bytes( + *, + receipt: ReceiptAuthority, + checked_at: datetime, + cargo_lock_sha256: str, + cargo_deny_version: str, +) -> bytes: + payload = { + "product": PRODUCT, + "advisory_cohort": ADVISORY_COHORT_ID, + "synced_commit": receipt.synced_commit, + "receipt_utc": receipt.utc, + "max_age": receipt.max_age, + "checked_at": _format_utc(checked_at), + "cargo_lock_sha256": cargo_lock_sha256, + "cargo_deny_version": cargo_deny_version, + "verdict": "pass", + } + return json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n" + + +def _remove_tree(path: Path) -> None: + shutil.rmtree(path) + + +def _materialize_and_check( + root: Path, + *, + bundle: Path, + receipt: ReceiptAuthority, + cargo_deny: str, + runner: Runner, + temp_root: Path, + config_path: Path, + secrets: set[str], +) -> None: + db_parent = temp_root / "db-root" + db_parent.mkdir(parents=True, exist_ok=True) + throwaway = db_parent / "bundle-clone" + secrets.update({str(temp_root), str(db_parent), str(throwaway)}) + + _run_git_checked( + runner, + ["git", "bundle", "verify", str(bundle)], + bundle=bundle, + error="advisory mirror bundle verification failed", + secrets=secrets, + ) + heads = _run_git_checked( + runner, + ["git", "bundle", "list-heads", str(bundle)], + bundle=bundle, + error="advisory mirror bundle list-heads failed", + secrets=secrets, + ) + _parse_bundle_heads(heads.stdout.strip(), synced_commit=receipt.synced_commit) + + _run_git_checked( + runner, + ["git", "clone", str(bundle), str(throwaway)], + bundle=bundle, + error="advisory mirror bundle clone failed", + secrets=secrets, + ) + head = _run_git_checked( + runner, + ["git", "-C", str(throwaway), "rev-parse", "HEAD"], + bundle=bundle, + error="advisory mirror clone HEAD read failed", + secrets=secrets, + ) + if head.stdout.strip() != receipt.synced_commit: + _raise_one( + "advisory mirror clone HEAD does not match the signed receipt", + expected="cloned HEAD equals synced_commit", + actual="clone HEAD mismatch", + ) + _validate_advisory_count(_count_advisories(throwaway)) + + cargo_env = _cargo_env() + argv = advisory_check_argv(cargo_deny, config_path, root) + discovery = _run_cargo_deny_unchecked(runner, argv, cwd=root, env=cargo_env) + derived = _scanned_advisory_db(discovery.stderr) + _assert_direct_child(derived, db_parent) + if derived.exists(): + _raise_one( + "advisory mirror derived database path already exists", + expected="cargo-deny derived database path absent before rename", + actual="preexisting derived path", + ) + secrets.add(str(derived)) + throwaway.rename(derived) + + final = _run_cargo_deny_checked( + runner, + argv, + cwd=root, + env=cargo_env, + secrets=secrets, + ) + _assert_final_scanned_snapshot(final.stderr, derived) + + +def audit_advisory_mirror( + root: Path, + *, + bundle: Path, + receipt: Path, + pubkey: Path, + locator: str, + cargo_deny: str = "cargo-deny", + runner: Runner = subprocess.run, + verifier: TransparencySigner | None = None, + minisign: str = "minisign", + clock: Clock = _utc_now, + temp_path_factory: TempPathFactory = _default_temp_path_factory, + cleanup_unlink: PathRemover = _unlink_path, + cleanup_rmdir: PathRemover = _remove_tree, + pinned_key_id: str = PINNED_KEY_ID, + pinned_pubkey_sha256: str = PINNED_PUBKEY_SHA256, +) -> bytes: + signature, input_failures = _validate_inputs( + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + ) + if input_failures: + raise ReleasePolicyError(input_failures) + + locator_failures = validate_locator(locator) + if locator_failures: + raise ReleasePolicyError(locator_failures) + + secrets = { + locator, + str(bundle), + str(receipt), + str(signature), + str(pubkey), + } + _validate_pubkey_binding( + pubkey, + pinned_key_id=pinned_key_id, + pinned_pubkey_sha256=pinned_pubkey_sha256, + ) + receipt_authority = _read_receipt_authority(receipt) + active_verifier = verifier or PublicKeyMinisignVerifier(pubkey, minisign=minisign) + _verify_signature( + verifier=active_verifier, + receipt=receipt, + signature=signature, + trusted_comment=receipt_authority.trusted_comment, + secrets=secrets, + ) + checked_at = _validate_receipt_freshness(receipt_authority, clock=clock) + + cargo_deny_observed = _cargo_deny_version( + cargo_deny, + runner=runner, + secrets=secrets, + ) + + temp_root = temp_path_factory("advisory-mirror-audit") + config_path: Path | None = None + result: bytes | None = None + primary_error: ReleasePolicyError | None = None + try: + db_parent = temp_root / "db-root" + config_path = _write_audit_config( + root, + temp_root, + db_root=db_parent, + db_urls=(locator,), + ) + secrets.update({str(temp_root), str(config_path)}) + _materialize_and_check( + root, + bundle=bundle, + receipt=receipt_authority, + cargo_deny=cargo_deny, + runner=runner, + temp_root=temp_root, + config_path=config_path, + secrets=secrets, + ) + result = _success_bytes( + receipt=receipt_authority, + checked_at=checked_at, + cargo_lock_sha256=_cargo_lock_sha256(root), + cargo_deny_version=cargo_deny_observed, + ) + except ReleasePolicyError as exc: + primary_error = exc + finally: + cleanup_error: ReleasePolicyError | None = None + try: + _cleanup_temp( + temp_root, + config_path, + unlink_path=cleanup_unlink, + remove_dir=cleanup_rmdir, + ) + except ReleasePolicyError as exc: + cleanup_error = exc + combined = _combined_release_policy_error(primary_error, cleanup_error) + if combined is not None: + raise combined + if result is None: + raise AssertionError("advisory mirror audit did not produce a result") + return result + + +def _raw_arg_failures(args: argparse.Namespace) -> list[Failure]: + failures: list[Failure] = [] + for name in ("bundle", "receipt", "pubkey", "locator"): + value = getattr(args, name) + if value == "": + failures.append( + _failure_record( + f"advisory mirror input {name} is empty", + expected=f"non-empty {name}", + actual="", + repair=INPUT_REPAIR, + ) + ) + return failures + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--bundle", required=True) + parser.add_argument("--receipt", required=True) + parser.add_argument("--pubkey", required=True) + parser.add_argument("--locator", required=True) + args = parser.parse_args(list(argv) if argv is not None else None) + + try: + raw_failures = _raw_arg_failures(args) + if raw_failures: + raise ReleasePolicyError(raw_failures) + output = audit_advisory_mirror( + ROOT, + bundle=Path(args.bundle), + receipt=Path(args.receipt), + pubkey=Path(args.pubkey), + locator=args.locator, + ) + except ReleasePolicyError as exc: + _format_failures(exc.failures) + return 1 + sys.stdout.buffer.write(output) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/release_advisory_policy.py b/scripts/release_advisory_policy.py index 2fd230e01..10a9238a8 100644 --- a/scripts/release_advisory_policy.py +++ b/scripts/release_advisory_policy.py @@ -24,7 +24,9 @@ To acquire a conforming snapshot, write a cargo-deny config that sets the same second run is intentional: cargo-deny 0.20.2 does not write ``.git/FETCH_HEAD`` on the first clone into an empty db root, but it does on subsequent fetches. Do not run manual ``git fetch`` or ``git reset``. ``make audit`` is not this acquisition -operation; it uses cargo-deny's default db path, not a controlled release db root. +operation. ``make audit`` is the separate signed-packet mirror-bound advisory +audit implemented by ``scripts/advisory_mirror_audit.py``; this module remains +the release-candidate advisory acquisition and receipt path. """ from __future__ import annotations diff --git a/tests/test_advisory_mirror_audit.py b/tests/test_advisory_mirror_audit.py new file mode 100644 index 000000000..72a951802 --- /dev/null +++ b/tests/test_advisory_mirror_audit.py @@ -0,0 +1,1237 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import base64 +import hashlib +import json +import subprocess +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +import scripts.advisory_mirror_audit as audit +from scripts.transparency_signing import FakeTransparencySigner + +NOW = datetime(2026, 7, 24, 12, 0, tzinfo=UTC) +RECEIPT_UTC = "2026-07-24T11:30:00Z" +TEST_KEY_ID = "A1B2C3D4E5F60708" +DERIVED_NAME = "rustsec-advisory-db.git-testderived" + + +def _run_git(repo: Path, argv: Sequence[str]) -> str: + result = subprocess.run( + ["git", *argv], + cwd=repo, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr + return result.stdout.strip() + + +def _audit_root(tmp_path: Path) -> Path: + root = tmp_path / "root" + (root / "core").mkdir(parents=True) + (root / "core" / "deny.toml").write_text( + '[licenses]\nallow = ["MIT"]\n', encoding="utf-8" + ) + (root / "core" / "Cargo.lock").write_text("fixture lock\n", encoding="utf-8") + (root / "core" / "Cargo.toml").write_text( + "[workspace]\nmembers = []\n", + encoding="utf-8", + ) + (root / "target" / "release-evidence").mkdir(parents=True) + (root / "target" / "release-evidence" / "preserve.txt").write_text( + "release evidence\n", + encoding="utf-8", + ) + (root / "dist" / "release-candidate").mkdir(parents=True) + (root / "dist" / "release-candidate" / "preserve.txt").write_text( + "release candidate\n", + encoding="utf-8", + ) + return root + + +def _advisory_repo( + tmp_path: Path, *, advisory_count: int = 1 +) -> tuple[Path, str, Path]: + repo = tmp_path / f"repo-{advisory_count}" + repo.mkdir() + _run_git(repo, ["init", "-b", "main"]) + _run_git(repo, ["config", "user.name", "Audit Test"]) + _run_git(repo, ["config", "user.email", "audit-test@example.invalid"]) + if advisory_count: + for index in range(advisory_count): + path = repo / "crates" / f"probe{index}" / f"RUSTSEC-2026-{index:04d}.md" + path.parent.mkdir(parents=True) + path.write_text( + "```toml\n" + "[advisory]\n" + f'id = "RUSTSEC-2026-{index:04d}"\n' + f'package = "probe{index}"\n' + 'date = "2026-01-01"\n' + 'url = "https://example.invalid/RUSTSEC-2026-0001"\n' + 'categories = ["unmaintained"]\n' + "keywords = []\n\n" + "[versions]\n" + "patched = []\n" + "```\n", + encoding="utf-8", + ) + else: + (repo / "README.md").write_text("empty advisory db\n", encoding="utf-8") + _run_git(repo, ["add", "."]) + _run_git(repo, ["commit", "-m", "fixture advisory db"]) + commit = _run_git(repo, ["rev-parse", "HEAD"]) + bundle = tmp_path / f"advisory-{advisory_count}.bundle" + _run_git(repo, ["bundle", "create", str(bundle), "HEAD", "refs/heads/main"]) + return repo, commit, bundle + + +def _pubkey_bytes(key_id: str = TEST_KEY_ID) -> bytes: + raw_id = bytes.fromhex(key_id)[::-1] + blob = b"Ed" + raw_id + (b"\x11" * 32) + return ( + f"untrusted comment: minisign public key {key_id}\n".encode("ascii") + + base64.b64encode(blob) + + b"\n" + ) + + +def _write_pubkey(path: Path, *, key_id: str = TEST_KEY_ID) -> str: + raw = _pubkey_bytes(key_id) + path.write_bytes(raw) + return hashlib.sha256(raw).hexdigest() + + +def _receipt_bytes( + commit: str, utc: str = RECEIPT_UTC, *, max_age: int = 86400 +) -> bytes: + return ( + f'{{"max_age":{max_age},"synced_commit":"{commit}","utc":"{utc}"}}\n' + ).encode("utf-8") + + +def _trusted_comment(commit: str, utc: str = RECEIPT_UTC) -> str: + return ( + f"{audit.TRUSTED_COMMENT_SCHEME} synced_commit={commit} utc={utc} max_age=86400" + ) + + +def _write_packet( + tmp_path: Path, + *, + commit: str, + utc: str = RECEIPT_UTC, + key_id: str = TEST_KEY_ID, + signer: FakeTransparencySigner | None = None, + trusted_comment: str | None = None, +) -> tuple[Path, Path, Path, str, FakeTransparencySigner]: + pubkey = tmp_path / "pub.key" + pubkey_sha = _write_pubkey(pubkey, key_id=key_id) + receipt = tmp_path / "freshness.json" + receipt.write_bytes(_receipt_bytes(commit, utc)) + signature = receipt.parent / f"{receipt.name}.minisig" + fake = signer or FakeTransparencySigner() + fake.sign_file( + receipt, + signature, + trusted_comment=trusted_comment or _trusted_comment(commit, utc), + ) + return receipt, signature, pubkey, pubkey_sha, fake + + +class NoCallRunner: + events: list[list[str]] + + def __init__(self) -> None: + self.events = [] + + def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]: + self.events.append(list(argv)) + raise AssertionError(f"unexpected command: {argv}") + + +class HybridRunner: + def __init__( + self, + *, + derived_name: str = DERIVED_NAME, + discovery_exit: int = 1, + final_exit: int = 0, + final_stderr_extra: str = "", + final_scanned_path: Path | None = None, + version: str = audit.CARGO_DENY_VERSION, + ) -> None: + self.derived_name = derived_name + self.discovery_exit = discovery_exit + self.final_exit = final_exit + self.final_stderr_extra = final_stderr_extra + self.final_scanned_path = final_scanned_path + self.version = version + self.events: list[list[str]] = [] + self.cargo_envs: list[Mapping[str, str]] = [] + self.cargo_cwds: list[Path | None] = [] + self.config_bytes: bytes | None = None + self.check_count = 0 + + def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]: + command = list(argv) + self.events.append(command) + if command[0] == "git": + return subprocess.run(command, **kwargs) + if command[0] == "cargo-deny": + if command == ["cargo-deny", "--version"]: + return subprocess.CompletedProcess( + command, 0, f"cargo-deny {self.version}\n", "" + ) + if command[-2:] == ["check", "advisories"]: + self.check_count += 1 + self.cargo_envs.append(kwargs.get("env", {})) + self.cargo_cwds.append(kwargs.get("cwd")) + config_path = Path(command[command.index("--config") + 1]) + self.config_bytes = config_path.read_bytes() + db_parent = _config_db_path(config_path) + scanned = self.final_scanned_path if self.check_count == 2 else None + if scanned is None: + scanned = db_parent / self.derived_name + stderr = ( + f"2026-07-24 [DEBUG] Opening advisory database at '{scanned}'\n" + ) + if self.check_count == 1: + return subprocess.CompletedProcess( + command, self.discovery_exit, "", stderr + ) + stderr += self.final_stderr_extra + return subprocess.CompletedProcess(command, self.final_exit, "", stderr) + raise AssertionError(f"unexpected cargo-deny command: {command}") + raise AssertionError(f"unexpected command: {command}") + + +class FakeGitRunner: + def __init__( + self, + *, + heads_commit: str, + clone_commit: str | None = None, + fail: str | None = None, + ) -> None: + self.heads_commit = heads_commit + self.clone_commit = clone_commit or heads_commit + self.fail = fail + self.events: list[list[str]] = [] + + def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]: + command = list(argv) + self.events.append(command) + if self.fail and self.fail in " ".join(command): + return subprocess.CompletedProcess(command, 1, "", "failed") + if command == ["cargo-deny", "--version"]: + return subprocess.CompletedProcess(command, 0, "cargo-deny 0.20.2\n", "") + if command[:3] == ["git", "bundle", "verify"]: + return subprocess.CompletedProcess(command, 0, "", "") + if command[:3] == ["git", "bundle", "list-heads"]: + return subprocess.CompletedProcess( + command, + 0, + f"{self.heads_commit} HEAD\n{self.heads_commit} refs/heads/main\n", + "", + ) + if command[:2] == ["git", "clone"]: + Path(command[3]).mkdir(parents=True) + return subprocess.CompletedProcess(command, 0, "", "") + if command[:3] == ["git", "-C", command[2]] and command[-2:] == [ + "rev-parse", + "HEAD", + ]: + return subprocess.CompletedProcess(command, 0, self.clone_commit + "\n", "") + if command[-2:] == ["check", "advisories"]: + config_path = Path(command[command.index("--config") + 1]) + db_parent = _config_db_path(config_path) + return subprocess.CompletedProcess( + command, + 1, + "", + f"Opening advisory database at '{db_parent / DERIVED_NAME}'\n", + ) + raise AssertionError(f"unexpected command: {command}") + + +def _config_db_path(config_path: Path) -> Path: + import tomllib + + parsed = tomllib.loads(config_path.read_text(encoding="utf-8")) + return Path(parsed["advisories"]["db-path"]) + + +def _invoke_green( + tmp_path: Path, + *, + runner: HybridRunner | None = None, + advisory_count: int = 1, + utc: str = RECEIPT_UTC, +) -> tuple[bytes, HybridRunner, Path, Path, Path, Path]: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path, advisory_count=advisory_count) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, + commit=commit, + utc=utc, + ) + active_runner = runner or HybridRunner() + output = audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=active_runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + return output, active_runner, root, bundle, receipt, pubkey + + +def _inventory(root: Path) -> dict[str, tuple[str, int]]: + result: dict[str, tuple[str, int]] = {} + if not root.exists(): + return result + for path in sorted(item for item in root.rglob("*") if item.is_file()): + rel = path.relative_to(root).as_posix() + raw = path.read_bytes() + result[rel] = (hashlib.sha256(raw).hexdigest(), len(raw)) + return result + + +def test_green_packet_emits_exact_success_json_and_uses_bound_snapshot( + tmp_path: Path, +) -> None: + output, runner, root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path) + + payload = json.loads(output) + assert list(payload) == [ + "product", + "advisory_cohort", + "synced_commit", + "receipt_utc", + "max_age", + "checked_at", + "cargo_lock_sha256", + "cargo_deny_version", + "verdict", + ] + assert payload["product"] == "solstone-journal" + assert payload["advisory_cohort"] == audit.ADVISORY_COHORT_ID + assert payload["receipt_utc"] == RECEIPT_UTC + assert payload["max_age"] == 86400 + assert payload["checked_at"] == "2026-07-24T12:00:00Z" + assert payload["cargo_deny_version"] == "0.20.2" + assert ( + payload["cargo_lock_sha256"] + == hashlib.sha256((root / "core" / "Cargo.lock").read_bytes()).hexdigest() + ) + expected = json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n" + assert output == expected + assert output.count(b"\n") == 1 + assert runner.check_count == 2 + assert all(env.get("CARGO_NET_OFFLINE") == "true" for env in runner.cargo_envs) + assert runner.config_bytes is not None + config_text = runner.config_bytes.decode("utf-8") + assert "git-fetch-with-cli" not in config_text + assert "maximum-db-staleness" not in config_text + + +def test_green_packet_with_real_git_bundle_materialization(tmp_path: Path) -> None: + output, runner, _root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path) + + assert json.loads(output)["verdict"] == "pass" + assert any(command[:3] == ["git", "bundle", "verify"] for command in runner.events) + assert any( + command[:3] == ["git", "bundle", "list-heads"] for command in runner.events + ) + assert any(command[:2] == ["git", "clone"] for command in runner.events) + assert runner.check_count == 2 + + +@pytest.mark.parametrize("missing", ["bundle", "receipt", "pubkey", "locator"]) +def test_required_inputs_fail_before_git_or_cargo(tmp_path: Path, missing: str) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + if missing == "bundle": + bundle = tmp_path / "missing.bundle" + elif missing == "receipt": + receipt = tmp_path / "missing.json" + elif missing == "pubkey": + pubkey = tmp_path / "missing.pub" + locator = ( + "" + if missing == "locator" + else "ssh://mirror.example.invalid/rustsec-advisory-db.git" + ) + runner = NoCallRunner() + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator=locator, + runner=runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert runner.events == [] + + +@pytest.mark.parametrize("kind", ["missing", "symlink", "directory"]) +def test_adjacent_signature_is_required_and_regular_before_git_or_cargo( + tmp_path: Path, + kind: str, +) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + signature.unlink() + if kind == "symlink": + target = tmp_path / "sig-target" + target.write_text("signature", encoding="utf-8") + signature.symlink_to(target) + elif kind == "directory": + signature.mkdir() + runner = NoCallRunner() + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert runner.events == [] + + +@pytest.mark.parametrize("target_name", ["bundle", "receipt", "pubkey"]) +@pytest.mark.parametrize("kind", ["missing", "symlink", "directory"]) +def test_unsafe_input_paths_and_symlinks_fail_before_git_or_cargo( + tmp_path: Path, + target_name: str, + kind: str, +) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + paths = {"bundle": bundle, "receipt": receipt, "pubkey": pubkey} + target = paths[target_name] + if target.exists() or target.is_symlink(): + target.unlink() + if kind == "symlink": + link_target = tmp_path / f"{target_name}-target" + link_target.write_text("target", encoding="utf-8") + target.symlink_to(link_target) + elif kind == "directory": + target.mkdir() + runner = NoCallRunner() + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=paths["bundle"], + receipt=paths["receipt"], + pubkey=paths["pubkey"], + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert runner.events == [] + + +@pytest.mark.parametrize( + ("locator", "accepted"), + [ + ("https://github.com/rustsec/advisory-db", False), + ("https://github.com/rustsec/advisory-db.git", False), + ("https://github.com/rustsec/advisory-db/", False), + ("https://github.com/rustsec/advisory-db.git/", False), + ("github.com:rustsec/advisory-db.git", False), + ("github.com:rustsec/advisory-db.git/", False), + ("ssh://github.com/rustsec/advisory-db.git", False), + ("ssh://github.com:22/rustsec/advisory-db.git", False), + ("git://github.com/rustsec/advisory-db.git", False), + ("http://github.com/rustsec/advisory-db", False), + ("git+ssh://github.com/rustsec/advisory-db.git", False), + ("https://raw.github.com/rustsec/advisory-db", False), + ("https://mirror.github.com/rustsec/advisory-db", False), + ("https://foo.github.com/rustsec/advisory-db.git", False), + ("https://github.com/rustsec/advisory-db?x=1", False), + ("https://github.com/rustsec/advisory-db#frag", False), + ("github.com/rustsec/advisory-db.git", False), + ("", False), + (" ", False), + ("https://mirror.example.invalid/rustsec/advisory-db", True), + ("https://mirror.example.invalid/rustsec/advisory-db.git", False), + ("https://mirror.example.invalid/rustsec/rustsec-advisory-db.git", True), + ("https://mirror.example.invalid/rustsec/rustsec-advisory-db", False), + ("https://mirror.example.invalid/rustsec/advisory-db/", False), + ("https://mirror.example.invalid/rustsec/rustsec-advisory-db.git/", False), + ("https://mirror.example.invalid/rustsec/advisory-db?x=1", False), + ("https://mirror.example.invalid/rustsec/advisory-db#frag", False), + ("ssh://git@mirror.example.invalid/rustsec/advisory-db", True), + ("git@mirror.example.invalid:rustsec/advisory-db", True), + ("git@mirror.example.invalid:rustsec/rustsec-advisory-db.git", True), + ("git@mirror.example.invalid:rustsec/advisory-db.git", False), + ("https://mirror.example.invalid/rustsec/advisory-db\n", False), + ("https://mirror.example.invalid/rustsec/advisory-db\t", False), + ("https://mirror.example.invalid/rustsec/advisory-db\x00", False), + ("https://github.com.evil/rustsec/advisory-db", True), + ], +) +def test_validate_locator_q3_oracle(locator: str, accepted: bool) -> None: + assert (audit.validate_locator(locator) == []) is accepted + + +@pytest.mark.parametrize( + "raw", + [ + b'{"synced_commit":"{commit}","max_age":86400,"utc":"2026-07-24T11:30:00Z"}\n', + b'{ "max_age": 86400, "synced_commit": "{commit}", "utc": "2026-07-24T11:30:00Z" }\n', + b'{"max_age":86400,"synced_commit":"{commit}","utc":"2026-07-24T11:30:00Z"}', + b'{"max_age":86400,"synced_commit":"{commit}","utc":"2026-07-24T11:30:00Z"}\n\n', + b'{"max_age":86400,"synced_commit":"{commit}","utc":"2026-07-24T11:30:00Z","x":1}\n', + ], +) +def test_receipt_body_requires_canonical_bytes(tmp_path: Path, raw: bytes) -> None: + _repo, commit, _bundle = _advisory_repo(tmp_path) + receipt = tmp_path / "freshness.json" + receipt.write_bytes(raw.replace(b"{commit}", commit.encode("ascii"))) + + with pytest.raises(audit.ReleasePolicyError): + audit._read_receipt_authority(receipt) + + +@pytest.mark.parametrize( + "payload", + [ + {"max_age": 1, "synced_commit": "a" * 40, "utc": RECEIPT_UTC}, + {"max_age": True, "synced_commit": "a" * 40, "utc": RECEIPT_UTC}, + {"max_age": "86400", "synced_commit": "a" * 40, "utc": RECEIPT_UTC}, + {"max_age": 86400, "synced_commit": "a" * 64, "utc": RECEIPT_UTC}, + {"max_age": 86400, "synced_commit": "A" * 40, "utc": RECEIPT_UTC}, + {"max_age": 86400, "synced_commit": "a" * 40, "utc": "2026-99-99T00:00:00Z"}, + { + "max_age": 86400, + "synced_commit": "a" * 40, + "utc": "2026-07-24T11:30:00+00:00", + }, + ], +) +def test_receipt_fields_are_strict(tmp_path: Path, payload: dict[str, Any]) -> None: + receipt = tmp_path / "freshness.json" + receipt.write_text( + json.dumps(payload, separators=(",", ":")) + "\n", encoding="utf-8" + ) + + with pytest.raises(audit.ReleasePolicyError): + audit._read_receipt_authority(receipt) + + +def test_trusted_comment_mismatch_fails(tmp_path: Path) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, + commit=commit, + trusted_comment="wrong", + ) + + with pytest.raises(audit.ReleasePolicyError) as exc: + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=NoCallRunner(), + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert "trusted comment mismatch" in exc.value.failures[0].error + + +def test_pubkey_sha256_mismatch_fails_before_minisign(tmp_path: Path) -> None: + _repo, commit, _bundle = _advisory_repo(tmp_path) + _receipt, _signature, pubkey, _pubkey_sha, _fake = _write_packet( + tmp_path, commit=commit + ) + + with pytest.raises(audit.ReleasePolicyError): + audit._validate_pubkey_binding( + pubkey, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256="0" * 64, + ) + + +def test_pubkey_key_id_mismatch_fails_before_minisign(tmp_path: Path) -> None: + pubkey = tmp_path / "pub.key" + pubkey_sha = _write_pubkey(pubkey) + + with pytest.raises(audit.ReleasePolicyError): + audit._validate_pubkey_binding( + pubkey, + pinned_key_id="0000000000000000", + pinned_pubkey_sha256=pubkey_sha, + ) + + +@pytest.mark.parametrize( + "raw", + [ + b"untrusted comment\nnot base64\n", + b"untrusted comment\nRWQ=\n", + b"untrusted comment\n" + base64.b64encode(b"XX" + b"\x00" * 40) + b"\n", + b"only one line\n", + ], +) +def test_pubkey_blob_shape_is_strict(tmp_path: Path, raw: bytes) -> None: + pubkey = tmp_path / "pub.key" + pubkey.write_bytes(raw) + pubkey_sha = hashlib.sha256(raw).hexdigest() + + with pytest.raises(audit.ReleasePolicyError): + audit._validate_pubkey_binding( + pubkey, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + + +def test_signature_mutation_fails(tmp_path: Path) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + signature.write_text( + signature.read_text(encoding="utf-8").replace("=", "A", 1), encoding="utf-8" + ) + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=NoCallRunner(), + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + + +@pytest.mark.parametrize( + "utc", + ["2026-07-24T12:06:00Z", "2026-07-23T11:59:59Z"], +) +def test_receipt_future_and_stale_times_fail(tmp_path: Path, utc: str) -> None: + _repo, commit, _bundle = _advisory_repo(tmp_path) + receipt = tmp_path / "freshness.json" + receipt.write_bytes(_receipt_bytes(commit, utc)) + authority = audit._read_receipt_authority(receipt) + + with pytest.raises(audit.ReleasePolicyError): + audit._validate_receipt_freshness(authority, clock=lambda: NOW) + + +def test_minisign_preflight_uses_product_binary_check( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[str, object]] = [] + + def check(minisign: str) -> str: + calls.append(("check", minisign)) + return "minisign 0.12" + + class Local: + def __init__( + self, *, secret_key: Path, public_key: Path, minisign: str + ) -> None: + calls.append(("init", (secret_key, public_key, minisign))) + + def verify_file( + self, + message_path: Path, + signature_path: Path, + *, + expected_trusted_comment: str, + ) -> None: + calls.append( + ("verify", (message_path, signature_path, expected_trusted_comment)) + ) + + monkeypatch.setattr(audit, "check_minisign_binary", check) + monkeypatch.setattr(audit, "LocalMinisignSigner", Local) + verifier = audit.PublicKeyMinisignVerifier( + tmp_path / "pub.key", minisign="minisign-test" + ) + verifier.check() + verifier.verify_file( + tmp_path / "freshness.json", + tmp_path / "freshness.json.minisig", + expected_trusted_comment="comment", + ) + + assert calls[0] == ("check", "minisign-test") + assert calls[1][0] == "init" + assert calls[2][0] == "verify" + + +def test_bundle_verify_failure_stops_before_clone_and_cargo(tmp_path: Path) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + runner = FakeGitRunner(heads_commit=commit, fail="bundle verify") + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert not any(command[:2] == ["git", "clone"] for command in runner.events) + assert not any(command[-2:] == ["check", "advisories"] for command in runner.events) + + +@pytest.mark.parametrize( + "stdout", + [ + "a" * 40 + " HEAD\n", + "a" * 40 + " refs/heads/main\n", + "a" * 40 + " HEAD\n" + "a" * 40 + " refs/heads/main\n" + "a" * 40 + " refs/x\n", + "b" * 40 + " HEAD\n" + "a" * 40 + " refs/heads/main\n", + "malformed\n" + "a" * 40 + " refs/heads/main\n", + ], +) +def test_bundle_heads_must_be_exact_head_and_main(stdout: str) -> None: + with pytest.raises(audit.ReleasePolicyError): + audit._parse_bundle_heads(stdout, synced_commit="a" * 40) + + +def test_clone_head_must_match_receipt_commit(tmp_path: Path) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + runner = FakeGitRunner(heads_commit=commit, clone_commit="b" * 40) + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + + +def test_zero_advisory_clone_fails_before_cargo(tmp_path: Path) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path, advisory_count=0) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + runner = HybridRunner() + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=runner, + verifier=fake, + clock=lambda: NOW, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert runner.check_count == 0 + + +def test_discovery_run_nonzero_is_expected_when_debug_line_present( + tmp_path: Path, +) -> None: + output, runner, _root, _bundle, _receipt, _pubkey = _invoke_green( + tmp_path, + runner=HybridRunner(discovery_exit=23), + ) + + assert json.loads(output)["verdict"] == "pass" + assert runner.check_count == 2 + + +@pytest.mark.parametrize("scanned", [Path("/outside/db"), Path("nested/child")]) +def test_discovery_path_must_be_direct_child_of_temp_parent( + tmp_path: Path, + scanned: Path, +) -> None: + parent = tmp_path / "parent" + parent.mkdir() + if not scanned.is_absolute(): + scanned = parent / scanned + + with pytest.raises(audit.ReleasePolicyError): + audit._assert_direct_child(scanned, parent) + + +def test_discovered_path_must_not_preexist(tmp_path: Path) -> None: + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + temp_root = tmp_path / "temp" + preexisting = temp_root / "db-root" / DERIVED_NAME + preexisting.mkdir(parents=True) + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=HybridRunner(), + verifier=fake, + clock=lambda: NOW, + temp_path_factory=lambda _label: temp_root, + cleanup_rmdir=lambda path: None, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + + +def test_alternate_or_ambient_database_substitution_is_rejected(tmp_path: Path) -> None: + runner = HybridRunner(final_scanned_path=tmp_path / "other-db") + + with pytest.raises(audit.ReleasePolicyError) as exc: + _invoke_green(tmp_path, runner=runner) + assert exc.value.failures[0].actual == "redacted" + + +def test_runner_never_sees_remote_git_or_cargo_fetch_operations(tmp_path: Path) -> None: + _output, runner, _root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path) + flattened = [" ".join(command) for command in runner.events] + + assert not any("fetch db" in item for item in flattened) + assert not any(item.startswith("git fetch") for item in flattened) + assert not any(item.startswith("git pull") for item in flattened) + assert not any(item.startswith("git ls-remote") for item in flattened) + assert not any("github.com" in item for item in flattened) + + +def test_final_cargo_deny_failure_is_redacted_and_no_success(tmp_path: Path) -> None: + runner = HybridRunner( + final_exit=1, + final_stderr_extra=( + "/private/path TOKEN=abc ghp_abcdefghijklmnopqrst localhost " + "ssh://mirror.example.invalid/rustsec-advisory-db.git" + ), + ) + + with pytest.raises(audit.ReleasePolicyError) as exc: + _invoke_green(tmp_path, runner=runner) + text = "\n".join(failure.actual for failure in exc.value.failures) + assert "mirror.example.invalid" not in text + assert "/private/path" not in text + assert "TOKEN=" not in text + assert "ghp_" not in text + + +def test_child_output_redaction_masks_locator_temp_path_and_token_canaries() -> None: + redacted = audit._redact_child_output( + "secret /tmp/private TOKEN=value ghp_abcdefghijklmnopqrst host.local", + secrets={"/tmp/private"}, + ) + + assert "/tmp/private" not in redacted + assert "TOKEN=" not in redacted + assert "ghp_" not in redacted + assert audit.validate_public_evidence_text("child-output", redacted) == [] + + +def test_cleanup_failure_suppresses_success_and_combines_errors(tmp_path: Path) -> None: + def fail_cleanup(_path: Path) -> None: + raise OSError("cleanup failed") + + root = _audit_root(tmp_path / "cleanup") + _repo, commit, bundle = _advisory_repo(tmp_path / "cleanup") + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path / "cleanup", commit=commit + ) + with pytest.raises(audit.ReleasePolicyError) as cleanup_exc: + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=HybridRunner(), + verifier=fake, + clock=lambda: NOW, + cleanup_rmdir=fail_cleanup, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + assert any( + "cleanup failed" in failure.error for failure in cleanup_exc.value.failures + ) + + +def test_cleanup_failure_combines_with_primary_error(tmp_path: Path) -> None: + def fail_cleanup(_path: Path) -> None: + raise OSError("cleanup failed") + + root = _audit_root(tmp_path) + _repo, commit, bundle = _advisory_repo(tmp_path) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + tmp_path, commit=commit + ) + + with pytest.raises(audit.ReleasePolicyError) as exc: + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=HybridRunner(final_exit=1), + verifier=fake, + clock=lambda: NOW, + cleanup_rmdir=fail_cleanup, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + errors = [failure.error for failure in exc.value.failures] + assert "advisory mirror cargo-deny final check failed" in errors + assert any("cleanup failed" in error for error in errors) + + +def test_exact_success_schema_and_witness_binding(tmp_path: Path) -> None: + output, _runner, root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path) + payload = json.loads(output) + + assert payload == { + "product": "solstone-journal", + "advisory_cohort": audit.ADVISORY_COHORT_ID, + "synced_commit": payload["synced_commit"], + "receipt_utc": RECEIPT_UTC, + "max_age": 86400, + "checked_at": "2026-07-24T12:00:00Z", + "cargo_lock_sha256": hashlib.sha256( + (root / "core" / "Cargo.lock").read_bytes() + ).hexdigest(), + "cargo_deny_version": "0.20.2", + "verdict": "pass", + } + serialized = json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n" + assert output == serialized + assert b"mirror.example.invalid" not in output + + +def test_success_inventory_is_non_destructive( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = _audit_root(tmp_path) + packet_root = tmp_path / "packet" + packet_root.mkdir() + cargo_home = tmp_path / "cargo-home" + cargo_home.mkdir() + (cargo_home / "preserve.txt").write_text("ambient cargo\n", encoding="utf-8") + monkeypatch.setenv("CARGO_HOME", str(cargo_home)) + temp_root = tmp_path / "audit-temp" + _repo, commit, bundle = _advisory_repo(packet_root) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + packet_root, commit=commit + ) + before_root = _inventory(root) + before_packet = _inventory(packet_root) + before_cargo = _inventory(cargo_home) + + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator="ssh://mirror.example.invalid/rustsec-advisory-db.git", + runner=HybridRunner(), + verifier=fake, + clock=lambda: NOW, + temp_path_factory=lambda _label: temp_root, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pubkey_sha, + ) + + assert _inventory(root) == before_root + assert _inventory(packet_root) == before_packet + assert _inventory(cargo_home) == before_cargo + assert not temp_root.exists() + + +@pytest.mark.parametrize( + "stage", + ["input", "locator", "pubkey", "signature", "time", "bundle", "cargo", "cleanup"], +) +def test_failure_inventory_is_non_destructive( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + stage: str, +) -> None: + root = _audit_root(tmp_path) + packet_root = tmp_path / "packet" + packet_root.mkdir() + cargo_home = tmp_path / "cargo-home" + cargo_home.mkdir() + (cargo_home / "preserve.txt").write_text("ambient cargo\n", encoding="utf-8") + monkeypatch.setenv("CARGO_HOME", str(cargo_home)) + temp_root = tmp_path / "audit-temp" + _repo, commit, bundle = _advisory_repo(packet_root) + receipt, _signature, pubkey, pubkey_sha, fake = _write_packet( + packet_root, commit=commit + ) + locator = "ssh://mirror.example.invalid/rustsec-advisory-db.git" + runner: object = HybridRunner(final_exit=1) + pinned_pubkey_sha = pubkey_sha + cleanup_rmdir = audit._remove_tree + if stage == "input": + bundle = packet_root / "missing.bundle" + runner = NoCallRunner() + elif stage == "locator": + locator = "" + runner = NoCallRunner() + elif stage == "pubkey": + pinned_pubkey_sha = "0" * 64 + runner = NoCallRunner() + elif stage == "signature": + (receipt.parent / f"{receipt.name}.minisig").write_text( + "bad signature\n", + encoding="utf-8", + ) + runner = NoCallRunner() + elif stage == "time": + utc = "2026-07-24T12:06:00Z" + receipt.write_bytes(_receipt_bytes(commit, utc)) + fake.sign_file( + receipt, + receipt.parent / f"{receipt.name}.minisig", + trusted_comment=_trusted_comment(commit, utc), + ) + runner = NoCallRunner() + elif stage == "bundle": + runner = FakeGitRunner(heads_commit=commit, fail="bundle verify") + elif stage == "cleanup": + + def fail_cleanup(_path: Path) -> None: + raise OSError("cleanup failed") + + runner = HybridRunner() + cleanup_rmdir = fail_cleanup + before_root = _inventory(root) + before_packet = _inventory(packet_root) + before_cargo = _inventory(cargo_home) + + with pytest.raises(audit.ReleasePolicyError): + audit.audit_advisory_mirror( + root, + bundle=bundle, + receipt=receipt, + pubkey=pubkey, + locator=locator, + runner=runner, + verifier=fake, + clock=lambda: NOW, + temp_path_factory=lambda _label: temp_root, + cleanup_rmdir=cleanup_rmdir, + pinned_key_id=TEST_KEY_ID, + pinned_pubkey_sha256=pinned_pubkey_sha, + ) + + assert _inventory(root) == before_root + assert _inventory(packet_root) == before_packet + assert _inventory(cargo_home) == before_cargo + if stage == "cleanup": + assert temp_root.exists() + else: + assert not temp_root.exists() + + +def test_audit_config_bytes_omits_fetch_head_and_staleness_fields( + tmp_path: Path, +) -> None: + cfg = audit.audit_config_bytes( + b'[licenses]\nallow = ["MIT"]\n', + db_root=tmp_path / "db-root", + db_urls=("ssh://mirror.example.invalid/rustsec-advisory-db.git",), + ).decode("utf-8") + + assert "git-fetch-with-cli" not in cfg + assert "maximum-db-staleness" not in cfg + assert "[advisories]" in cfg + with pytest.raises(audit.ReleasePolicyError): + audit.audit_config_bytes( + b'[advisories]\ndb-path = "x"\n', + db_root=tmp_path / "db-root", + db_urls=("ssh://mirror.example.invalid/rustsec-advisory-db.git",), + ) + + +def test_main_prints_success_bytes_only_on_success( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setattr( + audit, "audit_advisory_mirror", lambda *args, **kwargs: b'{"ok":1}\n' + ) + + result = audit.main( + [ + "--bundle", + "bundle", + "--receipt", + "receipt", + "--pubkey", + "pubkey", + "--locator", + "ssh://mirror.example.invalid/rustsec-advisory-db.git", + ] + ) + + captured = capsys.readouterr() + assert result == 0 + assert captured.out == '{"ok":1}\n' + assert captured.err == "" + + +@pytest.mark.parametrize("empty_name", ["bundle", "receipt", "pubkey", "locator"]) +def test_main_empty_inputs_fail_before_audit_orchestration( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + empty_name: str, +) -> None: + calls = [] + + def unexpected(*args, **kwargs): + calls.append((args, kwargs)) + raise AssertionError("audit should not run") + + monkeypatch.setattr(audit, "audit_advisory_mirror", unexpected) + values = { + "bundle": "bundle", + "receipt": "receipt", + "pubkey": "pubkey", + "locator": "ssh://mirror.example.invalid/rustsec-advisory-db.git", + } + values[empty_name] = "" + + result = audit.main( + [ + "--bundle", + values["bundle"], + "--receipt", + values["receipt"], + "--pubkey", + values["pubkey"], + "--locator", + values["locator"], + ] + ) + + captured = capsys.readouterr() + assert result == 1 + assert calls == [] + assert captured.out == "" + assert f"input {empty_name} is empty" in captured.err + + +def test_main_prints_redacted_failures_only_on_error( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + def fail(*args, **kwargs): + raise audit.ReleasePolicyError( + [ + audit._failure_record( + "failed", + expected="public", + actual="/private/path TOKEN=value", + ) + ] + ) + + monkeypatch.setattr(audit, "audit_advisory_mirror", fail) + result = audit.main( + [ + "--bundle", + "bundle", + "--receipt", + "receipt", + "--pubkey", + "pubkey", + "--locator", + "ssh://mirror.example.invalid/rustsec-advisory-db.git", + ] + ) + + captured = capsys.readouterr() + assert result == 1 + assert captured.out == "" + assert "/private/path" not in captured.err + assert "TOKEN=" not in captured.err diff --git a/tests/test_rust_policy_baseline.py b/tests/test_rust_policy_baseline.py index 365d280c8..fc35504f8 100644 --- a/tests/test_rust_policy_baseline.py +++ b/tests/test_rust_policy_baseline.py @@ -64,28 +64,39 @@ def test_check_rust_deny_recipe_is_version_asserted_locked_and_offline() -> None assert command in block -def test_audit_recipe_refreshes_then_checks_offline_fail_closed() -> None: +def test_audit_recipe_uses_signed_packet_without_fetch_db() -> None: block = _makefile_block("audit", "skills") - fetch = "cargo deny --manifest-path $(RUST_MANIFEST) fetch db" - check = ( - "cargo deny --manifest-path $(RUST_MANIFEST) --locked --offline " - "check advisories" - ) required_commands = [ "scripts/check_release_preflight.py cargo-deny", - fetch, - check, + "scripts/advisory_mirror_audit.py", + "AUDIT_ADVISORY_BUNDLE", + "AUDIT_ADVISORY_RECEIPT", + "AUDIT_ADVISORY_PUBKEY", + "AUDIT_ADVISORY_LOCATOR", + "--bundle", + "--receipt", + "--pubkey", + "--locator", ] assert required_commands for command in required_commands: assert command in block - assert block.index(fetch) < block.index(check) + assert "fetch db" not in block - fetch_line = next(line for line in block.splitlines() if fetch in line) - assert "no current advisory result was produced" in fetch_line - assert "ERROR: RustSec advisory refresh failed" in fetch_line - assert "exit 1" in fetch_line + # make audit stdout must be exactly the witness JSON; the private locator must never be echoed. + preflight_line = next( + line + for line in block.splitlines() + if "check_release_preflight.py cargo-deny" in line + ) + audit_line = next( + line + for line in block.splitlines() + if "scripts/advisory_mirror_audit.py" in line + ) + assert ">&2" in preflight_line + assert audit_line.lstrip("\t").startswith("@") def test_release_candidate_driver_binds_policy_before_artifact_construction() -> None: -- 2.51.2