diff --git a/strix/config/codex.py b/strix/config/codex.py index 94bc8470..cf34f003 100644 --- a/strix/config/codex.py +++ b/strix/config/codex.py @@ -24,6 +24,8 @@ from typing import TYPE_CHECKING, Any import requests +from strix.utils.secret_files import write_secret_text + if TYPE_CHECKING: from collections.abc import Iterator @@ -67,14 +69,7 @@ def _read_store() -> dict[str, Any]: def _write_store(data: dict[str, Any]) -> None: - AUTH_PATH.parent.mkdir(parents=True, exist_ok=True) - tmp = AUTH_PATH.with_suffix(".json.tmp") - tmp.write_text(json.dumps(data, indent=2), encoding="utf-8") - with contextlib.suppress(OSError): - tmp.chmod(0o600) - tmp.replace(AUTH_PATH) - with contextlib.suppress(OSError): - AUTH_PATH.chmod(0o600) + write_secret_text(AUTH_PATH, json.dumps(data, indent=2)) def read_record() -> dict[str, Any] | None: diff --git a/strix/config/loader.py b/strix/config/loader.py index 5b940760..fbcde898 100644 --- a/strix/config/loader.py +++ b/strix/config/loader.py @@ -2,7 +2,6 @@ from __future__ import annotations -import contextlib import json import logging import os @@ -12,6 +11,7 @@ from typing import TYPE_CHECKING, Any from pydantic import AliasChoices, BaseModel from strix.config.settings import Settings +from strix.utils.secret_files import write_secret_text if TYPE_CHECKING: @@ -71,9 +71,7 @@ def persist_current() -> None: env_block[alias.upper()] = value break - target.write_text(json.dumps({"env": env_block}, indent=2), encoding="utf-8") - with contextlib.suppress(OSError): - target.chmod(0o600) + write_secret_text(target, json.dumps({"env": env_block}, indent=2)) def _aliases_for(finfo: FieldInfo) -> list[str]: diff --git a/strix/interface/viewer/auth.py b/strix/interface/viewer/auth.py index 45710811..67a0db9d 100644 --- a/strix/interface/viewer/auth.py +++ b/strix/interface/viewer/auth.py @@ -22,6 +22,7 @@ from typing import Any import requests from strix.config.loader import load_settings +from strix.utils.secret_files import write_secret_text logger = logging.getLogger(__name__) @@ -115,15 +116,8 @@ def is_verified() -> bool: def write_auth(email: str, token: str, verified_at: str) -> None: """Atomically persist the auth record with 0600 permissions.""" - AUTH_PATH.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps({"email": email, "token": token, "verified_at": verified_at}) - tmp = AUTH_PATH.with_suffix(".json.tmp") - tmp.write_text(payload, encoding="utf-8") - with contextlib.suppress(OSError): - tmp.chmod(0o600) - tmp.replace(AUTH_PATH) - with contextlib.suppress(OSError): - AUTH_PATH.chmod(0o600) + write_secret_text(AUTH_PATH, payload) def forget() -> None: diff --git a/strix/utils/secret_files.py b/strix/utils/secret_files.py new file mode 100644 index 00000000..b2170bf9 --- /dev/null +++ b/strix/utils/secret_files.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +import contextlib +import os +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pathlib import Path + + +SECRET_FILE_MODE = 0o600 + + +def write_secret_text(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + + tmp = path.with_suffix(path.suffix + ".tmp") + with contextlib.suppress(FileNotFoundError): + tmp.unlink() + + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, SECRET_FILE_MODE) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(text) + except BaseException: + with contextlib.suppress(OSError): + tmp.unlink() + raise + + tmp.replace(path) diff --git a/tests/test_secret_files.py b/tests/test_secret_files.py new file mode 100644 index 00000000..f93c3541 --- /dev/null +++ b/tests/test_secret_files.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import json +import os +import stat +import sys +from typing import TYPE_CHECKING + +import pytest + +from strix.utils.secret_files import SECRET_FILE_MODE, write_secret_text + + +if TYPE_CHECKING: + from pathlib import Path + + +posix_only = pytest.mark.skipif( + sys.platform == "win32", reason="POSIX permission bits are not modelled on Windows" +) + + +def test_content_round_trips(tmp_path: Path) -> None: + target = tmp_path / "nested" / "auth.json" + payload = json.dumps({"token": "s3cret", "refresh": "r3fresh"}) + write_secret_text(target, payload) + assert json.loads(target.read_text(encoding="utf-8"))["token"] == "s3cret" # noqa: S105 + + +@posix_only +def test_file_is_owner_only(tmp_path: Path) -> None: + target = tmp_path / "auth.json" + write_secret_text(target, "{}") + assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE + + +@posix_only +def test_a_permissive_umask_cannot_widen_the_file(tmp_path: Path) -> None: + previous = os.umask(0) + try: + target = tmp_path / "auth.json" + write_secret_text(target, "{}") + assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE + finally: + os.umask(previous) + + +@posix_only +def test_a_stale_temporary_does_not_leak_its_mode(tmp_path: Path) -> None: + target = tmp_path / "auth.json" + stale = target.with_suffix(target.suffix + ".tmp") + stale.write_text("leftover", encoding="utf-8") + stale.chmod(0o666) + + write_secret_text(target, "{}") + assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE + + +def test_overwriting_an_existing_record_keeps_it_restricted(tmp_path: Path) -> None: + target = tmp_path / "auth.json" + write_secret_text(target, json.dumps({"v": 1})) + write_secret_text(target, json.dumps({"v": 2})) + assert json.loads(target.read_text(encoding="utf-8"))["v"] == 2 + if sys.platform != "win32": + assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE