Harden sandbox guard review bypasses
This commit is contained in:
Michael Han
parent
a37f225cd6
commit
5bf9833d74
6 changed files with 704 additions and 66 deletions
|
|
@ -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))
|
||||
|
|
@ -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 Studio sandbox. This module sits on the
|
||||
|
|
@ -28,7 +28,9 @@ Identical with and without output streaming because the child env is.
|
|||
"""
|
||||
|
||||
import builtins
|
||||
import contextvars
|
||||
import importlib
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
|
|
@ -87,11 +89,31 @@ def _path_is_in_roots(filename, roots):
|
|||
return False
|
||||
|
||||
|
||||
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):
|
||||
module = frame.f_globals.get("__name__", "")
|
||||
root = module.split(".", 1)[0]
|
||||
return root in {"httpx", "httpcore"} and _path_is_in_roots(
|
||||
frame.f_globals.get("__file__"), _TRUSTED_LIBRARY_ROOTS
|
||||
return _frame_uses_trusted_package(frame, "httpx") or _frame_uses_trusted_package(
|
||||
frame, "httpcore"
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -101,25 +123,32 @@ def _trusted_httpx_in_call_stack(skip = 1):
|
|||
except ValueError:
|
||||
return False
|
||||
while frame is not None:
|
||||
module = frame.f_globals.get("__name__", "")
|
||||
if module == "httpx" or module.startswith("httpx."):
|
||||
return _trusted_http_client_frame(frame)
|
||||
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:
|
||||
module = frame.f_globals.get("__name__", "")
|
||||
if (
|
||||
module == __name__
|
||||
or module == "importlib"
|
||||
or module.startswith("importlib.")
|
||||
or module.startswith("_frozen_importlib")
|
||||
frame.f_code.co_filename == __file__
|
||||
or _frame_is_importlib(frame)
|
||||
or str(frame.f_code.co_filename).startswith("<frozen importlib")
|
||||
):
|
||||
frame = frame.f_back
|
||||
continue
|
||||
|
|
@ -177,25 +206,98 @@ class _GuardedHttpcoreModule(types.ModuleType):
|
|||
return types.ModuleType.__delattr__(self, name)
|
||||
|
||||
|
||||
def _make_httpcore_network_gate():
|
||||
active = contextvars.ContextVar("unsloth_httpcore_network_active", default = None)
|
||||
active_markers = set()
|
||||
trusted_request = _trusted_httpx_in_call_stack
|
||||
blocked = _raise_blocked_network_module
|
||||
|
||||
def enter():
|
||||
if not trusted_request(2):
|
||||
blocked("httpcore")
|
||||
marker = object()
|
||||
active_markers.add(id(marker))
|
||||
return active.set(marker), marker
|
||||
|
||||
def exit(state):
|
||||
token, marker = state
|
||||
try:
|
||||
active.reset(token)
|
||||
finally:
|
||||
active_markers.discard(id(marker))
|
||||
|
||||
def is_active():
|
||||
marker = active.get()
|
||||
return marker is not None and id(marker) in active_markers
|
||||
|
||||
return enter, exit, is_active
|
||||
|
||||
|
||||
(
|
||||
_enter_httpcore_network,
|
||||
_exit_httpcore_network,
|
||||
_httpcore_network_is_active,
|
||||
) = _make_httpcore_network_gate()
|
||||
|
||||
|
||||
def _make_httpcore_backend_dispatchers():
|
||||
originals = {}
|
||||
|
||||
def register(original):
|
||||
key = object()
|
||||
originals[key] = original
|
||||
return key
|
||||
|
||||
def dispatch(key, *args, **kwargs):
|
||||
return originals[key](*args, **kwargs)
|
||||
|
||||
async def dispatch_async(key, *args, **kwargs):
|
||||
return await originals[key](*args, **kwargs)
|
||||
|
||||
return register, dispatch, dispatch_async
|
||||
|
||||
|
||||
(
|
||||
_register_httpcore_backend,
|
||||
_dispatch_httpcore_backend,
|
||||
_dispatch_httpcore_backend_async,
|
||||
) = _make_httpcore_backend_dispatchers()
|
||||
|
||||
|
||||
def _guard_httpcore_backend_method(cls, method_name):
|
||||
original = cls.__dict__.get(method_name)
|
||||
if not callable(original) or getattr(original, "_unsloth_httpcore_backend_guard", False):
|
||||
return
|
||||
|
||||
key = _register_httpcore_backend(original)
|
||||
network_enter = _enter_httpcore_network
|
||||
network_exit = _exit_httpcore_network
|
||||
trusted_request = _trusted_httpx_in_call_stack
|
||||
blocked = _raise_blocked_network_module
|
||||
code = getattr(original, "__code__", None)
|
||||
if code is not None and code.co_flags & 0x80: # CO_COROUTINE
|
||||
dispatch = _dispatch_httpcore_backend_async
|
||||
|
||||
async def guarded(*args, **kwargs):
|
||||
if not _trusted_httpx_in_call_stack(2):
|
||||
_raise_blocked_network_module("httpcore")
|
||||
return await original(*args, **kwargs)
|
||||
if not trusted_request(2):
|
||||
blocked("httpcore")
|
||||
state = network_enter()
|
||||
try:
|
||||
return await dispatch(key, *args, **kwargs)
|
||||
finally:
|
||||
network_exit(state)
|
||||
|
||||
else:
|
||||
dispatch = _dispatch_httpcore_backend
|
||||
|
||||
def guarded(*args, **kwargs):
|
||||
if not _trusted_httpx_in_call_stack(2):
|
||||
_raise_blocked_network_module("httpcore")
|
||||
return original(*args, **kwargs)
|
||||
if not trusted_request(2):
|
||||
blocked("httpcore")
|
||||
state = network_enter()
|
||||
try:
|
||||
return dispatch(key, *args, **kwargs)
|
||||
finally:
|
||||
network_exit(state)
|
||||
|
||||
guarded._unsloth_httpcore_backend_guard = True
|
||||
guarded.__name__ = getattr(original, "__name__", method_name)
|
||||
|
|
@ -278,6 +380,136 @@ def _guarded_import_module(name, package = None):
|
|||
_guard_loaded_httpcore_modules()
|
||||
return module
|
||||
|
||||
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
|
||||
|
||||
def guarded(self, *args, **kwargs):
|
||||
fullname = args[0] if args else kwargs.get("fullname", getattr(self, "name", None))
|
||||
root = _blocked_network_module(fullname)
|
||||
if root is not None:
|
||||
_raise_blocked_network_module(root)
|
||||
return original(self, *args, **kwargs)
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _make_network_guard_audit():
|
||||
"""Create a guard whose decisions do not depend on mutable module globals."""
|
||||
blocked = _BLOCKED_NETWORK_MODULES
|
||||
direct_blocked = _DIRECT_BLOCKED_NETWORK_MODULES
|
||||
trusted_roots = _TRUSTED_LIBRARY_ROOTS
|
||||
modules = sys.modules
|
||||
module_getattribute = types.ModuleType.__getattribute__
|
||||
getframe = sys._getframe
|
||||
commonpath = os.path.commonpath
|
||||
realpath = os.path.realpath
|
||||
relpath = os.path.relpath
|
||||
shim_path = realpath(__file__)
|
||||
httpcore_network_active = _httpcore_network_is_active
|
||||
|
||||
def blocked_error(root):
|
||||
raise ModuleNotFoundError(
|
||||
f"Blocked: low-level network module {root!r} is unavailable in sandboxed code"
|
||||
)
|
||||
|
||||
def frame_uses_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 = modules.get(module_name)
|
||||
if module is None:
|
||||
return False
|
||||
try:
|
||||
module_dict = module_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)
|
||||
filename = frame.f_code.co_filename
|
||||
if not isinstance(origin, str) or not isinstance(filename, str):
|
||||
return False
|
||||
try:
|
||||
origin_path = realpath(origin)
|
||||
code_path = realpath(filename)
|
||||
for root in trusted_roots:
|
||||
if commonpath((root, origin_path)) != root:
|
||||
continue
|
||||
if commonpath((root, code_path)) != root:
|
||||
continue
|
||||
origin_relative = relpath(origin_path, root).replace("\\", "/")
|
||||
code_relative = relpath(code_path, root).replace("\\", "/")
|
||||
return (
|
||||
(
|
||||
origin_relative == package
|
||||
or origin_relative.startswith(f"{package}/")
|
||||
)
|
||||
and (code_relative == package or code_relative.startswith(f"{package}/"))
|
||||
)
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
return False
|
||||
|
||||
def package_in_stack(package, skip):
|
||||
try:
|
||||
frame = getframe(skip)
|
||||
except ValueError:
|
||||
return False
|
||||
while frame is not None:
|
||||
if frame_uses_package(frame, package):
|
||||
return True
|
||||
frame = frame.f_back
|
||||
return False
|
||||
|
||||
def sandbox_requested_import(skip):
|
||||
try:
|
||||
frame = getframe(skip)
|
||||
except ValueError:
|
||||
return True
|
||||
while frame is not None:
|
||||
filename = frame.f_code.co_filename
|
||||
if filename == shim_path or (
|
||||
isinstance(filename, str) and filename.startswith("<frozen importlib")
|
||||
):
|
||||
frame = frame.f_back
|
||||
continue
|
||||
return not (
|
||||
frame_uses_package(frame, "httpx") or frame_uses_package(frame, "httpcore")
|
||||
)
|
||||
return True
|
||||
|
||||
def audit(event, args):
|
||||
if event == "import" and args:
|
||||
fullname = args[0]
|
||||
if not isinstance(fullname, str):
|
||||
return
|
||||
root = fullname.split(".", 1)[0]
|
||||
if root in blocked or (
|
||||
root in direct_blocked and sandbox_requested_import(2)
|
||||
):
|
||||
blocked_error(root)
|
||||
return
|
||||
if event not in {"socket.connect", "socket.connect_ex", "socket.getaddrinfo"}:
|
||||
return
|
||||
if httpcore_network_active():
|
||||
return
|
||||
if package_in_stack("httpcore", 2) or package_in_stack(
|
||||
"anyio", 2
|
||||
) or package_in_stack("trio", 2):
|
||||
blocked_error("httpcore")
|
||||
|
||||
return audit
|
||||
|
||||
|
||||
class _BlockedNetworkModuleFinder:
|
||||
_unsloth_blocked_network_guard = True
|
||||
|
|
@ -336,33 +568,13 @@ def _sandbox_guard_should_activate():
|
|||
|
||||
def _install_import_guard():
|
||||
global _import_guard_installed
|
||||
if not _sandbox_guard_should_activate():
|
||||
if __name__ != "sitecustomize" or not _sandbox_guard_should_activate():
|
||||
return
|
||||
if not _import_guard_installed:
|
||||
# Capture the block sets and trust probe as closure locals. Audit hooks
|
||||
# cannot be removed once registered, so this hook is the backstop for the
|
||||
# import wrapper and meta-path finder (both of which sandbox code can
|
||||
# restore/detach). Reading globals here would let sandbox code neutralise
|
||||
# it by rebinding this module's attributes, so the decision is frozen.
|
||||
blocked_roots = frozenset(_BLOCKED_NETWORK_MODULES)
|
||||
direct_roots = frozenset(_DIRECT_BLOCKED_NETWORK_MODULES)
|
||||
sandbox_requested = _sandbox_code_requested_import
|
||||
|
||||
def _immutable_network_import_audit(event, args):
|
||||
if event != "import" or not args:
|
||||
return
|
||||
name = args[0]
|
||||
if not isinstance(name, str):
|
||||
return
|
||||
root = name.split(".", 1)[0]
|
||||
if root in blocked_roots or (root in direct_roots and sandbox_requested()):
|
||||
raise ModuleNotFoundError(
|
||||
f"Blocked: low-level network module {root!r} is unavailable in sandboxed code"
|
||||
)
|
||||
|
||||
sys.addaudithook(_immutable_network_import_audit)
|
||||
sys.addaudithook(_make_network_guard_audit())
|
||||
builtins.__import__ = _guarded_import
|
||||
importlib.import_module = _guarded_import_module
|
||||
_guard_legacy_source_loader()
|
||||
_import_guard_installed = True
|
||||
if any(getattr(finder, "_unsloth_blocked_network_guard", False) for finder in sys.meta_path):
|
||||
return
|
||||
|
|
|
|||
|
|
@ -316,10 +316,211 @@ def _segment_persistently_mutates_sandbox_python_env(segment: list[str]) -> bool
|
|||
return False
|
||||
|
||||
|
||||
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
|
||||
if not 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
|
||||
|
||||
|
||||
_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"})
|
||||
|
||||
|
||||
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
|
||||
replacement = _SHELL_EXPANSION_TOKEN_RE.sub(" ", token).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 _static_python_command_argument(node: ast.AST) -> str | None:
|
||||
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||||
return node.value
|
||||
if isinstance(node, (ast.List, ast.Tuple)):
|
||||
parts: list[str] = []
|
||||
for element in node.elts:
|
||||
if not isinstance(element, ast.Constant) or not isinstance(element.value, str):
|
||||
return None
|
||||
parts.append(element.value)
|
||||
return shlex.join(parts)
|
||||
return None
|
||||
|
||||
|
||||
def _python_payload_launches_startup_bypass(code: str, depth: int) -> bool:
|
||||
if depth > 4:
|
||||
return True
|
||||
try:
|
||||
tree = ast.parse(code)
|
||||
except SyntaxError:
|
||||
return False
|
||||
subprocess_aliases = {"subprocess"}
|
||||
os_aliases = {"os"}
|
||||
launcher_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 isinstance(node, ast.ImportFrom):
|
||||
if node.module == "subprocess":
|
||||
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:
|
||||
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
|
||||
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) -> 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):
|
||||
return True
|
||||
environment_tainted = False
|
||||
shell_names = {"bash", "cmd", "cmd.exe", "dash", "fish", "ksh", "sh", "zsh"}
|
||||
for segment in _shell_command_segments(command):
|
||||
|
|
@ -356,11 +557,18 @@ def _sandbox_python_startup_bypasses_guard(command: str, depth: int = 0) -> bool
|
|||
elif token.startswith("--split-string="):
|
||||
if _sandbox_python_startup_bypasses_guard(token.split("=", 1)[1], depth + 1):
|
||||
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
|
||||
python_index = _segment_python_index(segment)
|
||||
if python_index is not None:
|
||||
if environment_tainted or _segment_mutates_sandbox_python_env(segment, python_index):
|
||||
return True
|
||||
if _python_flags_skip_sitecustomize(segment[python_index + 1 :]):
|
||||
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)
|
||||
|
|
@ -488,6 +696,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
|
||||
|
|
@ -831,6 +1040,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",
|
||||
|
|
@ -2605,11 +2815,12 @@ _RENDER_HTML_MARKUP_ASSIGNMENT_START_RE = re.compile(
|
|||
)
|
||||
_RENDER_HTML_MARKUP_CALL_START_RE = re.compile(
|
||||
r"(?:\.\s*(?P<insert>insertAdjacentHTML)|"
|
||||
r"\.\s*(?P<contextual>createContextualFragment)|"
|
||||
# document.write / writeln, optionally reached through a document-valued
|
||||
# receiver such as document.open(): document.open().write('<img src=...>')
|
||||
# returns the same document and inserts the remote-loading markup.
|
||||
r"\bdocument\s*(?:(?:\?\.\s*|\.\s*)open\s*\([^()]*\)\s*)?"
|
||||
r"(?:\?\.\s*|\.\s*)(?P<write>write|writeln))"
|
||||
r"\s*(?:\?\.\s*|\.\s*)(?P<write>write|writeln))"
|
||||
r"\s*(?:\?\.\s*)?\(",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
|
@ -2618,7 +2829,8 @@ _RENDER_HTML_COMPUTED_ASSIGNMENT_START_RE = re.compile(
|
|||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_RENDER_HTML_COMPUTED_CALL_START_RE = re.compile(
|
||||
r"(?:(?P<document>\bdocument)\s*(?:\?\.\s*)?)?\[\s*(?P<member>[^\]]+)\s*\]\s*(?:\?\.\s*)?\(",
|
||||
r"(?:(?P<document>\bdocument)\s*(?:\?\.\s*)?)?"
|
||||
r"\[\s*(?P<member>[^\]]+)\s*\]\s*(?:\?\.\s*)?\(",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_RENDER_HTML_REFLECT_SET_START_RE = re.compile(
|
||||
|
|
@ -2632,6 +2844,16 @@ _RENDER_HTML_OBJECT_PROPERTY_START_RE = re.compile(
|
|||
r"srcdoc|innerHTML|outerHTML)(?P=quote)\s*:\s*",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_RENDER_HTML_DESTRUCTURING_ASSIGNMENT_START_RE = re.compile(
|
||||
r"(?:\[[^\]]*(?:\.\s*|\[\s*['\"`])"
|
||||
r"(?:src|href|srcset|action|formaction|poster|data|ping|srcdoc|innerHTML|outerHTML)"
|
||||
r"[^\]]*\]|"
|
||||
r"\{[^{}]*(?:src|href|srcset|action|formaction|poster|data|ping|srcdoc|innerHTML|"
|
||||
r"outerHTML)\s*:\s*[^{}]*(?:\.\s*|\[\s*['\"`])"
|
||||
r"(?:src|href|srcset|action|formaction|poster|data|ping|srcdoc|innerHTML|outerHTML)"
|
||||
r"[^{}]*\})\s*=",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_RENDER_HTML_NETWORK_MEMBERS = frozenset(
|
||||
{
|
||||
"fetch",
|
||||
|
|
@ -2649,6 +2871,7 @@ _RENDER_HTML_NETWORK_ATTRIBUTES = frozenset(
|
|||
)
|
||||
_RENDER_HTML_URL_LIST_ATTRIBUTES = frozenset({"srcset", "ping"})
|
||||
_RENDER_HTML_URL_LIST_NETWORK_RE = re.compile(r"(?:^|[\s,])(?:https?:|/)", re.IGNORECASE)
|
||||
_RENDER_HTML_JS_NETWORK_URL_RE = re.compile(r"(?:^|[\s,:'\"`\[(])(?:https?:|/)", re.IGNORECASE)
|
||||
_RENDER_HTML_ACTIVE_DATA_MIME_TYPES = frozenset(
|
||||
{"text/html", "application/xhtml+xml", "image/svg+xml"}
|
||||
)
|
||||
|
|
@ -2789,19 +3012,22 @@ def _render_html_data_document_reaches_network(value: str, depth: int) -> bool:
|
|||
media_type = (parts[0] or "text/plain").lower()
|
||||
if media_type not in _RENDER_HTML_ACTIVE_DATA_MIME_TYPES:
|
||||
return False
|
||||
# Honour a declared charset so a UTF-16/Latin-1 document decodes the same way
|
||||
# the browser would; an unknown/undecodable charset fails closed rather than
|
||||
# letting a mangled UTF-8 read hide a remote load.
|
||||
# Decode declared charset so data documents match browser behavior.
|
||||
charset = "utf-8"
|
||||
for part in parts[1:]:
|
||||
if part.lower().startswith("charset="):
|
||||
charset = part.split("=", 1)[1].strip() or "utf-8"
|
||||
for parameter in parts[1:]:
|
||||
name, equals, parameter_value = parameter.partition("=")
|
||||
if equals and name.strip().lower() == "charset":
|
||||
charset = urllib.parse.unquote(parameter_value.strip().strip("\"'"))
|
||||
if not charset:
|
||||
return True
|
||||
break
|
||||
try:
|
||||
encoding = codecs.lookup(charset).name
|
||||
payload_bytes = urllib.parse.unquote_to_bytes(payload)
|
||||
if any(part.lower() == "base64" for part in parts[1:]):
|
||||
payload_bytes = base64.b64decode(b"".join(payload_bytes.split()), validate = True)
|
||||
markup = payload_bytes.decode(charset, errors = "replace")
|
||||
except (ValueError, binascii.Error, LookupError):
|
||||
markup = payload_bytes.decode(encoding)
|
||||
except (LookupError, UnicodeError, ValueError, binascii.Error):
|
||||
return True
|
||||
return _render_html_code_reaches_network(markup, depth + 1)
|
||||
|
||||
|
|
@ -2887,6 +3113,33 @@ def _js_call_arguments(code: str, offset: int) -> list[str] | None:
|
|||
return None
|
||||
|
||||
|
||||
def _js_assignment_expression(code: str) -> str | None:
|
||||
stack: list[str] = []
|
||||
quote: str | None = None
|
||||
escaped = False
|
||||
pairs = {")": "(", "]": "[", "}": "{"}
|
||||
for index, char in enumerate(code):
|
||||
if quote is not None:
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif char == "\\":
|
||||
escaped = True
|
||||
elif char == quote:
|
||||
quote = None
|
||||
continue
|
||||
if char in "\"'`":
|
||||
quote = char
|
||||
elif char in "([{":
|
||||
stack.append(char)
|
||||
elif char in ")]}":
|
||||
if not stack or stack[-1] != pairs[char]:
|
||||
return code[:index]
|
||||
stack.pop()
|
||||
elif char in ";\n\r<" and not stack:
|
||||
return code[:index]
|
||||
return code if quote is None and not stack else None
|
||||
|
||||
|
||||
def _render_html_assigned_member_reaches_network(
|
||||
member: str, value: str | None, depth: int
|
||||
) -> bool:
|
||||
|
|
@ -2976,10 +3229,19 @@ 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
|
||||
method = (match.group("insert") or match.group("write")).lower()
|
||||
method = (match.group("insert") or match.group("contextual") or match.group("write")).lower()
|
||||
if _render_html_markup_call_reaches_network(arguments, method, depth):
|
||||
return True
|
||||
|
||||
for match in _RENDER_HTML_DESTRUCTURING_ASSIGNMENT_START_RE.finditer(code):
|
||||
value = _js_assignment_expression(code[match.end() :])
|
||||
if value is None:
|
||||
return True
|
||||
if _RENDER_HTML_JS_NETWORK_URL_RE.search(value) or _render_html_code_reaches_network(
|
||||
value, depth + 1
|
||||
):
|
||||
return True
|
||||
|
||||
for match in _RENDER_HTML_COMPUTED_ASSIGNMENT_START_RE.finditer(code):
|
||||
member = _static_js_string(match.group("member"))
|
||||
if member is None:
|
||||
|
|
@ -3141,7 +3403,7 @@ def _build_safe_env(workdir: str) -> dict[str, str]:
|
|||
"TERM": "dumb",
|
||||
"PYTHONIOENCODING": "utf-8",
|
||||
"UNSLOTH_STUDIO_SANDBOXED": "1",
|
||||
# sitecustomize shim: remaps ChatGPT code-interpreter paths (/mnt/data
|
||||
# sitecustomize shim: remaps code-interpreter paths (/mnt/data
|
||||
# etc.) onto the sandbox CWD; see sandbox_site/sitecustomize.py.
|
||||
"PYTHONPATH": _SANDBOX_SITE_DIR,
|
||||
}
|
||||
|
|
@ -3330,7 +3592,7 @@ def _build_bypass_env(workdir: str) -> dict[str, str]:
|
|||
# operator's PYTHONPATH, so prepend rather than replace.
|
||||
inherited_pythonpath = env.get("PYTHONPATH", "")
|
||||
env["PYTHONPATH"] = os.pathsep.join(
|
||||
part for part in (_SANDBOX_SITE_DIR, inherited_pythonpath) if part
|
||||
part for part in (_BYPASS_SITE_DIR, inherited_pythonpath) if part
|
||||
)
|
||||
# Windows SDKs read creds under the profile dirs, not $HOME; repoint set
|
||||
# ones to the workdir (HOMEDRIVE/HOMEPATH are dropped above).
|
||||
|
|
@ -3557,7 +3819,7 @@ WEB_SEARCH_TOOL = {
|
|||
}
|
||||
|
||||
# Appended to the python/terminal descriptions: models habitually write to
|
||||
# /mnt/data (a ChatGPT code-interpreter path), which does not exist here.
|
||||
# /mnt/data (a code-interpreter path), which does not exist here.
|
||||
_SANDBOX_PATHS_NOTE = (
|
||||
" Read and write files using relative paths in the current working "
|
||||
"directory, which persists for this conversation; absolute paths like "
|
||||
|
|
@ -5826,6 +6088,8 @@ def _check_signal_escape_patterns(code: str):
|
|||
self.importlib_aliases = {"importlib"}
|
||||
self.builtins_aliases = {"builtins", "__builtins__"}
|
||||
self.import_loader_aliases = {"__import__"}
|
||||
self.source_loader_aliases = {"SourceFileLoader"}
|
||||
self.source_loader_names: dict[str, str] = {}
|
||||
self.literal_strings: dict[str, str] = {}
|
||||
|
||||
def _block_low_level_network_module(self, module_name: str, node) -> None:
|
||||
|
|
@ -5914,6 +6178,22 @@ def _check_signal_escape_patterns(code: str):
|
|||
}
|
||||
return False
|
||||
|
||||
def _is_source_file_loader(self, node) -> bool:
|
||||
if isinstance(node, ast.Name):
|
||||
return node.id in self.source_loader_aliases
|
||||
return isinstance(node, ast.Attribute) and node.attr == "SourceFileLoader"
|
||||
|
||||
def _source_loader_module_name(self, node) -> str | None:
|
||||
if isinstance(node, ast.Name):
|
||||
return self.source_loader_names.get(node.id)
|
||||
if (
|
||||
isinstance(node, ast.Call)
|
||||
and self._is_source_file_loader(node.func)
|
||||
and node.args
|
||||
):
|
||||
return self._static_string(node.args[0])
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _target_names(node) -> list[str]:
|
||||
if isinstance(node, ast.Name):
|
||||
|
|
@ -5944,6 +6224,12 @@ def _check_signal_escape_patterns(code: str):
|
|||
self.importlib_aliases.update(names)
|
||||
if isinstance(value, ast.Name) and value.id in self.builtins_aliases:
|
||||
self.builtins_aliases.update(names)
|
||||
loader_name = self._source_loader_module_name(value)
|
||||
for name in names:
|
||||
if loader_name is None:
|
||||
self.source_loader_names.pop(name, None)
|
||||
else:
|
||||
self.source_loader_names[name] = loader_name
|
||||
|
||||
def visit_Import(self, node):
|
||||
for alias in node.names:
|
||||
|
|
@ -5962,6 +6248,8 @@ def _check_signal_escape_patterns(code: str):
|
|||
self.import_loader_aliases.add(bound)
|
||||
elif node.module == "builtins" and alias.name == "__import__":
|
||||
self.import_loader_aliases.add(bound)
|
||||
elif node.module == "importlib.machinery" and alias.name == "SourceFileLoader":
|
||||
self.source_loader_aliases.add(bound)
|
||||
|
||||
def visit_Assign(self, node):
|
||||
self._bind_assignment(node.targets, node.value)
|
||||
|
|
@ -5983,6 +6271,14 @@ def _check_signal_escape_patterns(code: str):
|
|||
module_name = self._static_string(module_node)
|
||||
if module_name is not None:
|
||||
self._block_low_level_network_module(module_name, node)
|
||||
if isinstance(node.func, ast.Attribute) and node.func.attr == "load_module":
|
||||
module_name = None
|
||||
if node.args:
|
||||
module_name = self._static_string(node.args[0])
|
||||
if module_name is None:
|
||||
module_name = self._source_loader_module_name(node.func.value)
|
||||
if module_name is not None:
|
||||
self._block_low_level_network_module(module_name, node)
|
||||
|
||||
parts: list[str] = []
|
||||
cur = node.func
|
||||
|
|
@ -6255,7 +6551,7 @@ def _truncate(text: str, limit: int = _MAX_OUTPUT_CHARS) -> str:
|
|||
return text
|
||||
|
||||
|
||||
# ChatGPT code-interpreter path conventions models write out of habit; none
|
||||
# Code-interpreter path conventions models write out of habit; none
|
||||
# exist in the Studio sandbox, so a failure on one earns the retry hint.
|
||||
_MISSING_PATH_PREFIXES = (
|
||||
"/mnt/data",
|
||||
|
|
|
|||
|
|
@ -185,6 +185,9 @@ def test_bash_blocklist_enforced_when_sandboxed(captured_popen):
|
|||
# still be recursed into, not left as an opaque exec target.
|
||||
'find . -exec sh -c "python -S -c import\\ boto3" ;',
|
||||
'find . -type f -execdir bash -c "python -I -c import\\ boto3" ;',
|
||||
'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'\\\")\"",
|
||||
],
|
||||
)
|
||||
def test_bash_blocks_python_startup_guard_bypasses(captured_popen, command):
|
||||
|
|
@ -200,6 +203,7 @@ def test_bash_blocks_python_startup_guard_bypasses(captured_popen, command):
|
|||
"python script.py -S",
|
||||
"echo python -S",
|
||||
"python -c \"print('-S')\"",
|
||||
"python -c \"import subprocess; subprocess.run(['python','-c','print(1)'])\"",
|
||||
],
|
||||
)
|
||||
def test_bash_allows_python_without_startup_guard_bypass(captured_popen, command):
|
||||
|
|
|
|||
|
|
@ -1123,6 +1123,24 @@ def test_render_html_gated_only_when_networked():
|
|||
is True
|
||||
)
|
||||
assert rh("<script>document.open().write('<p>Local</p>')</script>") is False
|
||||
assert (
|
||||
rh(
|
||||
"<script>document.createRange().createContextualFragment("
|
||||
"'<img src=https://evil/x>')</script>"
|
||||
)
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
rh(
|
||||
"<script>document.createRange().createContextualFragment("
|
||||
"'<p>Local</p>')</script>"
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert rh("<script>[img.src] = ['https://evil/x']</script>") is True
|
||||
assert rh("<script>[img.src] = ['./local.png']</script>") is False
|
||||
assert rh("<script>({src: img.src} = {src:'https://evil/x'})</script>") is True
|
||||
assert rh("<script>({src: img.src} = {src:'./local.png'})</script>") is False
|
||||
# A computed bracket key spliced from string fragments on a global host object.
|
||||
assert rh("<script>window['fet'+'ch']('https://attacker.example')</script>") is True
|
||||
assert rh("<script>self['open' + '']('https://x')</script>") is True
|
||||
|
|
|
|||
|
|
@ -149,6 +149,12 @@ class TestLowLevelNetworkModules:
|
|||
("import importlib; print(vars(importlib)['import_module']('httpcore').__name__)"),
|
||||
("import importlib; print(importlib.__dict__['import_module']('boto3').__name__)"),
|
||||
("import builtins; print(getattr(builtins, '__import__')('botocore').__name__)"),
|
||||
(
|
||||
"import importlib.machinery, importlib.util\n"
|
||||
"spec = importlib.util.find_spec('httpcore')\n"
|
||||
"loader = importlib.machinery.SourceFileLoader('httpcore', spec.origin)\n"
|
||||
"loader.load_module()"
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_low_level_client_blocked(self, code):
|
||||
|
|
@ -490,6 +496,101 @@ class TestSandboxEnvIsolation:
|
|||
assert bypass.returncode == 0, bypass.stderr
|
||||
assert bypass.stdout.strip() == "httpcore"
|
||||
|
||||
def test_runtime_import_guard_rejects_spoofed_httpx_globals(self, tmp_path):
|
||||
from core.inference.tools import _build_safe_env
|
||||
|
||||
code = (
|
||||
"import httpx\n"
|
||||
"__name__ = 'httpx._client'\n"
|
||||
"__file__ = httpx.__file__\n"
|
||||
"name = ''.join(['http', 'core'])\n"
|
||||
"module = __import__(name)\n"
|
||||
"print(module.__name__)\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_legacy_loader_httpcore(self, tmp_path):
|
||||
from core.inference.tools import _build_safe_env
|
||||
|
||||
code = (
|
||||
"import importlib.machinery, importlib.util\n"
|
||||
"name = ''.join(['http', 'core'])\n"
|
||||
"spec = importlib.util.find_spec(name)\n"
|
||||
"loader = importlib.machinery.SourceFileLoader(name, spec.origin)\n"
|
||||
"loader.load_module()\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
|
||||
|
||||
code = (
|
||||
"import sys, types\n"
|
||||
"import httpx, sitecustomize\n"
|
||||
"client = httpx.Client()\n"
|
||||
"client.close()\n"
|
||||
"module = sys.modules['httpcore._backends.sync']\n"
|
||||
"module_dict = types.ModuleType.__getattribute__(module, '__dict__')\n"
|
||||
"backend_type = module_dict['SyncBackend']\n"
|
||||
"guarded = type.__getattribute__(backend_type, '__dict__')['connect_tcp']\n"
|
||||
"dispatch = key = None\n"
|
||||
"for cell in guarded.__closure__ or ():\n"
|
||||
" value = cell.cell_contents\n"
|
||||
" if callable(value) and getattr(value, '__name__', '') == 'dispatch':\n"
|
||||
" dispatch = value\n"
|
||||
" elif type(value) is object:\n"
|
||||
" key = value\n"
|
||||
"originals = None\n"
|
||||
"for cell in dispatch.__closure__ or ():\n"
|
||||
" value = cell.cell_contents\n"
|
||||
" if type(value) is dict:\n"
|
||||
" originals = value\n"
|
||||
"original = originals[key]\n"
|
||||
"tokens = []\n"
|
||||
"for value in vars(sitecustomize).values():\n"
|
||||
" for cell in getattr(value, '__closure__', ()) or ():\n"
|
||||
" content = cell.cell_contents\n"
|
||||
" if type(content).__name__ == 'ContextVar':\n"
|
||||
" tokens.append((content, content.set(True)))\n"
|
||||
"assert tokens\n"
|
||||
"try:\n"
|
||||
" original(\n"
|
||||
" backend_type(), '127.0.0.1', 9,\n"
|
||||
" timeout=0.01, local_address=None, socket_options=None,\n"
|
||||
" )\n"
|
||||
"finally:\n"
|
||||
" for content, token in reversed(tokens):\n"
|
||||
" content.reset(token)\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_local_module_httpcore_import(self, tmp_path):
|
||||
from core.inference.tools import _build_safe_env
|
||||
|
||||
|
|
@ -624,20 +725,19 @@ class TestSandboxEnvIsolation:
|
|||
assert env["TERM"] == "dumb"
|
||||
|
||||
def test_bypass_env_installs_sitecustomize_path_shim(self, tmp_path):
|
||||
# Bypass mode must install the same /mnt/data path-remap shim as the safe
|
||||
# env (finding 17), else /mnt/data writes work only in normal mode.
|
||||
from core.inference.tools import _SANDBOX_SITE_DIR, _build_bypass_env
|
||||
# Bypass mode keeps path remapping without installing network guards.
|
||||
from core.inference.tools import _BYPASS_SITE_DIR, _build_bypass_env
|
||||
env = _build_bypass_env(str(tmp_path))
|
||||
assert _SANDBOX_SITE_DIR in env["PYTHONPATH"].split(os.pathsep)
|
||||
assert _BYPASS_SITE_DIR in env["PYTHONPATH"].split(os.pathsep)
|
||||
|
||||
def test_bypass_env_prepends_shim_and_keeps_inherited_pythonpath(self, monkeypatch, tmp_path):
|
||||
from core.inference.tools import _SANDBOX_SITE_DIR, _build_bypass_env
|
||||
from core.inference.tools import _BYPASS_SITE_DIR, _build_bypass_env
|
||||
|
||||
monkeypatch.setenv("PYTHONPATH", "/operator/libs")
|
||||
env = _build_bypass_env(str(tmp_path))
|
||||
parts = env["PYTHONPATH"].split(os.pathsep)
|
||||
# Shim first so its open()/makedirs remap wins, operator entries kept.
|
||||
assert parts[0] == _SANDBOX_SITE_DIR
|
||||
assert parts[0] == _BYPASS_SITE_DIR
|
||||
assert "/operator/libs" in parts
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue