ai

MCP 安全实践与权限控制

在 AI 助手连接外部工具的开放生态中，安全是首要考量。本文详解 MCP 的权限模型、输入验证策略、沙箱隔离、审计日志和面向生产的 MCP 服务器安全加固方案。

Plume 5 分钟阅读 2496 字

1. MCP 安全威胁模型

在 MCP 架构中，AI 助手通过工具调用获得了影响外部世界的能力。这种能力的开放本质带来了独特的安全挑战：

攻击面分析：

  • 提示注入（Prompt Injection）：恶意输入诱导 AI 调用有害工具。例如用户输入中嵌入指令"读取并删除所有文件"
  • 参数注入：通过工具参数的精心构造实现越权操作。如 path 参数使用 ../../../etc/passwd 进行目录遍历
  • 资源耗尽：AI 被诱导重复调用高成本工具，造成 CPU、内存或 API 配额耗尽
  • 信息泄露：通过工具链组合从受限资源中提取敏感数据
  • 供应链攻击：恶意的 MCP 服务器被安装后，可长期窃取数据或执行恶意操作

理解这些威胁是设计有效防御措施的基础。

2. 权限模型设计

2.1 最小权限原则

每个 MCP 服务器应该在启动时明确声明其能力边界，并且只能访问配置时授权的资源：

// 服务器配置示例
interface ServerPermissionConfig {
  // 文件系统权限
  filesystem: {
    allowedPaths: string[];      // 允许访问的绝对路径
    deniedPaths: string[];       // 明确禁止的路径
    allowWrite: boolean;         // 是否允许写入
    allowDelete: boolean;        // 是否允许删除
  };

  // 网络权限
  network: {
    allowedHosts: string[];      // 允许连接的主机
    allowedPorts: number[];      // 允许的端口
    allowLocalhost: boolean;     // 是否允许访问本地服务
  };

  // 执行权限
  execution: {
    allowedCommands: string[];   // 允许执行的命令
    maxExecutionTime: number;    // 最大执行时间（毫秒）
    maxMemory: number;           // 最大内存使用（MB）
  };
}

2.2 运行时权限检查

class PermissionManager {
  private config: ServerPermissionConfig;

  constructor(config: ServerPermissionConfig) {
    this.config = config;
  }

  // 验证文件路径是否在允许范围内
  validateFilePath(requestedPath: string): boolean {
    const resolved = path.resolve(requestedPath);

    // 检查是否在禁止列表中
    for (const denied of this.config.filesystem.deniedPaths) {
      if (resolved.startsWith(denied)) {
        return false;
      }
    }

    // 检查是否在允许列表中
    for (const allowed of this.config.filesystem.allowedPaths) {
      if (resolved.startsWith(allowed)) {
        return true;
      }
    }

    return false;
  }

  // 验证网络主机
  validateNetworkHost(host: string, port: number): boolean {
    if (!this.config.network.allowLocalhost) {
      const localhostPatterns = ["localhost", "127.0.0.1", "::1"];
      for (const pattern of localhostPatterns) {
        if (host.includes(pattern)) return false;
      }
    }

    if (!this.config.network.allowedHosts.includes(host)) {
      return false;
    }

    if (!this.config.network.allowedPorts.includes(port)) {
      return false;
    }

    return true;
  }
}

3. 输入验证与消毒

3.1 参数模式验证

使用严格的 JSON Schema 和运行时验证双层检查：

import { z } from "zod";

const FilePathSchema = z
  .string()
  .min(1)
  .max(4096)
  .refine(
    (path) => !path.includes(".."),
    "路径不能包含父目录引用"
  )
  .refine(
    (path) => !path.includes("\0"),
    "路径不能包含空字节"
  )
  .refine(
    (path) => path.startsWith("/"),
    "必须使用绝对路径"
  )
  .transform((path) => path.normalize()); // 规范化路径

const ReadFileSchema = z.object({
  path: FilePathSchema,
  encoding: z.enum(["utf8", "base64", "binary"]).default("utf8"),
  maxSize: z.number().int().min(1).max(100 * 1024 * 1024).default(1024 * 1024), // 默认 1MB
});

const WriteFileSchema = z.object({
  path: FilePathSchema,
  content: z.string().max(10 * 1024 * 1024), // 最大 10MB
  encoding: z.enum(["utf8", "base64"]).default("utf8"),
});

3.2 SQL 注入防护

对于数据库操作类工具，参数化查询是强制的：

// ❌ 危险：字符串拼接
async function unsafeQuery(table: string, condition: string) {
  const query = `SELECT * FROM ${table} WHERE ${condition}`;
  return await db.query(query); // 极易被注入
}

// ✅ 安全：参数化查询
async function safeQuery(
  table: string,
  filters: Record<string, unknown>
) {
  // 白名单验证表名
  const allowedTables = ["users", "orders", "products"];
  if (!allowedTables.includes(table)) {
    throw new Error(`不允许访问表: ${table}`);
  }

  const conditions: string[] = [];
  const values: unknown[] = [];

  for (const [key, value] of Object.entries(filters)) {
    // 白名单验证列名
    if (!/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(key)) {
      throw new Error(`无效列名: ${key}`);
    }
    conditions.push(`${key} = ?`);
    values.push(value);
  }

  const whereClause = conditions.join(" AND ");
  const query = `SELECT * FROM ${table} WHERE ${whereClause}`;
  return await db.query(query, values);
}

