MCP path traversal: ontsnappingen uit het bestandssysteem in MCP-servers voorkomen

# Veelvoorkomende implementatie van een MCP-bestandstool
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_file":
        # Het pad komt van het LLM, dat mogelijk
        # door aanvaller gecontroleerde inhoud verwerkt
        path = arguments["path"]
        with open(path, "r") as f:
            return [{"type": "text", "text": f.read()}]

# KWETSBAAR patroon 1: helemaal geen padvalidatie
def read_file(path: str) -> str:
    with open(path, "r") as f:
        return f.read()
 
# KWETSBAAR patroon 2: stringprefix-controle (te omzeilen)
WORKSPACE = "/var/mcp/workspace"
def read_file(path: str) -> str:
    if not path.startswith(WORKSPACE):
        raise PermissionError("Access denied")
    with open(path, "r") as f:  # path: /var/mcp/workspace/../../../etc/shadow
        return f.read()
 
# KWETSBAAR patroon 3: alleen basename-controle
def read_file(path: str) -> str:
    filename = os.path.basename(path)
    full_path = os.path.join(WORKSPACE, filename)
    # Omzeild met: path = "/../../../etc/shadow/../../var/mcp/workspace/x/../../../etc/shadow"
    with open(full_path, "r") as f:
        return f.read()
 
# KWETSBAAR patroon 4: eerst resolven, dan controleren (TOCTOU-race)
def read_file(path: str) -> str:
    full_path = os.path.join(WORKSPACE, path)
    resolved = os.path.realpath(full_path)
    if not resolved.startswith(WORKSPACE):
        raise PermissionError("Access denied")
    # Tussen controle en open kan een symlink worden aangemaakt
    with open(resolved, "r") as f:
        return f.read()

"""
MCP Path Traversal Defense
Robuuste padvalidatie die alle bekende traversaltechnieken afhandelt.
"""
 
import os
import stat
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class PathValidationResult:
    """Resultaat van het valideren van een bestandspad."""
    safe: bool
    resolved_path: Optional[str] = None
    error: Optional[str] = None
    warnings: list[str] = None
 
    def __post_init__(self):
        if self.warnings is None:
            self.warnings = []
 
