Coverage for src/local_deep_research/research_library/services/pdf_storage_manager.py: 94%
211 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2PDF Storage Manager for Research Library
4Handles PDF storage across three modes:
5- none: Don't store PDFs (text-only)
6- filesystem: Store PDFs unencrypted on disk (fast, external tool compatible)
7- database: Store PDFs encrypted in SQLCipher database (secure, portable)
8"""
10import hashlib
11import re
12from datetime import datetime, UTC
13from pathlib import Path
14from typing import Optional, Tuple
15from urllib.parse import urlparse
17from loguru import logger
18from sqlalchemy.orm import Session
20from ...constants import FILE_PATH_SENTINELS
21from ...database.models.library import Document, DocumentBlob
22from ...security.path_validator import PathValidator
25# Default storage cap for individual PDFs (megabytes). Mirrors the
26# upload-validator cap (`FileUploadValidator.MAX_FILE_SIZE`, configurable
27# via `LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB`) so a file that passes the
28# upload step won't be silently dropped at storage time. The runtime
29# value comes from the `research_library.max_pdf_size_mb` setting; this
30# constant is the shared fallback used by every code-level default so the
31# limit doesn't drift across files.
32DEFAULT_MAX_PDF_SIZE_MB = 3072 # 3 GB
35def filesystem_pdf_storage_allowed() -> bool:
36 """Return whether the operator enabled unencrypted filesystem PDF storage.
38 Environment-only operator gate mirroring
39 ``policy.allow_unprotected_egress``: the ``filesystem`` mode writes
40 library-downloaded third-party PDFs as PLAINTEXT to disk (cleartext
41 storage of sensitive information, CWE-312), so it is disabled by default
42 and cannot be turned on through the user-writable settings API — only
43 via the environment variable
44 ``LDR_RESEARCH_LIBRARY_ALLOW_FILESYSTEM_PDF_STORAGE=true``.
45 """
46 from ...settings.env_registry import get_env_setting
48 return bool(
49 get_env_setting("research_library.allow_filesystem_pdf_storage", False)
50 )
53def resolve_pdf_storage_mode(mode):
54 """Resolve the effective PDF storage mode, enforcing the operator gate.
56 This is the runtime enforcement point (PEP) for the unencrypted-storage
57 gate: hiding the ``filesystem`` option in the settings UI is not enough,
58 because a value stored before the gate existed — or an
59 ``LDR_RESEARCH_LIBRARY_PDF_STORAGE_MODE=filesystem`` env override —
60 could still resolve to ``filesystem``. Callers that are about to WRITE
61 a PDF must resolve the mode through here first.
63 When the mode is ``filesystem`` and the gate is off, it is coerced to
64 the encrypted ``database`` default. Reads are unaffected:
65 ``PDFStorageManager.load_pdf`` checks the database first and then falls
66 back to the filesystem regardless of ``storage_mode``, so any
67 previously-written plaintext files stay readable after coercion.
68 """
69 if (
70 isinstance(mode, str)
71 and mode.strip().lower() == "filesystem"
72 and not filesystem_pdf_storage_allowed()
73 ):
74 logger.warning(
75 "Unencrypted filesystem PDF storage is operator-gated and "
76 "disabled; coercing pdf_storage_mode 'filesystem' -> 'database' "
77 "(set LDR_RESEARCH_LIBRARY_ALLOW_FILESYSTEM_PDF_STORAGE=true to "
78 "opt in)."
79 )
80 return "database"
81 return mode
84class PDFStorageManager:
85 """Unified interface for PDF storage across all modes."""
87 def __init__(
88 self,
89 library_root: Path,
90 storage_mode: str,
91 max_pdf_size_mb: int = DEFAULT_MAX_PDF_SIZE_MB,
92 legacy_root: Optional[Path] = None,
93 username: Optional[str] = None,
94 ):
95 """
96 Initialize PDF storage manager.
98 Args:
99 library_root: Base directory for filesystem storage. With per-user
100 library isolation (issue #5521) this is the per-user
101 directory; new PDFs are always WRITTEN here.
102 storage_mode: One of 'none', 'filesystem', 'database'
103 max_pdf_size_mb: Maximum PDF file size in MB. Should not
104 exceed `FileUploadValidator.MAX_FILE_SIZE` (the upload
105 validator's per-file cap, default 3 GB) — uploads above
106 that cap are rejected before they reach this layer.
107 legacy_root: Optional legacy shared root to fall back to when a
108 file is not found under ``library_root``. Lets PDFs
109 downloaded before per-user isolation still load; never
110 written to. Ignored when equal to ``library_root``.
111 username: Owning user for destructive operations. When falsy,
112 ``library_root`` cannot be guaranteed to be a per-user root
113 (``apply_user_subdir`` returns the bare shared root for an
114 empty username), so ``delete_pdf`` fails closed rather than
115 unlink within a possibly-shared directory.
116 """
117 self.library_root = Path(library_root).resolve()
118 self.storage_mode = storage_mode
119 self.username = username
120 self.max_pdf_size_bytes = max_pdf_size_mb * 1024 * 1024
121 resolved_legacy = (
122 Path(legacy_root).resolve() if legacy_root is not None else None
123 )
124 # Only keep a distinct legacy root — a legacy root equal to the
125 # per-user root is a no-op and would double-check the same directory.
126 self.legacy_root = (
127 resolved_legacy
128 if resolved_legacy is not None
129 and resolved_legacy != self.library_root
130 else None
131 )
133 if storage_mode not in ("none", "filesystem", "database"):
134 logger.warning(
135 f"Unknown storage mode '{storage_mode}', defaulting to 'none'"
136 )
137 self.storage_mode = "none"
139 def _safe_path_in_root(
140 self, relative_path: str, root: Path
141 ) -> Optional[Path]:
142 """Validate ``relative_path`` inside ``root`` (traversal + symlink
143 safe). Returns the validated Path (which may not exist), or None if
144 the path is invalid/unsafe."""
145 try:
146 # Use PathValidator to safely join and validate the path
147 safe_path = PathValidator.validate_safe_path(
148 relative_path, str(root)
149 )
150 safe_path = Path(safe_path)
151 # Block symbolic links to prevent symlink-based escapes
152 if safe_path.is_symlink():
153 logger.warning(f"Symlink blocked: {relative_path}")
154 return None
155 return safe_path
156 except ValueError:
157 logger.warning(f"Path traversal blocked: {relative_path}")
158 return None
160 def _get_safe_file_path(
161 self, relative_path: str, *, allow_legacy_fallback: bool = True
162 ) -> Optional[Path]:
163 """
164 Safely resolve a relative path within the library root.
166 Prevents path traversal attacks by validating the path stays within
167 the library root directory. Resolves against the per-user root first
168 and, when the file is absent there, falls back to the legacy shared
169 root (issue #5521) so pre-isolation downloads still load. When
170 neither location has the file, the per-user path is returned so
171 callers' ``is_file()``/``None`` handling is unchanged.
173 ``allow_legacy_fallback`` must be ``False`` for destructive callers
174 (unlink): the legacy shared root is not per-user namespaced, so a
175 colliding relative path can resolve to another tenant's file and be
176 unlinked. Delete only within this manager's own ``library_root``.
178 Even for read-only callers the legacy shared-root fallback is
179 cross-tenant read primitive on a multi-tenant instance (the shared
180 root is derived from the user-editable ``research_library.storage_path``
181 and per-user resource ids collide by construction), so it only fires
182 when the operator opted into it via
183 ``research_library.allow_legacy_read_fallback`` — OFF by default.
185 Args:
186 relative_path: Relative path from database
187 allow_legacy_fallback: Permit the read-only legacy shared-root
188 fallback. Pass ``False`` from delete/unlink paths. Even when
189 ``True``, the fallback only fires if the operator enabled
190 ``research_library.allow_legacy_read_fallback``.
192 Returns:
193 Validated absolute Path or None if path is invalid/unsafe
194 """
195 if not relative_path or relative_path in FILE_PATH_SENTINELS:
196 return None
198 primary = self._safe_path_in_root(relative_path, self.library_root)
199 if primary is not None and primary.is_file():
200 return primary
201 # Legacy shared-root fallback for files downloaded before per-user
202 # isolation. Never written to; read-only fallback. Enforced at the
203 # fallback-USE site (view_pdf_page passes legacy_root=base_root
204 # explicitly) so the operator gate covers every caller: only follow
205 # the legacy root when the caller permits it AND the operator opted in.
206 from ..utils import _legacy_read_fallback_allowed
208 if (
209 allow_legacy_fallback
210 and self.legacy_root is not None
211 and _legacy_read_fallback_allowed()
212 ):
213 legacy = self._safe_path_in_root(relative_path, self.legacy_root)
214 if legacy is not None and legacy.is_file(): 214 ↛ 216line 214 didn't jump to line 216 because the condition on line 214 was always true
215 return legacy
216 return primary
218 def save_pdf(
219 self,
220 pdf_content: bytes,
221 document: Document,
222 session: Session,
223 filename: str,
224 url: Optional[str] = None,
225 resource_id: Optional[int] = None,
226 ) -> Tuple[Optional[str], int]:
227 """
228 Save PDF based on configured storage mode.
230 Args:
231 pdf_content: Raw PDF bytes
232 document: Document model instance
233 session: Database session
234 filename: Filename to use for saving
235 url: Source URL (for generating better filenames)
236 resource_id: Resource ID (for generating better filenames)
238 Returns:
239 Tuple of (file_path or storage indicator, file_size)
240 - For filesystem: relative path string
241 - For database: "database"
242 - For none: None
243 """
244 file_size = len(pdf_content)
246 # Check file size limit
247 if file_size > self.max_pdf_size_bytes:
248 max_mb = self.max_pdf_size_bytes / (1024 * 1024)
249 logger.warning(
250 f"PDF size ({file_size / (1024 * 1024):.1f}MB) exceeds limit "
251 f"({max_mb:.0f}MB), skipping storage"
252 )
253 return None, file_size
255 if self.storage_mode == "none":
256 logger.debug("PDF storage mode is 'none' - skipping PDF save")
257 return None, file_size
259 if self.storage_mode == "filesystem":
260 file_path = self._save_to_filesystem(
261 pdf_content, filename, url, resource_id
262 )
263 relative_path = str(file_path.relative_to(self.library_root))
264 document.storage_mode = "filesystem"
265 document.file_path = relative_path
266 logger.info(f"PDF saved to filesystem: {relative_path}")
267 return relative_path, file_size
269 if self.storage_mode == "database": 269 ↛ 276line 269 didn't jump to line 276 because the condition on line 269 was always true
270 self._save_to_database(pdf_content, document, session)
271 document.storage_mode = "database"
272 document.file_path = None # No filesystem path
273 logger.info(f"PDF saved to database for document {document.id}")
274 return "database", file_size
276 return None, file_size
278 def load_pdf(self, document: Document, session: Session) -> Optional[bytes]:
279 """
280 Load PDF - check database first, then filesystem.
282 Smart retrieval: doesn't rely on storage_mode column, actually checks
283 where the PDF exists.
285 Args:
286 document: Document model instance
287 session: Database session
289 Returns:
290 PDF bytes or None if not available
291 """
292 # 1. Check database first
293 pdf_bytes = self._load_from_database(document, session)
294 if pdf_bytes:
295 logger.debug(f"Loaded PDF from database for document {document.id}")
296 return pdf_bytes
298 # 2. Fallback to filesystem
299 pdf_bytes = self._load_from_filesystem(document)
300 if pdf_bytes:
301 logger.debug(
302 f"Loaded PDF from filesystem for document {document.id}"
303 )
304 return pdf_bytes
306 logger.debug(f"No PDF available for document {document.id}")
307 return None
309 def has_pdf(self, document: Document, session: Session) -> bool:
310 """
311 Check if PDF is available without loading the actual bytes.
313 Args:
314 document: Document model instance
315 session: Database session
317 Returns:
318 True if PDF is available (in database or filesystem)
319 """
320 # Must be a PDF file type
321 if document.file_type != "pdf":
322 return False
324 # Check database first (has blob?)
325 from ...database.models.library import DocumentBlob
327 has_blob = (
328 session.query(DocumentBlob.document_id)
329 .filter_by(document_id=document.id)
330 .first()
331 is not None
332 )
333 if has_blob:
334 return True
336 # Check filesystem (with path traversal protection)
337 file_path = self._get_safe_file_path(document.file_path)
338 if file_path and file_path.is_file():
339 return True
341 return False
343 @classmethod
344 def pdf_exists(cls, library_root, document, session, legacy_root=None):
345 """Check if a PDF exists in any storage backend.
347 Use this when you need to check PDF availability without a specific
348 storage mode — e.g. generating document URLs in search results.
349 ``legacy_root`` enables the per-user -> legacy-shared read fallback
350 (issue #5521) so a pre-isolation filesystem PDF still reports present.
351 """
352 manager = cls(library_root, "none", legacy_root=legacy_root)
353 return manager.has_pdf(document, session)
355 def _infer_storage_mode(self, document: Document) -> str:
356 """
357 Infer storage mode for documents without explicit mode set.
358 Used for backward compatibility with existing documents.
359 """
360 # If there's a blob, it's database storage
361 if hasattr(document, "blob") and document.blob:
362 return "database"
363 # If there's a file_path (and not a sentinel), it's filesystem
364 if document.file_path and document.file_path not in FILE_PATH_SENTINELS:
365 return "filesystem"
366 # Otherwise no storage
367 return "none"
369 def _save_to_filesystem(
370 self,
371 pdf_content: bytes,
372 filename: str,
373 url: Optional[str] = None,
374 resource_id: Optional[int] = None,
375 ) -> Path:
376 """
377 Save PDF to filesystem with organized structure.
379 Returns:
380 Absolute path to saved file
381 """
382 # Generate better filename if URL is provided
383 if url:
384 filename = self._generate_filename(url, resource_id, filename)
386 # Create simple flat directory structure - all PDFs in one folder
387 from ...security.directory_creation import create_directory
389 pdf_path = self.library_root / "pdfs"
390 create_directory(
391 pdf_path,
392 context="library PDF storage directory",
393 )
395 # Use PathValidator with relative path from library_root
396 relative_path = f"pdfs/{filename}"
397 validated_path = PathValidator.validate_safe_path(
398 relative_path,
399 base_dir=str(self.library_root),
400 required_extensions=(".pdf",),
401 )
403 # Write the PDF file with security verification
404 # Pass current storage_mode as snapshot since we already validated it
405 from ...security.file_write_verifier import write_file_verified
407 write_file_verified(
408 validated_path,
409 pdf_content,
410 "research_library.pdf_storage_mode",
411 "filesystem",
412 "library PDF storage",
413 mode="wb",
414 settings_snapshot={
415 "research_library.pdf_storage_mode": self.storage_mode
416 },
417 )
419 return Path(validated_path)
421 def _save_to_database(
422 self, pdf_content: bytes, document: Document, session: Session
423 ) -> None:
424 """Store PDF in document_blobs table."""
425 # Check if blob already exists
426 existing_blob = (
427 session.query(DocumentBlob)
428 .filter_by(document_id=document.id)
429 .first()
430 )
432 if existing_blob:
433 # Update existing blob
434 existing_blob.pdf_binary = pdf_content
435 existing_blob.blob_hash = hashlib.sha256(pdf_content).hexdigest()
436 existing_blob.stored_at = datetime.now(UTC)
437 logger.debug(f"Updated existing blob for document {document.id}")
438 else:
439 # Create new blob
440 blob = DocumentBlob(
441 document_id=document.id,
442 pdf_binary=pdf_content,
443 blob_hash=hashlib.sha256(pdf_content).hexdigest(),
444 stored_at=datetime.now(UTC),
445 )
446 session.add(blob)
447 logger.debug(f"Created new blob for document {document.id}")
449 def _load_from_filesystem(self, document: Document) -> Optional[bytes]:
450 """Load PDF from filesystem with path traversal protection."""
451 # Use safe path resolution to prevent path traversal attacks
452 file_path = self._get_safe_file_path(document.file_path)
453 if not file_path:
454 return None
456 if not file_path.is_file():
457 logger.warning(f"PDF file not found: {file_path}")
458 return None
460 try:
461 return file_path.read_bytes()
462 except Exception:
463 logger.exception(f"Failed to read PDF from {file_path}")
464 return None
466 def _load_from_database(
467 self, document: Document, session: Session
468 ) -> Optional[bytes]:
469 """Load PDF from document_blobs table."""
470 blob = (
471 session.query(DocumentBlob)
472 .filter_by(document_id=document.id)
473 .first()
474 )
476 if not blob:
477 logger.debug(f"No blob found for document {document.id}")
478 return None
480 # Update last accessed timestamp
481 blob.last_accessed = datetime.now(UTC)
483 return blob.pdf_binary
485 def _generate_filename(
486 self, url: str, resource_id: Optional[int], fallback_filename: str
487 ) -> str:
488 """Generate a meaningful filename from URL."""
489 parsed_url = urlparse(url)
490 hostname = parsed_url.hostname or ""
491 timestamp = datetime.now(UTC).strftime("%Y%m%d")
493 if hostname == "arxiv.org" or hostname.endswith(".arxiv.org"):
494 # Extract arXiv ID
495 match = re.search(r"(\d{4}\.\d{4,5})", url)
496 if match:
497 return f"arxiv_{match.group(1)}.pdf"
498 return f"arxiv_{timestamp}_{resource_id or 'unknown'}.pdf"
500 if hostname == "ncbi.nlm.nih.gov" and "/pmc" in parsed_url.path:
501 # Extract PMC ID
502 match = re.search(r"(PMC\d+)", url)
503 if match:
504 return f"pmc_{match.group(1)}.pdf"
505 return f"pubmed_{timestamp}_{resource_id or 'unknown'}.pdf"
507 # Use fallback filename
508 return fallback_filename
510 def delete_pdf(self, document: Document, session: Session) -> bool:
511 """
512 Delete PDF for a document.
514 Args:
515 document: Document model instance
516 session: Database session
518 Returns:
519 True if deletion succeeded
520 """
521 storage_mode = document.storage_mode or self._infer_storage_mode(
522 document
523 )
525 try:
526 if storage_mode == "filesystem":
527 # Fail closed without a user context: an empty username means
528 # library_root may be the bare shared root (apply_user_subdir
529 # returns the shared base for an empty username), where a
530 # colliding relative path can belong to another tenant. Refuse
531 # the unlink rather than risk deleting another user's file.
532 if not self.username:
533 logger.warning(
534 "Refusing filesystem PDF delete for document "
535 f"{document.id}: no user context, cannot confirm the "
536 "library root is per-user (would risk unlinking a "
537 "shared-root file)."
538 )
539 return False
540 # Use safe path resolution to prevent path traversal attacks.
541 # Delete within our own root only — never the shared legacy
542 # root, where a colliding path can be another tenant's file.
543 file_path = self._get_safe_file_path(
544 document.file_path, allow_legacy_fallback=False
545 )
546 if file_path and file_path.is_file():
547 file_path.unlink()
548 logger.info(f"Deleted PDF file: {file_path}")
549 document.file_path = None
550 document.storage_mode = "none"
551 return True
553 if storage_mode == "database":
554 blob = (
555 session.query(DocumentBlob)
556 .filter_by(document_id=document.id)
557 .first()
558 )
559 if blob: 559 ↛ 562line 559 didn't jump to line 562 because the condition on line 559 was always true
560 session.delete(blob)
561 logger.info(f"Deleted PDF blob for document {document.id}")
562 document.storage_mode = "none"
563 return True
565 return True # Nothing to delete for 'none' mode
567 except Exception:
568 logger.exception(f"Failed to delete PDF for document {document.id}")
569 return False
571 def upgrade_to_pdf(
572 self, document: Document, pdf_content: bytes, session: Session
573 ) -> bool:
574 """
575 Upgrade a text-only document to include PDF storage.
577 If document already has a PDF stored, returns False (no action needed).
578 If document is text-only, adds the PDF blob and updates storage_mode.
580 Args:
581 document: Document model instance
582 pdf_content: Raw PDF bytes
583 session: Database session
585 Returns:
586 True if PDF was added, False if already had PDF or failed
587 """
588 # Only upgrade if document is currently text-only
589 if document.storage_mode not in (None, "none"):
590 logger.debug(
591 f"Document {document.id} already has storage_mode={document.storage_mode}"
592 )
593 return False
595 # Check if blob already exists (shouldn't happen, but be safe)
596 existing_blob = (
597 session.query(DocumentBlob)
598 .filter_by(document_id=document.id)
599 .first()
600 )
601 if existing_blob: 601 ↛ 602line 601 didn't jump to line 602 because the condition on line 601 was never true
602 logger.debug(f"Document {document.id} already has a blob")
603 return False
605 # Check file size
606 file_size = len(pdf_content)
607 if file_size > self.max_pdf_size_bytes:
608 max_mb = self.max_pdf_size_bytes / (1024 * 1024)
609 logger.warning(
610 f"PDF size ({file_size / (1024 * 1024):.1f}MB) exceeds limit "
611 f"({max_mb:.0f}MB), skipping upgrade"
612 )
613 return False
615 try:
616 # Add the PDF blob
617 self._save_to_database(pdf_content, document, session)
618 document.storage_mode = "database"
619 document.file_path = None
620 logger.info(f"Upgraded document {document.id} with PDF blob")
621 return True
622 except Exception:
623 logger.exception(
624 f"Failed to upgrade document {document.id} with PDF"
625 )
626 return False