Coverage for src/local_deep_research/config/paths.py: 99%
106 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Centralized path configuration for Local Deep Research.
3Handles database location using platformdirs for proper user data storage.
4"""
6import errno
7import hashlib
8import os
9import stat
10from pathlib import Path
12import platformdirs
13from loguru import logger
16def get_data_directory() -> Path:
17 """
18 Get the appropriate data directory for storing application data.
19 Uses platformdirs to get platform-specific user data directory.
21 Environment variable:
22 LDR_DATA_DIR: Override the default data directory location.
23 All subdirectories (research_outputs, cache, logs, database)
24 will be created under this directory.
26 Returns:
27 Path to data directory
28 """
29 # Check for explicit override via environment variable
30 custom_path = os.getenv("LDR_DATA_DIR")
31 if custom_path:
32 if not Path(custom_path).is_absolute():
33 raise ValueError("LDR_DATA_DIR must be an absolute path")
34 data_dir = Path(custom_path).resolve()
35 # Reject only control characters that are never valid in a path and
36 # could break a SQL/ATTACH statement (null byte, newline, carriage
37 # return). Quotes are intentionally NOT rejected here: legitimate POSIX
38 # home directories contain apostrophes (e.g. /home/O'Brien/ldr), and the
39 # actual ATTACH-DATABASE sink (backup_service) already validates the
40 # full interpolated path against a stricter denylist before use.
41 # The offending path is not echoed, mirroring the redaction below.
42 path_str = str(data_dir)
43 if any(c in path_str for c in ("\0", "\n", "\r")):
44 raise ValueError(
45 "LDR_DATA_DIR contains unsafe control characters "
46 "(null byte, newline, or carriage return)"
47 )
48 logger.debug(
49 f"Using custom data directory from LDR_DATA_DIR: {data_dir}"
50 )
51 return data_dir
53 # Use platformdirs for platform-specific user data directory
54 # Windows: C:\Users\Username\AppData\Local\local-deep-research
55 # macOS: ~/Library/Application Support/local-deep-research
56 # Linux: ~/.local/share/local-deep-research
57 data_dir = Path(platformdirs.user_data_dir("local-deep-research"))
58 # Log only the directory pattern, not the full path which may contain username
59 logger.debug(
60 f"Using platformdirs data directory pattern: .../{data_dir.name}"
61 )
63 return data_dir
66def get_custom_data_directory() -> Path | None:
67 """
68 Get the data directory that ``LDR_DATA_DIR`` selects, if it is set.
70 ``None`` means ``LDR_DATA_DIR`` is unset or empty, so
71 :func:`get_data_directory` uses the platform default. Otherwise the result
72 is exactly what :func:`get_data_directory` returns, including its
73 validation errors. ``LDR_DATA_DIR`` is the only environment variable that
74 moves the data directory, so this is what a status report should show
75 when it says the location is custom.
77 Returns:
78 Path selected by ``LDR_DATA_DIR``, or ``None`` for the platform default
79 """
80 if not os.getenv("LDR_DATA_DIR"):
81 return None
82 return get_data_directory()
85def _ensure_dir(
86 subdir: str, label: str | None = None, mode: int | None = None
87) -> Path:
88 """Create (if needed) and return ``<data_dir>/<subdir>``.
90 Args:
91 subdir: Path segment below the data directory. May be multi-segment
92 (e.g. ``"encrypted_databases/backups"``).
93 label: If provided, emit a debug log ``"Using {label} directory:
94 {path}"``. ``None`` suppresses logging.
95 mode: Optional creation mode for a new directory.
97 Returns:
98 The ensured directory path.
99 """
100 path = get_data_directory() / subdir
101 if mode is None:
102 path.mkdir( # Safe: bootstrap data subdirectory
103 parents=True, exist_ok=True
104 )
105 else:
106 path.mkdir( # Safe: bootstrap data subdirectory
107 parents=True, exist_ok=True, mode=mode
108 )
109 if label is not None:
110 logger.debug(f"Using {label} directory: {path}")
111 return path
114#: ``DirectoryNotPrivateError.reason`` values.
115OWNED_BY_ANOTHER_ACCOUNT = "owned_by_another_account"
116MODE_NOT_KEPT = "mode_not_kept"
118#: What an operator can do when report file backups are refused.
119REPORT_BACKUP_REMEDY = (
120 "Keep the data directory (LDR_DATA_DIR) on a Linux filesystem path or a "
121 "Docker named volume owned by the account that runs Local Deep Research, "
122 "or turn off report.enable_file_backup"
123)
126class DirectoryNotPrivateError(PermissionError):
127 """A directory cannot be restricted to the current OS account.
129 ``reason`` is ``OWNED_BY_ANOTHER_ACCOUNT`` or ``MODE_NOT_KEPT``.
130 """
132 def __init__(self, message: str, reason: str):
133 super().__init__(message)
134 self.reason = reason
137def restrict_directory_to_owner(path: Path) -> None:
138 """Set ``path`` to mode 0700 on POSIX and verify the result.
140 Other systems have no POSIX mode bits, so this does nothing there; the
141 directory's access-control list decides who can read it.
143 The result is checked because some mounts accept chmod without keeping
144 it, or show another owner: Docker Desktop bind mounts of Windows
145 folders, WSL drives such as /mnt/c, CIFS/SMB shares and exFAT.
147 Raises:
148 DirectoryNotPrivateError: another OS account owns the directory, or
149 its mode is not exactly 0700 after chmod.
150 OSError: chmod or stat failed for another reason.
151 """
152 if os.name != "posix": 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 return
154 try:
155 path.chmod(0o700)
156 except PermissionError:
157 # Only the owner may chmod. Report a foreign owner as such below;
158 # any other cause propagates unchanged.
159 if path.stat().st_uid == os.geteuid():
160 raise
161 info = path.stat()
162 if info.st_uid != os.geteuid():
163 raise DirectoryNotPrivateError(
164 f"{path} is not private: it is owned by uid {info.st_uid}, "
165 f"but this process runs as uid {os.geteuid()}",
166 OWNED_BY_ANOTHER_ACCOUNT,
167 )
168 mode = stat.S_IMODE(info.st_mode)
169 if mode != 0o700:
170 raise DirectoryNotPrivateError(
171 f"{path} is not private: its filesystem kept mode {mode:04o} "
172 "after chmod 0700",
173 MODE_NOT_KEPT,
174 )
177def describe_private_directory_refusal(error: OSError) -> str:
178 """Say why ``restrict_directory_to_owner`` refused a directory.
180 Fixed text for log lines: it never echoes the error message, only its
181 category and, for other OS errors, the symbolic errno.
182 """
183 reason = getattr(error, "reason", None)
184 if reason == OWNED_BY_ANOTHER_ACCOUNT:
185 return "another OS account owns it"
186 if reason == MODE_NOT_KEPT:
187 return "its filesystem did not keep mode 0700"
188 code = errno.errorcode.get(getattr(error, "errno", None) or 0)
189 return "it could not be created or set to mode 0700" + (
190 f" ({code})" if code else ""
191 )
194def get_research_outputs_directory() -> Path:
195 """Get the directory for research outputs and optional report backups.
197 A missing directory is created with mode 0700 (subject to the umask).
198 An existing directory is returned unchanged: the web application
199 restricts it at startup (``restrict_research_outputs_directory``), and
200 ``FileReportStorage`` restricts and verifies it before writing.
202 Returns:
203 Path to research outputs directory
204 """
205 return _ensure_dir("research_outputs", label="research outputs", mode=0o700)
208def restrict_research_outputs_directory() -> bool:
209 """Restrict the research outputs directory to this OS account.
211 The web application calls this once at startup, so a directory left at
212 0755 by an older release becomes 0700 even when no user has report file
213 backups enabled. An ``OSError`` is logged with its remedy, not raised:
214 ``FileReportStorage`` then refuses to write report backups there, and
215 database report storage is unaffected.
217 Returns:
218 False when the directory could not be made private, otherwise True
219 (always True on non-POSIX systems, where nothing is changed).
220 """
221 try:
222 restrict_directory_to_owner(get_research_outputs_directory())
223 except OSError as exc:
224 cause = describe_private_directory_refusal(exc)
225 logger.warning(
226 f"Could not make the research outputs directory private: {cause}. "
227 "Report file backups will be refused until this is fixed; "
228 f"reports are still saved in the database. {REPORT_BACKUP_REMEDY}."
229 )
230 return False
231 return True
234def get_journal_data_directory() -> Path:
235 """Get the directory for downloaded journal quality data files.
237 Contains openalex_sources.json.gz, doaj_journals.json, and the
238 compiled journal_reference.db. Fetched on first use from
239 OpenAlex and DOAJ APIs.
241 Returns:
242 Path to journal data directory
243 """
244 return _ensure_dir("journal_data")
247def get_cache_directory() -> Path:
248 """
249 Get the directory for storing cache files (search cache, etc.).
251 Returns:
252 Path to cache directory
253 """
254 return _ensure_dir("cache", label="cache")
257def get_logs_directory() -> Path:
258 """
259 Get the directory for storing log files.
261 Returns:
262 Path to logs directory
263 """
264 return _ensure_dir("logs", label="logs")
267def get_encrypted_database_path() -> Path:
268 """Get the path to the encrypted databases directory.
270 Returns:
271 Path to the encrypted databases directory
272 """
273 return _ensure_dir("encrypted_databases")
276def get_user_database_filename(username: str) -> str:
277 """Get the database filename for a specific user.
279 Args:
280 username: The username to generate a filename for
282 Returns:
283 The database filename (not full path) for the user
285 Raises:
286 ValueError: If username is empty/None. Registration already enforces
287 a non-empty username, but this is the actual hash chokepoint:
288 sha256("") is a well-known constant, so silently accepting an
289 empty username would let two different "no username" call sites
290 collide into one shared, deterministic database filename instead
291 of failing loudly. Defense-in-depth -- see #5481.
292 """
293 if not username:
294 raise ValueError(
295 "Cannot compute a user database filename for an empty username"
296 )
297 # Use username hash to avoid filesystem issues with special characters
298 username_hash = hashlib.sha256(username.encode()).hexdigest()[:16]
299 return f"ldr_user_{username_hash}.db"
302def get_library_directory() -> Path:
303 """
304 Get the directory for storing library files (documents, PDFs, etc.).
306 Returns:
307 Path to library directory
308 """
309 return _ensure_dir("library", label="library")
312def get_config_directory() -> Path:
313 """
314 Get the directory for storing configuration files.
316 Returns:
317 Path to config directory
318 """
319 return _ensure_dir("config", label="config")
322def get_models_directory() -> Path:
323 """
324 Get the directory for storing downloaded models.
326 Returns:
327 Path to models directory
328 """
329 return _ensure_dir("models", label="models")
332def get_sentence_transformers_home() -> str | None:
333 """
334 Get the sentence-transformers ecosystem cache override, if set.
336 Environment variable:
337 SENTENCE_TRANSFORMERS_HOME: Third-party override (honored by the
338 sentence-transformers library itself) for where model
339 snapshots are stored. Read at call time so the live
340 environment always wins, mirroring how the library
341 resolves its own cache root. The raw value is returned
342 (no Path coercion, empty string stays empty) so callers
343 can preserve the library's precedence exactly.
345 Returns:
346 The override path as a string, or None when unset (callers fall
347 back to the HuggingFace Hub default cache).
348 """
349 return os.getenv("SENTENCE_TRANSFORMERS_HOME")
352def get_backup_directory() -> Path:
353 """Get the base backup directory for all users."""
354 return _ensure_dir("encrypted_databases/backups")
357def get_user_backup_directory(username: str) -> Path:
358 """Get backup directory for a specific user.
360 Raises:
361 ValueError: If username is empty/None. Same fail-closed guard as
362 get_user_database_filename: sha256("") is a well-known constant,
363 so silently accepting an empty username would collapse every
364 "no username" call site into one shared, deterministic backup
365 directory. Defense-in-depth -- see #5481.
366 """
367 if not username:
368 raise ValueError(
369 "Cannot compute a user backup directory for an empty username"
370 )
371 from local_deep_research.security.directory_creation import create_directory
373 username_hash = hashlib.sha256(username.encode()).hexdigest()[:16]
374 backup_directory = get_backup_directory()
375 user_backup_dir = create_directory(
376 backup_directory / username_hash,
377 context="user backup directory",
378 root=backup_directory,
379 mode=0o700,
380 )
381 # Enforce 0o700 regardless of umask (mkdir mode is umask-masked)
382 user_backup_dir.chmod(0o700)
383 return user_backup_dir
386# Convenience functions for backward compatibility
387def get_data_dir() -> str:
388 """Get data directory as string for backward compatibility."""
389 return str(get_data_directory())