diff --git a/src/egressweave/release_evidence.py b/src/egressweave/release_evidence.py index d60ea5d..275708c 100644 --- a/src/egressweave/release_evidence.py +++ b/src/egressweave/release_evidence.py @@ -17,7 +17,9 @@ import re import stat import uuid +from contextlib import ExitStack from pathlib import Path +from tempfile import TemporaryDirectory from typing import Any DISTRIBUTION_NAME = "egressweave" @@ -213,6 +215,94 @@ def _select_evidence_paths( ) +def _evidence_root_identity(evidence_dir: Path) -> tuple[int, int]: + """Return the current filesystem identity of one admitted evidence root.""" + try: + state = evidence_dir.lstat() + except OSError as error: + raise SystemExit("release evidence directory changed") from error + return state.st_dev, state.st_ino + + +def _snapshot_selected_evidence( + evidence_dir: Path, + snapshot_root: Path, +) -> tuple[ + tuple[Path, Path, Path, Path, Path, Path], + tuple[Path, Path, Path, Path, Path, Path], + tuple[int, int], + dict[str, str], +]: + """Copy one descriptor-bound evidence authority into a private finite snapshot.""" + canonical_root = _require_canonical_evidence_root(evidence_dir) + try: + canonical_snapshot_root = snapshot_root.resolve(strict=True) + except (OSError, RuntimeError) as error: + raise SystemExit("release evidence snapshot directory is unavailable") from error + root_identity = _evidence_root_identity(canonical_root) + original_paths = _select_evidence_paths(canonical_root) + maximums = ( + MAX_ARTIFACT_BYTES, + MAX_ARTIFACT_BYTES, + MAX_SBOM_BYTES, + MAX_SBOM_BYTES, + MAX_SOURCE_IDENTITY_BYTES, + MAX_CHECKSUM_BYTES, + ) + labels = ( + "wheel", + "source distribution", + "wheel SBOM", + "source-distribution SBOM", + "sealed source identity", + "SHA256SUMS", + ) + source_digests: dict[str, str] = {} + + with ExitStack() as stack: + opened: list[tuple[Path, Any, int, str]] = [] + for path, maximum_bytes, label in zip( + original_paths, + maximums, + labels, + strict=True, + ): + try: + stream = stack.enter_context(path.open("rb")) + except OSError as error: + raise SystemExit(f"{label} is unreadable") from error + _require_open_regular_file(path, stream, label=label) + opened.append((path, stream, maximum_bytes, label)) + + if _evidence_root_identity(canonical_root) != root_identity: + raise SystemExit("release evidence directory changed") + _select_evidence_paths(canonical_root) + for path, stream, _, label in opened: + _require_open_regular_file(path, stream, label=label) + + for path, stream, maximum_bytes, label in opened: + digest = hashlib.sha256() + total_bytes = 0 + snapshot_path = canonical_snapshot_root / path.name + try: + with snapshot_path.open("xb") as output: + while True: + block = stream.read(1_048_576) + if not block: + break + total_bytes += len(block) + if total_bytes > maximum_bytes: + raise SystemExit(f"{label} exceeds the safety bound") + digest.update(block) + output.write(block) + except OSError as error: + raise SystemExit(f"{label} cannot be snapshotted safely") from error + source_digests[path.name] = digest.hexdigest() + + snapshot_paths = _select_evidence_paths(canonical_snapshot_root) + return snapshot_paths, original_paths, root_identity, source_digests + + def _load_checksums( checksum_path: Path, expected_names: set[str], @@ -498,90 +588,131 @@ def build_evidence_manifest( if SOURCE_SHA_PATTERN.fullmatch(source_sha) is None: raise SystemExit("source SHA must be exactly 40 lowercase hexadecimal characters") - ( - wheel_path, - sdist_path, - wheel_sbom, - sdist_sbom, - source_identity_path, - checksum_path, - ) = _select_evidence_paths(evidence_dir) - payload_specs = ( - (wheel_path, MAX_ARTIFACT_BYTES, "wheel"), - (sdist_path, MAX_ARTIFACT_BYTES, "source distribution"), - (wheel_sbom, MAX_SBOM_BYTES, "wheel SBOM"), - (sdist_sbom, MAX_SBOM_BYTES, "source-distribution SBOM"), + with TemporaryDirectory(prefix="egressweave-release-evidence-") as snapshot_directory: + snapshot_root = Path(snapshot_directory) + ( + snapshot_paths, + original_paths, + root_identity, + source_snapshot_digests, + ) = _snapshot_selected_evidence(evidence_dir, snapshot_root) ( + wheel_path, + sdist_path, + wheel_sbom, + sdist_sbom, source_identity_path, - MAX_SOURCE_IDENTITY_BYTES, - "sealed source identity", - ), - ) - payload_paths = tuple(path for path, _, _ in payload_specs) - checksums, checksum_digest = _load_checksums( - checksum_path, - {path.name for path in payload_paths}, - ) - observed_digests = _payload_digests(payload_specs) - if checksums != observed_digests: - raise SystemExit("release evidence digest mismatch") + checksum_path, + ) = snapshot_paths + payload_specs = ( + (wheel_path, MAX_ARTIFACT_BYTES, "wheel"), + (sdist_path, MAX_ARTIFACT_BYTES, "source distribution"), + (wheel_sbom, MAX_SBOM_BYTES, "wheel SBOM"), + (sdist_sbom, MAX_SBOM_BYTES, "source-distribution SBOM"), + ( + source_identity_path, + MAX_SOURCE_IDENTITY_BYTES, + "sealed source identity", + ), + ) + payload_paths = tuple(path for path, _, _ in payload_specs) + checksums, checksum_digest = _load_checksums( + checksum_path, + {path.name for path in payload_paths}, + ) + observed_digests = _payload_digests(payload_specs) + if checksums != observed_digests: + raise SystemExit("release evidence digest mismatch") - sealed_repository, sealed_source_sha = _load_source_identity( - source_identity_path, - expected_digest=observed_digests[source_identity_path.name], - ) - if sealed_repository != repository or sealed_source_sha != source_sha: - raise SystemExit("sealed source identity does not match caller expectations") - - version = WHEEL_PATTERN.fullmatch(wheel_path.name).group("version") - artifacts: list[dict[str, str]] = [] - for kind, artifact_path, sbom_path in ( - ("sdist", sdist_path, sdist_sbom), - ("wheel", wheel_path, wheel_sbom), - ): - artifact_digest = observed_digests[artifact_path.name] - serial_number = _verify_sbom( - sbom_path, - artifact_name=artifact_path.name, - artifact_digest=artifact_digest, - version=version, - expected_digest=observed_digests[sbom_path.name], + sealed_repository, sealed_source_sha = _load_source_identity( + source_identity_path, + expected_digest=observed_digests[source_identity_path.name], + ) + if sealed_repository != repository or sealed_source_sha != source_sha: + raise SystemExit("sealed source identity does not match caller expectations") + + version = WHEEL_PATTERN.fullmatch(wheel_path.name).group("version") + artifacts: list[dict[str, str]] = [] + for kind, artifact_path, sbom_path in ( + ("sdist", sdist_path, sdist_sbom), + ("wheel", wheel_path, wheel_sbom), + ): + artifact_digest = observed_digests[artifact_path.name] + serial_number = _verify_sbom( + sbom_path, + artifact_name=artifact_path.name, + artifact_digest=artifact_digest, + version=version, + expected_digest=observed_digests[sbom_path.name], + ) + artifacts.append( + { + "artifactFilename": artifact_path.name, + "artifactSha256": artifact_digest, + "kind": kind, + "sbomFilename": sbom_path.name, + "sbomSerialNumber": serial_number, + "sbomSha256": observed_digests[sbom_path.name], + } + ) + if _payload_digests(payload_specs) != observed_digests: + raise SystemExit("release evidence changed during verification") + if ( + _sha256_file( + checksum_path, + maximum_bytes=MAX_CHECKSUM_BYTES, + label="SHA256SUMS", + ) + != checksum_digest + ): + raise SystemExit("SHA256SUMS changed during verification") + + canonical_original_root = _require_canonical_evidence_root(evidence_dir) + if _evidence_root_identity(canonical_original_root) != root_identity: + raise SystemExit("release evidence directory changed") + _select_evidence_paths(canonical_original_root) + original_maximums = ( + MAX_ARTIFACT_BYTES, + MAX_ARTIFACT_BYTES, + MAX_SBOM_BYTES, + MAX_SBOM_BYTES, + MAX_SOURCE_IDENTITY_BYTES, + MAX_CHECKSUM_BYTES, ) - artifacts.append( - { - "artifactFilename": artifact_path.name, - "artifactSha256": artifact_digest, - "kind": kind, - "sbomFilename": sbom_path.name, - "sbomSerialNumber": serial_number, - "sbomSha256": observed_digests[sbom_path.name], - } + original_labels = ( + "wheel", + "source distribution", + "wheel SBOM", + "source-distribution SBOM", + "sealed source identity", + "SHA256SUMS", ) - if _payload_digests(payload_specs) != observed_digests: - raise SystemExit("release evidence changed during verification") - if ( - _sha256_file( - checksum_path, - maximum_bytes=MAX_CHECKSUM_BYTES, - label="SHA256SUMS", + original_specs = tuple( + (path, maximum_bytes, label) + for path, maximum_bytes, label in zip( + original_paths, + original_maximums, + original_labels, + strict=True, + ) ) - != checksum_digest - ): - raise SystemExit("SHA256SUMS changed during verification") - artifacts.sort(key=lambda item: item["artifactFilename"]) - return { - "artifacts": artifacts, - "checksumFilename": checksum_path.name, - "checksumSha256": checksum_digest, - "format": EVIDENCE_MANIFEST_FORMAT, - "formatVersion": EVIDENCE_MANIFEST_VERSION, - "cycloneDxSpecVersion": CYCLONEDX_SPEC_VERSION, - "predicateType": ATTESTATION_PREDICATE_TYPE, - "repository": sealed_repository, - "sourceIdentityFilename": source_identity_path.name, - "sourceIdentitySha256": observed_digests[source_identity_path.name], - "sourceSha": sealed_source_sha, - } + if _payload_digests(original_specs) != source_snapshot_digests: + raise SystemExit("release evidence changed during verification") + + artifacts.sort(key=lambda item: item["artifactFilename"]) + return { + "artifacts": artifacts, + "checksumFilename": checksum_path.name, + "checksumSha256": checksum_digest, + "format": EVIDENCE_MANIFEST_FORMAT, + "formatVersion": EVIDENCE_MANIFEST_VERSION, + "cycloneDxSpecVersion": CYCLONEDX_SPEC_VERSION, + "predicateType": ATTESTATION_PREDICATE_TYPE, + "repository": sealed_repository, + "sourceIdentityFilename": source_identity_path.name, + "sourceIdentitySha256": observed_digests[source_identity_path.name], + "sourceSha": sealed_source_sha, + } def _encode_evidence_manifest(manifest: dict[str, Any]) -> bytes: diff --git a/tests/test_sealed_release_evidence_checksum_snapshot.py b/tests/test_sealed_release_evidence_checksum_snapshot.py index 20b0106..01723f9 100644 --- a/tests/test_sealed_release_evidence_checksum_snapshot.py +++ b/tests/test_sealed_release_evidence_checksum_snapshot.py @@ -133,7 +133,7 @@ def verify_then_mutate( version: str, expected_digest: str, ) -> str: - """Replace SHA256SUMS only after both SBOMs and payloads were accepted.""" + """Replace source SHA256SUMS after the private snapshot was accepted.""" serial = original_verify( sbom_path, artifact_name=artifact_name, @@ -147,7 +147,7 @@ def verify_then_mutate( monkeypatch.setattr(release_evidence, "_verify_sbom", verify_then_mutate) - with pytest.raises(SystemExit, match="SHA256SUMS changed during verification"): + with pytest.raises(SystemExit, match="release evidence changed during verification"): release_evidence.build_evidence_manifest( root, repository=REPOSITORY, diff --git a/tests/test_sealed_release_evidence_root_membership.py b/tests/test_sealed_release_evidence_root_membership.py new file mode 100644 index 0000000..3680ed1 --- /dev/null +++ b/tests/test_sealed_release_evidence_root_membership.py @@ -0,0 +1,94 @@ +"""Regressions for sealed release-evidence root membership stability.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from egressweave import release_evidence + +VERSION = "0.3.0" + + +def _minimal_evidence(root: Path) -> None: + """Create the exact six direct-child names needed by the snapshot boundary.""" + root.mkdir() + names = ( + f"egressweave-{VERSION}-py3-none-any.whl", + f"egressweave-{VERSION}.tar.gz", + f"egressweave-{VERSION}-py3-none-any.whl.cdx.json", + f"egressweave-{VERSION}.tar.gz.cdx.json", + "SOURCE_IDENTITY.json", + "SHA256SUMS", + ) + for name in names: + (root / name).write_bytes(b"bounded fixture") + + +def test_snapshot_rejects_membership_change_after_selection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject an extra direct child added after the admitted member set is selected.""" + root = tmp_path / "evidence" + snapshot_root = tmp_path / "snapshot" + _minimal_evidence(root) + snapshot_root.mkdir() + original_select = release_evidence._select_evidence_paths + root_identity = release_evidence._evidence_root_identity(root) + source_selections = 0 + + def select_then_add_unexpected_member( + candidate: Path, + ) -> tuple[Path, Path, Path, Path, Path, Path]: + """Add one unreviewed direct child immediately after source selection.""" + nonlocal source_selections + selected = original_select(candidate) + if release_evidence._evidence_root_identity(candidate) == root_identity: + source_selections += 1 + if source_selections == 1: + (root / "unexpected.txt").write_bytes(b"unreviewed evidence") + return selected + + monkeypatch.setattr( + release_evidence, + "_select_evidence_paths", + select_then_add_unexpected_member, + ) + + with pytest.raises(SystemExit, match="cardinality mismatch"): + release_evidence._snapshot_selected_evidence(root, snapshot_root) + + +def test_snapshot_accepts_private_root_through_symlinked_parent(tmp_path: Path) -> None: + """Accept a private snapshot whose lexical parent aliases its real location.""" + root = tmp_path / "evidence" + _minimal_evidence(root) + real_parent = tmp_path / "private" + real_parent.mkdir() + alias_parent = tmp_path / "var" + alias_parent.symlink_to(real_parent, target_is_directory=True) + snapshot_root = alias_parent / "snapshot" + snapshot_root.mkdir() + + snapshot_paths, _, _, _ = release_evidence._snapshot_selected_evidence( + root, + snapshot_root, + ) + + canonical_snapshot_root = snapshot_root.resolve(strict=True) + assert {path.parent for path in snapshot_paths} == {canonical_snapshot_root} + + +def test_snapshot_rejects_unavailable_private_root(tmp_path: Path) -> None: + """Fail closed when the private snapshot root cannot be resolved strictly.""" + root = tmp_path / "evidence" + _minimal_evidence(root) + missing_snapshot_root = tmp_path / "missing-snapshot" + + with pytest.raises( + SystemExit, + match="release evidence snapshot directory is unavailable", + ): + release_evidence._snapshot_selected_evidence(root, missing_snapshot_root) diff --git a/tests/test_sealed_release_evidence_sbom_digest_binding.py b/tests/test_sealed_release_evidence_sbom_digest_binding.py index 38cc807..6da5213 100644 --- a/tests/test_sealed_release_evidence_sbom_digest_binding.py +++ b/tests/test_sealed_release_evidence_sbom_digest_binding.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json from pathlib import Path @@ -15,16 +16,8 @@ from egressweave import release_evidence -def test_semantic_sbom_snapshot_must_match_the_accepted_checksum_digest( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Reject a valid alternate SBOM parsed between matching outer snapshots.""" - root = tmp_path / "evidence" - _evidence(root) - wheel = root / "egressweave-0.3.0-py3-none-any.whl" - sbom = root / f"{wheel.name}.cdx.json" - accepted_bytes = sbom.read_bytes() +def _alternate_sbom_bytes(accepted_bytes: bytes) -> bytes: + """Return a valid but semantically different CycloneDX fixture.""" alternate_document = json.loads(accepted_bytes) alternate_document["components"] = [ { @@ -35,28 +28,77 @@ def test_semantic_sbom_snapshot_must_match_the_accepted_checksum_digest( ] alternate_document.pop("serialNumber") alternate_document["serialNumber"] = _serial(alternate_document) - alternate_bytes = json.dumps(alternate_document).encode("utf-8") - original_load = release_evidence._load_strict_json + return json.dumps(alternate_document).encode("utf-8") + - def load_alternate_then_restore( - path: Path, +def test_private_snapshot_ignores_transient_source_sbom_semantics( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep source-root ABA changes outside the admitted semantic snapshot.""" + root = tmp_path / "evidence" + _evidence(root) + wheel = root / "egressweave-0.3.0-py3-none-any.whl" + source_sbom = root / f"{wheel.name}.cdx.json" + accepted_bytes = source_sbom.read_bytes() + alternate_bytes = _alternate_sbom_bytes(accepted_bytes) + accepted_manifest = _build(root) + original_verify = release_evidence._verify_sbom + + def verify_while_source_is_alternate( + sbom_path: Path, *, - expected_digest: str | None = None, - ): - """Expose an ABA swap only during semantic parsing of the wheel SBOM.""" - if path != sbom: - if expected_digest is None: - return original_load(path) - return original_load(path, expected_digest=expected_digest) - path.write_bytes(alternate_bytes) + artifact_name: str, + artifact_digest: str, + version: str, + expected_digest: str, + ) -> str: + """Swap only the caller-owned source while the private snapshot is parsed.""" + if not artifact_name.endswith(".whl"): + return original_verify( + sbom_path, + artifact_name=artifact_name, + artifact_digest=artifact_digest, + version=version, + expected_digest=expected_digest, + ) + source_sbom.write_bytes(alternate_bytes) try: - if expected_digest is None: - return original_load(path) - return original_load(path, expected_digest=expected_digest) + return original_verify( + sbom_path, + artifact_name=artifact_name, + artifact_digest=artifact_digest, + version=version, + expected_digest=expected_digest, + ) finally: - path.write_bytes(accepted_bytes) + source_sbom.write_bytes(accepted_bytes) + + monkeypatch.setattr( + release_evidence, + "_verify_sbom", + verify_while_source_is_alternate, + ) + + assert _build(root) == accepted_manifest + - monkeypatch.setattr(release_evidence, "_load_strict_json", load_alternate_then_restore) +def test_strict_json_rejects_semantics_outside_sealed_digest(tmp_path: Path) -> None: + """Retain direct proof that semantic parsing cannot outrun its sealed digest.""" + path = tmp_path / "payload.cdx.json" + accepted_document = { + "bomFormat": "CycloneDX", + "components": [], + } + accepted_bytes = json.dumps(accepted_document).encode("utf-8") + accepted_digest = hashlib.sha256(accepted_bytes).hexdigest() + path.write_bytes(accepted_bytes) + alternate_document = dict(accepted_document) + alternate_document["components"] = [{"name": "alternate"}] + path.write_text(json.dumps(alternate_document), encoding="utf-8") with pytest.raises(SystemExit, match="sealed digest"): - _build(root) + release_evidence._load_strict_json( + path, + expected_digest=accepted_digest, + ) diff --git a/tests/test_sealed_release_evidence_snapshot_boundary.py b/tests/test_sealed_release_evidence_snapshot_boundary.py index 458b597..5759f80 100644 --- a/tests/test_sealed_release_evidence_snapshot_boundary.py +++ b/tests/test_sealed_release_evidence_snapshot_boundary.py @@ -85,13 +85,18 @@ def _source_identity() -> bytes: ).encode("utf-8") -def _evidence(root: Path) -> dict[str, Path]: +def _evidence( + root: Path, + *, + wheel_bytes: bytes = b"wheel", + sdist_bytes: bytes = b"sdist", +) -> dict[str, Path]: """Create one complete valid six-file release evidence set.""" root.mkdir() wheel = root / f"egressweave-{VERSION}-py3-none-any.whl" sdist = root / f"egressweave-{VERSION}.tar.gz" - wheel.write_bytes(b"wheel") - sdist.write_bytes(b"sdist") + wheel.write_bytes(wheel_bytes) + sdist.write_bytes(sdist_bytes) wheel_sbom = root / f"{wheel.name}.cdx.json" sdist_sbom = root / f"{sdist.name}.cdx.json" wheel_sbom.write_text( @@ -140,6 +145,19 @@ def test_hashing_rejects_a_symlink_even_when_target_bytes_are_valid( ) +def test_hashing_enforces_the_configured_safety_bound(tmp_path: Path) -> None: + """Fail closed before hashing bytes beyond a caller-selected finite bound.""" + path = tmp_path / "payload" + path.write_bytes(b"too large") + + with pytest.raises(SystemExit, match="safety bound"): + release_evidence._sha256_file( + path, + maximum_bytes=1, + label="payload", + ) + + def test_descriptor_identity_error_is_masked_as_unsafe( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -167,6 +185,27 @@ def fail_lstat(candidate: Path): ) +def test_root_identity_error_is_normalized( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Normalize root metadata loss at the snapshot admission boundary.""" + root = tmp_path / "evidence" + root.mkdir() + original_lstat = Path.lstat + + def fail_root_lstat(candidate: Path): + """Model the evidence root disappearing during identity capture.""" + if candidate == root: + raise OSError("gone") + return original_lstat(candidate) + + monkeypatch.setattr(Path, "lstat", fail_root_lstat) + + with pytest.raises(SystemExit, match="release evidence directory changed"): + release_evidence._evidence_root_identity(root) + + @pytest.mark.parametrize("mismatch", ["before", "after"]) def test_stable_read_rejects_each_digest_mismatch(mismatch: str) -> None: """Reject either a stale pre-read digest or a changed post-read digest.""" @@ -185,11 +224,190 @@ def test_stable_read_rejects_each_digest_mismatch(mismatch: str) -> None: ) +def test_snapshot_rejects_source_open_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Fail closed when an admitted direct child cannot be descriptor-opened.""" + root = tmp_path / "evidence" + snapshot_root = tmp_path / "snapshot" + paths = _evidence(root) + snapshot_root.mkdir() + original_open = Path.open + + def fail_wheel_open(candidate: Path, *args, **kwargs): + """Model one source file becoming unreadable after path selection.""" + mode = args[0] if args else kwargs.get("mode", "r") + if candidate == paths["wheel"] and mode == "rb": + raise OSError("unreadable") + return original_open(candidate, *args, **kwargs) + + monkeypatch.setattr(Path, "open", fail_wheel_open) + + with pytest.raises(SystemExit, match="wheel is unreadable"): + release_evidence._snapshot_selected_evidence(root, snapshot_root) + + +def test_snapshot_rejects_root_identity_change_after_opening_children( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject a root identity change after all direct children have opened.""" + root = tmp_path / "evidence" + snapshot_root = tmp_path / "snapshot" + _evidence(root) + snapshot_root.mkdir() + original_identity = release_evidence._evidence_root_identity + root_calls = 0 + + def change_second_root_identity(candidate: Path) -> tuple[int, int]: + """Return a distinct identity at the post-open root checkpoint.""" + nonlocal root_calls + identity = original_identity(candidate) + if candidate == root: + root_calls += 1 + if root_calls == 2: + return identity[0], identity[1] + 1 + return identity + + monkeypatch.setattr( + release_evidence, + "_evidence_root_identity", + change_second_root_identity, + ) + + with pytest.raises(SystemExit, match="release evidence directory changed"): + release_evidence._snapshot_selected_evidence(root, snapshot_root) + + +def test_snapshot_rejects_private_copy_creation_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Fail closed when the verifier cannot create its private finite snapshot.""" + root = tmp_path / "evidence" + snapshot_root = tmp_path / "snapshot" + _evidence(root) + snapshot_root.mkdir() + original_open = Path.open + + def fail_snapshot_open(candidate: Path, *args, **kwargs): + """Model local snapshot storage becoming unavailable at first write.""" + mode = args[0] if args else kwargs.get("mode", "r") + if candidate.parent == snapshot_root and mode == "xb": + raise OSError("snapshot unavailable") + return original_open(candidate, *args, **kwargs) + + monkeypatch.setattr(Path, "open", fail_snapshot_open) + + with pytest.raises(SystemExit, match="cannot be snapshotted safely"): + release_evidence._snapshot_selected_evidence(root, snapshot_root) + + +def test_manifest_rejects_release_root_replacement_after_selection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep a whole-directory replacement outside the sealed-evidence boundary.""" + root = tmp_path / "evidence" + parked_root = tmp_path / "evidence-original" + replacement_root = tmp_path / "evidence-replacement" + _evidence(root) + _evidence( + replacement_root, + wheel_bytes=b"replacement wheel bytes", + sdist_bytes=b"replacement sdist bytes", + ) + original_load_checksums = release_evidence._load_checksums + + def replace_root_then_load( + checksum_path: Path, + expected_names: set[str], + ) -> tuple[dict[str, str], str]: + """Replace the already-selected evidence directory before first hashing.""" + root.rename(parked_root) + replacement_root.rename(root) + return original_load_checksums(checksum_path, expected_names) + + monkeypatch.setattr(release_evidence, "_load_checksums", replace_root_then_load) + + with pytest.raises(SystemExit, match="release evidence directory changed"): + release_evidence.build_evidence_manifest( + root, + repository=REPOSITORY, + source_sha=SOURCE_SHA, + ) + + +def test_manifest_keeps_original_root_authority_across_transient_replacement( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Do not let replace-then-restore redirect one verification decision.""" + root = tmp_path / "evidence" + parked_root = tmp_path / "evidence-original" + replacement_root = tmp_path / "evidence-replacement" + original_paths = _evidence(root) + _evidence( + replacement_root, + wheel_bytes=b"replacement wheel bytes", + sdist_bytes=b"replacement sdist bytes", + ) + original_load_checksums = release_evidence._load_checksums + original_sha256_file = release_evidence._sha256_file + checksum_hash_count = 0 + + def replace_root_then_load( + checksum_path: Path, + expected_names: set[str], + ) -> tuple[dict[str, str], str]: + """Redirect later path opens to a self-consistent replacement root.""" + root.rename(parked_root) + replacement_root.rename(root) + return original_load_checksums(checksum_path, expected_names) + + def hash_then_restore_root( + path: Path, + *, + maximum_bytes: int, + label: str, + ) -> str: + """Restore the original pathname only after replacement verification.""" + nonlocal checksum_hash_count + digest = original_sha256_file( + path, + maximum_bytes=maximum_bytes, + label=label, + ) + if label == "SHA256SUMS": + checksum_hash_count += 1 + if checksum_hash_count == 3: + root.rename(replacement_root) + parked_root.rename(root) + return digest + + monkeypatch.setattr(release_evidence, "_load_checksums", replace_root_then_load) + monkeypatch.setattr(release_evidence, "_sha256_file", hash_then_restore_root) + + manifest = release_evidence.build_evidence_manifest( + root, + repository=REPOSITORY, + source_sha=SOURCE_SHA, + ) + artifact_digests = { + item["kind"]: item["artifactSha256"] for item in manifest["artifacts"] + } + assert artifact_digests == { + "sdist": _digest(original_paths["sdist"]), + "wheel": _digest(original_paths["wheel"]), + } + + def test_manifest_rejects_payload_mutation_during_sbom_verification( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - """Do not issue a manifest after verified payload bytes have changed.""" + """Do not issue a manifest after verified source payload bytes have changed.""" root = tmp_path / "evidence" paths = _evidence(root) original_verify = release_evidence._verify_sbom @@ -202,7 +420,7 @@ def verify_then_mutate( version: str, expected_digest: str, ) -> str: - """Change the wheel only after the verifier captured all initial digests.""" + """Change the source wheel after the verifier captured all initial digests.""" serial = original_verify( sbom_path, artifact_name=artifact_name, @@ -222,3 +440,94 @@ def verify_then_mutate( repository=REPOSITORY, source_sha=SOURCE_SHA, ) + + +def test_manifest_rejects_private_snapshot_payload_corruption( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Fail closed if local snapshot storage changes after semantic verification.""" + root = tmp_path / "evidence" + _evidence(root) + original_verify = release_evidence._verify_sbom + + def verify_then_corrupt_snapshot( + sbom_path: Path, + *, + artifact_name: str, + artifact_digest: str, + version: str, + expected_digest: str, + ) -> str: + """Corrupt a previously verified snapshot artifact after the final SBOM.""" + serial = original_verify( + sbom_path, + artifact_name=artifact_name, + artifact_digest=artifact_digest, + version=version, + expected_digest=expected_digest, + ) + if artifact_name.endswith(".whl"): + (sbom_path.parent / f"egressweave-{VERSION}.tar.gz").write_bytes( + b"local snapshot corruption" + ) + return serial + + monkeypatch.setattr( + release_evidence, + "_verify_sbom", + verify_then_corrupt_snapshot, + ) + + with pytest.raises(SystemExit, match="release evidence changed during verification"): + release_evidence.build_evidence_manifest( + root, + repository=REPOSITORY, + source_sha=SOURCE_SHA, + ) + + +def test_manifest_rejects_private_snapshot_checksum_corruption( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Fail closed if the private checksum snapshot changes before handoff.""" + root = tmp_path / "evidence" + _evidence(root) + original_verify = release_evidence._verify_sbom + + def verify_then_corrupt_checksum( + sbom_path: Path, + *, + artifact_name: str, + artifact_digest: str, + version: str, + expected_digest: str, + ) -> str: + """Corrupt snapshot SHA256SUMS only after both SBOMs were accepted.""" + serial = original_verify( + sbom_path, + artifact_name=artifact_name, + artifact_digest=artifact_digest, + version=version, + expected_digest=expected_digest, + ) + if artifact_name.endswith(".whl"): + (sbom_path.parent / "SHA256SUMS").write_text( + "local snapshot corruption\n", + encoding="ascii", + ) + return serial + + monkeypatch.setattr( + release_evidence, + "_verify_sbom", + verify_then_corrupt_checksum, + ) + + with pytest.raises(SystemExit, match="SHA256SUMS changed during verification"): + release_evidence.build_evidence_manifest( + root, + repository=REPOSITORY, + source_sha=SOURCE_SHA, + )