Coverage for src/local_deep_research/security/path

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 errno

9import ntpath

10import os

11import posixpath

12import re

13import unicodedata

14from pathlib import Path

15from typing import Optional, Union

16from urllib.parse import unquote

18from loguru import logger

19from werkzeug.security import safe_join

21from ..config.paths import get_models_directory

22from .log_sanitizer import redact_and_bound_for_log

25# Encoded forms of ".." and "." that path traversal attacks use to bypass

26# naive `..`-substring checks. The list covers both single and double URL

27# encoding (where the % itself is encoded as %25).

28_ENCODED_TRAVERSAL_TOKENS = (

29 "%2e%2e", # ..

30 "%2E%2E",

31 "%2e.", # .. (mixed)

32 ".%2e",

33 "%252e%252e", # double-encoded ..

34 "%252E%252E",

35 "%2f", # /

36 "%2F",

37)

40def _has_encoded_traversal(text: str) -> bool:

41 """Return True if `text` contains URL-encoded path-traversal tokens.

43 Checks for both single and double URL encoding of '..', '.', and '/'.

44 Decoders downstream may convert these into real '..' segments that

45 escape the base directory; reject the input before that happens.

46 """

47 lowered = text.lower()

48 for tok in _ENCODED_TRAVERSAL_TOKENS:

49 if tok.lower() in lowered:

50 return True

51 # Recursive decode: if unquote changes the string, the original

52 # contained encoded characters. Re-check the decoded form for ".."

53 # in case the encoding used is one not enumerated above.

54 decoded = unquote(text)

55 if decoded != text and ".." in decoded:

56 return True

57 # Double-decode catches %252e%252e -> %2e%2e -> ..

58 double_decoded = unquote(decoded)

59 if double_decoded != decoded and ".." in double_decoded: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 return True

61 return False

64def _has_unicode_traversal(text: str) -> bool:

65 """Return True if `text` contains unicode look-alikes for path traversal.

67 Many unicode characters NFKC-normalize to '.' or '/', so an attacker

68 can use full-width periods (U+FF0E '．') or other look-alikes to

69 bypass naive '..'-substring checks. After normalization, '..' or

70 '/..' segments indicate an attempt to traverse.

71 """

72 normalized = unicodedata.normalize("NFKC", text)

73 if normalized == text:

74 return False

75 return (

76 ".." in normalized

77 or normalized.startswith("/..")

78 or "/.." in normalized

79 )

82# ``os.path.ALLOW_MISSING`` exists on Python 3.14 and was backported to

83# 3.12.11 and 3.13.4. Older patch releases use the fallback below. It is

84# read from the platform's path module, picked the way ``os`` picks

85# ``os.path``, because the check-pathlib-usage hook forbids ``os.path``

86# and the sentinel has no pathlib spelling.

87_PLATFORM_PATH_MODULE = ntpath if os.name == "nt" else posixpath

88_ALLOW_MISSING = getattr(_PLATFORM_PATH_MODULE, "ALLOW_MISSING", None)

91def resolve_allow_missing(path: Union[str, Path]) -> Path:

92 """Resolve every symlink in ``path``. Only a missing tail is tolerated.

94 Use this, never a bare ``Path.resolve()``, wherever a resolved path is

95 compared against a containment boundary. Non-strict ``resolve()``

96 ignores every ``OSError`` that ``lstat`` raises, not only "does not

97 exist". A short in-base symlink to a deep in-base directory chain makes

98 the resolved prefix longer than PATH_MAX, so every later ``lstat`` fails

99 with ENAMETOOLONG. The remaining components, including a symlink that

100 leaves the base, are then appended without being resolved. The result

101 looks contained, but the kernel resolves the short path without that

102 limit and follows the link out (the CVE-2025-4517 class).

103

104 Here a component that does not exist is appended as written, so a leaf

105 that has not been created yet still resolves. Any other failure raises

106 ``OSError``: ENAMETOOLONG, ELOOP (a symlink loop), EACCES, and ENOTDIR

107 (a path that continues below a regular file). Python 3.12 reports a

108 loop as ``RuntimeError``, which is re-raised as ``OSError(ELOOP)``.

109

110 Without ``os.path.ALLOW_MISSING`` the longest prefix whose last entry

111 exists is resolved strictly and the missing tail is appended. A dangling

112 symlink, or a ``..`` in the missing tail, raises there as well. Both

113 are refusals that the ``ALLOW_MISSING`` path would not make, so the

114 fallback is only ever stricter.

115 """

