""" RUBRA — code_executor.py Safe sandboxed code execution engine. Supports: Python, JavaScript (Node.js), Bash Features: - Timeout protection (30s max) - Memory limit - AST-based dangerous-import/call blocking (Python) — not regex - Restricted builtins actually ENFORCED at exec() time, not just declared - JavaScript static safety check (dangerous modules/keywords) - stdout/stderr capture - Auto-fix loop: if error → AI fixes → re-run (max 3 attempts) SECURITY NOTE: regex-only checks (the old `BLOCKED_PATTERNS` approach) are trivially bypassed — `__import__('os').system(...)`, `getattr(__builtins__, 'ex'+'ec')(...)`, string concatenation, `\x5f\x5fimport\x5f\x5f`, etc. all sail past a substring/regex scan. AST inspection walks the actual parsed structure, so it can't be fooled by how the dangerous call is *spelled* — only by what it *is*. It's still not a perfect sandbox (a determined attacker with a Python interpreter and no import restrictions can chain allowed builtins in surprising ways), so this is defense-in-depth layered on top of: subprocess isolation, a timeout, output caps, and — the part that was previously missing — builtins that are actually restricted at the point of execution, not just defined and ignored. """ import os import re import ast import sys import json import time import signal import asyncio import logging import tempfile import subprocess from pathlib import Path from typing import Optional log = logging.getLogger("rubra.executor") # ── Safety config ───────────────────────────────────────────────────────────── TIMEOUT_SECONDS = 30 MAX_OUTPUT_CHARS = 8000 MAX_FIX_ATTEMPTS = 3 # ═══════════════════════════════════════════════════════ # PYTHON — AST-based safety check # ═══════════════════════════════════════════════════════ # Modules that are never importable from sandboxed code. This list mirrors # (and supersedes) the old regex BLOCKED_PATTERNS, but is enforced by # walking Import/ImportFrom AST nodes instead of grepping source text, so it # also catches `import os as o`, `from os import system as sys_call`, etc. BLOCKED_MODULES = { "os", "subprocess", "sys", "shutil", "socket", "ctypes", "multiprocessing", "threading", "pty", "pickle", "marshal", "shelve", "importlib", "requests", "urllib", "urllib2", "httpx", "aiohttp", "ftplib", "smtplib", "telnetlib", "poplib", "imaplib", "resource", "sysconfig", "code", "codeop", "pdb", "pip", "site", "distutils", "asyncio", # sandboxed code shouldn't be spinning its own event loop "signal", "fcntl", "mmap", "posix", "pwd", "grp", "tty", "termios", "webbrowser", "platform", } # `sys` is allowed by the SAFE_PYTHON_PRELUDE for things like sys.exit-free # introspection, but sys itself exposes far too much (sys.modules, # sys._getframe, etc.) to hand to untrusted code — keep it out of both the # prelude's real import and the blocklist exemptions. # Builtin functions never allowed to be *called*, even though some remain # reachable as names in the restricted builtins dict for introspection # reasons elsewhere in the codebase. BLOCKED_CALL_NAMES = { "eval", "exec", "compile", "__import__", "open", "input", "globals", "locals", "vars", "breakpoint", "help", "memoryview", "exit", "quit", } # Attribute/method names on ANY object that are classic sandbox-escape # vectors (walking the object graph back to builtins/subclasses/frames) # e.g. `().__class__.__bases__[0].__subclasses__()`. BLOCKED_ATTR_NAMES = { "__globals__", "__subclasses__", "__bases__", "__mro__", "__base__", "__builtins__", "__import__", "__loader__", "__spec__", "__code__", "__closure__", "__func__", "__self__", "__dict__", "__getattribute__", "__class__", } # os.system / subprocess.run / shutil.rmtree style dotted calls — kept as a # belt-and-suspenders check in case a blocked module somehow gets imported # indirectly (e.g. via a permitted module re-exporting it). BLOCKED_DOTTED_CALLS = { ("os", "system"), ("os", "popen"), ("os", "fork"), ("os", "kill"), ("subprocess", "run"), ("subprocess", "Popen"), ("subprocess", "call"), ("shutil", "rmtree"), ("shutil", "move"), } class _PySafetyVisitor(ast.NodeVisitor): """Walks the parsed AST once and collects every violation found, rather than bailing on the first one — a single clear error listing everything wrong is more useful than a whack-a-mole retry loop.""" def __init__(self): self.violations: list[str] = [] def visit_Import(self, node: ast.Import): for alias in node.names: root = alias.name.split(".")[0] if root in BLOCKED_MODULES: self.violations.append(f"import of blocked module '{alias.name}'") self.generic_visit(node) def visit_ImportFrom(self, node: ast.ImportFrom): root = (node.module or "").split(".")[0] if root in BLOCKED_MODULES: self.violations.append(f"import from blocked module '{node.module}'") self.generic_visit(node) def visit_Call(self, node: ast.Call): func = node.func if isinstance(func, ast.Name) and func.id in BLOCKED_CALL_NAMES: self.violations.append(f"call to blocked builtin '{func.id}()'") elif isinstance(func, ast.Attribute): attr = func.attr if attr in BLOCKED_ATTR_NAMES: self.violations.append(f"call to blocked attribute '.{attr}()'") if isinstance(func.value, ast.Name) and (func.value.id, attr) in BLOCKED_DOTTED_CALLS: self.violations.append(f"call to blocked '{func.value.id}.{attr}()'") self.generic_visit(node) def visit_Attribute(self, node: ast.Attribute): if node.attr in BLOCKED_ATTR_NAMES: self.violations.append(f"access to blocked attribute '.{node.attr}'") self.generic_visit(node) def visit_Name(self, node: ast.Name): if node.id == "__import__": self.violations.append("reference to '__import__'") self.generic_visit(node) def _python_ast_safety_check(code: str) -> tuple[bool, str]: """Returns (is_safe, reason). A SyntaxError is treated as unsafe rather than silently passed through to the interpreter — malformed input has no business reaching exec().""" try: tree = ast.parse(code) except SyntaxError as e: return False, f"Syntax error: {e}" visitor = _PySafetyVisitor() visitor.visit(tree) if visitor.violations: # De-dupe while preserving order, cap so the error stays readable. seen, uniq = set(), [] for v in visitor.violations: if v not in seen: seen.add(v) uniq.append(v) return False, "Blocked: " + "; ".join(uniq[:5]) return True, "" # ═══════════════════════════════════════════════════════ # JAVASCRIPT — static safety check # ═══════════════════════════════════════════════════════ # No JS parser dependency is available in this environment, so this is a # careful regex pass tuned to the specific escape hatches Node.js exposes # that Python doesn't: requiring child_process/fs/net, process.exit / # process.kill, dynamic require via a variable, vm/Function-based eval # equivalents, and Buffer-based shellcode-ish tricks. Like the old Python # regex layer this is bypassable by a sufficiently motivated obfuscator — # it's here to block the OBVIOUS 99% (subprocess spawn, filesystem access, # arbitrary eval), not to be airtight. _JS_BLOCKED_PATTERNS = [ (r'require\s*\(\s*[\'"]child_process[\'"]\s*\)', "require('child_process')"), (r'require\s*\(\s*[\'"]fs[\'"]\s*\)', "require('fs')"), (r'require\s*\(\s*[\'"]fs/promises[\'"]\s*\)', "require('fs/promises')"), (r'require\s*\(\s*[\'"]net[\'"]\s*\)', "require('net')"), (r'require\s*\(\s*[\'"]dgram[\'"]\s*\)', "require('dgram')"), (r'require\s*\(\s*[\'"]cluster[\'"]\s*\)', "require('cluster')"), (r'require\s*\(\s*[\'"]vm[\'"]\s*\)', "require('vm')"), (r'require\s*\(\s*[\'"]repl[\'"]\s*\)', "require('repl')"), (r'require\s*\(\s*[\'"]worker_threads[\'"]\s*\)', "require('worker_threads')"), (r'\bimport\s*\(\s*[\'"]child_process[\'"]', "dynamic import('child_process')"), (r'\bimport\s*\(\s*[\'"]fs[\'"]', "dynamic import('fs')"), (r'\bprocess\s*\.\s*exit\s*\(', "process.exit()"), (r'\bprocess\s*\.\s*kill\s*\(', "process.kill()"), (r'\bprocess\s*\.\s*binding\s*\(', "process.binding() (internal API escape hatch)"), (r'\bprocess\s*\.\s*mainModule\b', "process.mainModule"), (r'\bglobal\s*\.\s*process\s*\.\s*mainModule\b', "global.process.mainModule"), (r'\brequire\s*\.\s*cache\b', "require.cache tampering"), (r'\brequire\s*\.\s*resolve\s*\(', "require.resolve() (fs probing)"), (r'new\s+Function\s*\(', "new Function(...) (eval equivalent)"), (r'\beval\s*\(', "eval()"), (r'\bexecSync\b|\bspawnSync\b|\bexecFileSync\b', "sync child-process spawn"), (r'\b__dirname\b|\b__filename\b', "filesystem path introspection"), ] _JS_BLOCKED_IDENTIFIERS = {"child_process", "worker_threads"} def _js_safety_check(code: str) -> tuple[bool, str]: hits = [] for pattern, label in _JS_BLOCKED_PATTERNS: if re.search(pattern, code): hits.append(label) for ident in _JS_BLOCKED_IDENTIFIERS: if re.search(r'\b' + re.escape(ident) + r'\b', code): hits.append(f"reference to '{ident}'") if hits: seen, uniq = set(), [] for h in hits: if h not in seen: seen.add(h) uniq.append(h) return False, "Blocked: " + "; ".join(uniq[:5]) return True, "" # ═══════════════════════════════════════════════════════ # RESTRICTED BUILTINS — actually enforced # ═══════════════════════════════════════════════════════ # Names intentionally present so ordinary code still works. _ALLOWED_BUILTIN_NAMES = { "abs", "all", "any", "ascii", "bin", "bool", "bytearray", "bytes", "callable", "chr", "classmethod", "complex", "delattr", "dict", "dir", "divmod", "enumerate", "filter", "float", "format", "frozenset", "getattr", "hasattr", "hash", "hex", "id", "int", "isinstance", "issubclass", "iter", "len", "list", "map", "max", "min", "next", "object", "oct", "ord", "pow", "print", "property", "range", "repr", "reversed", "round", "set", "setattr", "slice", "sorted", "staticmethod", "str", "sum", "super", "tuple", "type", "zip", "True", "False", "None", "NotImplemented", "Ellipsis", # Common, non-dangerous exception types code legitimately catches/raises "Exception", "BaseException", "ValueError", "TypeError", "KeyError", "IndexError", "AttributeError", "StopIteration", "ZeroDivisionError", "RuntimeError", "NotImplementedError", "ArithmeticError", "OverflowError", "FloatingPointError", "AssertionError", "LookupError", "NameError", "UnboundLocalError", "RecursionError", "GeneratorExit", "KeyboardInterrupt", "StopAsyncIteration", } SAFE_PYTHON_PRELUDE = """ import sys as _sys import math import json import re import random import datetime import itertools import functools import collections import statistics from typing import * # ── Restricted builtins — this is the part that must actually be ENFORCED, # not just declared. Previously `_safe_builtins` was built here but never # wired into how the user code actually ran, so it was pure decoration. # The real enforcement now happens where this prelude is exec()'d: the # runner wraps USER CODE in exec(compile(...), {"__builtins__": _safe_builtins}) # so the restricted mapping is what Python's name resolution actually sees. import builtins as _builtins_module _ALLOWED = %r _safe_builtins = {k: v for k, v in vars(_builtins_module).items() if k in _ALLOWED} # BUG FIX (found via testing): a __builtins__ dict with NO '__import__' key # doesn't just block *dangerous* imports — it breaks the `import` statement # entirely, since Python's IMPORT_NAME opcode looks up __import__ on # __builtins__ unconditionally. Without this, even "import math" inside the # restricted exec() raised "ImportError: __import__ not found", making the # sandbox unable to run almost any real code. Fix: install a WHITELIST- # CHECKED __import__ — a second, independent enforcement layer alongside # the AST blacklist check in _python_ast_safety_check(). The two layers # check different things (static text vs. runtime call) and use different # list philosophies (blacklist vs. whitelist) on purpose — a gap or bug in # either one alone still leaves the other standing. _ALLOWED_RUNTIME_IMPORTS = { "math", "json", "re", "random", "datetime", "itertools", "functools", "collections", "statistics", "string", "time", "decimal", "fractions", "heapq", "bisect", "array", "copy", "typing", "dataclasses", "enum", "textwrap", "unicodedata", "difflib", "csv", "io", "operator", "uuid", "hashlib", "base64", "struct", "queue", "abc", "numbers", "calendar", } def _safe_import(name, globals=None, locals=None, fromlist=(), level=0): root = name.split(".")[0] if root not in _ALLOWED_RUNTIME_IMPORTS: raise ImportError(f"Import of '{name}' is not allowed in the sandbox") return __import__(name, globals, locals, fromlist, level) _safe_builtins["__import__"] = _safe_import del _sys, _builtins_module, _ALLOWED """ % (_ALLOWED_BUILTIN_NAMES,) # ── Language detection ──────────────────────────────────────────────────────── def detect_language(code: str, hint: str = "") -> str: """Detect programming language from code or hint.""" hint = hint.lower() if hint in ("python", "py"): return "python" if hint in ("javascript", "js", "node"): return "javascript" if hint in ("bash", "sh", "shell"): return "bash" # Auto-detect from code if re.search(r'^\s*(def |import |from .* import|print\(|if __name__)', code, re.M): return "python" if re.search(r'(const |let |var |=>|console\.log|require\(|async function)', code): return "javascript" if re.search(r'^#!(\/usr\/bin\/(env )?bash|\/bin\/sh)', code): return "bash" if re.search(r'function\s+\w+\s*\(', code): return "javascript" return "python" # default # ── Safety check ───────────────────────────────────────────────────────────── def safety_check(code: str, lang: str) -> tuple[bool, str]: """Returns (is_safe, reason)""" if lang == "python": return _python_ast_safety_check(code) if lang == "javascript": return _js_safety_check(code) # Block rm -rf style commands if lang == "bash": dangerous = [r'rm\s+-rf', r'mkfs', r'dd\s+if=', r':\(\)\{', r'fork bomb'] for p in dangerous: if re.search(p, code, re.IGNORECASE): return False, f"Dangerous bash command blocked" return True, "" # ── Python executor ─────────────────────────────────────────────────────────── async def run_python(code: str) -> dict: """ Run Python code in a subprocess with timeout. The subprocess itself is the outer sandbox boundary (process isolation, kill on timeout, capped output). Inside that subprocess, the user code is executed via exec() against a globals dict whose "__builtins__" is the RESTRICTED mapping built by SAFE_PYTHON_PRELUDE — this is what actually enforces the allowlist, rather than just defining it and letting the script run with full normal builtins regardless. """ # User code is embedded as a JSON string literal (not f-string-spliced # raw text) so there's no risk of the user's code breaking out of the # wrapper via a stray triple-quote or backslash sequence. user_code_literal = json.dumps(code) runner = ( SAFE_PYTHON_PRELUDE + "\n# === RUNNER ===\n" + "_user_code = " + user_code_literal + "\n" + "_restricted_globals = {\n" + " '__builtins__': _safe_builtins, '__name__': '__main__',\n" + " 'math': math, 'json': json, 're': re, 'random': random,\n" + " 'datetime': datetime, 'itertools': itertools,\n" + " 'functools': functools, 'collections': collections,\n" + " 'statistics': statistics,\n" + "}\n" + "exec(compile(_user_code, '', 'exec'), _restricted_globals)\n" ) with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, dir='/tmp') as f: f.write(runner) tmp_path = f.name try: proc = await asyncio.create_subprocess_exec( sys.executable, tmp_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd='/tmp', ) try: stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=TIMEOUT_SECONDS ) except asyncio.TimeoutError: proc.kill() return { "success": False, "stdout": "", "stderr": f"⏱ Execution timed out after {TIMEOUT_SECONDS}s", "exit_code": -1, "lang": "python", } out = stdout.decode("utf-8", errors="replace")[:MAX_OUTPUT_CHARS] err = stderr.decode("utf-8", errors="replace")[:2000] # Filter out runner-wrapper noise from stderr tracebacks so users see # their own code's error, not our wrapper's internal frame. err_lines = [l for l in err.splitlines() if 'SAFE_PYTHON_PRELUDE' not in l and 'tmp_path' not in l and '_restricted_globals' not in l and '/tmp/' not in l.lower()[:20]] err = "\n".join(err_lines) return { "success": proc.returncode == 0, "stdout": out, "stderr": err, "exit_code": proc.returncode, "lang": "python", } finally: try: os.unlink(tmp_path) except: pass # ── JavaScript executor ─────────────────────────────────────────────────────── async def run_javascript(code: str) -> dict: """Run JavaScript with Node.js.""" # Check node available try: check = await asyncio.create_subprocess_exec( "node", "--version", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) await check.communicate() except FileNotFoundError: return {"success": False, "stdout": "", "stderr": "Node.js not installed on this server", "exit_code": -1, "lang": "javascript"} with tempfile.NamedTemporaryFile(mode='w', suffix='.js', delete=False, dir='/tmp') as f: f.write(code) tmp_path = f.name try: proc = await asyncio.create_subprocess_exec( "node", "--max-old-space-size=128", tmp_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd='/tmp', ) try: stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=TIMEOUT_SECONDS ) except asyncio.TimeoutError: proc.kill() return {"success": False, "stdout": "", "stderr": f"⏱ Timeout after {TIMEOUT_SECONDS}s", "exit_code": -1, "lang": "javascript"} return { "success": proc.returncode == 0, "stdout": stdout.decode("utf-8", errors="replace")[:MAX_OUTPUT_CHARS], "stderr": stderr.decode("utf-8", errors="replace")[:2000], "exit_code": proc.returncode, "lang": "javascript", } finally: try: os.unlink(tmp_path) except: pass # ── Bash executor ───────────────────────────────────────────────────────────── async def run_bash(code: str) -> dict: """Run bash in restricted environment.""" with tempfile.NamedTemporaryFile(mode='w', suffix='.sh', delete=False, dir='/tmp') as f: f.write("#!/bin/bash\nset -euo pipefail\n" + code) tmp_path = f.name os.chmod(tmp_path, 0o700) try: proc = await asyncio.create_subprocess_exec( "bash", tmp_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd='/tmp', ) try: stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=15 # shorter for bash ) except asyncio.TimeoutError: proc.kill() return {"success": False, "stdout": "", "stderr": "⏱ Bash timeout after 15s", "exit_code": -1, "lang": "bash"} return { "success": proc.returncode == 0, "stdout": stdout.decode("utf-8", errors="replace")[:MAX_OUTPUT_CHARS], "stderr": stderr.decode("utf-8", errors="replace")[:2000], "exit_code": proc.returncode, "lang": "bash", } finally: try: os.unlink(tmp_path) except: pass # ── Main execute function ───────────────────────────────────────────────────── async def execute_code(code: str, lang: str = "") -> dict: """ Execute code safely. Returns execution result dict. """ lang = detect_language(code, lang) # Safety check safe, reason = safety_check(code, lang) if not safe: return { "success": False, "stdout": "", "stderr": f"🚫 Blocked: {reason}", "exit_code": -1, "lang": lang, "blocked": True, } start = time.time() if lang == "python": result = await run_python(code) elif lang == "javascript": result = await run_javascript(code) elif lang == "bash": result = await run_bash(code) else: result = {"success": False, "stdout": "", "stderr": f"Unsupported language: {lang}", "exit_code": -1, "lang": lang} result["elapsed_ms"] = int((time.time() - start) * 1000) return result # ── Auto-fix loop (Kimi-style) ──────────────────────────────────────────────── async def execute_with_autofix( code: str, lang: str = "", task_description: str = "", llm_func = None, ) -> dict: """ Execute code. If error → ask AI to fix → re-execute. Repeats up to MAX_FIX_ATTEMPTS times. Returns final result with fix history. """ lang = detect_language(code, lang) attempts = [] current_code = code for attempt in range(1, MAX_FIX_ATTEMPTS + 1): log.info(f"[EXECUTOR] Attempt {attempt}/{MAX_FIX_ATTEMPTS} — {lang}") result = await execute_code(current_code, lang) result["attempt"] = attempt result["code"] = current_code attempts.append(result) if result["success"]: log.info(f"[EXECUTOR] ✅ Success on attempt {attempt}") break if result.get("blocked"): break # Error — ask AI to fix (if llm_func provided) if llm_func and attempt < MAX_FIX_ATTEMPTS: log.info(f"[EXECUTOR] ⚡ Auto-fixing error...") error_msg = result.get("stderr", "") or result.get("stdout", "") fix_prompt = f"""Fix this {lang} code that has an error. ORIGINAL TASK: {task_description or 'Execute the code correctly'} CODE: ```{lang} {current_code} ``` ERROR: ``` {error_msg[:500]} ``` Return ONLY the fixed code. No explanation. No markdown fences.""" msgs = [ {"role": "system", "content": f"You are an expert {lang} debugger. Fix the error and return only the corrected code."}, {"role": "user", "content": fix_prompt}, ] fixed_code = "" try: async for tok in llm_func(msgs, mode="coding"): fixed_code += tok # Strip markdown fences if AI adds them fixed_code = re.sub(r'^```\w*\n?', '', fixed_code.strip()) fixed_code = re.sub(r'\n?```$', '', fixed_code.strip()) if fixed_code: current_code = fixed_code log.info(f"[EXECUTOR] AI fix applied, retrying...") else: break except Exception as e: log.warning(f"[EXECUTOR] AI fix failed: {e}") break final = attempts[-1].copy() final["attempts"] = len(attempts) final["fix_history"] = attempts if len(attempts) > 1 else [] final["auto_fixed"] = len(attempts) > 1 and attempts[-1]["success"] return final # ── Format result for SSE ───────────────────────────────────────────────────── def format_execution_result(result: dict) -> str: """Format execution result as readable text for frontend.""" lang = result.get("lang", "") success = result.get("success", False) stdout = result.get("stdout", "").strip() stderr = result.get("stderr", "").strip() elapsed = result.get("elapsed_ms", 0) attempts = result.get("attempts", 1) auto_fix = result.get("auto_fixed", False) lines = [] if auto_fix: lines.append(f"🔧 Auto-fixed after {attempts} attempt(s)\n") if success: lines.append(f"✅ **Executed** ({lang}, {elapsed}ms)") if stdout: lines.append(f"\n**Output:**\n```\n{stdout}\n```") else: lines.append("\n*(No output)*") else: if result.get("blocked"): lines.append(f"🚫 **Blocked:** {stderr}") else: lines.append(f"❌ **Error** ({lang}, exit {result.get('exit_code', '?')})") if stderr: lines.append(f"\n```\n{stderr[:800]}\n```") if stdout: lines.append(f"\n**Partial output:**\n```\n{stdout[:400]}\n```") return "\n".join(lines)