Coverage for src/lilbee/core/security.py: 100%

1"""Security helpers: path validation, input sanitization, secret-file writes.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import os 

7import stat 

8import sys 

9import tempfile 

10from collections.abc import Iterator 

11from contextlib import contextmanager 

12from pathlib import Path 

13 

14from filelock import FileLock 

15from filelock import Timeout as FileLockTimeout 

16 

17log = logging.getLogger(__name__) 

18 

19_OWNER_ONLY_MODE = 0o600 

20 

21 

22@contextmanager 

23def file_lock_or_warn(path: Path, timeout_s: float) -> Iterator[None]: 

24 """Serialize access to *path* across processes via a sibling ``.lock`` file. 

25 

26 On timeout the caller proceeds unserialized: losing coordination to a stale 

27 lock file is worse than the rare interleave the lock prevents. 

28 """ 

29 flock = FileLock(str(path) + ".lock") 

30 try: 

31 flock.acquire(timeout=timeout_s) 

32 except FileLockTimeout: 

33 log.warning("Timed out waiting for the %s lock; proceeding without it.", path.name) 

34 yield 

35 return 

36 try: 

37 yield 

38 finally: 

39 flock.release() 

40 

41 

42class PathTraversalError(ValueError): 

43 """Raised when a caller-supplied path escapes its allowed root. 

44 

45 Subclasses ``ValueError`` so existing ``except ValueError`` callers keep 

46 working, while letting handlers catch *only* a traversal (not an unrelated 

47 downstream ``ValueError`` such as a store dimension mismatch). 

48 """ 

49 

50 

51def validate_path_within(path: str | Path, root: Path) -> Path: 

52 """Resolve *path* under *root* and verify it stays within it. 

53 

54 A relative *path* is taken as relative to *root*. 

55 Raises :class:`PathTraversalError` if the resolved path escapes the root. 

56 Returns the resolved path on success. 

57 """ 

58 root_resolved = root.resolve() 

59 # Relative paths resolve against the CWD, not *root*, so anchor them here; 

60 # a traversal inside is still caught by the containment check below. 

61 candidate = Path(path) 

62 resolved = (candidate if candidate.is_absolute() else root_resolved / candidate).resolve() 

63 if not resolved.is_relative_to(root_resolved): 

64 raise PathTraversalError(f"Path escapes allowed directory: {path}") 

65 return resolved 

66 

67 

68def write_private_text(path: Path, text: str) -> None: 

69 """Write *text* to *path* so it is owner-only for its entire existence. 

70 

71 Writing under the umask and chmod'ing afterwards leaves a window where any 

72 local user can read the file, and these callers persist a bearer token and 

73 API keys. ``mkstemp`` creates at 0600 and ``os.replace`` keeps that mode, 

74 atomically. 

75 

76 Windows has no POSIX mode bits; there these rely on the inherited 

77 ``%LOCALAPPDATA%`` DACL. 

78 """ 

79 path.parent.mkdir(parents=True, exist_ok=True) 

80 fd, tmp_name = tempfile.mkstemp(dir=path.parent, suffix=".tmp") 

81 try: 

82 with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: 

83 handle.write(text) 

84 os.replace(tmp_name, path) 

85 except BaseException: 

86 Path(tmp_name).unlink(missing_ok=True) 

87 raise 

88 

89 

90def harden_private_file(path: Path) -> None: 

91 """Narrow *path* to owner-only, tolerating a file we do not own. 

92 

93 A secret file can arrive wider than :func:`write_private_text` leaves it 

94 (backup, older release) and is then read indefinitely without a rewrite, so 

95 callers narrow on every load. A refused chmod warns rather than raising: a 

96 file owned by someone else must not stop the caller from reading it. 

97 

98 No-op on Windows, which has no POSIX mode bits. 

99 """ 

100 if sys.platform == "win32": # pragma: no cover - Windows uses the DACL 

101 return 

102 if stat.S_IMODE(path.stat().st_mode) == _OWNER_ONLY_MODE: 

103 return 

104 try: 

105 path.chmod(_OWNER_ONLY_MODE) 

106 except OSError: 

107 log.warning("Could not restrict permissions on %s.", path, exc_info=True)