08e - MCP 安全最佳实践

// MCP Client 端：OAuth 2.1 + PKCE 认证流程
import crypto from 'crypto';
 
class MCPOAuthClient {
  private authServerUrl: string;
  private clientId: string;
  private redirectUri: string;
 
  constructor(config: {
    authServerUrl: string;
    clientId: string;
    redirectUri: string;
  }) {
    this.authServerUrl = config.authServerUrl;
    this.clientId = config.clientId;
    this.redirectUri = config.redirectUri;
  }
 
  // 步骤 1：生成 PKCE code_verifier 和 code_challenge
  generatePKCE(): { verifier: string; challenge: string } {
    const verifier = crypto.randomBytes(32)
      .toString('base64url');
    const challenge = crypto
      .createHash('sha256')
      .update(verifier)
      .digest('base64url');
    return { verifier, challenge };
  }
 
  // 步骤 2：构建授权 URL（含 Resource Indicator）
  buildAuthUrl(mcpServerUrl: string): {
    url: string;
    pkce: { verifier: string; challenge: string };
    state: string;
  } {
    const pkce = this.generatePKCE();
    const state = crypto.randomBytes(16).toString('hex');
 
    const params = new URLSearchParams({
      response_type: 'code',
      client_id: this.clientId,
      redirect_uri: this.redirectUri,
      code_challenge: pkce.challenge,
      code_challenge_method: 'S256',
      state,
      // RFC 8707: Resource Indicator — 防止 token 被用于非目标服务器
      resource: mcpServerUrl,
      scope: 'mcp:tools:read mcp:tools:execute mcp:resources:read',
    });
 
    return {
      url: `${this.authServerUrl}/authorize?${params}`,
      pkce,
      state,
    };
  }
 
  // 步骤 3：用授权码换取 access_token
  async exchangeCode(
    code: string,
    pkceVerifier: string,
    mcpServerUrl: string
  ): Promise<{ accessToken: string; refreshToken: string; expiresIn: number }> {
    const response = await fetch(`${this.authServerUrl}/token`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
      body: new URLSearchParams({
        grant_type: 'authorization_code',
        code,
        client_id: this.clientId,
        redirect_uri: this.redirectUri,
        code_verifier: pkceVerifier,  // PKCE 验证
        resource: mcpServerUrl,       // Resource Indicator
      }),
    });
 
    if (!response.ok) {
      throw new Error(`Token exchange failed: ${response.status}`);
    }
    return response.json();
  }
}

// MCP Server 端：验证 Bearer Token
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
import express from 'express';
import jwt from 'jsonwebtoken';
 
const app = express();
 
// Token 验证中间件
function validateMCPToken(req: express.Request, res: express.Response, next: express.NextFunction) {
  const authHeader = req.headers['authorization'];
  if (!authHeader?.startsWith('Bearer ')) {
    return res.status(401).json({
      error: 'missing_token',
      message: 'Authorization header with Bearer token required',
    });
  }
 
  const token = authHeader.slice(7);
 
  try {
    const decoded = jwt.verify(token, process.env.JWT_PUBLIC_KEY!, {
      algorithms: ['RS256'],
      // 验证 audience 匹配当前 MCP Server
      audience: process.env.MCP_SERVER_URL,
      issuer: process.env.OAUTH_ISSUER,
    });
 
    // 将用户身份附加到请求上下文
    (req as any).mcpIdentity = {
      sub: (decoded as any).sub,
      roles: (decoded as any).roles || [],
      scopes: (decoded as any).scope?.split(' ') || [],
      exp: (decoded as any).exp,
    };
 
    next();
  } catch (err) {
    return res.status(401).json({
      error: 'invalid_token',
      message: err instanceof Error ? err.message : 'Token validation failed',
    });
  }
}
 
// 应用到 MCP 端点
app.use('/sse', validateMCPToken);
app.use('/messages', validateMCPToken);

你是一个 MCP 安全审计助手。请审查以下 MCP Server 的认证配置：

[粘贴 MCP Server 代码或配置]

请检查以下安全要点：
1. 是否强制使用 PKCE（code_challenge_method: S256）？
2. Token 是否验证了 audience（防止 token 跨服务器滥用）？
3. 是否实现了 token 过期和刷新机制？
4. 是否使用了 Resource Indicators（RFC 8707）？
5. 敏感凭证是否通过环境变量或密钥管理器存储？

对每个要点给出 ✅ 通过 / ❌ 未通过 / ⚠️ 需改进 的评估，并提供修复建议。

┌─────────────────────────────────────────────────┐
│                  MCP RBAC 层次                    │
├─────────────────────────────────────────────────┤
│  Level 1: 服务器级别                              │
│  ├── 哪些 Agent/用户可以连接此 MCP Server？        │
│                                                   │
│  Level 2: 工具级别                                │
│  ├── 连接后可以看到/调用哪些工具？                  │
│                                                   │
│  Level 3: 参数级别                                │
│  ├── 调用工具时可以传递哪些参数值？                  │
│                                                   │
│  Level 4: 数据级别                                │
│  └── 工具返回的数据中哪些字段可见？                  │
└─────────────────────────────────────────────────┘

// 角色-工具权限矩阵定义
interface MCPRole {
  name: string;
  description: string;
  tools: {
    [toolName: string]: {
      allowed: boolean;
      paramConstraints?: Record<string, any>;  // 参数级约束
      dataFilter?: string[];                    // 可见字段白名单
    };
  };
}
 
const ROLES: Record<string, MCPRole> = {
  viewer: {
    name: 'viewer',
    description: '只读访问，仅可查询数据',
    tools: {
      'db_query': {
        allowed: true,
        paramConstraints: {
          query_type: ['SELECT'],  // 只允许 SELECT
          tables: ['public.*'],    // 只允许公开表
        },
      },
      'file_read': { allowed: true },
      'file_write': { allowed: false },
      'db_execute': { allowed: false },
    },
  },
  developer: {
    name: 'developer',
    description: '开发环境完整访问',
    tools: {
      'db_query': { allowed: true },
      'file_read': { allowed: true },
      'file_write': {
        allowed: true,
        paramConstraints: {
          path_pattern: ['^/workspace/.*'],  // 只允许工作区
        },
      },
      'db_execute': {
        allowed: true,
        paramConstraints: {
          environment: ['development', 'staging'],  // 禁止生产环境
        },
      },
    },
  },
  admin: {
    name: 'admin',
    description: '完整管理权限',
    tools: {
      'db_query': { allowed: true },
      'file_read': { allowed: true },
      'file_write': { allowed: true },
      'db_execute': { allowed: true },
      'server_config': { allowed: true },
    },
  },
};

// 工具调用前的 RBAC 检查
function createRBACMiddleware(roles: Record<string, MCPRole>) {
  return function checkToolAccess(
    identity: { roles: string[] },
    toolName: string,
    params: Record<string, any>
  ): { allowed: boolean; reason?: string } {
    // 合并用户所有角色的权限（取并集）
    for (const roleName of identity.roles) {
      const role = roles[roleName];
      if (!role) continue;
 
      const toolPerm = role.tools[toolName];
      if (!toolPerm?.allowed) continue;
 
      // 检查参数级约束
      if (toolPerm.paramConstraints) {
        for (const [param, allowedValues] of Object.entries(toolPerm.paramConstraints)) {
          const actualValue = params[param];
          if (actualValue && !matchesConstraint(actualValue, allowedValues as string[])) {
            return {
              allowed: false,
              reason: `Parameter '${param}' value '${actualValue}' not allowed for role '${roleName}'`,
            };
          }
        }
      }
 
      return { allowed: true };
    }
 
    return {
      allowed: false,
      reason: `No role grants access to tool '${toolName}'`,
    };
  };
}
 
function matchesConstraint(value: string, patterns: string[]): boolean {
  return patterns.some(pattern => {
    if (pattern.includes('*')) {
      const regex = new RegExp('^' + pattern.replace(/\*/g, '.*') + '$');
      return regex.test(value);
    }
    return value === pattern;
  });
}

# Nginx 反向代理配置：强制 TLS 1.3 for MCP Server
server {
    listen 443 ssl http2;
    server_name mcp.example.com;
 
    # TLS 1.3 only
    ssl_protocols TLSv1.3;
    ssl_ciphers TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256;
    ssl_prefer_server_ciphers off;
 
    # 证书配置
    ssl_certificate /etc/ssl/certs/mcp-server.crt;
    ssl_certificate_key /etc/ssl/private/mcp-server.key;
 
    # HSTS（强制浏览器使用 HTTPS）
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains" always;
 
    # 安全头
    add_header X-Content-Type-Options nosniff;
    add_header X-Frame-Options DENY;
 
    location / {
        proxy_pass http://localhost:3001;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-Proto $scheme;
 
        # SSE 特殊配置
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 86400s;
    }
}

// 敏感配置加密存储示例
import crypto from 'crypto';
 
class MCPSecretStore {
  private encryptionKey: Buffer;
 
