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

Harden sandbox: compile/FunctionType gadgets, lambda/comprehension aliases, IFS + symlink redirects, workdir import shadowing

Browse source 
Round 17 review follow-ups on the Studio code-exec sandbox classifier and runtime guard:

- Track a code object built through a LOCAL compile alias (cfn = compile; co = cfn(src);
  types.FunctionType(co, {})()) by resolving the callee through the scope exec-builtin map,
  so the FunctionType execution gadget still gets the recursive payload analysis.
- Recognize type(lambda: None) as the function constructor: type(lambda: None) IS
  types.FunctionType, so type(lambda: None)(code, {})() executed a compile() code object
  without the eval/exec gate.
- Include lambda and comprehension scopes in the alias index: (lambda e=exec: e(payload))()
  and [e(payload) for e in [exec]] now resolve e back to the exec sink. Lambdas/comprehensions
  become their own alias scopes, and a one-element comprehension generator binds its target.
- Record annotated single-assignment aliases (e: object = exec; e(payload)) alongside plain
  assignments, so the AnnAssign RHS is analyzed.
- Expand ${IFS} / $IFS to whitespace before shell command matching, so a separator-obfuscated
  writer/reader (rm${IFS}-rf${IFS}/, cat${IFS}/etc/shadow) is tokenized as bash runs it.
- Import the child-guard's stdlib deps (os/io/pathlib/re) with the workdir stripped from
  sys.path, then restore it, so a malicious workdir/os.py or pathlib.py cannot shadow a guard
  import and run unguarded at import time.
- Fail closed on shell redirection to any real-file target (the unguarded child follows a
  pre-existing symlink); only fd duplications (>&2) and the standard device sinks
  (/dev/null, ...) are allowed. This also fixes a pre-existing false positive where a benign
  redirect to /dev/null was blocked.

