Coverage for src/taipanstack/security/guards.py: 100%
273 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Runtime guards for protection against errors and AI hallucinations.
4These guards provide runtime protection against common security issues
5and programming errors that can occur from incorrect AI-generated code.
6All guards raise SecurityError on violation.
7"""
9import functools
10import ipaddress
11import os
12import re
13import socket
14import unicodedata
15from collections.abc import Sequence
16from pathlib import Path
17from urllib.parse import SplitResult, unquote, urlsplit
19from result import Err, Ok, Result
21from taipanstack.security.sanitizers import MAX_PATH_LENGTH
22from taipanstack.security.validators import MAX_ENV_VAR_LENGTH, MAX_URL_LENGTH
24# Build regex for traversal patterns.
25# Note: we handle ~ specially to only match at start of path or after a separator
26# to avoid false positives with Windows short paths (e.g., RUNNER~1).
27TRAVERSAL_REGEX = re.compile(
28 r"(?:\.\.|%2e%2e|%252e%252e)|(?:^|[\\/])~",
29 re.IGNORECASE,
30)
32_DANGEROUS_COMMAND_PATTERNS: tuple[tuple[str, str], ...] = (
33 (";", "command separator"),
34 ("|", "pipe"),
35 ("&", "background/and operator"),
36 ("$", "variable expansion"),
37 ("`", "command substitution"),
38 ("$(", "command substitution"),
39 ("${", "variable expansion"),
40 (">", "redirect"),
41 ("<", "redirect"),
42 (">>", "redirect append"),
43 ("||", "or operator"),
44 ("&&", "and operator"),
45 ("\n", "newline"),
46 ("\r", "carriage return"),
47 ("\x00", "null byte"),
48)
50# Pre-compiled regex and lookup map for fast-path command injection detection
51_DANGEROUS_COMMAND_RE = re.compile(
52 "|".join(re.escape(p) for p, _ in _DANGEROUS_COMMAND_PATTERNS),
53)
54_DANGEROUS_COMMAND_LOOKUP = dict(_DANGEROUS_COMMAND_PATTERNS)
56_DEFAULT_DENIED_EXTENSIONS = frozenset(
57 [
58 "exe",
59 "dll",
60 "so",
61 "dylib", # Executables
62 "sh",
63 "bash",
64 "zsh",
65 "ps1",
66 "bat",
67 "cmd", # Scripts
68 "php",
69 "jsp",
70 "asp",
71 "aspx", # Server-side scripts
72 ],
73)
75_DEFAULT_DENIED_ENV_VARS = frozenset(
76 [
77 "AWS_SECRET_ACCESS_KEY",
78 "AWS_SESSION_TOKEN",
79 "GITHUB_TOKEN",
80 "GH_TOKEN",
81 "GITLAB_TOKEN",
82 "DATABASE_URL",
83 "DB_PASSWORD",
84 "PASSWORD",
85 "SECRET_KEY",
86 "PRIVATE_KEY",
87 "API_KEY",
88 "API_SECRET",
89 ],
90)
92_SENSITIVE_ENV_VAR_PATTERN = re.compile(
93 r"SECRET|PASSWORD|TOKEN|PRIVATE.*?KEY|API.*?KEY",
94)
97class SecurityError(Exception):
98 """Raised when a security guard detects a violation.
100 Attributes:
101 guard_name: Name of the guard that was triggered.
102 message: Description of the violation.
103 value: The offending value (if safe to log).
105 """
107 def __init__(
108 self,
109 message: str,
110 guard_name: str = "unknown",
111 value: str | None = None,
112 ) -> None:
113 """Initialize SecurityError.
115 Args:
116 message: Description of the violation.
117 guard_name: Name of the guard that triggered.
118 value: The offending value (sanitized).
120 """
121 self.guard_name = guard_name
122 self.value = value
123 super().__init__(f"[{guard_name}] {message}")
126def _check_traversal_patterns(path_str: str) -> None:
127 """Check for explicit traversal patterns before resolution."""
128 match = TRAVERSAL_REGEX.search(path_str.lower())
129 if match:
130 raise SecurityError(
131 f"Path traversal pattern detected: {match.group(0)}",
132 guard_name="path_traversal",
133 value=path_str[:50], # Truncate for safety
134 )
137def _resolve_and_check_bounds(path: Path, base_dir: Path) -> tuple[Path, Path]:
138 """Resolve the path and check if it is within base_dir."""
139 try:
140 full_path = path if path.is_absolute() else (base_dir / path)
141 resolved = full_path.resolve()
142 except (OSError, ValueError, RuntimeError) as e:
143 raise SecurityError(
144 f"Invalid path: {e}",
145 guard_name="path_traversal",
146 ) from e
148 if not resolved.is_relative_to(base_dir):
149 raise SecurityError(
150 "Path escapes base directory",
151 guard_name="path_traversal",
152 )
153 return full_path, resolved
156def _check_symlink_safety(full_path: Path, base_dir: Path) -> None:
157 """Check for symlinks recursively up to the base directory."""
158 current = full_path
159 # Only check components from the user-provided path, not the base_dir
160 while current not in (base_dir, current.parent):
161 # We don't check .exists() because it returns False for broken symlinks
162 try:
163 is_symlink = current.is_symlink()
164 except OSError as e:
165 raise SecurityError(
166 f"Invalid path encountered during symlink check: {e}",
167 guard_name="path_traversal",
168 value=str(current)[:50],
169 ) from e
170 if is_symlink:
171 raise SecurityError(
172 "Symlinks are not allowed",
173 guard_name="path_traversal",
174 value=str(current),
175 )
176 current = current.parent
179def _check_path_types(path: object, base_dir: object) -> None:
180 """Validate types of path and base_dir."""
181 if not isinstance(path, (str, Path)):
182 raise TypeError(f"path must be str or Path, got {type(path).__name__}")
183 if base_dir is not None and not isinstance(base_dir, (str, Path)):
184 raise TypeError(f"base_dir must be str or Path, got {type(base_dir).__name__}")
187def _check_path_lengths(path: object, base_dir: object) -> None:
188 """Validate lengths of path and base_dir."""
189 if len(str(path)) > MAX_PATH_LENGTH:
190 raise SecurityError(
191 f"Path length exceeds maximum allowed limit of {MAX_PATH_LENGTH}",
192 guard_name="path_traversal",
193 )
195 if base_dir is not None and len(str(base_dir)) > MAX_PATH_LENGTH:
196 raise SecurityError(
197 f"Base directory length exceeds maximum allowed limit of {MAX_PATH_LENGTH}",
198 guard_name="path_traversal",
199 )
202def _validate_path_types(path: object, base_dir: object) -> None:
203 """Validate types and lengths of path and base_dir."""
204 _check_path_types(path, base_dir)
205 _check_path_lengths(path, base_dir)
208def _check_path_null_bytes(path: Path | str, base_dir: Path | str | None) -> None:
209 """Check for null bytes in path and base_dir."""
210 if "\x00" in str(path) or (base_dir is not None and "\x00" in str(base_dir)):
211 raise SecurityError(
212 "Path contains null bytes",
213 guard_name="path_traversal",
214 )
217def _resolve_base_dir(base_dir: Path | str | None) -> Path:
218 """Resolve the base directory."""
219 return Path(base_dir).resolve() if base_dir else Path.cwd().resolve()
222def guard_path_traversal(
223 path: Path | str,
224 base_dir: Path | str | None = None,
225 *,
226 allow_symlinks: bool = False,
227) -> Path:
228 """Prevent path traversal attacks.
230 Ensures that the given path does not escape the base directory
231 using techniques like '..' or symlinks.
233 Args:
234 path: The path to validate.
235 base_dir: The base directory to constrain to. Defaults to cwd.
236 allow_symlinks: Whether to allow symlinks (default: False).
238 Returns:
239 The resolved, validated path.
241 Raises:
242 SecurityError: If path traversal is detected.
244 Example:
245 >>> guard_path_traversal("../etc/passwd", Path("/app"))
246 SecurityError: [path_traversal] Path escapes base directory
248 """
249 _validate_path_types(path, base_dir)
250 _check_path_null_bytes(path, base_dir)
252 path_obj = Path(path) if isinstance(path, str) else path
253 base = _resolve_base_dir(base_dir)
255 _check_traversal_patterns(str(path_obj))
256 full_path, resolved = _resolve_and_check_bounds(path_obj, base)
258 if not allow_symlinks:
259 _check_symlink_safety(full_path, base)
261 return resolved
264def _check_command_not_empty(command: Sequence[str]) -> None:
265 if not command:
266 raise SecurityError(
267 "Empty command is not allowed",
268 guard_name="command_injection",
269 )
272def _check_command_null_bytes(cmd_list: list[str]) -> None:
273 for arg in cmd_list:
274 if isinstance(arg, str) and "\x00" in arg:
275 raise SecurityError(
276 "Dangerous shell character detected: null byte",
277 guard_name="command_injection",
278 value=arg[:50],
279 )
282def _check_command_patterns(cmd_list: list[str]) -> None:
283 for i, arg in enumerate(cmd_list):
284 if not isinstance(arg, str):
285 raise TypeError(
286 f"All command arguments must be strings, "
287 f"got {type(arg).__name__} at index {i}",
288 )
290 match = _DANGEROUS_COMMAND_RE.search(arg)
291 if match:
292 description = _DANGEROUS_COMMAND_LOOKUP[match.group(0)]
293 raise SecurityError(
294 f"Dangerous shell character detected: {description}",
295 guard_name="command_injection",
296 value=arg[:50],
297 )
300def _check_allowed_commands(
301 cmd_list: list[str],
302 allowed_commands: Sequence[str] | None,
303) -> None:
304 if allowed_commands is None:
305 return
307 base_command = cmd_list[0]
308 command_name = Path(base_command).name
309 cmd_not_allowed = (
310 command_name not in allowed_commands and base_command not in allowed_commands
311 )
312 if cmd_not_allowed:
313 raise SecurityError(
314 f"Command not in allowed list: {command_name}",
315 guard_name="command_injection",
316 value=command_name,
317 )
320def guard_command_injection(
321 command: Sequence[str],
322 *,
323 allowed_commands: Sequence[str] | None = None,
324) -> list[str]:
325 """Prevent command injection attacks.
327 Validates that command arguments don't contain shell metacharacters
328 that could lead to command injection.
330 Args:
331 command: The command and arguments as a sequence.
332 allowed_commands: Optional whitelist of allowed base commands.
334 Returns:
335 The validated command as a list.
337 Raises:
338 SecurityError: If command injection is detected.
340 Example:
341 >>> guard_command_injection(["echo", "hello; rm -rf /"])
342 SecurityError: [command_injection] Dangerous characters detected
344 """
345 cmd_list = list(command)
347 _check_command_not_empty(cmd_list)
349 _check_command_null_bytes(cmd_list)
350 _check_command_patterns(cmd_list)
351 _check_allowed_commands(cmd_list, allowed_commands)
353 return cmd_list
356def _check_filename_null_bytes(filename_str: str) -> None:
357 if "\x00" in filename_str:
358 raise SecurityError(
359 "Filename contains null bytes",
360 guard_name="file_extension",
361 value=filename_str,
362 )
365def _is_invalid_filename_end_char(char: str) -> bool:
366 return (
367 char == "."
368 or unicodedata.category(char).startswith(("Z", "C"))
369 or char == "\xad"
370 )
373def _clean_filename_end(clean_name: str) -> str:
374 end_idx = len(clean_name)
375 while end_idx > 0 and _is_invalid_filename_end_char(clean_name[end_idx - 1]):
376 end_idx -= 1
377 return clean_name[:end_idx]
380def _normalize_ext(e: str) -> str:
381 return e.lower().lstrip(".")
384def _check_denied_extension(
385 ext: str,
386 original_name: str,
387 denied_extensions: Sequence[str] | None,
388) -> None:
389 if denied_extensions is not None:
390 denied = frozenset(_normalize_ext(e) for e in denied_extensions)
391 else:
392 denied = _DEFAULT_DENIED_EXTENSIONS
394 if ext in denied:
395 raise SecurityError(
396 f"File extension '{ext}' is not allowed",
397 guard_name="file_extension",
398 value=original_name,
399 )
402def _check_allowed_extension(
403 ext: str,
404 original_name: str,
405 allowed_extensions: Sequence[str] | None,
406) -> None:
407 if allowed_extensions is not None:
408 allowed = {_normalize_ext(e) for e in allowed_extensions}
409 if ext not in allowed:
410 raise SecurityError(
411 f"File extension '{ext}' is not in allowed list",
412 guard_name="file_extension",
413 value=original_name,
414 )
417def guard_file_extension(
418 filename: str | Path,
419 *,
420 allowed_extensions: Sequence[str] | None = None,
421 denied_extensions: Sequence[str] | None = None,
422) -> Path:
423 """Validate file extension against allow/deny lists.
425 Args:
426 filename: The filename to check.
427 allowed_extensions: Extensions to allow (with or without dot).
428 denied_extensions: Extensions to deny (with or without dot).
430 Returns:
431 The filename as a Path.
433 Raises:
434 SecurityError: If extension is not allowed or is denied.
436 """
437 filename_str = str(filename)
438 if len(filename_str) > MAX_URL_LENGTH: # Reuse constant to avoid PLR2004
439 raise SecurityError(
440 f"Filename length exceeds maximum allowed limit of {MAX_URL_LENGTH}",
441 guard_name="file_extension",
442 value=filename_str[:80],
443 )
444 _check_filename_null_bytes(filename_str)
446 path = Path(filename)
447 clean_name = _clean_filename_end(path.name)
449 ext = "" if not clean_name else Path(clean_name).suffix.lower().lstrip(".")
451 _check_denied_extension(ext, str(path.name), denied_extensions)
452 _check_allowed_extension(ext, str(path.name), allowed_extensions)
454 return path
457def _check_env_denied(
458 name_upper: str,
459 name: str,
460 denied_names: Sequence[str] | None,
461) -> None:
462 """Check if the environment variable is in the denied list."""
463 if denied_names is not None:
464 denied = frozenset(n.upper() for n in denied_names)
465 else:
466 denied = _DEFAULT_DENIED_ENV_VARS
468 if name_upper in denied:
469 raise SecurityError(
470 f"Access to sensitive variable '{name}' is denied",
471 guard_name="env_variable",
472 value=name,
473 )
476def _is_env_allowed(name_upper: str, allowed_names: Sequence[str] | None) -> bool:
477 if allowed_names is None:
478 return False
479 allowed = {n.upper() for n in allowed_names}
480 return name_upper in allowed
483def _check_env_sensitive(
484 name_upper: str,
485 name: str,
486 allowed_names: Sequence[str] | None,
487) -> None:
488 """Check if the environment variable matches sensitive patterns."""
489 if not _SENSITIVE_ENV_VAR_PATTERN.search(name_upper):
490 return
492 if _is_env_allowed(name_upper, allowed_names):
493 return
495 raise SecurityError(
496 f"Access to potentially sensitive variable '{name}' is denied",
497 guard_name="env_variable",
498 value=name,
499 )
502def _check_env_var_type(name: object) -> str:
503 if not isinstance(name, str):
504 raise TypeError(f"Variable name must be str, got {type(name).__name__}")
505 return name
508def _check_env_var_length(name: str) -> None:
509 if len(name) > MAX_ENV_VAR_LENGTH:
510 raise SecurityError(
511 "Environment variable name exceeds maximum length",
512 guard_name="env_variable",
513 value=name[:80],
514 )
517def _check_env_var_content(name: str) -> None:
518 if not name or not name.strip():
519 raise SecurityError(
520 "Environment variable name cannot be empty or whitespace",
521 guard_name="env_variable",
522 )
524 if "\x00" in name:
525 raise SecurityError(
526 "Environment variable name cannot contain null bytes",
527 guard_name="env_variable",
528 )
531def _validate_env_var_name(name: object) -> str:
532 """Validate environment variable name."""
533 valid_name = _check_env_var_type(name)
534 _check_env_var_length(valid_name)
535 _check_env_var_content(valid_name)
537 return valid_name
540def guard_env_variable(
541 name: str,
542 *,
543 allowed_names: Sequence[str] | None = None,
544 denied_names: Sequence[str] | None = None,
545) -> str:
546 """Guard against accessing sensitive environment variables.
548 Args:
549 name: The environment variable name.
550 allowed_names: Variable names to allow.
551 denied_names: Variable names to deny.
553 Returns:
554 The environment variable value if safe.
556 Raises:
557 SecurityError: If variable access is not allowed.
559 """
560 # Validate input type and format
561 name = _validate_env_var_name(name)
563 name_upper = name.upper()
565 _check_env_denied(name_upper, name, denied_names)
566 _check_env_sensitive(name_upper, name, allowed_names)
568 # Get the variable
569 value = os.environ.get(name)
570 if value is None:
571 raise SecurityError(
572 f"Environment variable '{name}' is not set",
573 guard_name="env_variable",
574 value=name,
575 )
577 return value
580# ── SSRF Private-Range Constants ─────────────────────────────────────────────
581_ALLOWED_SSRF_SCHEMES: frozenset[str] = frozenset({"http", "https"})
584def _check_ssrf_url_length(url: str) -> Result[str, SecurityError]:
585 if not isinstance(url, str):
586 return Err( # type: ignore[unreachable]
587 SecurityError(
588 f"URL must be str, got {type(url).__name__}",
589 guard_name="ssrf",
590 ),
591 )
593 if not url:
594 return Err(SecurityError("URL cannot be empty", guard_name="ssrf"))
596 if len(url) > MAX_URL_LENGTH:
597 return Err(
598 SecurityError(
599 "URL length exceeds maximum allowed limit",
600 guard_name="ssrf",
601 value=url[:80],
602 ),
603 )
604 return Ok(url)
607def _is_invalid_url_char(c: str) -> bool:
608 return c <= "\x20" or c in {"\x7f", "\x00"} or not c.isprintable()
611def _has_invalid_url_chars(url: str) -> bool:
612 return any(_is_invalid_url_char(c) for c in url)
615def _fully_unquote_url(url: str) -> Result[str, SecurityError]:
616 current = url
617 for _ in range(10): # limit to prevent DoS
618 unquoted = unquote(current)
619 if unquoted == current:
620 return Ok(current)
621 current = unquoted
622 return Err(
623 SecurityError(
624 "URL exceeds maximum nested encoding limit",
625 guard_name="ssrf",
626 ),
627 )
630def _check_ssrf_url_characters(url: str) -> Result[str, SecurityError]:
631 if _has_invalid_url_chars(url):
632 return Err(
633 SecurityError(
634 "URL contains invalid characters",
635 guard_name="ssrf",
636 value=url[:80],
637 ),
638 )
640 unquoted_res = _fully_unquote_url(url)
641 if not isinstance(unquoted_res, Ok):
642 return unquoted_res
644 if _has_invalid_url_chars(unquoted_res.ok_value):
645 return Err(
646 SecurityError(
647 "URL contains invalid characters",
648 guard_name="ssrf",
649 value=url[:80],
650 ),
651 )
652 return Ok(url)
655def _validate_ssrf_url_type_and_length(url: str) -> Result[str, SecurityError]:
656 length_res = _check_ssrf_url_length(url)
657 if not isinstance(length_res, Ok):
658 return length_res
660 return _check_ssrf_url_characters(url)
663def _create_ssrf_error(msg: str, url: str) -> Err[SecurityError]:
664 return Err(
665 SecurityError(
666 msg,
667 guard_name="ssrf",
668 value=url[:80],
669 ),
670 )
673def _is_scheme_allowed(scheme: str, allowed_schemes: frozenset[str]) -> bool:
674 return bool(scheme and scheme.lower() in allowed_schemes)
677def _has_obfuscated_hostname(netloc: str, hostname: str) -> bool:
678 return "\\" in netloc or "@" in hostname
681def _validate_parsed_ssrf_url(
682 url: str,
683 parsed: SplitResult,
684 allowed_schemes: frozenset[str],
685) -> Result[str, SecurityError]:
686 if not _is_scheme_allowed(parsed.scheme, allowed_schemes):
687 return _create_ssrf_error(f"URL scheme '{parsed.scheme}' is not allowed", url)
689 if not parsed.hostname:
690 return _create_ssrf_error("URL has no resolvable hostname", url)
692 if _has_obfuscated_hostname(parsed.netloc, parsed.hostname):
693 return _create_ssrf_error(
694 "URL contains obfuscated characters in the hostname", url
695 )
697 return Ok(parsed.hostname)
700def _validate_ssrf_url_parse(
701 url: str,
702 allowed_schemes: frozenset[str],
703) -> Result[str, SecurityError]:
704 try:
705 parsed = urlsplit(url)
706 except ValueError as exc:
707 return _create_ssrf_error(f"Malformed URL: {exc}", url)
709 return _validate_parsed_ssrf_url(url, parsed, allowed_schemes)
712def _validate_ssrf_url(
713 url: str,
714 allowed_schemes: frozenset[str],
715) -> Result[str, SecurityError]:
716 """Validate the URL format, scheme, and presence of hostname."""
717 type_len_res = _validate_ssrf_url_type_and_length(url)
718 if not isinstance(type_len_res, Ok):
719 return type_len_res
721 return _validate_ssrf_url_parse(url, allowed_schemes)
724def _is_ip_address_unsafe_bounds(
725 addr: ipaddress.IPv4Address | ipaddress.IPv6Address,
726) -> bool:
727 return addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved
730def _is_ip_address_safe(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
731 """Evaluate if an ipaddress object represents a safe, public IP."""
732 if _is_ip_address_unsafe_bounds(addr):
733 return False
734 return not (
735 getattr(addr, "is_multicast", False) or getattr(addr, "is_unspecified", False) # type: ignore[misc]
736 )
739@functools.lru_cache(maxsize=1024) # type: ignore[misc]
740def _is_ip_safe(raw_ip: str) -> bool:
741 """Check if a single IP address is safe (not private/loopback/reserved)."""
742 try:
743 addr = ipaddress.ip_address(raw_ip)
744 except ValueError:
745 return True
747 return _is_ip_address_safe(addr)
750def _check_ip_safety(hostname: str) -> Result[None, SecurityError]:
751 """Resolve hostname to IP addresses and check for SSRF risk."""
752 try:
753 addr_infos = socket.getaddrinfo(hostname, None)
754 except (socket.gaierror, UnicodeError):
755 return Err(
756 SecurityError(
757 "Hostname could not be resolved or contains invalid characters",
758 guard_name="ssrf",
759 ),
760 )
762 for addr_info in addr_infos:
763 raw_ip = addr_info[4][0]
764 if not _is_ip_safe(raw_ip):
765 return Err(
766 SecurityError(
767 "SSRF detected: hostname resolves to private/reserved address",
768 guard_name="ssrf",
769 ),
770 )
772 return Ok(None)
775def guard_ssrf(
776 url: str,
777 *,
778 allowed_schemes: frozenset[str] = _ALLOWED_SSRF_SCHEMES,
779) -> Result[str, SecurityError]:
780 """Validate a URL against Server-Side Request Forgery (SSRF) attacks.
782 Parse the URL, resolve its hostname via DNS, and reject it when the
783 resulting IP address falls inside a private, loopback, link-local, or
784 otherwise reserved network range.
786 Args:
787 url: The URL string to validate.
788 allowed_schemes: Set of URL schemes considered safe.
789 Defaults to ``{"http", "https"}``.
791 Returns:
792 ``Ok(url)`` when the URL is safe to fetch.
793 ``Err(SecurityError)`` when an SSRF risk is detected.
795 Raises:
796 TypeError: If *url* is not a :class:`str`.
798 Example:
799 >>> guard_ssrf("https://example.com")
800 Ok('https://example.com')
801 >>> guard_ssrf("http://169.254.169.254/metadata")
802 Err(SecurityError('[ssrf] ...))
804 """
805 # 1. Validate format and scheme
806 val_res = _validate_ssrf_url(url, allowed_schemes)
807 if not isinstance(val_res, Ok):
808 return val_res
810 # 2. Check IP safety
811 ip_res = _check_ip_safety(val_res.ok_value)
812 if not isinstance(ip_res, Ok):
813 return ip_res
815 return Ok(url)