Coverage for src/taipanstack/security/sanitizers.py: 100%
222 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Input sanitizers for cleaning untrusted data.
4Provides functions to sanitize strings, filenames, and paths
5to remove potentially dangerous characters.
6"""
8import os
9import re
10from pathlib import Path
12# Constants to avoid magic values (PLR2004)
13MAX_PATH_LENGTH = 4096 # pragma: no mutate
14MAX_STRING_LENGTH = 1000000 # pragma: no mutate
16# Pre-compiled regex and sets for Performance Benchmarks
17_INVALID_FILENAME_CHARS_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f]') # pragma: no mutate
18_HTML_TAGS_RE = re.compile(r"<[^>]+>") # pragma: no mutate
19# Remove control characters (C0 and C1 sets)
20_CONTROL_CHARS_RE = re.compile(
21 r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]",
22) # pragma: no mutate
23_WINDOWS_RESERVED_NAMES = frozenset( # pragma: no mutate
24 {
25 "CON",
26 "PRN",
27 "AUX",
28 "NUL",
29 "COM1",
30 "COM2",
31 "COM3",
32 "COM4",
33 "COM5",
34 "COM6",
35 "COM7",
36 "COM8",
37 "COM9",
38 "LPT1",
39 "LPT2",
40 "LPT3",
41 "LPT4",
42 "LPT5",
43 "LPT6",
44 "LPT7",
45 "LPT8",
46 "LPT9",
47 },
48)
51def _handle_html(result: str, allow_html: bool) -> str:
52 """Remove HTML tags and escape entities if not allowed."""
53 if allow_html:
54 return result
55 result = _HTML_TAGS_RE.sub("", result)
56 result = result.replace("&", "&")
57 result = result.replace("<", "<")
58 return result.replace(">", ">")
61def _handle_unicode(result: str, allow_unicode: bool) -> str:
62 """Filter out non-ASCII characters if unicode is not allowed."""
63 if allow_unicode:
64 return result
65 return result.encode("ascii", errors="ignore").decode("ascii")
68def _check_string_length(value: str) -> None:
69 if len(value) > MAX_STRING_LENGTH:
70 raise ValueError("String length exceeds maximum allowed limit")
73def _check_max_length_param(max_length: int | None) -> None:
74 if max_length is not None and max_length < 0:
75 raise ValueError("max_length cannot be negative")
78def _is_valid_max_length_type(max_length: int | None) -> bool:
79 """Check if max_length is of valid integer type."""
80 if max_length is None:
81 return True
82 return isinstance(max_length, int) and not isinstance(max_length, bool)
85def _validate_string_params(value: str, max_length: int | None) -> None:
86 """Validate parameters for sanitize_string."""
87 if not isinstance(value, str):
88 raise TypeError(f"value must be str, got {type(value).__name__}")
90 if not _is_valid_max_length_type(max_length):
91 raise TypeError(f"max_length must be int, got {type(max_length).__name__}")
93 _check_string_length(value)
94 _check_max_length_param(max_length)
97def _enforce_max_length(value: str, max_length: int | None) -> str:
98 if max_length is not None and len(value) > max_length:
99 return value[:max_length]
100 return value
103def sanitize_string(
104 value: str,
105 *,
106 max_length: int | None = None,
107 allow_html: bool = False,
108 allow_unicode: bool = True,
109 strip_whitespace: bool = True,
110) -> str:
111 """Sanitize a string by removing dangerous characters.
113 Args:
114 value: The string to sanitize.
115 max_length: Maximum length to truncate to.
116 allow_html: Whether to keep HTML tags (default: False).
117 allow_unicode: Whether to keep non-ASCII characters.
118 strip_whitespace: Whether to strip leading/trailing whitespace.
120 Returns:
121 The sanitized string.
123 Example:
124 ```python
125 sanitize_string("<script>alert('xss')</script>Hello")
126 # Returns: "scriptalert('xss')/scriptHello"
127 ```
129 """
130 _validate_string_params(value, max_length)
132 if not value:
133 return ""
135 result = value.strip() if strip_whitespace else value
136 result = _CONTROL_CHARS_RE.sub("", result)
137 result = _handle_html(result, allow_html)
138 result = _handle_unicode(result, allow_unicode)
140 return _enforce_max_length(result, max_length)
143def _get_filename_from_path(filename: str) -> str:
144 """Extract the base filename from a full path."""
145 slash_idx = max(filename.rfind("/"), filename.rfind("\\"))
146 if slash_idx >= 0:
147 return filename[slash_idx + 1 :]
148 return filename
151def _has_valid_extension(name: str, idx: int) -> bool:
152 """Determine if a dot represents a valid extension."""
153 return idx > 0 and not all(c == "." for c in name) and name != ".."
156def _extract_stem_and_suffix(
157 filename: str,
158 preserve_extension: bool,
159) -> tuple[str, str]:
160 """Extract stem and suffix from a filename."""
161 name = _get_filename_from_path(filename)
162 idx = name.rfind(".")
164 if _has_valid_extension(name, idx):
165 stem = name[:idx]
166 suffix = name[idx:] if preserve_extension else ""
167 return stem, suffix
169 return name, ""
172def _remove_invalid_chars(stem: str, replacement: str) -> str:
173 """Remove or replace invalid characters in a filename stem."""
174 try:
175 if "\\" in replacement:
176 # Use lambda to avoid processing regex escape sequences in replacement
177 safe_stem = _INVALID_FILENAME_CHARS_RE.sub(lambda _: replacement, stem) # type: ignore[misc]
178 else:
179 safe_stem = _INVALID_FILENAME_CHARS_RE.sub(replacement, stem)
180 except re.error:
181 safe_stem = _INVALID_FILENAME_CHARS_RE.sub("_", stem)
183 # Remove leading/trailing dots and spaces (Windows issues)
184 safe_stem = safe_stem.strip(". ")
186 # Remove path separators that might have snuck through
187 safe_stem = safe_stem.replace("/", replacement)
188 safe_stem = safe_stem.replace("\\", replacement)
190 return safe_stem
193def _collapse_replacements(safe_stem: str, replacement: str) -> str:
194 """Collapse multiple consecutive replacement characters."""
195 if replacement:
196 double_replacement = replacement + replacement
197 while double_replacement in safe_stem:
198 safe_stem = safe_stem.replace(double_replacement, replacement)
199 safe_stem = safe_stem.strip(replacement)
200 return safe_stem
203def _truncate_filename(safe_stem: str, suffix: str, max_length: int) -> str:
204 """Truncate the filename while keeping the extension if possible."""
205 result = f"{safe_stem}{suffix}"
206 if len(result) > max_length:
207 available = max_length - len(suffix)
208 if available > 0:
209 safe_stem = safe_stem[:available]
210 result = f"{safe_stem}{suffix}"
211 else:
212 result = result[:max_length]
213 return result
216def _is_valid_length_and_not_dot(filename: str, max_length: int) -> bool:
217 """Check if filename length is valid and not a dot directory."""
218 if len(filename) > max_length:
219 return False
220 return filename not in {"..", "."}
223def _has_safe_characters(filename: str, stem: str) -> bool:
224 """Check if filename has safe characters and is not a reserved name."""
225 if stem.upper() in _WINDOWS_RESERVED_NAMES:
226 return False
227 if not filename.isascii():
228 return False
229 return filename.replace(".", "").replace("-", "").replace("_", "").isalnum()
232def _is_filename_safe(filename: str, max_length: int, stem: str) -> bool:
233 """Check if a filename is already safe without any modifications."""
234 if not _is_valid_length_and_not_dot(filename, max_length):
235 return False
236 return _has_safe_characters(filename, stem)
239def _handle_reserved_names(safe_stem: str, replacement: str) -> str:
240 """Handle Windows reserved names by prepending the replacement character."""
241 if safe_stem.upper() in _WINDOWS_RESERVED_NAMES:
242 return f"{replacement}{safe_stem}"
243 return safe_stem
246def _finalize_filename(
247 safe_stem: str,
248 replacement: str,
249 suffix: str,
250 max_length: int,
251) -> str:
252 """Finalize the sanitized filename by handling reserved names and empty results."""
253 safe_stem = _handle_reserved_names(safe_stem, replacement)
255 # Handle empty result
256 if not safe_stem:
257 safe_stem = "unnamed"
259 return _truncate_filename(safe_stem, suffix, max_length)
262def _check_filename_type_str(filename: object, replacement: object) -> None:
263 """Check str types of parameters for sanitize_filename."""
264 if not isinstance(filename, str):
265 raise TypeError(f"filename must be str, got {type(filename).__name__}")
266 if not isinstance(replacement, str):
267 raise TypeError(f"replacement must be str, got {type(replacement).__name__}")
270def _check_filename_types(
271 filename: object,
272 max_length: object,
273 replacement: object,
274 preserve_extension: object,
275) -> None:
276 """Check types of parameters for sanitize_filename."""
277 _check_filename_type_str(filename, replacement)
279 if not isinstance(max_length, int) or isinstance(max_length, bool):
280 raise TypeError(f"max_length must be int, got {type(max_length).__name__}")
282 if not isinstance(preserve_extension, bool):
283 raise TypeError(
284 f"preserve_extension must be bool, got {type(preserve_extension).__name__}"
285 )
288def _check_filename_length(filename: str, max_length: int) -> None:
289 """Check max_length param and filename length."""
290 _check_max_length_param(max_length)
291 if len(filename) > MAX_PATH_LENGTH:
292 raise ValueError("Filename length exceeds maximum allowed limit")
295def _validate_filename_params(
296 filename: str, max_length: int, replacement: str, preserve_extension: bool
297) -> None:
298 """Validate parameters for sanitize_filename."""
299 _check_filename_types(filename, max_length, replacement, preserve_extension)
300 _check_filename_length(filename, max_length)
303def sanitize_filename(
304 filename: str,
305 *,
306 max_length: int = 255,
307 replacement: str = "_",
308 preserve_extension: bool = True,
309) -> str:
310 """Sanitize a filename to be safe for filesystem use.
312 Removes or replaces characters that are:
313 - Not allowed in filenames on various OSes
314 - Potentially dangerous (path separators, etc.)
316 Args:
317 filename: The filename to sanitize.
318 max_length: Maximum length for the filename.
319 replacement: Character to replace invalid chars with.
320 preserve_extension: Keep original extension.
322 Returns:
323 The sanitized filename.
325 Example:
326 ```python
327 sanitize_filename("my/../file<>:name.txt")
328 # Returns: 'my_file_name.txt'
329 ```
331 """
332 _validate_filename_params(filename, max_length, replacement, preserve_extension)
334 if not filename:
335 filename = "unnamed"
337 stem, suffix = _extract_stem_and_suffix(filename, preserve_extension)
339 if _is_filename_safe(filename, max_length, stem):
340 return f"{stem}{suffix}"
342 safe_stem = _remove_invalid_chars(stem, replacement)
343 safe_stem = _collapse_replacements(safe_stem, replacement)
345 return _finalize_filename(safe_stem, replacement, suffix, max_length)
348def _get_stem(part: str) -> str:
349 """Get the stem of a path part."""
350 idx = part.rfind(".")
351 return part[:idx] if idx > 0 and not all(c == "." for c in part) else part
354def _is_safe_path_part(part: str, stem: str) -> bool:
355 """Check if a path part is safe."""
356 return (
357 len(part) <= 255 # noqa: PLR2004
358 and part.isascii()
359 and part.replace(".", "").replace("-", "").replace("_", "").isalnum()
360 and stem.upper() not in _WINDOWS_RESERVED_NAMES
361 )
364def _handle_dot_dot(parts: list[str], anchor: str) -> None:
365 """Handle '..' by popping the last part if safe."""
366 if parts and parts[-1] != ".." and parts[-1] != anchor:
367 parts.pop()
370def _handle_normal_part(part: str, parts: list[str]) -> None:
371 """Handle a normal part by checking if it's safe or sanitizing it."""
372 stem = _get_stem(part)
373 if _is_safe_path_part(part, stem):
374 parts.append(part)
375 else:
376 safe_part = sanitize_filename(part, preserve_extension=True)
377 if safe_part and safe_part != "..":
378 parts.append(safe_part)
381def _process_path_part(part: str, parts: list[str], anchor: str) -> None:
382 """Process a single path component, updating the parts list inline."""
383 if part == "..":
384 _handle_dot_dot(parts, anchor)
385 elif part != ".":
386 _handle_normal_part(part, parts)
389def _clean_path_parts(path: Path) -> list[str]:
390 """Clean and sanitize individual path components."""
391 parts: list[str] = []
392 anchor = path.anchor
393 for part in path.parts:
394 _process_path_part(part, parts, anchor)
395 return parts
398def _resolve_sanitized_path(sanitized: Path) -> Path:
399 """Resolve a sanitized path."""
400 try:
401 return sanitized.resolve()
402 except (OSError, RuntimeError) as e:
403 msg = f"Cannot resolve path: {e}"
404 raise ValueError(msg) from e
407def _apply_base_dir_constraint(
408 sanitized: Path,
409 base_dir: Path | str | None,
410 resolve: bool,
411) -> Path:
412 """Apply base directory constraints to a sanitized path."""
413 if base_dir is None:
414 return sanitized
416 base = Path(base_dir).resolve()
417 if resolve:
418 return _resolve_sanitized_path(sanitized)
420 # Make absolute relative to base
421 if not sanitized.is_absolute():
422 return base / sanitized
424 return sanitized
427def _process_string_path(path: str) -> Path:
428 """Process a string path, removing null bytes and validating length."""
429 if len(path) > MAX_PATH_LENGTH:
430 msg = "Path length exceeds maximum allowed"
431 raise ValueError(msg)
432 if "\x00" in path:
433 path = path.replace("\x00", "")
434 return Path(path)
437def _check_pathlike_length(path: os.PathLike[str]) -> None:
438 """Check the length of a PathLike object."""
439 if len(str(path)) > MAX_PATH_LENGTH:
440 msg = "Path length exceeds maximum allowed"
441 raise ValueError(msg)
444def _normalize_path_input(path: str | os.PathLike[str]) -> Path:
445 """Normalize input path string or Path object."""
446 if not isinstance(path, (str, os.PathLike)):
447 raise TypeError(
448 f"path must be a string or PathLike object, got {type(path).__name__}"
449 )
451 if isinstance(path, str):
452 return _process_string_path(path)
454 _check_pathlike_length(path)
455 return Path(path)
458def _reconstruct_path(original_path: Path, parts: list[str]) -> Path:
459 """Reconstruct a path from its sanitized parts."""
460 if original_path.is_absolute():
461 # Use path.anchor to correctly preserve absolute roots on Windows (e.g. C:\)
462 anchor = Path(original_path.anchor)
463 return anchor.joinpath(*parts) if parts else anchor
465 if parts:
466 return Path().joinpath(*parts)
468 return Path()
471def _validate_path_depth(path: Path, max_depth: int | None) -> None:
472 """Validate that the path depth does not exceed max_depth."""
473 depth = len(path.parts)
474 if max_depth is not None and depth > max_depth:
475 msg = f"Path depth {depth} exceeds maximum of {max_depth}"
476 raise ValueError(msg)
479def sanitize_path(
480 path: str | Path,
481 *,
482 base_dir: Path | None = None,
483 max_depth: int | None = 10,
484 resolve: bool = False,
485) -> Path:
486 """Sanitize a path to prevent traversal and normalize it.
488 Args:
489 path: The path to sanitize.
490 base_dir: Optional base directory to constrain to.
491 max_depth: Maximum directory depth allowed.
492 resolve: Whether to resolve the path (requires it to exist).
494 Returns:
495 The sanitized Path object.
497 Raises:
498 ValueError: If path is invalid or too deep.
500 """
501 if max_depth is not None and max_depth < 0:
502 raise ValueError("max_depth cannot be negative")
504 normalized_path = _normalize_path_input(path)
505 parts = _clean_path_parts(normalized_path)
506 sanitized = _reconstruct_path(normalized_path, parts)
508 _validate_path_depth(sanitized, max_depth)
509 return _apply_base_dir_constraint(sanitized, base_dir, resolve)