Require promotion manifests to bind consumed evidence

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`.",
     )

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