File size: 4,332 Bytes
9292c2d |
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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
"""
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)
|