From 3c677a13ab35c2e223bade7ad4d8577d58766d3d Mon Sep 17 00:00:00 2001 From: Irina Rueegg Date: Sun, 10 May 2026 21:44:53 +0200 Subject: [PATCH] feat MCP Sandboxing for python --- backend/agent/coding_agent.py | 16 ++- .../servers/mcp_server_code_execution.py | 130 +++++++++++++++++- .../agent/servers/mcp_server_file_search.py | 2 +- 3 files changed, 139 insertions(+), 9 deletions(-) diff --git a/backend/agent/coding_agent.py b/backend/agent/coding_agent.py index 0c6057d..12952e8 100644 --- a/backend/agent/coding_agent.py +++ b/backend/agent/coding_agent.py @@ -13,12 +13,9 @@ step-by-step methods so Streamlit can drive the loop via session_state: agent.reject(feedback) # skip action, inject user feedback """ -import ast -import inspect import json import os -import subprocess -import sys +import re from pathlib import Path import asyncio import pprint @@ -76,6 +73,8 @@ async def dispatch_tool(tool_name: str, arguments: dict) -> str: print(f"Trying to call tool '{tool_name}' in dispatch_tool through MCPToolAdapter...") result = await adapter.call_tool(tool_name, arguments) + print(f"Raw result from tool '{tool_name}': {result}") + if result.isError: texts = [block.text for block in result.content if block.type == "text"] return f"Tool error: {' '.join(texts)}" @@ -102,7 +101,7 @@ analyze code, and execute Python scripts. You can call tools to interact with the workspace and get feedback. You can write and read files, list directory contents, search for patterns, validate Python syntax, and run Python code. -Wou can access web search and page fetching tools to gather information from the internet. +You can access web search and page fetching tools to gather information from the internet. You can use these capabilities to iteratively work towards completing the user's task. @@ -239,7 +238,9 @@ def extract_json(text: str) -> str: Returns the cleaned JSON string, or the original text as a fallback (so json.loads can raise a meaningful error with context). """ - import re + + if text is None: + return "" # 1. Strip markdown fences fenced = re.sub(r"```(?:json)?\s*([\s\S]*?)\s*```", r"\1", text.strip()) @@ -282,6 +283,9 @@ def extract_json(text: str) -> str: def _strip_code_fences(text: str) -> str: """Remove markdown code fences (```json ... ```) from a string.""" + if text is None: + return "" + text = text.strip() if text.startswith("```"): lines = text.split("\n") diff --git a/backend/agent/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py index 4d04a75..e04308a 100644 --- a/backend/agent/servers/mcp_server_code_execution.py +++ b/backend/agent/servers/mcp_server_code_execution.py @@ -1,9 +1,32 @@ import ast +from datetime import datetime import subprocess 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" + +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() # ── Configuration ──────────────────────────────────────────────────────────── EXEC_TIMEOUT = 10 # seconds before killing the subprocess @@ -41,6 +64,10 @@ BLOCKED_BUILTINS = { "globals", "locals", "vars", "memoryview", "type" } +FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/", + "C:\\Windows", "C:\\Program Files", "C:\\Users", + "compile(", "__import__", "os.", "sys.", "subprocess."] + # ── Static Analysis ──────────────────────────────────────────────────── def check_code_safety(code: str) -> str | None: """ @@ -78,6 +105,10 @@ def check_code_safety(code: str) -> str | None: if isinstance(node.func, ast.Name): if node.func.id in BLOCKED_BUILTINS: return f"Blocked builtin: Use of builtin '{node.func.id}' is not allowed." + + for seq in FORBIDDEN_SEQUENCES: + if seq in code: + return f"Blocked: Suspect path sequence '{seq}' detected." return None # No violations found @@ -200,7 +231,84 @@ def lint_code(code: str) -> str: @mcp.tool() -def run_python_sandboxed(code: str) -> str: +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: """ Run Python code in a sandboxed environment. @@ -220,10 +328,24 @@ def run_python_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", "-c", code], + [PYTHON_EXE, "-c", code], + cwd=str(WORKSPACE_DIR), + env=custom_env, capture_output=True, text=True, timeout=EXEC_TIMEOUT) @@ -243,6 +365,10 @@ def run_python_sandboxed(code: str) -> str: except Exception as e: return f"Error during code execution: {e}" + finally: + if jail_dir.exists(): + shutil.rmtree(jail_dir) + @mcp.tool() def python_code_validation(code: str) -> str: diff --git a/backend/agent/servers/mcp_server_file_search.py b/backend/agent/servers/mcp_server_file_search.py index 7020593..65180bc 100644 --- a/backend/agent/servers/mcp_server_file_search.py +++ b/backend/agent/servers/mcp_server_file_search.py @@ -200,7 +200,7 @@ def create_new_directory(path: str) -> str: if resolved.exists(): return f"Error: File '{path}' already exists." - if resolved.suffix != None: + if resolved.suffix != None and resolved.suffix != "": return f"Error: can only create directories, got '{resolved.suffix}'." try: