Proof: seal shared source materializer

proof/region/v1/PROTOCOL.md

@@ -236,13 +236,15 @@ observation. Право на Arb receipt получает не executor, а от
 
 ## Общая граница BUILD
 
-`proof/region/v1/build/input.py` принимает уже нормализованные lane entries,
-кодирует один канонический USTAR и владеет точными input bytes. Он не
-импортирует и не перепроверяет source capability: это ответственность
-потребляющего lane. `SealedInputV1` структурно неизменяем, связывает
-целостность байтов с opaque caller digest и не утверждает recipe либо engine
-semantics. Resource bounds передаёт lane: общий encoder не вводит собственный
-fixture-specific cap.
+`provenance.materialize_admitted_source_files_v1` повторно допускает один
+admitted archive и выдаёт только exact relative regular files. Он не вводит
+USTAR namespace, recipe или engine semantics. Lane выбирает layout и связывает
+собственный aggregate source capability; общий materializer не создаёт generic
+source closure. `proof/region/v1/build/input.py` принимает уже нормализованные
+lane entries, кодирует один канонический USTAR и владеет точными input bytes.
+`SealedInputV1` структурно неизменяем, связывает целостность байтов с opaque
+caller digest и не утверждает recipe либо engine semantics. Resource bounds
+передаёт lane: общий encoder не вводит собственный fixture-specific cap.
 
 `proof/region/v1/build/transport.py` владеет immutable Docker policy,
 одноразовым probe→build lease, bounded stdin/stdout observation, cleanup и

proof/region/v1/arb/pipeline.py

@@ -11,8 +11,6 @@
 from __future__ import annotations
 
 import hashlib
-import io
-import tarfile
 from dataclasses import dataclass, fields
 from enum import StrEnum
 from functools import cached_property
