Coverage for src/lilbee/core/security.py: 100%
74 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Security helpers: path validation, input sanitization, secret-file writes."""
3from __future__ import annotations
5import logging
6import os
7import secrets
8import stat
9import sys
10from collections.abc import Callable, Iterator
11from contextlib import contextmanager, suppress
12from pathlib import Path
14from filelock import FileLock
15from filelock import Timeout as FileLockTimeout
17log = logging.getLogger(__name__)
19OWNER_ONLY_MODE = 0o600
20OWNER_ONLY_DIR_MODE = 0o700
22_Opener = Callable[[str, int], int]
25@contextmanager
26def file_lock_or_warn(path: Path, timeout_s: float) -> Iterator[None]:
27 """Serialize access to *path* across processes via a sibling ``.lock`` file.
29 On timeout the caller proceeds unserialized: losing coordination to a stale
30 lock file is worse than the rare interleave the lock prevents. filelock
31 applies the owner-only mode under its own lock on every acquire and
32 suppresses a refused chmod only for ``PermissionError``; some NFS/FUSE/SMB
33 mounts refuse it with a different ``OSError`` (e.g. ``ENOTSUP``), which
34 must degrade the same way rather than crash the caller.
35 """
36 flock = FileLock(str(path) + ".lock", mode=OWNER_ONLY_MODE)
37 try:
38 flock.acquire(timeout=timeout_s)
39 except FileLockTimeout:
40 log.warning("Timed out waiting for the %s lock; proceeding without it.", path.name)
41 yield
42 return
43 except OSError:
44 log.warning(
45 "Could not fully acquire the %s lock; proceeding without it.",
46 path.name,
47 exc_info=True,
48 )
49 yield
50 return
51 try:
52 yield
53 finally:
54 flock.release()
57class PathTraversalError(ValueError):
58 """Raised when a caller-supplied path escapes its allowed root.
60 Subclasses ``ValueError`` so existing ``except ValueError`` callers keep
61 working, while letting handlers catch *only* a traversal (not an unrelated
62 downstream ``ValueError`` such as a store dimension mismatch).
63 """
66def validate_path_within(path: str | Path, root: Path) -> Path:
67 """Resolve *path* under *root* and verify it stays within it.
69 A relative *path* is taken as relative to *root*.
70 Raises :class:`PathTraversalError` if the resolved path escapes the root.
71 Returns the resolved path on success.
72 """
73 root_resolved = root.resolve()
74 # Relative paths resolve against the CWD, not *root*, so anchor them here;
75 # a traversal inside is still caught by the containment check below.
76 candidate = Path(path)
77 resolved = (candidate if candidate.is_absolute() else root_resolved / candidate).resolve()
78 if not resolved.is_relative_to(root_resolved):
79 raise PathTraversalError(f"Path escapes allowed directory: {path}")
80 return resolved
83def write_private_text(path: Path, text: str) -> None:
84 """Write *text* to *path* so it is owner-only for its entire existence.
86 Writing under the umask and chmod'ing afterwards leaves a window where any
87 local user can read the file, and these callers persist a bearer token and
88 API keys. The temp file is created 0600 and ``os.replace`` keeps that mode.
90 Windows has no POSIX mode bits; there these rely on the inherited
91 ``%LOCALAPPDATA%`` DACL.
92 """
93 _write_atomically(path, text, private_opener)
96def write_text_atomically(path: Path, text: str) -> None:
97 """Write *text* to *path* atomically, with the mode the umask gives a new file."""
98 _write_atomically(path, text, None)
101def _write_atomically(path: Path, text: str, opener: _Opener | None) -> None:
102 """Write a temp file beside *path*, fsync it, then ``os.replace`` it onto *path*.
104 The data is fsynced before the rename, so a crash cannot leave the target empty.
105 """
106 path.parent.mkdir(parents=True, exist_ok=True)
107 tmp = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp")
108 try:
109 with open(tmp, "x", encoding="utf-8", newline="\n", opener=opener) as handle:
110 handle.write(text)
111 handle.flush()
112 os.fsync(handle.fileno())
113 os.replace(tmp, path)
114 except BaseException as exc:
115 with suppress(OSError):
116 tmp.unlink(missing_ok=True)
117 if isinstance(exc, OSError):
118 # The error names the file the user asked to write, not the hidden temp file.
119 raise OSError(exc.errno, exc.strerror, str(path)) from None
120 raise
123def private_opener(path: str, flags: int) -> int:
124 """``open(..., opener=)`` hook that creates a missing file owner-only."""
125 return os.open(path, flags, OWNER_ONLY_MODE)
128def harden_private_file(path: Path) -> None:
129 """Narrow *path* to owner-only, tolerating a file we do not own.
131 A secret file can arrive wider than :func:`write_private_text` leaves it
132 (backup, older release) and is then read indefinitely without a rewrite, so
133 callers narrow on every load. A refused chmod warns rather than raising: a
134 file owned by someone else must not stop the caller from reading it.
136 No-op on Windows, which has no POSIX mode bits.
137 """
138 _narrow_mode(path, OWNER_ONLY_MODE)
141def ensure_private_dir(path: Path) -> None:
142 """Create *path* owner-only, or narrow it when it already exists wider.
144 Same failure semantics as :func:`harden_private_file`.
145 """
146 path.mkdir(mode=OWNER_ONLY_DIR_MODE, parents=True, exist_ok=True)
147 _narrow_mode(path, OWNER_ONLY_DIR_MODE)
150def _narrow_mode(path: Path, mode: int) -> None:
151 """Clear the bits of *path* that *mode* lacks; a refused chmod warns.
153 Never adds a bit: a file its owner made read-only stays read-only.
154 """
155 if sys.platform == "win32": # pragma: no cover - Windows uses the DACL
156 return
157 current = stat.S_IMODE(path.stat().st_mode)
158 narrowed = current & mode
159 if narrowed == current:
160 return
161 try:
162 path.chmod(narrowed)
163 except OSError:
164 log.warning("Could not restrict permissions on %s.", path, exc_info=True)