diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 8e3019357b..fb2219d5e5 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1086,7 +1086,18 @@ def _find_blocked_commands(command: str) -> set[str]: for i in _cmd_word_idx: tok = tokens[i] _base = _token_basename(tok) - if _base not in ("sed", "gsed", "ssed", "perl", "sort", "find", "dd", "tee", "truncate"): + if _base not in ( + "sed", + "gsed", + "ssed", + "perl", + "sort", + "find", + "dd", + "tee", + "truncate", + "history", + ): continue if _base == "truncate": blocked.add("mutating:truncate") @@ -1124,6 +1135,14 @@ def _find_blocked_commands(command: str) -> set[str]: elif _base == "tee" and not a.startswith("-"): blocked.add("mutating:tee") break + elif _base == "history" and _short and any(_c in al[1:] for _c in "warn"): + # bash's history builtin reads/writes an arbitrary file: `history -w FILE` + # (or -a append) creates/overwrites an absolute host path, and `-r` / `-n` + # read a file into the history buffer. Even without a FILE operand it targets + # $HISTFILE, which the caller can point outside the workdir. -c / -d / -p / -s + # do not touch a file, so only w / a / r / n are blocked. + blocked.add("mutating:history") + break return blocked @@ -3790,6 +3809,22 @@ def _build_scope_alias_index(tree, const_env): and isinstance(n.target, ast.Name) ): assigns.append((n.target.id, n.value)) + elif ( + # A parallel unpacking assignment binds each name to the matching RHS element + # ((s,) = (os.system,); [e] = [exec]; a, b = os.system, 1), which then reaches + # a sink call the same way a plain alias does. Pair a literal tuple/list target + # with a literal tuple/list RHS of equal length element-wise so those aliases + # are recorded; a starred / mismatched / non-literal RHS is left alone. + isinstance(n, ast.Assign) + and len(n.targets) == 1 + and isinstance(n.targets[0], (ast.Tuple, ast.List)) + and isinstance(n.value, (ast.Tuple, ast.List)) + and len(n.targets[0].elts) == len(n.value.elts) + and not any(isinstance(_t, ast.Starred) for _t in n.targets[0].elts) + ): + for _tgt, _val in zip(n.targets[0].elts, n.value.elts): + if isinstance(_tgt, ast.Name): + assigns.append((_tgt.id, _val)) # A comprehension generator binds its target like a single-assignment alias when the # iterable is a one-element literal: [e(p) for e in [exec]] binds e to exec, so the # payload passed through e must still get eval/exec recursion. @@ -4307,12 +4342,16 @@ def _scan_command_string_for_reads( _cur_reader = False _wrapper = None _skip_operand = False + _chdir = None # env -C DIR / --chdir DIR sets the child's cwd for later relative reads + _pending_chdir = False for _pi, _pt in enumerate(ptoks): if _pt in _READ_SCAN_SEPARATORS: _at_cmd = True _cur_reader = False _wrapper = None _skip_operand = False + _chdir = None + _pending_chdir = False continue if _pt.startswith("<"): _rt = _pt.lstrip("<") or (ptoks[_pi + 1] if _pi + 1 < len(ptoks) else "") @@ -4323,6 +4362,9 @@ def _scan_command_string_for_reads( continue # output redirects are handled by _find_blocked_commands 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 = _pt + _pending_chdir = False _skip_operand = False continue if _ASSIGNMENT_RE.match(_pt): @@ -4331,7 +4373,15 @@ def _scan_command_string_for_reads( return _r continue # assignment prefix; the command word is still ahead if _pt.startswith("-"): - if _wrapper and _wrapper_flag_takes_operand(_wrapper, _pt): + # env -C DIR / --chdir DIR changes the child's cwd before the command runs, so + # a later relative reader arg (env -C /etc cat passwd -> /etc/passwd) resolves + # against DIR, not the workdir. Capture DIR instead of just skipping it. + if _wrapper == "env" and _pt in ("-C", "--chdir"): + _pending_chdir = True + _skip_operand = True + elif _wrapper == "env" and _pt.startswith("--chdir="): + _chdir = _pt.split("=", 1)[1] + elif _wrapper and _wrapper_flag_takes_operand(_wrapper, _pt): _skip_operand = True continue # wrapper flag; still before the command word _base = os.path.basename(_pt).lower() @@ -4365,12 +4415,14 @@ def _scan_command_string_for_reads( _at_cmd = False _wrapper = None continue - if ( - _cur_reader - and not _pt.startswith("-") - and ("$" in _pt or "`" in _pt or _escaping_glob(_pt)) - ): - return f"shell read command reads an expanded path {_pt!r}" + if _cur_reader and not _pt.startswith("-"): + if "$" in _pt or "`" in _pt or _escaping_glob(_pt): + return f"shell read command reads an expanded path {_pt!r}" + # Under an env -C DIR chdir, a relative reader arg resolves against DIR. + if _chdir and not _pt.startswith("/") and not _pt.startswith("~"): + _r = _flag(os.path.join(_chdir, _pt)) + if _r is not None: + return _r return None @@ -7559,16 +7611,27 @@ def _check_signal_escape_patterns( # A shell-command STRING sink: scan the command for embedded sensitive reads. if _scan_shell_string_reads(node, f): return + _is_child_exec = _is_subprocess_exec_callee(f) or _is_exec_family_callee(f) is_read_callee = ( _resolves_to_open(f) or fq in ("io.open", "os.open") or fq in _SHUTIL_COPY_SINKS or _is_shutil_copy_callee(f) or (isinstance(f, ast.Name) and f.id in _shutil_copy_from_aliases) - or _is_subprocess_exec_callee(f) - or _is_exec_family_callee(f) + or _is_child_exec or method in _READ_METHODS ) + # subprocess.run(['cat', 'passwd'], cwd='/etc') reads /etc/passwd in an unguarded + # child: the argv entry is relative and /etc alone is not sensitive, so combine a + # literal cwd= with each relative argv path before the sensitivity check. + _sub_cwd = None + if _is_child_exec: + for kw in node.keywords or []: + if kw.arg == "cwd": + _cv = _fold_read_arg(kw.value) + if isinstance(_cv, str): + _sub_cwd = _cv + break # Pathlib read on a Path(...) / join receiver: check the resolved path. if isinstance(f, ast.Attribute) and f.attr in _PATHLIB_READ_METHODS: rp = _pathlib_receiver_path(f.value) @@ -7617,6 +7680,11 @@ def _check_signal_escape_patterns( continue if _flag_read_path(node, s, is_read_callee): break + # Resolve a relative argv entry against a literal subprocess cwd= (cat passwd + # + cwd='/etc' -> /etc/passwd) so the combined host-secret read is caught. + if _sub_cwd is not None and not s.startswith("/") and not s.startswith("~"): + if _flag_read_path(node, os.path.join(_sub_cwd, s), is_read_callee): + break self.generic_visit(node) NetworkAndIoVisitor().visit(tree) @@ -7759,6 +7827,16 @@ import sys as _sys _saved_path = list(_sys.path) _sys.path = [_p for _p in _sys.path if _p not in ("", ".", __WORKDIR__, __WORKDIR__ + "/")] import os as _os, builtins as _bi, io as _io, pathlib as _pl, re as _re +# Pin the builtins the guard predicates consult (isinstance / int / bytes / str / any) into +# THIS namespace so a sandboxed `builtins.isinstance = lambda *a: True` (etc.) cannot make a +# guard check lie -- e.g. isinstance(path, int) treating an outside path as an fd and +# approving an absolute write. Every guard function below resolves these names from here, not +# the mutable builtins module. +isinstance = _bi.isinstance +int = _bi.int +bytes = _bi.bytes +str = _bi.str +any = _bi.any # NOTE: sys.path stays stripped for the WHOLE guard setup below (it also imports shutil, # which is pure-Python and equally shadowable); it is restored at the very END of this # prelude, just before user code runs, so ordinary user imports still resolve. @@ -7786,6 +7864,13 @@ _lstat = _os.lstat _readlink = _os.readlink _getcwd = _os.getcwd _stat = _os.stat +# posixpath.realpath decides whether to FOLLOW a component by calling os.path.stat.S_ISLNK +# on the live stat module. Sandboxed code can set os.path.stat.S_ISLNK = lambda mode: False +# (or reassign os.path.stat) so realpath stops following an in-workdir symlink that escapes, +# leaving the target under _WD while the real open() follows it outside. Capture the module + +# S_ISLNK so both can be re-pinned before each resolution. +_stat_mod = _os.path.stat +_S_ISLNK = _stat_mod.S_ISLNK _WD = _realpath(__WORKDIR__) def _within(p): @@ -7805,6 +7890,8 @@ def _within(p): _os.readlink = _readlink _os.getcwd = _getcwd _os.stat = _stat + _os.path.stat = _stat_mod + _stat_mod.S_ISLNK = _S_ISLNK rp = _realpath(_fspath(p)) # A bytes path resolves to bytes; normalize to str so the prefix compare against # the str _WD does not raise (which would deny a legitimate in-workdir bytes write @@ -7883,6 +7970,15 @@ def _is_sensitive_read(rp): return True if _SENS_PROC.match(n): return True + # Dotfiles / caches under a root home hold credentials (/root/.bashrc, /root/.cache/...); + # an opaque path the static /root/ rule cannot fold could read them at runtime. Restore + # the /root/ protection here, but carve out package / library trees so importing a library + # installed under a root home (site-packages, the stdlib) is not broken. + if n.startswith("/root/") and not any( + _seg in n + for _seg in ("/site-packages/", "/dist-packages/", "/lib/python", "/lib64/python") + ): + return True low = n.lower() return any(tok in low for tok in _SENS_TOKENS) @@ -7896,6 +7992,8 @@ def _read_realpath(p): _os.readlink = _readlink _os.getcwd = _getcwd _os.stat = _stat + _os.path.stat = _stat_mod + _stat_mod.S_ISLNK = _S_ISLNK rp = _realpath(_fspath(p)) if isinstance(rp, bytes): rp = _fsdecode(rp) diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index a47c87b26b..f7fec2bba4 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -15,6 +15,7 @@ if str(_BACKEND_ROOT) not in sys.path: from core.inference.tools import ( _BLOCKED_COMMANDS_COMMON, + _SANDBOX_GUARD_SRC, _bash_exec, _command_reads_sensitive, _python_exec, @@ -1266,3 +1267,65 @@ def test_bash_brace_expanded_writer_blocked(command): ) def test_bash_benign_prefixed_read_scan_allows(command): assert _command_reads_sensitive(command) is None, command + + +@_POSIX_ONLY +def test_sandboxed_poisoned_isinstance_write_denied(tmp_path): + # Reassigning builtins.isinstance must not make the guard's isinstance(path, int) fd check + # lie and approve an absolute write outside the workdir; the guard uses pinned builtins. + target = tmp_path / "poison_isinstance.txt" + out = _python_exec( + "import builtins\n" + "builtins.isinstance = lambda *a, **k: True\n" + f"open({str(target)!r}, 'w').write('x'); print('WROTE')\n", + None, + 30, + "backstop-poison-isinstance", + disable_sandbox = False, + ) + assert "sandbox:" in out or "PermissionError" in out + assert not target.exists() + + +@_POSIX_ONLY +def test_sandboxed_poisoned_s_islnk_symlink_write_denied(tmp_path): + # Reassigning os.path.stat.S_ISLNK so realpath stops following an in-workdir symlink that + # escapes must not let the write through; the guard re-pins S_ISLNK before each resolve. + session = "backstop-poison-islnk" + workdir = get_sandbox_workdir(session) + link = os.path.join(workdir, "islnk_escape") + if os.path.islink(link) or os.path.exists(link): + os.remove(link) + os.symlink(str(tmp_path), link) + victim = tmp_path / "poison_islnk.txt" + try: + out = _python_exec( + "import os.path\n" + "os.path.stat.S_ISLNK = lambda mode: False\n" + "open('islnk_escape/poison_islnk.txt', 'w').write('x'); print('WROTE')\n", + None, + 30, + session, + disable_sandbox = False, + ) + assert "sandbox:" in out or "PermissionError" in out + assert not victim.exists() + finally: + os.remove(link) + + +def test_runtime_is_sensitive_read_covers_root_home(): + # The runtime backstop's _is_sensitive_read must protect /root dotfiles/caches while + # carving out package/library trees so imports under a root home are not broken. + import re as _re + + src = _SANDBOX_GUARD_SRC + ns = {"_re": _re} + block = src[src.index("_SENS_EXACT = ") : src.index("def _read_realpath")] + exec(block, ns) + f = ns["_is_sensitive_read"] + assert f("/root/.bashrc") is True + assert f("/root/.cache/secret") is True + assert f("/root/.local/lib/python3.13/site-packages/certifi/cacert.pem") is False + assert f("/root/miniconda3/lib/python3.13/os.py") is False + assert f("/home/ubuntu/project/data.txt") is False diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index af9c370299..b0a42ecf3a 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -3000,3 +3000,71 @@ class TestRound25Bypasses: ) def test_round25_benign_allowed(self, code): _ok(code) + + +class TestRound26Bypasses: + """Twenty-sixth-round Codex findings (static portion): bash history file writes, subprocess + cwd + relative argv reads, env -C chdir before a relative read, and tuple/list unpacking + aliases. (The runtime-guard items -- pinned builtins / stat and /root reads -- are covered + in test_sandbox_runtime_backstop.)""" + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('history -s x; history -w /tmp/p')", + "import os\nos.system('history -r /etc/passwd')", + "import os\nos.system('history -a /tmp/p')", + ], + ) + def test_history_file_write_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import subprocess\nsubprocess.run(['cat', 'passwd'], cwd='/etc')", + "import subprocess\nsubprocess.run(['cat', 'shadow'], cwd='/etc')", + "import subprocess\nsubprocess.Popen(['cat', 'sshd_config'], cwd='/etc/ssh')", + ], + ) + def test_subprocess_cwd_relative_read_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('env -C /etc cat passwd')", + "import os\nos.system('env --chdir /etc head -1 passwd')", + "import os\nos.system('env --chdir=/etc cat passwd')", + ], + ) + def test_env_chdir_relative_read_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import os\n(s,) = (os.system,)\ns('touch /tmp/p')", + "import os\na, b = os.system, 1\na('rm -rf /tmp/x')", + "import os\ns, t = os.system, os.popen\nt('touch /tmp/x')", + "[e] = [exec]\ne('__import__(chr(111)+chr(115))')", + "import pickle\n(l,) = (pickle.loads,)\nl(b'x')", + ], + ) + def test_unpacking_alias_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + # Benign history / subprocess-cwd / env -C / unpacking forms must still pass. + "import os\nos.system('history -c')", + "import subprocess\nsubprocess.run(['cat', 'data.txt'], cwd='logs')", + "import os\nos.system('env -C build make')", + "import os\nos.system('env -C /app cat readme.md')", + "a, b = 1, 2\nprint(a + b)", + "a, b = 3, 4\na, b = b, a\nprint(a)", + ], + ) + def test_round26_benign_allowed(self, code): + _ok(code)