diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 36dba2aee..0259efd7a 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1705,6 +1705,21 @@ def _format_template_ok(template): return True +def _format_has_nested_spec(template): + """A replacement field whose spec itself contains a field ({:{}}): the width is + supplied by an argument, so a large numeric arg drives the allocation.""" + if not isinstance(template, str): + return False + try: + import string as _string + for _lit, _field, _spec, _conv in _string.Formatter().parse(template): + if _spec and "{" in _spec: + return True + except Exception: + return False + return False + + _PRINTF_WIDTH_RE = re.compile(r"%[-+ #0]*(\d+)?(?:\.(\d+))?[hlL]?[diouxXeEfFgGcrsab%]") @@ -2077,8 +2092,15 @@ 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 == "format" and not _format_template_ok(recv): - return None + if attr == "format": + if not _format_template_ok(recv): + return None + # Nested-width field ({:{}}) whose width comes from a large numeric + # arg: refuse before format() allocates. + if _format_has_nested_spec(recv) and any( + _too_wide(a) for a in list(call_args) + list(kwargs.values()) + ): + return None return _fold_cap(getattr(recv, attr)(*call_args, **kwargs)) except Exception: return None @@ -2413,30 +2435,68 @@ def _compile_mode(node, const_env): return "exec" +def _walk_scope_local(scope): + """Yield descendants of ``scope``'s body that share its namespace, WITHOUT + descending into nested def / lambda / class / comprehension (each of which is a + new scope). Used so single-assignment alias detection is scope-correct.""" + stack = list(getattr(scope, "body", [])) + _NESTED = ( + ast.FunctionDef, + ast.AsyncFunctionDef, + ast.Lambda, + ast.ClassDef, + ast.ListComp, + ast.SetComp, + ast.DictComp, + ast.GeneratorExp, + ) + while stack: + n = stack.pop() + yield n + for child in ast.iter_child_nodes(n): + if not isinstance(child, _NESTED): + 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] + for n in ast.walk(tree): + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)): + scopes.append(n) + for scope in scopes: + counts: dict[str, int] = {} + rebound: set[str] = set() + assigns: list[tuple[str, ast.expr]] = [] + 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 + elif isinstance(n, (ast.Global, ast.Nonlocal)): + rebound.update(n.names) + elif ( + isinstance(n, ast.Assign) + and len(n.targets) == 1 + and isinstance(n.targets[0], ast.Name) + ): + assigns.append((n.targets[0].id, n.value)) + 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.""" - store_counts: dict[str, int] = {} - for n in ast.walk(tree): - if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store): - store_counts[n.id] = store_counts.get(n.id, 0) + 1 - exec_aliases: dict[str, str] = {} compiled_env: dict[str, tuple] = {} - # 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 - and isinstance(stmt.targets[0], ast.Name) - ): - continue - name = stmt.targets[0].id - if store_counts.get(name, 0) != 1: - continue - rhs = stmt.value + # 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 ( @@ -2724,11 +2784,8 @@ def _build_shell_sink_aliases(tree): os_aliases = {"os"} subprocess_aliases = {"subprocess"} from_aliases: dict[str, str] = {} - store_counts: dict[str, int] = {} for n in ast.walk(tree): - if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store): - store_counts[n.id] = store_counts.get(n.id, 0) + 1 - elif isinstance(n, ast.Import): + if isinstance(n, ast.Import): for a in n.names: if a.name == "os": os_aliases.add(a.asname or "os") @@ -2741,21 +2798,11 @@ def _build_shell_sink_aliases(tree): from_aliases[a.asname or a.name] = fq aliases: dict[str, str] = {} - # Walk every Assign, not just the module body: a single-assignment alias created - # inside a function -- `def f(): s = os.system; s('rm -rf /')` -- is the common - # wrapper shape. store_counts is computed over the whole tree, so the "stored - # exactly once" guard still excludes any ambiguous re-binding (keeps it low-FP). - for stmt in ast.walk(tree): - if not ( - isinstance(stmt, ast.Assign) - and len(stmt.targets) == 1 - and isinstance(stmt.targets[0], ast.Name) - ): - continue - name = stmt.targets[0].id - if store_counts.get(name, 0) != 1: - continue # ambiguous reassignment -> do not alias (avoids FPs) - fq = _resolve_static_shell_sink(stmt.value, os_aliases, subprocess_aliases, from_aliases) + # 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 @@ -3335,6 +3382,12 @@ def _check_signal_escape_patterns( and func.attr in ("import_module", "reload", "__import__") and _ast_name_matches(func.value, self.importlib_aliases) ) + or ( + # builtins.__import__('os') / __builtins__.__import__(...) + isinstance(func, ast.Attribute) + and func.attr == "__import__" + and _ast_name_matches(func.value, self.builtins_aliases) + ) ) # Deserialization sinks reconstruct arbitrary objects/code from bytes. # Resolve aliased imports (from pickle import loads as l), module aliases @@ -4145,16 +4198,29 @@ def _check_signal_escape_patterns( ) 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 + # Path(...) or pathlib.Path(...) (module-qualified). Join ALL string args so a + # multi-component constructor -- Path('/etc', 'passwd') -- resolves to the full + # path rather than only its first (non-sensitive) component. + if not isinstance(recv, ast.Call) or not recv.args: + return None + rf = recv.func + ctor = (isinstance(rf, ast.Name) and rf.id in _PATHLIB_CTORS) or ( + isinstance(rf, ast.Attribute) and rf.attr in _PATHLIB_CTORS + ) + if not ctor: + return None + parts = [] + for a in recv.args: + v = _const_fold(a, _const_env) + if not isinstance(v, (str, bytes, bytearray)): + return None + parts.append(_to_text(v)) + if not parts: + return None + try: + return os.path.join(*parts) + except Exception: + return None def _flag_read_path(node, s, is_read_callee): norm = s.replace("\\", "/") @@ -4360,13 +4426,29 @@ def _gwraps(real): return w return _deco +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). + if isinstance(p, int): + return p + try: + return _os.fspath(p) + except Exception: + return p + +def _mode_is_write(mode): + # Coerce through the *base* str: a str-subclass __contains__/__str__ must not be + # able to lie about whether the mode requests a write. + m = str.__str__(mode) if isinstance(mode, str) else "r" + return any(c in m for c in "wax+") + def _guard_open_like(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): - _deny(file, "write") - return real(file, mode, *a, **k) + f = _fspath1(file) + if _mode_is_write(mode) and not _within(f): + _deny(f, "write") + return real(f, mode, *a, **k) return w _bi.open = _guard_open_like(_bi.open) @@ -4374,21 +4456,28 @@ _bi.open = _guard_open_like(_bi.open) # Low-level os.open: builtins.open does not route through it, so it needs its own # guard. Any mutating open flag confines the target; a mutating dir_fd call fails # closed (a string realpath against cwd is wrong for an fd-relative path). -_real_osopen = _os.open _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) ) -@_gwraps(_real_osopen) -def _guarded_osopen(path, flags, *a, **k): - mutating = (not isinstance(flags, int)) or bool(flags & _WRITE_OFLAGS) - if mutating: - if k.get("dir_fd") is not None: - _deny(path, "os.open (dir_fd)") - if not _within(path): - _deny(path, "os.open write") - return _real_osopen(path, flags, *a, **k) -_os.open = _guarded_osopen +def _make_osopen_guard(real_open): + @_gwraps(real_open) + def _guarded(path, flags, *a, **k): + try: + fi = int.__index__(flags) # base int: an int-subclass __and__ must not lie + except Exception: + fi = None + mutating = (fi is None) or bool(fi & _WRITE_OFLAGS) + if mutating: + if k.get("dir_fd") is not None: + _deny(path, "os.open (dir_fd)") + p = _fspath1(path) + if not _within(p): + _deny(p, "os.open write") + return real_open(p, flags, *a, **k) + return real_open(path, flags, *a, **k) + return _guarded +_os.open = _make_osopen_guard(_os.open) def _wrap1(mod, name, what): orig = getattr(mod, name, None) @@ -4398,14 +4487,20 @@ 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 not _within(path): - _deny(path, what) - return orig(path, *a, **k) + p = _fspath1(path) + if not _within(p): + _deny(p, what) + return orig(p if not isinstance(path, int) else path, *a, **k) setattr(mod, name, w) -# lchmod/lchown/chflags/mknod are platform-specific; _wrap1 no-ops when absent. -for _n in ("remove", "unlink", "rmdir", "removedirs", "truncate", "chmod", - "chown", "mkdir", "makedirs", "mknod", "lchmod", "lchown", "chflags"): +# Path-first single-arg mutators. mkfifo/utime/setxattr/removexattr create or mutate +# host files/metadata; the platform-specific ones no-op via _wrap1 when absent. +_OS_MUTATORS1 = ( + "remove", "unlink", "rmdir", "removedirs", "truncate", "chmod", "chown", + "mkdir", "makedirs", "mknod", "mkfifo", "utime", "setxattr", "removexattr", + "lchmod", "lchown", "chflags", "lchflags", +) +for _n in _OS_MUTATORS1: _wrap1(_os, _n, _n) def _wrap2(mod, name, both): @@ -4416,16 +4511,36 @@ def _wrap2(mod, name, both): 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 - if both and not _within(src): - _deny(src, name + " source") - if not _within(dst): - _deny(dst, name + " destination") - return orig(src, dst, *a, **k) + s, d = _fspath1(src), _fspath1(dst) + if both and not _within(s): + _deny(s, name + " source") + if not _within(d): + _deny(d, name + " destination") + return orig(s, d, *a, **k) setattr(mod, name, w) for _n in ("rename", "renames", "replace", "link", "symlink"): _wrap2(_os, _n, True) +# posix (POSIX) / nt (Windows) is the low-level C module os re-exports from; patching +# os.* leaves posix.open / posix.rename / ... importable with the originals, so apply +# the same guards to that module too. +for _lowosname in ("posix", "nt"): + try: + _lowos = __import__(_lowosname) + except Exception: + _lowos = None + if _lowos is not None: + try: + if hasattr(_lowos, "open"): + _lowos.open = _make_osopen_guard(_lowos.open) + for _n in _OS_MUTATORS1: + _wrap1(_lowos, _n, _lowosname + "." + _n) + for _n in ("rename", "renames", "replace", "link", "symlink"): + _wrap2(_lowos, _n, True) + except Exception: + pass + # io.open is a separate reference from the (now patched) builtins.open -- guard # direct io.open() writers (e.g. zipfile-based) the same way. (Path.open is handled # explicitly below, not via this patch.) @@ -4441,7 +4556,26 @@ try: import _io as _lowio _lowio.open = _guard_open_like(_lowio.open) except Exception: - pass + _lowio = None + +# io.FileIO / _io.FileIO is a C constructor that opens a file WITHOUT routing through +# open(), so `io.FileIO('/tmp/escape', 'w')` bypasses _guard_open_like. Subclass it to +# confine mutating modes (subclassing keeps guard-built objects real FileIO instances). +def _guard_fileio(_realcls): + class _GuardedFileIO(_realcls): + def __init__(self, name, mode="r", *a, **k): + f = _fspath1(name) + if _mode_is_write(mode) and not _within(f): + _deny(f, "FileIO write") + super().__init__(name, mode, *a, **k) + return _GuardedFileIO + +for _iomod in (_io, _lowio): + try: + if _iomod is not None and hasattr(_iomod, "FileIO"): + _iomod.FileIO = _guard_fileio(_iomod.FileIO) + 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. diff --git a/studio/backend/tests/test_sandbox_aliasing.py b/studio/backend/tests/test_sandbox_aliasing.py index 3fccfd575..d0f116b09 100644 --- a/studio/backend/tests/test_sandbox_aliasing.py +++ b/studio/backend/tests/test_sandbox_aliasing.py @@ -68,6 +68,31 @@ class TestFuncLocalAliasBlocked: _blocked("def f():\n r = eval\n r(\"__import__('os').system('rm -rf /')\")\nf()") +class TestPerScopeAliasCounting: + """Alias single-assignment is counted PER function scope: two functions binding + the same local name must not cancel each other out (a tree-wide count would treat + both as ambiguous and miss a real sink).""" + + def test_two_functions_same_shell_alias_name_blocked(self): + _blocked( + "import os\n" + "def a():\n s = os.system\n s('rm -rf /')\n" + "def b():\n s = print\na()" + ) + + def test_two_functions_same_exec_alias_name_blocked(self): + _blocked( + "def a():\n e = exec\n e(\"__import__('os').system('id')\")\n" + "def b():\n e = print\na()" + ) + + def test_two_functions_benign_aliases_allowed(self): + _ok( + "def a():\n s = sorted\n return s([3, 1])\n" + "def b():\n s = max\n return s([1, 2])\na()" + ) + + class TestAliasingLowFalsePositive: def test_reassigned_alias_not_treated_as_sink(self): # s is stored twice -> ambiguous -> NOT aliased. The literal arg is benign diff --git a/studio/backend/tests/test_sandbox_const_fold.py b/studio/backend/tests/test_sandbox_const_fold.py index 625dad705..c383cc28c 100644 --- a/studio/backend/tests/test_sandbox_const_fold.py +++ b/studio/backend/tests/test_sandbox_const_fold.py @@ -85,6 +85,15 @@ class TestConstFoldAllocationDoS: def test_percent_format_width_refused(self): assert _fold("'%1000000000d' % 1") is None + def test_nested_format_width_refused(self): + # '{:{}}'.format('x', 10**9): the width is supplied by a large arg, so the + # template check must refuse before format() allocates. + assert _fold("'{:{}}'.format('x', 1000000000)") is None + assert _fold("'{:>{w}}'.format('x', w = 10 ** 9)") is None + + def test_nested_format_small_width_folds(self): + assert _fold("'{:>{}}'.format('x', 5)") == " x" + 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 326998de1..0c745761a 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -369,6 +369,142 @@ def test_sandboxed_fd_metadata_mutator_denied(tmp_path): assert oct(os.stat(victim).st_mode & 0o777) == "0o600" +@_POSIX_ONLY +def test_sandboxed_posix_module_open_denied(tmp_path): + # os re-exports from the C module posix; posix.open must be guarded too. + target = tmp_path / "posix_escape.txt" + out = _python_exec( + "import posix, os\n" + f"fd = posix.open({str(target)!r}, os.O_CREAT | os.O_WRONLY, 0o600)\n" + "posix.write(fd, b'x')\nprint('DONE-OK')", + None, + 30, + "backstop-posix", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_extra_os_mutators_denied(tmp_path): + # os.mkfifo creates a host node; os.utime mutates host metadata. + fifo = tmp_path / "escape.fifo" + out = _python_exec( + f"import os; os.mkfifo({str(fifo)!r}); print('DONE-OK')", + None, + 30, + "backstop-mkfifo", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert not fifo.exists() + + victim = tmp_path / "utime_victim.txt" + victim.write_text("x") + before = victim.stat().st_mtime + out = _python_exec( + f"import os; os.utime({str(victim)!r}, (0, 0)); print('DONE-OK')", + None, + 30, + "backstop-utime", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert victim.stat().st_mtime == before + + +@_POSIX_ONLY +def test_sandboxed_io_fileio_write_denied(tmp_path): + # io.FileIO / _io.FileIO is a C constructor that opens a file without routing + # through open(), so it needs its own guard. + target = tmp_path / "fileio_escape.txt" + out = _python_exec( + f"import io; io.FileIO({str(target)!r}, 'w').write(b'x'); print('DONE-OK')", + None, + 30, + "backstop-fileio", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_stateful_fspath_toctou_denied(tmp_path): + # A stateful __fspath__ returning a local path for the check and an outside path + # for the real open must not escape: the guard materializes fspath once. + target = tmp_path / "fspath_escape.txt" + out = _python_exec( + "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" + "open(P(), 'w').write('x')\nprint('DONE')", + None, + 30, + "backstop-fspath", + disable_sandbox = False, + ) + # The escape target is never written (the single fspath call yields the local path). + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_str_subclass_mode_denied(tmp_path): + # A str-subclass mode whose __contains__ lies must not defeat the write check. + target = tmp_path / "mode_escape.txt" + out = _python_exec( + "class M(str):\n" + " def __contains__(self, c):\n" + " return False\n" + f"open({str(target)!r}, M('w')).write('x')\nprint('DONE-OK')", + None, + 30, + "backstop-mode", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_int_subclass_flags_denied(tmp_path): + # An int-subclass flags whose __and__ lies must not defeat the os.open write check. + target = tmp_path / "flags_escape.txt" + out = _python_exec( + "import os\n" + "class F(int):\n" + " def __and__(self, o):\n" + " return 0\n" + f"os.open({str(target)!r}, F(os.O_CREAT | os.O_WRONLY), 0o600)\nprint('DONE-OK')", + None, + 30, + "backstop-flags", + disable_sandbox = False, + ) + assert "sandbox:" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_in_workdir_ops_still_work(): + # The added guards must not break benign in-workdir writes. + out = _python_exec( + "import io, os\n" + "io.FileIO('fio.txt', 'w').write(b'z')\n" + "os.makedirs('subd', exist_ok = True)\n" + "print(io.FileIO('fio.txt', 'r').read())", + None, + 30, + "backstop-inworkdir", + disable_sandbox = False, + ) + assert "z" in out + assert "sandbox:" not in out + + @_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 a2d603afc..5ec770054 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -219,7 +219,7 @@ class TestUploadDenylist: ) def test_plain_post_json_not_blocked(self): - _ok("import requests\n" 'requests.post("https://api.weather.gov/lookup", json={"k": "v"})') + _ok('import requests\nrequests.post("https://api.weather.gov/lookup", json={"k": "v"})') class TestSandboxEnvIsolation: @@ -514,15 +514,11 @@ class TestHfUploadImportGate: def test_hf_bare_name_upload_folder_safe_allowed(self): _ok( - "from huggingface_hub import upload_folder;" - " upload_folder(folder_path='x', repo_id='r')" + "from huggingface_hub import upload_folder; upload_folder(folder_path='x', repo_id='r')" ) def test_hf_bare_name_create_commit_safe_allowed(self): - _ok( - "from huggingface_hub import create_commit;" - " create_commit(operations=[], repo_id='r')" - ) + _ok("from huggingface_hub import create_commit; create_commit(operations=[], repo_id='r')") def test_bare_name_upload_file_without_hf_import_allowed(self): # No HF import -- local helper named upload_file passes. @@ -894,6 +890,12 @@ class TestReceiverAndVarsAndDynImportBypasses: "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()", + # 602: multi-component / module-qualified pathlib receiver read. + "from pathlib import Path\nPath('/etc', 'passwd').read_text()", + "import pathlib\npathlib.Path('/etc', 'passwd').open().read()", + # 605: __import__ reached through the builtins module. + "import builtins\nbuiltins.__import__('os').system('rm -rf /')", + "__builtins__.__import__('subprocess').run(['id'])", ], ) def test_blocked(self, code): @@ -904,11 +906,15 @@ class TestReceiverAndVarsAndDynImportBypasses: [ "from pathlib import Path\nPath('data/out.txt').read_text()", "from pathlib import Path\nPath('model.json').open()", + # 602: in-workdir multi-component pathlib read stays allowed. + "from pathlib import Path\nPath('data', 'out.txt').read_text()", "vars(obj)", "vars()", "import pickle\npickle.dumps(x)", "import importlib\nimportlib.import_module('numpy')", "import os\nopen(os.path.join('sub', 'a.txt'))", + # 605: benign builtins attribute access stays allowed. + "import builtins\nx = builtins.len([1, 2, 3])", ], ) def test_benign_allowed(self, code):