""" env/executor.py — Secure sandboxed Python code executor for CodeDebugger. Runs untrusted code in isolated subprocesses. Never uses eval() in the main process. Enforces a forbidden-import safety check before execution. """ import re import subprocess import sys import tempfile import os import time from pathlib import Path # ───────────────────────────────────────────────────────────────────────────── # Forbidden patterns checked BEFORE execution # ───────────────────────────────────────────────────────────────────────────── _FORBIDDEN_PATTERNS: list[tuple[str, str]] = [ (r"\bimport\s+os\b", "import os"), (r"\bimport\s+sys\b", "import sys"), (r"\bimport\s+subprocess\b", "import subprocess"), (r"\bimport\s+shutil\b", "import shutil"), (r"\bimport\s+socket\b", "import socket"), (r"\bfrom\s+os\b", "from os"), (r"\bfrom\s+sys\b", "from sys"), (r"\bfrom\s+subprocess\b", "from subprocess"), (r"\bfrom\s+shutil\b", "from shutil"), (r"\bfrom\s+socket\b", "from socket"), (r"\bopen\s*\(", "open("), (r"\beval\s*\(", "eval("), (r"\bexec\s*\(", "exec("), (r"\b__import__\s*\(", "__import__("), ] class CodeExecutor: """ Secure sandboxed Python code executor. All code runs in a fresh subprocess — the main process never calls eval() or exec() on submitted code. A safety pre-check scans for forbidden imports and patterns before any subprocess is spawned. """ def __init__(self, timeout_seconds: int = 10): self.timeout = timeout_seconds # ───────────────────────────────────────────────────────────────────── # Public API # ───────────────────────────────────────────────────────────────────── def check_code_safety(self, code: str) -> dict: """ Scan code for forbidden imports / dangerous patterns. Forbidden tokens: os, sys, subprocess, open(, eval(, exec(, __import__, shutil, socket Returns: {"safe": bool, "violations": list[str]} """ violations: list[str] = [] for pattern, label in _FORBIDDEN_PATTERNS: if re.search(pattern, code): violations.append(label) return { "safe": len(violations) == 0, "violations": violations, } def run_test_case( self, code: str, function_name: str, test_input: str, expected: str, ) -> dict: """ Run *code* against a single test case in an isolated subprocess. Workflow -------- 1. Write a temp .py file: print(()) 2. Execute with subprocess.run(timeout=self.timeout). 3. Compare stripped stdout to stripped expected. 4. Delete the temp file (always, in a finally block). Parameters ---------- code : the submitted Python source function_name : name of the function to call test_input : raw Python expression(s) passed as argument(s) expected : expected string representation of the result Returns ------- { "passed": bool, "actual_output": str, "expected_output": str, "error": str | None, "execution_time": float (seconds) } """ tmp_path: str | None = None start = time.perf_counter() try: # ── 1. Write temp file ─────────────────────────────────────── script = self._build_script(code, function_name, test_input) with tempfile.NamedTemporaryFile( mode="w", suffix=".py", delete=False, encoding="utf-8", ) as f: f.write(script) tmp_path = f.name # ── 2. Execute in subprocess ───────────────────────────────── proc = subprocess.run( [sys.executable, "-u", tmp_path], capture_output=True, text=True, timeout=self.timeout, ) elapsed = time.perf_counter() - start # ── 3. Compare output ──────────────────────────────────────── actual = proc.stdout.strip() exp = expected.strip() passed = actual == exp error: str | None = None if proc.returncode != 0 and proc.stderr: error = proc.stderr.strip() passed = False actual = actual or error return { "passed": passed, "actual_output": actual, "expected_output": exp, "error": error, "execution_time": round(elapsed, 4), } except subprocess.TimeoutExpired: elapsed = time.perf_counter() - start return { "passed": False, "actual_output": "", "expected_output": expected.strip(), "error": f"Timeout after {self.timeout}s", "execution_time": round(elapsed, 4), } except Exception as exc: elapsed = time.perf_counter() - start return { "passed": False, "actual_output": "", "expected_output": expected.strip(), "error": str(exc), "execution_time": round(elapsed, 4), } finally: # ── 4. Always delete temp file ─────────────────────────────── if tmp_path is not None: try: os.unlink(tmp_path) except OSError: pass def run_all_tests( self, code: str, function_name: str, test_cases: list[dict], ) -> dict: """ Run *code* against every test case in *test_cases*. Each entry in test_cases must have "input" and "expected" keys. Returns ------- { "tests_passed": int, "tests_total": int, "pass_rate": float, (0.0–1.0) "results": list, (one dict per test case) "execution_time_total": float (seconds) } """ results: list[dict] = [] total_time = 0.0 for tc in test_cases: result = self.run_test_case( code=code, function_name=function_name, test_input=tc["input"], expected=tc["expected"], ) results.append(result) total_time += result["execution_time"] tests_passed = sum(1 for r in results if r["passed"]) tests_total = len(test_cases) pass_rate = tests_passed / tests_total if tests_total > 0 else 0.0 return { "tests_passed": tests_passed, "tests_total": tests_total, "pass_rate": round(pass_rate, 4), "results": results, "execution_time_total": round(total_time, 4), } # ───────────────────────────────────────────────────────────────────── # Internal helpers # ───────────────────────────────────────────────────────────────────── @staticmethod def _build_script(code: str, function_name: str, test_input: str) -> str: """ Build the temp script that will be executed in the subprocess. Structure: # --- test harness --- print(()) """ # Normalise line endings code = code.rstrip("\n") return ( f"{code}\n\n" f"# --- test harness (auto-generated) ---\n" f"print({function_name}({test_input}))\n" ) # ───────────────────────────────────────────────────────────────────────────── # Quick smoke-test when run directly # ───────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": executor = CodeExecutor(timeout_seconds=5) # Safety check safe_code = "def add(a, b):\n return a + b\n" unsafe_code = "import os\ndef add(a, b):\n return a + b\n" print("=== Safety Check ===") print("Safe code :", executor.check_code_safety(safe_code)) print("Unsafe code:", executor.check_code_safety(unsafe_code)) print() # Single test print("=== Single Test ===") result = executor.run_test_case( code=safe_code, function_name="add", test_input="2, 3", expected="5", ) print(result) print() # All tests print("=== All Tests ===") test_cases = [ {"input": "2, 3", "expected": "5"}, {"input": "0, 0", "expected": "0"}, {"input": "-1, 1", "expected": "0"}, {"input": "10, 20", "expected": "30"}, ] summary = executor.run_all_tests(safe_code, "add", test_cases) print(f"Passed {summary['tests_passed']}/{summary['tests_total']} " f"({summary['pass_rate']:.0%}) in {summary['execution_time_total']}s") for i, r in enumerate(summary["results"], 1): status = "PASS" if r["passed"] else "FAIL" print(f" Test {i}: [{status}] got={r['actual_output']!r} " f"expected={r['expected_output']!r}")