Upload app.py - Ubuntu Sandbox v2.0

app.py

@@ -0,0 +1,1440 @@
+#!/usr/bin/env python3
+"""
+Ubuntu Sandbox Environment for HuggingFace Spaces
+A production-grade, secure, and AI-accessible Ubuntu development environment.
+Version: 2.0.0
+Author: MiniMax Agent
+License: MIT
+
+Features:
+- Secure command execution with sandboxing
+- AI-friendly REST API
+- Real-time monitoring and logging
+- Resource management and limits
+- Session management
+- File system operations
+- System monitoring
+"""
+
+import os
+import sys
+import subprocess
+import pty
+import select
+import signal
+import json
+import time
+import threading
+import logging
+import hashlib
+import tempfile
+import contextlib
+from pathlib import Path
+from datetime import datetime, timezone
+from typing import Dict, List, Optional, Any, Union
+import uuid
+import re
+import functools
+import asyncio
+from dataclasses import dataclass
+from enum import Enum
+
+# Core dependencies
+import gradio as gr
+import psutil
+import requests
+from werkzeug.exceptions import HTTPException
+
+# Version info
+__version__ = "2.0.0"
+__author__ = "MiniMax Agent"
+__license__ = "MIT"
+
+# Configuration
+@dataclass
+class Config:
+    """Application configuration"""
+    # Security settings
+    MAX_COMMAND_LENGTH: int = 1000
+    MAX_FILE_SIZE: int = 10 * 1024 * 1024  # 10MB
+    COMMAND_TIMEOUT: int = 30
+    SESSION_TIMEOUT: int = 3600
+    RATE_LIMIT_REQUESTS: int = 100
+    RATE_LIMIT_WINDOW: int = 60  # seconds
+    
+    # Resource limits
+    MAX_MEMORY_MB: int = 4096
+    MAX_DISK_GB: int = 10
+    MAX_CPU_PERCENT: int = 80
+    
+    # File paths
+    WORKSPACE_PATH: str = "/workspace"
+    LOGS_PATH: str = "/workspace/logs"
+    
+    # Security - restricted commands
+    RESTRICTED_COMMANDS: List[str] = None
+    
+    def __post_init__(self):
+        if self.RESTRICTED_COMMANDS is None:
+            self.RESTRICTED_COMMANDS = [
+                "sudo", "su", "passwd", "useradd", "userdel", "usermod",
+                "chmod", "chown", "mount", "umount", "fdisk", "mkfs",
+                "dd", "shred", "docker", "systemctl", "service",
+                "init", "reboot", "shutdown", "halt", "poweroff",
+                "crontab", "at", "screen", "tmux", "vim", "emacs",
+                "nohup", "setsid", "exec", "source"
+            ]
+
+# Global configuration
+config = Config()
+
+# Setup logging
+def setup_logging():
+    """Setup comprehensive logging"""
+    os.makedirs(config.LOGS_PATH, exist_ok=True)
+    
+    # Create logger
+    logger = logging.getLogger("ubuntu_sandbox")
+    logger.setLevel(logging.INFO)
+    
+    # File handler
+    file_handler = logging.FileHandler(
+        Path(config.LOGS_PATH) / "sandbox.log"
+    )
+    file_handler.setLevel(logging.INFO)
+    
+    # Console handler
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.WARNING)
+    
+    # Formatter
+    formatter = logging.Formatter(
+        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+    file_handler.setFormatter(formatter)
+    console_handler.setFormatter(formatter)
+    
+    logger.addHandler(file_handler)
+    logger.addHandler(console_handler)
+    
+    return logger
+
+logger = setup_logging()
+
+# Session management
+class SessionManager:
+    """Manage user sessions with timeout and tracking"""
+    
+    def __init__(self):
+        self.sessions: Dict[str, Dict] = {}
+        self.command_history: Dict[str, List[str]] = {}
+        self.current_directories: Dict[str, str] = {}
+        self.last_activity: Dict[str, float] = {}
+    
+    def create_session(self, session_id: Optional[str] = None) -> str:
+        """Create a new session"""
+        if not session_id:
+            session_id = str(uuid.uuid4())
+        
+        self.sessions[session_id] = {
+            "id": session_id,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            "commands_executed": 0,
+            "files_created": 0,
+            "memory_usage": 0
+        }
+        self.command_history[session_id] = []
+        self.current_directories[session_id] = config.WORKSPACE_PATH
+        self.last_activity[session_id] = time.time()
+        
+        logger.info(f"Session created: {session_id}")
+        return session_id
+    
+    def get_session(self, session_id: str) -> Optional[Dict]:
+        """Get session information"""
+        return self.sessions.get(session_id)
+    
+    def update_activity(self, session_id: str):
+        """Update last activity timestamp"""
+        if session_id in self.last_activity:
+            self.last_activity[session_id] = time.time()
+    
+    def cleanup_expired_sessions(self):
+        """Clean up expired sessions"""
+        current_time = time.time()
+        expired_sessions = []
+        
+        for session_id, last_activity in self.last_activity.items():
+            if current_time - last_activity > config.SESSION_TIMEOUT:
+                expired_sessions.append(session_id)
+        
+        for session_id in expired_sessions:
+            self.delete_session(session_id)
+    
+    def delete_session(self, session_id: str):
+        """Delete a session and all associated data"""
+        self.sessions.pop(session_id, None)
+        self.command_history.pop(session_id, None)
+        self.current_directories.pop(session_id, None)
+        self.last_activity.pop(session_id, None)
+        
+        logger.info(f"Session deleted: {session_id}")
+    
+    def get_current_directory(self, session_id: str) -> str:
+        """Get current working directory for session"""
+        return self.current_directories.get(session_id, config.WORKSPACE_PATH)
+    
+    def set_current_directory(self, session_id: str, directory: str):
+        """Set current working directory for session"""
+        self.current_directories[session_id] = directory
+    
+    def add_command_to_history(self, session_id: str, command: str):
+        """Add command to session history"""
+        if session_id in self.command_history:
+            self.command_history[session_id].append(command)
+            # Keep only last 100 commands
+            if len(self.command_history[session_id]) > 100:
+                self.command_history[session_id] = self.command_history[session_id][-100:]
+
+# Security validation
+class SecurityValidator:
+    """Validate commands and inputs for security"""
+    
+    @staticmethod
+    def validate_command(command: str) -> tuple[bool, str]:
+        """Validate command for security issues"""
+        if not command or not command.strip():
+            return False, "Empty command"
+        
+        if len(command) > config.MAX_COMMAND_LENGTH:
+            return False, f"Command too long (max {config.MAX_COMMAND_LENGTH} characters)"
+        
+        # Check for restricted commands
+        command_lower = command.lower().strip()
+        for restricted in config.RESTRICTED_COMMANDS:
+            if command_lower.startswith(restricted):
+                return False, f"Restricted command: {restricted}"
+        
+        # Check for dangerous patterns
+        dangerous_patterns = [
+            r'rm\s+-rf\s+/',
+            r'mkdir\s+.*/\.\.',
+            r'cd\s+\.\.\.\s*/',
+            r'>\s*/dev/',
+            r'&\s*\|\s*',
+            r';\s*shutdown',
+            r';\s*reboot',
+            r';\s*halt',
+        ]
+        
+        for pattern in dangerous_patterns:
+            if re.search(pattern, command):
+                return False, f"Dangerous pattern detected: {pattern}"
+        
+        return True, "Command is valid"
+    
+    @staticmethod
+    def validate_filename(filename: str) -> tuple[bool, str]:
+        """Validate filename for security"""
+        if not filename or not filename.strip():
+            return False, "Empty filename"
+        
+        # Check for path traversal
+        if ".." in filename or filename.startswith("/"):
+            return False, "Path traversal not allowed"
+        
+        # Check for dangerous characters
+        dangerous_chars = ['<', '>', ':', '"', '|', '?', '*']
+        if any(char in filename for char in dangerous_chars):
+            return False, f"Invalid characters in filename: {dangerous_chars}"
+        
+        # Check length
+        if len(filename) > 255:
+            return False, "Filename too long"
+        
+        return True, "Filename is valid"
+    
+    @staticmethod
+    def validate_file_content(content: str) -> tuple[bool, str]:
+        """Validate file content for size"""
+        if len(content.encode('utf-8')) > config.MAX_FILE_SIZE:
+            return False, f"File too large (max {config.MAX_FILE_SIZE} bytes)"
+        
+        return True, "Content is valid"
+
+# Command execution
+class SecureExecutor:
+    """Secure command execution with resource management"""
+    
+    def __init__(self, session_manager: SessionManager):
+        self.session_manager = session_manager
+        self.running_processes: Dict[str, subprocess.Popen] = {}
+    
+    def execute_command(self, command: str, session_id: str) -> Dict[str, Any]:
+        """Execute command securely"""
+        # Validate command
+        is_valid, validation_message = SecurityValidator.validate_command(command)
+        if not is_valid:
+            logger.warning(f"Command validation failed for session {session_id}: {validation_message}")
+            return {
+                "success": False,
+                "error": f"Command validation failed: {validation_message}",
+                "output": "",
+                "exit_code": -1,
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+        
+        # Get session info
+        session = self.session_manager.get_session(session_id)
+        if not session:
+            return {
+                "success": False,
+                "error": "Invalid session",
+                "output": "",
+                "exit_code": -1,
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+        
+        # Update activity
+        self.session_manager.update_activity(session_id)
+        
+        # Get current directory
+        current_dir = self.session_manager.get_current_directory(session_id)
+        
+        start_time = time.time()
+        process_id = str(uuid.uuid4())
+        
+        try:
+            # Check resource usage before execution
+            memory_usage = psutil.virtual_memory().percent
+            cpu_usage = psutil.cpu_percent(interval=0.1)
+            
+            if memory_usage > config.MAX_CPU_PERCENT:
+                return {
+                    "success": False,
+                    "error": f"System resources overloaded: Memory {memory_usage:.1f}%",
+                    "output": "",
+                    "exit_code": -1,
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Execute command
+            process = subprocess.Popen(
+                command,
+                shell=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                cwd=current_dir,
+                preexec_fn=os.setsid
+            )
+            
+            self.running_processes[process_id] = process
+            
+            # Wait for completion with timeout
+            try:
+                stdout, stderr = process.communicate(timeout=config.COMMAND_TIMEOUT)
+                execution_time = time.time() - start_time
+                
+                # Update session stats
+                session["commands_executed"] += 1
+                self.session_manager.add_command_to_history(session_id, command)
+                
+                # Log execution
+                logger.info(
+                    f"Command executed - Session: {session_id}, "
+                    f"Command: {command[:50]}..., "
+                    f"Exit code: {process.returncode}, "
+                    f"Time: {execution_time:.2f}s"
+                )
+                
+                result = {
+                    "success": process.returncode == 0,
+                    "output": stdout + stderr if stderr else stdout,
+                    "exit_code": process.returncode,
+                    "execution_time": execution_time,
+                    "timestamp": datetime.now(timezone.utc).isoformat(),
+                    "process_id": process_id
+                }
+                
+                # Handle directory changes
+                if command.strip().startswith("cd "):
+                    new_path = command.strip()[3:].strip()
+                    if new_path:
+                        try:
+                            if new_path.startswith("/"):
+                                full_path = new_path
+                            else:
+                                full_path = str(Path(current_dir) / new_path).resolve()
+                            self.session_manager.set_current_directory(session_id, str(full_path))
+                            result["new_directory"] = str(full_path)
+                        except Exception as e:
+                            logger.warning(f"Directory change failed: {e}")
+                
+                return result
+                
+            except subprocess.TimeoutExpired:
+                # Kill the process
+                try:
+                    os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+                    process.wait(timeout=5)
+                except:
+                    process.kill()
+                
+                logger.warning(f"Command timeout - Session: {session_id}, Command: {command[:50]}...")
+                return {
+                    "success": False,
+                    "error": f"Command timed out after {config.COMMAND_TIMEOUT} seconds",
+                    "output": "",
+                    "exit_code": -1,
+                    "timestamp": datetime.now(timezone.utc).isoformat(),
+                    "process_id": process_id
+                }
+        
+        except Exception as e:
+            logger.error(f"Command execution error - Session: {session_id}, Error: {e}")
+            return {
+                "success": False,
+                "error": f"Execution error: {str(e)}",
+                "output": "",
+                "exit_code": -1,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "process_id": process_id
+            }
+        
+        finally:
+            # Clean up process
+            self.running_processes.pop(process_id, None)
+    
+    def kill_process(self, process_id: str) -> bool:
+        """Kill a running process"""
+        process = self.running_processes.get(process_id)
+        if process:
+            try:
+                os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+                process.wait(timeout=5)
+                self.running_processes.pop(process_id, None)
+                return True
+            except:
+                process.kill()
+                self.running_processes.pop(process_id, None)
+                return True
+        return False
+
+# File operations
+class FileManager:
+    """Secure file operations"""
+    
+    def __init__(self, session_manager: SessionManager):
+        self.session_manager = session_manager
+    
+    def list_directory(self, path: str, session_id: str) -> Dict[str, Any]:
+        """List directory contents"""
+        try:
+            # Validate path
+            if not path or path == ".":
+                path = self.session_manager.get_current_directory(session_id)
+            
+            directory = Path(path)
+            if not directory.exists():
+                return {
+                    "success": False,
+                    "error": "Directory not found",
+                    "files": [],
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            if not directory.is_dir():
+                return {
+                    "success": False,
+                    "error": "Not a directory",
+                    "files": [],
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            files = []
+            for item in directory.iterdir():
+                try:
+                    stat = item.stat()
+                    files.append({
+                        "name": item.name,
+                        "type": "directory" if item.is_dir() else "file",
+                        "size": stat.st_size if item.is_file() else 0,
+                        "modified": datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(),
+                        "permissions": oct(stat.st_mode)[-3:]
+                    })
+                except (PermissionError, OSError) as e:
+                    logger.warning(f"Error accessing file {item.name}: {e}")
+                    continue
+            
+            logger.info(f"Directory listed - Session: {session_id}, Path: {path}, Items: {len(files)}")
+            
+            return {
+                "success": True,
+                "files": files,
+                "path": str(path),
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+        
+        except Exception as e:
+            logger.error(f"Directory listing error - Session: {session_id}, Path: {path}, Error: {e}")
+            return {
+                "success": False,
+                "error": f"Directory listing failed: {str(e)}",
+                "files": [],
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+    
+    def read_file(self, filename: str, session_id: str) -> Dict[str, Any]:
+        """Read file content"""
+        try:
+            # Validate filename
+            is_valid, validation_message = SecurityValidator.validate_filename(filename)
+            if not is_valid:
+                return {
+                    "success": False,
+                    "error": f"Invalid filename: {validation_message}",
+                    "content": "",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Get file path
+            current_dir = self.session_manager.get_current_directory(session_id)
+            file_path = Path(current_dir) / filename
+            
+            if not file_path.exists():
+                return {
+                    "success": False,
+                    "error": "File not found",
+                    "content": "",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            if not file_path.is_file():
+                return {
+                    "success": False,
+                    "error": "Not a file",
+                    "content": "",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Check file size
+            if file_path.stat().st_size > config.MAX_FILE_SIZE:
+                return {
+                    "success": False,
+                    "error": f"File too large (max {config.MAX_FILE_SIZE} bytes)",
+                    "content": "",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Read file
+            content = file_path.read_text(encoding='utf-8')
+            
+            logger.info(f"File read - Session: {session_id}, File: {filename}, Size: {len(content)}")
+            
+            return {
+                "success": True,
+                "content": content,
+                "size": file_path.stat().st_size,
+                "filename": filename,
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+        
+        except Exception as e:
+            logger.error(f"File read error - Session: {session_id}, File: {filename}, Error: {e}")
+            return {
+                "success": False,
+                "error": f"File read failed: {str(e)}",
+                "content": "",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+    
+    def create_file(self, filename: str, content: str, session_id: str) -> Dict[str, Any]:
+        """Create a new file"""
+        try:
+            # Validate filename
+            is_valid, validation_message = SecurityValidator.validate_filename(filename)
+            if not is_valid:
+                return {
+                    "success": False,
+                    "error": f"Invalid filename: {validation_message}",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Validate content
+            is_valid, validation_message = SecurityValidator.validate_file_content(content)
+            if not is_valid:
+                return {
+                    "success": False,
+                    "error": f"Invalid content: {validation_message}",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Get file path
+            current_dir = self.session_manager.get_current_directory(session_id)
+            file_path = Path(current_dir) / filename
+            
+            # Check if file already exists
+            if file_path.exists():
+                return {
+                    "success": False,
+                    "error": "File already exists",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }
+            
+            # Create file
+            file_path.write_text(content, encoding='utf-8')
+            
+            # Update session stats
+            session = self.session_manager.get_session(session_id)
+            if session:
+                session["files_created"] += 1
+            
+            logger.info(f"File created - Session: {session_id}, File: {filename}, Size: {len(content)}")
+            
+            return {
+                "success": True,
+                "filename": filename,
+                "size": len(content.encode('utf-8')),
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+        
+        except Exception as e:
+            logger.error(f"File creation error - Session: {session_id}, File: {filename}, Error: {e}")
+            return {
+                "success": False,
+                "error": f"File creation failed: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+
+# System monitoring
+class SystemMonitor:
+    """Monitor system resources and health"""
+    
+    @staticmethod
+    def get_system_info() -> Dict[str, Any]:
+        """Get comprehensive system information"""
+        try:
+            # System info
+            info = {
+                "system": {
+                    "platform": sys.platform,
+                    "python_version": sys.version,
+                    "hostname": os.uname().nodename if hasattr(os, 'uname') else "unknown",
+                    "uptime": time.time() - psutil.boot_time()
+                },
+                "resources": {
+                    "cpu_count": psutil.cpu_count(),
+                    "cpu_usage": psutil.cpu_percent(interval=1),
+                    "memory": {
+                        "total": psutil.virtual_memory().total,
+                        "available": psutil.virtual_memory().available,
+                        "used": psutil.virtual_memory().used,
+                        "percent": psutil.virtual_memory().percent
+                    },
+                    "disk": {
+                        "total": psutil.disk_usage('/').total,
+                        "used": psutil.disk_usage('/').used,
+                        "free": psutil.disk_usage('/').free,
+                        "percent": (psutil.disk_usage('/').used / psutil.disk_usage('/').total) * 100
+                    }
+                },
+                "environment": {
+                    "workspace": config.WORKSPACE_PATH,
+                    "user": os.getenv("USER", "unknown"),
+                    "python_path": sys.executable,
+                    "cwd": os.getcwd()
+                },
+                "sandbox_features": {
+                    "security": "Command validation and sandboxing",
+                    "resource_limits": f"Max {config.MAX_MEMORY_MB}MB RAM, {config.MAX_DISK_GB}GB disk",
+                    "timeout": f"{config.COMMAND_TIMEOUT}s command timeout",
+                    "restricted_commands": len(config.RESTRICTED_COMMANDS)
+                },
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+            
+            return {
+                "success": True,
+                "info": info
+            }
+        
+        except Exception as e:
+            logger.error(f"System info error: {e}")
+            return {
+                "success": False,
+                "error": str(e),
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+
+# Initialize managers
+session_manager = SessionManager()
+executor = SecureExecutor(session_manager)
+file_manager = FileManager(session_manager)
+system_monitor = SystemMonitor()
+
+# Create default session
+default_session_id = session_manager.create_session()
+
+# API endpoints for AI integration
+class APIEndpoints:
+    """REST API endpoints for AI model integration"""
+    
+    @staticmethod
+    def execute_command(command: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+        """Execute a command via API"""
+        if not session_id:
+            session_id = default_session_id
+        
+        return executor.execute_command(command, session_id)
+    
+    @staticmethod
+    def create_file(filename: str, content: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+        """Create a file via API"""
+        if not session_id:
+            session_id = default_session_id
+        
+        return file_manager.create_file(filename, content, session_id)
+    
+    @staticmethod
+    def read_file(filename: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+        """Read a file via API"""
+        if not session_id:
+            session_id = default_session_id
+        
+        return file_manager.read_file(filename, session_id)
+    
+    @staticmethod
+    def list_directory(path: str = ".", session_id: Optional[str] = None) -> Dict[str, Any]:
+        """List directory via API"""
+        if not session_id:
+            session_id = default_session_id
+        
+        return file_manager.list_directory(path, session_id)
+    
+    @staticmethod
+    def get_system_info() -> Dict[str, Any]:
+        """Get system information via API"""
+        return system_monitor.get_system_info()
+    
+    @staticmethod
+    def create_session() -> Dict[str, Any]:
+        """Create a new session"""
+        session_id = session_manager.create_session()
+        return {
+            "success": True,
+            "session_id": session_id,
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+    
+    @staticmethod
+    def get_session_info(session_id: str) -> Dict[str, Any]:
+        """Get session information"""
+        session = session_manager.get_session(session_id)
+        if session:
+            return {
+                "success": True,
+                "session": session,
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+        else:
+            return {
+                "success": False,
+                "error": "Session not found",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }
+
+# Gradio Interface
+def create_interface():
+    """Create the main Gradio interface"""
+    
+    # Custom CSS for professional styling
+    css = """
+    .ubuntu-terminal {
+        background-color: #1e1e1e;
+        color: #d4d4d4;
+        font-family: 'Courier New', monospace;
+        border-radius: 8px;
+        padding: 20px;
+        min-height: 500px;
+        font-size: 14px;
+        line-height: 1.4;
+    }
+    
+    .header {
+        text-align: center;
+        padding: 25px;
+        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+        color: white;
+        border-radius: 12px;
+        margin-bottom: 25px;
+        box-shadow: 0 4px 15px rgba(0,0,0,0.2);
+    }
+    
+    .feature-box {
+        background: #f8f9fa;
+        border: 1px solid #e9ecef;
+        border-radius: 8px;
+        padding: 15px;
+        margin: 10px 0;
+        transition: all 0.3s ease;
+    }
+    
+    .feature-box:hover {
+        border-color: #667eea;
+        box-shadow: 0 2px 8px rgba(102, 126, 234, 0.15);
+    }
+    
+    .api-info {
+        background: #e3f2fd;
+        border: 1px solid #2196f3;
+        border-radius: 8px;
+        padding: 20px;
+        margin: 15px 0;
+    }
+    
+    .status-indicator {
+        display: inline-block;
+        width: 10px;
+        height: 10px;
+        border-radius: 50%;
+        margin-right: 8px;
+    }
+    
+    .status-active { background-color: #4caf50; }
+    .status-warning { background-color: #ff9800; }
+    .status-error { background-color: #f44336; }
+    """
+    
+    with gr.Blocks(css=css, title="Ubuntu Sandbox Environment v2.0", theme=gr.themes.Soft()) as app:
+        
+        # Header
+        gr.HTML("""
+        <div class="header">
+            <h1>🖥️ Ubuntu Sandbox Environment v2.0</h1>
+            <p><strong>Production-Grade, Secure, AI-Accessible Development Environment</strong></p>
+            <p>Perfect for AI models to build, ship, test, and deploy applications with complete security</p>
+        </div>
+        """)
+        
+        # Status bar
+        with gr.Row():
+            with gr.Column(scale=4):
+                gr.HTML('<div id="status-bar"><span class="status-indicator status-active"></span>System Online</div>')
+            with gr.Column(scale=1):
+                session_info = gr.Textbox(
+                    label="Session ID",
+                    value=default_session_id,
+                    interactive=False,
+                    elem_id="session-info"
+                )
+        
+        # Main terminal interface
+        with gr.Tab("💻 Terminal"):
+            with gr.Row():
+                with gr.Column(scale=3):
+                    command_input = gr.Textbox(
+                        label="Execute Command",
+                        placeholder="Enter any Ubuntu command (e.g., ls, pwd, python3 --version, git clone, docker build...)",
+                        lines=1,
+                        elem_id="command-input"
+                    )
+                    execute_btn = gr.Button("🚀 Execute", variant="primary", size="lg")
+                    
+                    terminal_output = gr.Textbox(
+                        label="Terminal Output",
+                        lines=25,
+                        max_lines=1000,
+                        info="Real-time output from executed commands",
+                        elem_classes=["ubuntu-terminal"],
+                        elem_id="terminal-output"
+                    )
+                    
+                    with gr.Row():
+                        clear_btn = gr.Button("🗑️ Clear Output", variant="secondary")
+                        new_session_btn = gr.Button("🆕 New Session", variant="secondary")
+                        help_btn = gr.Button("❓ Help", variant="secondary")
+        
+        # File management
+        with gr.Tab("📁 Files"):
+            with gr.Row():
+                with gr.Column(scale=1):
+                    gr.HTML("<h3>📄 File Operations</h3>")
+                    
+                    with gr.Tab("Create File"):
+                        create_filename = gr.Textbox(label="Filename", placeholder="example.py", elem_id="create-filename")
+                        create_content = gr.Textbox(
+                            label="Content", 
+                            lines=10, 
+                            placeholder="print('Hello from Ubuntu Sandbox!')",
+                            elem_id="create-content"
+                        )
+                        create_btn = gr.Button("📝 Create File", variant="primary")
+                        create_output = gr.Textbox(label="Result", lines=5, elem_id="create-output")
+                    
+                    with gr.Tab("Read File"):
+                        read_filename = gr.Textbox(label="Filename", placeholder="example.py", elem_id="read-filename")
+                        read_btn = gr.Button("📖 Read File", variant="primary")
+                        read_output = gr.Textbox(label="Content", lines=15, elem_id="read-output")
+                
+                with gr.Column(scale=1):
+                    gr.HTML("<h3>📂 Directory Browser</h3>")
+                    current_path = gr.Textbox(
+                        label="Current Path", 
+                        value="/workspace", 
+                        interactive=False, 
+                        elem_id="current-path"
+                    )
+                    refresh_btn = gr.Button("🔄 Refresh", variant="secondary")
+                    
+                    files_dataframe = gr.DataFrame(
+                        headers=["Name", "Type", "Size", "Modified", "Permissions"],
+                        datatype=["str", "str", "number", "str", "str"],
+                        label="Directory Contents",
+                        elem_id="files-df"
+                    )
+                    
+                    path_input = gr.Textbox(
+                        label="Navigate to Path", 
+                        placeholder="/workspace/your-folder",
+                        elem_id="path-input"
+                    )
+                    navigate_btn = gr.Button("🗂️ Navigate", variant="secondary")
+        
+        # System monitoring
+        with gr.Tab("📊 System"):
+            with gr.Row():
+                with gr.Column(scale=1):
+                    gr.HTML("<h3>🖥️ System Information</h3>")
+                    info_btn = gr.Button("📈 Update Info", variant="primary")
+                    system_json = gr.JSON(label="System Details", elem_id="system-json")
+                
+                with gr.Column(scale=1):
+                    gr.HTML("<h3>⚡ Quick Actions</h3>")
+                    
+                    quick_commands = [
+                        ("🐍 Python Version", "python3 --version"),
+                        ("📦 Node.js Version", "node --version"),
+                        ("🐙 Git Version", "git --version"),
+                        ("💾 Memory Info", "free -h"),
+                        ("💿 Disk Usage", "df -h"),
+                        ("🖥️ CPU Info", "lscpu | head -5"),
+                        ("🔍 Network Test", "ping -c 1 8.8.8.8"),
+                        ("📋 Process List", "ps aux --sort=-%mem | head -10")
+                    ]
+                    
+                    for label, command in quick_commands:
+                        gr.Button(
+                            label,
+                            variant="secondary"
+                        ).click(
+                            fn=lambda cmd=command: executor.execute_command(cmd, default_session_id)["output"],
+                            inputs=None,
+                            outputs=terminal_output
+                        )
+        
+        # API documentation
+        with gr.Tab("🔌 API"):
+            gr.HTML("""
+            <div class="api-info">
+                <h3>🤖 AI Model Integration API</h3>
+                <p>All endpoints support JSON requests and return JSON responses with timestamps and status information.</p>
+                
+                <h4>Base URL: <code>/api/v1/</code></h4>
+                
+                <h5>Execute Command</h5>
+                <p><strong>POST</strong> <code>/api/v1/execute</code></p>
+                <pre><code>{
+  "command": "ls -la",
+  "session_id": "optional-session-id"
+}</code></pre>
+                
+                <h5>Create File</h5>
+                <p><strong>POST</strong> <code>/api/v1/create-file</code></p>
+                <pre><code>{
+  "filename": "test.py",
+  "content": "print('Hello World!')",
+  "session_id": "optional-session-id"
+}</code></pre>
+                
+                <h5>Read File</h5>
+                <p><strong>POST</strong> <code>/api/v1/read-file</code></p>
+                <pre><code>{
+  "filename": "test.py",
+  "session_id": "optional-session-id"
+}</code></pre>
+                
+                <h5>List Directory</h5>
+                <p><strong>POST</strong> <code>/api/v1/list-directory</code></p>
+                <pre><code>{
+  "path": "/workspace",
+  "session_id": "optional-session-id"
+}</code></pre>
+                
+                <h5>Get System Info</h5>
+                <p><strong>GET</strong> <code>/api/v1/system-info</code></p>
+                
+                <h5>Create Session</h5>
+                <p><strong>POST</strong> <code>/api/v1/create-session</code></p>
+                
+                <h4>🔒 Security Features</h4>
+                <ul>
+                    <li>Command validation and sanitization</li>
+                    <li>Resource limits (memory, CPU, timeout)</li>
+                    <li>Restricted command list</li>
+                    <li>Session management with timeouts</li>
+                    <li>File system security checks</li>
+                    <li>Comprehensive logging</li>
+                </ul>
+            </div>
+            """)
+        
+        # Features showcase
+        with gr.Tab("✨ Features"):
+            gr.HTML("""
+            <div class="feature-box">
+                <h3>🛡️ Enterprise Security</h3>
+                <ul>
+                    <li>Command validation and sandboxing</li>
+                    <li>Resource limits and monitoring</li>
+                    <li>Session isolation and timeout</li>
+                    <li>Restricted command list</li>
+                    <li>Comprehensive audit logging</li>
+                </ul>
+            </div>
+            
+            <div class="feature-box">
+                <h3>🤖 AI-First Design</h3>
+                <ul>
+                    <li>RESTful API for programmatic access</li>
+                    <li>Session management for stateful interactions</li>
+                    <li>Real-time command execution</li>
+                    <li>Structured JSON responses</li>
+                    <li>Error handling and validation</li>
+                </ul>
+            </div>
+            
+            <div class="feature-box">
+                <h3>🛠️ Development Ready</h3>
+                <ul>
+                    <li>Pre-installed development tools</li>
+                    <li>Git version control</li>
+                    <li>Docker and container support</li>
+                    <li>Multiple programming languages</li>
+                    <li>Cloud CLI tools</li>
+                </ul>
+            </div>
+            
+            <div class="feature-box">
+                <h3>📊 Monitoring & Observability</h3>
+                <ul>
+                    <li>Real-time system monitoring</li>
+                    <li>Resource usage tracking</li>
+                    <li>Command execution logging</li>
+                    <li>Session analytics</li>
+                    <li>Health check endpoints</li>
+                </ul>
+            </div>
+            """)
+        
+        # Event handlers
+        def execute_command_ui(command):
+            if not command or not command.strip():
+                return "Please enter a command"
+            
+            result = executor.execute_command(command.strip(), default_session_id)
+            if result["success"]:
+                return f"✅ Command executed successfully\n\n{result['output']}"
+            else:
+                return f"❌ Command failed: {result['error']}\n\nOutput: {result.get('output', 'No output')}"
+        
+        def create_file_ui(filename, content):
+            if not filename or not content:
+                return "Please provide both filename and content"
+            
+            result = file_manager.create_file(filename, content, default_session_id)
+            if result["success"]:
+                return f"✅ File created successfully: {filename} ({result['size']} bytes)"
+            else:
+                return f"❌ File creation failed: {result['error']}"
+        
+        def read_file_ui(filename):
+            if not filename:
+                return "Please provide a filename"
+            
+            result = file_manager.read_file(filename, default_session_id)
+            if result["success"]:
+                return f"📄 {filename} ({result['size']} bytes):\n\n{result['content']}"
+            else:
+                return f"❌ File read failed: {result['error']}"
+        
+        def list_directory_ui(path="/workspace"):
+            result = file_manager.list_directory(path, default_session_id)
+            if result["success"]:
+                files = result["files"]
+                if files:
+                    data = [[f["name"], f["type"], f["size"], f["modified"], f["permissions"]] for f in files]
+                    return data, result["path"]
+                else:
+                    return [], result["path"]
+            else:
+                return [], path
+        
+        def show_help():
+            return get_help_text()
+        
+        def clear_output():
+            return ""
+        
+        def new_session_ui():
+            new_id = session_manager.create_session()
+            return new_id
+        
+        # Connect event handlers
+        execute_btn.click(
+            fn=execute_command_ui,
+            inputs=command_input,
+            outputs=terminal_output
+        )
+        
+        command_input.submit(
+            fn=execute_command_ui,
+            inputs=command_input,
+            outputs=terminal_output
+        )
+        
+        create_btn.click(
+            fn=create_file_ui,
+            inputs=[create_filename, create_content],
+            outputs=create_output
+        )
+        
+        read_btn.click(
+            fn=read_file_ui,
+            inputs=read_filename,
+            outputs=read_output
+        )
+        
+        refresh_btn.click(
+            fn=list_directory_ui,
+            inputs=None,
+            outputs=[files_dataframe, current_path]
+        )
+        
+        navigate_btn.click(
+            fn=list_directory_ui,
+            inputs=path_input,
+            outputs=[files_dataframe, current_path]
+        )
+        
+        info_btn.click(
+            fn=lambda: system_monitor.get_system_info()["info"],
+            inputs=None,
+            outputs=system_json
+        )
+        
+        help_btn.click(
+            fn=show_help,
+            inputs=None,
+            outputs=terminal_output
+        )
+        
+        clear_btn.click(
+            fn=clear_output,
+            inputs=None,
+            outputs=terminal_output
+        )
+        
+        new_session_btn.click(
+            fn=new_session_ui,
+            inputs=None,
+            outputs=session_info
+        )
+        
+        # Initialize with system info
+        system_json.update(
+            system_monitor.get_system_info()["info"]
+        )
+        
+        # Initialize file list
+        files_dataframe.update(
+            list_directory_ui()[0]
+        )
+    
+    return app
+
+def get_help_text():
+    """Get comprehensive help text"""
+    return """
+🔧 Ubuntu Sandbox Environment v2.0 - Help
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+📋 Available Commands:
+  • help         - Show this help text
+  • pwd          - Show current working directory
+  • ls           - List files and directories
+  • cd <dir>     - Change directory
+  • cat <file>   - Display file content
+  • touch <file> - Create empty file
+  • history      - Show command history
+  • new_session  - Start new terminal session
+
+🤖 AI Model Integration:
+  • Use REST API endpoints for programmatic access
+  • All commands are validated and sandboxed
+  • Session management for stateful interactions
+  • Structured JSON responses with error handling
+  • Rate limiting and resource monitoring
+
+🛠️ Development Tools:
+  • Python 3.x with scientific libraries
+  • Node.js and npm for JavaScript development
+  • Git for version control
+  • Docker for containerization
+  • Cloud CLI tools (AWS, GCP, Azure)
+  • Text editors (vim, nano)
+
+🛡️ Security Features:
+  • Command validation and sanitization
+  • Resource limits (CPU, memory, disk)
+  • 30-second command timeout
+  • Restricted command list
+  • Session isolation and cleanup
+  • Comprehensive audit logging
+
+📊 System Monitoring:
+  • Real-time resource usage
+  • Command execution statistics
+  • Session management
+  • Health monitoring
+  • Performance metrics
+
+💡 Tips:
+  • Use 'system_info' command for detailed system information
+  • All file operations are validated for security
+  • Sessions automatically clean up after inactivity
+  • Check logs in /workspace/logs/ for debugging
+  • Use the API for AI model integration
+
+━━━━━��━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Ready for AI agents to build, ship, and create anything! 🚀
+"""
+
+# Custom Flask routes
+def create_custom_routes(app):
+    """Add custom API routes to the Gradio app"""
+    
+    from flask import Flask, request, jsonify
+    
+    @app.route("/api/v1/execute", methods=["POST"])
+    def execute_endpoint():
+        """Execute command API endpoint"""
+        try:
+            data = request.get_json()
+            if not data or "command" not in data:
+                return jsonify({
+                    "success": False,
+                    "error": "Missing 'command' in request body",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }), 400
+            
+            command = data["command"]
+            session_id = data.get("session_id", default_session_id)
+            
+            result = executor.execute_command(command, session_id)
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API execute error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/api/v1/create-file", methods=["POST"])
+    def create_file_endpoint():
+        """Create file API endpoint"""
+        try:
+            data = request.get_json()
+            if not data or "filename" not in data or "content" not in data:
+                return jsonify({
+                    "success": False,
+                    "error": "Missing 'filename' or 'content' in request body",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }), 400
+            
+            filename = data["filename"]
+            content = data["content"]
+            session_id = data.get("session_id", default_session_id)
+            
+            result = file_manager.create_file(filename, content, session_id)
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API create file error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/api/v1/read-file", methods=["POST"])
+    def read_file_endpoint():
+        """Read file API endpoint"""
+        try:
+            data = request.get_json()
+            if not data or "filename" not in data:
+                return jsonify({
+                    "success": False,
+                    "error": "Missing 'filename' in request body",
+                    "timestamp": datetime.now(timezone.utc).isoformat()
+                }), 400
+            
+            filename = data["filename"]
+            session_id = data.get("session_id", default_session_id)
+            
+            result = file_manager.read_file(filename, session_id)
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API read file error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/api/v1/list-directory", methods=["POST"])
+    def list_directory_endpoint():
+        """List directory API endpoint"""
+        try:
+            data = request.get_json() or {}
+            path = data.get("path", "/workspace")
+            session_id = data.get("session_id", default_session_id)
+            
+            result = file_manager.list_directory(path, session_id)
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API list directory error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/api/v1/system-info", methods=["GET"])
+    def system_info_endpoint():
+        """System info API endpoint"""
+        try:
+            result = system_monitor.get_system_info()
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API system info error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/api/v1/create-session", methods=["POST"])
+    def create_session_endpoint():
+        """Create session API endpoint"""
+        try:
+            result = APIEndpoints.create_session()
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API create session error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/api/v1/session-info/<session_id>", methods=["GET"])
+    def session_info_endpoint(session_id):
+        """Get session info API endpoint"""
+        try:
+            result = APIEndpoints.get_session_info(session_id)
+            return jsonify(result)
+        
+        except Exception as e:
+            logger.error(f"API session info error: {e}")
+            return jsonify({
+                "success": False,
+                "error": f"Internal server error: {str(e)}",
+                "timestamp": datetime.now(timezone.utc).isoformat()
+            }), 500
+    
+    @app.route("/health", methods=["GET"])
+    def health_endpoint():
+        """Health check endpoint"""
+        return jsonify({
+            "status": "healthy",
+            "version": __version__,
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        })
+
+# Cleanup function
+def cleanup_on_exit():
+    """Cleanup on application exit"""
+    logger.info("Starting application cleanup...")
+    
+    # Kill any running processes
+    for process_id in list(executor.running_processes.keys()):
+        executor.kill_process(process_id)
+    
+    # Cleanup expired sessions
+    session_manager.cleanup_expired_sessions()
+    
+    logger.info("Application cleanup completed")
+
+# Main application
+if __name__ == "__main__":
+    try:
+        # Create workspace structure
+        os.makedirs(config.WORKSPACE_PATH, exist_ok=True)
+        os.makedirs(config.LOGS_PATH, exist_ok=True)
+        
+        # Setup signal handlers
+        def signal_handler(sig, frame):
+            logger.info("Received shutdown signal")
+            cleanup_on_exit()
+            sys.exit(0)
+        
+        signal.signal(signal.SIGINT, signal_handler)
+        signal.signal(signal.SIGTERM, signal_handler)
+        
+        # Log startup
+        logger.info(f"Starting Ubuntu Sandbox Environment v{__version__}")
+        logger.info(f"Workspace: {config.WORKSPACE_PATH}")
+        logger.info(f"Default session: {default_session_id}")
+        
+        # Create the interface
+        app = create_interface()
+        
+        # Add custom API routes
+        create_custom_routes(app)
+        
+        # Launch the app
+        logger.info("Launching Gradio application...")
+        app.launch(
+            server_name="0.0.0.0",
+            server_port=7860,
+            share=True,
+            debug=False,  # Set to False for production
+            show_error=True,
+            favicon_path=None,
+            height=800,
+            title="Ubuntu Sandbox Environment v2.0"
+        )
+    
+    except Exception as e:
+        logger.error(f"Application startup error: {e}")
+        sys.exit(1)
+    
+    finally:
+        cleanup_on_exit()