From ee94da2e0b2c6e798069d1e37e78f99f886c595b Mon Sep 17 00:00:00 2001 From: Mr-Neutr0n <64578610+Mr-Neutr0n@users.noreply.github.com> Date: Tue, 28 Jul 2026 17:35:03 +0530 Subject: [PATCH] security: create secret files private, never chmod them after (#143) Both writers in secrets.py created the temp with Path.write_text and only restricted it afterwards: tmp.write_text(json.dumps(store, indent=2)) # umask default -> 0644 _restrict_to_user(tmp, is_dir=False) # chmod 0600, too late os.replace(tmp, self.path) Measured on macOS by sampling the mode at the moment _restrict_to_user is called, i.e. while the file already holds the plaintext: before: temp held the plaintext at 0o644 -> WORLD/GROUP READABLE after: temp held the plaintext at 0o600 -> user-only Every local process could read every API key for the length of the write, as could anything indexing or backing up that directory. Fixed by routing both writers through _atomic_private_write, which uses tempfile.mkstemp: the file is created 0600 and empty, the Windows ACL is applied while it is still empty, and only then is the content written. Two further problems fall out of the same change, both proven by tests that fail on the previous code: The temp name was the fixed .tmp. Pre-creating that path as a symlink redirected the write, so an unrelated file was overwritten with the contents of secrets.json - an arbitrary overwrite that also discloses every key to a location the attacker picked. mkstemp uses O_EXCL and a random suffix, so a pre-existing path is not followed. A write that failed partway left the temp behind. It is now removed on any exception and the previous secrets.json is left intact. Tests: tests/test_secrets_file_mode.py. Three of the seven fail on the previous code (mid-write mode, hostile temp symlink, cleanup after a failed write); the rest pin the final mode and round-tripping. --- coworker/secrets.py | 54 +++++++++++----- tests/test_secrets_file_mode.py | 107 ++++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 16 deletions(-) create mode 100644 tests/test_secrets_file_mode.py diff --git a/coworker/secrets.py b/coworker/secrets.py index 6c1c0326..00f1d994 100644 --- a/coworker/secrets.py +++ b/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 `.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)) diff --git a/tests/test_secrets_file_mode.py b/tests/test_secrets_file_mode.py new file mode 100644 index 00000000..ae0c4f5d --- /dev/null +++ b/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 `.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"