diff --git a/src/tether/promote.py b/src/tether/promote.py index 09415d78..32911716 100644 --- a/src/tether/promote.py +++ b/src/tether/promote.py @@ -291,11 +291,13 @@ def _verify_manifest(packet_dir: Path, *, require_hashes: bool) -> dict[str, Any ) mismatches: list[dict[str, Any]] = [] + listed_names: set[str] = set() for item in manifest.get("files") or []: name = item.get("name") if not name: mismatches.append({"name": "", "reason": "missing name"}) continue + listed_names.add(str(name)) path = packet_dir / str(name) if not path.exists(): mismatches.append({"name": name, "reason": "missing file"}) @@ -305,15 +307,28 @@ def _verify_manifest(packet_dir: Path, *, require_hashes: bool) -> dict[str, Any if expected_size is not None and int(expected_size) != path.stat().st_size: mismatches.append({"name": name, "reason": "size mismatch"}) continue - if expected_sha and _sha256_file(path) != expected_sha: + if not expected_sha: + mismatches.append({"name": name, "reason": "missing sha256"}) + elif _sha256_file(path) != expected_sha: mismatches.append({"name": name, "reason": "sha256 mismatch"}) + # Bind each decision input, including the sidecar that overrides the + # embedded policy report. Unrelated operator files need not be listed. + required_names = {"deployment-proof.json"} + if (packet_dir / "policy-diff.json").exists(): + required_names.add("policy-diff.json") + for name in sorted(required_names - listed_names): + mismatches.append({"name": name, "reason": "missing manifest entry"}) + _add_check( checks, "packet_manifest_hashes", not mismatches or not require_hashes, category="reproducibility", - expected="all MANIFEST.json hashes match" if require_hashes else "optional", + expected=( + "decision inputs are listed and all MANIFEST.json hashes match" + if require_hashes else "optional" + ), actual=mismatches, remediation="Do not promote from a modified proof packet; rerun `tether prove`.", ) diff --git a/tests/test_promote_manifest.py b/tests/test_promote_manifest.py new file mode 100644 index 00000000..a7d4a772 --- /dev/null +++ b/tests/test_promote_manifest.py @@ -0,0 +1,167 @@ +"""Promotion must bind the files it consumes when hashes are required.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from tether.deploy_proof import write_deploy_proof_packet +from tether.promote import decide_promotion + + +def _policy_report(verdict: str = "pass") -> dict: + return { + "kind": "tether.policy_diff", + "summary": {"verdict": verdict, "compared": 1}, + } + + +def _packet(tmp_path: Path, *, policy: str | None = None, passed: bool = True) -> Path: + receipt = { + "schema_version": 1, + "kind": "tether.deployment_proof", + "passed": passed, + "checks": [{"name": "synthetic_fixture", "status": "pass"}], + } + if policy is not None: + receipt["policy_diff"] = {"enabled": True, "report": _policy_report(policy)} + packet = tmp_path / "proof" + write_deploy_proof_packet(receipt, packet) + return packet + + +def _read_manifest(packet: Path) -> dict: + return json.loads((packet / "MANIFEST.json").read_text(encoding="utf-8")) + + +def _write_manifest(packet: Path, manifest: dict) -> None: + (packet / "MANIFEST.json").write_text(json.dumps(manifest), encoding="utf-8") + + +def _omit(packet: Path, name: str) -> None: + manifest = _read_manifest(packet) + manifest["files"] = [item for item in manifest["files"] if item["name"] != name] + _write_manifest(packet, manifest) + + +def _assert_integrity_failure(packet: Path, *, active: bool = False) -> None: + report = decide_promotion(packet, candidate_active=active) + assert report["decision"] == ("ROLLBACK" if active else "BLOCK") + assert "packet_manifest_hashes" in report["summary"]["failed_checks"] + + +@pytest.mark.parametrize("variant", ["missing", "null", "empty"]) +def test_required_manifest_cannot_omit_file_list(tmp_path: Path, variant: str) -> None: + packet = _packet(tmp_path) + manifest = _read_manifest(packet) + if variant == "missing": + del manifest["files"] + else: + manifest["files"] = None if variant == "null" else [] + _write_manifest(packet, manifest) + _assert_integrity_failure(packet) + + +@pytest.mark.parametrize("variant", ["missing", "null", "empty"]) +def test_required_hash_cannot_be_absent(tmp_path: Path, variant: str) -> None: + packet = _packet(tmp_path) + manifest = _read_manifest(packet) + for item in manifest["files"]: + if item["name"] == "deployment-proof.json": + if variant == "missing": + del item["sha256"] + else: + item["sha256"] = None if variant == "null" else "" + _write_manifest(packet, manifest) + _assert_integrity_failure(packet) + + +@pytest.mark.parametrize("name", ["deployment-proof.json", "policy-diff.json"]) +def test_consumed_artifact_must_be_listed(tmp_path: Path, name: str) -> None: + packet = _packet(tmp_path, policy="pass") + _omit(packet, name) + _assert_integrity_failure(packet) + + +def test_unlisted_sidecar_cannot_override_embedded_failure(tmp_path: Path) -> None: + packet = _packet(tmp_path, policy="fail") + assert decide_promotion(packet)["decision"] == "BLOCK" + _omit(packet, "policy-diff.json") + (packet / "policy-diff.json").write_text(json.dumps(_policy_report("pass")), encoding="utf-8") + _assert_integrity_failure(packet) + + +def test_edited_proof_without_hash_or_size_is_rejected(tmp_path: Path) -> None: + packet = _packet(tmp_path, passed=False) + assert decide_promotion(packet)["decision"] == "BLOCK" + path = packet / "deployment-proof.json" + proof = json.loads(path.read_text(encoding="utf-8")) + proof["passed"] = True + path.write_text(json.dumps(proof), encoding="utf-8") + manifest = _read_manifest(packet) + for item in manifest["files"]: + if item["name"] == path.name: + del item["sha256"] + del item["size_bytes"] + _write_manifest(packet, manifest) + _assert_integrity_failure(packet) + + +def test_missing_binding_rolls_back_active_candidate(tmp_path: Path) -> None: + packet = _packet(tmp_path) + _omit(packet, "deployment-proof.json") + _assert_integrity_failure(packet, active=True) + + +@pytest.mark.parametrize("mode", ["none", "sidecar", "embedded"]) +def test_valid_packet_still_promotes(tmp_path: Path, mode: str) -> None: + packet = _packet(tmp_path, policy=None if mode == "none" else "pass") + if mode == "embedded": + (packet / "policy-diff.json").unlink() + _omit(packet, "policy-diff.json") + report = decide_promotion(packet) + assert report["decision"] == "PROMOTE" + assert report["summary"]["fail"] == 0 + + +@pytest.mark.parametrize("variant", ["missing_hash", "empty_list"]) +def test_explicit_optional_hash_profile_is_preserved(tmp_path: Path, variant: str) -> None: + packet = _packet(tmp_path) + manifest = _read_manifest(packet) + if variant == "empty_list": + manifest["files"] = [] + else: + for item in manifest["files"]: + del item["sha256"] + _write_manifest(packet, manifest) + profile = tmp_path / "optional.json" + profile.write_text( + json.dumps({"thresholds": {"require_manifest_hashes": False}}), + encoding="utf-8", + ) + report = decide_promotion(packet, profile_path=profile) + assert report["decision"] == "PROMOTE" + assert report["summary"]["fail"] == 0 + + +def test_unconsumed_operator_note_need_not_be_listed(tmp_path: Path) -> None: + packet = _packet(tmp_path) + (packet / "operator-notes.txt").write_text("Review pending.\n", encoding="utf-8") + assert decide_promotion(packet)["decision"] == "PROMOTE" + + +def test_wrong_digest_still_blocks(tmp_path: Path) -> None: + packet = _packet(tmp_path) + manifest = _read_manifest(packet) + manifest["files"][0]["sha256"] = "0" * 64 + _write_manifest(packet, manifest) + _assert_integrity_failure(packet) + + +def test_missing_manifest_still_blocks(tmp_path: Path) -> None: + packet = _packet(tmp_path) + (packet / "MANIFEST.json").unlink() + report = decide_promotion(packet) + assert report["decision"] == "BLOCK" + assert "packet_manifest_present" in report["summary"]["failed_checks"]