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

Studio sandbox: close fifth-round review bypasses (aliases, reads, path guards)

Browse source 
Scope-aware assignment aliases (extends the per-scope index):
- Resolve single-assignment aliases of dangerous callables in the call's own scope:
  e = builtins.eval, im = importlib.import_module (and imp = __import__), and
  l = pickle.loads (incl. aliased modules). Previously only bare-name and from-import
  aliases were recognized.
- Count function parameters as local bindings so a parameter lexically shadows an
  outer sink alias of the same name (fixes a false positive where def f(s): s(...)
  with a module-level s = os.system flagged the parameter call).

Sensitive-read scanner:
- Resolve pathlib join receivers -- (Path('/etc') / 'passwd').read_text() and
  Path('/etc').joinpath('passwd') -- not just a bare Path(...) constructor.
- Normalize path spellings (collapse redundant separators / '.' and resolve '..')
  before the exact / dir checks, so /etc//passwd, /etc/./passwd and
  /tmp/../etc/passwd are matched.
- Fold function-local single-assignment path constants (def f(): p = '/etc/passwd';
  open(p)), not only module-level constants.
- Flag sys.modules.get('os') as the method-call twin of sys.modules['os'].

Runtime realpath backstop:
- Path.open coerces a str-subclass mode through the base str (matching the other open
  wrappers) so a lying __contains__ cannot skip the write check.
- Path.rename/replace/link materialize the target once so a stateful __fspath__ cannot
  return an in-workdir path for the check and an outside one for the real call (the
  pre-3.11 accessor path where this wrapper is the only confinement).

