studio/sandbox: close round-4 bypass classes (aliasing, FileIO, walrus, copytree)

Closes additional bypass classes surfaced while exercising the gate:

1. Module / function aliasing (`m = os; m.system(...)`,
   `p = os.popen; p(...)`): `visit_Assign` now propagates the source
   alias when one tracked-module name is bound to another, and tracks
   bound method references into `shell_exec_aliases`. Previously only
   `m = __import__('os')` was handled.

2. Importlib from-alias (`from importlib import import_module as IM;
   IM('os').system(...)`): a new visitor-scope `import_module_aliases`
   set plus a `_resolve_dynamic_module` wrapper recognises the
   bound name in both inline-call and bound-name forms.

3. Shutil directory exfil (`shutil.copytree('~/.ssh', dst)`):
   `_matches_sensitive_dir()` adds a directory-only matcher used by
   the file-copy gate only. The boundary `(?=/?$|/?[\s'\";&|)<>])`
   matches the path AS the directory but NOT a single file inside it,
   so per-file allow-listed reads (`~/.ssh/known_hosts`, `~/.ssh/id_rsa.pub`)
   still pass. Covers `.ssh`, `.aws`, `.config/gcloud`, `.gnupg`,
   `.docker`, `.kube`, `.password-store`, plus `/etc`, `/etc/ssh`,
   `/var/spool/cron`, `/proc/<pid>`.

4. Explicit-reader / aliased file readers (`io.FileIO('/etc/shadow')`,
   `codecs.open('/etc/shadow')`, `from io import FileIO; FileIO(...)`):
   the open-call detector now recognises these qualified forms and
   the visitor tracks `from io|codecs import FileIO|open` aliases.

5. Bytes-literal paths (`open(b'/etc/shadow')`) and walrus
   expressions (`open((p := '/etc/shadow'))`): `_extract_string_literal`
   and `_extract_string_from_node` resolve `bytes` Constants via strict
   UTF-8 decode and `NamedExpr` via RHS extraction (recording the
   binding so later uses of the walrus target resolve too).

6. Tuple / list unpacking destructuring (`(a, b) = ('/etc', 'shadow');
   open(a + '/' + b)` and `p, = ['/etc/shadow']; open(p)`): the
   string-binding pre-pass now folds matched-length Tuple/List
   destructurings element-wise.

7. Pandas / numpy file readers (`pd.read_csv('/etc/shadow')`,
   `np.fromfile('/etc/shadow')`, etc.): suffix-match the common
   reader method names so any alias of the source module flows
   through the same sensitive-path gate as `open()`.

91 new regression tests cover each class, both blocked and legitimate
allow-list cases. Full sandbox suite: 473 passed.

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/<pid>/`` 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(
                             {

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}"