diff --git a/scripts/p4_hostile_acceptance.py b/scripts/p4_hostile_acceptance.py index b17dd0e..3ab0f36 100644 --- a/scripts/p4_hostile_acceptance.py +++ b/scripts/p4_hostile_acceptance.py @@ -136,6 +136,25 @@ def write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") +def write_acceptance_report(path: Path, report: dict[str, Any]) -> None: + """Publish phase completion only after checks and final cleanup succeed.""" + report.pop("completed_phase_gate", None) + checks = report.get("checks") + cleanup = report.get("cleanup") or {} + if ( + report.get("phase") == "P4" + and report.get("passed") is True + and report.get("run_outcome") == "COMPLETE" + and isinstance(checks, dict) + and checks + and all(value is True for value in checks.values()) + and cleanup.get("unrelated_container_ids_preserved") is True + and cleanup.get("ephemeral_private_files_removed") is True + ): + report["completed_phase_gate"] = "P4" + write_json(path, report) + + def docker_json(*arguments: str) -> Any: return json.loads(run(["docker", *arguments]).stdout) @@ -1859,7 +1878,7 @@ def main() -> int: report["error"] = {"type": "CleanupError", "message": "unrelated container identity changed"} return_code = 1 report["completed_at_utc"] = to_utc_iso(datetime.now(UTC)) - write_json(output, report) + write_acceptance_report(output, report) print( json.dumps( { diff --git a/scripts/p4_hostile_acceptance_exact.py b/scripts/p4_hostile_acceptance_exact.py index 0d6d61c..732ad99 100644 --- a/scripts/p4_hostile_acceptance_exact.py +++ b/scripts/p4_hostile_acceptance_exact.py @@ -147,6 +147,10 @@ def enrich_report( if not output_path.is_file(): raise RuntimeError(f"P4 harness did not produce an output report: {output_path}") report = json.loads(output_path.read_text(encoding="utf-8")) + # Enrichment is a further acceptance boundary; invalidate any earlier marker + # before parsing captured evidence, which can raise or be interrupted. + report.pop("completed_phase_gate", None) + output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") report["harness"] = { "path": str(HERE), "sha256": sha256_file(HERE), @@ -205,6 +209,10 @@ def enrich_report( ) return_code = 1 report["wrapper_return_code"] = return_code + if return_code != 0: + report["passed"] = False + report["run_outcome"] = "INFRA_FAILED" + report["release_verdict"] = "NOT_READY" output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") return report @@ -213,6 +221,15 @@ def main(arguments: list[str] | None = None) -> int: argv = list(sys.argv[1:] if arguments is None else arguments) sealed_manifest_path, output_path = parse_wrapper_paths(argv) harness = load_harness() + publish_report = harness.write_acceptance_report + + def write_pending_report(path: Path, report: dict[str, Any]) -> None: + report.pop("completed_phase_gate", None) + harness.write_json(path, report) + + # The core can finish before the exact verifier evidence is enriched. + # Keep its on-disk result pending until every wrapper layer has returned. + harness.write_acceptance_report = write_pending_report artifacts = resolve_runner_artifacts(harness, sealed_manifest_path) capture: dict[str, Any] = {} harness.execute_container = make_execute_container_override( @@ -234,6 +251,7 @@ def main(arguments: list[str] | None = None) -> int: return_code=return_code, ) final_code = 0 if report.get("passed") is True and report.get("run_outcome") == "COMPLETE" else 1 + publish_report(output_path, report) print( json.dumps( { diff --git a/tests/test_p4_master_report_contract.py b/tests/test_p4_master_report_contract.py new file mode 100644 index 0000000..4026343 --- /dev/null +++ b/tests/test_p4_master_report_contract.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import json +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +def load_script(name: str): + spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py") + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def completed_report() -> dict: + # Contract fixture only: no real P4 execution or certification is claimed. + return { + "phase": "P4", + "passed": True, + "run_outcome": "COMPLETE", + "release_verdict": "NOT_READY", + "source_commit": "a" * 40, + "checks": {"sealed_twelve_image_manifest_verified": True, "public_verifier_passed": True}, + "cleanup": { + "unrelated_container_ids_preserved": True, + "ephemeral_private_files_removed": True, + }, + } + + +def test_successful_producer_report_is_consumable_without_changing_master_gate(tmp_path: Path) -> None: + producer = load_script("p4_hostile_acceptance") + consumer = load_script("master_acceptance") + report = completed_report() + output = tmp_path / "p4.json" + + producer.write_acceptance_report(output, report) + + gate = consumer._gate("P4", output, consumer._p4) + assert gate == {"state": "COMPLETE", "evidence_sha256": hashlib.sha256(output.read_bytes()).hexdigest()} + emitted = json.loads(output.read_text()) + assert emitted["source_commit"] == "a" * 40 + assert emitted["release_verdict"] == "NOT_READY" + + +@pytest.mark.parametrize("failure", ["checks", "empty_checks", "outcome", "passed", "containers", "private_files"]) +def test_incomplete_producer_report_cannot_satisfy_master_gate(tmp_path: Path, failure: str) -> None: + producer = load_script("p4_hostile_acceptance") + consumer = load_script("master_acceptance") + report = completed_report() + # A stale marker must be removed if a later verification or cleanup fails. + report["completed_phase_gate"] = "P4" + if failure == "checks": + report["checks"]["public_verifier_passed"] = False + elif failure == "empty_checks": + report["checks"] = {} + elif failure == "outcome": + report["run_outcome"] = "INFRA_FAILED" + elif failure == "passed": + report["passed"] = False + elif failure == "containers": + report["cleanup"]["unrelated_container_ids_preserved"] = False + else: + report["cleanup"]["ephemeral_private_files_removed"] = False + output = tmp_path / "p4.json" + + producer.write_acceptance_report(output, report) + + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + assert "completed_phase_gate" not in json.loads(output.read_text()) + + +def test_wrapper_failure_removes_previously_completed_marker(tmp_path: Path) -> None: + producer = load_script("p4_hostile_acceptance") + wrapper = load_script("p4_hostile_acceptance_exact") + consumer = load_script("master_acceptance") + output = tmp_path / "p4.json" + producer.write_acceptance_report(output, completed_report()) + + result = wrapper.enrich_report(output_path=output, harness=producer, capture={}, return_code=1) + + assert result["passed"] is False + assert "completed_phase_gate" not in result + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + + +@pytest.mark.parametrize("scenario", ["success", "nonzero", "malformed", "interrupted"]) +def test_exact_entrypoint_publishes_only_after_complete_enrichment(tmp_path: Path, monkeypatch, scenario: str) -> None: + producer = load_script("p4_hostile_acceptance") + wrapper = load_script("p4_hostile_acceptance_exact") + consumer = load_script("master_acceptance") + output = tmp_path / "p4.json" + capture = { + "result": { + "exit_code": 0, "timed_out": False, "oom_killed": False, + "containment": {"network_none": True}, + "logs": json.dumps({"passed": True, "private_key_loaded": False, + "sbom_valid": True, "admission": {"allowed": True}}), + }, + "artifacts": {}, "execution_limits": {}, + } + if scenario == "malformed": + capture["result"] = None + + def core_main(): + producer.write_acceptance_report(output, completed_report()) + # Even a successful core result is pending while the wrapper is running. + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + if scenario == "interrupted": + raise RuntimeError("simulated interruption after core serialization") + return 1 if scenario == "nonzero" else 0 + + def capture_override(original, *, artifacts, capture: dict): + capture.update(captured) + return original + + captured = capture + monkeypatch.setattr(producer, "main", core_main) + monkeypatch.setattr(wrapper, "load_harness", lambda: producer) + monkeypatch.setattr(wrapper, "parse_wrapper_paths", lambda argv: (tmp_path / "manifest.json", output)) + monkeypatch.setattr(wrapper, "resolve_runner_artifacts", lambda *args: {}) + monkeypatch.setattr(wrapper, "make_execute_container_override", capture_override) + if scenario in {"malformed", "interrupted"}: + with pytest.raises((AttributeError, RuntimeError)): + wrapper.main([]) + else: + assert wrapper.main([]) == (0 if scenario == "success" else 1) + if scenario == "success": + assert consumer._gate("P4", output, consumer._p4)["state"] == "COMPLETE" + else: + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4)