MCP协议的安全机制:为什么AI工具调用需要护栏

SELECT*FROMordersWHEREcustomer_name='张总';

┌─────────────────────────────────────────────┐
│           MCP Host（AI应用层）               │
│                                             │
│   ┌─────────┐  ┌─────────┐  ┌──────────┐   │
│   │  Tool A │  │  Tool B │  │  Tool C  │   │
│   │ (读文件) │  │ (查数据库)│  │ (执行命令) │   │
│   └─────────┘  └─────────┘  └──────────┘   │
│         ↑           ↑           ↑           │
│    权限：只读    权限：查询    权限：无      │
└─────────────────────────────────────────────┘
                     │
              MCP Protocol
                     │
          ┌──────────┴──────────┐
          │   MCP Client        │
          │ (实际执行层)         │
          │ 真实文件系统/数据库   │
          └─────────────────────┘

defexecute_tool(tool_name: str, params: dict, capabilities: CapabilitySet):
    tool = registry.get(tool_name)

    if not capabilities.contains(tool.required_capability):
        raise SecurityError(f"No capability to use {tool_name}")

    validated_params = tool.schema.validate(params)

    if tool.operation_type == "read" and not capabilities.can_read():
        raise SecurityError("Read not permitted")

    return tool.execute(validated_params)

{
"name":"read_file",
"description":"读取指定路径的文本文件",
"inputSchema":{
"type":"object",
"properties":{
"path":{
"type":"string",
"pattern":"^[a-zA-Z0-9/_.-]+$",
"maxLength":255,
"description":"文件路径，仅允许字母数字下划线斜杠和点"
}
},
"required":["path"]
}
}

audit_log = {
    "timestamp": "2026-07-23T09:00:00Z",
    "tool": "read_file",
    "params": {"path": "/data/reports/q2.pdf"},
    "result_size": 2048,
    "capabilities_used": ["file:read"],
    "rejected": False
}

frommcp.serverimport MCPServer
frommcp.typesimport Tool, TextResource
frommcp.securityimport Capability, ResourcePolicy
importos

server = MCPServer(
    name="secure-file-search",
    version="1.0.0",

    capabilities=[
        Capability(
            name="file:search",
            description="搜索文档目录下的文本文件",
            operations=["read"],
            resource_scope="/data/docs"
        )
    ]
)

@server.tool(
    name="search_docs",
    description="在文档目录中搜索包含关键词的文件",
    input_schema={
        "type": "object",
        "properties": {
            "keyword": {
                "type": "string",
                "minLength": 2,
                "maxLength": 50
            },
            "extension": {
                "type": "string", 
                "enum": [".txt", ".md", ".json"],
                "default": ".txt"
            }
        },
        "required": ["keyword"]
    }
)
defsearch_docs(keyword: str, extension: str = ".txt"):
"""安全文档搜索——只允许搜索指定目录"""
    base_path = "/data/docs"

    if not os.path.abspath(base_path).startswith("/data/docs"):
        raise PermissionError("Path traversal detected")

    results = []
    for root, dirs, files in os.walk(base_path):
        real_root = os.path.realpath(root)
        if not real_root.startswith("/data/docs"):
            continue

        for file in files:
            if file.endswith(extension):
                filepath = os.path.join(root, file)
                try:
                    with open(filepath, 'r', encoding='utf-8') as f:
                        if keyword in f.read():
                            results.append(filepath)
                except PermissionError:
                    continue  # 静默跳过无权限文件

    return {"files": results, "count": len(results)}

frommcp.clientimport MCPClient

client = MCPClient(
    server_uri="http://localhost:5000",

    policies=[
        ResourcePolicy(
            allowed_tools=["search_docs", "get_file_info"],

            resource_patterns=[
                "file:/data/docs/**",
            ],

            allowed_operations=["read", "search"],
            denied_operations=["write", "delete", "execute"]
        )
    ]
)

try:
    result = client.call_tool("delete_file", {"path": "/data/docs/secret.txt"})
except SecurityError as e:
    print(f"拒绝调用：{e}")

try:
    result = client.call_tool("search_docs", {"keyword": "password", "extension": ".txt"})
    print("不应该成功！")
except SecurityError as e:
    print(f"拦截成功：{e}")

try:
    result = client.call_tool("execute_command", {"cmd": "rm -rf /"})
except SecurityError as e:
    print(f"拦截成功：{e}")

try:
    result = client.call_tool("search_docs", {
        "keyword": "secret",
        "extension": ".txt",
    })
except SecurityError as e:
    print(f"拦截成功：{e}")