From 52e68507ff0e0395d9cc6b7dc10cd06bc01834f5 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Fri, 10 Jul 2026 00:42:34 +0000 Subject: [PATCH] Harden sandbox: global/nonlocal aliases, wrapper-hidden commands, sys.modules aliases, descriptor gadgets, higher-order containers, keyword path guard Round 19 review follow-ups on the Studio code-exec sandbox classifier and runtime guard: - Record a global/nonlocal sink alias in its TARGET scope (module for global, enclosing for nonlocal): global s; s = os.system; s('...') now resolves s to the shell sink instead of being skipped as a rebound name (and a global name no longer shadows an outer alias). - Fail closed on any expansion in COMMAND POSITION, not just command substitution: a variable-expanded command word (p=python3; $p -c ...) / ${VAR} is unprovable. Argument position ($HOME, echo $(date)) stays allowed. - Honor wrapper/command-position logic in the mutating-utility and shell-script scans via a shared command-word index, so env sed -i / timeout 5 bash s.sh no longer hide behind a wrapper prefix. - Use the subprocess-exec callee resolver for the shell=True read scan, so from subprocess import run as r; r('head /etc/passwd', shell=True) and r = subprocess.run aliases are tokenized. - Resolve sys.modules aliases (m = sys.modules; m.pop('_io')) for the loader-table mutation checks (bound and unbound forms). - Treat a __dict__ subscript keyed by a gadget dunder (type(open).__dict__['__closure__']. __get__(open)) as gadget access, closing the descriptor-lookup route around the attribute gadget scan. - Unwrap inline literal containers in the higher-order sink check, so list(map([eval][0], [...])) / map({'e': exec}['e'], ...) are flagged. - Runtime: the mutator guard now accepts the path via its public keyword (os.makedirs(name=), os.mkdir(path=)) instead of raising TypeError, while still confining the write. Adds TestRound19Bypasses and keyword-path runtime tests; full sandbox suite green. --- studio/backend/core/inference/tools.py | 284 +++++++++++++----- .../tests/test_sandbox_runtime_backstop.py | 38 +++ studio/backend/tests/test_sandbox_tools.py | 84 ++++++ 3 files changed, 333 insertions(+), 73 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index b330e8c698..5279d1775a 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -591,45 +591,87 @@ def _find_blocked_commands(command: str) -> set[str]: blocked |= _find_blocked_commands(payload) break + def _command_word_indices(): + # Indices of the REAL command word at each command position, skipping FOO=bar + # assignments and wrapper prefixes (env / nice / timeout / xargs / ...) plus their + # numeric / separated-option arguments, so `env sed`, `timeout 5 bash` resolve to + # sed / bash. Mirrors the main command-position scan above. + out = [] + expect = True + pending = False + prev_flag = False + for _i, _tok in enumerate(tokens): + if _tok in _SHELL_SEPARATORS or _tok in _SHELL_KEYWORDS_AS_SEP: + expect = True + pending = False + prev_flag = False + continue + if _tok.startswith("-"): + if not pending: + expect = False + else: + prev_flag = True + continue + if not expect: + continue + if _ASSIGNMENT_RE.match(_tok): + continue + if pending and _is_wrapper_numeric_arg(_tok): + prev_flag = False + continue + _base = _token_basename(_tok) + if ( + pending + and prev_flag + and _base not in _BLOCKED_COMMANDS + and _base not in _COMMAND_PREFIXES + ): + prev_flag = False + continue + prev_flag = False + if _base in _COMMAND_PREFIXES: + pending = True + continue + out.append(_i) + expect = False + pending = False + return out + + _cmd_word_idx = _command_word_indices() + # A shell binary invoked with a SCRIPT FILE (`bash s.sh`) or `-s` (read the script from # stdin) runs unscanned shell code in the same unguarded environment; only the inline # `-c '...'` form is statically analyzable (handled above). Block a command-position - # shell whose operands include a non-flag argument (the script) and no -c/-lc flag. - _at_cmd_sh = True - for i, tok in enumerate(tokens): - if tok in _SHELL_SEPARATORS or tok in _SHELL_KEYWORDS_AS_SEP: - _at_cmd_sh = True + # shell whose operands include a non-flag argument (the script) and no -c/-lc flag. Using + # the wrapper-aware command-word indices so `env bash s.sh` / `timeout 5 bash s.sh` are + # not hidden behind the wrapper prefix. + for i in _cmd_word_idx: + tok = tokens[i] + if os.path.basename(tok).lower() not in _SHELLS: continue - if _at_cmd_sh and os.path.basename(tok).lower() in _SHELLS: - _has_c = False - _script = None - for k in range(i + 1, len(tokens)): - t = tokens[k] - if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: - break - tl = t.lower() - if tl == "-c" or ( - tl.startswith("-") and not tl.startswith("--") and tl.endswith("c") - ): - _has_c = True - break - if tl in ("-s", "--"): # -s reads the script from stdin (unscanned) - _script = t - break - if t.startswith("-"): - continue # other shell flags: -l, -x, --login, --norc, ... - _script = t # first non-flag operand is the script file + _has_c = False + _script = None + for k in range(i + 1, len(tokens)): + t = tokens[k] + if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break - # Any command-position shell WITHOUT an inline `-c` payload runs unscanned code: - # a script file (bash s.sh), stdin via -s, or a bare shell that reads stdin - # (`printf 'evil' | bash`). Only the `-c '...'` form is statically analyzable, so - # block everything else. - if not _has_c: - blocked.add("shell-script:" + (_script or _token_basename(tok))) - _at_cmd_sh = False - continue - if not tok.startswith("-"): - _at_cmd_sh = False + tl = t.lower() + if tl == "-c" or (tl.startswith("-") and not tl.startswith("--") and tl.endswith("c")): + _has_c = True + break + if tl in ("-s", "--"): # -s reads the script from stdin (unscanned) + _script = t + break + if t.startswith("-"): + continue # other shell flags: -l, -x, --login, --norc, ... + _script = t # first non-flag operand is the script file + break + # Any command-position shell WITHOUT an inline `-c` payload runs unscanned code: + # a script file (bash s.sh), stdin via -s, or a bare shell that reads stdin + # (`printf 'evil' | bash`). Only the `-c '...'` form is statically analyzable, so + # block everything else. + if not _has_c: + blocked.add("shell-script:" + (_script or _token_basename(tok))) # 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 @@ -696,26 +738,22 @@ def _find_blocked_commands(command: str) -> set[str]: if not tok.startswith("-"): _at_cmd = False - # A command substitution in COMMAND POSITION ($(cmd) / `cmd` as the command word) runs - # whatever it expands to as the command name; the inner command may be benign (printf - # touch) while the expansion is a writer/interpreter (touch). The scanner cannot prove - # the expansion safe, so fail closed. (An argument-position substitution -- echo $(date), - # x=$(cmd) -- is not command-position and stays allowed.) - if re.search(r"(?:^|[\n;&|(])\s*(?:\$\(|`)", command): - blocked.add("command-substitution") + # An EXPANSION in COMMAND POSITION runs whatever it expands to as the command name and + # cannot be proven safe: a command substitution ($(printf touch) / `printf touch`), a + # variable-expanded command word (p=python3; $p -c ...), or a ${VAR} parameter expansion. + # Fail closed. (An argument-position expansion -- echo $(date), echo $HOME, x=$(cmd) -- is + # not at command position, so it stays allowed. ${IFS} is already expanded to whitespace + # above, so a `cat${IFS}x` command word is not misread as an expansion here.) + if re.search(r"(?:^|[\n;&|(])\s*(?:\$|`)", command): + blocked.add("command-expansion") # Some normally read-only utilities MUTATE files with certain flags (sed -i, sort -o # FILE, find ... -delete, dd of=FILE, tee FILE, truncate), writing/deleting OUTSIDE the # workdir in an unguarded child that no redirect token exposes. Treat the mutating - # invocation as a child writer. - _at_cmd = True - for i, tok in enumerate(tokens): - if tok in _SHELL_SEPARATORS or tok in _SHELL_KEYWORDS_AS_SEP: - _at_cmd = True - continue - if not _at_cmd: - continue - _at_cmd = False + # invocation as a child writer. Uses the wrapper-aware command-word indices so a wrapper + # prefix (env sed -i ..., nice sed -i ...) does not hide the mutating utility. + for i in _cmd_word_idx: + tok = tokens[i] _base = _token_basename(tok) if _base not in ("sed", "gsed", "ssed", "perl", "sort", "find", "dd", "tee", "truncate"): continue @@ -3227,6 +3265,8 @@ def _build_scope_alias_index(tree, const_env): for scope in scopes: counts: dict[str, int] = {} rebound: set[str] = set() + global_names: set[str] = set() + nonlocal_names: 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 @@ -3242,8 +3282,12 @@ def _build_scope_alias_index(tree, const_env): if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store): counts[n.id] = counts.get(n.id, 0) + 1 allnames.add(n.id) - elif isinstance(n, (ast.Global, ast.Nonlocal)): + elif isinstance(n, ast.Global): rebound.update(n.names) + global_names.update(n.names) + elif isinstance(n, ast.Nonlocal): + rebound.update(n.names) + nonlocal_names.update(n.names) elif ( isinstance(n, ast.Assign) and len(n.targets) == 1 @@ -3271,6 +3315,9 @@ def _build_scope_alias_index(tree, const_env): counts[_tn] = counts.get(_tn, 0) + 1 allnames.add(_tn) assigns.append((_tn, _gen.iter.elts[0])) + # A `global`/`nonlocal`-declared name is NOT a local binding, so it must not shadow an + # outer alias here (its assignment rebinds the target scope instead). + allnames -= rebound idx.assigned[scope] = allnames smap: dict[str, str] = {} emap: dict[str, str] = {} @@ -3354,6 +3401,34 @@ def _build_scope_alias_index(tree, const_env): dmap[_pn] = _ddfq if _rhs_import_func(_de) and _pn not in imap: imap[_pn] = True + # A `global name = ` (or `nonlocal name = `) rebinds the name in the + # TARGET scope (module for global, nearest enclosing scope for nonlocal), NOT locally, + # so `global s; s = os.system; s('rm -rf /')` must record the alias in that target + # scope -- otherwise the local pass skips it (name in rebound) and the call resolves to + # nothing. Target scopes are processed before nested scopes, so setdefault preserves + # any alias they already hold. + if global_names or nonlocal_names: + for name, rhs in assigns: + if name in global_names: + _target = tree + elif name in nonlocal_names: + _target = idx.enclosing.get(scope) + else: + continue + if _target is None: + continue + _rhs_eff = _unwrap_container_index(rhs) + _gfq = _resolve_static_shell_sink( + _rhs_eff, os_aliases, subprocess_aliases, from_aliases + ) + if _gfq: + idx.shell.setdefault(_target, {}).setdefault(name, _gfq) + _geb = _rhs_exec_builtin(_rhs_eff) + if _geb is not None: + idx.execb.setdefault(_target, {}).setdefault(name, _geb) + _gdfq = _rhs_deserializer(_rhs_eff) + if _gdfq is not None: + idx.deser.setdefault(_target, {}).setdefault(name, _gdfq) if smap: idx.shell[scope] = smap if emap: @@ -4321,6 +4396,22 @@ def _check_signal_escape_patterns( sink (os.system / subprocess.*), a dynamic-import function, or a code deserializer. Returns a short description or None. The payloads such a sink runs never reach the recursive analyzer, so passing one by reference is unsafe.""" + if isinstance(n, ast.Subscript): + # An inline literal-container index hides the sink from the name/attribute + # checks: map([eval][0], [...]) / partial({'e': exec}['e'], ...). Resolve the + # element node and describe it, the same unwrap direct calls already apply. + container = n.value + ci = _const_fold(n.slice, _const_env) + elt = None + if isinstance(container, (ast.List, ast.Tuple)) and isinstance(ci, int): + if -len(container.elts) <= ci < len(container.elts): + elt = container.elts[ci] + elif 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: + elt = _v + break + return self._sink_ref_desc(elt) if elt is not None else None if isinstance(n, ast.Name): if n.id in _DYNAMIC_EXEC_BUILTINS: return f"{n.id} (dynamic exec)" @@ -4366,6 +4457,25 @@ def _check_signal_escape_patterns( return None return None + def _is_sys_modules(self, n): + # `sys.modules` as the attribute form, or a single-assignment alias of it + # (m = sys.modules; m.pop('_io')). Used by the loader-table mutation checks. + if ( + isinstance(n, ast.Attribute) + and n.attr == "modules" + and _ast_name_matches(n.value, self.sys_aliases) + ): + return True + if _analyzer_on and isinstance(n, ast.Name): + rhs = _scope_idx.resolve(n.id, n, "rhsnode") + if ( + isinstance(rhs, ast.Attribute) + and rhs.attr == "modules" + and _ast_name_matches(rhs.value, self.sys_aliases) + ): + return True + return False + def _is_compile_result(self, arg): """True when ``arg`` is a ``compile(...)`` code object (bare / builtins / single-assignment alias / inline-container unwrap). Used to catch code objects @@ -4911,9 +5021,7 @@ def _check_signal_escape_patterns( "__setitem__", "__delitem__", ) - and isinstance(func.value, ast.Attribute) - and func.value.attr == "modules" - and _ast_name_matches(func.value.value, self.sys_aliases) + and self._is_sys_modules(func.value) ): dynamic_desc = ( f"sys.modules.{func.attr}(...) mutates the loader table " @@ -4936,9 +5044,7 @@ def _check_signal_escape_patterns( "__delitem__", ) and node.args - and isinstance(node.args[0], ast.Attribute) - and node.args[0].attr == "modules" - and _ast_name_matches(node.args[0].value, self.sys_aliases) + and self._is_sys_modules(node.args[0]) and ( (isinstance(func.value, ast.Name) and func.value.id == "dict") or ( @@ -5170,6 +5276,25 @@ def _check_signal_escape_patterns( "description": f"subscripted {_mro_shape} extracts a base class (gadget)", } ) + # type(open).__dict__['__closure__'].__get__(open) / type(cell).__dict__[ + # 'cell_contents'].__get__(cell): fetch a gadget descriptor from a type's __dict__ + # BY NAME (a subscript, not an attribute node), then invoke __get__ to recover the + # guarded wrapper's original callable. Flag a __dict__ subscript keyed by a gadget + # dunder so the attribute-node gadget scan cannot be side-stepped this way. + if ( + isinstance(node.ctx, ast.Load) + and isinstance(node.value, ast.Attribute) + and node.value.attr == "__dict__" + ): + _dk = _const_fold(node.slice, _const_env) + if isinstance(_dk, str) and _dk in _GADGET_DUNDERS: + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": f"__dict__[{_dk!r}] descriptor lookup (gadget)", + } + ) # sys.modules['os'] pulls an already-loaded dangerous module out of the # loader table (os/subprocess are loaded by the host). Scope to a Load of a # dangerous LITERAL key so legit uses ("x" in sys.modules, sys.modules.get( @@ -6388,20 +6513,15 @@ def _check_signal_escape_patterns( if not _is_str: # subprocess.run/call/Popen/check_output/check_call(cmd, shell=True): a string # command with shell=True runs through /bin/sh (these are in _SHELL_EXEC_FUNCS - # but not in _STRING_SHELL_SINKS, so check the shell= kwarg explicitly). - if isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name): - if f.value.id in _subprocess_mod_aliases and f.attr in ( - "run", - "call", - "check_call", - "check_output", - "Popen", - ): - for kw in node.keywords or []: - if kw.arg == "shell" and not ( - isinstance(kw.value, ast.Constant) and kw.value.value is False - ): - _is_str = True + # but not in _STRING_SHELL_SINKS, so check the shell= kwarg explicitly). Uses the + # subprocess-exec callee resolver so the attribute, from-import (from subprocess + # import run as r) and single-assignment (r = subprocess.run) forms are all seen. + if _is_subprocess_exec_callee(f): + for kw in node.keywords or []: + if kw.arg == "shell" and not ( + isinstance(kw.value, ast.Constant) and kw.value.value is False + ): + _is_str = True if not _is_str or not node.args: return False return _scan_one_command(_fold_read_arg(node.args[0])) @@ -6822,7 +6942,19 @@ def _wrap1(mod, name, what): if orig is None: return @_gwraps(orig) - def w(path, *a, **k): + def w(*a, **k): + # The path may be positional OR a public keyword: os.mkdir(path=...), + # os.makedirs(name=...), os.removedirs(name=...). Extract it from whichever slot it + # arrived in so a keyword call is not broken (missing positional) while still confined. + _pk = None + if a: + path = a[0] + elif "path" in k: + path, _pk = k["path"], "path" + elif "name" in k: + path, _pk = k["name"], "name" + else: + return orig(*a, **k) # let the original raise its own TypeError if any(k.get(_f) is not None for _f in ("dir_fd", "src_dir_fd", "dst_dir_fd")): _deny(path, what + " (dir_fd)") # fd-relative target: a realpath check is meaningless if isinstance(path, int): @@ -6833,7 +6965,13 @@ def _wrap1(mod, name, what): p = _fspath1(path) if not _within(p): _deny(p, what) - return orig(p, *a, **k) + # Pass the MATERIALIZED path back in the same slot it arrived (a stateful __fspath__ + # cannot then return a different outside path to the real call). + if a: + return orig(p, *a[1:], **k) + k = dict(k) + k[_pk] = p + return orig(*a, **k) setattr(mod, name, w) # Path-first single-arg mutators. mkfifo/utime/setxattr/removexattr create or mutate diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 47684c25ce..49d3f56866 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -1012,3 +1012,41 @@ def test_sandboxed_workdir_module_shadowing_neutralized(tmp_path): shutil.rmtree(_pyc, ignore_errors = True) if os.path.exists(marker): os.remove(marker) + + +@_POSIX_ONLY +def test_sandboxed_keyword_path_mutator_allowed_and_confined(): + # The mutator guard must accept the path via its public keyword (os.makedirs(name=...), + # os.mkdir(path=...)) instead of raising TypeError on a missing positional argument, while + # still confining the write. Clean any leftover dir first so the test is idempotent. + session = "backstop-kwpath" + workdir = get_sandbox_workdir(session) + import shutil as _sh + + _sh.rmtree(os.path.join(workdir, "kwdir"), ignore_errors = True) + out = _python_exec( + "import os\nos.makedirs(name='kwdir/sub', exist_ok=True)\n" + "os.mkdir(path='kwdir/one')\nprint('KW-OK', os.path.isdir('kwdir/sub'))", + None, + 30, + session, + disable_sandbox = False, + ) + assert "KW-OK True" in out + assert "sandbox:" not in out + _sh.rmtree(os.path.join(workdir, "kwdir"), ignore_errors = True) + + +@_POSIX_ONLY +def test_sandboxed_keyword_path_mutator_escape_denied(tmp_path): + # A keyword path outside the workdir is still confined. + target = tmp_path / "kw_escape" + out = _python_exec( + f"import os\nos.makedirs(name={str(target)!r}, exist_ok=True)\nprint('MADE')", + None, + 30, + "backstop-kwpath-escape", + disable_sandbox = False, + ) + assert "sandbox:" in out or "PermissionError" in out + assert not target.exists() diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index fe85b92c2e..16e1c90456 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -2433,3 +2433,87 @@ class TestRound18Bypasses: def test_network_alias_trusted_allowed(self): _ok("import requests as r\nr.get('https://huggingface.co/x')") _ok("import requests\nrequests.get(url='https://huggingface.co/x')") + + +class TestRound19Bypasses: + """Nineteenth-round Codex findings: global/nonlocal sink aliases, variable-expanded and + wrapper-hidden command words, wrapper-hidden mutating utilities and shell scripts, + shell=True subprocess aliases, sys.modules aliases, descriptor-lookup gadgets, and + container-hidden sinks in higher-order calls.""" + + def test_global_alias_to_sink_blocked(self): + code = "def f():\n global s\n s = os.system\n s('touch /tmp/x')\nimport os\nf()" + assert _check_code_safety(code) is not None, code + + def test_benign_global_allowed(self): + _ok("def f():\n global s\n s = 5\n return s\nprint(f())") + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('p=python3; $p -c \"print(1)\"')", + "import os\nos.system('${CMD} -rf /')", + "import os\nos.system('cat f && $tool')", + ], + ) + def test_variable_expanded_command_word_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('env sed -i s/a/b/ /tmp/victim')", + "import os\nos.system('nice sed -i s/a/b/ /tmp/victim')", + "import os\nos.system('timeout 5 sort -o /tmp/f /tmp/f')", + ], + ) + def test_wrapper_hidden_mutating_util_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('env bash s.sh')", + "import os\nos.system('timeout 5 bash s.sh')", + "import os\nos.system('nice sh script.sh')", + ], + ) + def test_wrapper_hidden_shell_script_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "from subprocess import run as r\nr('head -1 /etc/passwd', shell=True)", + "import subprocess\nr = subprocess.run\nr('cat /etc/shadow', shell=True)", + ], + ) + def test_shell_true_subprocess_alias_read_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_sys_modules_alias_mutation_blocked(self): + code = "import sys\nm = sys.modules\nm.pop('_io', None)\nimport _io" + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "f = open\ntype(f).__dict__['__closure__'].__get__(f)", + "c = (lambda: x).__closure__[0]\ntype(c).__dict__['cell_contents'].__get__(c)", + ], + ) + def test_descriptor_lookup_gadget_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "list(map([eval][0], [\"__import__('os').system('touch /tmp/x')\"]))", + "list(map({'e': exec}['e'], [\"import os\\nos.system('id')\"]))", + ], + ) + def test_container_hidden_higher_order_sink_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_benign_higher_order_allowed(self): + _ok("print(list(map(str, [1, 2, 3])))")