Coverage for src/local_deep_research/research_library/utils/__init__.py: 93%
144 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""Shared utility functions for the Research Library."""
3import hashlib
4import os
5import subprocess
6import sys
7from pathlib import Path
8from typing import Optional
9from urllib.parse import urlparse
11from fastapi.responses import JSONResponse
12from loguru import logger
14from ...config.paths import get_library_directory
15from ...database.models.library import Document, DocumentCollection
16from ...security.path_validator import PathValidator
19def escape_like(text: str) -> str:
20 """Escape SQL LIKE/ILIKE wildcards so user input matches literally.
22 ``%`` and ``_`` are LIKE wildcards and ``\\`` is the escape character;
23 without escaping, a query like ``my_note`` would treat ``_`` as "any
24 character" and ``%`` would match anything. Use the result with
25 ``.like(pattern, escape="\\\\")`` / ``.ilike(pattern, escape="\\\\")``.
27 This is the single source of truth for LIKE-escaping across the library
28 and notes query paths (previously copy-pasted in NoteService,
29 LibraryService and unified_search_routes).
30 """
31 return (
32 (text or "")
33 .replace("\\", "\\\\")
34 .replace("%", "\\%")
35 .replace("_", "\\_")
36 )
39def is_downloadable_domain(url: str) -> bool:
40 """Check if URL is from a downloadable academic domain using proper URL parsing.
42 **This is a relevance filter, NOT the SSRF control.** It answers "is this
43 worth trying to download", not "is this safe to fetch". The security gate is
44 ``security.safe_requests.safe_get``, which every outbound fetch in
45 ``research_library/services/download_service.py`` goes through: it calls
46 ``ssrf_validator.validate_url()``, raises on failure, re-validates after
47 redirects, and always blocks cloud-metadata addresses
48 (``ssrf_validator.ALWAYS_BLOCKED_METADATA_IPS``) even when private IPs are
49 permitted. A resource URL also cannot enter the system unchecked —
50 ``web/routers/api.py::api_add_resource`` validates before inserting a row.
52 Saying so explicitly because the shape of this function invites the opposite
53 reading, and a 2026-08 security audit made exactly that mistake: the
54 PDF-shaped short-circuits below return True *before* the hostname allowlist
55 is consulted, which looks like an allowlist bypass. It is untidy — a
56 PDF-shaped URL does pass this filter whatever its host — but it grants no
57 reach, because nothing downstream treats a True from here as permission to
58 fetch. If you are adding a new outbound fetch, do NOT rely on this function
59 for safety; call ``safe_get``.
60 """
61 try:
62 if not url:
63 return False
65 parsed = urlparse(url.lower())
66 hostname = parsed.hostname or ""
67 path = parsed.path or ""
68 query = parsed.query or ""
70 # Check for direct PDF files
71 if path.endswith(".pdf") or ".pdf?" in url.lower():
72 return True
74 # List of downloadable academic domains
75 downloadable_domains = [
76 "arxiv.org",
77 "biorxiv.org",
78 "medrxiv.org",
79 "ncbi.nlm.nih.gov",
80 "pubmed.ncbi.nlm.nih.gov",
81 "europepmc.org",
82 "semanticscholar.org",
83 "researchgate.net",
84 "academia.edu",
85 "sciencedirect.com",
86 "springer.com",
87 "nature.com",
88 "wiley.com",
89 "ieee.org",
90 "acm.org",
91 "plos.org",
92 "frontiersin.org",
93 "mdpi.com",
94 "acs.org",
95 "rsc.org",
96 "tandfonline.com",
97 "sagepub.com",
98 "oxford.com",
99 "cambridge.org",
100 "bmj.com",
101 "nejm.org",
102 "thelancet.com",
103 "jamanetwork.com",
104 "annals.org",
105 "ahajournals.org",
106 "cell.com",
107 "science.org",
108 "pnas.org",
109 "elifesciences.org",
110 "embopress.org",
111 "journals.asm.org",
112 "microbiologyresearch.org",
113 "jvi.asm.org",
114 "genome.cshlp.org",
115 "genetics.org",
116 "g3journal.org",
117 "plantphysiol.org",
118 "plantcell.org",
119 "aspb.org",
120 "bioone.org",
121 "company-of-biologists.org",
122 "biologists.org",
123 "jeb.biologists.org",
124 "dmm.biologists.org",
125 "bio.biologists.org",
126 "doi.org",
127 "ssrn.com",
128 "openreview.net",
129 ]
131 # Check if hostname matches any downloadable domain
132 for domain in downloadable_domains:
133 if hostname == domain or hostname.endswith("." + domain):
134 return True
136 # Special case for PubMed which might appear in path
137 if "pubmed" in hostname or "/pubmed/" in path:
138 return True
140 # Check for PDF in path or query parameters
141 if "/pdf/" in path or "type=pdf" in query or "format=pdf" in query:
142 return True
144 return False
146 except Exception:
147 logger.warning(f"Error parsing URL {url}")
148 return False
151def is_downloadable_url(url: str) -> bool:
152 """Check if a URL is downloadable (academic domain or direct PDF link).
154 This is the single source of truth for downloadability checks.
155 Combines domain checking with PDF extension/path detection.
157 Args:
158 url: The URL to check
160 Returns:
161 True if the URL is from a downloadable academic domain or is a direct PDF link
162 """
163 return is_downloadable_domain(url)
166def get_document_for_resource(session, resource):
167 """Get Document for a ResearchResource.
169 Checks resource.document_id first (library resources point directly
170 to existing Documents), falls back to Document.resource_id lookup
171 (web downloads create Documents with resource_id set).
172 """
173 if resource.document_id:
174 return (
175 session.query(Document).filter_by(id=resource.document_id).first()
176 )
177 return session.query(Document).filter_by(resource_id=resource.id).first()
180def get_url_hash(url: str) -> str:
181 """
182 Generate a SHA256 hash of a URL.
184 Args:
185 url: The URL to hash
187 Returns:
188 The SHA256 hash of the URL
189 """
190 return hashlib.sha256(url.lower().encode()).hexdigest()
193def ensure_in_collection(
194 session, document_id: str, collection_id: str
195) -> "DocumentCollection":
196 """Get or create a DocumentCollection link between a document and a collection.
198 Args:
199 session: SQLAlchemy session
200 document_id: UUID of the document
201 collection_id: UUID of the collection
203 Returns:
204 The existing or newly created DocumentCollection row
205 """
206 existing = (
207 session.query(DocumentCollection)
208 .filter_by(document_id=document_id, collection_id=collection_id)
209 .first()
210 )
211 if existing:
212 return existing
214 doc_collection = DocumentCollection(
215 document_id=document_id,
216 collection_id=collection_id,
217 indexed=False,
218 )
219 session.add(doc_collection)
220 return doc_collection
223def _reject_unsafe_username_component(username: str) -> None:
224 """Fail closed unless ``username`` is a single, safe path component.
226 ``apply_user_subdir`` joins the username directly into a filesystem path,
227 so any value that could let a downstream ``.resolve()`` escape the
228 library base would be an arbitrary-directory read/write primitive. This
229 mirrors registration's *exact* predicate
230 (``username.replace('_','').replace('-','').isalnum()``) so the two
231 checks can never diverge: any Unicode letters/digits registration accepts
232 (e.g. accented Latin, Cyrillic, CJK) are accepted here too, alongside
233 plain ``[A-Za-z0-9_-]``. ``str.isalnum()`` still rejects every
234 path-escape vector — separators (``/``, ``\\``), ``..``, ``:``, dots,
235 spaces, and control characters are never alphanumeric — so this remains a
236 fail-closed traversal guard; it does not attempt to additionally block
237 reserved device names or homoglyphs, since neither is meaningfully closed
238 by matching registration's charset. We reject rather than sanitize/hash
239 so a valid account keeps a human-readable per-user directory, and reject
240 rather than fall back to the shared base (which would risk cross-user
241 co-mingling).
242 """
243 stripped = username.replace("_", "").replace("-", "")
244 if not stripped or not stripped.isalnum():
245 raise ValueError("Unsafe username for per-user library path")
248def apply_user_subdir(
249 base_path, username: Optional[str], shared_library: bool = False
250) -> Path:
251 """Return the per-user library directory under ``base_path``.
253 Single source of truth for the per-user library layout (issue #5521):
254 downloaded PDFs live under ``<base>/<username>`` so two users' per-user
255 autoincrement ``resource_id`` filenames (e.g. ``pdfs/5.pdf``) can no
256 longer collide in one shared directory. In shared-library mode — or when
257 no username is available — the base directory itself is returned.
259 The ``username`` is validated as a single safe path component before it is
260 joined (see :func:`_reject_unsafe_username_component`); an unsafe value
261 raises ``ValueError`` rather than escaping the base or silently sharing it.
263 Unlike :func:`get_library_storage_path`, this works from an
264 already-resolved base path plus an explicit ``shared_library`` flag, so
265 background threads (the scheduler download path) and snapshot-based
266 callers (collection/library search engines) get the same layout without
267 depending on request-context settings resolution.
268 """
269 # Expand $VARs (parity with the write path in DownloadService.__init__);
270 # the read-side resolvers previously only expanduser()'d, so a storage_path
271 # containing an env-var token resolved to a different, non-existent path on
272 # read than on write (files became unfindable, tracker state corrupted).
273 base = Path(os.path.expandvars(str(base_path))).expanduser().resolve()
274 # ``shared_library`` removes the per-user directory boundary. Because both
275 # ``research_library.shared_library`` and ``research_library.storage_path``
276 # are user-editable, a multi-tenant attacker could otherwise set their own
277 # shared_library=true and point storage_path at a victim's directory to
278 # read/overwrite their library PDFs. So shared mode only takes effect when
279 # an operator has explicitly enabled it via the environment; otherwise the
280 # per-user subdirectory is always enforced.
281 if (shared_library and _shared_library_allowed()) or not username:
282 return base
283 _reject_unsafe_username_component(username)
284 return base / username
287def _shared_library_allowed() -> bool:
288 """Whether the operator enabled shared-library mode (env-only gate).
290 Environment-only, mirroring ``filesystem_pdf_storage_allowed`` and
291 ``policy.allow_unprotected_egress``: shared mode drops the per-user
292 directory isolation, so it cannot be turned on through the user-writable
293 settings API — only via
294 ``LDR_RESEARCH_LIBRARY_ALLOW_SHARED_LIBRARY=true``.
295 """
296 from ...settings.env_registry import get_env_setting
298 return bool(get_env_setting("research_library.allow_shared_library", False))
301def _legacy_read_fallback_allowed() -> bool:
302 """Whether the operator opted into the legacy shared-root READ fallback.
304 Environment-only, mirroring ``_shared_library_allowed`` /
305 ``filesystem_pdf_storage_allowed``. The legacy fallback resolves a
306 per-user read miss against the shared root derived from the user-editable
307 ``research_library.storage_path``. Because a user could point their own
308 ``storage_path`` at another user's directory, and per-user autoincrement
309 resource ids collide by construction, that fallback is a cross-tenant
310 read primitive on a multi-tenant instance. It is therefore OFF by default
311 and can only be enabled by the operator via
312 ``LDR_RESEARCH_LIBRARY_ALLOW_LEGACY_READ_FALLBACK=true`` — never through
313 the user-writable settings API. When off, reads resolve strictly within
314 the caller's own per-user root.
315 """
316 from ...settings.env_registry import get_env_setting
318 return bool(
319 get_env_setting("research_library.allow_legacy_read_fallback", False)
320 )
323def get_library_storage_path(username: str) -> Path:
324 """
325 Get the storage path for a user's library.
327 Uses the settings system which respects environment variable overrides:
328 - research_library.storage_path: Base path for library storage
329 - research_library.shared_library: If true, all users share the same directory
331 Args:
332 username: The username
334 Returns:
335 Path to the library storage directory
336 """
337 from ...utilities.db_utils import get_settings_manager
339 settings = get_settings_manager()
341 # Get the base path from settings (uses centralized path, respects LDR_DATA_DIR)
342 base_path = settings.get_setting(
343 "research_library.storage_path",
344 str(get_library_directory()),
345 )
347 # Check if shared library mode is enabled
348 shared_library = settings.get_setting(
349 "research_library.shared_library", False
350 )
352 storage_path = apply_user_subdir(base_path, username, shared_library)
354 # Lazy import to avoid import cycles (this module is imported early by
355 # several bootstrap paths).
356 from ...security.directory_creation import create_directory
358 # ``apply_user_subdir`` has already expanded/resolved the base path and
359 # validated the per-user component; route the actual mkdir through the
360 # audited chokepoint, keeping containment to the resolved base so a
361 # symlinked parent that escapes it is still rejected.
362 resolved_base = (
363 Path(os.path.expandvars(str(base_path))).expanduser().resolve()
364 )
365 create_directory(
366 storage_path,
367 context="per-user library storage",
368 root=resolved_base,
369 )
370 return storage_path
373def open_file_location(file_path: str) -> bool:
374 """
375 Open the file location in the system file manager.
377 Args:
378 file_path: Path to the file
380 Returns:
381 True if successful, False otherwise
382 """
383 try:
384 # Validate path is safe (blocks system dirs, path traversal)
385 validated = PathValidator.validate_local_filesystem_path(file_path)
386 folder = str(validated.parent)
387 if sys.platform == "win32": 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 os.startfile(folder)
389 elif sys.platform == "darwin": # macOS
390 result = subprocess.run(
391 ["open", folder], capture_output=True, text=True, shell=False
392 )
393 if result.returncode != 0: 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true
394 logger.error(f"Failed to open folder on macOS: {result.stderr}")
395 return False
396 else: # Linux
397 result = subprocess.run(
398 ["xdg-open", folder],
399 capture_output=True,
400 text=True,
401 shell=False,
402 )
403 if result.returncode != 0:
404 logger.error(f"Failed to open folder on Linux: {result.stderr}")
405 return False
406 return True
407 except Exception:
408 logger.exception("Failed to open file location")
409 return False
412def get_absolute_library_path(
413 relative_path: str, username: str
414) -> Optional[Path]:
415 """
416 Get the absolute path from a relative library path.
418 Uses PathValidator to prevent path traversal attacks.
420 Args:
421 relative_path: The relative path from library root
422 username: The username
424 Returns:
425 The absolute path, or None if the path is unsafe
426 """
427 library_root = get_library_storage_path(username)
428 try:
429 # Use PathValidator to prevent path traversal attacks
430 safe_path = PathValidator.validate_safe_path(
431 relative_path, str(library_root)
432 )
433 if safe_path is None: 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true
434 return None
435 result = Path(safe_path)
436 if result.is_symlink():
437 logger.warning(f"Symlink blocked: {relative_path}")
438 return None
439 return result
440 except ValueError:
441 logger.warning(f"Path traversal blocked: {relative_path}")
442 return None
445def _resolve_within_root(
446 relative_path: str, library_root: Path
447) -> Optional[Path]:
448 """Validate ``relative_path`` inside ``library_root`` (traversal + symlink
449 safe). Returns the absolute path, or ``None`` if unsafe."""
450 try:
451 safe_path = PathValidator.validate_safe_path(
452 relative_path, str(library_root)
453 )
454 if safe_path is None: 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 return None
456 result = Path(safe_path)
457 if result.is_symlink():
458 logger.warning(f"Symlink blocked: {relative_path}")
459 return None
460 return result
461 except ValueError:
462 logger.warning(f"Path traversal blocked: {relative_path}")
463 return None
466def get_absolute_path_from_settings(
467 relative_path: str,
468 username: Optional[str] = None,
469 *,
470 allow_legacy_fallback: bool = True,
471 settings_manager=None,
472) -> Optional[Path]:
473 """
474 Get absolute path using settings manager for library root.
476 Uses PathValidator to prevent path traversal attacks.
478 ``settings_manager`` lets a caller supply an explicit, already-scoped
479 settings manager instead of the ambient ``get_settings_manager()``. This
480 matters in background/scheduler threads with no Flask request context:
481 there the ambient manager resolves to a *db-less* manager that reads only
482 env vars and the defaults file, so a user's UI-customized
483 ``research_library.storage_path`` / ``research_library.shared_library``
484 (stored only in their per-user encrypted DB) is invisible and the path
485 silently resolves against the default library location — the wrong root.
486 ``DownloadService`` passes its captured ``self.settings`` here so these
487 reads stay consistent with the per-user root it computed in ``__init__``
488 (issue #5521). When ``None`` the ambient manager is used, preserving the
489 request-context behavior for existing callers.
491 When ``username`` is provided the path is resolved against that user's
492 per-user library directory (issue #5521). If the file is not found there,
493 it falls back to the legacy shared root so PDFs downloaded before per-user
494 isolation still load correctly (no data loss). When ``username``
495 is ``None`` the legacy shared-root behavior is preserved unchanged for
496 callers that have no user context.
498 ``allow_legacy_fallback`` must be ``False`` for *destructive* callers
499 (delete / unlink). The legacy shared root is not per-user namespaced, and
500 per-user autoincrement resource ids collide by construction (each user has
501 an independently-numbered database), so filenames like ``pdfs/3.pdf`` can
502 map to *another* tenant's file in the shared root. Resolving-then-unlinking
503 that path would delete a different user's PDF. Because migration 0029 does
504 not move pre-isolation files into per-user subdirectories, the per-user
505 location is always empty for legacy documents and the fallback would fire
506 deterministically. Destructive callers therefore resolve only within the
507 caller's own per-user root; a missing file there means "nothing of mine to
508 delete" rather than reaching into shared storage. As a fail-closed guard,
509 a destructive call (``allow_legacy_fallback=False``) with a falsy
510 ``username`` raises ``ValueError`` rather than resolving to the bare
511 shared root: without a user context there is no per-user directory to
512 scope the unlink to, and silently resolving into the shared root could
513 unlink another tenant's file.
515 Even for read-only callers the legacy shared-root fallback is a
516 cross-tenant read primitive on a multi-tenant instance: the shared root is
517 derived from the user-editable ``research_library.storage_path``, so a user
518 can point their own ``storage_path`` at another user's directory and — via
519 the colliding-id fallback — read that user's PDFs. The read fallback is
520 therefore gated behind the operator-only
521 ``research_library.allow_legacy_read_fallback`` env setting and is OFF by
522 default; when off, reads resolve strictly within the caller's own per-user
523 root.
525 Args:
526 relative_path: The relative path from library root
527 username: Optional username for per-user resolution
528 allow_legacy_fallback: Permit the legacy shared-root fallback for
529 read-only callers. Pass ``False`` from delete/unlink paths. Even
530 when ``True``, the fallback only fires if the operator enabled
531 ``research_library.allow_legacy_read_fallback``.
532 settings_manager: Optional explicit settings manager to read
533 ``storage_path`` / ``shared_library`` from. Defaults to the
534 ambient ``get_settings_manager()``; pass a captured, user-scoped
535 manager from background/scheduler threads (see above).
537 Returns:
538 The absolute path, or None if the path is unsafe
540 Raises:
541 ValueError: If ``allow_legacy_fallback`` is ``False`` (destructive
542 caller) and ``username`` is falsy — fail closed rather than
543 resolve within the shared root.
544 """
545 # Fail closed: a destructive caller with no user context has no per-user
546 # directory to scope its unlink to. apply_user_subdir(base, "", ...) would
547 # otherwise return the bare shared root, so resolving+unlinking there could
548 # destroy another tenant's colliding file. Read-only callers
549 # (allow_legacy_fallback=True) keep the legacy no-username behavior.
550 if not allow_legacy_fallback and not username:
551 raise ValueError(
552 "Refusing to resolve a destructive library path without a user "
553 "context (would resolve within the shared root)."
554 )
556 if settings_manager is None:
557 from ...utilities.db_utils import get_settings_manager
559 settings = get_settings_manager()
560 else:
561 settings = settings_manager
562 base_path = (
563 Path(
564 os.path.expandvars(
565 settings.get_setting(
566 "research_library.storage_path",
567 str(get_library_directory()),
568 )
569 )
570 )
571 .expanduser()
572 .resolve()
573 )
574 shared_library = settings.get_setting(
575 "research_library.shared_library", False
576 )
577 per_user_root = apply_user_subdir(base_path, username, shared_library)
579 if not relative_path:
580 # An empty path asks for the (per-user) library root itself.
581 return per_user_root
583 primary = _resolve_within_root(relative_path, per_user_root)
584 # Legacy fallback: when the per-user location has no file, look in the
585 # legacy shared root where pre-isolation downloads still live. Gated three
586 # ways: (1) read-only callers only — destructive callers pass
587 # allow_legacy_fallback=False so they never unlink a colliding file that
588 # belongs to another tenant; (2) the operator must have opted into the
589 # shared-root read fallback, since it is a cross-tenant read primitive
590 # when a user points storage_path at another user's directory; (3) the
591 # per-user root must actually differ from the shared root.
592 if (
593 allow_legacy_fallback
594 and _legacy_read_fallback_allowed()
595 and per_user_root != base_path
596 and (primary is None or not primary.exists())
597 ):
598 legacy = _resolve_within_root(relative_path, base_path)
599 if legacy is not None and legacy.exists(): 599 ↛ 601line 599 didn't jump to line 601 because the condition on line 599 was always true
600 return legacy
601 return primary
604def handle_api_error(operation: str, error: Exception, status_code: int = 500):
605 """
606 Handle API errors consistently - log internally, return generic message to user.
608 This prevents information exposure by logging full error details internally
609 while returning a generic message to the user.
611 Args:
612 operation: Description of the operation that failed (for logging)
613 error: The exception that occurred
614 status_code: HTTP status code to return (default: 500)
616 Returns:
617 Flask JSON response tuple (response, status_code)
618 """
619 # Log the full error internally with stack trace
620 logger.exception(f"Error during {operation}")
622 # Return generic message to user (no internal details exposed)
623 return JSONResponse(
624 {
625 "success": False,
626 "error": "An internal error occurred. Please try again or contact support.",
627 },
628 status_code=status_code,
629 )