From 59fff4b86ba03947f3bce8383de9d281fbab82df Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Thu, 9 Jul 2026 22:34:19 +0000 Subject: [PATCH] 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 | 145 ++++++++++++++++-- .../tests/test_sandbox_runtime_backstop.py | 87 ++++++++++- studio/backend/tests/test_sandbox_tools.py | 70 +++++++++ 3 files changed, 290 insertions(+), 12 deletions(-) diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 7175fd6db1..eb02edf021 100644 --- a/studio/backend/core/inference/tools.py +++ b/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) diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 007c4d4129..665733a415 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/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 diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index b6f2a851bc..ea7853979f 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/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')")