feat MCP Sandboxing for python

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 re
-import sys
 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.
 </capabilities>
 
@@ -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")

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:

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: