From 0c17f074aef932cccb0a38a2c6ca1c19e0c462e3 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Thu, 9 Jul 2026 18:45:38 +0000 Subject: [PATCH] Studio sandbox: close eighth-round review bypasses (closures, class scopes, redirects, aliases) Static classifier: - flag cell_contents (the only closure-cell reader) as a gadget so recovering a guarded wrapper's original callable via __closure__ fails closed even when the __closure__ name is built at runtime - treat a class body as its own alias scope (class C: e = eval; e(...) now recognized) while keeping methods lexically skipping the class scope, so a same-named class attr does not shadow the module-level sink a method reaches - block output redirection (> / >> / &> / N>) to an absolute / ~ / .. target: a child shell runs unguarded; relative in-workdir redirects stay allowed - resolve pathlib constructor import aliases (from pathlib import Path as P) before traversal reads - flag an integer-indexed __mro__ (io.FileIO.__mro__[1]) that extracts the unguarded FileIO C base class; plain iteration / slicing stays allowed - resolve aliased read callees before traversal checks: o = open and import shutil as sh; sh.copy(...) - recurse into env -S / --split-string operands so env -S 'python3 -c ...' still detects the interpreter --- studio/backend/core/inference/tools.py | 138 ++++++++++++++++-- .../tests/test_sandbox_runtime_backstop.py | 34 +++++ studio/backend/tests/test_sandbox_tools.py | 112 ++++++++++++++ 3 files changed, 274 insertions(+), 10 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index e1ff3e483e..ef04b5ba3b 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -374,6 +374,45 @@ def _find_blocked_commands(command: str) -> set[str]: blocked |= _find_blocked_commands(tokens[i + 1]) break # stop at first non-flag token + # `env -S 'cmd ...'` / `env --split-string='cmd'` splits the string and runs it as a + # fresh command, so a bare `env -S` operand is NOT just a flag value -- recurse into + # it (it can invoke an unguarded interpreter or another blocked command). + for i, token in enumerate(tokens): + tl = token.lower() + payload = None + if tl in ("-s", "--split-string") and i + 1 < len(tokens): + payload = tokens[i + 1] + elif tl.startswith("-s") and tl != "-s" and not tl.startswith("--"): + payload = token[2:] # glued short form: env -S'cmd' / -Scmd + elif tl.startswith("--split-string="): + payload = token[len("--split-string=") :] + if not payload: + continue + for j in range(i - 1, -1, -1): + prev = tokens[j] + if prev.startswith("-"): + continue + if os.path.basename(prev).lower() == "env": + blocked |= _find_blocked_commands(payload) + break + + # Output redirection (> / >> / &> / N>) to a path OUTSIDE the workdir: a child shell + # runs unguarded, so `echo x > /tmp/p` / `>> ../p` / `> ~/p` writes past the session + # workdir. A relative target (> out.txt) stays in the workdir cwd and is allowed. + # Scanning tokens (not the raw string) avoids matching a `>` inside a quoted argument. + for i, tok in enumerate(tokens): + rm = re.search(r">{1,2}([^\s>]*)$", tok) + if rm is None: + continue + tgt = rm.group(1) + if not tgt and i + 1 < len(tokens): + tgt = tokens[i + 1] + if not tgt: + continue + tn = tgt.replace("\\", "/") + if tgt.startswith("~") or tn.startswith("/") or ".." in tn.split("/"): + blocked.add("redirect:" + tgt) + return blocked @@ -2638,16 +2677,27 @@ class _ScopeAliasIndex: def _build_scope_alias_index(tree, const_env): idx = _ScopeAliasIndex(tree) - def _rec(node, scope): + def _rec(node, scope, func_enclose): + # scope: namespace the direct children belong to (for node_scope + counting). + # func_enclose: the scope a nested FUNCTION / class body encloses to. Python skips + # class scope for nested functions, so inside a class body this stays the class's + # own lexical function/module parent rather than the class. for child in ast.iter_child_nodes(node): idx.node_scope[child] = scope if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): - idx.enclosing[child] = scope - _rec(child, child) + idx.enclosing[child] = func_enclose + _rec(child, child, child) + elif isinstance(child, ast.ClassDef): + # A class body executes immediately with its OWN namespace, so it is a + # real alias scope (class C: e = eval; e(...) runs eval), but its names + # are not visible to methods defined inside it -- those enclose to + # func_enclose, skipping the class. + idx.enclosing[child] = func_enclose + _rec(child, child, func_enclose) else: - _rec(child, scope) + _rec(child, scope, func_enclose) - _rec(tree, tree) + _rec(tree, tree, tree) # os / subprocess import + from-import aliases are collected tree-wide (imports # are lexically visible module-wide in practice) and shared across scopes. @@ -2737,7 +2787,9 @@ def _build_scope_alias_index(tree, const_env): return None scopes = [tree] + [ - n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + n + for n in ast.walk(tree) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) ] for scope in scopes: counts: dict[str, int] = {} @@ -3299,11 +3351,16 @@ def _check_signal_escape_patterns( # 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. - # __mro__ and __code__ are deliberately EXCLUDED: on their own they do not reach + # __mro__ and __code__ are deliberately EXCLUDED here: on their own they do not reach # an execution primitive, and they are read by ordinary ML/debugging code # (trainer_class.__mro__, fn.__code__), so flagging them over-blocks legitimate # snippets. The terminal escape primitives below still trip on the real gadget - # chains (().__class__.__bases__[0].__subclasses__(), f.__globals__['os']). + # chains (().__class__.__bases__[0].__subclasses__(), f.__globals__['os']). A + # SUBSCRIPTED __mro__ (cls.__mro__[1], the base-class extraction shape) is flagged + # separately in visit_Subscript so plain iteration stays allowed. + # cell_contents is the ONLY way to read a closure cell's value, so it is the terminal + # step of recovering a guarded wrapper's original callable via __closure__; flagging + # it closes that recovery even when the __closure__ name was built dynamically. _GADGET_DUNDERS = frozenset( { "__subclasses__", @@ -3312,6 +3369,7 @@ def _check_signal_escape_patterns( "__globals__", "__builtins__", "__closure__", + "cell_contents", } ) @@ -3955,6 +4013,24 @@ def _check_signal_escape_patterns( self.generic_visit(node) def visit_Subscript(self, node): + # An INTEGER-indexed __mro__ (cls.__mro__[1]) extracts a specific base class the + # way __bases__[0] does -- the shape used to reach the original FileIO C base + # class (io.FileIO.__mro__[1]) or walk to object/subclasses. Plain iteration + # (for c in cls.__mro__) and slicing (cls.__mro__[1:]) yield the tuple/list for + # legitimate introspection, so only a non-slice index is flagged. + if ( + isinstance(node.ctx, ast.Load) + and isinstance(node.value, ast.Attribute) + and node.value.attr == "__mro__" + and not isinstance(node.slice, ast.Slice) + ): + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": "subscripted __mro__ extracts a base class (gadget)", + } + ) # sys.modules['os'] pulls an already-loaded dangerous module out of the # loader table (os/subprocess are loaded by the host). Scope to a Load of a # dangerous LITERAL key so legit uses ("x" in sys.modules, sys.modules.get( @@ -4661,6 +4737,47 @@ def _check_signal_escape_patterns( "shutil.copytree", "shutil.move", ) + _SHUTIL_COPY_METHODS = ("copy", "copy2", "copyfile", "copytree", "move") + # Import aliases so the traversal check still recognizes a renamed callee: + # from pathlib import Path as P -> P('../../etc/passwd').read_text() + # import shutil as sh -> sh.copy('../../etc/passwd', 'x') + _pathlib_ctor_aliases = set(_PATHLIB_CTORS) + _shutil_aliases = {"shutil"} + for _imp in ast.walk(tree): + if isinstance(_imp, ast.ImportFrom) and _imp.module == "pathlib": + for _a in _imp.names: + if _a.name in _PATHLIB_CTORS: + _pathlib_ctor_aliases.add(_a.asname or _a.name) + elif isinstance(_imp, ast.Import): + for _a in _imp.names: + if _a.name == "shutil": + _shutil_aliases.add(_a.asname or "shutil") + + def _resolves_to_open(fn): + # A callee that is `open`, or a single-assignment alias of it (o = open; + # o('../../etc/passwd').read()), or builtins.open / io.open / os.open. + if isinstance(fn, ast.Name): + if fn.id == "open": + return True + rhs = _scope_idx.resolve(fn.id, fn, "rhsnode") + if isinstance(rhs, ast.Name) and rhs.id == "open": + return True + if ( + isinstance(rhs, ast.Attribute) + and rhs.attr == "open" + and isinstance(rhs.value, ast.Name) + and rhs.value.id in ("builtins", "__builtins__", "io", "os") + ): + return True + return False + + def _is_shutil_copy_callee(fn): + return ( + isinstance(fn, ast.Attribute) + and fn.attr in _SHUTIL_COPY_METHODS + and isinstance(fn.value, ast.Name) + and fn.value.id in _shutil_aliases + ) def _fold_read_arg(arg): # Fold a read-path argument to a concrete string, resolving a module-level @@ -4717,7 +4834,7 @@ def _check_signal_escape_patterns( return os.path.join(*parts) except Exception: return None - ctor = (isinstance(rf, ast.Name) and rf.id in _PATHLIB_CTORS) or ( + ctor = (isinstance(rf, ast.Name) and rf.id in _pathlib_ctor_aliases) or ( isinstance(rf, ast.Attribute) and rf.attr in _PATHLIB_CTORS ) if not ctor or not recv.args: @@ -4762,9 +4879,10 @@ def _check_signal_escape_patterns( else (f.id if isinstance(f, ast.Name) else "") ) is_read_callee = ( - (isinstance(f, ast.Name) and f.id == "open") + _resolves_to_open(f) or fq in ("io.open", "os.open") or fq in _SHUTIL_COPY_SINKS + or _is_shutil_copy_callee(f) or method in _READ_METHODS ) # Pathlib read on a Path(...) / join receiver: check the resolved path. diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 3567059792..5efae67efb 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -708,3 +708,37 @@ def test_sandboxed_closure_recovery_of_open_blocked(): ) assert "unsafe code detected" in out or "sandbox:" in out or "PermissionError" in out assert not os.path.exists("/tmp/studio_closure_escape.txt") + + +@_POSIX_ONLY +def test_sandboxed_dynamic_closure_name_recovery_blocked(): + # __closure__ built at runtime via chr(): the static gate must block the + # .cell_contents recovery step so the original open is never reached. + name = "''.join(map(chr,[95,95,99,108,111,115,117,114,101,95,95]))" + out = _python_exec( + f"getattr(open, {name})[0].cell_contents('/tmp/studio_dyn_closure.txt', 'w').write('x')\n" + "print('DYN_CLOSURE_WROTE')\n", + None, + 30, + "backstop-dyn-closure", + disable_sandbox = False, + ) + assert "unsafe code detected" in out or "sandbox:" in out or "PermissionError" in out + assert not os.path.exists("/tmp/studio_dyn_closure.txt") + + +@_POSIX_ONLY +def test_sandboxed_fileio_base_via_mro_blocked(): + # The guarded io.FileIO subclass exposes the unguarded C base at __mro__[1]; the + # static gate now blocks the integer-indexed __mro__ base extraction. + out = _python_exec( + "import io\n" + "io.FileIO.__mro__[1]('/tmp/studio_mro_escape.txt', 'w').write(b'x')\n" + "print('MRO_WROTE')\n", + None, + 30, + "backstop-mro", + disable_sandbox = False, + ) + assert "unsafe code detected" in out or "sandbox:" in out or "PermissionError" in out + assert not os.path.exists("/tmp/studio_mro_escape.txt") diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index 01760f14ef..639807da08 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -1334,3 +1334,115 @@ class TestRound7Bypasses: def test_shutil_copy_benign_allowed(self): _ok("import shutil\nshutil.copy('data.csv', 'backup.csv')") + + +class TestRound8Bypasses: + """Eighth-round Codex findings: dynamic closure recovery, class-body exec aliases, + shell redirection escapes, pathlib/read-callee/shutil aliases, FileIO base via + __mro__, and env -S split strings.""" + + def test_dynamic_closure_name_lookup_blocked(self): + # __closure__ built at runtime then .cell_contents to recover the guarded open. + name = "''.join(map(chr,[95,95,99,108,111,115,117,114,101,95,95]))" + assert ( + _check_code_safety(f"getattr(open, {name})[0].cell_contents('/tmp/x','w')") is not None + ) + # cell_contents is flagged directly and via getattr, regardless of how __closure__ + # was reached. + assert _check_code_safety("open.__closure__[0].cell_contents('/tmp/x','w')") is not None + assert _check_code_safety("getattr(f, 'cell_contents')") is not None + + @pytest.mark.parametrize( + "code", + [ + "class C:\n e = eval\n e(\"__import__('os').system('rm -rf /')\")", + "class C:\n r = exec\n r(\"__import__('os').system('id')\")", + "import os\n\n\nclass C:\n s = os.system\n s('rm -rf /')", + ], + ) + def test_class_body_exec_alias_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_method_still_resolves_module_alias(self): + # A method skips the class scope (Python semantics), so a same-named class attr + # must NOT shadow the module-level sink alias the method actually reaches. + assert ( + _check_code_safety( + "import os\ns = os.system\n" + "class C:\n s = 1\n def m(self):\n s('rm -rf /')\n" + "C().m()" + ) + is not None + ) + + def test_class_body_benign_alias_allowed(self): + _ok("class C:\n f = sorted\n y = f([3, 1, 2])") + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('echo x > /tmp/p')", + "import os\nos.system('echo x >> /etc/passwd')", + "import os\nos.system('echo x > ~/p')", + "import os\nos.system('echo x > ../escape')", + "exec(\"import os\\nos.system('printf x > /tmp/p')\")", + ], + ) + def test_shell_redirect_escape_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_benign_relative_redirect_allowed(self): + # A relative redirect stays in the workdir cwd. + _ok("import os\nos.system('echo hi > out.txt')") + _ok("import os\nos.system('ls 2>&1')") + + @pytest.mark.parametrize( + "code", + [ + "from pathlib import Path as P\nP('../../../etc/passwd').read_text()", + "from pathlib import PurePath as PP\nPP('../../../etc/passwd').read_text()", + ], + ) + def test_pathlib_ctor_alias_read_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import io\nio.FileIO.__mro__[1]('/tmp/x', 'w')", + "import _io\n_io.FileIO.__mro__[1]('/tmp/x', 'w')", + ], + ) + def test_fileio_base_via_mro_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_mro_iteration_and_slice_allowed(self): + _ok("cls = int\nfor c in cls.__mro__:\n pass") + _ok("for c in int.__mro__[1:]:\n pass") + + @pytest.mark.parametrize( + "code", + [ + "o = open\no('../../../etc/passwd').read()", + "import shutil as sh\nsh.copy('../../../etc/passwd', 'x')", + "import shutil as sh\nsh.copyfile('../../../etc/passwd', 'x')", + ], + ) + def test_aliased_read_callee_traversal_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system(\"env -S 'python3 -c print(1)'\")", + "import os\nos.system('env -Spython3 evil.py')", + "import os\nos.system(\"env -S 'rm -rf /'\")", + ], + ) + def test_env_split_string_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_benign_env_and_getattr_allowed(self): + # No false positive on a benign env invocation or a benign dynamic getattr. + _ok("import os\nos.system('env PYTHONPATH=. echo hi')") + _ok("obj = {}\nname = 'keys'\ngetattr(obj, name)()")