tests/test_repl_sandbox.py

"""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"
    )