diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index a1213b193a..b087cb46c7 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -251,6 +251,65 @@ _ABSOLUTE_SENSITIVE_RE = re.compile( re.IGNORECASE, ) +# Whole-directory variants of the credential roots above. Only used by +# the shutil / file-copy gate -- ``ls ~/.ssh`` and ``find ~/.aws -type f`` +# are legitimate, but ``shutil.copytree('~/.ssh', dst)`` and +# ``cp -r ~/.aws /tmp/out`` exfil every file in those dirs in one call. +# +# The end anchor matches the path AS the directory (``~/.ssh`` or +# ``~/.ssh/``) and not a file inside it (``~/.ssh/known_hosts`` — +# the per-file allow-list already governs whether that single read +# is OK). It also rejects similar-name prefixes (``~/.ssh_backup``). +_DIR_END = r"(?=/?$|/?[\s'\";&|)<>])" +_HOME_RELATIVE_SENSITIVE_DIRS = ( + rf"\.ssh{_DIR_END}", + rf"\.aws{_DIR_END}", + rf"\.config/gcloud{_DIR_END}", + rf"\.gnupg{_DIR_END}", + rf"\.docker{_DIR_END}", + rf"\.kube{_DIR_END}", + rf"\.password-store{_DIR_END}", +) +_ABSOLUTE_SENSITIVE_DIRS = ( + rf"/etc{_DIR_END}", + rf"/etc/ssh{_DIR_END}", + rf"/var/spool/cron{_DIR_END}", + # Same Linux process-state roots as the per-file regex — copying + # ``/proc/self/`` or ``/proc//`` recursively drags the entire + # process state (environ, mem, maps, cmdline) out. + rf"/proc/(?:self|thread-self|\d+){_DIR_END}", +) +_HOME_SENSITIVE_DIR_RE = re.compile( + _PATH_TOKEN_START + + _HOME_PREFIX_RE + + r"(?:" + + "|".join(_HOME_RELATIVE_SENSITIVE_DIRS) + + r")", + re.IGNORECASE, +) +_ABSOLUTE_SENSITIVE_DIR_RE = re.compile( + _PATH_TOKEN_START + r"(?:" + "|".join(_ABSOLUTE_SENSITIVE_DIRS) + r")", + re.IGNORECASE, +) + + +def _matches_sensitive_dir(path: str) -> bool: + """Return True if *path* names a sensitive credential / key directory + (rather than a single file). Used by the shutil-copy gate so + ``shutil.copytree('~/.ssh', dst)`` and ``shutil.copy('~/.aws', dst)`` + are caught even though ``~/.ssh`` itself isn't a single sensitive + file in ``_HOME_RELATIVE_SENSITIVE``.""" + if not path: + return False + for cand in {path, path.replace("\\", "/")}: + norm = _normalize_path_separators(cand) + for projection in {cand, norm}: + if _HOME_SENSITIVE_DIR_RE.search(projection): + return True + if _ABSOLUTE_SENSITIVE_DIR_RE.search(projection): + return True + return False + # Sensitive root prefix immediately followed by a shell substitution # (``$(...)`` or backticks). Catches dynamic-path constructions like # ``cat /etc/$(printf shadow)`` or ``cat /proc/1/$(echo environ)`` that @@ -1211,9 +1270,21 @@ def _check_signal_escape_patterns(code: str): if isinstance(node, ast.Constant): if isinstance(node.value, str): return node.value + if isinstance(node.value, bytes): + # ``open(b'/etc/shadow')`` — bytes are valid path-like + # objects to ``open()`` so the literal must reach the + # sensitive-path gate too. Strict UTF-8 to avoid + # masking junk. + try: + return node.value.decode("utf-8") + except UnicodeDecodeError: + return None if isinstance(node.value, (int, float)): return str(node.value) return None + if isinstance(node, ast.NamedExpr): + # Walrus (``open((p := '/etc/shadow'))``): resolve the RHS. + return _extract_string_literal(node.value, _depth + 1) if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): left = _extract_string_literal(node.left, _depth + 1) right = _extract_string_literal(node.right, _depth + 1) @@ -1265,11 +1336,26 @@ def _check_signal_escape_patterns(code: str): if isinstance(node, ast.Constant): if isinstance(node.value, str): return node.value + if isinstance(node.value, bytes): + # ``open(b'/etc/shadow')`` -- bytes paths are valid + # PathLike for ``open()``. Decode strictly so non-UTF-8 + # junk does not mask the gate. + try: + return node.value.decode("utf-8") + except UnicodeDecodeError: + return None if isinstance(node.value, (int, float)): return str(node.value) return None if isinstance(node, ast.Name): return string_bindings.get(node.id) + if isinstance(node, ast.NamedExpr): + # Walrus ``(p := '/etc/shadow')``: resolve and record the + # binding so later uses of ``p`` also resolve. + val = _extract_string_from_node(node.value, _depth + 1) + if val is not None and isinstance(node.target, ast.Name): + string_bindings.setdefault(node.target.id, val) + return val if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): left = _extract_string_from_node(node.left, _depth + 1) right = _extract_string_from_node(node.right, _depth + 1) @@ -1327,10 +1413,15 @@ def _check_signal_escape_patterns(code: str): # Pre-pass: collect simple ``name = 'literal'`` string assignments # and ``name = eval`` / ``name = exec`` function aliases so the # visitors and ``_extract_string_from_node`` can resolve later uses. + # Also handles tuple / list unpacking (``a, b = '/etc', 'shadow'; + # open(a + '/' + b)`` and ``p, = ['/etc/shadow']; open(p)``) so that + # statically resolvable destructuring isn't a free bypass channel. # Walks the AST in one pass; first assignment wins (mirrors actual # execution order well enough for the static gate). for _assign in ast.walk(tree): - if isinstance(_assign, ast.Assign) and len(_assign.targets) == 1: + if not isinstance(_assign, ast.Assign): + continue + if len(_assign.targets) == 1: _target = _assign.targets[0] if isinstance(_target, ast.Name) and _target.id not in string_bindings: _val = _extract_string_from_node(_assign.value) @@ -1341,6 +1432,19 @@ def _check_signal_escape_patterns(code: str): "exec", ): eval_exec_aliases[_target.id] = _assign.value.id + elif isinstance(_target, (ast.Tuple, ast.List)) and isinstance( + _assign.value, (ast.Tuple, ast.List) + ): + # ``(a, b) = ('/etc', 'shadow')`` / ``p, = ['/etc/shadow']``. + if len(_target.elts) == len(_assign.value.elts): + for _tgt_e, _val_e in zip(_target.elts, _assign.value.elts): + if ( + isinstance(_tgt_e, ast.Name) + and _tgt_e.id not in string_bindings + ): + _v = _extract_string_from_node(_val_e) + if _v is not None: + string_bindings[_tgt_e.id] = _v def _extract_strings_from_list(node): """Extract string elements from an AST List or Tuple node.""" @@ -1577,6 +1681,11 @@ def _check_signal_escape_patterns(code: str): # and ``import builtins as b; b.exec(...)`` flow through the # same recursion guard as the bare-name forms. self.builtins_aliases = {"builtins", "__builtins__"} + # Names that resolve to ``importlib.import_module`` so + # ``from importlib import import_module as IM; IM('os')...`` + # flows through ``_resolve_dynamic_module`` the same as + # ``import importlib; importlib.import_module('os')...``. + self.import_module_aliases = {"import_module"} self.loop_depth = 0 # Cap recursion into nested eval/exec literals; an adversarial # ``eval("eval('eval(...)')")`` should not blow the stack. @@ -1629,6 +1738,13 @@ def _check_signal_escape_patterns(code: str): for alias in node.names: if alias.name in ("eval", "exec"): eval_exec_aliases[alias.asname or alias.name] = alias.name + elif node.module == "importlib": + # ``from importlib import import_module as IM`` so a + # later ``IM('os').system(...)`` flows through the same + # dynamic-import gate as ``importlib.import_module('os')``. + for alias in node.names: + if alias.name == "import_module": + self.import_module_aliases.add(alias.asname or alias.name) self.generic_visit(node) def visit_While(self, node): @@ -1646,7 +1762,7 @@ def _check_signal_escape_patterns(code: str): # ``m = importlib.import_module('os')`` so a subsequent # ``m.system(...)`` / ``m.popen(...)`` flows through the # os/subprocess alias detection unchanged. - dyn = _resolve_dynamic_module_name(node.value) + dyn = self._resolve_dynamic_module(node.value) if dyn == "os": for tgt in node.targets: if isinstance(tgt, ast.Name): @@ -1655,8 +1771,62 @@ def _check_signal_escape_patterns(code: str): for tgt in node.targets: if isinstance(tgt, ast.Name): self.subprocess_aliases.add(tgt.id) + + # Bare module rebinding (``m = os`` / ``r = subprocess``): + # propagate the source alias set so a later ``m.system(...)`` + # is caught by the same os/subprocess gate as the direct call. + if isinstance(node.value, ast.Name): + src = node.value.id + if src in self.os_aliases: + for tgt in node.targets: + if isinstance(tgt, ast.Name): + self.os_aliases.add(tgt.id) + elif src in self.subprocess_aliases: + for tgt in node.targets: + if isinstance(tgt, ast.Name): + self.subprocess_aliases.add(tgt.id) + + # Method rebinding (``p = os.popen`` / ``r = subprocess.run``): + # the bound name now points at a shell-exec function so a + # later ``p('sudo whoami')`` must flow through the + # shell-escape gate. Track it under ``shell_exec_aliases`` + # alongside the existing from-import path. + elif ( + isinstance(node.value, ast.Attribute) + and isinstance(node.value.value, ast.Name) + ): + recv = node.value.value.id + attr = node.value.attr + fq = None + if recv in self.os_aliases: + fq = f"os.{attr}" + elif recv in self.subprocess_aliases: + fq = f"subprocess.{attr}" + if fq and fq in _SHELL_EXEC_FUNCS: + for tgt in node.targets: + if isinstance(tgt, ast.Name): + self.shell_exec_aliases[tgt.id] = fq + self.generic_visit(node) + def _resolve_dynamic_module(self, node): + """Visitor-aware dynamic-import detection: recognises + everything :func:`_resolve_dynamic_module_name` does plus + tracked ``from importlib import import_module as IM`` + aliases stored on ``self.import_module_aliases``.""" + mod = _resolve_dynamic_module_name(node) + if mod is not None: + return mod + if isinstance(node, ast.Call) and node.args: + arg0 = node.args[0] + if isinstance(arg0, ast.Constant) and isinstance(arg0.value, str): + if ( + isinstance(node.func, ast.Name) + and node.func.id in self.import_module_aliases + ): + return arg0.value + return None + def visit_Call(self, node): func = node.func @@ -1773,9 +1943,10 @@ def _check_signal_escape_patterns(code: str): # Inline dynamic import: # __import__('os').system(...) # importlib.import_module('os').popen(...) + # IM('os').system(...) (IM is a from-import alias) # No intermediate name binding so the Name branch # above misses it; resolve the receiver here. - dyn = _resolve_dynamic_module_name(func.value) + dyn = self._resolve_dynamic_module(func.value) if dyn == "os": shell_func = f"os.{func.attr}" elif dyn == "subprocess": @@ -2396,6 +2567,10 @@ def _check_signal_escape_patterns(code: str): self.builtins_aliases = {"builtins", "__builtins__"} self.path_aliases = set(_PATHLIB_PATH_CLASSES) self.pathlib_aliases = {"pathlib"} + # ``from io import FileIO as X`` and ``from codecs import open + # as X``: a later bare ``X('/etc/shadow')`` flows through the + # same file-read gate as the qualified call. + self.file_reader_aliases: set[str] = set() def visit_Import(self, node): for alias in node.names: @@ -2414,6 +2589,17 @@ def _check_signal_escape_patterns(code: str): for alias in node.names: if alias.name in ("eval", "exec"): eval_exec_aliases[alias.asname or alias.name] = alias.name + elif node.module in ("io", "codecs"): + # ``from io import FileIO`` / ``from codecs import open`` + # bind a bare name that is otherwise indistinguishable + # from any other ``FileIO(...)`` / ``open(...)`` call. + # The reader's gate uses this set to recognise the + # alias as a file-read. + for alias in node.names: + if ( + node.module == "io" and alias.name in ("FileIO", "open") + ) or (node.module == "codecs" and alias.name == "open"): + self.file_reader_aliases.add(alias.asname or alias.name) self.generic_visit(node) def visit_Call(self, node): @@ -2639,10 +2825,51 @@ def _check_signal_escape_patterns(code: str): ): receiver_read_method = node.func.attr + # ``io.FileIO`` and ``codecs.open`` are the two stdlib + # file-reader call shapes that don't end in ``.open`` / + # ``open()`` but still read an arbitrary path. Treat them + # as the same gate so ``io.FileIO('/etc/shadow').read()`` is + # blocked alongside ``open('/etc/shadow')``. + _EXPLICIT_FILE_READERS = ("io.FileIO", "codecs.open") + # Third-party file-reader method names that any reasonable + # ``pandas``/``numpy`` alias exposes (``pd.read_csv`` / + # ``pandas.read_csv`` / ``np.fromfile`` / ``numpy.loadtxt``). + # Matched by suffix so the receiver alias does not need to + # be tracked separately. + _DATAFRAME_READERS = ( + ".read_csv", + ".read_table", + ".read_excel", + ".read_json", + ".read_parquet", + ".read_pickle", + ".read_feather", + ".read_orc", + ".read_hdf", + ".read_sas", + ".read_stata", + ".read_xml", + ".read_fwf", + ".read_sql", + ".fromfile", + ".loadtxt", + ".genfromtxt", + ) + looks_like_dataframe_reader = isinstance( + node.func, ast.Attribute + ) and any(fq.endswith(s) for s in _DATAFRAME_READERS) is_open_call = ( - (isinstance(node.func, ast.Name) and node.func.id == "open") + ( + isinstance(node.func, ast.Name) + and ( + node.func.id == "open" + or node.func.id in self.file_reader_aliases + ) + ) or fq in ("io.open", "pathlib.Path.open") + or fq in _EXPLICIT_FILE_READERS or fq.endswith(".open") + or looks_like_dataframe_reader or receiver_read_method is not None ) if is_open_call: @@ -2771,6 +2998,15 @@ def _check_signal_escape_patterns(code: str): if _find_sensitive_paths(cand): flagged = True break + # Whole-directory exfil: shutil.copytree('~/.ssh', + # dst) drags every key out in one call. Reusing + # `_find_sensitive_paths` would miss it because + # `~/.ssh` (no filename) isn't in the per-file + # list. The dir matcher is shutil-specific so + # `ls ~/.ssh` (legit) stays allowed. + if _matches_sensitive_dir(cand): + flagged = True + break if flagged: sensitive_file_reads.append( { diff --git a/studio/backend/tests/test_sandbox_hardening.py b/studio/backend/tests/test_sandbox_hardening.py index 67db06bef3..0172507c5b 100644 --- a/studio/backend/tests/test_sandbox_hardening.py +++ b/studio/backend/tests/test_sandbox_hardening.py @@ -1409,3 +1409,272 @@ class TestFollowup_ProcSelfSymlinkTraversal: ) def test_proc_legit_introspection_allowed(self, code): assert not _is_blocked(code), f"legit /proc read blocked: {code!r}" + + +class TestFollowup_BareAndMethodAliases: + """Module-rebinding bypass class: + + * ``m = os; m.system('sudo whoami')`` (bare module alias) + * ``p = os.popen; p('sudo whoami')`` (method alias) + * Same shape for ``subprocess`` and its dangerous attrs. + + Previously the alias tracker only handled ``m = __import__('os')`` + / ``m = importlib.import_module('os')`` and ``from os import system`` + -- the simple ``m = os`` and ``p = os.popen`` chains slipped through + because the static gate never propagated the source alias to ``m`` + or registered ``p`` as a shell-exec callable. + """ + + @pytest.mark.parametrize( + "code", + [ + # bare module rebinding + "import os\nm = os\nm.system('s' + 'udo whoami')", + "import os\nx = os\nx.popen('s' + 'udo whoami')", + "import subprocess\nr = subprocess\nr.run(['s'+'udo','whoami'], shell=True)", + "import subprocess\nsp = subprocess\nsp.Popen('s'+'udo whoami', shell=True)", + # chained rebinding + "import os\nm = os\nn = m\nn.system('s' + 'udo whoami')", + # method (callable) aliasing + "import os\np = os.popen\np('s'+'udo whoami')", + "import os\nss = os.system\nss('s'+'udo whoami')", + "import subprocess\nr = subprocess.run\nr(['s'+'udo','whoami'], shell=True)", + "import subprocess\np = subprocess.Popen\np('s'+'udo whoami', shell=True)", + ], + ) + def test_bare_and_method_alias_blocked(self, code): + assert _is_blocked(code), f"alias bypass leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + # Same shapes but with safe targets must keep working. + "import os\nm = os\nm.listdir('.')", + "import os\nm = os\nm.getcwd()", + "import os\nj = os.path.join\nj('a', 'b')", + "import subprocess\nr = subprocess\nr.list2cmdline(['ls'])", + # Aliasing a non-dangerous module is unrelated to the gate. + "import json\nj = json\nj.dumps({})", + # Aliasing a function we never tracked is fine. + "import os\nl = os.listdir\nl('.')", + ], + ) + def test_legit_aliases_allowed(self, code): + assert not _is_blocked(code), f"legit alias blocked: {code!r}" + + +class TestFollowup_ImportlibFromImportAlias: + """``from importlib import import_module as IM; IM('os').system(...)`` + and ``m = IM('os'); m.system(...)``. Previously the alias was + untracked, so ``IM('os')`` was not recognised as a dynamic os import.""" + + @pytest.mark.parametrize( + "code", + [ + "from importlib import import_module as IM\nIM('os').system('s'+'udo whoami')", + "from importlib import import_module as IM\nm = IM('os')\nm.system('s'+'udo whoami')", + "from importlib import import_module as IM\nIM('subprocess').run(['s'+'udo','whoami'], shell=True)", + "from importlib import import_module as load_it\nload_it('os').popen('cat ~/.ssh/id_rsa')", + "from importlib import import_module as IM\nm = IM('os')\nm.popen('cat /etc/shadow')", + ], + ) + def test_importlib_from_import_alias_blocked(self, code): + assert _is_blocked(code), f"importlib alias leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + "from importlib import import_module as IM\nIM('json').dumps({})", + "from importlib import import_module as IM\np = IM('pathlib')\np.Path('/tmp/x').exists()", + ], + ) + def test_importlib_from_import_alias_legit_allowed(self, code): + assert not _is_blocked(code), f"legit importlib alias blocked: {code!r}" + + +class TestFollowup_ShutilDirectoryExfil: + """``shutil.copytree('/home/u/.ssh', '/tmp/out')`` and + ``shutil.copy('/etc', '/tmp/out')`` drag every file out of a + sensitive directory in one call. Previously the gate only matched + per-file sensitive paths so the bare directory slipped through.""" + + @pytest.mark.parametrize( + "code", + [ + "import shutil; shutil.copytree('/home/u/.ssh', '/tmp/out')", + "import shutil; shutil.copytree('/root/.ssh', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.aws', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.config/gcloud', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.gnupg', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.docker', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.kube', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.password-store', '/tmp/out')", + "import shutil; shutil.copytree('/etc', '/tmp/out')", + "import shutil; shutil.copytree('/etc/ssh', '/tmp/out')", + "import shutil; shutil.copytree('/proc/self', '/tmp/out')", + "import shutil; shutil.copytree('/proc/1', '/tmp/out')", + "import shutil; shutil.move('/home/u/.aws', '/tmp/out')", + "import shutil; shutil.copy('/home/u/.aws', '/tmp/out')", + # trailing slash + "import shutil; shutil.copytree('/home/u/.ssh/', '/tmp/out')", + # tilde + home prefix + "import shutil; shutil.copytree('~/.ssh', '/tmp/out')", + ], + ) + def test_shutil_dir_exfil_blocked(self, code): + assert _is_blocked(code), f"shutil dir exfil leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + # Single-file legit reads under a sensitive directory -- + # the per-file allow-list governs these, NOT the dir gate. + "import shutil; shutil.copy('/home/u/.ssh/known_hosts', './b.txt')", + "import shutil; shutil.copy('/home/u/.ssh/id_rsa.pub', './b.txt')", + "import shutil; shutil.copy('/home/u/.ssh/config', './b.txt')", + # Lookalike directory names (different dir, similar prefix) + "import shutil; shutil.copytree('/home/u/.ssh_backup', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.sshconfig', '/tmp/out')", + "import shutil; shutil.copytree('/home/u/.awsd', '/tmp/out')", + # Project-local lookalikes + "import shutil; shutil.copytree('./workspace/home/u/.ssh', '/tmp/out')", + "import shutil; shutil.copytree('./project/.aws', '/tmp/out')", + # Safe directories with sensitive-looking suffix in path + "import shutil; shutil.copytree('./src', '/tmp/out')", + "import shutil; shutil.copy('./data.txt', './backup.txt')", + ], + ) + def test_shutil_dir_legit_allowed(self, code): + assert not _is_blocked(code), f"legit shutil dir blocked: {code!r}" + + +class TestFollowup_ExplicitFileReaders: + """``io.FileIO`` and ``codecs.open`` are file-reader call shapes + that do not match ``open()`` / ``.open`` but read arbitrary paths. + Treat them the same as ``open()``.""" + + @pytest.mark.parametrize( + "code", + [ + "import io; io.FileIO('/etc/shadow').read()", + "import io; io.FileIO('/etc/shadow', 'r')", + "from io import FileIO; FileIO('/etc/shadow')", + "import codecs; codecs.open('/etc/shadow').read()", + "import codecs; codecs.open('/home/u/.aws/credentials', 'r').read()", + "import io; io.FileIO('/proc/self/environ')", + ], + ) + def test_explicit_readers_blocked(self, code): + assert _is_blocked(code), f"explicit reader leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + "import io; io.FileIO('./data.bin').read()", + "import codecs; codecs.open('./input.txt', encoding='utf-8').read()", + "import io; io.FileIO('/etc/hosts').read()", # allow-listed + ], + ) + def test_explicit_readers_legit_allowed(self, code): + assert not _is_blocked(code), f"legit reader blocked: {code!r}" + + +class TestFollowup_BytesAndWalrus: + """``open(b'/etc/shadow')`` (bytes path) and + ``open((p := '/etc/shadow'))`` (walrus). Bytes are valid PathLike; + walrus must resolve to the RHS literal.""" + + @pytest.mark.parametrize( + "code", + [ + "open(b'/etc/shadow')", + "open(b'/etc/' + b'shadow')", + "import io; io.FileIO(b'/etc/shadow')", + "open((p := '/etc/shadow'))", + "p = (q := '/etc/shadow')\nopen(p)", + "open((p := '/etc/' + 'shadow'))", + ], + ) + def test_bytes_and_walrus_blocked(self, code): + assert _is_blocked(code), f"bytes/walrus path leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + "open(b'data.bin')", + "open((p := 'data.txt'))", + "x = (y := 5)\nprint(x)", + ], + ) + def test_bytes_and_walrus_legit_allowed(self, code): + assert not _is_blocked(code), f"legit bytes/walrus blocked: {code!r}" + + +class TestFollowup_TupleAndListUnpack: + """Statically-resolvable tuple / list unpacking destructuring: + ``(a, b) = ('/etc', 'shadow'); open(a + '/' + b)`` and + ``p, = ['/etc/shadow']; open(p)``. The pre-pass that backs + ``_extract_string_from_node`` now folds these into ``string_bindings``.""" + + @pytest.mark.parametrize( + "code", + [ + "(a, b) = ('/etc', 'shadow')\nopen(a + '/' + b)", + "a, b = '/etc', 'shadow'\nopen(a + '/' + b)", + "p, = ['/etc/shadow']\nopen(p)", + "[p] = ['/etc/shadow']\nopen(p)", + "(a, b, c) = ('/', 'etc/', 'shadow')\nopen(a + b + c)", + "(a, b) = ('/home/u/.aws', '/credentials')\nopen(a + b)", + ], + ) + def test_tuple_unpack_blocked(self, code): + assert _is_blocked(code), f"tuple-unpack path leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + "(a, b) = ('hello', 'world')\nprint(a + b)", + "a, b = 1, 2\nprint(a + b)", + "(a, b) = ('./input', '.txt')\nopen(a + b)", + ], + ) + def test_tuple_unpack_legit_allowed(self, code): + assert not _is_blocked(code), f"legit tuple-unpack blocked: {code!r}" + + +class TestFollowup_DataframeReaders: + """``pd.read_csv('/etc/shadow')`` and friends are file-reader calls + that bypass the ``open()`` gate. Match the common pandas / numpy + reader method names by suffix so any alias of the module is caught.""" + + @pytest.mark.parametrize( + "code", + [ + "import pandas as pd; pd.read_csv('/etc/shadow')", + "import pandas as pd; pd.read_csv('/proc/self/environ')", + "import pandas; pandas.read_csv('/home/u/.aws/credentials')", + "import pandas as pd; pd.read_excel('/proc/1/environ')", + "import pandas as pd; pd.read_json('/etc/shadow')", + "import pandas as pd; pd.read_parquet('/etc/shadow')", + "import pandas as pd; pd.read_table('/etc/shadow')", + "import pandas as pd; pd.read_pickle('/home/u/.ssh/id_rsa')", + "import numpy as np; np.fromfile('/etc/shadow')", + "import numpy as np; np.loadtxt('/etc/shadow')", + "import numpy as np; np.genfromtxt('/proc/self/environ')", + ], + ) + def test_dataframe_readers_blocked(self, code): + assert _is_blocked(code), f"dataframe reader leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + "import pandas as pd; pd.read_csv('./data.csv')", + "import pandas as pd; pd.read_excel('input.xlsx')", + "import pandas as pd; pd.read_csv('/etc/hosts')", # allow-listed + "import numpy as np; np.fromfile('./weights.bin')", + "import numpy as np; np.loadtxt('train.txt')", + ], + ) + def test_dataframe_readers_legit_allowed(self, code): + assert not _is_blocked(code), f"legit dataframe reader blocked: {code!r}"