Skip to content
🔗 分享本题
查看我的学习进度 →

企业MCP身份工具数据与执行环境最小权限审计图解

🧠 图解记忆: 最小权限贯穿身份、工具、数据和执行环境,每次调用都要可追踪。

💡 答案要点

MCP安全分层:

┌─────────────────────────────────────────────────────────┐
│                  MCP 安全分层体系                          │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  Layer 1: 传输层安全                                     │
│    - stdio(本地进程,零网络风险)                        │
│    - HTTPS + TLS(云端通信)                             │
│    - OAuth 2.1(企业认证)                               │
│                                                          │
│  Layer 2: 协议层安全                                     │
│    - 工具注册白名单                                      │
│    - 输入Schema验证                                      │
│    - 输出大小限制                                        │
│                                                          │
│  Layer 3: 应用层安全                                     │
│    - Scope权限控制                                       │
│    - 审计日志                                            │
│    - 限流保护                                            │
│                                                          │
│  Layer 4: 工具执行层                                     │
│    - 沙箱隔离                                            │
│    - 超时控制                                            │
│    - 错误处理                                            │
└─────────────────────────────────────────────────────────┘

Scope权限模型(OAuth 2.1):

python
# MCP Server定义工具的权限Scope
@mcp.tool(
    name="read_user_data",
    scopes=["user:read"],      # 需要user:read权限
    description="读取用户数据"
)
def read_user_data(user_id: str) -> dict:
    return db.query("SELECT * FROM users WHERE id = ?", user_id)

@mcp.tool(
    name="write_user_data",
    scopes=["user:write"],     # 需要user:write权限(更高级别)
    description="写入用户数据"
)
def write_user_data(user_id: str, data: dict) -> bool:
    return db.execute("UPDATE users SET ...", user_id, data)

@mcp.tool(
    name="delete_user",
    scopes=["user:admin"],    # 需要admin权限(最高级别)
    description="删除用户"
)
def delete_user(user_id: str) -> bool:
    return db.execute("DELETE FROM users WHERE id = ?", user_id)

企业级MCP权限配置:

展开 Json 代码示例(32 行)
json
// 企业MCP Server权限配置
{
  "server": {
    "name": "company-mcp-server",
    "auth": {
      "type": "oauth2.1",
      "issuer": "https://auth.company.com",
      "client_id": "company-mcp-server"
    }
  },

  "scopes": {
    "user:read": {
      "description": "读取用户数据",
      "tools": ["read_user_data", "list_users"]
    },
    "user:write": {
      "description": "修改用户数据",
      "tools": ["write_user_data", "reset_password"]
    },
    "user:admin": {
      "description": "管理用户(最高权限)",
      "tools": ["delete_user", "impersonate_user"],
      "requires": ["user:read", "user:write"]
    }
  },

  "rate_limits": {
    "default": {"calls": 100, "window": 60},
    "expensive_operations": {"calls": 10, "window": 60}
  }
}

沙箱隔离(工具执行层):

展开 Python 代码示例(38 行)
python
import subprocess
import tempfile
import os

# 危险工具(如执行Shell命令)需要沙箱隔离
@mcp.tool(
    name="safe_bash",
    description="安全执行bash命令(受限环境)",
    annotations={"dangerousHint": "高风险工具,需沙箱隔离"}
)
def safe_bash(command: str, timeout: int = 30) -> dict:
    # 白名单允许的命令
    ALLOWED_COMMANDS = ["grep", "awk", "sed", "find", "ls", "cat"]

    # 解析命令
    cmd_parts = command.strip().split()
    base_cmd = cmd_parts[0]

    if base_cmd not in ALLOWED_COMMANDS:
        raise ValueError(f"Command '{base_cmd}' not allowed. Use: {ALLOWED_COMMANDS}")

    # 超时控制
    try:
        result = subprocess.run(
            cmd_parts,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=tempfile.gettempdir(),  # 工作目录限制
            env={"PATH": "/usr/bin:/bin"}  # PATH限制
        )
        return {
            "returncode": result.returncode,
            "stdout": result.stdout[:10000],  # 输出截断
            "stderr": result.stderr[:1000]
        }
    except subprocess.TimeoutExpired:
        return {"error": f"Command timed out after {timeout}s"}

审计日志:

展开 Python 代码示例(30 行)
python
import logging
from datetime import datetime

audit_log = logging.getLogger("mcp.audit")

@mcp.tool()
def audited_tool(user_id: str, action: str, **kwargs):
    """带审计日志的工具"""

    # 记录调用
    audit_log.info({
        "timestamp": datetime.utcnow().isoformat(),
        "user_id": user_id,
        "tool": "audited_tool",
        "action": action,
        "args": {k: v for k, v in kwargs.items() if k != "sensitive"},
        "client_ip": request.headers.get("X-Forwarded-For"),
        "user_agent": request.headers.get("User-Agent")
    })

    # 执行实际操作
    result = actual_implementation(action, **kwargs)

    # 记录响应
    audit_log.info({
        "timestamp": datetime.utcnow().isoformat(),
        "result": "success" if not result.get("error") else "failure"
    })

    return result

面试话术:

"MCP安全机制是四层防护:传输层(stdio本地/HTTPS云端)、协议层(Schema验证+输出限制)、应用层(OAuth Scope权限)、执行层(沙箱+超时)。企业最看重的是OAuth 2.1+Scope权限——不同AI应用/用户拿到的Token Scope不同,能读数据的不能写,能写的不能删。审计日志记录每次调用,做合规追溯。沙箱隔离确保危险工具(如bash执行)不会破坏系统。"

📚 参考:MCP 官方规范:Security Best Practices