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

Studio sandbox: close sixth-round review bypasses (obfuscation, reads, child procs, guard pinning)

Browse source 
Static classifier:
- resolve pathlib expressions passed to open()/read callees so open(Path('/etc') / 'passwd') blocks like open('/etc/passwd')
- flag getattr()/setattr() of an introspection gadget dunder (__globals__, __subclasses__, ...) regardless of receiver
- add the .get() twin of the globals()/locals()/vars() namespace-dict subscript guard
- constant-fold sys.modules[...] and sys.modules.get(...) keys so a concatenated key is caught
- track 'from builtins import __import__ as imp' as a dynamic import alias
- treat deserializer module aliases (pickle, dill, ...) as sensitive targets for getattr/vars/__dict__
- flag code objects executed through types.FunctionType(compile(src, ...), ...), including the c = compile(src); FunctionType(c) two-step
- block language interpreters (python/perl/ruby/node/...) at shell command position: a spawned child runs without the in-process write guard

Runtime backstop:
- pin os.fspath/os.path.realpath to captured originals inside _within so a sandboxed reassignment of os.fspath cannot make realpath resolve an outside write target to an in-workdir path
This commit is contained in:
danielhanchen 2026-07-09 17:32:27 +00:00
commit 0f4b4b3d36
3 changed files with 369 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -113,6 +113,32 @@ _BLOCKED_COMMANDS_COMMON = frozenset(
"ln",
}
)
# Language interpreters that run inline / file / stdin code in a FRESH child process.
# The runtime filesystem backstop only patches the current interpreter, so a spawned
# `python -c '...'`, `perl -e '...'`, `node -e '...'`, etc. runs with none of the guard
# monkeypatches and can write/delete outside the session workdir. Blocking the
# interpreter at shell command position closes that child-process escape; the sandbox's
# own python_execute tool is the supported way to run Python (it IS guarded). Argument
# position (`echo python`, `ls /usr/bin/python3`) is unaffected by the command-position
# scanner.
_INTERPRETER_COMMANDS = frozenset(
{
"python",
"python2",
"python3",
"pythonw",
"perl",
"ruby",
"node",
"nodejs",
"php",
"deno",
"lua",
"luajit",
"rscript",
}
)
_BLOCKED_COMMANDS_COMMON = _BLOCKED_COMMANDS_COMMON | _INTERPRETER_COMMANDS
_BLOCKED_COMMANDS_WIN = frozenset(
{
"rmdir",
@ -2511,6 +2537,7 @@ class _ScopeAliasIndex:
"shell",
"execb",
"compiled",
"compiledany",
"impf",
"deser",
"strconst",
@ -2524,6 +2551,9 @@ class _ScopeAliasIndex:
self.shell: dict = {}
self.execb: dict = {}
self.compiled: dict = {}
# name -> True: bound to a compile() result (foldable OR dynamic). Used to catch
# a code object executed through types.FunctionType(c, ...) after `c = compile(...)`.
self.compiledany: dict = {}
self.impf: dict = {} # name -> True: alias of __import__ / importlib.import_module
self.deser: dict = {} # name -> fq deserializer sink (pickle.loads, ...)
self.strconst: dict = {} # name -> folded str/bytes constant (for read scanning)
@ -2667,6 +2697,7 @@ def _build_scope_alias_index(tree, const_env):
smap: dict[str, str] = {}
emap: dict[str, str] = {}
cmap: dict[str, tuple] = {}
camap: dict[str, bool] = {}
imap: dict[str, bool] = {}
dmap: dict[str, str] = {}
scmap: dict[str, object] = {}
@ -2685,6 +2716,9 @@ def _build_scope_alias_index(tree, const_env):
and rhs.func.id == "compile"
and rhs.args
):
# Any `c = compile(...)` binds a code object, tracked for the
# types.FunctionType(c) execution gadget below (dynamic or foldable).
camap[name] = True
v = _const_fold(rhs.args[0], const_env)
if isinstance(v, (str, bytes, bytearray)):
cmap[name] = (
@ -2708,6 +2742,8 @@ def _build_scope_alias_index(tree, const_env):
idx.execb[scope] = emap
if cmap:
idx.compiled[scope] = cmap
if camap:
idx.compiledany[scope] = camap
if imap:
idx.impf[scope] = imap
if dmap:
@ -3259,6 +3295,11 @@ def _check_signal_escape_patterns(
# import pickle as p -> {"p": "pickle"}; from pickle import loads as l -> {"l": "pickle.loads"}
self.deserialize_module_aliases: dict[str, str] = {}
self.deserialize_aliases: dict[str, str] = {}
# import types as t -> {"types", "t"}; from types import FunctionType as F -> {"F"}.
# FunctionType(code, globals)() runs a code object WITHOUT eval/exec, so a
# dynamic compile() result reaches execution through it (see visit_Call).
self.types_aliases = {"types"}
self.functiontype_aliases: set[str] = set()
self.loop_depth = 0
def visit_Import(self, node):
@ -3277,6 +3318,8 @@ def _check_signal_escape_patterns(
self.sys_aliases.add(alias.asname or "sys")
elif alias.name == "builtins":
self.builtins_aliases.add(alias.asname or "builtins")
elif alias.name == "types":
self.types_aliases.add(alias.asname or "types")
if alias.name in _DESERIALIZE_MODULES:
self.deserialize_module_aliases[alias.asname or alias.name] = alias.name
self.generic_visit(node)
@ -3314,11 +3357,19 @@ def _check_signal_escape_patterns(
for alias in node.names:
if alias.name in _DYNAMIC_EXEC_BUILTINS:
self.exec_from_aliases[alias.asname or alias.name] = alias.name
elif alias.name == "__import__":
# `from builtins import __import__ as imp; imp('os').system(...)`
# is a dynamic import exactly like a bare __import__ call.
self.import_func_aliases.add(alias.asname or alias.name)
elif node.module in _DESERIALIZE_MODULES:
for alias in node.names:
fq = f"{node.module}.{alias.name}"
if fq in _CODE_DESERIALIZE_SINKS:
self.deserialize_aliases[alias.asname or alias.name] = fq
elif node.module == "types":
for alias in node.names:
if alias.name == "FunctionType":
self.functiontype_aliases.add(alias.asname or alias.name)
self.generic_visit(node)
def visit_While(self, node):
@ -3358,6 +3409,43 @@ def _check_signal_escape_patterns(
return _elt(v)
return None
def _is_compile_result(self, arg):
"""True when ``arg`` is a ``compile(...)`` code object (bare / builtins /
single-assignment alias). Used to catch code objects executed through
``types.FunctionType`` instead of eval/exec."""
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 _attr_obfuscation_targets(self):
# Modules whose DYNAMIC attribute / dict access (getattr, vars, __dict__) is
# obfuscation that reaches code execution: the exec / import / shell modules
# PLUS the deserializer modules -- getattr(pickle, 'loads')(x) and
# vars(pickle)['loads'](x) are just pickle.loads(x) with the name hidden.
return (
_DYNAMIC_ATTR_TARGETS
| self.os_aliases
| self.subprocess_aliases
| self.importlib_aliases
| self.sys_aliases
| self.builtins_aliases
| set(self.deserialize_module_aliases)
)
def visit_Call(self, node):
func = node.func
func_name = None
@ -3617,19 +3705,26 @@ def _check_signal_escape_patterns(
# targets too: __import__('pickle').loads(blob) executes a reduce
# payload even though a plain `import pickle` is benign.
dynamic_desc = "dynamic import of a computed or sensitive module name"
elif (
isinstance(func, ast.Name)
and func.id in ("getattr", "setattr")
and len(node.args) >= 2
and isinstance(_const_fold(node.args[1], _const_env), str)
and _const_fold(node.args[1], _const_env) in _GADGET_DUNDERS
):
# getattr(anything, '__globals__' / '__subclasses__' / ...) reaches an
# introspection gadget with no ast.Attribute for visit_Attribute to catch.
# Direct x.__globals__ is already flagged for ANY receiver, so flag the
# getattr-string form regardless of receiver too.
dynamic_desc = (
"getattr() of an introspection gadget dunder "
f"({_const_fold(node.args[1], _const_env)})"
)
elif (
isinstance(func, ast.Name)
and func.id == "vars"
and node.args
and _ast_name_matches(
node.args[0],
_DYNAMIC_ATTR_TARGETS
| self.os_aliases
| self.subprocess_aliases
| self.importlib_aliases
| self.sys_aliases
| self.builtins_aliases,
)
and _ast_name_matches(node.args[0], self._attr_obfuscation_targets())
):
# vars(os) / vars(__builtins__) returns the module __dict__, the same
# obfuscation as os.__dict__['system'] but without the attribute access.
@ -3638,15 +3733,7 @@ def _check_signal_escape_patterns(
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
| self.importlib_aliases
| self.sys_aliases
| self.builtins_aliases,
)
and _ast_name_matches(node.args[0], self._attr_obfuscation_targets())
):
# Stage 2 refinement: a benign constant attr (getattr(os, "getpid"))
# is allowed; only a dynamic attr or a dangerous constant attr blocks.
@ -3675,9 +3762,50 @@ def _check_signal_escape_patterns(
and _ast_name_matches(func.value.value, self.sys_aliases)
and node.args
):
_key = _extract_string_from_node(node.args[0])
if _key is not None and _key.split(".")[0] in _DANGEROUS_IMPORT_NAMES:
# Constant-fold the key so sys.modules.get('o' + 's') is caught, not
# just a bare literal (the module is already loaded by the prelude).
_key = _const_fold(node.args[0], _const_env)
if isinstance(_key, str) and _key.split(".")[0] in _DANGEROUS_IMPORT_NAMES:
dynamic_desc = "sys.modules.get(...) access to a sensitive module"
elif (
# globals().get('__builtins__') / locals().get(...) / vars().get(...)
# -- the .get() twin of the globals()['__builtins__'] subscript form.
isinstance(func, ast.Attribute)
and func.attr == "get"
and isinstance(func.value, ast.Call)
and isinstance(func.value.func, ast.Name)
and func.value.func.id in ("globals", "locals", "vars")
and not func.value.args
and node.args
):
_key = _const_fold(node.args[0], _const_env)
if isinstance(_key, str) and (
_key in ("__builtins__", "__builtin__")
or _key.split(".")[0] in _DANGEROUS_IMPORT_NAMES
):
dynamic_desc = (
"namespace-dict .get() access to builtins / a sensitive module"
)
elif (
(
# types.FunctionType(compile(src, ...), {})() runs a code object WITHOUT
# eval/exec, so a dynamic compile() payload reaches execution here even
# though compile() alone is allowed. Flag when a FunctionType call takes
# a compile()-derived code object as its first argument.
(
isinstance(func, ast.Attribute)
and func.attr == "FunctionType"
and _ast_name_matches(func.value, self.types_aliases)
)
or (isinstance(func, ast.Name) and func.id in self.functiontype_aliases)
)
and node.args
and self._is_compile_result(node.args[0])
):
dynamic_desc = (
"types.FunctionType() executes a compile() code object "
"(bypasses the eval/exec gate)"
)
if dynamic_desc:
dynamic_exec.append(
{
@ -3702,13 +3830,7 @@ def _check_signal_escape_patterns(
}
)
elif node.attr == "__dict__" and _ast_name_matches(
node.value,
_DYNAMIC_ATTR_TARGETS
| self.os_aliases
| self.subprocess_aliases
| self.importlib_aliases
| self.sys_aliases
| self.builtins_aliases,
node.value, self._attr_obfuscation_targets()
):
# os.__dict__['system']('id') reaches the sink with no getattr call for
# the name-based checks to see. __dict__ on ordinary objects stays allowed.
@ -3742,8 +3864,10 @@ def _check_signal_escape_patterns(
and _extract_string_from_node(v.args[1]) == "modules"
)
if isinstance(node.ctx, ast.Load) and is_sys_modules:
key = _extract_string_from_node(node.slice)
if key is not None and key.split(".")[0] in _DANGEROUS_IMPORT_NAMES:
# Constant-fold the key so sys.modules['o' + 's'] is caught, not just a
# bare literal; a truly dynamic key (sys.modules[name]) stays allowed.
key = _const_fold(node.slice, _const_env)
if isinstance(key, str) and key.split(".")[0] in _DANGEROUS_IMPORT_NAMES:
dynamic_exec.append(
{
"type": "dynamic_exec",
@ -3761,8 +3885,8 @@ def _check_signal_escape_patterns(
and v.func.id in ("globals", "locals", "vars")
and not v.args
):
key = _extract_string_from_node(node.slice)
if key is not None and (
key = _const_fold(node.slice, _const_env)
if isinstance(key, str) and (
key in ("__builtins__", "__builtin__")
or key.split(".")[0] in _DANGEROUS_IMPORT_NAMES
):
@ -4517,6 +4641,12 @@ def _check_signal_escape_patterns(
for arg in list(node.args) + [kw.value for kw in (node.keywords or [])]:
s = _fold_read_arg(arg)
if s is None:
# A pathlib expression carries no foldable string constant
# (open(Path('/etc') / 'passwd')), so resolve it the same way a
# Path(...).read_text() receiver is resolved before skipping.
rp = _pathlib_receiver_path(arg)
if rp is not None and _flag_read_path(node, rp, is_read_callee):
break
continue
if _flag_read_path(node, s, is_read_callee):
break
@ -4658,16 +4788,33 @@ import os as _os, builtins as _bi, io as _io, pathlib as _pl
# wrapper does (self shifts into the next arg), which would corrupt Path.open /
# Path.write_text. Importing here makes the accessor capture the originals; the
# confinement is applied on the public os / io / Path.* APIs below instead.
_WD = _os.path.realpath(__WORKDIR__)
# Capture the path helpers/separator into IMMUTABLE guard-local names BEFORE user code
# runs. _within() otherwise reads os.path.realpath / os.fspath / os.sep off the live
# module every call, so sandboxed code could reassign os.path.realpath (e.g. to a lambda
# that echoes an in-workdir path) right before a write and have the guard approve an
# outside target while the real open() still writes there. These references cannot be
# rebound by mutating the os module.
_realpath = _os.path.realpath
_fspath = _os.fspath
_sep = _os.sep
_WD = _realpath(__WORKDIR__)
def _within(p):
try:
if isinstance(p, int):
return True
rp = _os.path.realpath(_os.fspath(p))
# os.path.realpath internally calls the LIVE os.fspath (posixpath.realpath does
# `filename = os.fspath(filename)`), so a sandboxed reassignment of os.fspath
# would still poison the resolution even though we hold the original realpath.
# Re-pin os.fspath to the captured original before resolving; a str target then
# resolves truthfully. (Re-pinning per check keeps it self-healing if user code
# re-patches; the real open() call receives the already-materialized str and does
# not route through os.fspath, so restoring it has no effect on the write itself.)
_os.fspath = _fspath
rp = _realpath(_fspath(p))
except Exception:
return False
return rp == _WD or rp.startswith(_WD + _os.sep)
return rp == _WD or rp.startswith(_WD + _sep)
def _deny(p, what):
raise PermissionError(
@ -4691,10 +4838,11 @@ def _gwraps(real):
def _fspath1(p):
# Materialize a path-like ONCE so a stateful __fspath__ cannot return a workdir
# path for the _within() check and a different path for the real syscall (TOCTOU).
# Uses the captured _fspath so a reassigned os.fspath cannot interpose here.
if isinstance(p, int):
return p
try:
return _os.fspath(p)
return _fspath(p)
except Exception:
return p

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

@ -607,4 +607,53 @@ def test_sandboxed_imports_still_work_under_guard():
disable_sandbox = False,
)
assert '{"a": 1}' in out
assert "sandbox:" not in out
@_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
# path, then writes to an absolute path outside the workdir. If the guard read
# os.path.realpath off the live module every call, the reassignment would fool
# _within() into approving the outside write. The guard captures the original path
# helpers at prelude time, so the write is still denied.
session = "backstop-realpath-monkeypatch"
workdir = get_sandbox_workdir(session)
target = tmp_path / "realpath_escape_probe.txt"
if target.exists():
target.unlink()
out = _python_exec(
"import os\n"
f"os.path.realpath = lambda p: {workdir!r} + '/ok'\n"
f"os.fspath = lambda p: {workdir!r} + '/ok'\n"
f"open({str(target)!r}, 'w').write('escaped')\n"
"print('WROTE-VIA-MONKEYPATCH')\n",
None,
30,
session,
disable_sandbox = False,
)
assert "sandbox:" in out or "PermissionError" in out
assert not target.exists()
@_POSIX_ONLY
def test_sandboxed_fspath_monkeypatch_write_escape_denied(tmp_path):
# The os.fspath twin of the realpath monkeypatch: reassigning os.fspath must not
# let a materialized outside path slip past the confinement check.
session = "backstop-fspath-monkeypatch"
workdir = get_sandbox_workdir(session)
target = tmp_path / "fspath_escape_probe.txt"
if target.exists():
target.unlink()
out = _python_exec(
"import os\n"
f"os.fspath = lambda p: {workdir!r} + '/ok'\n"
f"open({str(target)!r}, 'w').write('escaped')\n"
"print('WROTE-VIA-MONKEYPATCH')\n",
None,
30,
session,
disable_sandbox = False,
)
assert "sandbox:" in out or "PermissionError" in out
assert not target.exists()

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

@ -1083,3 +1083,139 @@ class TestEvalExecRecursion:
b"+AG8AcwAuAHMAeQBzAHQAZQBtACgAJwBpAGQAJwAp-"
)
_ok("exec(%r)" % benign)
class TestRound6Bypasses:
"""Sixth-round Codex findings: pathlib read args, getattr gadget dunders, namespace
.get() lookups, folded sys.modules keys, builtins __import__ aliases, deserializer
obfuscation, and code objects executed through types.FunctionType."""
def test_pathlib_open_read_resolved(self):
# open(Path('/etc') / 'passwd') carries no foldable string constant, but the
# pathlib resolver must reconstruct the path so it blocks like open('/etc/passwd').
assert (
_check_code_safety("from pathlib import Path\nopen(Path('/etc') / 'passwd').read()")
is not None
)
assert (
_check_code_safety(
"from pathlib import Path\nopen(Path('/etc').joinpath('passwd')).read()"
)
is not None
)
# A benign relative pathlib read stays allowed (no false positive).
_ok("from pathlib import Path\nopen(Path('data') / 'train.csv').read()")
@pytest.mark.parametrize(
"code",
[
"getattr(object, '__subclasses__')()",
"getattr(lambda: 0, '__globals__')",
"setattr(object, '__bases__', ())",
"getattr(getattr(object, '__subclasses__')()[0], '__init__')",
"getattr(().__class__, '__bases__')",
],
)
def test_getattr_gadget_dunder_any_receiver_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_getattr_benign_attr_allowed(self):
# A non-gadget attribute name via getattr on an ordinary object stays allowed.
_ok("getattr(object, 'mro')")
_ok("import numpy as np\ngetattr(np, 'zeros')((3, 3))")
@pytest.mark.parametrize(
"code",
[
"globals().get('__builtins__').__import__('os').system('id')",
"locals().get('__builtins__')",
"vars().get('os')",
],
)
def test_namespace_get_builtins_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_namespace_get_benign_key_allowed(self):
_ok("d = {'x': 1}\nd.get('x')")
_ok("globals().get('my_var')")
@pytest.mark.parametrize(
"code",
[
"import sys\nsys.modules['o' + 's'].system('id')",
"import sys\nsys.modules.get('o' + 's').system('id')",
"import sys\nk = 'o' + 's'\nsys.modules[k].system('id')",
],
)
def test_sys_modules_folded_key_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_sys_modules_dynamic_key_allowed(self):
# A genuinely dynamic key (not constant-foldable) stays allowed -- legit uses
# like sys.modules[name] for an unknown name must not be over-blocked.
_ok("import sys\ndef f(name):\n return sys.modules.get(name)\nf('json')")
def test_builtins_import_alias_blocked(self):
assert (
_check_code_safety("from builtins import __import__ as imp\nimp('os').system('id')")
is not None
)
assert (
_check_code_safety("import builtins\nbuiltins.__import__('os').system('id')")
is not None
)
@pytest.mark.parametrize(
"code",
[
"import pickle\ngetattr(pickle, 'loads')(b'x')",
"import pickle\nvars(pickle)['loads'](b'x')",
"import pickle\npickle.__dict__['loads'](b'x')",
"import pickle as p\ngetattr(p, 'loads')(b'x')",
],
)
def test_deserializer_attr_obfuscation_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_deserializer_benign_attr_allowed(self):
# getattr(pickle, 'dumps') (serialize) is not a code-exec sink -> allowed.
_ok("import pickle\ngetattr(pickle, 'dumps')({'a': 1})")
@pytest.mark.parametrize(
"code",
[
"import types\n"
"def f(src):\n types.FunctionType(compile(src, '<s>', 'exec'), {})()\nf('import os')",
"from types import FunctionType as F\n"
"def f(src):\n F(compile(src, '<s>', 'exec'), {})()\nf('x')",
"import types\n"
"def f(src):\n c = compile(src, '<s>', 'exec')\n types.FunctionType(c, {})()\nf('x')",
],
)
def test_functiontype_compile_result_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_functiontype_without_compile_allowed(self):
# types.FunctionType on an ordinary code object (fn.__code__) is not the
# dynamic-compile gadget; keep it allowed to avoid over-blocking metaprogramming.
_ok("import types\ndef g():\n return 1\ntypes.FunctionType(g.__code__, {})")
@pytest.mark.parametrize(
"code",
[
"import os\nos.system(\"python -c 'print(1)'\")",
"import os\nos.system('python3 evil.py')",
"import subprocess\nsubprocess.run(['python3', '-c', 'print(1)'])",
"import os\nos.system('perl -e \"print 1\"')",
"import os\nos.system('node -e \"1\"')",
],
)
def test_interpreter_child_process_blocked(self, code):
# A child interpreter runs WITHOUT the in-process write guard, so spawning one
# escapes the sandbox; interpreters are blocked at shell command position.
assert _check_code_safety(code) is not None, code
def test_benign_shell_still_allowed(self):
_ok("import os\nos.system('echo hello')")
_ok("import os\nos.system('ls -la')")
_ok("import subprocess\nsubprocess.run(['echo', 'hi'])")