app/services/repl.py

import sys
import io
import contextlib
import multiprocessing
import traceback
from typing import Dict, Any, Optional

def _restricted_execute(code: str, queue: multiprocessing.Queue):
    """
    Executes code in a completely isolated process environment.
    """
    # Redirect stdout to capture it
    stdout = io.StringIO()
    
    # Define restricted globals
    # We remove dangerous modules like 'os', 'sys', 'shutil', etc.
    safe_globals = {
        "__builtins__": __builtins__.copy(),
        "math": __import__("math"),
        "datetime": __import__("datetime"),
        "json": __import__("json"),
    }
    # Remove dangerous builtins
    dangerous_builtins = ["open", "exec", "eval", "getattr", "setattr", "delattr", "help", "input", "compile"]
    for b in dangerous_builtins:
        if b in safe_globals["__builtins__"]:
            del safe_globals["__builtins__"][b]

    try:
        with contextlib.redirect_stdout(stdout):
            exec(code, safe_globals)
        queue.put({"success": True, "output": stdout.getvalue(), "error": None})
    except Exception:
        queue.put({"success": False, "output": stdout.getvalue(), "error": traceback.format_exc()})

class REPLService:
    def execute(self, code: str, timeout: int = 2) -> Dict[str, Any]:
        """
        Executes Python code in a restricted global environment.
        """
        stdout = io.StringIO()
        
        # Define restricted globals
        safe_globals = {
            "__builtins__": {
                "print": print,
                "range": range,
                "len": len,
                "list": list,
                "dict": dict,
                "str": str,
                "int": int,
                "float": float,
                "bool": bool,
                "abs": abs,
                "sum": sum,
                "min": min,
                "max": max,
                "reversed": reversed,
                "sorted": sorted,
                "set": set,
                "enumerate": enumerate,
                "zip": zip,
            },
            "math": __import__("math"),
            "datetime": __import__("datetime"),
            "json": __import__("json"),
        }

        try:
            with contextlib.redirect_stdout(stdout):
                # Using a wrapper to avoid affecting the main thread too much if it hangs
                # Note: This doesn't actually provide timeout protection in same-thread
                # but works around the multiprocess hang in local dev
                exec(code, safe_globals)
            
            return {
                "success": True, 
                "output": stdout.getvalue(), 
                "error": None
            }
        except Exception:
            return {
                "success": False, 
                "output": stdout.getvalue(), 
                "error": traceback.format_exc()
            }

repl_service = REPLService()