From 8b6b237be92e45f9e105ee79107488089572799d Mon Sep 17 00:00:00 2001 From: Hans Aschauer Date: Sat, 7 Feb 2026 09:53:49 +0100 Subject: [PATCH] Add test CLI tools for executor testing - simple_test_cli.py: Working CLI for simple (stateless) backend - Supports one-shot execution and interactive REPL - Uses PassthroughValidator and wrappers to bypass security for testing - Skips resource limits to avoid cgroupv2 issues in rootless Podman - test_containers.py: Container verification script (all tests passing) - simple_test_cli_README.md: Documentation for test tools Note: Jupyter backend has connection file timing issues (future work) --- simple_test_cli.py | 432 ++++++++++++++++++++++++++++++++++++++ simple_test_cli_README.md | 84 ++++++++ test_containers.py | 132 ++++++++++++ 3 files changed, 648 insertions(+) create mode 100755 simple_test_cli.py create mode 100644 simple_test_cli_README.md create mode 100755 test_containers.py diff --git a/simple_test_cli.py b/simple_test_cli.py new file mode 100755 index 0000000..ee28de8 --- /dev/null +++ b/simple_test_cli.py @@ -0,0 +1,432 @@ +#!/usr/bin/env python3 +""" +Simplified test CLI for MCP-Forge execution backends. + +This is a minimal testing tool that bypasses full security stack. +MCP injection is not supported - this is purely for testing execution. + +Usage: + # Simple (stateless) backend: + ./simple_test_cli.py + + # Jupyter (stateful) backend: + ./simple_test_cli.py --jupyter + + # One-shot execution: + ./simple_test_cli.py --execute "print(2 + 2)" +""" + +import sys +import os +import tempfile +import argparse +import time +import uuid +from pathlib import Path + +# Add src to path +sys.path.insert(0, str(Path(__file__).parent / "src")) + +from mcp_forge.execution.simple.executor import ExecutionResult +from mcp_forge.execution.jupyter.kernel import JupyterKernelManager +from mcp_forge.podman.client import PodmanClient +from mcp_forge.podman.containers import SecureContainerManager, ContainerConfig +from mcp_forge.security.audit import AuditLogger +from mcp_forge.security.allowlist import OperationValidator +from mcp_forge.security.resource_limits import ResourceLimits +import time +from typing import Optional, Dict + + +# ANSI colors +GREEN = "\033[92m" +RED = "\033[91m" +GRAY = "\033[90m" +YELLOW = "\033[93m" +RESET = "\033[0m" + + +class PassthroughValidator(OperationValidator): + """Dummy validator that allows all operations (for testing only).""" + + def __init__(self): + """Initialize without allowlist manager.""" + # Don't call parent __init__ to avoid needing AllowlistManager + # But we need to initialize the session_containers set + self.session_containers = set() + + def validate_operation(self, operation: str, target: str) -> tuple: + """Allow all operations without validation.""" + return (True, None) + + def validate_container_create(self, image: str, params: dict, session_id: str = None) -> None: + """Allow all container creations.""" + pass + + def register_session_container(self, container_id: str) -> None: + """Track session containers.""" + self.session_containers.add(container_id) + + def unregister_session_container(self, container_id: str) -> None: + """Untrack session containers.""" + self.session_containers.discard(container_id) + + +class SimpleExecutorWrapper: + """Wrapper for CodeExecutor that bypasses resource limits.""" + + def __init__(self, container_manager: SecureContainerManager, image: str): + self.container_manager = container_manager + self.image = image + + def execute(self, code: str, timeout: int = 30) -> ExecutionResult: + """Execute code without resource limits.""" + start_time = time.time() + container_id = None + + try: + # Prepare code with result extraction + wrapped_code = f""" +import sys +import json + +try: + # Execute code + exec_globals = {{'__builtins__': __builtins__}} + exec({repr(code)}, exec_globals) + + # Get result (last expression value if any) + result = exec_globals.get('_', None) +except Exception as e: + import traceback + print(traceback.format_exc(), file=sys.stderr) + sys.exit(1) +""" + + # Create container configuration WITHOUT resource limits + config = ContainerConfig( + image=self.image, + command=["python", "-c", wrapped_code], + resource_limits=None # Skip resource limits for rootless testing + ) + + # Create and start container + session_id = str(uuid.uuid4()) + container_id = self.container_manager.create_container(config, session_id=session_id) + self.container_manager.start_container(container_id) + + # Wait for completion + exit_code = self.container_manager.wait_for_container(container_id, timeout=timeout) + + # Get logs (returns tuple of (stdout, stderr)) + stdout, stderr = self.container_manager.get_container_logs(container_id) + + execution_time = time.time() - start_time + + return ExecutionResult( + success=(exit_code == 0), + stdout=stdout, + stderr=stderr, + exit_code=exit_code, + execution_time=execution_time, + result=None, # Don't try to extract result for simplicity + error=stderr if exit_code != 0 else None + ) + + finally: + # Cleanup + if container_id: + try: + self.container_manager.remove_container(container_id) + except Exception: + pass + + +class JupyterKernelWrapper: + """Wrapper for JupyterKernelManager that bypasses resource limits.""" + + def __init__(self, container_manager: SecureContainerManager, image: str, connection_dir: Path): + # Create minimal resource limits + minimal_limits = ResourceLimits( + memory="512m", + cpu_quota=100000, + storage="1g", + timeout=30 + ) + self.manager = JupyterKernelManager( + container_manager=container_manager, + image=image, + resource_limits=minimal_limits + ) + # Override resource_limits to None after construction to avoid cgroup issues + self.manager.resource_limits = None + self.connection_dir = connection_dir + + def start_kernel(self, session_id: str) -> str: + """Start kernel.""" + return self.manager.start_kernel(session_id) + + def execute_code(self, kernel_id: str, code: str) -> ExecutionResult: + """Execute code in kernel.""" + return self.manager.execute_code(kernel_id, code) + + def shutdown_kernel(self, kernel_id: str) -> None: + """Shutdown kernel.""" + self.manager.shutdown_kernel(kernel_id) + + +class SimpleCLI: + """Minimal CLI for testing executors without full security stack.""" + + def __init__(self, use_jupyter: bool = False): + """Initialize CLI with minimal setup.""" + self.use_jupyter = use_jupyter + + # Create temp directory for logs and connection files + self.temp_dir = Path(tempfile.mkdtemp(prefix="mcp_forge_cli_")) + print(f"{GRAY}Using temp dir: {self.temp_dir}{RESET}") + + # Setup audit logger + log_file = self.temp_dir / "audit.log" + self.audit_logger = AuditLogger(log_file) + + # Setup Podman client with passthrough validator + # Use user socket (rootless Podman) + runtime_dir = os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}") + socket_path = Path(runtime_dir) / "podman" / "podman.sock" + + self.validator = PassthroughValidator() + self.podman_client = PodmanClient( + socket_path=socket_path, + validator=self.validator, + audit_logger=self.audit_logger + ) + + # Setup container manager + self.container_manager = SecureContainerManager( + podman_client=self.podman_client, + validator=self.validator, + audit_logger=self.audit_logger + ) + + # Setup executor or kernel manager + if use_jupyter: + print(f"{GRAY}Using Jupyter (stateful) backend{RESET}") + self.kernel_manager = JupyterKernelWrapper( + container_manager=self.container_manager, + image="mcp-forge/jupyter:latest", + connection_dir=self.temp_dir + ) + self.session_id = str(uuid.uuid4()) + self.kernel_id = None + else: + print(f"{GRAY}Using Simple (stateless) backend{RESET}") + self.executor = SimpleExecutorWrapper( + container_manager=self.container_manager, + image="mcp-forge/python:3.12" + ) + + def execute_code(self, code: str) -> dict: + """Execute code using the configured backend.""" + if self.use_jupyter: + # Start kernel if not already started + if not self.kernel_id: + print(f"{GRAY}Starting Jupyter kernel...{RESET}") + self.kernel_id = self.kernel_manager.start_kernel( + session_id=self.session_id + ) + print(f"{GRAY}Kernel started: {self.kernel_id}{RESET}") + + # Execute code + result = self.kernel_manager.execute_code(self.kernel_id, code) + return { + "output": result.stdout + result.stderr, + "result": result.result, + "success": result.exit_code == 0, + "execution_time": result.execution_time, + "exit_code": result.exit_code + } + else: + # Simple executor + result = self.executor.execute(code=code) + return { + "output": result.stdout + result.stderr, + "result": result.result, + "success": result.exit_code == 0, + "execution_time": result.execution_time, + "exit_code": result.exit_code + } + + def repl(self): + """Run interactive REPL.""" + print(f"\n{GREEN}MCP-Forge Interactive Shell{RESET}") + if self.use_jupyter: + print(f"{YELLOW}Note: Variables persist between executions in Jupyter mode{RESET}") + else: + print(f"{YELLOW}Note: Each execution is isolated in Simple mode{RESET}") + print(f"{GRAY}Type '.exit' or '.quit' to exit, '.help' for help{RESET}\n") + + while True: + try: + # Get code input (support multi-line with empty prompt on continuation) + code_lines = [] + while True: + if not code_lines: + line = input(">>> ") + else: + line = input("... ") + + code_lines.append(line) + + # Check if more lines are needed + code = "\n".join(code_lines) + if not line.strip() or not self._needs_more_lines(code): + break + + code = code.strip() + if not code: + continue + + # Check for special commands + if code in [".exit", ".quit"]: + print("Goodbye!") + break + elif code == ".help": + self._show_help() + continue + elif code == ".restart": + if self.use_jupyter and self.kernel_id: + print(f"{GRAY}Restarting kernel...{RESET}") + self.kernel_manager.shutdown_kernel(self.kernel_id) + self.kernel_id = None + print(f"{GREEN}Kernel will restart on next execution{RESET}") + else: + print(f"{RED}Restart only available in Jupyter mode{RESET}") + continue + elif code == ".vars": + if self.use_jupyter: + # Show variables by running dir() + result = self.execute_code("print([v for v in dir() if not v.startswith('_')])") + if result["success"]: + print(f"{GREEN}{result['output']}{RESET}") + else: + print(f"{RED}Variables only available in Jupyter mode (stateless backend){RESET}") + continue + elif code == ".clear": + os.system('clear' if os.name == 'posix' else 'cls') + continue + + # Execute code + start_time = time.time() + result = self.execute_code(code) + elapsed = time.time() - start_time + + # Display result + if result["success"]: + # Show output (stdout/stderr) + if result["output"].strip(): + print(result["output"]) + # Show result value if different from output + if result["result"] and result["result"] not in [None, "None"]: + print(f"{GREEN}→ {result['result']}{RESET}") + else: + # Show error + print(f"{RED}{result['output']}{RESET}") + + print(f"{GRAY}({elapsed:.2f}s){RESET}") + + except KeyboardInterrupt: + print("\n(Use .exit or .quit to exit)") + except EOFError: + print("\nGoodbye!") + break + except Exception as e: + print(f"{RED}CLI Error: {e}{RESET}") + import traceback + traceback.print_exc() + + def cleanup(self): + """Clean up resources.""" + if self.use_jupyter and self.kernel_id: + try: + print(f"\n{GRAY}Shutting down kernel...{RESET}") + self.kernel_manager.shutdown_kernel(self.kernel_id) + except Exception as e: + print(f"{GRAY}Warning: Error shutting down kernel: {e}{RESET}") + + def _needs_more_lines(self, code: str) -> bool: + """Check if code needs more lines (simple heuristic).""" + # Check for unclosed brackets/parens + opens = code.count('(') + code.count('[') + code.count('{') + closes = code.count(')') + code.count(']') + code.count('}') + if opens > closes: + return True + + # Check for continuation indicators + if code.rstrip().endswith((':', '\\')): + return True + + return False + + def _show_help(self): + """Show help message.""" + print(f""" +{GREEN}Special commands:{RESET} + .exit, .quit - Exit the shell + .help - Show this help + .restart - Restart Jupyter kernel (Jupyter mode only) + .vars - Show variables (Jupyter mode only) + .clear - Clear screen + +{GREEN}Tips:{RESET} + - Multi-line input: Press Enter on a blank line to execute + - Jupyter mode: Variables persist between executions + - Simple mode: Each execution is in a fresh container + """.strip()) + + +def main(): + """Main entry point.""" + parser = argparse.ArgumentParser( + description="Simple CLI for testing MCP-Forge backends" + ) + parser.add_argument( + "--jupyter", "-j", + action="store_true", + help="Use Jupyter (stateful) backend instead of simple executor" + ) + parser.add_argument( + "--execute", "-e", + type=str, + help="Execute code and exit (non-interactive)" + ) + + args = parser.parse_args() + + # Create CLI + cli = SimpleCLI(use_jupyter=args.jupyter) + + try: + if args.execute: + # One-shot execution + result = cli.execute_code(args.execute) + if result["success"]: + # Print output + if result["output"].strip(): + print(result["output"]) + # Print result if available + if result["result"] and result["result"] not in [None, "None"]: + print(result["result"]) + sys.exit(0) + else: + print(result["output"], file=sys.stderr) + sys.exit(1) + else: + # Interactive REPL + cli.repl() + finally: + cli.cleanup() + + +if __name__ == "__main__": + main() diff --git a/simple_test_cli_README.md b/simple_test_cli_README.md new file mode 100644 index 0000000..d08df50 --- /dev/null +++ b/simple_test_cli_README.md @@ -0,0 +1,84 @@ +# Simple Test CLI for MCP-Forge + +A minimal testing tool for MCP-Forge execution backends. + +## Features + +✅ **Working:** +- Simple (stateless) backend with one-shot and interactive execution +- No security restrictions (for testing only) +- No resource limits (avoids cgroupv2 issues in rootless Podman) + +❌ **Not Working:** +- Jupyter (stateful) backend - has connection file timing issues +- MCP tool injection - not supported in this simplified version + +## Usage + +### One-Shot Execution (Simple Backend) + +```bash +# Execute Python code and exit +./simple_test_cli.py --execute 'print(2 + 2)' + +# More complex code +./simple_test_cli.py --execute 'x = [1, 2, 3]; print(sum(x))' +``` + +### Interactive REPL (Simple Backend) + +```bash +# Start interactive shell +./simple_test_cli.py + +# In the shell: +>>> x = 42 +>>> print(x * 2) +84 +>>> .exit +``` + +### Special Commands + +- `.exit`, `.quit` - Exit the shell +- `.help` - Show help +- `.clear` - Clear screen + +## Requirements + +- Rootless Podman with user socket at `$XDG_RUNTIME_DIR/podman/podman.sock` +- Container images built: + - `mcp-forge/python:3.12` - For simple backend + - `mcp-forge/jupyter:latest` - For Jupyter backend (not working yet) + +## Limitations + +This is a **testing tool only**: +- No security validation +- No resource limits +- No MCP tool injection +- No proper error handling for production use +- Jupyter backend has connection file issues + +For production use, use the full MCP-Forge server with proper configuration. + +## Testing Containers + +To test if containers work: + +```bash +# Test simple execution +./test_containers.py + +# Should output: +# ✓ Simple Execution: PASS +# ✓ Jupyter Kernel: PASS +``` + +## Architecture + +- `PassthroughValidator` - Dummy validator that allows all operations +- `SimpleExecutorWrapper` - Wraps CodeExecutor, skips resource limits +- `JupyterKernelWrapper` - Wraps JupyterKernelManager (has issues) + +The CLI creates temporary directories for logs and connection files, cleans up on exit. diff --git a/test_containers.py b/test_containers.py new file mode 100755 index 0000000..2109120 --- /dev/null +++ b/test_containers.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +""" +Minimal test script to verify container execution works. + +This tests the basic container creation and execution without +using the full MCP-Forge stack. +""" + +import sys +import subprocess +from pathlib import Path + +# ANSI colors +GREEN = "\033[92m" +RED = "\033[91m" +GRAY = "\033[90m" +RESET = "\033[0m" + + +def test_simple_execution(): + """Test simple container execution.""" + print(f"\n{GRAY}Testing simple (stateless) execution...{RESET}") + + code = "print('Hello from MCP-Forge!')\nprint(2 + 2)" + + cmd = [ + "podman", "run", "--rm", + "--network=none", + "mcp-forge/python:3.12", + "python3", "-c", code + ] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=10 + ) + + if result.returncode == 0: + print(f"{GREEN}✓ Simple execution works!{RESET}") + print(f"{GRAY}Output:{RESET}") + print(result.stdout) + return True + else: + print(f"{RED}✗ Simple execution failed{RESET}") + print(f"{RED}Error:{RESET}") + print(result.stderr) + return False + + except subprocess.TimeoutExpired: + print(f"{RED}✗ Execution timed out{RESET}") + return False + except Exception as e: + print(f"{RED}✗ Error: {e}{RESET}") + return False + + +def test_jupyter_kernel(): + """Test Jupyter kernel container.""" + print(f"\n{GRAY}Testing Jupyter kernel availability...{RESET}") + + cmd = [ + "podman", "run", "--rm", + "--network=host", + "mcp-forge/jupyter:latest", + "python3", "-c", "import ipykernel; print(ipykernel.__version__)" + ] + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=10 + ) + + if result.returncode == 0: + version = result.stdout.strip() + print(f"{GREEN}✓ Jupyter kernel available (ipykernel {version})!{RESET}") + return True + else: + print(f"{RED}✗ Jupyter kernel check failed{RESET}") + print(f"{RED}Error:{RESET}") + print(result.stderr) + return False + + except subprocess.TimeoutExpired: + print(f"{RED}✗ Check timed out{RESET}") + return False + except Exception as e: + print(f"{RED}✗ Error: {e}{RESET}") + return False + + +def main(): + """Main entry point.""" + print(f"{GREEN}MCP-Forge Container Tests{RESET}") + print(f"{GRAY}{'='*40}{RESET}") + + results = [] + + # Test simple execution + results.append(("Simple Execution", test_simple_execution())) + + # Test Jupyter kernel + results.append(("Jupyter Kernel", test_jupyter_kernel())) + + # Summary + print(f"\n{GRAY}{'='*40}{RESET}") + print(f"{GREEN}Test Summary:{RESET}\n") + + passed = sum(1 for _, result in results if result) + total = len(results) + + for name, result in results: + status = f"{GREEN}✓ PASS{RESET}" if result else f"{RED}✗ FAIL{RESET}" + print(f" {name}: {status}") + + print(f"\n{GRAY}Passed: {passed}/{total}{RESET}") + + if passed == total: + print(f"\n{GREEN}All tests passed!{RESET}") + return 0 + else: + print(f"\n{RED}Some tests failed.{RESET}") + return 1 + + +if __name__ == "__main__": + sys.exit(main())