MCP 跨客戶端資料外洩：防止工作階段隔離失效

# 易受攻擊模式——1.26.0 前 MCP SDK
# 這是教學與文件中的模式
from mcp.server import Server
 
server = Server("my-server")
 
# 伺服器層狀態——由所有客戶端共享
conversation_history = []
user_files = {}
cached_results = {}
 
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "remember":
        # 此資料對所有已連線客戶端可見
        conversation_history.append(arguments["text"])
        return [{"type": "text", "text": "Remembered."}]
 
    if name == "recall":
        # 任何客戶端可讀取任何其他客戶端的資料
        return [{"type": "text", "text": "\n".join(conversation_history)}]
 
    if name == "save_file":
        # 客戶端 A 儲存的檔案客戶端 B 可存取
        user_files[arguments["name"]] = arguments["content"]
        return [{"type": "text", "text": f"Saved {arguments['name']}"}]
 
    if name == "read_file":
        # 客戶端 B 讀取客戶端 A 的私有檔案
        content = user_files.get(arguments["name"], "Not found")
        return [{"type": "text", "text": content}]

# 1.26.0 前 SDK 內部（簡化）
# 單一 Server 執行個體處理所有工作階段
 
class SseServerTransport:
    def __init__(self, endpoint: str):
        self.endpoint = endpoint
        self._sessions = {}  # session_id -> SSE connection
 
    async def connect_sse(self, scope, receive, send):
        session_id = generate_session_id()
        # 建立新的 SSE 連線，但路由到同一個 server
        connection = SSEConnection(session_id, send)
        self._sessions[session_id] = connection
        # 所有工作階段共享 server.state
        return (ReadStream(connection), WriteStream(connection))

# 已防禦——工作階段隔離的狀態（SDK >= 1.26.0）
from mcp.server import Server
from mcp.server.session import ServerSession
from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Any
import asyncio
 
# 儲存當前工作階段的上下文變數
current_session: ContextVar[str] = ContextVar("current_session")
 
@dataclass
class SessionState:
    """每工作階段狀態容器。"""
    session_id: str
    conversation_history: list[str] = field(default_factory=list)
    user_files: dict[str, str] = field(default_factory=dict)
    cached_results: dict[str, Any] = field(default_factory=dict)
    created_at: float = 0.0
    last_accessed: float = 0.0
 
