MCP-安全（代码实例）

安全最佳实践

介绍

学习目标

身份验证和授权

.NET 身份集成

public class SecureMcpStartup
{public void ConfigureServices(IServiceCollection services){// Add ASP.NET Core Identityservices.AddIdentity<ApplicationUser, IdentityRole>().AddEntityFrameworkStores<ApplicationDbContext>().AddDefaultTokenProviders();// Configure JWT authenticationservices.AddAuthentication(options =>{options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;options.DefaultChallengeScheme = JwtBearerDefaults.AuthenticationScheme;}).AddJwtBearer(options =>{options.TokenValidationParameters = new TokenValidationParameters{ValidateIssuer = true,ValidateAudience = true,ValidateLifetime = true,ValidateIssuerSigningKey = true,ValidIssuer = Configuration["Jwt:Issuer"],ValidAudience = Configuration["Jwt:Audience"],IssuerSigningKey = new SymmetricSecurityKey(Encoding.UTF8.GetBytes(Configuration["Jwt:Key"]))};});// Add authorization policiesservices.AddAuthorization(options =>{options.AddPolicy("CanUseAdminTools", policy =>policy.RequireRole("Admin"));options.AddPolicy("CanUseBasicTools", policy =>policy.RequireAuthenticatedUser());});// Configure MCP server with securityservices.AddMcpServer(options =>{options.ServerName = "Secure MCP Server";options.ServerVersion = "1.0.0";options.RequireAuthentication = true;});// Register tools with authorization requirementsservices.AddMcpTool<BasicTool>(options => options.RequirePolicy("CanUseBasicTools"));services.AddMcpTool<AdminTool>(options => options.RequirePolicy("CanUseAdminTools"));}public void Configure(IApplicationBuilder app){// Use authentication and authorizationapp.UseAuthentication();app.UseAuthorization();// Use MCP server middlewareapp.UseMcpServer();}
}

Java Spring 安全集成

@Configuration
@EnableWebSecurity
public class SecurityConfig extends WebSecurityConfigurerAdapter {@Overrideprotected void configure(HttpSecurity http) throws Exception {http.csrf().disable().authorizeRequests().antMatchers("/mcp/discovery").permitAll() // Allow tool discovery.antMatchers("/mcp/tools/**").hasAnyRole("USER", "ADMIN") // Require authentication for tools.antMatchers("/mcp/admin/**").hasRole("ADMIN") // Admin-only endpoints.anyRequest().authenticated().and().oauth2ResourceServer().jwt();}@Beanpublic McpSecurityInterceptor mcpSecurityInterceptor() {return new McpSecurityInterceptor();}
}// MCP Security Interceptor for tool authorization
public class McpSecurityInterceptor implements ToolExecutionInterceptor {@Autowiredprivate JwtDecoder jwtDecoder;@Overridepublic void beforeToolExecution(ToolRequest request, Authentication authentication) {String toolName = request.getToolName();// Check if user has permissions for this toolif (toolName.startsWith("admin") && !authentication.getAuthorities().contains("ROLE_ADMIN")) {throw new AccessDeniedException("You don't have permission to use this tool");}// Additional security checks based on tool or parametersif ("sensitiveDataAccess".equals(toolName)) {validateDataAccessPermissions(request, authentication);}}private void validateDataAccessPermissions(ToolRequest request, Authentication auth) {// Implementation to check fine-grained data access permissions}
}

数据保护和隐私

Python 数据保护示例

from mcp_server import McpServer
from mcp_tools import Tool, ToolRequest, ToolResponse
from cryptography.fernet import Fernet
import os
import json
from functools import wraps# PII Detector - identifies and protects sensitive information
class PiiDetector:def __init__(self):# Load patterns for different types of PIIwith open("pii_patterns.json", "r") as f:self.patterns = json.load(f)def scan_text(self, text):"""Scans text for PII and returns detected PII types"""detected_pii = []# Implementation to detect PII using regex or ML modelsreturn detected_piidef scan_parameters(self, parameters):"""Scans request parameters for PII"""detected_pii = []for key, value in parameters.items():if isinstance(value, str):pii_in_value = self.scan_text(value)if pii_in_value:detected_pii.append((key, pii_in_value))return detected_pii# Encryption Service for protecting sensitive data
class EncryptionService:def __init__(self, key_path=None):if key_path and os.path.exists(key_path):with open(key_path, "rb") as key_file:self.key = key_file.read()else:self.key = Fernet.generate_key()if key_path:with open(key_path, "wb") as key_file:key_file.write(self.key)self.cipher = Fernet(self.key)def encrypt(self, data):"""Encrypt data"""if isinstance(data, str):return self.cipher.encrypt(data.encode()).decode()else:return self.cipher.encrypt(json.dumps(data).encode()).decode()def decrypt(self, encrypted_data):"""Decrypt data"""if encrypted_data is None:return Nonedecrypted = self.cipher.decrypt(encrypted_data.encode())try:return json.loads(decrypted)except:return decrypted.decode()# Security decorator for tools
def secure_tool(requires_encryption=False, log_access=True):def decorator(cls):original_execute = cls.execute_async if hasattr(cls, 'execute_async') else cls.execute@wraps(original_execute)async def secure_execute(self, request):# Check for PII in requestpii_detector = PiiDetector()pii_found = pii_detector.scan_parameters(request.parameters)# Log access if requiredif log_access:tool_name = self.get_name()user_id = request.context.get("user_id", "anonymous")log_entry = {"timestamp": datetime.now().isoformat(),"tool": tool_name,"user": user_id,"contains_pii": bool(pii_found),"parameters": {k: "***" for k in request.parameters.keys()}  # Don't log actual values}logging.info(f"Tool access: {json.dumps(log_entry)}")# Handle detected PIIif pii_found:# Either encrypt sensitive data or reject the requestif requires_encryption:encryption_service = EncryptionService("keys/tool_key.key")for param_name, pii_types in pii_found:# Encrypt the sensitive parameterrequest.parameters[param_name] = encryption_service.encrypt(request.parameters[param_name])else:# If encryption not available but PII found, you might reject the requestraise ToolExecutionException("Request contains sensitive data that cannot be processed securely")# Execute the original methodreturn await original_execute(self, request)# Replace the execute methodif hasattr(cls, 'execute_async'):cls.execute_async = secure_executeelse:cls.execute = secure_executereturn clsreturn decorator# Example of a secure tool with the decorator
@secure_tool(requires_encryption=True, log_access=True)
class SecureCustomerDataTool(Tool):def get_name(self):return "customerData"def get_description(self):return "Accesses customer data securely"def get_schema(self):# Schema definitionreturn {}async def execute_async(self, request):# Implementation would access customer data securely# Since we used the decorator, PII is already detected and encryptedreturn ToolResponse(result={"status": "success"})