Harden sandbox: fail closed on unvetted __code__ stores; workdir getattr sink obfuscation; honor subprocess cwd= and GNU env glued -C/-u for git writes
Close four bypasses Codex found on the round-44 branch:
- fn.__code__ = <code object>: rebinding a function's code runs it via fn()
WITHOUT eval / exec, the __code__ twin of the FunctionType gadget. The
assignment visitor only checked container-stored exec aliases, so
co = codeop.compile_command('...'); f.__code__ = co; f() ran unanalyzed
source. Flag a __code__ store whose RHS is not a vetted code object; an
in-source function's code (g.__code__) and a compile() result (analyzed at
the compile site) stay allowed.
- workdir-module getattr obfuscation: the import vetter caught direct
os.system(...) but not getattr(os, 'system')('...') in an imported helper,
so the top-level analyzer saw only the file write / import and the vetter
passed. Refuse getattr on a sink-module receiver (os / posix / builtins /
deserializers) -- a constant sink attribute name, and a non-constant name
that cannot be proven benign.
- subprocess cwd= ignored for child writes: the argv scan reconstructed the
git command but dropped cwd=, so subprocess.run(['git','init','repo'],
cwd='/tmp') created /tmp/repo outside the workdir. Model a literal escaping
cwd= as a synthetic `env -C <cwd>` wrapper on the reconstructed command so
the existing git cwd backscan resolves the escape; a workdir-relative cwd
adds no wrapper and stays allowed.
- GNU env glued -C / -u operands: env -C/tmp git init repo (and
env -uGIT_CONFIG_COUNT git ...) glue the chdir / unset operand directly onto
the short flag, which the separated and --long= scans missed, so the git cwd
and hook-suppression backscan never saw the escape. Parse the glued short
forms alongside the separated ones.
Regression coverage: TestRound45Bypasses in tests/test_sandbox_tools.py
(__code__ store of a producer / opaque code object blocked while a compile()
result and g.__code__ stay allowed; subprocess git under an escaping cwd
blocked while a workdir-relative cwd is allowed; env -C/tmp and
-uGIT_CONFIG_COUNT before git blocked while plain env git init is allowed) and
two workdir-module vetter cases in tests/test_sandbox_runtime_backstop.py
(getattr(os,'system') helper denied, benign getattr on a plain object allowed).
This commit is contained in:
danielhanchen
parent
8384cea660
commit
8814d7de3b
3 changed files with 216 additions and 4 deletions
|
|
@ -1526,6 +1526,10 @@ def _find_blocked_commands(command: str) -> set[str]:
|
|||
_git_cwd_escapes = True
|
||||
elif _bt.startswith("--chdir=") and _arg_escapes_workdir(_bt.split("=", 1)[1]):
|
||||
_git_cwd_escapes = True
|
||||
# GNU env glues the short chdir operand directly onto the flag (env -C/tmp git init),
|
||||
# which the separated / --chdir= forms above miss. Only -C takes a dir here.
|
||||
elif _bt.startswith("-C") and len(_bt) > 2 and _arg_escapes_workdir(_bt[2:]):
|
||||
_git_cwd_escapes = True
|
||||
# env -i / --ignore-environment / a bare `-` start git with an EMPTY environment, and
|
||||
# env -u NAME / --unset NAME / --unset=NAME strip just the suppression var; either
|
||||
# removes the injected core.hooksPath suppression so a planted .git/hooks/* runs in
|
||||
|
|
@ -1540,6 +1544,9 @@ def _find_blocked_commands(command: str) -> set[str]:
|
|||
_env_suppress_dropped = True
|
||||
elif _bt.startswith("--unset=") and _bt.split("=", 1)[1].startswith("GIT_CONFIG"):
|
||||
_env_suppress_dropped = True
|
||||
# GNU env glues the short unset operand onto the flag (env -uGIT_CONFIG git ...).
|
||||
elif _bt.startswith("-u") and len(_bt) > 2 and _bt[2:].startswith("GIT_CONFIG"):
|
||||
_env_suppress_dropped = True
|
||||
elif _token_basename(_bt) == "env":
|
||||
_seg_has_env = True
|
||||
if _git_cwd_escapes:
|
||||
|
|
@ -6049,8 +6056,11 @@ def _check_signal_escape_patterns(
|
|||
found.add("shell-script:" + first)
|
||||
return found
|
||||
|
||||
def _check_args_for_blocked(args_nodes, shell_maybe_true = False):
|
||||
"""Check if any call arguments contain blocked commands."""
|
||||
def _check_args_for_blocked(args_nodes, shell_maybe_true = False, cwd_prefix = ""):
|
||||
"""Check if any call arguments contain blocked commands. ``cwd_prefix`` is a synthetic
|
||||
``env -C <dir> `` wrapper string prepended to a reconstructed argv command when the call
|
||||
has a literal escaping ``cwd=`` (subprocess.run(['git','init','repo'], cwd='/tmp')), so the
|
||||
git cwd backscan resolves the child's real working directory."""
|
||||
found = set()
|
||||
for arg in args_nodes:
|
||||
s = _extract_string_from_node(arg)
|
||||
|
|
@ -6092,7 +6102,8 @@ def _check_signal_escape_patterns(
|
|||
# env -C /tmp before git -- is still seen by the git cwd backscan.
|
||||
elif _cmd_base in _ARGV_TAIL_SCAN_COMMANDS:
|
||||
found |= _find_blocked_commands(
|
||||
" ".join(shlex.quote(s) for s in str_elts if s is not None)
|
||||
cwd_prefix
|
||||
+ " ".join(shlex.quote(s) for s in str_elts if s is not None)
|
||||
)
|
||||
# An env WRAPPER in the argv applies NAME=value assignments before the command
|
||||
# (env PATH=. evil, env BASH_ENV=env.sh bash -c ..., env GIT_DIR=/tmp git init);
|
||||
|
|
@ -6756,6 +6767,34 @@ def _check_signal_escape_patterns(
|
|||
return True
|
||||
return False
|
||||
|
||||
def _code_store_rhs_vetted(self, rhs):
|
||||
# A value assigned to fn.__code__ that we can prove is safe to execute. An in-source
|
||||
# function's code (g.__code__ / meth.__func__) is analyzed normally, and a compile()
|
||||
# result -- direct call or a c = compile(...) alias -- has its SOURCE analyzed at the
|
||||
# compile site (a malicious / opaque source is flagged there). Everything else (a
|
||||
# producer code object from codeop / a loader's get_code() / marshal, or an opaque
|
||||
# name) is unvetted and fails closed.
|
||||
if isinstance(rhs, ast.Attribute) and rhs.attr in ("__code__", "__func__"):
|
||||
return True
|
||||
if isinstance(rhs, ast.Call):
|
||||
rf = rhs.func
|
||||
if isinstance(rf, ast.Name) and (
|
||||
rf.id == "compile" or self.exec_from_aliases.get(rf.id) == "compile"
|
||||
):
|
||||
return True
|
||||
if (
|
||||
isinstance(rf, ast.Attribute)
|
||||
and rf.attr == "compile"
|
||||
and _ast_name_matches(rf.value, self.builtins_aliases)
|
||||
):
|
||||
return True
|
||||
if _analyzer_on and isinstance(rhs, ast.Name):
|
||||
if _scope_idx.resolve(rhs.id, rhs, "compiledany"):
|
||||
return True
|
||||
if _scope_idx.resolve(rhs.id, rhs, "execb") == "compile":
|
||||
return True
|
||||
return False
|
||||
|
||||
def visit_Assign(self, node):
|
||||
# d['e'] = exec / lst[0] = eval -- storing a dynamic-exec builtin into a CONTAINER
|
||||
# element (not a plain name, which the alias tracker already follows) hides the sink
|
||||
|
|
@ -6775,6 +6814,24 @@ def _check_signal_escape_patterns(
|
|||
),
|
||||
}
|
||||
)
|
||||
# fn.__code__ = <code object> rebinds a function's body, so fn() then runs that code
|
||||
# WITHOUT eval / exec. A code object from an unvetted producer (codeop.compile_command,
|
||||
# a loader's get_code(), marshal) runs source the recursive analysis never saw, the
|
||||
# __code__ twin of the FunctionType gadget. Flag a __code__ store whose RHS is not a
|
||||
# vetted in-source / compile()-analyzed code object.
|
||||
if any(
|
||||
isinstance(t, ast.Attribute) and t.attr == "__code__" for t in node.targets
|
||||
) and not self._code_store_rhs_vetted(node.value):
|
||||
dynamic_exec.append(
|
||||
{
|
||||
"type": "dynamic_exec",
|
||||
"line": getattr(node, "lineno", -1),
|
||||
"description": (
|
||||
"an unvetted code object assigned to __code__ "
|
||||
"(executes via the function without eval/exec)"
|
||||
),
|
||||
}
|
||||
)
|
||||
self.generic_visit(node)
|
||||
|
||||
def visit_Call(self, node):
|
||||
|
|
@ -6930,7 +6987,19 @@ def _check_signal_escape_patterns(
|
|||
_shell_node is None
|
||||
or (isinstance(_shell_node, ast.Constant) and _shell_node.value is False)
|
||||
)
|
||||
blocked_in_args = _check_args_for_blocked(all_call_args, _shell_maybe_true)
|
||||
# A literal cwd= that escapes the workdir sets the child's real working directory,
|
||||
# so a relative write operand (subprocess.run(['git','init','repo'], cwd='/tmp'))
|
||||
# lands OUTSIDE the session. Model it as a synthetic `env -C <cwd>` wrapper so the
|
||||
# git cwd backscan resolves the escape; a workdir-relative / in-tree cwd adds no
|
||||
# prefix and stays allowed.
|
||||
_cwd_node = expanded_kwargs.get("cwd")
|
||||
_cwd_str = _extract_string_from_node(_cwd_node) if _cwd_node is not None else None
|
||||
_cwd_prefix = ""
|
||||
if _cwd_str is not None and _arg_escapes_workdir(_cwd_str):
|
||||
_cwd_prefix = "env -C " + shlex.quote(_cwd_str) + " "
|
||||
blocked_in_args = _check_args_for_blocked(
|
||||
all_call_args, _shell_maybe_true, _cwd_prefix
|
||||
)
|
||||
|
||||
# The argv sequence can be given positionally (run(['bash', ...])) or through the
|
||||
# public args= keyword (run(args=['bash', ...])), which this analyzer already
|
||||
|
|
@ -10346,6 +10415,34 @@ try:
|
|||
and _guard_attr_root(_nd.func.value) in _deser
|
||||
):
|
||||
return True
|
||||
# getattr(os, 'system')(...) / getattr(builtins, 'eval')(...) /
|
||||
# getattr(pickle, 'loads')(...) -- dynamic attribute access is the obfuscated twin
|
||||
# of the direct sink attribute (the name-based checks above never see it). A
|
||||
# constant sink name on a sink-module receiver is refused; a NON-constant name on
|
||||
# such a receiver is refused too (the attribute cannot be proven benign).
|
||||
if (
|
||||
isinstance(_nd.func, _gast.Name)
|
||||
and _nd.func.id == "getattr"
|
||||
and len(_nd.args) >= 2
|
||||
and isinstance(_nd.args[0], (_gast.Name, _gast.Attribute))
|
||||
):
|
||||
_grecv = _guard_attr_root(_nd.args[0])
|
||||
_gname = (
|
||||
_nd.args[1].value
|
||||
if isinstance(_nd.args[1], _gast.Constant)
|
||||
and isinstance(_nd.args[1].value, str)
|
||||
else None
|
||||
)
|
||||
if _gname is None:
|
||||
if _grecv in _recv or _grecv in _bi or _grecv in _deser:
|
||||
return True
|
||||
else:
|
||||
if _grecv in _recv and _gname in _GUARD_EXEC_ATTRS:
|
||||
return True
|
||||
if _grecv in _bi and _gname in ("eval", "exec", "compile", "__import__"):
|
||||
return True
|
||||
if _grecv in _deser and _gname in _GUARD_DESER_ATTRS:
|
||||
return True
|
||||
elif isinstance(_nd, _gast.Attribute):
|
||||
# A sink-named attribute REFERENCE (even uncalled) rooted at os / posix / an
|
||||
# os alias (x = os.system, s = o.system). A same-named attribute on an unrelated
|
||||
|
|
|
|||
|
|
@ -1054,6 +1054,53 @@ def test_sandboxed_benign_json_workdir_module_allowed():
|
|||
os.remove(os.path.join(workdir, "okjson.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_getattr_obfuscated_sink_workdir_module_denied():
|
||||
# getattr(os, 'system')('...') in a workdir helper is the obfuscated twin of os.system, which
|
||||
# the direct-attribute checks miss -- the dynamic-attribute sink must be refused at import.
|
||||
session = "backstop-workdir-getattr"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
with open(os.path.join(workdir, "evilga.py"), "w") as f:
|
||||
f.write("import os\nprint('OBF_REACHED')\ngetattr(os, 'system')('echo PWNED_GA')\n")
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import evilga; print('REACHED_' + 'BODY')",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "PWNED_GA" not in out
|
||||
assert "OBF_REACHED" not in out
|
||||
assert "REACHED_BODY" not in out
|
||||
assert "sandbox:" in out or "ImportError" in out
|
||||
finally:
|
||||
os.remove(os.path.join(workdir, "evilga.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_benign_getattr_workdir_module_allowed():
|
||||
# getattr on a non-sink receiver (a plain object attribute) is ordinary reflection, not a sink,
|
||||
# so a workdir helper using it must still import.
|
||||
session = "backstop-workdir-okgetattr"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
with open(os.path.join(workdir, "okga.py"), "w") as f:
|
||||
f.write("class K:\n v = 7\nVALUE = getattr(K, 'v')\nprint('GA_OK')\n")
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import okga; print('REACHED', okga.VALUE)",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "GA_OK" in out
|
||||
assert "REACHED 7" in out
|
||||
assert "sandbox:" not in out
|
||||
finally:
|
||||
os.remove(os.path.join(workdir, "okga.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_realpath_monkeypatch_write_escape_denied(tmp_path):
|
||||
# Sandboxed code reassigns os.path.realpath to a lambda that echoes an in-workdir
|
||||
|
|
|
|||
|
|
@ -4564,3 +4564,71 @@ class TestRound44Bypasses:
|
|||
)
|
||||
def test_shell_keyword_compound_header_still_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
|
||||
class TestRound45Bypasses:
|
||||
"""Forty-fifth-round Codex findings. The __code__ store gadget, the subprocess cwd= escape, and
|
||||
the GNU env glued -C/-u forms are static; the workdir-module getattr obfuscation item is covered
|
||||
in test_sandbox_runtime_backstop.py."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# fn.__code__ = <code object>; fn() runs that code WITHOUT eval/exec. A code object
|
||||
# from an unvetted producer (codeop / loader.get_code / marshal) runs unanalyzed source.
|
||||
"import codeop\n"
|
||||
"co = codeop.compile_command(\"__import__('os').system('touch /tmp/x')\")\n"
|
||||
"f = lambda: None\nf.__code__ = co\nf()",
|
||||
"import codeop\nf = lambda: None\n"
|
||||
"f.__code__ = codeop.compile_command(\"__import__('os').system('touch /tmp/x')\")\nf()",
|
||||
"import types\nf = lambda: None\n"
|
||||
"f.__code__ = types.FunctionType.__call__ # opaque non-compile code object\nf()",
|
||||
],
|
||||
)
|
||||
def test_code_attr_store_unvetted_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# A compile() result (analyzed at the compile site) or an in-source function's code are
|
||||
# vetted, so binding them to __code__ stays allowed.
|
||||
"c = compile(source='X = 1', filename='<s>', mode='exec')\nf = lambda: None\nf.__code__ = c\nf()",
|
||||
"def g():\n return 1\nf = lambda: None\nf.__code__ = g.__code__\nf()",
|
||||
],
|
||||
)
|
||||
def test_code_attr_store_vetted_allowed(self, code):
|
||||
_ok(code)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# A literal escaping cwd= sets the child's real working directory, so a relative git
|
||||
# write operand lands outside the workdir (git init repo -> /tmp/repo).
|
||||
"import subprocess\nsubprocess.run(['git', 'init', 'repo'], cwd='/tmp')",
|
||||
"import subprocess\nsubprocess.run(['git', 'clone', 'u', 'repo'], cwd='/var/tmp')",
|
||||
],
|
||||
)
|
||||
def test_subprocess_escaping_cwd_git_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
||||
def test_subprocess_workdir_cwd_git_allowed(self):
|
||||
# A workdir-relative cwd (or no cwd) keeps a relative git operand in-tree -- still allowed.
|
||||
_ok("import subprocess\nsubprocess.run(['git', 'init', 'repo'], cwd='sub')")
|
||||
_ok("import subprocess\nsubprocess.run(['git', 'init', 'repo'])")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"cmd",
|
||||
[
|
||||
# GNU env glues the short chdir / unset operand onto the flag; the git cwd / suppression
|
||||
# backscan must parse the glued forms, not only the separated / --long= ones.
|
||||
"import os\nos.system('env -C/tmp git init repo')",
|
||||
"import os\nos.system('env -uGIT_CONFIG_COUNT git init repo')",
|
||||
],
|
||||
)
|
||||
def test_env_glued_operand_git_blocked(self, cmd):
|
||||
assert _check_code_safety(cmd) is not None, cmd
|
||||
|
||||
def test_env_no_chdir_git_allowed(self):
|
||||
# env with no -C / -u before a workdir-relative git op stays allowed.
|
||||
_ok("import os\nos.system('env git init repo')")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue