MCP 路徑穿越：防止 MCP 伺服器檔案系統逃逸

# 常見 MCP 檔案工具實作
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_file":
        # 路徑來自 LLM，可能正處理
        # 攻擊者控制的內容
        path = arguments["path"]
        with open(path, "r") as f:
            return [{"type": "text", "text": f.read()}]

# 易受攻擊模式 1：完全沒有路徑驗證
def read_file(path: str) -> str:
    with open(path, "r") as f:
        return f.read()
 
# 易受攻擊模式 2：字串前綴檢查（可繞過）
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()
 
# 易受攻擊模式 3：僅檢查 basename
def read_file(path: str) -> str:
    filename = os.path.basename(path)
    full_path = os.path.join(WORKSPACE, filename)
    # 可被：path = "/../../../etc/shadow/../../var/mcp/workspace/x/../../../etc/shadow" 繞過
    with open(full_path, "r") as f:
        return f.read()
 
# 易受攻擊模式 4：先解析再檢查（TOCTOU 競態）
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")
    # 檢查與開啟之間，symlink 可能被建立
    with open(resolved, "r") as f:
        return f.read()

"""
MCP 路徑穿越防禦
能處理所有已知穿越技術的穩健路徑驗證。
"""
 
import os
import stat
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class PathValidationResult:
    """檔案路徑驗證結果。"""
    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:
    """
    為 MCP 檔案操作解析並驗證檔案路徑。
    防範所有已知的路徑穿越技術。
    """
 
    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):
        # 解析基底目錄本身
        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:
        """
        驗證所請求的檔案路徑。
 
        步驟：
        1. 拒絕 null byte 與控制字元
        2. 與基底目錄合併
        3. 解析所有 symlink 與相對成分
        4. 確認解析後的路徑在基底目錄內
        5. 檢查檔案類型與權限
        6. 驗證副檔名允許清單
        """
        warnings = []
 
        # 步驟 1：拒絕危險字元
        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"
            )
 
        # 步驟 2：正規化並合併
        # 移除前導斜線以防絕對路徑注入
        clean_path = requested_path.lstrip('/')
        clean_path = clean_path.lstrip('\\')
 
        # 與基底目錄合併
        candidate = self.base_dir / clean_path
 
        # 步驟 3：解析所有成分
        try:
            if self.follow_symlinks:
                resolved = candidate.resolve()
            else:
                # resolve() 會跟隨 symlink；使用非嚴格解析
                # 以偵測斷裂的 symlink 並處理競態
                resolved = candidate.resolve(strict=False)
 
                # 若不跟隨 symlink，驗證每個成分都不是 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}"
            )
 
        # 步驟 4：驗證包含關係（關鍵檢查）
        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}'"
                ),
            )
 
        # 步驟 5：檢查深度
        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}"
            )
 
        # 步驟 6：檢查副檔名允許清單
        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}"
                )
 
        # 步驟 7：若檔案存在則檢查其屬性
        if resolved.exists():
            file_stat = resolved.stat()
 
            # 檢查是否為一般檔案
            if not stat.S_ISREG(file_stat.st_mode):
                return PathValidationResult(
                    safe=False,
                    error=f"Not a regular file: {resolved}"
                )
 
            # 檢查檔案大小
            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"
                    ),
                )
 
            # 對全域可讀的敏感檔發出警告
            if file_stat.st_mode & stat.S_IROTH:
                warnings.append("File is world-readable")
 
        return PathValidationResult(
            safe=True,
            resolved_path=str(resolved),
            warnings=warnings,
        )
 
# 在 MCP 伺服器中使用
def create_secure_file_handler(workspace: str):
    """建立具正確路徑驗證的檔案處理器。"""
 
    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 — 為 MCP 檔案伺服器建立 chroot 監獄
set -euo pipefail
 
CHROOT_BASE="/var/mcp/jails"
SERVER_NAME="${1:?Usage: setup-mcp-chroot.sh <server-name>}"
JAIL="${CHROOT_BASE}/${SERVER_NAME}"
 
echo "[*] 為 MCP 伺服器 ${SERVER_NAME} 建立 chroot 監獄"
 
# 建立目錄結構
mkdir -p "${JAIL}"/{bin,lib,lib64,usr/lib,usr/bin,dev,proc,tmp,workspace}
chmod 1777 "${JAIL}/tmp"
chmod 750 "${JAIL}/workspace"
 