116 if isinstance(path, str):

117 path = Path(path)

118 try:

119 if _ALLOW_MISSING is not None:

120 return path.resolve(strict=_ALLOW_MISSING)

121 return _resolve_allow_missing_fallback(path)

122 except RuntimeError as exc:

123 raise OSError(errno.ELOOP, "Symlink loop", str(path)) from exc

124

125

126def _resolve_allow_missing_fallback(path: Path) -> Path:

127 """``resolve_allow_missing`` for a Python without ``ALLOW_MISSING``."""

128 path = path.absolute()

129 missing: list[str] = []

130 head = path

131 while True:

132 try:

133 # lstat, not exists(): only ENOENT means "missing". Every

134 # other error (ENAMETOOLONG, ENOTDIR, EACCES) propagates.

135 os.lstat(head)

136 break

137 except FileNotFoundError:

138 if head.parent == head: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 raise

140 missing.append(head.name)

141 head = head.parent

142 if ".." in missing:

143 # ".." below a missing component cannot be resolved against the

144 # filesystem, and is_relative_to() would compare it lexically.

145 raise OSError(errno.ENOENT, "Unresolvable '..' segment", str(path))

146 # strict=True raises for a dangling symlink, a loop and ENAMETOOLONG.

147 return head.resolve(strict=True).joinpath(*reversed(missing))

148

149

150class PathValidator:

151 """Centralized path validation for security."""

152

153 # Regex for safe filename/path characters

154 SAFE_PATH_PATTERN = re.compile(r"^[a-zA-Z0-9._/\-]+$")

155

156 # Allowed config file extensions

157 CONFIG_EXTENSIONS = (".json", ".yaml", ".yml", ".toml", ".ini", ".conf")

158

159 @staticmethod

160 def validate_safe_path(

161 user_input: str,

162 base_dir: Union[str, Path],

163 allow_absolute: bool = False,

164 required_extensions: Optional[tuple] = None,

165 ) -> Optional[Path]:

166 """

167 Validate and sanitize a user-provided path.

168

169 Args:

170 user_input: The user-provided path string

171 base_dir: The safe base directory to contain paths within

172 allow_absolute: Whether to allow absolute paths (with restrictions)

173 required_extensions: Tuple of required file extensions (e.g., ('.json', '.yaml'))

174

175 Returns:

176 The validated Path. It is never None: every rejection raises.

177

178 Raises:

179 ValueError: If the path is invalid or unsafe

180 """

181 if not user_input or not isinstance(user_input, str):

182 raise ValueError("Invalid path input")

183

184 if "\x00" in user_input:

185 raise ValueError("Null bytes are not allowed in path")

186

187 # Strip whitespace

188 user_input = user_input.strip()

189 # (#6938: the input is a client-chosen path — e.g. the

190 # unauthenticated /static route — so the rejection log lines

191 # below interpolate a redacted, bounded form, never the raw value.

192 # It is built only in those rejection branches: this runs on every

193 # static-file lookup, and redaction costs up to tens of

194 # milliseconds on a long adversarial path.)

195

196 # Reject URL-encoded traversal tokens (single or double encoded).

197 # safe_join's literal-".."-check doesn't catch %2e%2e or %252e%252e,

198 # so do it explicitly here before any decoding takes place.

