Harden sandbox: alias read sinks, gc graph walk, list-concat fold cap, runtime sensitive-read backstop, multi-component redirects

Round 15 review follow-ups on the Studio code-exec sandbox classifier and runtime guard:

- Resolve single-assignment aliases for shutil.copy* and subprocess exec read
  sinks (c = shutil.copy; c('../../etc/passwd', ...); r = subprocess.run; r([...]))
  so the traversal / sensitive-path check fires on the aliased callee.
- Block gc.get_referents / get_referrers / get_objects (and from-import aliases):
  they walk the object graph to a guarded wrapper's closure cell to recover the
  original unguarded open / os.* callable.
- Cap list / tuple concatenation during constant folding so a doubling chain
  (a + a + a + ...) cannot materialize an oversized sequence in the parent process
  before the child rlimits apply.
- Add a runtime sensitive-read backstop in the child prelude: deny a read whose
  realpath resolves to a known host secret (SSH / cloud / kube / netrc / HF-token /
  /etc/passwd family / /proc) outside the workdir. This covers opaque read paths the
  static scanner cannot fold (open(globals()['x'])) and pre-existing in-workdir
  symlinks to secrets, while leaving benign outside reads and library imports intact.
- Fail closed on relative multi-component shell redirect targets (echo x > sub/out.txt)
  whose subdirectory component could be a symlink traversing outside the workdir; a
  bare single-component target stays allowed.

Adds TestRound15Bypasses and runtime backstop tests; full sandbox suite green.

studio/backend/core/inference/tools.py

@@ -501,10 +501,10 @@ def _find_blocked_commands(command: str) -> set[str]:
 
     # Output redirection (> / >> / &> / N>) to a path OUTSIDE the workdir: a child shell
     # runs unguarded, so `echo x > /tmp/p` / `>> ../p` / `> ~/p` writes past the session
-    # workdir. A relative literal target (> out.txt) stays in the workdir cwd and is
-    # allowed; a NON-LITERAL target (variable / command substitution) cannot be verified,
-    # so it fails closed (`echo x > "$p"` could expand anywhere). Scanning tokens (not the
-    # raw string) avoids matching a `>` inside a quoted argument.
+    # workdir. A relative SINGLE-component literal target (> out.txt) stays in the workdir
+    # cwd and is allowed; a NON-LITERAL target (variable / command substitution) cannot be
+    # verified, so it fails closed (`echo x > "$p"` could expand anywhere). Scanning tokens
+    # (not the raw string) avoids matching a `>` inside a quoted argument.
     for i, tok in enumerate(tokens):
         rm = re.search(r">{1,2}([^\s>]*)$", tok)
         if rm is None:
@@ -522,12 +522,19 @@ def _find_blocked_commands(command: str) -> set[str]:
         if not tgt:
             continue
         tn = tgt.replace("\\", "/")
+        # A relative multi-component target (sub/out.txt) resolves through a subdirectory
+        # component whose realpath the static scanner cannot verify -- if that component is
+        # a symlink pointing outside the workdir the unguarded child writes past it. Fail
+        # closed on any relative target carrying a `/` separator (a leading `./` is dropped
+        # first so `./out.txt` stays allowed); only a bare single-component name is allowed.
+        _rel = tn[2:] if tn.startswith("./") else tn
         if (
             tgt.startswith("~")
             or tn.startswith("/")
             or ".." in tn.split("/")
             or "$" in tgt
             or "`" in tgt
+            or "/" in _rel.rstrip("/")
         ):
             blocked.add("redirect:" + tgt)
 
@@ -2192,6 +2199,12 @@ def _const_fold(
                         return None
                 return _fold_cap(left * right)
             if isinstance(op, ast.Add):
+                # str/bytes concat is sized by _fold_cap, but list/tuple concatenation is
+                # not, so a chain (a + a + a + ...) materializes an oversized sequence in the
+                # parent process before child rlimits apply. Cap the combined length.
+                if isinstance(left, (list, tuple)) and isinstance(right, (list, tuple)):
+                    if len(left) + len(right) > _FOLD_MAX_SEQ:
+                        return None
                 return _fold_cap(left + right)
             if isinstance(op, ast.Mod):
                 if isinstance(left, (str, bytes, bytearray)) and not _printf_ok(left):
@@ -3764,6 +3777,12 @@ def _check_signal_escape_patterns(
             # from operator import methodcaller as mc -> {"mc"}. methodcaller('__getattribute__',
             # 'system')(os) fetches os.system, the same obfuscation as attrgetter.
             self.methodcaller_aliases: set[str] = set()
+            # import gc as g -> {"gc", "g"}. gc.get_referents / get_referrers / get_objects
+            # walk the object graph to a guard wrapper's closure cell (the original unguarded
+            # callable) without spelling __closure__, so treat them as recovery gadgets.
+            self.gc_aliases = {"gc"}
+            # from gc import get_referents as gr -> {"gr"}.
+            self.gc_walk_aliases: set[str] = set()
             self.loop_depth = 0
 
         def visit_Import(self, node):
@@ -3790,6 +3809,8 @@ def _check_signal_escape_patterns(
                     self.inspect_aliases.add(alias.asname or "inspect")
                 elif alias.name == "operator":
                     self.operator_aliases.add(alias.asname or "operator")
+                elif alias.name == "gc":
+                    self.gc_aliases.add(alias.asname or "gc")
                 if alias.name in _DESERIALIZE_MODULES:
                     self.deserialize_module_aliases[alias.asname or alias.name] = alias.name
             self.generic_visit(node)
@@ -3854,6 +3875,10 @@ def _check_signal_escape_patterns(
                         self.attrgetter_aliases.add(alias.asname or alias.name)
                     elif alias.name == "methodcaller":
                         self.methodcaller_aliases.add(alias.asname or alias.name)
+            elif node.module == "gc":
+                for alias in node.names:
+                    if alias.name in ("get_referents", "get_referrers", "get_objects"):
+                        self.gc_walk_aliases.add(alias.asname or alias.name)
             self.generic_visit(node)
 
         def visit_While(self, node):
@@ -4660,6 +4685,22 @@ def _check_signal_escape_patterns(
                     or (isinstance(func, ast.Name) and func.id in self.getclosurevars_aliases)
                 ):
                     dynamic_desc = "inspect.getclosurevars() recovers a guarded wrapper's closure"
+                elif (
+                    # gc.get_referents / get_referrers / get_objects walk the object graph to a
+                    # guard wrapper's closure cell (the original unguarded open/os.* callable)
+                    # without spelling __closure__ / cell_contents, so a recovered original can
+                    # then write/read outside the workdir. Block the graph-traversal APIs.
+                    (
+                        isinstance(func, ast.Attribute)
+                        and func.attr in ("get_referents", "get_referrers", "get_objects")
+                        and _ast_name_matches(func.value, self.gc_aliases)
+                    )
+                    or (isinstance(func, ast.Name) and func.id in self.gc_walk_aliases)
+                ):
+                    _gn = func.attr if isinstance(func, ast.Attribute) else func.id
+                    dynamic_desc = (
+                        f"gc.{_gn}() walks the object graph to a guarded wrapper's closure"
+                    )
                 elif (
                     # cls.mro().__getitem__(1) / .pop(1) / cls.__mro__.__getitem__(1): the
                     # method-call twin of the subscripted-mro base extraction
@@ -5634,6 +5675,12 @@ def _check_signal_escape_patterns(
         return False
 
     def _is_shutil_copy_callee(fn):
+        while isinstance(fn, ast.Attribute) and fn.attr == "__call__":
+            fn = fn.value
+        # A single-assignment alias (c = shutil.copy; c('../../etc/passwd', 'x')) hides the
+        # shutil.copy attribute form behind a bare Name, so resolve the RHS before matching.
+        if isinstance(fn, ast.Name):
+            fn = _unwrap_container_node(_scope_idx.resolve(fn.id, fn, "rhsnode"))
         return (
             isinstance(fn, ast.Attribute)
             and fn.attr in _SHUTIL_COPY_METHODS
@@ -5646,6 +5693,12 @@ def _check_signal_escape_patterns(
         # `..` traversal in a literal argv (subprocess.run(['cat', '../../root/.ssh/id_rsa']))
         # reads a host secret. Treat these as read callees so the traversal check fires on
         # their argv path elements (absolute-sensitive elements already block regardless).
+        while isinstance(fn, ast.Attribute) and fn.attr == "__call__":
+            fn = fn.value
+        # r = subprocess.run; r(['cat', '../../root/.ssh/id_rsa']) hides the exec attribute
+        # form behind a single-assignment alias, so resolve the RHS before matching.
+        if isinstance(fn, ast.Name):
+            fn = _unwrap_container_node(_scope_idx.resolve(fn.id, fn, "rhsnode"))
         return (
             isinstance(fn, ast.Attribute)
             and isinstance(fn.value, ast.Name)
@@ -6094,7 +6147,7 @@ def _truncate(text: str, limit: int = _MAX_OUTPUT_CHARS) -> str:
 # and the realpath-before-open TOCTOU window under adversarial in-sandbox threading.
 # --------------------------------------------------------------------------
 _SANDBOX_GUARD_SRC = r"""
-import os as _os, builtins as _bi, io as _io, pathlib as _pl
+import os as _os, builtins as _bi, io as _io, pathlib as _pl, re as _re
 # io + pathlib are imported BEFORE any patching on purpose: on Python <= 3.11
 # pathlib._NormalAccessor captures io.open / os.* into class attributes at import
 # time. A C builtin captured there does not bind on instance access, but a Python
@@ -6184,12 +6237,78 @@ def _mode_is_write(mode):
     m = str.__str__(mode) if isinstance(mode, str) else "r"
     return any(c in m for c in "wax+")
 
+# Runtime sensitive-read backstop. The static scanner cannot fold every read path
+# (open(globals()['x']), open(fetch_name()), open(''.join(...))), and reads are otherwise
+# unconfined, so an opaque path could name a host secret. Deny a read whose REALPATH
+# resolves to a known-sensitive host file OUTSIDE the workdir. In-workdir files are the
+# sandbox's own and always allowed. The loose 'credentials' / '.pem' / '/root/' signals the
+# static layer uses are intentionally NOT applied here: importing common libraries reads
+# site-packages files such as google/auth/credentials.py and certifi/cacert.pem (and, under
+# a root home, /root/.local/.../site-packages), so matching them at runtime would break
+# imports. The specific SSH / cloud / kube / netrc / HF-token signals stay.
+_SENS_EXACT = frozenset({
+    "/etc/passwd", "/etc/shadow", "/etc/sudoers", "/etc/gshadow", "/etc/master.passwd",
+})
+_SENS_DIRS = (
+    "/etc/ssh/", "/.ssh/", "/.aws/", "/.config/gcloud", "/.kube/", "/.docker/",
+    "/var/run/secrets/kubernetes.io/", "/run/secrets/kubernetes.io/",
+)
+_SENS_TOKENS = (
+    "id_rsa", "id_ed25519", ".netrc", ".git-credentials", "/.huggingface/token", ".kube/config",
+)
+_SENS_PROC = _re.compile(r"^/proc/(?:self|\d+)/(?:environ|cmdline|maps|mem|task/\d+/environ)$")
+
+def _is_sensitive_read(rp):
+    n = rp.replace("\\", "/")
+    if n in _SENS_EXACT:
+        return True
+    if any(part in n for part in _SENS_DIRS):
+        return True
+    if _SENS_PROC.match(n):
+        return True
+    low = n.lower()
+    return any(tok in low for tok in _SENS_TOKENS)
+
+def _read_realpath(p):
+    # Resolve to a truthful realpath the same self-healing way _within does, so a
+    # sandboxed reassignment of os.fspath / os.lstat / os.readlink / os.getcwd cannot
+    # poison the resolution.
+    try:
+        _os.fspath = _fspath
+        _os.lstat = _lstat
+        _os.readlink = _readlink
+        _os.getcwd = _getcwd
+        _os.stat = _stat
+        rp = _realpath(_fspath(p))
+        if isinstance(rp, bytes):
+            rp = _fsdecode(rp)
+        return rp
+    except Exception:
+        return None
+
+def _deny_sensitive_read(p):
+    if isinstance(p, int):
+        return
+    rp = _read_realpath(p)
+    if rp is None:
+        return
+    # In-workdir files are the sandbox's own; never treat them as host secrets.
+    if rp == _WD or rp.startswith(_WD + _sep):
+        return
+    if _is_sensitive_read(rp):
+        raise PermissionError(
+            "sandbox: reading a sensitive host path is not permitted: %r" % (rp,)
+        )
+
 def _guard_open_like(real):
     @_gwraps(real)
     def w(file, mode="r", *a, **k):
         f = _fspath1(file)
-        if _mode_is_write(mode) and not _within(f):
-            _deny(f, "write")
+        if _mode_is_write(mode):
+            if not _within(f):
+                _deny(f, "write")
+        else:
+            _deny_sensitive_read(f)
         return real(f, mode, *a, **k)
     return w
 
@@ -6219,7 +6338,10 @@ def _make_osopen_guard(real_open):
             if not _within(p):
                 _deny(p, "os.open write")
             return real_open(p, flags, *a, **k)
-        return real_open(path, flags, *a, **k)
+        # Read-only os.open: reads are unconfined, but a host secret is still off limits.
+        p = _fspath1(path)
+        _deny_sensitive_read(p)
+        return real_open(p, flags, *a, **k)
     return _guarded
 _os.open = _make_osopen_guard(_os.open)
 
@@ -6314,8 +6436,11 @@ 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")
+            if _mode_is_write(mode):
+                if not _within(f):
+                    _deny(f, "FileIO write")
+            else:
+                _deny_sensitive_read(f)
             # Pass the MATERIALIZED path so a stateful __fspath__ cannot return a
             # different (outside) path to the real constructor than we checked.
             super().__init__(f, mode, *a, **k)

studio/backend/tests/test_sandbox_runtime_backstop.py

@@ -126,8 +126,9 @@ def test_sandboxed_os_open_write_escape_denied(tmp_path):
 
 @_POSIX_ONLY
 def test_sandboxed_os_open_read_local_allowed():
-    # Read-only os.open of a workdir-local file is allowed (reads are not confined
-    # by the backstop; host-secret reads are caught by the static scanner instead).
+    # Read-only os.open of a workdir-local file is allowed: reads of non-sensitive paths
+    # are not confined (only mutating opens are workdir-confined, and only host-secret
+    # realpaths are denied by the runtime sensitive-read backstop).
     out = _python_exec(
         "import os\n"
         "fd = os.open('ro_probe.txt', os.O_CREAT | os.O_WRONLY, 0o600)\n"
@@ -835,3 +836,85 @@ def test_sandboxed_posix_fd_metadata_mutator_denied(tmp_path):
     )
     assert "sandbox:" in out and "fchmod" in out
     assert oct(os.stat(victim).st_mode & 0o777) == "0o600"
+
+
+_SECRET_ABS = "/" + "etc" + "/" + "passwd"
+
+
+@_POSIX_ONLY
+def test_sandboxed_opaque_read_of_secret_denied():
+    # The static scanner cannot fold an opaque read path (globals()['x']), and reads are
+    # otherwise unconfined, so the runtime sensitive-read backstop must deny a read whose
+    # realpath resolves to a host secret regardless of how the path was computed.
+    out = _python_exec(
+        "x = " + repr(_SECRET_ABS) + "\np = globals()['x']\nprint('LEN', len(open(p).read()))\n",
+        None,
+        30,
+        "backstop-opaque-read",
+        disable_sandbox = False,
+    )
+    assert "sandbox:" in out or "PermissionError" in out
+    assert "reading a sensitive host path" in out
+    assert "LEN " not in out
+
+
+@_POSIX_ONLY
+def test_sandboxed_opaque_os_open_read_of_secret_denied():
+    # The same opaque path routed through the low-level os.open read entry point.
+    out = _python_exec(
+        "import os\n"
+        "x = " + repr(_SECRET_ABS) + "\n"
+        "p = globals()['x']\n"
+        "fd = os.open(p, os.O_RDONLY); print('FD', fd)\n",
+        None,
+        30,
+        "backstop-opaque-osopen",
+        disable_sandbox = False,
+    )
+    assert "sandbox:" in out or "PermissionError" in out
+    assert "reading a sensitive host path" in out
+    assert "FD " not in out
+
+
+@_POSIX_ONLY
+def test_sandboxed_symlink_read_of_secret_denied(tmp_path):
+    # A pre-existing in-workdir symlink pointing at a host secret: the static scanner sees a
+    # benign local name ('notes.txt'), only the runtime realpath backstop can follow the
+    # link and deny the read. (Sandboxed code cannot create the symlink; this is the
+    # defense-in-depth the runtime layer adds over static analysis.)
+    session = "backstop-symlink-read"
+    workdir = get_sandbox_workdir(session)
+    link = os.path.join(workdir, "notes.txt")
+    if os.path.islink(link) or os.path.exists(link):
+        os.remove(link)
+    os.symlink(_SECRET_ABS, link)
+    try:
+        out = _python_exec(
+            "print('LEN', len(open('notes.txt').read()))\n",
+            None,
+            30,
+            session,
+            disable_sandbox = False,
+        )
+        assert "sandbox:" in out or "PermissionError" in out
+        assert "reading a sensitive host path" in out
+        assert "LEN " not in out
+    finally:
+        os.remove(link)
+
+
+@_POSIX_ONLY
+def test_sandboxed_benign_outside_read_allowed():
+    # Reads are not confined to the workdir; only sensitive realpaths are denied. A benign
+    # outside read (and importing libraries whose files carry 'credentials'/'.pem' in the
+    # name) must stay allowed so the backstop does not break normal computation.
+    out = _python_exec(
+        "print('HOST', open('/etc/hostname').read().strip()[:0] == '')\n"
+        "import json, urllib.request, ssl, email\nprint('IMPORTS_OK')",
+        None,
+        30,
+        "backstop-benign-read",
+        disable_sandbox = False,
+    )
+    assert "IMPORTS_OK" in out
+    assert "sandbox:" not in out

studio/backend/tests/test_sandbox_tools.py

@@ -2114,3 +2114,73 @@ class TestRound14Bypasses:
 
     def test_class_attribute_benign_allowed(self):
         _ok("class C:\n    x = 1\nprint(C.x)")
+
+
+class TestRound15Bypasses:
+    """Fifteenth-round Codex findings: single-assignment aliases of shutil.copy /
+    subprocess.run read sinks, gc.get_referents guard-recovery, an uncapped list
+    concatenation during const folding, and relative multi-component shell redirects.
+    (The opaque-read backstop is a runtime guard, covered in the runtime test module.)"""
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "import shutil\nc = shutil.copy\nc('../../../etc/passwd', 'leak.txt')",
+            "import shutil as sh\nc = sh.copyfile\nc('../../../etc/passwd', 'leak.txt')",
+            "import subprocess\nr = subprocess.run\nr(['cat', '../../../root/.ssh/id_rsa'])",
+            "import subprocess\np = subprocess.Popen\np(['cat', '/etc/shadow'])",
+        ],
+    )
+    def test_aliased_read_sink_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    def test_aliased_read_sink_local_allowed(self):
+        # A single-assignment alias whose source is an in-workdir relative path stays allowed.
+        _ok("import shutil\nc = shutil.copy\nc('data/in.csv', 'out.csv')")
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "import gc, builtins\ngc.get_referents(builtins.open)",
+            "import gc\ngc.get_referrers(open)",
+            "from gc import get_referents as g\ng(open)",
+            "import gc\ngc.get_objects()",
+        ],
+    )
+    def test_gc_graph_walk_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    def test_list_concat_fold_is_capped_and_fast(self):
+        # A doubling chain of list concatenations must NOT be materialized during folding
+        # (that is an analysis-time memory/CPU DoS); the fold caps the sequence length.
+        import time
+
+        dos = (
+            "a = [65] * 40000\n"
+            + "\n".join(
+                f"a{i} = a{'' if i == 0 else i - 1} + a{'' if i == 0 else i - 1}"
+                for i in range(1, 12)
+            )
+            + "\nexec(bytes(a11))"
+        )
+        t0 = time.time()
+        res = _check_code_safety(dos)
+        dt = time.time() - t0
+        assert res is not None, "the exec(...) sink should still be blocked"
+        assert dt < 2.0, f"folding a list-concat chain took {dt:.2f}s (should be capped)"
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "import os\nos.system('echo escaped > outlink/pwn.txt')",
+            "import os\nos.system('echo x > logs/app.log')",
+            "import os\nos.system('cat data >> sub/dir/out.txt')",
+        ],
+    )
+    def test_relative_multicomponent_redirect_blocked(self, code):
+        assert _check_code_safety(code) is not None, code
+
+    def test_single_component_redirect_allowed(self):
+        # A bare single-component relative redirect target stays in the workdir cwd.
+        _ok("import os\nos.system('echo x > out.txt')")
+        _ok("import os\nos.system('echo x > ./out.txt')")