Create credential files with owner-only permissions (#945)

Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>

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:

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]:

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:

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)

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