@@ -418,13 +416,20 @@ def _seal_build_input_bundle_v1(
     if not build_transport.docker_policy_is_valid_v1(exact_policy):
         raise TypeError("exact_policy must be canonical DockerBuildPolicyV1")
     source_entries = tuple(
-        entry
+        (
+            f"inputs/{lock.root_prefix[:-1]}/{relative}",
+            mode,
+            contents,
+        )
         for lock, admitted in zip(
             request.source_lock.sources,
             request.admitted_sources.sources,
             strict=True,
         )
-        for entry in _normalized_source_entries_v1(lock, admitted)
+        for relative, mode, contents in provenance.materialize_admitted_source_files_v1(
+            lock,
+            admitted,
+        )
     )
     workspace_entries = tuple(
         (
@@ -509,89 +514,6 @@ def __str__(self) -> str:
         return f"{self.reason.value}: {self.field}"
 
 
-def _normalized_source_entries_v1(
-    lock: provenance.SourceReleaseLockV1,
-    admitted: provenance.SafeSourceArchiveV1,
-) -> tuple[tuple[str, int, bytes], ...]:
-    """Replay Arb-owned source authority into generic canonical-tree entries."""
-
-    def reject(field_name: str) -> NoReturn:
-        raise PipelineInputErrorV1(
-            PipelineInputReasonV1.FOREIGN_SOURCE_CAPABILITY,
-            field_name,
-        )
-
-    if type(lock) is not provenance.SourceReleaseLockV1:
-        reject("lock")
-    if type(admitted) is not provenance.SafeSourceArchiveV1:
-        reject("admitted")
-    try:
-        replayed, raw_tar = provenance.replay_admitted_source_archive_v1(
-            lock,
-            admitted,
-        )
-    except Exception:
-        reject("admitted")
-    if (
-        replayed.source_lock_identity != admitted.source_lock_identity
-        or replayed.archive_sha256 != admitted.archive_sha256
-        or replayed.tree_identity != admitted.tree_identity
-        or replayed.regular_file_count != admitted.regular_file_count
-        or replayed.regular_file_bytes != admitted.regular_file_bytes
-        or replayed.files != admitted.files
-    ):
-        reject("admitted")
-    expected = {item.path: item for item in replayed.files}
-    values: list[tuple[str, int, bytes]] = []
-    seen: set[str] = set()
-    try:
-        with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r:") as archive:
-            for member in archive:
-                if member.isdir():
-                    continue
-                if not member.isreg() or not member.name.startswith(lock.root_prefix):
-                    reject("member")
-                relative = member.name[len(lock.root_prefix) :]
-                coordinate = expected.get(relative)
-                if coordinate is None or relative in seen:
-                    reject("file set")
-                stream = archive.extractfile(member)
-                if stream is None:
-                    reject("regular file")
-                chunks: list[bytes] = []
-                length = 0
-                hasher = hashlib.sha256()
-                while True:
-                    chunk = stream.read(provenance.READ_CHUNK_BYTES)
-                    if not chunk:
-                        break
-                    length += len(chunk)
-                    if length > coordinate.length:
-                        reject("file length")
-                    chunks.append(chunk)
-                    hasher.update(chunk)
-                if (
-                    length != coordinate.length
-                    or hasher.digest() != coordinate.sha256
-                ):
-                    reject("file contents")
-                values.append(
-                    (
-                        f"inputs/{lock.root_prefix[:-1]}/{relative}",
-                        coordinate.mode,
-                        b"".join(chunks),
-                    )
-                )
-                seen.add(relative)
-    except PipelineInputErrorV1:
-        raise
-    except (OSError, tarfile.TarError, ValueError):
-        reject("archive")
-    if seen != set(expected):
-        reject("incomplete archive")
-    return tuple(sorted(values))
-
-
 @dataclass(frozen=True)
 class FlintSourceContentPartitionV1:
     """Structural FLINT archive partition, not an origin assertion.
@@ -1580,7 +1502,6 @@ def build(self, request: PipelineRequestV1) -> BuildResultV1:
             OSError,
             TypeError,
             ValueError,
-            tarfile.TarError,
             BuildSourceAdmissionErrorV1,
             provenance.ProvenanceErrorV1,
             build_input.InputErrorV1,

proof/region/v1/arb/tests/gate.py

@@ -15,9 +15,9 @@
 REPO = Path(__file__).resolve().parents[5]
 sys.path.insert(0, str(REPO))
 EXPECTED_TEST_INVENTORY_SHA256 = (
-    "75462b6e595a3642705ce5135ca6a38b5c634be61b0ef33ea81f73abe17b564b"
+    "721fcceb07c3d73e30032814a181e9c3d86f2185cfb3382cc79b34e05618fa48"
 )
-EXPECTED_TEST_COUNT = 182
+EXPECTED_TEST_COUNT = 184
 _EVALUATOR_REASON = "set LABCOLORS_ARB_EVALUATOR to the controlled C17 binary"
 EXPECTED_SKIPS = frozenset(
     {

proof/region/v1/arb/tests/test_pipeline.py

@@ -831,6 +831,49 @@ def test_input_transport_or_invalid_binary_is_typed_failure(self) -> None:
                 self.assertEqual(result.attempt, 1)
                 self.assertEqual(result.reason, reason)
 
+    def test_forged_source_coordinate_is_rejected_before_build_without_escape(self) -> None:
+        request = _request()
+        source = request.admitted_sources.sources[0]
+        original_tree_identity = source.tree_identity
+        object.__setattr__(source, "tree_identity", _digest("foreign-tree"))
+        backend = _BuildBackend((_static_elf(), _static_elf()))
+        try:
+            result = pipeline.ControlledPipelineV1(build_backend=backend).build(request)
+        finally:
+            object.__setattr__(source, "tree_identity", original_tree_identity)
+
+        self.assertIs(type(result), build_transport.BuildRejectedV1)
+        self.assertEqual(result.attempt, 1)
+        self.assertIs(
+            result.reason,
+            build_transport.BuildFailureReasonV1.CONTRACT_VIOLATION,
+        )
+        self.assertEqual(backend.requests, [])
+
+    def test_hostile_nominal_source_coordinate_is_typed_before_build(self) -> None:
+        request = _request()
+        source = request.source_lock.sources[0]
+        original_length = source.archive_length
+
+        class ExplodingCoordinate:
+            def __ne__(self, _other: object) -> bool:
+                raise RuntimeError("hostile coordinate comparison")
+
+        object.__setattr__(source, "archive_length", ExplodingCoordinate())
+        backend = _BuildBackend((_static_elf(), _static_elf()))
+        try:
+            result = pipeline.ControlledPipelineV1(build_backend=backend).build(request)
+        finally:
+            object.__setattr__(source, "archive_length", original_length)
+
+        self.assertIs(type(result), build_transport.BuildRejectedV1)
+        self.assertEqual(result.attempt, 1)
+        self.assertIs(
+            result.reason,
+            build_transport.BuildFailureReasonV1.CONTRACT_VIOLATION,
+        )
+        self.assertEqual(backend.requests, [])
+
     def test_job_that_exceeds_exact_run_limits_is_rejected_before_build(self) -> None:
         with self.assertRaises(pipeline.PipelineInputErrorV1) as caught:
             _request(

proof/region/v1/arb/tests/test_transport.py

@@ -30,6 +30,7 @@
 from build import input as build_input  # noqa: E402
 from build import transport as build_transport  # noqa: E402
 import pipeline  # noqa: E402
+import provenance  # noqa: E402
 from test_pipeline import (  # noqa: E402
     _docker_capability,
     _probe_native_backend,
@@ -278,15 +279,27 @@ def test_bundle_is_reproducible_normalized_ustar_with_no_host_authority(self) ->
         self.assertEqual(first.sha256, second.sha256)
         self.assertEqual(first.binding_identity, second.binding_identity)
         self.assertTrue(pipeline.arb_input_is_bound_v1(request, policy, first))
+        self.assertEqual(
+            first.sha256.hex(),
+            "5d6e789a721aeed1a8ff023f0af5389711f85f6fe95294d8290b20301235f4df",
+        )
+        self.assertEqual(first.length, 174_080)
 
         source_entries = tuple(
-            entry
+            (
+                f"inputs/{lock.root_prefix[:-1]}/{relative}",
+                mode,
+                contents,
+            )
             for lock, admitted in zip(
                 request.source_lock.sources,
                 request.admitted_sources.sources,
                 strict=True,
             )
-            for entry in pipeline._normalized_source_entries_v1(lock, admitted)
+            for relative, mode, contents in provenance.materialize_admitted_source_files_v1(
+                lock,
+                admitted,
+            )
         )
         workspace_entries = tuple(
             (
@@ -491,8 +504,8 @@ def test_replayed_source_coordinates_must_match_the_admitted_capability(self) ->
         original = admitted.tree_identity
         object.__setattr__(admitted, "tree_identity", _digest("mutated-tree"))
         try:
-            with self.assertRaises(pipeline.PipelineInputErrorV1):
-                pipeline._normalized_source_entries_v1(
+            with self.assertRaises(provenance.ProvenanceErrorV1):
+                provenance.materialize_admitted_source_files_v1(
                     request.source_lock.sources[0],
                     admitted,
                 )