Harden sandbox: aliased open modules, os shell from-imports, subprocess shell cwd, device sink writes
- Recognize an aliased open-module receiver (import builtins as b; b.open(...), import io as i;
i.open(...), import os as o; o.open(...)) as a read callee via a new _open_mod_aliases set, so
an aliased traversal / sensitive read is caught like the literal builtins/io/os.open forms.
- Record os shell aliases from `from os import system as s` / popen: the import-walk elif that
consumed the os module only recorded `open`, so the later shell-alias branch never saw it and
the read scan skipped s('cat /etc/passwd'). Handle os shell functions in that branch and split
the subprocess from-import handling into its own branch.
- Combine a subprocess cwd= with a shell payload's relative reads: subprocess.run('cat passwd',
shell=True, cwd='/etc') is resolved to /etc/passwd (the shared read scanner now takes a cwd
seed, overridable per-command by env -C), and a NON-literal cwd fails closed for a relative
reader.
- Allow a Python write to a standard device sink (/dev/null, /dev/stdout, ...) in the runtime
guard, checked on the requested path (not its realpath, so /dev/stdout is not followed to a
redirected outside file); benign output-suppression patterns are no longer denied.
Adds TestRound28Bypasses plus runtime device-sink write tests.
This commit is contained in:
danielhanchen
parent
e6d24369a9
commit
475c06c968
3 changed files with 147 additions and 16 deletions
|
|
@ -4371,6 +4371,8 @@ def _scan_command_string_for_reads(
|
|||
command,
|
||||
*,
|
||||
strict_traversal,
|
||||
cwd = None,
|
||||
cwd_dynamic = False,
|
||||
_depth = 0,
|
||||
):
|
||||
"""Scan a shell command STRING for an embedded host-secret read; return a short reason or
|
||||
|
|
@ -4475,7 +4477,9 @@ 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
|
||||
# The child's cwd for a relative read: seeded from an ambient cwd (a subprocess cwd=), and
|
||||
# overridable per-command by env -C DIR. Resets to the ambient cwd at each separator.
|
||||
_chdir = cwd
|
||||
_pending_chdir = False
|
||||
for _pi, _pt in enumerate(ptoks):
|
||||
if _pt in _READ_SCAN_SEPARATORS:
|
||||
|
|
@ -4483,7 +4487,7 @@ def _scan_command_string_for_reads(
|
|||
_cur_reader = False
|
||||
_wrapper = None
|
||||
_skip_operand = False
|
||||
_chdir = None
|
||||
_chdir = cwd
|
||||
_pending_chdir = False
|
||||
continue
|
||||
if _pt.startswith("<"):
|
||||
|
|
@ -4535,6 +4539,8 @@ def _scan_command_string_for_reads(
|
|||
_r = _scan_command_string_for_reads(
|
||||
ptoks[_k + 1],
|
||||
strict_traversal = strict_traversal,
|
||||
cwd = _chdir,
|
||||
cwd_dynamic = cwd_dynamic,
|
||||
_depth = _depth + 1,
|
||||
)
|
||||
if _r is not None:
|
||||
|
|
@ -4551,11 +4557,17 @@ def _scan_command_string_for_reads(
|
|||
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("~"):
|
||||
_rel = not _pt.startswith("/") and not _pt.startswith("~")
|
||||
# Under a known chdir (env -C DIR or an ambient subprocess cwd=), a relative reader
|
||||
# arg resolves against DIR (cat passwd + cwd=/etc -> /etc/passwd).
|
||||
if _chdir and _rel:
|
||||
_r = _flag(os.path.join(_chdir, _pt))
|
||||
if _r is not None:
|
||||
return _r
|
||||
# A relative reader arg under a NON-literal cwd cannot be proven sandbox-local, so
|
||||
# fail closed (subprocess.run('cat passwd', shell=True, cwd=P)).
|
||||
if cwd_dynamic and _chdir is None and _rel:
|
||||
return f"shell read command reads {_pt!r} under a non-literal cwd"
|
||||
return None
|
||||
|
||||
|
||||
|
|
@ -7261,6 +7273,10 @@ def _check_signal_escape_patterns(
|
|||
# the read-only os.open is deliberately allowed OUTSIDE the workdir by the runtime
|
||||
# guard, so a traversal read via such an alias must be caught statically.
|
||||
_open_from_aliases: set[str] = set()
|
||||
# Receiver-module aliases for the open() attribute form (import builtins as b; b.open(...),
|
||||
# import io as i; i.open(...), import os as o; o.open(...)), so an aliased-module read is
|
||||
# recognized like the literal builtins/io/os.open forms.
|
||||
_open_mod_aliases = {"builtins", "__builtins__", "io", "os"}
|
||||
# os/subprocess module aliases + from-import shell-name aliases, so a shell command
|
||||
# string that reads a host secret (os.system('cat /etc/passwd')) is scanned even when
|
||||
# os/subprocess is renamed.
|
||||
|
|
@ -7288,22 +7304,22 @@ def _check_signal_escape_patterns(
|
|||
for _a in _imp.names:
|
||||
if _a.name == "open":
|
||||
_open_from_aliases.add(_a.asname or "open")
|
||||
# `from os import system as s` / popen: record the os shell-exec alias here too
|
||||
# (this elif consumes the `os` module, so the subprocess branch below never sees
|
||||
# it), else _scan_shell_string_reads skips s('cat /etc/passwd').
|
||||
_fq = f"{_imp.module}.{_a.name}"
|
||||
if _fq in _SHELL_EXEC_FUNCS:
|
||||
_shell_name_aliases[_a.asname or _a.name] = _fq
|
||||
elif isinstance(_imp, ast.ImportFrom) and _imp.module == "shutil":
|
||||
for _a in _imp.names:
|
||||
if _a.name in _SHUTIL_COPY_METHODS:
|
||||
_shutil_copy_from_aliases.add(_a.asname or _a.name)
|
||||
elif isinstance(_imp, ast.ImportFrom) and _imp.module in ("os", "subprocess"):
|
||||
elif isinstance(_imp, ast.ImportFrom) and _imp.module == "subprocess":
|
||||
for _a in _imp.names:
|
||||
_fq = f"{_imp.module}.{_a.name}"
|
||||
_fq = f"subprocess.{_a.name}"
|
||||
if _fq in _SHELL_EXEC_FUNCS:
|
||||
_shell_name_aliases[_a.asname or _a.name] = _fq
|
||||
if _imp.module == "subprocess" and _a.name in (
|
||||
"run",
|
||||
"call",
|
||||
"check_call",
|
||||
"check_output",
|
||||
"Popen",
|
||||
):
|
||||
if _a.name in ("run", "call", "check_call", "check_output", "Popen"):
|
||||
_subprocess_exec_from_aliases.add(_a.asname or _a.name)
|
||||
elif isinstance(_imp, ast.ImportFrom) and _imp.module in (
|
||||
"os.path",
|
||||
|
|
@ -7325,10 +7341,13 @@ def _check_signal_escape_patterns(
|
|||
_operator_mod_aliases.add(_a.asname or "operator")
|
||||
elif _a.name == "os":
|
||||
_os_mod_aliases.add(_a.asname or "os")
|
||||
_open_mod_aliases.add(_a.asname or "os") # o.open(...)
|
||||
elif _a.name in ("posix", "nt"):
|
||||
# posix / nt are the os C backend (posix.system == os.system), so a shell
|
||||
# string passed to them must be scanned for embedded secret reads too.
|
||||
_os_mod_aliases.add(_a.asname or _a.name)
|
||||
elif _a.name in ("io", "builtins"):
|
||||
_open_mod_aliases.add(_a.asname or _a.name) # i.open(...) / b.open(...)
|
||||
elif _a.name == "subprocess":
|
||||
_subprocess_mod_aliases.add(_a.asname or "subprocess")
|
||||
|
||||
|
|
@ -7407,14 +7426,14 @@ def _check_signal_escape_patterns(
|
|||
isinstance(rhs, ast.Attribute)
|
||||
and rhs.attr == "open"
|
||||
and isinstance(rhs.value, ast.Name)
|
||||
and rhs.value.id in ("builtins", "__builtins__", "io", "os")
|
||||
and rhs.value.id in _open_mod_aliases
|
||||
):
|
||||
return True
|
||||
if (
|
||||
isinstance(fn, ast.Attribute)
|
||||
and fn.attr == "open"
|
||||
and isinstance(fn.value, ast.Name)
|
||||
and fn.value.id in ("builtins", "__builtins__", "io", "os")
|
||||
and fn.value.id in _open_mod_aliases
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
|
@ -7683,6 +7702,21 @@ def _check_signal_escape_patterns(
|
|||
return _kw.value
|
||||
return None
|
||||
|
||||
# subprocess.run('cat passwd', shell=True, cwd='/etc') runs the payload in an unguarded
|
||||
# shell whose cwd is /etc, so a relative reader arg reads /etc/passwd; a NON-literal cwd
|
||||
# cannot be proven sandbox-local. Extract cwd= once and thread it into the payload scan.
|
||||
_cwd_lit = None
|
||||
_cwd_dyn = False
|
||||
if _is_subprocess_exec_callee(f):
|
||||
for _kw in node.keywords or []:
|
||||
if _kw.arg == "cwd":
|
||||
_cv = _fold_read_arg(_kw.value)
|
||||
if isinstance(_cv, str):
|
||||
_cwd_lit = _cv
|
||||
elif not (isinstance(_kw.value, ast.Constant) and _kw.value.value is None):
|
||||
_cwd_dyn = True
|
||||
break
|
||||
|
||||
def _scan_one_command(cmd):
|
||||
# Scan a shell command STRING (folded to a literal) for an embedded host-secret
|
||||
# read and record a violation. Delegates to the shared scanner in strict-traversal
|
||||
|
|
@ -7690,7 +7724,9 @@ def _check_signal_escape_patterns(
|
|||
# resolves the reader past assignment / wrapper prefixes and recurses nested shells.
|
||||
if cmd is None:
|
||||
return False
|
||||
_r = _scan_command_string_for_reads(cmd, strict_traversal = True)
|
||||
_r = _scan_command_string_for_reads(
|
||||
cmd, strict_traversal = True, cwd = _cwd_lit, cwd_dynamic = _cwd_dyn
|
||||
)
|
||||
if _r is not None:
|
||||
_fs_block(node, _r)
|
||||
return True
|
||||
|
|
@ -8050,11 +8086,22 @@ _stat = _os.stat
|
|||
_stat_mod = _os.path.stat
|
||||
_S_ISLNK = _stat_mod.S_ISLNK
|
||||
_WD = _realpath(__WORKDIR__)
|
||||
# Standard device sinks cannot persist data outside the workspace, so a write to one is
|
||||
# allowed (mirrors the terminal shell redirect allowlist); benign patterns like
|
||||
# open('/dev/null', 'w') to suppress output would otherwise be denied by the workdir check.
|
||||
_SAFE_DEV_SINKS = frozenset(
|
||||
{"/dev/null", "/dev/zero", "/dev/full", "/dev/stdout", "/dev/stderr", "/dev/tty"}
|
||||
)
|
||||
|
||||
def _within(p):
|
||||
try:
|
||||
if isinstance(p, int):
|
||||
return True
|
||||
# Allow a write to an exact device sink. Checked on the REQUESTED path, not its
|
||||
# realpath, so /dev/stdout is not followed to a redirected outside file.
|
||||
_ps = p if isinstance(p, str) else (_fsdecode(p) if isinstance(p, (bytes, bytearray)) else None)
|
||||
if _ps is not None and _ps.replace("\\", "/") in _SAFE_DEV_SINKS:
|
||||
return True
|
||||
# os.path.realpath internally calls the LIVE os.fspath (posixpath.realpath does
|
||||
# `filename = os.fspath(filename)`) and os.lstat / os.readlink / os.getcwd, so a
|
||||
# sandboxed reassignment of any of them would poison the resolution even though we
|
||||
|
|
|
|||
|
|
@ -1371,3 +1371,32 @@ def test_sandboxed_pathlib_glob_sensitive_dir_denied(meth):
|
|||
assert "sandbox:" in out or "PermissionError" in out
|
||||
finally:
|
||||
os.remove(link)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
@pytest.mark.parametrize("path", ["/dev/null", "/dev/stdout", "/dev/stderr"])
|
||||
def test_sandboxed_device_sink_write_allowed(path):
|
||||
# A write to a standard device sink cannot persist data outside the workspace, so it is
|
||||
# allowed (mirrors the terminal redirect allowlist) rather than denied by the workdir check.
|
||||
out = _python_exec(
|
||||
f"open({path!r}, 'w').write('x'); print('WROTE_SINK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-devsink",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "WROTE_SINK" in out
|
||||
assert "sandbox:" not in out
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_os_devnull_write_allowed():
|
||||
out = _python_exec(
|
||||
"import os\nopen(os.devnull, 'w').write('x'); print('WROTE_OSDEVNULL')",
|
||||
None,
|
||||
30,
|
||||
"backstop-osdevnull",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "WROTE_OSDEVNULL" in out
|
||||
assert "sandbox:" not in out
|
||||
|
|
|
|||
|
|
@ -3126,3 +3126,58 @@ class TestRound27Bypasses:
|
|||
)
|
||||
def test_round27_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
||||
|
||||
class TestRound28Bypasses:
|
||||
"""Twenty-eighth-round Codex findings: aliased open-module receivers, os shell from-import
|
||||
aliases dropped by an exclusive elif, and a subprocess shell payload not combined with a
|
||||
literal / dynamic cwd. (The device-sink write FP is covered in test_sandbox_runtime_backstop.)"""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import builtins as b\nb.open('../../../etc/passwd').read()",
|
||||
"import io as i\ni.open('../../../etc/passwd').read()",
|
||||
"import os as o\no.open('../../../etc/passwd', 0)",
|
||||
"import builtins as b\nb.open('/etc/passwd').read()",
|
||||
],
|
||||
)
|
||||
def test_aliased_open_module_read_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"from os import system as s\ns('cat /etc/passwd')",
|
||||
"from os import popen as p\np('head -1 /etc/shadow')",
|
||||
],
|
||||
)
|
||||
def test_os_shell_from_import_alias_read_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import subprocess\nsubprocess.run('cat passwd', shell=True, cwd='/etc')",
|
||||
"import subprocess\nsubprocess.run(['sh', '-c', 'cat passwd'], cwd='/etc')",
|
||||
"import subprocess\nsubprocess.run('cat passwd', shell=True, cwd=P)",
|
||||
"import subprocess\nsubprocess.check_output('cat sshd_config', shell=True, cwd='/etc/ssh')",
|
||||
],
|
||||
)
|
||||
def test_subprocess_shell_cwd_read_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# Benign aliased-open / from-import / shell-cwd forms must still pass.
|
||||
"import builtins as b\nb.open('data.txt').read()",
|
||||
"from os import getcwd as g\nprint(g())",
|
||||
"from subprocess import run as r\nr(['echo', 'hi'])",
|
||||
"import subprocess\nsubprocess.run('cat notes.txt', shell=True, cwd='logs')",
|
||||
"import subprocess\nsubprocess.run('echo hi', shell=True, cwd=P)",
|
||||
"import subprocess\nsubprocess.run('echo hi', shell=True)",
|
||||
],
|
||||
)
|
||||
def test_round28_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue