Harden sandbox: dynamic/environb/update PATH mutations; sqlite URI decode + shell/pipe dot-commands; getattr gadget dunders; find -exec in argv

Close seven bypasses Codex found on the round-50 branch.

- dynamic PATH assignment: os.environ['PATH'] = '.:' + os.environ['PATH'] (or an
  f-string) was accepted because the value is non-literal. Fold what we can and fail
  closed when a COMPLETE, fully-literal PATH entry the value contributes is a
  relative / cwd / empty entry; a dynamic ABSOLUTE extension ('/usr/local/bin:' +
  $PATH, venv + ':' + $PATH) stays allowed.

- os.environb mutations: os.environb[b'PATH'] = b'.:...' updates the same inherited
  environment, but only os.environ[...] was recognized. Match environ / environb
  (attribute and bare) and decode a bytes key / value before the policy check.

- os.environ.update / setdefault: a mapping mutator
  (os.environ.update({'PATH': '.:...'}), .update(PATH=...), .setdefault('PATH', ...))
  never hit the subscript check. Run each (key, value) pair through the mutation
  policy in visit_Call.

- sqlite URI percent-decode: sqlite3.connect('file:%2Ftmp%2Fescape.db', uri=True)
  passed the runtime guard as a relative-looking string while SQLite decodes the
  filename and opens /tmp/escape.db. Percent-decode the URI path (with the guard's
  captured chr/int) before the workdir check.

- sqlite shell / pipe dot-commands: the CLI scanner only path-checked file
  dot-commands, but .shell CMD / .system CMD run a system shell and .output |CMD
  opens a pipe. Block a .shell / .system / .excel dot-command and an .output/.once
  target that begins with '|'.

- getattr gadget dunders in the workdir vetter: a helper module could call
  getattr(open, '__closure__') / getattr(cell, 'cell_contents') to recover the guard
  wrapper's original unguarded open, because the getattr branch only rejected a few
  sensitive receivers. Reject a gadget-dunder name on ANY receiver (mirrors the
  direct-attribute check).

- find -exec in subprocess argv: the read scanner flattened the argv and checked
  each element independently, missing subprocess.run(['find','/etc',...,'-exec',
  'cat','{}',';']) reading /etc/passwd (the {} placeholder loses the escaping search
  root). Reconstruct a find child-exec argv into a shell string and run it through
  the read scanner, which carries the find-root + -exec logic.

Regression coverage: TestRound51Bypasses in tests/test_sandbox_tools.py (dynamic /
environb / update PATH mutations, sqlite .shell/.system/.output-pipe, find -exec
argv, plus a round51 benign-allowed set: absolute dynamic PATH, benign env vars,
local sqlite .output/.dump, workdir find -exec) and, in
tests/test_sandbox_runtime_backstop.py, the sqlite percent-encoded URI escape (with
a benign local URI) and the getattr gadget-dunder workdir-module denial.
This commit is contained in:
danielhanchen 2026-07-10 21:49:54 +00:00
commit 81bb65f4e0
3 changed files with 300 additions and 15 deletions

View file

