From 006bf4479ed7f1a581f29ef21d8888e05d5c749f Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 8 Jul 2026 10:46:12 +0000 Subject: [PATCH] Harden Studio Python-tool AST checks against obfuscated escapes The python tool's static safety analysis (_check_signal_escape_patterns) was purely name-based with no attribute visitor, so obfuscated routes to the shell / network / file policies it already enforces slipped past: eval / exec / compile, __import__ / importlib with a computed or dangerous module name, getattr / setattr aimed at os / subprocess / sys / builtins, and dunder gadget chains (().__class__.__bases__[0].__subclasses__()). Add a dynamic_exec category covering those, surfaced through _check_code_safety alongside the existing categories. Dynamic import stays allowed for a benign literal module name (huggingface_hub, json, numpy) so real workflows and the HF upload gate keep working; ordinary getattr(obj, "field") and __class__ access stay benign. Bypass Permissions (disable_sandbox) still skips the check. Tests: TestDynamicExecObfuscation in test_sandbox_tools.py with matching benign cases, giving _check_signal_escape_patterns its first direct coverage. --- studio/backend/core/inference/tools.py | 111 ++++++++++++++++++++- studio/backend/tests/test_sandbox_tools.py | 43 ++++++++ 2 files changed, 153 insertions(+), 1 deletion(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 82c50933fc..69cee0d9be 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1608,6 +1608,7 @@ def _check_signal_escape_patterns(code: str): signal_tampering = [] exception_catching = [] shell_escapes = [] + dynamic_exec = [] warnings = [] def _ast_name_matches(node, names): @@ -1661,6 +1662,60 @@ def _check_signal_escape_patterns(code: str): } ) + # Dynamic-execution / obfuscation primitives that defeat the static (name-based) checks + # above: they build or reach a dangerous callable at runtime, so a bare name match cannot + # see the payload. eval/exec/compile are direct code-execution builtins; __import__ / + # importlib load a module by (possibly computed) name; getattr/setattr on a sensitive + # module implement `getattr(os, 'sys'+'tem')(...)`. + _DYNAMIC_EXEC_BUILTINS = frozenset({"eval", "exec", "compile"}) + _DYNAMIC_IMPORT_FUNCS = frozenset( + {"importlib.import_module", "importlib.reload", "importlib.__import__"} + ) + # Dynamic import is a real workflow (e.g. importing huggingface_hub), so it is flagged only + # when the target is computed (non-literal name = obfuscation) or names a module that can + # reach code execution / shell / builtins. A benign literal (json, numpy, huggingface_hub) + # passes; the HF upload gate below still validates its call args separately. + _DANGEROUS_IMPORT_NAMES = frozenset( + { + "os", + "subprocess", + "sys", + "builtins", + "importlib", + "ctypes", + "pty", + "socket", + "signal", + "resource", + "shutil", + "multiprocessing", + "runpy", + "code", + "codeop", + "pdb", + "mmap", + "fcntl", + } + ) + # Attribute-name obfuscation via getattr/setattr is only flagged when aimed at a module + # that can execute code or reach builtins (keeps ordinary getattr(obj, "field") benign). + _DYNAMIC_ATTR_TARGETS = frozenset({"os", "subprocess", "sys", "builtins", "importlib"}) + # Introspection "gadget" dunders used to walk from a harmless object to os/builtins + # (``().__class__.__bases__[0].__subclasses__()``). ``__class__`` / ``__dict__`` are + # intentionally excluded (too common); the chain still trips on the others. + _GADGET_DUNDERS = frozenset( + { + "__subclasses__", + "__bases__", + "__base__", + "__mro__", + "__globals__", + "__builtins__", + "__code__", + "__closure__", + } + ) + def _extract_string_from_node(node): """Extract a plain string value from an AST node, if it is a constant.""" if isinstance(node, ast.Constant) and isinstance(node.value, str): @@ -1901,6 +1956,52 @@ def _check_signal_escape_patterns(code: str): } ) + # --- Dynamic execution / obfuscation primitives --- + dynamic_desc = None + is_dynamic_import = _ast_name_matches(func, _DYNAMIC_IMPORT_FUNCS) or ( + isinstance(func, ast.Name) and func.id in ("__import__", "import_module") + ) + if isinstance(func, ast.Name) and func.id in _DYNAMIC_EXEC_BUILTINS: + dynamic_desc = f"dynamic code execution via {func.id}()" + elif is_dynamic_import: + # Computed module name (obfuscation) or a dangerous target is unsafe; a benign + # literal import (huggingface_hub, json, ...) passes. + mod = _extract_string_from_node(node.args[0]) if node.args else None + if mod is None or mod.split(".")[0] in _DANGEROUS_IMPORT_NAMES: + dynamic_desc = "dynamic import of a computed or sensitive module name" + elif ( + isinstance(func, ast.Name) + and func.id in ("getattr", "setattr") + and node.args + and _ast_name_matches( + node.args[0], + _DYNAMIC_ATTR_TARGETS | self.os_aliases | self.subprocess_aliases, + ) + ): + dynamic_desc = f"{func.id}() on a sensitive module (attribute-name obfuscation)" + if dynamic_desc: + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": dynamic_desc, + } + ) + + self.generic_visit(node) + + def visit_Attribute(self, node): + # Introspection gadget dunders (``__subclasses__``, ``__globals__``, ...) are the + # standard way to walk from a benign object to os/builtins, bypassing the name-based + # checks. Flag the attribute access itself, then keep descending. + if node.attr in _GADGET_DUNDERS: + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": f"introspection gadget attribute {node.attr}", + } + ) self.generic_visit(node) def visit_ExceptHandler(self, node): @@ -2520,6 +2621,7 @@ def _check_signal_escape_patterns(code: str): len(signal_tampering) == 0 and len(exception_catching) == 0 and len(shell_escapes) == 0 + and len(dynamic_exec) == 0 and len(network_calls) == 0 and len(sensitive_file_reads) == 0 ) @@ -2527,6 +2629,7 @@ def _check_signal_escape_patterns(code: str): "signal_tampering": signal_tampering, "exception_catching": exception_catching, "shell_escapes": shell_escapes, + "dynamic_exec": dynamic_exec, "network_calls": network_calls, "sensitive_file_reads": sensitive_file_reads, "warnings": warnings, @@ -2550,13 +2653,19 @@ def _check_code_safety(code: str) -> str | None: exception_reasons = [ item.get("description", "") for item in info.get("exception_catching", []) ] + dynamic_reasons = [item.get("description", "") for item in info.get("dynamic_exec", [])] network_reasons = [item.get("description", "") for item in info.get("network_calls", [])] file_reasons = [ item.get("description", "") for item in info.get("sensitive_file_reads", []) ] all_reasons = [ r - for r in reasons + shell_reasons + exception_reasons + network_reasons + file_reasons + for r in reasons + + shell_reasons + + exception_reasons + + dynamic_reasons + + network_reasons + + file_reasons if r ] if all_reasons: diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index 24b1da1772..a1b3a4fdbd 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -781,3 +781,46 @@ class TestHfUploadEnvAndSecretLeakBlock: ' operations=[], token="hf_xxx")', expect_phrase = "HF upload token= cannot be set", ) + + +class TestDynamicExecObfuscation: + """The python AST checker must flag runtime code-execution / obfuscation primitives that + defeat its name-based analysis, while ordinary dynamic-attribute code stays allowed.""" + + @pytest.mark.parametrize( + "code, phrase", + [ + ("eval('1+1')", "dynamic code execution"), + ("exec('import os')", "dynamic code execution"), + ("compile('x', '', 'exec')", "dynamic code execution"), + ("__import__('os').system('id')", "dynamic import"), + ("__import__('o'+'s')", "dynamic import"), + ("__import__(chr(111) + chr(115))", "dynamic import"), + ("import importlib; importlib.import_module('subprocess')", "dynamic import"), + ("from importlib import import_module; import_module(name)", "dynamic import"), + ("getattr(os, 'system')('id')", "attribute-name obfuscation"), + ("import os as o; getattr(o, 'sys' + 'tem')('id')", "attribute-name obfuscation"), + ("().__class__.__bases__[0].__subclasses__()", "introspection gadget"), + ("[].__class__.__mro__", "introspection gadget"), + ("f.__globals__['os']", "introspection gadget"), + ], + ) + def test_dynamic_exec_blocked(self, code, phrase): + _blocked(code, expect_phrase = phrase) + + @pytest.mark.parametrize( + "code", + [ + "import json; json.loads('{}')", + "d = {'k': 1}; getattr(d, 'get')('k')", + "getattr(obj, 'name', None)", + "setattr(config, 'debug', True)", + "class A: pass\nprint(A().__class__.__name__)", + "import math; print(math.sqrt(2))", + "hf = __import__('huggingface_hub'); hf.HfApi()", + "import importlib; importlib.import_module('numpy')", + "__import__('json')", + ], + ) + def test_benign_dynamic_code_allowed(self, code): + _ok(code)