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

Studio sandbox: close ninth-round review bypasses (module table, exec sinks, aliases, budget)

Browse source 
Static classifier:
- deny sys.modules Store/Del: del sys.modules['posix']; import posix drops the guard-patched module for a fresh unwrapped C module
- flag code.InteractiveInterpreter().runcode / InteractiveConsole().runsource as code-object execution sinks (opaque compile results run un-analyzed)
- resolve indirect read-only open callees: from os/io/builtins import open as X (os.open read-only is deliberately allowed outside the workdir, so traversal must be caught statically)
- fold os.path.normpath / abspath on literals so a traversal that only emerges after normalization is scanned
- resolve inline-container-hidden exec/eval: ({'e': exec}['e'])(...) / [exec][0](...)
- normalize the bound one-arg __getattribute__ form obj.__getattribute__('name') (builtins.open.__getattribute__('__closure__'))
- descend into a literal list/tuple argv so subprocess.run(['cat', '/etc/passwd']) is caught
- enforce the analyzer node budget: charge each tree's node count against _MAX_ANALYZER_NODES and fail closed above it (parent-process DoS guard)

Runtime backstop:
- capture and re-pin os.lstat / os.readlink / os.getcwd / os.stat before realpath, since it consults the live symlink helpers -- monkeypatching os.lstat to fail could stop realpath following an in-workdir symlink that points outside
This commit is contained in:
danielhanchen 2026-07-09 19:17:07 +00:00
commit ccac1f0293
3 changed files with 297 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2187,6 +2187,19 @@ def _fold_call(node, _state, _depth):
except Exception:
return None
return None
# os.path.normpath / abspath on a literal reveal the same sensitive / traversal
# path a direct string would (open(os.path.normpath('a/../../etc/passwd')) must
# not stay opaque). normpath is a pure string transform; abspath is only foldable
# for an already-absolute arg (a relative abspath depends on the runtime cwd,
# which is the in-workdir sandbox cwd, so it need not be folded).
if attr in ("normpath", "abspath") and _is_path_join_owner(owner):
if len(args) == 1 and isinstance(args[0], str):
try:
if attr == "normpath" or os.path.isabs(args[0]):
return _fold_cap(os.path.normpath(args[0]))
except Exception:
return None
return None
if isinstance(owner, ast.Name):
mod = owner.id
try:
@ -3161,6 +3174,38 @@ def _check_signal_escape_patterns(
"warnings": [],
}
# Charge this tree's node count against the shared analyzer budget BEFORE the several
# unbounded ast.walk / visitor passes below. A syntactically valid file (or a huge
# recovered exec/eval payload, since the budget is shared across the recursion) with
# hundreds of thousands of nodes would otherwise tie up the Studio parent process
# before the child rlimits apply. Fail closed (block) when the budget is exceeded.
if _budget is None:
_budget = _AnalyzerBudget()
try:
_budget.nodes += sum(1 for _ in ast.walk(tree))
except Exception: # pragma: no cover - defensive
pass
if _budget.nodes > _MAX_ANALYZER_NODES:
return False, {
"error": None,
"signal_tampering": [],
"exception_catching": [],
"shell_escapes": [],
"dynamic_exec": [
{
"type": "analyzer_budget",
"line": -1,
"description": (
"code exceeds the static-analysis node budget (too large to verify safely)"
),
}
],
"network_calls": [],
"sensitive_file_reads": [],
"filesystem_violations": [],
"warnings": [],
}
signal_tampering = []
exception_catching = []
shell_escapes = []
@ -3172,8 +3217,6 @@ def _check_signal_escape_patterns(
# Default on; UNSLOTH_STUDIO_SINK_ANALYZER=0 reverts to the legacy blanket
# eval/exec ban and disables filesystem-confinement + aliasing analysis.
_analyzer_on = os.environ.get("UNSLOTH_STUDIO_SINK_ANALYZER", "1") != "0"
if _budget is None:
_budget = _AnalyzerBudget()
if _analyzer_on:
try:
_const_env = _build_const_prop_env(tree)
@ -3547,6 +3590,39 @@ def _check_signal_escape_patterns(
return _elt(v)
return None
def _resolve_container_exec(self, sub):
"""Resolve an inline literal-container index callee to a dynamic-exec builtin.
Covers ({'e': exec}['e'])(...), [exec][0](...), (eval,)[0](...): an inline
container hiding an eval/exec/compile sink from the bare-name recursion."""
def _elt(elt):
if isinstance(elt, ast.Name):
if elt.id in _DYNAMIC_EXEC_BUILTINS:
return elt.id
if elt.id in self.exec_from_aliases:
return self.exec_from_aliases[elt.id]
if _analyzer_on:
return _scope_idx.resolve(elt.id, elt, "execb")
if (
isinstance(elt, ast.Attribute)
and elt.attr in _DYNAMIC_EXEC_BUILTINS
and _ast_name_matches(elt.value, self.builtins_aliases)
):
return elt.attr
return None
container = sub.value
ci = _const_fold(sub.slice, _const_env)
if isinstance(container, (ast.List, ast.Tuple)) and isinstance(ci, int):
if -len(container.elts) <= ci < len(container.elts):
return _elt(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 _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
@ -3755,6 +3831,10 @@ def _check_signal_escape_patterns(
and _ast_name_matches(func.value, self.builtins_aliases)
):
exec_func_id = func.attr # builtins.eval(...) / __builtins__.exec(...)
elif isinstance(func, ast.Subscript):
# ({'e': exec}['e'])(...) / [exec][0](...): an inline container hides the
# sink from the bare-name / attribute checks above.
exec_func_id = self._resolve_container_exec(func)
if exec_func_id is not None:
if _analyzer_on:
@ -3784,12 +3864,17 @@ def _check_signal_escape_patterns(
and len(node.args) >= 2
):
_attr_call = (node.args[0], node.args[1])
elif (
isinstance(func, ast.Attribute)
and func.attr in ("__getattribute__", "__getattr__")
and len(node.args) >= 2
elif isinstance(func, ast.Attribute) and func.attr in (
"__getattribute__",
"__getattr__",
):
_attr_call = (node.args[0], node.args[1])
# Unbound form object.__getattribute__(obj, 'name') carries the receiver
# as arg0; the BOUND form obj.__getattribute__('name') carries it as the
# attribute's own value (builtins.open.__getattribute__('__closure__')).
if len(node.args) >= 2:
_attr_call = (node.args[0], node.args[1])
elif len(node.args) == 1:
_attr_call = (func.value, node.args[0])
is_dynamic_import = (
_ast_name_matches(func, _DYNAMIC_IMPORT_FUNCS)
or (
@ -3975,6 +4060,19 @@ def _check_signal_escape_patterns(
dynamic_desc = (
f"runpy.{func.attr}() executes a file/module without static analysis"
)
elif isinstance(func, ast.Attribute) and func.attr in (
"runcode",
"runsource",
):
# code.InteractiveInterpreter().runcode(c) / InteractiveConsole()
# .runsource(src) execute a code object / source string without the
# recursive analysis exec/eval receive, so an opaque compile() result
# (or raw source) runs un-analyzed. These method names are unique to the
# code module's interpreters, so flag the call regardless of receiver.
dynamic_desc = (
f"{func.attr}() executes code without static analysis "
"(code.InteractiveInterpreter / InteractiveConsole)"
)
if dynamic_desc:
dynamic_exec.append(
{
@ -4062,6 +4160,18 @@ def _check_signal_escape_patterns(
"description": "sys.modules[...] access to a sensitive module",
}
)
if isinstance(node.ctx, (ast.Store, ast.Del)) and is_sys_modules:
# `del sys.modules['posix']; import posix` (or reassigning the entry) drops
# the guard-patched module object so a fresh, UNWRAPPED C module is imported,
# bypassing the Stage 5 wrappers. Mutating the loader table has no legitimate
# use in sandboxed compute, so deny any Store/Del on sys.modules[...].
dynamic_exec.append(
{
"type": "dynamic_exec",
"line": getattr(node, "lineno", -1),
"description": "sys.modules mutation (del / assign) can drop a guarded module",
}
)
# globals()['__builtins__'] / locals()[...] / vars()[...] pulls the builtins
# namespace (or a dangerous module) out of the namespace dict, e.g.
# getattr(globals()['__builtins__'], '__import__')('os'). Flag a Load of a
@ -4743,21 +4853,30 @@ def _check_signal_escape_patterns(
# import shutil as sh -> sh.copy('../../etc/passwd', 'x')
_pathlib_ctor_aliases = set(_PATHLIB_CTORS)
_shutil_aliases = {"shutil"}
# from os import open as oo / from io import open as X / from builtins import open as X:
# the read-only os.open is deliberately allowed OUTSIDE the workdir by the runtime
# guard, so a traversal read via such an alias must be caught statically.
_open_from_aliases: set[str] = set()
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.ImportFrom) and _imp.module in ("os", "io", "builtins"):
for _a in _imp.names:
if _a.name == "open":
_open_from_aliases.add(_a.asname or "open")
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.
# A callee that is `open`, a `from os/io/builtins import open as X` alias, a
# single-assignment alias (o = open; o('../../etc/passwd').read()), or
# builtins.open / io.open / os.open.
if isinstance(fn, ast.Name):
if fn.id == "open":
if fn.id == "open" or fn.id in _open_from_aliases:
return True
rhs = _scope_idx.resolve(fn.id, fn, "rhsnode")
if isinstance(rhs, ast.Name) and rhs.id == "open":
@ -4899,6 +5018,16 @@ def _check_signal_escape_patterns(
scan_args.extend(v for v in kw.value.values if v is not None)
else:
scan_args.append(kw.value)
# Descend into a literal list/tuple argv so a sensitive path element is
# scanned: subprocess.run(['cat', '/etc/passwd']) reads the host file in an
# unguarded child even though the top-level arg is a list, not a string.
_expanded = []
for a in scan_args:
if isinstance(a, (ast.List, ast.Tuple)):
_expanded.extend(a.elts)
else:
_expanded.append(a)
scan_args = _expanded
for arg in scan_args:
s = _fold_read_arg(arg)
if s is None:
@ -5059,6 +5188,14 @@ _realpath = _os.path.realpath
_fspath = _os.fspath
_fsdecode = _os.fsdecode
_sep = _os.sep
# os.path.realpath resolves symlinks by consulting the LIVE os.lstat / os.readlink (and
# os.getcwd for relative paths). Capture the originals so a monkeypatch of any of them --
# e.g. os.lstat raising so realpath stops FOLLOWING an in-workdir symlink that points
# outside -- cannot make _within() approve a path the real open() then escapes through.
_lstat = _os.lstat
_readlink = _os.readlink
_getcwd = _os.getcwd
_stat = _os.stat
_WD = _realpath(__WORKDIR__)
def _within(p):
@ -5066,13 +5203,18 @@ def _within(p):
if isinstance(p, int):
return True
# 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.)
# `filename = os.fspath(filename)`) and os.lstat / os.readlink / os.getcwd, so a
# sandboxed reassignment of any of them would poison the resolution even though we
# hold the original realpath. Re-pin them to the captured originals before
# resolving; the target then resolves truthfully (symlinks followed, cwd honest).
# Re-pinning per check keeps it self-healing if user code re-patches; the real
# open() receives the already-materialized str and does not route through these,
# so restoring them has no effect on the write itself.
_os.fspath = _fspath
_os.lstat = _lstat
_os.readlink = _readlink
_os.getcwd = _getcwd
_os.stat = _stat
rp = _realpath(_fspath(p))
# A bytes path resolves to bytes; normalize to str so the prefix compare against
# the str _WD does not raise (which would deny a legitimate in-workdir bytes write

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

@ -742,3 +742,38 @@ def test_sandboxed_fileio_base_via_mro_blocked():
)
assert "unsafe code detected" in out or "sandbox:" in out or "PermissionError" in out
assert not os.path.exists("/tmp/studio_mro_escape.txt")
@_POSIX_ONLY
def test_sandboxed_lstat_monkeypatch_symlink_escape_denied(tmp_path):
# A pre-existing in-workdir symlink points outside. Sandboxed code monkeypatches
# os.lstat to raise so os.path.realpath stops FOLLOWING the link, which would make
# _within() resolve to the in-workdir link path while the real open() escapes through
# it. The guard captures os.lstat/os.readlink and re-pins them before resolving, so
# the write is still denied.
session = "backstop-lstat-monkeypatch"
workdir = get_sandbox_workdir(session)
link = os.path.join(workdir, "lstat_escape_link")
if os.path.islink(link) or os.path.exists(link):
os.remove(link)
os.symlink(str(tmp_path), link)
target = tmp_path / "lstat_escape_probe.txt"
if target.exists():
target.unlink()
try:
out = _python_exec(
"import os\n"
"def _boom(*a, **k):\n"
" raise OSError('nope')\n"
"os.lstat = _boom\n"
"open('lstat_escape_link/lstat_escape_probe.txt', 'w').write('escaped')\n"
"print('LSTAT_WROTE')\n",
None,
30,
session,
disable_sandbox = False,
)
assert "sandbox:" in out or "PermissionError" in out
assert not target.exists()
finally:
os.remove(link)

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

@ -1446,3 +1446,107 @@ class TestRound8Bypasses:
# 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)()")
class TestRound9Bypasses:
"""Ninth-round Codex findings: sys.modules mutation, code-object execution sinks,
indirect open aliases, path-builder folding, container-hidden exec, bound
__getattribute__, literal sequence reads, and the analyzer node budget."""
@pytest.mark.parametrize(
"code",
[
"import sys, os\ndel sys.modules['posix']\nimport posix\nposix.open('/tmp/x', os.O_CREAT)",
"import sys\nsys.modules['os'] = None",
"import sys\ndel sys.modules['_io']",
],
)
def test_sys_modules_mutation_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
"import code\nc = compile(open('e.py').read(), '<s>', 'exec')\n"
"code.InteractiveInterpreter().runcode(c)",
"import code\ncode.InteractiveConsole().runsource('import os; os.system(\"id\")')",
],
)
def test_code_object_execution_sinks_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_indirect_open_alias_traversal_blocked(self):
assert (
_check_code_safety(
"from os import open as oo, O_RDONLY\noo('../../../etc/passwd', O_RDONLY)"
)
is not None
)
assert (
_check_code_safety(
"from io import open as io_open\nio_open('../../../etc/passwd').read()"
)
is not None
)
@pytest.mark.parametrize(
"code",
[
"import os\nopen(os.path.normpath('a/../../../../etc/passwd')).read()",
"import os\nopen(os.path.abspath('/etc/passwd')).read()",
"import os\nopen(os.path.normpath('/tmp/../etc/shadow')).read()",
],
)
def test_path_builder_fold_read_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_path_builder_benign_allowed(self):
_ok("import os\nopen(os.path.normpath('data/train.csv')).read()")
_ok("import os\nopen(os.path.abspath('out.txt'), 'w')")
@pytest.mark.parametrize(
"code",
[
"({'e': exec}['e'])(\"__import__('os').system('id')\")",
"[exec][0](\"__import__('os').system('rm -rf /')\")",
"(eval,)[0](\"__import__('os').system('id')\")",
],
)
def test_container_hidden_exec_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_bound_getattribute_gadget_blocked(self):
assert (
_check_code_safety("import builtins\nbuiltins.open.__getattribute__('__closure__')")
is not None
)
assert (
_check_code_safety(
"import builtins\nc = builtins.open.__getattribute__('__closure__')\n"
"c[0].__getattribute__('cell_contents')('/tmp/x', 'w')"
)
is not None
)
@pytest.mark.parametrize(
"code",
[
"import subprocess\nsubprocess.run(['cat', '/etc/passwd'])",
"import subprocess\nsubprocess.check_output(['cat', '/etc/shadow'])",
"import subprocess\nsubprocess.run(('cat', '/etc/passwd'))",
],
)
def test_literal_sequence_secret_read_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_literal_sequence_benign_allowed(self):
_ok("import subprocess\nsubprocess.run(['echo', 'hi'])")
_ok("import subprocess\nsubprocess.run(['ls', 'data'])")
def test_analyzer_node_budget_enforced(self):
big = "\n".join(f"a{i} = {i} + {i}" for i in range(60000))
msg = _check_code_safety(big)
assert msg is not None
assert "node budget" in msg
# A normal-sized program is unaffected.
_ok("x = 1 + 2\ny = [i for i in range(10)]")