fix(p4): make hash-pinned public verifier records readable to non-root

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,52 +1246,53 @@ 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",
         "policy": "/input/runner.admission-policy.json",
         "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")
 

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:

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)
 

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())