fix(config): restore read-only-target protection for atomic writes
Review finding 2 (10.07 gate): a deliberately locked config (0444) in a writable directory was silently overwritten — atomic replace creates a fresh temp inode and renames over the read-only file, defeating the old in-place Path.write_text PermissionError contract. Probe the existing target with a non-truncating O_WRONLY open before any replacement work and let the PermissionError propagate. The probe fstat()s the fd it actually opened and hands that stat to the rest of the write, so a concurrent writer replacing the inode between stat and probe refreshes the metadata instead of failing (keeps concurrent-writer semantics). Regression: writable parent + 0444 target now raises and keeps the original bytes and mode. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
83bd7125bd
commit
0dc16dcce8
2 files changed
+55
-7
No files matched your search
+35
-7
@@ -83,6 +83,28 @@ def _has_extended_attributes(path: Path) -> bool:
|
||||
raise
|
||||
|
||||
|
||||
def _require_writable_target(write_path: Path) -> os.stat_result | None:
|
||||
"""Reject writes to a read-only target before any replacement work.
|
||||
|
||||
``os.replace`` happily swaps a fresh inode over a ``0444`` file when the
|
||||
directory is writable, which would silently defeat a deliberately locked
|
||||
config. The old in-place ``Path.write_text`` raised ``PermissionError``
|
||||
there; preserve that contract with a non-truncating ``O_WRONLY`` probe,
|
||||
letting the failure propagate. Returns the ``fstat`` of the inode the
|
||||
probe actually opened (so the caller's metadata describes the verified
|
||||
file even if a concurrent writer replaced it since its own ``stat``), or
|
||||
``None`` if the target vanished concurrently.
|
||||
"""
|
||||
try:
|
||||
probe_fd = os.open(write_path, os.O_WRONLY)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
try:
|
||||
return os.fstat(probe_fd)
|
||||
finally:
|
||||
os.close(probe_fd)
|
||||
|
||||
|
||||
def _atomic_write_text(path: Path, text: str, *, encoding: str = "utf-8") -> None:
|
||||
"""Atomically replace *path* with *text*.
|
||||
|
||||
@@ -168,13 +190,19 @@ def _atomic_write_text(path: Path, text: str, *, encoding: str = "utf-8") -> Non
|
||||
if owns_fallback_fd:
|
||||
os.close(fallback_fd)
|
||||
|
||||
if (
|
||||
existing_stat is not None
|
||||
and stat.S_ISREG(existing_stat.st_mode)
|
||||
and (existing_stat.st_nlink > 1 or _has_extended_attributes(write_path))
|
||||
):
|
||||
_write_in_place()
|
||||
return
|
||||
if existing_stat is not None and stat.S_ISREG(existing_stat.st_mode):
|
||||
probed_stat = _require_writable_target(write_path)
|
||||
if probed_stat is not None and stat.S_ISREG(probed_stat.st_mode):
|
||||
existing_stat = probed_stat
|
||||
mode = stat.S_IMODE(existing_stat.st_mode)
|
||||
else:
|
||||
existing_stat = probed_stat
|
||||
mode = None
|
||||
if existing_stat is not None and (
|
||||
existing_stat.st_nlink > 1 or _has_extended_attributes(write_path)
|
||||
):
|
||||
_write_in_place()
|
||||
return
|
||||
|
||||
try:
|
||||
fd, tmp = _create_atomic_temp_file(write_path, existing=existing_stat is not None)
|
||||
|
||||
@@ -786,6 +786,26 @@ def test_readonly_parent_without_writable_target_still_raises(
|
||||
|
||||
|
||||
@_ROOT_SKIP
|
||||
def test_writable_parent_with_readonly_file_raises_and_keeps_bytes(tmp_path: Path) -> None:
|
||||
"""A deliberately locked config (0444) in a writable directory must keep
|
||||
rejecting writes: without the target-writability probe, atomic replace
|
||||
creates a fresh temp inode and renames over the read-only file."""
|
||||
cfg_dir = tmp_path / "open-dir"
|
||||
cfg_dir.mkdir()
|
||||
target = cfg_dir / "config.yaml"
|
||||
original = "model:\n default: keep-me\n"
|
||||
target.write_text(original, encoding="utf-8")
|
||||
os.chmod(target, 0o444)
|
||||
os.chmod(cfg_dir, 0o755)
|
||||
try:
|
||||
with pytest.raises(PermissionError):
|
||||
_atomic_write_text(target, "model:\n default: new\n")
|
||||
assert target.read_text(encoding="utf-8") == original
|
||||
assert stat.S_IMODE(os.stat(target).st_mode) == 0o444
|
||||
finally:
|
||||
os.chmod(target, 0o644)
|
||||
|
||||
|
||||
def test_readonly_parent_with_unwritable_file_still_raises(tmp_path: Path) -> None:
|
||||
cfg_dir = tmp_path / "locked"
|
||||
cfg_dir.mkdir()
|
||||
|
||||
Reference in new issue
Block a user