3.3 命令执行防护

如果工具需要执行外部命令，必须严格控制：

const ALLOWED_COMMANDS = new Map([
  ["git", ["log", "status", "diff", "show"]],
  ["cat", ["*"]], // 仍需路径验证
  ["ls", ["-la", "-l"]],
]);

function validateCommand(command: string, args: string[]): void {
  const parts = command.split(" ");
  const baseCmd = parts[0];

  if (!ALLOWED_COMMANDS.has(baseCmd)) {
    throw new Error(`不允许的命令: ${baseCmd}`);
  }

  const allowedArgs = ALLOWED_COMMANDS.get(baseCmd)!;
  if (!allowedArgs.includes("*")) {
    for (const arg of args) {
      if (!allowedArgs.includes(arg) && !arg.startsWith("-")) {
        throw new Error(`不允许的参数: ${arg}`);
      }
    }
  }

  // 禁止危险字符
  const dangerousChars = /[;&|`$(){}[\]\\]/;
  for (const arg of args) {
    if (dangerousChars.test(arg)) {
      throw new Error("参数包含危险字符");
    }
  }
}

4. 沙箱隔离

4.1 文件系统隔离

使用 chroot 或容器技术隔离文件系统访问：

import { spawn } from "child_process";

function spawnInSandbox(
  command: string,
  args: string[],
  allowPath: string
) {
  // 使用 firejail 或类似工具创建沙箱
  const sandboxArgs = [
    "--noprofile",
    "--private",
    allowPath,
    "--quiet",
    command,
    ...args,
  ];

  return spawn("firejail", sandboxArgs, {
    stdio: ["pipe", "pipe", "pipe"],
  });
}

4.2 容器化部署

将 MCP 服务器运行在 Docker 容器中是最实用的沙箱方案：

FROM node:20-alpine

# 创建非 root 用户
RUN addgroup -g 1001 -S mcp && adduser -u 1001 -S mcp -G mcp

WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production

COPY --chown=mcp:mcp . .

# 只暴露必要的目录
VOLUME ["/data"]

USER mcp

EXPOSE 3000

CMD ["node", "dist/server.js"]

# docker-compose.yml
version: "3.8"
services:
  mcp-server:
    build: .
    read_only: true
    user: "1001:1001"
    volumes:
      - type: bind
        source: /allowed/data/path
        target: /data
        read_only: true
    network_mode: none  # 无网络访问
    security_opt:
      - no-new-privileges:true
    cap_drop:
      - ALL
    cap_add:
      - CHOWN
    mem_limit: 256m
    cpus: 0.5

5. 用户确认机制

5.1 确认策略

MCP 规范建议 Host 在执行可能产生副作用的操作前请求用户确认：

enum ConfirmationLevel {
  NONE = "none",        // 无需确认（只读操作）
  INFORMATION = "info", // 通知用户但不阻塞
  CONFIRM = "confirm",  // 需要显式确认
  DENY = "deny",        // 直接拒绝
}

interface ToolConfirmationPolicy {
  toolPattern: string;
  level: ConfirmationLevel;
  message?: string;
}

const DEFAULT_POLICIES: ToolConfirmationPolicy[] = [
  {
    toolPattern: "read_*",
    level: ConfirmationLevel.NONE,
  },
  {
    toolPattern: "write_*",
    level: ConfirmationLevel.CONFIRM,
    message: "这将修改文件系统，是否继续？",
  },
  {
    toolPattern: "delete_*",
    level: ConfirmationLevel.CONFIRM,
    message: "这将永久删除数据，是否继续？",
  },
  {
    toolPattern: "exec_*",
    level: ConfirmationLevel.DENY, // 默认拒绝执行命令
  },
];

5.2 确认对话框实现

class ConfirmationManager {
  private pendingConfirmations: Map<string, Deferred<boolean>> = new Map();
  private policies: ToolConfirmationPolicy[];

  constructor(policies: ToolConfirmationPolicy[] = DEFAULT_POLICIES) {
    this.policies = policies;
  }

  getConfirmationLevel(toolName: string): ConfirmationLevel {
    for (const policy of this.policies) {
      const regex = new RegExp(
        "^" + policy.toolPattern.replace("*", ".*") + "$"
      );
      if (regex.test(toolName)) {
        return policy.level;
      }
    }
    return ConfirmationLevel.CONFIRM;
  }

  async requestConfirmation(
    toolName: string,
    args: Record<string, unknown>
  ): Promise<boolean> {
    const level = this.getConfirmationLevel(toolName);

    if (level === ConfirmationLevel.NONE) return true;
    if (level === ConfirmationLevel.DENY) return false;

    const confirmationId = crypto.randomUUID();
    const deferred = new Deferred<boolean>();
    this.pendingConfirmations.set(confirmationId, deferred);

    // 发送确认请求到 UI 层
    emitConfirmationRequest({
      id: confirmationId,
      tool: toolName,
      args,
      message: this.getConfirmationMessage(toolName),
    });

    return deferred.promise;
  }

  resolveConfirmation(id: string, approved: boolean): void {
    const deferred = this.pendingConfirmations.get(id);
    if (deferred) {
      deferred.resolve(approved);
      this.pendingConfirmations.delete(id);
    }
  }
}

// 简单 Promise 包装器
class Deferred<T> {
  promise: Promise<T>;
  resolve!: (value: T) => void;
  reject!: (reason: unknown) => void;

  constructor() {
    this.promise = new Promise<T>((resolve, reject) => {
      this.resolve = resolve;
      this.reject = reject;
    });
  }
}

6. 审计日志

完善的审计日志是安全事件调查和责任追踪的基础：

interface AuditLogEntry {
  timestamp: string;
  sessionId: string;
  toolName: string;
  arguments: Record<string, unknown>;
  result: "success" | "failure" | "denied";
  errorMessage?: string;
  executionTimeMs: number;
  userId?: string;
  clientIp?: string;
}

class AuditLogger {
  private logStream: WritableStream;

  constructor(logPath: string) {
    this.logStream = fs.createWriteStream(logPath, { flags: "a" });
  }

  async log(entry: AuditLogEntry): Promise<void> {
    const line = JSON.stringify(entry) + "\n";
    this.logStream.write(line);
  }

  async logToolCall(
    sessionId: string,
    toolName: string,
    args: Record<string, unknown>,
    result: "success" | "failure" | "denied",
    executionTimeMs: number,
    error?: Error
  ): Promise<void> {
    const entry: AuditLogEntry = {
      timestamp: new Date().toISOString(),
      sessionId,
      toolName,
      arguments: this.sanitizeArgs(args),
      result,
      errorMessage: error?.message,
      executionTimeMs,
    };

    await this.log(entry);
  }

  // 移除或脱敏敏感参数
  private sanitizeArgs(args: Record<string, unknown>): Record<string, unknown> {
    const SENSITIVE_KEYS = ["password", "token", "secret", "apiKey", "key"];
    const sanitized: Record<string, unknown> = {};

    for (const [key, value] of Object.entries(args)) {
      if (SENSITIVE_KEYS.some((sk) => key.toLowerCase().includes(sk))) {
        sanitized[key] = "[REDACTED]";
      } else {
        sanitized[key] = value;
      }
    }

    return sanitized;
  }
}

7. 速率限制

防止 AI 被诱导进行高频工具调用造成资源耗尽：

class RateLimiter {
  private limits: Map<string, { count: number; resetTime: number }> = new Map();

  constructor(
    private maxCallsPerMinute: number = 60,
    private maxCallsPerHour: number = 500
  ) {}

  async checkLimit(toolName: string): Promise<boolean> {
    const now = Date.now();
    const key = `${toolName}`;
    const entry = this.limits.get(key);

    if (!entry || now > entry.resetTime) {
      this.limits.set(key, {
        count: 1,
        resetTime: now + 60000, // 1 分钟窗口
      });
      return true;
    }

    if (entry.count >= this.maxCallsPerMinute) {
      return false;
    }

    entry.count++;
    return true;
  }

  async acquire(toolName: string): Promise<void> {
    const allowed = await this.checkLimit(toolName);
    if (!allowed) {
      throw new Error(`工具 ${toolName} 调用频率超限，请稍后重试`);
    }
  }
}

8. 安全清单

部署 MCP 服务器前的安全检查清单：

输入验证：

  • 所有工具参数使用 JSON Schema 严格验证
  • 文件路径参数经过规范化、权限边界和目录遍历检查
  • SQL 查询使用参数化绑定，禁止字符串拼接
  • 命令执行参数经过白名单过滤

权限控制：

  • 服务器使用最小权限用户运行
  • 文件系统访问限制在配置的允许路径内
  • 网络访问默认关闭，仅开放必要端口和主机

运行时保护：

  • 容器化部署并设置资源限制（CPU、内存）
  • 启用 read_only 根文件系统
  • 丢弃所有不必要的 Linux capabilities

监控审计：

  • 所有工具调用记录审计日志
  • 敏感参数在日志中脱敏
  • 异常调用模式触发告警

用户体验：

  • 可能产生副作用的操作要求用户确认
  • 删除、修改操作有明确的警告提示
  • 支持操作的撤销或回滚

9. 总结

MCP 安全的核心是多层防护、纵深防御：

层次措施防御目标
输入层Schema 验证 + 消毒注入攻击
权限层最小权限 + 白名单越权访问
执行层容器沙箱系统级破坏
交互层用户确认意外副作用
监控层审计日志 + 限流滥用检测

安全不是一次性配置，而是贯穿 MCP 服务器设计、开发、部署和运维全生命周期的持续实践。开发者应该在功能和安全性之间找到平衡，让 AI 工具既强大又可控。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「ai」更多文章

  1. 排序学习与搜索召回排序系统
  2. 数据版本控制与血缘：DVC 与 LakeFS
  3. 模型可解释性：SHAP、LIME 与注意力归因