from fastapi import FastAPI, HTTPException, status, UploadFile, File, Path as FastAPIPath from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response, HTMLResponse import logging from contextlib import asynccontextmanager from typing import List from sandbox.executor import SandboxExecutor from sandbox.session_manager import SessionManager from sandbox.file_manager import FileManager from sandbox.container_builder import ContainerBuilder from sandbox.models import ( ExecutionRequest, ExecutionResponse, SandboxConfig, Language, CreateSessionRequest, SessionResponse, FileInfo, FileUploadResponse, ExecuteInSessionRequest, ExecuteFileRequest ) from sandbox.language_runners import LanguageRunner # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Global instances executor: SandboxExecutor = None session_manager: SessionManager = None file_manager: FileManager = None @asynccontextmanager async def lifespan(app: FastAPI): """Initialize and cleanup resources""" global executor, session_manager, file_manager docker_available = False try: # Try to initialize Docker services logger.info("Checking Docker availability...") import docker try: docker_client = docker.from_env() docker_client.ping() docker_available = True logger.info("Docker is available, initializing full services...") except Exception as docker_error: logger.warning(f"Docker is not available: {docker_error}") logger.warning("Running in limited mode without Docker support") logger.warning("Session management and file operations will be disabled") docker_available = False if docker_available: # Build/verify devenv image logger.info("Checking development environment image...") try: builder = ContainerBuilder() if not builder.ensure_devenv_image(): logger.warning("Dev environment image not available, sessions will fail") except Exception as e: logger.warning(f"Could not check/build devenv image: {e}") # Initialize services logger.info("Initializing sandbox services...") try: executor = SandboxExecutor(SandboxConfig()) session_manager = SessionManager(SandboxConfig()) file_manager = FileManager() logger.info("All services initialized successfully") except Exception as e: logger.error(f"Failed to initialize some services: {e}") logger.warning("Some features may be unavailable") else: logger.info("Skipping Docker-dependent service initialization") yield except Exception as e: logger.error(f"Lifespan error: {e}", exc_info=True) # Don't raise - allow app to start even with limited functionality yield finally: # Cleanup on shutdown if session_manager: try: logger.info("Shutting down session manager...") session_manager.shutdown() except: pass app = FastAPI( title="isolated-sandbox", description="Execute code in isolated containers with persistent VM-like sessions and file system operations", version="2.0.0", lifespan=lifespan ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/") def root(): """Root endpoint with API information""" docker_available = executor is not None and hasattr(executor, 'client') and executor.client is not None if docker_available: try: executor.client.ping() docker_available = True except: docker_available = False return { "name": "isolated-sandbox", "version": "2.0.0", "status": "running", "docker_available": docker_available, "features": { "stateless_execution": "/execute" if docker_available else "unavailable (Docker required)", "persistent_sessions": "/sessions" if docker_available else "unavailable (Docker required)", "file_operations": docker_available, "multi_language": docker_available }, "supported_languages": ["python", "javascript", "bash"] if docker_available else [], "endpoints": { "execute": "/execute (stateless)" if docker_available else "unavailable", "sessions": "/sessions (create/list)" if docker_available else "unavailable", "session_detail": "/sessions/{session_id}" if docker_available else "unavailable", "files": "/sessions/{session_id}/files" if docker_available else "unavailable", "execute_in_session": "/sessions/{session_id}/execute" if docker_available else "unavailable", "languages": "/languages", "health": "/health", "docs": "/docs", "ui-rules": "/ui-rules" }, "notice": "Docker is not available. Full functionality requires Docker. See /docs for API documentation." if not docker_available else None } @app.get("/health") def health_check(): """Health check endpoint""" try: docker_status = "unavailable" session_count = 0 if executor and hasattr(executor, 'client') and executor.client: try: executor.client.ping() docker_status = "connected" # Check session manager session_count = len(session_manager.sessions) if session_manager else 0 except Exception: docker_status = "disconnected" # App is healthy even without Docker return { "status": "healthy", "docker": docker_status, "active_sessions": session_count, "features_available": docker_status == "connected" } except Exception as e: logger.error(f"Health check failed: {e}") # Still return healthy status, just with error info return { "status": "healthy", "docker": "unknown", "active_sessions": 0, "features_available": False, "warning": str(e) } @app.get("/languages") def list_languages(): """List all supported programming languages""" return { "languages": LanguageRunner.get_all_languages() } @app.get("/ui-rules", response_class=HTMLResponse) def ui_rules_docs(): """UI/UX Design Rules Documentation""" html_content = """ UI/UX Rules - isolated-sandbox

