diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 6294f35eb8..92050b194c 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1671,6 +1671,59 @@ def _fold_cap(value): return value +def _too_wide(n): + """A format width / precision / size arg large enough to OOM the folder.""" + return isinstance(n, int) and not isinstance(n, bool) and n > _FOLD_MAXLEN + + +# Format-spec mini-language: reject an oversized width or precision BEFORE format() +# allocates the padded string. format()/str.format()/f-strings all run in the Studio +# process during static analysis, ahead of the child-subprocess rlimits. +_FMT_SPEC_RE = re.compile(r"^(?:.?[<>=^])?[+\- ]?z?#?0?(\d+)?[,_]?(?:\.(\d+))?[a-zA-Z%]?$") + + +def _format_spec_ok(spec): + if not isinstance(spec, str) or not spec: + return True + m = _FMT_SPEC_RE.match(spec) + if not m: + return True # unrecognized spec: let format() itself decide at runtime + return not any(g and _too_wide(int(g)) for g in m.groups()) + + +def _format_template_ok(template): + """Every replacement field of a str.format template has a bounded width.""" + if not isinstance(template, str): + return True + try: + import string as _string + + for _lit, _field, _spec, _conv in _string.Formatter().parse(template): + if _spec and not _format_spec_ok(_spec): + return False + except Exception: + return True + return True + + +_PRINTF_WIDTH_RE = re.compile(r"%[-+ #0]*(\d+)?(?:\.(\d+))?[hlL]?[diouxXeEfFgGcrsab%]") + + +def _printf_ok(fmt): + """Percent-format string with no oversized field width / precision.""" + if isinstance(fmt, (bytes, bytearray)): + try: + fmt = fmt.decode("latin-1") + except Exception: + return True + if not isinstance(fmt, str): + return True + 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 _fold_apply_codec(name, data): """Pure data transforms only (rot13/hex/base64/zlib/text codecs). Bounded zlib.""" name = name.lower().replace("-", "_") @@ -1798,6 +1851,8 @@ def _const_fold( v = {114: repr, 115: str, 97: ascii}[part.conversion](v) except Exception: return None + if not _format_spec_ok(spec if isinstance(spec, str) else ""): + return None # oversized f-string width/precision: refuse pre-format try: out.append(format(v, spec if isinstance(spec, str) else "")) except Exception: @@ -1820,10 +1875,20 @@ def _const_fold( if isinstance(right, (str, bytes, bytearray)) and isinstance(left, int): if len(right) * max(left, 0) > _FOLD_MAXLEN: return None + # list/tuple repetition allocates len(seq)*n elements before _fold_cap + # (which only sizes str/bytes) can reject it -- cap it here too. + if isinstance(left, (list, tuple)) and isinstance(right, int): + if len(left) * max(right, 0) > _FOLD_MAX_SEQ: + return None + if isinstance(right, (list, tuple)) and isinstance(left, int): + if len(right) * max(left, 0) > _FOLD_MAX_SEQ: + return None return _fold_cap(left * right) if isinstance(op, ast.Add): return _fold_cap(left + right) if isinstance(op, ast.Mod): + if isinstance(left, (str, bytes, bytearray)) and not _printf_ok(left): + return None # oversized %-format width/precision: refuse pre-format return _fold_cap(left % right) if isinstance(op, ast.Sub): return _fold_cap(left - right) @@ -1889,6 +1954,18 @@ def _const_fold( return None +def _is_path_join_owner(nv): + """AST for ``os.path`` (Attribute) or ``posixpath`` / ``ntpath`` (Name).""" + if ( + isinstance(nv, ast.Attribute) + and nv.attr == "path" + and isinstance(nv.value, ast.Name) + and nv.value.id == "os" + ): + return True + return isinstance(nv, ast.Name) and nv.id in ("posixpath", "ntpath") + + def _fold_call(node, _state, _depth): """Fold a whitelisted pure builtin / method / decode call, else None.""" f = node.func @@ -1946,6 +2023,16 @@ def _fold_call(node, _state, _depth): if isinstance(f, ast.Attribute): attr = f.attr owner = f.value + # os.path.join('/etc', 'passwd') / posixpath.join(...) / ntpath.join(...): + # fold literal path builders so the sensitive-read scanner sees the concrete + # path (open(os.path.join('/etc','passwd')) must not be treated as opaque). + if attr == "join" and _is_path_join_owner(owner): + if args and all(isinstance(x, str) for x in args): + try: + return _fold_cap(os.path.join(*args)) + except Exception: + return None + return None if isinstance(owner, ast.Name): mod = owner.id try: @@ -1985,6 +2072,14 @@ def _fold_call(node, _state, _depth): call_args.append( list(a) if attr == "join" and isinstance(a, (list, tuple)) else a ) + # Padding methods take a width as their first arg; str.format takes a + # template with per-field widths. Reject an oversized width before the + # method allocates the padded string during folding. + if attr in ("center", "ljust", "rjust", "zfill") and call_args: + if _too_wide(call_args[0]): + return None + if attr == "format" and not _format_template_ok(recv): + return None return _fold_cap(getattr(recv, attr)(*call_args, **kwargs)) except Exception: return None @@ -2258,6 +2353,53 @@ def _to_text(value): return value +# PEP 263 source-encoding cookie ("# -*- coding: utf-8 -*-", "# coding: utf_7"). +_CODING_COOKIE_RE = re.compile(rb"coding[:=]\s*([-\w.]+)") +_CODING_COOKIE_TEXT_RE = re.compile(r"coding[:=]\s*([-\w.]+)") + + +def _decode_source_bytes(data): + """Decode an exec/compile *bytes* payload the way CPython would. + + exec()/eval()/compile() honor a PEP 263 coding cookie on bytes, so the analyzer + must decode with that cookie's codec (not a fixed UTF-8 view) or a snippet like + ``exec(b"# coding: utf_7\\n#+AAo-__import__('os').system('id')")`` reads as pure + comments under UTF-8 while actually running hidden code. Detect the encoding, + decode, then neutralize the cookie so ast.parse(str) does not reject the decoded + text (a str carrying a coding declaration raises SyntaxError), preserving line + numbers so the recursive analysis sees the real source. + """ + data = bytes(data) + enc = "utf-8" + try: + import io as _io_mod + import tokenize as _tok + + enc, _ = _tok.detect_encoding(_io_mod.BytesIO(data).readline) + except Exception: + enc = "utf-8" + for _cand in (enc, "utf-8"): + try: + text = data.decode(_cand) + break + except Exception: + text = None + if text is None: + text = data.decode("latin-1", "replace") + lines = text.split("\n") + for _i in range(min(2, len(lines))): + if _CODING_COOKIE_TEXT_RE.search(lines[_i]): + lines[_i] = _CODING_COOKIE_TEXT_RE.sub("coding_neutralized", lines[_i], count=1) + return "\n".join(lines) + + +def _recovered_source(v): + """Text an exec/compile sink actually runs: cookie-aware decode for bytes.""" + if isinstance(v, (bytes, bytearray)): + return _decode_source_bytes(v) + return _to_text(v) + + def _compile_mode(node, const_env): """Recover a compile()'s literal mode= (3rd positional or keyword), else 'exec'.""" mode_node = None @@ -2283,7 +2425,10 @@ def _build_exec_env(tree, const_env): exec_aliases: dict[str, str] = {} compiled_env: dict[str, tuple] = {} - for stmt in getattr(tree, "body", []): + # Walk the whole tree, not just tree.body: an alias assigned inside a function + # (def f(): e = exec; e("...")) must still be unwrapped. store_counts spans the + # tree, so the "stored exactly once" guard keeps this single-assignment (low-FP). + for stmt in ast.walk(tree): if not ( isinstance(stmt, ast.Assign) and len(stmt.targets) == 1 @@ -2305,7 +2450,7 @@ def _build_exec_env(tree, const_env): v = _const_fold(rhs.args[0], const_env) if isinstance(v, (str, bytes, bytearray)): compiled_env[name] = ( - _to_text(v), + _recovered_source(v), _compile_mode(rhs, const_env), isinstance(v, (bytes, bytearray)), ) @@ -2440,7 +2585,7 @@ def _recover_exec_payload(node, func_id, const_env, exec_aliases, compiled_env): if isinstance(v, (str, bytes, bytearray)): return ( "RECOVERED", - _to_text(v), + _recovered_source(v), _compile_mode(arg0, const_env), isinstance(v, (bytes, bytearray)), ) @@ -2454,7 +2599,7 @@ def _recover_exec_payload(node, func_id, const_env, exec_aliases, compiled_env): v = _const_fold(arg0, const_env) if isinstance(v, (str, bytes, bytearray)): mode = _compile_mode(node, const_env) if func_id == "compile" else base_mode - return ("RECOVERED", _to_text(v), mode, isinstance(v, (bytes, bytearray))) + return ("RECOVERED", _recovered_source(v), mode, isinstance(v, (bytes, bytearray))) return ("DYNAMIC", None, None, False) @@ -3225,8 +3370,33 @@ def _check_signal_escape_patterns( mod = _extract_string_from_node(node.args[0]) else: mod = None - if mod is None or mod.split(".")[0] in _DANGEROUS_IMPORT_NAMES: + _mod_top = mod.split(".")[0] if mod else None + if ( + mod is None + or _mod_top in _DANGEROUS_IMPORT_NAMES + or _mod_top in _DESERIALIZE_MODULES + ): + # Deserializer modules (pickle/marshal/...) are dangerous import + # 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 == "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, + ) + ): + # vars(os) / vars(__builtins__) returns the module __dict__, the same + # obfuscation as os.__dict__['system'] but without the attribute access. + dynamic_desc = "vars() on a sensitive module (dict obfuscation)" elif ( isinstance(func, ast.Name) and func.id in ("getattr", "setattr") @@ -3963,6 +4133,40 @@ def _check_signal_escape_patterns( # open()/read callees, so benign relative-path building (os.path.join('..','x')) # is not caught. Dynamic (non-foldable) paths are left to the runtime backstop. _READ_METHODS = ("read_text", "read_bytes") + # Pathlib read methods carry the path on the RECEIVER, not in an argument: + # Path('../../.ssh/id_rsa').read_text() has no call args, so the constructor path + # must be inspected separately. + _PATHLIB_READ_METHODS = ("read_text", "read_bytes", "open") + _PATHLIB_CTORS = ( + "Path", + "PurePath", + "PosixPath", + "PurePosixPath", + "WindowsPath", + "PureWindowsPath", + ) + + def _pathlib_receiver_path(recv): + if ( + isinstance(recv, ast.Call) + and isinstance(recv.func, ast.Name) + and recv.func.id in _PATHLIB_CTORS + and recv.args + ): + v = _const_fold(recv.args[0], _const_env) + if isinstance(v, (str, bytes, bytearray)): + return _to_text(v) + return None + + def _flag_read_path(node, s, is_read_callee): + norm = s.replace("\\", "/") + if _is_sensitive_abs_path(norm): + _fs_block(node, f"{s!r} is a sensitive host identity / credential file") + return True + if is_read_callee and (s[:1] == "~" or ".." in norm.split("/")): + _fs_block(node, f"{s!r} escapes the session workdir via path traversal") + return True + return False class _SensitiveReadVisitor(ast.NodeVisitor): def visit_Call(self, node): @@ -3978,17 +4182,17 @@ def _check_signal_escape_patterns( or fq in ("io.open", "os.open") or method in _READ_METHODS ) + # Pathlib read on a literal Path(...) receiver: check the constructor path. + if isinstance(f, ast.Attribute) and f.attr in _PATHLIB_READ_METHODS: + rp = _pathlib_receiver_path(f.value) + if rp is not None and _flag_read_path(node, rp, True): + return for arg in list(node.args) + [kw.value for kw in (node.keywords or [])]: v = _const_fold(arg, _const_env) s = _to_text(v) if isinstance(v, (str, bytes, bytearray)) else None if s is None: continue - norm = s.replace("\\", "/") - if _is_sensitive_abs_path(norm): - _fs_block(node, f"{s!r} is a sensitive host identity / credential file") - break - if is_read_callee and (s[:1] == "~" or ".." in norm.split("/")): - _fs_block(node, f"{s!r} escapes the session workdir via path traversal") + if _flag_read_path(node, s, is_read_callee): break self.generic_visit(node) @@ -4121,7 +4325,7 @@ def _truncate(text: str, limit: int = _MAX_OUTPUT_CHARS) -> str: # and the realpath-before-open TOCTOU window under adversarial in-sandbox threading. # -------------------------------------------------------------------------- _SANDBOX_GUARD_SRC = r""" -import os as _os, builtins as _bi, functools as _ft, io as _io, pathlib as _pl +import os as _os, builtins as _bi, io as _io, pathlib as _pl # io + pathlib are imported BEFORE any patching on purpose: on Python <= 3.11 # pathlib._NormalAccessor captures io.open / os.* into class attributes at import # time. A C builtin captured there does not bind on instance access, but a Python @@ -4144,8 +4348,22 @@ def _deny(p, what): "sandbox: %s outside the session workdir is not permitted: %r" % (what, p) ) +def _gwraps(real): + # Like functools.wraps but WITHOUT publishing __wrapped__: functools.wraps stores + # the ORIGINAL unguarded callable on w.__wrapped__, and sandboxed code could reach + # it (builtins.open.__wrapped__('/etc/x', 'w'), os.rename.__wrapped__(...)) to call + # straight through every confinement below. Copy only the cosmetic metadata. + def _deco(w): + for _a in ("__module__", "__name__", "__qualname__", "__doc__"): + try: + setattr(w, _a, getattr(real, _a)) + except Exception: + pass + return w + return _deco + def _guard_open_like(real): - @_ft.wraps(real) + @_gwraps(real) def w(file, mode="r", *a, **k): m = mode if isinstance(mode, str) else "r" if any(c in m for c in "wax+") and not _within(file): @@ -4163,7 +4381,7 @@ _WRITE_OFLAGS = ( getattr(_os, "O_WRONLY", 0) | getattr(_os, "O_RDWR", 0) | getattr(_os, "O_CREAT", 0) | getattr(_os, "O_TRUNC", 0) | getattr(_os, "O_APPEND", 0) ) -@_ft.wraps(_real_osopen) +@_gwraps(_real_osopen) def _guarded_osopen(path, flags, *a, **k): mutating = (not isinstance(flags, int)) or bool(flags & _WRITE_OFLAGS) if mutating: @@ -4178,7 +4396,7 @@ def _wrap1(mod, name, what): orig = getattr(mod, name, None) if orig is None: return - @_ft.wraps(orig) + @_gwraps(orig) 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 @@ -4196,7 +4414,7 @@ def _wrap2(mod, name, both): orig = getattr(mod, name, None) if orig is None: return - @_ft.wraps(orig) + @_gwraps(orig) def w(src, dst, *a, **k): if any(k.get(_f) is not None for _f in ("dir_fd", "src_dir_fd", "dst_dir_fd")): _deny(dst, name + " (dir_fd)") # fd-relative target: a realpath check is meaningless @@ -4218,6 +4436,42 @@ try: except Exception: pass +# The low-level C module _io is where io.open / builtins.open originate; patching the +# io alias above leaves _io.open untouched, so `import _io; _io.open(p, 'w')` would +# escape. Guard the underlying entry point too. +try: + import _io as _lowio + _lowio.open = _guard_open_like(_lowio.open) +except Exception: + pass + +# Confine the current working directory: os.chdir to a dir outside the workdir would +# let a later relative write/read (which the static read scan treats as local) escape. +# os.fchdir takes an fd whose target we cannot cheaply realpath, so deny it outright. +_wrap1(_os, "chdir", "chdir") +try: + _real_fchdir = _os.fchdir + @_gwraps(_real_fchdir) + def _guarded_fchdir(fd): + _deny(fd, "fchdir") + _os.fchdir = _guarded_fchdir +except Exception: + pass + +# fd-based metadata mutators operate on an already-open descriptor, so a read-only +# os.open of an outside file (allowed -- reads are not confined) could still be reused +# to mutate host metadata. Deny them; sandboxed compute has no need to chmod/chown by fd. +def _make_fd_denier(_name, _orig): + @_gwraps(_orig) + def _w(fd, *a, **k): + _deny(fd, _name) + return _w +for _n in ("fchmod", "fchown"): + try: + setattr(_os, _n, _make_fd_denier(_n, getattr(_os, _n))) + except Exception: + pass + try: import shutil as _sh _wrap1(_sh, "rmtree", "rmtree") @@ -4232,7 +4486,7 @@ try: # Path.open("w"): wrap the public method directly (mode-aware). Version-robust # because pathlib's accessor holds the original io.open (captured at the top). _real_path_open = _pl.Path.open - @_ft.wraps(_real_path_open) + @_gwraps(_real_path_open) def _guarded_path_open(self, mode="r", *a, **k): m = mode if isinstance(mode, str) else "r" if any(c in m for c in "wax+") and not _within(self): @@ -4244,7 +4498,7 @@ try: orig = getattr(_pl.Path, name, None) if orig is None: return - @_ft.wraps(orig) + @_gwraps(orig) def w(self, *a, **k): if not _within(self): _deny(str(self), "Path." + name) diff --git a/studio/backend/tests/test_sandbox_aliasing.py b/studio/backend/tests/test_sandbox_aliasing.py index bebce34cf6..3fccfd575e 100644 --- a/studio/backend/tests/test_sandbox_aliasing.py +++ b/studio/backend/tests/test_sandbox_aliasing.py @@ -61,6 +61,12 @@ class TestFuncLocalAliasBlocked: _ok("def run():\n f = sorted\n return f([3, 1, 2])\nrun()") _ok("import os\ndef run():\n s = os.system\n s('echo done')\nrun()") + def test_func_local_exec_alias_blocked(self): + # An exec builtin aliased inside a function must still be unwrapped and its + # recovered payload analyzed (exec-env aliasing walks the whole tree). + _blocked("def f():\n e = exec\n e(\"__import__('os').system('id')\")\nf()") + _blocked("def f():\n r = eval\n r(\"__import__('os').system('rm -rf /')\")\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 247558e837..625dad7056 100644 --- a/studio/backend/tests/test_sandbox_const_fold.py +++ b/studio/backend/tests/test_sandbox_const_fold.py @@ -71,6 +71,55 @@ class TestConstFoldArithAndConcat: assert _fold("bytes(10 ** 9)") is None +class TestConstFoldAllocationDoS: + """Oversized format widths / sequence repetitions must refuse BEFORE the folder + allocates the result (folding runs in the Studio process, ahead of subprocess + rlimits).""" + + def test_fstring_width_refused(self): + assert _fold("f'{1:1000000000}'") is None + + def test_str_format_width_refused(self): + assert _fold("'{:1000000000}'.format(1)") is None + + def test_percent_format_width_refused(self): + assert _fold("'%1000000000d' % 1") is None + + def test_pad_method_width_refused(self): + assert _fold("'x'.ljust(1000000000)") is None + assert _fold("'x'.rjust(10 ** 9)") is None + assert _fold("'x'.center(2000000000)") is None + assert _fold("'x'.zfill(10 ** 9)") is None + + def test_list_tuple_repeat_refused(self): + assert _fold("[0] * 1000000000") is None + assert _fold("(1,) * 10 ** 9") is None + + def test_benign_format_and_repeat_still_fold(self): + assert _fold("f'{2 + 2}'") == "4" + assert _fold("'{:>8}'.format('hi')") == " hi" + assert _fold("'%05d' % 7") == "00007" + assert _fold("'x'.ljust(10)") == "x " + assert _fold("[0] * 8") == [0] * 8 + + +class TestConstFoldPathJoin: + """os.path.join / posixpath.join of string literals fold so the sensitive-read + scanner sees the concrete path (628).""" + + def test_os_path_join_literal(self): + assert _fold("os.path.join('/etc', 'passwd')") == "/etc/passwd" + + def test_posixpath_join_literal(self): + assert _fold("posixpath.join('/etc', 'shadow')") == "/etc/shadow" + + def test_relative_join_literal(self): + assert _fold("os.path.join('sub', 'a.txt')") == "sub/a.txt" + + def test_join_nonliteral_unknown(self): + assert _fold("os.path.join('/etc', x)") is None + + class TestConstFoldJoinFormatFstring: def test_sep_join(self): assert _fold('".".join(["os", "system"])') == "os.system" diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 125af179ae..326998de1d 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -289,6 +289,86 @@ def test_inject_sandbox_guard_plain_prepend_without_future(): assert _inject_sandbox_guard(code, prelude) == prelude + code +@_POSIX_ONLY +def test_sandboxed_open_wrapped_attr_removed(tmp_path): + # functools.wraps would publish the ORIGINAL unguarded callable on __wrapped__; + # the guard must not expose it (open.__wrapped__(outside, 'w') would bypass). + target = tmp_path / "wrapped_escape.txt" + out = _python_exec( + f"open.__wrapped__({str(target)!r}, 'w').write('x'); print('WROTE')", + None, + 30, + "backstop-wrapped", + disable_sandbox = False, + ) + assert not target.exists() + assert "AttributeError" in out or "sandbox:" in out + + +@_POSIX_ONLY +def test_sandboxed_low_level_io_open_denied(tmp_path): + # io.open / builtins.open originate from the C module _io; patching io.open leaves + # _io.open untouched, so it must be guarded too. + target = tmp_path / "lowio_escape.txt" + out = _python_exec( + f"import _io; _io.open({str(target)!r}, 'w').write('x'); print('WROTE')", + None, + 30, + "backstop-lowio", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_chdir_escape_denied(): + # os.chdir outside the workdir would let a later relative read/write (which the + # static scan treats as local) escape, so cwd changes are confined. + out = _python_exec( + "import os\nos.chdir('/etc')\nprint('CWD', os.getcwd())", + None, + 30, + "backstop-chdir", + disable_sandbox = False, + ) + assert "sandbox:" in out and "chdir" in out + + +@_POSIX_ONLY +def test_sandboxed_chdir_within_workdir_allowed(): + out = _python_exec( + "import os\nos.chdir('.')\nprint('CWD-OK')", + None, + 30, + "backstop-chdir-ok", + disable_sandbox = False, + ) + assert "CWD-OK" in out + assert "sandbox:" not in out + + +@_POSIX_ONLY +def test_sandboxed_fd_metadata_mutator_denied(tmp_path): + # A read-only os.open of an outside file is allowed (reads are not confined), but + # fd-based metadata mutators (os.fchmod/fchown) must be denied so they cannot be + # reused to mutate host files. + victim = tmp_path / "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.fchmod(fd, 0o644); print('CHMODDED')", + None, + 30, + "backstop-fchmod", + disable_sandbox = False, + ) + assert "sandbox:" in out and "fchmod" in out + assert oct(os.stat(victim).st_mode & 0o777) == "0o600" + + @_POSIX_ONLY def test_sandboxed_imports_still_work_under_guard(): # The guard must not break library imports (bytecode caching failures are diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index 1260362624..a2d603afc4 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -874,6 +874,47 @@ class TestAliasIntrospectionBypasses: assert _check_code_safety(code) is not None, code +class TestReceiverAndVarsAndDynImportBypasses: + """Second-round bypasses: sensitive reach through a pathlib receiver, vars() on a + module, and dynamic import of a deserializer module.""" + + @pytest.mark.parametrize( + "code", + [ + # 572: sensitive path on the pathlib receiver, not in a call arg. + "from pathlib import Path\nPath('../../.ssh/id_rsa').read_text()", + "from pathlib import Path\nPath('/etc/passwd').read_bytes()", + "from pathlib import Path\nPath('/etc/passwd').open().read()", + # 617: vars(module) exposes the module __dict__. + "import os\nvars(os)['system']('rm -rf /')", + "vars(__builtins__)['eval']('x')", + # 596: dynamic import of a deserializer module runs a reduce payload. + "__import__('pickle').loads(blob)", + "__import__('marshal').loads(b)", + "import importlib\nimportlib.import_module('pickle').loads(b)", + # 628: literal os.path.join to a host secret. + "import os\nopen(os.path.join('/etc', 'passwd')).read()", + ], + ) + def test_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "from pathlib import Path\nPath('data/out.txt').read_text()", + "from pathlib import Path\nPath('model.json').open()", + "vars(obj)", + "vars()", + "import pickle\npickle.dumps(x)", + "import importlib\nimportlib.import_module('numpy')", + "import os\nopen(os.path.join('sub', 'a.txt'))", + ], + ) + def test_benign_allowed(self, code): + assert _check_code_safety(code) is None, code + + class TestEvalExecRecursion: """Stage 2: eval/exec/compile are unwrapped, not blanket-banned. A safe (constant-recoverable) payload is allowed; an obfuscated escape blocks.""" @@ -946,27 +987,40 @@ class TestEvalExecRecursion: def test_import_concat_benign_module_allowed(self): _ok('__import__("hugging" + "face_hub")') - def test_exec_utf7_bytes_coding_cookie_blocked(self): - # exec()/eval()/compile() honor PEP 263 coding cookies on *bytes*: a UTF-7 - # payload behind "# coding: utf-7" decodes to real Python that the UTF-8 - # static view (which is SYNTAX_BAD) never sees. An unparseable *bytes* - # payload for an executing sink must block. - payload = ( - b"# coding: utf-7\n" - b"+AGkAbQBwAG8AcgB0ACAAbwBz-\n" - b"+AG8AcwAuAHMAeQBzAHQAZQBtACgAJwBpAGQAJwAp-" - ) - # self-check: the cookie-decoded payload really is os.system ACE. - assert "os.system" in payload.decode("utf-7") + def test_exec_utf7_comment_cookie_smuggle_blocked(self): + # The exec/eval/compile sinks honor a PEP 263 coding cookie on *bytes*. Here + # the UTF-8 view is TWO comment lines (safe), but "+AAo-" decodes (UTF-7) to a + # newline, so exec(bytes) actually runs the hidden __import__('os') call. The + # analyzer must decode with the cookie's codec, not read the UTF-8 view. + sneaky = b"# coding: utf_7\n#+AAo-__import__('os').system('id')\n" + # self-check: UTF-8 view is pure comments; the cookie decode reveals the call. + import ast as _ast + + _ast.parse(sneaky.decode("utf-8")) # parses (comments only) under UTF-8 + assert "__import__('os')" in sneaky.decode("utf-7") + for sink in ("exec(%r)", "exec(compile(%r, '', 'exec'))"): + assert _check_code_safety(sink % sneaky) is not None, sink + + def test_exec_utf7_bytes_decodes_to_blocked_op(self): + # A bytes payload behind a coding cookie whose decoded source reaches a blocked + # operation must block for every executing sink (eval sees a statement -> the + # SYNTAX_BAD-bytes backstop still trips). + payload = b"# coding: utf-7\n" + "import os\nos.system('rm -rf /')\n".encode("utf-7") + assert "rm -rf" in payload.decode("utf-7") for sink in ("exec(%r)", "eval(%r)", "exec(compile(%r, '', 'exec'))"): - code = sink % payload - assert _check_code_safety(code) is not None, code - # bare compile() does not run, so it stays allowed (the exec of its result - # is where the block lands). - assert _check_code_safety("compile(%r, '', 'exec')" % payload) is None + assert _check_code_safety(sink % payload) is not None, sink def test_exec_plain_bytes_payload_allowed(self): # Legitimate exec/eval of ASCII/UTF-8 bytes that parse cleanly stay allowed. _ok('exec(b"x = 1")') _ok('exec(b"print(1)")') _ok('eval(b"2 + 2")') + # A UTF-7 payload that decodes to a benign, non-blocked call stays allowed too + # (os.system('id') is benign -- 'id' is not a blocked command), matching the + # plain-text exec("import os; os.system('id')") behavior. + benign = ( + b"# coding: utf-7\n" + b"+AGkAbQBwAG8AcgB0ACAAbwBz-\n" + b"+AG8AcwAuAHMAeQBzAHQAZQBtACgAJwBpAGQAJwAp-" + ) + _ok("exec(%r)" % benign)