class SecurePathResolver:
    """
    Resolvt en valideert bestandspaden voor MCP-bestandsoperaties.
    Voorkomt alle bekende path traversal-technieken.
    """
 
    def __init__(self, base_directory: str,
                 allowed_extensions: set[str] = None,
                 max_depth: int = 10,
                 follow_symlinks: bool = False,
                 max_file_size: int = 10_000_000):
        # Resolve de basisdirectory zelf
        self.base_dir = Path(base_directory).resolve()
        if not self.base_dir.is_dir():
            raise ValueError(f"Base directory does not exist: {base_directory}")
 
        self.allowed_extensions = allowed_extensions
        self.max_depth = max_depth
        self.follow_symlinks = follow_symlinks
        self.max_file_size = max_file_size
 
    def validate(self, requested_path: str) -> PathValidationResult:
        """
        Valideer een opgevraagd bestandspad.
 
        Stappen:
        1. Weiger null bytes en stuurtekens
        2. Voeg samen met de basisdirectory
        3. Resolve alle symlinks en relatieve componenten
        4. Verifieer dat het geresolvde pad binnen de basisdirectory ligt
        5. Controleer bestandstype en -rechten
        6. Valideer de allowlist van extensies
        """
        warnings = []
 
        # Stap 1: weiger gevaarlijke tekens
        if '\x00' in requested_path:
            return PathValidationResult(
                safe=False,
                error="Null byte in path (null byte injection attempt)"
            )
 
        if any(c in requested_path for c in '\r\n'):
            return PathValidationResult(
                safe=False,
                error="Control characters in path"
            )
 
        # Stap 2: normaliseren en samenvoegen
        # Verwijder voorloopslashes om injectie van absolute paden te voorkomen
        clean_path = requested_path.lstrip('/')
        clean_path = clean_path.lstrip('\\')
 
        # Voeg samen met de basisdirectory
        candidate = self.base_dir / clean_path
 
        # Stap 3: resolve alle componenten
        try:
            if self.follow_symlinks:
                resolved = candidate.resolve()
            else:
                # resolve() volgt symlinks; gebruik strict resolve
                # om kapotte symlinks te detecteren en raceconditions af te handelen
                resolved = candidate.resolve(strict=False)
 
                # Als symlinks niet worden gevolgd, verifieer dat geen enkel component een symlink is
                current = self.base_dir
                for part in Path(clean_path).parts:
                    current = current / part
                    if current.is_symlink():
                        return PathValidationResult(
                            safe=False,
                            error=f"Symlink detected at: {current.relative_to(self.base_dir)}"
                        )
        except (OSError, ValueError) as e:
            return PathValidationResult(
                safe=False,
                error=f"Path resolution failed: {e}"
            )
 
        # Stap 4: verifieer dat het pad binnen de grenzen blijft (de cruciale controle)
        try:
            resolved.relative_to(self.base_dir)
        except ValueError:
            return PathValidationResult(
                safe=False,
                error=(
                    f"Path traversal detected: '{requested_path}' "
                    f"resolves to '{resolved}' which is outside "
                    f"base directory '{self.base_dir}'"
                ),
            )
 
        # Stap 5: controleer de diepte
        depth = len(resolved.relative_to(self.base_dir).parts)
        if depth > self.max_depth:
            return PathValidationResult(
                safe=False,
                error=f"Path depth {depth} exceeds maximum {self.max_depth}"
            )
 
        # Stap 6: controleer de allowlist van extensies
        if self.allowed_extensions is not None:
            ext = resolved.suffix.lower()
            if ext not in self.allowed_extensions:
                return PathValidationResult(
                    safe=False,
                    error=f"File extension '{ext}' not in allowed set: {self.allowed_extensions}"
                )
 
        # Stap 7: als het bestand bestaat, controleer de eigenschappen ervan
        if resolved.exists():
            file_stat = resolved.stat()
 
            # Controleer dat het een gewoon bestand is
            if not stat.S_ISREG(file_stat.st_mode):
                return PathValidationResult(
                    safe=False,
                    error=f"Not a regular file: {resolved}"
                )
 
            # Controleer de bestandsgrootte
            if file_stat.st_size > self.max_file_size:
                return PathValidationResult(
                    safe=False,
                    error=(
                        f"File size {file_stat.st_size} bytes exceeds "
                        f"maximum {self.max_file_size} bytes"
                    ),
                )
 
            # Waarschuw bij voor iedereen leesbare gevoelige bestanden
            if file_stat.st_mode & stat.S_IROTH:
                warnings.append("File is world-readable")
 
        return PathValidationResult(
            safe=True,
            resolved_path=str(resolved),
            warnings=warnings,
        )
 
# Gebruik in een MCP-server
def create_secure_file_handler(workspace: str):
    """Maak een bestands-handler met correcte padvalidatie."""
 
    resolver = SecurePathResolver(
        base_directory=workspace,
        allowed_extensions={".txt", ".md", ".json", ".yaml", ".yml",
                           ".py", ".js", ".ts", ".csv", ".xml", ".toml"},
        max_depth=10,
        follow_symlinks=False,
        max_file_size=10_000_000,
    )
 
    async def handle_read_file(arguments: dict) -> list[dict]:
        path = arguments.get("path", "")
        result = resolver.validate(path)
 
        if not result.safe:
            return [{"type": "text", "text": f"Access denied: {result.error}"}]
 
        try:
            with open(result.resolved_path, "r") as f:
                content = f.read()
            return [{"type": "text", "text": content}]
        except (PermissionError, OSError) as e:
            return [{"type": "text", "text": f"Read error: {e}"}]
 
    async def handle_write_file(arguments: dict) -> list[dict]:
        path = arguments.get("path", "")
        content = arguments.get("content", "")
 
        result = resolver.validate(path)
        if not result.safe:
            return [{"type": "text", "text": f"Access denied: {result.error}"}]
 
        try:
            os.makedirs(os.path.dirname(result.resolved_path), exist_ok=True)
            with open(result.resolved_path, "w") as f:
                f.write(content)
            return [{"type": "text", "text": f"Written: {path}"}]
        except (PermissionError, OSError) as e:
            return [{"type": "text", "text": f"Write error: {e}"}]
 
    return handle_read_file, handle_write_file

#!/bin/bash
# setup-mcp-chroot.sh -- Maak een chroot-jail voor MCP-bestandsservers
set -euo pipefail
 
CHROOT_BASE="/var/mcp/jails"
SERVER_NAME="${1:?Usage: setup-mcp-chroot.sh <server-name>}"
JAIL="${CHROOT_BASE}/${SERVER_NAME}"
 
