Coverage for src/local_deep_research/security/path_validator.py: 97%
170 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Centralized path validation utilities for security.
4This module provides secure path validation to prevent path traversal attacks
5and other filesystem-based security vulnerabilities.
6"""
8import os
9import re
10import unicodedata
11from pathlib import Path
12from typing import Optional, Union
13from urllib.parse import unquote
15from loguru import logger
16from werkzeug.security import safe_join
18from ..config.paths import get_models_directory
21# Encoded forms of ".." and "." that path traversal attacks use to bypass
22# naive `..`-substring checks. The list covers both single and double URL
23# encoding (where the % itself is encoded as %25).
24_ENCODED_TRAVERSAL_TOKENS = (
25 "%2e%2e", # ..
26 "%2E%2E",
27 "%2e.", # .. (mixed)
28 ".%2e",
29 "%252e%252e", # double-encoded ..
30 "%252E%252E",
31 "%2f", # /
32 "%2F",
33)
36def _has_encoded_traversal(text: str) -> bool:
37 """Return True if `text` contains URL-encoded path-traversal tokens.
39 Checks for both single and double URL encoding of '..', '.', and '/'.
40 Decoders downstream may convert these into real '..' segments that
41 escape the base directory; reject the input before that happens.
42 """
43 lowered = text.lower()
44 for tok in _ENCODED_TRAVERSAL_TOKENS:
45 if tok.lower() in lowered:
46 return True
47 # Recursive decode: if unquote changes the string, the original
48 # contained encoded characters. Re-check the decoded form for ".."
49 # in case the encoding used is one not enumerated above.
50 decoded = unquote(text)
51 if decoded != text and ".." in decoded:
52 return True
53 # Double-decode catches %252e%252e -> %2e%2e -> ..
54 double_decoded = unquote(decoded)
55 if double_decoded != decoded and ".." in double_decoded: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 return True
57 return False
60def _has_unicode_traversal(text: str) -> bool:
61 """Return True if `text` contains unicode look-alikes for path traversal.
63 Many unicode characters NFKC-normalize to '.' or '/', so an attacker
64 can use full-width periods (U+FF0E '．') or other look-alikes to
65 bypass naive '..'-substring checks. After normalization, '..' or
66 '/..' segments indicate an attempt to traverse.
67 """
68 normalized = unicodedata.normalize("NFKC", text)
69 if normalized == text:
70 return False
71 return (
72 ".." in normalized
73 or normalized.startswith("/..")
74 or "/.." in normalized
75 )
78class PathValidator:
79 """Centralized path validation for security."""
81 # Regex for safe filename/path characters
82 SAFE_PATH_PATTERN = re.compile(r"^[a-zA-Z0-9._/\-]+$")
84 # Allowed config file extensions
85 CONFIG_EXTENSIONS = (".json", ".yaml", ".yml", ".toml", ".ini", ".conf")
87 @staticmethod
88 def validate_safe_path(
89 user_input: str,
90 base_dir: Union[str, Path],
91 allow_absolute: bool = False,
92 required_extensions: Optional[tuple] = None,
93 ) -> Optional[Path]:
94 """
95 Validate and sanitize a user-provided path.
97 Args:
98 user_input: The user-provided path string
99 base_dir: The safe base directory to contain paths within
100 allow_absolute: Whether to allow absolute paths (with restrictions)
101 required_extensions: Tuple of required file extensions (e.g., ('.json', '.yaml'))
103 Returns:
104 Path object if valid, None if invalid
106 Raises:
107 ValueError: If the path is invalid or unsafe
108 """
109 if not user_input or not isinstance(user_input, str):
110 raise ValueError("Invalid path input")
112 if "\x00" in user_input:
113 raise ValueError("Null bytes are not allowed in path")
115 # Strip whitespace
116 user_input = user_input.strip()
118 # Reject URL-encoded traversal tokens (single or double encoded).
119 # safe_join's literal-".."-check doesn't catch %2e%2e or %252e%252e,
120 # so do it explicitly here before any decoding takes place.
121 if _has_encoded_traversal(user_input):
122 logger.warning(
123 f"Encoded path-traversal attempt blocked: {user_input!r}"
124 )
125 raise ValueError(
126 "Invalid path - encoded traversal pattern detected"
127 )
129 # Reject unicode look-alike traversal (e.g. full-width '．．').
130 # NFKC normalization is the same form most filesystems / browsers
131 # apply; if normalizing produces a '..' segment the original was a
132 # disguised traversal attempt.
133 if _has_unicode_traversal(user_input):
134 logger.warning(
135 f"Unicode path-traversal attempt blocked: {user_input!r}"
136 )
137 raise ValueError(
138 "Invalid path - unicode traversal pattern detected"
139 )
141 # Use werkzeug's safe_join for secure path joining
142 # This handles path traversal attempts automatically
143 base_dir = Path(base_dir).resolve()
145 try:
146 # safe_join returns None if the path tries to escape base_dir
147 safe_path = safe_join(str(base_dir), user_input)
148 except ValueError:
149 raise
150 except Exception as e:
151 logger.warning(f"Path validation failed for input '{user_input}'")
152 raise ValueError(f"Invalid path: {e}") from e
154 if safe_path is None:
155 logger.warning(f"Path traversal attempt blocked: {user_input}")
156 raise ValueError("Invalid path - potential traversal attempt")
158 result_path = Path(safe_path)
160 # Check extensions if required
161 if (
162 required_extensions
163 and result_path.suffix not in required_extensions
164 ):
165 raise ValueError(
166 f"Invalid file type. Allowed: {required_extensions}"
167 )
169 return result_path
171 @staticmethod
172 def validate_local_filesystem_path(
173 user_path: str,
174 restricted_dirs: Optional[list[Path]] = None,
175 ) -> Path:
176 """
177 Validate a user-provided absolute filesystem path for local indexing.
179 This is for features like local folder indexing where users need to
180 access files anywhere on their own machine, but system directories
181 should be blocked.
183 Args:
184 user_path: User-provided path string (absolute or with ~)
185 restricted_dirs: List of restricted directories to block
187 Returns:
188 Validated and resolved Path object
190 Raises:
191 ValueError: If path is invalid or points to restricted location
192 """
193 import sys
195 if not user_path or not isinstance(user_path, str):
196 raise ValueError("Invalid path input")
198 user_path = user_path.strip()
200 # Basic sanitation: forbid null bytes and control characters
201 if "\x00" in user_path:
202 raise ValueError("Null bytes are not allowed in path")
203 if any(ord(ch) < 32 for ch in user_path):
204 raise ValueError("Control characters are not allowed in path")
206 # Expand ~ to home directory
207 if user_path.startswith("~"):
208 home_dir = Path.home()
209 relative_part = user_path[2:].lstrip("/")
210 if relative_part:
211 user_path = str(home_dir / relative_part)
212 else:
213 user_path = str(home_dir)
215 # Disallow malformed Windows paths (e.g. "/C:/Windows")
216 if (
217 sys.platform == "win32"
218 and user_path.startswith(("/", "\\"))
219 and ":" in user_path
220 ):
221 raise ValueError("Malformed Windows path input")
223 # Block path traversal patterns before resolving
224 # This explicit check helps static analyzers understand the security intent
225 if ".." in user_path:
226 raise ValueError("Path traversal patterns not allowed")
228 # Use safe_join to sanitize the path - this is recognized by static analyzers
229 # For absolute paths, we validate against the root directory
230 if user_path.startswith("/"):
231 # Unix absolute path - use safe_join with root
232 safe_path = safe_join("/", user_path.lstrip("/"))
233 if safe_path is None:
234 raise ValueError("Invalid path - failed security validation")
235 validated_path = Path(safe_path).resolve()
236 elif len(user_path) > 2 and user_path[1] == ":":
237 # Windows absolute path (e.g., C:\Users\...)
238 drive = user_path[:2]
239 rest = user_path[2:].lstrip("\\").lstrip("/")
240 safe_path = safe_join(drive + "\\", rest)
241 if safe_path is None:
242 raise ValueError("Invalid path - failed security validation")
243 validated_path = Path(safe_path).resolve()
244 else:
245 # Relative path - resolve relative to current directory
246 # Use safe_join to validate
247 cwd = os.getcwd()
248 safe_path = safe_join(cwd, user_path)
249 if safe_path is None:
250 raise ValueError("Invalid path - failed security validation")
251 validated_path = Path(safe_path).resolve()
253 # Default restricted directories
254 if restricted_dirs is None:
255 restricted_dirs = [
256 Path("/etc"),
257 Path("/sys"),
258 Path("/proc"),
259 Path("/dev"),
260 Path("/root"),
261 Path("/boot"),
262 Path("/var/log"),
263 ]
264 # Add Windows system directories if on Windows
265 if sys.platform == "win32":
266 for drive in ["C:", "D:", "E:"]:
267 restricted_dirs.extend(
268 [
269 Path(f"{drive}\\Windows"),
270 Path(f"{drive}\\System32"),
271 Path(f"{drive}\\Program Files"),
272 Path(f"{drive}\\Program Files (x86)"),
273 ]
274 )
276 # Check against restricted directories. Resolve each restricted dir so
277 # the containment check holds on platforms where these are symlinks
278 # (e.g. macOS /etc -> /private/etc, /var -> /private/var) — validated_path
279 # is already resolved, so an unresolved "/etc" would never match
280 # "/private/etc/...". resolve(strict=False) does not raise for a missing
281 # or unreadable dir; fall back to the literal path if it ever does.
282 for restricted in restricted_dirs:
283 try:
284 restricted_resolved = restricted.resolve()
285 except OSError:
286 restricted_resolved = restricted
287 if validated_path.is_relative_to(restricted_resolved):
288 # Log WHICH restricted dir was hit, not the user's submitted
289 # path — a resolved local path can contain a username.
290 logger.error(
291 f"Security: blocked access to a restricted directory ({restricted})"
292 )
293 raise ValueError("Cannot access system directories")
295 return validated_path
297 @staticmethod
298 def sanitize_for_filesystem_ops(validated_path: Path) -> Path:
299 """
300 Re-sanitize a validated path for static analyzer recognition.
302 This method takes an already-validated Path and passes it through
303 werkzeug's safe_join to create a path that static analyzers like
304 CodeQL recognize as sanitized.
306 Note: This exists because CodeQL doesn't trace through custom validation
307 functions. The path is already secure from validate_local_filesystem_path(),
308 but safe_join makes that explicit to static analyzers.
310 Args:
311 validated_path: A Path that has already been validated by
312 validate_local_filesystem_path()
314 Returns:
315 A Path object safe for filesystem operations
317 Raises:
318 ValueError: If the path fails sanitization
319 """
320 if not validated_path.is_absolute():
321 raise ValueError("Path must be absolute")
323 # Use safe_join to create a sanitized path that static analyzers recognize
324 # safe_join handles path traversal detection properly (not substring matching)
325 path_str = str(validated_path)
326 safe_path_str = safe_join("/", path_str.lstrip("/"))
327 if safe_path_str is None:
328 raise ValueError("Path failed security sanitization")
330 return Path(safe_path_str)
332 # A `confine_to_base()` helper once lived here — see #4868. It filtered a list
333 # of paths down to those whose real (symlink-resolved) location stayed inside a
334 # base directory, dropping symlink escapes and loops.
336 @staticmethod
337 def validate_model_path(
338 model_path: str, model_root: Optional[str] = None
339 ) -> Path:
340 """
341 Validate a model file path specifically.
343 Args:
344 model_path: Path to the model file
345 model_root: Root directory for models (defaults to ~/.local/share/llm_models)
347 Returns:
348 Validated Path object
350 Raises:
351 ValueError: If the path is invalid
352 """
353 if model_root is None:
354 # Default model root - uses centralized path config (respects LDR_DATA_DIR)
355 model_root = str(get_models_directory())
357 # Create model root if it doesn't exist
358 model_root_path = Path(model_root).resolve()
359 model_root_path.mkdir(parents=True, exist_ok=True)
361 # Validate the path
362 validated_path = PathValidator.validate_safe_path(
363 model_path,
364 model_root_path,
365 allow_absolute=False, # Models should always be relative to model root
366 required_extensions=None, # Models can have various extensions
367 )
369 if not validated_path:
370 raise ValueError("Invalid model path")
372 # Check if the file exists
373 if not validated_path.exists():
374 raise ValueError(f"Model file not found: {validated_path}")
376 if not validated_path.is_file():
377 raise ValueError(f"Model path is not a file: {validated_path}")
379 return validated_path
381 @staticmethod
382 def validate_data_path(file_path: str, data_root: str) -> Path:
383 """
384 Validate a path within the data directory.
386 Args:
387 file_path: Path relative to data root
388 data_root: The data root directory
390 Returns:
391 Validated Path object
393 Raises:
394 ValueError: If the path is invalid
395 """
396 validated_path = PathValidator.validate_safe_path(
397 file_path,
398 data_root,
399 allow_absolute=False, # Data paths should be relative
400 required_extensions=None,
401 )
403 if not validated_path:
404 raise ValueError("Invalid data path")
406 return validated_path
408 @staticmethod
409 def validate_config_path(
410 config_path: str, config_root: Optional[Union[str, Path]] = None
411 ) -> Path:
412 """
413 Validate a configuration file path.
415 Args:
416 config_path: Path to config file
417 config_root: Root directory for configs (optional for absolute paths)
419 Returns:
420 Validated Path object
422 Raises:
423 ValueError: If the path is invalid
424 """
425 # Validate input: reject null bytes, then normalize whitespace
426 if not config_path or not isinstance(config_path, str):
427 raise ValueError("Invalid config path input")
429 if "\x00" in config_path:
430 raise ValueError("Null bytes are not allowed in config path")
432 config_path = config_path.strip()
434 # Check for path traversal attempts in the string itself
435 # Define restricted system directories that should never be accessed
436 RESTRICTED_PREFIXES = ("etc", "proc", "sys", "dev")
438 if ".." in config_path:
439 raise ValueError("Invalid path - potential traversal attempt")
441 # Check if path starts with any restricted system directory
442 normalized_path = config_path.lstrip("/").lower()
443 for restricted in RESTRICTED_PREFIXES:
444 if (
445 normalized_path.startswith(restricted + "/")
446 or normalized_path == restricted
447 ):
448 raise ValueError(
449 f"Invalid path - restricted system directory: {restricted}"
450 )
452 # For config files, we might allow absolute paths with restrictions
453 # Check if path starts with / or drive letter (Windows) to detect absolute paths
454 # This avoids using Path() or os.path on user input
455 is_absolute = (
456 config_path.startswith("/") # Unix absolute
457 or (
458 len(config_path) > 2 and config_path[1] == ":"
459 ) # Windows absolute
460 )
462 if is_absolute:
463 # For absolute paths, use safe_join with root directory
464 # This validates the path without using Path() directly on user input
465 # Use safe_join to validate the absolute path
466 safe_path = safe_join("/", config_path)
467 if safe_path is None:
468 raise ValueError("Invalid absolute path")
470 # Now it's safe to create Path object from validated string
471 path_obj = Path(safe_path)
473 # Additional validation for config files
474 if path_obj.suffix not in PathValidator.CONFIG_EXTENSIONS:
475 raise ValueError(f"Invalid config file type: {path_obj.suffix}")
477 # Check existence using validated path
478 if not path_obj.exists():
479 raise ValueError(f"Config file not found: {path_obj}")
481 return path_obj
482 # For relative paths, use the config root
483 if config_root is None:
484 from ..config.paths import get_data_directory
486 config_root = get_data_directory()
488 validated = PathValidator.validate_safe_path(
489 config_path,
490 config_root,
491 allow_absolute=False,
492 required_extensions=PathValidator.CONFIG_EXTENSIONS,
493 )
494 if validated is None:
495 raise ValueError(f"Invalid config path: {config_path}")
496 return validated