From c8299f89c308fe420d38a241f4636df5833c839c Mon Sep 17 00:00:00 2001 From: Irina Rueegg Date: Thu, 21 May 2026 15:39:34 +0200 Subject: [PATCH] mcp code execution fix --- .../servers/mcp_server_code_execution.py | 349 ++++++++++++------ 1 file changed, 228 insertions(+), 121 deletions(-) diff --git a/backend/agent/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py index e04308a..c613653 100644 --- a/backend/agent/servers/mcp_server_code_execution.py +++ b/backend/agent/servers/mcp_server_code_execution.py @@ -1,35 +1,34 @@ import ast -from datetime import datetime import subprocess +import tempfile import io from pyflakes.api import check from pyflakes.reporter import Reporter from mcp.server.fastmcp import FastMCP from pathlib import Path -import venv -import shutil # ── Sandbox venv ──────────────────────────────────────────────────────────── -SERVER_BASE_DIR = Path(__file__).parent.resolve() -SANDBOX_DIR = SERVER_BASE_DIR / ".mcp_sandbox" -WORKSPACE_DIR = SERVER_BASE_DIR.parent.parent.parent.parent / "workspace" +#SERVER_BASE_DIR = Path(__file__).parent.resolve() +#SANDBOX_DIR = SERVER_BASE_DIR / ".mcp_sandbox" +#WORKSPACE_DIR = SERVER_BASE_DIR.parent.parent.parent.parent / "workspace" -def get_sandbox_paths(): - """Bestimmt die Executables innerhalb der Venv ohne os-Modul.""" - if not SANDBOX_DIR.exists(): - venv.create(SANDBOX_DIR, with_pip=True) - - bin_folder = "Scripts" if Path("C:/").exists() else "bin" # Einfacher Check für Windows - - python_exe = SANDBOX_DIR / bin_folder / "python" - pip_exe = SANDBOX_DIR / bin_folder / "pip" - - return str(python_exe), str(pip_exe) - -PYTHON_EXE, PIP_EXE = get_sandbox_paths() +#def get_sandbox_paths(): +# """Bestimmt die Executables innerhalb der Venv ohne os-Modul.""" +# if not SANDBOX_DIR.exists(): +# venv.create(SANDBOX_DIR, with_pip=True) +# +# bin_folder = "Scripts" if os.name == "nt" else "bin" # Einfacher Check für Windows +# +# exe_suffix = ".exe" if os.name == "nt" else "" +# python_exe = SANDBOX_DIR / bin_folder / f"python{exe_suffix}" +# pip_exe = SANDBOX_DIR / bin_folder / f"pip{exe_suffix}" +# +# return str(python_exe), str(pip_exe) +# +#PYTHON_EXE, PIP_EXE = get_sandbox_paths() # ── Configuration ──────────────────────────────────────────────────────────── -EXEC_TIMEOUT = 10 # seconds before killing the subprocess +EXEC_TIMEOUT = 45 # seconds before killing the subprocess MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return # ── Create the MCP server ──────────────────────────────────────────────────── @@ -68,6 +67,8 @@ FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/", "C:\\Windows", "C:\\Program Files", "C:\\Users", "compile(", "__import__", "os.", "sys.", "subprocess."] +ALLOWED_PACKAGES = ["pygame", "numpy", "pandas"] + # ── Static Analysis ──────────────────────────────────────────────────── def check_code_safety(code: str) -> str | None: """ @@ -188,7 +189,7 @@ def analyse_structure(code: str) -> str: @mcp.tool() -def lint_code(code: str) -> str: +def lint_code(code: str) -> tuple[str, bool]: """ Runs a fast static analysis check to catch syntax errors, unused imports, or undefined variables without executing the code. @@ -206,14 +207,14 @@ def lint_code(code: str) -> str: try: check(code, filename="", reporter=reporter) except Exception as e: - return f"Critical error during linting: {str(e)}" + return (f"Critical error during linting: {str(e)}", False) errors = error_buffer.getvalue().strip() warnings = warning_buffer.getvalue().strip() # Ergebnis-String zusammenbauen if not errors and not warnings: - return "Linting complete: No issues found. The code is syntactically sound." + return ("Linting complete: No issues found. The code is syntactically sound.", True) report = ["--- Linting Report ---"] @@ -227,88 +228,158 @@ def lint_code(code: str) -> str: report.append("\nAdvice: Please fix these issues before attempting to execute the code.") - return "\n".join(report) + return ("\n".join(report), False) +#@mcp.tool() +#def list_sandbox_packages() -> str: +# """ +# Lists all Python-Packages, that are installed in the Sandbox and their Version. +# Helpful to determine if packages like 'pygame', 'numpy' or similair are already available +# """ +# try: +# result = subprocess.run( +# [PIP_EXE, "list"], +# capture_output=True, +# text=True, +# timeout=10 +# ) +# +# if result.returncode != 0: +# return f"Error while listing the packages: {result.stderr}" +# +# if not result.stdout.strip(): +# return "The Sandbox environment is empty (only Standard-Libraries are available)." +# +# return f"Installed Packages: {result.stdout}" +# +# except Exception as e: +# return f"Error trying to list packages from the Sandbox venv: {str(e)}" +# +# +#@mcp.tool() +#def install_package_into_sandbox(package_name: str) -> str: +# """ +# Install a Python package into the sandbox environment using pip. +# +# Args: +# package_name: The name of the package to install (e.g., "requests"). +# +# Returns: +# A success message or an error message if installation fails. +# """ +# clean_name = "".join(e for e in package_name if e.isalnum() or e in "-_.") +# +# if clean_name in BLOCKED_IMPORTS: +# return f"Error: Installation of package '{clean_name}' is blocked due to security policies." +# +# if clean_name in BLOCKED_BUILTINS: +# return f"Error: Installation of package '{clean_name}' is blocked due to security policies." +# +# if not clean_name: +# return "Error: Invalid package name provided." +# +# try: +# result = subprocess.run( +# [PIP_EXE, "install", clean_name], +# capture_output=True, +# text=True, +# timeout=EXEC_TIMEOUT +# ) +# +# if result.returncode == 0: +# return f"Package '{clean_name}' installed successfully in the sandbox." +# else: +# return (f"Error installing package '{clean_name}':\n" +# f"{result.stdout}\n{result.stderr}") +# +# except subprocess.TimeoutExpired: +# return f"Error: Package installation exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated." +# except Exception as e: +# return f"Error during package installation: {e}" +# +# +#@mcp.tool() +#def reset_sandbox() -> str: +# """Deletes the complete Sandbox and reopens it from Zero (Full Reset).""" +# global PYTHON_EXE, PIP_EXE +# +# try: +# if SANDBOX_DIR.exists(): +# shutil.rmtree(SANDBOX_DIR) +# +# PYTHON_EXE, PIP_EXE = get_sandbox_paths() +# +# return "Sandbox was reseet completely" +# +# except Exception as e: +# return f"Reset failed: {e}" + + +#@mcp.tool() +#def run_python_code_sandboxed(code: str) -> str: +# """ +# Runs Python code in a sandboxed environment. +# +# The sandbox blocks dangerous operations (filesystem, network, process +# control). Code is killed after 45 seconds. Use print() to produce +# output, which is captured and returned (up to 3000 chars). If the code +# is deemed unsafe by static analysis, it will not be executed and an error +# message will be returned instead. +# +# Args: +# code: The Python code or the File content of a python file to be executed in str format +# +# Returns: +# Combined stdout+stderr, or an error message in str format. +# """ +# +# static_safety = check_code_safety(code) +# if static_safety: +# return f"Code rejected:{static_safety}" +# +# run_id = datetime.now().strftime("%Y%m%d_%H%M%S") +# jail_dir = WORKSPACE_DIR / f"sandbox_run_{run_id}" +# +# try: +# jail_dir.mkdir(parents=True, exist_ok=True) +# +# custom_env = { +# "PYTHONPATH": str(WORKSPACE_DIR), +# "PATH": str(Path(PYTHON_EXE).parent), +# "HOME": str(jail_dir), +# "TMPDIR": str(jail_dir) +# } +# +# result = subprocess.run( +# [PYTHON_EXE, "-c", code], +# cwd=str(WORKSPACE_DIR), +# env=custom_env, +# capture_output=True, +# text=True, +# timeout=EXEC_TIMEOUT) +# +# output = result.stdout + result.stderr +# +# if len(output) > MAX_OUTPUT_LENGTH: +# output = output[:MAX_OUTPUT_LENGTH] + "\n...[output truncated]..." +# +# if not output.strip(): +# return "Code executed successfully (no output)." +# +# return output +# +# except subprocess.TimeoutExpired: +# return f"Error: Code execution exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated." +# except Exception as e: +# return f"Error during code execution: {e}" +# +# finally: +# if jail_dir.exists(): +# shutil.rmtree(jail_dir) + @mcp.tool() -def list_sandbox_packages() -> str: - """ - Lists all Python-Packages, that are installed in the Sandbox and their Version. - Helpful to determine if packages like 'pygame', 'numpy' or similair are already available - """ - try: - result = subprocess.run( - [PIP_EXE, "list"], - capture_output=True, - text=True, - timeout=10 - ) - - if result.returncode != 0: - return f"Error while listing the packages: {result.stderr}" - - if not result.stdout.strip(): - return "The Sandbox environment is empty (only Standard-Libraries are available)." - - return f"Installed Packages: {result.stdout}" - - except Exception as e: - return f"Error trying to list packages from the Sandbox venv: {str(e)}" - - -@mcp.tool() -def install_package_into_sandbox(package_name: str) -> str: - """ - Install a Python package into the sandbox environment using pip. - - Args: - package_name: The name of the package to install (e.g., "requests"). - - Returns: - A success message or an error message if installation fails. - """ - clean_name = "".join(e for e in package_name if e.isalnum() or e in "-_.") - - if clean_name in BLOCKED_IMPORTS: - return f"Error: Installation of package '{clean_name}' is blocked due to security policies." - - if clean_name in BLOCKED_BUILTINS: - return f"Error: Installation of package '{clean_name}' is blocked due to security policies." - - if not clean_name: - return "Error: Invalid package name provided." - - try: - result = subprocess.run( - [PIP_EXE, "install", clean_name], - capture_output=True, - text=True, - timeout=EXEC_TIMEOUT - ) - - if result.returncode == 0: - return f"Package '{clean_name}' installed successfully in the sandbox." - else: - return (f"Error installing package '{clean_name}':\n" - f"{result.stdout}\n{result.stderr}") - - except subprocess.TimeoutExpired: - return f"Error: Package installation exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated." - except Exception as e: - return f"Error during package installation: {e}" - - -@mcp.tool() -def reset_sandbox() -> str: - """Löscht die gesamte Sandbox und erstellt sie neu (Full Reset).""" - if SANDBOX_DIR.exists(): - shutil.rmtree(SANDBOX_DIR) - get_sandbox_paths() - return "Sandbox wurde komplett zurückgesetzt." - - -@mcp.tool() -def run_python_code_sandboxed(code: str) -> str: +def run_python_sandboxed(code: str) -> str: """ Run Python code in a sandboxed environment. @@ -328,24 +399,10 @@ def run_python_code_sandboxed(code: str) -> str: static_safety = check_code_safety(code) if static_safety: return f"Code rejected:{static_safety}" - - run_id = datetime.now().strftime("%Y%m%d_%H%M%S") - jail_dir = WORKSPACE_DIR / f"sandbox_run_{run_id}" try: - jail_dir.mkdir(parents=True, exist_ok=True) - - custom_env = { - "PYTHONPATH": str(WORKSPACE_DIR), - "PATH": str(Path(PYTHON_EXE).parent), - "HOME": str(jail_dir), - "TMPDIR": str(jail_dir) - } - result = subprocess.run( - [PYTHON_EXE, "-c", code], - cwd=str(WORKSPACE_DIR), - env=custom_env, + ["python", "-c", code], capture_output=True, text=True, timeout=EXEC_TIMEOUT) @@ -364,11 +421,61 @@ def run_python_code_sandboxed(code: str) -> str: return f"Error: Code execution exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated." except Exception as e: return f"Error during code execution: {e}" - - finally: - if jail_dir.exists(): - shutil.rmtree(jail_dir) +#@mcp.tool() +#def run_python_code(code: str) -> str: +# """ +# Execute Python code safely inside a temporary directory. +# +# Args: +# code: The Python code or the File content of a python file to be executed in str format +# +# Returns: +# Combined stdout+stderr, or an error message in str format. +# """ +# +# static_safety = check_code_safety(code) +# if static_safety: +# return f"Code rejected:{static_safety}" +# +# linted_code = lint_code(code) +# if not linted_code[1]: +# return f"Code is not executable. Errror while linting:{linted_code[0]}" +# +# # 2. Create isolated temp directory +# with tempfile.TemporaryDirectory() as tmp_dir: +# +# tmp_path = Path(tmp_dir) +# +# script_file = tmp_path / "main.py" +# +# script_file.write_text(code, encoding="utf-8") +# +# try: +# result = subprocess.run( +# ["python", str(script_file)], +# capture_output=True, +# text=True, +# timeout=EXEC_TIMEOUT, +# cwd=tmp_dir +# ) +# +# output = result.stdout + result.stderr +# +# if len(output) > MAX_OUTPUT_LENGTH: +# output = output[:MAX_OUTPUT_LENGTH] +# output += "\n...[output truncated]..." +# +# if not output.strip(): +# return "Code executed successfully." +# +# return output +# +# except subprocess.TimeoutExpired: +# return f"Execution stopped: Timeout after {EXEC_TIMEOUT} seconds." +# +# except Exception as e: +# return f"Execution error: {e}" @mcp.tool() def python_code_validation(code: str) -> str: