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.
This commit is contained in:
danielhanchen
parent
934f879043
commit
006bf4479e
2 changed files with 153 additions and 1 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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', '<s>', '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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue