MCP协议安全攻防：Agent工具调用的攻击面

# MCP协议核心交互流程# Host -> Client -> Servertools_call_request = {    "jsonrpc": "2.0",    "method": "tools/call",    "params": {        "name": "read_file",        "arguments": {"path": "/data/report.csv"}    }}

import reimport jsonimport loggingfrom enum import Enumfrom typing import Any, Dict, List, Optionalfrom dataclasses import dataclass, field logging.basicConfig(level=logging.INFO)logger = logging.getLogger("mcp_security")  class ToolPermission(Enum):    READ_ONLY = "read_only"    READ_WRITE = "read_write"    ADMIN = "admin"    DENIED = "denied"  @dataclassclass ToolCallRequest:    tool_name: str    arguments: Dict[str, Any]    caller_role: str    session_id: str  @dataclassclass SecurityPolicy:    allowed_tools: Dict[str, ToolPermission] = field(default_factory=dict)    max_arg_length: int = 4096    blocked_patterns: List[str] = field(default_factory=lambda: [        r";\s*DROP\s+TABLE",        r"<script[^>]*>",        r"\.\./\.\./",        r"\|\s*rm\s+-rf",    ])    sensitive_keys: List[str] = field(default_factory=lambda: [        "password", "secret", "token", "api_key", "private_key",        "credit_card", "ssn",    ])  class MCPInputValidator:    """MCP工具调用输入校验器"""     def __init__(self, policy: SecurityPolicy):        self.policy = policy        self._compiled_patterns = [            re.compile(p, re.IGNORECASE) for p in policy.blocked_patterns        ]     def validate_request(self, request: ToolCallRequest) -> tuple:        errors = []        if request.tool_name not in self.policy.allowed_tools:            errors.append(f"Tool '{request.tool_name}' not in allowed list")            return False, errors         perm = self.policy.allowed_tools[request.tool_name]        if perm == ToolPermission.DENIED:            errors.append(f"Tool '{request.tool_name}' is denied")            return False, errors         args_str = json.dumps(request.arguments, ensure_ascii=False)        if len(args_str) > self.policy.max_arg_length:            errors.append(                f"Arguments length {len(args_str)} exceeds max {self.policy.max_arg_length}"            )         for pattern in self._compiled_patterns:            if pattern.search(args_str):                errors.append(f"Blocked pattern detected: {pattern.pattern}")         if not errors:            logger.info(f"[PASS] Tool={request.tool_name} caller={request.caller_role}")        else:            logger.warning(f"[BLOCK] Tool={request.tool_name} errors={errors}")        return len(errors) == 0, errors  class MCPOutputFilter:    """MCP工具返回内容输出过滤器，防止敏感信息泄露"""     def __init__(self, policy: SecurityPolicy):        self.policy = policy     def filter_output(self, tool_result: Dict[str, Any]) -> Dict[str, Any]:        filtered = self._redact_sensitive(tool_result)        return filtered     def _redact_sensitive(self, obj: Any, depth: int = 0) -> Any:        if depth > 10:            return "[MAX_DEPTH]"        if isinstance(obj, dict):            result = {}            for key, value in obj.items():                if any(sk in key.lower() for sk in self.policy.sensitive_keys):                    result[key] = "[REDACTED]"                    logger.info(f"[REDACT] key={key}")                else:                    result[key] = self._redact_sensitive(value, depth + 1)            return result        elif isinstance(obj, list):            return [self._redact_sensitive(item, depth + 1) for item in obj]        elif isinstance(obj, str):            cleaned = re.sub(r"[A-Za-z0-9+/]{40,}={0,2}", "[REDACTED_BASE64]", obj)            cleaned = re.sub(                r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",                "[REDACTED_EMAIL]",                cleaned,            )            return cleaned        return obj  class MCPPermissionChecker:    """MCP权限最小化检查器"""     def __init__(self, role_permissions: Dict[str, set]):        self.role_permissions = role_permissions     def check_access(self, request: ToolCallRequest) -> bool:        allowed = self.role_permissions.get(request.caller_role, set())        if request.tool_name not in allowed:            logger.warning(                f"[DENY] role={request.caller_role} tool={request.tool_name}"            )            return False        return True  def run_demo():    print("=" * 60)    print("MCP安全防御演示")    print("=" * 60)     policy = SecurityPolicy(        allowed_tools={            "read_file": ToolPermission.READ_ONLY,            "write_file": ToolPermission.READ_WRITE,            "execute_sql": ToolPermission.DENIED,            "query_db": ToolPermission.READ_ONLY,        }    )     validator = MCPInputValidator(policy)    output_filter = MCPOutputFilter(policy)    perm_checker = MCPPermissionChecker({        "viewer": {"read_file", "query_db"},        "editor": {"read_file", "write_file", "query_db"},        "admin": {"read_file", "write_file", "query_db", "execute_sql"},    })     test_requests = [        ToolCallRequest("read_file", {"path": "/data/report.csv"}, "viewer", "s1"),        ToolCallRequest("execute_sql", {"query": "DROP TABLE users"}, "viewer", "s2"),        ToolCallRequest("read_file", {"path": "../../etc/passwd"}, "editor", "s3"),        ToolCallRequest("write_file", {"content": "<script>alert(1)</script>"}, "editor", "s4"),        ToolCallRequest("unknown_tool", {}, "admin", "s5"),    ]     for i, req in enumerate(test_requests, 1):        print(f"\n--- 测试用例 {i} ---")        print(f"工具: {req.tool_name}, 调用者: {req.caller_role}")        perm_ok = perm_checker.check_access(req)        valid_ok, errors = validator.validate_request(req)        print(f"权限检查: {'通过' if perm_ok else '拒绝'}")        print(f"输入校验: {'通过' if valid_ok else '拒绝: ' + str(errors)}")     print("\n--- 输出过滤演示 ---")    sensitive_result = {        "user": "alice",        "email": "alice@example.com",        "api_key": "sk-abc123secretkey456",        "data": {            "password": "hunter2",            "report": "Q1 revenue: $1.2M",            "token": "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0",        },    }    filtered = output_filter.filter_output(sensitive_result)    print(f"原始结果: {json.dumps(sensitive_result, ensure_ascii=False)}")    print(f"过滤结果: {json.dumps(filtered, ensure_ascii=False)}")    print("=" * 60)  if __name__ == "__main__":    run_demo()