diff --git a/studio/backend/core/inference/bypass_site/sitecustomize.py b/studio/backend/core/inference/bypass_site/sitecustomize.py new file mode 100644 index 0000000000..36ff0d197a --- /dev/null +++ b/studio/backend/core/inference/bypass_site/sitecustomize.py @@ -0,0 +1,8 @@ +"""Load path remapping without sandbox network guards.""" + +import runpy +from pathlib import Path + + +_SHIM = Path(__file__).resolve().parents[1] / "sandbox_site" / "sitecustomize.py" +runpy.run_path(str(_SHIM)) diff --git a/studio/backend/core/inference/sandbox_site/sitecustomize.py b/studio/backend/core/inference/sandbox_site/sitecustomize.py index 244fa95145..511e4e8cbf 100644 --- a/studio/backend/core/inference/sandbox_site/sitecustomize.py +++ b/studio/backend/core/inference/sandbox_site/sitecustomize.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 -"""Sandbox-side compatibility shim for ChatGPT code-interpreter paths. +"""Sandbox-side compatibility shim for code-interpreter path conventions. Models habitually write to /mnt/data (or /mnt/outputs, /home/sandbox, /workspace), none of which exist in the Unsloth sandbox. This module sits on the @@ -28,10 +28,14 @@ Identical with and without output streaming because the child env is. """ import builtins +import importlib +import importlib.machinery +import importlib.util import io import json import os import sys +import types # Code-interpreter convention prefixes. Remapping is gated on the prefix being # ABSENT (see _remap) so a genuine host mount / user dir is never shadowed. @@ -46,6 +50,576 @@ _remapped_writes: dict = {} # on-disk sidecar carries the map across runs. It records only sources the # fallback healed, so an unrelated same-basename file is never adopted. _REMAP_SIDECAR = ".unsloth_sandbox_remap.json" +_BLOCKED_NETWORK_MODULES = frozenset({"boto3", "botocore"}) +# httpx imports httpcore internally, so block only sandbox-user requests. +_DIRECT_BLOCKED_NETWORK_MODULES = frozenset({"httpcore"}) +_import_guard_installed = False +_original_import = builtins.__import__ +_original_import_module = importlib.import_module + + +def _initial_trusted_library_roots(): + """Capture interpreter-managed package roots before sandbox code can edit sys.path.""" + roots = [] + for entry in sys.path: + if not isinstance(entry, str) or not entry: + continue + try: + path = os.path.realpath(entry) + except OSError: + continue + if os.path.basename(path).lower() not in {"site-packages", "dist-packages"}: + continue + if path not in roots: + roots.append(path) + return tuple(roots) + + +_TRUSTED_LIBRARY_ROOTS = _initial_trusted_library_roots() + + +def _path_is_in_roots(filename, roots): + if not isinstance(filename, str) or filename.startswith("<"): + return False + try: + path = os.path.realpath(filename) + return any(os.path.commonpath((root, path)) == root for root in roots) + except (OSError, ValueError): + return False + + +def _blocked_network_module_origin(filename): + if not isinstance(filename, str) or filename.startswith("<"): + return None + try: + path = os.path.realpath(filename) + for root in _TRUSTED_LIBRARY_ROOTS: + if os.path.commonpath((root, path)) != root: + continue + relative = os.path.relpath(path, root).replace("\\", "/") + package = relative.split("/", 1)[0].removesuffix(".py") + if package in _BLOCKED_NETWORK_MODULES or package in _DIRECT_BLOCKED_NETWORK_MODULES: + return package + except (OSError, ValueError): + return None + return None + + +def _frame_uses_trusted_package(frame, package): + module_name = frame.f_globals.get("__name__", "") + if not isinstance(module_name, str): + return False + if module_name != package and not module_name.startswith(f"{package}."): + return False + module = sys.modules.get(module_name) + if module is None: + return False + try: + module_dict = types.ModuleType.__getattribute__(module, "__dict__") + except TypeError: + module_dict = getattr(module, "__dict__", None) + if module_dict is not frame.f_globals: + return False + spec = getattr(module, "__spec__", None) + origin = getattr(spec, "origin", None) or getattr(module, "__file__", None) + if not _path_is_in_roots(origin, _TRUSTED_LIBRARY_ROOTS): + return False + return _path_is_in_roots(frame.f_code.co_filename, _TRUSTED_LIBRARY_ROOTS) + + +def _trusted_http_client_frame(frame): + return _frame_uses_trusted_package(frame, "httpx") or _frame_uses_trusted_package( + frame, "httpcore" + ) + + +def _trusted_httpx_in_call_stack(skip = 1): + try: + frame = sys._getframe(skip) + except ValueError: + return False + while frame is not None: + if _frame_uses_trusted_package(frame, "httpx"): + return True + frame = frame.f_back + return False + + +def _frame_is_importlib(frame): + module_name = frame.f_globals.get("__name__", "") + if not isinstance(module_name, str) or ( + module_name != "importlib" and not module_name.startswith("importlib.") + ): + return False + module = sys.modules.get(module_name) + return module is not None and getattr(module, "__dict__", None) is frame.f_globals + + +def _sandbox_code_requested_import(skip = 1): + try: + frame = sys._getframe(skip) + except ValueError: + return True + while frame is not None: + if ( + frame.f_code.co_filename == __file__ + or _frame_is_importlib(frame) + or str(frame.f_code.co_filename).startswith(" bool: _FIND_EXEC_FLAGS = frozenset({"-exec", "-execdir", "-ok", "-okdir"}) +_SANDBOX_PYTHON_ENV_VARS = frozenset( + {"PATH", "PYTHONHOME", "PYTHONPATH", "UNSLOTH_STUDIO_SANDBOXED"} +) +_PYTHON_LAUNCH_WRAPPERS = _COMMAND_PREFIXES | frozenset({"conda", "hatch", "pipx", "poetry", "uv"}) +_SHELL_SCRIPT_INTERPRETERS = frozenset( + {"bash", "cmd", "cmd.exe", "dash", "fish", "ksh", "sh", "zsh"} +) + + +def _python_executable_token(token: str) -> bool: + base = os.path.basename(token.replace("\\", "/")).lower() + if base.endswith(".exe"): + base = base[:-4] + return bool(re.fullmatch(r"(?:python(?:w)?(?:\d+(?:\.\d+)*)?|py)", base)) + + +def _shell_command_segments(command: str) -> list[list[str]]: + try: + lexer = shlex.shlex(command, posix = sys.platform != "win32", punctuation_chars = ";&|()`") + lexer.whitespace_split = True + tokens = list(lexer) + except ValueError: + tokens = command.split() + segments: list[list[str]] = [[]] + expect_command = True + for token in tokens: + stripped = token.strip("\"'") + if token in _SHELL_SEPARATORS or ( + expect_command and stripped.lower() in _SHELL_KEYWORDS_AS_SEP + ): + if segments[-1]: + segments.append([]) + expect_command = True + continue + segments[-1].append(stripped) + expect_command = False + return [segment for segment in segments if segment] + + +def _shell_command_with_unquoted_newlines_as_separators(command: str) -> str: + if "\n" not in command and "\r" not in command: + return command + out: list[str] = [] + quote: str | None = None + escaped = False + index = 0 + while index < len(command): + char = command[index] + if escaped: + out.append(char) + escaped = False + index += 1 + continue + if char == "\\": + out.append(char) + escaped = True + index += 1 + continue + if char == "'" and quote is None: + quote = "'" + elif char == "'" and quote == "'": + quote = None + elif char == '"' and quote is None: + quote = '"' + elif char == '"' and quote == '"': + quote = None + if char in "\r\n" and quote is None: + out.append(" ; ") + if char == "\r" and index + 1 < len(command) and command[index + 1] == "\n": + index += 1 + else: + out.append(char) + index += 1 + return "".join(out) + + +def _shell_command_without_line_continuations(command: str) -> str: + if "\\\n" not in command and "\\\r" not in command: + return command + out: list[str] = [] + quote: str | None = None + escaped = False + index = 0 + while index < len(command): + char = command[index] + if escaped: + if char in "\r\n" and quote != "'": + escaped = False + if char == "\r" and index + 1 < len(command) and command[index + 1] == "\n": + index += 1 + index += 1 + continue + out.append("\\") + out.append(char) + escaped = False + index += 1 + continue + if char == "\\": + escaped = True + index += 1 + continue + if char == "'" and quote is None: + quote = "'" + elif char == "'" and quote == "'": + quote = None + elif char == '"' and quote is None: + quote = '"' + elif char == '"' and quote == '"': + quote = None + out.append(char) + index += 1 + if escaped: + out.append("\\") + return "".join(out) + + +def _segment_python_index(segment: list[str]) -> int | None: + command_index = 0 + while command_index < len(segment) and _ASSIGNMENT_RE.match(segment[command_index]): + command_index += 1 + if command_index >= len(segment): + return None + command = os.path.basename(segment[command_index].replace("\\", "/")).lower() + if _python_executable_token(segment[command_index]): + return command_index + if command not in _PYTHON_LAUNCH_WRAPPERS: + if command not in {"find", "fd"}: + return None + for index in range(command_index + 1, len(segment)): + if _python_executable_token(segment[index]) and any( + token in _FIND_EXEC_FLAGS for token in segment[command_index:index] + ): + return index + return None + for index in range(command_index + 1, len(segment)): + if _python_executable_token(segment[index]): + return index + return None + + +def _segment_shell_index(segment: list[str]) -> int | None: + if not segment: + return None + first = os.path.basename(segment[0].replace("\\", "/")).lower() + wrapper_context = first in _PYTHON_LAUNCH_WRAPPERS + find_exec = first in {"find", "fd"} + for index, token in enumerate(segment): + shell = os.path.basename(token.replace("\\", "/")).lower() + find_exec_context = find_exec and any( + token in _FIND_EXEC_FLAGS for token in segment[:index] + ) + if shell in _SHELL_SCRIPT_INTERPRETERS and ( + index == 0 or wrapper_context or find_exec_context + ): + return index + return None + + +def _segment_shell_reads_stdin(segment: list[str], shell_index: int) -> bool: + saw_stdin_flag = False + for token in segment[shell_index + 1 :]: + lowered = token.lower() + if lowered == "/c" or (lowered.startswith("-") and lowered.endswith("c")): + return False + if lowered in {"-s", "--stdin"} or ( + lowered.startswith("-") and not lowered.startswith("--") and "s" in lowered[1:] + ): + saw_stdin_flag = True + continue + if lowered.startswith("-"): + continue + if saw_stdin_flag: + continue + return False + return True + + +def _python_flags_skip_sitecustomize(arguments: list[str]) -> bool: + skip_next = False + for argument in arguments: + if skip_next: + skip_next = False + continue + if argument in {"-c", "-m"} or not argument.startswith("-"): + break + if argument == "--": + break + if argument in {"--ignore-environment", "--isolated", "--no-site"}: + return True + if re.fullmatch(r"-[^-]*[SEI][^-]*", argument): + return True + if argument in {"-W", "-X", "--check-hash-based-pycs"}: + skip_next = True + return False + + +def _segment_mutates_sandbox_python_env(segment: list[str], python_index: int) -> bool: + before_python = segment[:python_index] + for index, token in enumerate(before_python): + assignment = _ASSIGNMENT_RE.match(token) + if assignment and token.split("=", 1)[0].upper() in _SANDBOX_PYTHON_ENV_VARS: + return True + lowered = token.lower() + # GNU env treats a lone ``-`` as ``-i`` (clear the environment), so + # ``env - python`` starts the child without PYTHONPATH / the sandbox flag. + if lowered in {"-i", "--ignore-environment", "-"} and before_python: + if any( + os.path.basename(candidate.replace("\\", "/")).lower() == "env" + for candidate in before_python[:index] + ): + return True + if lowered.startswith("--unset="): + if token.split("=", 1)[1].upper() in _SANDBOX_PYTHON_ENV_VARS: + return True + if lowered.startswith("-u") and lowered != "-u" and not lowered.startswith("--"): + if token[2:].upper() in _SANDBOX_PYTHON_ENV_VARS: + return True + if lowered in {"-u", "--unset"} and index + 1 < len(before_python): + if before_python[index + 1].upper() in _SANDBOX_PYTHON_ENV_VARS: + return True + # GNU env also accepts grouped short options: ``-i`` clears the whole + # environment and ``-uNAME`` unsets a variable, so a bundle like + # ``-iuPYTHONPATH`` both resets the environment and would drop PYTHONPATH. + # ``-u`` consumes the remainder of the cluster as the NAME. + if ( + token.startswith("-") + and not token.startswith("--") + and len(token) > 1 + and any( + os.path.basename(candidate.replace("\\", "/")).lower() == "env" + for candidate in before_python[:index] + ) + ): + cluster = token[1:] + flags = cluster.split("u", 1)[0] if "u" in cluster else cluster + if "i" in flags: + return True + if "u" in cluster: + name = cluster.split("u", 1)[1] + if name and name.upper() in _SANDBOX_PYTHON_ENV_VARS: + return True + return False + + +def _segment_persistently_mutates_sandbox_python_env(segment: list[str]) -> bool: + if not segment: + return False + first = os.path.basename(segment[0].replace("\\", "/")).lower() + if all(_ASSIGNMENT_RE.match(token) for token in segment): + return any(token.split("=", 1)[0].upper() in _SANDBOX_PYTHON_ENV_VARS for token in segment) + if first in {"unset", "unsetenv"}: + return any(token.upper() in _SANDBOX_PYTHON_ENV_VARS for token in segment[1:]) + if first in {"export", "set", "setenv"}: + return any( + token.split("=", 1)[0].upper() in _SANDBOX_PYTHON_ENV_VARS for token in segment[1:] + ) + # ``declare -x``/``typeset -x`` export to the child environment, so an empty + # ``declare -x PYTHONPATH=`` drops the guard var. Only the exporting form + # (``-x`` present) reaches the child; a bare ``declare NAME=`` stays a shell + # local and is inherited-safe. + if first in {"declare", "typeset"}: + exported = any( + token.startswith("-") and not token.startswith("--") and "x" in token[1:] + for token in segment[1:] + ) + if exported: + return any( + token.split("=", 1)[0].upper() in _SANDBOX_PYTHON_ENV_VARS + for token in segment[1:] + if not token.startswith("-") + ) + return False + + +def _shell_alias_definitions(command: str) -> dict[str, str]: + """Collect ``alias NAME=VALUE`` definitions so a later use of NAME can be + expanded (``alias python='python -S'`` makes a subsequent ``python -c`` run + ``python -S -c``, skipping sitecustomize). Fail-open: only adds detections.""" + aliases: dict[str, str] = {} + for segment in _shell_command_segments(command): + if not segment or os.path.basename(segment[0].replace("\\", "/")).lower() != "alias": + continue + for token in segment[1:]: + if token.startswith("-") or "=" not in token: + continue + name, value = token.split("=", 1) + if name: + aliases[name] = value + return aliases + + +def _segment_with_alias_expansion(segment: list[str], aliases: dict[str, str]) -> list[str] | None: + if not aliases: + return None + cmd_index = 0 + while cmd_index < len(segment) and _ASSIGNMENT_RE.match(segment[cmd_index]): + cmd_index += 1 + if cmd_index >= len(segment) or segment[cmd_index] not in aliases: + return None + try: + value_tokens = shlex.split(aliases[segment[cmd_index]]) + except ValueError: + return None + if not value_tokens: + return None + return segment[:cmd_index] + value_tokens + segment[cmd_index + 1 :] + + +def _shell_command_substitutions(command: str) -> list[str]: + """Extract executable command substitutions, including quoted forms.""" + substitutions: list[str] = [] + index = 0 + quote: str | None = None + while index < len(command): + char = command[index] + if quote == "'": + if char == "'": + quote = None + index += 1 + continue + if char == "'" and quote is None: + quote = "'" + index += 1 + continue + if char == '"': + quote = None if quote == '"' else '"' + index += 1 + continue + if char == "\\": + index += 2 + continue + if char == "\x60": + end = index + 1 + while end < len(command): + if command[end] == "\\": + end += 2 + continue + if command[end] == "\x60": + substitutions.append(command[index + 1 : end]) + index = end + 1 + break + end += 1 + else: + return substitutions + continue + if command.startswith("$((", index): + index += 3 + continue + # $(...) command substitution and <(...)/>(...) process substitution all + # run their inner command; recurse into each. + if not ( + command.startswith("$(", index) + or command.startswith("<(", index) + or command.startswith(">(", index) + ): + index += 1 + continue + start = index + 2 + end = start + depth = 1 + nested_quote: str | None = None + while end < len(command): + nested = command[end] + if nested_quote == "'": + if nested == "'": + nested_quote = None + end += 1 + continue + if nested == "'" and nested_quote is None: + nested_quote = "'" + end += 1 + continue + if nested == '"': + nested_quote = None if nested_quote == '"' else '"' + end += 1 + continue + if nested == "\\": + end += 2 + continue + if command.startswith("$(", end) and not command.startswith("$((", end): + depth += 1 + end += 2 + continue + if nested == "(": + depth += 1 + elif nested == ")": + depth -= 1 + if depth == 0: + substitutions.append(command[start:end]) + index = end + 1 + break + end += 1 + else: + return substitutions + return substitutions + + +_HEREDOC_START_RE = re.compile(r"<<-?\s*(?P['\"]?)(?P[A-Za-z_][A-Za-z0-9_.-]*)") +# A Python interpreter reading its program from a process substitution +# (python <(printf ...)) runs a generated script this static scan cannot see; +# the inner generator's output is the program, so fail closed on this shape. +_PYTHON_PROCESS_SUB_SCRIPT_RE = re.compile( + r"(?:^|[\s;&|(])(?:[\w./\\-]*/)?python(?:w)?[0-9.]*(?:\.exe)?(?:\s+-[^\s]*)*\s+<\(", + re.IGNORECASE, +) + + +def _shell_here_doc_entries(command: str) -> tuple[list[tuple[str, str]], bool]: + if "<<" not in command: + return [], False + lines = command.splitlines() + entries: list[tuple[str, str]] = [] + malformed = False + index = 0 + while index < len(lines): + line = lines[index] + matches = list(_HEREDOC_START_RE.finditer(line)) + if not matches: + index += 1 + continue + for match in matches: + quote = match.group("quote") + delimiter = match.group("name") + end = match.end() + if quote and (end >= len(line) or line[end] != quote): + continue + body_start = index + 1 + body_end = body_start + while body_end < len(lines): + candidate = lines[body_end] + if candidate == delimiter or candidate.lstrip("\t") == delimiter: + break + body_end += 1 + if body_end >= len(lines): + malformed = True + continue + # Keep the command text after the delimiter so a here-doc piped into + # another process (cat <<'PY' | python) still exposes that consumer: + # the body is the consumer's stdin program. Only reconstruct the tail + # for a single here-doc on the line to avoid mis-pairing multi-doc + # openers (cmd < tuple[list[tuple[str, str]], bool]: + if "<<<" not in command: + return [], False + try: + lexer = shlex.shlex(command, posix = sys.platform != "win32", punctuation_chars = ";&|()`{}<") + lexer.whitespace_split = True + tokens = list(lexer) + except ValueError: + return [], True + segments: list[list[str]] = [[]] + expect_command = True + for token in tokens: + if token in _SHELL_SEPARATORS or ( + expect_command and token.lower() in _SHELL_KEYWORDS_AS_SEP + ): + if segments[-1]: + segments.append([]) + expect_command = True + continue + segments[-1].append(token) + expect_command = False + + entries: list[tuple[str, str]] = [] + malformed = False + for segment in segments: + index = 0 + while index < len(segment): + if segment[index] != "<<<": + index += 1 + continue + if index + 1 >= len(segment): + malformed = True + break + opener = segment[:index] + segment[index + 2 :] + if not opener: + malformed = True + else: + entries.append((shlex.join(opener), segment[index + 1])) + index += 2 + return entries, malformed + + +_SHELL_EXPANSION_TOKEN_RE = re.compile( + r"\$(?:[A-Za-z_][A-Za-z0-9_]*|\{[^}]*\}|[0-9#?*$!@_-])|%[A-Za-z_][A-Za-z0-9_]*%" +) +_PYTHON_CHILD_LAUNCHERS = frozenset({"run", "call", "check_call", "check_output", "Popen"}) +_PYTHON_OS_EXEC_LAUNCHERS = frozenset( + { + "execl", + "execle", + "execlp", + "execlpe", + "execv", + "execve", + "execvp", + "execvpe", + "posix_spawn", + "posix_spawnp", + } +) +# argv-list launchers whose first positional is the full ``[program, *args]`` +# vector (``pty.spawn(argv)``). Modelled like a subprocess sequence. +_PYTHON_ARGV_LIST_LAUNCHERS = frozenset({"spawn"}) +# ``asyncio.create_subprocess_exec(program, *args)`` passes the program and its +# arguments as separate positionals; ``create_subprocess_shell(cmd)`` passes a +# single shell-script string. +_PYTHON_ASYNCIO_EXEC_LAUNCHERS = frozenset({"create_subprocess_exec"}) +_PYTHON_ASYNCIO_SHELL_LAUNCHERS = frozenset({"create_subprocess_shell"}) + + +def _segment_with_shell_expansions_split(segment: list[str]) -> list[str] | None: + expanded: list[str] = [] + changed = False + for token in segment: + if not _SHELL_EXPANSION_TOKEN_RE.search(token): + expanded.append(token) + continue + changed = True + # Resolve default/alternate parameter expansions to their operand first + # (``${PYTHON:-python} -S`` runs ``python -S`` when PYTHON is unset), so + # a command word supplied entirely by a defaulted expansion is still + # modelled instead of dropped. + resolved = _expand_param_defaults(token) + replacement = _SHELL_EXPANSION_TOKEN_RE.sub(" ", resolved).split() + expanded.extend(replacement) + return expanded if changed else None + + +def _python_inline_payload(arguments: list[str]) -> str | None: + skip_next = False + for index, argument in enumerate(arguments): + if skip_next: + skip_next = False + continue + if argument == "-c": + return arguments[index + 1] if index + 1 < len(arguments) else "" + if argument.startswith("-c") and argument != "-c": + return argument[2:] + if argument == "-m" or not argument.startswith("-"): + return None + if argument == "--": + return None + if argument in {"-W", "-X", "--check-hash-based-pycs"}: + skip_next = True + return None + + +def _python_reads_program_from_stdin(arguments: list[str]) -> bool: + skip_next = False + for argument in arguments: + if skip_next: + skip_next = False + continue + if argument == "-c" or argument.startswith("-c"): + return False + if argument == "-m": + return False + if argument == "--": + return False + if argument == "-": + return True + if not argument.startswith("-"): + return False + if argument in {"-W", "-X", "--check-hash-based-pycs"}: + skip_next = True + return True + + +def _static_python_command_argument(node: ast.AST) -> str | None: + value = _static_python_string(node) + if value is not None: + return value + if isinstance(node, (ast.List, ast.Tuple)): + parts: list[str] = [] + for element in node.elts: + piece = _static_python_string(element) + if piece is None: + return None + parts.append(piece) + return shlex.join(parts) + return None + + +def _static_python_string(node: ast.AST | None) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + # Fold statically concatenated string literals (``'py' + 'thon'``) and + # constant-only f-strings so an argv element assembled from pieces is still + # recognised as ``python`` rather than treated as dynamic. + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = _static_python_string(node.left) + right = _static_python_string(node.right) + if left is not None and right is not None: + return left + right + return None + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for value in node.values: + piece = _static_python_string(value) + if piece is None: + return None + parts.append(piece) + return "".join(parts) + return None + + +def _static_python_string_list(node: ast.AST | None) -> list[str] | None: + if not isinstance(node, (ast.List, ast.Tuple)): + return None + parts: list[str] = [] + for element in node.elts: + value = _static_python_string(element) + if value is None: + return None + parts.append(value) + return parts + + +def _python_call_keyword(node: ast.Call, name: str) -> ast.AST | None: + for keyword in node.keywords or []: + if keyword.arg == name: + return keyword.value + return None + + +def _python_call_is_shell_true(node: ast.Call) -> bool: + keyword = _python_call_keyword(node, "shell") + return isinstance(keyword, ast.Constant) and keyword.value is True + + +def _python_env_key(node: ast.AST) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value.upper() + return None + + +def _python_env_mapping_taints_guard(value: ast.AST) -> bool: + """True if merging ``value`` into ``os.environ`` could clear/redirect a + guard variable. A literal dict is safe only when every key is a known, + non-guard name; any unknown mapping fails closed.""" + if isinstance(value, ast.Dict): + for key in value.keys: + name = _python_env_key(key) if key is not None else None + if name is None or name in _SANDBOX_PYTHON_ENV_VARS: + return True + return False + return True + + +def _python_payload_mutates_sandbox_env(node: ast.AST, os_aliases: set[str]) -> bool: + def is_os_environ(candidate: ast.AST) -> bool: + return ( + isinstance(candidate, ast.Attribute) + and candidate.attr == "environ" + and isinstance(candidate.value, ast.Name) + and candidate.value.id in os_aliases + ) + + if isinstance(node, (ast.Assign, ast.AnnAssign, ast.AugAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + for target in targets: + if isinstance(target, ast.Subscript) and is_os_environ(target.value): + key = _python_env_key(target.slice) + if key is None or key in _SANDBOX_PYTHON_ENV_VARS: + return True + # ``os.environ |= {...}`` (PEP 584) merges a mapping in place; its target is + # an Attribute, not a Subscript, so it is checked separately here. + if ( + isinstance(node, ast.AugAssign) + and isinstance(node.op, ast.BitOr) + and is_os_environ(node.target) + ): + if _python_env_mapping_taints_guard(node.value): + return True + if isinstance(node, ast.Delete): + for target in node.targets: + if isinstance(target, ast.Subscript) and is_os_environ(target.value): + key = _python_env_key(target.slice) + if key is None or key in _SANDBOX_PYTHON_ENV_VARS: + return True + if not isinstance(node, ast.Call): + return False + if isinstance(node.func, ast.Attribute): + if is_os_environ(node.func.value): + if node.func.attr in {"clear", "popitem"}: + return True + if node.func.attr in {"pop", "setdefault", "update", "__delitem__", "__setitem__"}: + key = _python_env_key(node.args[0]) if node.args else None + return key is None or key in _SANDBOX_PYTHON_ENV_VARS + if ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in os_aliases + and node.func.attr in {"putenv", "unsetenv"} + ): + key = _python_env_key(node.args[0]) if node.args else None + return key is None or key in _SANDBOX_PYTHON_ENV_VARS + return False + + +def _python_payload_child_env_taints_guard(node: ast.Call, os_aliases: set[str]) -> bool: + env_node = _python_call_keyword(node, "env") + if env_node is None: + return False + return _python_env_node_taints_guard(env_node, os_aliases) + + +def _python_env_node_taints_guard(env_node: ast.AST, os_aliases: set[str]) -> bool: + if isinstance(env_node, ast.Constant) and env_node.value is None: + return False + if ( + isinstance(env_node, ast.Attribute) + and env_node.attr == "environ" + and isinstance(env_node.value, ast.Name) + and env_node.value.id in os_aliases + ): + return False + if ( + isinstance(env_node, ast.Call) + and isinstance(env_node.func, ast.Attribute) + and env_node.func.attr == "copy" + and isinstance(env_node.func.value, ast.Attribute) + and env_node.func.value.attr == "environ" + and isinstance(env_node.func.value.value, ast.Name) + and env_node.func.value.value.id in os_aliases + ): + return False + return True + + +def _python_subprocess_command_argument(node: ast.Call, command_node: ast.AST) -> str | None: + if _python_call_is_shell_true(node): + # ``shell=True`` runs the command through ``/bin/sh -c``. A string is the + # script verbatim; for a sequence POSIX passes element 0 as the script + # (later elements become ``$0``, ``$1`` ...). Either way the value is + # shell input, so return it raw for the shell-aware recursion rather than + # ``shlex.join``-ing the whole vector into one quoted word. + script = _static_python_string(command_node) + if script is not None: + return script + if isinstance(command_node, (ast.List, ast.Tuple)) and command_node.elts: + return _static_python_string(command_node.elts[0]) + return None + executable = _static_python_string(_python_call_keyword(node, "executable")) + parts = _static_python_string_list(command_node) + if executable and _python_executable_token(executable): + if parts is not None: + return shlex.join([executable, *parts[1:]]) + nested = _static_python_command_argument(command_node) + if nested is not None: + return f"{shlex.quote(executable)} {nested}" + return None + return _static_python_command_argument(command_node) + + +def _python_os_exec_command_argument(node: ast.Call, launcher: str) -> str | None: + args = list(node.args) + if not args: + return None + executable = _static_python_string(args[0]) + if executable is None: + return None + if launcher in {"execl", "execlp"}: + parts = [_static_python_string(arg) for arg in args[2:]] + elif launcher in {"execle", "execlpe"}: + parts = [_static_python_string(arg) for arg in args[2:-1]] + elif launcher in {"execv", "execvp"}: + argv = _static_python_string_list(args[1] if len(args) > 1 else None) + parts = None if argv is None else argv[1:] + elif launcher in {"execve", "execvpe", "posix_spawn", "posix_spawnp"}: + argv = _static_python_string_list(args[1] if len(args) > 1 else None) + parts = None if argv is None else argv[1:] + else: + return None + if parts is None or any(part is None for part in parts): + return None + return shlex.join([executable, *parts]) + + +def _python_os_exec_env_taints_guard(node: ast.Call, launcher: str, os_aliases: set[str]) -> bool: + if launcher in {"execle", "execlpe"}: + env_node = node.args[-1] if len(node.args) >= 2 else None + elif launcher in {"execve", "execvpe", "posix_spawn", "posix_spawnp"}: + env_node = node.args[2] if len(node.args) > 2 else None + else: + env_node = None + return env_node is not None and _python_env_node_taints_guard(env_node, os_aliases) + + +def _python_payload_launches_startup_bypass( + code: str, + depth: int, + environment_tainted: bool = False, +) -> bool: + if depth > 4: + return True + try: + tree = ast.parse(code) + except SyntaxError: + return False + subprocess_aliases = {"subprocess"} + os_aliases = {"os"} + pty_aliases = {"pty"} + asyncio_aliases = {"asyncio"} + launcher_aliases: set[str] = set() + os_exec_aliases: dict[str, str] = {} + argv_list_launcher_aliases: set[str] = set() + asyncio_exec_aliases: set[str] = set() + asyncio_shell_aliases: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "subprocess": + subprocess_aliases.add(alias.asname or "subprocess") + elif alias.name == "os": + os_aliases.add(alias.asname or "os") + elif alias.name == "pty": + pty_aliases.add(alias.asname or "pty") + elif alias.name == "asyncio": + asyncio_aliases.add(alias.asname or "asyncio") + elif isinstance(node, ast.ImportFrom): + if node.module == "subprocess": + for alias in node.names: + if alias.name == "*": + launcher_aliases.update(_PYTHON_CHILD_LAUNCHERS) + elif alias.name in _PYTHON_CHILD_LAUNCHERS: + launcher_aliases.add(alias.asname or alias.name) + elif node.module == "os": + for alias in node.names: + if alias.name == "*": + os_exec_aliases.update( + (launcher, launcher) for launcher in _PYTHON_OS_EXEC_LAUNCHERS + ) + elif alias.name in _PYTHON_OS_EXEC_LAUNCHERS: + os_exec_aliases[alias.asname or alias.name] = alias.name + elif node.module == "pty": + for alias in node.names: + if alias.name == "*": + argv_list_launcher_aliases.update(_PYTHON_ARGV_LIST_LAUNCHERS) + elif alias.name in _PYTHON_ARGV_LIST_LAUNCHERS: + argv_list_launcher_aliases.add(alias.asname or alias.name) + elif node.module == "asyncio": + for alias in node.names: + if alias.name == "*" or alias.name in _PYTHON_ASYNCIO_EXEC_LAUNCHERS: + if alias.name == "*": + asyncio_exec_aliases.update(_PYTHON_ASYNCIO_EXEC_LAUNCHERS) + asyncio_shell_aliases.update(_PYTHON_ASYNCIO_SHELL_LAUNCHERS) + else: + asyncio_exec_aliases.add(alias.asname or alias.name) + elif alias.name in _PYTHON_ASYNCIO_SHELL_LAUNCHERS: + asyncio_shell_aliases.add(alias.asname or alias.name) + if _python_payload_mutates_sandbox_env(node, os_aliases): + environment_tainted = True + # Propagate simple launcher aliases assigned by name + # (``r = subprocess.run``; ``r([...])``) to a fixpoint so chained rebinds + # (``a = subprocess.run; b = a``) resolve too. + assigns = [ + n + for n in ast.walk(tree) + if isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name) + ] + changed = True + while changed: + changed = False + for assign in assigns: + name = assign.targets[0].id + value = assign.value + if isinstance(value, ast.Attribute) and isinstance(value.value, ast.Name): + base, attr = value.value.id, value.attr + if ( + base in subprocess_aliases + and attr in _PYTHON_CHILD_LAUNCHERS + and name not in launcher_aliases + ): + launcher_aliases.add(name) + changed = True + elif ( + base in os_aliases + and attr in _PYTHON_OS_EXEC_LAUNCHERS + and os_exec_aliases.get(name) != attr + ): + os_exec_aliases[name] = attr + changed = True + elif ( + base in pty_aliases + and attr in _PYTHON_ARGV_LIST_LAUNCHERS + and name not in argv_list_launcher_aliases + ): + argv_list_launcher_aliases.add(name) + changed = True + elif ( + base in asyncio_aliases + and attr in _PYTHON_ASYNCIO_EXEC_LAUNCHERS + and name not in asyncio_exec_aliases + ): + asyncio_exec_aliases.add(name) + changed = True + elif ( + base in asyncio_aliases + and attr in _PYTHON_ASYNCIO_SHELL_LAUNCHERS + and name not in asyncio_shell_aliases + ): + asyncio_shell_aliases.add(name) + changed = True + elif isinstance(value, ast.Name): + if value.id in launcher_aliases and name not in launcher_aliases: + launcher_aliases.add(name) + changed = True + elif ( + value.id in os_exec_aliases + and os_exec_aliases.get(name) != os_exec_aliases[value.id] + ): + os_exec_aliases[name] = os_exec_aliases[value.id] + changed = True + elif ( + value.id in argv_list_launcher_aliases + and name not in argv_list_launcher_aliases + ): + argv_list_launcher_aliases.add(name) + changed = True + elif value.id in asyncio_exec_aliases and name not in asyncio_exec_aliases: + asyncio_exec_aliases.add(name) + changed = True + elif value.id in asyncio_shell_aliases and name not in asyncio_shell_aliases: + asyncio_shell_aliases.add(name) + changed = True + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + nested = None + env_tainted = _python_payload_child_env_taints_guard(node, os_aliases) + # Recurse into a statically-known ``exec``/``eval`` string payload so a + # child launch hidden inside ``exec("... subprocess.run([...]) ...")`` is + # scanned rather than treated as an opaque call. + if isinstance(node.func, ast.Name) and node.func.id in {"exec", "eval"} and node.args: + inner = _static_python_string(node.args[0]) + if inner is not None and _python_payload_launches_startup_bypass( + inner, depth + 1, environment_tainted + ): + return True + if isinstance(node.func, ast.Name) and node.func.id in launcher_aliases: + command_node = node.args[0] if node.args else _python_call_keyword(node, "args") + if command_node is not None: + nested = _python_subprocess_command_argument(node, command_node) + elif isinstance(node.func, ast.Name) and node.func.id in argv_list_launcher_aliases: + command_node = node.args[0] if node.args else None + if command_node is not None: + nested = _static_python_command_argument(command_node) + elif isinstance(node.func, ast.Name) and node.func.id in asyncio_exec_aliases: + parts = [_static_python_string(arg) for arg in node.args] + if parts and all(part is not None for part in parts): + nested = shlex.join(parts) + elif isinstance(node.func, ast.Name) and node.func.id in asyncio_shell_aliases: + nested = _static_python_string(node.args[0]) if node.args else None + elif isinstance(node.func, ast.Name) and node.func.id in os_exec_aliases: + launcher = os_exec_aliases[node.func.id] + nested = _python_os_exec_command_argument(node, launcher) + env_tainted = _python_os_exec_env_taints_guard(node, launcher, os_aliases) + elif isinstance(node.func, ast.Attribute): + if ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in subprocess_aliases + and node.func.attr in _PYTHON_CHILD_LAUNCHERS + ): + command_node = node.args[0] if node.args else _python_call_keyword(node, "args") + if command_node is not None: + nested = _python_subprocess_command_argument(node, command_node) + elif ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in os_aliases + and node.func.attr in {"system", "popen"} + ): + command_node = node.args[0] if node.args else None + if command_node is not None: + nested = _static_python_command_argument(command_node) + elif ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in os_aliases + and node.func.attr in _PYTHON_OS_EXEC_LAUNCHERS + ): + nested = _python_os_exec_command_argument(node, node.func.attr) + env_tainted = _python_os_exec_env_taints_guard(node, node.func.attr, os_aliases) + elif ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in pty_aliases + and node.func.attr in _PYTHON_ARGV_LIST_LAUNCHERS + ): + command_node = node.args[0] if node.args else None + if command_node is not None: + nested = _static_python_command_argument(command_node) + elif ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in asyncio_aliases + and node.func.attr in _PYTHON_ASYNCIO_EXEC_LAUNCHERS + ): + parts = [_static_python_string(arg) for arg in node.args] + if parts and all(part is not None for part in parts): + nested = shlex.join(parts) + elif ( + isinstance(node.func.value, ast.Name) + and node.func.value.id in asyncio_aliases + and node.func.attr in _PYTHON_ASYNCIO_SHELL_LAUNCHERS + ): + nested = _static_python_string(node.args[0]) if node.args else None + if nested is None: + continue + if _sandbox_python_startup_bypasses_guard( + nested, + depth + 1, + environment_tainted or env_tainted, + ): + return True + return False + + +def _segment_python_launch_bypasses_guard( + segment: list[str], python_index: int, environment_tainted: bool, depth: int +) -> bool: + if environment_tainted or _segment_mutates_sandbox_python_env(segment, python_index): + return True + arguments = segment[python_index + 1 :] + if _python_flags_skip_sitecustomize(arguments): + return True + payload = _python_inline_payload(arguments) + return payload is not None and _python_payload_launches_startup_bypass(payload, depth + 1) + + +def _sandbox_python_startup_bypasses_guard( + command: str, + depth: int = 0, + environment_tainted: bool = False, +) -> bool: + """Detect terminal-launched Python that suppresses the sandbox sitecustomize guard.""" + if depth > 4: + return True + if _PYTHON_PROCESS_SUB_SCRIPT_RE.search(command): + return True + here_doc_entries, malformed_here_doc = _shell_here_doc_entries(command) + if malformed_here_doc: + return True + for opener, payload in here_doc_entries: + opener_command = _shell_command_with_unquoted_newlines_as_separators( + _shell_command_without_line_continuations(opener) + ) + for segment in _shell_command_segments(opener_command): + python_index = _segment_python_index(segment) + if python_index is None: + continue + if _segment_python_launch_bypasses_guard( + segment, python_index, environment_tainted, depth + ): + return True + arguments = segment[python_index + 1 :] + if _python_inline_payload(arguments) is None and _python_reads_program_from_stdin( + arguments + ): + if _python_payload_launches_startup_bypass(payload, depth + 1, environment_tainted): + return True + # Preserve the previous fail-safe for nested shell bodies. + if _sandbox_python_startup_bypasses_guard(payload, depth + 1, environment_tainted): + return True + here_string_entries, malformed_here_string = _shell_here_string_entries(command) + if malformed_here_string: + return True + for opener, payload in here_string_entries: + dynamic_payload = _SHELL_EXPANSION_TOKEN_RE.search(payload) is not None + opener_command = _shell_command_with_unquoted_newlines_as_separators( + _shell_command_without_line_continuations(opener) + ) + for segment in _shell_command_segments(opener_command): + python_index = _segment_python_index(segment) + if python_index is not None: + if _segment_python_launch_bypasses_guard( + segment, python_index, environment_tainted, depth + ): + return True + arguments = segment[python_index + 1 :] + if _python_inline_payload(arguments) is None and _python_reads_program_from_stdin( + arguments + ): + if dynamic_payload: + return True + if _python_payload_launches_startup_bypass( + payload, depth + 1, environment_tainted + ): + return True + shell_index = _segment_shell_index(segment) + if shell_index is not None and _segment_shell_reads_stdin(segment, shell_index): + if dynamic_payload: + return True + if _sandbox_python_startup_bypasses_guard(payload, depth + 1, environment_tainted): + return True + command = _shell_command_with_unquoted_newlines_as_separators( + _shell_command_without_line_continuations(command) + ) + for nested in _shell_command_substitutions(command): + if _sandbox_python_startup_bypasses_guard(nested, depth + 1, environment_tainted): + return True + aliases = _shell_alias_definitions(command) + for segment in _shell_command_segments(command): + first = os.path.basename(segment[0].replace("\\", "/")).lower() + wrapper_context = first in _PYTHON_LAUNCH_WRAPPERS + find_exec = first in {"find", "fd"} + for shell_index, shell_token in enumerate(segment): + shell = os.path.basename(shell_token.replace("\\", "/")).lower() + # A shell after the first token is only a real launch when it follows + # a launch wrapper (env/xargs/...) or a find/fd -exec flag; otherwise + # it is an argument (e.g. a path) and is ignored. + find_exec_context = find_exec and any( + token in _FIND_EXEC_FLAGS for token in segment[:shell_index] + ) + if shell not in _SHELL_SCRIPT_INTERPRETERS or ( + shell_index and not wrapper_context and not find_exec_context + ): + continue + for index in range(shell_index + 1, len(segment) - 1): + token = segment[index] + if token.lower() == "/c" or (token.startswith("-") and token.lower().endswith("c")): + nested = segment[index + 1] + if _segment_mutates_sandbox_python_env(segment, shell_index): + nested = f"PYTHONPATH=; {nested}" + if _sandbox_python_startup_bypasses_guard( + nested, depth + 1, environment_tainted + ): + return True + break + break + # ``env -S "python -S ..."`` splits its argument into a command; this + # applies wherever env sits in the launcher chain (env first, or behind a + # wrapper such as ``timeout 1 env -S ...`` / ``find . -exec env -S ...``), + # not only when env is the first token. + for env_index, env_token in enumerate(segment): + if os.path.basename(env_token.replace("\\", "/")).lower() != "env": + continue + env_launch = ( + env_index == 0 + or wrapper_context + or (find_exec and any(t in _FIND_EXEC_FLAGS for t in segment[:env_index])) + ) + if not env_launch: + continue + for index in range(env_index + 1, len(segment)): + token = segment[index] + if token in {"-S", "--split-string"} and index + 1 < len(segment): + if _sandbox_python_startup_bypasses_guard( + segment[index + 1], depth + 1, environment_tainted + ): + return True + elif token.startswith("--split-string="): + if _sandbox_python_startup_bypasses_guard( + token.split("=", 1)[1], depth + 1, environment_tainted + ): + return True + expanded_segment = _segment_with_shell_expansions_split(segment) + if expanded_segment: + expanded_python_index = _segment_python_index(expanded_segment) + if expanded_python_index is not None and _segment_python_launch_bypasses_guard( + expanded_segment, expanded_python_index, environment_tainted, depth + ): + return True + aliased_segment = _segment_with_alias_expansion(segment, aliases) + if aliased_segment is not None: + aliased_python_index = _segment_python_index(aliased_segment) + if aliased_python_index is not None and _segment_python_launch_bypasses_guard( + aliased_segment, aliased_python_index, environment_tainted, depth + ): + return True + python_index = _segment_python_index(segment) + if python_index is not None: + if _segment_python_launch_bypasses_guard( + segment, python_index, environment_tainted, depth + ): + return True + environment_tainted = ( + environment_tainted or _segment_persistently_mutates_sandbox_python_env(segment) + ) + return False def _find_blocked_commands(command: str) -> set[str]: @@ -328,6 +1519,7 @@ def _find_blocked_commands(command: str) -> set[str]: # Directory holding the sandbox ``sitecustomize.py`` shim (code-interpreter # path remap); placed on the sandboxed child's PYTHONPATH in _build_safe_env. _SANDBOX_SITE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sandbox_site") +_BYPASS_SITE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bypass_site") # ── "Approve for me" (permission_mode="auto") safety detection ────────────── # Auto mode pauses only calls classified here as potentially unsafe. The sandbox # and hard blocks (blocklist, rlimits) still apply at run time; this gate only @@ -552,6 +1744,10 @@ _AUTO_SENSITIVE_MCP_NOUN_RE = re.compile( re.IGNORECASE, ) +# Low-level clients bypass the sandbox host scanner, so sandboxed Python blocks +# them and auto mode asks before they can run. +_SANDBOX_BLOCKED_NETWORK_MODULES = frozenset({"httpcore", "boto3", "botocore"}) + # Python: modules whose import alone signals side effects auto mode should ask # about (process spawning, network, bulk file ops, low-level memory). _AUTO_UNSAFE_PY_MODULES = frozenset( @@ -605,6 +1801,7 @@ _AUTO_UNSAFE_PY_MODULES = frozenset( "venv", } ) +_AUTO_UNSAFE_PY_MODULES |= _SANDBOX_BLOCKED_NETWORK_MODULES # Attribute calls that mutate the filesystem / spawn processes (os.remove, # Path.write_text, sock.connect, ...) regardless of how the module was bound. _AUTO_UNSAFE_PY_ATTRS = frozenset( @@ -666,6 +1863,7 @@ _AUTO_UNSAFE_PY_ATTRS = frozenset( # extractall/extract write arbitrary files (zip-slip): extract takes a # single member but an attacker-controlled member path still escapes. "exec_module", + "load_module", "extractall", "extract", "FileIO", @@ -2402,22 +3600,21 @@ _RENDER_HTML_NETWORK_RE = re.compile( r"\bnew\s+(?:Shared)?Worker\s*\(|" r"@import|" r"url\(\s*[\"']?\s*(?:https?:|/)|" + # ES module loads of a remote/root URL: dynamic import('https://...') and + # static `import ... from 'https://...'` / side-effect `import 'https://...'` + # (a relative './mod.js' specifier has no https:/root prefix and stays static). + r"\bimport\s*\(\s*[\"'`]?\s*(?:https?:|/)|" + r"\bimport\b[^;\n]*?[\"'`]\s*(?:https?:|/)|" + # Module re-exports (export * from '...' / export {a} from '/...') fetch the + # referenced module just like a static import. + r"\bexport\b[^;\n]*?\bfrom\s*[\"'`]\s*(?:https?:|/)|" r"]*\bsrc\s*=|" - r"\b(?:src|href|srcset)\s*=\s*[\"']?\s*(?:https?:|/)|" # Self-navigation sinks: location.assign/replace(...), window.open(...), and # assigning a URL to (window.)location(.href). location.reload()/history.back # do not navigate to a new URL, so they stay static. r"\blocation\s*\.\s*(?:assign|replace)\s*\(|" r"\bwindow\s*\.\s*open\s*\(|" r"\b(?:window\s*\.\s*)?location(?:\s*\.\s*href)?\s*=\s*[\"'`]?\s*(?:https?:|/)|" - # Bracket-access obfuscation: window['fetch'](...), self["open"](...). - r"\[\s*[\"'](?:fetch|open|XMLHttpRequest|WebSocket|EventSource|importScripts|" - r"sendBeacon|serviceWorker)[\"']\s*\]|" - # Computed bracket key spliced at runtime on a global host object - # (window['fet'+'ch'](...)): a quoted fragment adjacent to a + inside the - # index. Anchored to a host object so a plain obj['a'+'b'] key stays safe. - r"\b(?:window|self|globalThis|top|parent|frames)\s*\[[^\]]*" - r"(?:[\"']\s*\+|\+\s*[\"'])[^\]]*\]|" # Declarative meta-refresh navigation to a URL (order-tolerant); a bare # content="30" self-reload has no url= and stays static. r"]*http-equiv\s*=\s*[\"']?\s*refresh)(?=[^>]*\burl\s*=)|" @@ -2428,13 +3625,744 @@ _RENDER_HTML_NETWORK_RE = re.compile( # matching. Line // comments are left alone -- stripping them would eat the // in # an https:// URL and hide a real load. _JS_BLOCK_COMMENT_RE = re.compile(r"/\*.*?\*/", re.DOTALL) +_RENDER_HTML_GLOBAL_BRACKET_RE = re.compile( + r"\b(?:window|self|globalThis|top|parent|frames|this|navigator|" + r"document\s*\.\s*defaultView)\s*(?:\?\.\s*)?" + r"(?:\[\s*\d+\s*\]\s*(?:\?\.\s*)?)*\[([^\]]*)\]", + re.IGNORECASE | re.DOTALL, +) +_RENDER_HTML_SET_ATTRIBUTE_START_RE = re.compile( + r"\.\s*(?PsetAttribute(?:NS)?)\s*(?:\?\.\s*)?\(", + re.IGNORECASE, +) +_RENDER_HTML_SET_ATTRIBUTE_INDIRECT_CALL_START_RE = re.compile( + r"\.\s*(?PsetAttribute(?:NS)?)\s*(?:\?\.\s*|\.\s*)" + r"(?Pcall|apply)\s*(?:\?\.\s*)?\(", + re.IGNORECASE, +) +_RENDER_HTML_PROPERTY_ASSIGNMENT_START_RE = re.compile( + r"\.\s*(?Psrc|href|srcset|action|formaction|poster|data|ping|srcdoc)\s*" + r"(?P\+=|&&=|\|\|=|\?\?=|=(?!=))", + re.IGNORECASE, +) +_RENDER_HTML_MARKUP_ASSIGNMENT_START_RE = re.compile( + r"\.\s*(?:innerHTML|outerHTML)\s*(?:\+=|&&=|\|\|=|\?\?=|=(?!=))", + re.IGNORECASE, +) +_RENDER_HTML_MARKUP_CALL_START_RE = re.compile( + r"(?:\.\s*(?PinsertAdjacentHTML)|" + r"\.\s*(?PcreateContextualFragment)|" + # document.write / writeln, optionally reached through a document-valued + # receiver such as document.open(): document.open().write('') + # returns the same document and inserts the remote-loading markup. + r"\bdocument\s*(?:(?:\?\.\s*|\.\s*)open\s*\([^()]*\)\s*)?" + r"\s*(?:\?\.\s*|\.\s*)(?Pwrite|writeln))" + r"\s*(?:\?\.\s*)?\(", + re.IGNORECASE, +) +_RENDER_HTML_MARKUP_INDIRECT_CALL_START_RE = re.compile( + r"(?:\.\s*(?PinsertAdjacentHTML)|" + r"\.\s*(?PcreateContextualFragment)|" + r"\bdocument\s*(?:(?:\?\.\s*|\.\s*)open\s*\([^()]*\)\s*)?" + r"\s*(?:\?\.\s*|\.\s*)(?Pwrite|writeln))" + r"\s*(?:\?\.\s*|\.\s*)(?Pcall|apply)\s*(?:\?\.\s*)?\(", + re.IGNORECASE, +) +_RENDER_HTML_MARKUP_TAGGED_TEMPLATE_START_RE = re.compile( + r"(?:\.\s*(?PinsertAdjacentHTML)|" + r"\.\s*(?PcreateContextualFragment)|" + r"\bdocument\s*(?:(?:\?\.\s*|\.\s*)open\s*\([^()]*\)\s*)?" + r"\s*(?:\?\.\s*|\.\s*)(?Pwrite|writeln))" + r"\s*(?P