fix(security): sandbox RLM REPL exec — CRITICAL prompt-injection RCE

afterburn/passes.py

@@ -1,6 +1,7 @@
 """Analysis passes — extract findings from session JSONL data."""
 
 import json
+import os
 import re
 import sys
 from collections import Counter, defaultdict
@@ -264,7 +265,7 @@
             import re
 
             breakdown_match = re.search(
                 r"Breakdown:\s*(.+?)\.(?:\s*Remediations:|$)", f.evidence
             )
             if breakdown_match:
                 pairs = breakdown_match.group(1).split(",")
@@ -601,8 +602,39 @@
     return findings
 
 
+# RLM analysis executes model-authored Python code (see
+# afterburn.vendor.rlm_repl.sandbox). The code is derived from the root LLM
+# analyzing untrusted session-transcript content, so — even though the
+# sandbox restricts builtins and rejects import/dunder-escape patterns — this
+# path is opt-in only. `afterburn discover` MUST NOT silently exec by
+# default; set AFTERBURN_ENABLE_RLM_EXEC=1 to enable it.
+_RLM_EXEC_ENV_VAR = "AFTERBURN_ENABLE_RLM_EXEC"
+_RLM_EXEC_TRUE_VALUES = frozenset({"1", "true", "yes", "on"})
+
+
+def _rlm_exec_enabled() -> bool:
+    """Whether the operator has opted in to executing model-authored REPL code."""
+    return os.environ.get(_RLM_EXEC_ENV_VAR, "").strip().lower() in _RLM_EXEC_TRUE_VALUES
+
+
 def _rlm_friction_analysis(large_sessions: list[SessionInfo]) -> list[Finding]:
-    """Use RLM REPL to analyze sessions too large for direct parsing."""
+    """Use RLM REPL to analyze sessions too large for direct parsing.
+
+    SECURITY: disabled by default. This runs model-authored Python (inside a
+    restricted sandbox) derived from analyzing untrusted transcript content;
+    an attacker can plant a ```repl payload in a transcript. Set
+    AFTERBURN_ENABLE_RLM_EXEC=1 to opt in.
+    """
+    if not _rlm_exec_enabled():
+        print(
+            f"  [warn] Skipping RLM analysis of {len(large_sessions)} large "
+            f"session(s): this executes model-authored code derived from "
+            f"untrusted transcript content. Set {_RLM_EXEC_ENV_VAR}=1 to "
+            f"opt in.",
+            file=sys.stderr,
+        )
+        return []
+
     try:
         from afterburn.vendor.rlm_repl import RLM_REPL
     except ImportError as e:

afterburn/vendor/rlm_repl/sandbox.py

@@ -1,9 +1,109 @@
-"""REPL sandbox — executes Python code with injected tools."""
-
+"""REPL sandbox — executes Python code with injected tools.
+
+SECURITY: the code executed here is model-authored, and the model is
+analyzing untrusted session-transcript content (see
+afterburn.passes._rlm_friction_analysis). A transcript can contain an
+adversarial payload (e.g. a fake ```repl block) that the analyzing LLM
+faithfully reproduces, so this sandbox must not grant unrestricted access
+to the interpreter. Two layers of defense are applied:
+
+1. A minimal `__builtins__` allowlist — no `__import__`, `open`, `eval`,
+   `exec`, `compile`, `input`, or introspection builtins (`globals`,
+   `locals`, `vars`, `dir`, `getattr`, `setattr`, `delattr`) that could be
+   used to reach `os`/`sys`.
+2. A static AST check that rejects `import` statements, dunder attribute
+   access (blocks the classic `().__class__.__bases__[0].__subclasses__()`
+   escape idiom), and direct references to forbidden names.
+
+This is defense-in-depth, not a formally complete sandbox — CPython has no
+fully-safe `exec()` mode. Combined, the two layers block the well-known
+escape idioms while preserving the benign analysis subset (arithmetic,
+string/list/dict operations, comprehensions, etc).
+"""
+
+import ast
+import builtins as _builtins_module
 import io
 import traceback
 from contextlib import redirect_stderr, redirect_stdout
 
