ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Harden sandbox: history writes, cwd-relative reads, unpacking aliases, pin guard builtins/stat, root-home reads

Browse source 
- Block bash's history builtin when it reads/writes a file (history -w / -a / -r / -n): it can
  create or overwrite an arbitrary host path (or read a file into the buffer) in the unguarded
  shell child. Bare history / -c / -d / -p / -s stay allowed.
- Combine a subprocess cwd= with relative argv paths in the sensitive-read scan, so
  subprocess.run(['cat', 'passwd'], cwd='/etc') is seen as a /etc/passwd read.
- Track env -C DIR / --chdir DIR in the shell-string read scan so a later relative reader
  argument (env -C /etc cat passwd) resolves against DIR.
- Record aliases created by tuple/list unpacking assignments ((s,) = (os.system,); a, b =
  os.system, 1; [e] = [exec]) in the scope alias index, pairing a literal target with a literal
  RHS element-wise, so the shell/exec/deserializer sink checks see them.
- Pin the builtins the runtime path guard consults (isinstance / int / bytes / str / any) into
  the guard namespace, so sandboxed code cannot reassign builtins.isinstance to make
  isinstance(path, int) treat an outside path as an fd and approve an absolute write.
- Re-pin os.path.stat + S_ISLNK before each realpath resolution in the guard, so a
  os.path.stat.S_ISLNK = lambda mode: False (stopping realpath from following an in-workdir
  symlink that escapes) cannot approve a write the real open() then routes outside.
- Restore the /root/ protection in the runtime sensitive-read backstop for an opaque path,
  carving out package / library trees (site-packages, dist-packages, the stdlib) so imports
  under a root home are not broken.

Adds TestRound26Bypasses plus runtime tests for the pinned builtins / stat and root-home reads.
This commit is contained in:
danielhanchen 2026-07-10 05:12:39 +00:00
commit fbc67fd490
3 changed files with 239 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

118
studio/backend/core/inference/tools.py
View file

@ -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)

63
studio/backend/tests/test_sandbox_runtime_backstop.py
View file

@ -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

68
studio/backend/tests/test_sandbox_tools.py
View file

@ -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)