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 

19OWNER_ONLY_MODE = 0o600 

20OWNER_ONLY_DIR_MODE = 0o700 

21 

22 

23@contextmanager 

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

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

26 

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

28 lock file is worse than the rare interleave the lock prevents. filelock 

29 applies the owner-only mode under its own lock on every acquire and 

30 suppresses a refused chmod only for ``PermissionError``; some NFS/FUSE/SMB 

31 mounts refuse it with a different ``OSError`` (e.g. ``ENOTSUP``), which 

32 must degrade the same way rather than crash the caller. 

33 """ 

34 flock = FileLock(str(path) + ".lock", mode=OWNER_ONLY_MODE) 

35 try: 

36 flock.acquire(timeout=timeout_s) 

37 except FileLockTimeout: 

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

39 yield 

40 return 

41 except OSError: 

42 log.warning( 

43 "Could not fully acquire the %s lock; proceeding without it.", 

44 path.name, 

45 exc_info=True, 

46 ) 

47 yield 

48 return 

49 try: 

50 yield 

51 finally: 

52 flock.release() 

53 

54 

55class PathTraversalError(ValueError): 

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

57 

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

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

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

61 """ 

62 

63 

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

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

66 

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

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

69 Returns the resolved path on success. 

70 """ 

71 root_resolved = root.resolve() 

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

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

74 candidate = Path(path) 

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

76 if not resolved.is_relative_to(root_resolved): 

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

78 return resolved 

79 

80 

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

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

83 

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

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

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

87 atomically. The data is fsynced before the rename, so a crash cannot leave 

88 the target empty. 

89 

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

91 ``%LOCALAPPDATA%`` DACL. 

92 """ 

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

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

95 try: 

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

97 handle.write(text) 

98 handle.flush() 

99 os.fsync(handle.fileno()) 

100 os.replace(tmp_name, path) 

101 except BaseException: 

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

103 raise 

104 

105 

106def private_opener(path: str, flags: int) -> int: 

107 """``open(..., opener=)`` hook that creates a missing file owner-only.""" 

108 return os.open(path, flags, OWNER_ONLY_MODE) 

109 

110 

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

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

113 

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

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

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

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

118 

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

120 """ 

121 _narrow_mode(path, OWNER_ONLY_MODE) 

122 

123 

124def ensure_private_dir(path: Path) -> None: 

125 """Create *path* owner-only, or narrow it when it already exists wider. 

126 

127 Same failure semantics as :func:`harden_private_file`. 

128 """ 

129 path.mkdir(mode=OWNER_ONLY_DIR_MODE, parents=True, exist_ok=True) 

130 _narrow_mode(path, OWNER_ONLY_DIR_MODE) 

131 

132 

133def _narrow_mode(path: Path, mode: int) -> None: 

134 """chmod *path* to *mode* unless it already has it; a refused chmod warns.""" 

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

136 return 

137 if stat.S_IMODE(path.stat().st_mode) == mode: 

138 return 

139 try: 

140 path.chmod(mode) 

141 except OSError: 

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