env/executor.py

"""
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:
               <submitted code>
               print(<function_name>(<test_input>))
        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:
            <submitted code>

            # --- test harness ---
            print(<function_name>(<test_input>))
        """
        # 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}")