From f11fbdcb7fdd3227deaeaf4b37b81e5fffb4d04e Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Fri, 10 Jul 2026 16:05:17 +0000 Subject: [PATCH] Harden sandbox: compile(source=) payload; workdir import vetter -- PEP263 decode, ignore bytecode cache, refuse symlinks, network sinks, os import aliases Close six P1 bypasses Codex found on the round-41 branch: - compile(source=...) keyword payload: compile() accepts its source as the source= keyword, but the analyzer only recovered the 1st positional arg, so a keyword-only compile feeding types.FunctionType(code)() (or exec(compile(source=...))) was treated as having no payload and ran unscanned. Recover the source= keyword too (new _compile_source_node), at both the code-object tracking and exec(compile()) sites. The remaining five harden the workdir-module import vetter (a helper .py the user wrote is vetted before import; each of these slipped a payload past it): - PEP 263 source encoding: the vetter read modules as fixed UTF-8, but Python's loader honors an encoding cookie. A `# coding: utf_7` module hides os.system in what the UTF-8 scan sees as a comment (raw +AAo- bytes are a newline under UTF-7). Decode with importlib.util.decode_source so the vetter sees what the loader will run. - bytecode cache: after scanning the source, returning the original spec let SourceFileLoader satisfy the import from a planted __pycache__ .pyc whose header matches the harmless source. Run the EXACT vetted source via a dedicated loader (_GuardVettedSourceLoader) so the bytecode cache is never consulted. - symlinked module: a workdir module that is a symlink to an outside file had a realpath outside the workdir, so it was treated as not-workdir and handed to the default loader unvetted. Decide workdir-membership by the origin path, then fail closed when the realpath escapes. - network sinks: the vetter only checked command-exec/eval, so a helper doing socket.create_connection(...) bypassed the static network policy (no runtime network backstop). Refuse a workdir module that imports a network primitive (socket / ssl / ftplib / smtplib / requests / httpx / aiohttp / ...). - os import aliases: sink references were only recognized when rooted at literal os / posix, so import os as o; s = o.system; s(...) passed (the assignment, not a direct call). Record os / posix import aliases before checking sink references. Regression coverage: TestRound42Bypasses in tests/test_sandbox_tools.py (compile source= keyword, positional, and exec(compile()) forms) and five workdir-module vetter tests in tests/test_sandbox_runtime_backstop.py (utf-7 encoding denied, forged pyc ignored while the vetted source runs, symlinked module denied, network sink denied, os import alias denied). --- studio/backend/core/inference/tools.py | 115 ++++++++++++--- .../tests/test_sandbox_runtime_backstop.py | 139 ++++++++++++++++++ studio/backend/tests/test_sandbox_tools.py | 26 ++++ 3 files changed, 263 insertions(+), 17 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 5ba62ce72f..19e3648689 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -4227,6 +4227,18 @@ def _recovered_source(v): return _to_text(v) +def _compile_source_node(node): + """The SOURCE argument of a compile() call: the 1st positional arg or the ``source=`` + keyword. compile() accepts its payload either way, so a keyword-only call + (compile(source='...', filename='

', mode='exec')) must still be recovered.""" + if node.args: + return node.args[0] + for kw in node.keywords or []: + if kw.arg == "source": + return kw.value + return None + + def _compile_mode(node, const_env): """Recover a compile()'s literal mode= (3rd positional or keyword), else 'exec'.""" mode_node = None @@ -4678,12 +4690,13 @@ def _build_scope_alias_index(tree, const_env): and isinstance(rhs_eff.func, ast.Name) and emap.get(rhs_eff.func.id) == "compile" ) - ) and rhs_eff.args: + ) and _compile_source_node(rhs_eff) is not None: # Any `c = compile(...)` (bare / builtins.compile / from-import alias) # binds a code object, tracked for the types.FunctionType(c) execution - # gadget below (dynamic or foldable payload). + # gadget below (dynamic or foldable payload). The source may be positional + # OR the source= keyword. camap[name] = True - v = _const_fold(rhs_eff.args[0], const_env) + v = _const_fold(_compile_source_node(rhs_eff), const_env) if isinstance(v, (str, bytes, bytearray)): cmap[name] = ( _recovered_source(v), @@ -4949,14 +4962,15 @@ def _recover_exec_payload(node, func_id, const_env, compiled_env): arg0 = node.args[0] base_mode = "eval" if func_id == "eval" else "exec" - # exec(compile("...", ...)) / eval(compile("...", "", "eval")) + # exec(compile("...", ...)) / eval(compile("...", "", "eval")) -- the compile source may be + # positional or the source= keyword. if ( isinstance(arg0, ast.Call) and isinstance(arg0.func, ast.Name) and arg0.func.id == "compile" - and arg0.args + and _compile_source_node(arg0) is not None ): - v = _const_fold(arg0.args[0], const_env) + v = _const_fold(_compile_source_node(arg0), const_env) if isinstance(v, (str, bytes, bytearray)): return ( "RECOVERED", @@ -10168,7 +10182,16 @@ except Exception: try: import ast as _gast import importlib.machinery as _gimach + import importlib.util as _gimportutil _GUARD_WORKDIR_REAL = _os.path.realpath(__WORKDIR__) + # Network-capable modules: a workdir helper that opens a socket / HTTP client bypasses the + # static network policy (there is no runtime network backstop), so refuse importing one. + # (urllib / http bare tops are left out -- urllib.parse etc. are benign, OS isolation remains + # the boundary for the dotted network submodules.) + _GUARD_NET_MODS = frozenset({ + "socket", "ssl", "ftplib", "smtplib", "telnetlib", "poplib", "imaplib", "nntplib", + "requests", "httpx", "aiohttp", "urllib3", "pycurl", "websocket", "websockets", "paramiko", + }) _GUARD_EXEC_ATTRS = frozenset({ "system", "popen", "popen2", "popen3", "popen4", "startfile", "execl", "execle", "execlp", "execlpe", "execv", "execve", "execvp", "execvpe", @@ -10189,14 +10212,23 @@ try: _tree = _gast.parse(_src) except _bi.BaseException: return True # unparseable workdir module -> fail closed + # Pre-pass: record os / posix import ALIASES (import os as o) so an aliased sink reference + # that is only assigned (s = o.system) -- not directly called -- is still recognized. + _recv = set(_GUARD_EXEC_RECEIVERS) for _nd in _gast.walk(_tree): if isinstance(_nd, _gast.Import): for _al in _nd.names: - if _al.name.split(".")[0] in _GUARD_EXEC_MODS: + if _al.name in ("os", "posix"): + _recv.add(_al.asname or _al.name) + for _nd in _gast.walk(_tree): + if isinstance(_nd, _gast.Import): + for _al in _nd.names: + _top = _al.name.split(".")[0] + if _top in _GUARD_EXEC_MODS or _top in _GUARD_NET_MODS: return True elif isinstance(_nd, _gast.ImportFrom): _mroot = (_nd.module or "").split(".")[0] - if _mroot in _GUARD_EXEC_MODS: + if _mroot in _GUARD_EXEC_MODS or _mroot in _GUARD_NET_MODS: return True # `from os import system` / `from os import *` binds a BARE sink name into the # module namespace; a later bare system('id') call has no os. attribute to catch. @@ -10213,10 +10245,10 @@ try: "eval", "exec", "compile", "__import__"): return True elif isinstance(_nd, _gast.Attribute): - # A sink-named attribute REFERENCE (even uncalled) rooted at os / posix - # (x = os.system). A same-named attribute on an unrelated object - # (p.system = 'linux') is NOT a sink, so require a sink-module receiver root. - if _nd.attr in _GUARD_EXEC_ATTRS and _guard_attr_root(_nd.value) in _GUARD_EXEC_RECEIVERS: + # A sink-named attribute REFERENCE (even uncalled) rooted at os / posix / an + # os alias (x = os.system, s = o.system). A same-named attribute on an unrelated + # object (p.system = 'linux') is NOT a sink, so require a sink-module receiver. + if _nd.attr in _GUARD_EXEC_ATTRS and _guard_attr_root(_nd.value) in _recv: return True # A workdir module that touches the import machinery (sys.meta_path / # sys.path_hooks / sys.path_importer_cache) can remove THIS vetter, then a @@ -10225,6 +10257,34 @@ try: if _nd.attr in ("meta_path", "path_hooks", "path_importer_cache"): return True return False + def _guard_under_workdir(_p): + return _p == _GUARD_WORKDIR_REAL or _p.startswith(_GUARD_WORKDIR_REAL + _os.sep) + + class _GuardVettedSourceLoader: + # Executes the EXACT source string the vetter scanned, so the loader can never satisfy + # the import from a planted bytecode cache (.pyc) or re-decode the file differently than + # it was vetted. __path__ for a package still comes from the spec's search locations. + def __init__(self, _name, _path, _src, _is_pkg): + self._n = _name + self._p = _path + self._s = _src + self._pkg = _is_pkg + + def create_module(self, _spec): + return None + + def exec_module(self, _module): + exec(compile(self._s, self._p, "exec"), _module.__dict__) + + def get_filename(self, _name=None): + return self._p + + def is_package(self, _name=None): + return self._pkg + + def get_source(self, _name=None): + return self._s + class _GuardWorkdirImportVetter: def find_spec(self, _name, _path=None, _target=None): try: @@ -10235,11 +10295,21 @@ try: if not _orig: return None # namespace / builtin / frozen: no file to vet, not workdir-sourced try: + _abs = _os.path.abspath(_orig) _rp = _os.path.realpath(_orig) except _bi.BaseException: return None - if not (_rp == _GUARD_WORKDIR_REAL or _rp.startswith(_GUARD_WORKDIR_REAL + _os.sep)): - return None # not a workdir module; let the default finders load it + _orig_in_wd = _guard_under_workdir(_abs) or _guard_under_workdir( + _os.path.dirname(_abs) + ) + _rp_in_wd = _guard_under_workdir(_rp) + if not _orig_in_wd and not _rp_in_wd: + return None # genuinely not a workdir module; let the default finders load it + # A workdir-sourced origin whose REALPATH escapes the workdir (a symlink to an outside + # file) must fail closed, not be handed to the default loader unvetted. + if not _rp_in_wd: + raise _bi.ImportError( + "sandbox: refusing symlinked workdir module " + _name) # A workdir module must be a .py we can read + scan. A sourceless .pyc / native .so / # any other non-source file under the workdir cannot be statically vetted, so refuse # it: a planted legacy evil.pyc would otherwise run its bytecode via the default @@ -10247,17 +10317,28 @@ try: if not _orig.endswith(".py"): raise _bi.ImportError( "sandbox: refusing to import non-source workdir module " + _name) + # Decode with Python's PEP 263 source encoding (importlib.util.decode_source), NOT a + # fixed utf-8: a `# coding: utf_7` module the loader would decode as UTF-7 must be + # vetted as UTF-7, or a payload hidden in what a UTF-8 scan sees as a comment runs. try: - _fh = _io.open(_orig, "r", encoding="utf-8", errors="replace") + _fb = _io.open(_orig, "rb") try: - _msrc = _fh.read() + _raw = _fb.read() finally: - _fh.close() + _fb.close() + _msrc = _gimportutil.decode_source(_raw) except _bi.BaseException: raise _bi.ImportError("sandbox: cannot vet workdir module " + _name) if _guard_module_src_unsafe(_msrc): raise _bi.ImportError( "sandbox: refusing to import unvetted workdir module " + _name) + # Run the EXACT vetted source via our loader so a planted matching .pyc can never be + # executed instead (the default SourceFileLoader would satisfy the import from a + # __pycache__ .pyc whose header matches the harmless source). + _is_pkg = _spec.submodule_search_locations is not None or _os.path.basename( + _orig + ) == "__init__.py" + _spec.loader = _GuardVettedSourceLoader(_name, _orig, _msrc, _is_pkg) return _spec _sys.meta_path.insert(0, _GuardWorkdirImportVetter()) except _bi.BaseException: diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 3b4f4db7b0..9456c55f79 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -789,6 +789,145 @@ def test_sandboxed_workdir_module_meta_path_mutation_denied(): os.remove(os.path.join(workdir, "mp_popper.py")) +@_POSIX_ONLY +def test_sandboxed_utf7_encoded_workdir_module_denied(): + # A `# coding: utf_7` module hides os.system in what a UTF-8 scan reads as a comment (the raw + # +AAo- bytes are a newline under UTF-7). The vetter must decode with PEP 263 like the loader + # will, so the real os.system is seen and refused. + session = "backstop-workdir-utf7" + workdir = get_sandbox_workdir(session) + data = b"# coding: utf_7\nimport os\npass #+AAo-os.system('echo PWNED_UTF7')\n" + with open(os.path.join(workdir, "evilenc.py"), "wb") as f: + f.write(data) + try: + out = _python_exec( + "import evilenc; print('REACHED_' + 'BODY')", + None, + 30, + session, + disable_sandbox = False, + ) + assert "PWNED_UTF7" not in out + assert "REACHED_BODY" not in out + assert "sandbox:" in out or "ImportError" in out + finally: + os.remove(os.path.join(workdir, "evilenc.py")) + + +@_POSIX_ONLY +def test_sandboxed_forged_pyc_workdir_module_ignored(): + # A harmless source plus a planted __pycache__ .pyc whose header matches it but whose body is + # malicious: the vetter scans the safe source, but the module must run the VETTED SOURCE + # (not the cached bytecode), so the planted payload never executes. + import importlib.util + import marshal + import struct + + session = "backstop-workdir-forgedpyc" + workdir = get_sandbox_workdir(session) + src_path = os.path.join(workdir, "forged.py") + with open(src_path, "w") as f: + f.write("VALUE = 7\nprint('SOURCE_RAN')\n") + st = os.stat(src_path) + mal = compile("import os\nos.system('echo PWNED_FORGEDPYC')\n", "forged.py", "exec") + pyc_dir = os.path.join(workdir, "__pycache__") + os.makedirs(pyc_dir, exist_ok = True) + pyc_path = os.path.join(pyc_dir, f"forged.{sys.implementation.cache_tag}.pyc") + with open(pyc_path, "wb") as f: + f.write(importlib.util.MAGIC_NUMBER) + f.write(struct.pack("', 'exec')\n" + "types.FunctionType(c, {})()", + # exec(compile(source=...)) keyword form. + "exec(compile(source='import os\\nos.system(\"touch /tmp/x\")', filename='

', mode='exec'))", + ], + ) + def test_compile_source_keyword_payload_blocked(self, code): + assert _check_code_safety(code) is not None, code