Adds TestRound17Bypasses and a workdir-shadowing runtime test; full sandbox suite green.
This commit is contained in:
danielhanchen 2026-07-09 23:35:46 +00:00
commit cf7503da21
3 changed files with 261 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -208,6 +208,13 @@ _SHELL_KEYWORDS_AS_SEP = frozenset({"then", "do", "else", "elif"})
# POSIX / common shell binaries. A shell without an inline `-c` payload runs unscanned
# code (a script file, -s / stdin, or a bare stdin-reading shell), so it is denied.
_SHELL_BINARIES = frozenset({"bash", "sh", "zsh", "dash", "ksh", "csh", "tcsh", "fish"})
# The only shell redirection targets trusted without a realpath check: standard device
# sinks that cannot escape the workdir. Every other target (relative or absolute) fails
# closed, because the unguarded child follows symlinks and resolves relative names against a
# cwd the static scanner cannot verify (a pre-existing `out -> /tmp/host` symlink escapes).
_SAFE_REDIRECT_TARGETS = frozenset(
{"/dev/null", "/dev/zero", "/dev/full", "/dev/stdout", "/dev/stderr", "/dev/tty"}
)
# Coreutils that read + print file contents. A shell-expanded ($VAR / `cmd`) path passed
# to one of these can exfiltrate a host secret whose name the static scan cannot resolve.
_SHELL_READ_COMMANDS = frozenset(
@ -399,6 +406,18 @@ def _normalize_ansi_c_quotes(command: str) -> str:
return "".join(res)
_IFS_RE = re.compile(r"\$\{IFS[^}]*\}|\$IFS\b")
def _expand_ifs(command: str) -> str:
"""bash expands ${IFS} / $IFS to whitespace (default space/tab/newline) BEFORE word
splitting, so cat${IFS}/etc/shadow runs `cat /etc/shadow` in the child. Replace an IFS
reference with a space so the scanner tokenizes the command bash actually executes."""
if "IFS" not in command:
return command
return _IFS_RE.sub(" ", command)
def _find_blocked_commands(command: str) -> set[str]:
"""Detect blocked commands at shell command position only.
@ -413,8 +432,9 @@ def _find_blocked_commands(command: str) -> set[str]:
# Normalize bash ANSI-C ($'...') / locale ($"...") quoting first: shlex leaves
# `$'touch'` as `$touch`, so a writer/interpreter hidden behind ANSI-C quoting would
# never match the blocklist even though bash decodes and runs it.
command = _normalize_ansi_c_quotes(command)
# never match the blocklist even though bash decodes and runs it. Then expand ${IFS} to
# whitespace so a separator-obfuscated command (rm${IFS}-rf${IFS}/) is tokenized.
command = _expand_ifs(_normalize_ansi_c_quotes(command))
# punctuation_chars splits separators into their own tokens, so command
# position is detected even in `echo done; rm -rf x` (no whitespace) or
@ -611,12 +631,14 @@ def _find_blocked_commands(command: str) -> set[str]:
if not tok.startswith("-"):
_at_cmd_sh = False
# 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 SINGLE-component literal target (> out.txt) stays in the workdir
# cwd and is allowed; a NON-LITERAL target (variable / command substitution) cannot be
# verified, so it fails closed (`echo x > "$p"` could expand anywhere). Scanning tokens
# (not the raw string) avoids matching a `>` inside a quoted argument.
# Output redirection (> / >> / &> / N>) runs in an unguarded child shell that follows
# symlinks before any Python guard, so no filename target can be trusted: a relative
# single-component name (> out) may be a pre-existing symlink to an outside file, a
# relative multi-component name (> sub/out) may traverse a symlinked subdir, an absolute
# / ~ / .. target is plainly outside, and a $ / backtick target can expand anywhere.
# Fail closed on every real-file target; only fd duplications (>&2) and the standard
# device sinks (/dev/null, ...) are 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:
@ -625,8 +647,7 @@ def _find_blocked_commands(command: str) -> set[str]:
j = i
# `>|` (noclobber override) and `>&` (stdout+stderr / fd-or-file redirect) tokenize
# as `>` then `|` / `&`, so that punctuation is part of the redirect operator, not a
# pipeline / background op; skip it and take the real target after. A pure fd target
# (`>&2`) is a bare number that fails the path checks below and stays allowed.
# pipeline / background op; skip it and take the real target after.
if not tgt and j + 1 < len(tokens) and tokens[j + 1] in ("|", "&"):
j += 1
if not tgt and j + 1 < len(tokens):
@ -634,21 +655,11 @@ def _find_blocked_commands(command: str) -> set[str]:
if not tgt:
continue
tn = tgt.replace("\\", "/")
# A relative multi-component target (sub/out.txt) resolves through a subdirectory
# component whose realpath the static scanner cannot verify -- if that component is
# a symlink pointing outside the workdir the unguarded child writes past it. Fail
# closed on any relative target carrying a `/` separator (a leading `./` is dropped
# first so `./out.txt` stays allowed); only a bare single-component name is allowed.
_rel = tn[2:] if tn.startswith("./") else tn
if (
tgt.startswith("~")
or tn.startswith("/")
or ".." in tn.split("/")
or "$" in tgt
or "`" in tgt
or "/" in _rel.rstrip("/")
):
blocked.add("redirect:" + tgt)
# Allowed: a pure fd duplication (>&2, >&1 -> `&2` / a bare digit) and the safe
# device sinks. Everything else is a file target that fails closed.
if tgt.startswith("&") or tgt.isdigit() or tn in _SAFE_REDIRECT_TARGETS:
continue
blocked.add("redirect:" + tgt)
# `cd` / `pushd` to a dir OUTSIDE the workdir moves the child shell's cwd so a later
# relative redirect / write escapes (`cd /tmp; echo x > p`, `pushd /tmp; echo x > p`).
@ -2873,8 +2884,22 @@ def _compile_mode(node, const_env):
def _walk_scope_local(scope):
"""Yield descendants of ``scope``'s body that share its namespace, WITHOUT
descending into nested def / lambda / class / comprehension (each of which is a
new scope). Used so single-assignment alias detection is scope-correct."""
stack = list(getattr(scope, "body", []))
new scope). Used so single-assignment alias detection is scope-correct.
When ``scope`` is itself a lambda or comprehension, walk its own namespace: a lambda
body is a single expression, and a comprehension's namespace holds its element
expression plus the generator iterables / conditions (target bindings are collected
separately). ``getattr(scope, "body", [])`` only applies to def / class / module."""
if isinstance(scope, ast.Lambda):
stack = [scope.body]
elif isinstance(scope, (ast.ListComp, ast.SetComp, ast.GeneratorExp)):
stack = [scope.elt] + [g for gen in scope.generators for g in [gen.iter, *gen.ifs]]
elif isinstance(scope, ast.DictComp):
stack = [scope.key, scope.value] + [
g for gen in scope.generators for g in [gen.iter, *gen.ifs]
]
else:
stack = list(getattr(scope, "body", []))
_NESTED = (
ast.FunctionDef,
ast.AsyncFunctionDef,
@ -2989,9 +3014,17 @@ def _build_scope_alias_index(tree, const_env):
# 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)):
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)):
# A lambda, like a def, opens its own scope: a default-bound param alias
# ((lambda e=exec: e(payload))()) lives in the lambda body's namespace, and
# anything nested inside encloses to the lambda itself.
idx.enclosing[child] = func_enclose
_rec(child, child, child)
elif isinstance(child, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)):
# A comprehension opens its own scope in Python 3; its target bindings
# ([e(p) for e in [exec]]) belong to that scope, enclosing to func_enclose.
idx.enclosing[child] = func_enclose
_rec(child, child, func_enclose)
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
@ -3112,7 +3145,19 @@ def _build_scope_alias_index(tree, const_env):
scopes = [tree] + [
n
for n in ast.walk(tree)
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
if isinstance(
n,
(
ast.FunctionDef,
ast.AsyncFunctionDef,
ast.ClassDef,
ast.Lambda,
ast.ListComp,
ast.SetComp,
ast.DictComp,
ast.GeneratorExp,
),
)
]
for scope in scopes:
counts: dict[str, int] = {}
@ -3140,6 +3185,27 @@ def _build_scope_alias_index(tree, const_env):
and isinstance(n.targets[0], ast.Name)
):
assigns.append((n.targets[0].id, n.value))
elif (
# An annotated single-assignment (e: object = exec) is a real binding whose
# RHS must be recorded like a plain Assign, else e(payload) skips analysis.
isinstance(n, ast.AnnAssign)
and n.value is not None
and isinstance(n.target, ast.Name)
):
assigns.append((n.target.id, n.value))
# A comprehension generator binds its target like a single-assignment alias when the
# iterable is a one-element literal: [e(p) for e in [exec]] binds e to exec, so the
# payload passed through e must still get eval/exec recursion.
for _gen in getattr(scope, "generators", []):
if (
isinstance(_gen.target, ast.Name)
and isinstance(_gen.iter, (ast.List, ast.Tuple, ast.Set))
and len(_gen.iter.elts) == 1
):
_tn = _gen.target.id
counts[_tn] = counts.get(_tn, 0) + 1
allnames.add(_tn)
assigns.append((_tn, _gen.iter.elts[0]))
idx.assigned[scope] = allnames
smap: dict[str, str] = {}
emap: dict[str, str] = {}
@ -3164,7 +3230,17 @@ def _build_scope_alias_index(tree, const_env):
eb = _rhs_exec_builtin(rhs_eff)
if eb is not None:
emap[name] = eb
elif _rhs_is_compile_call(rhs_eff) and rhs_eff.args:
elif (
_rhs_is_compile_call(rhs_eff)
or (
# A local alias of compile (cfn = compile; co = cfn(src, ...)) is not in
# compile_aliases, so resolve the callee through this scope's exec-builtin
# map (built in source order, cfn precedes co) before giving up.
isinstance(rhs_eff, ast.Call)
and isinstance(rhs_eff.func, ast.Name)
and emap.get(rhs_eff.func.id) == "compile"
)
) and rhs_eff.args:
# Any `c = compile(...)` (bare / builtins.compile / from-import alias)
# binds a code object, tracked for the types.FunctionType(c) execution
# gadget below (dynamic or foldable payload).
@ -4761,6 +4837,23 @@ def _check_signal_escape_patterns(
and _ast_name_matches(func.value, self.types_aliases)
)
or (isinstance(func, ast.Name) and func.id in self.functiontype_aliases)
or (
# The same constructor is reachable as type(lambda: None): the
# type of any function IS types.FunctionType, so
# type(lambda: None)(code, {})() executes a code object too.
isinstance(func, ast.Call)
and not func.keywords
and len(func.args) == 1
and isinstance(func.args[0], ast.Lambda)
and (
(isinstance(func.func, ast.Name) and func.func.id == "type")
or (
isinstance(func.func, ast.Attribute)
and func.func.attr == "type"
and _ast_name_matches(func.func.value, self.builtins_aliases)
)
)
)
)
and node.args
and self._is_compile_result(node.args[0])
@ -6013,8 +6106,9 @@ def _check_signal_escape_patterns(
# not runtime-confined, so these must be blocked statically.
if cmd is None:
return False
# Normalize ANSI-C ($'...') quoting so an obfuscated reader / path is seen.
cmd = _normalize_ansi_c_quotes(cmd)
# Normalize ANSI-C ($'...') quoting and expand ${IFS} so an obfuscated reader /
# path (cat${IFS}/etc/shadow) is seen the way bash runs it.
cmd = _expand_ifs(_normalize_ansi_c_quotes(cmd))
# Literal-path scan (absolute-sensitive + traversal) on plain whitespace tokens.
try:
toks = shlex.split(cmd, posix = True)
@ -6316,7 +6410,18 @@ def _truncate(text: str, limit: int = _MAX_OUTPUT_CHARS) -> str:
# and the realpath-before-open TOCTOU window under adversarial in-sandbox threading.
# --------------------------------------------------------------------------
_SANDBOX_GUARD_SRC = r"""
import sys as _sys
# The exec script lives INSIDE the workdir, so Python prepends the workdir to sys.path[0].
# A malicious workdir/os.py / io.py / pathlib.py / re.py (dropped by a prior run or upload)
# would otherwise shadow the guard's OWN imports below and execute unguarded at import time,
# before any patch is installed. Import the guard's stdlib deps with the workdir / cwd
# stripped from the path, then restore it so ordinary user imports still resolve (os / io /
# pathlib / re are now cached as the real, patched modules). `import sys` is safe: sys is a
# built-in module, never loaded from a file.
_saved_path = list(_sys.path)
_sys.path = [_p for _p in _sys.path if _p not in ("", ".", __WORKDIR__, __WORKDIR__ + "/")]
import os as _os, builtins as _bi, io as _io, pathlib as _pl, re as _re
_sys.path = _saved_path
# io + pathlib are imported BEFORE any patching on purpose: on Python <= 3.11
# pathlib._NormalAccessor captures io.open / os.* into class attributes at import
# time. A C builtin captured there does not bind on instance access, but a Python

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

@ -970,3 +970,44 @@ def test_sandboxed_pathlib_local_read_allowed():
)
assert "GOT hi" in out
assert "sandbox:" not in out
@_POSIX_ONLY
def test_sandboxed_workdir_module_shadowing_neutralized(tmp_path):
# The exec script lives in the workdir, so Python prepends the workdir to sys.path[0].
# A malicious re.py / pathlib.py / os.py / io.py dropped in the workdir must NOT shadow
# the guard's own imports (which would run unguarded at import time before any patch).
session = "backstop-shadow"
workdir = get_sandbox_workdir(session)
marker = os.path.join(str(tmp_path), "shadow_ran.marker")
evil = "import builtins as _b\n_b.open(%r, 'w').write('pwned')\nraise SystemExit\n" % marker
written = []
for name in ("re.py", "pathlib.py", "os.py", "io.py"):
p = os.path.join(workdir, name)
with open(p, "w") as fh:
fh.write(evil)
written.append(p)
try:
# A benign snippet: if any guard import is shadowed, evil runs and writes the marker.
out = _python_exec("print('OK', 1 + 1)", None, 30, session, disable_sandbox = False)
assert "OK 2" in out
assert not os.path.exists(marker), "workdir module shadowed a guard import"
# The real, patched modules stay usable for ordinary user imports.
out2 = _python_exec(
"import re, pathlib\nprint('REOK', bool(re.match('a', 'abc')))\n",
None,
30,
session,
disable_sandbox = False,
)
assert "REOK True" in out2
finally:
for p in written:
if os.path.exists(p):
os.remove(p)
_pyc = os.path.join(workdir, "__pycache__")
if os.path.isdir(_pyc):
import shutil
shutil.rmtree(_pyc, ignore_errors = True)
if os.path.exists(marker):
os.remove(marker)

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

@ -1392,9 +1392,10 @@ class TestRound8Bypasses:
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')")
# Redirects fail closed on file targets (an unguarded child follows symlinks), but
# fd duplications and the safe device sinks stay allowed.
_ok("import os\nos.system('ls 2>&1')")
_ok("import os\nos.system('echo hi > /dev/null')")
@pytest.mark.parametrize(
"code",
@ -1586,8 +1587,10 @@ class TestRound10Bypasses:
assert _check_code_safety(code) is not None, code
def test_benign_relative_redirect_and_cd_allowed(self):
_ok("import os\nos.system('cd data && echo x > out.txt')")
_ok("import os\nos.system('echo hi > local.txt')")
# cd to a relative in-workdir dir stays allowed; the redirect itself must target a
# safe device sink now that file targets fail closed.
_ok("import os\nos.system('cd data && echo x > /dev/null')")
_ok("import os\nos.system('cd data && ls')")
@pytest.mark.parametrize(
"code",
@ -1936,7 +1939,9 @@ class TestRound13Bypasses:
assert _check_code_safety(code) is not None, code
def test_pushd_relative_allowed(self):
_ok("import os\nos.system('pushd data; echo hi > out.txt')")
# pushd to a relative in-workdir dir stays allowed; a file redirect now fails closed,
# so pair it with a safe device sink.
_ok("import os\nos.system('pushd data; echo hi > /dev/null')")
@pytest.mark.parametrize(
"code",
@ -2180,11 +2185,6 @@ class TestRound15Bypasses:
def test_relative_multicomponent_redirect_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_single_component_redirect_allowed(self):
# A bare single-component relative redirect target stays in the workdir cwd.
_ok("import os\nos.system('echo x > out.txt')")
_ok("import os\nos.system('echo x > ./out.txt')")
class TestRound16Bypasses:
"""Sixteenth-round Codex findings: a dynamic Path read, sensitive reads inside a
@ -2258,3 +2258,76 @@ class TestRound16Bypasses:
# A relative glob expands only within the workdir cwd, so it stays allowed.
_ok("import os\nos.system('grep foo *.txt')")
_ok("import os\nos.system('echo *.py')")
class TestRound17Bypasses:
"""Seventeenth-round Codex findings: compile via local alias, type(lambda) function
constructor, lambda / comprehension alias scopes, annotated single-assignment aliases,
${IFS}-obfuscated shell words, workdir-shadowed guard imports, and redirects that follow
a pre-existing symlink."""
_SH = r"import os\nos.system('cat /etc/shadow')"
def test_compile_local_alias_functiontype_blocked(self):
code = (
"import types\ncfn = compile\nco = cfn(\"%s\", '<s>', 'exec')\n"
"types.FunctionType(co, {})()" % self._SH
)
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"ctor",
["type(lambda: None)", "type(lambda: 0)", "(lambda: None).__class__"],
)
def test_type_lambda_function_constructor_blocked(self, ctor):
# type(lambda: None) IS types.FunctionType; running a compile() code object through
# it bypasses the eval/exec gate. (__class__ is covered by the gadget-dunder scan.)
code = "co = compile(\"%s\", '<s>', 'exec')\n%s(co, {})()" % (self._SH, ctor)
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
'(lambda e=exec: e("' + _SH + '"))()',
'[e("' + _SH + '") for e in [exec]]',
'list(e("' + _SH + '") for e in (exec,))',
'{e("' + _SH + '") for e in [exec]}',
],
)
def test_lambda_comprehension_alias_scopes_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_annotated_assignment_alias_blocked(self):
assert _check_code_safety('e: object = exec\ne("' + self._SH + '")') is not None
@pytest.mark.parametrize(
"code",
[
"import os\nos.system('cat${IFS}/etc/shadow')",
"import os\nos.system('head$IFS/etc/passwd')",
"import os\nos.system('cat${IFS%?}/etc/shadow')",
"import os\nos.system('rm${IFS}-rf${IFS}/tmp/x')",
],
)
def test_ifs_expanded_shell_words_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_benign_ifs_echo_allowed(self):
_ok("import os\nos.system('echo $IFS')")
@pytest.mark.parametrize(
"code",
[
"import os\nos.system('echo x > out')",
"import os\nos.system('echo x > out.txt')",
"import os\nos.system('echo x >> log')",
],
)
def test_symlink_prone_redirect_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_safe_device_redirect_allowed(self):
# fd duplications and the standard device sinks are the only trusted targets.
_ok("import os\nos.system('echo hi > /dev/null')")
_ok("import os\nos.system('ls 2>&1')")
_ok("import os\nos.system('echo hi >> /dev/null 2>&1')")