  constructor(masterKey: string) {
    // 从主密钥派生加密密钥
    this.encryptionKey = crypto.scryptSync(masterKey, 'mcp-salt', 32);
  }
 
  encrypt(plaintext: string): string {
    const iv = crypto.randomBytes(16);
    const cipher = crypto.createCipheriv('aes-256-gcm', this.encryptionKey, iv);
    const encrypted = Buffer.concat([
      cipher.update(plaintext, 'utf8'),
      cipher.final(),
    ]);
    const authTag = cipher.getAuthTag();
    // 格式：iv:authTag:ciphertext（全部 base64）
    return `${iv.toString('base64')}:${authTag.toString('base64')}:${encrypted.toString('base64')}`;
  }
 
  decrypt(ciphertext: string): string {
    const [ivB64, tagB64, dataB64] = ciphertext.split(':');
    const iv = Buffer.from(ivB64, 'base64');
    const authTag = Buffer.from(tagB64, 'base64');
    const data = Buffer.from(dataB64, 'base64');
    const decipher = crypto.createDecipheriv('aes-256-gcm', this.encryptionKey, iv);
    decipher.setAuthTag(authTag);
    return decipher.update(data) + decipher.final('utf8');
  }
}
 
// 使用示例：加密存储 MCP Server 的数据库凭证
const store = new MCPSecretStore(process.env.MASTER_KEY!);
const encryptedDbUrl = store.encrypt('postgresql://user:pass@host:5432/db');
// 存储 encryptedDbUrl 到配置文件，运行时解密

请审查以下 MCP Server 的数据加密配置：

[粘贴配置或代码]

检查要点：
1. 传输层是否强制 TLS 1.3？是否禁用了 TLS 1.0/1.1？
2. 敏感数据（API Key、数据库凭证）是否加密存储？
3. 加密算法是否使用 AES-256-GCM 或 ChaCha20-Poly1305？
4. 密钥是否通过环境变量或密钥管理器（如 AWS KMS、HashiCorp Vault）管理？
5. 日志中是否存在敏感信息明文输出？

请给出安全评级（A/B/C/D）和改进建议。

用户输入 → [AI 模型] → 工具调用参数 → [输入验证层] → MCP Server 执行
     ↑                        ↑                    ↑
  直接注入              间接注入/XPIA          最后防线

import { z } from 'zod';
 
// 层 1：Schema 验证（结构正确性）
const FileReadSchema = z.object({
  path: z.string()
    .min(1)
    .max(500)
    .regex(/^[a-zA-Z0-9_\-\/\.]+$/, '路径只允许字母数字和 /_-. 字符'),
  encoding: z.enum(['utf-8', 'ascii', 'base64']).default('utf-8'),
});
 
const DbQuerySchema = z.object({
  query: z.string().min(1).max(2000),
  params: z.array(z.unknown()).max(20).optional(),
});
 
// 层 2：语义验证（业务逻辑安全）
function validateFilePath(path: string): { safe: boolean; reason?: string } {
  // 防止路径遍历
  if (path.includes('..') || path.includes('~')) {
    return { safe: false, reason: '路径遍历攻击：包含 .. 或 ~' };
  }
  // 限制访问范围
  const allowedPrefixes = ['/workspace/', '/tmp/mcp/'];
  if (!allowedPrefixes.some(prefix => path.startsWith(prefix))) {
    return { safe: false, reason: `路径不在允许范围内：${allowedPrefixes.join(', ')}` };
  }
  // 禁止敏感文件
  const blockedPatterns = [/\.env$/, /\.ssh\//, /\.aws\//, /password/i, /secret/i];
  if (blockedPatterns.some(p => p.test(path))) {
    return { safe: false, reason: '尝试访问敏感文件' };
  }
  return { safe: true };
}
 
// 层 3：SQL 注入防御
function validateSQLQuery(query: string): { safe: boolean; reason?: string } {
  const upperQuery = query.toUpperCase().trim();
  // 只允许 SELECT 语句
  if (!upperQuery.startsWith('SELECT')) {
    return { safe: false, reason: '只允许 SELECT 查询' };
  }
  // 检测危险关键字
  const dangerous = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'ALTER',
                     'EXEC', 'EXECUTE', 'UNION', '--', '/*', 'xp_'];
  for (const keyword of dangerous) {
    if (upperQuery.includes(keyword)) {
      return { safe: false, reason: `检测到危险关键字：${keyword}` };
    }
  }
  return { safe: true };
}

import crypto from 'crypto';
 
interface ToolDefinition {
  name: string;
  description: string;
  inputSchema: Record<string, any>;
}
 
class ToolIntegrityVerifier {
  private signingKey: string;
 
  constructor(signingKey: string) {
    this.signingKey = signingKey;
  }
 
  // 为工具定义生成签名
  signTool(tool: ToolDefinition): string {
    const canonical = JSON.stringify({
      name: tool.name,
      description: tool.description,
      inputSchema: tool.inputSchema,
    });
    return crypto
      .createHmac('sha256', this.signingKey)
      .update(canonical)
      .digest('hex');
  }
 
  // 验证工具定义未被篡改
  verifyTool(tool: ToolDefinition, expectedSignature: string): boolean {
    const actualSignature = this.signTool(tool);
    return crypto.timingSafeEqual(
      Buffer.from(actualSignature, 'hex'),
      Buffer.from(expectedSignature, 'hex')
    );
  }
}
 
// 在 MCP Server 启动时签名所有工具，运行时验证
const verifier = new ToolIntegrityVerifier(process.env.TOOL_SIGNING_KEY!);

┌──────────┐     ┌─────────────────────────────────┐     ┌──────────────┐
│  Agent A │────▶│                                 │────▶│ MCP Server 1 │
├──────────┤     │        MCP Gateway              │     ├──────────────┤
│  Agent B │────▶│                                 │────▶│ MCP Server 2 │
├──────────┤     │  ┌─────────┐ ┌──────────────┐  │     ├──────────────┤
│  Agent C │────▶│  │ 认证/授权│ │ 工具 ACL 过滤 │  │────▶│ MCP Server 3 │
└──────────┘     │  └─────────┘ └──────────────┘  │     └──────────────┘
                 │  ┌─────────┐ ┌──────────────┐  │
                 │  │ 审计日志 │ │ 速率限制     │  │
                 │  └─────────┘ └──────────────┘  │
                 └─────────────────────────────────┘

// MCP Gateway OBO 认证流程
class MCPGateway {
  private tokenExchangeUrl: string;
 
  constructor(config: { tokenExchangeUrl: string }) {
    this.tokenExchangeUrl = config.tokenExchangeUrl;
  }
 
  // 将用户的 access_token 交换为目标 MCP Server 的 OBO token
  async exchangeForOBOToken(
    userToken: string,
    targetMCPServer: string
  ): Promise<string> {
    const response = await fetch(this.tokenExchangeUrl, {
      method: 'POST',
      headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
      body: new URLSearchParams({
        grant_type: 'urn:ietf:params:oauth:grant-type:token-exchange',
        subject_token: userToken,
        subject_token_type: 'urn:ietf:params:oauth:token-type:access_token',
        requested_token_type: 'urn:ietf:params:oauth:token-type:access_token',
        resource: targetMCPServer,  // 目标 MCP Server
        scope: 'mcp:tools:execute',
      }),
    });
 
    if (!response.ok) {
      throw new Error(`OBO token exchange failed: ${response.status}`);
    }
 
    const data = await response.json();
    return data.access_token;
  }
 
  // 转发工具调用请求，使用 OBO token
  async forwardToolCall(
    userToken: string,
    targetServer: string,
    toolName: string,
    params: Record<string, any>
  ) {
    const oboToken = await this.exchangeForOBOToken(userToken, targetServer);
 
    return fetch(`${targetServer}/messages`, {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${oboToken}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        jsonrpc: '2.0',
        method: 'tools/call',
        params: { name: toolName, arguments: params },
        id: crypto.randomUUID(),
      }),
    });
  }
}

# mcp-gateway-config.yaml — 工具级 ACL 配置
gateway:
  listen: 0.0.0.0:8443
  tls:
    cert: /etc/ssl/mcp-gateway.crt
    key: /etc/ssl/mcp-gateway.key
 
# 后端 MCP Server 注册
servers:
  - name: database-server
    url: http://mcp-db:3001
    health_check: /health
  - name: filesystem-server
    url: http://mcp-fs:3002
    health_check: /health
 
# 工具级 ACL 规则
acl:
  default_policy: deny  # 默认拒绝所有工具访问
 
  rules:
    - role: data-analyst
      servers:
        - name: database-server
          tools:
            - name: query_database
              allow: true
              constraints:
                query_type: [SELECT]
            - name: modify_database
              allow: false
        - name: filesystem-server
          tools:
            - name: read_file
              allow: true
              constraints:
                path_prefix: /reports/
            - name: write_file
              allow: false
 
    - role: developer
      servers:
        - name: database-server
          tools: [allow_all]
        - name: filesystem-server
          tools: [allow_all]
 