class SessionIsolatedServer:
    """
    具嚴格工作階段隔離的 MCP 伺服器。
    每個客戶端工作階段擁有自己的狀態容器。
    """
 
    def __init__(self):
        self.server = Server("isolated-server")
        self._sessions: dict[str, SessionState] = {}
        self._lock = asyncio.Lock()
 
        # 註冊工具
        self.server.call_tool()(self._handle_tool_call)
        self.server.list_tools()(self._list_tools)
 
    def _get_session_state(self) -> SessionState:
        """取得當前工作階段的狀態。若無則拋出例外。"""
        session_id = current_session.get()
        state = self._sessions.get(session_id)
        if state is None:
            raise RuntimeError(f"No state for session: {session_id}")
        import time
        state.last_accessed = time.time()
        return state
 
    async def create_session(self, session_id: str) -> SessionState:
        """建立新的隔離工作階段狀態。"""
        import time
        async with self._lock:
            if session_id in self._sessions:
                raise ValueError(f"Session already exists: {session_id}")
            state = SessionState(
                session_id=session_id,
                created_at=time.time(),
                last_accessed=time.time(),
            )
            self._sessions[session_id] = state
            return state
 
    async def destroy_session(self, session_id: str):
        """客戶端中斷連線時清理工作階段狀態。"""
        async with self._lock:
            state = self._sessions.pop(session_id, None)
            if state:
                # 安全清除敏感資料
                state.conversation_history.clear()
                state.user_files.clear()
                state.cached_results.clear()
 
    async def _list_tools(self):
        return [
            {
                "name": "remember",
                "description": "Store text in session memory",
                "inputSchema": {
                    "type": "object",
                    "properties": {"text": {"type": "string"}},
                    "required": ["text"],
                },
            },
            {
                "name": "recall",
                "description": "Retrieve stored session memory",
                "inputSchema": {"type": "object", "properties": {}},
            },
            {
                "name": "save_file",
                "description": "Save a file in the session workspace",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "content": {"type": "string"},
                    },
                    "required": ["name", "content"],
                },
            },
            {
                "name": "read_file",
                "description": "Read a file from the session workspace",
                "inputSchema": {
                    "type": "object",
                    "properties": {"name": {"type": "string"}},
                    "required": ["name"],
                },
            },
        ]
 
    async def _handle_tool_call(self, name: str, arguments: dict):
        # 所有狀態存取皆限縮至當前工作階段
        state = self._get_session_state()
 
        if name == "remember":
            state.conversation_history.append(arguments["text"])
            return [{"type": "text", "text": "Remembered (session-scoped)."}]
 
        if name == "recall":
            # 僅回傳此工作階段的歷史
            return [{"type": "text", "text": "\n".join(state.conversation_history)}]
 
        if name == "save_file":
            state.user_files[arguments["name"]] = arguments["content"]
            return [{"type": "text", "text": f"Saved {arguments['name']} (session-scoped)."}]
 
        if name == "read_file":
            content = state.user_files.get(arguments["name"], "Not found in this session")
            return [{"type": "text", "text": content}]
 
        return [{"type": "text", "text": f"Unknown tool: {name}"}]

"""
確保每次請求都設定工作階段上下文的傳輸包裝器。
"""
 
import uuid
import logging
from contextvars import copy_context
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.requests import Request
from mcp.server.sse import SseServerTransport
 
logger = logging.getLogger("mcp.session")
 
class SessionAwareSseTransport:
    """
    包裝 SseServerTransport 以管理工作階段生命週期，
    並為每個連線設定工作階段上下文。
    """
 
    def __init__(self, server: "SessionIsolatedServer", endpoint: str = "/messages/"):
        self.server = server
        self.sse = SseServerTransport(endpoint)
        self._active_sessions: dict[str, str] = {}  # connection -> session_id
 
    async def handle_sse(self, request: Request):
        """處理新 SSE 連線並建立工作階段。"""
        session_id = str(uuid.uuid4())
 
        # 建立隔離的工作階段狀態
        await self.server.create_session(session_id)
        logger.info("Session created: %s from %s", session_id, request.client.host)
 
        try:
            async with self.sse.connect_sse(
                request.scope, request.receive, request._send
            ) as streams:
                # 為此連線設定工作階段上下文
                current_session.set(session_id)
                await self.server.server.run(
                    streams[0], streams[1],
                    self.server.server.create_initialization_options()
                )
        finally:
            # 中斷連線時清理工作階段狀態
            await self.server.destroy_session(session_id)
            logger.info("Session destroyed: %s", session_id)
 
    def create_app(self) -> Starlette:
        """建立具工作階段感知路由的 Starlette app。"""
        return Starlette(
            routes=[
                Route("/sse", endpoint=self.handle_sse),
                Mount("/messages/", app=self.sse.handle_post_message),
            ]
        )

"""
MCP 伺服器的檔案系統工作階段隔離。
每個工作階段擁有專屬工作空間目錄。
"""
 
import os
import shutil
import tempfile
import hashlib
from pathlib import Path
from contextlib import contextmanager
 
class SessionFilesystem:
    """
    為每個 MCP 工作階段提供隔離的檔案系統存取。
    """
 
    def __init__(self, base_dir: str = "/var/mcp/sessions"):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)
        os.chmod(str(self.base_dir), 0o700)
 
    def create_workspace(self, session_id: str) -> Path:
        """為工作階段建立隔離的工作空間目錄。"""
        # 使用雜湊後的 session ID 作為目錄名（避免路徑注入）
        safe_name = hashlib.sha256(session_id.encode()).hexdigest()[:16]
        workspace = self.base_dir / f"session_{safe_name}"
        workspace.mkdir(exist_ok=True)
        os.chmod(str(workspace), 0o700)
        return workspace
 
    def destroy_workspace(self, session_id: str):
        """安全刪除工作階段的工作空間。"""
        safe_name = hashlib.sha256(session_id.encode()).hexdigest()[:16]
        workspace = self.base_dir / f"session_{safe_name}"
        if workspace.exists():
            # 敏感資料在刪除前先覆寫
            for file_path in workspace.rglob("*"):
                if file_path.is_file():
                    size = file_path.stat().st_size
                    with open(file_path, 'wb') as f:
                        f.write(b'\x00' * size)
            shutil.rmtree(str(workspace))
 
    def resolve_path(self, session_id: str, relative_path: str) -> Path:
        """
        在工作階段工作空間中解析路徑。
        防止路徑穿越攻擊。
        """
        safe_name = hashlib.sha256(session_id.encode()).hexdigest()[:16]
        workspace = self.base_dir / f"session_{safe_name}"
 
        if not workspace.exists():
            raise FileNotFoundError(f"Session workspace does not exist")
 
        # 解析並驗證
        resolved = (workspace / relative_path).resolve()
        if not str(resolved).startswith(str(workspace.resolve())):
            raise PermissionError(
                f"Path traversal detected: {relative_path} resolves outside workspace"
            )
 
        return resolved
 
    def get_usage(self, session_id: str) -> dict:
        """取得工作階段的檔案系統使用統計。"""
        safe_name = hashlib.sha256(session_id.encode()).hexdigest()[:16]
        workspace = self.base_dir / f"session_{safe_name}"
 
        if not workspace.exists():
            return {"exists": False}
 
        total_size = 0
        file_count = 0
        for file_path in workspace.rglob("*"):
            if file_path.is_file():
                total_size += file_path.stat().st_size
                file_count += 1
 
        return {
            "exists": True,
            "total_size_bytes": total_size,
            "file_count": file_count,
            "path": str(workspace),
        }
 
# 配額執行
MAX_SESSION_SIZE_BYTES = 100 * 1024 * 1024  # 每工作階段 100MB
MAX_SESSION_FILES = 1000
 
def enforce_quota(fs: SessionFilesystem, session_id: str) -> bool:
    """檢查工作階段是否在檔案系統配額內。"""
    usage = fs.get_usage(session_id)
    if not usage["exists"]:
        return True
    if usage["total_size_bytes"] > MAX_SESSION_SIZE_BYTES:
        return False
    if usage["file_count"] > MAX_SESSION_FILES:
        return False
    return True

"""
MCP 伺服器的資料庫工作階段隔離。
每個工作階段擁有自己的連線，並帶有資料列層級安全上下文。
"""
 
import asyncpg
import logging
from contextlib import asynccontextmanager
 
logger = logging.getLogger("mcp.db.isolation")
 
class SessionDatabasePool:
    """
    管理每工作階段資料庫連線，搭配資料列層級安全。
    """
 
    def __init__(self, dsn: str, max_connections_per_session: int = 2):
        self.dsn = dsn
        self.max_per_session = max_connections_per_session
        self._pools: dict[str, asyncpg.Pool] = {}
 
    async def get_connection(self, session_id: str):
        """取得限縮至工作階段的資料庫連線。"""
        if session_id not in self._pools:
            # 為此工作階段建立專屬連線池
            self._pools[session_id] = await asyncpg.create_pool(
                self.dsn,
                min_size=1,
                max_size=self.max_per_session,
                command_timeout=30,
            )
 
        pool = self._pools[session_id]
        conn = await pool.acquire()
 
        # 設定工作階段層級變數供資料列層級安全使用
        await conn.execute(
            "SET mcp.session_id = $1", session_id
        )
        # 為安全設為唯讀
        await conn.execute("SET default_transaction_read_only = ON")
 
        return conn
 
    async def release_session(self, session_id: str):
        """釋放工作階段的所有資料庫資源。"""
        pool = self._pools.pop(session_id, None)
        if pool:
            await pool.close()
            logger.info("Database pool closed for session: %s", session_id)
 
