app/services/code_executor_service.py

from __future__ import annotations

import asyncio
import logging
import re
import shutil
import tempfile
import time
from pathlib import Path
from typing import Optional

from app.config import get_settings
from app.utils.subprocess_utils import run_subprocess

_logger = logging.getLogger(__name__)

_APP_DIR = Path(__file__).resolve().parent
_settings = get_settings()


class CodeSanitizer:
    _BLOCKED_PATTERNS: dict[str, list[tuple[str, str]]] = {
        "python": [
            (r"import\s+subprocess\b", "subprocess not allowed"),
            (r"from\s+subprocess\b", "subprocess not allowed"),
            (r"import\s+ctypes\b", "ctypes not allowed"),
            (r"from\s+ctypes\b", "ctypes not allowed"),
            (r"import\s+os\b", "os not allowed"),
            (r"from\s+os\b", "os not allowed"),
            (r"import\s+sys\b", "sys not allowed"),
            (r"from\s+sys\b", "sys not allowed"),
            (r"import\s+socket\b", "socket not allowed"),
            (r"from\s+socket\b", "socket not allowed"),
            (r"import\s+builtins\b", "builtins not allowed"),
            (r"from\s+builtins\b", "builtins not allowed"),
            (r"import\s+signal\b", "signal not allowed"),
            (r"from\s+signal\b", "signal not allowed"),
            (r"import\s+shutil\b", "shutil not allowed"),
            (r"from\s+shutil\b", "shutil not allowed"),
            (r"__import__\s*\(", "__import__() not allowed"),
            (r"exec\s*\(", "exec() not allowed"),
            (r"eval\s*\(", "eval() not allowed"),
            (r"compile\s*\(", "compile() not allowed"),
            (r"\.__subclasses__\s*\(\)", "subclass escape not allowed"),
            (r"\.__bases__", "__bases__ not allowed"),
            (r"\.__mro__", "__mro__ not allowed"),
            (r"\.__globals__", "__globals__ not allowed"),
            (r"\.__code__", "__code__ not allowed"),
            (r"\.__closure__", "__closure__ not allowed"),
            (r"\.__dict__", "__dict__ not allowed"),
            (r"\.__builtins__", "__builtins__ not allowed"),
            (r"\.__class__", "__class__ not allowed"),
        ],
        "javascript": [
            (r"require\s*\(", "require() not allowed"),
            (r"process\s*\.", "process not allowed"),
            (r"__dirname", "__dirname not allowed"),
            (r"__filename", "__filename not allowed"),
            (r"global\s*\.", "global not allowed"),
            (r"globalThis\s*\.", "globalThis not allowed"),
            (r"eval\s*\(", "eval() not allowed"),
            (r"Function\s*\(", "Function() constructor not allowed"),
            (r"child_process", "child_process not allowed"),
            (r"fs\s*\.", "fs not allowed"),
            (r"net\s*\.", "net not allowed"),
            (r"http\s*\.", "http not allowed"),
            (r"https\s*\.", "https not allowed"),
            (r"worker_threads", "worker_threads not allowed"),
            (r"Buffer\s*\.", "Buffer not allowed"),
        ],
    }

    @classmethod
    def sanitize(cls, code: str, language: str) -> tuple[bool, Optional[str]]:
        patterns = cls._BLOCKED_PATTERNS.get(language, [])
        for pattern, message in patterns:
            if re.search(pattern, code):
                return False, f"Forbidden: {message}"
        return True, None



class CodeExecutorService:
    def __init__(self) -> None:
        self._max_execution_time = _settings.code_exec_max_time
        self._max_output_bytes = _settings.code_exec_max_output
        self._max_memory_mb = _settings.code_exec_max_memory
        self._semaphore = asyncio.Semaphore(_settings.code_exec_max_concurrent)
        self._workdir = Path(_settings.code_exec_workdir)
        self._workdir.mkdir(parents=True, exist_ok=True)

    async def execute(
        self,
        code: str,
        language: str,
        timeout: Optional[int] = None,
    ) -> dict:
        sanitized, err = CodeSanitizer.sanitize(code, language)
        if not sanitized:
            return {
                "success": False, "output": "", "error": err,
                "exit_code": None, "execution_time_ms": None,
                "language": language, "timed_out": False,
            }

        exec_timeout = min(timeout or self._max_execution_time, self._max_execution_time)

        async with self._semaphore:
            run_dir = None
            start_time = time.monotonic()
            try:
                run_dir = Path(tempfile.mkdtemp(dir=self._workdir))
                filename = self._filename_for(language)
                src_path = run_dir / filename
                src_path.write_text(code, encoding="utf-8")

                cmd = self._build_command(language, run_dir, filename)
                result = await asyncio.to_thread(
                    self._run_subprocess, cmd, exec_timeout,
                )

                elapsed_ms = (time.monotonic() - start_time) * 1000

                if result["timed_out"]:
                    return {
                        "success": False,
                        "output": result["stdout"],
                        "error": f"Execution timed out after {exec_timeout}s",
                        "exit_code": result["exit_code"],
                        "execution_time_ms": round(elapsed_ms, 2),
                        "language": language,
                        "timed_out": True,
                    }

                return {
                    "success": result["exit_code"] == 0 if result["exit_code"] is not None else False,
                    "output": result["stdout"],
                    "error": result["stderr"] or None,
                    "exit_code": result["exit_code"],
                    "execution_time_ms": round(elapsed_ms, 2),
                    "language": language,
                    "timed_out": False,
                }

            except FileNotFoundError as exc:
                _logger.error("Runtime not found: %s", exc)
                return {
                    "success": False, "output": "", "error": f"Runtime not found: {exc.filename}",
                    "execution_time_ms": None, "language": language, "timed_out": False,
                }
            except Exception as exc:
                _logger.exception("Executor error")
                return {
                    "success": False, "output": "", "error": f"Internal error: {exc}",
                    "execution_time_ms": None, "language": language, "timed_out": False,
                }
            finally:
                if run_dir and run_dir.exists():
                    try:
                        shutil.rmtree(run_dir, ignore_errors=True)
                    except Exception:
                        pass

    def _run_subprocess(
        self,
        cmd: list[str],
        timeout: float,
    ) -> dict:
        return run_subprocess(cmd, timeout, self._max_output_bytes)

    async def check_runtimes(self) -> dict[str, str]:
        status = {}
        for lang, runtime in [("python", "python3"), ("javascript", "node")]:
            path = shutil.which(runtime)
            status[lang] = f"found at {path}" if path else "missing"
        return status

    @staticmethod
    def _filename_for(language: str) -> str:
        return {"python": "code.py", "javascript": "code.js"
               }[language]

    @staticmethod
    def _build_command(language: str, run_dir: Path, filename: str) -> list[str]:
        sandbox_py = _APP_DIR / "py_sandbox.py"
        sandbox_js = _APP_DIR / "js_sandbox.js"
        if language == "python":
            return ["python3", str(sandbox_py), str(run_dir / filename)]
        elif language == "javascript":
            return ["node", str(sandbox_js), str(run_dir / filename)]
        return []