+# Builtins explicitly allowed inside the sandbox. Anything not listed here
+# is unreachable via plain name lookup.
+_SAFE_BUILTIN_NAMES = frozenset(
+    {
+        "abs", "all", "any", "bool", "bytearray", "bytes", "callable", "chr",
+        "complex", "dict", "divmod", "enumerate", "filter", "float", "format",
+        "frozenset", "hasattr", "hash", "hex", "int", "isinstance",
+        "issubclass", "iter",
+        "len", "list", "map", "max", "min", "next", "oct", "ord", "pow",
+        "print", "range", "repr", "reversed", "round", "set", "slice",
+        "sorted", "str", "sum", "tuple", "type", "zip",
+        "True", "False", "None", "NotImplemented",
+        "Exception", "ValueError", "TypeError", "KeyError", "IndexError",
+        "AttributeError", "StopIteration", "StopAsyncIteration",
+        "RuntimeError", "ZeroDivisionError", "ArithmeticError",
+        "OverflowError", "NotImplementedError", "LookupError",
+        "AssertionError", "GeneratorExit", "UnicodeError", "UnicodeDecodeError",
+        "UnicodeEncodeError",
+    }
+)
+
+# Names that must never be reachable, even indirectly, because they grant
+# filesystem, process, or interpreter escape hatches. Removing them from
+# __builtins__ (below) already makes plain lookups fail with NameError;
+# the AST check additionally rejects source code that even *names* them,
+# so the failure is an explicit SandboxViolation instead of a confusing
+# NameError deep inside model-authored code.
+_FORBIDDEN_NAMES = frozenset(
+    {
+        "__import__", "eval", "exec", "compile", "open", "input", "exit",
+        "quit", "breakpoint", "globals", "locals", "vars", "dir", "getattr",
+        "setattr", "delattr", "help", "copyright", "credits", "license",
+        "memoryview", "__loader__", "__build_class__", "__debug__",
+    }
+)
+
+
+class SandboxViolation(Exception):
+    """Raised when sandboxed code attempts a forbidden operation."""
+
+
+def _build_safe_builtins() -> dict:
+    """Construct a minimal __builtins__ mapping with dangerous names removed."""
+    return {
+        name: getattr(_builtins_module, name)
+        for name in _SAFE_BUILTIN_NAMES
+        if hasattr(_builtins_module, name)
+    }
+
+
+def _check_ast_safety(code: str) -> None:
+    """Static check: reject imports, dunder attribute access, forbidden names.
+
+    Blocks the classic Python sandbox-escape idiom
+    (`().__class__.__bases__[0].__subclasses__()` and friends) and explicit
+    `import os` / `__import__('os')` statements before the code is ever
+    compiled or executed.
+    """
+    try:
+        tree = ast.parse(code, mode="exec")
+    except SyntaxError as exc:
+        raise SandboxViolation(f"code does not parse: {exc}") from None
+
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.Import, ast.ImportFrom)):
+            raise SandboxViolation(
+                "import statements are not allowed in the sandbox"
+            )
+        if isinstance(node, ast.Attribute) and node.attr.startswith("__"):
+            raise SandboxViolation(
+                f"dunder attribute access is not allowed: .{node.attr}"
+            )
+        if isinstance(node, ast.Name) and node.id in _FORBIDDEN_NAMES:
+            raise SandboxViolation(
+                f"name is not allowed in the sandbox: {node.id}"
+            )
+
 
 class REPLSandbox:
     """Python REPL environment with injected context and tools.
@@ -15,13 +115,15 @@ class REPLSandbox:
     - `FINAL_VAR(name)`: signal completion, return a variable's value
 
     All state persists across exec() calls within the same sandbox.
+    Execution is restricted per the module docstring (safe builtins + AST
+    checks) — see SandboxViolation.
     """
 
     def __init__(self, llm_query_fn=None):
         self._final_answer = None
         self._final_var_name = None
         self._globals: dict = {
-            "__builtins__": __builtins__,
+            "__builtins__": _build_safe_builtins(),
             "llm_query": llm_query_fn or (lambda p: "[no LLM configured]"),
             "FINAL": self._handle_final,
             "FINAL_VAR": self._handle_final_var,
@@ -48,6 +150,12 @@ def execute(self, code: str, timeout_chars: int = 500_000) -> tuple[str, str, bo
         stdout_buf = io.StringIO()
         stderr_buf = io.StringIO()
 
+        try:
+            _check_ast_safety(code)
+        except SandboxViolation as exc:
+            stderr_buf.write(f"SandboxViolation: {exc}\n")
+            return stdout_buf.getvalue(), stderr_buf.getvalue(), False
+
         try:
             with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
                 exec(code, self._globals)

tests/test_sandbox.py

@@ -0,0 +1,171 @@
+"""Tests for the RLM REPL sandbox — blocks RCE payloads, preserves benign use.
+
+Covers the two defense-in-depth layers in afterburn.vendor.rlm_repl.sandbox:
+1. A minimal __builtins__ allowlist (no __import__, open, eval, exec, compile).
+2. A static AST check rejecting import statements and dunder attribute access
+   (the `().__class__.__bases__[0].__subclasses__()` escape idiom).
+
+Also covers the opt-in gate in afterburn.passes._rlm_friction_analysis: the
+exec path must be disabled by default and only run when
+AFTERBURN_ENABLE_RLM_EXEC is set to a truthy value.
+"""
+
+import pytest
+
+from afterburn.passes import _rlm_exec_enabled, _rlm_friction_analysis
+from afterburn.vendor.rlm_repl.sandbox import REPLSandbox
+
+
+class TestSandboxBlocksMaliciousPayloads:
+    """A transcript-planted payload must never actually execute."""
+
+    def test_blocks_import_os(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute(
+            "import os\nprint(os.system('id'))"
+        )
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_blocks_dunder_import(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute(
+            "os_mod = __import__('os')\nprint(os_mod.system('id'))"
+        )
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_blocks_open(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute(
+            "print(open('/etc/passwd').read())"
+        )
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_blocks_class_hierarchy_escape(self) -> None:
+        """The classic no-import RCE idiom: climb from () to os via __class__."""
+        sandbox = REPLSandbox()
+        payload = (
+            "base = ().__class__.__bases__[0]\n"
+            "for sub in base.__subclasses__():\n"
+            "    pass\n"
+        )
+        stdout, stderr, _ = sandbox.execute(payload)
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_blocks_eval(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute("eval('__import__(\"os\").system(\"id\")')")
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_blocks_exec_builtin(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute("exec('import os')")
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_blocks_globals_introspection(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute("print(globals())")
+        assert "SandboxViolation" in stderr
+        assert stdout == ""
+
+    def test_state_not_corrupted_after_blocked_payload(self) -> None:
+        """A blocked payload must not leave __builtins__ or globals tampered."""
+        sandbox = REPLSandbox()
+        sandbox.execute("import os\nprint(os.system('id'))")
+        # Sandbox should still be usable and safe afterward.
+        stdout, stderr, has_final = sandbox.execute("FINAL(1 + 1)")
+        assert stderr == ""
+        assert has_final
+        assert sandbox.get_final_answer() == "2"
+
+
+class TestSandboxAllowsBenignAnalysis:
+    """Legitimate analysis code (arithmetic, list/dict/str ops) still works."""
+
+    def test_arithmetic_and_print(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, _ = sandbox.execute("print(1 + 2 * 3)")
+        assert stderr == ""
+        assert stdout.strip() == "7"
+
+    def test_list_and_string_ops(self) -> None:
+        sandbox = REPLSandbox()
+        code = (
+            "data = ['a', 'bb', 'ccc']\n"
+            "lengths = [len(x) for x in data]\n"
+            "print(sum(lengths), sorted(lengths), max(lengths))\n"
+        )
+        stdout, stderr, _ = sandbox.execute(code)
+        assert stderr == ""
+        assert stdout.strip() == "6 [1, 2, 3] 3"
+
+    def test_context_inspection(self) -> None:
+        """Matches the pattern from engine.SYSTEM_PROMPT's own example."""
+        sandbox = REPLSandbox()
+        sandbox.load_context([{"role": "user", "content": "hi"}] * 5)
+        stdout, stderr, _ = sandbox.execute("print(type(context), len(context))")
+        assert stderr == ""
+        assert "5" in stdout
+
+    def test_final_var(self) -> None:
+        sandbox = REPLSandbox()
+        stdout, stderr, has_final = sandbox.execute(
+            "findings = [{'theme': 'x'}]\nFINAL_VAR('findings')"
+        )
+        assert stderr == ""
+        assert has_final
+        assert sandbox.get_final_answer() == "[{'theme': 'x'}]"
+
+    def test_dict_and_counter_style_aggregation(self) -> None:
+        sandbox = REPLSandbox()
+        code = (
+            "counts = {}\n"
+            "for word in ['a', 'b', 'a', 'c', 'b', 'a']:\n"
+            "    counts[word] = counts.get(word, 0) + 1\n"
+            "print(sorted(counts.items()))\n"
+        )
+        stdout, stderr, _ = sandbox.execute(code)
+        assert stderr == ""
+        assert stdout.strip() == "[('a', 3), ('b', 2), ('c', 1)]"
+
+
+class TestRLMExecOptInGate:
+    """The exec path must default to disabled and require explicit opt-in."""
+
+    def test_disabled_by_default(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("AFTERBURN_ENABLE_RLM_EXEC", raising=False)
+        assert _rlm_exec_enabled() is False
+
+    @pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", "on"])
+    def test_enabled_by_truthy_values(
+        self, monkeypatch: pytest.MonkeyPatch, value: str
+    ) -> None:
+        monkeypatch.setenv("AFTERBURN_ENABLE_RLM_EXEC", value)
+        assert _rlm_exec_enabled() is True
+
+    @pytest.mark.parametrize("value", ["0", "false", "", "no"])
+    def test_not_enabled_by_falsy_values(
+        self, monkeypatch: pytest.MonkeyPatch, value: str
+    ) -> None:
+        monkeypatch.setenv("AFTERBURN_ENABLE_RLM_EXEC", value)
+        assert _rlm_exec_enabled() is False
+
+    def test_friction_analysis_skips_when_disabled(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """discover MUST NOT silently exec by default: no sessions, no exec."""
+        monkeypatch.delenv("AFTERBURN_ENABLE_RLM_EXEC", raising=False)
+
+        class _FakeSession:
+            session_id = "fake"
+            size_bytes = 20 * 1024 * 1024
+            file_path = "/nonexistent/path.jsonl"
+            project_slug = "fake"
+
+        result = _rlm_friction_analysis([_FakeSession()])
+        assert result == []