# PostgreSQL 資料列層級安全設定
RLS_SETUP_SQL = """
-- 對 MCP 存取的資料表啟用資料列層級安全
ALTER TABLE user_documents ENABLE ROW LEVEL SECURITY;
 
-- 政策：工作階段僅能看到被標記為其 session_id 的資料列
CREATE POLICY mcp_session_isolation ON user_documents
    USING (session_tag = current_setting('mcp.session_id'));
 
-- 建立 MCP 專屬唯讀角色
CREATE ROLE mcp_reader NOLOGIN;
GRANT SELECT ON user_documents TO mcp_reader;
GRANT USAGE ON SCHEMA public TO mcp_reader;
 
-- 工作階段使用者繼承受限角色
CREATE ROLE mcp_session LOGIN PASSWORD 'generated_secure_password';
GRANT mcp_reader TO mcp_session;
"""

"""
偵測 MCP 伺服器潛在跨工作階段資料外洩的監控器。
"""
 
import json
import logging
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, field
 
logger = logging.getLogger("mcp.session.monitor")
 
@dataclass
class SessionAccessEvent:
    """工作階段中資料存取的紀錄。"""
    timestamp: datetime
    session_id: str
    tool_name: str
    resource_key: str  # 檔案名稱、查詢等
    access_type: str   # read、write、delete
    data_size: int = 0
 
class CrossSessionLeakDetector:
    """
    透過追蹤資料存取模式並標記異常，
    偵測 MCP 工作階段間潛在的資料外洩。
    """
 
    def __init__(self, alert_callback=None):
        self.alert_callback = alert_callback or self._default_alert
        # 追蹤每個資源由哪個工作階段寫入
        self._resource_owners: dict[str, str] = {}
        # 追蹤每工作階段的存取模式
        self._session_reads: dict[str, list[SessionAccessEvent]] = defaultdict(list)
        self._session_writes: dict[str, list[SessionAccessEvent]] = defaultdict(list)
 
    def record_access(self, event: SessionAccessEvent):
        """記錄並分析資料存取事件。"""
        if event.access_type == "write":
            self._resource_owners[event.resource_key] = event.session_id
            self._session_writes[event.session_id].append(event)
 
        elif event.access_type == "read":
            self._session_reads[event.session_id].append(event)
 
            # 檢查此工作階段是否在讀取另一工作階段擁有的資料
            owner = self._resource_owners.get(event.resource_key)
            if owner and owner != event.session_id:
                self.alert_callback({
                    "alert_type": "cross_session_data_access",
                    "severity": "critical",
                    "reading_session": event.session_id,
                    "owning_session": owner,
                    "resource": event.resource_key,
                    "tool": event.tool_name,
                    "timestamp": event.timestamp.isoformat(),
                    "description": (
                        f"Session {event.session_id} read resource "
                        f"'{event.resource_key}' owned by session {owner}"
                    ),
                })
 
    def session_ended(self, session_id: str):
        """工作階段結束時清理追蹤資料。"""
        # 移除此工作階段資源的擁有權紀錄
        self._resource_owners = {
            k: v for k, v in self._resource_owners.items()
            if v != session_id
        }
        self._session_reads.pop(session_id, None)
        self._session_writes.pop(session_id, None)
 
    def get_isolation_report(self) -> dict:
        """產生工作階段隔離健康度報表。"""
        return {
            "active_sessions": len(set(
                list(self._session_reads.keys()) +
                list(self._session_writes.keys())
            )),
            "tracked_resources": len(self._resource_owners),
            "total_reads": sum(
                len(events) for events in self._session_reads.values()
            ),
            "total_writes": sum(
                len(events) for events in self._session_writes.values()
            ),
        }
 
    def _default_alert(self, alert: dict):
        logger.critical(json.dumps({
            "event": "mcp_cross_session_leak",
            **alert,
        }))

"""
驗證 MCP 伺服器實作中工作階段隔離的測試套件。
"""
 
import pytest
import asyncio
import aiohttp
import json
 
class TestSessionIsolation:
    """
    驗證資料不會在工作階段間外洩的整合測試。
    對您的 MCP 伺服器部署執行這些測試。
    """
 
    MCP_SERVER_URL = "http://localhost:8080"
 
    async def _create_session(self) -> aiohttp.ClientSession:
        """向 MCP 伺服器建立新客戶端工作階段。"""
        session = aiohttp.ClientSession()
        # 初始化 MCP 連線
        async with session.get(f"{self.MCP_SERVER_URL}/sse") as resp:
            assert resp.status == 200
        return session
 
    async def _call_tool(self, session: aiohttp.ClientSession,
                         tool: str, args: dict, msg_id: int = 1) -> dict:
        """呼叫 MCP 工具並回傳結果。"""
        payload = {
            "jsonrpc": "2.0",
            "method": "tools/call",
            "id": msg_id,
            "params": {"name": tool, "arguments": args},
        }
        async with session.post(
            f"{self.MCP_SERVER_URL}/messages/",
            json=payload
        ) as resp:
            return await resp.json()
 
    @pytest.mark.asyncio
    async def test_session_write_not_visible_to_other_session(self):
        """工作階段 A 寫入的資料不可被工作階段 B 讀取。"""
        session_a = await self._create_session()
        session_b = await self._create_session()
 
        try:
            # 工作階段 A 寫入秘密
            await self._call_tool(session_a, "save_file", {
                "name": "secret.txt",
                "content": "TOP_SECRET_DATA_12345",
            })
 
            # 工作階段 B 嘗試讀取
            result = await self._call_tool(session_b, "read_file", {
                "name": "secret.txt",
            })
 
            # 工作階段 B 不應看到工作階段 A 的資料
            response_text = result["result"]["content"][0]["text"]
            assert "TOP_SECRET_DATA_12345" not in response_text, \
                "CRITICAL: Cross-session data leak detected!"
        finally:
            await session_a.close()
            await session_b.close()
 
    @pytest.mark.asyncio
    async def test_session_memory_not_shared(self):
        """對話歷史必須限縮於工作階段。"""
        session_a = await self._create_session()
        session_b = await self._create_session()
 
        try:
            # 工作階段 A 儲存記憶
            await self._call_tool(session_a, "remember", {
                "text": "My API key is sk-secret-abc123",
            })
 
            # 工作階段 B 喚起
            result = await self._call_tool(session_b, "recall", {})
            response_text = result["result"]["content"][0]["text"]
 
            assert "sk-secret-abc123" not in response_text, \
                "CRITICAL: Cross-session memory leak detected!"
        finally:
            await session_a.close()
            await session_b.close()
 
    @pytest.mark.asyncio
    async def test_session_cleanup_on_disconnect(self):
        """客戶端中斷連線時工作階段資料必須被清理。"""
        session_a = await self._create_session()
 
        # 寫入資料
        await self._call_tool(session_a, "save_file", {
            "name": "temp_data.txt",
            "content": "SENSITIVE_DATA",
        })
 
        # 中斷連線
        await session_a.close()
        await asyncio.sleep(1)  # 允許清理時間
 
        # 以新工作階段連線
        session_c = await self._create_session()
        try:
            result = await self._call_tool(session_c, "read_file", {
                "name": "temp_data.txt",
            })
            response_text = result["result"]["content"][0]["text"]
            assert "SENSITIVE_DATA" not in response_text, \
                "CRITICAL: Stale session data accessible after disconnect!"
        finally:
            await session_c.close()
 
    @pytest.mark.asyncio
    async def test_concurrent_sessions_isolated(self):
        """多個並行工作階段必須維持隔離。"""
        sessions = [await self._create_session() for _ in range(5)]
 
        try:
            # 每個工作階段寫入獨特資料
            for i, session in enumerate(sessions):
                await self._call_tool(session, "save_file", {
                    "name": "identity.txt",
                    "content": f"I am session {i}",
                })
 
            # 每個工作階段只應讀到自己的資料
            for i, session in enumerate(sessions):
                result = await self._call_tool(session, "read_file", {
                    "name": "identity.txt",
                })
                text = result["result"]["content"][0]["text"]
                assert f"I am session {i}" in text, \
                    f"Session {i} got wrong data: {text}"
                # 驗證不會出現其他工作階段的資料
                for j in range(5):
                    if j != i:
                        assert f"I am session {j}" not in text, \
                            f"Session {i} can see session {j}'s data!"
        finally:
            for session in sessions:
                await session.close()

