MCP Path Traversal: Preventing File System Escapes in MCP Servers

# Common MCP file tool implementation
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_file":
        # The path comes from the LLM, which may be processing
        # attacker-controlled content
        path = arguments["path"]
        with open(path, "r") as f:
            return [{"type": "text", "text": f.read()}]

# VULNERABLE Pattern 1: No path validation at all
def read_file(path: str) -> str:
    with open(path, "r") as f:
        return f.read()
 
# VULNERABLE Pattern 2: String prefix check (bypassable)
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()
 
# VULNERABLE Pattern 3: Basename check only
def read_file(path: str) -> str:
    filename = os.path.basename(path)
    full_path = os.path.join(WORKSPACE, filename)
    # Bypassed by: path = "/../../../etc/shadow/../../var/mcp/workspace/x/../../../etc/shadow"
    with open(full_path, "r") as f:
        return f.read()
 
# VULNERABLE Pattern 4: Resolve then check (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")
    # Between check and open, a symlink could be created
    with open(resolved, "r") as f:
        return f.read()

"""
MCP Path Traversal Defense
Robust path validation that handles all known traversal techniques.
"""
 
import os
import stat
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
 
 
@dataclass
class PathValidationResult:
    """Result of validating a file path."""
    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:
    """
    Resolves and validates file paths for MCP file operations.
    Prevents all known path traversal techniques.
    """
 
    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 the base directory itself
        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:
        """
        Validate a requested file path.
 
        Steps:
        1. Reject null bytes and control characters
        2. Join with base directory
        3. Resolve all symlinks and relative components
        4. Verify resolved path is within base directory
        5. Check file type and permissions
        6. Validate extension allowlist
        """
        warnings = []
 
        # Step 1: Reject dangerous characters
        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"
            )
 
        # Step 2: Normalize and join
        # Remove leading slashes to prevent absolute path injection
        clean_path = requested_path.lstrip('/')
        clean_path = clean_path.lstrip('\\')
 
        # Join with base directory
        candidate = self.base_dir / clean_path
 
        # Step 3: Resolve all components
        try:
            if self.follow_symlinks:
                resolved = candidate.resolve()
            else:
                # resolve() follows symlinks; use strict resolve
                # to detect broken symlinks and handle race conditions
                resolved = candidate.resolve(strict=False)
 
                # If not following symlinks, verify no component is a symlink
                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}"
            )
 
        # Step 4: Verify containment (the critical check)
        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}'"
                ),
            )
 
        # Step 5: Check depth
        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}"
            )
 
        # Step 6: Check extension allowlist
        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}"
                )
 
        # Step 7: If file exists, check its properties
        if resolved.exists():
            file_stat = resolved.stat()
 
            # Check it is a regular file
            if not stat.S_ISREG(file_stat.st_mode):
                return PathValidationResult(
                    safe=False,
                    error=f"Not a regular file: {resolved}"
                )
 
            # Check file size
            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"
                    ),
                )
 
            # Warn about world-readable sensitive files
            if file_stat.st_mode & stat.S_IROTH:
                warnings.append("File is world-readable")
 
        return PathValidationResult(
            safe=True,
            resolved_path=str(resolved),
            warnings=warnings,
        )
 
 
# Usage in an MCP server
def create_secure_file_handler(workspace: str):
    """Create a file handler with proper path validation."""
 
    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 -- Create a chroot jail for MCP file servers
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}"
 
# Create directory structure
mkdir -p "${JAIL}"/{bin,lib,lib64,usr/lib,usr/bin,dev,proc,tmp,workspace}
chmod 1777 "${JAIL}/tmp"
chmod 750 "${JAIL}/workspace"
 
# Copy required binaries
for bin in /bin/sh /usr/bin/python3 /usr/bin/env; do
    if [ -f "$bin" ]; then
        cp "$bin" "${JAIL}${bin}"
    fi
done
 
# Copy required libraries (for 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
 
# Copy Python standard library
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}/"
 
# Create minimal /dev entries
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
 
# Create MCP server runner script
cat > "${JAIL}/run-mcp-server.sh" << 'RUNEOF'
#!/bin/sh
# This runs INSIDE the 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 with Linux namespace isolation.
Provides filesystem, network, and PID isolation.
"""
 
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:
    """
    Launch an MCP server in an isolated Linux namespace.
    Uses unshare for filesystem and optionally network isolation.
    """
    workspace = Path(workspace_path).resolve()
    workspace.mkdir(parents=True, exist_ok=True)
 
    # Build unshare command for namespace isolation
    unshare_args = [
        "unshare",
        "--mount",      # New mount namespace
        "--pid",        # New PID namespace
        "--fork",       # Fork before exec
        "--mount-proc", # Mount fresh /proc
    ]
 
    if not network_access:
        unshare_args.append("--net")  # New network namespace (no network)
 
    # Build the sandbox setup script
    sandbox_script = f"""
    set -e
    # Make root filesystem read-only
    mount --make-rprivate /
    mount -o remount,ro /
 
    # Create writable workspace overlay
    mkdir -p /tmp/mcp-workspace
    mount --bind {workspace} /tmp/mcp-workspace
    mount -o remount,rw /tmp/mcp-workspace
 
    # Create writable /tmp
    mount -t tmpfs -o size=100M tmpfs /tmp/mcp-tmp
 
    # Set restrictive umask
    umask 077
 
    # Drop to unprivileged user
    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
# Minimal Docker container for MCP file servers with filesystem isolation
 
FROM python:3.12-slim AS base
 
# Security: Run as non-root
RUN groupadd -r mcp && useradd -r -g mcp -d /home/mcp -s /bin/false mcp
 
# Install only required packages
RUN pip install --no-cache-dir mcp>=1.26.0
 
# Copy server code
COPY --chown=mcp:mcp server.py /app/server.py
 
# Create workspace directory
RUN mkdir -p /workspace && chown mcp:mcp /workspace
 
# Security hardening
RUN chmod 750 /app && chmod 750 /workspace
 
# Drop capabilities
USER mcp
 
# Read-only root filesystem (workspace is mounted as volume)
# --read-only flag is set in 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 -- Run MCP file server in Docker with security controls
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)"

"""
Detection rules for MCP path traversal monitoring.
"""
 
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:
    """Monitors MCP tool calls for path traversal attempts."""
 
    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 a path parameter for traversal indicators."""
        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(),
        }))

"""
Test suite for MCP path traversal defenses.
"""
 
import pytest
import os
import tempfile
from mcp_path_security import SecurePathResolver
 
 
@pytest.fixture
def workspace(tmp_path):
    """Create a test workspace with sample files."""
    (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):
        # Create a symlink pointing outside workspace
        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):
        # Create deeply nested path
        deep_path = "a/b/c/d/e/f/g/h/i/j/deep.txt"
        result = resolver.validate(deep_path)
        assert not result.safe  # Exceeds max_depth of 5
 
    def test_absolute_path_normalized(self, resolver):
        result = resolver.validate("/etc/passwd")
        assert not result.safe