ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
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).
This commit is contained in:
danielhanchen 2026-07-10 17:17:32 +00:00
commit 8384cea660
3 changed files with 218 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

130
studio/backend/core/inference/tools.py
View file

@ -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

56
studio/backend/tests/test_sandbox_runtime_backstop.py
View file

@ -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

70
studio/backend/tests/test_sandbox_tools.py
View file

@ -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', '<s>', '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', '<s>', '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