Cover nested canvas and dynamic import paths

studio/backend/core/inference/sandbox_site/sitecustomize.py

@@ -28,6 +28,8 @@ Identical with and without output streaming because the child env is.
 """
 
 import builtins
+import importlib
+import importlib.util
 import io
 import json
 import os
@@ -47,14 +49,102 @@ _remapped_writes: dict = {}
 # 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 _path_is_in_sandbox(filename):
+    if not isinstance(filename, str) or filename.startswith("<"):
+        return False
+    try:
+        cwd = os.path.realpath(os.getcwd())
+        path = os.path.realpath(filename)
+        return os.path.commonpath((cwd, path)) == cwd
+    except (OSError, ValueError):
+        return False
+
+
+def _sandbox_code_requested_import():
+    try:
+        frame = sys._getframe(1)
+    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 = frame.f_back
+            continue
+        if module == "__main__" or _path_is_in_sandbox(frame.f_globals.get("__file__")):
+            return True
+        return False
+    return True
 
 
 def _blocked_network_module(fullname):
     if not isinstance(fullname, str):
         return None
     root = fullname.split(".", 1)[0]
-    return root if root in _BLOCKED_NETWORK_MODULES else None
+    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"
+    )
+
+
+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
+    root = _blocked_network_module(_absolute_import_name(name, package, level))
+    if root is not None:
+        _raise_blocked_network_module(root)
+    return _original_import(name, globals, locals, fromlist, level)
+
+
+def _guarded_import_module(name, package = None):
+    root = _blocked_network_module(_absolute_import_name(name, package))
+    if root is not None:
+        _raise_blocked_network_module(root)
+    return _original_import_module(name, package)
 
 
 def _network_import_audit(event, args):
@@ -62,9 +152,7 @@ def _network_import_audit(event, args):
         return
     root = _blocked_network_module(args[0])
     if root is not None:
-        raise ModuleNotFoundError(
-            f"Blocked: low-level network module {root!r} is unavailable in sandboxed code"
-        )
+        _raise_blocked_network_module(root)
 
 
 class _BlockedNetworkModuleFinder:
@@ -78,9 +166,7 @@ class _BlockedNetworkModuleFinder:
     ):
         root = _blocked_network_module(fullname)
         if root is not None:
-            raise ModuleNotFoundError(
-                f"Blocked: low-level network module {root!r} is unavailable in sandboxed code"
-            )
+            _raise_blocked_network_module(root)
         return None
 
 
@@ -90,6 +176,8 @@ def _install_import_guard():
         return
     if not _import_guard_installed:
         sys.addaudithook(_network_import_audit)
+        builtins.__import__ = _guarded_import
+        importlib.import_module = _guarded_import_module
         _import_guard_installed = True
     if any(getattr(finder, "_unsloth_blocked_network_guard", False) for finder in sys.meta_path):
         return

studio/backend/core/inference/tools.py

@@ -2426,17 +2426,29 @@ _RENDER_HTML_NETWORK_RE = re.compile(
 # an https:// URL and hide a real load.
 _JS_BLOCK_COMMENT_RE = re.compile(r"/\*.*?\*/", re.DOTALL)
 _RENDER_HTML_GLOBAL_BRACKET_RE = re.compile(
-    r"\b(?:window|self|globalThis|top|parent|frames|this)\s*(?:\?\.\s*)?\[([^\]]*)\]",
+    r"\b(?:window|self|globalThis|top|parent|frames|this|navigator|"
+    r"document\s*\.\s*defaultView)\s*(?:\?\.\s*)?"
+    r"(?:\[\s*\d+\s*\]\s*(?:\?\.\s*)?)*\[([^\]]*)\]",
     re.IGNORECASE | re.DOTALL,
 )
 _RENDER_HTML_SET_ATTRIBUTE_START_RE = re.compile(
-    r"\.\s*setAttribute\s*(?:\?\.\s*)?\(",
+    r"\.\s*(?P<method>setAttribute(?:NS)?)\s*(?:\?\.\s*)?\(",
     re.IGNORECASE,
 )
 _RENDER_HTML_PROPERTY_ASSIGNMENT_START_RE = re.compile(
     r"\.\s*(?P<attr>src|href|srcset|action|formaction|poster|data|ping)\s*=(?!=)",
     re.IGNORECASE,
 )
+_RENDER_HTML_MARKUP_ASSIGNMENT_START_RE = re.compile(
+    r"\.\s*(?:innerHTML|outerHTML)\s*=(?!=)",
+    re.IGNORECASE,
+)
+_RENDER_HTML_MARKUP_CALL_START_RE = re.compile(
+    r"(?:\.\s*(?P<insert>insertAdjacentHTML)|"
+    r"\bdocument\s*(?:\?\.\s*|\.\s*)(?P<write>write|writeln))"
+    r"\s*(?:\?\.\s*)?\(",
+    re.IGNORECASE,
+)
 _RENDER_HTML_NETWORK_MEMBERS = frozenset(
     {
         "fetch",
@@ -2472,16 +2484,66 @@ def _leading_js_string(expression: str) -> tuple[str, int] | None:
             i += 1
             if i >= len(expression):
                 return None
-            if expression[i] not in ("\\", '"', "'", "`"):
-                return None
-            value.append(expression[i])
+            escape = expression[i]
+            if escape in ("\\", "/", '"', "'", "`"):
+                value.append(escape)
+                i += 1
+                continue
+            escapes = {
+                "b": "\b",
+                "f": "\f",
+                "n": "\n",
+                "r": "\r",
+                "t": "\t",
+                "v": "\v",
+            }
+            if escape in escapes:
+                value.append(escapes[escape])
+                i += 1
+                continue
+            if escape == "0":
+                if i + 1 < len(expression) and expression[i + 1].isdigit():
+                    return None
+                value.append("\0")
+                i += 1
+                continue
+            if escape == "x":
+                digits = expression[i + 1 : i + 3]
+                if len(digits) != 2 or not re.fullmatch(r"[0-9a-fA-F]{2}", digits):
+                    return None
+                value.append(chr(int(digits, 16)))
+                i += 3
+                continue
+            if escape == "u":
+                if i + 1 < len(expression) and expression[i + 1] == "{":
+                    end = expression.find("}", i + 2)
+                    digits = expression[i + 2 : end] if end != -1 else ""
+                    if not digits or not re.fullmatch(r"[0-9a-fA-F]{1,6}", digits):
+                        return None
+                    codepoint = int(digits, 16)
+                    if codepoint > 0x10FFFF:
+                        return None
+                    value.append(chr(codepoint))
+                    i = end + 1
+                    continue
+                digits = expression[i + 1 : i + 5]
+                if len(digits) != 4 or not re.fullmatch(r"[0-9a-fA-F]{4}", digits):
+                    return None
+                value.append(chr(int(digits, 16)))
+                i += 5
+                continue
+            if escape in ("\n", "\r"):
+                if escape == "\r" and i + 1 < len(expression) and expression[i + 1] == "\n":
+                    i += 1
+                i += 1
+                continue
+            value.append(escape)
             i += 1
             continue
         if char == quote:
-            text = "".join(value)
-            if quote == "`" and "${" in text:
-                return None
-            return text, i + 1
+            return "".join(value), i + 1
+        if quote == "`" and char == "$" and i + 1 < len(expression) and expression[i + 1] == "{":
+            return None
         value.append(char)
         i += 1
     return None
@@ -2518,6 +2580,18 @@ def _static_js_string(expression: str) -> str | None:
     return value
 
 
+def _static_js_assignment_string(expression: str) -> str | None:
+    """Fold a static assignment value and reject trailing transformations."""
+    parsed = _static_js_string_prefix(expression)
+    if parsed is None:
+        return None
+    value, end = parsed
+    rest = expression[end:].lstrip()
+    if not rest or rest.startswith("//") or rest[0] in ";,)]}\n\r<":
+        return value
+    return None
+
+
 def _render_html_attribute_reaches_network(name: str, value: str | None) -> bool:
     if value is None:
         return False
@@ -2528,13 +2602,20 @@ def _render_html_attribute_reaches_network(name: str, value: str | None) -> bool
 
 
 class _RenderHtmlAttributeParser(HTMLParser):
-    def __init__(self):
+    def __init__(self, depth: int):
         super().__init__(convert_charrefs = True)
+        self.depth = depth
         self.reaches_network = False
 
     def handle_starttag(self, tag, attrs):
         for name, value in attrs:
             name = name.lower()
+            if name == "srcdoc" and value:
+                if self.depth >= 8 or _render_html_code_reaches_network(value, self.depth + 1):
+                    self.reaches_network = True
+                    return
+            if name == "xlink:href":
+                name = "href"
             if name in _RENDER_HTML_NETWORK_ATTRIBUTES and _render_html_attribute_reaches_network(
                 name, value
             ):
@@ -2542,8 +2623,8 @@ class _RenderHtmlAttributeParser(HTMLParser):
                 return
 
 
-def _render_html_attributes_reach_network(code: str) -> bool:
-    parser = _RenderHtmlAttributeParser()
+def _render_html_attributes_reach_network(code: str, depth: int = 0) -> bool:
+    parser = _RenderHtmlAttributeParser(depth)
     try:
         parser.feed(code)
         parser.close()
@@ -2586,7 +2667,7 @@ def _js_call_arguments(code: str, offset: int) -> list[str] | None:
     return None
 
 
-def _render_html_computed_network_access(code: str) -> bool:
+def _render_html_computed_network_access(code: str, depth: int = 0) -> bool:
     for match in _RENDER_HTML_GLOBAL_BRACKET_RE.finditer(code):
         expression = match.group(1)
         member = _static_js_string(expression)
@@ -2608,42 +2689,75 @@ def _render_html_computed_network_access(code: str) -> bool:
         arguments = _js_call_arguments(code, match.end())
         if arguments is None:
             return True
-        if len(arguments) < 2:
+        if match.group("method").lower() == "setattributens":
+            name_index, value_index = 1, 2
+        else:
+            name_index, value_index = 0, 1
+        if len(arguments) <= value_index:
             continue
-        name = _static_js_string(arguments[0])
-        value = _static_js_string(arguments[1])
+        name = _static_js_string(arguments[name_index])
+        value = _static_js_string(arguments[value_index])
         if name is None:
             if value is None or _RENDER_HTML_URL_LIST_NETWORK_RE.search(value.lstrip()):
                 return True
             continue
-        name = name.lower()
+        name = name.lower().rsplit(":", 1)[-1]
         if name not in _RENDER_HTML_NETWORK_ATTRIBUTES:
             continue
         if value is None or _render_html_attribute_reaches_network(name, value):
             return True
 
     for match in _RENDER_HTML_PROPERTY_ASSIGNMENT_START_RE.finditer(code):
-        parsed = _static_js_string_prefix(code[match.end() :])
-        if parsed is None:
-            continue
-        value, _ = parsed
+        value = _static_js_assignment_string(code[match.end() :])
+        if value is None:
+            return True
         if _render_html_attribute_reaches_network(match.group("attr").lower(), value):
             return True
+
+    for match in _RENDER_HTML_MARKUP_ASSIGNMENT_START_RE.finditer(code):
+        markup = _static_js_assignment_string(code[match.end() :])
+        if markup is None:
+            return True
+        if _render_html_code_reaches_network(markup, depth + 1):
+            return True
+
+    for match in _RENDER_HTML_MARKUP_CALL_START_RE.finditer(code):
+        arguments = _js_call_arguments(code, match.end())
+        if arguments is None:
+            return True
+        method = (match.group("insert") or match.group("write")).lower()
+        if method == "insertadjacenthtml":
+            if len(arguments) < 2:
+                continue
+            markup = _static_js_string(arguments[1])
+        else:
+            if not arguments:
+                continue
+            parts = [_static_js_string(argument) for argument in arguments]
+            markup = "".join(part for part in parts if part is not None)
+            if any(part is None for part in parts):
+                markup = None
+        if markup is None or _render_html_code_reaches_network(markup, depth + 1):
+            return True
     return False
 
 
-def _render_html_reaches_network(arguments: dict) -> bool:
-    code = arguments.get("code")
-    if not isinstance(code, str):
-        return False
+def _render_html_code_reaches_network(code: str, depth: int = 0) -> bool:
+    if depth > 8:
+        return True
     code = _JS_BLOCK_COMMENT_RE.sub("", code)
     return bool(
         _RENDER_HTML_NETWORK_RE.search(code)
-        or _render_html_attributes_reach_network(code)
-        or _render_html_computed_network_access(code)
+        or _render_html_attributes_reach_network(code, depth)
+        or _render_html_computed_network_access(code, depth)
     )
 
 
+def _render_html_reaches_network(arguments: dict) -> bool:
+    code = arguments.get("code")
+    return isinstance(code, str) and _render_html_code_reaches_network(code)
+
+
 # Tools that are read-only regardless of their arguments, so auto mode never has
 # to pause them and their safety needs no argument scan. render_html is handled
 # separately above because a networked canvas does need approval.

studio/backend/tests/test_permission_mode.py

@@ -973,6 +973,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.
@@ -1014,6 +1018,10 @@ def test_render_html_gated_only_when_networked():
     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');"
@@ -1049,9 +1057,36 @@ def test_render_html_gated_only_when_networked():
         )
         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='<p>Local</p>'</script>") is False
+    assert rh("<script>document.body.innerHTML=markup</script>") is True
+    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.writeln('<p>Local</p>')</script>") is False
+    assert rh("<script>writer.write('<img 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

studio/backend/tests/test_sandbox_tools.py

@@ -403,6 +403,65 @@ class TestSandboxEnvIsolation:
         assert bypass.returncode == 0, bypass.stderr
         assert bypass.stdout.strip() == "7"
 
+    @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_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", ["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