diff --git a/backend/tests/test_delivery_io.py b/backend/tests/test_delivery_io.py index 577b2f6..8fd2a29 100644 --- a/backend/tests/test_delivery_io.py +++ b/backend/tests/test_delivery_io.py @@ -8,6 +8,7 @@ import os import sys import threading import time +from collections import Counter from collections.abc import Sequence from pathlib import Path @@ -86,15 +87,82 @@ def ordinary_bundle(tmp_path): def observed_digests(monkeypatch): reads = [] original = manifest.digest + original_payload = manifest._read_payload def observed(path): reads.append(path) return original(path) monkeypatch.setattr(manifest, "digest", observed) + + def payload(row): + reads.append(row[0]) + return original_payload(row) + + monkeypatch.setattr(manifest, "_read_payload", payload) return reads +def test_payload_byte_reads_are_bounded_joined_and_complete(ordinary_bundle, monkeypatch): + # Ordinary owned bytes only. This observes the inventory's actual reads, + # not service permissions, isolation, or VM recovery latency. + bundle, _ = ordinary_bundle + payload = bundle / "payload" + expected = manifest.inventory(payload) + original = manifest._read_payload + release, saturated = threading.Event(), threading.Event() + lock = threading.Lock() + active = maximum = 0 + started, results, errors = [], [], [] + + def read(row): + nonlocal active, maximum + with lock: + started.append(row[0]) + active += 1 + maximum = max(active, maximum) + if active == 4: + saturated.set() + try: + assert release.wait(5) + return original(row) + finally: + with lock: + active -= 1 + + def collect(): + try: + results.extend(manifest.inventory(payload)) + except BaseException as error: + errors.append(error) + + monkeypatch.setattr(manifest, "_read_payload", read) + collector = threading.Thread(target=collect) + collector.start() + try: + assert saturated.wait(5) + with lock: + assert len(started) == maximum == 4 + assert collector.is_alive() and not results + finally: + release.set() + collector.join(10) + assert not collector.is_alive() and not errors and active == 0 + assert results == expected + assert Counter(started) == Counter(payload / row["path"] for row in expected) + + +@pytest.mark.parametrize("index", range(9)) +def test_payload_reader_refuses_identity_drift_model(ordinary_bundle, monkeypatch, index): + bundle, _ = ordinary_bundle + path = bundle / "payload/frontend/index.html" + info = path.lstat() + expected = list(manifest._file_stamp(info)) + expected[index] += 1 + with pytest.raises(policy.DeliveryError, match="changed before byte verification"): + manifest._read_payload((path, tuple(expected), {"path": "frontend/index.html"})) + + def test_same_operation_reuses_bytes_but_not_mutable_metadata(ordinary_bundle, monkeypatch): bundle, expected = ordinary_bundle reads = observed_digests(monkeypatch) @@ -109,7 +177,7 @@ def test_same_operation_reuses_bytes_but_not_mutable_metadata(ordinary_bundle, m assert second["files"] == original_files and reads == first_reads assert manifest._READ_SCOPE.get() is None manifest.verify(bundle, expected) - assert reads == first_reads * 2 + assert Counter(reads) == Counter(first_reads * 2) def test_failed_operation_cannot_keep_a_byte_proof(ordinary_bundle, monkeypatch): @@ -123,7 +191,7 @@ def test_failed_operation_cannot_keep_a_byte_proof(ordinary_bundle, monkeypatch) assert manifest._READ_SCOPE.get() is None with manifest.verification_scope(): manifest.verify(bundle, expected) - assert reads == first_reads * 2 + assert Counter(reads) == Counter(first_reads * 2) @pytest.mark.parametrize("operator", ["factory", "proxy"]) @@ -154,11 +222,11 @@ def test_installer_refusal_reuses_only_its_own_complete_byte_proof( arguments.extend([tmp_path / "unread-certificate", tmp_path / "unread-key"]) with pytest.raises(RuntimeError, match="ordinary installer refusal probe"): entry(*arguments) - assert reads == complete_read + assert Counter(reads) == Counter(complete_read) assert manifest._READ_SCOPE.get() is None assert not source.exists() manifest.verify(bundle, expected) - assert reads == complete_read * 2 + assert Counter(reads) == Counter(complete_read * 2) @pytest.mark.parametrize("index", range(9)) @@ -200,7 +268,7 @@ def test_inherited_scope_owner_model_requires_fresh_bytes(ordinary_bundle, monke first_reads = list(reads) manifest._READ_SCOPE.get().pid = os.getpid() - 1 manifest.verify(bundle, expected) - assert reads == first_reads * 2 + assert Counter(reads) == Counter(first_reads * 2) @pytest.mark.parametrize( @@ -214,7 +282,7 @@ def test_scope_capacity_model_falls_back_to_full_reads(ordinary_bundle, monkeypa manifest.verify(bundle, expected) first_reads = list(reads) manifest.verify(bundle, expected) - assert reads == first_reads * 2 + assert Counter(reads) == Counter(first_reads * 2) @pytest.mark.parametrize("length", [0, 17, 3 * 1024**2 + 17]) @@ -372,7 +440,7 @@ def test_setup_session_runs_real_cli_and_releases_complete_proofs(ordinary_bundl assert manifest._READ_SCOPE.get() is None with setup_session.SetupSession(maximum_seconds=5) as another: another.invoke("scripts.release", arguments) - assert reads == initial * 2 + assert Counter(reads) == Counter(initial * 2) with pytest.raises(policy.DeliveryError, match="single-use"): with operation: pass diff --git a/scripts/release/README.md b/scripts/release/README.md index 2108c81..fd4e72d 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -408,6 +408,12 @@ failure. Further bundles receive full verification; separate commands start fres Fixed-size regular-file copies can use Linux kernel transfer with bounded chunks. Unsupported transfer falls back at the same offsets. Source size/EOF, file fsync and independent destination byte verification remain required. +Public bundle inventories read every payload byte with at most four readers in +flight. Each reader opens without following a final symlink and verifies the +file identity and metadata before and after hashing. The inventory checks those +identities again before returning. All readers are joined on success or failure; +separate processes still obtain their own complete byte proofs. Service limits, +startup deadlines and the recovery target are unchanged. Verified installer and stopped-node recovery inventories can keep at most four file operations in flight. Each file retains its durability and byte checks; directory finalization and success receipts wait for all dispatched work. On diff --git a/scripts/release/manifest.py b/scripts/release/manifest.py index 36a96e9..40165e7 100644 --- a/scripts/release/manifest.py +++ b/scripts/release/manifest.py @@ -1,5 +1,6 @@ """Explicit directory bundle inventories, verified before installation.""" +import hashlib import json import os import re @@ -10,6 +11,7 @@ from dataclasses import dataclass, field from functools import wraps from pathlib import Path +from . import file_copy from .policy import ( FORMAT, MAX_BUNDLE_BYTES, @@ -105,7 +107,11 @@ def _bundle_signature(bundle: Path) -> dict[str, tuple[int, ...]]: add("payload/" + name, info, directory=is_directory) if is_directory: pending.append( - (os.path.join(directory, entry.name), name + "/", identity(info)) + ( + os.path.join(directory, entry.name), + name + "/", + identity(info), + ) ) finally: os.close(descriptor) @@ -144,6 +150,33 @@ def verified_operation(function): return wrapped +def _file_stamp(info: os.stat_result) -> tuple[int, ...]: + return ( + info.st_dev, + info.st_ino, + info.st_mode, + info.st_nlink, + info.st_uid, + info.st_gid, + info.st_size, + info.st_mtime_ns, + info.st_ctime_ns, + ) + + +def _read_payload(row: tuple[Path, tuple[int, ...], dict]) -> dict: + path, expected, record = row + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + with os.fdopen(descriptor, "rb") as stream: + if _file_stamp(os.fstat(stream.fileno())) != expected: + raise DeliveryError("payload file changed before byte verification") + checked = hashlib.file_digest(stream, "sha256").hexdigest() + after = _file_stamp(os.fstat(stream.fileno())) + if after != expected or _file_stamp(path.lstat()) != expected: + raise DeliveryError("payload file changed during byte verification") + return {**record, "sha256": checked} + + def inventory(root: Path, *, path_prefix: str = "") -> list[dict]: """No links/devices or inferred inclusion; bounds apply before hashing.""" if not root.is_dir() or root.is_symlink(): @@ -173,14 +206,23 @@ def inventory(root: Path, *, path_prefix: str = "") -> list[dict]: relative = path.relative_to(root).as_posix() relative_name(path_prefix + relative) entries.append( - { - "path": relative, - "bytes": size, - "sha256": digest(path), - "mode": "0555" if info.st_mode & 0o111 else "0444", - } + ( + path, + _file_stamp(info), + { + "path": relative, + "bytes": size, + "mode": "0555" if info.st_mode & 0o111 else "0444", + }, + ) ) - return sorted(entries, key=lambda x: x["path"]) + # The same finite four-call scheduler used for durable file copies also + # bounds reads. It joins every dispatched reader before returning or raising. + verified = file_copy.bounded_copies(entries, _read_payload) + for path, expected, _ in entries: + if _file_stamp(path.lstat()) != expected: + raise DeliveryError("payload file changed before inventory completion") + return sorted(verified, key=lambda x: x["path"]) def directories(root: Path, *, path_prefix: str = "") -> list[str]: @@ -249,7 +291,13 @@ def validate(data: dict) -> dict: for image in data["images"].values(): fields( image, - {"toolchain_id", "image_kind", "image_revision", "runtime_version", "image_digest"}, + { + "toolchain_id", + "image_kind", + "image_revision", + "runtime_version", + "image_digest", + }, "runtime identity", ) sha256(image["image_digest"])