mirror of
https://github.com/usestrix/strix.git
synced 2026-10-02 02:13:43 +00:00
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
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, open_secret_file, 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
|
|
|
|
|
|
@posix_only
|
|
def test_stream_is_private_before_writing_and_atomic_on_completion(tmp_path: Path) -> None:
|
|
target = tmp_path / "shared" / "result.zip"
|
|
target.parent.mkdir(mode=0o777)
|
|
previous = os.umask(0)
|
|
try:
|
|
with open_secret_file(target) as stream:
|
|
assert stat.S_IMODE(os.fstat(stream.fileno()).st_mode) == SECRET_FILE_MODE
|
|
assert not target.exists()
|
|
stream.write(b"private customer source\x00\xff")
|
|
stream.flush()
|
|
assert not target.exists()
|
|
assert target.read_bytes() == b"private customer source\x00\xff"
|
|
assert stat.S_IMODE(target.stat().st_mode) == SECRET_FILE_MODE
|
|
finally:
|
|
os.umask(previous)
|
|
|
|
|
|
def test_failed_stream_preserves_previous_result_and_removes_partial_file(tmp_path: Path) -> None:
|
|
target = tmp_path / "result.zip"
|
|
write_secret_text(target, "previous result")
|
|
with pytest.raises(RuntimeError, match="interrupted write"), open_secret_file(target) as stream:
|
|
stream.write(b"partial customer source")
|
|
raise RuntimeError("interrupted write")
|
|
assert target.read_text() == "previous result"
|
|
assert list(tmp_path.iterdir()) == [target]
|