# 複製所需的二進位檔
for bin in /bin/sh /usr/bin/python3 /usr/bin/env; do
    if [ -f "$bin" ]; then
        cp "$bin" "${JAIL}${bin}"
    fi
done
 
# 複製 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
 
# 複製 Python 標準函式庫
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}/"
 
# 建立最小的 /dev 條目
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
 
# 建立 MCP 伺服器執行腳本
cat > "${JAIL}/run-mcp-server.sh" << 'RUNEOF'
#!/bin/sh
# 此腳本在 chroot 內部執行
cd /workspace
exec python3 /usr/bin/mcp-server "$@"
RUNEOF
chmod +x "${JAIL}/run-mcp-server.sh"
 
echo "[+] Chroot 監獄已建立於：${JAIL}"
echo "    工作空間：${JAIL}/workspace"
echo ""
echo "    執行方式：chroot --userspec=mcp-server:mcp-server ${JAIL} /run-mcp-server.sh"

"""
具 Linux 命名空間隔離的 MCP 伺服器啟動器。
提供檔案系統、網路與 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:
    """
    在隔離的 Linux 命名空間中啟動 MCP 伺服器。
    使用 unshare 進行檔案系統（以及可選的網路）隔離。
    """
    workspace = Path(workspace_path).resolve()
    workspace.mkdir(parents=True, exist_ok=True)
 
    # 建構 unshare 命令進行命名空間隔離
    unshare_args = [
        "unshare",
        "--mount",      # 新的 mount namespace
        "--pid",        # 新的 PID namespace
        "--fork",       # exec 前先 fork
        "--mount-proc", # 掛載新的 /proc
    ]
 
    if not network_access:
        unshare_args.append("--net")  # 新的網路 namespace（無網路）
 
    # 建構沙箱設定腳本
    sandbox_script = f"""
    set -e
    # 將根檔案系統設為唯讀
    mount --make-rprivate /
    mount -o remount,ro /
 
    # 建立可寫入的工作空間覆蓋
    mkdir -p /tmp/mcp-workspace
    mount --bind {workspace} /tmp/mcp-workspace
    mount -o remount,rw /tmp/mcp-workspace
 
    # 建立可寫入的 /tmp
    mount -t tmpfs -o size=100M tmpfs /tmp/mcp-tmp
 
    # 設定嚴格的 umask
    umask 077
 
    # 降權至非特權使用者
    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
# 具檔案系統隔離的 MCP 檔案伺服器最小 Docker 容器
 
FROM python:3.12-slim AS base
 
# 安全性：以非 root 執行
RUN groupadd -r mcp && useradd -r -g mcp -d /home/mcp -s /bin/false mcp
 
# 只安裝必要的套件
RUN pip install --no-cache-dir mcp>=1.26.0
 
# 複製伺服器程式碼
COPY --chown=mcp:mcp server.py /app/server.py
 
# 建立工作空間目錄
RUN mkdir -p /workspace && chown mcp:mcp /workspace
 
# 安全強化
RUN chmod 750 /app && chmod 750 /workspace
 
# 卸下能力
USER mcp
 
# 唯讀根檔案系統（工作空間以 volume 掛載）
# --read-only 旗標在 docker run 時設定
 
WORKDIR /app
 
# 健康檢查
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 — 以 Docker 與安全控制執行 MCP 檔案伺服器
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 檔案伺服器 '${SERVER_NAME}' 已啟動。"
echo "工作空間：${WORKSPACE} -> /workspace（容器內）"

"""
MCP 路徑穿越監控的偵測規則。
"""
 
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:
    """監控 MCP 工具呼叫的路徑穿越嘗試。"""
 
    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]:
        """掃描路徑參數以偵測穿越指標。"""
        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(),
        }))

"""
MCP 路徑穿越防禦測試套件。
"""
 
import pytest
import os
import tempfile
from mcp_path_security import SecurePathResolver
 
@pytest.fixture
def workspace(tmp_path):
    """建立含樣本檔案的測試工作空間。"""
    (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):
        # 建立指向工作空間外的 symlink
        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):
        # 建立深層巢狀路徑
        deep_path = "a/b/c/d/e/f/g/h/i/j/deep.txt"
        result = resolver.validate(deep_path)
        assert not result.safe  # 超過 max_depth 5
 
    def test_absolute_path_normalized(self, resolver):
        result = resolver.validate("/etc/passwd")
        assert not result.safe