Adds regression tests across the classifier, aliasing and runtime-backstop suites.
This commit is contained in:
danielhanchen 2026-07-09 16:50:43 +00:00
commit 5dcb93f57d
4 changed files with 282 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2504,7 +2504,18 @@ class _ScopeAliasIndex:
(a real sink would be missed) nor cross-contaminate (a benign call in one function
would be flagged, or a dynamic exec in another wrongly treated as a safe alias)."""
__slots__ = ("tree", "node_scope", "enclosing", "shell", "execb", "compiled", "assigned")
__slots__ = (
"tree",
"node_scope",
"enclosing",
"shell",
"execb",
"compiled",
"impf",
"deser",
"strconst",
"assigned",
)
def __init__(self, tree):
self.tree = tree
@ -2513,6 +2524,9 @@ class _ScopeAliasIndex:
self.shell: dict = {}
self.execb: dict = {}
self.compiled: 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)
self.assigned: dict = {}
def _chain(self, node):
@ -2562,6 +2576,9 @@ def _build_scope_alias_index(tree, const_env):
os_aliases = {"os"}
subprocess_aliases = {"subprocess"}
from_aliases: dict[str, str] = {}
builtins_aliases = {"builtins", "__builtins__"}
importlib_aliases = {"importlib"}
deser_module_aliases: dict[str, str] = {m: m for m in _DESERIALIZE_MODULES}
for n in ast.walk(tree):
if isinstance(n, ast.Import):
for a in n.names:
@ -2569,12 +2586,54 @@ def _build_scope_alias_index(tree, const_env):
os_aliases.add(a.asname or "os")
elif a.name == "subprocess":
subprocess_aliases.add(a.asname or "subprocess")
elif a.name == "builtins":
builtins_aliases.add(a.asname or "builtins")
elif a.name == "importlib":
importlib_aliases.add(a.asname or "importlib")
if a.name in _DESERIALIZE_MODULES:
deser_module_aliases[a.asname or a.name] = a.name
elif isinstance(n, ast.ImportFrom) and n.module in ("os", "subprocess"):
for a in n.names:
fq = f"{n.module}.{a.name}"
if fq in _SHELL_SINK_FUNCS:
from_aliases[a.asname or a.name] = fq
def _rhs_exec_builtin(rhs):
# bare `exec` / `eval` / `compile`, or `builtins.eval` (attribute form).
if isinstance(rhs, ast.Name) and rhs.id in _EXEC_BUILTINS:
return rhs.id
if (
isinstance(rhs, ast.Attribute)
and rhs.attr in _EXEC_BUILTINS
and isinstance(rhs.value, ast.Name)
and rhs.value.id in builtins_aliases
):
return rhs.attr
return None
def _rhs_import_func(rhs):
# `__import__` / `importlib.import_module` (+ reload) bound to a name.
if isinstance(rhs, ast.Name) and rhs.id in ("__import__", "import_module"):
return True
if (
isinstance(rhs, ast.Attribute)
and rhs.attr in ("import_module", "reload", "__import__")
and isinstance(rhs.value, ast.Name)
and rhs.value.id in importlib_aliases
):
return True
return False
def _rhs_deserializer(rhs):
# `pickle.loads` (+ aliased module) bound to a name.
if isinstance(rhs, ast.Attribute) and isinstance(rhs.value, ast.Name):
canon = deser_module_aliases.get(rhs.value.id)
if canon is not None:
fq = f"{canon}.{rhs.attr}"
if fq in _CODE_DESERIALIZE_SINKS:
return fq
return None
scopes = [tree] + [
n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
]
@ -2583,6 +2642,15 @@ def _build_scope_alias_index(tree, const_env):
rebound: set[str] = set()
assigns: list[tuple[str, ast.expr]] = []
allnames: set[str] = set()
# Function parameters bind local names that lexically shadow an outer alias of
# the same name, so count them as local assignments for the shadowing rules.
_sargs = getattr(scope, "args", None)
if _sargs is not None:
for _a in list(_sargs.posonlyargs) + list(_sargs.args) + list(_sargs.kwonlyargs):
allnames.add(_a.arg)
for _extra in (_sargs.vararg, _sargs.kwarg):
if _extra is not None:
allnames.add(_extra.arg)
for n in _walk_scope_local(scope):
if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store):
counts[n.id] = counts.get(n.id, 0) + 1
@ -2599,14 +2667,18 @@ def _build_scope_alias_index(tree, const_env):
smap: dict[str, str] = {}
emap: dict[str, str] = {}
cmap: dict[str, tuple] = {}
imap: dict[str, bool] = {}
dmap: dict[str, str] = {}
scmap: dict[str, object] = {}
for name, rhs in assigns:
if counts.get(name) != 1 or name in rebound:
continue
fq = _resolve_static_shell_sink(rhs, os_aliases, subprocess_aliases, from_aliases)
if fq:
smap[name] = fq
if isinstance(rhs, ast.Name) and rhs.id in _EXEC_BUILTINS:
emap[name] = rhs.id
eb = _rhs_exec_builtin(rhs)
if eb is not None:
emap[name] = eb
elif (
isinstance(rhs, ast.Call)
and isinstance(rhs.func, ast.Name)
@ -2620,12 +2692,28 @@ def _build_scope_alias_index(tree, const_env):
_compile_mode(rhs, const_env),
isinstance(v, (bytes, bytearray)),
)
if _rhs_import_func(rhs):
imap[name] = True
dfq = _rhs_deserializer(rhs)
if dfq is not None:
dmap[name] = dfq
# Single-assignment string/bytes path constant (p = '/etc/passwd'), used by
# the sensitive-read scanner to fold function-local read paths.
cv = _const_fold(rhs, const_env)
if isinstance(cv, (str, bytes, bytearray)):
scmap[name] = cv
if smap:
idx.shell[scope] = smap
if emap:
idx.execb[scope] = emap
if cmap:
idx.compiled[scope] = cmap
if imap:
idx.impf[scope] = imap
if dmap:
idx.deser[scope] = dmap
if scmap:
idx.strconst[scope] = scmap
return idx
@ -3477,6 +3565,12 @@ def _check_signal_escape_patterns(
and func.attr == "__import__"
and _ast_name_matches(func.value, self.builtins_aliases)
)
or (
# single-assignment `im = importlib.import_module` in scope.
_analyzer_on
and isinstance(func, ast.Name)
and bool(_scope_idx.resolve(func.id, func, "impf"))
)
)
# Deserialization sinks reconstruct arbitrary objects/code from bytes.
# Resolve aliased imports (from pickle import loads as l), module aliases
@ -3491,6 +3585,9 @@ def _check_signal_escape_patterns(
_deser_fq = _cand
elif isinstance(func, ast.Name):
_deser_fq = self.deserialize_aliases.get(func.id)
if _deser_fq is None and _analyzer_on:
# single-assignment `l = pickle.loads` in the call's scope.
_deser_fq = _scope_idx.resolve(func.id, func, "deser")
if _deser_fq is None:
_fq_func = _fq_attr_name(func)
if _fq_func in _CODE_DESERIALIZE_SINKS:
@ -3569,6 +3666,18 @@ def _check_signal_escape_patterns(
dynamic_desc = (
f"{func.id}() on a sensitive module (attribute-name obfuscation)"
)
elif (
# sys.modules.get('os') -- the .get() twin of sys.modules['os'].
isinstance(func, ast.Attribute)
and func.attr == "get"
and isinstance(func.value, ast.Attribute)
and func.value.attr == "modules"
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:
dynamic_desc = "sys.modules.get(...) access to a sensitive module"
if dynamic_desc:
dynamic_exec.append(
{
@ -4308,24 +4417,60 @@ def _check_signal_escape_patterns(
"PureWindowsPath",
)
def _fold_read_arg(arg):
# Fold a read-path argument to a concrete string, resolving a module-level
# constant (via _const_env) OR a function-local single-assignment string
# constant (p = '/etc/passwd' inside a def) via the scope index.
v = _const_fold(arg, _const_env)
if isinstance(v, (str, bytes, bytearray)):
return _to_text(v)
if isinstance(arg, ast.Name):
sv = _scope_idx.resolve(arg.id, arg, "strconst")
if isinstance(sv, (str, bytes, bytearray)):
return _to_text(sv)
return None
def _pathlib_receiver_path(recv):
# Path(...) or pathlib.Path(...) (module-qualified). Join ALL string args so a
# multi-component constructor -- Path('/etc', 'passwd') -- resolves to the full
# path rather than only its first (non-sensitive) component.
if not isinstance(recv, ast.Call) or not recv.args:
# Resolve a pathlib receiver to a concrete path: Path(...) / pathlib.Path(...)
# (all constructor args joined), a `/` join (Path('/etc') / 'passwd'), or a
# .joinpath(...) chain.
if isinstance(recv, ast.BinOp) and isinstance(recv.op, ast.Div):
base = _pathlib_receiver_path(recv.left)
rv = _fold_read_arg(recv.right)
if base is None or rv is None:
return None
try:
return os.path.join(base, rv)
except Exception:
return None
if not isinstance(recv, ast.Call):
return None
rf = recv.func
if isinstance(rf, ast.Attribute) and rf.attr == "joinpath":
base = _pathlib_receiver_path(rf.value)
if base is None:
return None
parts = [base]
for a in recv.args:
v = _fold_read_arg(a)
if v is None:
return None
parts.append(v)
try:
return os.path.join(*parts)
except Exception:
return None
ctor = (isinstance(rf, ast.Name) and rf.id in _PATHLIB_CTORS) or (
isinstance(rf, ast.Attribute) and rf.attr in _PATHLIB_CTORS
)
if not ctor:
if not ctor or not recv.args:
return None
parts = []
for a in recv.args:
v = _const_fold(a, _const_env)
if not isinstance(v, (str, bytes, bytearray)):
v = _fold_read_arg(a)
if v is None:
return None
parts.append(_to_text(v))
parts.append(v)
if not parts:
return None
try:
@ -4335,7 +4480,14 @@ def _check_signal_escape_patterns(
def _flag_read_path(node, s, is_read_callee):
norm = s.replace("\\", "/")
if _is_sensitive_abs_path(norm):
# Collapse redundant separators / '.' segments and resolve '..' so equivalent
# spellings (/etc//passwd, /etc/./passwd, /tmp/../etc/passwd) still match the
# sensitive exact / dir checks. Keep the raw form for the traversal check below.
try:
canon = os.path.normpath(norm)
except Exception:
canon = norm
if _is_sensitive_abs_path(norm) or _is_sensitive_abs_path(canon):
_fs_block(node, f"{s!r} is a sensitive host identity / credential file")
return True
if is_read_callee and (s[:1] == "~" or ".." in norm.split("/")):
@ -4357,14 +4509,13 @@ def _check_signal_escape_patterns(
or fq in ("io.open", "os.open")
or method in _READ_METHODS
)
# Pathlib read on a literal Path(...) receiver: check the constructor path.
# Pathlib read on a Path(...) / join receiver: check the resolved path.
if isinstance(f, ast.Attribute) and f.attr in _PATHLIB_READ_METHODS:
rp = _pathlib_receiver_path(f.value)
if rp is not None and _flag_read_path(node, rp, True):
return
for arg in list(node.args) + [kw.value for kw in (node.keywords or [])]:
v = _const_fold(arg, _const_env)
s = _to_text(v) if isinstance(v, (str, bytes, bytearray)) else None
s = _fold_read_arg(arg)
if s is None:
continue
if _flag_read_path(node, s, is_read_callee):
@ -4738,8 +4889,8 @@ try:
_real_path_open = _pl.Path.open
@_gwraps(_real_path_open)
def _guarded_path_open(self, mode="r", *a, **k):
m = mode if isinstance(mode, str) else "r"
if any(c in m for c in "wax+") and not _within(self):
# Coerce mode through the base str (a str-subclass __contains__ must not lie).
if _mode_is_write(mode) and not _within(self):
_deny(str(self), "Path.open")
return _real_path_open(self, mode, *a, **k)
_pl.Path.open = _guarded_path_open
@ -4758,8 +4909,17 @@ try:
# accessor's ORIGINAL os.rename, so this wrapper is the only
# confinement -- check the keyword as well as the positional arg.
_t = a[0] if a else k.get("target")
if _t is not None and not _within(_t):
_deny(str(_t), "Path." + name + " target")
if _t is not None:
# Materialize the target once so a stateful __fspath__ cannot return
# an in-workdir path here and an outside one to the real call.
_tm = _fspath1(_t)
if not _within(_tm):
_deny(str(_tm), "Path." + name + " target")
if a:
a = (_tm,) + tuple(a[1:])
else:
k = dict(k)
k["target"] = _tm
return orig(self, *a, **k)
setattr(_pl.Path, name, w)
for _n in ("write_text", "write_bytes", "unlink", "mkdir", "rmdir", "chmod", "touch"):

7
studio/backend/tests/test_sandbox_aliasing.py
View file

@ -116,6 +116,13 @@ class TestPerScopeAliasCounting:
# the local binding inside that function.
_ok("import os\ns = os.system\ndef f():\n s = print\n s('please rm the files')\nf()")
def test_parameter_shadows_module_alias_allowed(self):
# A function parameter lexically shadows an outer sink alias of the same name,
# so a call through the parameter is not the sink (must not be a false positive).
_ok("import os\ns = os.system\ndef f(s):\n s('rm -rf /')\nf(print)")
_ok("import os\ns = os.system\ndef f(s, /):\n s('rm -rf /')\nf(print)")
_ok("def e_outer():\n pass\ne = exec\ndef g(e):\n e('x')\ng(print)")
class TestAliasingLowFalsePositive:
def test_reassigned_alias_not_treated_as_sink(self):

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

@ -528,6 +528,56 @@ def test_sandboxed_os_chmod_fd_denied(tmp_path):
assert oct(os.stat(victim).st_mode & 0o777) == "0o600"
@_POSIX_ONLY
def test_sandboxed_path_open_str_subclass_mode_denied(tmp_path):
# Path.open must coerce a str-subclass mode through the base str (a lying
# __contains__ must not defeat the write check). On 3.13 the underlying io.open
# guard also catches it; this asserts the write never lands regardless.
target = tmp_path / "pathopen_mode_escape.txt"
out = _python_exec(
"from pathlib import Path\n"
"class M(str):\n"
" def __contains__(self, c):\n"
" return False\n"
f"Path({str(target)!r}).open(M('w')).write('x')\nprint('DONE')",
None,
30,
"backstop-pathmode",
disable_sandbox = False,
)
assert not target.exists()
@_POSIX_ONLY
def test_sandboxed_path_rename_stateful_target_denied(tmp_path):
# Path.rename must materialize the target once so a stateful __fspath__ cannot
# return an in-workdir path for the check and an outside one for the real call.
target = tmp_path / "pathrename_target_escape.txt"
session = "backstop-pathrename-stateful"
workdir = get_sandbox_workdir(session)
src = os.path.join(workdir, "stateful_src.txt")
with open(src, "w") as f:
f.write("x")
try:
out = _python_exec(
"from pathlib import Path\n"
"class T:\n"
" n = 0\n"
" def __fspath__(self):\n"
" T.n += 1\n"
f" return 'okp.txt' if T.n == 1 else {str(target)!r}\n"
"Path('stateful_src.txt').rename(T())\nprint('DONE')",
None,
30,
session,
disable_sandbox = False,
)
assert not target.exists()
finally:
if os.path.exists(src):
os.remove(src)
@_POSIX_ONLY
def test_sandboxed_in_workdir_ops_still_work():
# The added guards must not break benign in-workdir writes.

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

@ -928,6 +928,52 @@ class TestReceiverAndVarsAndDynImportBypasses:
assert _check_code_safety(code) is None, code
class TestAssignedAliasesAndNormalization:
"""Fifth-round refinements: assignment aliases to dangerous callables, pathlib
join receivers, path normalization, function-local path constants, and the
sys.modules.get twin."""
@pytest.mark.parametrize(
"code",
[
# 978: pathlib join receivers (/ operator and joinpath).
"from pathlib import Path\n(Path('/etc') / 'passwd').read_text()",
"from pathlib import Path\nPath('/etc').joinpath('passwd').read_bytes()",
# 984: eval/exec aliased from the builtins module.
"import builtins\ne = builtins.eval\ne(\"__import__('os').system('rm -rf /')\")",
# 988: equivalent path spellings normalize to a sensitive file.
"open('/etc//passwd').read()",
"open('/etc/./passwd').read()",
"open('/tmp/../etc/passwd').read()",
# 996: function-local path constant.
"def f():\n p = '/etc/passwd'\n return open(p).read()\nf()",
# 998: assignment alias of a dynamic-import function.
"import importlib\nim = importlib.import_module\nim('os').system('rm -rf /')",
"imp = __import__\nimp('os').system('rm -rf /')",
# 003: assignment alias of a deserializer.
"import pickle\nl = pickle.loads\nl(payload)",
"import pickle as pk\nl = pk.loads\nl(data)",
# 005: sys.modules.get twin of the subscript form.
"import sys\nsys.modules.get('os').system('rm -rf /')",
],
)
def test_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
"from pathlib import Path\n(Path('data') / 'out.txt').read_text()",
"def f():\n p = 'data/out.txt'\n return open(p).read()\nf()",
"import importlib\nm = importlib.import_module\nm('numpy')",
"import sys\nm = sys.modules.get('numpy')",
"open('output/result.txt').read()",
],
)
def test_benign_allowed(self, code):
assert _check_code_safety(code) is None, code
class TestEvalExecRecursion:
"""Stage 2: eval/exec/compile are unwrapped, not blanket-banned. A safe
(constant-recoverable) payload is allowed; an obfuscated escape blocks."""