Spaces:
Sleeping
Sleeping
File size: 2,936 Bytes
5e0532d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
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()
|