diff --git a/studio/backend/core/inference/sandbox_site/sitecustomize.py b/studio/backend/core/inference/sandbox_site/sitecustomize.py index 3b6e2aaee7..5bf8a136a6 100644 --- a/studio/backend/core/inference/sandbox_site/sitecustomize.py +++ b/studio/backend/core/inference/sandbox_site/sitecustomize.py @@ -89,6 +89,23 @@ def _path_is_in_roots(filename, roots): 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): @@ -167,6 +184,15 @@ def _blocked_network_module(fullname): return None +def _blocked_network_loader_origin(filename): + root = _blocked_network_module_origin(filename) + if root in _BLOCKED_NETWORK_MODULES: + return root + if root in _DIRECT_BLOCKED_NETWORK_MODULES and _sandbox_code_requested_import(): + return root + return None + + def _raise_blocked_network_module(root): raise ModuleNotFoundError( f"Blocked: low-level network module {root!r} is unavailable in sandboxed code" @@ -383,22 +409,52 @@ def _guarded_import_module(name, package = None): def _guard_legacy_source_loader(): cls = importlib.machinery.SourceFileLoader - original = getattr(cls, "load_module", None) - if not callable(original) or getattr(original, "_unsloth_network_guard", False): - return + original_load_module = getattr(cls, "load_module", None) + original_exec_module = getattr(cls, "exec_module", None) - def guarded(self, *args, **kwargs): - fullname = args[0] if args else kwargs.get("fullname", getattr(self, "name", None)) + def blocked_loader_root(self, fullname = None): root = _blocked_network_module(fullname) - if root is not None: - _raise_blocked_network_module(root) - return original(self, *args, **kwargs) + if root is None: + root = _blocked_network_loader_origin(getattr(self, "path", None)) + return root - guarded._unsloth_network_guard = True - guarded.__name__ = getattr(original, "__name__", "load_module") - guarded.__qualname__ = getattr(original, "__qualname__", guarded.__name__) - guarded.__doc__ = getattr(original, "__doc__", None) - cls.load_module = guarded + if callable(original_load_module) and not getattr( + original_load_module, "_unsloth_network_guard", False + ): + + def guarded_load_module(self, *args, **kwargs): + fullname = args[0] if args else kwargs.get("fullname", getattr(self, "name", None)) + root = blocked_loader_root(self, fullname) + if root is not None: + _raise_blocked_network_module(root) + return original_load_module(self, *args, **kwargs) + + guarded_load_module._unsloth_network_guard = True + guarded_load_module.__name__ = getattr(original_load_module, "__name__", "load_module") + guarded_load_module.__qualname__ = getattr( + original_load_module, "__qualname__", guarded_load_module.__name__ + ) + guarded_load_module.__doc__ = getattr(original_load_module, "__doc__", None) + cls.load_module = guarded_load_module + + if callable(original_exec_module) and not getattr( + original_exec_module, "_unsloth_network_guard", False + ): + + def guarded_exec_module(self, module): + fullname = getattr(module, "__name__", getattr(self, "name", None)) + root = blocked_loader_root(self, fullname) + if root is not None: + _raise_blocked_network_module(root) + return original_exec_module(self, module) + + guarded_exec_module._unsloth_network_guard = True + guarded_exec_module.__name__ = getattr(original_exec_module, "__name__", "exec_module") + guarded_exec_module.__qualname__ = getattr( + original_exec_module, "__qualname__", guarded_exec_module.__name__ + ) + guarded_exec_module.__doc__ = getattr(original_exec_module, "__doc__", None) + cls.exec_module = guarded_exec_module def _make_network_guard_audit(): @@ -420,6 +476,34 @@ def _make_network_guard_audit(): f"Blocked: low-level network module {root!r} is unavailable in sandboxed code" ) + def blocked_origin(filename): + if not isinstance(filename, str) or filename.startswith("<"): + return None + try: + path = realpath(filename) + for root in trusted_roots: + if commonpath((root, path)) != root: + continue + relative = relpath(path, root).replace("\\", "/") + package = relative.split("/", 1)[0].removesuffix(".py") + if package in blocked or package in direct_blocked: + return package + except (OSError, ValueError): + return None + return None + + def blocked_origin_in_stack(skip): + try: + frame = getframe(skip) + except ValueError: + return None + while frame is not None: + root = blocked_origin(frame.f_code.co_filename) + if root is not None: + return root + frame = frame.f_back + return None + def frame_uses_package(frame, package): module_name = frame.f_globals.get("__name__", "") if not isinstance(module_name, str): @@ -496,6 +580,9 @@ def _make_network_guard_audit(): return if httpcore_network_active(): return + root = blocked_origin_in_stack(2) + if root is not None: + blocked_error(root) if ( package_in_stack("httpcore", 2) or package_in_stack("anyio", 2) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 26c6600612..a074c422f7 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -239,6 +239,43 @@ def _shell_command_segments(command: str) -> list[list[str]]: 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 _segment_python_index(segment: list[str]) -> int | None: command_index = 0 while command_index < len(segment) and _ASSIGNMENT_RE.match(segment[command_index]): @@ -290,7 +327,10 @@ def _segment_mutates_sandbox_python_env(segment: list[str], python_index: int) - return True lowered = token.lower() if lowered in {"-i", "--ignore-environment"} and before_python: - if os.path.basename(before_python[0]).lower() == "env": + 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: @@ -399,6 +439,44 @@ def _shell_command_substitutions(command: str) -> list[str]: return substitutions +_HEREDOC_START_RE = re.compile(r"<<-?\s*(?P['\"]?)(?P[A-Za-z_][A-Za-z0-9_]*)") + + +def _shell_here_doc_payloads(command: str) -> tuple[list[str], bool]: + if "<<" not in command: + return [], False + lines = command.splitlines() + payloads: list[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 + payloads.append("\n".join(lines[body_start:body_end])) + index = max(index, body_end) + index += 1 + return payloads, malformed + + _SHELL_EXPANSION_TOKEN_RE = re.compile( r"\$(?:[A-Za-z_][A-Za-z0-9_]*|\{[^}]*\}|[0-9#?*$!@_-])|%[A-Za-z_][A-Za-z0-9_]*%" ) @@ -450,7 +528,56 @@ def _static_python_command_argument(node: ast.AST) -> str | None: return None -def _python_payload_launches_startup_bypass(code: str, depth: int) -> bool: +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_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 + 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_launches_startup_bypass( + code: str, depth: int, environment_tainted: bool = False +) -> bool: if depth > 4: return True try: @@ -472,28 +599,34 @@ def _python_payload_launches_startup_bypass(code: str, depth: int) -> bool: for alias in node.names: if alias.name in _PYTHON_CHILD_LAUNCHERS: launcher_aliases.add(alias.asname or alias.name) - elif isinstance(node, ast.Call): - command_node = None - if isinstance(node.func, ast.Name) and node.func.id in launcher_aliases: + if _python_payload_mutates_sandbox_env(node, os_aliases): + environment_tainted = True + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + command_node = None + if isinstance(node.func, ast.Name) and node.func.id in launcher_aliases: + command_node = node.args[0] if node.args else None + 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 None - 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 None - 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 None: - continue - nested = _static_python_command_argument(command_node) - if nested is not None and _sandbox_python_startup_bypasses_guard(nested, depth + 1): - return True + 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 None: + continue + nested = _static_python_command_argument(command_node) + if nested is not None and _sandbox_python_startup_bypasses_guard( + nested, depth + 1, environment_tainted + ): + return True return False @@ -509,14 +642,22 @@ def _segment_python_launch_bypasses_guard( 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) -> bool: +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 - for nested in _shell_command_substitutions(command): - if _sandbox_python_startup_bypasses_guard(nested, depth + 1): + payloads, malformed_here_doc = _shell_here_doc_payloads(command) + if malformed_here_doc: + return True + for payload in payloads: + if _sandbox_python_startup_bypasses_guard(payload, depth + 1, environment_tainted): + return True + command = _shell_command_with_unquoted_newlines_as_separators(command) + for nested in _shell_command_substitutions(command): + if _sandbox_python_startup_bypasses_guard(nested, depth + 1, environment_tainted): return True - environment_tainted = False shell_names = {"bash", "cmd", "cmd.exe", "dash", "fish", "ksh", "sh", "zsh"} for segment in _shell_command_segments(command): first = os.path.basename(segment[0].replace("\\", "/")).lower() @@ -540,17 +681,23 @@ def _sandbox_python_startup_bypasses_guard(command: str, depth: int = 0) -> bool 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): + if _sandbox_python_startup_bypasses_guard( + nested, depth + 1, environment_tainted + ): return True break break if first == "env": for index, token in enumerate(segment): if token in {"-S", "--split-string"} and index + 1 < len(segment): - if _sandbox_python_startup_bypasses_guard(segment[index + 1], depth + 1): + 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): + 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: @@ -2834,6 +2981,20 @@ _RENDER_HTML_REFLECT_SET_START_RE = re.compile( _RENDER_HTML_OBJECT_ASSIGN_START_RE = re.compile( r"\bObject\s*\.\s*assign\s*(?:\?\.\s*)?\(", re.IGNORECASE ) +_RENDER_HTML_JS_STATIC_NAME_START_RE = re.compile( + r"\b(?:const|let|var)\s+(?P[A-Za-z_$][A-Za-z0-9_$]*)\s*=\s*", + re.IGNORECASE, +) +_RENDER_HTML_WITH_BLOCK_RE = re.compile( + r"\bwith\s*\((?:[^()]|\([^()]*\))*\)\s*\{(?P[^{}]*)\}", + re.IGNORECASE | re.DOTALL, +) +_RENDER_HTML_BARE_PROPERTY_ASSIGNMENT_START_RE = re.compile( + r"(?src|href|srcset|action|formaction|poster|data|ping|srcdoc|innerHTML|outerHTML)" + r"\s*(?:\+=|&&=|\|\|=|\?\?=|=(?!=))", + re.IGNORECASE, +) _RENDER_HTML_OBJECT_PROPERTY_START_RE = re.compile( r"(?P['\"]?)(?Psrc|href|srcset|action|formaction|poster|data|ping|" r"srcdoc|innerHTML|outerHTML)(?P=quote)\s*:\s*", @@ -2996,6 +3157,25 @@ def _static_js_assignment_string(expression: str) -> str | None: return None +def _render_html_static_js_name_aliases(code: str) -> dict[str, str]: + aliases: dict[str, str] = {} + for match in _RENDER_HTML_JS_STATIC_NAME_START_RE.finditer(code): + value = _static_js_assignment_string(code[match.end() :]) + if value is not None: + aliases[match.group("name")] = value + return aliases + + +def _render_html_resolve_member(expression: str, aliases: dict[str, str]) -> str | None: + member = _static_js_string(expression) + if member is not None: + return member + name = expression.strip() + if re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*", name): + return aliases.get(name) + return None + + def _render_html_data_document_reaches_network(value: str, depth: int) -> bool: """Inspect executable document payloads embedded in data: URLs.""" if not value.lower().startswith("data:"): @@ -3146,21 +3326,30 @@ def _render_html_assigned_member_reaches_network( return False -def _render_html_set_attribute_arguments(arguments: list[str], method: str, depth: int) -> bool: +def _render_html_value_reaches_network_without_member(value: str | None, depth: int) -> bool: + if value is None: + return True + value = value.lstrip() + return bool( + _RENDER_HTML_URL_LIST_NETWORK_RE.search(value) + or _render_html_data_document_reaches_network(value, depth) + or _render_html_code_reaches_network(value, depth + 1) + ) + + +def _render_html_set_attribute_arguments( + arguments: list[str], method: str, depth: int, aliases: dict[str, str] +) -> bool: if method == "setattributens": name_index, value_index = 1, 2 else: name_index, value_index = 0, 1 if len(arguments) <= value_index: return False - name = _static_js_string(arguments[name_index]) + name = _render_html_resolve_member(arguments[name_index], aliases) value = _static_js_string(arguments[value_index]) if name is None: - return bool( - value is None - or _RENDER_HTML_URL_LIST_NETWORK_RE.search(value.lstrip()) - or _render_html_code_reaches_network(value, depth + 1) - ) + return _render_html_value_reaches_network_without_member(value, depth) name = name.lower().rsplit(":", 1)[-1] return _render_html_assigned_member_reaches_network(name, value, depth) @@ -3181,9 +3370,10 @@ def _render_html_markup_call_reaches_network(arguments: list[str], method: str, def _render_html_computed_network_access(code: str, depth: int = 0) -> bool: + aliases = _render_html_static_js_name_aliases(code) for match in _RENDER_HTML_GLOBAL_BRACKET_RE.finditer(code): expression = match.group(1) - member = _static_js_string(expression) + member = _render_html_resolve_member(expression, aliases) if member is not None: if member.lower() in _RENDER_HTML_NETWORK_MEMBERS: return True @@ -3202,7 +3392,9 @@ def _render_html_computed_network_access(code: str, depth: int = 0) -> bool: arguments = _js_call_arguments(code, match.end()) if arguments is None: return True - if _render_html_set_attribute_arguments(arguments, match.group("method").lower(), depth): + if _render_html_set_attribute_arguments( + arguments, match.group("method").lower(), depth, aliases + ): return True for match in _RENDER_HTML_PROPERTY_ASSIGNMENT_START_RE.finditer(code): @@ -3240,15 +3432,20 @@ def _render_html_computed_network_access(code: str, depth: int = 0) -> bool: return True for match in _RENDER_HTML_COMPUTED_ASSIGNMENT_START_RE.finditer(code): - member = _static_js_string(match.group("member")) - if member is None: - continue value = _static_js_assignment_string(code[match.end() :]) + member_expression = match.group("member") + member = _render_html_resolve_member(member_expression, aliases) + if member is None: + if "." in member_expression: + continue + if _render_html_value_reaches_network_without_member(value, depth): + return True + continue if _render_html_assigned_member_reaches_network(member, value, depth): return True for match in _RENDER_HTML_COMPUTED_CALL_START_RE.finditer(code): - method = _static_js_string(match.group("member")) + method = _render_html_resolve_member(match.group("member"), aliases) if method is None: continue method = method.lower() @@ -3256,7 +3453,7 @@ def _render_html_computed_network_access(code: str, depth: int = 0) -> bool: if arguments is None: return True if method in {"setattribute", "setattributens"}: - if _render_html_set_attribute_arguments(arguments, method, depth): + if _render_html_set_attribute_arguments(arguments, method, depth, aliases): return True elif method == "insertadjacenthtml" or ( method in {"write", "writeln"} and match.group("document") @@ -3270,8 +3467,11 @@ def _render_html_computed_network_access(code: str, depth: int = 0) -> bool: return True if len(arguments) < 3: continue - member = _static_js_string(arguments[1]) + member = _render_html_resolve_member(arguments[1], aliases) if member is None: + value = _static_js_string(arguments[2]) + if _render_html_value_reaches_network_without_member(value, depth): + return True continue value = _static_js_string(arguments[2]) if _render_html_assigned_member_reaches_network(member, value, depth): @@ -3288,6 +3488,17 @@ def _render_html_computed_network_access(code: str, depth: int = 0) -> bool: property_match.group("member"), value, depth ): return True + for match in _RENDER_HTML_WITH_BLOCK_RE.finditer(code): + body = match.group("body") + for assignment in _RENDER_HTML_BARE_PROPERTY_ASSIGNMENT_START_RE.finditer(body): + prefix = body[max(0, assignment.start() - 12) : assignment.start()] + if re.search(r"\b(?:const|let|var)\s+$", prefix, re.IGNORECASE): + continue + value = _static_js_assignment_string(body[assignment.end() :]) + if _render_html_assigned_member_reaches_network( + assignment.group("attr"), value, depth + ): + return True return False diff --git a/studio/backend/tests/test_bypass_permissions.py b/studio/backend/tests/test_bypass_permissions.py index 79f5957bec..20ae6adbb1 100644 --- a/studio/backend/tests/test_bypass_permissions.py +++ b/studio/backend/tests/test_bypass_permissions.py @@ -188,6 +188,15 @@ def test_bash_blocklist_enforced_when_sandboxed(captured_popen): 'python$IFS-S -c "import boto3"', "python -c \"import subprocess; subprocess.run(['python','-S','-c','import boto3'])\"", 'python -c "import os; os.system(\\"python -S -c \'import boto3\'\\")"', + 'echo ok\npython -S -c "import boto3"', + 'timeout 1 env -i python -c "import boto3"', + 'find . -exec env -i python -c "import boto3" ;', + 'bash <<\'EOF\'\npython -S -c "import boto3"\nEOF', + ( + "python -c \"import os,subprocess; os.environ.pop('PYTHONPATH',None); " + "os.environ['UNSLOTH_STUDIO_SANDBOXED']='0'; " + "subprocess.run(['python','-c','import boto3'])\"" + ), ], ) def test_bash_blocks_python_startup_guard_bypasses(captured_popen, command): diff --git a/studio/backend/tests/test_permission_mode.py b/studio/backend/tests/test_permission_mode.py index cde473561e..5c0762a91a 100644 --- a/studio/backend/tests/test_permission_mode.py +++ b/studio/backend/tests/test_permission_mode.py @@ -1138,6 +1138,31 @@ def test_render_html_gated_only_when_networked(): assert rh("") is False assert rh("") is True assert rh("") is False + assert rh("") is True + assert rh("") is False + assert rh("") is False + assert rh("") is True + assert ( + rh( + "" + ) + is True + ) + assert ( + rh("") + is False + ) + assert ( + rh( + "" + ) + is True + ) + assert rh("") is True + assert rh("") is False + assert rh("") is False # A computed bracket key spliced from string fragments on a global host object. assert rh("") is True assert rh("") is True diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index 66aa5f71e1..6d55cb1dce 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -539,6 +539,32 @@ class TestSandboxEnvIsolation: assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr + def test_runtime_import_guard_blocks_aliased_httpcore_origin(self, tmp_path): + from core.inference.tools import _build_safe_env + + code = ( + "import importlib.machinery, importlib.util, sys\n" + "spec = importlib.machinery.PathFinder.find_spec('httpcore')\n" + "alias = importlib.util.spec_from_file_location(\n" + " 'hc', spec.origin,\n" + " submodule_search_locations=list(spec.submodule_search_locations or []),\n" + ")\n" + "module = importlib.util.module_from_spec(alias)\n" + "sys.modules['hc'] = module\n" + "alias.loader.exec_module(module)\n" + "module.request('GET', 'http://127.0.0.1:9')\n" + ) + result = subprocess.run( + [sys.executable, "-c", code], + cwd = tmp_path, + env = _build_safe_env(str(tmp_path)), + capture_output = True, + text = True, + check = False, + ) + assert result.returncode != 0 + assert "Blocked: low-level network module 'httpcore'" in result.stderr + def test_runtime_import_guard_blocks_httpcore_context_flag_tampering(self, tmp_path): from core.inference.tools import _build_safe_env