echo "[*] Creating chroot jail for MCP server: ${SERVER_NAME}"
 
# Maak de directorystructuur aan
mkdir -p "${JAIL}"/{bin,lib,lib64,usr/lib,usr/bin,dev,proc,tmp,workspace}
chmod 1777 "${JAIL}/tmp"
chmod 750 "${JAIL}/workspace"
 
# Kopieer de benodigde binaries
for bin in /bin/sh /usr/bin/python3 /usr/bin/env; do
    if [ -f "$bin" ]; then
        cp "$bin" "${JAIL}${bin}"
    fi
done
 
# Kopieer de benodigde bibliotheken (voor Python)
for lib in $(ldd /usr/bin/python3 | grep -oP '/[^\s]+'); do
    dir=$(dirname "$lib")
    mkdir -p "${JAIL}${dir}"
    cp "$lib" "${JAIL}${lib}" 2>/dev/null || true
done
 
# Kopieer de Python-standaardbibliotheek
PYVER=$(python3 -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')")
mkdir -p "${JAIL}/usr/lib/python${PYVER}"
cp -r "/usr/lib/python${PYVER}"/* "${JAIL}/usr/lib/python${PYVER}/"
 
# Maak minimale /dev-entries aan
mknod -m 666 "${JAIL}/dev/null" c 1 3 2>/dev/null || true
mknod -m 666 "${JAIL}/dev/zero" c 1 5 2>/dev/null || true
mknod -m 666 "${JAIL}/dev/urandom" c 1 9 2>/dev/null || true
 
# Maak het runner-script voor de MCP-server aan
cat > "${JAIL}/run-mcp-server.sh" << 'RUNEOF'
#!/bin/sh
# Dit draait BINNEN de chroot
cd /workspace
exec python3 /usr/bin/mcp-server "$@"
RUNEOF
chmod +x "${JAIL}/run-mcp-server.sh"
 
echo "[+] Chroot jail created at: ${JAIL}"
echo "    Workspace: ${JAIL}/workspace"
echo ""
echo "    To run: chroot --userspec=mcp-server:mcp-server ${JAIL} /run-mcp-server.sh"

"""
MCP-server-launcher met isolatie via Linux-namespaces.
Biedt isolatie van bestandssysteem, netwerk en PID.
"""
 
import os
import subprocess
import json
import logging
from pathlib import Path
 
logger = logging.getLogger("mcp.sandbox")
 
def launch_sandboxed_mcp_server(
    server_command: list[str],
    workspace_path: str,
    server_name: str,
    network_access: bool = False,
) -> subprocess.Popen:
    """
    Start een MCP-server in een geïsoleerde Linux-namespace.
    Gebruikt unshare voor bestandssysteem- en optioneel netwerkisolatie.
    """
    workspace = Path(workspace_path).resolve()
    workspace.mkdir(parents=True, exist_ok=True)
 
    # Bouw het unshare-commando voor namespace-isolatie
    unshare_args = [
        "unshare",
        "--mount",      # Nieuwe mount-namespace
        "--pid",        # Nieuwe PID-namespace
        "--fork",       # Fork vóór exec
        "--mount-proc", # Mount een verse /proc
    ]
 
    if not network_access:
        unshare_args.append("--net")  # Nieuwe netwerk-namespace (geen netwerk)
 
    # Bouw het setup-script voor de sandbox
    sandbox_script = f"""
    set -e
    # Maak het root-bestandssysteem alleen-lezen
    mount --make-rprivate /
    mount -o remount,ro /
 
    # Maak een beschrijfbare workspace-overlay
    mkdir -p /tmp/mcp-workspace
    mount --bind {workspace} /tmp/mcp-workspace
    mount -o remount,rw /tmp/mcp-workspace
 
    # Maak een beschrijfbare /tmp
    mount -t tmpfs -o size=100M tmpfs /tmp/mcp-tmp
 
    # Stel een restrictieve umask in
    umask 077
 
    # Val terug naar een gebruiker zonder privileges
    exec su -s /bin/sh mcp-server -c '{" ".join(server_command)}'
    """
 
    process = subprocess.Popen(
        unshare_args + ["sh", "-c", sandbox_script],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env={
            "PATH": "/usr/bin:/bin",
            "HOME": "/tmp",
            "MCP_WORKSPACE": "/tmp/mcp-workspace",
        },
    )
 
    logger.info(
        "Launched sandboxed MCP server '%s' (PID %d, workspace: %s, network: %s)",
        server_name, process.pid, workspace, network_access,
    )
 
    return process

# Dockerfile.mcp-file-server
# Minimale Docker-container voor MCP-bestandsservers met filesystem-isolatie
 
FROM python:3.12-slim AS base
 
# Beveiliging: draai als non-root
RUN groupadd -r mcp && useradd -r -g mcp -d /home/mcp -s /bin/false mcp
 
# Installeer alleen de benodigde pakketten
RUN pip install --no-cache-dir mcp>=1.26.0
 
# Kopieer de servercode
COPY --chown=mcp:mcp server.py /app/server.py
 
# Maak de workspace-directory aan
RUN mkdir -p /workspace && chown mcp:mcp /workspace
 
# Beveiligingsharding
RUN chmod 750 /app && chmod 750 /workspace
 
# Laat capabilities vallen
USER mcp
 
# Alleen-lezen root-bestandssysteem (de workspace wordt als volume gemount)
# De --read-only-vlag wordt ingesteld bij docker run
 
WORKDIR /app
 
# Health check
HEALTHCHECK --interval=30s --timeout=5s \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"
 
ENTRYPOINT ["python", "server.py"]

#!/bin/bash
# run-mcp-file-server.sh -- Draai de MCP-bestandsserver in Docker met beveiligingscontroles
set -euo pipefail
 
SERVER_NAME="${1:?Usage: run-mcp-file-server.sh <server-name> <workspace-path>}"
WORKSPACE="${2:?Usage: run-mcp-file-server.sh <server-name> <workspace-path>}"
 
docker run -d \
    --name "mcp-${SERVER_NAME}" \
    --read-only \
    --tmpfs /tmp:size=100M,noexec,nosuid \
    --volume "${WORKSPACE}:/workspace:rw" \
    --cap-drop ALL \
    --security-opt no-new-privileges \
    --security-opt seccomp=mcp-seccomp.json \
    --memory 256m \
    --cpus 0.5 \
    --pids-limit 50 \
    --network none \
    --user mcp:mcp \
    "mcp-file-server:latest"
 
echo "MCP file server '${SERVER_NAME}' started."
echo "Workspace: ${WORKSPACE} -> /workspace (container)"

"""
Detectieregels voor het monitoren van MCP path traversal.
"""
 
import re
import json
import logging
from datetime import datetime
from dataclasses import dataclass, field
 
logger = logging.getLogger("mcp.security.traversal")
 
TRAVERSAL_PATTERNS = [
    {
        "id": "MCP-PT-001",
        "name": "Directory traversal sequences",
        "pattern": re.compile(r'\.\.(/|\\)'),
        "severity": "high",
        "description": "Path contains ../ or ..\\ sequences",
    },
    {
        "id": "MCP-PT-002",
        "name": "Absolute path in parameter",
        "pattern": re.compile(r'^(/|[A-Za-z]:\\)'),
        "severity": "medium",
        "description": "Absolute path provided (expected relative)",
    },
    {
        "id": "MCP-PT-003",
        "name": "Sensitive file targets",
        "pattern": re.compile(
            r'(etc/(passwd|shadow|hosts|sudoers)|'
            r'\.ssh/(id_rsa|authorized_keys)|'
            r'\.aws/credentials|'
            r'\.env|\.git/config|'
            r'proc/self/(environ|cmdline))'
        ),
        "severity": "critical",
        "description": "Attempt to access known sensitive files",
    },
    {
        "id": "MCP-PT-004",
        "name": "URL-encoded traversal",
        "pattern": re.compile(r'(%2[eE]){2}(%2[fF]|%5[cC])'),
        "severity": "high",
        "description": "URL-encoded path traversal attempt",
    },
    {
        "id": "MCP-PT-005",
        "name": "Null byte injection",
        "pattern": re.compile(r'%00|\x00'),
        "severity": "critical",
        "description": "Null byte in path (truncation attack)",
    },
    {
        "id": "MCP-PT-006",
        "name": "Double encoding",
        "pattern": re.compile(r'%25(2[eE]|2[fF]|5[cC])'),
        "severity": "high",
        "description": "Double URL-encoded traversal characters",
    },
    {
        "id": "MCP-PT-007",
        "name": "Unicode normalization bypass",
        "pattern": re.compile(r'[\uff0e]{2}[\uff0f]|%c0%ae|%c0%af'),
        "severity": "high",
        "description": "Unicode-based path traversal bypass",
    },
]
 
@dataclass
class TraversalAlert:
    rule_id: str
    severity: str
    tool_name: str
    path_value: str
    description: str
    timestamp: datetime = field(default_factory=datetime.utcnow)
 
class MCPTraversalDetector:
    """Monitort MCP-tool-aanroepen op path traversal-pogingen."""
 
    def __init__(self, alert_callback=None):
        self.alert_callback = alert_callback or self._log_alert
        self.alert_count = 0
 
    def scan_path_parameter(self, tool_name: str,
                            param_name: str,
                            path_value: str) -> list[TraversalAlert]:
        """Scan een padparameter op traversal-indicatoren."""
        alerts = []
 
        for rule in TRAVERSAL_PATTERNS:
            if rule["pattern"].search(path_value):
                alert = TraversalAlert(
                    rule_id=rule["id"],
                    severity=rule["severity"],
                    tool_name=tool_name,
                    path_value=path_value[:200],
                    description=rule["description"],
                )
                alerts.append(alert)
                self.alert_count += 1
                self.alert_callback(alert)
 
        return alerts
 
    def _log_alert(self, alert: TraversalAlert):
        logger.warning(json.dumps({
            "event": "mcp_traversal_detection",
            "rule_id": alert.rule_id,
            "severity": alert.severity,
            "tool": alert.tool_name,
            "path": alert.path_value,
            "description": alert.description,
            "timestamp": alert.timestamp.isoformat(),
        }))

"""
Testsuite voor MCP path traversal-verdedigingen.
"""
 
import pytest
import os
import tempfile
from mcp_path_security import SecurePathResolver
 
@pytest.fixture
def workspace(tmp_path):
    """Maak een test-workspace met voorbeeldbestanden."""
    (tmp_path / "allowed.txt").write_text("allowed content")
    (tmp_path / "subdir").mkdir()
    (tmp_path / "subdir" / "nested.txt").write_text("nested content")
    return str(tmp_path)
 
@pytest.fixture
def resolver(workspace):
    return SecurePathResolver(
        base_directory=workspace,
        allowed_extensions={".txt", ".md", ".json"},
        max_depth=5,
        follow_symlinks=False,
    )
 
class TestPathTraversalPrevention:
 
    TRAVERSAL_PAYLOADS = [
        "../../../etc/passwd",
        "..\\..\\..\\windows\\system32\\config\\sam",
        "....//....//....//etc/passwd",
        "..%2f..%2f..%2fetc/passwd",
        "%2e%2e/%2e%2e/%2e%2e/etc/passwd",
        "..%252f..%252f..%252fetc/passwd",
        "/etc/passwd",
        "\\etc\\passwd",
        "subdir/../../etc/passwd",
        "allowed.txt/../../../etc/passwd",
        "./././../../../etc/passwd",
        "...\\.\\...\\.\\etc\\passwd",
    ]
 
    @pytest.mark.parametrize("payload", TRAVERSAL_PAYLOADS)
    def test_traversal_blocked(self, resolver, payload):
        result = resolver.validate(payload)
        assert not result.safe, f"Traversal payload was not blocked: {payload}"
 
    def test_valid_file_accepted(self, resolver):
        result = resolver.validate("allowed.txt")
        assert result.safe
        assert result.resolved_path is not None
 
    def test_valid_nested_file_accepted(self, resolver):
        result = resolver.validate("subdir/nested.txt")
        assert result.safe
 
    def test_null_byte_blocked(self, resolver):
        result = resolver.validate("allowed.txt\x00../../../etc/passwd")
        assert not result.safe
 
    def test_disallowed_extension_blocked(self, resolver):
        result = resolver.validate("script.sh")
        assert not result.safe
 
    def test_symlink_blocked(self, resolver, workspace):
        # Maak een symlink die buiten de workspace wijst
        link_path = os.path.join(workspace, "evil_link.txt")
        os.symlink("/etc/passwd", link_path)
        result = resolver.validate("evil_link.txt")
        assert not result.safe
 
    def test_depth_limit_enforced(self, resolver, workspace):
        # Maak een diep genest pad
        deep_path = "a/b/c/d/e/f/g/h/i/j/deep.txt"
        result = resolver.validate(deep_path)
        assert not result.safe  # Overschrijdt max_depth van 5
 
    def test_absolute_path_normalized(self, resolver):
        result = resolver.validate("/etc/passwd")
        assert not result.safe