From 0f4b4b3d36d30932fbb327a561d64216c21c9ca4 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Thu, 9 Jul 2026 17:32:27 +0000 Subject: [PATCH] Studio sandbox: close sixth-round review bypasses (obfuscation, reads, child procs, guard pinning) Static classifier: - resolve pathlib expressions passed to open()/read callees so open(Path('/etc') / 'passwd') blocks like open('/etc/passwd') - flag getattr()/setattr() of an introspection gadget dunder (__globals__, __subclasses__, ...) regardless of receiver - add the .get() twin of the globals()/locals()/vars() namespace-dict subscript guard - constant-fold sys.modules[...] and sys.modules.get(...) keys so a concatenated key is caught - track 'from builtins import __import__ as imp' as a dynamic import alias - treat deserializer module aliases (pickle, dill, ...) as sensitive targets for getattr/vars/__dict__ - flag code objects executed through types.FunctionType(compile(src, ...), ...), including the c = compile(src); FunctionType(c) two-step - block language interpreters (python/perl/ruby/node/...) at shell command position: a spawned child runs without the in-process write guard Runtime backstop: - pin os.fspath/os.path.realpath to captured originals inside _within so a sandboxed reassignment of os.fspath cannot make realpath resolve an outside write target to an in-workdir path --- studio/backend/core/inference/tools.py | 218 +++++++++++++++--- .../tests/test_sandbox_runtime_backstop.py | 51 +++- studio/backend/tests/test_sandbox_tools.py | 136 +++++++++++ 3 files changed, 369 insertions(+), 36 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index bdb844e066..47af1e26d8 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -113,6 +113,32 @@ _BLOCKED_COMMANDS_COMMON = frozenset( "ln", } ) +# Language interpreters that run inline / file / stdin code in a FRESH child process. +# The runtime filesystem backstop only patches the current interpreter, so a spawned +# `python -c '...'`, `perl -e '...'`, `node -e '...'`, etc. runs with none of the guard +# monkeypatches and can write/delete outside the session workdir. Blocking the +# interpreter at shell command position closes that child-process escape; the sandbox's +# own python_execute tool is the supported way to run Python (it IS guarded). Argument +# position (`echo python`, `ls /usr/bin/python3`) is unaffected by the command-position +# scanner. +_INTERPRETER_COMMANDS = frozenset( + { + "python", + "python2", + "python3", + "pythonw", + "perl", + "ruby", + "node", + "nodejs", + "php", + "deno", + "lua", + "luajit", + "rscript", + } +) +_BLOCKED_COMMANDS_COMMON = _BLOCKED_COMMANDS_COMMON | _INTERPRETER_COMMANDS _BLOCKED_COMMANDS_WIN = frozenset( { "rmdir", @@ -2511,6 +2537,7 @@ class _ScopeAliasIndex: "shell", "execb", "compiled", + "compiledany", "impf", "deser", "strconst", @@ -2524,6 +2551,9 @@ class _ScopeAliasIndex: self.shell: dict = {} self.execb: dict = {} self.compiled: dict = {} + # name -> True: bound to a compile() result (foldable OR dynamic). Used to catch + # a code object executed through types.FunctionType(c, ...) after `c = compile(...)`. + self.compiledany: 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) @@ -2667,6 +2697,7 @@ def _build_scope_alias_index(tree, const_env): smap: dict[str, str] = {} emap: dict[str, str] = {} cmap: dict[str, tuple] = {} + camap: dict[str, bool] = {} imap: dict[str, bool] = {} dmap: dict[str, str] = {} scmap: dict[str, object] = {} @@ -2685,6 +2716,9 @@ def _build_scope_alias_index(tree, const_env): and rhs.func.id == "compile" and rhs.args ): + # Any `c = compile(...)` binds a code object, tracked for the + # types.FunctionType(c) execution gadget below (dynamic or foldable). + camap[name] = True v = _const_fold(rhs.args[0], const_env) if isinstance(v, (str, bytes, bytearray)): cmap[name] = ( @@ -2708,6 +2742,8 @@ def _build_scope_alias_index(tree, const_env): idx.execb[scope] = emap if cmap: idx.compiled[scope] = cmap + if camap: + idx.compiledany[scope] = camap if imap: idx.impf[scope] = imap if dmap: @@ -3259,6 +3295,11 @@ def _check_signal_escape_patterns( # import pickle as p -> {"p": "pickle"}; from pickle import loads as l -> {"l": "pickle.loads"} self.deserialize_module_aliases: dict[str, str] = {} self.deserialize_aliases: dict[str, str] = {} + # import types as t -> {"types", "t"}; from types import FunctionType as F -> {"F"}. + # FunctionType(code, globals)() runs a code object WITHOUT eval/exec, so a + # dynamic compile() result reaches execution through it (see visit_Call). + self.types_aliases = {"types"} + self.functiontype_aliases: set[str] = set() self.loop_depth = 0 def visit_Import(self, node): @@ -3277,6 +3318,8 @@ def _check_signal_escape_patterns( self.sys_aliases.add(alias.asname or "sys") elif alias.name == "builtins": self.builtins_aliases.add(alias.asname or "builtins") + elif alias.name == "types": + self.types_aliases.add(alias.asname or "types") if alias.name in _DESERIALIZE_MODULES: self.deserialize_module_aliases[alias.asname or alias.name] = alias.name self.generic_visit(node) @@ -3314,11 +3357,19 @@ def _check_signal_escape_patterns( for alias in node.names: if alias.name in _DYNAMIC_EXEC_BUILTINS: self.exec_from_aliases[alias.asname or alias.name] = alias.name + elif alias.name == "__import__": + # `from builtins import __import__ as imp; imp('os').system(...)` + # is a dynamic import exactly like a bare __import__ call. + self.import_func_aliases.add(alias.asname or alias.name) elif node.module in _DESERIALIZE_MODULES: for alias in node.names: fq = f"{node.module}.{alias.name}" if fq in _CODE_DESERIALIZE_SINKS: self.deserialize_aliases[alias.asname or alias.name] = fq + elif node.module == "types": + for alias in node.names: + if alias.name == "FunctionType": + self.functiontype_aliases.add(alias.asname or alias.name) self.generic_visit(node) def visit_While(self, node): @@ -3358,6 +3409,43 @@ def _check_signal_escape_patterns( return _elt(v) return None + def _is_compile_result(self, arg): + """True when ``arg`` is a ``compile(...)`` code object (bare / builtins / + single-assignment alias). Used to catch code objects executed through + ``types.FunctionType`` instead of eval/exec.""" + if isinstance(arg, ast.Call): + af = arg.func + if isinstance(af, ast.Name): + if af.id == "compile" or self.exec_from_aliases.get(af.id) == "compile": + return True + if _analyzer_on and _scope_idx.resolve(af.id, arg, "execb") == "compile": + return True + elif ( + isinstance(af, ast.Attribute) + and af.attr == "compile" + and _ast_name_matches(af.value, self.builtins_aliases) + ): + return True + if _analyzer_on and isinstance(arg, ast.Name): + if _scope_idx.resolve(arg.id, arg, "compiledany"): + return True + return False + + def _attr_obfuscation_targets(self): + # Modules whose DYNAMIC attribute / dict access (getattr, vars, __dict__) is + # obfuscation that reaches code execution: the exec / import / shell modules + # PLUS the deserializer modules -- getattr(pickle, 'loads')(x) and + # vars(pickle)['loads'](x) are just pickle.loads(x) with the name hidden. + return ( + _DYNAMIC_ATTR_TARGETS + | self.os_aliases + | self.subprocess_aliases + | self.importlib_aliases + | self.sys_aliases + | self.builtins_aliases + | set(self.deserialize_module_aliases) + ) + def visit_Call(self, node): func = node.func func_name = None @@ -3617,19 +3705,26 @@ def _check_signal_escape_patterns( # targets too: __import__('pickle').loads(blob) executes a reduce # payload even though a plain `import pickle` is benign. dynamic_desc = "dynamic import of a computed or sensitive module name" + elif ( + isinstance(func, ast.Name) + and func.id in ("getattr", "setattr") + and len(node.args) >= 2 + and isinstance(_const_fold(node.args[1], _const_env), str) + and _const_fold(node.args[1], _const_env) in _GADGET_DUNDERS + ): + # getattr(anything, '__globals__' / '__subclasses__' / ...) reaches an + # introspection gadget with no ast.Attribute for visit_Attribute to catch. + # Direct x.__globals__ is already flagged for ANY receiver, so flag the + # getattr-string form regardless of receiver too. + dynamic_desc = ( + "getattr() of an introspection gadget dunder " + f"({_const_fold(node.args[1], _const_env)})" + ) elif ( isinstance(func, ast.Name) and func.id == "vars" and node.args - and _ast_name_matches( - node.args[0], - _DYNAMIC_ATTR_TARGETS - | self.os_aliases - | self.subprocess_aliases - | self.importlib_aliases - | self.sys_aliases - | self.builtins_aliases, - ) + and _ast_name_matches(node.args[0], self._attr_obfuscation_targets()) ): # vars(os) / vars(__builtins__) returns the module __dict__, the same # obfuscation as os.__dict__['system'] but without the attribute access. @@ -3638,15 +3733,7 @@ def _check_signal_escape_patterns( isinstance(func, ast.Name) and func.id in ("getattr", "setattr") and node.args - and _ast_name_matches( - node.args[0], - _DYNAMIC_ATTR_TARGETS - | self.os_aliases - | self.subprocess_aliases - | self.importlib_aliases - | self.sys_aliases - | self.builtins_aliases, - ) + and _ast_name_matches(node.args[0], self._attr_obfuscation_targets()) ): # Stage 2 refinement: a benign constant attr (getattr(os, "getpid")) # is allowed; only a dynamic attr or a dangerous constant attr blocks. @@ -3675,9 +3762,50 @@ def _check_signal_escape_patterns( 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: + # Constant-fold the key so sys.modules.get('o' + 's') is caught, not + # just a bare literal (the module is already loaded by the prelude). + _key = _const_fold(node.args[0], _const_env) + if isinstance(_key, str) and _key.split(".")[0] in _DANGEROUS_IMPORT_NAMES: dynamic_desc = "sys.modules.get(...) access to a sensitive module" + elif ( + # globals().get('__builtins__') / locals().get(...) / vars().get(...) + # -- the .get() twin of the globals()['__builtins__'] subscript form. + isinstance(func, ast.Attribute) + and func.attr == "get" + and isinstance(func.value, ast.Call) + and isinstance(func.value.func, ast.Name) + and func.value.func.id in ("globals", "locals", "vars") + and not func.value.args + and node.args + ): + _key = _const_fold(node.args[0], _const_env) + if isinstance(_key, str) and ( + _key in ("__builtins__", "__builtin__") + or _key.split(".")[0] in _DANGEROUS_IMPORT_NAMES + ): + dynamic_desc = ( + "namespace-dict .get() access to builtins / a sensitive module" + ) + elif ( + ( + # types.FunctionType(compile(src, ...), {})() runs a code object WITHOUT + # eval/exec, so a dynamic compile() payload reaches execution here even + # though compile() alone is allowed. Flag when a FunctionType call takes + # a compile()-derived code object as its first argument. + ( + isinstance(func, ast.Attribute) + and func.attr == "FunctionType" + and _ast_name_matches(func.value, self.types_aliases) + ) + or (isinstance(func, ast.Name) and func.id in self.functiontype_aliases) + ) + and node.args + and self._is_compile_result(node.args[0]) + ): + dynamic_desc = ( + "types.FunctionType() executes a compile() code object " + "(bypasses the eval/exec gate)" + ) if dynamic_desc: dynamic_exec.append( { @@ -3702,13 +3830,7 @@ def _check_signal_escape_patterns( } ) elif node.attr == "__dict__" and _ast_name_matches( - node.value, - _DYNAMIC_ATTR_TARGETS - | self.os_aliases - | self.subprocess_aliases - | self.importlib_aliases - | self.sys_aliases - | self.builtins_aliases, + node.value, self._attr_obfuscation_targets() ): # os.__dict__['system']('id') reaches the sink with no getattr call for # the name-based checks to see. __dict__ on ordinary objects stays allowed. @@ -3742,8 +3864,10 @@ def _check_signal_escape_patterns( and _extract_string_from_node(v.args[1]) == "modules" ) if isinstance(node.ctx, ast.Load) and is_sys_modules: - key = _extract_string_from_node(node.slice) - if key is not None and key.split(".")[0] in _DANGEROUS_IMPORT_NAMES: + # Constant-fold the key so sys.modules['o' + 's'] is caught, not just a + # bare literal; a truly dynamic key (sys.modules[name]) stays allowed. + key = _const_fold(node.slice, _const_env) + if isinstance(key, str) and key.split(".")[0] in _DANGEROUS_IMPORT_NAMES: dynamic_exec.append( { "type": "dynamic_exec", @@ -3761,8 +3885,8 @@ def _check_signal_escape_patterns( 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 = _const_fold(node.slice, _const_env) + if isinstance(key, str) and ( key in ("__builtins__", "__builtin__") or key.split(".")[0] in _DANGEROUS_IMPORT_NAMES ): @@ -4517,6 +4641,12 @@ def _check_signal_escape_patterns( for arg in list(node.args) + [kw.value for kw in (node.keywords or [])]: s = _fold_read_arg(arg) if s is None: + # A pathlib expression carries no foldable string constant + # (open(Path('/etc') / 'passwd')), so resolve it the same way a + # Path(...).read_text() receiver is resolved before skipping. + rp = _pathlib_receiver_path(arg) + if rp is not None and _flag_read_path(node, rp, is_read_callee): + break continue if _flag_read_path(node, s, is_read_callee): break @@ -4658,16 +4788,33 @@ import os as _os, builtins as _bi, io as _io, pathlib as _pl # wrapper does (self shifts into the next arg), which would corrupt Path.open / # Path.write_text. Importing here makes the accessor capture the originals; the # confinement is applied on the public os / io / Path.* APIs below instead. -_WD = _os.path.realpath(__WORKDIR__) +# Capture the path helpers/separator into IMMUTABLE guard-local names BEFORE user code +# runs. _within() otherwise reads os.path.realpath / os.fspath / os.sep off the live +# module every call, so sandboxed code could reassign os.path.realpath (e.g. to a lambda +# that echoes an in-workdir path) right before a write and have the guard approve an +# outside target while the real open() still writes there. These references cannot be +# rebound by mutating the os module. +_realpath = _os.path.realpath +_fspath = _os.fspath +_sep = _os.sep +_WD = _realpath(__WORKDIR__) def _within(p): try: if isinstance(p, int): return True - rp = _os.path.realpath(_os.fspath(p)) + # os.path.realpath internally calls the LIVE os.fspath (posixpath.realpath does + # `filename = os.fspath(filename)`), so a sandboxed reassignment of os.fspath + # would still poison the resolution even though we hold the original realpath. + # Re-pin os.fspath to the captured original before resolving; a str target then + # resolves truthfully. (Re-pinning per check keeps it self-healing if user code + # re-patches; the real open() call receives the already-materialized str and does + # not route through os.fspath, so restoring it has no effect on the write itself.) + _os.fspath = _fspath + rp = _realpath(_fspath(p)) except Exception: return False - return rp == _WD or rp.startswith(_WD + _os.sep) + return rp == _WD or rp.startswith(_WD + _sep) def _deny(p, what): raise PermissionError( @@ -4691,10 +4838,11 @@ def _gwraps(real): def _fspath1(p): # Materialize a path-like ONCE so a stateful __fspath__ cannot return a workdir # path for the _within() check and a different path for the real syscall (TOCTOU). + # Uses the captured _fspath so a reassigned os.fspath cannot interpose here. if isinstance(p, int): return p try: - return _os.fspath(p) + return _fspath(p) except Exception: return p diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 07cd263207..ae9de2ca7f 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -607,4 +607,53 @@ def test_sandboxed_imports_still_work_under_guard(): disable_sandbox = False, ) assert '{"a": 1}' in out - assert "sandbox:" not in out + + +@_POSIX_ONLY +def test_sandboxed_realpath_monkeypatch_write_escape_denied(tmp_path): + # Sandboxed code reassigns os.path.realpath to a lambda that echoes an in-workdir + # path, then writes to an absolute path outside the workdir. If the guard read + # os.path.realpath off the live module every call, the reassignment would fool + # _within() into approving the outside write. The guard captures the original path + # helpers at prelude time, so the write is still denied. + session = "backstop-realpath-monkeypatch" + workdir = get_sandbox_workdir(session) + target = tmp_path / "realpath_escape_probe.txt" + if target.exists(): + target.unlink() + out = _python_exec( + "import os\n" + f"os.path.realpath = lambda p: {workdir!r} + '/ok'\n" + f"os.fspath = lambda p: {workdir!r} + '/ok'\n" + f"open({str(target)!r}, 'w').write('escaped')\n" + "print('WROTE-VIA-MONKEYPATCH')\n", + None, + 30, + session, + disable_sandbox = False, + ) + assert "sandbox:" in out or "PermissionError" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_fspath_monkeypatch_write_escape_denied(tmp_path): + # The os.fspath twin of the realpath monkeypatch: reassigning os.fspath must not + # let a materialized outside path slip past the confinement check. + session = "backstop-fspath-monkeypatch" + workdir = get_sandbox_workdir(session) + target = tmp_path / "fspath_escape_probe.txt" + if target.exists(): + target.unlink() + out = _python_exec( + "import os\n" + f"os.fspath = lambda p: {workdir!r} + '/ok'\n" + f"open({str(target)!r}, 'w').write('escaped')\n" + "print('WROTE-VIA-MONKEYPATCH')\n", + None, + 30, + session, + 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 11753361d2..39df457f17 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -1083,3 +1083,139 @@ class TestEvalExecRecursion: b"+AG8AcwAuAHMAeQBzAHQAZQBtACgAJwBpAGQAJwAp-" ) _ok("exec(%r)" % benign) + + +class TestRound6Bypasses: + """Sixth-round Codex findings: pathlib read args, getattr gadget dunders, namespace + .get() lookups, folded sys.modules keys, builtins __import__ aliases, deserializer + obfuscation, and code objects executed through types.FunctionType.""" + + def test_pathlib_open_read_resolved(self): + # open(Path('/etc') / 'passwd') carries no foldable string constant, but the + # pathlib resolver must reconstruct the path so it blocks like open('/etc/passwd'). + assert ( + _check_code_safety("from pathlib import Path\nopen(Path('/etc') / 'passwd').read()") + is not None + ) + assert ( + _check_code_safety( + "from pathlib import Path\nopen(Path('/etc').joinpath('passwd')).read()" + ) + is not None + ) + # A benign relative pathlib read stays allowed (no false positive). + _ok("from pathlib import Path\nopen(Path('data') / 'train.csv').read()") + + @pytest.mark.parametrize( + "code", + [ + "getattr(object, '__subclasses__')()", + "getattr(lambda: 0, '__globals__')", + "setattr(object, '__bases__', ())", + "getattr(getattr(object, '__subclasses__')()[0], '__init__')", + "getattr(().__class__, '__bases__')", + ], + ) + def test_getattr_gadget_dunder_any_receiver_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_getattr_benign_attr_allowed(self): + # A non-gadget attribute name via getattr on an ordinary object stays allowed. + _ok("getattr(object, 'mro')") + _ok("import numpy as np\ngetattr(np, 'zeros')((3, 3))") + + @pytest.mark.parametrize( + "code", + [ + "globals().get('__builtins__').__import__('os').system('id')", + "locals().get('__builtins__')", + "vars().get('os')", + ], + ) + def test_namespace_get_builtins_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_namespace_get_benign_key_allowed(self): + _ok("d = {'x': 1}\nd.get('x')") + _ok("globals().get('my_var')") + + @pytest.mark.parametrize( + "code", + [ + "import sys\nsys.modules['o' + 's'].system('id')", + "import sys\nsys.modules.get('o' + 's').system('id')", + "import sys\nk = 'o' + 's'\nsys.modules[k].system('id')", + ], + ) + def test_sys_modules_folded_key_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_sys_modules_dynamic_key_allowed(self): + # A genuinely dynamic key (not constant-foldable) stays allowed -- legit uses + # like sys.modules[name] for an unknown name must not be over-blocked. + _ok("import sys\ndef f(name):\n return sys.modules.get(name)\nf('json')") + + def test_builtins_import_alias_blocked(self): + assert ( + _check_code_safety("from builtins import __import__ as imp\nimp('os').system('id')") + is not None + ) + assert ( + _check_code_safety("import builtins\nbuiltins.__import__('os').system('id')") + is not None + ) + + @pytest.mark.parametrize( + "code", + [ + "import pickle\ngetattr(pickle, 'loads')(b'x')", + "import pickle\nvars(pickle)['loads'](b'x')", + "import pickle\npickle.__dict__['loads'](b'x')", + "import pickle as p\ngetattr(p, 'loads')(b'x')", + ], + ) + def test_deserializer_attr_obfuscation_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_deserializer_benign_attr_allowed(self): + # getattr(pickle, 'dumps') (serialize) is not a code-exec sink -> allowed. + _ok("import pickle\ngetattr(pickle, 'dumps')({'a': 1})") + + @pytest.mark.parametrize( + "code", + [ + "import types\n" + "def f(src):\n types.FunctionType(compile(src, '', 'exec'), {})()\nf('import os')", + "from types import FunctionType as F\n" + "def f(src):\n F(compile(src, '', 'exec'), {})()\nf('x')", + "import types\n" + "def f(src):\n c = compile(src, '', 'exec')\n types.FunctionType(c, {})()\nf('x')", + ], + ) + def test_functiontype_compile_result_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_functiontype_without_compile_allowed(self): + # types.FunctionType on an ordinary code object (fn.__code__) is not the + # dynamic-compile gadget; keep it allowed to avoid over-blocking metaprogramming. + _ok("import types\ndef g():\n return 1\ntypes.FunctionType(g.__code__, {})") + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system(\"python -c 'print(1)'\")", + "import os\nos.system('python3 evil.py')", + "import subprocess\nsubprocess.run(['python3', '-c', 'print(1)'])", + "import os\nos.system('perl -e \"print 1\"')", + "import os\nos.system('node -e \"1\"')", + ], + ) + def test_interpreter_child_process_blocked(self, code): + # A child interpreter runs WITHOUT the in-process write guard, so spawning one + # escapes the sandbox; interpreters are blocked at shell command position. + assert _check_code_safety(code) is not None, code + + def test_benign_shell_still_allowed(self): + _ok("import os\nos.system('echo hello')") + _ok("import os\nos.system('ls -la')") + _ok("import subprocess\nsubprocess.run(['echo', 'hi'])")