File size: 2,546 Bytes
47fc3ad | 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 91 92 93 94 95 96 97 98 | # code executor
# runs python snippets in a sandboxed subprocess
# hard timeout enforced via subprocess.run(..., timeout=...)
import subprocess
import sys
import tempfile
import os
from typing import Optional
# blocked imports that could harm or infinite loops
BLOCKED_IMPORTS = {
"os.system",
"subprocess.Popen",
"__import__",
"eval(",
"exec(",
"open(",
"socket",
"requests",
"urllib",
"http",
}
def _contains_blocked(code: str) -> Optional[str]:
for blocked in BLOCKED_IMPORTS:
if blocked in code:
return blocked
return None
def execute_code(code: str, timeout: int = 15) -> dict:
"""
Execute a Python code snippet in an isolated subprocess.
Args:
code: Python source code to run.
timeout: Maximum execution time in seconds (default 15).
Returns:
dict with keys 'success', 'stdout', 'stderr', 'exit_code', 'error'.
"""
if not code or not code.strip():
return {
"success": False,
"stdout": "",
"stderr": "",
"exit_code": -1,
"error": "No code provided"
}
blocked = _contains_blocked(code)
if blocked:
return {
"success": False,
"stdout": "",
"stderr": "",
"exit_code": -1,
"error": f"Blocked construct detected: '{blocked}'"
}
# temp file to traceback line numbers
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False, encoding="utf-8"
) as tmp_file:
tmp_file.write(code)
tmp_file_path = tmp_file.name
try:
result = subprocess.run(
[sys.executable, tmp_file_path],
capture_output=True,
text=True,
timeout=timeout
)
stdout = result.stdout[:2000]
stderr = result.stderr[:1000]
return {
"success": result.returncode == 0,
"stdout": stdout,
"stderr": stderr,
"exit_code": result.returncode,
"error": None if result.returncode == 0 else f"Process exited with code {result.returncode}"
}
except subprocess.TimeoutExpired:
return {
"success": False,
"stdout": "",
"stderr": "",
"exit_code": -1,
"error": f"Code execution timed out after {timeout} seconds"
}
finally:
if os.path.exists(tmp_file_path):
os.unlink(tmp_file_path) |