199 if _has_encoded_traversal(user_input):

200 logger.warning(

201 f"Encoded path-traversal attempt blocked: {redact_and_bound_for_log(user_input, 200)!r}"

202 )

203 raise ValueError(

204 "Invalid path - encoded traversal pattern detected"

205 )

206

207 # Reject unicode look-alike traversal (e.g. full-width '．．').

208 # NFKC normalization is the same form most filesystems / browsers

209 # apply; if normalizing produces a '..' segment the original was a

210 # disguised traversal attempt.

211 if _has_unicode_traversal(user_input):

212 logger.warning(

213 f"Unicode path-traversal attempt blocked: {redact_and_bound_for_log(user_input, 200)!r}"

214 )

215 raise ValueError(

216 "Invalid path - unicode traversal pattern detected"

217 )

218

219 # Use werkzeug's safe_join for secure path joining

220 # This handles path traversal attempts automatically. The base is

221 # resolved the same way as the joined path below, so the

222 # containment comparison is between two fully resolved paths.

223 try:

224 base_dir = resolve_allow_missing(Path(base_dir))

225 except OSError:

226 logger.warning("Path validation base directory is unresolvable")

227 raise ValueError(

228 "Invalid base directory - could not be resolved safely"

229 )

230

231 try:

232 # safe_join returns None if the path tries to escape base_dir

233 safe_path = safe_join(str(base_dir), user_input)

234 except ValueError:

235 raise

236 except Exception as e:

237 logger.warning(

238 f"Path validation failed for input '{redact_and_bound_for_log(user_input, 200)}'"

239 )

240 raise ValueError(f"Invalid path: {e}") from e

241

242 if safe_path is None:

243 logger.warning(

244 f"Path traversal attempt blocked: {redact_and_bound_for_log(user_input, 200)}"

245 )

246 raise ValueError("Invalid path - potential traversal attempt")

247

248 result_path = Path(safe_path)

249

250 # Filesystem-level containment (GHSA-gv87-qcw7-gjr8). safe_join is

251 # a *string* operation: it rejects ``..`` and absolute inputs but

252 # never touches the filesystem, so a symlinked directory or leaf

253 # inside base_dir that points outside it is joined and returned

254 # verbatim — the caller then reads through the link, outside the

255 # base. Resolve the joined path and require its real location to

256 # stay under the (already resolved) base. This is deliberately

257 # NOT #3090's removed check: that one rejected *every* symlink

258 # and broke legitimate layouts (bind mounts, macOS

259 # /tmp -> /private/tmp); containment accepts any symlink whose

260 # target resolves within base_dir and refuses only the escapes.

261 # resolve_allow_missing() tolerates only a missing component, so

262 # a not-yet-existing leaf is fine. Every other resolution error

263 # raises and fails closed here: a symlink loop (ELOOP), a path

264 # below a regular file (ENOTDIR), and ENAMETOOLONG. A bare

265 # non-strict resolve() ignores ENAMETOOLONG (and, on 3.13+, a

266 # loop) and appends the rest of the path unresolved, so a symlink

267 # chain whose resolved form outgrows PATH_MAX could hide an escape.

268 try:

269 resolved_path = resolve_allow_missing(result_path)

270 except OSError:

271 logger.warning(

272 f"Path resolution failed for input {redact_and_bound_for_log(user_input, 200)!r}"

273 )

274 raise ValueError("Invalid path - could not be resolved safely")

275 if not resolved_path.is_relative_to(base_dir):

276 logger.warning(

277 f"Symlink containment violation blocked: {redact_and_bound_for_log(user_input, 200)!r}"

278 )

279 raise ValueError(

280 "Invalid path - resolved path escapes base directory"

281 )

282 # Return the *lexical* (pre-resolve) path, not the resolved one:

283 # the containment check has already run, and callers key external

284 # behaviour off the requested name (e.g. FileResponse guesses the

285 # content type from the served filename, and the storage manager

286 # stores paths relative to the library root it passed in). Callers

287 # that need the real location resolve it themselves.

288

289 # Check extensions if required

290 if (

291 required_extensions

292 and result_path.suffix not in required_extensions

293 ):

294 raise ValueError(

295 f"Invalid file type. Allowed: {required_extensions}"

296 )

297

298 return result_path

299

300 @staticmethod

301 def validate_local_filesystem_path(

302 user_path: str,

303 restricted_dirs: Optional[list[Path]] = None,

304 ) -> Path:

305 """

306 Validate a user-provided absolute filesystem path for local indexing.

307

308 This is for features like local folder indexing where users need to

309 access files anywhere on their own machine, but system directories

310 should be blocked.

311

312 Args:

313 user_path: User-provided path string (absolute or with ~)

314 restricted_dirs: List of restricted directories to block

315

316 Returns:

317 Validated and resolved Path object

318

319 Raises:

320 ValueError: If path is invalid or points to restricted location

321 """

322 import sys

323

324 if not user_path or not isinstance(user_path, str):

325 raise ValueError("Invalid path input")

326

327 user_path = user_path.strip()

328

329 # Basic sanitation: forbid null bytes and control characters

330 if "\x00" in user_path:

331 raise ValueError("Null bytes are not allowed in path")

332 if any(ord(ch) < 32 for ch in user_path):

333 raise ValueError("Control characters are not allowed in path")

334

335 # Expand ~ to home directory

336 if user_path.startswith("~"):

337 home_dir = Path.home()

338 relative_part = user_path[2:].lstrip("/")

339 if relative_part:

340 user_path = str(home_dir / relative_part)

341 else:

342 user_path = str(home_dir)

343

344 # Disallow malformed Windows paths (e.g. "/C:/Windows")

345 if (

346 sys.platform == "win32"

347 and user_path.startswith(("/", "\\"))

348 and ":" in user_path

349 ):

350 raise ValueError("Malformed Windows path input")

351

352 # Block path traversal patterns before resolving

353 # This explicit check helps static analyzers understand the security intent

354 if ".." in user_path:

355 raise ValueError("Path traversal patterns not allowed")

356

357 # Use safe_join to sanitize the path - this is recognized by static analyzers

358 # For absolute paths, we validate against the root directory

359 if user_path.startswith("/"):

360 # Unix absolute path - use safe_join with root

361 safe_path = safe_join("/", user_path.lstrip("/"))

362 if safe_path is None:

363 raise ValueError("Invalid path - failed security validation")

364 validated_path = Path(safe_path).resolve()

365 elif len(user_path) > 2 and user_path[1] == ":":

366 # Windows absolute path (e.g., C:\Users\...)

367 drive = user_path[:2]

368 rest = user_path[2:].lstrip("\\").lstrip("/")

369 safe_path = safe_join(drive + "\\", rest)

370 if safe_path is None:

371 raise ValueError("Invalid path - failed security validation")

372 validated_path = Path(safe_path).resolve()

373 else:

374 # Relative path - resolve relative to current directory

375 # Use safe_join to validate

376 cwd = os.getcwd()

377 safe_path = safe_join(cwd, user_path)

378 if safe_path is None:

379 raise ValueError("Invalid path - failed security validation")

380 validated_path = Path(safe_path).resolve()

381

382 # Default restricted directories

383 if restricted_dirs is None:

384 restricted_dirs = [

385 Path("/etc"),

386 Path("/sys"),

387 Path("/proc"),

388 Path("/dev"),

389 Path("/root"),

390 Path("/boot"),

391 Path("/var/log"),

392 ]

393 # Add Windows system directories if on Windows

394 if sys.platform == "win32":

395 for drive in ["C:", "D:", "E:"]:

