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

Harden sandbox classifier against round-12 mro/attrgetter/loader bypasses

Browse source 
Static classifier:
- Flag mro().pop(i) / __mro__.pop(i) base-class extraction alongside the subscript and __getitem__ forms.
- Normalize operator.attrgetter('name')(obj) as attribute access whether or not the result is immediately invoked, so attrgetter('__closure__')(open)[0] gadget recovery is caught.
- Resolve a container-hidden open alias (o = [open][0]; o('../../etc/passwd').read()) as a read callee.
- Fail closed on an opaque read path assembled from obfuscation primitives (open(''.join(map(chr, ...))).read()), matching the exec-payload obfuscation policy.
- Treat cd behind the command / builtin shell wrappers as a cwd escape before allowing a relative redirect.
- Block importlib file loaders as execution sinks (SourceFileLoader(...).load_module(), spec.loader.exec_module(...)).
- Add the in-cluster Kubernetes service-account credential path to the sensitive-read list.

Adds TestRound12Bypasses covering each vector plus benign controls.
This commit is contained in:
danielhanchen 2026-07-09 20:55:05 +00:00
commit 389303d3de
2 changed files with 158 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -434,6 +434,11 @@ def _find_blocked_commands(command: str) -> set[str]:
if tok in _SHELL_SEPARATORS or tok in _SHELL_KEYWORDS_AS_SEP:
_at_cmd = True
continue
if _at_cmd and _token_basename(tok) in ("command", "builtin"):
# `command` / `builtin` run the following shell builtin with its args, so a
# `command cd /tmp` still changes the cwd. Stay at command position so the cd
# behind the wrapper is inspected (bash `help command`/`help builtin`).
continue
if _at_cmd and _token_basename(tok) == "cd":
for k in range(i + 1, len(tokens)):
t = tokens[k]
@ -3152,6 +3157,10 @@ _SANDBOX_SENSITIVE_DIR_PARTS = (
"/.config/gcloud",
"/.kube/",
"/.docker/",
# In-cluster Kubernetes service-account credentials (token / ca.crt / namespace)
# mounted into every pod; reading the token impersonates the pod to the API server.
"/var/run/secrets/kubernetes.io/",
"/run/secrets/kubernetes.io/",
)
_SANDBOX_SENSITIVE_TOKENS = (
"id_rsa",
@ -4173,13 +4182,14 @@ def _check_signal_escape_patterns(
_attr_call = (node.args[0], node.args[1])
elif len(node.args) == 1:
_attr_call = (func.value, node.args[0])
elif isinstance(_ecf, ast.Call):
# operator.attrgetter('system')(os)(...) / attrgetter('eval')(builtins):
# attrgetter is the same attribute-fetch obfuscation as getattr, so map
# attrgetter('name')(obj) to the (obj, 'name') pair.
_ag_name = self._attrgetter_name(_ecf.func)
if _ag_name is not None and len(_ecf.args) == 1:
_attr_call = (_ecf.args[0], ast.Constant(value = _ag_name))
elif self._attrgetter_name(func) is not None and len(node.args) == 1:
# operator.attrgetter('name')(obj) evaluates to obj.name -- the same
# attribute-fetch obfuscation as getattr(obj, 'name'). Detect the
# attrgetter APPLICATION call itself (node.func is the attrgetter,
# node.args[0] is the object) so it is caught whether or not the result
# is immediately invoked: attrgetter('__closure__')(open)[0] and the
# chained attrgetter('system')(os)('rm -rf /') both normalize here.
_attr_call = (node.args[0], ast.Constant(value = self._attrgetter_name(func)))
is_dynamic_import = (
_ast_name_matches(_ecf, _DYNAMIC_IMPORT_FUNCS)
or (
@ -4428,14 +4438,13 @@ def _check_signal_escape_patterns(
):
dynamic_desc = "inspect.getclosurevars() recovers a guarded wrapper's closure"
elif (
# cls.mro().__getitem__(1) / cls.__mro__.__getitem__(1): the method-call
# twin of the subscripted-mro base extraction (visit_Subscript). Same
# gadget shape (io.FileIO.mro().__getitem__(1) recovers the original
# FileIO base), so flag a non-slice integer index via __getitem__.
# cls.mro().__getitem__(1) / .pop(1) / cls.__mro__.__getitem__(1): the
# method-call twin of the subscripted-mro base extraction
# (visit_Subscript). Same gadget shape (io.FileIO.mro().pop(1) recovers
# the original FileIO base), so flag an element-extraction method on an
# mro()/__mro__ receiver.
isinstance(func, ast.Attribute)
and func.attr == "__getitem__"
and len(node.args) == 1
and isinstance(_const_fold(node.args[0], _const_env), int)
and func.attr in ("__getitem__", "pop")
and (
(
isinstance(func.value, ast.Call)
@ -4445,20 +4454,33 @@ def _check_signal_escape_patterns(
)
or (isinstance(func.value, ast.Attribute) and func.value.attr == "__mro__")
)
and (
# pop() / pop(i) always extract an element; __getitem__ only when the
# index is a plain integer (not a slice object).
func.attr == "pop"
or (
len(node.args) == 1
and isinstance(_const_fold(node.args[0], _const_env), int)
)
)
):
dynamic_desc = "mro().__getitem__(i) extracts a base class (gadget)"
dynamic_desc = f"mro().{func.attr}(...) extracts a base class (gadget)"
elif isinstance(func, ast.Attribute) and func.attr in (
"runcode",
"runsource",
"load_module",
"exec_module",
):
# 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.
# .runsource(src) execute a code object / source string; an importlib file
# loader (SourceFileLoader(...).load_module() / spec.loader.exec_module(m))
# executes a local file. None run through the recursive analysis exec/eval
# receive, so an opaque payload (a written evil.py, a compile() result, or
# raw source) runs un-analyzed. These method names are unique to those
# interpreters / loaders, so flag the call regardless of receiver.
dynamic_desc = (
f"{func.attr}() executes code without static analysis "
"(code.InteractiveInterpreter / InteractiveConsole)"
f"{func.attr}() executes code / a file without static analysis "
"(code interpreter / importlib file loader)"
)
if dynamic_desc:
dynamic_exec.append(
@ -5290,14 +5312,31 @@ def _check_signal_escape_patterns(
elif _a.name == "subprocess":
_subprocess_mod_aliases.add(_a.asname or "subprocess")
def _unwrap_container_node(n):
# `[open][0]` / `(open,)[0]` / `{'k': open}['k']`: resolve an inline literal-container
# index to the element node so a container-hidden alias is seen through.
if not isinstance(n, ast.Subscript):
return n
container = n.value
ci = _const_fold(n.slice, _const_env)
if isinstance(container, (ast.List, ast.Tuple)) and isinstance(ci, int):
if -len(container.elts) <= ci < len(container.elts):
return 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 v
return n
def _resolves_to_open(fn):
# 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.
# single-assignment alias (o = open; o('../../etc/passwd').read()), a
# container-hidden alias (o = [open][0]; o(...)), or builtins.open / io.open /
# os.open.
if isinstance(fn, ast.Name):
if fn.id == "open" or fn.id in _open_from_aliases:
return True
rhs = _scope_idx.resolve(fn.id, fn, "rhsnode")
rhs = _unwrap_container_node(_scope_idx.resolve(fn.id, fn, "rhsnode"))
if isinstance(rhs, ast.Name) and rhs.id == "open":
return True
if (
@ -5538,6 +5577,14 @@ def _check_signal_escape_patterns(
rp = _pathlib_receiver_path(arg)
if rp is not None and _flag_read_path(node, rp, is_read_callee):
break
# An opaque read path assembled from obfuscation primitives
# (open(''.join(map(chr, [...]))).read()) can still target a host
# secret, and reads are not runtime-confined. Apply the same
# fail-closed obfuscation policy exec payloads get: block a read
# callee whose path is built from chr/join(map)/decode/fetch/... .
if is_read_callee and _payload_has_obfuscation_primitive(arg):
_fs_block(node, "read path assembled from obfuscation primitives")
break
continue
if _flag_read_path(node, s, is_read_callee):
break

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

@ -1808,3 +1808,90 @@ class TestRound11Bypasses:
def test_pathlib_wrapper_and_shutil_benign_allowed(self):
_ok("from pathlib import Path\nPath('data').joinpath('train.csv').resolve().read_text()")
_ok("from shutil import copy as c\nc('a.txt', 'b.txt')")
class TestRound12Bypasses:
"""Twelfth-round Codex findings: mro().pop base extraction, attrgetter not immediately
invoked, container-hidden open alias, opaque obfuscated read paths, cd behind
command/builtin, importlib file loaders, and Kubernetes service-account tokens."""
@pytest.mark.parametrize(
"code",
[
"import io\nio.FileIO.mro().pop(1)('/tmp/x', 'w')",
"import io\nio.FileIO.mro().pop()",
"import io\nio.FileIO.__mro__.pop(1)",
],
)
def test_mro_pop_base_extraction_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
"import operator\noperator.attrgetter('__closure__')(open)[0]",
"import operator\noperator.attrgetter('cell_contents')"
"(operator.attrgetter('__closure__')(open)[0])('/tmp/x','w')",
"from operator import attrgetter\nattrgetter('__globals__')(open)",
],
)
def test_attrgetter_gadget_not_invoked_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_container_hidden_open_alias_read_blocked(self):
assert _check_code_safety("o = [open][0]\no('../../../etc/passwd').read()") is not None
# Benign local write through the same alias stays allowed.
_ok("o = [open][0]\no('out.txt', 'w')")
@pytest.mark.parametrize(
"code",
[
"open(''.join(map(chr, [47,101,116,99,47,112,97,115,115,119,100]))).read()",
"import base64\nopen(base64.b64decode('L2V0Yy9wYXNzd2Q=').decode()).read()",
],
)
def test_opaque_obfuscated_read_path_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_opaque_read_path_benign_allowed(self):
_ok("fn = 'data/train.csv'\nopen(fn).read()")
_ok("import os\nopen(os.path.join('data', 'train.csv')).read()")
@pytest.mark.parametrize(
"code",
[
"import os\nos.system('command cd /tmp; printf x > p')",
"import os\nos.system('builtin cd /tmp && printf x > p')",
],
)
def test_cd_behind_shell_builtin_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_command_builtin_benign_allowed(self):
_ok("import os\nos.system('command ls')")
_ok("import os\nos.system('builtin echo hi')")
@pytest.mark.parametrize(
"code",
[
"import importlib.machinery\n"
"importlib.machinery.SourceFileLoader('m', 'evil.py').load_module()",
"spec.loader.exec_module(mod)",
],
)
def test_importlib_file_loader_blocked(self, code):
assert _check_code_safety(code) is not None, code
def test_importlib_import_module_benign_allowed(self):
_ok("import importlib\nimportlib.import_module('json')")
@pytest.mark.parametrize(
"code",
[
"open('/var/run/secrets/kubernetes.io/serviceaccount/token').read()",
"open('/var/run/secrets/kubernetes.io/serviceaccount/ca.crt').read()",
"open('/run/secrets/kubernetes.io/serviceaccount/token').read()",
],
)
def test_kubernetes_service_account_token_blocked(self, code):
assert _check_code_safety(code) is not None, code