{
    "minimum_safe_versions": {
        "python_mcp_sdk": ">=1.26.0",
        "typescript_mcp_sdk": ">=1.26.0",
        "description": "1.26.0 以下版本預設使用共享伺服器狀態"
    },
    "verification_commands": {
        "python": "pip show mcp | grep Version",
        "npm": "npm list @modelcontextprotocol/sdk | grep sdk"
    },
    "changelog_notes": {
        "1.26.0": [
            "新增工作階段範圍的狀態管理",
            "ServerSession 現在維持每連線狀態",
            "Breaking：server.state 不再跨工作階段共享",
            "新增工作階段生命週期鉤子（on_connect、on_disconnect）"
        ]
    }
}

#!/bin/bash
# check-mcp-sdk-version.sh — 驗證 MCP SDK 是否滿足最低安全需求
set -euo pipefail
 
MIN_VERSION="1.26.0"
 
echo "=== MCP SDK 安全版本檢查 ==="
 
# 檢查 Python SDK
if command -v pip &> /dev/null; then
    PY_VERSION=$(pip show mcp 2>/dev/null | grep "^Version:" | awk '{print $2}' || echo "not installed")
    if [ "$PY_VERSION" = "not installed" ]; then
        echo "[SKIP] Python MCP SDK 未安裝"
    else
        if python3 -c "
from packaging.version import Version
import sys
sys.exit(0 if Version('$PY_VERSION') >= Version('$MIN_VERSION') else 1)
" 2>/dev/null; then
            echo "[PASS] Python MCP SDK: $PY_VERSION (>= $MIN_VERSION)"
        else
            echo "[FAIL] Python MCP SDK: $PY_VERSION (< $MIN_VERSION — 易受 CVE-2026-25536 影響)"
            echo "       請執行：pip install --upgrade mcp>=$MIN_VERSION"
        fi
    fi
fi
 
# 檢查 TypeScript SDK
if command -v npm &> /dev/null; then
    TS_VERSION=$(npm list @modelcontextprotocol/sdk 2>/dev/null | grep sdk | grep -oP '[\d.]+' | head -1 || echo "not installed")
    if [ "$TS_VERSION" = "not installed" ]; then
        echo "[SKIP] TypeScript MCP SDK 未安裝"
    else
        if npx semver -r ">=$MIN_VERSION" "$TS_VERSION" &>/dev/null; then
            echo "[PASS] TypeScript MCP SDK: $TS_VERSION (>= $MIN_VERSION)"
        else
            echo "[FAIL] TypeScript MCP SDK: $TS_VERSION (< $MIN_VERSION — 易受攻擊)"
            echo "       請執行：npm install @modelcontextprotocol/sdk@latest"
        fi
    fi
fi
 
echo "=== 檢查完成 ==="