fix(utils): tighten create_mode semantics and close the yaml 0600 transit window

Post-review fixes on the preserve_mode/create_mode follow-up:

- create_mode is now applied ONLY when the target does not exist, on
  both atomic_write_text and atomic_yaml_write. Previously
  atomic_write_text(path, s, create_mode=X) without preserve_mode would
  silently chmod an EXISTING file to X (docstring/code mismatch, latent
  trap -- no caller relied on it), and a stat failure on an existing
  file could fall through to create_mode instead of leaving the mode
  alone.

- atomic_yaml_write now fchmods the temp fd BEFORE the replace when a
  mode is known, matching atomic_write_text: a freshly created
  distribution.yaml no longer transits through mkstemp's 0600 (a crash
  between replace and chmod could previously leave it 0600 forever).
  The post-replace _restore_file_mode stays as the Windows path.

- fchmod moved inside the fdopen context in atomic_write_text, so a
  raising fchmod can no longer leak the fd.

Tests: create_mode-never-rewrites-existing guard (mutation-checked) and
a monkeypatch.delattr(os, 'fchmod') test covering the Windows
post-replace branch that the win32 module skip left uncovered.
This commit is contained in:
kshitij
2026-08-06 04:55:13 +05:30
committed by kshitij
parent 3556728a54
commit 43fc86562c
2 changed files with 46 additions and 10 deletions
+27
View File
@@ -138,6 +138,33 @@ class TestCreateMode:
assert stat.S_IMODE(target.stat().st_mode) == 0o600
def test_create_mode_never_rewrites_an_existing_file(
self, tmp_path: Path
) -> None:
"""create_mode without preserve_mode must not chmod an existing file."""
target = tmp_path / "notes.md"
target.write_text("old\n", encoding="utf-8")
os.chmod(target, 0o640)
atomic_write_text(target, "new\n", create_mode=0o644)
# The write is a plain (non-preserving) atomic rewrite: mkstemp 0600.
assert stat.S_IMODE(target.stat().st_mode) == 0o600
def test_windows_fallback_branch_applies_mode_after_replace(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Without os.fchmod (Windows), the mode is applied post-replace."""
target = tmp_path / "config.yaml"
target.write_text("old\n", encoding="utf-8")
os.chmod(target, 0o640)
monkeypatch.delattr(os, "fchmod")
atomic_write_text(target, "new\n", preserve_mode=True)
assert target.read_text(encoding="utf-8") == "new\n"
assert stat.S_IMODE(target.stat().st_mode) == 0o640
def test_atomic_yaml_write_create_mode(self, tmp_path: Path) -> None:
"""write_manifest's create path: new file lands 0644, not 0600."""
target = tmp_path / "distribution.yaml"
+19 -10
View File
@@ -165,25 +165,28 @@ def atomic_write_text(
default: the historical callers (memory store, skill manager,
cron) own their 0600-is-fine files.
create_mode: Permission bits to apply when the target does not yet
exist (otherwise the new file keeps mkstemp's 0600). Ignored
when ``preserve_mode`` found an existing mode to carry over.
exist (otherwise the new file keeps mkstemp's 0600). Never
applied to an existing file.
"""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
original_mode = _preserve_file_mode(path) if preserve_mode else None
original_owner = _preserve_file_owner(path) if preserve_mode else None
effective_mode = original_mode if original_mode is not None else create_mode
effective_mode = original_mode
if effective_mode is None and create_mode is not None and not path.exists():
effective_mode = create_mode
fd, tmp_path = tempfile.mkstemp(
dir=str(path.parent), prefix=tmp_prefix, suffix=".tmp"
)
try:
if effective_mode is not None and hasattr(os, "fchmod"):
# fchmod is Unix-only; on Windows the post-replace chmod below
# applies the final mode instead.
os.fchmod(fd, effective_mode)
with os.fdopen(fd, "w", encoding=encoding) as handle:
if effective_mode is not None and hasattr(os, "fchmod"):
# fchmod the temp fd BEFORE the replace so the target never
# transits through mkstemp's 0600. fchmod is Unix-only; on
# Windows the post-replace chmod below applies the mode.
os.fchmod(handle.fileno(), effective_mode)
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
@@ -352,15 +355,15 @@ def atomic_yaml_write(
extra_content: Optional string to append after the YAML dump
(e.g. commented-out sections for user reference).
create_mode: Permission bits to apply when the target does not yet
exist (a created file otherwise keeps mkstemp's 0600). An
existing file's mode is always preserved and wins over this.
exist (a created file otherwise keeps mkstemp's 0600). Never
applied to an existing file, whose mode is always preserved.
"""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
original_mode = _preserve_file_mode(path)
original_owner = _preserve_file_owner(path)
if original_mode is None:
if original_mode is None and create_mode is not None and not path.exists():
original_mode = create_mode
fd, tmp_path = tempfile.mkstemp(
@@ -370,6 +373,12 @@ def atomic_yaml_write(
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
if original_mode is not None and hasattr(os, "fchmod"):
# Apply the mode to the temp fd BEFORE the replace so the
# target never transits through mkstemp's 0600 (the
# post-replace _restore_file_mode below then re-applies it
# harmlessly, and remains the sole path on Windows).
os.fchmod(f.fileno(), original_mode)
# allow_unicode=True writes emoji/kaomoji (e.g. personalities, skin
# cursors) as real UTF-8 instead of fragile escape sequences. Without
# it, PyYAML emits astral-plane chars as `\UXXXXXXXX` (8-digit) escapes