Studio sandbox: close seventh-round review bypasses (scope counts, obfuscation, child writers, reads)
Static classifier:
- fix scope-local walker so a nested def/class reassigning an alias name no longer inflates the outer single-assignment count and drops a real module-level sink alias
- track non-bare compile aliases (builtins.compile, from builtins import compile as comp) for the types.FunctionType(c) code-object gadget
- block child-process file writers at shell command position (touch/tee/cp/mv/mkdir/install/truncate/mkfifo/mknod/shred/unlink): a spawned child runs without the in-process write guard
- expand a literal **{...} unpack in the read scanner so open(**{'file': '../../etc/passwd'}) is resolved
- resolve a pathlib expression bound to a single-assignment name before read methods (p = Path('..')/'etc'/'passwd'; p.read_text())
- keep a wrapper's separated option argument in command position so stdbuf -o L python -c ... still detects the interpreter (env -i rm still caught; no FP on grep patterns)
- treat object.__getattribute__ / type.__getattribute__ as attribute obfuscation, covering gadget dunders and sensitive-module attrs (also closes __closure__ recovery of a guarded wrapper's original callable)
- block runpy.run_path / runpy.run_module execution sinks
- treat shutil.copy*/move SOURCE as a read callee so a .. traversal source is caught
Runtime backstop:
- normalize a bytes realpath (fsdecode) before the workdir prefix compare so a legitimate in-workdir bytes write is not denied by a TypeError; outside bytes writes still denied
This commit is contained in:
danielhanchen
parent
0f4b4b3d36
commit
bf4bc7e449
3 changed files with 353 additions and 38 deletions
|
|
@ -138,7 +138,28 @@ _INTERPRETER_COMMANDS = frozenset(
|
|||
"rscript",
|
||||
}
|
||||
)
|
||||
_BLOCKED_COMMANDS_COMMON = _BLOCKED_COMMANDS_COMMON | _INTERPRETER_COMMANDS
|
||||
# File-creating / writing coreutils. Same rationale as the interpreters: a spawned child
|
||||
# runs without the in-process realpath backstop, so subprocess.run(['touch', '/tmp/x']),
|
||||
# tee, cp, mv, ... write / create / delete outside the session workdir. In-workdir file
|
||||
# work should go through the guarded Python file APIs. (dd / ln / rm are already denied
|
||||
# above.) Native / unknown binaries the sandbox cannot enumerate remain an OS-isolation
|
||||
# residual.
|
||||
_CHILD_WRITE_COMMANDS = frozenset(
|
||||
{
|
||||
"touch",
|
||||
"tee",
|
||||
"cp",
|
||||
"mv",
|
||||
"mkdir",
|
||||
"install",
|
||||
"truncate",
|
||||
"mkfifo",
|
||||
"mknod",
|
||||
"shred",
|
||||
"unlink",
|
||||
}
|
||||
)
|
||||
_BLOCKED_COMMANDS_COMMON = _BLOCKED_COMMANDS_COMMON | _INTERPRETER_COMMANDS | _CHILD_WRITE_COMMANDS
|
||||
_BLOCKED_COMMANDS_WIN = frozenset(
|
||||
{
|
||||
"rmdir",
|
||||
|
|
@ -244,16 +265,20 @@ def _find_blocked_commands(command: str) -> set[str]:
|
|||
|
||||
expect_command = True # start of string is a command position
|
||||
prefix_pending = False # last cmd-position token was a wrapper (env/time/xargs/...)
|
||||
prev_was_flag = False # previous token (while a wrapper is pending) was an option flag
|
||||
for token in tokens:
|
||||
if token in _SHELL_SEPARATORS or token in _SHELL_KEYWORDS_AS_SEP:
|
||||
expect_command = True
|
||||
prefix_pending = False
|
||||
prev_was_flag = False
|
||||
continue
|
||||
if token.startswith("-"):
|
||||
# Flags belong to the active command, but keep expect_command while a
|
||||
# wrapper prefix awaits its command (`stdbuf -oL cmd`, `xargs -- cmd`).
|
||||
if not prefix_pending:
|
||||
expect_command = False
|
||||
else:
|
||||
prev_was_flag = True
|
||||
continue
|
||||
if not expect_command:
|
||||
continue
|
||||
|
|
@ -266,8 +291,23 @@ def _find_blocked_commands(command: str) -> set[str]:
|
|||
# accepting a numeric-looking token is safe (we only skip, never stop scanning),
|
||||
# whereas the old int-only check let `timeout 5m rm -rf /` slip through.
|
||||
if prefix_pending and _is_wrapper_numeric_arg(token):
|
||||
prev_was_flag = False
|
||||
continue
|
||||
base = _token_basename(token)
|
||||
# A wrapper's separated option ARGUMENT (`stdbuf -o L cmd`, `ionice -c 2 cmd`):
|
||||
# an operand right after a wrapper flag that is NOT itself a blocked command /
|
||||
# prefix is the flag's value, so skip it and keep scanning for the real command
|
||||
# instead of mistaking it for the command and stopping. If it IS a blocked
|
||||
# command / prefix it is treated as the command below (never miss `env -i rm`).
|
||||
if (
|
||||
prefix_pending
|
||||
and prev_was_flag
|
||||
and base not in _BLOCKED_COMMANDS
|
||||
and base not in _COMMAND_PREFIXES
|
||||
):
|
||||
prev_was_flag = False
|
||||
continue
|
||||
prev_was_flag = False
|
||||
if base in _BLOCKED_COMMANDS:
|
||||
blocked.add(base)
|
||||
# Wrappers (env/time/xargs/sudo) consume one command; the next non-flag,
|
||||
|
|
@ -2517,10 +2557,16 @@ def _walk_scope_local(scope):
|
|||
)
|
||||
while stack:
|
||||
n = stack.pop()
|
||||
# A nested def / lambda / class / comprehension opens its OWN scope: its body
|
||||
# neither shares this namespace nor should its stores be counted here. Skip it
|
||||
# entirely -- do not yield it or descend into it. (Checking the popped node,
|
||||
# not just its children, is what keeps a nested `def f(): s = print` from
|
||||
# inflating the outer count of an `s = os.system` single-assignment alias.)
|
||||
if isinstance(n, _NESTED):
|
||||
continue
|
||||
yield n
|
||||
for child in ast.iter_child_nodes(n):
|
||||
if not isinstance(child, _NESTED):
|
||||
stack.append(child)
|
||||
stack.append(child)
|
||||
|
||||
|
||||
class _ScopeAliasIndex:
|
||||
|
|
@ -2541,6 +2587,7 @@ class _ScopeAliasIndex:
|
|||
"impf",
|
||||
"deser",
|
||||
"strconst",
|
||||
"rhsnode",
|
||||
"assigned",
|
||||
)
|
||||
|
||||
|
|
@ -2557,6 +2604,7 @@ class _ScopeAliasIndex:
|
|||
self.impf: dict = {} # name -> True: alias of __import__ / importlib.import_module
|
||||
self.deser: dict = {} # name -> fq deserializer sink (pickle.loads, ...)
|
||||
self.strconst: dict = {} # name -> folded str/bytes constant (for read scanning)
|
||||
self.rhsnode: dict = {} # name -> single-assignment RHS node (for pathlib reads)
|
||||
self.assigned: dict = {}
|
||||
|
||||
def _chain(self, node):
|
||||
|
|
@ -2608,6 +2656,8 @@ def _build_scope_alias_index(tree, const_env):
|
|||
from_aliases: dict[str, str] = {}
|
||||
builtins_aliases = {"builtins", "__builtins__"}
|
||||
importlib_aliases = {"importlib"}
|
||||
# `compile` bound by name (bare builtin or `from builtins import compile as comp`).
|
||||
compile_aliases = {"compile"}
|
||||
deser_module_aliases: dict[str, str] = {m: m for m in _DESERIALIZE_MODULES}
|
||||
for n in ast.walk(tree):
|
||||
if isinstance(n, ast.Import):
|
||||
|
|
@ -2627,6 +2677,28 @@ def _build_scope_alias_index(tree, const_env):
|
|||
fq = f"{n.module}.{a.name}"
|
||||
if fq in _SHELL_SINK_FUNCS:
|
||||
from_aliases[a.asname or a.name] = fq
|
||||
elif isinstance(n, ast.ImportFrom) and n.module == "builtins":
|
||||
for a in n.names:
|
||||
if a.name == "compile":
|
||||
compile_aliases.add(a.asname or "compile")
|
||||
|
||||
def _rhs_is_compile_call(rhs):
|
||||
# `compile(...)` reached as the bare builtin, `builtins.compile(...)`, or a
|
||||
# `from builtins import compile as comp` alias -- the callee forms that produce a
|
||||
# code object bound to a name (for the types.FunctionType(c) execution gadget).
|
||||
if not isinstance(rhs, ast.Call):
|
||||
return False
|
||||
f = rhs.func
|
||||
if isinstance(f, ast.Name):
|
||||
return f.id in compile_aliases
|
||||
if (
|
||||
isinstance(f, ast.Attribute)
|
||||
and f.attr == "compile"
|
||||
and isinstance(f.value, ast.Name)
|
||||
and f.value.id in builtins_aliases
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _rhs_exec_builtin(rhs):
|
||||
# bare `exec` / `eval` / `compile`, or `builtins.eval` (attribute form).
|
||||
|
|
@ -2701,23 +2773,23 @@ def _build_scope_alias_index(tree, const_env):
|
|||
imap: dict[str, bool] = {}
|
||||
dmap: dict[str, str] = {}
|
||||
scmap: dict[str, object] = {}
|
||||
rnmap: dict[str, ast.expr] = {}
|
||||
for name, rhs in assigns:
|
||||
if counts.get(name) != 1 or name in rebound:
|
||||
continue
|
||||
# Single-assignment RHS node, used by the read scanner to resolve a pathlib
|
||||
# expression bound to a name (p = Path('..') / 'etc' / 'passwd'; p.read_text()).
|
||||
rnmap[name] = rhs
|
||||
fq = _resolve_static_shell_sink(rhs, os_aliases, subprocess_aliases, from_aliases)
|
||||
if fq:
|
||||
smap[name] = fq
|
||||
eb = _rhs_exec_builtin(rhs)
|
||||
if eb is not None:
|
||||
emap[name] = eb
|
||||
elif (
|
||||
isinstance(rhs, ast.Call)
|
||||
and isinstance(rhs.func, ast.Name)
|
||||
and rhs.func.id == "compile"
|
||||
and rhs.args
|
||||
):
|
||||
# Any `c = compile(...)` binds a code object, tracked for the
|
||||
# types.FunctionType(c) execution gadget below (dynamic or foldable).
|
||||
elif _rhs_is_compile_call(rhs) and rhs.args:
|
||||
# Any `c = compile(...)` (bare / builtins.compile / from-import alias)
|
||||
# binds a code object, tracked for the types.FunctionType(c) execution
|
||||
# gadget below (dynamic or foldable payload).
|
||||
camap[name] = True
|
||||
v = _const_fold(rhs.args[0], const_env)
|
||||
if isinstance(v, (str, bytes, bytearray)):
|
||||
|
|
@ -2750,6 +2822,8 @@ def _build_scope_alias_index(tree, const_env):
|
|||
idx.deser[scope] = dmap
|
||||
if scmap:
|
||||
idx.strconst[scope] = scmap
|
||||
if rnmap:
|
||||
idx.rhsnode[scope] = rnmap
|
||||
return idx
|
||||
|
||||
|
||||
|
|
@ -3300,6 +3374,10 @@ def _check_signal_escape_patterns(
|
|||
# dynamic compile() result reaches execution through it (see visit_Call).
|
||||
self.types_aliases = {"types"}
|
||||
self.functiontype_aliases: set[str] = set()
|
||||
# import runpy as r -> {"runpy", "r"}. runpy.run_path/run_module execute a
|
||||
# file/module in the guarded interpreter without the recursive source
|
||||
# analysis exec/eval receive, so treat those calls as execution sinks.
|
||||
self.runpy_aliases = {"runpy"}
|
||||
self.loop_depth = 0
|
||||
|
||||
def visit_Import(self, node):
|
||||
|
|
@ -3320,6 +3398,8 @@ def _check_signal_escape_patterns(
|
|||
self.builtins_aliases.add(alias.asname or "builtins")
|
||||
elif alias.name == "types":
|
||||
self.types_aliases.add(alias.asname or "types")
|
||||
elif alias.name == "runpy":
|
||||
self.runpy_aliases.add(alias.asname or "runpy")
|
||||
if alias.name in _DESERIALIZE_MODULES:
|
||||
self.deserialize_module_aliases[alias.asname or alias.name] = alias.name
|
||||
self.generic_visit(node)
|
||||
|
|
@ -3633,6 +3713,25 @@ def _check_signal_escape_patterns(
|
|||
)
|
||||
else:
|
||||
dynamic_desc = None
|
||||
# An attribute-access call whose (receiver, attr-name) pair is the same
|
||||
# obfuscation as getattr(): the builtin getattr/setattr, or the dunder
|
||||
# forms object.__getattribute__(obj, 'name') / type.__getattribute__(...)
|
||||
# / obj.__getattr__('name') that fetch an attribute without matching the
|
||||
# bare getattr name. Normalized here so the gadget + sensitive-module
|
||||
# checks below cover all of them.
|
||||
_attr_call = None
|
||||
if (
|
||||
isinstance(func, ast.Name)
|
||||
and func.id in ("getattr", "setattr")
|
||||
and len(node.args) >= 2
|
||||
):
|
||||
_attr_call = (node.args[0], node.args[1])
|
||||
elif (
|
||||
isinstance(func, ast.Attribute)
|
||||
and func.attr in ("__getattribute__", "__getattr__")
|
||||
and len(node.args) >= 2
|
||||
):
|
||||
_attr_call = (node.args[0], node.args[1])
|
||||
is_dynamic_import = (
|
||||
_ast_name_matches(func, _DYNAMIC_IMPORT_FUNCS)
|
||||
or (
|
||||
|
|
@ -3706,19 +3805,19 @@ def _check_signal_escape_patterns(
|
|||
# payload even though a plain `import pickle` is benign.
|
||||
dynamic_desc = "dynamic import of a computed or sensitive module name"
|
||||
elif (
|
||||
isinstance(func, ast.Name)
|
||||
and func.id in ("getattr", "setattr")
|
||||
and len(node.args) >= 2
|
||||
and isinstance(_const_fold(node.args[1], _const_env), str)
|
||||
and _const_fold(node.args[1], _const_env) in _GADGET_DUNDERS
|
||||
_attr_call is not None
|
||||
and isinstance(_const_fold(_attr_call[1], _const_env), str)
|
||||
and _const_fold(_attr_call[1], _const_env) in _GADGET_DUNDERS
|
||||
):
|
||||
# getattr(anything, '__globals__' / '__subclasses__' / ...) reaches an
|
||||
# introspection gadget with no ast.Attribute for visit_Attribute to catch.
|
||||
# Direct x.__globals__ is already flagged for ANY receiver, so flag the
|
||||
# getattr-string form regardless of receiver too.
|
||||
# getattr(anything, '__globals__' / '__subclasses__' / ...) or the
|
||||
# object.__getattribute__ equivalent reaches an introspection gadget
|
||||
# with no ast.Attribute for visit_Attribute to catch. Direct
|
||||
# x.__globals__ is already flagged for ANY receiver, so flag the
|
||||
# dynamic-attr-name form regardless of receiver too. (Also closes the
|
||||
# __closure__ recovery of a guarded wrapper's original callable.)
|
||||
dynamic_desc = (
|
||||
"getattr() of an introspection gadget dunder "
|
||||
f"({_const_fold(node.args[1], _const_env)})"
|
||||
"dynamic attribute access of an introspection gadget dunder "
|
||||
f"({_const_fold(_attr_call[1], _const_env)})"
|
||||
)
|
||||
elif (
|
||||
isinstance(func, ast.Name)
|
||||
|
|
@ -3729,29 +3828,29 @@ def _check_signal_escape_patterns(
|
|||
# vars(os) / vars(__builtins__) returns the module __dict__, the same
|
||||
# obfuscation as os.__dict__['system'] but without the attribute access.
|
||||
dynamic_desc = "vars() on a sensitive module (dict obfuscation)"
|
||||
elif (
|
||||
isinstance(func, ast.Name)
|
||||
and func.id in ("getattr", "setattr")
|
||||
and node.args
|
||||
and _ast_name_matches(node.args[0], self._attr_obfuscation_targets())
|
||||
elif _attr_call is not None and _ast_name_matches(
|
||||
_attr_call[0], self._attr_obfuscation_targets()
|
||||
):
|
||||
# Stage 2 refinement: a benign constant attr (getattr(os, "getpid"))
|
||||
# is allowed; only a dynamic attr or a dangerous constant attr blocks.
|
||||
if _analyzer_on and len(node.args) >= 2:
|
||||
attr_val = _const_fold(node.args[1], _const_env)
|
||||
# Covers getattr/setattr and object.__getattribute__(builtins, 'eval').
|
||||
if _analyzer_on:
|
||||
attr_val = _const_fold(_attr_call[1], _const_env)
|
||||
if isinstance(attr_val, str):
|
||||
if attr_val in _DANGEROUS_ATTR_NAMES:
|
||||
dynamic_desc = (
|
||||
f"{func.id}() on a sensitive module "
|
||||
"dynamic attribute access on a sensitive module "
|
||||
"(attribute-name obfuscation)"
|
||||
)
|
||||
else:
|
||||
dynamic_desc = (
|
||||
f"{func.id}() on a sensitive module (attribute-name obfuscation)"
|
||||
"dynamic attribute access on a sensitive module "
|
||||
"(attribute-name obfuscation)"
|
||||
)
|
||||
else:
|
||||
dynamic_desc = (
|
||||
f"{func.id}() on a sensitive module (attribute-name obfuscation)"
|
||||
"dynamic attribute access on a sensitive module "
|
||||
"(attribute-name obfuscation)"
|
||||
)
|
||||
elif (
|
||||
# sys.modules.get('os') -- the .get() twin of sys.modules['os'].
|
||||
|
|
@ -3806,6 +3905,18 @@ def _check_signal_escape_patterns(
|
|||
"types.FunctionType() executes a compile() code object "
|
||||
"(bypasses the eval/exec gate)"
|
||||
)
|
||||
elif (
|
||||
# runpy.run_path('evil.py') / runpy.run_module('evil') execute a
|
||||
# file/module in the guarded interpreter WITHOUT the recursive source
|
||||
# analysis exec/eval receive, so a sandboxed snippet can write a local
|
||||
# evil.py and run it. Treat these as direct execution sinks.
|
||||
isinstance(func, ast.Attribute)
|
||||
and func.attr in ("run_path", "run_module")
|
||||
and _ast_name_matches(func.value, self.runpy_aliases)
|
||||
):
|
||||
dynamic_desc = (
|
||||
f"runpy.{func.attr}() executes a file/module without static analysis"
|
||||
)
|
||||
if dynamic_desc:
|
||||
dynamic_exec.append(
|
||||
{
|
||||
|
|
@ -4540,6 +4651,16 @@ def _check_signal_escape_patterns(
|
|||
"WindowsPath",
|
||||
"PureWindowsPath",
|
||||
)
|
||||
# shutil.copy*/move read their SOURCE (first arg) from the host, so a `..` traversal
|
||||
# or ~ source copies a host secret into the workdir even though it is not an open()/
|
||||
# read callee. Treat them as read callees so the traversal/sensitive check applies.
|
||||
_SHUTIL_COPY_SINKS = (
|
||||
"shutil.copy",
|
||||
"shutil.copy2",
|
||||
"shutil.copyfile",
|
||||
"shutil.copytree",
|
||||
"shutil.move",
|
||||
)
|
||||
|
||||
def _fold_read_arg(arg):
|
||||
# Fold a read-path argument to a concrete string, resolving a module-level
|
||||
|
|
@ -4554,12 +4675,24 @@ def _check_signal_escape_patterns(
|
|||
return _to_text(sv)
|
||||
return None
|
||||
|
||||
def _pathlib_receiver_path(recv):
|
||||
def _pathlib_receiver_path(recv, _seen = None):
|
||||
# Resolve a pathlib receiver to a concrete path: Path(...) / pathlib.Path(...)
|
||||
# (all constructor args joined), a `/` join (Path('/etc') / 'passwd'), or a
|
||||
# .joinpath(...) chain.
|
||||
# (all constructor args joined), a `/` join (Path('/etc') / 'passwd'), a
|
||||
# .joinpath(...) chain, or a single-assignment name bound to any of these
|
||||
# (p = Path('..') / 'etc' / 'passwd'; p.read_text()).
|
||||
if isinstance(recv, ast.Name):
|
||||
# Resolve the name to its single-assignment RHS (cycle-guarded).
|
||||
if _seen is None:
|
||||
_seen = set()
|
||||
if recv.id in _seen:
|
||||
return None
|
||||
_seen.add(recv.id)
|
||||
rhs = _scope_idx.resolve(recv.id, recv, "rhsnode")
|
||||
if rhs is None:
|
||||
return None
|
||||
return _pathlib_receiver_path(rhs, _seen)
|
||||
if isinstance(recv, ast.BinOp) and isinstance(recv.op, ast.Div):
|
||||
base = _pathlib_receiver_path(recv.left)
|
||||
base = _pathlib_receiver_path(recv.left, _seen)
|
||||
rv = _fold_read_arg(recv.right)
|
||||
if base is None or rv is None:
|
||||
return None
|
||||
|
|
@ -4571,7 +4704,7 @@ def _check_signal_escape_patterns(
|
|||
return None
|
||||
rf = recv.func
|
||||
if isinstance(rf, ast.Attribute) and rf.attr == "joinpath":
|
||||
base = _pathlib_receiver_path(rf.value)
|
||||
base = _pathlib_receiver_path(rf.value, _seen)
|
||||
if base is None:
|
||||
return None
|
||||
parts = [base]
|
||||
|
|
@ -4631,6 +4764,7 @@ def _check_signal_escape_patterns(
|
|||
is_read_callee = (
|
||||
(isinstance(f, ast.Name) and f.id == "open")
|
||||
or fq in ("io.open", "os.open")
|
||||
or fq in _SHUTIL_COPY_SINKS
|
||||
or method in _READ_METHODS
|
||||
)
|
||||
# Pathlib read on a Path(...) / join receiver: check the resolved path.
|
||||
|
|
@ -4638,7 +4772,16 @@ def _check_signal_escape_patterns(
|
|||
rp = _pathlib_receiver_path(f.value)
|
||||
if rp is not None and _flag_read_path(node, rp, True):
|
||||
return
|
||||
for arg in list(node.args) + [kw.value for kw in (node.keywords or [])]:
|
||||
# Build the arg list, expanding a literal **{...} unpack so its path value is
|
||||
# scanned (open(**{'file': '../../etc/passwd'}) reads the same file that
|
||||
# open('../../etc/passwd') would, which is otherwise treated as opaque).
|
||||
scan_args = list(node.args)
|
||||
for kw in node.keywords or []:
|
||||
if kw.arg is None and isinstance(kw.value, ast.Dict):
|
||||
scan_args.extend(v for v in kw.value.values if v is not None)
|
||||
else:
|
||||
scan_args.append(kw.value)
|
||||
for arg in scan_args:
|
||||
s = _fold_read_arg(arg)
|
||||
if s is None:
|
||||
# A pathlib expression carries no foldable string constant
|
||||
|
|
@ -4796,6 +4939,7 @@ import os as _os, builtins as _bi, io as _io, pathlib as _pl
|
|||
# rebound by mutating the os module.
|
||||
_realpath = _os.path.realpath
|
||||
_fspath = _os.fspath
|
||||
_fsdecode = _os.fsdecode
|
||||
_sep = _os.sep
|
||||
_WD = _realpath(__WORKDIR__)
|
||||
|
||||
|
|
@ -4812,6 +4956,11 @@ def _within(p):
|
|||
# not route through os.fspath, so restoring it has no effect on the write itself.)
|
||||
_os.fspath = _fspath
|
||||
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
|
||||
# such as open(b'local.txt', 'w')).
|
||||
if isinstance(rp, bytes):
|
||||
rp = _fsdecode(rp)
|
||||
except Exception:
|
||||
return False
|
||||
return rp == _WD or rp.startswith(_WD + _sep)
|
||||
|
|
|
|||
|
|
@ -657,3 +657,54 @@ def test_sandboxed_fspath_monkeypatch_write_escape_denied(tmp_path):
|
|||
)
|
||||
assert "sandbox:" in out or "PermissionError" in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_bytes_path_in_workdir_write_allowed():
|
||||
# A bytes path resolves to bytes from os.path.realpath; the guard must normalize it
|
||||
# (fsdecode) so a legitimate in-workdir bytes write is not denied by a str/bytes
|
||||
# prefix-compare TypeError.
|
||||
out = _python_exec(
|
||||
"f = open(b'bytes_local.txt', 'w'); f.write('hi'); f.close(); print('BYTES_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-bytes-path",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "BYTES_OK" in out
|
||||
assert "sandbox:" not in out
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_bytes_path_out_of_workdir_write_denied(tmp_path):
|
||||
# The bytes-path normalization must not weaken confinement: an outside bytes write
|
||||
# is still denied.
|
||||
target = tmp_path / "bytes_escape.txt"
|
||||
out = _python_exec(
|
||||
f"open({bytes(str(target), 'utf-8')!r}, 'w').write('x'); print('WROTE')",
|
||||
None,
|
||||
30,
|
||||
"backstop-bytes-escape",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "sandbox:" in out or "PermissionError" in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_closure_recovery_of_open_blocked():
|
||||
# object.__getattribute__(builtins.open, '__closure__') recovers the original
|
||||
# unguarded open from the wrapper closure. The static gate now blocks the
|
||||
# introspection (gadget dunder via __getattribute__), so it never runs.
|
||||
out = _python_exec(
|
||||
"import builtins\n"
|
||||
"object.__getattribute__(builtins.open, '__closure__')[0].cell_contents"
|
||||
"('/tmp/studio_closure_escape.txt', 'w').write('x')\n"
|
||||
"print('CLOSURE_WROTE')\n",
|
||||
None,
|
||||
30,
|
||||
"backstop-closure",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "unsafe code detected" in out or "sandbox:" in out or "PermissionError" in out
|
||||
assert not os.path.exists("/tmp/studio_closure_escape.txt")
|
||||
|
|
|
|||
|
|
@ -1219,3 +1219,118 @@ class TestRound6Bypasses:
|
|||
_ok("import os\nos.system('echo hello')")
|
||||
_ok("import os\nos.system('ls -la')")
|
||||
_ok("import subprocess\nsubprocess.run(['echo', 'hi'])")
|
||||
|
||||
|
||||
class TestRound7Bypasses:
|
||||
"""Seventh-round Codex findings: nested-scope alias counting, non-bare compile
|
||||
aliases, child-process writers, literal **kwargs reads, assigned pathlib reads,
|
||||
wrapper option arguments, object.__getattribute__ obfuscation, runpy sinks, and
|
||||
shutil copy-source traversal reads."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# A nested reassignment of an alias name must NOT inflate the outer scope's
|
||||
# single-assignment count and drop the real module-level sink alias.
|
||||
"import os\ns = os.system\ndef f():\n s = 1\ns('rm -rf /')",
|
||||
"e = exec\ndef f():\n e = 1\ne(\"__import__('os').system('id')\")",
|
||||
"import os\ns = os.system\nclass C:\n s = 1\ns('rm -rf /')",
|
||||
],
|
||||
)
|
||||
def test_nested_reassignment_keeps_outer_alias(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import types, builtins\n"
|
||||
"def f(src):\n c = builtins.compile(src, '<s>', 'exec')\n"
|
||||
" types.FunctionType(c, {})()\nf('x')",
|
||||
"import types\nfrom builtins import compile as comp\n"
|
||||
"def f(src):\n c = comp(src, '<s>', 'exec')\n types.FunctionType(c, {})()\nf('x')",
|
||||
],
|
||||
)
|
||||
def test_non_bare_compile_functiontype_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import subprocess\nsubprocess.run(['touch', '/tmp/x'])",
|
||||
"import os\nos.system('tee /tmp/x')",
|
||||
"import os\nos.system('cp a /tmp/x')",
|
||||
"import os\nos.system('mv a /tmp/x')",
|
||||
"import os\nos.system('mkdir /tmp/x')",
|
||||
"import os\nos.system('truncate -s 0 /tmp/x')",
|
||||
],
|
||||
)
|
||||
def test_child_process_writers_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
def test_open_literal_kwargs_unpack_read_blocked(self):
|
||||
assert _check_code_safety("open(**{'file': '../../../etc/passwd'}).read()") is not None
|
||||
# A benign relative kwargs read stays allowed.
|
||||
_ok("open(**{'file': 'data.csv'}).read()")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"from pathlib import Path\np = Path('..') / '..' / '..' / 'etc' / 'passwd'\np.read_text()",
|
||||
"from pathlib import Path\nbase = Path('..') / '..'\np = base / 'etc' / 'passwd'\np.read_text()",
|
||||
],
|
||||
)
|
||||
def test_assigned_pathlib_read_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
def test_assigned_benign_pathlib_read_allowed(self):
|
||||
_ok("from pathlib import Path\np = Path('data') / 'train.csv'\np.read_text()")
|
||||
|
||||
def test_wrapper_option_argument_interpreter_blocked(self):
|
||||
# `stdbuf -o L python -c ...`: the option argument L must not be mistaken for the
|
||||
# command, so the interpreter that follows is still detected.
|
||||
assert _check_code_safety("import os\nos.system('stdbuf -o L python -c \"x\"')") is not None
|
||||
assert _check_code_safety("import os\nos.system('ionice -c 2 python evil.py')") is not None
|
||||
# env -i rm must still be caught (blocked command is not treated as a flag arg).
|
||||
assert _check_code_safety("import os\nos.system('env -i rm -rf /')") is not None
|
||||
# No false positive: grep's search pattern is not a command.
|
||||
_ok("import os\nos.system('timeout 5 grep -r curl .')")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import builtins\nobject.__getattribute__(builtins, 'eval')(\"open('/etc/passwd').read()\")",
|
||||
"import subprocess\ntype.__getattribute__(subprocess, 'call')(['id'])",
|
||||
"import builtins\nobject.__getattribute__(builtins.open, '__closure__')",
|
||||
],
|
||||
)
|
||||
def test_object_getattribute_obfuscation_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import runpy\nrunpy.run_path('evil.py')",
|
||||
"import runpy\nrunpy.run_module('evil')",
|
||||
"import runpy as r\nr.run_path('evil.py')",
|
||||
],
|
||||
)
|
||||
def test_runpy_execution_sinks_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
def test_runpy_non_exec_allowed(self):
|
||||
_ok("import runpy\nx = runpy.__doc__")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import shutil\nshutil.copy('../../../etc/passwd', 'p')",
|
||||
"import shutil\nshutil.copyfile('../../../etc/passwd', 'p')",
|
||||
"import shutil\nshutil.copy('/etc/passwd', 'p')",
|
||||
"import shutil\nshutil.move('../../../etc/shadow', 'p')",
|
||||
],
|
||||
)
|
||||
def test_shutil_copy_source_traversal_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
def test_shutil_copy_benign_allowed(self):
|
||||
_ok("import shutil\nshutil.copy('data.csv', 'backup.csv')")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue