security: create secret files private, never chmod them after (#143)

coworker/secrets.py

@@ -14,6 +14,7 @@
 import os
 import re
 import subprocess
+import tempfile
 import sys
 import threading
 import time
@@ -88,21 +89,50 @@ def _restrict_to_user(path: Path, *, is_dir: bool) -> None:
     os.chmod(path, 0o700 if is_dir else 0o600)
 
 
-def write_private_text(path: str | Path, content: str) -> Path:
-    """Atomically write a user-only text file using the SecretStore's OS protections."""
-    target = Path(path).expanduser()
+def _atomic_private_write(target: Path, content: str) -> Path:
+    """Write `content` to `target` atomically, never exposing it through a readable temp.
+
+    The temp file used to be created by `Path.write_text` and only chmod-ed afterwards, so
+    the plaintext sat on disk at the umask default (0644 on a normal box) for the length of
+    the write — readable by every local process and by anything backing the directory up.
+    That is issue #143; the same pattern was in both writers here.
+
+    `tempfile.mkstemp` creates with 0600 and O_EXCL before a byte is written, which also
+    removes the fixed `<name>.tmp` filename. That name was predictable, so a local attacker
+    could pre-create it as a symlink and have the write land wherever the link pointed.
+
+    Windows gets no mode bits from mkstemp, so the ACL is applied to the still-empty file
+    before the content goes in.
+    """
     target.parent.mkdir(parents=True, exist_ok=True)
     try:
         _restrict_to_user(target.parent, is_dir=True)
     except OSError:
         pass
-    tmp = target.with_name(target.name + ".tmp")
-    tmp.write_text(content, encoding="utf-8")
-    _restrict_to_user(tmp, is_dir=False)
-    os.replace(tmp, target)
+
+    fd, tmp_name = tempfile.mkstemp(
+        dir=str(target.parent), prefix=f".{target.name}.", suffix=".tmp"
+    )
+    tmp = Path(tmp_name)
+    try:
+        _restrict_to_user(tmp, is_dir=False)
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            fh.write(content)
+        os.replace(tmp, target)
+    except BaseException:
+        try:
+            tmp.unlink()
+        except OSError:
+            pass
+        raise
     return target
 
 
+def write_private_text(path: str | Path, content: str) -> Path:
+    """Atomically write a user-only text file using the SecretStore's OS protections."""
+    return _atomic_private_write(Path(path).expanduser(), content)
+
+
 class SecretStore:
     """File-backed secret store. Reads resolve `${VAR}` refs; status never leaks values."""
 
@@ -182,12 +212,4 @@ def _read(self) -> dict[str, Any]:
             return {}
 
     def _write(self, store: dict[str, Any]) -> None:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        try:
-            _restrict_to_user(self.path.parent, is_dir=True)
-        except OSError:
-            pass
-        tmp = self.path.with_name(self.path.name + ".tmp")
-        tmp.write_text(json.dumps(store, indent=2), encoding="utf-8")
-        _restrict_to_user(tmp, is_dir=False)
-        os.replace(tmp, self.path)
+        _atomic_private_write(self.path, json.dumps(store, indent=2))

tests/test_secrets_file_mode.py

@@ -0,0 +1,107 @@
+"""Secrets must never touch the disk world-readable, even briefly (#143).
+
+The write used to be: create the temp with `Path.write_text` (umask default, 0644 on a
+normal box), then `chmod 0600`, then rename. The plaintext existed at 0644 for the length
+of the write, which is readable by every other local process.
+"""
+
+import json
+import os
+import stat
+import sys
+
+import pytest
+
+from coworker import secrets as secrets_mod
+from coworker.secrets import SecretStore, write_private_text
+
+posix_only = pytest.mark.skipif(
+    sys.platform == "win32", reason="POSIX mode bits; Windows uses the icacls ACL path"
+)
+
+
+@posix_only
+def test_written_secret_file_is_user_only(tmp_path):
+    store = SecretStore(tmp_path / "secrets.json")
+    store.put("openai", {"api_key": "sk-live-secret"})
+    assert stat.S_IMODE((tmp_path / "secrets.json").stat().st_mode) == 0o600
+
+
+@posix_only
+def test_write_private_text_is_user_only(tmp_path):
+    path = write_private_text(tmp_path / "token.txt", "sk-live-secret")
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+
+
+@posix_only
+def test_temp_file_is_never_group_or_world_readable_mid_write(tmp_path, monkeypatch):
+    """The regression itself.
+
+    Hooks `_restrict_to_user`, which both the old and the new writer call, and samples the
+    temp's mode at that moment. Old order was write_text (umask default) -> chmod 0600 ->
+    rename, so the plaintext was on disk at 0644 first and this observes it. New order
+    creates the file 0600 and empty, so the same sample sees 0600.
+    """
+    observed = {}
+    real = secrets_mod._restrict_to_user
+
+    def spy(path, *, is_dir):
+        if not is_dir and path.exists():
+            observed[path.name] = stat.S_IMODE(path.stat().st_mode)
+        return real(path, is_dir=is_dir)
+
+    monkeypatch.setattr(secrets_mod, "_restrict_to_user", spy)
+    SecretStore(tmp_path / "secrets.json").put("openai", {"api_key": "sk-live"})
+
+    assert observed, "expected the writer to restrict a temp file"
+    for name, mode in observed.items():
+        assert mode & (stat.S_IRGRP | stat.S_IROTH) == 0, (
+            f"{name} existed at {oct(mode)} while holding the plaintext"
+        )
+
+
+def test_no_temp_file_is_left_behind(tmp_path):
+    store = SecretStore(tmp_path / "secrets.json")
+    store.put("openai", {"api_key": "sk-live"})
+    leftovers = [p.name for p in tmp_path.iterdir() if p.name != "secrets.json"]
+    assert leftovers == []
+
+
+def test_content_round_trips(tmp_path):
+    store = SecretStore(tmp_path / "secrets.json")
+    store.put("openai", {"api_key": "sk-live"})
+    store.put("anthropic", {"api_key": "sk-ant"})
+    on_disk = json.loads((tmp_path / "secrets.json").read_text())
+    assert on_disk["openai"]["api_key"] == "sk-live"
+    assert on_disk["anthropic"]["api_key"] == "sk-ant"
+    assert SecretStore(tmp_path / "secrets.json").get("openai")["api_key"] == "sk-live"
+
+
+def test_a_failed_write_leaves_the_previous_file_intact(tmp_path, monkeypatch):
+    path = tmp_path / "secrets.json"
+    store = SecretStore(path)
+    store.put("openai", {"api_key": "sk-original"})
+
+    def boom(*a, **kw):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(secrets_mod.os, "replace", boom)
+    with pytest.raises(OSError):
+        store.put("openai", {"api_key": "sk-replacement"})
+
+    assert json.loads(path.read_text())["openai"]["api_key"] == "sk-original"
+    assert [p.name for p in tmp_path.iterdir()] == ["secrets.json"], "temp must be cleaned up"
+
+
+def test_a_hostile_preexisting_temp_name_cannot_redirect_the_write(tmp_path):
+    """The old fixed `<name>.tmp` was predictable; a symlink there redirected the write."""
+    victim = tmp_path / "victim.txt"
+    victim.write_text("do not clobber")
+    decoy = tmp_path / "secrets.json.tmp"
+    try:
+        decoy.symlink_to(victim)
+    except (OSError, NotImplementedError):
+        pytest.skip("symlinks unavailable")
+
+    SecretStore(tmp_path / "secrets.json").put("openai", {"api_key": "sk-live"})
+    assert victim.read_text() == "do not clobber"