From 8384cea660facbe263a73b8eca0b8eb0bf5a7ee0 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Fri, 10 Jul 2026 17:17:32 +0000 Subject: [PATCH] Harden sandbox: fail closed on unvetted FunctionType code objects, subscript-stored exec aliases, and workdir deserializers; only treat shell keywords as separators at command position Close three bypasses and one false positive Codex found on the round-43 branch: - types.FunctionType() of an unvetted code object: the gadget was only flagged when its first arg was a compile() result, so a code object from any other producer (codeop.compile_command(), a loader's get_code(), or an opaque name) ran source the recursive eval/exec analysis never saw. Replace the compile-only denylist with an allowlist: FunctionType is allowed only when its first arg is an ordinary in-source function's code object (fn.__code__ / meth.__func__), whose body is analyzed normally, and fails closed for everything else. Robust against new producers instead of chasing each one. - subscript-stored exec alias: storing a dynamic-exec builtin into a container element (d['e'] = exec; d['e'](payload)) hid the sink from the name / attribute call checks -- the alias tracker only followed plain-name targets -- so the later subscript call ran an unreviewed payload. Flag the store itself: there is no benign reason to stash exec / eval / compile / __import__ in a container slot. - deserializer in an imported workdir module: the module import vetter (the only scan of a helper .py the user wrote) checked shell / eval / import / network sinks but not deserializers, so a helper calling pickle.loads on bytes whose reducer runs posix.system spawned an unguarded child. Refuse a workdir module that calls a reduce-executing deserializer (pickle / marshal / dill / cloudpickle / jsonpickle load / loads / Unpickler / decode) or binds one via from-import. json / importing pickle for dumps stay allowed. - false positive: shell keywords as separators regardless of position. if / while / until (and then / do / else / elif) were treated as command separators everywhere, so `echo if touch` was rejected as if `touch` ran even though it is just data passed to echo. Only reset command position for these keywords when they appear AT command position (the compound-statement header); real separators (; | && ...) still reset everywhere, so `if touch x; then :; fi` stays blocked. Regression coverage: TestRound44Bypasses in tests/test_sandbox_tools.py (FunctionType of codeop / loader / producer code objects blocked while fn.__code__ stays allowed; subscript-stored exec / eval / compile blocked while a benign container store is allowed; `echo if touch` allowed while the compound headers stay blocked) and two workdir-module vetter cases in tests/test_sandbox_runtime_backstop.py (pickle.loads reduce payload denied, json.loads still allowed). --- studio/backend/core/inference/tools.py | 130 +++++++++++++----- .../tests/test_sandbox_runtime_backstop.py | 56 ++++++++ studio/backend/tests/test_sandbox_tools.py | 70 ++++++++++ 3 files changed, 218 insertions(+), 38 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 1ba69e8dba..70bc2b3b91 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1167,12 +1167,23 @@ def _find_blocked_commands(command: str) -> set[str]: prev_was_flag = False # previous token (while a wrapper is pending) takes an operand cur_wrapper = None # the active wrapper's basename (drives per-wrapper option arity) for token in tokens: - if token in _SHELL_SEPARATORS or token in _SHELL_KEYWORDS_AS_SEP: + if token in _SHELL_SEPARATORS: expect_command = True prefix_pending = False prev_was_flag = False cur_wrapper = None continue + if token in _SHELL_KEYWORDS_AS_SEP: + # if / while / until / then / do / else / elif start a NEW command position ONLY when + # they appear at command position (the compound-statement header: `if touch x; then`). + # After a command word they are ordinary arguments -- bash does not run the next word as + # a command in `echo if touch`, so only reset there. Real separators (; | && ...) above + # always reset regardless of position. + if expect_command: + prefix_pending = False + prev_was_flag = False + cur_wrapper = None + continue if token.startswith("-"): # Flags belong to the active command, but keep expect_command while a # wrapper prefix awaits its command. Only a flag that actually takes a SEPARATED @@ -6701,41 +6712,16 @@ def _check_signal_escape_patterns( return True return False - def _is_compile_result(self, arg): - """True when ``arg`` is a ``compile(...)`` code object (bare / builtins / - single-assignment alias / inline-container unwrap). Used to catch code objects - executed through ``types.FunctionType`` instead of eval/exec.""" - # (compile(src, ...),)[0] / [compile(...)][0] / {'k': compile(...)}['k']: a - # trivial container unwrap hiding the compile() code object from the direct-call - # check. Resolve the indexed element and recurse. - if isinstance(arg, ast.Subscript): - container = arg.value - ci = _const_fold(arg.slice, _const_env) - if isinstance(container, (ast.List, ast.Tuple)) and isinstance(ci, int): - if -len(container.elts) <= ci < len(container.elts): - return self._is_compile_result(container.elts[ci]) - if isinstance(container, ast.Dict) and ci is not None: - for k, v in zip(container.keys, container.values): - if k is not None and _const_fold(k, _const_env) == ci: - return self._is_compile_result(v) - return False - if isinstance(arg, ast.Call): - af = arg.func - if isinstance(af, ast.Name): - if af.id == "compile" or self.exec_from_aliases.get(af.id) == "compile": - return True - if _analyzer_on and _scope_idx.resolve(af.id, arg, "execb") == "compile": - return True - elif ( - isinstance(af, ast.Attribute) - and af.attr == "compile" - and _ast_name_matches(af.value, self.builtins_aliases) - ): - return True - if _analyzer_on and isinstance(arg, ast.Name): - if _scope_idx.resolve(arg.id, arg, "compiledany"): - return True - return False + def _functiontype_arg_is_vetted(self, arg): + """True only when a ``types.FunctionType(code, ...)`` first arg is statically KNOWN to + be an ordinary in-source function's code object -- ``fn.__code__`` / ``meth.__func__`` + -- whose body the analyzer already walked. Fails CLOSED for everything else: a + ``compile(...)`` result, a loader's ``get_code()``, ``codeop.compile_command()``, + ``marshal.loads()``, a bare name / alias, or a container unwrap all yield a code object + running source the recursive eval/exec analysis never saw, so FunctionType is the + execution gadget and must be blocked. (Denylisting only ``compile`` left other producers + open; an allowlist of the one benign shape is robust against new producers.)""" + return isinstance(arg, ast.Attribute) and arg.attr in ("__code__", "__func__") def _attr_obfuscation_targets(self): # Modules whose DYNAMIC attribute / dict access (getattr, vars, __dict__) is @@ -6752,6 +6738,45 @@ def _check_signal_escape_patterns( | set(self.deserialize_module_aliases) ) + def _stores_hidden_sink(self, value): + # A dynamic-exec builtin (exec / eval / compile / __import__), bare / builtins-attr / + # scope alias, being stashed for later obfuscated invocation. + if isinstance(value, ast.Name): + if value.id in _DYNAMIC_EXEC_BUILTINS or value.id == "__import__": + return True + if value.id in self.exec_from_aliases: + return True + if _analyzer_on and _scope_idx.resolve(value.id, value, "execb"): + return True + if ( + isinstance(value, ast.Attribute) + and value.attr in (_DYNAMIC_EXEC_BUILTINS | {"__import__"}) + and _ast_name_matches(value.value, self.builtins_aliases) + ): + return True + return False + + def visit_Assign(self, node): + # d['e'] = exec / lst[0] = eval -- storing a dynamic-exec builtin into a CONTAINER + # element (not a plain name, which the alias tracker already follows) hides the sink + # from the name / attribute call checks, and the later d['e'](payload) then runs + # unreviewed. There is no benign reason to stash exec / eval / compile / __import__ in + # a container slot, so flag the store itself. + if any(isinstance(t, ast.Subscript) for t in node.targets) and self._stores_hidden_sink( + node.value + ): + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": ( + "a dynamic-exec builtin stored into a container element " + "(obfuscated exec alias)" + ), + } + ) + self.generic_visit(node) + def visit_Call(self, node): # operator.methodcaller('system', 'rm -rf /')(os) applies a deferred method to a # module receiver; rewrite it to the direct os.system('rm -rf /') call and analyze @@ -7673,10 +7698,10 @@ def _check_signal_escape_patterns( ) ) and node.args - and self._is_compile_result(node.args[0]) + and not self._functiontype_arg_is_vetted(node.args[0]) ): dynamic_desc = ( - "types.FunctionType() executes a compile() code object " + "types.FunctionType() executes an unvetted code object " "(bypasses the eval/exec gate)" ) elif ( @@ -10223,6 +10248,15 @@ try: # os / posix expose the exec-attr sinks (os.system, os.execv, ...); a sink attribute rooted at # one of these is a command-exec sink even without a direct call (x = os.system; x('id')). _GUARD_EXEC_RECEIVERS = frozenset({"os", "posix"}) + # Deserializers that run an attacker-controlled reduce payload (which can spawn an unguarded + # child via posix.system in the reducer): pickle & friends, marshal, dill, cloudpickle, + # jsonpickle. The malicious bytes live OUTSIDE this source, so a workdir helper that calls one + # is refused. (yaml / torch / numpy have safe modes and are left to the top-level analyzer; + # importing pickle for pickle.dumps stays allowed -- only the load sinks are refused.) + _GUARD_DESER_MODS = frozenset( + {"pickle", "_pickle", "cpickle", "marshal", "dill", "cloudpickle", "jsonpickle"} + ) + _GUARD_DESER_ATTRS = frozenset({"loads", "load", "Unpickler", "decode"}) def _guard_attr_root(_v): # Base Name id of an attribute chain (os.path -> 'os'); None if not Name-rooted. while isinstance(_v, _gast.Attribute): @@ -10239,6 +10273,7 @@ try: # as an attribute (builtins.eval / b.exec) are recognized alongside the bare names. _recv = set(_GUARD_EXEC_RECEIVERS) _bi = {"builtins", "__builtins__"} + _deser = set(_GUARD_DESER_MODS) for _nd in _gast.walk(_tree): if isinstance(_nd, _gast.Import): for _al in _nd.names: @@ -10246,6 +10281,8 @@ try: _recv.add(_al.asname or _al.name) elif _al.name == "builtins": _bi.add(_al.asname or _al.name) + elif _al.name in _GUARD_DESER_MODS: + _deser.add(_al.asname or _al.name) for _nd in _gast.walk(_tree): if isinstance(_nd, _gast.Import): for _al in _nd.names: @@ -10274,6 +10311,13 @@ try: for _al in _nd.names: if _al.name == "*" or _al.name in _GUARD_EXEC_ATTRS: return True + # `from pickle import loads` / `from pickle import *` binds a bare deserializer + # sink; a later bare loads(evil) has no module attribute to catch. (dumps stays + # allowed -- only the load sinks are refused.) + if _mroot in _GUARD_DESER_MODS: + for _al in _nd.names: + if _al.name == "*" or _al.name in _GUARD_DESER_ATTRS: + return True elif isinstance(_nd, _gast.Call): # An ACTUAL invocation of a sink-named method on any receiver (os.system(...), # or an aliased o.system(...)) is a command-exec call regardless of receiver. @@ -10292,6 +10336,16 @@ try: and _guard_attr_root(_nd.func.value) in _bi ): return True + # A deserializer call (pickle.loads / marshal.load / dill.load / + # pickle.Unpickler(f) / jsonpickle.decode) runs an attacker-controlled reduce + # payload whose bytes live outside this source, so refuse it -- rooted at a + # deserializer module / alias so a benign json.load / config.load is untouched. + if ( + isinstance(_nd.func, _gast.Attribute) + and _nd.func.attr in _GUARD_DESER_ATTRS + and _guard_attr_root(_nd.func.value) in _deser + ): + return True elif isinstance(_nd, _gast.Attribute): # A sink-named attribute REFERENCE (even uncalled) rooted at os / posix / an # os alias (x = os.system, s = o.system). A same-named attribute on an unrelated diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 1a3ead09a7..064d9a2246 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -998,6 +998,62 @@ def test_sandboxed_benign_urllib_parse_workdir_module_allowed(): os.remove(os.path.join(workdir, "okparse.py")) +@_POSIX_ONLY +def test_sandboxed_deserializer_workdir_module_denied(): + # A workdir helper that calls pickle.loads runs an attacker-controlled reduce payload (whose + # bytes live outside the helper source) -- here the reducer spawns os.system -- so the import + # vetter must refuse it even though the malicious call is not spelled in the source. + import pickle + + class _Evil: + def __reduce__(self): + return (os.system, ("echo PWNED_DESER",)) + + evil = pickle.dumps(_Evil()) + helper = "import pickle\nprint('DESER_REACHED')\npickle.loads(%r)\n" % (evil,) + session = "backstop-workdir-deser" + workdir = get_sandbox_workdir(session) + with open(os.path.join(workdir, "evildeser.py"), "w") as f: + f.write(helper) + try: + out = _python_exec( + "import evildeser; print('REACHED_' + 'BODY')", + None, + 30, + session, + disable_sandbox = False, + ) + assert "PWNED_DESER" not in out + assert "DESER_REACHED" not in out + assert "REACHED_BODY" not in out + assert "sandbox:" in out or "ImportError" in out + finally: + os.remove(os.path.join(workdir, "evildeser.py")) + + +@_POSIX_ONLY +def test_sandboxed_benign_json_workdir_module_allowed(): + # json.load / json.loads do NOT run a reduce payload and are not in the deserializer set, so a + # workdir helper using json must still import. + session = "backstop-workdir-json" + workdir = get_sandbox_workdir(session) + with open(os.path.join(workdir, "okjson.py"), "w") as f: + f.write("import json\nVALUE = json.loads('[1, 2, 3]')\nprint('JSON_OK')\n") + try: + out = _python_exec( + "import okjson; print('REACHED', okjson.VALUE)", + None, + 30, + session, + disable_sandbox = False, + ) + assert "JSON_OK" in out + assert "REACHED [1, 2, 3]" in out + assert "sandbox:" not in out + finally: + os.remove(os.path.join(workdir, "okjson.py")) + + @_POSIX_ONLY def test_sandboxed_realpath_monkeypatch_write_escape_denied(tmp_path): # Sandboxed code reassigns os.path.realpath to a lambda that echoes an in-workdir diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index f563e73b3c..3f16d3d98f 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -4494,3 +4494,73 @@ class TestRound43Bypasses: ) def test_round43_benign_compile_allowed(self, code): _ok(code) + + +class TestRound44Bypasses: + """Forty-fourth-round Codex findings. The FunctionType producer gadget, the subscript-stored + exec alias, and the command-position shell-keyword FP are static; the workdir-module + deserializer item is covered in test_sandbox_runtime_backstop.py.""" + + @pytest.mark.parametrize( + "code", + [ + # types.FunctionType() executes ANY code object, not only a compile() one. A code + # object from another producer (codeop.compile_command, a loader's get_code(), or an + # opaque name) runs source the recursive analysis never saw, so the gadget fails closed. + "import types, codeop\n" + "types.FunctionType(codeop.compile_command('import os', '', 'exec'), globals())()", + "import types\n" + "from importlib.machinery import SourceFileLoader\n" + "types.FunctionType(SourceFileLoader('m', 'x.py').get_code(), {})()", + "import types\ndef producer():\n return _external\n" + "types.FunctionType(producer(), {})()", + ], + ) + def test_functiontype_unvetted_code_object_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_functiontype_ordinary_code_object_allowed(self): + # fn.__code__ is an ordinary in-source function's code, analyzed normally -- still allowed. + _ok("import types\ndef g():\n return 1\ntypes.FunctionType(g.__code__, {})") + + @pytest.mark.parametrize( + "code", + [ + # Storing a dynamic-exec builtin into a CONTAINER element hides it from the name / + # attribute call checks; the later subscript call runs an unreviewed payload. + "d = {}\nd['e'] = exec\nd['e'](\"import os\\nos.system('touch /tmp/pwn')\")", + "lst = [None]\nlst[0] = eval\nlst[0](\"__import__('os').system('touch /tmp/pwn')\")", + "d = {}\nd['c'] = compile\nc = d['c']('import os', '', 'exec')", + ], + ) + def test_subscript_stored_exec_alias_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_subscript_stored_benign_value_allowed(self): + # A non-sink value stored in a container slot is ordinary code -- not flagged. + _ok("d = {}\nd['x'] = 5\nd['y'] = len\nprint(d['x'], d['y']([1, 2]))") + + @pytest.mark.parametrize( + "code", + [ + # if / while / until are compound-statement HEADERS only at command position; after a + # command word they are ordinary arguments, so `echo if touch` is not a `touch` run. + "import os\nos.system('echo if touch')", + "import os\nos.system('echo while rm')", + "import os\nos.system('printf %s until cp')", + ], + ) + def test_shell_keyword_as_argument_allowed(self, code): + _ok(code) + + @pytest.mark.parametrize( + "code", + [ + # A real compound header still runs its condition / body command -- must stay blocked. + "import os\nos.system('if touch x; then :; fi')", + "import os\nos.system('while true; do rm -rf /tmp/x; done')", + "import os\nos.system('touch x')", + ], + ) + def test_shell_keyword_compound_header_still_blocked(self, code): + assert _check_code_safety(code) is not None, code