diff --git a/coworker/secrets.py b/coworker/secrets.py index 6c1c0326..faebdc22 100644 --- a/coworker/secrets.py +++ b/coworker/secrets.py @@ -88,6 +88,21 @@ def _restrict_to_user(path: Path, *, is_dir: bool) -> None: os.chmod(path, 0o700 if is_dir else 0o600) +def _write_text_born_private(tmp: Path, content: str) -> None: + """Write `content` to a file that is user-only from its first byte. + + `Path.write_text` creates the file umask-wide (typically 0644) and the chmod that + follows leaves a window where the content is world-readable — and no protection at + all on the paths where that step is best-effort. Passing 0600 at open covers the + content for its whole lifetime instead. O_EXCL refuses to write through anything + pre-planted at the predictable `.tmp` name (write_text would follow a symlink and + pour the secret into its target); a stale tmp left by a crash is removed first.""" + tmp.unlink(missing_ok=True) + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(content) + + 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() @@ -97,8 +112,8 @@ def write_private_text(path: str | Path, content: str) -> Path: except OSError: pass tmp = target.with_name(target.name + ".tmp") - tmp.write_text(content, encoding="utf-8") - _restrict_to_user(tmp, is_dir=False) + _write_text_born_private(tmp, content) + _restrict_to_user(tmp, is_dir=False) # still needed on Windows: ACL, not mode bits os.replace(tmp, target) return target @@ -188,6 +203,6 @@ def _write(self, store: dict[str, Any]) -> None: 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) + _write_text_born_private(tmp, json.dumps(store, indent=2)) + _restrict_to_user(tmp, is_dir=False) # still needed on Windows: ACL, not mode bits os.replace(tmp, self.path) diff --git a/tests/test_secrets.py b/tests/test_secrets.py index c723fa95..b10c0b7a 100644 --- a/tests/test_secrets.py +++ b/tests/test_secrets.py @@ -85,3 +85,56 @@ def test_delete(tmp_path): assert store.delete("x") is True assert store.delete("x") is False assert store.get("x") is None + + +# -- files must be private from creation, not privatized after the fact ---------- + + +def test_secrets_file_private_even_if_restrict_step_does_nothing(tmp_path, monkeypatch): + """The 0600 must come from the open itself. If it only arrived via the follow-up + _restrict_to_user, the content would sit umask-wide between write and chmod.""" + if sys.platform == "win32": + return # POSIX mode bits; Windows privacy is the ACL, covered above + import coworker.secrets as secrets_mod + + monkeypatch.setattr(secrets_mod, "_restrict_to_user", lambda *a, **k: None) + old_umask = os.umask(0) + try: + store = SecretStore(tmp_path / "secrets.json") + store.put("slack:default", {"type": "token", "bot_token": "xoxb-123"}) + finally: + os.umask(old_umask) + assert stat.S_IMODE(os.stat(store.path).st_mode) == 0o600 + + +def test_write_private_text_private_even_if_restrict_step_does_nothing( + tmp_path, monkeypatch +): + if sys.platform == "win32": + return # POSIX mode bits; Windows privacy is the ACL, covered above + import coworker.secrets as secrets_mod + from coworker.secrets import write_private_text + + monkeypatch.setattr(secrets_mod, "_restrict_to_user", lambda *a, **k: None) + old_umask = os.umask(0) + try: + out = write_private_text(tmp_path / "sidecar-8123.token", "tok\n") + finally: + os.umask(old_umask) + assert stat.S_IMODE(os.stat(out).st_mode) == 0o600 + + +def test_write_private_text_does_not_write_through_a_preplanted_tmp(tmp_path): + """A symlink parked at the predictable `.tmp` name must not receive the secret. + Path.write_text follows it — the secret lands in the symlink's target.""" + if sys.platform == "win32": + return # symlink creation needs privileges on Windows runners + from coworker.secrets import write_private_text + + victim = tmp_path / "victim.txt" + victim.write_text("keep me", encoding="utf-8") + target = tmp_path / "cred.token" + (tmp_path / "cred.token.tmp").symlink_to(victim) + out = write_private_text(target, "s3cret") + assert victim.read_text(encoding="utf-8") == "keep me" + assert out.read_text(encoding="utf-8") == "s3cret"