mirror of
https://github.com/andrewyng/openworker.git
synced 2026-11-04 13:27:11 +00:00
Mr-Neutr0n
ee94da2e0b
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 <name>.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.
108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
"""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"
|