Harden sandbox: compile(source=) payload; workdir import vetter -- PEP263 decode, ignore bytecode cache, refuse symlinks, network sinks, os import aliases
Close six P1 bypasses Codex found on the round-41 branch: - compile(source=...) keyword payload: compile() accepts its source as the source= keyword, but the analyzer only recovered the 1st positional arg, so a keyword-only compile feeding types.FunctionType(code)() (or exec(compile(source=...))) was treated as having no payload and ran unscanned. Recover the source= keyword too (new _compile_source_node), at both the code-object tracking and exec(compile()) sites. The remaining five harden the workdir-module import vetter (a helper .py the user wrote is vetted before import; each of these slipped a payload past it): - PEP 263 source encoding: the vetter read modules as fixed UTF-8, but Python's loader honors an encoding cookie. A `# coding: utf_7` module hides os.system in what the UTF-8 scan sees as a comment (raw +AAo- bytes are a newline under UTF-7). Decode with importlib.util.decode_source so the vetter sees what the loader will run. - bytecode cache: after scanning the source, returning the original spec let SourceFileLoader satisfy the import from a planted __pycache__ .pyc whose header matches the harmless source. Run the EXACT vetted source via a dedicated loader (_GuardVettedSourceLoader) so the bytecode cache is never consulted. - symlinked module: a workdir module that is a symlink to an outside file had a realpath outside the workdir, so it was treated as not-workdir and handed to the default loader unvetted. Decide workdir-membership by the origin path, then fail closed when the realpath escapes. - network sinks: the vetter only checked command-exec/eval, so a helper doing socket.create_connection(...) bypassed the static network policy (no runtime network backstop). Refuse a workdir module that imports a network primitive (socket / ssl / ftplib / smtplib / requests / httpx / aiohttp / ...). - os import aliases: sink references were only recognized when rooted at literal os / posix, so import os as o; s = o.system; s(...) passed (the assignment, not a direct call). Record os / posix import aliases before checking sink references. Regression coverage: TestRound42Bypasses in tests/test_sandbox_tools.py (compile source= keyword, positional, and exec(compile()) forms) and five workdir-module vetter tests in tests/test_sandbox_runtime_backstop.py (utf-7 encoding denied, forged pyc ignored while the vetted source runs, symlinked module denied, network sink denied, os import alias denied).
This commit is contained in:
danielhanchen
parent
6f54040042
commit
f11fbdcb7f
3 changed files with 263 additions and 17 deletions
|
|
@ -4227,6 +4227,18 @@ def _recovered_source(v):
|
|||
return _to_text(v)
|
||||
|
||||
|
||||
def _compile_source_node(node):
|
||||
"""The SOURCE argument of a compile() call: the 1st positional arg or the ``source=``
|
||||
keyword. compile() accepts its payload either way, so a keyword-only call
|
||||
(compile(source='...', filename='<p>', mode='exec')) must still be recovered."""
|
||||
if node.args:
|
||||
return node.args[0]
|
||||
for kw in node.keywords or []:
|
||||
if kw.arg == "source":
|
||||
return kw.value
|
||||
return None
|
||||
|
||||
|
||||
def _compile_mode(node, const_env):
|
||||
"""Recover a compile()'s literal mode= (3rd positional or keyword), else 'exec'."""
|
||||
mode_node = None
|
||||
|
|
@ -4678,12 +4690,13 @@ def _build_scope_alias_index(tree, const_env):
|
|||
and isinstance(rhs_eff.func, ast.Name)
|
||||
and emap.get(rhs_eff.func.id) == "compile"
|
||||
)
|
||||
) and rhs_eff.args:
|
||||
) and _compile_source_node(rhs_eff) is not None:
|
||||
# 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).
|
||||
# gadget below (dynamic or foldable payload). The source may be positional
|
||||
# OR the source= keyword.
|
||||
camap[name] = True
|
||||
v = _const_fold(rhs_eff.args[0], const_env)
|
||||
v = _const_fold(_compile_source_node(rhs_eff), const_env)
|
||||
if isinstance(v, (str, bytes, bytearray)):
|
||||
cmap[name] = (
|
||||
_recovered_source(v),
|
||||
|
|
@ -4949,14 +4962,15 @@ def _recover_exec_payload(node, func_id, const_env, compiled_env):
|
|||
arg0 = node.args[0]
|
||||
base_mode = "eval" if func_id == "eval" else "exec"
|
||||
|
||||
# exec(compile("...", ...)) / eval(compile("...", "<s>", "eval"))
|
||||
# exec(compile("...", ...)) / eval(compile("...", "<s>", "eval")) -- the compile source may be
|
||||
# positional or the source= keyword.
|
||||
if (
|
||||
isinstance(arg0, ast.Call)
|
||||
and isinstance(arg0.func, ast.Name)
|
||||
and arg0.func.id == "compile"
|
||||
and arg0.args
|
||||
and _compile_source_node(arg0) is not None
|
||||
):
|
||||
v = _const_fold(arg0.args[0], const_env)
|
||||
v = _const_fold(_compile_source_node(arg0), const_env)
|
||||
if isinstance(v, (str, bytes, bytearray)):
|
||||
return (
|
||||
"RECOVERED",
|
||||
|
|
@ -10168,7 +10182,16 @@ except Exception:
|
|||
try:
|
||||
import ast as _gast
|
||||
import importlib.machinery as _gimach
|
||||
import importlib.util as _gimportutil
|
||||
_GUARD_WORKDIR_REAL = _os.path.realpath(__WORKDIR__)
|
||||
# Network-capable modules: a workdir helper that opens a socket / HTTP client bypasses the
|
||||
# static network policy (there is no runtime network backstop), so refuse importing one.
|
||||
# (urllib / http bare tops are left out -- urllib.parse etc. are benign, OS isolation remains
|
||||
# the boundary for the dotted network submodules.)
|
||||
_GUARD_NET_MODS = frozenset({
|
||||
"socket", "ssl", "ftplib", "smtplib", "telnetlib", "poplib", "imaplib", "nntplib",
|
||||
"requests", "httpx", "aiohttp", "urllib3", "pycurl", "websocket", "websockets", "paramiko",
|
||||
})
|
||||
_GUARD_EXEC_ATTRS = frozenset({
|
||||
"system", "popen", "popen2", "popen3", "popen4", "startfile",
|
||||
"execl", "execle", "execlp", "execlpe", "execv", "execve", "execvp", "execvpe",
|
||||
|
|
@ -10189,14 +10212,23 @@ try:
|
|||
_tree = _gast.parse(_src)
|
||||
except _bi.BaseException:
|
||||
return True # unparseable workdir module -> fail closed
|
||||
# Pre-pass: record os / posix import ALIASES (import os as o) so an aliased sink reference
|
||||
# that is only assigned (s = o.system) -- not directly called -- is still recognized.
|
||||
_recv = set(_GUARD_EXEC_RECEIVERS)
|
||||
for _nd in _gast.walk(_tree):
|
||||
if isinstance(_nd, _gast.Import):
|
||||
for _al in _nd.names:
|
||||
if _al.name.split(".")[0] in _GUARD_EXEC_MODS:
|
||||
if _al.name in ("os", "posix"):
|
||||
_recv.add(_al.asname or _al.name)
|
||||
for _nd in _gast.walk(_tree):
|
||||
if isinstance(_nd, _gast.Import):
|
||||
for _al in _nd.names:
|
||||
_top = _al.name.split(".")[0]
|
||||
if _top in _GUARD_EXEC_MODS or _top in _GUARD_NET_MODS:
|
||||
return True
|
||||
elif isinstance(_nd, _gast.ImportFrom):
|
||||
_mroot = (_nd.module or "").split(".")[0]
|
||||
if _mroot in _GUARD_EXEC_MODS:
|
||||
if _mroot in _GUARD_EXEC_MODS or _mroot in _GUARD_NET_MODS:
|
||||
return True
|
||||
# `from os import system` / `from os import *` binds a BARE sink name into the
|
||||
# module namespace; a later bare system('id') call has no os. attribute to catch.
|
||||
|
|
@ -10213,10 +10245,10 @@ try:
|
|||
"eval", "exec", "compile", "__import__"):
|
||||
return True
|
||||
elif isinstance(_nd, _gast.Attribute):
|
||||
# A sink-named attribute REFERENCE (even uncalled) rooted at os / posix
|
||||
# (x = os.system). A same-named attribute on an unrelated object
|
||||
# (p.system = 'linux') is NOT a sink, so require a sink-module receiver root.
|
||||
if _nd.attr in _GUARD_EXEC_ATTRS and _guard_attr_root(_nd.value) in _GUARD_EXEC_RECEIVERS:
|
||||
# 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
|
||||
# object (p.system = 'linux') is NOT a sink, so require a sink-module receiver.
|
||||
if _nd.attr in _GUARD_EXEC_ATTRS and _guard_attr_root(_nd.value) in _recv:
|
||||
return True
|
||||
# A workdir module that touches the import machinery (sys.meta_path /
|
||||
# sys.path_hooks / sys.path_importer_cache) can remove THIS vetter, then a
|
||||
|
|
@ -10225,6 +10257,34 @@ try:
|
|||
if _nd.attr in ("meta_path", "path_hooks", "path_importer_cache"):
|
||||
return True
|
||||
return False
|
||||
def _guard_under_workdir(_p):
|
||||
return _p == _GUARD_WORKDIR_REAL or _p.startswith(_GUARD_WORKDIR_REAL + _os.sep)
|
||||
|
||||
class _GuardVettedSourceLoader:
|
||||
# Executes the EXACT source string the vetter scanned, so the loader can never satisfy
|
||||
# the import from a planted bytecode cache (.pyc) or re-decode the file differently than
|
||||
# it was vetted. __path__ for a package still comes from the spec's search locations.
|
||||
def __init__(self, _name, _path, _src, _is_pkg):
|
||||
self._n = _name
|
||||
self._p = _path
|
||||
self._s = _src
|
||||
self._pkg = _is_pkg
|
||||
|
||||
def create_module(self, _spec):
|
||||
return None
|
||||
|
||||
def exec_module(self, _module):
|
||||
exec(compile(self._s, self._p, "exec"), _module.__dict__)
|
||||
|
||||
def get_filename(self, _name=None):
|
||||
return self._p
|
||||
|
||||
def is_package(self, _name=None):
|
||||
return self._pkg
|
||||
|
||||
def get_source(self, _name=None):
|
||||
return self._s
|
||||
|
||||
class _GuardWorkdirImportVetter:
|
||||
def find_spec(self, _name, _path=None, _target=None):
|
||||
try:
|
||||
|
|
@ -10235,11 +10295,21 @@ try:
|
|||
if not _orig:
|
||||
return None # namespace / builtin / frozen: no file to vet, not workdir-sourced
|
||||
try:
|
||||
_abs = _os.path.abspath(_orig)
|
||||
_rp = _os.path.realpath(_orig)
|
||||
except _bi.BaseException:
|
||||
return None
|
||||
if not (_rp == _GUARD_WORKDIR_REAL or _rp.startswith(_GUARD_WORKDIR_REAL + _os.sep)):
|
||||
return None # not a workdir module; let the default finders load it
|
||||
_orig_in_wd = _guard_under_workdir(_abs) or _guard_under_workdir(
|
||||
_os.path.dirname(_abs)
|
||||
)
|
||||
_rp_in_wd = _guard_under_workdir(_rp)
|
||||
if not _orig_in_wd and not _rp_in_wd:
|
||||
return None # genuinely not a workdir module; let the default finders load it
|
||||
# A workdir-sourced origin whose REALPATH escapes the workdir (a symlink to an outside
|
||||
# file) must fail closed, not be handed to the default loader unvetted.
|
||||
if not _rp_in_wd:
|
||||
raise _bi.ImportError(
|
||||
"sandbox: refusing symlinked workdir module " + _name)
|
||||
# A workdir module must be a .py we can read + scan. A sourceless .pyc / native .so /
|
||||
# any other non-source file under the workdir cannot be statically vetted, so refuse
|
||||
# it: a planted legacy evil.pyc would otherwise run its bytecode via the default
|
||||
|
|
@ -10247,17 +10317,28 @@ try:
|
|||
if not _orig.endswith(".py"):
|
||||
raise _bi.ImportError(
|
||||
"sandbox: refusing to import non-source workdir module " + _name)
|
||||
# Decode with Python's PEP 263 source encoding (importlib.util.decode_source), NOT a
|
||||
# fixed utf-8: a `# coding: utf_7` module the loader would decode as UTF-7 must be
|
||||
# vetted as UTF-7, or a payload hidden in what a UTF-8 scan sees as a comment runs.
|
||||
try:
|
||||
_fh = _io.open(_orig, "r", encoding="utf-8", errors="replace")
|
||||
_fb = _io.open(_orig, "rb")
|
||||
try:
|
||||
_msrc = _fh.read()
|
||||
_raw = _fb.read()
|
||||
finally:
|
||||
_fh.close()
|
||||
_fb.close()
|
||||
_msrc = _gimportutil.decode_source(_raw)
|
||||
except _bi.BaseException:
|
||||
raise _bi.ImportError("sandbox: cannot vet workdir module " + _name)
|
||||
if _guard_module_src_unsafe(_msrc):
|
||||
raise _bi.ImportError(
|
||||
"sandbox: refusing to import unvetted workdir module " + _name)
|
||||
# Run the EXACT vetted source via our loader so a planted matching .pyc can never be
|
||||
# executed instead (the default SourceFileLoader would satisfy the import from a
|
||||
# __pycache__ .pyc whose header matches the harmless source).
|
||||
_is_pkg = _spec.submodule_search_locations is not None or _os.path.basename(
|
||||
_orig
|
||||
) == "__init__.py"
|
||||
_spec.loader = _GuardVettedSourceLoader(_name, _orig, _msrc, _is_pkg)
|
||||
return _spec
|
||||
_sys.meta_path.insert(0, _GuardWorkdirImportVetter())
|
||||
except _bi.BaseException:
|
||||
|
|
|
|||
|
|
@ -789,6 +789,145 @@ def test_sandboxed_workdir_module_meta_path_mutation_denied():
|
|||
os.remove(os.path.join(workdir, "mp_popper.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_utf7_encoded_workdir_module_denied():
|
||||
# A `# coding: utf_7` module hides os.system in what a UTF-8 scan reads as a comment (the raw
|
||||
# +AAo- bytes are a newline under UTF-7). The vetter must decode with PEP 263 like the loader
|
||||
# will, so the real os.system is seen and refused.
|
||||
session = "backstop-workdir-utf7"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
data = b"# coding: utf_7\nimport os\npass #+AAo-os.system('echo PWNED_UTF7')\n"
|
||||
with open(os.path.join(workdir, "evilenc.py"), "wb") as f:
|
||||
f.write(data)
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import evilenc; print('REACHED_' + 'BODY')",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "PWNED_UTF7" not in out
|
||||
assert "REACHED_BODY" not in out
|
||||
assert "sandbox:" in out or "ImportError" in out
|
||||
finally:
|
||||
os.remove(os.path.join(workdir, "evilenc.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_forged_pyc_workdir_module_ignored():
|
||||
# A harmless source plus a planted __pycache__ .pyc whose header matches it but whose body is
|
||||
# malicious: the vetter scans the safe source, but the module must run the VETTED SOURCE
|
||||
# (not the cached bytecode), so the planted payload never executes.
|
||||
import importlib.util
|
||||
import marshal
|
||||
import struct
|
||||
|
||||
session = "backstop-workdir-forgedpyc"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
src_path = os.path.join(workdir, "forged.py")
|
||||
with open(src_path, "w") as f:
|
||||
f.write("VALUE = 7\nprint('SOURCE_RAN')\n")
|
||||
st = os.stat(src_path)
|
||||
mal = compile("import os\nos.system('echo PWNED_FORGEDPYC')\n", "forged.py", "exec")
|
||||
pyc_dir = os.path.join(workdir, "__pycache__")
|
||||
os.makedirs(pyc_dir, exist_ok = True)
|
||||
pyc_path = os.path.join(pyc_dir, f"forged.{sys.implementation.cache_tag}.pyc")
|
||||
with open(pyc_path, "wb") as f:
|
||||
f.write(importlib.util.MAGIC_NUMBER)
|
||||
f.write(struct.pack("<I", 0))
|
||||
f.write(struct.pack("<I", int(st.st_mtime) & 0xFFFFFFFF))
|
||||
f.write(struct.pack("<I", st.st_size & 0xFFFFFFFF))
|
||||
f.write(marshal.dumps(mal))
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import forged; print('REACHED', forged.VALUE)",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "PWNED_FORGEDPYC" not in out # the cached malicious bytecode never runs
|
||||
assert "SOURCE_RAN" in out and "REACHED 7" in out # the vetted source runs
|
||||
finally:
|
||||
os.remove(src_path)
|
||||
if os.path.exists(pyc_path):
|
||||
os.remove(pyc_path)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_symlinked_workdir_module_denied(tmp_path):
|
||||
# A workdir module that is a symlink to a file OUTSIDE the workdir: its realpath escapes, so
|
||||
# the vetter must fail closed rather than hand the outside file to the default loader unvetted.
|
||||
session = "backstop-workdir-symlinkmod"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
outside = tmp_path / "outside_evil.py"
|
||||
outside.write_text("import os\nos.system('echo PWNED_SYMLINKMOD')\n")
|
||||
link = os.path.join(workdir, "evillink.py")
|
||||
if os.path.islink(link) or os.path.exists(link):
|
||||
os.remove(link)
|
||||
os.symlink(str(outside), link)
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import evillink; print('REACHED_' + 'BODY')",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "PWNED_SYMLINKMOD" not in out
|
||||
assert "REACHED_BODY" not in out
|
||||
assert "sandbox:" in out or "ImportError" in out
|
||||
finally:
|
||||
os.remove(link)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_network_workdir_module_denied():
|
||||
# A workdir helper that opens a socket bypasses the static network policy (no runtime network
|
||||
# backstop), so the vetter refuses a module importing a network primitive.
|
||||
session = "backstop-workdir-net"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
with open(os.path.join(workdir, "evilnet.py"), "w") as f:
|
||||
f.write("print('NET_REACHED')\nimport socket\ns = socket.socket()\ns.close()\n")
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import evilnet; print('REACHED_' + 'BODY')",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "NET_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, "evilnet.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_os_alias_workdir_module_denied():
|
||||
# import os as o; s = o.system; s(...) -- an os import ALIAS whose sink is only assigned (not
|
||||
# directly called) must be recognized: record the alias before checking sink references.
|
||||
session = "backstop-workdir-osalias"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
with open(os.path.join(workdir, "evilalias.py"), "w") as f:
|
||||
f.write("import os as o\ns = o.system\ns('echo PWNED_OSALIAS')\n")
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import evilalias; print('REACHED_' + 'BODY')",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "PWNED_OSALIAS" not in out
|
||||
assert "REACHED_BODY" not in out
|
||||
assert "sandbox:" in out or "ImportError" in out
|
||||
finally:
|
||||
os.remove(os.path.join(workdir, "evilalias.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
|
||||
|
|
|
|||
|
|
@ -4416,3 +4416,29 @@ class TestRound41Bypasses:
|
|||
)
|
||||
def test_round41_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
||||
|
||||
class TestRound42Bypasses:
|
||||
"""Forty-second-round Codex findings. The compile(source=...) keyword payload is static; the
|
||||
five workdir-module import-vetter items (utf-7 source encoding, forged bytecode cache,
|
||||
symlinked module, network sink, os import alias) are covered in
|
||||
test_sandbox_runtime_backstop.py."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# compile() accepts its payload as source=..., so a keyword-only compile feeding the
|
||||
# types.FunctionType(code)() execution gadget must still be recovered and analyzed.
|
||||
"import types\n"
|
||||
"c = compile(source='import os\\nos.system(\"touch /tmp/x\")', filename='<p>', mode='exec')\n"
|
||||
"types.FunctionType(c, {})()",
|
||||
# positional control (already caught) -- same gadget, positional source.
|
||||
"import types\n"
|
||||
"c = compile('import os\\nos.system(\"touch /tmp/x\")', '<p>', 'exec')\n"
|
||||
"types.FunctionType(c, {})()",
|
||||
# exec(compile(source=...)) keyword form.
|
||||
"exec(compile(source='import os\\nos.system(\"touch /tmp/x\")', filename='<p>', mode='exec'))",
|
||||
],
|
||||
)
|
||||
def test_compile_source_keyword_payload_blocked(self, code):
|
||||
assert _check_code_safety(code) is not None, code
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue