Compare commits

...
6 changed files with 104 additions and 22 deletions
+3 -8
View File
@@ -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:
+2 -4
View File
@@ -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]:
+1 -2
View File
@@ -496,8 +496,7 @@ func (m Model) setupHintsView(width int) string {
key := lipgloss.NewStyle().Foreground(white).Render
label := render.Dim().Render
hint := func(k, text string) string { return key(k) + label(" "+text) }
left := hint("enter", "launch scan") + label(" ") + hint("shift+enter", "newline") +
label(" ") + hint("ctrl+c", "quit")
left := hint("enter", "launch scan") + label(" ") + hint("ctrl+c", "quit")
if lipgloss.Width(left) > inner {
left = hint("enter", "launch scan")
}
+2 -8
View File
@@ -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:
+31
View File
@@ -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)
+65
View File
@@ -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