|
|
""" |
|
|
HuggingFace Space: Code Execution |
|
|
Sandboxed Python code execution service |
|
|
""" |
|
|
from fastapi import FastAPI, HTTPException |
|
|
from pydantic import BaseModel |
|
|
import subprocess |
|
|
import tempfile |
|
|
import os |
|
|
from typing import Optional |
|
|
|
|
|
app = FastAPI( |
|
|
title="Code Execution Space", |
|
|
description="Sandboxed Python code execution" |
|
|
) |
|
|
|
|
|
|
|
|
class CodeRequest(BaseModel): |
|
|
code: str |
|
|
language: str = "python" |
|
|
timeout: int = 10 |
|
|
|
|
|
|
|
|
class CodeResponse(BaseModel): |
|
|
stdout: str |
|
|
stderr: str |
|
|
returncode: int |
|
|
error: Optional[str] = None |
|
|
|
|
|
|
|
|
|
|
|
ALLOWED_MODULES = { |
|
|
"math", "datetime", "json", "re", "random", "collections", |
|
|
"itertools", "functools", "string", "statistics", "decimal", |
|
|
"fractions", "numbers", "operator", "copy", "pprint", |
|
|
"textwrap", "unicodedata", "difflib", "enum", "typing", |
|
|
"dataclasses", "abc", "contextlib", "hashlib", "base64" |
|
|
} |
|
|
|
|
|
|
|
|
def validate_code(code: str) -> Optional[str]: |
|
|
"""Basic security validation""" |
|
|
|
|
|
|
|
|
blocked = [ |
|
|
"import os", "import sys", "import subprocess", |
|
|
"__import__", "eval(", "exec(", "compile(", |
|
|
"open(", "file(", "input(", |
|
|
"globals(", "locals(", "vars(", |
|
|
"__builtins__", "__class__", "__bases__", |
|
|
"breakpoint", "exit", "quit" |
|
|
] |
|
|
|
|
|
for pattern in blocked: |
|
|
if pattern in code: |
|
|
return f"Blocked: {pattern} not allowed" |
|
|
|
|
|
return None |
|
|
|
|
|
|
|
|
@app.get("/") |
|
|
async def root(): |
|
|
return { |
|
|
"status": "running", |
|
|
"service": "code_execution", |
|
|
"allowed_modules": list(ALLOWED_MODULES) |
|
|
} |
|
|
|
|
|
|
|
|
@app.post("/api/execute", response_model=CodeResponse) |
|
|
async def execute_code(request: CodeRequest): |
|
|
"""Execute Python code in sandbox""" |
|
|
|
|
|
|
|
|
error = validate_code(request.code) |
|
|
if error: |
|
|
return CodeResponse( |
|
|
stdout="", |
|
|
stderr=error, |
|
|
returncode=1, |
|
|
error=error |
|
|
) |
|
|
|
|
|
|
|
|
if request.language != "python": |
|
|
return CodeResponse( |
|
|
stdout="", |
|
|
stderr=f"Unsupported language: {request.language}", |
|
|
returncode=1, |
|
|
error="Only Python is supported" |
|
|
) |
|
|
|
|
|
try: |
|
|
|
|
|
with tempfile.NamedTemporaryFile( |
|
|
mode="w", |
|
|
suffix=".py", |
|
|
delete=False |
|
|
) as f: |
|
|
f.write(request.code) |
|
|
temp_path = f.name |
|
|
|
|
|
try: |
|
|
|
|
|
result = subprocess.run( |
|
|
["python", temp_path], |
|
|
capture_output=True, |
|
|
text=True, |
|
|
timeout=request.timeout, |
|
|
cwd=tempfile.gettempdir(), |
|
|
env={ |
|
|
"PATH": os.environ.get("PATH", ""), |
|
|
"PYTHONDONTWRITEBYTECODE": "1" |
|
|
} |
|
|
) |
|
|
|
|
|
return CodeResponse( |
|
|
stdout=result.stdout[:10000], |
|
|
stderr=result.stderr[:10000], |
|
|
returncode=result.returncode |
|
|
) |
|
|
|
|
|
finally: |
|
|
|
|
|
os.unlink(temp_path) |
|
|
|
|
|
except subprocess.TimeoutExpired: |
|
|
return CodeResponse( |
|
|
stdout="", |
|
|
stderr=f"Execution timeout ({request.timeout}s)", |
|
|
returncode=1, |
|
|
error="Timeout" |
|
|
) |
|
|
except Exception as e: |
|
|
return CodeResponse( |
|
|
stdout="", |
|
|
stderr=str(e), |
|
|
returncode=1, |
|
|
error=str(e) |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
def gradio_interface(): |
|
|
import gradio as gr |
|
|
|
|
|
def execute_wrapper(code): |
|
|
response = execute_code(CodeRequest(code=code)) |
|
|
if response.returncode == 0: |
|
|
return response.stdout |
|
|
else: |
|
|
return f"Error: {response.stderr}" |
|
|
|
|
|
iface = gr.Interface( |
|
|
fn=execute_wrapper, |
|
|
inputs=gr.Textbox(lines=10, label="Python Code"), |
|
|
outputs=gr.Textbox(lines=10, label="Output"), |
|
|
title="Code Execution", |
|
|
description="Run Python code in a sandbox" |
|
|
) |
|
|
|
|
|
return iface |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
import uvicorn |
|
|
uvicorn.run(app, host="0.0.0.0", port=7860) |
|
|
|