Coverage for src/local_deep_research/security/path_validator.py: 97%
171 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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. Lazy import: this sibling
358 # module can be imported very early (e.g. from `security/__init__.py`
359 # during package bootstrap), so import at call time rather than at
360 # module top level to avoid any import-cycle risk.
361 from .directory_creation import create_directory
363 model_root_path = Path(model_root).resolve()
364 create_directory(
365 model_root_path,
366 context="model root directory",
367 )
369 # Validate the path
370 validated_path = PathValidator.validate_safe_path(
371 model_path,
372 model_root_path,
373 allow_absolute=False, # Models should always be relative to model root
374 required_extensions=None, # Models can have various extensions
375 )
377 if not validated_path:
378 raise ValueError("Invalid model path")
380 # Check if the file exists
381 if not validated_path.exists():
382 raise ValueError(f"Model file not found: {validated_path}")
384 if not validated_path.is_file():
385 raise ValueError(f"Model path is not a file: {validated_path}")
387 return validated_path
389 @staticmethod
390 def validate_data_path(file_path: str, data_root: str) -> Path:
391 """
392 Validate a path within the data directory.
394 Args:
395 file_path: Path relative to data root
396 data_root: The data root directory
398 Returns:
399 Validated Path object
401 Raises:
402 ValueError: If the path is invalid
403 """
404 validated_path = PathValidator.validate_safe_path(
405 file_path,
406 data_root,
407 allow_absolute=False, # Data paths should be relative
408 required_extensions=None,
409 )
411 if not validated_path:
412 raise ValueError("Invalid data path")
414 return validated_path
416 @staticmethod
417 def validate_config_path(
418 config_path: str, config_root: Optional[Union[str, Path]] = None
419 ) -> Path:
420 """
421 Validate a configuration file path.
423 Args:
424 config_path: Path to config file
425 config_root: Root directory for configs (optional for absolute paths)
427 Returns:
428 Validated Path object
430 Raises:
431 ValueError: If the path is invalid
432 """
433 # Validate input: reject null bytes, then normalize whitespace
434 if not config_path or not isinstance(config_path, str):
435 raise ValueError("Invalid config path input")
437 if "\x00" in config_path:
438 raise ValueError("Null bytes are not allowed in config path")
440 config_path = config_path.strip()
442 # Check for path traversal attempts in the string itself
443 # Define restricted system directories that should never be accessed
444 RESTRICTED_PREFIXES = ("etc", "proc", "sys", "dev")
446 if ".." in config_path:
447 raise ValueError("Invalid path - potential traversal attempt")
449 # Check if path starts with any restricted system directory
450 normalized_path = config_path.lstrip("/").lower()
451 for restricted in RESTRICTED_PREFIXES:
452 if (
453 normalized_path.startswith(restricted + "/")
454 or normalized_path == restricted
455 ):
456 raise ValueError(
457 f"Invalid path - restricted system directory: {restricted}"
458 )
460 # For config files, we might allow absolute paths with restrictions
461 # Check if path starts with / or drive letter (Windows) to detect absolute paths
462 # This avoids using Path() or os.path on user input
463 is_absolute = (
464 config_path.startswith("/") # Unix absolute
465 or (
466 len(config_path) > 2 and config_path[1] == ":"
467 ) # Windows absolute
468 )
470 if is_absolute:
471 # For absolute paths, use safe_join with root directory
472 # This validates the path without using Path() directly on user input
473 # Use safe_join to validate the absolute path
474 safe_path = safe_join("/", config_path)
475 if safe_path is None:
476 raise ValueError("Invalid absolute path")
478 # Now it's safe to create Path object from validated string
479 path_obj = Path(safe_path)
481 # Additional validation for config files
482 if path_obj.suffix not in PathValidator.CONFIG_EXTENSIONS:
483 raise ValueError(f"Invalid config file type: {path_obj.suffix}")
485 # Check existence using validated path
486 if not path_obj.exists():
487 raise ValueError(f"Config file not found: {path_obj}")
489 return path_obj
490 # For relative paths, use the config root
491 if config_root is None:
492 from ..config.paths import get_data_directory
494 config_root = get_data_directory()
496 validated = PathValidator.validate_safe_path(
497 config_path,
498 config_root,
499 allow_absolute=False,
500 required_extensions=PathValidator.CONFIG_EXTENSIONS,
501 )
502 if validated is None:
503 raise ValueError(f"Invalid config path: {config_path}")
504 return validated