code-exec / app.py
ndwdgda's picture
Upload folder using huggingface_hub
9292c2d verified
"""
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 # seconds
class CodeResponse(BaseModel):
stdout: str
stderr: str
returncode: int
error: Optional[str] = None
# Allowed modules for security
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 patterns
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"""
# Validate
error = validate_code(request.code)
if error:
return CodeResponse(
stdout="",
stderr=error,
returncode=1,
error=error
)
# Only Python supported
if request.language != "python":
return CodeResponse(
stdout="",
stderr=f"Unsupported language: {request.language}",
returncode=1,
error="Only Python is supported"
)
try:
# Write to temp file
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".py",
delete=False
) as f:
f.write(request.code)
temp_path = f.name
try:
# Execute with resource limits
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], # Limit output
stderr=result.stderr[:10000],
returncode=result.returncode
)
finally:
# Clean up
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)
)
# For Gradio interface (HF Spaces requirement)
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)