diff --git a/scripts/p4_hostile_acceptance.py b/scripts/p4_hostile_acceptance.py index b17dd0e..07daaab 100644 --- a/scripts/p4_hostile_acceptance.py +++ b/scripts/p4_hostile_acceptance.py @@ -39,7 +39,11 @@ sha256_bytes, to_utc_iso, ) -from echo_certification_forge.p4_manifest import P4ImageManifest, verify_p4_image_manifest # noqa: E402 +from echo_certification_forge.p4_manifest import ( # noqa: E402 + P4ImageManifest, + staged_public_verifier_artifacts, + verify_p4_image_manifest, +) from echo_certification_forge.hostile import ( # noqa: E402 HostileCaseKind, HostileCaseObservation, @@ -1242,6 +1246,7 @@ def public_verifier_probe( image_digest: str, now: datetime, ownership_token: str, + expected_sha256: dict[str, str] | None = None, ) -> dict[str, Any]: targets = { "attestation": "/input/runner.attestation.json", @@ -1249,45 +1254,45 @@ def public_verifier_probe( "sbom": "/input/runner.spdx.json", } require(set(artifacts) == set(targets), f"public verifier artifact set mismatch: {sorted(artifacts)}") - mounts: list[tuple[Path, str, bool]] = [] artifact_identities: dict[str, dict[str, Any]] = {} for name, target in targets.items(): source = artifacts[name].resolve(strict=True) require(source.is_file(), f"public verifier artifact is not a file: {name}") - mounts.append((source, target, True)) artifact_identities[name] = { "source": str(source), "target": target, "sha256": sha256_file(source), "read_only": True, } - result = execute_container( - image=image, - role="verifier", - case="public-verify", - ownership_token=ownership_token, - entrypoint="python", - command=[ - "-m", - "echo_certification_forge.verifier_cli", - "--attestation", - targets["attestation"], - "--policy", - targets["policy"], - "--sbom", - targets["sbom"], - "--image-digest", - image_digest, - "--now", - to_utc_iso(now), - ], - mounts=mounts, - timeout_seconds=60, - memory="256m", - pids=64, - workspace_size="8m", - nofile=256, - ) + with staged_public_verifier_artifacts(artifacts, expected_sha256=expected_sha256) as readable: + mounts = [(readable[name], target, True) for name, target in targets.items()] + result = execute_container( + image=image, + role="verifier", + case="public-verify", + ownership_token=ownership_token, + entrypoint="python", + command=[ + "-m", + "echo_certification_forge.verifier_cli", + "--attestation", + targets["attestation"], + "--policy", + targets["policy"], + "--sbom", + targets["sbom"], + "--image-digest", + image_digest, + "--now", + to_utc_iso(now), + ], + mounts=mounts, + timeout_seconds=60, + memory="256m", + pids=64, + workspace_size="8m", + nofile=256, + ) output = parse_last_json(result["logs"]) passed = ( result["exit_code"] == 0 @@ -1725,6 +1730,11 @@ def main() -> int: image_digest=attestations[ImageRole.RUNNER].identity.image_digest, now=started + timedelta(seconds=1), ownership_token=ownership_token, + expected_sha256={ + "attestation": sealed_records[(ImageRole.RUNNER.value, "a")].attestation.sha256, + "policy": sealed_records[(ImageRole.RUNNER.value, "a")].admission_policy.sha256, + "sbom": sealed_records[(ImageRole.RUNNER.value, "a")].sbom.sha256, + }, ) require(report["service_runtime"]["verifier"]["passed"], "public verifier image failed") diff --git a/scripts/p4_hostile_acceptance_exact.py b/scripts/p4_hostile_acceptance_exact.py index 0d6d61c..72c1297 100644 --- a/scripts/p4_hostile_acceptance_exact.py +++ b/scripts/p4_hostile_acceptance_exact.py @@ -21,6 +21,11 @@ HERE = Path(__file__).resolve() ROOT = HERE.parents[1] LEGACY_HARNESS = ROOT / "scripts" / "p4_hostile_acceptance.py" +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from echo_certification_forge.p4_manifest import staged_public_verifier_artifacts # noqa: E402 def sha256_file(path: Path) -> str: @@ -44,7 +49,11 @@ def parse_wrapper_paths(arguments: list[str]) -> tuple[Path, Path]: return parsed.sealed_manifest.resolve(strict=True), parsed.output.resolve() -def resolve_runner_artifacts(harness: Any, sealed_manifest_path: Path) -> dict[str, Path]: +def resolve_runner_artifacts( + harness: Any, + sealed_manifest_path: Path, + expected_sha256_out: dict[str, str] | None = None, +) -> dict[str, Path]: document = harness.P4ImageManifest.model_validate_json( sealed_manifest_path.read_text(encoding="utf-8") ) @@ -76,6 +85,9 @@ def resolve_runner_artifacts(harness: Any, sealed_manifest_path: Path) -> dict[s raise RuntimeError( f"public verifier artifact hash mismatch: {name}:expected={expected[name]}:observed={observed}" ) + if expected_sha256_out is not None: + expected_sha256_out.clear() + expected_sha256_out.update(expected) return artifacts @@ -84,6 +96,7 @@ def make_execute_container_override( *, artifacts: dict[str, Path], capture: dict[str, Any], + expected_sha256: dict[str, str] | None = None, ) -> Callable[..., dict[str, Any]]: targets = { "attestation": "/input/runner.attestation.json", @@ -96,42 +109,41 @@ def make_execute_container_override( def execute_container_exact(**kwargs: Any) -> dict[str, Any]: if kwargs.get("role") != "verifier" or kwargs.get("case") != "public-verify": return original(**kwargs) - mounts = [ - (artifacts[name].resolve(strict=True), target, True) - for name, target in targets.items() - ] - kwargs.update( - mounts=mounts, - timeout_seconds=60, - memory="256m", - pids=64, - workspace_size="8m", - nofile=256, - ) - result = original(**kwargs) - capture.clear() - capture.update( - { - "result": result, - "artifacts": { - name: { - "source": str(artifacts[name]), - "target": targets[name], - "sha256": sha256_file(artifacts[name]), - "read_only": True, - } - for name in targets - }, - "execution_limits": { - "timeout_seconds": 60, - "memory": "256m", - "pids": 64, - "workspace_size": "8m", - "nofile": 256, - }, - "private_key_mounted": False, - } - ) + with staged_public_verifier_artifacts(artifacts, expected_sha256=expected_sha256) as readable: + mounts = [(readable[name], target, True) for name, target in targets.items()] + kwargs.update( + mounts=mounts, + timeout_seconds=60, + memory="256m", + pids=64, + workspace_size="8m", + nofile=256, + ) + result = original(**kwargs) + capture.clear() + capture.update( + { + "result": result, + "artifacts": { + name: { + "source": str(artifacts[name]), + "target": targets[name], + "sha256": sha256_file(readable[name]), + "mount_mode": "0444", + "read_only": True, + } + for name in targets + }, + "execution_limits": { + "timeout_seconds": 60, + "memory": "256m", + "pids": 64, + "workspace_size": "8m", + "nofile": 256, + }, + "private_key_mounted": False, + } + ) return result return execute_container_exact @@ -213,12 +225,14 @@ 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() - artifacts = resolve_runner_artifacts(harness, sealed_manifest_path) + expected_sha256: dict[str, str] = {} + artifacts = resolve_runner_artifacts(harness, sealed_manifest_path, expected_sha256) capture: dict[str, Any] = {} harness.execute_container = make_execute_container_override( harness.execute_container, artifacts=artifacts, capture=capture, + expected_sha256=expected_sha256, ) original_argv = sys.argv try: diff --git a/src/echo_certification_forge/p4_manifest.py b/src/echo_certification_forge/p4_manifest.py index cbab638..215b4b5 100644 --- a/src/echo_certification_forge/p4_manifest.py +++ b/src/echo_certification_forge/p4_manifest.py @@ -3,9 +3,13 @@ import hashlib import json +import shutil +import stat +import tempfile +from contextlib import contextmanager from datetime import UTC, datetime from pathlib import Path -from typing import Any, Literal +from typing import Any, Iterator, Literal, Mapping from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -21,6 +25,11 @@ P4_ROLES = tuple(role.value for role in ImageRole) P4_VARIANTS = ("a", "b") +PUBLIC_VERIFIER_FILES = { + "attestation": "runner.attestation.json", + "policy": "runner.admission-policy.json", + "sbom": "runner.spdx.json", +} def sha256_file(path: Path) -> str: @@ -31,6 +40,48 @@ def sha256_file(path: Path) -> str: return digest.hexdigest() +@contextmanager +def staged_public_verifier_artifacts( + artifacts: Mapping[str, Path], + *, + expected_sha256: Mapping[str, str] | None = None, +) -> Iterator[dict[str, Path]]: + """Expose only hash-pinned public records to an unprivileged verifier. + + Sealed owner records may be mode 0600 under a restrictive umask. Docker + preserves that mode on a read-only file bind, so UID 65532 cannot open the + file. Make short-lived, mode-0444 copies without modifying sealed originals + or copying the private signing key. Remove them after container execution. + """ + if set(artifacts) != set(PUBLIC_VERIFIER_FILES): + raise ValueError("public verifier artifact set mismatch") + if expected_sha256 is not None and set(expected_sha256) != set(PUBLIC_VERIFIER_FILES): + raise ValueError("public verifier expected-hash set mismatch") + + with tempfile.TemporaryDirectory(prefix="certforge-p4-public-") as directory: + staging = Path(directory) + staging.chmod(0o755) + copies: dict[str, Path] = {} + for name, filename in PUBLIC_VERIFIER_FILES.items(): + source = Path(artifacts[name]).resolve(strict=True) + if not source.is_file(): + raise RuntimeError(f"public verifier artifact is not a file: {name}") + original_hash = sha256_file(source) + if expected_sha256 is not None and original_hash != expected_sha256[name]: + raise RuntimeError(f"public verifier source hash mismatch: {name}") + copy = staging / filename + shutil.copyfile(source, copy) + copy.chmod(0o444) + if ( + stat.S_IMODE(copy.stat().st_mode) != 0o444 + or sha256_file(copy) != original_hash + or sha256_file(source) != original_hash + ): + raise RuntimeError(f"public verifier staged artifact mismatch: {name}") + copies[name] = copy + yield copies + + class P4ArtifactRef(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) diff --git a/tests/test_p4_harness_diagnostics.py b/tests/test_p4_harness_diagnostics.py index 847113b..0be79a4 100644 --- a/tests/test_p4_harness_diagnostics.py +++ b/tests/test_p4_harness_diagnostics.py @@ -3,6 +3,7 @@ import importlib.util from datetime import UTC, datetime import json +import stat import subprocess from pathlib import Path @@ -149,11 +150,16 @@ def test_public_verifier_probe_uses_only_exact_read_only_artifacts(tmp_path: Pat for name in ("attestation", "policy", "sbom"): path = tmp_path / f"{name}.json" path.write_text("{}", encoding="utf-8") + path.chmod(0o600) artifacts[name] = path captured = {} def fake_execute_container(**kwargs): captured.update(kwargs) + for path, _target, read_only in kwargs["mounts"]: + assert read_only is True + assert path.read_text(encoding="utf-8") == "{}" + assert stat.S_IMODE(path.stat().st_mode) == 0o444 return { "exit_code": 0, "timed_out": False, @@ -189,11 +195,13 @@ def fake_execute_container(**kwargs): assert captured["pids"] == 64 assert captured["workspace_size"] == "8m" assert captured["nofile"] == 256 - assert captured["mounts"] == [ - (artifacts["attestation"].resolve(), "/input/runner.attestation.json", True), - (artifacts["policy"].resolve(), "/input/runner.admission-policy.json", True), - (artifacts["sbom"].resolve(), "/input/runner.spdx.json", True), + assert [(path.name, target, readonly) for path, target, readonly in captured["mounts"]] == [ + ("runner.attestation.json", "/input/runner.attestation.json", True), + ("runner.admission-policy.json", "/input/runner.admission-policy.json", True), + ("runner.spdx.json", "/input/runner.spdx.json", True), ] + assert all(not path.exists() for path, _target, _readonly in captured["mounts"]) + assert all(stat.S_IMODE(path.stat().st_mode) == 0o600 for path in artifacts.values()) assert all(item["read_only"] for item in result["artifacts"].values()) diff --git a/tests/test_p4_hostile_acceptance_exact.py b/tests/test_p4_hostile_acceptance_exact.py index 43b24d9..89d2295 100644 --- a/tests/test_p4_hostile_acceptance_exact.py +++ b/tests/test_p4_hostile_acceptance_exact.py @@ -2,6 +2,7 @@ import importlib.util import json +import stat from pathlib import Path from types import SimpleNamespace @@ -39,11 +40,17 @@ def test_override_uses_only_three_exact_read_only_artifacts(tmp_path: Path) -> N for name in ("attestation", "policy", "sbom"): path = tmp_path / f"{name}.json" path.write_text("{}", encoding="utf-8") + path.chmod(0o600) artifacts[name] = path observed: dict[str, object] = {} def original(**kwargs): observed.update(kwargs) + for path, _target, read_only in kwargs["mounts"]: + assert read_only is True + assert path.read_text(encoding="utf-8") == "{}" + assert stat.S_IMODE(path.stat().st_mode) == 0o444 + assert stat.S_IMODE(path.parent.stat().st_mode) == 0o755 return { "exit_code": 0, "timed_out": False, @@ -70,11 +77,13 @@ def original(**kwargs): ) assert result["exit_code"] == 0 - assert observed["mounts"] == [ - (artifacts["attestation"].resolve(), "/input/runner.attestation.json", True), - (artifacts["policy"].resolve(), "/input/runner.admission-policy.json", True), - (artifacts["sbom"].resolve(), "/input/runner.spdx.json", True), + assert [(path.name, target, readonly) for path, target, readonly in observed["mounts"]] == [ + ("runner.attestation.json", "/input/runner.attestation.json", True), + ("runner.admission-policy.json", "/input/runner.admission-policy.json", True), + ("runner.spdx.json", "/input/runner.spdx.json", True), ] + assert all(not path.exists() for path, _target, _readonly in observed["mounts"]) + assert all(stat.S_IMODE(path.stat().st_mode) == 0o600 for path in artifacts.values()) assert observed["timeout_seconds"] == 60 assert observed["memory"] == "256m" assert observed["pids"] == 64 @@ -82,6 +91,30 @@ def original(**kwargs): assert observed["nofile"] == 256 assert capture["private_key_mounted"] is False assert all(item["read_only"] for item in capture["artifacts"].values()) + assert all(item["mount_mode"] == "0444" for item in capture["artifacts"].values()) + + +def test_override_rejects_hash_drift_before_verifier_execution(tmp_path: Path) -> None: + wrapper = load_wrapper() + artifacts = {} + expected = {} + for name in ("attestation", "policy", "sbom"): + path = tmp_path / f"{name}.json" + path.write_text("{}", encoding="utf-8") + path.chmod(0o600) + artifacts[name] = path + expected[name] = wrapper.sha256_file(path) + expected["attestation"] = "0" * 64 + calls = [] + override = wrapper.make_execute_container_override( + lambda **kwargs: calls.append(kwargs), + artifacts=artifacts, + expected_sha256=expected, + capture={}, + ) + with pytest.raises(RuntimeError, match="source hash mismatch: attestation"): + override(role="verifier", case="public-verify", mounts=[]) + assert calls == [] def test_override_does_not_change_non_verifier_calls(tmp_path: Path) -> None: diff --git a/tests/test_p4_manifest.py b/tests/test_p4_manifest.py index c811949..2779da2 100644 --- a/tests/test_p4_manifest.py +++ b/tests/test_p4_manifest.py @@ -1,14 +1,18 @@ from __future__ import annotations import json +import stat from datetime import UTC, datetime, timedelta from pathlib import Path +import pytest + from echo_certification_forge.p4_manifest import ( P4ArtifactRef, P4ImageManifest, P4ImageRecord, sha256_file, + staged_public_verifier_artifacts, verify_p4_image_manifest, write_manifest, ) @@ -215,3 +219,39 @@ def test_manifest_rejects_duplicate_immutable_image_identity(tmp_path: Path) -> assert "unexpected duplicate image_id" in str(exc) else: raise AssertionError("duplicate image identity was accepted") + + +def test_public_verifier_staging_handles_mode_0600_without_mutating_seals(tmp_path: Path) -> None: + artifacts = {} + for name in ("attestation", "policy", "sbom"): + path = tmp_path / f"sealed-{name}.json" + path.write_text('{"public":"synthetic"}\n', encoding="utf-8") + path.chmod(0o600) + artifacts[name] = path + expected = {name: sha256_file(path) for name, path in artifacts.items()} + + with staged_public_verifier_artifacts(artifacts, expected_sha256=expected) as copies: + paths = list(copies.values()) + assert len(paths) == 3 + for name, path in copies.items(): + assert stat.S_IMODE(path.stat().st_mode) == 0o444 + assert stat.S_IMODE(path.parent.stat().st_mode) == 0o755 + assert sha256_file(path) == expected[name] + assert all(not path.exists() for path in paths) + assert all(stat.S_IMODE(path.stat().st_mode) == 0o600 for path in artifacts.values()) + + +def test_public_verifier_staging_rejects_unpinned_and_extra_artifacts(tmp_path: Path) -> None: + artifacts = {} + for name in ("attestation", "policy", "sbom"): + path = tmp_path / f"{name}.json" + path.write_text("{}", encoding="utf-8") + artifacts[name] = path + expected = {name: sha256_file(path) for name, path in artifacts.items()} + expected["policy"] = "0" * 64 + with pytest.raises(RuntimeError, match="source hash mismatch: policy"): + with staged_public_verifier_artifacts(artifacts, expected_sha256=expected): + pass + with pytest.raises(ValueError, match="artifact set mismatch"): + with staged_public_verifier_artifacts({**artifacts, "private_key": tmp_path / "key"}): + pass