From 44f9e48c2a5ac1cb890d1eb66e110fd0a14fe93f Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 31 Jul 2026 13:47:02 -0700 Subject: [PATCH] [eric] security: close the Output executor sandbox escape via sys, dunders and module chains --- backend/apps/outputs/code_safety.py | 226 ++++++++++++++++++ backend/apps/outputs/executor.py | 171 ++++--------- backend/apps/outputs/outputs.py | 7 +- backend/apps/outputs/publish_scan.py | 2 +- backend/apps/swarm/scan_app_files.py | 2 +- .../tests/test_outputs_executor_sandbox.py | 178 ++++++++++++++ 6 files changed, 461 insertions(+), 125 deletions(-) create mode 100644 backend/apps/outputs/code_safety.py create mode 100644 backend/tests/test_outputs_executor_sandbox.py diff --git a/backend/apps/outputs/code_safety.py b/backend/apps/outputs/code_safety.py new file mode 100644 index 00000000..8c7ddda0 --- /dev/null +++ b/backend/apps/outputs/code_safety.py @@ -0,0 +1,226 @@ +"""Static safety gate for user-supplied Output backend code. + +`executor.py` owns the subprocess; this file owns the verdict on whether the +code may run without asking the user first. The gate is allowlist-shaped: code +may import a data-shaping module and touch its ordinary attributes, and that is +all. Every other way of getting hold of a module the allowlist withholds (a bare +handle the sandbox preamble binds, an attribute chain that lands on a module, a +dunder walk, an attribute name spelled as a string) is a warning, so it reaches +the user as a consent prompt instead of auto-running. +""" + +import ast +import importlib +import types +from typing import Dict, List, Optional, Set + +# Modules backend code is allowed to import. This is not an OS-level jail, so keep the list to "data shaping" libraries; no I/O, no networking, no subprocess. It pairs with cwd=tempdir + minimal env so the blast radius stays small. +ALLOWED_MODULES = frozenset({ + "json", "math", "re", "datetime", "collections", "itertools", + "functools", "statistics", "decimal", "fractions", "random", + "string", "textwrap", "unicodedata", "csv", "copy", "enum", + "dataclasses", "typing", "abc", "numbers", "uuid", "hashlib", + "base64", "binascii", "operator", "heapq", "bisect", "array", +}) + +# Builtins that punch holes through the allowlist or do I/O. Most are also deleted off `builtins` inside the subprocess; exec/compile/__import__ can't be, because the import machinery runs on them. +P_BLOCKED_BUILTINS = frozenset({ + "exec", "eval", "compile", "__import__", "open", "input", + "breakpoint", "exit", "quit", +}) + +# These hand back a live namespace dict, which is every blocked name again through a different door. Warned about but never deleted: library code calls them constantly, and a scrubbed `builtins` would break `import csv` itself. +P_NAMESPACE_BUILTINS = frozenset({"vars", "globals", "locals"}) + +# Spell an attribute as a string and the AST can't read it, so these are allowed only with a plain literal that would have passed written out longhand. +P_DYNAMIC_ATTR_BUILTINS = frozenset({"getattr", "setattr", "delattr"}) + +# Modules the executor preamble binds into the user's namespace with no import. `json` stays usable (it is allowlisted anyway); these three were the free handles that made the whole allowlist decorative. +P_SANDBOX_MODULE_HANDLES = frozenset({"sys", "io", "builtins"}) + +# Dunders that hold a reference to nothing at all, and `if __name__ == "__main__"` is far too common to punish. +P_INERT_DUNDERS = frozenset({"__name__", "__file__", "__doc__"}) + + +class UnsafeCodeError(Exception): + """Raised when the static gate rejects user-supplied backend code.""" + + +def p_is_dunder(name: str) -> bool: + return len(name) > 4 and name.startswith("__") and name.endswith("__") + + +def p_dotted_chain(node: ast.expr) -> Optional[List[str]]: + """['json', 'codecs', 'open'] for `json.codecs.open`; None when the chain + doesn't start at a plain name.""" + parts: List[str] = [] + current: ast.expr = node + while isinstance(current, ast.Attribute): + parts.append(current.attr) + current = current.value + if not isinstance(current, ast.Name): + return None + parts.append(current.id) + parts.reverse() + return parts + + +def p_resolved_module(chain: List[str], aliases: Dict[str, str]) -> Optional[str]: + """The name of the module an attribute chain resolves to, or None if it + resolves to something that isn't a module. + + `json.codecs` is a module and `datetime.time` is a class, and only the live + object knows which; matching attribute names against a list of module names + would flag both. So resolve against the module actually imported. Safe to + import here because `aliases` only ever holds allowlisted stdlib roots. + """ + root = aliases.get(chain[0]) + if root is None: + return None + try: + value: object = importlib.import_module(root) + for attr in chain[1:]: + value = getattr(value, attr) + except Exception: + return None + return value.__name__ if isinstance(value, types.ModuleType) else None + + +def p_module_aliases(tree: ast.Module) -> Dict[str, str]: + """Local name -> allowlisted module it holds. Plain assignment counts, so + `m = json` doesn't launder `m.codecs` past the chain check. Modules outside + the allowlist are never recorded, which is what keeps the resolver above + from importing anything a hostile file names.""" + aliases: Dict[str, str] = {"json": "json"} + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + root = alias.name.split(".")[0] + if root in ALLOWED_MODULES: + aliases[alias.asname or root] = root + assigns = [n for n in ast.walk(tree) if isinstance(n, ast.Assign)] + for _ in range(len(assigns)): + before = len(aliases) + for node in assigns: + if len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name): + continue + chain = p_dotted_chain(node.value) + resolved = p_resolved_module(chain, aliases) if chain else None + if resolved and resolved.split(".")[0] in ALLOWED_MODULES: + aliases[node.targets[0].id] = resolved + if len(aliases) == before: + break + return aliases + + +def p_dunder_warning(attr: str, prefix: str) -> Optional[str]: + if p_is_dunder(attr) and attr not in P_INERT_DUNDERS: + return f"Uses dunder '{prefix}{attr}', which walks the object graph past the allowlist" + return None + + +def p_attribute_warning(chain: List[str], aliases: Dict[str, str]) -> Optional[str]: + """The verdict on one resolved attribute chain, dunders first.""" + for attr in chain[1:]: + dunder = p_dunder_warning(attr, ".") + if dunder: + return dunder + for depth in range(2, len(chain) + 1): + reached = p_resolved_module(chain[:depth], aliases) + if reached and reached.split(".")[0] not in ALLOWED_MODULES: + return f"Reaches module '{reached}' via '{'.'.join(chain[:depth])}' (outside the safe-data-shaping allowlist)" + return None + + +def p_call_warning(node: ast.Call, aliases: Dict[str, str]) -> Optional[str]: + if not isinstance(node.func, ast.Name): + return None + name = node.func.id + if name in P_BLOCKED_BUILTINS: + return f"Calls builtin '{name}()' which can escape the sandbox" + if name in P_NAMESPACE_BUILTINS: + return f"Calls '{name}()', which hands back the sandbox's own namespace" + if name not in P_DYNAMIC_ATTR_BUILTINS: + return None + attr = node.args[1] if len(node.args) > 1 else None + if not isinstance(attr, ast.Constant) or not isinstance(attr.value, str): + return f"Computes an attribute name for '{name}()', which can spell any escape as a string" + base = p_dotted_chain(node.args[0]) + if base is None: + return p_dunder_warning(attr.value, ".") + return p_attribute_warning(base + [attr.value], aliases) + + +def p_star_import_warning(module: str, aliases: Dict[str, str]) -> Optional[str]: + """`from json import *` binds whatever json's __all__ names, which is a + short list of functions today but is not ours to assume.""" + try: + imported = importlib.import_module(module) + except Exception: + return None + names = getattr(imported, "__all__", None) or [n for n in dir(imported) if not n.startswith("_")] + for name in names: + warning = p_attribute_warning([module, str(name)], aliases) + if warning: + return warning + return None + + +def get_code_warnings(code: str) -> List[str]: + """Return human-readable warnings for every static risk, without raising. + + `/api/outputs/execute` surfaces these in the run dialog, so an Output that + genuinely needs `pandas` gets a "review and click Run Anyway" affordance + instead of a silent 500. An empty list is what buys the no-prompt auto-run + path, so anything that could reach past the allowlist has to land in it. A + syntax error is reported as a warning rather than raised, so the dialog can + show it next to the code. + """ + try: + tree = ast.parse(code) + except SyntaxError as e: + return [f"Syntax error: {e}"] + + aliases = p_module_aliases(tree) + warnings: List[str] = [] + seen: Set[str] = set() + + def note(msg: Optional[str]) -> None: + if msg and msg not in seen: + seen.add(msg) + warnings.append(msg) + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name.split(".")[0] not in ALLOWED_MODULES: + note(f"Imports '{alias.name}' (outside the safe-data-shaping allowlist)") + elif isinstance(node, ast.ImportFrom): + root = (node.module or "").split(".")[0] + if root not in ALLOWED_MODULES: + note(f"Imports from '{node.module}' (outside the safe-data-shaping allowlist)") + continue + for alias in node.names: + if alias.name == "*": + note(p_star_import_warning(root, aliases)) + else: + note(p_attribute_warning([root, alias.name], aliases)) + elif isinstance(node, ast.Call): + note(p_call_warning(node, aliases)) + elif isinstance(node, ast.Attribute): + chain = p_dotted_chain(node) + note(p_attribute_warning(chain, aliases) if chain else p_dunder_warning(node.attr, ".")) + elif isinstance(node, ast.Name): + if node.id in P_SANDBOX_MODULE_HANDLES: + note(f"References '{node.id}', a live module the sandbox binds but the allowlist withholds") + else: + note(p_dunder_warning(node.id, "")) + return warnings + + +def validate_code_safety(code: str) -> None: + """Raise UnsafeCodeError on the first static risk. The strict wrapper around + get_code_warnings, for callers with no user to ask.""" + warnings = get_code_warnings(code) + if warnings: + raise UnsafeCodeError(warnings[0]) diff --git a/backend/apps/outputs/executor.py b/backend/apps/outputs/executor.py index 51167376..c79da34b 100644 --- a/backend/apps/outputs/executor.py +++ b/backend/apps/outputs/executor.py @@ -1,4 +1,3 @@ -import ast import asyncio import json import logging @@ -7,88 +6,13 @@ import sys import tempfile from dataclasses import dataclass +from backend.apps.outputs.code_safety import ALLOWED_MODULES, validate_code_safety + logger = logging.getLogger(__name__) TIMEOUT_SECONDS = 30 -# Modules backend code is allowed to import. Trade-off: a determined attacker can find ways around this (e.g. string-encoded imports via tricks the AST validator can't see), but the allowlist kills the easy paths cheaply and pairs with cwd=tempdir + minimal env so the blast radius is small even if a payload slips past. Keep this list to "data shaping" libraries; no I/O, no networking, no subprocess. -P_ALLOWED_MODULES = frozenset({ - "json", "math", "re", "datetime", "collections", "itertools", - "functools", "statistics", "decimal", "fractions", "random", - "string", "textwrap", "unicodedata", "csv", "copy", "enum", - "dataclasses", "typing", "abc", "numbers", "uuid", "hashlib", - "base64", "binascii", "operator", "heapq", "bisect", "array", -}) - -# Builtin functions that punch holes through the allowlist or do I/O. Direct calls (e.g. `eval(...)`) are caught here. Attribute-style calls (`__builtins__.eval(...)`) are blocked by the preamble's `delattr` loop in the subprocess. -P_BLOCKED_BUILTINS = frozenset({ - "exec", "eval", "compile", "__import__", "open", "input", - "breakpoint", "exit", "quit", -}) - - -class UnsafeCodeError(Exception): - """Raised when AST validation rejects user-supplied backend code.""" - - -def get_code_warnings(code: str) -> list[str]: - """Return human-readable warnings for AST-visible risks, without raising. - - Used by `/api/outputs/execute` to surface risks to the user in the run - dialog before executing; so a legit Output that needs `pandas` doesn't - silently 500 with "import not allowed," it gets a "this Output uses - unsafe imports; review and click Run Anyway" affordance. - - Returns [] for code that's fully inside the allowlist. A syntax error - is reported as a single warning rather than raised so the dialog can - show it next to the code. - """ - try: - tree = ast.parse(code) - except SyntaxError as e: - return [f"Syntax error: {e}"] - - warnings: list[str] = [] - seen: set[str] = set() - for node in ast.walk(tree): - if isinstance(node, ast.Import): - for alias in node.names: - root = alias.name.split(".")[0] - if root not in P_ALLOWED_MODULES: - msg = f"Imports '{alias.name}' (outside the safe-data-shaping allowlist)" - if msg not in seen: - seen.add(msg) - warnings.append(msg) - elif isinstance(node, ast.ImportFrom): - if node.module: - root = node.module.split(".")[0] - if root not in P_ALLOWED_MODULES: - msg = f"Imports from '{node.module}' (outside the safe-data-shaping allowlist)" - if msg not in seen: - seen.add(msg) - warnings.append(msg) - elif isinstance(node, ast.Call): - if isinstance(node.func, ast.Name) and node.func.id in P_BLOCKED_BUILTINS: - msg = f"Calls builtin '{node.func.id}()' which can escape the sandbox" - if msg not in seen: - seen.add(msg) - warnings.append(msg) - return warnings - - -def p_validate_code_safety(code: str) -> None: - """Raise UnsafeCodeError on the first AST-visible risk. Thin wrapper - around get_code_warnings for callers that want the strict-reject - behavior (the default `execute_backend_code` path). Callers that want - to show warnings to a user and let them override should call - get_code_warnings directly and pass `skip_validation=True` to - execute_backend_code.""" - warnings = get_code_warnings(code) - if warnings: - raise UnsafeCodeError(warnings[0]) - - -# Env vars we always scrub from the subprocess, regardless of strict-vs-force. These are the keys an attacker would actually want; install token, provider API keys, cloud credentials. Everything else is local-machine convenience. +# Env vars we always scrub from the subprocess, approved or not. These are the keys an attacker would actually want; install token, provider API keys, cloud credentials. Everything else is local-machine convenience. P_SCRUBBED_ENV_KEYS = frozenset({ "OPENSWARM_AUTH_TOKEN", "ANTHROPIC_API_KEY", @@ -106,24 +30,23 @@ P_SCRUBBED_ENV_KEYS = frozenset({ }) -def p_minimal_env(force: bool = False) -> dict: +def exec_env(approved: bool = False) -> dict: """Build the env for the executor subprocess. - Strict mode (force=False): only language essentials. AST-validated code - is data-shaping only; `import os` and `open()` are blocked, so the - subprocess can't read env vars or expand `~` anyway. Minimal env is - correct here. + Sandboxed (approved=False): only language essentials. Gate-passing code is + data-shaping only; `import os` and `open()` are blocked, so the subprocess + can't read env vars or expand `~` anyway. No PATH, no HOME. - Force mode (force=True): user has explicitly approved unsafe imports - via the HITL preview. They expect the code to behave like a normal - Python process; read HOME, find files, etc. Inherit the real env - minus credentials, so an `open(os.path.expanduser("~/data.csv"))` - actually works instead of silently misbehaving. + Approved (approved=True): the user has explicitly okayed unsafe imports via + the HITL preview. They expect the code to behave like a normal Python + process; read HOME, find files, etc. Inherit the real env minus credentials, + so an `open(os.path.expanduser("~/data.csv"))` actually works instead of + silently misbehaving. - Both modes scrub P_SCRUBBED_ENV_KEYS so even force-mode code never - sees the install token or provider API keys. + Both modes scrub P_SCRUBBED_ENV_KEYS so even approved code never sees the + install token or provider API keys. """ - if force: + if approved: env = {k: v for k, v in os.environ.items() if k not in P_SCRUBBED_ENV_KEYS} env["PYTHONDONTWRITEBYTECODE"] = "1" # Force UTF-8 even if the parent somehow lacked it (dev mode where Electron didn't inject PYTHONUTF8). Without this, a child reading non-ASCII stdin/files on a cp1252 Windows machine raises UnicodeDecodeError, the "works on my laptop, not theirs" failure. @@ -146,6 +69,30 @@ def p_minimal_env(force: bool = False) -> dict: return env +# The subprocess bootstrap: capture stdout, read the input, and hand the code a `result` to fill. Bound objects rather than modules carry the answer back out, so the hardening below can take the modules away. +P_PREAMBLE = ( + "import json, sys, io, builtins\n" + "p_stdout = sys.stdout\n" + "p_capture = io.StringIO()\n" + "sys.stdout = p_capture\n" + "input_data = json.loads(sys.stdin.read())\n" + "result = {}\n" +) +# Warm the allowlist BEFORE scrubbing builtins, because half the stdlib borrows the builtins the scrub deletes while it loads (tokenize does `from builtins import open`, which is how `import dataclasses` dies). Warm imports are cache hits, so gate-passing code never touches the loader again. Then the module handles go: leaving `sys` bound handed gate-passing code a live `sys.modules['os']` with no import statement in sight. exec/eval/compile/__import__ stay put whatever we'd like: the import statement, the loader and namedtuple all run on them, and calling them by name is a static-gate warning anyway. +P_SANDBOX_HARDENING = ( + f"for p_name in {tuple(sorted(ALLOWED_MODULES))!r}:\n" + " try: __import__(p_name)\n" + " except ImportError: pass\n" + "for p_name in ('open','input','breakpoint','exit','quit'):\n" + " try: delattr(builtins, p_name)\n" + " except AttributeError: pass\n" + "del sys, io, builtins, p_name\n" +) +P_POSTAMBLE = ( + "\np_stdout.write(json.dumps({\"__stdout__\": p_capture.getvalue(), \"__result__\": result}))\n" +) + + @dataclass class BackendExecResult: result: dict @@ -154,7 +101,7 @@ class BackendExecResult: async def execute_backend_code( - code: str, input_data: dict, *, skip_validation: bool = False + code: str, input_data: dict, *, approved: bool = False ) -> BackendExecResult: """Execute user-provided Python code in a subprocess. @@ -163,39 +110,23 @@ async def execute_backend_code( separately from the result via an in-process StringIO redirect. Security boundaries (defense in depth; none alone is sufficient): - 1. AST allowlist on imports + blocked-builtin call list. + 1. The static gate in code_safety.py, on every run that isn't approved. 2. Subprocess cwd = fresh temp dir (not the OpenSwarm process cwd). 3. Subprocess env strips PATH, all *TOKEN / *_API_KEY inheritance. - 4. Preamble scrubs dangerous attrs off `builtins` inside the subprocess - to catch AST-bypass tricks (e.g. metaclass shenanigans). + 4. Preamble scrubs the I/O builtins and drops the module handles it + needed, so gate-passing code starts with no reachable module. 5. 30s wall-clock timeout, killed on overrun. - `skip_validation=True` bypasses #1; intended ONLY for callers that - have already surfaced the warnings to a user and gotten explicit - consent (the `/api/outputs/execute` HITL flow). #2, #5 always run. + `approved=True` means a user saw the warnings and clicked Run Anyway, and + it relaxes 1, 3 and 4 together. It is the ONLY thing that relaxes them: a + caller that has already run the gate itself still gets the sandbox, because + "we checked" must never be the reason the walls come down. """ - if not skip_validation: - p_validate_code_safety(code) + if not approved: + validate_code_safety(code) - preamble = ( - "import json, sys, io, builtins\n" - # Defense-in-depth: scrub dangerous attrs off `builtins` so attribute-style accesses (metaclass.__subclasses__ chains) can't reach them. NOTE: __import__ is deliberately NOT scrubbed, Python's `import` statement bytecode reads `__import__` from builtins, so removing it makes EVERY import (including allowlisted ones like `import math`) fail with "ImportError: __import__ not found". The AST allowlist on the host is what blocks `import subprocess`; the per-subprocess scrub just plugs the named-builtin attack vectors that the AST can't see (eval/exec via attribute access on objects, etc.). - "for _b in ('exec','eval','compile','open','input',\n" - " 'breakpoint','exit','quit'):\n" - " try: delattr(builtins, _b)\n" - " except AttributeError: pass\n" - "_orig_stdout = sys.stdout\n" - "_capture = io.StringIO()\n" - "sys.stdout = _capture\n" - "input_data = json.loads(sys.stdin.read())\n" - "result = {}\n" - ) - postamble = ( - "\nsys.stdout = _orig_stdout\n" - 'json.dump({"__stdout__": _capture.getvalue(), "__result__": result}, sys.stdout)\n' - ) - wrapper = preamble + code + postamble + wrapper = P_PREAMBLE + ("" if approved else P_SANDBOX_HARDENING) + code + P_POSTAMBLE with tempfile.TemporaryDirectory(prefix="openswarm-exec-") as workdir: proc = await asyncio.create_subprocess_exec( @@ -204,7 +135,7 @@ async def execute_backend_code( stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workdir, - env=p_minimal_env(force=skip_validation), + env=exec_env(approved=approved), ) try: diff --git a/backend/apps/outputs/outputs.py b/backend/apps/outputs/outputs.py index 2e4dddef..cd7861fc 100644 --- a/backend/apps/outputs/outputs.py +++ b/backend/apps/outputs/outputs.py @@ -15,7 +15,8 @@ from backend.apps.outputs.models import ( PublishPreflightRequest, PublishRequest, PublishPreflightResponse, PublishResult, PublishReview, ) -from backend.apps.outputs.executor import execute_backend_code, get_code_warnings +from backend.apps.outputs.code_safety import get_code_warnings +from backend.apps.outputs.executor import execute_backend_code from backend.apps.outputs.publish_common import slugify, PublishError from backend.apps.outputs.publish_scan import scan_for_publish, quick_ast_gate from backend.apps.outputs.publish_build import build_static, collect_bundle @@ -723,9 +724,9 @@ async def execute_output(body: OutputExecute): code_preview = output.backend_code if not warnings_out: try: - # We've either already vetted (no warnings above) or the user explicitly opted in with force=True. Pass skip_validation=True so we don't pay for a redundant AST walk inside execute_backend_code. + # `approved` is body.force alone: code that merely passed the gate above still runs sandboxed, because a clean scan is not consent. exec_result = await execute_backend_code( - output.backend_code, body.input_data, skip_validation=True + output.backend_code, body.input_data, approved=bool(body.force) ) backend_result = exec_result.result stdout_text = exec_result.stdout diff --git a/backend/apps/outputs/publish_scan.py b/backend/apps/outputs/publish_scan.py index a9ca3829..d74794fa 100644 --- a/backend/apps/outputs/publish_scan.py +++ b/backend/apps/outputs/publish_scan.py @@ -15,7 +15,7 @@ import os from collections import OrderedDict from typing import Literal -from backend.apps.outputs.executor import get_code_warnings +from backend.apps.outputs.code_safety import get_code_warnings from backend.apps.outputs.models import Output, PublishReview from backend.apps.outputs.publish_common import is_webapp, workspace_dir from backend.apps.outputs.workspace_io import WALK_SKIP_DIRS diff --git a/backend/apps/swarm/scan_app_files.py b/backend/apps/swarm/scan_app_files.py index 98313fea..2fe872de 100644 --- a/backend/apps/swarm/scan_app_files.py +++ b/backend/apps/swarm/scan_app_files.py @@ -6,7 +6,7 @@ user choosing to open/run the app and the flat-app /execute HITL. A full semanti LLM scan is the separate App Publishing feature, not this.""" from __future__ import annotations -from backend.apps.outputs.executor import get_code_warnings +from backend.apps.outputs.code_safety import get_code_warnings from backend.apps.swarm.models import ReviewSummary diff --git a/backend/tests/test_outputs_executor_sandbox.py b/backend/tests/test_outputs_executor_sandbox.py new file mode 100644 index 00000000..118bc04a --- /dev/null +++ b/backend/tests/test_outputs_executor_sandbox.py @@ -0,0 +1,178 @@ +"""Sandbox escape in the Output backend-code executor (issue #134). + +The gate only ever looked at import statements and calls to a bare builtin +name, while the subprocess preamble handed user code live `sys`, `io` and +`builtins`. `sys.modules['os']` therefore scanned clean, and clean means the +`/api/outputs/execute` auto-run path with no consent prompt. Live before the +fix: arbitrary file read, arbitrary file write, and `os.system`, all with +`AST warnings: []`. + +The payloads below are the class, not the one string: bare handles, attribute +chains that land on a module, dunder traversal, `getattr` indirection, and +aliasing. The legit block underneath is the other half of the bar; a gate that +warns about `datetime.time` would just train users to click through. + +Run: + backend/.venv/bin/python -m pytest backend/tests/test_outputs_executor_sandbox.py -v +""" + +import asyncio +import os +from typing import Any + +import pytest + +from backend.apps.outputs import executor +from backend.apps.outputs.code_safety import UnsafeCodeError, get_code_warnings +from backend.apps.outputs.executor import execute_backend_code, exec_env + + +def p_run(coro: Any) -> Any: + return asyncio.new_event_loop().run_until_complete(coro) + + +P_ESCAPES = [ + # The reported exploit, verbatim. + "result = {'escaped': sys.modules['os'].getcwd(), 'uid': sys.modules['os'].getuid()}", + # The other two handles the preamble left lying around. + "result = {'x': str(io.open)}", + "result = {'b': str(builtins.__dict__)}", + # Attribute chains that land on a module the allowlist withholds. + "result = {'c': str(json.codecs)}", + "import random\nresult = {'cwd': random._os.getcwd()}", + "import collections\nresult = {'m': str(collections._sys.modules)}", + "import hashlib\nresult = {'h': str(hashlib._hashlib)}", + "import base64\nresult = {'s': str(base64.struct)}", + "from json import codecs\nresult = {'c': str(codecs)}", + # Aliasing the module first. + "m = json\nresult = {'c': str(m.codecs)}", + # Dunder traversal. + "result = {'n': len(().__class__.__bases__[0].__subclasses__())}", + "f = lambda: 0\nresult = {'g': str(f.__globals__)}", + "result = {'i': str(__import__('os'))}", + # exec/eval/compile survive in the subprocess (the import machinery needs them), so the gate is the only thing standing here. + "exec('import os')\nresult = {}", + "result = {'e': eval('__import__(\"os\").getcwd()')}", + "result = {'f': open('/etc/hosts').read()}", + # getattr indirection, literal and computed. + "result = {'c': str(getattr(json, 'codecs'))}", + "n = 'codecs'\nresult = {'c': str(getattr(json, n))}", + "result = {'c': str(getattr((), '__class__'))}", + # Namespace grabs. + "result = {'g': str(globals())}", + "result = {'v': str(vars())}", + "result = {'l': str(locals())}", +] + +P_LEGIT = [ + 'import math, json\nresult = {"a": math.floor(1.9), "b": json.dumps([1, 2])}', + # datetime.time is a CLASS, not the `time` module: a name-matching gate would flag it. + 'import datetime\nresult = {"t": datetime.time(12, 0).isoformat()}', + 'import collections\nresult = {"c": dict(collections.Counter("aab"))}', + 'import collections.abc\nresult = {"ok": isinstance({}, collections.abc.Mapping)}', + 'import re\nresult = {"m": re.findall(r"\\d+", "a1b22")}', + 'import csv\nresult = {"rows": [r for r in csv.reader("a,b\\n1,2".splitlines())]}', + 'import statistics\nresult = {"mean": statistics.mean([1, 2, 3])}', + 'import textwrap\nresult = {"w": textwrap.wrap("word " * 20, 10)}', + 'import decimal\nresult = {"d": str(decimal.Decimal("1.10") + decimal.Decimal("2.20"))}', + 'import uuid\nresult = {"u": str(uuid.uuid5(uuid.NAMESPACE_DNS, "x"))}', + # dataclasses and namedtuple exec their generated methods, so they prove the builtins scrub left the plumbing alone. + 'import dataclasses\n@dataclasses.dataclass\nclass Row:\n n: int\nresult = dataclasses.asdict(Row(1))', + 'import collections\nPoint = collections.namedtuple("Point", "x y")\nresult = {"s": sum(Point(1, 2))}', + 'import functools, operator\nresult = {"s": functools.reduce(operator.add, [1, 2, 3])}', + 'from json import dumps\nresult = {"j": dumps({"a": 1})}', + 'from math import *\nresult = {"p": floor(pi)}', + 'if __name__ == "__main__":\n pass\nresult = {"ok": True}', + 'result = {"n": len(input_data), "keys": sorted(input_data)}', + 'result = {"v": getattr(input_data, "get")("missing", 7)}', +] + + +# --- the gate must see every escape ----------------------------------------- + +@pytest.mark.parametrize("code", P_ESCAPES) +def test_escape_is_warned(code: str) -> None: + assert get_code_warnings(code), f"no warning for: {code!r}" + + +@pytest.mark.parametrize("code", P_ESCAPES) +def test_escape_is_refused_before_it_runs(code: str) -> None: + with pytest.raises(UnsafeCodeError): + p_run(execute_backend_code(code, {})) + + +# --- and must stay quiet about ordinary data shaping ------------------------- + +@pytest.mark.parametrize("code", P_LEGIT) +def test_legit_code_is_clean(code: str) -> None: + assert get_code_warnings(code) == [] + + +@pytest.mark.parametrize("code", P_LEGIT) +def test_legit_code_still_runs(code: str) -> None: + out = p_run(execute_backend_code(code, {"a": 1})) + assert isinstance(out.result, dict) and out.result + + +def test_syntax_error_is_reported_not_raised() -> None: + assert get_code_warnings("result = {")[0].startswith("Syntax error") + + +def test_print_output_is_still_captured() -> None: + out = p_run(execute_backend_code('print("hi")\nresult = {"ok": 1}', {})) + assert out.stdout.strip() == "hi" + assert out.result == {"ok": 1} + + +# --- second wall: the subprocess itself, with the gate bypassed --------------- + +@pytest.fixture +def gate_bypassed(monkeypatch: pytest.MonkeyPatch) -> None: + """Pretend a future payload beats the static gate, and check the subprocess + still has nothing to grab. Defense in depth is only real if it holds alone.""" + monkeypatch.setattr(executor, "validate_code_safety", lambda code: None) + + +@pytest.mark.parametrize("handle", ["sys", "io", "builtins"]) +def test_module_handles_are_gone_from_the_subprocess(gate_bypassed: None, handle: str) -> None: + with pytest.raises(RuntimeError) as e: + p_run(execute_backend_code(f"result = {{'x': str({handle})}}", {})) + assert "NameError" in str(e.value) + + +def test_open_builtin_is_gone_from_the_subprocess(gate_bypassed: None) -> None: + with pytest.raises(RuntimeError) as e: + p_run(execute_backend_code("result = {'x': open('/etc/hosts').read()}", {})) + assert "NameError" in str(e.value) + + +def test_no_credentials_or_shell_reachable_when_not_approved(gate_bypassed: None, tmp_path: Any) -> None: + """The end-to-end version of the report: read HOME, then shell out.""" + marker = tmp_path / "pwned.txt" + code = ( + "os_mod = sys.modules['os']\n" + f"result = {{'home': os_mod.environ.get('HOME'), 'rc': os_mod.system('echo x > {marker}')}}" + ) + with pytest.raises(RuntimeError): + p_run(execute_backend_code(code, {})) + assert not marker.exists() + + +def test_sandboxed_env_carries_no_path_or_home() -> None: + env = exec_env(approved=False) + assert "PATH" not in env and "HOME" not in env + + +def test_approved_env_inherits_but_scrubs_credentials(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-should-not-leak") + monkeypatch.setenv("OPENSWARM_AUTH_TOKEN", "should-not-leak") + env = exec_env(approved=True) + assert env.get("PATH") == os.environ.get("PATH") + assert "ANTHROPIC_API_KEY" not in env and "OPENSWARM_AUTH_TOKEN" not in env + + +def test_approved_run_still_gets_its_escape_hatch() -> None: + """The HITL "Run Anyway" path must keep working, or the fix just breaks the + feature instead of securing it.""" + out = p_run(execute_backend_code("import os\nresult = {'sep': os.sep}", {}, approved=True)) + assert out.result == {"sep": os.sep}