added all files

Dockerfile

@@ -0,0 +1,42 @@
+# Use official Python runtime as base image
+FROM python:3.9-slim
+
+# Set working directory in container
+WORKDIR /app
+
+# Set environment variables
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV GRADIO_SERVER_NAME=0.0.0.0
+ENV GRADIO_SERVER_PORT=7860
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    gcc \
+    g++ \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements first for better Docker layer caching
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY app.py .
+
+# Create non-root user for security
+RUN useradd --create-home --shell /bin/bash app && \
+    chown -R app:app /app
+USER app
+
+# Expose the port that Gradio will run on
+EXPOSE 7860
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
+    CMD curl -f http://localhost:7860/health || exit 1
+
+# Run the application
+CMD ["python", "app.py"]

app.py

@@ -0,0 +1,442 @@
+import gradio as gr
+import sys
+import io
+import traceback
+import ast
+import time
+import json
+import os
+from contextlib import redirect_stdout, redirect_stderr
+from typing import Dict, Any
+import threading
+import signal
+
+# Set up proper signal handling for Docker
+def signal_handler(signum, frame):
+    print(f"Received signal {signum}, shutting down gracefully...")
+    sys.exit(0)
+
+signal.signal(signal.SIGTERM, signal_handler)
+signal.signal(signal.SIGINT, signal_handler)
+
+class PythonSyntaxFSM:
+    """Finite State Machine for Python syntax validation"""
+    
+    def __init__(self):
+        self.reset()
+        
+    def reset(self):
+        self.state = 'normal'
+        self.bracket_stack = []
+        self.in_string = False
+        self.string_delimiter = None
+        self.line_number = 0
+        self.errors = []
+        self.warnings = []
+        
+    def validate_code(self, code: str) -> Dict[str, Any]:
+        """Validate Python code using FSM approach"""
+        self.reset()
+        
+        if not code or not code.strip():
+            return {
+                'valid': False,
+                'errors': [{'line': 0, 'message': 'Empty code input', 'type': 'input_error'}],
+                'warnings': [],
+                'total_issues': 1
+            }
+        
+        lines = code.split('\n')
+        
+        for i, line in enumerate(lines):
+            self.line_number = i + 1
+            self.validate_line(line)
+            
+        # Check for unclosed brackets at end
+        if self.bracket_stack:
+            self.errors.append({
+                'line': self.line_number,
+                'type': 'syntax_error',
+                'message': f'Unclosed bracket: {self.bracket_stack[-1]}',
+                'severity': 'error'
+            })
+            
+        return {
+            'valid': len(self.errors) == 0,
+            'errors': self.errors,
+            'warnings': self.warnings,
+            'total_issues': len(self.errors) + len(self.warnings)
+        }
+    
+    def validate_line(self, line: str):
+        """Validate a single line using FSM logic"""
+        if not line.strip():
+            return
+            
+        # Check indentation
+        leading_spaces = len(line) - len(line.lstrip())
+        if leading_spaces % 4 != 0 and line.strip():
+            self.warnings.append({
+                'line': self.line_number,
+                'type': 'style_warning',
+                'message': 'Inconsistent indentation (PEP 8 recommends 4 spaces)',
+                'severity': 'warning'
+            })
+        
+        # Process character by character for brackets and strings
+        i = 0
+        while i < len(line):
+            char = line[i]
+            
+            # Handle string states
+            if self.in_string:
+                if char == self.string_delimiter and (i == 0 or line[i-1] != '\\'):
+                    self.in_string = False
+                    self.string_delimiter = None
+                i += 1
+                continue
+                
+            # Handle comment detection
+            if char == '#':
+                break   # Rest of line is comment
+                
+            # Handle string detection
+            if char in ['"', "'"]:
+                # Check for triple quotes
+                if i + 2 < len(line) and line[i:i+3] == char * 3:
+                    self.string_delimiter = char * 3
+                    i += 3
+                else:
+                    self.string_delimiter = char
+                    i += 1
+                self.in_string = True
+                continue
+                
+            # Handle brackets
+            if char in '([{':
+                self.bracket_stack.append(char)
+            elif char in ')]}':
+                if not self.bracket_stack:
+                    self.errors.append({
+                        'line': self.line_number,
+                        'type': 'syntax_error',
+                        'message': f'Unmatched closing bracket: {char}',
+                        'severity': 'error'
+                    })
+                else:
+                    expected = {'(': ')', '[': ']', '{': '}'}
+                    last_open = self.bracket_stack[-1]
+                    if expected[last_open] != char:
+                        self.errors.append({
+                            'line': self.line_number,
+                            'type': 'syntax_error',
+                            'message': f'Mismatched brackets: expected {expected[last_open]}, got {char}',
+                            'severity': 'error'
+                        })
+                    else:
+                        self.bracket_stack.pop()
+                        
+            i += 1
+        
+        # Check Python syntax patterns
+        stripped = line.strip()
+        control_keywords = ['def ', 'class ', 'if ', 'elif ', 'for ', 'while ', 'try:', 'except']
+        
+        for keyword in control_keywords:
+            if stripped.startswith(keyword):
+                if not stripped.endswith(':') and keyword != 'except':
+                    self.errors.append({
+                        'line': self.line_number,
+                        'type': 'syntax_error',
+                        'message': f'{keyword.strip()} statement must end with colon',
+                        'severity': 'error'
+                    })
+                break
+
+
+class SecurePythonExecutor:
+    """Secure Python code executor with sandboxing and timeout"""
+    
+    def __init__(self, timeout: int = 10):
+        self.timeout = timeout
+        self.restricted_modules = {
+            'os', 'sys', 'subprocess', 'socket', 'urllib', 'requests',
+            'shutil', 'glob', 'pickle', 'marshal', 'shelve', 'dbm',
+            'sqlite3', 'threading', 'multiprocessing', 'ctypes', 'importlib',
+            'builtins', '__builtin__', 'imp', 'zipimport'
+        }
+        
+        # Create safe builtins
+        self.safe_builtins = {
+            'print': print,
+            'len': len,
+            'range': range,
+            'str': str,
+            'int': int,
+            'float': float,
+            'bool': bool,
+            'list': list,
+            'dict': dict,
+            'tuple': tuple,
+            'set': set,
+            'frozenset': frozenset,
+            'abs': abs,
+            'max': max,
+            'min': min,
+            'sum': sum,
+            'sorted': sorted,
+            'reversed': reversed,
+            'enumerate': enumerate,
+            'zip': zip,
+            'map': map,
+            'filter': filter,
+            'all': all,
+            'any': any,
+            'type': type,
+            'isinstance': isinstance,
+            'hasattr': hasattr,
+            'getattr': getattr,
+            'setattr': setattr,
+            'delattr': delattr,
+            'round': round,
+            'pow': pow,
+            'divmod': divmod,
+            'chr': chr,
+            'ord': ord,
+            'hex': hex,
+            'oct': oct,
+            'bin': bin,
+            'format': format,
+            'repr': repr,
+            'ascii': ascii,
+            'iter': iter,
+            'next': next,
+            'slice': slice,
+            'callable': callable,
+            'id': id,
+            'hash': hash,
+            'vars': vars,
+            'dir': dir,
+            'help': help,
+            'Exception': Exception,
+            'ValueError': ValueError,
+            'TypeError': TypeError,
+            'IndexError': IndexError,
+            'KeyError': KeyError,
+            'AttributeError': AttributeError,
+            'NameError': NameError,
+            'ZeroDivisionError': ZeroDivisionError,
+        }
+        
+    def execute_code(self, code: str) -> Dict[str, Any]:
+        """Execute Python code in a secure environment"""
+        start_time = time.time()
+        
+        # Input validation
+        if not code or not code.strip():
+            return {
+                'success': False,
+                'output': '',
+                'error': 'No code provided',
+                'execution_time': 0,
+                'validation': {
+                    'valid': False,
+                    'errors': [{'message': 'Empty code input', 'line': 0}],
+                    'warnings': [],
+                    'total_issues': 1
+                }
+            }
+        
+        # Pre-execution validation
+        fsm = PythonSyntaxFSM()
+        validation_result = fsm.validate_code(code)
+        
+        # Check for restricted imports
+        restricted_check = self.check_restricted_imports(code)
+        if not restricted_check['allowed']:
+            return {
+                'success': False,
+                'output': '',
+                'error': f"Security violation: Restricted module '{restricted_check['module']}' is not allowed",
+                'execution_time': 0,
+                'validation': validation_result
+            }
+        
+        # Execute code with timeout
+        result = self.run_with_timeout(code)
+        execution_time = time.time() - start_time
+        
+        return {
+            'success': result['success'],
+            'output': result['output'],
+            'error': result['error'],
+            'execution_time': round(execution_time, 3),
+            'validation': validation_result
+        }
+    
+    def check_restricted_imports(self, code: str) -> Dict[str, Any]:
+        """Check for restricted module imports"""
+        try:
+            tree = ast.parse(code)
+            for node in ast.walk(tree):
+                if isinstance(node, ast.Import):
+                    for alias in node.names:
+                        if alias.name.split('.')[0] in self.restricted_modules:
+                            return {'allowed': False, 'module': alias.name}
+                elif isinstance(node, ast.ImportFrom):
+                    if node.module and node.module.split('.')[0] in self.restricted_modules:
+                        return {'allowed': False, 'module': node.module}
+            return {'allowed': True, 'module': None}
+        except Exception:
+            return {'allowed': True, 'module': None}   # If parsing fails, let execution handle it
+    
+    def run_with_timeout(self, code: str) -> Dict[str, Any]:
+        """Run code with timeout protection"""
+        output_buffer = io.StringIO()
+        error_buffer = io.StringIO()
+        result = {'success': False, 'output': '', 'error': ''}
+        
+        def target():
+            try:
+                with redirect_stdout(output_buffer), redirect_stderr(error_buffer):
+                    # Create restricted execution environment
+                    restricted_globals = {
+                        '__builtins__': self.safe_builtins,
+                        '__name__': '__main__',
+                        '__doc__': None,
+                    }
+                    
+                    # Execute the code
+                    exec(code, restricted_globals)
+                    result['success'] = True
+                    
+            except Exception as e:
+                error_msg = f"{type(e).__name__}: {str(e)}"
+                error_buffer.write(error_msg)
+                result['success'] = False
+        
+        # Run with timeout
+        thread = threading.Thread(target=target)
+        thread.daemon = True
+        thread.start()
+        thread.join(timeout=self.timeout)
+        
+        if thread.is_alive():
+            result['error'] = f"Code execution timed out after {self.timeout} seconds"
+            result['success'] = False
+        else:
+            result['output'] = output_buffer.getvalue()
+            if error_buffer.getvalue():
+                result['error'] = error_buffer.getvalue()
+                result['success'] = False
+            else:
+                result['success'] = True
+                
+        return result
+
+
+# Initialize the executor
+executor = SecurePythonExecutor(timeout=10)
+
+def execute_python_code_api(code: str) -> Dict[str, Any]:
+    """
+    API function to execute Python code.
+    This function is directly exposed as an API endpoint.
+    """
+    try:
+        result = executor.execute_code(code)
+        return result
+    except Exception as e:
+        return {
+            'success': False,
+            'output': '',
+            'error': f"Execution engine error: {str(e)}",
+            'execution_time': 0,
+            'validation': {
+                'valid': False,
+                'errors': [{'message': str(e), 'line': 0, 'type': 'engine_error'}],
+                'warnings': [],
+                'total_issues': 1
+            }
+        }
+
+def health_check_api() -> Dict[str, Any]:
+    """
+    Health check API endpoint.
+    """
+    try:
+        test_result = execute_python_code_api("print('Health check OK')")
+        return {
+            'status': 'healthy' if test_result['success'] else 'unhealthy',
+            'timestamp': time.time(),
+            'test_execution_success': test_result['success']
+        }
+    except Exception as e:
+        return {
+            'status': 'unhealthy',
+            'timestamp': time.time(),
+            'error': str(e),
+            'test_execution_success': False
+        }
+
+# Create the Gradio interface specifically for API exposure
+# We use gr.Interface for direct API endpoint mapping, not gr.Blocks for this scenario.
+# We also create a separate interface for the health check.
+
+# Main code execution API
+api_code_executor = gr.Interface(
+    fn=execute_python_code_api,
+    inputs=[
+        gr.Code(label="Python Code", language="python", lines=10, value="print('Hello from API!')")
+    ],
+    outputs=[
+        gr.JSON(label="Execution Result")
+    ],
+    # To enable direct API calls without relying on implicit UI interactions
+    api_name="predict",
+    title="Python Code Executor API (Main)",
+    description="Submits Python code for secure execution and returns the result."
+)
+
+# Health check API
+api_health_check = gr.Interface(
+    fn=health_check_api,
+    inputs=[],
+    outputs=[
+        gr.JSON(label="Health Status")
+    ],
+    # You can give this a different API name if you want a separate endpoint
+    api_name="health",
+    title="Service Health Check",
+    description="Checks the health and basic functionality of the API service."
+)
+
+
+if __name__ == "__main__":
+    print("Starting Python Code Executor API...")
+    print(f"Python version: {sys.version}")
+    print(f"Working directory: {os.getcwd()}")
+    
+    server_name = os.getenv('GRADIO_SERVER_NAME', '0.0.0.0')
+    server_port = int(os.getenv('GRADIO_SERVER_PORT', 7860))
+    
+    print(f"Starting server on {server_name}:{server_port}")
+    
+    try:
+        # Launch the main API interface
+        # Gradio will automatically create API endpoints for functions defined in gr.Interface
+        # The /api/predict endpoint will now directly map to execute_python_code_api
+        api_code_executor.launch(
+            server_name=server_name,
+            server_port=server_port,
+            share=False,
+            show_error=True,
+            quiet=False,
+            # show_api=True is default for Interface, and the API_NAME makes it explicit
+        )
+        # Note: If you launch multiple interfaces, they might conflict on the same port
+        # For simplicity for this API-only case, we'll focus on the primary endpoint.
+        # If you truly need both, you'd integrate FastAPI more deeply or use separate ports.
+        
+    except Exception as e:
+        print(f"Failed to start server: {e}")

requirements.txt

@@ -0,0 +1,4 @@
+gradio==4.44.0
+fastapi==0.104.1
+uvicorn==0.24.0
+pydantic==2.5.0