# 速率限制
rate_limits:
  per_user:
    requests_per_minute: 60
    requests_per_hour: 1000
  per_tool:
    query_database:
      requests_per_minute: 30
    modify_database:
      requests_per_minute: 5

interface MCPAuditEvent {
  timestamp: string;
  eventType: 'tool_call' | 'tool_result' | 'auth_success' | 'auth_failure'
    | 'access_denied' | 'rate_limited' | 'session_start' | 'session_end';
  sessionId: string;
  identity: {
    userId: string;
    roles: string[];
    clientId: string;
  };
  tool?: {
    name: string;
    server: string;
    params: Record<string, any>;  // 脱敏后的参数
  };
  result?: {
    status: 'success' | 'error' | 'denied';
    durationMs: number;
    errorCode?: string;
  };
  context: {
    sourceIp: string;
    userAgent: string;
    traceId: string;
  };
}
 
class MCPAuditLogger {
  private logStream: NodeJS.WritableStream;
 
  constructor(logPath: string) {
    const fs = require('fs');
    this.logStream = fs.createWriteStream(logPath, { flags: 'a' });
  }
 
  // 记录工具调用事件
  logToolCall(event: Omit<MCPAuditEvent, 'timestamp'>) {
    const auditEvent: MCPAuditEvent = {
      ...event,
      timestamp: new Date().toISOString(),
    };
 
    // 脱敏处理：移除敏感参数值
    if (auditEvent.tool?.params) {
      auditEvent.tool.params = this.sanitizeParams(auditEvent.tool.params);
    }
 
    this.logStream.write(JSON.stringify(auditEvent) + '\n');
  }
 
  private sanitizeParams(params: Record<string, any>): Record<string, any> {
    const sensitiveKeys = ['password', 'secret', 'token', 'key', 'credential'];
    const sanitized = { ...params };
    for (const key of Object.keys(sanitized)) {
      if (sensitiveKeys.some(s => key.toLowerCase().includes(s))) {
        sanitized[key] = '[REDACTED]';
      }
    }
    return sanitized;
  }
}

# Grafana Loki + Promtail 配置：MCP 审计日志聚合
# promtail-config.yaml
server:
  http_listen_port: 9080
 
positions:
  filename: /tmp/positions.yaml
 
clients:
  - url: http://loki:3100/loki/api/v1/push
 
scrape_configs:
  - job_name: mcp-audit
    static_configs:
      - targets: [localhost]
        labels:
          job: mcp-audit
          __path__: /var/log/mcp/audit/*.jsonl
    pipeline_stages:
      - json:
          expressions:
            eventType: eventType
            toolName: tool.name
            resultStatus: result.status
            userId: identity.userId
      - labels:
          eventType:
          toolName:
          resultStatus:
          userId:

# Grafana 告警规则：MCP 安全异常检测
# mcp-alert-rules.yaml
groups:
  - name: mcp-security-alerts
    rules:
      # 告警：认证失败率过高
      - alert: MCPAuthFailureSpike
        expr: |
          rate(mcp_auth_failures_total[5m]) > 10
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "MCP 认证失败率异常升高"
          description: "过去 5 分钟内认证失败超过 10 次/分钟"
 
      # 告警：工具调用被拒绝率过高
      - alert: MCPAccessDeniedSpike
        expr: |
          rate(mcp_access_denied_total[5m]) > 5
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "MCP 工具访问拒绝率异常"
 
      # 告警：单用户调用量异常
      - alert: MCPUserAnomalyDetected
        expr: |
          rate(mcp_tool_calls_total[10m]) by (userId) > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "用户 {{ $labels.userId }} 工具调用量异常"

第 1 天：部署 MCP Gateway（Lasso 开源版）
  ├── 配置 TLS 终止
  ├── 接入 Auth0 作为 OAuth 2.1 提供商
  └── 启用审计日志

第 2 天：实施 RBAC
  ├── 定义 3 个角色：analyst（只读）、developer（读写）、admin（全部）
  ├── 配置工具级 ACL
  └── 数据库工具限制 analyst 只能 SELECT

第 3 天：输入验证加固
  ├── 为每个工具添加 Zod Schema 验证
  ├── 实现路径遍历防护
  └── SQL 查询参数化强制

第 4 天：监控与告警
  ├── 部署 Grafana + Loki 日志聚合
  ├── 配置认证失败告警
  └── 配置异常调用量告警

第 5 天：安全测试
  ├── 工具投毒模拟测试
  ├── 路径遍历渗透测试
  └── Token 过期和刷新测试