@ -301,6 +301,10 @@ _SQLITE_DOTFILE_RE = re.compile(
r"(?m)^\s*\.(?:output|once|excel|import|backup|save|dump|clone|log|read)\b\s+(?:-{1,2}\S+\s+)*"
r"(?P<f>(?:'[^']*'|\"[^\"]*\"|\S+))"
)
# sqlite3 dot-commands that RUN a system shell command in the unguarded child: `.shell CMD` /
# `.system CMD` ("Run CMD ARGS... in a system shell"), and `.excel` (opens the result in a
# system program). These execute regardless of any path check, so match the command itself.
_SQLITE_SHELL_RE = re.compile(r"(?m)^\s*\.(?:shell|system|excel)\b")
# sqlite3 CLI options that consume a SEPARATED operand (so the value after them is NOT the
# database filename). Only -init also reads a file (its value is path-checked at the call site).
_SQLITE_OPERAND_OPTS = frozenset(
@ -467,6 +471,58 @@ def _path_value_is_unsafe(value: str, assignments = None) -> bool:
return False
def _dynamic_path_value_unsafe(value_node, env) -> bool:
"""A NON-literal PATH assignment value (os.environ['PATH'] = '.:' + os.environ['PATH'],
f'.:{x}') is unsafe when a COMPLETE, fully-literal PATH entry it contributes is a relative /
cwd / empty entry. Operands are const-folded; an OPAQUE segment (os.environ['PATH'], a
variable) taints only the entry that spans it, so a dynamic ABSOLUTE extension
(venv + ':' + $PATH, '/usr/local/bin:' + $PATH) stays allowed. Returns True only for a
provable unsafe entry -- the folded literal case is handled by the caller."""
segments: list = [] # ("lit", str) or ("opaque",)
def _flatten(n):
folded = _const_fold(n, env)
if isinstance(folded, str):
segments.append(("lit", folded))
return
if isinstance(n, ast.BinOp) and isinstance(n.op, ast.Add):
_flatten(n.left)
_flatten(n.right)
return
if isinstance(n, ast.JoinedStr):
for _p in n.values:
if isinstance(_p, ast.Constant) and isinstance(_p.value, str):
segments.append(("lit", _p.value))
else:
_fv = _const_fold(getattr(_p, "value", _p), env)
segments.append(("lit", _fv) if isinstance(_fv, str) else ("opaque",))
return
segments.append(("opaque",))
_flatten(value_node)
entries: list = [] # (text, complete, tainted)
cur = ""
tainted = False
for seg in segments:
if seg[0] == "opaque":
tainted = True
continue
parts = seg[1].split(":")
for j, part in enumerate(parts):
if j == 0:
cur += part
else:
entries.append((cur, True, tainted))
cur = part
tainted = False
_last_opaque = bool(segments) and segments[-1][0] == "opaque"
entries.append((cur, not _last_opaque, tainted))
for text, complete, taint in entries:
if complete and not taint and _path_value_is_unsafe(text):
return True
return False
def _arg_escapes_workdir(tok: str) -> bool:
"""True when a path-like argument can point OUTSIDE the session workdir: an absolute path
(``/tmp/x``), a ``~`` / ``~user`` home path (home == workdir, but a shell child follows the
@ -1928,11 +1984,17 @@ def _find_blocked_commands(command: str) -> set[str]:
_unq = t
if len(_unq) >= 2 and _unq[0] == _unq[-1] and _unq[0] in ("'", '"'):
_unq = _unq[1:-1]
# .shell CMD / .system CMD run an arbitrary command in the unguarded child shell.
if _SQLITE_SHELL_RE.search(_unq):
blocked.add("sqlite3-shell")
for _m in _SQLITE_DOTFILE_RE.finditer(_unq):
_dot_f = _m.group("f")
if len(_dot_f) >= 2 and _dot_f[0] == _dot_f[-1] and _dot_f[0] in ("'", '"'):
_dot_f = _dot_f[1:-1]
if _dot_f not in ("stdout", "stderr", "off") and _git_operand_escapes(
# .output |CMD / .once |CMD open CMD as a PIPE (a shell command), not a file.
if _dot_f.startswith("|"):
blocked.add("sqlite3-shell")
elif _dot_f not in ("stdout", "stderr", "off") and _git_operand_escapes(
_dot_f, _local_assigns
):
blocked.add("sqlite3-write-outside")
@ -6609,6 +6671,19 @@ def _check_signal_escape_patterns(
return node.value
return None
def _extract_env_scalar(node):
"""A str constant, or a bytes constant decoded to str (os.environb byte keys / values are
the same inherited environment as os.environ), else None."""
if isinstance(node, ast.Constant):
if isinstance(node.value, str):
return node.value
if isinstance(node.value, (bytes, bytearray)):
try:
return bytes(node.value).decode("utf-8", "surrogateescape")
except Exception:
return None
return None
def _extract_strings_from_list(node):
"""Extract string elements from an AST List or Tuple node."""
if isinstance(node, (ast.List, ast.Tuple)):
@ -7419,30 +7494,39 @@ def _check_signal_escape_patterns(
return True
return False
def _environ_subscript_key(self, target):
# The literal key of an os.environ[...] (or a bare `environ[...]` from
# `from os import environ`) subscript assignment target; None otherwise.
if not isinstance(target, ast.Subscript):
return None
_v = target.value
_is_environ = (
def _is_environ_receiver(self, _v):
# os.environ / os.environb (or a bare `environ` / `environb` from `from os import ...`).
# environb is the SAME inherited process environment via byte keys/values.
return (
isinstance(_v, ast.Attribute)
and _v.attr == "environ"
and _v.attr in ("environ", "environb")
and isinstance(_v.value, ast.Name)
and _v.value.id in self.os_aliases
) or (isinstance(_v, ast.Name) and _v.id == "environ")
if not _is_environ:
) or (isinstance(_v, ast.Name) and _v.id in ("environ", "environb"))
def _environ_subscript_key(self, target):
# The literal key of an os.environ[...] / os.environb[...] (or a bare environ[...] /
# environb[...]) subscript assignment target; None otherwise. A bytes key is decoded.
if not isinstance(target, ast.Subscript):
return None
return _extract_string_from_node(target.slice)
if not self._is_environ_receiver(target.value):
return None
return _extract_env_scalar(target.slice)
def _env_mutation_escape(self, key, value_node):
# A short reason when setting env var ``key`` to ``value_node`` is a child-escape
# prelude (mirrors the subprocess env={...} mapping analysis), else None. The mutated
# process environment is inherited by a later unguarded child.
_vs = _extract_string_from_node(value_node)
_vs = _extract_env_scalar(value_node)
if key == "PATH":
if isinstance(_vs, str) and _path_value_is_unsafe(_vs):
return "PATH set to a relative / cwd entry (a bare argv resolves to a workdir exec)"
if isinstance(_vs, str):
if _path_value_is_unsafe(_vs):
return "PATH set to a relative / cwd entry (a bare argv resolves to a workdir exec)"
return None
# A non-literal PATH value that provably prepends / embeds a relative / cwd entry
# ('.:' + os.environ['PATH'], f'.:{x}'); a dynamic ABSOLUTE extension stays allowed.
if _dynamic_path_value_unsafe(value_node, _const_env):
return "PATH prepends a relative / cwd entry (dynamic value)"
return None
if key in ("BASH_ENV", "ENV"):
return None if _vs == "" else "a shell startup file a child shell sources"
@ -7517,6 +7601,40 @@ def _check_signal_escape_patterns(
if _mc_rewrite is not None:
self.visit_Call(_mc_rewrite)
return
# os.environ.update({'PATH': '.:...'}) / .update(PATH='...') / .setdefault('PATH', ...)
# (and the os.environb byte forms) mutate the inherited environment WITHOUT a subscript
# assignment, the same child escape as os.environ['PATH'] = ...; run each (key, value)
# pair through the mutation policy.
_mf = node.func
if isinstance(_mf, ast.Attribute) and _mf.attr in ("update", "setdefault"):
if self._is_environ_receiver(_mf.value):
_pairs = []
if _mf.attr == "setdefault" and len(node.args) >= 2:
_sk = _extract_env_scalar(node.args[0])
if _sk is not None:
_pairs.append((_sk, node.args[1]))
elif _mf.attr == "update":
if node.args and isinstance(node.args[0], ast.Dict):
for _kn, _vn in zip(node.args[0].keys, node.args[0].values):
if _kn is not None:
_dk = _extract_env_scalar(_kn)
if _dk is not None:
_pairs.append((_dk, _vn))
for _kw in node.keywords:
if _kw.arg is not None:
_pairs.append((_kw.arg, _kw.value))
for _pk, _pvn in _pairs:
_preason = self._env_mutation_escape(_pk, _pvn)
if _preason is not None:
shell_escapes.append(
{
"type": "shell_escape",
"line": getattr(node, "lineno", -1),
"description": (
f"os.environ.{_mf.attr}({_pk!r}) mutation: {_preason}"
),
}
)
if self._is_unbound_mro_gadget(node):
# type.mro(io.FileIO) / type.__getattribute__(io.FileIO, '__mro__') /
# getattr(io.FileIO, 'mro'): reaches the unguarded MRO without a .mro / .__mro__
@ -10171,6 +10289,31 @@ def _check_signal_escape_patterns(
break
if _argv0:
return
# A `find` child-exec argv (subprocess.run(['find','/etc','-name','passwd','-exec',
# 'cat','{}',';'])) reads host files the flat per-element scan misses: the {} placeholder
# loses the escaping search root. Reconstruct the argv from the find command word into a
# shell string and run it through the read scanner, which carries the find-root + -exec
# logic. Only when every reconstructed element folds to a literal (else best-effort skip).
if _is_child_exec:
_fargv = node.args[0] if node.args else None
if _fargv is None:
for _name, _val in _iter_call_kwargs(node):
if _name == "args":
_fargv = _val
break
if isinstance(_fargv, (ast.List, ast.Tuple)):
_ffolded = [_fold_read_arg(_e) for _e in _fargv.elts]
_fci = _argv_command_word_index(_ffolded)
if (
_fci is not None
and isinstance(_ffolded[_fci], str)
and os.path.basename(_ffolded[_fci]).lower() == "find"
and all(isinstance(_x, str) for _x in _ffolded[_fci:])
):
_freason = _command_reads_sensitive(shlex.join(_ffolded[_fci:]))
if _freason is not None:
_fs_block(node, f"find child-exec reads a host file ({_freason})")
return
# 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)
@ -10910,6 +11053,15 @@ try:
if _pth.startswith("//"):
_slash = _pth.find("/", 2)
_pth = _pth[_slash:] if _slash != -1 else ""
# SQLite percent-decodes the URI filename (file:%2Ftmp%2Fx -> /tmp/x), so decode
# BEFORE the workdir check -- otherwise an encoded absolute path passes _within()
# as a relative-looking string while SQLite opens the escaping path. Use the
# captured _bi.chr / _bi.int so a sandboxed rebind of chr/int cannot skew the decode.
_pth = _re.sub(
"%([0-9A-Fa-f]{2})",
lambda _m: _bi.chr(_bi.int(_m.group(1), 16)),
_pth,
)
return _within(_pth)
return _within(_db)
@ -11256,6 +11408,13 @@ try:
if _grecv in _obf:
return True
else:
# An introspection / frame gadget dunder via getattr reaches an escape on
# ANY receiver -- getattr(open, '__closure__'), getattr(cell,
# 'cell_contents') recover the guard wrapper's original unguarded open --
# so reject the gadget name regardless of receiver (mirrors the direct
# attribute check below).
if _gname in _GUARD_GADGET_ATTRS:
return True
if _grecv in _recv and _gname in _GUARD_EXEC_ATTRS:
return True
if _grecv in _bi and _gname in ("eval", "exec", "compile", "__import__"):

View file

@ -295,6 +295,50 @@ def test_sandboxed_low_level_sqlite3_connect_escape_denied(tmp_path):
assert not target.exists()
@_POSIX_ONLY
def test_sandboxed_sqlite3_uri_percent_encoded_escape_denied(tmp_path):
# SQLite percent-decodes the URI filename, so an encoded absolute path (file:%2Ftmp%2Fx,
# uri=True) must be decoded before the workdir check or it slips through as relative-looking.
target = tmp_path / "uri_escape.db"
enc = str(target).replace("/", "%2F")
out = _python_exec(
f"import sqlite3; sqlite3.connect('file:{enc}', uri=True); print('OPENED')",
None,
30,
"backstop-sqlite-uri-escape",
disable_sandbox = False,
)
assert "sandbox:" in out or "PermissionError" in out
assert not target.exists()
def test_sandboxed_sqlite3_uri_local_allowed():
# A workdir-local file: URI (no escaping percent-decode) still opens.
out = _python_exec(
"import sqlite3\n"
"c = sqlite3.connect('file:uri_local.db', uri=True)\n"
"c.execute('create table if not exists t(x)'); c.close(); print('URI_OK')",
None,
30,
"backstop-sqlite-uri-local",
disable_sandbox = False,
)
assert "URI_OK" in out
assert "sandbox:" not in out
@_POSIX_ONLY
def test_sandboxed_getattr_gadget_dunder_workdir_module_denied():
# A workdir helper recovering the guard wrapper's original open via a getattr gadget dunder
# (getattr(open, '__closure__')) must be refused by the vetter, like the direct attribute form.
_assert_workdir_module_denied(
"backstop-workdir-getattr-gadget",
"gadgetdunder",
"C = getattr(open, '__closure__')\nprint('GADGET_RAN')\n",
"GADGET_RAN",
)
@_POSIX_ONLY
def test_sandboxed_os_rename_dir_fd_denied(tmp_path):
# os.rename / os.replace with src_dir_fd / dst_dir_fd is fd-relative; a string

View file

@ -5000,3 +5000,85 @@ class TestRound50Bypasses:
)
def test_round50_benign_allowed(self, code):
_ok(code)
class TestRound51Bypasses:
"""Fifty-first-round Codex findings: dynamic / os.environb / os.environ.update PATH mutations,
sqlite shell + pipe dot-commands, and find -exec over an escaping root in a subprocess argv.
(The sqlite URI percent-decode and the getattr gadget-dunder helper are runtime concerns,
covered in test_sandbox_runtime_backstop.py.)"""
@pytest.mark.parametrize(
"code",
[
# A non-literal PATH value that prepends a relative / cwd entry.
"import os, subprocess\nos.environ['PATH'] = '.:' + os.environ['PATH']\nsubprocess.run(['evil'])",
"import os, subprocess\nos.environ['PATH'] = f'.:{os.environ[\"PATH\"]}'\nsubprocess.run(['evil'])",
],
)
def test_dynamic_path_mutation_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
# os.environb byte-key mutations are the same inherited environment as os.environ.
"import os, subprocess\nos.environb[b'PATH'] = b'.:/usr/bin'\nsubprocess.run(['evil'])",
"import os, subprocess\nos.environb[b'BASH_ENV'] = b'e.sh'\nsubprocess.run(['bash','-c','echo hi'])",
],
)
def test_environb_mutation_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
# os.environ.update(...) / .setdefault(...) mapping mutators reach PATH too.
"import os, subprocess\nos.environ.update({'PATH': '.:/usr/bin'})\nsubprocess.run(['evil'])",
"import os, subprocess\nos.environ.update(PATH='.:/usr/bin')\nsubprocess.run(['evil'])",
"import os, subprocess\nos.environ.setdefault('PATH', '.:/usr/bin')\nsubprocess.run(['evil'])",
],
)
def test_environ_update_mutation_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
# sqlite .shell / .system run a system shell; .output |CMD opens a pipe (a shell cmd).
"import os\nos.system(\"sqlite3 ':memory:' '.shell touch /tmp/x'\")",
"import os\nos.system(\"sqlite3 ':memory:' '.system rm -rf /'\")",
"import os\nos.system(\"sqlite3 local.db '.output |touch /tmp/x' 'select 1;'\")",
],
)
def test_sqlite_shell_dotcommand_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
# find -exec reader over an escaping root, expressed as a subprocess argv vector.
"import subprocess\nsubprocess.run(['find','/etc','-maxdepth','1','-name','passwd','-exec','cat','{}',';'])",
"import subprocess\nsubprocess.run(['find','/','-name','id_rsa','-exec','head','{}',';'])",
],
)
def test_find_exec_argv_escaping_root_blocked(self, code):
assert _check_code_safety(code) is not None, code
@pytest.mark.parametrize(
"code",
[
# A dynamic ABSOLUTE PATH extension, a benign environb / update var, a local sqlite
# .output / .dump, and a workdir-scoped find -exec argv all stay allowed.
"import os, subprocess\nos.environ['PATH'] = '/usr/local/bin:' + os.environ['PATH']\nsubprocess.run(['ls'])",
"import os, subprocess\nos.environ['MYVAR'] = 'a' + 'b'\nsubprocess.run(['ls'])",
"import os, subprocess\nos.environb[b'MYVAR'] = b'x'\nsubprocess.run(['ls'])",
"import os, subprocess\nos.environ.update({'MYVAR': 'x'})\nsubprocess.run(['ls'])",
"import os, subprocess\nos.environ.update({'PATH': '/usr/bin:/bin'})\nsubprocess.run(['ls'])",
"import os\nos.system(\"sqlite3 ':memory:' '.output out.txt' 'select 1;'\")",
"import os\nos.system(\"sqlite3 local.db '.dump'\")",
"import subprocess\nsubprocess.run(['find','.','-name','*.py','-exec','cat','{}',';'])",
],
)
def test_round51_benign_allowed(self, code):
_ok(code)