diff --git a/.github/workflows/engineering.yml b/.github/workflows/engineering.yml index ac9921e..9cd158c 100644 --- a/.github/workflows/engineering.yml +++ b/.github/workflows/engineering.yml @@ -109,6 +109,7 @@ jobs: tests/test_production_logging.py \ tests/test_delivery.py \ tests/test_delivery_io.py \ + tests/test_delivery_validation_work.py \ tests/test_delivery_backup.py \ tests/test_delivery_restore_io.py \ tests/test_restore_io_contracts.py \ diff --git a/backend/tests/test_delivery_validation_work.py b/backend/tests/test_delivery_validation_work.py new file mode 100644 index 0000000..708a34a --- /dev/null +++ b/backend/tests/test_delivery_validation_work.py @@ -0,0 +1,191 @@ +"""Ordinary string/worker models, not Root, durability or recovery acceptance.""" + +import importlib +import random +import re +import sys +import threading +import time +from collections.abc import Sequence +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) +policy = importlib.import_module("scripts.release.policy") +file_copy = importlib.import_module("scripts.release.file_copy") + + +def old_path_result(value): + """Frozen pre-optimization lexical policy, independent of the new cache.""" + if not isinstance(value, str) or len(value) > 512: + return "invalid payload path" + if not all(re.fullmatch(r"[A-Za-z0-9_.+@-]{1,255}", p) for p in value.split("/")): + return "noncanonical payload path" + if any( + p in {".", "..", ".git", ".localplan", "docs", "__pycache__", ".env", ".ssh"} + for p in value.split("/") + ): + return "private or unsafe payload path" + if not value.startswith(policy.ROOTS) or value.endswith((".pyc", ".pyo")): + return "payload path outside delivery allowlist" + return None + + +def test_lexical_policy_preserves_boundaries_and_refusal_categories(): + values = [None, True, 1, [], {}, b"backend/a", ""] + components = [ + ".", + "..", + ".git", + ".localplan", + "docs", + "__pycache__", + ".env", + ".ssh", + "a//b", + "a/", + "a.pyc", + "a.pyo", + "a\x00b", + "a\nb", + "é", + "a" * 255, + "a" * 256, + "a" * 250 + "/" + "b" * 250, + "a" * 250 + "/" + "b" * 254, + ] + values.extend("backend/" + component for component in components) + rng = random.Random(20261028) + for _ in range(3000): + value = "".join(rng.choices("abcXYZ019_.+@-/ \\:\n\x00é", k=rng.randrange(90))) + values.append(rng.choice(["backend/", "images/gnu/", "site/x/", ""]) + value) + for value in values: + refused = old_path_result(value) + if refused is None: + assert policy.relative_name(value) == value + else: + with pytest.raises(policy.DeliveryError) as observed: + policy.relative_name(value) + assert str(observed.value) == refused + + +def test_lexical_cache_is_bounded_and_custom_hash_cannot_bypass_policy(): + class Collision(str): + def __hash__(self): + return hash("backend/valid.py") + + def __eq__(self, other): + return True + + policy._cached_relative_name.cache_clear() + try: + policy.relative_name("backend/valid.py") + before = policy._cached_relative_name.cache_info() + with pytest.raises(policy.DeliveryError, match="private or unsafe"): + policy.relative_name(Collision("backend/../escape")) + assert policy._cached_relative_name.cache_info() == before + for index in range(9000): + policy.relative_name("backend/member-" + str(index)) + assert policy._cached_relative_name.cache_info().currsize == 8192 + finally: + policy._cached_relative_name.cache_clear() + + +def test_deadline_stops_dispatch_but_waits_for_running_operations(): + started = threading.Barrier(5) + release = threading.Event() + finished = threading.Event() + active = set() + errors = [] + lock = threading.Lock() + + def operation(index): + with lock: + active.add(index) + try: + started.wait(timeout=5) + assert release.wait(5) + return index + finally: + with lock: + active.remove(index) + + def collect(): + try: + file_copy.bounded_copies(list(range(32)), operation, deadline=time.monotonic() + 0.2) + except BaseException as error: + errors.append(error) + finally: + finished.set() + + collector = threading.Thread(target=collect) + collector.start() + try: + started.wait(timeout=5) + assert not finished.wait(0.3) + with lock: + assert active == {0, 1, 2, 3} + finally: + release.set() + collector.join(5) + assert not collector.is_alive() and not active + assert len(errors) == 1 and isinstance(errors[0], policy.DeliveryError) + assert "copy deadline exceeded" in str(errors[0]) + + +def test_single_operation_cannot_publish_after_its_deadline(monkeypatch): + current = [10.0] + monkeypatch.setattr(file_copy, "time", SimpleNamespace(monotonic=lambda: current[0])) + calls = [] + + def operation(index): + calls.append(index) + current[0] = 12.0 + return "completed after deadline" + + with pytest.raises(policy.DeliveryError, match="copy deadline exceeded"): + file_copy.bounded_copies([0], operation, deadline=11.0) + assert calls == [0] + + +def test_iterator_failure_joins_in_flight_work_without_dispatching_rest(): + started, release, finished = (threading.Event() for _ in range(3)) + calls, errors = [], [] + + class Broken(Sequence): + def __len__(self): + return 32 + + def __getitem__(self, index): + if index == 0: + return index + assert started.wait(5) + raise ValueError("finite inventory refusal") + + def operation(index): + calls.append(index) + started.set() + assert release.wait(5) + return index + + def collect(): + try: + file_copy.bounded_copies(Broken(), operation) + except BaseException as error: + errors.append(error) + finally: + finished.set() + + collector = threading.Thread(target=collect) + collector.start() + try: + assert started.wait(5) + assert not finished.wait(0.05) + finally: + release.set() + collector.join(5) + assert not collector.is_alive() and calls == [0] + assert len(errors) == 1 and str(errors[0]) == "finite inventory refusal" diff --git a/scripts/release/README.md b/scripts/release/README.md index c34d962..9e7cb59 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -464,7 +464,14 @@ earlier formats retain per-file fsync. 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. +identities again before returning. Four persistent workers take one item at a +time from a locked iterator; there is no Future for each file or eagerly queued +inventory. Failure or deadline expiry stops dispatch, and all running calls join +before an error or result returns. Results retain the input order. +Payload path spelling checks use the same immutable allowlist and may retain at +most 8192 successful, length-bounded strings per process. Custom string subclasses +bypass this cache. Byte reads, filesystem topology, ownership and all metadata +checks remain live; no filesystem or execution result enters the lexical cache. During one `start`, `resume` or `restart`, the fixed Root maintenance helper reads the complete bundle once. Its Root startup checks and node-health helpers can diff --git a/scripts/release/file_copy.py b/scripts/release/file_copy.py index a571115..1ee50cb 100644 --- a/scripts/release/file_copy.py +++ b/scripts/release/file_copy.py @@ -3,10 +3,10 @@ import errno import os import stat +import threading import time from collections.abc import Callable, Sequence -from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait -from itertools import islice +from concurrent.futures import FIRST_EXCEPTION, ThreadPoolExecutor, wait from typing import BinaryIO, cast from .policy import MAX_FILE_BYTES, DeliveryError @@ -34,32 +34,58 @@ def bounded_copies[Item, Result]( remaining() if len(rows) < 2: - return [operation(row) for row in rows] + single = [operation(row) for row in rows] + remaining() + return single results: list[Result | None] = [None] * len(rows) source = iter(enumerate(rows)) - with ThreadPoolExecutor(max_workers=min(IN_FLIGHT, len(rows))) as executor: - pending = { - executor.submit(operation, row): index for index, row in islice(source, IN_FLIGHT) - } + lock = threading.Lock() + stopped = False + + def worker() -> None: + nonlocal stopped try: - while pending: - ready, _ = wait(pending, timeout=remaining(), return_when=FIRST_COMPLETED) - if not ready: - raise DeliveryError("copy deadline exceeded; private partial target retained") - # Observe every completed failure before dispatching more work. - completed = [(pending.pop(future), future.result()) for future in ready] - for index, result in completed: - results[index] = result + while True: + with lock: + if stopped: + return + remaining() following = next(source, None) - if following is not None: - index, row = following - remaining() - pending[executor.submit(operation, row)] = index + if following is None: + return + index, row = following + result = operation(row) + with lock: + results[index] = result + except BaseException: + # Stop dispatch before this future publishes failure. Running calls + # are still joined; their partial results cannot become a receipt. + with lock: + stopped = True + raise + + # One future per worker, not per file. The locked iterator still admits + # only four individual operations, without an eagerly queued inventory. + count = min(IN_FLIGHT, len(rows)) + with ThreadPoolExecutor(max_workers=count) as executor: + pending = [] + try: + for _ in range(count): + pending.append(executor.submit(worker)) + ready, incomplete = wait(pending, timeout=remaining(), return_when=FIRST_EXCEPTION) + for future in ready: + future.result() + if incomplete: + raise DeliveryError("copy deadline exceeded; private partial target retained") + remaining() finally: + with lock: + stopped = True for future in pending: future.cancel() - # The executor joins running calls before this function can return - # or propagate an error. No success receipt can outrun file I/O. + # The executor joins every running call even after cancellation or + # deadline expiry. No success receipt can outrun file I/O. + return cast(list[Result], results) diff --git a/scripts/release/policy.py b/scripts/release/policy.py index ebf8d22..d96deb5 100644 --- a/scripts/release/policy.py +++ b/scripts/release/policy.py @@ -3,6 +3,7 @@ import hashlib import json import re +from functools import lru_cache from pathlib import Path from typing import Any @@ -25,7 +26,9 @@ ROOTS = ( "delivery/", "proxy/", ) -FORBIDDEN = {".git", ".localplan", "docs", "__pycache__", ".env", ".ssh"} +FORBIDDEN = frozenset({".git", ".localplan", "docs", "__pycache__", ".env", ".ssh"}) +_UNSAFE_PARTS = FORBIDDEN | {".", ".."} +_PAYLOAD_PATH = re.compile(r"[A-Za-z0-9_.+@-]{1,255}(?:/[A-Za-z0-9_.+@-]{1,255})*") class DeliveryError(ValueError): @@ -57,18 +60,29 @@ def integer(value: Any, lower: int, upper: int, label: str) -> int: return value -def relative_name(value: Any) -> str: - if not isinstance(value, str) or len(value) > 512: - raise DeliveryError("invalid payload path") - if not all(re.fullmatch(r"[A-Za-z0-9_.+@-]{1,255}", p) for p in value.split("/")): +def _relative_name(value: str) -> str: + if _PAYLOAD_PATH.fullmatch(value) is None: raise DeliveryError("noncanonical payload path") - if any(p in {".", "..", *FORBIDDEN} for p in value.split("/")): + if _UNSAFE_PARTS.intersection(value.split("/")): raise DeliveryError("private or unsafe payload path") if not value.startswith(ROOTS) or value.endswith((".pyc", ".pyo")): raise DeliveryError("payload path outside delivery allowlist") return value +# Only immutable lexical policy results are reused. No byte, path existence, +# metadata, ownership, topology or execution decision is cached here. At most +# 8192 already length-bounded strings survive in one process. +_cached_relative_name = lru_cache(maxsize=8192)(_relative_name) + + +def relative_name(value: Any) -> str: + if not isinstance(value, str) or len(value) > 512: + raise DeliveryError("invalid payload path") + # Custom string equality/hash implementations must never address the cache. + return _cached_relative_name(value) if type(value) is str else _relative_name(value) + + def unique_object(pairs): result = {} for key, value in pairs: