"""Adversarial sandbox test suite — the R3/H3 ruin-mode gate. Each test below is named after a specific attack documented in STACK.md §7 "Adversarial unit tests" and CONTEXT.md D-13. All seven core tests MUST pass before Phase 1's HF Space deploy gate opens (SANDBOX_CERT.md is the artifact). A "pass" for each test means: the attack attempted inside ``run_repl(...)`` either raised an exception (RestrictedPython rejection, ImportError from ``_safe_import``, AttributeError/NameError from the guarded builtins), or was killed by the wall-clock timeout — AND produced no observable host side effect (no file created, no network connection opened, no child process spawned, no forbidden module bound into globals). Structure: one ``def test_*`` per attack, verbatim-named per D-13 so that the ROADMAP literal ``pytest tests/test_repl_sandbox.py -v`` produces matching verdict rows in ``.planning/SANDBOX_CERT.md``. """ from __future__ import annotations import os import sys import time from pathlib import Path import pytest from env.server.repl import ( ALLOWED_IMPORTS, BLOCKED_IMPORTS, ReplResult, run_repl, ) # ─────────────────────────────────────────────────────────────────────────── # Shared assertion # ─────────────────────────────────────────────────────────────────────────── def _assert_contained(result: ReplResult, attack_name: str) -> None: """Shared assertion: the attack did not escape. A containment is a ``ReplResult`` where either an exception fired or the wall-clock timeout killed the child. An escape is both: no exception AND no timeout — the attack code ran to completion unrecorded. """ escaped = (result.exception is None and not result.timed_out) assert not escaped, ( f"SANDBOX ESCAPE — {attack_name}: run_repl returned exception=None, " f"timed_out=False, stdout={result.stdout!r}, " f"stderr={result.stderr!r}, globals={result.globals_dict!r}" ) # ─────────────────────────────────────────────────────────────────────────── # Core 7 attacks (STACK.md §7 + CONTEXT.md D-13 — EXACT names) # ─────────────────────────────────────────────────────────────────────────── def test_attribute_access_escape() -> None: """Attack: ``().__class__.__base__.__subclasses__()`` — dunder traversal. Defense: RestrictedPython ``compile_restricted`` rejects ``__class__`` attribute access at the AST level before any code executes. """ code = ( "subclasses = ().__class__.__base__.__subclasses__()\n" "bad = [c for c in subclasses if c.__name__ == 'BuiltinImporter']\n" ) result = run_repl(code, {}) _assert_contained(result, "attribute_access_escape") # The escape artefact must not have been bound. assert "subclasses" not in result.globals_dict, ( "Escape succeeded — `subclasses` leaked into globals" ) assert "bad" not in result.globals_dict, ( "Escape succeeded — `bad` (BuiltinImporter hunter) leaked into globals" ) def test_builtins_monkeypatch() -> None: """Attack: ``__builtins__['open']`` / ``getattr(__builtins__, 'open')``. Defense: ``_build_safe_builtins()`` returns a minimal dict; ``open`` is in ``_FORBIDDEN_NAMES``. ``__builtins__`` itself is also forbidden, and RestrictedPython blocks ``getattr`` at compile time. """ code = ( "err = None\n" "data = None\n" "try:\n" " f = __builtins__['open']('/etc/passwd', 'r')\n" " data = f.read()\n" "except Exception as e:\n" " err = type(e).__name__\n" ) result = run_repl(code, {}) # Critical: `data` must never be populated — that would be a real read. assert not result.globals_dict.get("data"), ( "ESCAPE — builtins['open'] returned readable file" ) # Either RestrictedPython rejected at compile, OR the runtime hit # NameError/KeyError/TypeError/AttributeError. if result.exception is not None: # Compile-time rejection or other parent-side containment. return # Runtime containment: err should be a contained exception name. err = result.globals_dict.get("err") assert err in {"KeyError", "TypeError", "AttributeError", "NameError"}, ( f"Unexpected containment path: err={err!r}, result={result!r}" ) def test_eval_via_reflection() -> None: """Attack: ``eval("__import__('os').system('id')")`` — reflective exec. Defense: ``eval``, ``exec``, ``compile``, ``__import__`` are all in ``_FORBIDDEN_NAMES``. ``eval`` is not bound in the child's builtins, so the call raises ``NameError`` at runtime (or RestrictedPython may also reject at compile time depending on the version). """ code = "eval(\"__import__('os').system('id')\")" result = run_repl(code, {}) _assert_contained(result, "eval_via_reflection") # Exception MUST exist — this attack cannot possibly time out. assert result.exception is not None, ( f"eval() somehow ran to completion: {result!r}" ) # The exception must be about eval not being callable or about # RestrictedPython/compile rejection. Accept any of the known containment # signatures (NameError is the most common for missing builtins). exc = result.exception or "" assert any( token in exc for token in ( "NameError", "eval", "__import__", "not defined", "RestrictedPython", "SyntaxError", ) ), f"Unexpected exception type: {exc!r}" def test_network_egress() -> None: """Attack: ``import socket; s.connect(('8.8.8.8', 53))`` — outbound net. Defense: ``socket`` is in ``BLOCKED_IMPORTS``. ``_safe_import`` raises ``ImportError`` before ``s = socket.socket(...)`` can execute. """ # Precondition: the blocklist is actually wired. assert "socket" in BLOCKED_IMPORTS assert "socket" not in ALLOWED_IMPORTS code = ( "import socket\n" "s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n" "s.connect(('8.8.8.8', 53))\n" ) result = run_repl(code, {}) _assert_contained(result, "network_egress") assert result.exception is not None, ( f"socket import somehow succeeded: {result!r}" ) # No socket object leaked. assert "s" not in result.globals_dict, ( "ESCAPE — socket object `s` bound in globals" ) assert "socket" not in result.globals_dict, ( "ESCAPE — `socket` module object leaked into globals" ) def test_file_write() -> None: """Attack: ``open(path, 'w').write('pwned')`` — host filesystem write. Defense: ``open`` is absent from the child's builtins (in ``_FORBIDDEN_NAMES``). On POSIX ``RLIMIT_FSIZE=0`` is a second layer. Verified by checking that the target file does not exist after the run. """ if sys.platform == "win32": tmp_root = Path(os.environ.get("TEMP") or os.environ.get("TMP") or ".") else: tmp_root = Path("/tmp") # Unique per-pid-per-test file; cleaned up regardless of test outcome. target = tmp_root / f"fathom_sandbox_escape_{os.getpid()}.txt" if target.exists(): target.unlink() try: code = f"open({str(target)!r}, 'w').write('pwned')\n" result = run_repl(code, {}) _assert_contained(result, "file_write") # Hard containment check: no host side effect. assert not target.exists(), ( f"ESCAPE — file write succeeded at {target}" ) finally: if target.exists(): try: target.unlink() except Exception: pass def test_os_system() -> None: """Attack: ``import os; os.system('echo PWNED')`` — shell exec. Defense: ``os`` is in ``BLOCKED_IMPORTS``. ``_safe_import`` raises ``ImportError`` before ``os.system`` is reachable. """ assert "os" in BLOCKED_IMPORTS code = ( "import os\n" "rc = os.system('echo PWNED')\n" ) result = run_repl(code, {}) _assert_contained(result, "os_system") assert result.exception is not None, ( f"os import somehow succeeded: {result!r}" ) assert "rc" not in result.globals_dict, ( "ESCAPE — os.system return code bound in globals" ) assert "os" not in result.globals_dict, ( "ESCAPE — `os` module object leaked into globals" ) def test_subprocess_of_subprocess() -> None: """Attack: ``import subprocess; subprocess.Popen([...])`` — nested child. Defense: ``subprocess`` is in ``BLOCKED_IMPORTS``. Even if somehow an attacker reached Popen, the outer env server's PID-1 tini + process group would contain it; the import-layer block is the first wall. """ assert "subprocess" in BLOCKED_IMPORTS assert "multiprocessing" in BLOCKED_IMPORTS code = ( "import subprocess\n" "p = subprocess.Popen(['sh', '-c', 'echo nested'], stdout=subprocess.PIPE)\n" ) result = run_repl(code, {}) _assert_contained(result, "subprocess_of_subprocess") assert result.exception is not None, ( f"subprocess import somehow succeeded: {result!r}" ) assert "p" not in result.globals_dict, ( "ESCAPE — Popen handle `p` leaked into globals" ) assert "subprocess" not in result.globals_dict, ( "ESCAPE — `subprocess` module object leaked into globals" ) # ─────────────────────────────────────────────────────────────────────────── # Bonus attacks (STACK §7 — belong in the same file per D-13 spirit) # ─────────────────────────────────────────────────────────────────────────── def test_traceback_walk() -> None: """Attack: walk ``e.__traceback__.tb_frame.f_back.f_builtins`` after raise. Defense: RestrictedPython blocks access to ``__traceback__`` / ``f_back`` / ``f_builtins`` at AST time (dunder attribute access). Even if the walk partially evaluated, ``_FORBIDDEN_NAMES`` strips the captured builtins of ``open``. """ code = ( "leaked = None\n" "opener = None\n" "try:\n" " 1 / 0\n" "except Exception as e:\n" " tb = e.__traceback__\n" " frame = tb.tb_frame.f_back\n" " leaked = frame.f_builtins\n" " opener = leaked.get('open') if leaked else None\n" ) result = run_repl(code, {}) # Either the compile rejected at AST time, or the dunder-attr access # raised at runtime. Either way, `opener` must not be callable and no # open() leaked. assert not result.globals_dict.get("leaked"), ( "ESCAPE — frame f_builtins leaked into globals" ) assert not result.globals_dict.get("opener"), ( "ESCAPE — `open` builtin leaked via traceback walk" ) def test_infinite_loop() -> None: """Attack: ``while True: pass`` — DoS via non-terminating code. Defense: wall-clock timeout in ``run_repl`` kills the child. D-03 says globals are preserved unchanged on timeout. """ started = time.monotonic() important = {"important_state": "must survive timeout"} result = run_repl("while True:\n pass\n", dict(important), timeout_s=2.0) elapsed = time.monotonic() - started assert result.timed_out, f"Expected timed_out=True, got {result!r}" assert elapsed < 8.0, ( f"Timeout enforcement too slow: {elapsed:.2f}s (expected < 8s, " f"incl. startup + kill-wait)" ) # D-03 — prior state preserved on timeout. assert result.globals_dict == important, ( "D-03 violation — globals mutated on timeout" ) @pytest.mark.skipif( sys.platform == "win32", reason="RLIMIT_AS unsupported on Windows; memory-bomb containment via " "RestrictedPython allocation is not representative on this platform", ) def test_memory_bomb() -> None: """Attack: ``x = [0] * 10**9`` — memory exhaustion. Defense (POSIX only): ``RLIMIT_AS = 512 MiB`` kills the child process. Test is skipped on Windows where ``resource.setrlimit`` is unavailable. """ code = "x = [0] * (10 ** 9)\n" result = run_repl(code, {}, timeout_s=10.0) # Either MemoryError from the Python allocator or the OS kills the child # via RLIMIT_AS (manifests as a ChildProtocolError / non-JSON stdout). assert result.exception is not None or result.timed_out, ( f"Memory bomb not contained: {result!r}" ) assert "x" not in result.globals_dict, ( "ESCAPE — gigantic list `x` bound in globals" )