Harden sandbox: workdir-vetter alias / namespace / MRO / sys.modules gaps, network client entry points, request() URL arg, sqlite URI mode parsing
Close eight gaps Codex found on the round-60 branch (seven inline P1 plus one
review-body P1). Five are in the runtime workdir-module import vetter (a planted
workdir helper is not scanned by the outer static pass), two are in the static
network scanner, and one is the runtime sqlite URI parser.
- assigned os aliases in the vetter: the pre-pass only recorded `import os as o`
aliases, so a workdir helper doing `import os; o = os; o.system(...)` passed
vetting and spawned an unguarded child. Follow simple whole-module assignments
(o = os; b = builtins; s = sys) to a fixpoint before the sink checks.
- module namespace-dict subscripts in the vetter: `os.__dict__['system'](...)`
reached the sink because __dict__ was not a gadget and the subscript branch only
failed closed for builtins. Fail closed on a `<module>.__dict__[...]` subscript
for os / posix / deserializer / sys / importlib, like vars(<module>).
- module __getattribute__ / __getattr__ in the vetter: `os.__getattribute__(
'system')(...)` reached the sink because only the builtin getattr(...) form was
recognized. Classify bound `module.__getattribute__('name')` and unbound
`object.__getattribute__(module, 'name')` lookups the same way as getattr.
- MRO base recovery in the vetter: `io.FileIO.__mro__[1]('/tmp/x','w')` /
`sqlite3.Connection.__mro__[1](...)` recovered an unguarded base class because the
helper gadget set omitted __mro__ / mro. Add both to the gadget attributes.
- sys.modules in the vetter: `sys.modules['os'].system(...)` recovered the
guard-cached os module without an import, bypassing the denied-import path. Deny
sys.modules access (direct attribute and getattr form) in a vetted workdir module.
- public network client entry points: `requests.api.get(...)`, `ftplib.FTP(...)`,
and `smtplib.SMTP(...)` were not in the network prefix table, so a metadata /
untrusted host reached through them bypassed the allowlist. Add requests.api.*,
ftplib.FTP / FTP_TLS, and smtplib.SMTP / SMTP_SSL / LMTP (the ftplib / smtplib
clients take a bare host), and track ftplib / smtplib import aliases.
- request(method, url) URL argument: module-level requests.request / httpx.request /
urllib3.request (and requests.api.request) carry the URL at arg1, but the code
passed arg0 (the HTTP method) to the host check, so the URL was never inspected.
Read the URL from arg1 for these method-first APIs, like the client-instance
.request() branch.
- sqlite URI mode=memory parsing: a `file:/tmp/escape.db?xmode=memory` URI was
treated as in-memory by a substring test and skipped path confinement, but SQLite
ignores the unknown xmode key and opens the on-disk file. Parse the query exactly
(split on &, first occurrence of a repeated key, percent-decoded) and treat only a
genuine mode=memory parameter as in-memory, in the runtime guard and the two static
sqlite operand checks.
Regression coverage: TestRound61Bypasses in tests/test_sandbox_tools.py (request()
URL at arg1 for requests / httpx / urllib3 / requests.api, network client entry
points against metadata and untrusted hosts, sqlite shell xmode=memory on an
escaping absolute path, plus a benign-allowed set: trusted-host requests / api /
ftplib / smtplib, a genuine in-memory URI, and a workdir-relative db) and, in
tests/test_sandbox_runtime_backstop.py, workdir-module denials for the assigned os
alias, os.__dict__ subscript, os.__getattribute__, io.FileIO.__mro__, and
sys.modules['os'] forms, a benign os-alias helper that still imports, and the sqlite
URI xmode=memory escape denial plus a genuine mode=memory allowance.
This commit is contained in:
danielhanchen
parent
5343e2e993
commit
236e89d0c9
3 changed files with 357 additions and 32 deletions
|
|
@ -587,6 +587,24 @@ def _cwd_wrapper_escapes(tokens, cmd_idx) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def _sqlite_uri_mode_is_memory(_s: str) -> bool:
|
||||
"""True only when a sqlite URI query string has a genuine mode=memory parameter. SQLite splits
|
||||
query parameters on ``&`` and uses the FIRST occurrence of a repeated key, so an unknown key
|
||||
(``xmode=memory``) or a later ``mode=`` is NOT in-memory -- a substring test wrongly treated
|
||||
``file:/tmp/escape.db?xmode=memory`` as in-memory and skipped path confinement. Percent-decodes
|
||||
each key / value so a ``mode=m%65mory`` (which SQLite decodes) is still recognized."""
|
||||
|
||||
def _dec(_x):
|
||||
return re.sub("%([0-9A-Fa-f]{2})", lambda _m: chr(int(_m.group(1), 16)), _x)
|
||||
|
||||
_q = _s.partition("?")[2]
|
||||
for _pair in _q.split("&"):
|
||||
_k, _sep, _v = _pair.partition("=")
|
||||
if _dec(_k) == "mode":
|
||||
return _dec(_v) == "memory"
|
||||
return False
|
||||
|
||||
|
||||
def _operand_relative_local(tok: str) -> bool:
|
||||
"""A literal RELATIVE path operand that resolves under the child cwd, so it escapes the workdir
|
||||
when the cwd itself escapes (paired with _cwd_wrapper_escapes). Absolute (``/x``), home (``~``),
|
||||
|
|
@ -600,7 +618,7 @@ def _operand_relative_local(tok: str) -> bool:
|
|||
if not _u or _u[0] in ("/", "~", "-") or "$" in _u or "`" in _u:
|
||||
return False
|
||||
_ul = _u.lower()
|
||||
if _u == ":memory:" or _ul.startswith("file::memory:") or "mode=memory" in _ul:
|
||||
if _u == ":memory:" or _ul.startswith("file::memory:") or _sqlite_uri_mode_is_memory(_ul):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
|
@ -2337,7 +2355,7 @@ def _find_blocked_commands(command: str) -> set[str]:
|
|||
_is_mem = (
|
||||
_dbn in ("", ":memory:")
|
||||
or _dblow.startswith("file::memory:")
|
||||
or "mode=memory" in _dblow
|
||||
or _sqlite_uri_mode_is_memory(_dblow)
|
||||
)
|
||||
if not _is_mem and (
|
||||
_git_operand_escapes(_dbn, _local_assigns)
|
||||
|
|
@ -10125,7 +10143,11 @@ def _check_signal_escape_patterns(
|
|||
"requests.patch",
|
||||
"requests.head",
|
||||
"requests.request",
|
||||
"requests.api.",
|
||||
"requests.Session",
|
||||
"ftplib.FTP",
|
||||
"smtplib.SMTP",
|
||||
"smtplib.LMTP",
|
||||
"http.client.HTTPConnection",
|
||||
"http.client.HTTPSConnection",
|
||||
"socket.gethostbyname",
|
||||
|
|
@ -10550,7 +10572,17 @@ def _check_signal_escape_patterns(
|
|||
# u -> {"u": "urllib.request"}, from urllib import request as req -> {"req":
|
||||
# "urllib.request"}. Without this, r.get('http://169.254.169.254/') builds fq="r.get"
|
||||
# and skips every metadata / allowlist / upload check.
|
||||
_NET_TOP_MODULES = ("socket", "urllib", "urllib3", "requests", "http", "httpx", "aiohttp")
|
||||
_NET_TOP_MODULES = (
|
||||
"socket",
|
||||
"urllib",
|
||||
"urllib3",
|
||||
"requests",
|
||||
"http",
|
||||
"httpx",
|
||||
"aiohttp",
|
||||
"ftplib",
|
||||
"smtplib",
|
||||
)
|
||||
_net_aliases: dict[str, str] = {}
|
||||
for _n in ast.walk(tree):
|
||||
if isinstance(_n, ast.Import):
|
||||
|
|
@ -10577,6 +10609,23 @@ def _check_signal_escape_patterns(
|
|||
"socket.getaddrinfo",
|
||||
"socket.gethostbyname",
|
||||
"socket.gethostbyname_ex",
|
||||
# ftplib / smtplib clients take a bare host (or host= keyword), not a URL.
|
||||
"ftplib.FTP",
|
||||
"ftplib.FTP_TLS",
|
||||
"smtplib.SMTP",
|
||||
"smtplib.SMTP_SSL",
|
||||
"smtplib.LMTP",
|
||||
}
|
||||
)
|
||||
# Module-level request(method, url, ...) APIs whose URL is the SECOND positional argument
|
||||
# (the first is the HTTP method), so the host is read from args[1] like the client-instance
|
||||
# .request() branch -- not args[0], which is just the method string.
|
||||
_NET_REQUEST_METHOD_APIS = frozenset(
|
||||
{
|
||||
"requests.request",
|
||||
"requests.api.request",
|
||||
"httpx.request",
|
||||
"urllib3.request",
|
||||
}
|
||||
)
|
||||
_NET_HOST_KWARGS = ("host",)
|
||||
|
|
@ -10873,7 +10922,10 @@ def _check_signal_escape_patterns(
|
|||
# host=...)). Bare-host callees (HTTPConnection, getaddrinfo) treat the literal
|
||||
# arg as a host directly; everything else parses a scheme://host URL. A target
|
||||
# that stays fully opaque fails closed. See _net_check_target.
|
||||
a0 = node.args[0] if node.args else None
|
||||
# requests.request('GET', url) / httpx.request(...) / urllib3.request(...) carry
|
||||
# the URL at arg1 (arg0 is the HTTP method); every other API carries it at arg0.
|
||||
_url_idx = 1 if fq in _NET_REQUEST_METHOD_APIS else 0
|
||||
a0 = node.args[_url_idx] if len(node.args) > _url_idx else None
|
||||
if a0 is None:
|
||||
for _kw in node.keywords or []:
|
||||
if _kw.arg in _NET_URL_KWARGS or _kw.arg in _NET_ADDR_KWARGS:
|
||||
|
|
@ -12397,18 +12449,33 @@ try:
|
|||
# same re-exported _sqlite3.connect, so wrap once and reassign every reachable attribute.
|
||||
import sqlite3 as _sq3
|
||||
|
||||
def _sqlite_uri_pct(_s):
|
||||
# Percent-decode a URI component with the captured _bi.chr / _bi.int so a sandboxed rebind
|
||||
# of chr / int cannot skew the decode (SQLite decodes file:%2Ftmp%2Fx -> /tmp/x itself).
|
||||
return _re.sub("%([0-9A-Fa-f]{2})", lambda _m: _bi.chr(_bi.int(_m.group(1), 16)), _s)
|
||||
|
||||
def _sqlite_uri_is_memory(_params):
|
||||
# True only when the query has a genuine mode=memory parameter. SQLite splits query
|
||||
# parameters on '&' and uses the FIRST occurrence of a repeated key, so an unknown key
|
||||
# (xmode=memory) or a later mode= is NOT in-memory -- a substring test wrongly treated
|
||||
# file:/tmp/escape.db?xmode=memory as in-memory and skipped path confinement.
|
||||
for _pair in _params.split("&"):
|
||||
_k, _sep, _v = _pair.partition("=")
|
||||
if _sqlite_uri_pct(_k) == "mode":
|
||||
return _sqlite_uri_pct(_v) == "memory"
|
||||
return False
|
||||
|
||||
def _sqlite_uri_path(_body):
|
||||
# Resolve a file: URI body (already stripped of the 'file:' prefix) to the concrete path
|
||||
# SQLite opens, or None for an in-memory / private target. Strips a //authority and
|
||||
# percent-decodes the filename (SQLite decodes file:%2Ftmp%2Fx -> /tmp/x itself), using
|
||||
# the captured _bi.chr / _bi.int so a sandboxed rebind of chr/int cannot skew the decode.
|
||||
# percent-decodes the filename.
|
||||
_pth, _, _params = _body.partition("?")
|
||||
if _pth == ":memory:" or _pth == "" or "mode=memory" in _params.lower():
|
||||
if _pth == ":memory:" or _pth == "" or _sqlite_uri_is_memory(_params):
|
||||
return None
|
||||
if _pth.startswith("//"):
|
||||
_slash = _pth.find("/", 2)
|
||||
_pth = _pth[_slash:] if _slash != -1 else ""
|
||||
return _re.sub("%([0-9A-Fa-f]{2})", lambda _m: _bi.chr(_bi.int(_m.group(1), 16)), _pth)
|
||||
return _sqlite_uri_pct(_pth)
|
||||
|
||||
def _sqlite_target_path(_db, _uri):
|
||||
# The concrete filesystem path to confine for a sqlite database argument, or None when it
|
||||
|
|
@ -12692,14 +12759,18 @@ try:
|
|||
# unguarded callable (open.__closure__[0].cell_contents, frame.f_locals['real']) or walk to
|
||||
# os / builtins. Mirrors the top-level _GADGET_DUNDERS; refuse them in a workdir helper too.
|
||||
_GUARD_GADGET_ATTRS = frozenset({
|
||||
"__subclasses__", "__bases__", "__base__", "__globals__", "__builtins__",
|
||||
"__subclasses__", "__bases__", "__base__", "__mro__", "mro", "__globals__", "__builtins__",
|
||||
"__closure__", "cell_contents", "f_locals", "f_globals", "f_back", "f_builtins",
|
||||
"tb_frame", "tb_next", "gi_frame", "cr_frame", "ag_frame",
|
||||
"settrace", "setprofile", "_getframe", "_current_frames", "currentframe",
|
||||
})
|
||||
# sys attributes that reach the import machinery: mutating them removes the guard's import
|
||||
# vetter so a sibling `import evil` loads unscanned.
|
||||
_GUARD_IMPORT_MACHINERY = frozenset({"meta_path", "path_hooks", "path_importer_cache"})
|
||||
# sys attributes that reach the import machinery: reading sys.modules recovers a guard-cached
|
||||
# module (sys.modules['os']) without an import, and mutating meta_path / path_hooks /
|
||||
# path_importer_cache removes the guard's import vetter so a sibling `import evil` loads
|
||||
# unscanned.
|
||||
_GUARD_IMPORT_MACHINERY = frozenset(
|
||||
{"modules", "meta_path", "path_hooks", "path_importer_cache"}
|
||||
)
|
||||
def _guard_attr_root(_v):
|
||||
# Base Name id of an attribute chain (os.path -> 'os'); None if not Name-rooted.
|
||||
while isinstance(_v, _gast.Attribute):
|
||||
|
|
@ -12744,8 +12815,45 @@ try:
|
|||
_sysmod.add(_al.asname or _al.name)
|
||||
elif _al.name == "importlib":
|
||||
_implib.add(_al.asname or _al.name)
|
||||
# Follow simple whole-module assignments (o = os; b = builtins; s = sys) so an aliased
|
||||
# receiver reached only through assignment -- not `import os as o` -- is tracked too.
|
||||
# Iterate to a fixpoint so a chain (o = os; p = o) is fully resolved before the sink checks.
|
||||
_alias_groups = (_recv, _bi, _deser, _sysmod, _implib)
|
||||
_changed = True
|
||||
while _changed:
|
||||
_changed = False
|
||||
for _nd in _gast.walk(_tree):
|
||||
if isinstance(_nd, _gast.Assign) and isinstance(_nd.value, _gast.Name):
|
||||
_srcid = _nd.value.id
|
||||
for _tgt in _nd.targets:
|
||||
if not isinstance(_tgt, _gast.Name):
|
||||
continue
|
||||
for _grp in _alias_groups:
|
||||
if _srcid in _grp and _tgt.id not in _grp:
|
||||
_grp.add(_tgt.id)
|
||||
_changed = True
|
||||
# Modules whose dynamic attribute / namespace-dict access (getattr / vars) is obfuscation.
|
||||
_obf = _recv | _bi | _deser | _sysmod | _implib
|
||||
def _guard_dyn_attr_hit(_grecv, _gname):
|
||||
# Classify a (receiver-root, attribute-name) dynamic lookup -- from getattr(recv, name)
|
||||
# or recv.__getattribute__(name) -- against the guarded sink sets. A non-constant name
|
||||
# (_gname is None) on a guarded receiver fails closed; a gadget dunder escapes on ANY
|
||||
# receiver; a sink name is refused only on its matching guarded receiver.
|
||||
if _gname is None:
|
||||
return _grecv in _obf
|
||||
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__"):
|
||||
return True
|
||||
if _grecv in _deser and _gname in _GUARD_DESER_ATTRS:
|
||||
return True
|
||||
if _grecv in _sysmod and _gname in _GUARD_IMPORT_MACHINERY:
|
||||
return True
|
||||
if _grecv in _implib and _gname in ("import_module", "reload", "__import__"):
|
||||
return True
|
||||
return False
|
||||
for _nd in _gast.walk(_tree):
|
||||
if isinstance(_nd, _gast.Import):
|
||||
for _al in _nd.names:
|
||||
|
|
@ -12871,27 +12979,41 @@ try:
|
|||
and isinstance(_nd.args[1].value, str)
|
||||
else None
|
||||
)
|
||||
if _gname is None:
|
||||
if _grecv in _obf:
|
||||
return True
|
||||
if _guard_dyn_attr_hit(_grecv, _gname):
|
||||
return True
|
||||
# os.__getattribute__('system')('id') / sys.__getattr__('modules') (bound), and the
|
||||
# unbound object.__getattribute__(os, 'system') / type.__getattribute__(...) forms:
|
||||
# a dynamic attribute lookup that reaches a guarded sink the builtin getattr(...)
|
||||
# branch and the direct-attribute checks miss. Classify it the same way.
|
||||
if (
|
||||
isinstance(_nd.func, _gast.Attribute)
|
||||
and _nd.func.attr in ("__getattribute__", "__getattr__")
|
||||
):
|
||||
_baseroot = _guard_attr_root(_nd.func.value)
|
||||
if _baseroot in ("object", "type") and len(_nd.args) >= 2:
|
||||
_grecv = (
|
||||
_guard_attr_root(_nd.args[0])
|
||||
if isinstance(_nd.args[0], (_gast.Name, _gast.Attribute))
|
||||
else None
|
||||
)
|
||||
_gnamenode = _nd.args[1]
|
||||
elif (
|
||||
isinstance(_nd.func.value, (_gast.Name, _gast.Attribute))
|
||||
and len(_nd.args) >= 1
|
||||
):
|
||||
_grecv = _baseroot
|
||||
_gnamenode = _nd.args[0]
|
||||
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__"):
|
||||
return True
|
||||
if _grecv in _deser and _gname in _GUARD_DESER_ATTRS:
|
||||
return True
|
||||
if _grecv in _sysmod and _gname in _GUARD_IMPORT_MACHINERY:
|
||||
return True
|
||||
if _grecv in _implib and _gname in (
|
||||
"import_module", "reload", "__import__"):
|
||||
_grecv = None
|
||||
_gnamenode = None
|
||||
if _grecv is not None:
|
||||
_gnm = (
|
||||
_gnamenode.value
|
||||
if isinstance(_gnamenode, _gast.Constant)
|
||||
and isinstance(_gnamenode.value, str)
|
||||
else None
|
||||
)
|
||||
if _guard_dyn_attr_hit(_grecv, _gnm):
|
||||
return True
|
||||
# vars(sys) / vars(os) / vars(builtins) exposes the module namespace dict for
|
||||
# indirect access (vars(sys)['meta_path'][:] = [...], vars(os)['system']).
|
||||
|
|
@ -12915,6 +13037,18 @@ try:
|
|||
return True
|
||||
if _skey in ("eval", "exec", "compile", "__import__"):
|
||||
return True
|
||||
# os.__dict__['system'] / sys.__dict__['modules'] / pickle.__dict__['loads'] --
|
||||
# a namespace-dict subscript reached through a guarded module's __dict__ is the
|
||||
# obfuscated twin of the direct sink attribute (the attribute checks miss the
|
||||
# subscript key). Fail closed wholesale, exactly like vars(<module>) above.
|
||||
# builtins is handled by the key-specific branch above (its dict legitimately
|
||||
# exposes many benign names), so exclude it here.
|
||||
if (
|
||||
isinstance(_nd.value, _gast.Attribute)
|
||||
and _nd.value.attr == "__dict__"
|
||||
and _guard_attr_root(_nd.value.value) in (_recv | _deser | _sysmod | _implib)
|
||||
):
|
||||
return True
|
||||
elif isinstance(_nd, _gast.Attribute):
|
||||
# An introspection / frame gadget attribute (open.__closure__[0].cell_contents,
|
||||
# frame.f_locals['real'], ().__class__.__bases__[0].__subclasses__()) recovers a
|
||||
|
|
@ -12940,6 +13074,12 @@ try:
|
|||
# mutation in submitted code; refuse it inside a vetted workdir module too.
|
||||
if _nd.attr in ("meta_path", "path_hooks", "path_importer_cache"):
|
||||
return True
|
||||
# sys.modules['os'].system(...) recovers a guard-cached module without an import,
|
||||
# bypassing both the denied-import path and the sink-root check (the receiver is a
|
||||
# subscript, not an os name). Deny access to sys.modules in a vetted workdir module.
|
||||
# Require a sys root so a benign .modules attribute (torch model.modules()) is kept.
|
||||
if _nd.attr == "modules" and _guard_attr_root(_nd.value) in _sysmod:
|
||||
return True
|
||||
return False
|
||||
def _guard_under_workdir(_p):
|
||||
return _p == _GUARD_WORKDIR_REAL or _p.startswith(_GUARD_WORKDIR_REAL + _os.sep)
|
||||
|
|
|
|||
|
|
@ -2461,3 +2461,117 @@ def test_sandboxed_docstring_same_line_write_denied(tmp_path):
|
|||
)
|
||||
assert "sandbox:" in out or "PermissionError" in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_assigned_os_alias_workdir_module_denied():
|
||||
# import os; o = os; o.system(...) -- a whole-module assignment alias (not `import os as o`)
|
||||
# must be followed in the vetter pre-pass so the aliased sink receiver is recognized.
|
||||
_assert_workdir_module_denied(
|
||||
"backstop-workdir-osassign",
|
||||
"evilassign",
|
||||
"import os\no = os\nprint('R61_ASSIGN')\no.system('echo PWN')\n",
|
||||
"R61_ASSIGN",
|
||||
)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_module_dict_subscript_workdir_module_denied():
|
||||
# import os; os.__dict__['system'](...) -- a namespace-dict subscript through a guarded
|
||||
# module's __dict__ reaches the sink the attribute checks miss; fail closed like vars(os).
|
||||
_assert_workdir_module_denied(
|
||||
"backstop-workdir-osdict",
|
||||
"evildict",
|
||||
"import os\nprint('R61_DICT')\nos.__dict__['system']('echo PWN')\n",
|
||||
"R61_DICT",
|
||||
)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_module_getattribute_workdir_module_denied():
|
||||
# import os; os.__getattribute__('system')(...) -- a dynamic attribute lookup via the module
|
||||
# dunder reaches the sink the builtin getattr(...) branch misses.
|
||||
_assert_workdir_module_denied(
|
||||
"backstop-workdir-osgetattr",
|
||||
"evilga",
|
||||
"import os\nprint('R61_GA')\nos.__getattribute__('system')('echo PWN')\n",
|
||||
"R61_GA",
|
||||
)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_mro_recovery_workdir_module_denied():
|
||||
# io.FileIO.__mro__[1](...) recovers an unguarded base class in a vetted workdir helper; the
|
||||
# vetter must treat __mro__ / mro as a gadget.
|
||||
_assert_workdir_module_denied(
|
||||
"backstop-workdir-mro",
|
||||
"evilmro",
|
||||
"import io\nprint('R61_MRO')\nc = io.FileIO.__mro__[1]\n",
|
||||
"R61_MRO",
|
||||
)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_sys_modules_subscript_workdir_module_denied():
|
||||
# import sys; sys.modules['os'].system(...) recovers the guard-cached os module without an
|
||||
# import, bypassing the denied-import path; access to sys.modules must be denied.
|
||||
_assert_workdir_module_denied(
|
||||
"backstop-workdir-sysmods",
|
||||
"evilsysmods",
|
||||
"import sys\nprint('R61_SYSMODS')\nsys.modules['os'].system('echo PWN')\n",
|
||||
"R61_SYSMODS",
|
||||
)
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_benign_os_alias_workdir_module_allowed():
|
||||
# A benign whole-module alias that only calls a NON-sink os attribute (o = os; o.getcwd())
|
||||
# must still import -- the alias-following must not over-block ordinary os use.
|
||||
session = "backstop-workdir-benignalias"
|
||||
workdir = get_sandbox_workdir(session)
|
||||
with open(os.path.join(workdir, "okalias.py"), "w") as f:
|
||||
f.write("import os\no = os\nCWD = o.getcwd()\nprint('OKALIAS_' + 'BODY')\n")
|
||||
try:
|
||||
out = _python_exec(
|
||||
"import okalias; print('IMPORTED_' + 'OK')",
|
||||
None,
|
||||
30,
|
||||
session,
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "IMPORTED_OK" in out
|
||||
assert "sandbox:" not in out
|
||||
finally:
|
||||
os.remove(os.path.join(workdir, "okalias.py"))
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_sqlite_uri_xmode_memory_escape_denied(tmp_path):
|
||||
# file:<path>?xmode=memory is an on-disk file (SQLite ignores the unknown xmode key), so the
|
||||
# in-memory skip must NOT apply -- an escaping path via a uri connection is confined.
|
||||
target = tmp_path / "sqlite_uri_escape.db"
|
||||
out = _python_exec(
|
||||
f"import sqlite3\nsqlite3.connect('file:{target}?xmode=memory', uri=True)\nprint('OPENED')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-uri-xmode",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "sandbox:" in out or "PermissionError" in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_sqlite_uri_real_memory_allowed():
|
||||
# A genuine mode=memory URI parameter is a real in-memory database and must stay allowed.
|
||||
out = _python_exec(
|
||||
"import sqlite3\n"
|
||||
"c = sqlite3.connect('file:r61mem?mode=memory&cache=shared', uri=True)\n"
|
||||
"c.execute('create table t(x)')\nprint('MEM_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-uri-mem",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "MEM_OK" in out
|
||||
assert "sandbox:" not in out
|
||||
|
|
|
|||
|
|
@ -5931,3 +5931,74 @@ class TestRound60Bypasses:
|
|||
)
|
||||
def test_round60_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
||||
|
||||
class TestRound61Bypasses:
|
||||
# Module-level request(method, url) APIs carry the URL at arg1 (arg0 is the HTTP method), so
|
||||
# the host must be read from the second argument -- not the method string.
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import requests\nrequests.request('GET', 'http://169.254.169.254/')",
|
||||
"import httpx\nhttpx.request('GET', 'http://169.254.169.254/')",
|
||||
"import urllib3\nurllib3.request('GET', 'http://169.254.169.254/')",
|
||||
"import requests\nrequests.api.request('GET', 'http://169.254.169.254/')",
|
||||
],
|
||||
)
|
||||
def test_request_method_url_at_arg1_blocked(self, code):
|
||||
_blocked(code, expect_phrase = "cloud-metadata host")
|
||||
|
||||
# Public network client entry points (requests.api.*, ftplib, smtplib) were not in the prefix
|
||||
# table, so a metadata / untrusted host reached through them bypassed the allowlist.
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import requests\nrequests.api.get('http://169.254.169.254/')",
|
||||
"import ftplib\nftplib.FTP('169.254.169.254')",
|
||||
"import ftplib\nftplib.FTP_TLS('169.254.169.254')",
|
||||
"import ftplib\nftplib.FTP(host='169.254.169.254')",
|
||||
"import smtplib\nsmtplib.SMTP('169.254.169.254')",
|
||||
"import smtplib\nsmtplib.SMTP_SSL('169.254.169.254')",
|
||||
],
|
||||
)
|
||||
def test_network_client_entry_points_metadata_blocked(self, code):
|
||||
_blocked(code, expect_phrase = "cloud-metadata host")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
"import ftplib\nftplib.FTP('untrusted.example')",
|
||||
"import smtplib\nsmtplib.SMTP('untrusted.example', 587)",
|
||||
"import requests\nrequests.api.get('http://untrusted.example/')",
|
||||
],
|
||||
)
|
||||
def test_network_client_entry_points_untrusted_blocked(self, code):
|
||||
_blocked(code, expect_phrase = "not in sandbox allowlist")
|
||||
|
||||
# A sqlite operand whose absolute path carries an UNKNOWN query key that merely contains the
|
||||
# text mode=memory (?xmode=memory) is still an on-disk file, so the escaping path must block --
|
||||
# the in-memory skip only applies to a genuine first mode=memory parameter.
|
||||
@pytest.mark.parametrize(
|
||||
"cmd",
|
||||
[
|
||||
"sqlite3 '/tmp/escape.db?xmode=memory' 'create table t(x)'",
|
||||
"sqlite3 '/tmp/escape.db?cache=shared&xmode=memory' 'create table t(x)'",
|
||||
],
|
||||
)
|
||||
def test_sqlite_shell_xmode_memory_blocked(self, cmd):
|
||||
_blocked(_sh(cmd), expect_phrase = "blocked command")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# Benign round-61 forms stay allowed.
|
||||
"import requests\nrequests.request('GET', 'https://huggingface.co/x')",
|
||||
"import requests\nrequests.api.get('https://huggingface.co/x')",
|
||||
"import ftplib\nftplib.FTP('huggingface.co')",
|
||||
"import smtplib\nsmtplib.SMTP('huggingface.co')",
|
||||
_sh("sqlite3 '/tmp/x?mode=memory' 'create table t(x)'"), # genuine in-memory URI
|
||||
_sh("sqlite3 'local.db' 'create table t(x)'"), # workdir-relative db
|
||||
],
|
||||
)
|
||||
def test_round61_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue