Harden sandbox: global/nonlocal chained aliases, wrapper-hidden env -C / readers, meta_path.__class__, getattr re-exports, git -o/x, find/ls expanded reads

Close eight P1 bypasses Codex found on the round-39 branch:

- global / nonlocal chained alias: a `global t; t = s` (with s = os.system bound in
  the module / an enclosing scope) rebinds the target-scope name from an EXISTING
  alias, but the global/nonlocal indexer only resolved a DIRECT sink RHS, so t stayed
  unlinked and t('...') ran unguarded. Resolve a bare-name RHS through the local map
  and the already-indexed enclosing / module scopes, matching the local pass.
- env -C behind a wrapper operand (argv): _argv_env_chdir stopped at a wrapper's own
  operand (timeout 1 ...) before reaching env, so run(['timeout','1','env','-C','/etc',
  'cat','passwd']) escaped. Skip wrapper flags / numeric operands, mirroring
  _blocked_in_argv, so the trailing env -C is found.
- env -C shell operand with an expansion: `P=/etc; env -C $P cat passwd` stored $P
  literally as the cwd and never combined it. Track local VAR=value bindings and
  resolve a $VAR chdir operand against them; an unknown $ / backtick expansion fails
  closed for the following relative reader.
- wrapper-hidden reader under a dynamic cwd: the cwd=P (non-literal) reader check only
  tested argv[0], so run(['timeout','1','cat','passwd'], cwd=P) hid the reader behind
  the wrapper. Resolve the executed command word past wrappers before the relative-arg
  fail-closed decision.
- sys.meta_path.__class__ mutation: the unbound list-method guard recognized `list.*`
  and `type(sys.meta_path).*` but not `sys.meta_path.__class__.pop(sys.meta_path, 0)`,
  which removes the workdir import vetter. Add the `.__class__` receiver form (and the
  same for the sys.modules loader-table guard).
- getattr / vars re-export of a call-returned module: `<mod>.os.system` was caught, but
  getattr(__import__('pathlib'), 'os').system(...) / vars(...)['os'].system(...) /
  <mod>.__dict__['os'].system(...) were not. Map an os / posix / subprocess fetched by
  name off any expression to the sink module.
- git stuck short path option: git archive -o/tmp/x glues the escaping output path onto
  the short flag with no space, which the separated / --opt=val scans missed. Handle the
  -o / -O / -C stuck short form (a non-escaping value like -oout.tar stays allowed).
- find / ls on an expanded path: only _SHELL_READ_COMMANDS ran the expansion check, so
  find ${P:-/root/.ssh} -exec cat {} \; and ls $SECRET enumerated an unresolved host
  root. Add find / ls as enumerator readers; literal find / ls (find . -name '*.py',
  ls -la) carry no expansion and stay allowed.

Regression coverage: TestRound40Bypasses in tests/test_sandbox_tools.py (per-item
blocked cases plus a benign-allowed set: benign global reassignment / alias, literal
find / ls, a relative git output path, benign git archive, a non-reader wrapped command
under a dynamic cwd, and a benign getattr on a non-module object).

studio/backend/core/inference/tools.py

@@ -516,6 +516,12 @@ _SHELL_READ_COMMANDS = frozenset(
         "diff3",
         "colordiff",
         "cmp",
+        # directory / file enumerators: an EXPANDED root (find ${P:-/root/.ssh} -exec cat {} \;,
+        # ls $SECRET) enumerates a host path the static scan cannot resolve, and find's -exec
+        # can then read every match. Literal find / ls (find . -name '*.py', ls -la) carry no
+        # expansion and stay allowed; only a $ / backtick / escaping-glob operand fails closed.
+        "find",
+        "ls",
     }
 )
 # Wrappers whose next non-flag argument is the command Bash will exec.
@@ -1551,6 +1557,15 @@ def _find_blocked_commands(command: str) -> set[str]:
                     blocked.add("git-write-outside")
                 _gk += 2
                 continue
