Harden sandbox: taskset wrapper, sqlite3.Connection ctor + durable authorizer, namespace-dict sink aliases, injected subprocess, shelve reads
Close six issues Codex found on the round-56 branch (all P1).
- taskset exec wrapper: taskset [options] <mask | -c cpu-list> <command> execs the
following command, but taskset was not a command-prefix wrapper, so taskset 1
touch /tmp/p resolved to nothing and the write slipped. Add taskset to the
wrapper set with its -c / --cpu-list (and -p / --pid) operand flags, and extend
the wrapper numeric-arg skip to cover a hex affinity mask (0x3) and a cpu-list
(0,1 / 0-3), so the wrapped command is resolved and scanned.
- sqlite3.Connection constructor confinement: wrapping only connect() left the
public constructors unconfined, so sqlite3.Connection('/tmp/escape.db') /
_sqlite3.Connection(...) created a database outside the workdir via the native
extension. Route construction through a guarded Connection subclass whose
__init__ confines the database path, and replace the module Connection attribute
with it (isinstance stays valid); connect() forces the subclass as its factory.
- durable ATTACH / VACUUM authorizer: installing the authorizer once on the
returned connection was not durable -- sandboxed code could call
conn.set_authorizer(None) and then ATTACH DATABASE '/tmp/escape.db' / VACUUM
INTO an outside file. The guarded Connection overrides set_authorizer to compose
the workdir confinement ahead of any caller callback and keep it on
set_authorizer(None), so the confinement cannot be removed.
- namespace-dict sink aliases: globals()/locals()/vars()[key] only blocked literal
builtins / dangerous-module keys, so import os; f = os.system;
globals()['f']('touch /tmp/p') passed. Resolve the key through the scope alias
index too -- a shell / exec-builtin / deserializer sink alias makes the
namespace-dict lookup the sink itself.
- dependency-injected subprocess / pty: a workdir helper receiving the module as
an argument (def f(subprocess): subprocess.run([...])) has no import to reject,
and the vetter's call check only rooted os / posix. Reject a subprocess / pty
child-spawn method rooted at a receiver named subprocess / pty in the vetter, and
-- robust to the callee's parameter name -- flag the subprocess / pty module
passed by reference (f(subprocess)) as a first-class dangerous value in the
submitted code, mirroring the existing os.system-by-reference block.
- shelve reads as pickle deserialization: shelve is a dbm-backed dict that
unpickles a value on every read (shelf[key], shelf.get(key)), so shelve.open()
on an attacker-planted dbm runs a pickle reduce payload just like pickle.load
(which is already blocked). Model shelve.open as a deserialization sink. The
read can be aliased (d = shelve.open(...); d[k]), so the open() gateway call is
flagged; a pure write-only shelf never unpickles, so blocking it is an accepted
narrow tradeoff.
Regression coverage: TestRound57Bypasses in tests/test_sandbox_tools.py (taskset
mask / cpu-list / nested wrappers + benign taskset; namespace-dict sink aliases
via globals / locals / vars incl. folded key and a pickle alias; injected
subprocess / pty module by reference incl. an aliased import; shelve.open read /
aliased read / get / import alias / from-import; and a round57 benign-allowed set)
and test_sandbox_runtime_backstop.py (sqlite3.Connection and _sqlite3.Connection
constructor escape denied + local allowed; set_authorizer(None) ATTACH / VACUUM
escape still denied; a caller authorizer still composes).
This commit is contained in:
danielhanchen
parent
570c3ff2f2
commit
cb243a23f1
3 changed files with 360 additions and 36 deletions
|
|
@ -739,6 +739,11 @@ _COMMAND_PREFIXES = frozenset(
|
|||
# watch [options] command: repeatedly runs command (via sh -c, or exec with -x), so
|
||||
# watch -x touch /tmp/x / watch -n 2 rm -rf / must resolve to the wrapped command.
|
||||
"watch",
|
||||
# taskset [options] <mask | -c cpu-list> <command> [<arg>...]: util-linux affinity
|
||||
# wrapper that execs the following command, so taskset 1 touch /tmp/x / taskset -c 0,1
|
||||
# rm -rf must resolve to the wrapped command (the mask / cpu-list is skipped as a
|
||||
# numeric operand). The -p PID form operates on an existing process and execs nothing.
|
||||
"taskset",
|
||||
}
|
||||
)
|
||||
# A shell assignment prefix: NAME=value or NAME+=value (bash append). The optional `+` is part
|
||||
|
|
@ -806,6 +811,9 @@ _WRAPPER_OPERAND_FLAGS = {
|
|||
"time": frozenset({"-f", "--format", "-o", "--output"}),
|
||||
"chrt": frozenset({"-T", "--sched-runtime", "-P", "--sched-period", "-D", "--sched-deadline"}),
|
||||
"watch": frozenset({"-n", "--interval"}),
|
||||
# taskset -c CPU-LIST cmd (the cpu-list is a separated operand); -p PID targets an existing
|
||||
# process (no command follows). The bare hex / decimal mask form is skipped as a numeric arg.
|
||||
"taskset": frozenset({"-c", "--cpu-list", "-p", "--pid"}),
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -858,8 +866,23 @@ def _is_wrapper_numeric_arg(token: str) -> bool:
|
|||
t = token.lstrip("-")
|
||||
if not t:
|
||||
return False
|
||||
# A hex affinity mask (taskset 0x3 cmd).
|
||||
if t[:2].lower() == "0x" and len(t) > 2:
|
||||
try:
|
||||
int(t, 16)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
# Strip a single trailing GNU timeout duration unit (timeout 5m / 0.5s).
|
||||
if len(t) > 1 and t[-1] in "smhd":
|
||||
t = t[:-1]
|
||||
# A cpu-list / affinity mask of digits with , and - separators (taskset -c 0,1 / 0-3 cmd).
|
||||
if (
|
||||
any(c in ",-" for c in t)
|
||||
and all(c in "0123456789,-" for c in t)
|
||||
and any(c.isdigit() for c in t)
|
||||
):
|
||||
return True
|
||||
try:
|
||||
float(t)
|
||||
return True
|
||||
|
|
@ -4698,12 +4721,18 @@ _CODE_DESERIALIZE_SINKS = frozenset(
|
|||
"yaml.unsafe_load_all",
|
||||
"yaml.full_load",
|
||||
"yaml.full_load_all",
|
||||
# shelve is a dbm-backed dict that UNPICKLES a value on every read (shelf[key],
|
||||
# shelf.get(key)), so shelve.open() on an attacker-planted dbm runs a pickle reduce
|
||||
# payload just like pickle.load. The read can be aliased (d = shelve.open(...); d[k]),
|
||||
# so the open() gateway call is flagged rather than only the direct-chain subscript.
|
||||
# (A pure write-only shelf never unpickles; blocking it is an accepted narrow tradeoff.)
|
||||
"shelve.open",
|
||||
}
|
||||
)
|
||||
# Modules whose load/loads/decode entry points run a pickle reduce payload; used to
|
||||
# resolve `import pickle as p; p.loads(x)` and `from pickle import loads as l`.
|
||||
_DESERIALIZE_MODULES = frozenset(
|
||||
{"pickle", "marshal", "dill", "cloudpickle", "_pickle", "jsonpickle", "yaml"}
|
||||
{"pickle", "marshal", "dill", "cloudpickle", "_pickle", "jsonpickle", "yaml", "shelve"}
|
||||
)
|
||||
# Modules exposing an Unpickler class whose .load() runs the same reduce payload as *.load:
|
||||
# pickle.Unpickler(f).load() / dill.Unpickler(f).load() bypass the *.load sink-name check.
|
||||
|
|
@ -7904,6 +7933,14 @@ def _check_signal_escape_patterns(
|
|||
_ds = self.deserialize_aliases.get(n.id)
|
||||
if _ds is not None:
|
||||
return f"{_ds} (deserialize)"
|
||||
# The subprocess / pty MODULE passed by reference (f(subprocess)) is a child-spawn
|
||||
# primitive a callee can invoke as subprocess.run(...) with an unguarded escape;
|
||||
# the recursive analyzer never sees that call, and a workdir helper receiving the
|
||||
# module as a parameter cannot resolve it. Flag the module reference itself.
|
||||
if n.id in self.subprocess_aliases:
|
||||
return "subprocess module (child spawn)"
|
||||
if n.id in self.pty_aliases:
|
||||
return "pty module (child spawn)"
|
||||
if _analyzer_on:
|
||||
_r = _scope_idx.resolve(n.id, n, "shell")
|
||||
if _r in _SHELL_EXEC_FUNCS:
|
||||
|
|
@ -9643,17 +9680,34 @@ def _check_signal_escape_patterns(
|
|||
# dangerous literal key off a bare globals()/locals()/vars() call.
|
||||
if isinstance(node.ctx, ast.Load) and self._is_namespace_dict_expr(v):
|
||||
key = _const_fold(node.slice, _const_env)
|
||||
if isinstance(key, str) and (
|
||||
key in ("__builtins__", "__builtin__")
|
||||
or key.split(".")[0] in _DANGEROUS_IMPORT_NAMES
|
||||
):
|
||||
dynamic_exec.append(
|
||||
{
|
||||
"type": "dynamic_exec",
|
||||
"line": getattr(node, "lineno", -1),
|
||||
"description": "namespace-dict access to builtins / a sensitive module",
|
||||
}
|
||||
if isinstance(key, str):
|
||||
_ns_hit = (
|
||||
key in ("__builtins__", "__builtin__")
|
||||
or key.split(".")[0] in _DANGEROUS_IMPORT_NAMES
|
||||
)
|
||||
if not _ns_hit and _analyzer_on:
|
||||
# The namespace dict also exposes a module-level / local ALIAS bound to a
|
||||
# sink (import os; f = os.system; globals()['f']('touch /tmp/p')), which
|
||||
# the literal-key check above misses. Resolve the key through the alias
|
||||
# index -- a shell / exec-builtin / deserializer sink alias makes the
|
||||
# namespace-dict lookup the sink itself. resolve() walks local->module,
|
||||
# matching globals() (module) and locals()/vars() (local) in the usual case.
|
||||
_ns_hit = (
|
||||
_scope_idx.resolve(key, node, "shell") is not None
|
||||
or _scope_idx.resolve(key, node, "execb") is not None
|
||||
or _scope_idx.resolve(key, node, "deser") is not None
|
||||
)
|
||||
if _ns_hit:
|
||||
dynamic_exec.append(
|
||||
{
|
||||
"type": "dynamic_exec",
|
||||
"line": getattr(node, "lineno", -1),
|
||||
"description": (
|
||||
"namespace-dict access to builtins / a sensitive module "
|
||||
"or a sink alias"
|
||||
),
|
||||
}
|
||||
)
|
||||
self.generic_visit(node)
|
||||
|
||||
def visit_ExceptHandler(self, node):
|
||||
|
|
@ -11914,52 +11968,109 @@ try:
|
|||
return 0 # SQLITE_OK
|
||||
return _auth
|
||||
|
||||
def _guard_sqlite_connect(_orig):
|
||||
@_gwraps(_orig)
|
||||
def w(*a, **k):
|
||||
_SqliteConnBase = _sq3.Connection # the original (unguarded) Connection class
|
||||
|
||||
class _GuardedSqliteConnection(_SqliteConnBase):
|
||||
# A Connection subclass that (1) confines the database path AT CONSTRUCTION, so the direct
|
||||
# constructor forms sqlite3.Connection('/tmp/x') / _sqlite3.Connection(...) are guarded just
|
||||
# like connect(); and (2) makes the ATTACH / VACUUM INTO authorizer DURABLE:
|
||||
# set_authorizer(cb) composes the workdir confinement AHEAD of the caller's callback, and
|
||||
# set_authorizer(None) keeps the confinement -- so sandboxed code cannot drop the hook and
|
||||
# then ATTACH DATABASE '/tmp/escape.db' / VACUUM INTO an outside file via native code.
|
||||
def __init__(self, *a, **k):
|
||||
if a:
|
||||
_db = a[0]
|
||||
elif "database" in k:
|
||||
_db = k["database"]
|
||||
else:
|
||||
return _orig(*a, **k) # let sqlite3 raise its own TypeError
|
||||
_db = None
|
||||
_uri = bool(k.get("uri", False))
|
||||
# Materialize a path-like once so a stateful __fspath__ cannot pass the check
|
||||
# with an in-workdir value and then hand sqlite a different outside path.
|
||||
if not isinstance(_db, (str, bytes)):
|
||||
# Materialize a path-like once so a stateful __fspath__ cannot pass the check with an
|
||||
# in-workdir value and then hand sqlite a different outside path.
|
||||
if _db is not None and not isinstance(_db, (str, bytes)):
|
||||
_db = _fspath1(_db)
|
||||
if not _sqlite_path_ok(_db, _uri):
|
||||
_deny(_db, "sqlite3.connect")
|
||||
if a:
|
||||
_conn = _orig(_db, *a[1:], **k)
|
||||
else:
|
||||
k = dict(k)
|
||||
k["database"] = _db
|
||||
_conn = _orig(**k)
|
||||
# Confine ATTACH / VACUUM INTO targets on the live connection too. Best-effort: a
|
||||
# build without set_authorizer simply lacks this extra confinement.
|
||||
if a:
|
||||
a = (_db,) + tuple(a[1:])
|
||||
else:
|
||||
k = dict(k)
|
||||
k["database"] = _db
|
||||
if _db is not None and not _sqlite_path_ok(_db, _uri):
|
||||
_deny(_db, "sqlite3.Connection")
|
||||
_SqliteConnBase.__init__(self, *a, **k)
|
||||
self._sandbox_uri_on = _uri
|
||||
# Install the initial confinement authorizer through the durable override below.
|
||||
try:
|
||||
_conn.set_authorizer(_make_sqlite_authorizer(_uri))
|
||||
self.set_authorizer(None)
|
||||
except Exception:
|
||||
pass
|
||||
return _conn
|
||||
|
||||
def set_authorizer(self, callback, *a, **k):
|
||||
_confine = _make_sqlite_authorizer(getattr(self, "_sandbox_uri_on", False))
|
||||
|
||||
def _composed(_action, _a1, _a2, _dbname, _source):
|
||||
if _confine(_action, _a1, _a2, _dbname, _source) != 0:
|
||||
return 1 # SQLITE_DENY -- an escaping ATTACH / VACUUM INTO target
|
||||
if callback is None:
|
||||
return 0 # SQLITE_OK
|
||||
return callback(_action, _a1, _a2, _dbname, _source)
|
||||
|
||||
# Route through the ORIGINAL C method (not the possibly-reassigned module attribute) so
|
||||
# the confinement is always reinstalled and this override cannot recurse.
|
||||
return _SqliteConnBase.set_authorizer(self, _composed, *a, **k)
|
||||
|
||||
_guard_conn_cache = {}
|
||||
|
||||
def _combined_guard_conn(_user):
|
||||
# A caller-supplied Connection factory is COMBINED with the guard subclass (guard methods
|
||||
# take MRO precedence) so the path confinement + durable authorizer still apply.
|
||||
_g = _guard_conn_cache.get(_user)
|
||||
if _g is None:
|
||||
try:
|
||||
_g = type("SandboxGuardedConnection", (_GuardedSqliteConnection, _user), {})
|
||||
except Exception:
|
||||
_g = _GuardedSqliteConnection
|
||||
_guard_conn_cache[_user] = _g
|
||||
return _g
|
||||
|
||||
def _guard_sqlite_connect(_orig):
|
||||
@_gwraps(_orig)
|
||||
def w(*a, **k):
|
||||
# Force our guarded Connection subclass as the factory so the returned connection is
|
||||
# path-confined and carries the durable authorizer; a caller factory is combined in.
|
||||
_fac = k.get("factory")
|
||||
if _fac is None:
|
||||
k = dict(k)
|
||||
k["factory"] = _GuardedSqliteConnection
|
||||
elif not (isinstance(_fac, type) and issubclass(_fac, _GuardedSqliteConnection)):
|
||||
k = dict(k)
|
||||
k["factory"] = _combined_guard_conn(_fac)
|
||||
return _orig(*a, **k)
|
||||
return w
|
||||
|
||||
_sq3_orig_connect = _sq3.connect
|
||||
_sq3_guarded_connect = _guard_sqlite_connect(_sq3_orig_connect)
|
||||
_sq3_guarded_connect = _guard_sqlite_connect(_sq3.connect)
|
||||
_sq3.connect = _sq3_guarded_connect
|
||||
try:
|
||||
_sq3.dbapi2.connect = _sq3_guarded_connect
|
||||
except Exception:
|
||||
pass
|
||||
# The native _sqlite3 C extension still exposes the ORIGINAL connect, and it is importable
|
||||
# directly (import _sqlite3; _sqlite3.connect('/tmp/escape.db')), bypassing the two Python
|
||||
# bindings above. Wrap it too so the low-level entry point is confined; module attribute
|
||||
# assignment on a C extension is allowed, but guard it in case a build disallows it.
|
||||
# Confine the direct constructor forms too (sqlite3.Connection('/tmp/escape.db') /
|
||||
# sqlite3.dbapi2.Connection / _sqlite3.Connection), which never go through connect(). Replacing
|
||||
# the module attribute with the guarded subclass keeps isinstance() working (it IS a Connection)
|
||||
# while routing construction through the confining __init__.
|
||||
_sq3.Connection = _GuardedSqliteConnection
|
||||
try:
|
||||
_sq3.dbapi2.Connection = _GuardedSqliteConnection
|
||||
except Exception:
|
||||
pass
|
||||
# The native _sqlite3 C extension still exposes the ORIGINAL connect / Connection, importable
|
||||
# directly (import _sqlite3; _sqlite3.connect('/tmp/escape.db') / _sqlite3.Connection(...)),
|
||||
# bypassing the bindings above. Wrap them too; module attribute assignment on a C extension is
|
||||
# allowed, but guard it in case a build disallows it.
|
||||
try:
|
||||
import _sqlite3 as _lowsq3
|
||||
|
||||
_lowsq3.connect = _guard_sqlite_connect(_lowsq3.connect)
|
||||
_lowsq3.Connection = _GuardedSqliteConnection
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
|
|
@ -12071,6 +12182,16 @@ try:
|
|||
"posix_spawn", "posix_spawnp",
|
||||
})
|
||||
_GUARD_EXEC_MODS = frozenset({"subprocess", "pty"})
|
||||
# Child-spawning methods of the exec modules above. A workdir helper that IMPORTS subprocess /
|
||||
# pty is already refused, but one that receives the module as an argument (def f(subprocess):
|
||||
# subprocess.run([...])) has no import to reject, so a call rooted at a receiver literally named
|
||||
# subprocess / pty (the injected module) is refused here regardless of import.
|
||||
_GUARD_EXEC_MOD_ATTRS = {
|
||||
"subprocess": frozenset(
|
||||
{"run", "Popen", "call", "check_call", "check_output", "getoutput", "getstatusoutput"}
|
||||
),
|
||||
"pty": frozenset({"spawn", "fork"}),
|
||||
}
|
||||
# os / posix expose the exec-attr sinks (os.system, os.execv, ...); a sink attribute rooted at
|
||||
# one of these is a command-exec sink even without a direct call (x = os.system; x('id')).
|
||||
_GUARD_EXEC_RECEIVERS = frozenset({"os", "posix"})
|
||||
|
|
@ -12209,6 +12330,17 @@ try:
|
|||
and _guard_attr_root(_nd.func.value) in _recv
|
||||
):
|
||||
return True
|
||||
# A subprocess / pty child-spawn (subprocess.run([...]) / pty.spawn(...)) rooted at
|
||||
# a receiver literally named subprocess / pty. A helper that IMPORTS these is already
|
||||
# refused above; this catches the dependency-injected form (def f(subprocess):
|
||||
# subprocess.run(...)) that has no import statement to reject.
|
||||
if isinstance(_nd.func, _gast.Attribute):
|
||||
_mroot = _guard_attr_root(_nd.func.value)
|
||||
if (
|
||||
_mroot in _GUARD_EXEC_MOD_ATTRS
|
||||
and _nd.func.attr in _GUARD_EXEC_MOD_ATTRS[_mroot]
|
||||
):
|
||||
return True
|
||||
if isinstance(_nd.func, _gast.Name) and _nd.func.id in (
|
||||
"eval", "exec", "compile", "__import__"):
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -382,6 +382,113 @@ def test_sandboxed_sqlite3_attach_local_allowed():
|
|||
assert "sandbox:" not in out
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_sqlite3_connection_constructor_escape_denied(tmp_path):
|
||||
# The public sqlite3.Connection('/outside.db') constructor creates the DB via the native
|
||||
# extension without going through the guarded connect(); the guarded Connection subclass must
|
||||
# confine the path at construction.
|
||||
target = tmp_path / "conn_ctor_escape.db"
|
||||
out = _python_exec(
|
||||
f"import sqlite3\nc = sqlite3.Connection({str(target)!r})\n"
|
||||
"c.execute('create table t(x)'); c.commit(); print('CTOR_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-ctor-escape",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "CTOR_OK" not in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_low_level_sqlite3_connection_constructor_escape_denied(tmp_path):
|
||||
# _sqlite3.Connection is the raw C constructor, importable directly; it must be guarded too.
|
||||
target = tmp_path / "low_conn_ctor_escape.db"
|
||||
out = _python_exec(
|
||||
f"import _sqlite3\nc = _sqlite3.Connection({str(target)!r})\n"
|
||||
"c.execute('create table t(x)'); print('LOW_CTOR_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-low-ctor-escape",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "LOW_CTOR_OK" not in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
def test_sandboxed_sqlite3_connection_constructor_local_allowed():
|
||||
# A workdir-relative sqlite3.Connection(...) opens and is usable.
|
||||
out = _python_exec(
|
||||
"import sqlite3\n"
|
||||
"c = sqlite3.Connection('ctor_local.db')\n"
|
||||
"c.execute('create table if not exists t(x)'); c.close(); print('CTOR_LOCAL_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-ctor-local",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "CTOR_LOCAL_OK" in out
|
||||
assert "sandbox:" not in out
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_sqlite3_set_authorizer_none_attach_escape_denied(tmp_path):
|
||||
# Removing the confinement authorizer (set_authorizer(None)) must NOT re-open the ATTACH
|
||||
# escape: the guarded Connection composes its workdir confinement ahead of any caller
|
||||
# callback and keeps it on set_authorizer(None).
|
||||
target = tmp_path / "auth_removed_attach_escape.db"
|
||||
out = _python_exec(
|
||||
"import sqlite3\n"
|
||||
"c = sqlite3.connect('backstop_authrm.db')\n"
|
||||
"c.set_authorizer(None)\n"
|
||||
f"c.execute(\"ATTACH DATABASE '{target}' AS ext\")\n"
|
||||
"print('AUTH_REMOVED_ATTACH_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-authrm-attach",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "AUTH_REMOVED_ATTACH_OK" not in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_sqlite3_set_authorizer_none_vacuum_escape_denied(tmp_path):
|
||||
# The same durability holds for VACUUM INTO after set_authorizer(None).
|
||||
target = tmp_path / "auth_removed_vacuum_escape.db"
|
||||
out = _python_exec(
|
||||
"import sqlite3\n"
|
||||
"c = sqlite3.connect('backstop_authrm_v.db')\n"
|
||||
"c.execute('create table t(x)')\n"
|
||||
"c.set_authorizer(None)\n"
|
||||
f"c.execute(\"VACUUM INTO '{target}'\")\n"
|
||||
"print('AUTH_REMOVED_VACUUM_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-authrm-vacuum",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "AUTH_REMOVED_VACUUM_OK" not in out
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
def test_sandboxed_sqlite3_user_authorizer_still_runs():
|
||||
# A caller-supplied authorizer still composes (benign work is not broken by the confinement).
|
||||
out = _python_exec(
|
||||
"import sqlite3\n"
|
||||
"c = sqlite3.connect('backstop_userauth.db')\n"
|
||||
"def ok(*a):\n return sqlite3.SQLITE_OK\n"
|
||||
"c.set_authorizer(ok)\n"
|
||||
"c.execute('create table if not exists t(x)'); c.close(); print('USERAUTH_OK')",
|
||||
None,
|
||||
30,
|
||||
"backstop-sqlite-userauth",
|
||||
disable_sandbox = False,
|
||||
)
|
||||
assert "USERAUTH_OK" in out
|
||||
assert "sandbox:" not in out
|
||||
|
||||
|
||||
@_POSIX_ONLY
|
||||
def test_sandboxed_getattr_gadget_dunder_workdir_module_denied():
|
||||
# A workdir helper recovering the guard wrapper's original open via a getattr gadget dunder
|
||||
|
|
|
|||
|
|
@ -5570,3 +5570,88 @@ class TestRound56Bypasses:
|
|||
)
|
||||
def test_round56_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
||||
|
||||
class TestRound57Bypasses:
|
||||
# taskset [options] <mask | -c cpu-list> <command> execs the following command, so it must be
|
||||
# a command-prefix wrapper resolving to the wrapped command (the mask / cpu-list is skipped).
|
||||
@pytest.mark.parametrize(
|
||||
"cmd",
|
||||
[
|
||||
"taskset 1 touch /tmp/p",
|
||||
"taskset 0x3 touch /tmp/p",
|
||||
"taskset -c 0,1 touch /tmp/p",
|
||||
"taskset -c 0-3 rm -rf /tmp/x",
|
||||
"taskset 1 nice -n 5 touch /tmp/p",
|
||||
],
|
||||
)
|
||||
def test_taskset_wrapper_resolves_command(self, cmd):
|
||||
_blocked(_sh(cmd), expect_phrase = "blocked command")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"cmd",
|
||||
[
|
||||
# taskset wrapping a benign command stays allowed; the -p PID form execs nothing.
|
||||
"taskset 1 echo hi",
|
||||
"taskset -c 0,1 echo hi",
|
||||
"taskset -p 1234",
|
||||
],
|
||||
)
|
||||
def test_taskset_benign_allowed(self, cmd):
|
||||
_ok(_sh(cmd))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# A namespace-dict lookup of an alias bound to a sink (f = os.system; globals()['f'](...))
|
||||
# resolves through the alias index, not just literal builtins / module keys.
|
||||
"import os\nf = os.system\nglobals()['f']('touch /tmp/p')",
|
||||
"import os\nf = os.system\nlocals()['f']('touch /tmp/p')",
|
||||
"import os\nf = os.system\nvars()['f']('touch /tmp/p')",
|
||||
"import os\nf = os.system\nglobals()['f' + '']('touch /tmp/p')",
|
||||
"import pickle\np = pickle.loads\nglobals()['p'](b'x')",
|
||||
],
|
||||
)
|
||||
def test_namespace_dict_sink_alias_blocked(self, code):
|
||||
_blocked(code, expect_phrase = "namespace-dict access")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# Passing the subprocess / pty MODULE by reference to a helper (which can spawn a child
|
||||
# the recursive analyzer never sees) is blocked regardless of the callee's parameter name.
|
||||
"import subprocess\ndef f(m):\n m.run(['touch', '/tmp/p'])\nf(subprocess)",
|
||||
"import subprocess as sp\ndef f(m):\n m.run(['id'])\nf(sp)",
|
||||
"import pty\ndef f(m):\n m.spawn(['/bin/sh'])\nf(pty)",
|
||||
],
|
||||
)
|
||||
def test_injected_subprocess_module_blocked(self, code):
|
||||
_blocked(code, expect_phrase = "child spawn")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# shelve.open() unpickles values on read, so it is a deserialization sink like
|
||||
# pickle.load; the open() gateway is flagged to cover aliased reads.
|
||||
"import shelve\nx = shelve.open('db')['k']",
|
||||
"import shelve\nd = shelve.open('db')\nx = d['k']",
|
||||
"import shelve\nd = shelve.open('db')\nx = d.get('k')",
|
||||
"import shelve as s\ns.open('db')",
|
||||
"from shelve import open as o\no('db')",
|
||||
],
|
||||
)
|
||||
def test_shelve_open_deserialize_blocked(self, code):
|
||||
_blocked(code, expect_phrase = "shelve.open")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
# Benign forms across the round-57 checks stay allowed.
|
||||
"import os\nf = os.system\n", # a bare alias assignment is not a namespace-dict call
|
||||
"x = 5\nprint(globals()['x'])", # benign namespace-dict read of a non-sink
|
||||
"import subprocess\nsubprocess.run(['ls'])", # a direct benign-command subprocess run
|
||||
"import os\ndef f(m):\n return m.getcwd()\nf(os)", # passing os (not subprocess/pty)
|
||||
],
|
||||
)
|
||||
def test_round57_benign_allowed(self, code):
|
||||
_ok(code)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue