diff --git a/src/format_bench/equivalence_run.py b/src/format_bench/equivalence_run.py index da0ed23..4b46835 100644 --- a/src/format_bench/equivalence_run.py +++ b/src/format_bench/equivalence_run.py @@ -16,7 +16,7 @@ pair_evidence, ) from .fair import expected_rows, operations_for -from .json_contract import strict_json_dumps +from .json_contract import atomic_write_json from .model import Comparability, ExecutionState, transition from .runner import ( Job, @@ -29,7 +29,7 @@ def _write_json(path: Path, payload: dict) -> None: - path.write_text(strict_json_dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + atomic_write_json(path, payload) def run_equivalence( diff --git a/src/format_bench/fair_run.py b/src/format_bench/fair_run.py index ec7d682..cb9d47a 100644 --- a/src/format_bench/fair_run.py +++ b/src/format_bench/fair_run.py @@ -5,7 +5,7 @@ from pathlib import Path from .fair import expected_rows, operations_for -from .json_contract import strict_json_dumps +from .json_contract import atomic_write_json from .model import ExecutionState, Lane, transition from .runner import Job, MeasurementConfig, measurement_metadata, new_results, run_jobs @@ -56,9 +56,7 @@ def run_fair(root: Path, run_dir: Path, config: MeasurementConfig | None = None) # LLM contract precondition: estimand metadata is durable before # ROUNDTRIP_VERIFIED -> BENCHMARKED. run_manifest["measurement"] = measurement_record - manifest_path.write_text( - strict_json_dumps(run_manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) + atomic_write_json(manifest_path, run_manifest) measured = run_jobs(jobs, measurement, root) successful_entries = 0 failed_entries = 0 @@ -99,8 +97,8 @@ def run_fair(root: Path, run_dir: Path, config: MeasurementConfig | None = None) } ) results_path = run_dir / "results.json" - results_path.write_text(strict_json_dumps(results, indent=2, sort_keys=True) + "\n") + atomic_write_json(results_path, results) run_manifest["state"] = run_state run_manifest["profile"] = "fair" - manifest_path.write_text(strict_json_dumps(run_manifest, indent=2, sort_keys=True) + "\n") + atomic_write_json(manifest_path, run_manifest) return results_path diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 637a29f..2857b01 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -2,9 +2,16 @@ from __future__ import annotations +import ctypes +import errno import json import math -from typing import Never, TypedDict, Unpack, cast +import os +import secrets +import stat +import sys +from pathlib import Path +from typing import BinaryIO, Never, TypedDict, Unpack, cast class JsonDumpOptions(TypedDict, total=False): @@ -14,6 +21,139 @@ class JsonDumpOptions(TypedDict, total=False): sort_keys: bool +class _TemporaryOwnershipLost(ValueError): + """The temporary basename no longer identifies this writer's inode.""" + + +def _darwin_has_extended_acl(directory_fd: int) -> bool: + if sys.platform != "darwin": + return False + + libc = ctypes.CDLL(None, use_errno=True) + acl_get_fd = libc.acl_get_fd_np + acl_get_fd.argtypes = (ctypes.c_int, ctypes.c_int) + acl_get_fd.restype = ctypes.c_void_p + acl_free = libc.acl_free + acl_free.argtypes = (ctypes.c_void_p,) + acl_free.restype = ctypes.c_int + + # Darwin : ACL_TYPE_EXTENDED = 0x00000100. + ctypes.set_errno(0) + acl = acl_get_fd(directory_fd, 0x00000100) + if not acl: + error = ctypes.get_errno() + if error in {errno.ENOENT, errno.EOPNOTSUPP}: + return False + raise OSError(error, os.strerror(error)) + try: + return True + finally: + if acl_free(acl) != 0: + error = ctypes.get_errno() + raise OSError(error, os.strerror(error)) + + +def _write_atomic_file(destination: BinaryIO, payload: bytes) -> None: + offset = 0 + while offset < len(payload): + written = destination.write(payload[offset:]) + if written <= 0: + raise OSError("temporary JSON file write made no progress") + offset += written + + +def _flush_atomic_file(destination: BinaryIO) -> None: + destination.flush() + os.fsync(destination.fileno()) + + +def _destination_mode(directory_fd: int, name: str) -> int | None: + try: + destination = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError: + return None + if stat.S_ISLNK(destination.st_mode): + raise ValueError("JSON destination must not be a symbolic link") + if not stat.S_ISREG(destination.st_mode): + raise ValueError("JSON destination must be a regular file") + return stat.S_IMODE(destination.st_mode) + + +def _validate_publication_directory(directory_fd: int) -> None: + directory = os.fstat(directory_fd) + shared_write = directory.st_mode & (stat.S_IWGRP | stat.S_IWOTH) + if directory.st_uid != os.geteuid() or shared_write: + raise PermissionError( + "JSON destination directory must be owned by the effective user " + "and not writable by group or other principals" + ) + if _darwin_has_extended_acl(directory_fd): + raise PermissionError( + "JSON destination directory must not have a macOS extended ACL" + ) + + +def _create_temporary( + directory_fd: int, + destination_name: str, + creation_mode: int, +) -> tuple[int, str]: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW + for _ in range(100): + name = f".{destination_name}.{secrets.token_hex(8)}.tmp" + try: + descriptor = os.open( + name, + flags, + creation_mode, + dir_fd=directory_fd, + ) + except FileExistsError: + continue + return descriptor, name + raise FileExistsError("could not create an exclusive temporary JSON file") + + +def _temporary_name( + directory_fd: int, + temporary_name: str, + temporary_identity: tuple[int, int], +) -> str: + try: + entry = os.stat( + temporary_name, + dir_fd=directory_fd, + follow_symlinks=False, + ) + except FileNotFoundError as error: + raise _TemporaryOwnershipLost( + "temporary JSON file no longer identifies the opened entry" + ) from error + if ( + not stat.S_ISREG(entry.st_mode) + or entry.st_dev != temporary_identity[0] + or entry.st_ino != temporary_identity[1] + ): + raise _TemporaryOwnershipLost( + "temporary JSON file no longer identifies the opened entry" + ) + return temporary_name + + +def _replace_same_directory( + directory_fd: int, + temporary_name: str, + destination_name: str, +) -> None: + _destination_mode(directory_fd, destination_name) + os.replace( + temporary_name, + destination_name, + src_dir_fd=directory_fd, + dst_dir_fd=directory_fd, + ) + + def _reject_nonfinite(token: str) -> Never: raise json.JSONDecodeError( f"non-finite JSON number is not permitted: {token}", @@ -64,3 +204,72 @@ def strict_json_dumps(value: object, **kwargs: Unpack[JsonDumpOptions]) -> str: if "allow_nan" in kwargs: raise TypeError("allow_nan is fixed to False") return json.dumps(value, allow_nan=False, **kwargs) + + +def atomic_write_json(path: Path, value: object) -> None: + """Write one deterministic JSON document through a same-directory replace.""" + payload = (strict_json_dumps(value, indent=2, sort_keys=True) + "\n").encode( + "utf-8" + ) + destination = path.absolute() + destination_parent = destination.parent.resolve(strict=True) + destination_name = destination.name + if not destination_name or destination_name in {".", ".."}: + raise ValueError("JSON destination must name a regular file") + + directory_fd = -1 + descriptor = -1 + try: + directory_fd = os.open( + destination_parent, + os.O_RDONLY | os.O_DIRECTORY, + ) + # LLM contract: DIRECTORY_OPENED -> PRINCIPAL_BOUNDARY_VERIFIED. + # A basename can identify publication only while other OS principals + # cannot rename entries in the retained directory. + _validate_publication_directory(directory_fd) + mode = _destination_mode(directory_fd, destination_name) + descriptor, temporary_name = _create_temporary( + directory_fd, + destination_name, + mode if mode is not None else 0o666, + ) + opened_temporary = os.fstat(descriptor) + temporary_identity = (opened_temporary.st_dev, opened_temporary.st_ino) + temporary_name = _temporary_name( + directory_fd, + temporary_name, + temporary_identity, + ) + final_mode = ( + mode if mode is not None else stat.S_IMODE(opened_temporary.st_mode) + ) + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "wb") as stream: + descriptor = -1 + _write_atomic_file(stream, payload) + _flush_atomic_file(stream) + # LLM contract: PRIVATE_COMPLETE -> PUBLISHED_PRIVATE -> + # PUBLISHED_INTENDED_MODE. The complete unpublished payload stays + # private even when replace fails in a searchable directory. + _replace_same_directory( + directory_fd, + temporary_name, + destination_name, + ) + # A mode-restoration failure is post-publication: new bytes remain + # installed at 0600. A later fsync failure may occur after the + # intended mode is installed; neither failure restores old bytes. + os.fchmod(stream.fileno(), final_mode) + os.fsync(stream.fileno()) + # LLM contract: TEMPORARY_OWNED -> PUBLISHED | RETAINED_FAILED. + # Portable POSIX has no atomic compare-and-unlink operation. A failed + # write retains its narrowed temporary entry instead of risking deletion + # of an unrelated file that concurrently reuses the basename. + finally: + try: + if descriptor >= 0: + os.close(descriptor) + finally: + if directory_fd >= 0: + os.close(directory_fd) diff --git a/src/format_bench/profile_run.py b/src/format_bench/profile_run.py index ae27e12..4a8f6d4 100644 --- a/src/format_bench/profile_run.py +++ b/src/format_bench/profile_run.py @@ -9,7 +9,7 @@ from .claims.tsfile import run_tsfile_claim from .claims.vortex import run_vortex_stress from .formats.lance import build_fts, query_fts -from .json_contract import strict_json_dumps +from .json_contract import atomic_write_json from .model import Comparability, ExecutionState, Lane, TargetTier, transition from .prompt import token_metrics, write_prompt_artifacts from .research import load_research_records @@ -38,12 +38,10 @@ def _finish(root: Path, run_dir: Path, run: dict, profile: Lane, evidence: dict) "results": evidence, } path = run_dir / "results.json" - path.write_text(strict_json_dumps(results, indent=2, sort_keys=True) + "\n", encoding="utf-8") + atomic_write_json(path, results) run["state"] = transition(ExecutionState.ROUNDTRIP_VERIFIED, ExecutionState.BENCHMARKED) run["profile"] = profile - (run_dir / "manifest.json").write_text( - strict_json_dumps(run, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) + atomic_write_json(run_dir / "manifest.json", run) return path diff --git a/src/format_bench/report.py b/src/format_bench/report.py index 3735a03..22f2513 100644 --- a/src/format_bench/report.py +++ b/src/format_bench/report.py @@ -6,7 +6,7 @@ from pathlib import Path from .model import Comparability, ExecutionState, Lane, RobustnessVerdict, transition -from .json_contract import strict_json_dumps +from .json_contract import atomic_write_json, strict_json_dumps from .robustness.summary import summarize_cases @@ -780,6 +780,6 @@ def render_report(run_dir: Path) -> Path: payload["state"] = transition( ExecutionState.BENCHMARKED, ExecutionState.REPORTED ) - json_path.write_bytes(_json_bytes(payload)) + atomic_write_json(json_path, payload) # LLM contract: transition in memory, then persist after durable report output. return path diff --git a/src/format_bench/shards.py b/src/format_bench/shards.py index 0d0db38..9e984f5 100644 --- a/src/format_bench/shards.py +++ b/src/format_bench/shards.py @@ -7,17 +7,11 @@ from typing import TypeAlias, cast from .artifact_digest import artifact_sha256 -from .json_contract import strict_json_dumps +from .json_contract import atomic_write_json, strict_json_dumps from .model import ExecutionState, transition JSONValue: TypeAlias = ( - None - | bool - | int - | float - | str - | list["JSONValue"] - | dict[str, "JSONValue"] + None | bool | int | float | str | list["JSONValue"] | dict[str, "JSONValue"] ) JSONObject = dict[str, JSONValue] @@ -54,7 +48,7 @@ def _object_map(value: JSONValue, label: str) -> dict[str, JSONObject]: def _write_json(path: Path, value: JSONObject) -> None: - path.write_text(strict_json_dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") + atomic_write_json(path, value) def _safe_run_path( @@ -66,12 +60,16 @@ def _safe_run_path( if relative.is_absolute() or ".." in relative.parts: raise ValueError(f"run path must be relative: {label}") candidate = run / relative - if candidate.is_symlink() or any(parent.is_symlink() for parent in candidate.parents): + if candidate.is_symlink() or any( + parent.is_symlink() for parent in candidate.parents + ): raise ValueError(f"run path must not resolve through a symlink: {label}") try: candidate.resolve(strict=not allow_missing).relative_to(run.resolve()) except (FileNotFoundError, ValueError) as error: - raise ValueError(f"run path is missing or escapes run directory: {label}") from error + raise ValueError( + f"run path is missing or escapes run directory: {label}" + ) from error return candidate @@ -127,7 +125,7 @@ def _format_identities(run: Path, manifest: JSONObject) -> dict[str, JSONObject] def _hardlink_tree(source: Path, destination: Path) -> None: if source.is_symlink(): raise ValueError(f"symlink is not allowed in shard input: {source}") - destination.mkdir(parents=True, exist_ok=True) + destination.mkdir(mode=0o700, parents=True, exist_ok=True) for entry in source.iterdir(): target = destination / entry.name if entry.is_dir(): @@ -138,8 +136,26 @@ def _hardlink_tree(source: Path, destination: Path) -> None: raise ValueError(f"unsupported artifact entry: {entry}") +def _mkdir_private_tree(destination: Path) -> None: + missing: list[Path] = [] + current = destination + while not current.exists(): + missing.append(current) + parent = current.parent + if parent == current: + raise FileNotFoundError("shard output has no existing ancestor") + current = parent + if not current.is_dir(): + raise NotADirectoryError(f"shard output ancestor is not a directory: {current}") + # LLM contract: ANCESTOR_DISCOVERED -> PRIVATE_CREATED | FAILED. + # Create each missing level explicitly so pathlib cannot apply umask-only + # parent creation before the private output leaf exists. + for directory in reversed(missing): + directory.mkdir(mode=0o700) + + def _copy_run_files(base_run: Path, output_run: Path) -> None: - output_run.mkdir(parents=True) + _mkdir_private_tree(output_run) for name in ("artifacts", "input"): _hardlink_tree(base_run / name, output_run / name) shutil.copy2(base_run / "manifest.json", output_run / "manifest.json") @@ -237,9 +253,7 @@ def merge_equivalence_shards( if pair in primary_endpoints and primary_endpoints[pair] != endpoint: raise ValueError(f"conflicting primary endpoint: {pair}") primary_endpoints[pair] = endpoint - pairs = _object_map( - equivalence.get("pairs"), f"{shard}/equivalence.pairs" - ) + pairs = _object_map(equivalence.get("pairs"), f"{shard}/equivalence.pairs") if not pairs: raise ValueError(f"shard has no equivalence pair evidence: {shard}") for job_id, evidence in _object_map( diff --git a/src/format_bench/workflow.py b/src/format_bench/workflow.py index 9453085..c8427ca 100644 --- a/src/format_bench/workflow.py +++ b/src/format_bench/workflow.py @@ -10,7 +10,7 @@ from .canonical import canonical_hash, query_counts, read_csv, verify_table from .datasets import load_manifest from .formats.base import Artifact, FormatAdapter -from .json_contract import strict_json_dumps +from .json_contract import atomic_write_json from .model import ExecutionState, transition from .registry import adapter_map, adapters from .runner import environment_info @@ -23,9 +23,7 @@ def _write_json(path: Path, payload: dict) -> None: - path.write_text( - strict_json_dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) + atomic_write_json(path, payload) def _sha256(path: Path) -> str: @@ -143,9 +141,15 @@ def prepare_run( f"expected {manifest['source_sha256']}, got {actual_source_sha256}" ) destination = run_dir or _default_run_dir(root, dataset_id) - destination.mkdir(parents=True, exist_ok=False) + if run_dir is None: + destination.parent.mkdir(mode=0o700, exist_ok=True) + destination.mkdir( + mode=0o700, + parents=run_dir is not None, + exist_ok=False, + ) input_dir = destination / "input" - input_dir.mkdir() + input_dir.mkdir(mode=0o700) table = read_csv(source, manifest) effective = _fixture_manifest(manifest, table, source) if fixture else manifest @@ -155,9 +159,9 @@ def prepare_run( entries = [] artifact_root = destination / "artifacts" - artifact_root.mkdir() + artifact_root.mkdir(mode=0o700) observation_root = destination / ".size-observations" - observation_root.mkdir() + observation_root.mkdir(mode=0o700) _ensure_contained(destination, artifact_root) _ensure_contained(destination, observation_root) # LLM contract: DISCOVERED -> ENCODED -> ROUNDTRIP_VERIFIED -> BENCHMARKED -> REPORTED. @@ -201,7 +205,7 @@ def prepare_run( observation_dir = _ensure_contained( destination, observation_root / format_name ) - observation_dir.mkdir(exist_ok=True) + observation_dir.mkdir(mode=0o700, exist_ok=True) repeated_path = _ensure_contained( destination, observation_dir / f"{attempt_index}{format_extension}", diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index 58c90cf..1e2a3e2 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -1,13 +1,22 @@ import ast import json import math +import os +import stat +import subprocess +import sys from collections.abc import Callable from pathlib import Path -from typing import cast +from typing import BinaryIO, cast import pytest -from format_bench.json_contract import strict_json_dumps, strict_json_loads +from format_bench import json_contract +from format_bench.json_contract import ( + atomic_write_json, + strict_json_dumps, + strict_json_loads, +) def _unsafe_json_writer_lines(source: str, filename: str) -> list[int]: @@ -68,6 +77,550 @@ def test_strict_json_dumps_accepts_finite_numbers() -> None: assert strict_json_dumps({"value": 1.25}) == '{"value": 1.25}' +def test_atomic_write_json_creates_and_replaces_deterministic_document( + tmp_path: Path, +) -> None: + destination = tmp_path / "manifest.json" + + atomic_write_json(destination, {"value": 1}) + assert destination.read_bytes() == b'{\n "value": 1\n}\n' + + atomic_write_json(destination, {"value": 2}) + assert destination.read_bytes() == b'{\n "value": 2\n}\n' + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX umask contract") +@pytest.mark.parametrize(("mask", "expected"), ((0o022, 0o644), (0o077, 0o600))) +def test_atomic_write_json_honors_umask_on_first_write( + tmp_path: Path, + mask: int, + expected: int, +) -> None: + destination = tmp_path / "manifest.json" + + previous_mask = os.umask(mask) + try: + atomic_write_json(destination, {"value": 1}) + finally: + os.umask(previous_mask) + + assert stat.S_IMODE(destination.stat().st_mode) == expected + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX evidence-mode contract") +def test_atomic_write_json_preserves_existing_evidence_mode(tmp_path: Path) -> None: + destination = tmp_path / "manifest.json" + destination.write_bytes(b'{"value":1}\n') + destination.chmod(0o640) + + atomic_write_json(destination, {"value": 2}) + + assert stat.S_IMODE(destination.stat().st_mode) == 0o640 + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX evidence-mode contract") +@pytest.mark.parametrize( + ("mode", "mask"), ((0o600, 0o022), (0o640, 0o022), (0o660, 0o077)) +) +def test_atomic_write_json_never_creates_replacement_broader_than_destination( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: int, + mask: int, +) -> None: + destination = tmp_path / "manifest.json" + destination.write_bytes(b'{"value":1}\n') + destination.chmod(mode) + observed_modes: list[int] = [] + real_fchmod = os.fchmod + + def inspect_initial_mode(descriptor: int, requested_mode: int) -> None: + observed_modes.append(stat.S_IMODE(os.fstat(descriptor).st_mode)) + real_fchmod(descriptor, requested_mode) + + monkeypatch.setattr(json_contract.os, "fchmod", inspect_initial_mode) + previous_mask = os.umask(mask) + try: + atomic_write_json(destination, {"value": 2}) + finally: + os.umask(previous_mask) + + assert len(observed_modes) == 2 + assert observed_modes[0] & ~mode == 0 + assert stat.S_IMODE(destination.stat().st_mode) == mode + + +def test_atomic_write_json_mode_failure_preserves_previous_document( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination = tmp_path / "manifest.json" + previous = b'{"state":"ENCODED"}\n' + destination.write_bytes(previous) + + def fail_mode(*_args: object, **_kwargs: object) -> None: + raise OSError("injected mode failure") + + monkeypatch.setattr(json_contract.os, "fchmod", fail_mode) + + with pytest.raises(OSError, match="injected mode failure"): + atomic_write_json(destination, {"state": "BENCHMARKED"}) + + assert destination.read_bytes() == previous + assert len(list(tmp_path.glob(".manifest.json.*.tmp"))) == 1 + + +def test_atomic_write_json_serialization_failure_preserves_previous_document( + tmp_path: Path, +) -> None: + destination = tmp_path / "manifest.json" + previous = b'{"state":"ENCODED"}\n' + destination.write_bytes(previous) + + with pytest.raises(ValueError, match="Out of range float values"): + atomic_write_json(destination, {"value": math.nan}) + + assert destination.read_bytes() == previous + assert list(tmp_path.glob(".manifest.json.*.tmp")) == [] + + +def test_atomic_file_writer_completes_short_writes() -> None: + class ShortWriter: + def __init__(self) -> None: + self.payload = bytearray() + + def write(self, payload: bytes) -> int: + accepted = min(2, len(payload)) + self.payload.extend(payload[:accepted]) + return accepted + + destination = ShortWriter() + json_contract._write_atomic_file(cast(BinaryIO, destination), b"evidence") + + assert destination.payload == b"evidence" + + +@pytest.mark.parametrize("fault", ("write", "flush", "replace")) +def test_atomic_write_json_failure_preserves_previous_document( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fault: str, +) -> None: + destination = tmp_path / "manifest.json" + previous = b'{"state":"ROUNDTRIP_VERIFIED"}\n' + destination.write_bytes(previous) + + def fail(*_args: object, **_kwargs: object) -> None: + raise OSError(f"injected {fault} failure") + + if fault == "replace": + monkeypatch.setattr(json_contract.os, "replace", fail) + else: + monkeypatch.setattr(json_contract, f"_{fault}_atomic_file", fail) + + with pytest.raises(OSError, match=f"injected {fault} failure"): + atomic_write_json(destination, {"state": "BENCHMARKED"}) + + assert destination.read_bytes() == previous + assert len(list(tmp_path.glob(".manifest.json.*.tmp"))) == 1 + assert ( + stat.S_IMODE(next(tmp_path.glob(".manifest.json.*.tmp")).stat().st_mode) + == 0o600 + ) + + +def test_atomic_write_json_widens_mode_only_after_replace( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination = tmp_path / "manifest.json" + destination.write_bytes(b'{"state":"ENCODED"}\n') + destination.chmod(0o644) + real_replace = json_contract._replace_same_directory + + def inspect_private_completed_temp( + directory_fd: int, + temporary_name: str, + destination_name: str, + ) -> None: + temporary = os.stat( + temporary_name, + dir_fd=directory_fd, + follow_symlinks=False, + ) + assert stat.S_IMODE(temporary.st_mode) == 0o600 + real_replace(directory_fd, temporary_name, destination_name) + + monkeypatch.setattr( + json_contract, + "_replace_same_directory", + inspect_private_completed_temp, + ) + + atomic_write_json(destination, {"state": "REPORTED"}) + + assert destination.read_bytes() == b'{\n "state": "REPORTED"\n}\n' + assert stat.S_IMODE(destination.stat().st_mode) == 0o644 + + +def test_atomic_write_json_post_publish_mode_failure_stays_private( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination = tmp_path / "manifest.json" + destination.write_bytes(b'{"state":"ENCODED"}\n') + destination.chmod(0o644) + calls = 0 + real_fchmod = os.fchmod + + def fail_final_mode(descriptor: int, mode: int) -> None: + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("injected final mode failure") + real_fchmod(descriptor, mode) + + monkeypatch.setattr(json_contract.os, "fchmod", fail_final_mode) + + with pytest.raises(OSError, match="injected final mode failure"): + atomic_write_json(destination, {"state": "REPORTED"}) + + assert destination.read_bytes() == b'{\n "state": "REPORTED"\n}\n' + assert stat.S_IMODE(destination.stat().st_mode) == 0o600 + assert list(tmp_path.glob(".manifest.json.*.tmp")) == [] + + +def test_atomic_write_json_post_publish_fsync_failure_keeps_intended_mode( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination = tmp_path / "manifest.json" + destination.write_bytes(b'{"state":"ENCODED"}\n') + destination.chmod(0o644) + calls = 0 + real_fsync = os.fsync + + def fail_final_fsync(descriptor: int) -> None: + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("injected final fsync failure") + real_fsync(descriptor) + + monkeypatch.setattr(json_contract.os, "fsync", fail_final_fsync) + + with pytest.raises(OSError, match="injected final fsync failure"): + atomic_write_json(destination, {"state": "REPORTED"}) + + assert destination.read_bytes() == b'{\n "state": "REPORTED"\n}\n' + assert stat.S_IMODE(destination.stat().st_mode) == 0o644 + assert list(tmp_path.glob(".manifest.json.*.tmp")) == [] + + +def test_atomic_write_json_rejects_symlink_destination(tmp_path: Path) -> None: + outside = tmp_path / "outside.json" + outside.write_bytes(b'{"outside":true}\n') + destination = tmp_path / "manifest.json" + destination.symlink_to(outside) + + with pytest.raises(ValueError, match="symbolic link"): + atomic_write_json(destination, {"outside": False}) + + assert destination.is_symlink() + assert outside.read_bytes() == b'{"outside":true}\n' + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX directory-mode contract") +@pytest.mark.parametrize("directory_mode", (0o770, 0o707)) +def test_atomic_write_json_rejects_shared_writable_directory( + tmp_path: Path, + directory_mode: int, +) -> None: + destination_dir = tmp_path / "shared" + destination_dir.mkdir() + destination = destination_dir / "manifest.json" + previous = b'{"state":"ENCODED"}\n' + destination.write_bytes(previous) + destination_dir.chmod(directory_mode) + + with pytest.raises(PermissionError, match="not writable by group or other"): + atomic_write_json(destination, {"state": "REPORTED"}) + + assert destination.read_bytes() == previous + assert list(destination_dir.glob(".manifest.json.*.tmp")) == [] + + +@pytest.mark.skipif(sys.platform != "darwin", reason="macOS extended ACL contract") +def test_atomic_write_json_rejects_macos_extended_acl(tmp_path: Path) -> None: + destination_dir = tmp_path / "acl" + destination_dir.mkdir(mode=0o700) + destination = destination_dir / "manifest.json" + previous = b'{"state":"ENCODED"}\n' + destination.write_bytes(previous) + subprocess.run( + [ + "/bin/chmod", + "+a", + "everyone allow add_file,delete_child,search", + str(destination_dir), + ], + check=True, + ) + try: + with pytest.raises(PermissionError, match="extended ACL"): + atomic_write_json(destination, {"state": "REPORTED"}) + finally: + subprocess.run(["/bin/chmod", "-N", str(destination_dir)], check=True) + + assert destination.read_bytes() == previous + assert list(destination_dir.glob(".manifest.json.*.tmp")) == [] + + +def test_atomic_write_json_accepts_same_directory_with_parent_segments( + tmp_path: Path, +) -> None: + destination_dir = tmp_path / "destination" + nested = destination_dir / "nested" + nested.mkdir(parents=True) + destination = nested / ".." / "manifest.json" + + atomic_write_json(destination, {"state": "ENCODED"}) + + assert (destination_dir / "manifest.json").read_bytes() == ( + b'{\n "state": "ENCODED"\n}\n' + ) + + +def test_atomic_write_json_anchors_parent_identity_during_replace( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination_dir = tmp_path / "destination" + nested = destination_dir / "nested" + nested.mkdir(parents=True) + intended = destination_dir / "manifest.json" + intended.write_bytes(b"INTENDED-OLD") + outside = tmp_path / "outside" + outside_child = outside / "child" + outside_child.mkdir(parents=True) + outside_manifest = outside / "manifest.json" + outside_manifest.write_bytes(b"OUTSIDE-OLD") + destination = nested / ".." / "manifest.json" + real_replace = os.replace + + def retarget_parent( + source: str | bytes | Path, + target: str | bytes | Path, + **kwargs: object, + ) -> None: + nested.rmdir() + nested.symlink_to(outside_child, target_is_directory=True) + real_replace(source, target, **kwargs) + + monkeypatch.setattr(json_contract.os, "replace", retarget_parent) + + atomic_write_json(destination, {"state": "BENCHMARKED"}) + + assert intended.read_bytes() == b'{\n "state": "BENCHMARKED"\n}\n' + assert outside_manifest.read_bytes() == b"OUTSIDE-OLD" + + +def test_atomic_write_json_rejects_cross_directory_temp( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination_dir = tmp_path / "destination" + destination_dir.mkdir() + destination = destination_dir / "manifest.json" + previous = b'{"state":"ENCODED"}\n' + destination.write_bytes(previous) + other_dir = tmp_path / "other" + other_dir.mkdir() + temporary_name = ".manifest.json.external.tmp" + same_name_victim = destination_dir / temporary_name + same_name_victim.write_bytes(b"DO-NOT-DELETE") + + def cross_directory_temporary( + directory_fd: int, + destination_name: str, + creation_mode: int, + ) -> tuple[int, str]: + del directory_fd, destination_name, creation_mode + descriptor = os.open( + other_dir / temporary_name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + ) + return descriptor, temporary_name + + monkeypatch.setattr( + json_contract, + "_create_temporary", + cross_directory_temporary, + ) + + with pytest.raises(ValueError, match="no longer identifies"): + atomic_write_json(destination, {"state": "BENCHMARKED"}) + + assert destination.read_bytes() == previous + assert len(list(other_dir.iterdir())) == 1 + assert same_name_victim.read_bytes() == b"DO-NOT-DELETE" + + +def test_atomic_write_json_retries_exclusive_name_collision( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + collision = tmp_path / ".manifest.json.collision.tmp" + collision.write_bytes(b"DO-NOT-TOUCH") + tokens = iter(("collision", "fresh")) + monkeypatch.setattr(json_contract.secrets, "token_hex", lambda _size: next(tokens)) + + destination = tmp_path / "manifest.json" + atomic_write_json(destination, {"state": "ENCODED"}) + + assert collision.read_bytes() == b"DO-NOT-TOUCH" + assert destination.read_bytes() == b'{\n "state": "ENCODED"\n}\n' + assert not (tmp_path / ".manifest.json.fresh.tmp").exists() + + +@pytest.mark.parametrize("raise_after_publish", (False, True)) +def test_atomic_write_json_drops_temporary_name_after_replace( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + raise_after_publish: bool, +) -> None: + token = "published" + temporary_name = f".manifest.json.{token}.tmp" + destination = tmp_path / "manifest.json" + captured_directory_fds: list[int] = [] + real_create = json_contract._create_temporary + real_replace = json_contract._replace_same_directory + + def capture_create( + directory_fd: int, + destination_name: str, + creation_mode: int, + ) -> tuple[int, str]: + return real_create(directory_fd, destination_name, creation_mode) + + def replace_and_reuse( + directory_fd: int, + source_name: str, + destination_name: str, + ) -> None: + captured_directory_fds.append(directory_fd) + real_replace(directory_fd, source_name, destination_name) + reused = os.open( + source_name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=directory_fd, + ) + try: + os.write(reused, b"REUSED") + finally: + os.close(reused) + if raise_after_publish: + raise RuntimeError("injected post-publication failure") + + monkeypatch.setattr(json_contract.secrets, "token_hex", lambda _size: token) + monkeypatch.setattr(json_contract, "_create_temporary", capture_create) + monkeypatch.setattr(json_contract, "_replace_same_directory", replace_and_reuse) + + if raise_after_publish: + with pytest.raises(RuntimeError, match="post-publication failure"): + atomic_write_json(destination, {"state": "REPORTED"}) + else: + atomic_write_json(destination, {"state": "REPORTED"}) + + assert destination.read_bytes() == b'{\n "state": "REPORTED"\n}\n' + assert (tmp_path / temporary_name).read_bytes() == b"REUSED" + for descriptor in captured_directory_fds: + with pytest.raises(OSError): + os.fstat(descriptor) + + +def test_atomic_write_json_retains_failed_temp_without_unlink( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination = tmp_path / "manifest.json" + captured_directory_fds: list[int] = [] + real_create = json_contract._create_temporary + + def capture_create( + directory_fd: int, + destination_name: str, + creation_mode: int, + ) -> tuple[int, str]: + captured_directory_fds.append(directory_fd) + return real_create(directory_fd, destination_name, creation_mode) + + def fail_write(_destination: BinaryIO, _payload: bytes) -> None: + raise OSError("injected write failure") + + def reject_unlink(*_args: object, **_kwargs: object) -> None: + raise AssertionError("failed temporary entries must not be unlinked by name") + + monkeypatch.setattr(json_contract, "_create_temporary", capture_create) + monkeypatch.setattr(json_contract, "_write_atomic_file", fail_write) + monkeypatch.setattr(json_contract.os, "unlink", reject_unlink) + + with pytest.raises(OSError, match="injected write failure"): + atomic_write_json(destination, {"state": "REPORTED"}) + + retained = list(tmp_path.glob(".manifest.json.*.tmp")) + assert len(retained) == 1 + assert stat.S_IMODE(retained[0].stat().st_mode) == 0o600 + for descriptor in captured_directory_fds: + with pytest.raises(OSError): + os.fstat(descriptor) + + +def test_atomic_write_json_preserves_name_reused_during_validation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + token = "reused" + temporary_name = f".manifest.json.{token}.tmp" + destination = tmp_path / "manifest.json" + previous = b'{"state":"ENCODED"}\n' + destination.write_bytes(previous) + real_temporary_name = json_contract._temporary_name + + def replace_before_validation( + directory_fd: int, + name: str, + identity: tuple[int, int], + ) -> str: + os.unlink(name, dir_fd=directory_fd) + replacement = os.open( + name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=directory_fd, + ) + try: + os.write(replacement, b"UNRELATED") + finally: + os.close(replacement) + return real_temporary_name( + directory_fd, + name, + identity, + ) + + monkeypatch.setattr(json_contract.secrets, "token_hex", lambda _size: token) + monkeypatch.setattr(json_contract, "_temporary_name", replace_before_validation) + + with pytest.raises(ValueError, match="no longer identifies"): + atomic_write_json(destination, {"state": "REPORTED"}) + + assert destination.read_bytes() == previous + assert (tmp_path / temporary_name).read_bytes() == b"UNRELATED" + + @pytest.mark.parametrize( "token", ("NaN", "Infinity", "-Infinity", "1e9999", "-1e9999"), diff --git a/tests/test_shards.py b/tests/test_shards.py index c5195ea..39df1c9 100644 --- a/tests/test_shards.py +++ b/tests/test_shards.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import os +import stat from pathlib import Path import pytest @@ -76,9 +78,7 @@ def _shard_results(pair: str, name: str) -> dict: "interval_method": "bootstrap_percentile", "coverage_claim": "none", }, - "primary_endpoints": { - pair: {"scope": "storage", "metric": "native_bytes"} - }, + "primary_endpoints": {pair: {"scope": "storage", "metric": "native_bytes"}}, "multiplicity_control": { "contract_version": "1", "error_control_target": "FWER", @@ -111,8 +111,11 @@ def _shard_results(pair: str, name: str) -> dict: } +@pytest.mark.skipif(os.name != "posix", reason="POSIX directory-mode contract") +@pytest.mark.parametrize("mask", (0o002, 0o000)) def test_merge_equivalence_shards_reuses_artifacts_and_unions_results( tmp_path: Path, + mask: int, ) -> None: base = tmp_path / "base" _write(base / "input" / "manifest.json", {"rows": 4}) @@ -133,11 +136,24 @@ def test_merge_equivalence_shards_reuses_artifacts_and_unions_results( _write(shard / "manifest.json", shard_manifest) _write(shard / "results.json", _shard_results(pair, name)) - output = tmp_path / "merged" - merge_equivalence_shards(base, shard_root, output) + output_parent = tmp_path / "missing" / "nested" + output = output_parent / "merged" + previous_mask = os.umask(mask) + try: + merge_equivalence_shards(base, shard_root, output) + finally: + os.umask(previous_mask) merged = json.loads((output / "results.json").read_text(encoding="utf-8")) manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8")) + for directory in ( + tmp_path / "missing", + output_parent, + output, + output / "input", + output / "artifacts", + ): + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 assert merged["status"] == "MEASURED" assert set(merged["results"]) == {"arrow_ipc/read_all", "feather_v2/read_all"} assert set(merged["equivalence"]["pairs"]) == {"arrow-feather", "csv-tsv"} @@ -145,9 +161,7 @@ def test_merge_equivalence_shards_reuses_artifacts_and_unions_results( "arrow-feather", "csv-tsv", } - assert merged["equivalence"]["multiplicity_control"][ - "planned_comparisons" - ] == 2 + assert merged["equivalence"]["multiplicity_control"]["planned_comparisons"] == 2 assert merged["equivalence"]["storage_estimand"]["numerator"] == ( "candidate_group_median" ) @@ -369,7 +383,9 @@ def test_merge_equivalence_shards_hashes_directory_artifacts(tmp_path: Path) -> merge_equivalence_shards(base, tmp_path / "shards", tmp_path / "merged") -def test_merge_equivalence_shards_rejects_artifact_kind_mismatch(tmp_path: Path) -> None: +def test_merge_equivalence_shards_rejects_artifact_kind_mismatch( + tmp_path: Path, +) -> None: base = tmp_path / "base" _write(base / "input" / "manifest.json", {"rows": 4}) (base / "input" / "source.csv").write_text("id\n1\n", encoding="utf-8") diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 72b8e00..cb0ee9b 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -1,12 +1,14 @@ import json +import os import shutil +import stat from dataclasses import replace from pathlib import Path import pyarrow as pa import pytest -from format_bench import cli +from format_bench import cli, workflow from format_bench.artifact_digest import artifact_sha256 from format_bench.fair import Operation from format_bench.formats.base import Artifact, FormatDescription @@ -169,6 +171,78 @@ def test_prepare_and_verify_fixture_record_relative_evidence(tmp_path: Path) -> ) +@pytest.mark.skipif(os.name != "posix", reason="POSIX directory-mode contract") +@pytest.mark.parametrize("mask", (0o002, 0o000)) +def test_prepare_run_creates_private_lifecycle_directories_under_open_umask( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mask: int, +) -> None: + root = Path(__file__).parents[1] + run_dir = tmp_path / "run" + requested_modes: dict[str, int] = {} + real_mkdir = Path.mkdir + + def record_mode( + path: Path, + mode: int = 0o777, + parents: bool = False, + exist_ok: bool = False, + ) -> None: + if path == run_dir or run_dir in path.parents: + requested_modes.setdefault(str(path.relative_to(run_dir.parent)), mode) + real_mkdir(path, mode=mode, parents=parents, exist_ok=exist_ok) + + monkeypatch.setattr(Path, "mkdir", record_mode) + previous_mask = os.umask(mask) + try: + prepare_run( + root, + DATASET, + run_dir, + fixture=True, + selected=(CsvAdapter(),), + size_observations=2, + ) + finally: + os.umask(previous_mask) + + for directory in (run_dir, run_dir / "input", run_dir / "artifacts"): + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 + assert requested_modes["run/.size-observations"] == 0o700 + assert requested_modes["run/.size-observations/csv"] == 0o700 + assert (run_dir / "input" / "manifest.json").is_file() + assert (run_dir / "manifest.json").is_file() + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX directory-mode contract") +@pytest.mark.parametrize("mask", (0o002, 0o000)) +def test_prepare_run_creates_missing_default_parent_privately( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mask: int, +) -> None: + root = Path(__file__).parents[1] + destination_root = tmp_path / "root" + destination_root.mkdir(mode=0o700) + destination = destination_root / "runs" / "run" + monkeypatch.setattr( + workflow, + "_default_run_dir", + lambda _root, _dataset_id: destination, + ) + + previous_mask = os.umask(mask) + try: + prepared = prepare_run(root, DATASET, fixture=True, selected=()) + finally: + os.umask(previous_mask) + + assert prepared == destination + for directory in (destination.parent, destination): + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 + + def test_artifact_digest_rejects_root_symlink(tmp_path: Path) -> None: target = tmp_path / "target" target.mkdir()