[eric] security: mirror the executor sandbox gate into the public edge, with a drift test

This commit is contained in:
ciregenz
2026-07-31 13:47:11 -07:00
parent 44f9e48c2a
commit 02cda4c47a
6 changed files with 318 additions and 55 deletions
@@ -0,0 +1,29 @@
"""The public edge vendors the desktop sandbox's static gate, and a gate that
only gets tightened on the desktop leaves the internet-facing copy open. So the
two files must stay byte-identical below their docstrings; SECURITY.md has
carried "drift risk" as an open note on this pair since the edge shipped.
Run:
backend/.venv/bin/python -m pytest backend/tests/test_edge_sandbox_mirrors_backend.py -v
"""
import ast
from pathlib import Path
P_ROOT = Path(__file__).resolve().parents[2]
P_DESKTOP = P_ROOT / "backend" / "apps" / "outputs" / "code_safety.py"
P_EDGE = P_ROOT / "openswarm-edge" / "app" / "code_safety.py"
def p_body(path: Path) -> str:
"""The file with its module docstring (the only licensed difference) removed."""
source = path.read_text()
docstring = ast.parse(source).body[0]
return "".join(source.splitlines(keepends=True)[docstring.end_lineno:])
def test_edge_gate_is_a_verbatim_copy_of_the_desktop_gate() -> None:
assert p_body(P_EDGE) == p_body(P_DESKTOP), (
"openswarm-edge/app/code_safety.py has drifted from "
"backend/apps/outputs/code_safety.py; re-copy it below the docstring."
)
+2 -1
View File
@@ -19,7 +19,7 @@
"eslint-knip": "Node tooling deferred to a later pass.", "eslint-knip": "Node tooling deferred to a later pass.",
"classes": "Placeholder check, not wired up. endpoints: orphaned-endpoint triage deferred.", "classes": "Placeholder check, not wired up. endpoints: orphaned-endpoint triage deferred.",
"max-file-lines-exceptions": "Grandfather list of pre-existing >300-line files (existing debt, not new). Paths updated after the folder-tree restructure moved several of them. The two manager/prompt/* entries are from the agent_manager decomposition: prompt_context.py aggregates the system-prompt context builders and attachments.py is one cohesive 230-line attachment resolver; both are single-responsibility and a few lines over, not splittable without an artificial seam.", "max-file-lines-exceptions": "Grandfather list of pre-existing >300-line files (existing debt, not new). Paths updated after the folder-tree restructure moved several of them. The two manager/prompt/* entries are from the agent_manager decomposition: prompt_context.py aggregates the system-prompt context builders and attachments.py is one cohesive 230-line attachment resolver; both are single-responsibility and a few lines over, not splittable without an artificial seam.",
"max-folder-items-exceptions": "Exact-path allow for folders intentionally over the cap. The rule trips at >7 (7 items is fine, the 8th tips it), so only genuinely 8+ folders are listed. backend/ and backend/apps are FastAPI feature-package registries (each child is an app mounted in main.py); agents/ aggregates agent subsystems; agents/manager/ is the agent_manager god-object decomposition (cohesive AgentManager mixins + standalone run helpers + the streaming/permissions/prompt/session subtrees), conventionally flat like agents/ and core/ since its standalone helpers are heterogeneous and don't group cleanly; agents/manager/streaming and agents/manager/session are flat peer collections of one-module-per-concern handlers; core/, tools_lib/, tests/ are conventionally flat. Frontend: app/pages is the page registry, AgentChat/ChatInput/Settings-sections/Onboarding are organizational parents, and shared/state (Redux slices) plus hooks/steps/mcp-cards/Views are flat peer collections. scripts/, electron/, linter/checks/ are flat tool dirs. These replaced blanket .lintignore-max-folder-items sentinels (backend, frontend, scripts, electron, linter/checks) so the rule still catches NEW unplanned bloat everywhere else. Kept as whole-subtree sentinels on purpose: debugger/ (self-contained injected sub-tool with its own Vite GUI), webapp_template (Vite scaffold payload), and vendored mcp-bundles. 2026-07 desktop-shell additions: Dashboard canvas/cards/desktop + hooks/interaction + hooks/lifecycle, AgentChat bubbles/tool-ui, and shared/styles are flat peer collections (one component or hook per concern) that crossed 7 as the redesign surface grew. frontend/src/toolui carries a whole-subtree .lintignore: vendored tool-ui component library (pierre), same treatment as mcp-bundles.", "max-folder-items-exceptions": "Exact-path allow for folders intentionally over the cap. The rule trips at >7 (7 items is fine, the 8th tips it), so only genuinely 8+ folders are listed. backend/ and backend/apps are FastAPI feature-package registries (each child is an app mounted in main.py); agents/ aggregates agent subsystems; agents/manager/ is the agent_manager god-object decomposition (cohesive AgentManager mixins + standalone run helpers + the streaming/permissions/prompt/session subtrees), conventionally flat like agents/ and core/ since its standalone helpers are heterogeneous and don't group cleanly; agents/manager/streaming and agents/manager/session are flat peer collections of one-module-per-concern handlers; core/, tools_lib/, tests/ are conventionally flat. Frontend: app/pages is the page registry, AgentChat/ChatInput/Settings-sections/Onboarding are organizational parents, and shared/state (Redux slices) plus hooks/steps/mcp-cards/Views are flat peer collections. scripts/, electron/, linter/checks/ are flat tool dirs. These replaced blanket .lintignore-max-folder-items sentinels (backend, frontend, scripts, electron, linter/checks) so the rule still catches NEW unplanned bloat everywhere else. Kept as whole-subtree sentinels on purpose: debugger/ (self-contained injected sub-tool with its own Vite GUI), webapp_template (Vite scaffold payload), and vendored mcp-bundles. 2026-07 desktop-shell additions: Dashboard canvas/cards/desktop + hooks/interaction + hooks/lifecycle, AgentChat bubbles/tool-ui, and shared/styles are flat peer collections (one component or hook per concern) that crossed 7 as the redesign surface grew. frontend/src/toolui carries a whole-subtree .lintignore: vendored tool-ui component library (pierre), same treatment as mcp-bundles. openswarm-edge/app is the edge's flat one-module-per-concern set (routing, bundles, inject, ratelimit, sandbox, and the vendored code_safety gate); it crossed 7 when the sandbox's static gate was split out to mirror the desktop file byte for byte.",
"import-cycles": "Flags RUNTIME circular imports only (SCC>1). Skips type-only imports (import type / export type) and dynamic import() since neither runs at module init, which is why the idiomatic Redux store<->hooks type cycle is not flagged. Frontend alias resolution comes from import-cycle-aliases. Zero cycles today; the check keeps it that way.", "import-cycles": "Flags RUNTIME circular imports only (SCC>1). Skips type-only imports (import type / export type) and dynamic import() since neither runs at module init, which is why the idiomatic Redux store<->hooks type cycle is not flagged. Frontend alias resolution comes from import-cycle-aliases. Zero cycles today; the check keeps it that way.",
"ruff + pyright": "Ported from Haik's linter (haik/feat/ingest). ruff is narrowed to F401/F811/F841 (unused imports/redefs/locals) and intentionally DROPS Haik's ARG001/ARG002 (unused args): our SDK-callback signatures require unused params (can_use_tool/pre_tool_hook take a `context` they don't use) and we ban the `_unused` prefix, so ARG is noise here. pyright runs Haik's existence-only config (typeCheckingMode off) with reportAttributeAccessIssue ENABLED: the AgentManager behavior classes now inherit a typing-only AgentManagerProtocol base (manager/AgentManagerProtocol.py) that declares the composed __init__ state + cross-class methods, so the checker sees self.sessions etc. from inside a mixin. pyright caught real bugs: a dangling `_conns` ref + TWO broken lazy imports (`_load_all`/`_load` from outputs.py, renamed to load_all/load in workspace_io but the import sites weren't updated \u2014 App Builder workspace seeding/name-sync was silently failing in a try/except). The one grandfathered SURFACE file (handle_assistant_message) is the SDK-optional try/except-import boundary (TextBlock=object fallback defeats isinstance narrowing). Both grandfather pre-existing debt by file; the refactor surface is clean. Requires `ruff` + `pyright` on PATH (added to requirements-dev.txt); pyright's config expects the venv at backend/.venv.", "ruff + pyright": "Ported from Haik's linter (haik/feat/ingest). ruff is narrowed to F401/F811/F841 (unused imports/redefs/locals) and intentionally DROPS Haik's ARG001/ARG002 (unused args): our SDK-callback signatures require unused params (can_use_tool/pre_tool_hook take a `context` they don't use) and we ban the `_unused` prefix, so ARG is noise here. pyright runs Haik's existence-only config (typeCheckingMode off) with reportAttributeAccessIssue ENABLED: the AgentManager behavior classes now inherit a typing-only AgentManagerProtocol base (manager/AgentManagerProtocol.py) that declares the composed __init__ state + cross-class methods, so the checker sees self.sessions etc. from inside a mixin. pyright caught real bugs: a dangling `_conns` ref + TWO broken lazy imports (`_load_all`/`_load` from outputs.py, renamed to load_all/load in workspace_io but the import sites weren't updated \u2014 App Builder workspace seeding/name-sync was silently failing in a try/except). The one grandfathered SURFACE file (handle_assistant_message) is the SDK-optional try/except-import boundary (TextBlock=object fallback defeats isinstance narrowing). Both grandfather pre-existing debt by file; the refactor surface is clean. Requires `ruff` + `pyright` on PATH (added to requirements-dev.txt); pyright's config expects the venv at backend/.venv.",
"no-underscore-names + p-private": "Convention checks ported verbatim from Haik's linter (haik/feat/ingest): no-underscore-names bans leading-underscore names (a dead-code-tooling blind spot; use p_ for private), p-private enforces that p_-prefixed names are accessed only inside their owning file/class (cross-file/class use means the name should be public). Backend Python only. The exception lists grandfather pre-existing debt that landed with the workflows/analytics forward-ports (eric's 'don't mass-migrate untouched files' rule); the agent_manager refactor surface is clean. NOTE: Haik's full linter (his branch also adds pyright + ruff and runs a different enabled set) should eventually supersede this; these two were lifted to enforce the p_ conventions on eric/dev now. browser_cookies.py is excepted for `_fields_` only: a ctypes.Structure protocol name required by the ctypes metaclass, not our naming." "no-underscore-names + p-private": "Convention checks ported verbatim from Haik's linter (haik/feat/ingest): no-underscore-names bans leading-underscore names (a dead-code-tooling blind spot; use p_ for private), p-private enforces that p_-prefixed names are accessed only inside their owning file/class (cross-file/class use means the name should be public). Backend Python only. The exception lists grandfather pre-existing debt that landed with the workflows/analytics forward-ports (eric's 'don't mass-migrate untouched files' rule); the agent_manager refactor surface is clean. NOTE: Haik's full linter (his branch also adds pyright + ruff and runs a different enabled set) should eventually supersede this; these two were lifted to enforce the p_ conventions on eric/dev now. browser_cookies.py is excepted for `_fields_` only: a ctypes.Structure protocol name required by the ctypes metaclass, not our naming."
@@ -165,6 +165,7 @@
"backend/apps/agents/manager/session", "backend/apps/agents/manager/session",
"backend/apps/agents/manager/streaming", "backend/apps/agents/manager/streaming",
"backend/apps/outputs", "backend/apps/outputs",
"openswarm-edge/app",
"backend/apps/tools_lib", "backend/apps/tools_lib",
"backend/apps/service", "backend/apps/service",
"backend/tests", "backend/tests",
+223
View File
@@ -0,0 +1,223 @@
"""Static safety gate for published apps' backend.py compute.
VENDORED, verbatim below this docstring, from backend/apps/outputs/code_safety.py
(the desktop App Builder gate). test_edge_sandbox_mirrors_backend.py in the
backend suite fails if the two drift, because a gate that is only tightened on
the desktop leaves the internet-facing copy open.
"""
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])
+2 -1
View File
@@ -20,7 +20,8 @@ from .bundles import get_bundle, resolve_file
from .fallback import apex_page, not_found_page from .fallback import apex_page, not_found_page
from .inject import inject_runtime from .inject import inject_runtime
from .ratelimit import RateLimiter from .ratelimit import RateLimiter
from .sandbox import UnsafeCodeError, run_backend from .code_safety import UnsafeCodeError
from .sandbox import run_backend
APPS_BASE_DOMAIN = os.environ.get("APPS_BASE_DOMAIN", "openswarm.host") APPS_BASE_DOMAIN = os.environ.get("APPS_BASE_DOMAIN", "openswarm.host")
# The metered-LLM call goes to the cloud over Fly's PRIVATE 6PN mesh (encrypted, # The metered-LLM call goes to the cloud over Fly's PRIVATE 6PN mesh (encrypted,
+19 -52
View File
@@ -1,13 +1,13 @@
"""Sandboxed Python runner for published apps' backend.py compute. """Sandboxed Python runner for published apps' backend.py compute.
VENDORED from backend/apps/outputs/executor.py (the desktop App Builder runtime). VENDORED from backend/apps/outputs/executor.py (the desktop App Builder runtime).
Keep the allow/deny lists + the subprocess hardening in sync with that file; this Keep the subprocess hardening in sync with that file; this is the same
is the same data-shaping sandbox, just running in the edge instead of on the data-shaping sandbox, just running in the edge instead of on the desktop. The
desktop. Pure compute only: no network, no disk, no subprocess, no secrets. Safe static gate it runs on every call lives in the vendored app/code_safety.py. Pure
to run multi-tenant on one machine because nothing here can reach shared state.""" compute only: no network, no disk, no subprocess, no secrets. Safe to run
multi-tenant on one machine because nothing here can reach shared state."""
from __future__ import annotations from __future__ import annotations
import ast
import asyncio import asyncio
import json import json
import os import os
@@ -15,46 +15,10 @@ import sys
import tempfile import tempfile
from dataclasses import dataclass from dataclasses import dataclass
from app.code_safety import ALLOWED_MODULES, validate_code_safety
TIMEOUT_SECONDS = 30 TIMEOUT_SECONDS = 30
_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",
})
_BLOCKED_BUILTINS = frozenset({
"exec", "eval", "compile", "__import__", "open", "input",
"breakpoint", "exit", "quit",
})
class UnsafeCodeError(Exception):
"""AST validation rejected the backend code."""
def validate_code_safety(code: str) -> None:
"""Raise UnsafeCodeError on the first AST-visible risk. Published apps are
vetted at publish time, but we re-check here: the edge never trusts that the
bundle in storage matches what was scanned."""
try:
tree = ast.parse(code)
except SyntaxError as e:
raise UnsafeCodeError(f"Syntax error: {e}")
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:
raise UnsafeCodeError(f"import '{alias.name}' is not allowed")
elif isinstance(node, ast.ImportFrom):
if node.module and node.module.split(".")[0] not in _ALLOWED_MODULES:
raise UnsafeCodeError(f"import from '{node.module}' is not allowed")
elif isinstance(node, ast.Call):
if isinstance(node.func, ast.Name) and node.func.id in _BLOCKED_BUILTINS:
raise UnsafeCodeError(f"builtin '{node.func.id}()' is not allowed")
def _minimal_env() -> dict: def _minimal_env() -> dict:
return { return {
@@ -79,19 +43,22 @@ async def run_backend(code: str, input_data: dict) -> ComputeResult:
preamble = ( preamble = (
"import json, sys, io, builtins\n" "import json, sys, io, builtins\n"
"for _b in ('exec','eval','compile','open','input',\n" "p_stdout = sys.stdout\n"
" 'breakpoint','exit','quit'):\n" "p_capture = io.StringIO()\n"
" try: delattr(builtins, _b)\n" "sys.stdout = p_capture\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" "input_data = json.loads(sys.stdin.read())\n"
"result = {}\n" "result = {}\n"
# Warm the allowlist BEFORE scrubbing builtins: half the stdlib borrows the builtins the scrub deletes while it loads (tokenize does `from builtins import open`, taking `import dataclasses` with it). Then the module handles go, because leaving `sys` bound hands gate-passing code a live `sys.modules['os']` with no import statement in sight.
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"
) )
postamble = ( postamble = (
"\nsys.stdout = _orig_stdout\n" "\np_stdout.write(json.dumps({\"__stdout__\": p_capture.getvalue(), \"__result__\": result}))\n"
'json.dump({"__stdout__": _capture.getvalue(), "__result__": result}, sys.stdout)\n'
) )
wrapper = preamble + code + postamble wrapper = preamble + code + postamble
+43 -1
View File
@@ -21,7 +21,9 @@ from app.main import slug_from_host
from app.bundles import unpack, resolve_file from app.bundles import unpack, resolve_file
from app.inject import inject_runtime from app.inject import inject_runtime
from app.ratelimit import RateLimiter from app.ratelimit import RateLimiter
from app.sandbox import validate_code_safety, run_backend, UnsafeCodeError from app.code_safety import validate_code_safety, UnsafeCodeError
from app import sandbox as edge_sandbox
from app.sandbox import run_backend
def test_slug_from_host(): def test_slug_from_host():
@@ -179,11 +181,51 @@ def test_sandbox_rejects_unsafe_and_allows_safe():
validate_code_safety("import math\nresult={'x': math.pi}") # no raise validate_code_safety("import math\nresult={'x': math.pi}") # no raise
def test_sandbox_rejects_the_module_handle_escapes():
"""Issue #134 at the public tier: the preamble's own `sys`/`io` handles, an
attribute chain onto a withheld module, and the dunder walk."""
for code in (
"result = {'cwd': sys.modules['os'].getcwd()}",
"result = {'x': str(io.open)}",
"result = {'c': str(json.codecs)}",
"result = {'n': len(().__class__.__bases__[0].__subclasses__())}",
"result = {'c': str(getattr(json, 'codecs'))}",
):
try:
validate_code_safety(code)
assert False, f"expected UnsafeCodeError for {code!r}"
except UnsafeCodeError:
pass
def test_sandbox_runs_safe_code(): def test_sandbox_runs_safe_code():
res = asyncio.run(run_backend("result = {'sum': sum(input_data['nums'])}", {"nums": [1, 2, 3]})) res = asyncio.run(run_backend("result = {'sum': sum(input_data['nums'])}", {"nums": [1, 2, 3]}))
assert res.result == {"sum": 6} assert res.result == {"sum": 6}
def test_sandbox_runs_allowlisted_imports():
"""The builtins scrub used to delete exec/eval, which broke `import statistics`
and every namedtuple; a sandbox that can't run real code isn't secure, it's off."""
res = asyncio.run(run_backend(
"import statistics, datetime\n"
"result = {'mean': statistics.mean(input_data['nums']), 'd': datetime.time(9, 0).isoformat()}",
{"nums": [1, 2, 3]},
))
assert res.result == {"mean": 2, "d": "09:00:00"}
def test_sandbox_subprocess_has_no_module_handles(monkeypatch):
"""Second wall: pretend a payload beats the gate, and the subprocess still
has no module left to grab."""
monkeypatch.setattr(edge_sandbox, "validate_code_safety", lambda code: None)
for handle in ("sys", "io", "builtins"):
try:
asyncio.run(run_backend(f"result = {{'x': str({handle})}}", {}))
assert False, f"{handle} was still reachable"
except RuntimeError as e:
assert "NameError" in str(e)
def test_inject_runtime(): def test_inject_runtime():
out = inject_runtime(b"<html><head><title>x</title></head><body>hi</body></html>").decode() out = inject_runtime(b"<html><head><title>x</title></head><body>hi</body></html>").decode()
assert "OUTPUT_COMPUTE" in out and "OUTPUT_LLM" in out assert "OUTPUT_COMPUTE" in out and "OUTPUT_LLM" in out