app.py

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}")