From a8b2a09c7b9d5317c74ad96b6bed5faba8c6c358 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:05:06 +0900 Subject: [PATCH 01/15] fix: persist lifecycle JSON atomically (#475) --- src/format_bench/equivalence_run.py | 4 +- src/format_bench/fair_run.py | 10 +-- src/format_bench/json_contract.py | 65 ++++++++++++++- src/format_bench/profile_run.py | 8 +- src/format_bench/report.py | 4 +- src/format_bench/shards.py | 4 +- src/format_bench/workflow.py | 6 +- tests/test_json_contract.py | 121 +++++++++++++++++++++++++++- 8 files changed, 198 insertions(+), 24 deletions(-) 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..10b1b93 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -4,7 +4,10 @@ import json import math -from typing import Never, TypedDict, Unpack, cast +import os +import tempfile +from pathlib import Path +from typing import BinaryIO, Never, TypedDict, Unpack, cast class JsonDumpOptions(TypedDict, total=False): @@ -14,6 +17,30 @@ class JsonDumpOptions(TypedDict, total=False): sort_keys: bool +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 _replace_same_directory(temporary: Path, destination: Path) -> None: + if temporary.parent != destination.parent: + raise ValueError( + "temporary JSON file must use the destination's same directory" + ) + if destination.is_symlink(): + raise ValueError("JSON destination must not be a symbolic link") + os.replace(temporary, destination) + + def _reject_nonfinite(token: str) -> Never: raise json.JSONDecodeError( f"non-finite JSON number is not permitted: {token}", @@ -64,3 +91,39 @@ 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() + if destination.is_symlink(): + raise ValueError("JSON destination must not be a symbolic link") + if destination.exists() and not destination.is_file(): + raise ValueError("JSON destination must be a regular file") + + descriptor = -1 + temporary: Path | None = None + try: + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{destination.name}.", + suffix=".tmp", + dir=destination.parent, + ) + temporary = Path(temporary_name) + with os.fdopen(descriptor, "wb") as stream: + descriptor = -1 + _write_atomic_file(stream, payload) + _flush_atomic_file(stream) + # LLM contract: DISCOVERED -> ENCODED -> ROUNDTRIP_VERIFIED -> + # BENCHMARKED -> REPORTED becomes persistent only at this replace boundary. + # os.replace is atomic on success; file fsync does not claim directory or + # power-loss durability. See https://docs.python.org/3/library/os.html#os.replace + _replace_same_directory(temporary, destination) + finally: + if descriptor >= 0: + os.close(descriptor) + if temporary is not None: + temporary.unlink(missing_ok=True) 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..4bb1959 100644 --- a/src/format_bench/shards.py +++ b/src/format_bench/shards.py @@ -7,7 +7,7 @@ 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 = ( @@ -54,7 +54,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( diff --git a/src/format_bench/workflow.py b/src/format_bench/workflow.py index 9453085..be2eecb 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: diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index 58c90cf..b0928b1 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -1,13 +1,19 @@ import ast import json import math +import tempfile 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 +74,117 @@ 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' + + +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 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' + + +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() + real_mkstemp = tempfile.mkstemp + + def cross_directory_mkstemp( + *, + prefix: str, + suffix: str, + dir: str | Path, + ) -> tuple[int, str]: + del dir + return real_mkstemp(prefix=prefix, suffix=suffix, dir=other_dir) + + monkeypatch.setattr(json_contract.tempfile, "mkstemp", cross_directory_mkstemp) + + with pytest.raises(ValueError, match="same directory"): + atomic_write_json(destination, {"state": "BENCHMARKED"}) + + assert destination.read_bytes() == previous + assert list(other_dir.iterdir()) == [] + + @pytest.mark.parametrize( "token", ("NaN", "Infinity", "-Infinity", "1e9999", "-1e9999"), From 40a2e1a6a7ca168c26add90e23c16afab4e31d53 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:15:18 +0900 Subject: [PATCH 02/15] fix: compare atomic writer directories by identity (#475) --- src/format_bench/json_contract.py | 2 +- tests/test_json_contract.py | 15 +++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 10b1b93..e6a281d 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -32,7 +32,7 @@ def _flush_atomic_file(destination: BinaryIO) -> None: def _replace_same_directory(temporary: Path, destination: Path) -> None: - if temporary.parent != destination.parent: + if not temporary.parent.samefile(destination.parent): raise ValueError( "temporary JSON file must use the destination's same directory" ) diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index b0928b1..cbdac94 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -154,6 +154,21 @@ def test_atomic_write_json_rejects_symlink_destination(tmp_path: Path) -> None: assert outside.read_bytes() == b'{"outside":true}\n' +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_rejects_cross_directory_temp( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From f3f4815e85f3a81233641291d9c53c6acb02d86b Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:24:05 +0900 Subject: [PATCH 03/15] fix: anchor atomic JSON publication (#478 #479) --- src/format_bench/json_contract.py | 76 ++++++++++++++++++++++++++----- tests/test_json_contract.py | 70 ++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+), 12 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index e6a281d..3490f79 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -5,6 +5,7 @@ import json import math import os +import stat import tempfile from pathlib import Path from typing import BinaryIO, Never, TypedDict, Unpack, cast @@ -31,14 +32,49 @@ def _flush_atomic_file(destination: BinaryIO) -> None: os.fsync(destination.fileno()) -def _replace_same_directory(temporary: Path, destination: Path) -> None: - if not temporary.parent.samefile(destination.parent): +def _destination_mode(directory_fd: int, name: str) -> int: + try: + destination = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError: + return 0o644 + 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 _temporary_name(directory_fd: int, descriptor: int, temporary: Path) -> str: + try: + entry = os.stat(temporary.name, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError as error: raise ValueError( "temporary JSON file must use the destination's same directory" + ) from error + opened = os.fstat(descriptor) + if ( + not stat.S_ISREG(entry.st_mode) + or entry.st_dev != opened.st_dev + or entry.st_ino != opened.st_ino + ): + raise ValueError( + "temporary JSON file must be a regular same-directory entry" ) - if destination.is_symlink(): - raise ValueError("JSON destination must not be a symbolic link") - os.replace(temporary, destination) + 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: @@ -99,20 +135,29 @@ def atomic_write_json(path: Path, value: object) -> None: "utf-8" ) destination = path.absolute() - if destination.is_symlink(): - raise ValueError("JSON destination must not be a symbolic link") - if destination.exists() and not destination.is_file(): - raise ValueError("JSON destination must be a regular file") + 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 temporary: Path | None = None + temporary_name: str | None = None try: + directory_fd = os.open( + destination_parent, + os.O_RDONLY | os.O_DIRECTORY, + ) + mode = _destination_mode(directory_fd, destination_name) descriptor, temporary_name = tempfile.mkstemp( prefix=f".{destination.name}.", suffix=".tmp", - dir=destination.parent, + dir=destination_parent, ) temporary = Path(temporary_name) + temporary_name = _temporary_name(directory_fd, descriptor, temporary) + os.fchmod(descriptor, mode) with os.fdopen(descriptor, "wb") as stream: descriptor = -1 _write_atomic_file(stream, payload) @@ -121,9 +166,16 @@ def atomic_write_json(path: Path, value: object) -> None: # BENCHMARKED -> REPORTED becomes persistent only at this replace boundary. # os.replace is atomic on success; file fsync does not claim directory or # power-loss durability. See https://docs.python.org/3/library/os.html#os.replace - _replace_same_directory(temporary, destination) + _replace_same_directory(directory_fd, temporary_name, destination_name) finally: if descriptor >= 0: os.close(descriptor) - if temporary is not None: + if temporary_name is not None and directory_fd >= 0: + try: + os.unlink(temporary_name, dir_fd=directory_fd) + except FileNotFoundError: + pass + elif temporary is not None: temporary.unlink(missing_ok=True) + if directory_fd >= 0: + os.close(directory_fd) diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index cbdac94..e649f4b 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -1,6 +1,8 @@ import ast import json import math +import os +import stat import tempfile from collections.abc import Callable from pathlib import Path @@ -86,6 +88,40 @@ def test_atomic_write_json_creates_and_replaces_deterministic_document( assert destination.read_bytes() == b'{\n "value": 2\n}\n' +@pytest.mark.skipif(os.name != "posix", reason="POSIX evidence-mode contract") +def test_atomic_write_json_declares_and_preserves_evidence_mode( + tmp_path: Path, +) -> None: + destination = tmp_path / "manifest.json" + + atomic_write_json(destination, {"value": 1}) + assert stat.S_IMODE(destination.stat().st_mode) == 0o644 + + destination.chmod(0o640) + atomic_write_json(destination, {"value": 2}) + assert stat.S_IMODE(destination.stat().st_mode) == 0o640 + + +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 list(tmp_path.glob(".manifest.json.*.tmp")) == [] + + def test_atomic_write_json_serialization_failure_preserves_previous_document( tmp_path: Path, ) -> None: @@ -169,6 +205,40 @@ def test_atomic_write_json_accepts_same_directory_with_parent_segments( ) +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, From c6c7bfd0b90fa1a4155a27e7986c44ed6dd89f85 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:31:40 +0900 Subject: [PATCH 04/15] fix: honor umask for atomic JSON creation (#481) --- src/format_bench/json_contract.py | 39 +++++++++++++----- tests/test_json_contract.py | 66 +++++++++++++++++++++++++------ 2 files changed, 83 insertions(+), 22 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 3490f79..d79f0b1 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -5,8 +5,8 @@ import json import math import os +import secrets import stat -import tempfile from pathlib import Path from typing import BinaryIO, Never, TypedDict, Unpack, cast @@ -32,11 +32,11 @@ def _flush_atomic_file(destination: BinaryIO) -> None: os.fsync(destination.fileno()) -def _destination_mode(directory_fd: int, name: str) -> int: +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 0o644 + 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): @@ -44,6 +44,27 @@ def _destination_mode(directory_fd: int, name: str) -> int: return stat.S_IMODE(destination.st_mode) +def _create_temporary( + directory_fd: int, + directory: Path, + destination_name: str, +) -> tuple[int, Path]: + 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, + 0o666, + dir_fd=directory_fd, + ) + except FileExistsError: + continue + return descriptor, directory / name + raise FileExistsError("could not create an exclusive temporary JSON file") + + def _temporary_name(directory_fd: int, descriptor: int, temporary: Path) -> str: try: entry = os.stat(temporary.name, dir_fd=directory_fd, follow_symlinks=False) @@ -150,14 +171,14 @@ def atomic_write_json(path: Path, value: object) -> None: os.O_RDONLY | os.O_DIRECTORY, ) mode = _destination_mode(directory_fd, destination_name) - descriptor, temporary_name = tempfile.mkstemp( - prefix=f".{destination.name}.", - suffix=".tmp", - dir=destination_parent, + descriptor, temporary = _create_temporary( + directory_fd, + destination_parent, + destination_name, ) - temporary = Path(temporary_name) temporary_name = _temporary_name(directory_fd, descriptor, temporary) - os.fchmod(descriptor, mode) + if mode is not None: + os.fchmod(descriptor, mode) with os.fdopen(descriptor, "wb") as stream: descriptor = -1 _write_atomic_file(stream, payload) diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index e649f4b..a6ebe9f 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -88,17 +88,32 @@ def test_atomic_write_json_creates_and_replaces_deterministic_document( assert destination.read_bytes() == b'{\n "value": 2\n}\n' -@pytest.mark.skipif(os.name != "posix", reason="POSIX evidence-mode contract") -def test_atomic_write_json_declares_and_preserves_evidence_mode( +@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" - atomic_write_json(destination, {"value": 1}) - assert stat.S_IMODE(destination.stat().st_mode) == 0o644 + 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 @@ -252,16 +267,24 @@ def test_atomic_write_json_rejects_cross_directory_temp( other_dir.mkdir() real_mkstemp = tempfile.mkstemp - def cross_directory_mkstemp( - *, - prefix: str, - suffix: str, - dir: str | Path, - ) -> tuple[int, str]: - del dir - return real_mkstemp(prefix=prefix, suffix=suffix, dir=other_dir) + def cross_directory_temporary( + directory_fd: int, + directory: Path, + destination_name: str, + ) -> tuple[int, Path]: + del directory_fd, directory + descriptor, name = real_mkstemp( + prefix=f".{destination_name}.", + suffix=".tmp", + dir=other_dir, + ) + return descriptor, Path(name) - monkeypatch.setattr(json_contract.tempfile, "mkstemp", cross_directory_mkstemp) + monkeypatch.setattr( + json_contract, + "_create_temporary", + cross_directory_temporary, + ) with pytest.raises(ValueError, match="same directory"): atomic_write_json(destination, {"state": "BENCHMARKED"}) @@ -270,6 +293,23 @@ def cross_directory_mkstemp( assert list(other_dir.iterdir()) == [] +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( "token", ("NaN", "Infinity", "-Infinity", "1e9999", "-1e9999"), From 15fe4fa4e48179913d9acb2ef4754ffe02d9072f Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:42:03 +0900 Subject: [PATCH 05/15] fix: retain atomic temporary ownership (#482 #483) --- src/format_bench/json_contract.py | 109 +++++++++++++++++++---------- tests/test_json_contract.py | 112 +++++++++++++++++++++++++++--- 2 files changed, 173 insertions(+), 48 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index d79f0b1..ad502d5 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -46,28 +46,50 @@ def _destination_mode(directory_fd: int, name: str) -> int | None: def _create_temporary( directory_fd: int, - directory: Path, destination_name: str, -) -> tuple[int, Path]: - 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, - 0o666, - dir_fd=directory_fd, - ) - except FileExistsError: - continue - return descriptor, directory / name - raise FileExistsError("could not create an exclusive temporary JSON file") - - -def _temporary_name(directory_fd: int, descriptor: int, temporary: Path) -> str: +) -> tuple[int, int, str]: + cleanup_directory_fd = os.dup(directory_fd) + try: + 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, + 0o666, + dir_fd=cleanup_directory_fd, + ) + except FileExistsError: + continue + return descriptor, cleanup_directory_fd, name + raise FileExistsError("could not create an exclusive temporary JSON file") + except BaseException: + os.close(cleanup_directory_fd) + raise + + +def _temporary_name( + directory_fd: int, + cleanup_directory_fd: int, + descriptor: int, + temporary_name: str, +) -> str: + destination_directory = os.fstat(directory_fd) + temporary_directory = os.fstat(cleanup_directory_fd) + if ( + destination_directory.st_dev != temporary_directory.st_dev + or destination_directory.st_ino != temporary_directory.st_ino + ): + raise ValueError( + "temporary JSON file must use the destination's same directory" + ) try: - entry = os.stat(temporary.name, dir_fd=directory_fd, follow_symlinks=False) + entry = os.stat( + temporary_name, + dir_fd=cleanup_directory_fd, + follow_symlinks=False, + ) except FileNotFoundError as error: raise ValueError( "temporary JSON file must use the destination's same directory" @@ -78,10 +100,8 @@ def _temporary_name(directory_fd: int, descriptor: int, temporary: Path) -> str: or entry.st_dev != opened.st_dev or entry.st_ino != opened.st_ino ): - raise ValueError( - "temporary JSON file must be a regular same-directory entry" - ) - return temporary.name + raise ValueError("temporary JSON file must be a regular same-directory entry") + return temporary_name def _replace_same_directory( @@ -162,8 +182,8 @@ def atomic_write_json(path: Path, value: object) -> None: raise ValueError("JSON destination must name a regular file") directory_fd = -1 + cleanup_directory_fd = -1 descriptor = -1 - temporary: Path | None = None temporary_name: str | None = None try: directory_fd = os.open( @@ -171,12 +191,16 @@ def atomic_write_json(path: Path, value: object) -> None: os.O_RDONLY | os.O_DIRECTORY, ) mode = _destination_mode(directory_fd, destination_name) - descriptor, temporary = _create_temporary( + descriptor, cleanup_directory_fd, temporary_name = _create_temporary( directory_fd, - destination_parent, destination_name, ) - temporary_name = _temporary_name(directory_fd, descriptor, temporary) + temporary_name = _temporary_name( + directory_fd, + cleanup_directory_fd, + descriptor, + temporary_name, + ) if mode is not None: os.fchmod(descriptor, mode) with os.fdopen(descriptor, "wb") as stream: @@ -188,15 +212,24 @@ def atomic_write_json(path: Path, value: object) -> None: # os.replace is atomic on success; file fsync does not claim directory or # power-loss durability. See https://docs.python.org/3/library/os.html#os.replace _replace_same_directory(directory_fd, temporary_name, destination_name) + # LLM contract: TEMPORARY_OWNED -> PUBLISHED transfers the basename; + # this call must never clean a later entry that reuses the vacated name. + temporary_name = None finally: - if descriptor >= 0: - os.close(descriptor) - if temporary_name is not None and directory_fd >= 0: + try: + if descriptor >= 0: + os.close(descriptor) + finally: try: - os.unlink(temporary_name, dir_fd=directory_fd) - except FileNotFoundError: - pass - elif temporary is not None: - temporary.unlink(missing_ok=True) - if directory_fd >= 0: - os.close(directory_fd) + if temporary_name is not None and cleanup_directory_fd >= 0: + try: + os.unlink(temporary_name, dir_fd=cleanup_directory_fd) + except FileNotFoundError: + pass + finally: + try: + if cleanup_directory_fd >= 0: + os.close(cleanup_directory_fd) + finally: + if directory_fd >= 0: + os.close(directory_fd) diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index a6ebe9f..f84dd1e 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -3,7 +3,6 @@ import math import os import stat -import tempfile from collections.abc import Callable from pathlib import Path from typing import BinaryIO, cast @@ -265,20 +264,26 @@ def test_atomic_write_json_rejects_cross_directory_temp( destination.write_bytes(previous) other_dir = tmp_path / "other" other_dir.mkdir() - real_mkstemp = tempfile.mkstemp + 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, - directory: Path, destination_name: str, - ) -> tuple[int, Path]: - del directory_fd, directory - descriptor, name = real_mkstemp( - prefix=f".{destination_name}.", - suffix=".tmp", - dir=other_dir, + ) -> tuple[int, int, str]: + del directory_fd, destination_name + cleanup_directory_fd = os.open( + other_dir, + os.O_RDONLY | os.O_DIRECTORY, ) - return descriptor, Path(name) + descriptor = os.open( + temporary_name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=cleanup_directory_fd, + ) + return descriptor, cleanup_directory_fd, temporary_name monkeypatch.setattr( json_contract, @@ -291,6 +296,7 @@ def cross_directory_temporary( assert destination.read_bytes() == previous assert list(other_dir.iterdir()) == [] + assert same_name_victim.read_bytes() == b"DO-NOT-DELETE" def test_atomic_write_json_retries_exclusive_name_collision( @@ -310,6 +316,92 @@ def test_atomic_write_json_retries_exclusive_name_collision( assert not (tmp_path / ".manifest.json.fresh.tmp").exists() +def test_atomic_write_json_drops_temporary_name_after_replace( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + token = "published" + temporary_name = f".manifest.json.{token}.tmp" + destination = tmp_path / "manifest.json" + captured_directory_fds: list[int] = [] + captured_cleanup_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, + ) -> tuple[int, int, str]: + result = real_create(directory_fd, destination_name) + captured_cleanup_fds.append(result[1]) + return result + + 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) + + 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) + + atomic_write_json(destination, {"state": "REPORTED"}) + + assert (tmp_path / temporary_name).read_bytes() == b"REUSED" + for descriptor in captured_directory_fds + captured_cleanup_fds: + with pytest.raises(OSError): + os.fstat(descriptor) + + +def test_atomic_write_json_closes_directories_when_cleanup_fails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + destination = tmp_path / "manifest.json" + captured_directory_fds: list[int] = [] + captured_cleanup_fds: list[int] = [] + real_create = json_contract._create_temporary + + def capture_create( + directory_fd: int, + destination_name: str, + ) -> tuple[int, int, str]: + captured_directory_fds.append(directory_fd) + result = real_create(directory_fd, destination_name) + captured_cleanup_fds.append(result[1]) + return result + + def fail_write(_destination: BinaryIO, _payload: bytes) -> None: + raise OSError("injected write failure") + + def fail_cleanup(*_args: object, **_kwargs: object) -> None: + raise PermissionError("injected cleanup failure") + + monkeypatch.setattr(json_contract, "_create_temporary", capture_create) + monkeypatch.setattr(json_contract, "_write_atomic_file", fail_write) + monkeypatch.setattr(json_contract.os, "unlink", fail_cleanup) + + with pytest.raises(PermissionError, match="injected cleanup failure"): + atomic_write_json(destination, {"state": "REPORTED"}) + + for descriptor in captured_directory_fds + captured_cleanup_fds: + with pytest.raises(OSError): + os.fstat(descriptor) + + @pytest.mark.parametrize( "token", ("NaN", "Infinity", "-Infinity", "1e9999", "-1e9999"), From 2aca47e22f7802962bbd2c52fb02bf33181e702c Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:44:39 +0900 Subject: [PATCH 06/15] fix: restrict atomic replacement temp mode (#484) --- src/format_bench/json_contract.py | 4 ++- tests/test_json_contract.py | 41 ++++++++++++++++++++++++++++--- 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index ad502d5..7c6b2fe 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -47,6 +47,7 @@ def _destination_mode(directory_fd: int, name: str) -> int | None: def _create_temporary( directory_fd: int, destination_name: str, + creation_mode: int, ) -> tuple[int, int, str]: cleanup_directory_fd = os.dup(directory_fd) try: @@ -57,7 +58,7 @@ def _create_temporary( descriptor = os.open( name, flags, - 0o666, + creation_mode, dir_fd=cleanup_directory_fd, ) except FileExistsError: @@ -194,6 +195,7 @@ def atomic_write_json(path: Path, value: object) -> None: descriptor, cleanup_directory_fd, temporary_name = _create_temporary( directory_fd, destination_name, + mode if mode is not None else 0o666, ) temporary_name = _temporary_name( directory_fd, diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index f84dd1e..49e7515 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -116,6 +116,38 @@ def test_atomic_write_json_preserves_existing_evidence_mode(tmp_path: Path) -> N 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) == 1 + 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, @@ -271,8 +303,9 @@ def test_atomic_write_json_rejects_cross_directory_temp( def cross_directory_temporary( directory_fd: int, destination_name: str, + creation_mode: int, ) -> tuple[int, int, str]: - del directory_fd, destination_name + del directory_fd, destination_name, creation_mode cleanup_directory_fd = os.open( other_dir, os.O_RDONLY | os.O_DIRECTORY, @@ -331,8 +364,9 @@ def test_atomic_write_json_drops_temporary_name_after_replace( def capture_create( directory_fd: int, destination_name: str, + creation_mode: int, ) -> tuple[int, int, str]: - result = real_create(directory_fd, destination_name) + result = real_create(directory_fd, destination_name, creation_mode) captured_cleanup_fds.append(result[1]) return result @@ -378,9 +412,10 @@ def test_atomic_write_json_closes_directories_when_cleanup_fails( def capture_create( directory_fd: int, destination_name: str, + creation_mode: int, ) -> tuple[int, int, str]: captured_directory_fds.append(directory_fd) - result = real_create(directory_fd, destination_name) + result = real_create(directory_fd, destination_name, creation_mode) captured_cleanup_fds.append(result[1]) return result From 6258ec08290a3c0e2b7798f671e1a33014286760 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:48:31 +0900 Subject: [PATCH 07/15] fix: bind temp cleanup to opened inode (#485) --- src/format_bench/json_contract.py | 85 ++++++++++++++++++++++--------- tests/test_json_contract.py | 45 ++++++++++++++++ 2 files changed, 106 insertions(+), 24 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 7c6b2fe..2a28aa3 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -18,6 +18,10 @@ class JsonDumpOptions(TypedDict, total=False): sort_keys: bool +class _TemporaryOwnershipLost(ValueError): + """The temporary basename no longer identifies this writer's inode.""" + + def _write_atomic_file(destination: BinaryIO, payload: bytes) -> None: offset = 0 while offset < len(payload): @@ -73,9 +77,27 @@ def _create_temporary( def _temporary_name( directory_fd: int, cleanup_directory_fd: int, - descriptor: int, temporary_name: str, + temporary_identity: tuple[int, int], ) -> str: + try: + entry = os.stat( + temporary_name, + dir_fd=cleanup_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" + ) destination_directory = os.fstat(directory_fd) temporary_directory = os.fstat(cleanup_directory_fd) if ( @@ -85,24 +107,25 @@ def _temporary_name( raise ValueError( "temporary JSON file must use the destination's same directory" ) + return temporary_name + + +def _unlink_owned_temporary( + cleanup_directory_fd: int, + temporary_name: str, + temporary_identity: tuple[int, int], +) -> None: try: entry = os.stat( temporary_name, dir_fd=cleanup_directory_fd, follow_symlinks=False, ) - except FileNotFoundError as error: - raise ValueError( - "temporary JSON file must use the destination's same directory" - ) from error - opened = os.fstat(descriptor) - if ( - not stat.S_ISREG(entry.st_mode) - or entry.st_dev != opened.st_dev - or entry.st_ino != opened.st_ino - ): - raise ValueError("temporary JSON file must be a regular same-directory entry") - return temporary_name + except FileNotFoundError: + return + if entry.st_dev != temporary_identity[0] or entry.st_ino != temporary_identity[1]: + return + os.unlink(temporary_name, dir_fd=cleanup_directory_fd) def _replace_same_directory( @@ -186,6 +209,7 @@ def atomic_write_json(path: Path, value: object) -> None: cleanup_directory_fd = -1 descriptor = -1 temporary_name: str | None = None + temporary_identity: tuple[int, int] | None = None try: directory_fd = os.open( destination_parent, @@ -197,12 +221,19 @@ def atomic_write_json(path: Path, value: object) -> None: destination_name, mode if mode is not None else 0o666, ) - temporary_name = _temporary_name( - directory_fd, - cleanup_directory_fd, - descriptor, - temporary_name, - ) + opened_temporary = os.fstat(descriptor) + temporary_identity = (opened_temporary.st_dev, opened_temporary.st_ino) + try: + temporary_name = _temporary_name( + directory_fd, + cleanup_directory_fd, + temporary_name, + temporary_identity, + ) + except _TemporaryOwnershipLost: + temporary_name = None + temporary_identity = None + raise if mode is not None: os.fchmod(descriptor, mode) with os.fdopen(descriptor, "wb") as stream: @@ -217,17 +248,23 @@ def atomic_write_json(path: Path, value: object) -> None: # LLM contract: TEMPORARY_OWNED -> PUBLISHED transfers the basename; # this call must never clean a later entry that reuses the vacated name. temporary_name = None + temporary_identity = None finally: try: if descriptor >= 0: os.close(descriptor) finally: try: - if temporary_name is not None and cleanup_directory_fd >= 0: - try: - os.unlink(temporary_name, dir_fd=cleanup_directory_fd) - except FileNotFoundError: - pass + if ( + temporary_name is not None + and temporary_identity is not None + and cleanup_directory_fd >= 0 + ): + _unlink_owned_temporary( + cleanup_directory_fd, + temporary_name, + temporary_identity, + ) finally: try: if cleanup_directory_fd >= 0: diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index 49e7515..b788edf 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -437,6 +437,51 @@ def fail_cleanup(*_args: object, **_kwargs: object) -> None: 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, + cleanup_directory_fd: int, + name: str, + identity: tuple[int, int], + ) -> str: + os.unlink(name, dir_fd=cleanup_directory_fd) + replacement = os.open( + name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=cleanup_directory_fd, + ) + try: + os.write(replacement, b"UNRELATED") + finally: + os.close(replacement) + return real_temporary_name( + directory_fd, + cleanup_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"), From cccf8babc2998b1a0bab16bda337bd13f435f50f Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 13:49:30 +0900 Subject: [PATCH 08/15] test: cover post-publication replacement error (#485) --- tests/test_json_contract.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index b788edf..7d88449 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -349,9 +349,11 @@ def test_atomic_write_json_retries_exclusive_name_collision( 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" @@ -387,13 +389,20 @@ def replace_and_reuse( 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) - atomic_write_json(destination, {"state": "REPORTED"}) + 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 + captured_cleanup_fds: with pytest.raises(OSError): From 84ddd0d11fd5a39d56d84b50ed24c56b9b9961ad Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:00:44 +0900 Subject: [PATCH 09/15] fix: retain failed atomic temps safely (#485) --- src/format_bench/json_contract.py | 122 +++++++++--------------------- tests/test_json_contract.py | 58 ++++++-------- 2 files changed, 58 insertions(+), 122 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 2a28aa3..6abea28 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -52,38 +52,32 @@ def _create_temporary( directory_fd: int, destination_name: str, creation_mode: int, -) -> tuple[int, int, str]: - cleanup_directory_fd = os.dup(directory_fd) - try: - 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=cleanup_directory_fd, - ) - except FileExistsError: - continue - return descriptor, cleanup_directory_fd, name - raise FileExistsError("could not create an exclusive temporary JSON file") - except BaseException: - os.close(cleanup_directory_fd) - raise +) -> 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, - cleanup_directory_fd: int, temporary_name: str, temporary_identity: tuple[int, int], ) -> str: try: entry = os.stat( temporary_name, - dir_fd=cleanup_directory_fd, + dir_fd=directory_fd, follow_symlinks=False, ) except FileNotFoundError as error: @@ -98,36 +92,9 @@ def _temporary_name( raise _TemporaryOwnershipLost( "temporary JSON file no longer identifies the opened entry" ) - destination_directory = os.fstat(directory_fd) - temporary_directory = os.fstat(cleanup_directory_fd) - if ( - destination_directory.st_dev != temporary_directory.st_dev - or destination_directory.st_ino != temporary_directory.st_ino - ): - raise ValueError( - "temporary JSON file must use the destination's same directory" - ) return temporary_name -def _unlink_owned_temporary( - cleanup_directory_fd: int, - temporary_name: str, - temporary_identity: tuple[int, int], -) -> None: - try: - entry = os.stat( - temporary_name, - dir_fd=cleanup_directory_fd, - follow_symlinks=False, - ) - except FileNotFoundError: - return - if entry.st_dev != temporary_identity[0] or entry.st_ino != temporary_identity[1]: - return - os.unlink(temporary_name, dir_fd=cleanup_directory_fd) - - def _replace_same_directory( directory_fd: int, temporary_name: str, @@ -206,69 +173,48 @@ def atomic_write_json(path: Path, value: object) -> None: raise ValueError("JSON destination must name a regular file") directory_fd = -1 - cleanup_directory_fd = -1 descriptor = -1 - temporary_name: str | None = None - temporary_identity: tuple[int, int] | None = None try: directory_fd = os.open( destination_parent, os.O_RDONLY | os.O_DIRECTORY, ) mode = _destination_mode(directory_fd, destination_name) - descriptor, cleanup_directory_fd, temporary_name = _create_temporary( + 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) - try: - temporary_name = _temporary_name( - directory_fd, - cleanup_directory_fd, - temporary_name, - temporary_identity, - ) - except _TemporaryOwnershipLost: - temporary_name = None - temporary_identity = None - raise - if mode is not None: - os.fchmod(descriptor, mode) + 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) + os.fchmod(stream.fileno(), final_mode) + os.fsync(stream.fileno()) # LLM contract: DISCOVERED -> ENCODED -> ROUNDTRIP_VERIFIED -> # BENCHMARKED -> REPORTED becomes persistent only at this replace boundary. # os.replace is atomic on success; file fsync does not claim directory or # power-loss durability. See https://docs.python.org/3/library/os.html#os.replace _replace_same_directory(directory_fd, temporary_name, destination_name) - # LLM contract: TEMPORARY_OWNED -> PUBLISHED transfers the basename; - # this call must never clean a later entry that reuses the vacated name. - temporary_name = None - temporary_identity = None + # 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: - try: - if ( - temporary_name is not None - and temporary_identity is not None - and cleanup_directory_fd >= 0 - ): - _unlink_owned_temporary( - cleanup_directory_fd, - temporary_name, - temporary_identity, - ) - finally: - try: - if cleanup_directory_fd >= 0: - os.close(cleanup_directory_fd) - finally: - if directory_fd >= 0: - os.close(directory_fd) + if directory_fd >= 0: + os.close(directory_fd) diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index 7d88449..aab2b9c 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -143,7 +143,7 @@ def inspect_initial_mode(descriptor: int, requested_mode: int) -> None: finally: os.umask(previous_mask) - assert len(observed_modes) == 1 + assert len(observed_modes) == 2 assert observed_modes[0] & ~mode == 0 assert stat.S_IMODE(destination.stat().st_mode) == mode @@ -165,7 +165,7 @@ def fail_mode(*_args: object, **_kwargs: object) -> None: atomic_write_json(destination, {"state": "BENCHMARKED"}) assert destination.read_bytes() == previous - assert list(tmp_path.glob(".manifest.json.*.tmp")) == [] + assert len(list(tmp_path.glob(".manifest.json.*.tmp"))) == 1 def test_atomic_write_json_serialization_failure_preserves_previous_document( @@ -220,7 +220,7 @@ def fail(*_args: object, **_kwargs: object) -> None: atomic_write_json(destination, {"state": "BENCHMARKED"}) assert destination.read_bytes() == previous - assert list(tmp_path.glob(".manifest.json.*.tmp")) == [] + assert len(list(tmp_path.glob(".manifest.json.*.tmp"))) == 1 def test_atomic_write_json_rejects_symlink_destination(tmp_path: Path) -> None: @@ -304,19 +304,14 @@ def cross_directory_temporary( directory_fd: int, destination_name: str, creation_mode: int, - ) -> tuple[int, int, str]: + ) -> tuple[int, str]: del directory_fd, destination_name, creation_mode - cleanup_directory_fd = os.open( - other_dir, - os.O_RDONLY | os.O_DIRECTORY, - ) descriptor = os.open( - temporary_name, + other_dir / temporary_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, - dir_fd=cleanup_directory_fd, ) - return descriptor, cleanup_directory_fd, temporary_name + return descriptor, temporary_name monkeypatch.setattr( json_contract, @@ -324,11 +319,11 @@ def cross_directory_temporary( cross_directory_temporary, ) - with pytest.raises(ValueError, match="same directory"): + with pytest.raises(ValueError, match="no longer identifies"): atomic_write_json(destination, {"state": "BENCHMARKED"}) assert destination.read_bytes() == previous - assert list(other_dir.iterdir()) == [] + assert len(list(other_dir.iterdir())) == 1 assert same_name_victim.read_bytes() == b"DO-NOT-DELETE" @@ -359,7 +354,6 @@ def test_atomic_write_json_drops_temporary_name_after_replace( temporary_name = f".manifest.json.{token}.tmp" destination = tmp_path / "manifest.json" captured_directory_fds: list[int] = [] - captured_cleanup_fds: list[int] = [] real_create = json_contract._create_temporary real_replace = json_contract._replace_same_directory @@ -367,10 +361,8 @@ def capture_create( directory_fd: int, destination_name: str, creation_mode: int, - ) -> tuple[int, int, str]: - result = real_create(directory_fd, destination_name, creation_mode) - captured_cleanup_fds.append(result[1]) - return result + ) -> tuple[int, str]: + return real_create(directory_fd, destination_name, creation_mode) def replace_and_reuse( directory_fd: int, @@ -404,44 +396,44 @@ def replace_and_reuse( 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 + captured_cleanup_fds: + for descriptor in captured_directory_fds: with pytest.raises(OSError): os.fstat(descriptor) -def test_atomic_write_json_closes_directories_when_cleanup_fails( +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] = [] - captured_cleanup_fds: list[int] = [] real_create = json_contract._create_temporary def capture_create( directory_fd: int, destination_name: str, creation_mode: int, - ) -> tuple[int, int, str]: + ) -> tuple[int, str]: captured_directory_fds.append(directory_fd) - result = real_create(directory_fd, destination_name, creation_mode) - captured_cleanup_fds.append(result[1]) - return result + return real_create(directory_fd, destination_name, creation_mode) def fail_write(_destination: BinaryIO, _payload: bytes) -> None: raise OSError("injected write failure") - def fail_cleanup(*_args: object, **_kwargs: object) -> None: - raise PermissionError("injected cleanup 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", fail_cleanup) + monkeypatch.setattr(json_contract.os, "unlink", reject_unlink) - with pytest.raises(PermissionError, match="injected cleanup failure"): + with pytest.raises(OSError, match="injected write failure"): atomic_write_json(destination, {"state": "REPORTED"}) - for descriptor in captured_directory_fds + captured_cleanup_fds: + 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) @@ -459,16 +451,15 @@ def test_atomic_write_json_preserves_name_reused_during_validation( def replace_before_validation( directory_fd: int, - cleanup_directory_fd: int, name: str, identity: tuple[int, int], ) -> str: - os.unlink(name, dir_fd=cleanup_directory_fd) + 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=cleanup_directory_fd, + dir_fd=directory_fd, ) try: os.write(replacement, b"UNRELATED") @@ -476,7 +467,6 @@ def replace_before_validation( os.close(replacement) return real_temporary_name( directory_fd, - cleanup_directory_fd, name, identity, ) From e848e725d64fb93be8b4a52e408dfadce298f6dc Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:07:04 +0900 Subject: [PATCH 10/15] fix: bind publication to owner-private directory (#487) --- src/format_bench/json_contract.py | 14 ++++++++++++++ tests/test_json_contract.py | 20 ++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 6abea28..8833666 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -48,6 +48,16 @@ def _destination_mode(directory_fd: int, name: str) -> int | None: 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" + ) + + def _create_temporary( directory_fd: int, destination_name: str, @@ -179,6 +189,10 @@ def atomic_write_json(path: Path, value: object) -> None: 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, diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index aab2b9c..3437239 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -236,6 +236,26 @@ def test_atomic_write_json_rejects_symlink_destination(tmp_path: Path) -> None: 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")) == [] + + def test_atomic_write_json_accepts_same_directory_with_parent_segments( tmp_path: Path, ) -> None: From 5ef7c1ffe5200ffb68f5d1b5b425c63961790c37 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:16:42 +0900 Subject: [PATCH 11/15] fix: reject Darwin extended ACL publication (#487) --- src/format_bench/json_contract.py | 35 +++++++++++++++++++++++++++++++ tests/test_json_contract.py | 28 +++++++++++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 8833666..c47d889 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -2,11 +2,14 @@ from __future__ import annotations +import ctypes +import errno import json import math import os import secrets import stat +import sys from pathlib import Path from typing import BinaryIO, Never, TypedDict, Unpack, cast @@ -22,6 +25,34 @@ 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): @@ -56,6 +87,10 @@ def _validate_publication_directory(directory_fd: int) -> None: "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( diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index 3437239..a8091e3 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -3,6 +3,8 @@ import math import os import stat +import subprocess +import sys from collections.abc import Callable from pathlib import Path from typing import BinaryIO, cast @@ -256,6 +258,32 @@ def test_atomic_write_json_rejects_shared_writable_directory( 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: From 7a1be7cad7fcc9744b32bbf476e0d4c256905bfa Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:30:59 +0900 Subject: [PATCH 12/15] fix: secure atomic publication lifecycle directories (#489 #490) --- src/format_bench/json_contract.py | 15 ++++--- src/format_bench/shards.py | 22 +++++------ src/format_bench/workflow.py | 4 +- tests/test_json_contract.py | 65 +++++++++++++++++++++++++++++++ tests/test_shards.py | 24 ++++++++---- tests/test_workflow.py | 22 +++++++++++ 6 files changed, 124 insertions(+), 28 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index c47d889..282193b 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -249,13 +249,18 @@ def atomic_write_json(path: Path, value: object) -> None: 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 later failure is post-publication: new bytes remain installed + # at the restrictive mode instead of exposing unpublished evidence. os.fchmod(stream.fileno(), final_mode) os.fsync(stream.fileno()) - # LLM contract: DISCOVERED -> ENCODED -> ROUNDTRIP_VERIFIED -> - # BENCHMARKED -> REPORTED becomes persistent only at this replace boundary. - # os.replace is atomic on success; file fsync does not claim directory or - # power-loss durability. See https://docs.python.org/3/library/os.html#os.replace - _replace_same_directory(directory_fd, temporary_name, destination_name) # 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 diff --git a/src/format_bench/shards.py b/src/format_bench/shards.py index 4bb1959..09ed2e4 100644 --- a/src/format_bench/shards.py +++ b/src/format_bench/shards.py @@ -11,13 +11,7 @@ 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] @@ -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 @@ -139,7 +137,7 @@ def _hardlink_tree(source: Path, destination: Path) -> None: def _copy_run_files(base_run: Path, output_run: Path) -> None: - output_run.mkdir(parents=True) + output_run.mkdir(mode=0o700, parents=True) 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 +235,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 be2eecb..1f0f49f 100644 --- a/src/format_bench/workflow.py +++ b/src/format_bench/workflow.py @@ -141,9 +141,9 @@ 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) + destination.mkdir(mode=0o700, parents=True, 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 diff --git a/tests/test_json_contract.py b/tests/test_json_contract.py index a8091e3..6447db4 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -223,6 +223,71 @@ def fail(*_args: object, **_kwargs: object) -> None: 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_rejects_symlink_destination(tmp_path: Path) -> None: diff --git a/tests/test_shards.py b/tests/test_shards.py index c5195ea..be1a87d 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}) @@ -134,10 +137,15 @@ def test_merge_equivalence_shards_reuses_artifacts_and_unions_results( _write(shard / "results.json", _shard_results(pair, name)) output = tmp_path / "merged" - merge_equivalence_shards(base, shard_root, output) + 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")) + assert stat.S_IMODE(output.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 +153,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 +375,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..76b0c38 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -1,5 +1,7 @@ import json +import os import shutil +import stat from dataclasses import replace from pathlib import Path @@ -169,6 +171,26 @@ 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, + mask: int, +) -> None: + root = Path(__file__).parents[1] + run_dir = tmp_path / "run" + previous_mask = os.umask(mask) + try: + prepare_run(root, DATASET, run_dir, fixture=True, selected=()) + finally: + os.umask(previous_mask) + + for directory in (run_dir, run_dir / "input"): + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 + assert (run_dir / "input" / "manifest.json").is_file() + assert (run_dir / "manifest.json").is_file() + + def test_artifact_digest_rejects_root_symlink(tmp_path: Path) -> None: target = tmp_path / "target" target.mkdir() From da6b5a09504f709f42a66e2d9683088815ea4e21 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:39:44 +0900 Subject: [PATCH 13/15] fix: complete private directory and post-publish contracts (#489 #490) --- src/format_bench/json_contract.py | 5 +++-- src/format_bench/shards.py | 2 +- src/format_bench/workflow.py | 6 +++--- tests/test_json_contract.py | 27 +++++++++++++++++++++++++++ tests/test_shards.py | 3 ++- tests/test_workflow.py | 28 ++++++++++++++++++++++++++-- 6 files changed, 62 insertions(+), 9 deletions(-) diff --git a/src/format_bench/json_contract.py b/src/format_bench/json_contract.py index 282193b..2857b01 100644 --- a/src/format_bench/json_contract.py +++ b/src/format_bench/json_contract.py @@ -257,8 +257,9 @@ def atomic_write_json(path: Path, value: object) -> None: temporary_name, destination_name, ) - # A later failure is post-publication: new bytes remain installed - # at the restrictive mode instead of exposing unpublished evidence. + # 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. diff --git a/src/format_bench/shards.py b/src/format_bench/shards.py index 09ed2e4..91017e0 100644 --- a/src/format_bench/shards.py +++ b/src/format_bench/shards.py @@ -125,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(): diff --git a/src/format_bench/workflow.py b/src/format_bench/workflow.py index 1f0f49f..b47a807 100644 --- a/src/format_bench/workflow.py +++ b/src/format_bench/workflow.py @@ -153,9 +153,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. @@ -199,7 +199,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 6447db4..1e2a3e2 100644 --- a/tests/test_json_contract.py +++ b/tests/test_json_contract.py @@ -290,6 +290,33 @@ def fail_final_mode(descriptor: int, mode: int) -> None: 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') diff --git a/tests/test_shards.py b/tests/test_shards.py index be1a87d..eec2515 100644 --- a/tests/test_shards.py +++ b/tests/test_shards.py @@ -145,7 +145,8 @@ def test_merge_equivalence_shards_reuses_artifacts_and_unions_results( merged = json.loads((output / "results.json").read_text(encoding="utf-8")) manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8")) - assert stat.S_IMODE(output.stat().st_mode) == 0o700 + for directory in (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"} diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 76b0c38..b5f3922 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -175,18 +175,42 @@ def test_prepare_and_verify_fixture_record_relative_evidence(tmp_path: Path) -> @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=()) + 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"): + 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() From 05e72a7a689524d0b8d74ea7e30096ac2a1c2cac Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:46:23 +0900 Subject: [PATCH 14/15] fix: create default run parent privately (#493) --- src/format_bench/workflow.py | 8 +++++++- tests/test_workflow.py | 30 +++++++++++++++++++++++++++++- 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/src/format_bench/workflow.py b/src/format_bench/workflow.py index b47a807..c8427ca 100644 --- a/src/format_bench/workflow.py +++ b/src/format_bench/workflow.py @@ -141,7 +141,13 @@ def prepare_run( f"expected {manifest['source_sha256']}, got {actual_source_sha256}" ) destination = run_dir or _default_run_dir(root, dataset_id) - destination.mkdir(mode=0o700, 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(mode=0o700) diff --git a/tests/test_workflow.py b/tests/test_workflow.py index b5f3922..cb0ee9b 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -8,7 +8,7 @@ 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 @@ -215,6 +215,34 @@ def record_mode( 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() From 4f1c84efdb8e052f0e57332b18196c11ca6e5b77 Mon Sep 17 00:00:00 2001 From: Anionix Date: Fri, 24 Jul 2026 14:59:57 +0900 Subject: [PATCH 15/15] fix: create shard-output ancestors privately (#494) --- src/format_bench/shards.py | 20 +++++++++++++++++++- tests/test_shards.py | 11 +++++++++-- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/src/format_bench/shards.py b/src/format_bench/shards.py index 91017e0..9e984f5 100644 --- a/src/format_bench/shards.py +++ b/src/format_bench/shards.py @@ -136,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(mode=0o700, 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") diff --git a/tests/test_shards.py b/tests/test_shards.py index eec2515..39df1c9 100644 --- a/tests/test_shards.py +++ b/tests/test_shards.py @@ -136,7 +136,8 @@ 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" + output_parent = tmp_path / "missing" / "nested" + output = output_parent / "merged" previous_mask = os.umask(mask) try: merge_equivalence_shards(base, shard_root, output) @@ -145,7 +146,13 @@ def test_merge_equivalence_shards_reuses_artifacts_and_unions_results( 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 (output, output / "input", output / "artifacts"): + 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"}