396 restricted_dirs.extend(

397 [

398 Path(f"{drive}\\Windows"),

399 Path(f"{drive}\\System32"),

400 Path(f"{drive}\\Program Files"),

401 Path(f"{drive}\\Program Files (x86)"),

402 ]

403 )

404

405 # Check against restricted directories. Resolve each restricted dir so

406 # the containment check holds on platforms where these are symlinks

407 # (e.g. macOS /etc -> /private/etc, /var -> /private/var) — validated_path

408 # is already resolved, so an unresolved "/etc" would never match

409 # "/private/etc/...". resolve(strict=False) does not raise for a missing

410 # or unreadable dir; fall back to the literal path if it ever does.

411 for restricted in restricted_dirs:

412 try:

413 restricted_resolved = restricted.resolve()

414 except OSError:

415 restricted_resolved = restricted

416 if validated_path.is_relative_to(restricted_resolved):

417 # Log WHICH restricted dir was hit, not the user's submitted

418 # path — a resolved local path can contain a username.

419 logger.error(

420 f"Security: blocked access to a restricted directory ({restricted})"

421 )

422 raise ValueError("Cannot access system directories")

423

424 return validated_path

425

426 @staticmethod

427 def sanitize_for_filesystem_ops(validated_path: Path) -> Path:

428 """

429 Re-sanitize a validated path for static analyzer recognition.

430

431 This method takes an already-validated Path and passes it through

432 werkzeug's safe_join to create a path that static analyzers like

433 CodeQL recognize as sanitized.

434

435 Note: This exists because CodeQL doesn't trace through custom validation

436 functions. The path is already secure from validate_local_filesystem_path(),

437 but safe_join makes that explicit to static analyzers.

438

439 Args:

440 validated_path: A Path that has already been validated by

441 validate_local_filesystem_path()

442

443 Returns:

444 A Path object safe for filesystem operations

445

446 Raises:

447 ValueError: If the path fails sanitization

448 """

449 if not validated_path.is_absolute():

450 raise ValueError("Path must be absolute")

451

452 # Use safe_join to create a sanitized path that static analyzers recognize

453 # safe_join handles path traversal detection properly (not substring matching)

454 path_str = str(validated_path)

455 safe_path_str = safe_join("/", path_str.lstrip("/"))

456 if safe_path_str is None:

457 raise ValueError("Path failed security sanitization")

458

459 return Path(safe_path_str)

460

461 # A `confine_to_base()` helper once lived here — see #4868. It filtered a list

462 # of paths down to those whose real (symlink-resolved) location stayed inside a

463 # base directory, dropping symlink escapes and loops.

464

465 @staticmethod

466 def validate_model_path(

467 model_path: str, model_root: Optional[str] = None

468 ) -> Path:

469 """

470 Validate a model file path specifically.

471

472 Args:

473 model_path: Path to the model file

474 model_root: Root directory for models (defaults to ~/.local/share/llm_models)

475

476 Returns:

477 Validated Path object

478

479 Raises:

480 ValueError: If the path is invalid

481 """

482 if model_root is None:

483 # Default model root - uses centralized path config (respects LDR_DATA_DIR)

484 model_root = str(get_models_directory())

485

486 # Create model root if it doesn't exist. Lazy import: this sibling

487 # module can be imported very early (e.g. from `security/__init__.py`

488 # during package bootstrap), so import at call time rather than at

489 # module top level to avoid any import-cycle risk.

490 from .directory_creation import create_directory

491

492 model_root_path = Path(model_root).resolve()

493 create_directory(

494 model_root_path,

495 context="model root directory",

496 )

497

498 # Validate the path

499 validated_path = PathValidator.validate_safe_path(

500 model_path,

501 model_root_path,

502 allow_absolute=False, # Models should always be relative to model root

503 required_extensions=None, # Models can have various extensions

504 )

505

506 if not validated_path:

507 raise ValueError("Invalid model path")

508

509 # Check if the file exists

