[rajarshidattapy] fix(outputs): close sandbox escape in backend-code executor (#134)

backend/apps/outputs/executor.py

@@ -21,11 +21,21 @@
 })
 
 # 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.
+# getattr/setattr/delattr/vars/globals/locals are dynamic-attribute escapes: they let `getattr(obj, "__class__")` reconstruct the dunder-walk the static check below forbids, so they're blocked too.
 P_BLOCKED_BUILTINS = frozenset({
     "exec", "eval", "compile", "__import__", "open", "input",
     "breakpoint", "exit", "quit",
+    "getattr", "setattr", "delattr", "vars", "globals", "locals",
 })
 
+# Bare names that hand back a live module the import allowlist meant to withhold. `sys` is the headline: `sys.modules["os"]` retrieves the already-loaded os with no import statement and no blocked-builtin call, so the AST never saw it (issue #134). `__import__`/`__builtins__` are the attribute-free spellings of the same escape.
+P_BLOCKED_NAMES = frozenset({"sys", "__import__", "__builtins__", "__loader__", "__spec__"})
+
+
+def p_is_dunder(name: str) -> bool:
+    """True for `__x__` names. Blocking dunder ATTRIBUTE access (`().__class__`, `f.__globals__`, `.__subclasses__`) severs the object-graph walk that reaches os/subprocess without ever importing them; data-shaping code never needs a dunder."""
+    return len(name) > 4 and name.startswith("__") and name.endswith("__")
+
 
 class UnsafeCodeError(Exception):
     """Raised when AST validation rejects user-supplied backend code."""
@@ -73,6 +83,18 @@ def get_code_warnings(code: str) -> list[str]:
                 if msg not in seen:
                     seen.add(msg)
                     warnings.append(msg)
+        elif isinstance(node, ast.Attribute):
+            if p_is_dunder(node.attr):
+                msg = f"Accesses dunder attribute '.{node.attr}' which can walk the object graph to a blocked module"
+                if msg not in seen:
+                    seen.add(msg)
+                    warnings.append(msg)
+        elif isinstance(node, ast.Name):
+            if node.id in P_BLOCKED_NAMES or p_is_dunder(node.id):
+                msg = f"References '{node.id}' which exposes a live module the import allowlist withholds"
+                if msg not in seen:
+                    seen.add(msg)
+                    warnings.append(msg)
     return warnings
 
 
@@ -154,7 +176,7 @@ class BackendExecResult:
 
 
 async def execute_backend_code(
-    code: str, input_data: dict, *, skip_validation: bool = False
+    code: str, input_data: dict, *, skip_validation: bool = False, force_env: bool = False
 ) -> BackendExecResult:
     """Execute user-provided Python code in a subprocess.
 
@@ -173,6 +195,13 @@ async def execute_backend_code(
     `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.
+
+    `force_env` selects the privileged inherit-real-env mode (#3 relaxed):
+    ONLY pass it when the user has consented to unsafe imports. It is a
+    SEPARATE decision from `skip_validation`: code that merely passed the
+    allowlist (empty warnings) must still run in the minimal env, else
+    every vetted Output silently runs with PATH/COMSPEC and `os.system`
+    becomes reachable with no consent (issue #134).
     """
 
     if not skip_validation:
@@ -204,7 +233,7 @@ async def execute_backend_code(
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
             cwd=workdir,
-            env=p_minimal_env(force=skip_validation),
+            env=p_minimal_env(force=force_env),
         )
 
         try:

backend/apps/outputs/outputs.py

@@ -725,9 +725,10 @@ 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.
+                # 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. force_env is gated on body.force ALONE: vetted-but-not-forced code stays in the minimal env, so `os.system` can't reach a shell without explicit consent (issue #134).
                 exec_result = await execute_backend_code(
-                    output.backend_code, body.input_data, skip_validation=True
+                    output.backend_code, body.input_data,
+                    skip_validation=True, force_env=bool(body.force),
                 )
                 backend_result = exec_result.result
                 stdout_text = exec_result.stdout

backend/tests/test_outputs_executor_sandbox.py

@@ -0,0 +1,87 @@
+"""Regression tests for the Output backend-code sandbox (issue #134).
+
+The AST allowlist and builtins-scrub sandbox was bypassable via
+`sys.modules["os"]` (a bare-dict lookup the import allowlist never sees) and
+the classic `().__class__.__base__.__subclasses__()` / `__globals__` object
+walk. Separately, `/api/outputs/execute` ran vetted (empty-warning) code in the
+privileged force-mode env, making `os.system` reachable with no consent.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_outputs_executor_sandbox.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import tempfile
+
+import pytest
+
+from backend.apps.outputs.executor import (
+    execute_backend_code,
+    get_code_warnings,
+    UnsafeCodeError,
+)
+
+
+def p_run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+# --- the #134 escape payloads must all be flagged ----------------------------
+
+@pytest.mark.parametrize("code", [
+    "os_mod = sys.modules['os']\nresult = {}",
+    "result = {'x': ().__class__.__base__.__subclasses__()}",
+    "f = (lambda: 0)\nresult = {'g': f.__globals__}",
+    "result = {'c': getattr((), '__class__')}",
+    "result = {'m': __import__('os')}",
+    "result = {'m': sys.modules}",
+    "b = __builtins__\nresult = {}",
+])
+def test_escape_payloads_are_rejected(code):
+    assert get_code_warnings(code), f"expected a warning for: {code!r}"
+    # The strict path (default) runs p_validate_code_safety and must refuse.
+    with pytest.raises(UnsafeCodeError):
+        p_run(execute_backend_code(code, {}))
+
+
+# --- legitimate data-shaping must stay inside the allowlist (no warnings) -----
+
+@pytest.mark.parametrize("code", [
+    'import math, json\nresult = {"a": math.floor(1.9), "b": json.dumps([1, 2])}',
+    'import datetime\nresult = {"t": datetime.date(2026, 7, 20).isoformat()}',
+    'result = {"u": "a,b,c".split(","), "n": len([1, 2, 3])}',
+])
+def test_legit_data_shaping_stays_clean(code):
+    assert get_code_warnings(code) == []
+
+
+# --- end-to-end: the escape no longer reads/writes/execs ----------------------
+
+def test_strict_mode_rejects_before_execution():
+    code = "os_mod = sys.modules['os']\nresult = {'h': os_mod.path.expanduser('~')}"
+    with pytest.raises(UnsafeCodeError):
+        p_run(execute_backend_code(code, {}))
+
+
+def test_minimal_env_blocks_os_system_even_if_ast_bypassed():
+    """Mirrors the `/execute` path: skip_validation=True but force_env=False.
+    Even reaching os via the bypass, `os.system` finds no shell (PATH/COMSPEC
+    stripped by the minimal env), so no command runs."""
+    marker = os.path.join(tempfile.gettempdir(), "openswarm_sandbox_test_marker.txt")
+    if os.path.exists(marker):
+        os.remove(marker)
+    code = (
+        "os_mod = sys.modules['os']\n"
+        f"result = {{'rc': os_mod.system('echo x > {marker.replace(os.sep, '/')}')}}"
+    )
+    p_run(execute_backend_code(code, {}, skip_validation=True, force_env=False))
+    assert not os.path.exists(marker), "os.system reached a shell under the minimal env"
+
+
+def test_allowlisted_code_runs_and_returns_result():
+    code = 'import math\nresult = {"floored": math.floor(input_data["x"])}'
+    out = p_run(execute_backend_code(code, {"x": 3.7}))
+    assert out.result == {"floored": 3}