+            # Stuck short form: git archive -o/tmp/x (and -O.. / -C/outside) glue the path value
+            # directly onto the short option with no space, which the separated / --opt=val scans
+            # above miss. Only the path-valued SHORT options take a glued value; a non-escaping
+            # value (-oout.tar, -C90 for find-copies) is left alone by _git_operand_escapes.
+            if len(_gt) > 2 and _gt[:2] in ("-C", "-o", "-O"):
+                if _git_operand_escapes(_gt[2:], _local_assigns):
+                    blocked.add("git-write-outside")
+                _gk += 1
+                continue
             _oeq = None
             for _opt in _GIT_PATH_VALUE_OPTIONS:
                 if _gt.startswith(_opt + "="):
@@ -4067,6 +4082,39 @@ class _AnalyzerBudget:
         self.nodes = 0
 
 
+def _reexport_dangerous_module_name(expr):
+    """For a re-export gadget that fetches a submodule by NAME off some object and then calls a
+    sink on it, return the fetched module name normalized to 'os' / 'subprocess' (or None).
+
+    Covers ``getattr(<mod>, 'os')``, ``vars(<mod>)['os']`` and ``<mod>.__dict__['os']`` -- the
+    call / subscript twins of the plain ``<mod>.os`` attribute form (pathlib.os.system), which
+    stay reachable off a call-returned module (``getattr(__import__('pathlib'), 'os').system``)."""
+    def _str_const(n):
+        return n.value if isinstance(n, ast.Constant) and isinstance(n.value, str) else None
+
+    key = None
+    if (
+        isinstance(expr, ast.Call)
+        and isinstance(expr.func, ast.Name)
+        and expr.func.id == "getattr"
+        and len(expr.args) >= 2
+    ):
+        key = _str_const(expr.args[1])
+    elif isinstance(expr, ast.Subscript):
+        base = expr.value
+        if (
+            isinstance(base, ast.Call)
+            and isinstance(base.func, ast.Name)
+            and base.func.id == "vars"
+        ) or (isinstance(base, ast.Attribute) and base.attr == "__dict__"):
+            key = _str_const(expr.slice)
+    if key in ("os", "posix"):
+        return "os"
+    if key == "subprocess":
+        return "subprocess"
+    return None
+
+
 def _fq_attr_name(node):
     """Return the dotted name for a Name/Attribute chain, else ''."""
     parts = []
@@ -4657,6 +4705,20 @@ def _build_scope_alias_index(tree, const_env):
         # nothing. Target scopes are processed before nested scopes, so setdefault preserves
         # any alias they already hold.
         if global_names or nonlocal_names:
+
+            def _chain_lookup(nm, table, local_map):
+                # Resolve a bare-name RHS alias (global t; t = s) against this scope's just-built
+                # local map, then the already-indexed enclosing / module scopes -- so a
+                # `s = os.system` at module scope copied into a `global t; t = s` is propagated,
+                # matching the chained-alias resolution the local pass does via smap[rhs.id].
+                _sc = scope
+                while _sc is not None:
+                    _m = local_map if _sc is scope else table.get(_sc, {})
+                    if nm in _m:
+                        return _m[nm]
+                    _sc = idx.enclosing.get(_sc)
+                return None
+
             for name, rhs in assigns:
                 if name in global_names:
                     _target = tree
@@ -4670,12 +4732,18 @@ def _build_scope_alias_index(tree, const_env):
                 _gfq = _resolve_static_shell_sink(
                     _rhs_eff, os_aliases, subprocess_aliases, from_aliases
                 )
+                if _gfq is None and isinstance(_rhs_eff, ast.Name):
+                    _gfq = _chain_lookup(_rhs_eff.id, idx.shell, smap)
                 if _gfq:
                     idx.shell.setdefault(_target, {}).setdefault(name, _gfq)
                 _geb = _rhs_exec_builtin(_rhs_eff)
+                if _geb is None and isinstance(_rhs_eff, ast.Name):
+                    _geb = _chain_lookup(_rhs_eff.id, idx.execb, emap)
                 if _geb is not None:
                     idx.execb.setdefault(_target, {}).setdefault(name, _geb)
                 _gdfq = _rhs_deserializer(_rhs_eff)
+                if _gdfq is None and isinstance(_rhs_eff, ast.Name):
+                    _gdfq = _chain_lookup(_rhs_eff.id, idx.deser, dmap)
                 if _gdfq is not None:
                     idx.deser.setdefault(_target, {}).setdefault(name, _gdfq)
         if smap:
@@ -5021,6 +5089,24 @@ def _extract_command_subs(s):
     return subs
 
 
+def _resolve_read_chdir(operand, assigns):
+    """Resolve an ``env -C DIR`` operand for the read scanner. Returns ``(dir, dynamic)``:
+
+    - a ``$VAR`` / ``${VAR}`` that a preceding assignment in the same command bound (``P=/etc;
+      env -C $P cat passwd``) resolves to that value so the read is combined and caught;
+    - any other operand carrying ``$`` / backtick is an UNKNOWN expansion that the unguarded
+      child can point outside the workdir, so ``dynamic=True`` (fail closed for relative reads);
+    - a plain literal DIR resolves to itself."""
+    m = re.fullmatch(r"\$\{?([A-Za-z_][A-Za-z0-9_]*)\}?", operand)
+    if m:
+        if assigns and m.group(1) in assigns:
+            return assigns[m.group(1)], False
+        return None, True
+    if "$" in operand or "`" in operand:
+        return None, True
+    return operand, False
+
+
 def _argv_env_chdir(str_elts):
     """Extract an ``env -C DIR`` / ``--chdir[=DIR]`` target from a folded argv vector.
 
@@ -5043,9 +5129,11 @@ def _argv_env_chdir(str_elts):
                     return str_elts[i + 1] if i + 1 < n else None
                 if tok.startswith("--chdir="):
                     return tok.split("=", 1)[1]
-                if _wrapper_flag_takes_operand("env", tok):
-                    i += 2
-                    continue
+            # A wrapper flag that consumes the NEXT token (env -u NAME, sudo -u user, nice -n 5,
+            # timeout -k 5): skip both so the operand is not mistaken for the command word.
+            if wrapper is not None and _wrapper_flag_takes_operand(wrapper, tok):
+                i += 2
+                continue
             i += 1
             continue
         base = os.path.basename(tok).lower()
@@ -5053,10 +5141,49 @@ def _argv_env_chdir(str_elts):
             wrapper = base
             i += 1
             continue
+        # A wrapper's numeric / duration operand (timeout 1, nice 5): skip it and keep scanning
+        # for a following env -C, rather than stopping at it as the executed command word. Without
+        # this, `env -C` hidden behind `timeout 1 env -C /etc ...` was never reached.
+        if wrapper is not None and _is_wrapper_numeric_arg(tok):
+            i += 1
+            continue
         return None  # reached the executed command word before any env -C
     return None
 
 
+def _argv_command_word_index(str_elts):
+    """Index of the EXECUTED command word in a folded argv vector, skipping VAR=value
+    assignments and command wrappers (env / sudo / nice / timeout / ...) with their flags and
+    numeric / separated operands. None if a token is unresolved (None) or no command word is
+    reached. Lets a wrapper-hidden reader (['timeout', '1', 'cat', 'x']) be found at its real
+    position instead of stopping at argv[0]."""
+    i, n = 0, len(str_elts)
+    wrapper = None
+    while i < n:
+        tok = str_elts[i]
+        if tok is None:
+            return None
+        if _ASSIGNMENT_RE.match(tok):
+            i += 1
+            continue
+        if tok.startswith("-"):
+            if wrapper is not None and _wrapper_flag_takes_operand(wrapper, tok):
+                i += 2
+                continue
+            i += 1
+            continue
+        base = os.path.basename(tok).lower()
+        if base in _COMMAND_PREFIXES:
+            wrapper = base
+            i += 1
+            continue
+        if wrapper is not None and _is_wrapper_numeric_arg(tok):
+            i += 1
+            continue
+        return i
+    return None
+
+
 def _scan_command_string_for_reads(
     command,
     *,
@@ -5264,6 +5391,12 @@ def _scan_command_string_for_reads(
     _chdir = cwd
     _pending_chdir = False
     _pending_argfile = False
+    # env -C DIR whose DIR is an unknown expansion ($UNRESOLVED / backtick): the child's cwd is
+    # unprovable, so a later relative reader arg fails closed. Reset per command, like _chdir.
+    _chdir_dynamic = False
+    # Shell VAR=value bindings seen so far (P=/etc; env -C $P ...), so an env -C $P operand
+    # resolves to /etc and the read is combined + caught. Persists across separators.
+    _local_assigns = {}
     for _pi, _pt in enumerate(ptoks):
         if _pt in _READ_SCAN_SEPARATORS:
             _at_cmd = True
@@ -5272,6 +5405,7 @@ def _scan_command_string_for_reads(
             _skip_operand = False
             _pending_argfile = False
             _chdir = cwd
+            _chdir_dynamic = False
             _pending_chdir = False
             continue
         if _pt.startswith("<"):
@@ -5284,7 +5418,11 @@ def _scan_command_string_for_reads(
         if _at_cmd:
             if _skip_operand:  # a wrapper flag's separated operand (env -u NAME)
                 if _pending_chdir:  # ...but env -C DIR's operand is the child cwd
-                    _chdir = _join_chdir(_chdir, _pt)
+                    _rdir, _rdyn = _resolve_read_chdir(_pt, _local_assigns)
+                    if _rdyn:
+                        _chdir_dynamic = True
+                    else:
+                        _chdir = _join_chdir(_chdir, _rdir)
                     _pending_chdir = False
                 elif _pending_argfile:  # ...and xargs -a FILE reads FILE
                     _pending_argfile = False
@@ -5296,6 +5434,8 @@ def _scan_command_string_for_reads(
                 _r = _check_assignment_rhs(_pt)
                 if _r is not None:
                     return _r
+                _an, _, _av = _pt.partition("=")
+                _local_assigns[_an.rstrip("+")] = _av
                 continue  # assignment prefix; the command word is still ahead
             if _pt.startswith("-"):
                 # env -C DIR / --chdir DIR changes the child's cwd before the command runs, so
@@ -5305,7 +5445,11 @@ def _scan_command_string_for_reads(
                     _pending_chdir = True
                     _skip_operand = True
                 elif _wrapper == "env" and _pt.startswith("--chdir="):
-                    _chdir = _join_chdir(_chdir, _pt.split("=", 1)[1])
+                    _rdir, _rdyn = _resolve_read_chdir(_pt.split("=", 1)[1], _local_assigns)
+                    if _rdyn:
+                        _chdir_dynamic = True
+                    else:
+                        _chdir = _join_chdir(_chdir, _rdir)
                 # xargs -a FILE / --arg-file[=]FILE reads its argument list FROM that file, so a
                 # sensitive / expanded target is a host-file read even though xargs is a wrapper.
                 elif _wrapper == "xargs" and _pt in ("-a", "--arg-file"):
@@ -5360,6 +5504,10 @@ def _scan_command_string_for_reads(
             if "$" in _pt or "`" in _pt or _escaping_glob(_pt):
                 return f"shell read command reads an expanded path {_pt!r}"
             _rel = not _pt.startswith("/") and not _pt.startswith("~")
+            # A relative reader arg under an env -C whose DIR was an UNKNOWN expansion (env -C
+            # $UNRESOLVED cat passwd) cannot be proven sandbox-local, so fail closed.
+            if _chdir_dynamic and _rel:
+                return f"shell read command reads {_pt!r} under an expanded chdir"
             # Under a known chdir (env -C DIR or an ambient subprocess cwd=), a relative reader
             # arg resolves against DIR (cat passwd + cwd=/etc -> /etc/passwd).
             if _chdir and _rel:
@@ -6615,6 +6763,15 @@ def _check_signal_escape_patterns(
                 elif isinstance(_ecf.value, ast.Attribute) and _ecf.value.attr == "subprocess":
                     # ...and *.subprocess.run (a module re-exporting subprocess).
                     shell_func = f"subprocess.{_ecf.attr}"
+                if shell_func is None:
+                    # getattr(<mod>, 'os').system(...) / vars(<mod>)['os'].system(...) /
+                    # <mod>.__dict__['subprocess'].run(...): a re-export fetched by NAME -- the
+                    # call / subscript twin of the <mod>.os.system attribute form, and reachable
+                    # off a call-returned module (getattr(__import__('pathlib'), 'os')). Map the
+                    # fetched os / posix / subprocess to the sink module so the chain is a sink.
+                    _rx = _reexport_dangerous_module_name(_ecf.value)
+                    if _rx:
+                        shell_func = f"{_rx}.{_ecf.attr}"
             elif isinstance(_ecf, ast.Name):
                 # from-import aliases: from os import system; system(...)
                 shell_func = self.shell_exec_aliases.get(_ecf.id)
@@ -7263,6 +7420,13 @@ def _check_signal_escape_patterns(
                             and isinstance(func.value.func, ast.Name)
                             and func.value.func.id == "type"
                         )
+                        # sys.modules.__class__.pop(sys.modules, ...): the receiver is the dict
+                        # TYPE reached via .__class__, not the bare `dict` name / type(...) call.
+                        or (
+                            isinstance(func.value, ast.Attribute)
+                            and func.value.attr == "__class__"
+                            and self._is_sys_modules(func.value.value)
+                        )
                     )
                 ):
                     dynamic_desc = (
@@ -7324,6 +7488,13 @@ def _check_signal_escape_patterns(
                             and isinstance(func.value.func, ast.Name)
                             and func.value.func.id == "type"
                         )
+                        # sys.meta_path.__class__.pop(sys.meta_path, 0): the receiver is the list
+                        # TYPE reached via .__class__, not the bare `list` name / type(...) call.
+                        or (
+                            isinstance(func.value, ast.Attribute)
+                            and func.value.attr == "__class__"
+                            and self._is_sys_meta_path(func.value.value)
+                        )
                     )
                 ):
                     dynamic_desc = (
@@ -9083,12 +9254,17 @@ def _check_signal_escape_patterns(
                             break
                 if isinstance(_argv_node, (ast.List, ast.Tuple)):
                     _av = _argv_node.elts
-                    _p0 = _fold_read_arg(_av[0]) if _av else None
+                    # Resolve the real command word past wrappers (timeout / env / nice / ...), so
+                    # a wrapper-hidden reader (['timeout', '1', 'cat', 'passwd']) is checked, not
+                    # just argv[0]. Then scan the reader's own relative args for a host read.
+                    _folded = [_fold_read_arg(_e) for _e in _av]
+                    _ci = _argv_command_word_index(_folded)
+                    _p0 = _folded[_ci] if _ci is not None else None
                     if (
                         isinstance(_p0, str)
                         and os.path.basename(_p0).lower() in _SHELL_READ_COMMANDS
                     ):
-                        for _ae in _av[1:]:
+                        for _ae in _av[_ci + 1 :]:
                             _av_s = _fold_read_arg(_ae)
                             if (
                                 isinstance(_av_s, str)

studio/backend/tests/test_sandbox_tools.py

@@ -4189,3 +4189,129 @@ class TestRound39Bypasses:
     )
     def test_round39_benign_allowed(self, code):
         _ok(code)
+
+
+class TestRound40Bypasses:
+    """Fortieth-round Codex findings: a global/nonlocal name rebound from an existing sink alias,
+    env -C hidden behind a wrapper operand in argv, an env -C shell operand with an expansion, a
+    wrapper-hidden reader under a dynamic cwd, sys.meta_path.__class__ list-method mutation, a
+    re-export fetched via getattr / vars on a call-returned module, a git stuck short path option
+    (-o/tmp/x), and find / ls enumerators on an expanded path operand."""
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # global/nonlocal name rebound from an EXISTING alias (s = os.system; t = s), not the
+            # sink directly: the target-scope alias must inherit the chained sink identity.
+            "import os\ns = os.system\ndef f():\n    global t\n    t = s\n    t('touch /tmp/x')\nf()",
+            "import os\ndef outer():\n    s = os.system\n    t = None\n    def f():\n        nonlocal t\n        t = s\n        t('touch /tmp/x')\n    f()",
+        ],
+    )
+    def test_global_nonlocal_chained_alias_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # env -C hidden behind a wrapper's operand (timeout 1 env -C /etc ...): the argv env-C
+            # scan must skip the wrapper operand to reach env.
+            "import subprocess\nsubprocess.run(['timeout', '1', 'env', '-C', '/etc', 'cat', 'passwd'])",
+            "import subprocess\nsubprocess.run(['nice', '5', 'env', '-C', '/etc', 'cat', 'passwd'])",
+        ],
+    )
+    def test_argv_env_c_behind_wrapper_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # A shell env -C operand with an expansion: a locally-assigned $P resolves to /etc, and
+            # an unknown expansion fails closed.
+            "import os\nos.system('P=/etc; env -C $P cat passwd')",
+            "import os\nos.system('env -C $UNKNOWN cat passwd')",
+        ],
+    )
+    def test_shell_env_c_expansion_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # A wrapper-hidden reader under a NON-literal cwd (timeout 1 cat passwd, cwd=P): resolve
+            # the command word past the wrapper before the relative-arg fail-closed decision.
+            "import subprocess\ndef f(P):\n    subprocess.run(['timeout', '1', 'cat', 'passwd'], cwd=P)",
+        ],
+    )
+    def test_wrapped_reader_dynamic_cwd_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # sys.meta_path.__class__.pop(sys.meta_path, 0): the receiver is the list type reached
+            # via .__class__, not the bare `list` name / type(...) call.
+            "import sys\nsys.meta_path.__class__.pop(sys.meta_path, 0)",
+            "import sys\nsys.modules.__class__.pop(sys.modules, '_io')",
+        ],
+    )
+    def test_class_unbound_method_mutation_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # A dangerous re-export fetched by NAME off a call-returned module, then a sink call.
+            "getattr(__import__('pathlib'), 'os').system('touch /tmp/x')",
+            "vars(__import__('pathlib'))['os'].system('touch /tmp/x')",
+            "__import__('pathlib').__dict__['os'].system('touch /tmp/x')",
+            "import importlib\ngetattr(importlib.import_module('pathlib'), 'subprocess').run(['rm', '-rf', '/tmp/x'])",
+        ],
+    )
+    def test_reexport_getattr_vars_module_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # git stuck short path option: git archive -o/tmp/x glues the escaping output path onto
+            # the short flag with no space.
+            "import os\nos.system('git archive -o/tmp/x HEAD')",
+            "import os\nos.system('git archive -O/tmp/outdir HEAD')",
+        ],
+    )
+    def test_git_stuck_short_output_option_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # find / ls enumerators on an EXPANDED path operand: the root the child enumerates
+            # cannot be resolved statically (find ${P:-/root/.ssh} ..., ls $SECRET).
+            "import os\nos.system('find ${P:-/root/.ssh} -type f')",
+            "import os\nos.system('ls $SECRET')",
+            "import os\nos.system('find $DIR -type f')",
+        ],
+    )
+    def test_find_ls_expanded_path_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # A benign global reassignment, a benign alias, literal find / ls (no expansion), a
+            # relative git output path, benign git archive, a wrapped enumerator under a dynamic
+            # cwd, and a benign getattr on a non-module object must all still pass.
+            "def f():\n    global t\n    t = 5\n    return t",
+            "def f():\n    global t\n    t = print\n    t('hi')\nf()",
+            "import os\nos.system('find . -name \"*.py\"')",
+            "import os\nos.system('ls -la')",
+            "import os\nos.system('ls data/')",
+            "import os\nos.system('find build -type f -name \"*.o\"')",
+            "import os\nos.system('git archive -oout.tar HEAD')",
+            "import os\nos.system('git archive HEAD')",
+            "import subprocess\ndef f(P):\n    subprocess.run(['timeout', '1', 'echo', 'hi'], cwd=P)",
+            "obj = Foo()\ngetattr(obj, 'run')()",
+        ],
+    )
+    def test_round40_benign_allowed(self, code):
+        _ok(code)