510 if not validated_path.exists():

511 raise ValueError(f"Model file not found: {validated_path}")

512

513 if not validated_path.is_file():

514 raise ValueError(f"Model path is not a file: {validated_path}")

515

516 return validated_path

517

518 @staticmethod

519 def validate_data_path(file_path: str, data_root: str) -> Path:

520 """

521 Validate a path within the data directory.

522

523 Args:

524 file_path: Path relative to data root

525 data_root: The data root directory

526

527 Returns:

528 Validated Path object

529

530 Raises:

531 ValueError: If the path is invalid

532 """

533 validated_path = PathValidator.validate_safe_path(

534 file_path,

535 data_root,

536 allow_absolute=False, # Data paths should be relative

537 required_extensions=None,

538 )

539

540 if not validated_path:

541 raise ValueError("Invalid data path")

542

543 return validated_path

544

545 @staticmethod

546 def validate_config_path(

547 config_path: str, config_root: Optional[Union[str, Path]] = None

548 ) -> Path:

549 """

550 Validate a configuration file path.

551

552 Args:

553 config_path: Path to config file

554 config_root: Root directory for configs (optional for absolute paths)

555

556 Returns:

557 Validated Path object

558

559 Raises:

560 ValueError: If the path is invalid

561 """

562 # Validate input: reject null bytes, then normalize whitespace

563 if not config_path or not isinstance(config_path, str):

564 raise ValueError("Invalid config path input")

565

566 if "\x00" in config_path:

567 raise ValueError("Null bytes are not allowed in config path")

568

569 config_path = config_path.strip()

570

571 # Check for path traversal attempts in the string itself

572 # Define restricted system directories that should never be accessed

573 RESTRICTED_PREFIXES = ("etc", "proc", "sys", "dev")

574

575 if ".." in config_path:

576 raise ValueError("Invalid path - potential traversal attempt")

577

578 # Check if path starts with any restricted system directory

579 normalized_path = config_path.lstrip("/").lower()

580 for restricted in RESTRICTED_PREFIXES:

581 if (

582 normalized_path.startswith(restricted + "/")

583 or normalized_path == restricted

584 ):

585 raise ValueError(

586 f"Invalid path - restricted system directory: {restricted}"

587 )

588

589 # For config files, we might allow absolute paths with restrictions

590 # Check if path starts with / or drive letter (Windows) to detect absolute paths

591 # This avoids using Path() or os.path on user input

592 is_absolute = (

593 config_path.startswith("/") # Unix absolute

594 or (

595 len(config_path) > 2 and config_path[1] == ":"

596 ) # Windows absolute

597 )

598

599 if is_absolute:

600 # For absolute paths, use safe_join with root directory

601 # This validates the path without using Path() directly on user input

602 # Use safe_join to validate the absolute path

603 safe_path = safe_join("/", config_path)

604 if safe_path is None:

605 raise ValueError("Invalid absolute path")

606

607 # Now it's safe to create Path object from validated string

608 path_obj = Path(safe_path)

609

610 # Additional validation for config files

611 if path_obj.suffix not in PathValidator.CONFIG_EXTENSIONS:

612 raise ValueError(f"Invalid config file type: {path_obj.suffix}")

613

614 # Check existence using validated path

615 if not path_obj.exists():

616 raise ValueError(f"Config file not found: {path_obj}")

617

618 return path_obj

619 # For relative paths, use the config root

620 if config_root is None:

621 from ..config.paths import get_data_directory

622

623 config_root = get_data_directory()

624

625 validated = PathValidator.validate_safe_path(

626 config_path,

627 config_root,

628 allow_absolute=False,

629 required_extensions=PathValidator.CONFIG_EXTENSIONS,

630 )

631 if validated is None:

632 raise ValueError(f"Invalid config path: {config_path}")

633 return validated

Coverage for src/local_deep_research/security/path_validator.py: 96%

214 statements