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"