Spaces:

Srevarshan1502
/

collaborative-python-executor

Paused

added all files

6521e91 about 1 year ago

15.4 kB

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

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