一、为什么需要安全通信前面六讲构建了应用层的纵深防御体系但所有这些防御都依赖于一个前提通信信道是可信的。如果通信本身不安全攻击者可以中间人攻击篡改 Agent 与 Server 之间的消息绕过所有应用层检测凭证窃取截获长期有效的 Token冒充合法 Agent重放攻击捕获一次合法请求重复发送以执行未授权操作通信窃听获取传输中的敏感数据即使 L4 做了脱敏元数据仍有价值MCP 协议默认支持 Streamable HTTP但安全传输需要额外加固。二、安全通信的四层保障层级技术手段解决的问题L7a传输加密TLS/mTLS防窃听、防篡改、防中间人L7b身份认证JWT SPIFFE确认通信双方的身份L7c短期凭证动态 Token 颁发限制凭证泄露的影响范围L7dSecret 管理Vault/K8s Secret安全分发和管理密钥三、Go 实现mTLS 通信与动态凭证管理package main import ( crypto/ecdsa crypto/elliptic crypto/rand crypto/tls crypto/x509 crypto/x509/pkix encoding/json encoding/pem fmt log math/big net net/http sync time ) // ---- SPIFFE 风格的 Identity ---- // SPIFFE (Secure Production Identity Framework for Everyone) // 标准格式: spiffe://trust-domain/path type SPIFFEID struct { TrustDomain string json:trust_domain Path string json:path } func (s *SPIFFEID) String() string { return fmt.Sprintf(spiffe://%s%s, s.TrustDomain, s.Path) } // 从证书的 URI SAN 中解析 SPIFFE ID func ExtractSPIFFEFromCert(cert *x509.Certificate) *SPIFFEID { for _, uri : cert.URIs { uriStr : uri.String() if len(uriStr) 9 uriStr[:9] spiffe:// { remainder : uriStr[9:] slashIdx : indexOf(remainder, /) if slashIdx -1 { return SPIFFEID{TrustDomain: remainder, Path: /} } return SPIFFEID{ TrustDomain: remainder[:slashIdx], Path: remainder[slashIdx:], } } } return nil } func indexOf(s string, c byte) int { for i : 0; i len(s); i { if s[i] c { return i } } return -1 } // ---- 动态凭证颁发器 ---- type TokenIssuer struct { mu sync.Mutex privateKey *ecdsa.PrivateKey publicKey *ecdsa.PublicKey validTokens map[string]*TokenInfo // token_hash - info ttl time.Duration } type TokenInfo struct { AgentID string json:agent_id SPIFFEID *SPIFFEID json:spiffe_id IssuedAt time.Time json:issued_at ExpiresAt time.Time json:expires_at Scopes []string json:scopes } func NewTokenIssuer() (*TokenIssuer, error) { privateKey, err : ecdsa.GenerateKey(elliptic.P256(), rand.Reader) if err ! nil { return nil, fmt.Errorf(生成密钥对失败: %w, err) } return TokenIssuer{ privateKey: privateKey, publicKey: privateKey.PublicKey, validTokens: make(map[string]*TokenInfo), ttl: 15 * time.Minute, // 默认 15 分钟过期 }, nil } // 颁发短期 Token func (ti *TokenIssuer) IssueToken(agentID string, spiffeID *SPIFFEID, scopes []string) (string, error) { ti.mu.Lock() defer ti.mu.Unlock() now : time.Now() info : TokenInfo{ AgentID: agentID, SPIFFEID: spiffeID, IssuedAt: now, ExpiresAt: now.Add(ti.ttl), Scopes: scopes, } // 序列化并签名 data, _ : json.Marshal(info) signature, err : ecdsa.SignASN1(rand.Reader, ti.privateKey, hashData(data)) if err ! nil { return , fmt.Errorf(签名失败: %w, err) } token : base64Encode(data) . base64Encode(signature) tokenHash : hashString(token) ti.validTokens[tokenHash] info log.Printf([TOKEN] 颁发 Token: Agent%s SPIFFE%s TTL%v Scopes%v, agentID, spiffeID, ti.ttl, scopes) return token, nil } // 验证 Token func (ti *TokenIssuer) VerifyToken(token string) (*TokenInfo, error) { ti.mu.Lock() defer ti.mu.Unlock() // 检查是否在有效期内 tokenHash : hashString(token) info, exists : ti.validTokens[tokenHash] if !exists { return nil, fmt.Errorf(未知 Token) } if time.Now().After(info.ExpiresAt) { delete(ti.validTokens, tokenHash) return nil, fmt.Errorf(Token 已过期) } return info, nil } // 撤销 TokenAgent 登出或检测到异常时调用 func (ti *TokenIssuer) RevokeToken(token string) { tokenHash : hashString(token) ti.mu.Lock() defer ti.mu.Unlock() delete(ti.validTokens, tokenHash) log.Printf([TOKEN] 撤销 Token: Hash%s, tokenHash[:16]) } // ---- mTLS 证书生成器用于开发和测试 ---- type CertGenerator struct { caCert *x509.Certificate caKey interface{} } func NewCertGenerator() (*CertGenerator, error) { // 生成 CA 密钥对 caKey, err : ecdsa.GenerateKey(elliptic.P384(), rand.Reader) if err ! nil { return nil, err } // 自签名 CA 证书 caTemplate : x509.Certificate{ SerialNumber: big.NewInt(1), Subject: pkix.Name{ CommonName: MCP Security CA, }, NotBefore: time.Now(), NotAfter: time.Now().AddDate(10, 0, 0), IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, } caDER, err : x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, caKey.PublicKey, caKey) if err ! nil { return nil, err } caCert, err : x509.ParseCertificate(caDER) if err ! nil { return nil, err } return CertGenerator{caCert: caCert, caKey: caKey}, nil } // 为 Agent 签发证书 func (cg *CertGenerator) GenerateAgentCert(spiffeID *SPIFFEID) (tls.Certificate, error) { key, err : ecdsa.GenerateKey(elliptic.P256(), rand.Reader) if err ! nil { return tls.Certificate{}, err } serial, _ : rand.Int(rand.Reader, big.NewInt(162)) template : x509.Certificate{ SerialNumber: serial, Subject: pkix.Name{ CommonName: spiffeID.String(), }, URIs: []*url.URL{{Scheme: spiffe, Host: spiffeID.TrustDomain, Path: spiffeID.Path}}, NotBefore: time.Now(), NotAfter: time.Now().AddDate(1, 0, 0), KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment, ExtKeyUsage: []x509.ExtKeyUsage{ x509.ExtKeyUsageClientAuth, x509.ExtKeyUsageServerAuth, }, } certDER, err : x509.CreateCertificate(rand.Reader, template, cg.caCert, key.PublicKey, cg.caKey) if err ! nil { return tls.Certificate{}, err } return tls.Certificate{ Certificate: [][]byte{certDER, cg.caCert.Raw}, PrivateKey: key, }, nil } // ---- 安全的 MCP Gateway带 mTLS Token 认证 ---- type SecureMCPGateway struct { issuer *TokenIssuer certGen *CertGenerator serverCert tls.Certificate agentCerts map[string]tls.Certificate // agent_id - cert mu sync.RWMutex handler http.Handler } func NewSecureMCPGateway(handler http.Handler) (*SecureMCPGateway, error) { issuer, err : NewTokenIssuer() if err ! nil { return nil, err } certGen, err : NewCertGenerator() if err ! nil { return nil, err } // 生成 Gateway 自身的证书 gatewayID : SPIFFEID{TrustDomain: mcp.example.com, Path: /gateway/main} serverCert, err : certGen.GenerateAgentCert(gatewayID) if err ! nil { return nil, err } return SecureMCPGateway{ issuer: issuer, certGen: certGen, serverCert: serverCert, agentCerts: make(map[string]tls.Certificate), handler: handler, }, nil } // 注册 Agent 并为其签发证书和 Token func (sg *SecureMCPGateway) RegisterAgent(agentID string, roles []string) (tls.Certificate, string, error) { spiffeID : SPIFFEID{ TrustDomain: mcp.example.com, Path: fmt.Sprintf(/agent/%s, agentID), } // 签发证书 cert, err : sg.certGen.GenerateAgentCert(spiffeID) if err ! nil { return tls.Certificate{}, , err } // 颁发 Token token, err : sg.issuer.IssueToken(agentID, spiffeID, roles) if err ! nil { return tls.Certificate{}, , err } sg.mu.Lock() sg.agentCerts[agentID] cert sg.mu.Unlock() log.Printf([REGISTER] Agent %s 注册成功SPIFFE ID: %s, agentID, spiffeID) return cert, token, nil } // 创建 mTLS 配置 func (sg *SecureMCPGateway) TLSConfig() *tls.Config { return tls.Config{ Certificates: []tls.Certificate{sg.serverCert}, ClientAuth: tls.RequireAndVerifyClientCert, ClientCAs: sg.certGen.caCertPool(), MinVersion: tls.VersionTLS13, CipherSuites: []uint16{ tls.TLS_AES_128_GCM_SHA256, tls.TLS_AES_256_GCM_SHA384, }, VerifyPeerCertificate: sg.verifyPeerCertificate, } } // 验证客户端证书 func (sg *SecureMCPGateway) verifyPeerCertificate(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error { if len(verifiedChains) 0 || len(verifiedChains[0]) 0 { return fmt.Errorf(未提供客户端证书) } clientCert : verifiedChains[0][0] spiffeID : ExtractSPIFFEFromCert(clientCert) if spiffeID nil { return fmt.Errorf(客户端证书缺少 SPIFFE ID) } log.Printf([MTLS] 验证客户端: SPIFFE%s CN%s, spiffeID, clientCert.Subject.CommonName) return nil } // HTTP 中间件提取并验证 Token func (sg *SecureMCPGateway) AuthMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // 从 mTLS 连接中获取客户端身份 if r.TLS nil || len(r.TLS.PeerCertificates) 0 { http.Error(w, {error:need_mtls}, http.StatusUnauthorized) return } clientCert : r.TLS.PeerCertificates[0] spiffeID : ExtractSPIFFEFromCert(clientCert) if spiffeID nil { http.Error(w, {error:invalid_spiffe}, http.StatusUnauthorized) return } // 从 Header 提取 Token token : r.Header.Get(Authorization) if token || !strings.HasPrefix(token, Bearer ) { http.Error(w, {error:missing_token}, http.StatusUnauthorized) return } token strings.TrimPrefix(token, Bearer ) // 验证 Token tokenInfo, err : sg.issuer.VerifyToken(token) if err ! nil { http.Error(w, fmt.Sprintf({error:invalid_token,detail:%s}, err.Error()), http.StatusUnauthorized) return } // 将身份信息注入请求上下文 ctx : context.WithValue(r.Context(), agent_id, tokenInfo.AgentID) ctx context.WithValue(ctx, spiffe_id, tokenInfo.SPIFFEID) ctx context.WithValue(ctx, scopes, tokenInfo.Scopes) next.ServeHTTP(w, r.WithContext(ctx)) }) } // ---- Secret 管理 ---- type SecretManager struct { mu sync.RWMutex secrets map[string]*SecretEntry // 轮换策略 rotationInterval time.Duration } type SecretEntry struct { Value string json:value Version int json:version CreatedAt time.Time json:created_at ExpiresAt time.Time json:expires_at // 访问控制 AllowedAgents []string json:allowed_agents } func NewSecretManager(rotationInterval time.Duration) *SecretManager { return SecretManager{ secrets: make(map[string]*SecretEntry), rotationInterval: rotationInterval, } } // 存储 Secret func (sm *SecretManager) StoreSecret(name string, value string, allowedAgents []string) { sm.mu.Lock() defer sm.mu.Unlock() entry : SecretEntry{ Value: value, Version: 1, CreatedAt: time.Now(), ExpiresAt: time.Now().Add(sm.rotationInterval), AllowedAgents: allowedAgents, } if existing, ok : sm.secrets[name]; ok { entry.Version existing.Version 1 } sm.secrets[name] entry log.Printf([SECRET] 存储 Secret: %s (版本 %d, 过期 %v), name, entry.Version, entry.ExpiresAt) } // 获取 Secret带 Agent 身份验证 func (sm *SecretManager) GetSecret(name string, agentID string) (string, error) { sm.mu.RLock() defer sm.mu.RUnlock() entry, exists : sm.secrets[name] if !exists { return , fmt.Errorf(Secret %s 不存在, name) } // 检查 Agent 是否有权限 if len(entry.AllowedAgents) 0 { allowed : false for _, allowedID : range entry.AllowedAgents { if allowedID agentID || allowedID * { allowed true break } } if !allowed { return , fmt.Errorf(Agent %s 无权访问 Secret %s, agentID, name) } } // 检查是否过期 if time.Now().After(entry.ExpiresAt) { return , fmt.Errorf(Secret %s 已过期版本 %d, name, entry.Version) } return entry.Value, nil } // 自动轮换 Secret func (sm *SecretManager) AutoRotate(name string, generator func() string) { go func() { for { time.Sleep(sm.rotationInterval / 2) // 在过期前一半时间轮换 sm.mu.Lock() entry, exists : sm.secrets[name] if !exists { sm.mu.Unlock() continue } newValue : generator() entry.Value newValue entry.Version entry.CreatedAt time.Now() entry.ExpiresAt time.Now().Add(sm.rotationInterval) sm.mu.Unlock() log.Printf([ROTATE] Secret %s 已轮换 (版本 %d), name, entry.Version) } }() } // ---- 辅助函数 ---- func hashData(data []byte) []byte { h : sha256.New() h.Write(data) return h.Sum(nil) } func hashString(s string) string { return hex.EncodeToString(hashData([]byte(s))) } func base64Encode(data []byte) string { return base64.URLEncoding.WithPadding(base64.NoPadding).EncodeToString(data) } // ---- 演示 ---- func main() { // 初始化安全网关 handler : http.NewServeMux() handler.HandleFunc(/mcp/v1/tools/list, func(w http.ResponseWriter, r *http.Request) { agentID : r.Context().Value(agent_id).(string) spiffeID : r.Context().Value(spiffe_id).(*SPIFFEID) w.Header().Set(Content-Type, application/json) json.NewEncoder(w).Encode(map[string]interface{}{ agent_id: agentID, spiffe_id: spiffeID.String(), tools: []map[string]string{ {name: query_orders, description: 查询订单}, }, }) }) gateway, err : NewSecureMCPGateway(handler) if err ! nil { log.Fatal(err) } // 注册两个 Agent agent1Cert, agent1Token, _ : gateway.RegisterAgent(agent-cs-001, []string{cs_agent, readonly}) agent2Cert, agent2Token, _ : gateway.RegisterAgent(agent-ops-001, []string{ops_admin}) fmt.Println( Agent 注册信息 ) fmt.Printf(Agent 1 证书序列号: %x\n, agent1Cert.Certificate[0][len(agent1Cert.Certificate[0])-8:]) fmt.Printf(Agent 1 Token: %s...\n, agent1Token[:32]) fmt.Printf(Agent 2 证书序列号: %x\n, agent2Cert.Certificate[0][len(agent2Cert.Certificate[0])-8:]) fmt.Printf(Agent 2 Token: %s...\n, agent2Token[:32]) // Secret 管理演示 secretManager : NewSecretManager(24 * time.Hour) secretManager.StoreSecret(db_password, s3cr3t!pass, []string{agent-ops-001}) secretManager.StoreSecret(api_key_openai, sk-xxxxxxxxxx, []string{agent-cs-001, agent-ops-001}) fmt.Println(\n Secret 访问测试 ) // Agent 1 访问 db_password无权限 _, err secretManager.GetSecret(db_password, agent-cs-001) fmt.Printf(Agent cs-001 访问 db_password: %v\n, err) // Agent 2 访问 db_password有权限 val, err : secretManager.GetSecret(db_password, agent-ops-001) fmt.Printf(Agent ops-001 访问 db_password: 成功 (value%s)\n, val[:8]...) // 两个 Agent 都可以访问 api_key val1, _ : secretManager.GetSecret(api_key_openai, agent-cs-001) val2, _ : secretManager.GetSecret(api_key_openai, agent-ops-001) fmt.Printf(Agent cs-001 访问 api_key: 成功\nAgent ops-001 访问 api_key: 成功\n) // Token 过期测试 fmt.Println(\n Token 过期测试 ) // 修改 issuer 的 TTL 为 1 秒 gateway.issuer.ttl 1 * time.Second shortToken, _ : gateway.issuer.IssueToken(test-agent, SPIFFEID{TrustDomain: test, Path: /test}, []string{test}) fmt.Printf(短期 Token 刚颁发: valid%v\n, gateway.issuer.VerifyToken(shortToken) nil) time.Sleep(2 * time.Second) fmt.Printf(2 秒后: valid%v\n, gateway.issuer.VerifyToken(shortToken) nil) // Token 撤销测试 fmt.Println(\n Token 撤销测试 ) revokeToken, _ : gateway.issuer.IssueToken(evil-agent, SPIFFEID{TrustDomain: evil, Path: /hacker}, []string{admin}) fmt.Printf(撤销前: valid%v\n, gateway.issuer.VerifyToken(revokeToken) nil) gateway.issuer.RevokeToken(revokeToken) fmt.Printf(撤销后: valid%v\n, gateway.issuer.VerifyToken(revokeToken) nil) }四、生产环境的安全通信架构┌─────────────────┐ mTLS ┌──────────────────┐ │ Agent A │◄─────────────►│ MCP Gateway │ │ (SPIFFE ID) │ 短期 Token │ (证书验证Token) │ └─────────────────┘ └────────┬─────────┘ │ ┌──────────────────────┼──────────────────────┐ │ │ │ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ Vault/Hashi │ │ K8s CA │ │ External │ │ Secret Mgmt │ │ (Pod Identity)│ │ PKI │ └──────────────┘ └──────────────┘ └──────────────┘五、关键设计原则原则说明实现方式短期凭证Token 有效期短15分钟降低泄露影响JWT 自动续期双向认证不仅 Server 验证 ClientClient 也验证 ServermTLS最小凭证每个 Token 只包含所需的最小 ScopeScope 白名单自动轮换证书和 Secret 定期自动更换TTL 后台 Goroutine即时撤销发现异常立即吊销凭证Revoke API不可否认性所有操作都可追溯到具体的凭证审计日志关联 Token ID六、安全分层L7本讲在 L1-L6 基础上叠加通信安全层层级防御手段本讲新增L1Prompt 注入检测输入过滤L2输出验证与对齐参数校验、业务规则L3工具权限与最小特权RBAC/ABAC 策略引擎L4数据脱敏与隔离字段级脱敏L5行为审计与异常检测审计日志、异常评分L6速率限制与熔断保护限流、熔断、降级L7amTLS 传输加密✅ 双向 TLS 认证L7bSPIFFE 身份体系✅ 标准化身份标识L7c短期 Token 管理✅ 动态颁发/验证/撤销L7dSecret 安全存储✅ 访问控制 自动轮换七、延伸阅读SPIFFE/SPIRE 标准CNCF 孵化的生产环境身份认证框架Lets Encrypt ACME 协议自动化证书管理的行业标准HashiCorp VaultSecret 管理和动态凭证颁发的工业级方案RFC 5280 (X.509 PKI)公钥基础设施的标准规范八、下一讲预告第8讲运行时安全与沙箱隔离——Agent 能做什么不能做什么通信安全解决了「谁在说话」的问题但没有解决「Agent 能做什么」的问题。如果 Agent 本身被攻破它可以利用宿主机的所有能力。下一讲实现 Agent 运行时沙箱文件系统隔离、网络策略、进程限制和资源配额确保被攻破的 Agent 无法逃逸到宿主机或其他 Agent。
