Coverage for src/taipanstack/security/guards.py: 100%
250 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 15:01 +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 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)
96_SAFE_HASH_ALGORITHMS = frozenset(
97 [
98 "sha256",
99 "sha384",
100 "sha512",
101 "sha3_256",
102 "sha3_384",
103 "sha3_512",
104 "blake2b",
105 "blake2s",
106 ],
107)
110class SecurityError(Exception):
111 """Raised when a security guard detects a violation.
113 Attributes:
114 guard_name: Name of the guard that was triggered.
115 message: Description of the violation.
116 value: The offending value (if safe to log).
118 """
120 def __init__(
121 self,
122 message: str,
123 guard_name: str = "unknown",
124 value: str | None = None,
125 ) -> None:
126 """Initialize SecurityError.
128 Args:
129 message: Description of the violation.
130 guard_name: Name of the guard that triggered.
131 value: The offending value (sanitized).
133 """
134 self.guard_name = guard_name
135 self.value = value
136 super().__init__(f"[{guard_name}] {message}")
139def _check_traversal_patterns(path_str: str) -> None:
140 """Check for explicit traversal patterns before resolution."""
141 match = TRAVERSAL_REGEX.search(path_str.lower())
142 if match:
143 raise SecurityError(
144 f"Path traversal pattern detected: {match.group(0)}",
145 guard_name="path_traversal",
146 value=path_str[:50], # Truncate for safety
147 )
150def _resolve_and_check_bounds(path: Path, base_dir: Path) -> tuple[Path, Path]:
151 """Resolve the path and check if it is within base_dir."""
152 try:
153 full_path = path if path.is_absolute() else (base_dir / path)
154 resolved = full_path.resolve()
155 except (OSError, ValueError, RuntimeError) as e:
156 raise SecurityError(
157 f"Invalid path: {e}",
158 guard_name="path_traversal",
159 ) from e
161 if not resolved.is_relative_to(base_dir):
162 raise SecurityError(
163 "Path escapes base directory",
164 guard_name="path_traversal",
165 )
166 return full_path, resolved
169def _check_symlink_safety(full_path: Path, base_dir: Path) -> None:
170 """Check for symlinks recursively up to the base directory."""
171 current = full_path
172 # Only check components from the user-provided path, not the base_dir
173 while current not in (base_dir, current.parent):
174 # We don't check .exists() because it returns False for broken symlinks
175 try:
176 is_symlink = current.is_symlink()
177 except OSError as e:
178 raise SecurityError(
179 f"Invalid path encountered during symlink check: {e}",
180 guard_name="path_traversal",
181 value=str(current)[:50],
182 ) from e
183 if is_symlink:
184 raise SecurityError(
185 "Symlinks are not allowed",
186 guard_name="path_traversal",
187 value=str(current),
188 )
189 current = current.parent
192def _check_path_types(path: object, base_dir: object) -> None:
193 """Validate types of path and base_dir."""
194 if not isinstance(path, (str, Path)):
195 raise TypeError(f"path must be str or Path, got {type(path).__name__}")
196 if base_dir is not None and not isinstance(base_dir, (str, Path)):
197 raise TypeError(f"base_dir must be str or Path, got {type(base_dir).__name__}")
200def _check_path_lengths(path: object, base_dir: object) -> None:
201 """Validate lengths of path and base_dir."""
202 if len(str(path)) > MAX_PATH_LENGTH:
203 raise SecurityError(
204 f"Path length exceeds maximum allowed limit of {MAX_PATH_LENGTH}",
205 guard_name="path_traversal",
206 )
208 if base_dir is not None and len(str(base_dir)) > MAX_PATH_LENGTH:
209 raise SecurityError(
210 f"Base directory length exceeds maximum allowed limit of {MAX_PATH_LENGTH}",
211 guard_name="path_traversal",
212 )
215def _validate_path_types(path: object, base_dir: object) -> None:
216 """Validate types and lengths of path and base_dir."""
217 _check_path_types(path, base_dir)
218 _check_path_lengths(path, base_dir)
221def _check_path_null_bytes(path: Path | str, base_dir: Path | str | None) -> None:
222 """Check for null bytes in path and base_dir."""
223 if "\x00" in str(path) or (base_dir is not None and "\x00" in str(base_dir)):
224 raise SecurityError(
225 "Path contains null bytes",
226 guard_name="path_traversal",
227 )
230def _resolve_base_dir(base_dir: Path | str | None) -> Path:
231 """Resolve the base directory."""
232 return Path(base_dir).resolve() if base_dir else Path.cwd().resolve()
235def guard_path_traversal(
236 path: Path | str,
237 base_dir: Path | str | None = None,
238 *,
239 allow_symlinks: bool = False,
240) -> Path:
241 """Prevent path traversal attacks.
243 Ensures that the given path does not escape the base directory
244 using techniques like '..' or symlinks.
246 Args:
247 path: The path to validate.
248 base_dir: The base directory to constrain to. Defaults to cwd.
249 allow_symlinks: Whether to allow symlinks (default: False).
251 Returns:
252 The resolved, validated path.
254 Raises:
255 SecurityError: If path traversal is detected.
257 Example:
258 >>> guard_path_traversal("../etc/passwd", Path("/app"))
259 SecurityError: [path_traversal] Path escapes base directory
261 """
262 _validate_path_types(path, base_dir)
263 _check_path_null_bytes(path, base_dir)
265 path_obj = Path(path) if isinstance(path, str) else path
266 base = _resolve_base_dir(base_dir)
268 _check_traversal_patterns(str(path_obj))
269 full_path, resolved = _resolve_and_check_bounds(path_obj, base)
271 if not allow_symlinks:
272 _check_symlink_safety(full_path, base)
274 return resolved
277def _check_command_not_empty(command: Sequence[str]) -> None:
278 if not command:
279 raise SecurityError(
280 "Empty command is not allowed",
281 guard_name="command_injection",
282 )
285def _check_command_null_bytes(cmd_list: list[str]) -> None:
286 for arg in cmd_list:
287 if isinstance(arg, str) and "\x00" in arg:
288 raise SecurityError(
289 "Dangerous shell character detected: null byte",
290 guard_name="command_injection",
291 value=arg[:50],
292 )
295def _check_command_patterns(cmd_list: list[str]) -> None:
296 for i, arg in enumerate(cmd_list):
297 if not isinstance(arg, str):
298 raise TypeError(
299 f"All command arguments must be strings, "
300 f"got {type(arg).__name__} at index {i}",
301 )
303 match = _DANGEROUS_COMMAND_RE.search(arg)
304 if match:
305 description = _DANGEROUS_COMMAND_LOOKUP[match.group(0)]
306 raise SecurityError(
307 f"Dangerous shell character detected: {description}",
308 guard_name="command_injection",
309 value=arg[:50],
310 )
313def _check_allowed_commands(
314 cmd_list: list[str],
315 allowed_commands: Sequence[str] | None,
316) -> None:
317 if allowed_commands is None:
318 return
320 base_command = cmd_list[0]
321 command_name = Path(base_command).name
322 cmd_not_allowed = (
323 command_name not in allowed_commands and base_command not in allowed_commands
324 )
325 if cmd_not_allowed:
326 raise SecurityError(
327 f"Command not in allowed list: {command_name}",
328 guard_name="command_injection",
329 value=command_name,
330 )
333def guard_command_injection(
334 command: Sequence[str],
335 *,
336 allowed_commands: Sequence[str] | None = None,
337) -> list[str]:
338 """Prevent command injection attacks.
340 Validates that command arguments don't contain shell metacharacters
341 that could lead to command injection.
343 Args:
344 command: The command and arguments as a sequence.
345 allowed_commands: Optional whitelist of allowed base commands.
347 Returns:
348 The validated command as a list.
350 Raises:
351 SecurityError: If command injection is detected.
353 Example:
354 >>> guard_command_injection(["echo", "hello; rm -rf /"])
355 SecurityError: [command_injection] Dangerous characters detected
357 """
358 cmd_list = list(command)
360 _check_command_not_empty(cmd_list)
362 _check_command_null_bytes(cmd_list)
363 _check_command_patterns(cmd_list)
364 _check_allowed_commands(cmd_list, allowed_commands)
366 return cmd_list
369def _check_filename_null_bytes(filename_str: str) -> None:
370 if "\x00" in filename_str:
371 raise SecurityError(
372 "Filename contains null bytes",
373 guard_name="file_extension",
374 value=filename_str,
375 )
378def _clean_filename_end(clean_name: str) -> str:
379 end_idx = len(clean_name)
380 while end_idx > 0:
381 char = clean_name[end_idx - 1]
382 if (
383 char == "."
384 or unicodedata.category(char).startswith(("Z", "C"))
385 or char == "\xad"
386 ):
387 end_idx -= 1
388 else:
389 break
390 return clean_name[:end_idx]
393def _normalize_ext(e: str) -> str:
394 return e.lower().lstrip(".")
397def _check_denied_extension(
398 ext: str,
399 original_name: str,
400 denied_extensions: Sequence[str] | None,
401) -> None:
402 if denied_extensions is not None:
403 denied = frozenset(_normalize_ext(e) for e in denied_extensions)
404 else:
405 denied = _DEFAULT_DENIED_EXTENSIONS
407 if ext in denied:
408 raise SecurityError(
409 f"File extension '{ext}' is not allowed",
410 guard_name="file_extension",
411 value=original_name,
412 )
415def _check_allowed_extension(
416 ext: str,
417 original_name: str,
418 allowed_extensions: Sequence[str] | None,
419) -> None:
420 if allowed_extensions is not None:
421 allowed = {_normalize_ext(e) for e in allowed_extensions}
422 if ext not in allowed:
423 raise SecurityError(
424 f"File extension '{ext}' is not in allowed list",
425 guard_name="file_extension",
426 value=original_name,
427 )
430def guard_file_extension(
431 filename: str | Path,
432 *,
433 allowed_extensions: Sequence[str] | None = None,
434 denied_extensions: Sequence[str] | None = None,
435) -> Path:
436 """Validate file extension against allow/deny lists.
438 Args:
439 filename: The filename to check.
440 allowed_extensions: Extensions to allow (with or without dot).
441 denied_extensions: Extensions to deny (with or without dot).
443 Returns:
444 The filename as a Path.
446 Raises:
447 SecurityError: If extension is not allowed or is denied.
449 """
450 filename_str = str(filename)
451 if len(filename_str) > MAX_URL_LENGTH: # Reuse constant to avoid PLR2004
452 raise SecurityError(
453 f"Filename length exceeds maximum allowed limit of {MAX_URL_LENGTH}",
454 guard_name="file_extension",
455 value=filename_str[:80],
456 )
457 _check_filename_null_bytes(filename_str)
459 path = Path(filename)
460 clean_name = _clean_filename_end(path.name)
462 ext = "" if not clean_name else Path(clean_name).suffix.lower().lstrip(".")
464 _check_denied_extension(ext, str(path.name), denied_extensions)
465 _check_allowed_extension(ext, str(path.name), allowed_extensions)
467 return path
470def _check_env_denied(
471 name_upper: str,
472 name: str,
473 denied_names: Sequence[str] | None,
474) -> None:
475 """Check if the environment variable is in the denied list."""
476 if denied_names is not None:
477 denied = frozenset(n.upper() for n in denied_names)
478 else:
479 denied = _DEFAULT_DENIED_ENV_VARS
481 if name_upper in denied:
482 raise SecurityError(
483 f"Access to sensitive variable '{name}' is denied",
484 guard_name="env_variable",
485 value=name,
486 )
489def _check_env_sensitive(
490 name_upper: str,
491 name: str,
492 allowed_names: Sequence[str] | None,
493) -> None:
494 """Check if the environment variable matches sensitive patterns."""
495 if not _SENSITIVE_ENV_VAR_PATTERN.search(name_upper):
496 return
498 # Only block if not explicitly allowed
499 if allowed_names is not None:
500 allowed = {n.upper() for n in allowed_names}
501 if name_upper in allowed:
502 return
504 raise SecurityError(
505 f"Access to potentially sensitive variable '{name}' is denied",
506 guard_name="env_variable",
507 value=name,
508 )
511def _check_env_var_type(name: object) -> str:
512 if not isinstance(name, str):
513 raise TypeError(f"Variable name must be str, got {type(name).__name__}")
514 return name
517def _check_env_var_length(name: str) -> None:
518 if len(name) > MAX_ENV_VAR_LENGTH:
519 raise SecurityError(
520 "Environment variable name exceeds maximum length",
521 guard_name="env_variable",
522 value=name[:80],
523 )
526def _check_env_var_content(name: str) -> None:
527 if not name or not name.strip():
528 raise SecurityError(
529 "Environment variable name cannot be empty or whitespace",
530 guard_name="env_variable",
531 )
533 if "\x00" in name:
534 raise SecurityError(
535 "Environment variable name cannot contain null bytes",
536 guard_name="env_variable",
537 )
540def _validate_env_var_name(name: object) -> str:
541 """Validate environment variable name."""
542 valid_name = _check_env_var_type(name)
543 _check_env_var_length(valid_name)
544 _check_env_var_content(valid_name)
546 return valid_name
549def guard_env_variable(
550 name: str,
551 *,
552 allowed_names: Sequence[str] | None = None,
553 denied_names: Sequence[str] | None = None,
554) -> str:
555 """Guard against accessing sensitive environment variables.
557 Args:
558 name: The environment variable name.
559 allowed_names: Variable names to allow.
560 denied_names: Variable names to deny.
562 Returns:
563 The environment variable value if safe.
565 Raises:
566 SecurityError: If variable access is not allowed.
568 """
569 # Validate input type and format
570 name = _validate_env_var_name(name)
572 name_upper = name.upper()
574 _check_env_denied(name_upper, name, denied_names)
575 _check_env_sensitive(name_upper, name, allowed_names)
577 # Get the variable
578 value = os.environ.get(name)
579 if value is None:
580 raise SecurityError(
581 f"Environment variable '{name}' is not set",
582 guard_name="env_variable",
583 value=name,
584 )
586 return value
589# ── SSRF Private-Range Constants ─────────────────────────────────────────────
590_ALLOWED_SSRF_SCHEMES: frozenset[str] = frozenset({"http", "https"})
593def _check_ssrf_url_length(url: str) -> Result[str, SecurityError]:
594 if not isinstance(url, str):
595 return Err( # type: ignore[unreachable]
596 SecurityError(
597 f"URL must be str, got {type(url).__name__}",
598 guard_name="ssrf",
599 ),
600 )
602 if not url:
603 return Err(SecurityError("URL cannot be empty", guard_name="ssrf"))
605 if len(url) > MAX_URL_LENGTH:
606 return Err(
607 SecurityError(
608 "URL length exceeds maximum allowed limit",
609 guard_name="ssrf",
610 value=url[:80],
611 ),
612 )
613 return Ok(url)
616def _has_invalid_url_chars(url: str) -> bool:
617 if any(c <= "\x20" or c == "\x7f" for c in url):
618 return True
619 return bool("\x00" in url or not url.isprintable())
622def _check_ssrf_url_characters(url: str) -> Result[str, SecurityError]:
623 if _has_invalid_url_chars(url) or _has_invalid_url_chars(unquote(url)):
624 return Err(
625 SecurityError(
626 "URL contains invalid characters",
627 guard_name="ssrf",
628 value=url[:80],
629 ),
630 )
631 return Ok(url)
634def _validate_ssrf_url_type_and_length(url: str) -> Result[str, SecurityError]:
635 length_res = _check_ssrf_url_length(url)
636 if not isinstance(length_res, Ok):
637 return length_res
639 return _check_ssrf_url_characters(url)
642def _validate_ssrf_url_parse(
643 url: str,
644 allowed_schemes: frozenset[str],
645) -> Result[str, SecurityError]:
646 try:
647 parsed = urlsplit(url)
648 except ValueError as exc:
649 return Err(
650 SecurityError(
651 f"Malformed URL: {exc}",
652 guard_name="ssrf",
653 value=url[:80],
654 ),
655 )
657 if not parsed.scheme or parsed.scheme.lower() not in allowed_schemes:
658 return Err(
659 SecurityError(
660 f"URL scheme '{parsed.scheme}' is not allowed",
661 guard_name="ssrf",
662 value=url[:80],
663 ),
664 )
666 hostname = parsed.hostname
667 if not hostname:
668 return Err(
669 SecurityError(
670 "URL has no resolvable hostname",
671 guard_name="ssrf",
672 value=url[:80],
673 ),
674 )
676 return Ok(hostname)
679def _validate_ssrf_url(
680 url: str,
681 allowed_schemes: frozenset[str],
682) -> Result[str, SecurityError]:
683 """Validate the URL format, scheme, and presence of hostname."""
684 type_len_res = _validate_ssrf_url_type_and_length(url)
685 if not isinstance(type_len_res, Ok):
686 return type_len_res
688 return _validate_ssrf_url_parse(url, allowed_schemes)
691def _is_ip_address_unsafe_bounds(
692 addr: ipaddress.IPv4Address | ipaddress.IPv6Address,
693) -> bool:
694 return addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved
697def _is_ip_address_safe(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
698 """Evaluate if an ipaddress object represents a safe, public IP."""
699 if _is_ip_address_unsafe_bounds(addr):
700 return False
701 return not (
702 getattr(addr, "is_multicast", False) or getattr(addr, "is_unspecified", False)
703 )
706@functools.lru_cache(maxsize=1024)
707def _is_ip_safe(raw_ip: str) -> bool:
708 """Check if a single IP address is safe (not private/loopback/reserved)."""
709 try:
710 addr = ipaddress.ip_address(raw_ip)
711 except ValueError:
712 return True
714 return _is_ip_address_safe(addr)
717def _check_ip_safety(hostname: str) -> Result[None, SecurityError]:
718 """Resolve hostname to IP addresses and check for SSRF risk."""
719 try:
720 addr_infos = socket.getaddrinfo(hostname, None)
721 except (socket.gaierror, UnicodeError):
722 return Err(
723 SecurityError(
724 "Hostname could not be resolved or contains invalid characters",
725 guard_name="ssrf",
726 ),
727 )
729 for addr_info in addr_infos:
730 raw_ip = addr_info[4][0]
731 if not _is_ip_safe(raw_ip):
732 return Err(
733 SecurityError(
734 "SSRF detected: hostname resolves to private/reserved address",
735 guard_name="ssrf",
736 ),
737 )
739 return Ok(None)
742def guard_ssrf(
743 url: str,
744 *,
745 allowed_schemes: frozenset[str] = _ALLOWED_SSRF_SCHEMES,
746) -> Result[str, SecurityError]:
747 """Validate a URL against Server-Side Request Forgery (SSRF) attacks.
749 Parse the URL, resolve its hostname via DNS, and reject it when the
750 resulting IP address falls inside a private, loopback, link-local, or
751 otherwise reserved network range.
753 Args:
754 url: The URL string to validate.
755 allowed_schemes: Set of URL schemes considered safe.
756 Defaults to ``{"http", "https"}``.
758 Returns:
759 ``Ok(url)`` when the URL is safe to fetch.
760 ``Err(SecurityError)`` when an SSRF risk is detected.
762 Raises:
763 TypeError: If *url* is not a :class:`str`.
765 Example:
766 >>> guard_ssrf("https://example.com")
767 Ok('https://example.com')
768 >>> guard_ssrf("http://169.254.169.254/metadata")
769 Err(SecurityError('[ssrf] ...))
771 """
772 # 1. Validate format and scheme
773 val_res = _validate_ssrf_url(url, allowed_schemes)
774 if not isinstance(val_res, Ok):
775 return val_res
777 # 2. Check IP safety
778 ip_res = _check_ip_safety(val_res.ok_value)
779 if not isinstance(ip_res, Ok):
780 return ip_res
782 return Ok(url)