Coverage for src/local_deep_research/security/file_upload_validator.py: 100%
79 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Centralized file upload validation for security.
4Provides validation for file uploads to prevent:
5- Memory exhaustion attacks (file size limits)
6- Malicious file uploads (structure validation)
7- Resource abuse (file count limits)
8- Type confusion attacks (MIME validation)
9"""
11import io
12from typing import Optional, Tuple
14import pdfplumber
15from loguru import logger
16from pdfminer.pdfpage import PDFPage
17from pdfplumber.page import Page
19from ..settings.manager import check_env_setting
21# Built-in default for the per-file upload cap, in megabytes.
22# Override via the LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB environment variable
23# (or its server-config alias `security.upload_max_file_size_mb`).
24#
25# A separate library-side cap (`research_library.max_pdf_size_mb`,
26# default 3072) gates whether a PDF can be *stored* after upload. Both
27# values default to 3 GB so the two boundaries stay aligned; raising the
28# library cap above this value has no effect because uploads above this
29# cap are rejected before they reach storage.
30_DEFAULT_MAX_FILE_SIZE_MB = 3072 # 3 GB
33def _format_size(size_bytes: int) -> str:
34 """Render a byte count as ``X.YGB`` when >=1 GB, else ``X.YMB``."""
35 one_gb = 1024 * 1024 * 1024
36 if size_bytes >= one_gb:
37 return f"{size_bytes / one_gb:.1f}GB"
38 return f"{size_bytes / (1024 * 1024):.1f}MB"
41def _resolve_max_file_size() -> int:
42 """Resolve the per-file upload cap, in bytes.
44 Reads ``LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB`` (via
45 ``settings.manager.check_env_setting``) so deployments can lower the
46 cap without code changes. Falls back to ``_DEFAULT_MAX_FILE_SIZE_MB``
47 when the variable is unset or unparseable.
48 """
49 env_value = check_env_setting("security.upload_max_file_size_mb")
50 if env_value is not None:
51 try:
52 mb = int(env_value)
53 except ValueError:
54 logger.warning(
55 "LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB={!r} is not an integer; "
56 "falling back to {} MB default.",
57 env_value,
58 _DEFAULT_MAX_FILE_SIZE_MB,
59 )
60 else:
61 # Reject zero / negative values — they would silently break
62 # all uploads. Fall back to the default and log so the
63 # operator can spot the misconfiguration.
64 if mb > 0:
65 return mb * 1024 * 1024
66 logger.warning(
67 "LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB={!r} must be > 0; "
68 "falling back to {} MB default.",
69 env_value,
70 _DEFAULT_MAX_FILE_SIZE_MB,
71 )
72 return _DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
75class FileUploadValidator:
76 """Centralized file upload validation for security."""
78 # Security constants. ``MAX_FILE_SIZE`` is loaded from server config so
79 # it can be lowered per-deployment via the env var or UI setting; the
80 # built-in default is 3 GB (see ``server_config._DEFAULTS``).
81 MAX_FILE_SIZE = _resolve_max_file_size()
82 MAX_FILES_PER_REQUEST = 200 # Maximum number of files in single request
83 PDF_MAGIC_BYTES = b"%PDF" # PDF file signature
84 ALLOWED_MIME_TYPES = {"application/pdf"}
86 @staticmethod
87 def validate_file_size(
88 content_length: Optional[int], file_content: Optional[bytes] = None
89 ) -> Tuple[bool, Optional[str]]:
90 """
91 Validate file size to prevent memory exhaustion attacks.
93 Args:
94 content_length: Content-Length header value (if available)
95 file_content: Actual file bytes (if already read)
97 Returns:
98 Tuple of (is_valid, error_message)
99 """
100 # Check Content-Length header first (before reading file)
101 if content_length is not None:
102 if content_length > FileUploadValidator.MAX_FILE_SIZE:
103 return (
104 False,
105 f"File too large: {_format_size(content_length)} "
106 f"(max: {_format_size(FileUploadValidator.MAX_FILE_SIZE)})",
107 )
109 # Check actual file size if content is provided
110 if file_content is not None:
111 actual_size = len(file_content)
112 if actual_size > FileUploadValidator.MAX_FILE_SIZE:
113 return (
114 False,
115 f"File too large: {_format_size(actual_size)} "
116 f"(max: {_format_size(FileUploadValidator.MAX_FILE_SIZE)})",
117 )
119 return True, None
121 @staticmethod
122 def validate_file_count(file_count: int) -> Tuple[bool, Optional[str]]:
123 """
124 Validate number of files to prevent resource abuse.
126 Args:
127 file_count: Number of files in the request
129 Returns:
130 Tuple of (is_valid, error_message)
131 """
132 if file_count > FileUploadValidator.MAX_FILES_PER_REQUEST:
133 return (
134 False,
135 f"Too many files: {file_count} (max: {FileUploadValidator.MAX_FILES_PER_REQUEST})",
136 )
138 if file_count <= 0:
139 return False, "No files provided"
141 return True, None
143 @staticmethod
144 def validate_mime_type(
145 filename: str, file_content: bytes
146 ) -> Tuple[bool, Optional[str]]:
147 """
148 Validate file MIME type and extension.
150 Args:
151 filename: Original filename
152 file_content: File content bytes
154 Returns:
155 Tuple of (is_valid, error_message)
156 """
157 # Check file extension
158 if not filename.lower().endswith(".pdf"):
159 return (
160 False,
161 f"Invalid file type: {filename}. Only PDF files allowed",
162 )
164 # Check PDF magic bytes (file signature)
165 if not file_content.startswith(FileUploadValidator.PDF_MAGIC_BYTES):
166 return (
167 False,
168 f"Invalid PDF file: {filename}. File signature mismatch",
169 )
171 return True, None
173 @staticmethod
174 def validate_pdf_structure(
175 filename: str, file_content: bytes
176 ) -> Tuple[bool, Optional[str]]:
177 """
178 Validate PDF structure to detect malicious or corrupted files.
180 This goes beyond just checking the magic bytes and actually attempts
181 to parse the PDF structure. Only the first page is inspected, so the
182 cost does not grow with the number of pages the file declares.
184 Args:
185 filename: Original filename
186 file_content: File content bytes
188 Returns:
189 Tuple of (is_valid, error_message)
190 """
191 try:
192 # Not a ``with`` block: pdfplumber's ``PDF.pages`` -- which
193 # ``PDF.close()`` walks too -- parses and builds every page up
194 # front, so an 11 MB upload declaring 100,000 pages cost hundreds
195 # of MB and over ten seconds here (#6771). The page tree is walked
196 # lazily instead and stops at the first page; nothing needs
197 # closing for an in-memory stream.
198 pdf = pdfplumber.open(io.BytesIO(file_content))
199 page_obj = next(PDFPage.create_pages(pdf.doc), None)
200 if page_obj is None:
201 return False, f"Invalid PDF: {filename}. No pages found"
203 # pdfplumber's Page validates the first page's boxes exactly as
204 # building ``pdf.pages`` would.
205 first_page = Page(pdf, page_obj, page_number=1)
206 _ = first_page.width # Access basic metadata
207 _ = first_page.height
209 return True, None
211 except Exception:
212 logger.warning(f"PDF structure validation failed for {filename}")
213 return False, f"Invalid or corrupted PDF file: {filename}"
215 @classmethod
216 def validate_upload(
217 cls,
218 filename: str,
219 file_content: bytes,
220 content_length: Optional[int] = None,
221 ) -> Tuple[bool, Optional[str]]:
222 """
223 Comprehensive validation for a single file upload.
225 Runs all validation checks in sequence. Stops at first failure.
227 Args:
228 filename: Original filename
229 file_content: File content bytes
230 content_length: Content-Length header (if available)
232 Returns:
233 Tuple of (is_valid, error_message)
234 """
235 # 1. Validate file size
236 is_valid, error = cls.validate_file_size(content_length, file_content)
237 if not is_valid:
238 return is_valid, error
240 # 2. Validate MIME type and extension
241 is_valid, error = cls.validate_mime_type(filename, file_content)
242 if not is_valid:
243 return is_valid, error
245 # 3. Validate PDF structure (more thorough check)
246 is_valid, error = cls.validate_pdf_structure(filename, file_content)
247 if not is_valid:
248 return is_valid, error
250 return True, None