From b2ae60434290daeeebf0a40b73b275949115e3f4 Mon Sep 17 00:00:00 2001 From: Valentin Vladescu Date: Fri, 3 Jul 2026 18:15:55 +0300 Subject: [PATCH] =?UTF-8?q?fix(security):=20sandbox=20RLM=20REPL=20exec=20?= =?UTF-8?q?=E2=80=94=20CRITICAL=20prompt-injection=20RCE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit vendor/rlm_repl/sandbox.py exec()'d LLM-authored Python with unrestricted __builtins__ on untrusted session-transcript content — an attacker planting a repl block in a transcript got RCE on the developer's machine merely by running `afterburn discover`. Defense in depth: - safe-builtins allowlist (~50 names; no __import__/open/eval/exec/compile/getattr) - AST check rejecting imports, dunder-attribute escapes (().__class__.__bases__...), forbidden names - opt-in gate AFTERBURN_ENABLE_RLM_EXEC; default discover SKIPS model-authored exec with a warning - 24 tests: blocks import os/__import__/open/subclasses-escape/eval/exec/globals; benign analysis still works Found by the cal adversarial N-reviewer harness (3/3 majority-verify). --- afterburn/passes.py | 34 +++++- afterburn/vendor/rlm_repl/sandbox.py | 114 +++++++++++++++++- tests/test_sandbox.py | 171 +++++++++++++++++++++++++++ 3 files changed, 315 insertions(+), 4 deletions(-) create mode 100644 tests/test_sandbox.py diff --git a/afterburn/passes.py b/afterburn/passes.py index 635e6a6..e7665fa 100644 --- a/afterburn/passes.py +++ b/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 @@ -601,8 +602,39 @@ def run_friction_pass( 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: diff --git a/afterburn/vendor/rlm_repl/sandbox.py b/afterburn/vendor/rlm_repl/sandbox.py index cf44311..0d22c20 100644 --- a/afterburn/vendor/rlm_repl/sandbox.py +++ b/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) diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py new file mode 100644 index 0000000..138cdae --- /dev/null +++ b/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 == []