Coverage for src/taipanstack/security/sanitizers.py: 100%
200 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 15:01 +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_VALID_SQL_PREFIX = frozenset(
24 "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_",
25) # pragma: no mutate
26_WINDOWS_RESERVED_NAMES = frozenset( # pragma: no mutate
27 {
28 "CON",
29 "PRN",
30 "AUX",
31 "NUL",
32 "COM1",
33 "COM2",
34 "COM3",
35 "COM4",
36 "COM5",
37 "COM6",
38 "COM7",
39 "COM8",
40 "COM9",
41 "LPT1",
42 "LPT2",
43 "LPT3",
44 "LPT4",
45 "LPT5",
46 "LPT6",
47 "LPT7",
48 "LPT8",
49 "LPT9",
50 },
51)
54def _handle_html(result: str, allow_html: bool) -> str:
55 """Remove HTML tags and escape entities if not allowed."""
56 if allow_html:
57 return result
58 result = _HTML_TAGS_RE.sub("", result)
59 result = result.replace("&", "&")
60 result = result.replace("<", "<")
61 return result.replace(">", ">")
64def _handle_unicode(result: str, allow_unicode: bool) -> str:
65 """Filter out non-ASCII characters if unicode is not allowed."""
66 if allow_unicode:
67 return result
68 return result.encode("ascii", errors="ignore").decode("ascii")
71def _check_string_length(value: str) -> None:
72 if len(value) > MAX_STRING_LENGTH:
73 raise ValueError("String length exceeds maximum allowed limit")
76def _check_max_length_param(max_length: int | None) -> None:
77 if max_length is not None and max_length < 0:
78 raise ValueError("max_length cannot be negative")
81def _validate_string_params(value: str, max_length: int | None) -> None:
82 """Validate parameters for sanitize_string."""
83 if not isinstance(value, str):
84 raise TypeError(f"value must be str, got {type(value).__name__}")
86 if max_length is not None and (
87 not isinstance(max_length, int) or isinstance(max_length, bool)
88 ):
89 raise TypeError(f"max_length must be int, got {type(max_length).__name__}")
91 _check_string_length(value)
92 _check_max_length_param(max_length)
95def sanitize_string(
96 value: str,
97 *,
98 max_length: int | None = None,
99 allow_html: bool = False,
100 allow_unicode: bool = True,
101 strip_whitespace: bool = True,
102) -> str:
103 """Sanitize a string by removing dangerous characters.
105 Args:
106 value: The string to sanitize.
107 max_length: Maximum length to truncate to.
108 allow_html: Whether to keep HTML tags (default: False).
109 allow_unicode: Whether to keep non-ASCII characters.
110 strip_whitespace: Whether to strip leading/trailing whitespace.
112 Returns:
113 The sanitized string.
115 Example:
116 ```python
117 sanitize_string("<script>alert('xss')</script>Hello")
118 # Returns: "scriptalert('xss')/scriptHello"
119 ```
121 """
122 _validate_string_params(value, max_length)
124 if not value:
125 return ""
127 result = value.strip() if strip_whitespace else value
128 result = _CONTROL_CHARS_RE.sub("", result)
129 result = _handle_html(result, allow_html)
130 result = _handle_unicode(result, allow_unicode)
132 if max_length is not None and len(result) > max_length:
133 return result[:max_length]
134 return result
137def _get_filename_from_path(filename: str) -> str:
138 """Extract the base filename from a full path."""
139 slash_idx = max(filename.rfind("/"), filename.rfind("\\"))
140 if slash_idx >= 0:
141 return filename[slash_idx + 1 :]
142 return filename
145def _has_valid_extension(name: str, idx: int) -> bool:
146 """Determine if a dot represents a valid extension."""
147 return idx > 0 and not all(c == "." for c in name) and name != ".."
150def _extract_stem_and_suffix(
151 filename: str,
152 preserve_extension: bool,
153) -> tuple[str, str]:
154 """Extract stem and suffix from a filename."""
155 name = _get_filename_from_path(filename)
156 idx = name.rfind(".")
158 if _has_valid_extension(name, idx):
159 stem = name[:idx]
160 suffix = name[idx:] if preserve_extension else ""
161 return stem, suffix
163 return name, ""
166def _remove_invalid_chars(stem: str, replacement: str) -> str:
167 """Remove or replace invalid characters in a filename stem."""
168 try:
169 if "\\" in replacement:
170 # Use lambda to avoid processing regex escape sequences in replacement
171 safe_stem = _INVALID_FILENAME_CHARS_RE.sub(lambda _: replacement, stem)
172 else:
173 safe_stem = _INVALID_FILENAME_CHARS_RE.sub(replacement, stem)
174 except re.error:
175 safe_stem = _INVALID_FILENAME_CHARS_RE.sub("_", stem)
177 # Remove leading/trailing dots and spaces (Windows issues)
178 safe_stem = safe_stem.strip(". ")
180 # Remove path separators that might have snuck through
181 safe_stem = safe_stem.replace("/", replacement)
182 safe_stem = safe_stem.replace("\\", replacement)
184 return safe_stem
187def _collapse_replacements(safe_stem: str, replacement: str) -> str:
188 """Collapse multiple consecutive replacement characters."""
189 if replacement:
190 double_replacement = replacement + replacement
191 while double_replacement in safe_stem:
192 safe_stem = safe_stem.replace(double_replacement, replacement)
193 safe_stem = safe_stem.strip(replacement)
194 return safe_stem
197def _truncate_filename(safe_stem: str, suffix: str, max_length: int) -> str:
198 """Truncate the filename while keeping the extension if possible."""
199 result = f"{safe_stem}{suffix}"
200 if len(result) > max_length:
201 available = max_length - len(suffix)
202 if available > 0:
203 safe_stem = safe_stem[:available]
204 result = f"{safe_stem}{suffix}"
205 else:
206 result = result[:max_length]
207 return result
210def _is_filename_safe(filename: str, max_length: int, stem: str) -> bool:
211 """Check if a filename is already safe without any modifications."""
212 return (
213 len(filename) <= max_length
214 and filename not in {"..", "."}
215 and stem.upper() not in _WINDOWS_RESERVED_NAMES
216 and filename.isascii()
217 and filename.replace(".", "").replace("-", "").replace("_", "").isalnum()
218 )
221def _finalize_filename(
222 safe_stem: str,
223 replacement: str,
224 suffix: str,
225 max_length: int,
226) -> str:
227 """Finalize the sanitized filename by handling reserved names and empty results."""
228 # Handle reserved names (Windows)
229 if safe_stem.upper() in _WINDOWS_RESERVED_NAMES:
230 safe_stem = f"{replacement}{safe_stem}"
232 # Handle empty result
233 if not safe_stem:
234 safe_stem = "unnamed"
236 return _truncate_filename(safe_stem, suffix, max_length)
239def _check_filename_type_str(filename: object, replacement: object) -> None:
240 """Check str types of parameters for sanitize_filename."""
241 if not isinstance(filename, str):
242 raise TypeError(f"filename must be str, got {type(filename).__name__}")
243 if not isinstance(replacement, str):
244 raise TypeError(f"replacement must be str, got {type(replacement).__name__}")
247def _check_filename_types(
248 filename: object,
249 max_length: object,
250 replacement: object,
251 preserve_extension: object,
252) -> None:
253 """Check types of parameters for sanitize_filename."""
254 _check_filename_type_str(filename, replacement)
256 if not isinstance(max_length, int) or isinstance(max_length, bool):
257 raise TypeError(f"max_length must be int, got {type(max_length).__name__}")
259 if not isinstance(preserve_extension, bool):
260 raise TypeError(
261 f"preserve_extension must be bool, got {type(preserve_extension).__name__}"
262 )
265def _check_filename_length(filename: str, max_length: int) -> None:
266 """Check max_length param and filename length."""
267 _check_max_length_param(max_length)
268 if len(filename) > MAX_PATH_LENGTH:
269 raise ValueError("Filename length exceeds maximum allowed limit")
272def _validate_filename_params(
273 filename: str, max_length: int, replacement: str, preserve_extension: bool
274) -> None:
275 """Validate parameters for sanitize_filename."""
276 _check_filename_types(filename, max_length, replacement, preserve_extension)
277 _check_filename_length(filename, max_length)
280def sanitize_filename(
281 filename: str,
282 *,
283 max_length: int = 255,
284 replacement: str = "_",
285 preserve_extension: bool = True,
286) -> str:
287 """Sanitize a filename to be safe for filesystem use.
289 Removes or replaces characters that are:
290 - Not allowed in filenames on various OSes
291 - Potentially dangerous (path separators, etc.)
293 Args:
294 filename: The filename to sanitize.
295 max_length: Maximum length for the filename.
296 replacement: Character to replace invalid chars with.
297 preserve_extension: Keep original extension.
299 Returns:
300 The sanitized filename.
302 Example:
303 ```python
304 sanitize_filename("my/../file<>:name.txt")
305 # Returns: 'my_file_name.txt'
306 ```
308 """
309 _validate_filename_params(filename, max_length, replacement, preserve_extension)
311 if not filename:
312 filename = "unnamed"
314 stem, suffix = _extract_stem_and_suffix(filename, preserve_extension)
316 if _is_filename_safe(filename, max_length, stem):
317 return f"{stem}{suffix}"
319 safe_stem = _remove_invalid_chars(stem, replacement)
320 safe_stem = _collapse_replacements(safe_stem, replacement)
322 return _finalize_filename(safe_stem, replacement, suffix, max_length)
325def _get_stem(part: str) -> str:
326 """Get the stem of a path part."""
327 idx = part.rfind(".")
328 return part[:idx] if idx > 0 and not all(c == "." for c in part) else part
331def _is_safe_path_part(part: str, stem: str) -> bool:
332 """Check if a path part is safe."""
333 return (
334 len(part) <= 255 # noqa: PLR2004
335 and part.isascii()
336 and part.replace(".", "").replace("-", "").replace("_", "").isalnum()
337 and stem.upper() not in _WINDOWS_RESERVED_NAMES
338 )
341def _handle_dot_dot(parts: list[str], anchor: str) -> None:
342 """Handle '..' by popping the last part if safe."""
343 if parts and parts[-1] != ".." and parts[-1] != anchor:
344 parts.pop()
347def _handle_normal_part(part: str, parts: list[str]) -> None:
348 """Handle a normal part by checking if it's safe or sanitizing it."""
349 stem = _get_stem(part)
350 if _is_safe_path_part(part, stem):
351 parts.append(part)
352 else:
353 safe_part = sanitize_filename(part, preserve_extension=True)
354 if safe_part and safe_part != "..":
355 parts.append(safe_part)
358def _process_path_part(part: str, parts: list[str], anchor: str) -> None:
359 """Process a single path component, updating the parts list inline."""
360 if part == "..":
361 _handle_dot_dot(parts, anchor)
362 elif part != ".":
363 _handle_normal_part(part, parts)
366def _clean_path_parts(path: Path) -> list[str]:
367 """Clean and sanitize individual path components."""
368 parts: list[str] = []
369 anchor = path.anchor
370 for part in path.parts:
371 _process_path_part(part, parts, anchor)
372 return parts
375def _apply_base_dir_constraint(
376 sanitized: Path,
377 base_dir: Path | str | None,
378 resolve: bool,
379) -> Path:
380 """Apply base directory constraints to a sanitized path."""
381 if base_dir is None:
382 return sanitized
384 base = Path(base_dir).resolve()
385 if resolve:
386 try:
387 return sanitized.resolve()
388 except (OSError, RuntimeError) as e:
389 msg = f"Cannot resolve path: {e}"
390 raise ValueError(msg) from e
392 # Make absolute relative to base
393 if not sanitized.is_absolute():
394 return base / sanitized
396 return sanitized
399def _process_string_path(path: str) -> Path:
400 """Process a string path, removing null bytes and validating length."""
401 if len(path) > MAX_PATH_LENGTH:
402 msg = "Path length exceeds maximum allowed"
403 raise ValueError(msg)
404 if "\x00" in path:
405 path = path.replace("\x00", "")
406 return Path(path)
409def _check_pathlike_length(path: os.PathLike[str]) -> None:
410 """Check the length of a PathLike object."""
411 if len(str(path)) > MAX_PATH_LENGTH:
412 msg = "Path length exceeds maximum allowed"
413 raise ValueError(msg)
416def _normalize_path_input(path: str | os.PathLike[str]) -> Path:
417 """Normalize input path string or Path object."""
418 if not isinstance(path, (str, os.PathLike)):
419 raise TypeError(
420 f"path must be a string or PathLike object, got {type(path).__name__}"
421 )
423 if isinstance(path, str):
424 return _process_string_path(path)
426 _check_pathlike_length(path)
427 return Path(path)
430def _reconstruct_path(original_path: Path, parts: list[str]) -> Path:
431 """Reconstruct a path from its sanitized parts."""
432 if original_path.is_absolute():
433 # Use path.anchor to correctly preserve absolute roots on Windows (e.g. C:\)
434 anchor = Path(original_path.anchor)
435 return anchor.joinpath(*parts) if parts else anchor
437 if parts:
438 return Path().joinpath(*parts)
440 return Path()
443def _validate_path_depth(path: Path, max_depth: int | None) -> None:
444 """Validate that the path depth does not exceed max_depth."""
445 depth = len(path.parts)
446 if max_depth is not None and depth > max_depth:
447 msg = f"Path depth {depth} exceeds maximum of {max_depth}"
448 raise ValueError(msg)
451def sanitize_path(
452 path: str | Path,
453 *,
454 base_dir: Path | None = None,
455 max_depth: int | None = 10,
456 resolve: bool = False,
457) -> Path:
458 """Sanitize a path to prevent traversal and normalize it.
460 Args:
461 path: The path to sanitize.
462 base_dir: Optional base directory to constrain to.
463 max_depth: Maximum directory depth allowed.
464 resolve: Whether to resolve the path (requires it to exist).
466 Returns:
467 The sanitized Path object.
469 Raises:
470 ValueError: If path is invalid or too deep.
472 """
473 if max_depth is not None and max_depth < 0:
474 raise ValueError("max_depth cannot be negative")
476 normalized_path = _normalize_path_input(path)
477 parts = _clean_path_parts(normalized_path)
478 sanitized = _reconstruct_path(normalized_path, parts)
480 _validate_path_depth(sanitized, max_depth)
481 return _apply_base_dir_constraint(sanitized, base_dir, resolve)