diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 59f495902b..00c16d411e 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -111,6 +111,10 @@ _BLOCKED_COMMANDS_COMMON = frozenset( "eval", "source", "ln", + # flock [options] | (or flock -c ) runs an arbitrary + # command in an unguarded child while holding a lock; its file/fd operand + -c forms + # make the command word hard to resolve, so block the wrapper outright. + "flock", } ) # Language interpreters that run inline / file / stdin code in a FRESH child process. @@ -218,7 +222,7 @@ _SHELL_SEPARATORS = frozenset({";", "&&", "||", "|", "&", "\n", "(", ")", "`", " # (then / do / else). `if touch x; then :; fi` executes `touch` as the condition command, so # these must reset command position -- otherwise the header word is mistaken for the command # and the real command it precedes is skipped as an argument. -_SHELL_KEYWORDS_AS_SEP = frozenset({"then", "do", "else", "elif", "if", "while", "until"}) +_SHELL_KEYWORDS_AS_SEP = frozenset({"then", "do", "else", "elif", "if", "while", "until", "coproc"}) # POSIX / common shell binaries. A shell without an inline `-c` payload runs unscanned # code (a script file, -s / stdin, or a bare stdin-reading shell), so it is denied. _SHELL_BINARIES = frozenset({"bash", "sh", "zsh", "dash", "ksh", "csh", "tcsh", "fish"}) @@ -250,7 +254,12 @@ def _is_local_executable_path(tok: str) -> bool: t = tok.replace("\\", "/") if "/" not in t: return False - return not t.startswith(_SYSTEM_BIN_PREFIXES) + # Collapse .. before the system-bin exemption so a workdir shebang cannot masquerade as a + # trusted binary via /usr/bin/../..//evil (normpath -> //evil, not exempt). + # normpath keeps the leading ./ -> bare-name collapse harmless: the "/" check above already + # ran on the original token, so ./evil (has a slash) still reaches here and stays local. + norm = os.path.normpath(t) + return not norm.startswith(_SYSTEM_BIN_PREFIXES) # The only shell redirection targets trusted without a realpath check: standard device @@ -1127,6 +1136,16 @@ def _find_blocked_commands(command: str) -> set[str]: _cmd_word_idx = _command_word_indices() + # trap 'CMD' SIGSPEC registers CMD to run (in the unguarded shell) on EXIT / a signal, so + # the quoted handler is unscanned shell code. Scan the handler operand of a command-position + # `trap` recursively; a reset (trap - EXIT) / ignore (trap '' EXIT) has nothing to run. + for i in _cmd_word_idx: + if _token_basename(tokens[i]) != "trap" or i + 1 >= len(tokens): + continue + _h = tokens[i + 1] + if _h and _h != "-" and _h not in _SHELL_SEPARATORS and _h not in _SHELL_KEYWORDS_AS_SEP: + blocked |= _find_blocked_commands(_h) + # A shell binary invoked with a SCRIPT FILE (`bash s.sh`) or `-s` (read the script from # stdin) runs unscanned shell code in the same unguarded environment; only the inline # `-c '...'` form is statically analyzable (handled above). Block a command-position @@ -4439,6 +4458,38 @@ def _join_chdir(base, newdir): return os.path.join(base, newdir) if base else newdir +def _extract_command_subs(s): + """Extract the inner payloads of ``$(...)`` and backtick command substitutions from a + shell string, INCLUDING those inside double quotes (bash runs a substitution regardless + of surrounding quotes: ``echo "$(head /etc/passwd)"``). Returns a list of inner command + strings for recursive read scanning. ``$((arith))`` yields a harmless ``(arith)`` payload + that scans clean.""" + subs = [] + i, n = 0, len(s) + while i < n: + c = s[i] + if c == "`": + j = s.find("`", i + 1) + if j == -1: + break + subs.append(s[i + 1 : j]) + i = j + 1 + elif c == "$" and i + 1 < n and s[i + 1] == "(": + depth = 1 + k = i + 2 + while k < n and depth: + if s[k] == "(": + depth += 1 + elif s[k] == ")": + depth -= 1 + k += 1 + subs.append(s[i + 2 : k - 1] if depth == 0 else s[i + 2 : k]) + i = k + else: + i += 1 + return subs + + def _argv_env_chdir(str_elts): """Extract an ``env -C DIR`` / ``--chdir[=DIR]`` target from a folded argv vector. @@ -4593,6 +4644,35 @@ def _scan_command_string_for_reads( ) if _r is not None: return _r + # trap 'CMD' SIG: the quoted handler runs as shell code on EXIT / a signal; scan it. + if _ft == "trap" and _fi + 1 < len(ptoks): + _th = ptoks[_fi + 1] + if _th and _th != "-" and _th not in _READ_SCAN_SEPARATORS: + _r = _scan_command_string_for_reads( + _th, + strict_traversal = strict_traversal, + cwd = cwd, + cwd_dynamic = cwd_dynamic, + _depth = _depth + 1, + ) + if _r is not None: + return _r + + # A command substitution ($(...) / `...`) runs its payload as a shell command regardless of + # surrounding quotes, so `echo "$(head /etc/passwd)"` reads the file even though the outer + # command is not a reader and the tokenizer keeps the quoted substitution as one argument. + # Scan each substitution payload recursively, independent of the outer command word. + for _cs in _extract_command_subs(cmd): + if _cs.strip(): + _r = _scan_command_string_for_reads( + _cs, + strict_traversal = strict_traversal, + cwd = cwd, + cwd_dynamic = cwd_dynamic, + _depth = _depth + 1, + ) + if _r is not None: + return _r def _risky_read_target(tgt): if not tgt: @@ -8659,8 +8739,8 @@ for _n in _OS_MUTATORS1: # open-like backstop. Apply the same sensitive-read check to the directory path. A bare # call (cwd), in-workdir paths, and an fd argument (os.open already screens the fd's read) # stay allowed. -def _guard_dir_reader(name): - orig = getattr(_os, name, None) +def _guard_dir_reader(name, mod=_os): + orig = getattr(mod, name, None) if orig is None: return @_gwraps(orig) @@ -8673,7 +8753,7 @@ def _guard_dir_reader(name): p = _fspath1(path) _deny_sensitive_read(p) return orig(p, *a, **k) - setattr(_os, name, w) + setattr(mod, name, w) for _n in ("listdir", "scandir"): _guard_dir_reader(_n) @@ -8777,6 +8857,17 @@ def _reguard_created(m): _wrap1(m, _rn, _nm + "." + _rn) for _rn in ("rename", "renames", "replace", "link", "symlink"): _wrap2(m, _rn, True) + # A fresh posix/nt module also re-exposes the ORIGINAL fd metadata mutators and + # directory readers; reapply the same fd deniers + read confinement applied to the + # already-loaded module (else fresh fchmod(fd, ...) / fresh listdir('/root') slip). + if hasattr(m, "chdir"): + _wrap1(m, "chdir", _nm + ".chdir") + for _rn in ("fchmod", "fchown"): + if hasattr(m, _rn): + setattr(m, _rn, _make_fd_denier(_nm + "." + _rn, getattr(m, _rn))) + for _rn in ("listdir", "scandir"): + if hasattr(m, _rn): + _guard_dir_reader(_rn, m) elif _nm in ("_io", "io"): if hasattr(m, "open"): m.open = _guard_open_like(m.open) @@ -8848,6 +8939,12 @@ for _lowosname in ("posix", "nt"): for _n in ("fchmod", "fchown"): if hasattr(_lowos, _n): setattr(_lowos, _n, _make_fd_denier(_lowosname + "." + _n, getattr(_lowos, _n))) + # posix.listdir / posix.scandir re-export the ORIGINAL enumerators, so the os.* dir + # guard leaves them reachable (posix.listdir('/root')); apply the same sensitive-read + # confinement to the low-level module. + for _n in ("listdir", "scandir"): + if hasattr(_lowos, _n): + _guard_dir_reader(_n, _lowos) except Exception: pass diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index e1f654d1c5..6f1e18965e 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -1418,3 +1418,71 @@ def test_sandboxed_device_sink_str_subclass_escape_denied(tmp_path): out = _python_exec(code, None, 30, "backstop-devsink-subclass", disable_sandbox = False) assert "sandbox:" in out or "PermissionError" in out assert not target.exists() + + +# '/root' assembled from chr() codepoints so the static scanner cannot const-fold it, proving +# the RUNTIME guard on the low-level posix module (not the static layer). +_OPAQUE_ROOT = "P=''.join(chr(c) for c in [47,114,111,111,116])\n" + + +@_POSIX_ONLY +@pytest.mark.parametrize( + "reader", + [ + "import posix\nposix.listdir(P)", + "import posix\nlist(posix.scandir(P))", + ], +) +def test_sandboxed_low_level_posix_dir_read_denied(reader): + # posix.listdir / posix.scandir re-export the ORIGINAL enumerators, so the os.* dir guard + # left them reachable for an opaque sensitive path; the runtime guard now confines them too. + out = _python_exec(_OPAQUE_ROOT + reader, None, 30, "backstop-posixdir", disable_sandbox = False) + assert "sandbox:" in out or "PermissionError" in out + + +@_POSIX_ONLY +def test_sandboxed_fresh_posix_module_fd_denier_reapplied(): + # A fresh posix module built via _imp.create_builtin re-exposes the original fd metadata + # mutators; _reguard_created must reapply the fchmod/fchown deniers so a read-only open of an + # outside file cannot be reused to mutate host metadata. + code = ( + "import _imp, posix\n" + "m = _imp.create_builtin(posix.__spec__)\n" + "try:\n" + " fd = m.open('/etc/hostname', 0)\n" + " m.fchmod(fd, 0o777)\n" + " print('MUTATED')\n" + "except Exception as e:\n" + " print(repr(e))" + ) + out = _python_exec(code, None, 30, "backstop-freshfchmod", disable_sandbox = False) + assert "MUTATED" not in out + assert "sandbox:" in out or "PermissionError" in out + + +@_POSIX_ONLY +def test_sandboxed_fresh_posix_module_dir_read_denied(): + # The fresh posix module's directory readers are guarded too (opaque sensitive path). + code = ( + "import _imp, posix\n" + _OPAQUE_ROOT + "m = _imp.create_builtin(posix.__spec__)\n" + "try:\n" + " print(m.listdir(P))\n" + "except Exception as e:\n" + " print(repr(e))" + ) + out = _python_exec(code, None, 30, "backstop-freshlistdir", disable_sandbox = False) + assert "sandbox:" in out or "PermissionError" in out + + +@_POSIX_ONLY +def test_sandboxed_low_level_posix_workdir_read_allowed(): + # Enumerating the sandbox's own workdir through the low-level module stays allowed. + out = _python_exec( + "import posix, os\nos.makedirs('subd', exist_ok=True)\nprint('LS', posix.listdir('subd'))", + None, + 30, + "backstop-posixdir-ok", + disable_sandbox = False, + ) + assert "LS" 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 ec716ca420..336adabbde 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -3519,3 +3519,63 @@ class TestRound32Bypasses: ) def test_round32_benign_allowed(self, code): _ok(code) + + +class TestRound33Bypasses: + """Thirty-third-round Codex findings: quoted command substitutions unscanned when the outer + command is not a reader, a system-bin path escaped via .., and the flock wrapper / coproc + keyword / trap handler slipping past the command scan. (The low-level posix directory-reader + and fresh-module fd-denier gaps are covered in test_sandbox_runtime_backstop.)""" + + @pytest.mark.parametrize( + "code", + [ + # $()/backtick run regardless of quotes; the payload reads a host secret. + "import os\nos.system('echo \"$(head -1 /etc/passwd)\"')", + "import os\nos.system('echo `cat /etc/shadow`')", + "import subprocess\nsubprocess.run('printf %s \"$(cat /etc/passwd)\"', shell=True)", + ], + ) + def test_quoted_command_sub_read_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + # /usr/bin/../../evil must normalize before the system-bin exemption. + "import os\nos.system('/usr/bin/../../tmp/evil.sh')", + "import subprocess\nsubprocess.run(['/usr/bin/../../tmp/evil.sh'])", + ], + ) + def test_system_bin_dotdot_escape_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + # flock runs a command in an unguarded child; coproc / trap execute their operands. + "import os\nos.system('flock lockfile touch /tmp/escape')", + "import os\nos.system(\"flock /tmp/l -c 'rm -rf /'\")", + "import subprocess\nsubprocess.run(['flock', 'lock', 'touch', '/tmp/x'])", + "import os\nos.system('coproc touch /tmp/escape')", + "import os\nos.system('coproc rm -rf /')", + "import os\nos.system(\"trap 'touch /tmp/escape' EXIT\")", + "import os\nos.system(\"trap 'rm -rf /' EXIT\")", + ], + ) + def test_flock_coproc_trap_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + # Benign quoted subs (no read), trap reset, compound headers, scheduler view. + "import os\nos.system('echo \"$(date)\"')", + "import os\nos.system('echo $(ls data)')", + "import os\nos.system('trap - EXIT')", + "import os\nos.system('if ls data; then echo ok; fi')", + "import os\nos.system('chrt -p 1234')", + ], + ) + def test_round33_benign_allowed(self, code): + _ok(code)