UI/UX Design Rules

Concise rules for building accessible, fast, delightful UIs. Use MUST/SHOULD/NEVER to guide decisions.

Interactions

Keyboard

Targets & Input

Inputs & Forms (Behavior)

State & Navigation

Feedback

Touch/Drag/Scroll

Autofocus

Animation

Layout

Content & Accessibility

Performance

Design

""" return html_content # ========== Stateless Execution (backward compatible) ========== @app.post("/execute", response_model=ExecutionResponse) def execute_code(request: ExecutionRequest): """ Execute code in an isolated ephemeral container (stateless). This is the original execution method - creates a fresh container, executes code, and destroys the container immediately. Note: Requires Docker to be available. On Hugging Face Spaces, Docker-in-Docker is not supported, so this endpoint will not work. """ if not executor: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Code execution requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) try: logger.info(f"Stateless execution: {request.language} code") result = executor.execute(request) return result except Exception as e: logger.error(f"Execution failed: {e}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Execution failed: {str(e)}" ) # ========== Session Management ========== @app.post("/sessions", response_model=SessionResponse, status_code=status.HTTP_201_CREATED) def create_session(request: CreateSessionRequest): """ Create a new persistent VM-like session. The session is a long-running container with persistent storage, supporting file uploads and multiple code executions. """ if not session_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Session management requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) try: logger.info(f"Creating new session with metadata: {request.metadata}") session = session_manager.create_session(request) return session except RuntimeError as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) ) @app.get("/sessions", response_model=List[SessionResponse]) def list_sessions(): """List all active sessions""" if not session_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Session management requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) sessions = session_manager.list_sessions() return sessions @app.get("/sessions/{session_id}", response_model=SessionResponse) def get_session(session_id: str = FastAPIPath(..., description="Session ID")): """Get session details by ID""" if not session_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Session management requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) session = session_manager.get_session(session_id) if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) # Update file count if file_manager: try: files = file_manager.list_files(session.container_id) session.files_count = len(files) except: pass return session @app.delete("/sessions/{session_id}") def destroy_session(session_id: str = FastAPIPath(..., description="Session ID")): """Destroy a session and cleanup all resources""" if not session_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Session management requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) success = session_manager.destroy_session(session_id) if not success: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) return {"message": f"Session {session_id} destroyed successfully"} # ========== File Operations ========== @app.post("/sessions/{session_id}/files", response_model=FileUploadResponse) async def upload_file( session_id: str = FastAPIPath(..., description="Session ID"), file: UploadFile = File(..., description="File to upload") ): """Upload a file to session workspace""" if not session_manager or not file_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. File operations require Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) # Get session session = session_manager.get_session(session_id) if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) try: # Read file data file_data = await file.read() # Upload to container result = file_manager.upload_file( container_id=session.container_id, filename=file.filename, file_data=file_data ) # Update session activity session_manager.update_activity(session_id) return result except ValueError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e) ) except Exception as e: logger.error(f"File upload failed: {e}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"File upload failed: {str(e)}" ) @app.get("/sessions/{session_id}/files", response_model=List[FileInfo]) def list_files(session_id: str = FastAPIPath(..., description="Session ID")): """List files in session workspace""" if not session_manager or not file_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. File operations require Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) # Get session session = session_manager.get_session(session_id) if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) try: files = file_manager.list_files(session.container_id) return files except Exception as e: logger.error(f"File listing failed: {e}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"File listing failed: {str(e)}" ) @app.get("/sessions/{session_id}/files/{filepath:path}") def download_file( session_id: str = FastAPIPath(..., description="Session ID"), filepath: str = FastAPIPath(..., description="File path relative to workspace") ): """Download a file from session workspace""" if not session_manager or not file_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. File operations require Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) # Get session session = session_manager.get_session(session_id) if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) try: file_data = file_manager.download_file(session.container_id, filepath) # Determine content type import mimetypes content_type, _ = mimetypes.guess_type(filepath) return Response( content=file_data, media_type=content_type or "application/octet-stream", headers={ "Content-Disposition": f"attachment; filename={filepath.split('/')[-1]}" } ) except ValueError as e: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(e) ) except Exception as e: logger.error(f"File download failed: {e}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"File download failed: {str(e)}" ) # ========== Execute in Session ========== @app.post("/sessions/{session_id}/execute", response_model=ExecutionResponse) def execute_in_session( session_id: str = FastAPIPath(..., description="Session ID"), request: ExecuteInSessionRequest = None ): """Execute code in an existing session (persistent state)""" if not session_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Session management requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) # Get session session = session_manager.get_session(session_id) if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) try: import time from docker.errors import DockerException container = executor.client.containers.get(session.container_id) runner_config = LanguageRunner.get_runner_config(request.language) start_time = time.time() # Execute command in running container exec_result = container.exec_run( cmd=runner_config["command"] + [request.code], workdir=request.working_dir, demux=True, stream=False ) execution_time = time.time() - start_time # Parse output stdout = exec_result.output[0].decode('utf-8', errors='replace') if exec_result.output[0] else "" stderr = exec_result.output[1].decode('utf-8', errors='replace') if exec_result.output[1] else "" # Update session activity session_manager.update_activity(session_id) return ExecutionResponse( stdout=stdout, stderr=stderr, exit_code=exec_result.exit_code, execution_time=round(execution_time, 3), error=None if exec_result.exit_code == 0 else "Execution failed" ) except DockerException as e: logger.error(f"Docker error: {e}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Execution failed: {str(e)}" ) @app.post("/sessions/{session_id}/execute-file", response_model=ExecutionResponse) def execute_file_in_session( session_id: str = FastAPIPath(..., description="Session ID"), request: ExecuteFileRequest = None ): """Execute an uploaded file in session""" if not session_manager: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Docker is not available. Session management requires Docker which is not supported on this platform. Please deploy to a Docker-capable platform for full functionality." ) # Get session session = session_manager.get_session(session_id) if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) try: import time container = executor.client.containers.get(session.container_id) runner_config = LanguageRunner.get_runner_config(request.language) # Build command based on language if request.language == Language.PYTHON: cmd = ["python", request.filepath] + request.args elif request.language == Language.JAVASCRIPT: cmd = ["node", request.filepath] + request.args elif request.language == Language.BASH: cmd = ["bash", request.filepath] + request.args else: cmd = runner_config["command"] + [request.filepath] + request.args start_time = time.time() # Execute file exec_result = container.exec_run( cmd=cmd, workdir="/workspace", demux=True, stream=False ) execution_time = time.time() - start_time # Parse output stdout = exec_result.output[0].decode('utf-8', errors='replace') if exec_result.output[0] else "" stderr = exec_result.output[1].decode('utf-8', errors='replace') if exec_result.output[1] else "" # Update session activity session_manager.update_activity(session_id) return ExecutionResponse( stdout=stdout, stderr=stderr, exit_code=exec_result.exit_code, execution_time=round(execution_time, 3), error=None if exec_result.exit_code == 0 else "Execution failed" ) except Exception as e: logger.error(f"File execution failed: {e}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"File execution failed: {str(e)}" ) @app.exception_handler(Exception) async def global_exception_handler( request, exc): """Global exception handler""" logger.error(f"Unhandled exception: {exc}", exc_info=True) return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={ "error": "Internal server error", "detail": str(exc) } ) if __name__ == "__main__": import uvicorn uvicorn.run( "app:app", host="0.0.0.0", port=7860, reload=False, log_level="info" )