From 0441be5e1131a3d107aaeabd1c2beb31959353cc Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Thu, 9 Jul 2026 16:22:40 +0000 Subject: [PATCH] Studio sandbox: close fourth-round review bypasses (scope-aware aliases + guards) Scope-aware alias resolution (replaces the flat, module-wide alias maps): - A new per-scope index resolves shell-sink, exec-builtin and compiled-code aliases with Python lexical scoping. This fixes two problems the flat maps had: a safe `c = compile('1+1')` in one function no longer shadows a dynamic `exec(c)` in another (a real bypass), and a `s = os.system` in one function no longer makes a benign `s = print` call in another look like a shell sink (a false positive), while still catching a genuine function-local sink and honoring local shadowing of a module-level alias. Runtime realpath backstop: - io.FileIO now passes the MATERIALIZED fspath to the real constructor (a stateful __fspath__ could otherwise return an outside path to the C constructor). - Deny an integer fd path for the mutating single-path wrappers (os.chmod(fd) etc.): a read-only fd opened on an outside file could otherwise mutate host metadata. Constant-folder allocation DoS: - Refuse dynamic printf widths/precisions ('%*s', '%.*f') that draw their size from a runtime argument. - Bound str.replace / str.join output before it allocates (a long replacement over many occurrences, or joining many long parts, can build a multi-gigabyte string). Static classifier: - Flag builtins / a sensitive module reached through the namespace dict: globals()['__builtins__'], locals()[...] and globals()['os']. Adds regression tests across the aliasing, runtime-backstop, const-fold and classifier suites for every item above. --- studio/backend/core/inference/tools.py | 284 +++++++++++++----- studio/backend/tests/test_sandbox_aliasing.py | 24 ++ .../backend/tests/test_sandbox_const_fold.py | 16 + .../tests/test_sandbox_runtime_backstop.py | 40 +++ studio/backend/tests/test_sandbox_tools.py | 7 + 5 files changed, 288 insertions(+), 83 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 0259efd7a5..f21a042f4d 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1724,7 +1724,7 @@ _PRINTF_WIDTH_RE = re.compile(r"%[-+ #0]*(\d+)?(?:\.(\d+))?[hlL]?[diouxXeEfFgGcr def _printf_ok(fmt): - """Percent-format string with no oversized field width / precision.""" + """Percent-format string with no oversized (or dynamic '*') width / precision.""" if isinstance(fmt, (bytes, bytearray)): try: fmt = fmt.decode("latin-1") @@ -1732,12 +1732,47 @@ def _printf_ok(fmt): return True if not isinstance(fmt, str): return True + # A '*' width or precision ('%*s', '%.*f') pulls its size from a runtime argument, + # so it cannot be bounded statically -- refuse rather than risk a large allocation. + for m in re.finditer(r"%[-+ #0]*(\*)?(?:\.(\*)?\d*)?", fmt): + if m.group(1) == "*" or m.group(2) == "*": + return False for m in _PRINTF_WIDTH_RE.finditer(fmt): if any(g and _too_wide(int(g)) for g in m.groups()): return False return True +def _replace_output_ok(recv, call_args): + """Bound str.replace/bytes.replace output before it allocates: replacing many + occurrences with a long replacement can build a multi-gigabyte string.""" + if len(call_args) < 2: + return True + old, new = call_args[0], call_args[1] + if not isinstance(new, (str, bytes, bytearray)): + return True + lo = len(old) if isinstance(old, (str, bytes, bytearray)) else 1 + n_repl = (len(recv) + 1) if lo == 0 else (len(recv) // max(lo, 1) + 1) + if len(call_args) >= 3 and isinstance(call_args[2], int) and call_args[2] >= 0: + n_repl = min(n_repl, call_args[2]) + return len(recv) + n_repl * len(new) <= _FOLD_MAXLEN + + +def _join_output_ok(sep, call_args): + """Bound str.join/bytes.join output before it allocates.""" + if not call_args or not isinstance(call_args[0], (list, tuple)): + return True + items = call_args[0] + total = len(sep) * max(len(items) - 1, 0) + for x in items: + if not isinstance(x, (str, bytes, bytearray)): + return True # a real join would TypeError; not an allocation concern + total += len(x) + if total > _FOLD_MAXLEN: + return False + return True + + def _fold_apply_codec(name, data): """Pure data transforms only (rot13/hex/base64/zlib/text codecs). Bounded zlib.""" name = name.lower().replace("-", "_") @@ -2092,6 +2127,10 @@ def _fold_call(node, _state, _depth): if attr in ("center", "ljust", "rjust", "zfill") and call_args: if _too_wide(call_args[0]): return None + if attr == "replace" and not _replace_output_ok(recv, call_args): + return None + if attr == "join" and not _join_output_ok(recv, call_args): + return None if attr == "format": if not _format_template_ok(recv): return None @@ -2458,23 +2497,96 @@ def _walk_scope_local(scope): stack.append(child) -def _iter_scope_single_assignments(tree): - """Yield (name, rhs) for each Name assigned exactly once within its OWN function - (or module) scope and not declared global / nonlocal there. Counting per scope -- - not tree-wide -- means a name reused independently in two functions is still a - single-assignment alias in each (a tree-wide count would wrongly treat both as - ambiguous and miss a real sink alias).""" - scopes = [tree] +class _ScopeAliasIndex: + """Per-scope single-assignment aliases (shell sink / exec builtin / compiled + source) resolved with Python lexical scoping. Counting and resolution are per + function scope, so two functions binding the same local name neither cancel out + (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") + + def __init__(self, tree): + self.tree = tree + self.node_scope: dict = {tree: tree} + self.enclosing: dict = {tree: None} + self.shell: dict = {} + self.execb: dict = {} + self.compiled: dict = {} + self.assigned: dict = {} + + def _chain(self, node): + s = self.node_scope.get(node, self.tree) + while s is not None: + yield s + s = self.enclosing.get(s) + + def resolve(self, name, node, kind): + maps = getattr(self, kind) + for s in self._chain(node): + m = maps.get(s) + if m and name in m: + return m[name] + if name in self.assigned.get(s, ()): # locally shadowed by a non-alias + return None + return None + + def effective(self, node, kind): + maps = getattr(self, kind) + result: dict = {} + shadowed: set = set() + for s in self._chain(node): + for k, v in maps.get(s, {}).items(): + if k not in shadowed and k not in result: + result[k] = v + shadowed |= self.assigned.get(s, set()) + return result + + +def _build_scope_alias_index(tree, const_env): + idx = _ScopeAliasIndex(tree) + + def _rec(node, scope): + for child in ast.iter_child_nodes(node): + idx.node_scope[child] = scope + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): + idx.enclosing[child] = scope + _rec(child, child) + else: + _rec(child, scope) + + _rec(tree, tree) + + # os / subprocess import + from-import aliases are collected tree-wide (imports + # are lexically visible module-wide in practice) and shared across scopes. + os_aliases = {"os"} + subprocess_aliases = {"subprocess"} + from_aliases: dict[str, str] = {} for n in ast.walk(tree): - if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)): - scopes.append(n) + if isinstance(n, ast.Import): + for a in n.names: + if a.name == "os": + os_aliases.add(a.asname or "os") + elif a.name == "subprocess": + subprocess_aliases.add(a.asname or "subprocess") + 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 + + scopes = [tree] + [ + n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) + ] for scope in scopes: counts: dict[str, int] = {} rebound: set[str] = set() assigns: list[tuple[str, ast.expr]] = [] + allnames: set[str] = set() 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 + allnames.add(n.id) elif isinstance(n, (ast.Global, ast.Nonlocal)): rebound.update(n.names) elif ( @@ -2483,36 +2595,38 @@ def _iter_scope_single_assignments(tree): and isinstance(n.targets[0], ast.Name) ): assigns.append((n.targets[0].id, n.value)) + idx.assigned[scope] = allnames + smap: dict[str, str] = {} + emap: dict[str, str] = {} + cmap: dict[str, tuple] = {} for name, rhs in assigns: - if counts.get(name) == 1 and name not in rebound: - yield name, rhs - - -def _build_exec_env(tree, const_env): - """Map single-assignment names to exec builtins (`e = exec`) and to a compiled - source (`c = compile("...")`) so a later call through the alias is unwrapped.""" - exec_aliases: dict[str, str] = {} - compiled_env: dict[str, tuple] = {} - # Per-scope single assignments: an alias assigned inside a function (def f(): - # e = exec; e("...")) is unwrapped, and two functions sharing a local name do not - # cancel each other out (that would be a false negative). - for name, rhs in _iter_scope_single_assignments(tree): - if isinstance(rhs, ast.Name) and rhs.id in _EXEC_BUILTINS: - exec_aliases[name] = rhs.id - elif ( - isinstance(rhs, ast.Call) - and isinstance(rhs.func, ast.Name) - and rhs.func.id == "compile" - and rhs.args - ): - v = _const_fold(rhs.args[0], const_env) - if isinstance(v, (str, bytes, bytearray)): - compiled_env[name] = ( - _recovered_source(v), - _compile_mode(rhs, const_env), - isinstance(v, (bytes, bytearray)), - ) - return exec_aliases, compiled_env + 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 + elif ( + isinstance(rhs, ast.Call) + and isinstance(rhs.func, ast.Name) + and rhs.func.id == "compile" + and rhs.args + ): + v = _const_fold(rhs.args[0], const_env) + if isinstance(v, (str, bytes, bytearray)): + cmap[name] = ( + _recovered_source(v), + _compile_mode(rhs, const_env), + isinstance(v, (bytes, bytearray)), + ) + if smap: + idx.shell[scope] = smap + if emap: + idx.execb[scope] = emap + if cmap: + idx.compiled[scope] = cmap + return idx def _payload_has_obfuscation_primitive(node): @@ -2618,7 +2732,7 @@ def _first_unsafe_reason(info): return "unsafe operation" -def _recover_exec_payload(node, func_id, const_env, exec_aliases, compiled_env): +def _recover_exec_payload(node, func_id, const_env, compiled_env): """Recover a statically foldable source string for eval/exec/compile. Returns ("RECOVERED", src, mode, is_bytes) / ("DYNAMIC", None, None, False) / @@ -2779,35 +2893,6 @@ def _resolve_static_shell_sink(node, os_aliases, subprocess_aliases, from_aliase return None -def _build_shell_sink_aliases(tree): - """Single-assignment names (stored exactly once) bound to a resolved shell sink.""" - os_aliases = {"os"} - subprocess_aliases = {"subprocess"} - from_aliases: dict[str, str] = {} - for n in ast.walk(tree): - if isinstance(n, ast.Import): - for a in n.names: - if a.name == "os": - os_aliases.add(a.asname or "os") - elif a.name == "subprocess": - subprocess_aliases.add(a.asname or "subprocess") - 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 - - aliases: dict[str, str] = {} - # Per-scope single assignments: a function-local `s = os.system` is aliased, and - # two functions each binding their own local `s` do not cancel out (a tree-wide - # store count would treat both as ambiguous and miss a real sink alias). - for name, rhs in _iter_scope_single_assignments(tree): - fq = _resolve_static_shell_sink(rhs, os_aliases, subprocess_aliases, from_aliases) - if fq: - aliases[name] = fq - return aliases - - def _check_signal_escape_patterns( code: str, _depth: int = 0, @@ -2842,22 +2927,24 @@ def _check_signal_escape_patterns( if _analyzer_on: try: _const_env = _build_const_prop_env(tree) - _exec_aliases, _compiled_env = _build_exec_env(tree, _const_env) - _sink_aliases = _build_shell_sink_aliases(tree) + _scope_idx = _build_scope_alias_index(tree, _const_env) except Exception: # pragma: no cover - defensive: never crashier than legacy logger.warning("sandbox analyzer context build failed; legacy fallback", exc_info = True) _analyzer_on = False - _const_env, _exec_aliases, _compiled_env = {}, {}, {} - _sink_aliases = {} + _const_env = {} + _scope_idx = _ScopeAliasIndex(tree) else: - _const_env, _exec_aliases, _compiled_env = {}, {}, {} - _sink_aliases = {} + _const_env = {} + _scope_idx = _ScopeAliasIndex(tree) def _analyze_exec_call(node, func_id): """Stage 2 driver: recover + recurse a foldable payload, else dynamic policy.""" try: + # Resolve compiled-code aliases (c = compile(...)) in the CALL's scope so a + # safe alias in one function cannot shadow a dynamic exec(c) in another. + _compiled_here = _scope_idx.effective(node, "compiled") kind, src, mode, is_bytes = _recover_exec_payload( - node, func_id, _const_env, _exec_aliases, _compiled_env + node, func_id, _const_env, _compiled_here ) if kind == "NO_PAYLOAD": return @@ -3169,7 +3256,7 @@ def _check_signal_escape_patterns( if fq: return fq if isinstance(elt, ast.Name): - return _sink_aliases.get(elt.id) + return _scope_idx.resolve(elt.id, elt, "shell") return None container = sub.value @@ -3244,9 +3331,10 @@ def _check_signal_escape_patterns( elif isinstance(func, ast.Name): # from-import aliases: from os import system; system(...) shell_func = self.shell_exec_aliases.get(func.id) - # Stage 4: single-assignment alias `s = os.system; s('rm -rf /')`. + # Stage 4: single-assignment alias `s = os.system; s('rm -rf /')`, + # resolved in the call's own scope (per-function). if shell_func is None and _analyzer_on: - shell_func = _sink_aliases.get(func.id) + shell_func = _scope_idx.resolve(func.id, func, "shell") elif _analyzer_on and isinstance(func, ast.Subscript): # Stage 4: inline literal container index `[os.system][0](...)`. shell_func = self._resolve_container_sink(func) @@ -3344,8 +3432,9 @@ def _check_signal_escape_patterns( exec_func_id = func.id elif func.id in self.exec_from_aliases: exec_func_id = self.exec_from_aliases[func.id] # from builtins import exec as e - elif _analyzer_on and func.id in _exec_aliases: - exec_func_id = _exec_aliases[func.id] + elif _analyzer_on: + # single-assignment `e = exec` alias, resolved in the call's scope. + exec_func_id = _scope_idx.resolve(func.id, func, "execb") elif ( isinstance(func, ast.Attribute) and func.attr in _DYNAMIC_EXEC_BUILTINS @@ -3553,6 +3642,28 @@ def _check_signal_escape_patterns( "description": "sys.modules[...] access to a sensitive 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 + # dangerous literal key off a bare globals()/locals()/vars() call. + if isinstance(node.ctx, ast.Load) and ( + isinstance(v, ast.Call) + and isinstance(v.func, ast.Name) + and v.func.id in ("globals", "locals", "vars") + and not v.args + ): + key = _extract_string_from_node(node.slice) + if key is not None and ( + key in ("__builtins__", "__builtin__") + or key.split(".")[0] in _DANGEROUS_IMPORT_NAMES + ): + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": "namespace-dict access to builtins / a sensitive module", + } + ) self.generic_visit(node) def visit_ExceptHandler(self, node): @@ -4487,10 +4598,15 @@ def _wrap1(mod, name, what): def w(path, *a, **k): 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): + # A mutating op given an fd (os.chmod(fd), os.truncate(fd), ...) can hit a + # file opened read-only outside the workdir; a string realpath cannot + # confine an fd, so deny it (fchmod/fchown are already denied separately). + _deny(path, what + " (fd)") p = _fspath1(path) if not _within(p): _deny(p, what) - return orig(p if not isinstance(path, int) else path, *a, **k) + return orig(p, *a, **k) setattr(mod, name, w) # Path-first single-arg mutators. mkfifo/utime/setxattr/removexattr create or mutate @@ -4567,7 +4683,9 @@ def _guard_fileio(_realcls): f = _fspath1(name) if _mode_is_write(mode) and not _within(f): _deny(f, "FileIO write") - super().__init__(name, mode, *a, **k) + # Pass the MATERIALIZED path so a stateful __fspath__ cannot return a + # different (outside) path to the real constructor than we checked. + super().__init__(f, mode, *a, **k) return _GuardedFileIO for _iomod in (_io, _lowio): diff --git a/studio/backend/tests/test_sandbox_aliasing.py b/studio/backend/tests/test_sandbox_aliasing.py index d0f116b093..ba69adb4bd 100644 --- a/studio/backend/tests/test_sandbox_aliasing.py +++ b/studio/backend/tests/test_sandbox_aliasing.py @@ -92,6 +92,30 @@ class TestPerScopeAliasCounting: "def b():\n s = max\n return s([1, 2])\na()" ) + def test_sink_alias_does_not_leak_into_other_scope(self): + # A `s = os.system` in one function must NOT make a benign `s = print` call in + # another function look like a shell sink (would be a false positive). + _ok( + "import os\n" + "def a():\n s = os.system\n s('echo hi')\n" + "def b():\n s = print\n s('remove the rm temp files')\n" + "b()" + ) + + def test_safe_compiled_alias_does_not_shadow_dynamic_exec(self): + # A safe `c = compile('1+1')` in one function must NOT let a dynamic + # `c = compile(src); exec(c)` in another function be treated as safe. + _blocked( + "def a():\n c = compile('1 + 1', '', 'eval')\n eval(c)\n" + "def b(src):\n c = compile(src, '', 'exec')\n exec(c)\n" + "b('x')" + ) + + def test_function_local_shadow_of_module_alias_allowed(self): + # A module-level `s = os.system` shadowed by a local `s = print` resolves to + # the local binding inside that function. + _ok("import os\ns = os.system\ndef f():\n s = print\n s('please rm the files')\nf()") + class TestAliasingLowFalsePositive: def test_reassigned_alias_not_treated_as_sink(self): diff --git a/studio/backend/tests/test_sandbox_const_fold.py b/studio/backend/tests/test_sandbox_const_fold.py index c383cc28ca..7ba9e68b60 100644 --- a/studio/backend/tests/test_sandbox_const_fold.py +++ b/studio/backend/tests/test_sandbox_const_fold.py @@ -94,6 +94,22 @@ class TestConstFoldAllocationDoS: def test_nested_format_small_width_folds(self): assert _fold("'{:>{}}'.format('x', 5)") == " x" + def test_dynamic_percent_width_refused(self): + # '%*s' / '%.*f' take the width/precision from a runtime arg; cannot be bounded. + assert _fold("'%*s' % (1000000000, 'x')") is None + assert _fold("'%.*f' % (1000000000, 1.0)") is None + + def test_replace_expansion_refused(self): + assert _fold("('x' * 65536).replace('x', 'y' * 65536)") is None + + def test_join_expansion_refused(self): + assert _fold("','.join(['y' * 65536] * 4096)") is None + + def test_benign_replace_and_join_fold(self): + assert _fold("'aaa'.replace('a', 'b')") == "bbb" + assert _fold("','.join(['a', 'b', 'c'])") == "a,b,c" + assert _fold("'%05d' % 7") == "00007" + def test_pad_method_width_refused(self): assert _fold("'x'.ljust(1000000000)") is None assert _fold("'x'.rjust(10 ** 9)") is None diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 0c745761a7..4c681c4f89 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -488,6 +488,46 @@ def test_sandboxed_int_subclass_flags_denied(tmp_path): assert not target.exists() +@_POSIX_ONLY +def test_sandboxed_fileio_stateful_fspath_denied(tmp_path): + # io.FileIO must pass the MATERIALIZED path to the real constructor so a stateful + # __fspath__ cannot return a different (outside) path than was checked. + target = tmp_path / "fileio_fspath_escape.txt" + out = _python_exec( + "import io\n" + "class P:\n" + " n = 0\n" + " def __fspath__(self):\n" + " P.n += 1\n" + f" return 'ok.txt' if P.n == 1 else {str(target)!r}\n" + "io.FileIO(P(), 'w').write(b'x')\nprint('DONE')", + None, + 30, + "backstop-fileio-fspath", + disable_sandbox = False, + ) + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_os_chmod_fd_denied(tmp_path): + # A read-only fd opened on an outside file must not be reusable via os.chmod(fd). + victim = tmp_path / "chmod_fd_victim.txt" + victim.write_text("x") + os.chmod(victim, 0o600) + out = _python_exec( + "import os\n" + f"fd = os.open({str(victim)!r}, os.O_RDONLY)\n" + "os.chmod(fd, 0o644); print('CHMODDED')", + None, + 30, + "backstop-chmod-fd", + disable_sandbox = False, + ) + assert "sandbox:" in out and "(fd)" in out + assert oct(os.stat(victim).st_mode & 0o777) == "0o600" + + @_POSIX_ONLY def test_sandboxed_in_workdir_ops_still_work(): # The added guards must not break benign in-workdir writes. diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index 5ec7700541..384051af97 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -896,6 +896,10 @@ class TestReceiverAndVarsAndDynImportBypasses: # 605: __import__ reached through the builtins module. "import builtins\nbuiltins.__import__('os').system('rm -rf /')", "__builtins__.__import__('subprocess').run(['id'])", + # 158: builtins / sensitive module reached through the namespace dict. + "getattr(globals()['__builtins__'], '__import__')('os').system('rm -rf /')", + "getattr(locals()['__builtins__'], 'eval')('x')", + "globals()['os'].system('rm -rf /')", ], ) def test_blocked(self, code): @@ -915,6 +919,9 @@ class TestReceiverAndVarsAndDynImportBypasses: "import os\nopen(os.path.join('sub', 'a.txt'))", # 605: benign builtins attribute access stays allowed. "import builtins\nx = builtins.len([1, 2, 3])", + # 158: a benign globals() lookup of a normal variable stays allowed. + "g = globals()\nx = g['some_var']", + "globals()['my_config']", ], ) def test_benign_allowed(self, code):