去中心化 AI 身份:用 DID 绑定模型调用的责任方
去中心化 AI 身份:用 DID 绑定模型调用的责任方
一、AI 生成了一段恶意代码,谁来负责
这个问题不是一个哲学问题,而是工程问题。
去年我们团队上线了一个代码生成 Agent,上线第二周就出了事。Agent 自动在代码仓库里提交了一段数据库迁移脚本,这段脚本被部署到生产环境后直接把一张核心业务表给改坏了。事故复盘的时候,我们需要追溯到谁授权了这个 Agent 执行这次操作。但翻遍操作日志,只能看到操作者是 API Key 对应的服务账号svc-code-agent,而持有这个 API Key 的是整个平台团队共 8 个人。
传统 OAuth/API Key 只能证明"谁调用了 API"。但无法证明"谁授权了这次特定的 Agent 操作"。Agent 可能在一小时前被授权,一小时后的行为已经超出预期。操作日志显示操作者 ID,但这个 ID 是 API Key 对应的服务账号,真正的授权者被淹没了。
这不是个别现象。随着 Agent 自动化程度越来越高,"谁批准了这次操作"这个问题会越来越频繁地出现。合规审计要求每一次 AI 调用都可追溯到具体的人。传统的身份体系是在"账号层级"做授权,而 Agent 需要在"调用层级"做授权。
DID(Decentralized Identifier)提供了一个可验证的身份层。每次模型调用绑定一个 DID 签名的授权凭证。这把身份从"账号层级"下沉到"调用层级"。凭证包含权限范围、有效期和调用次数限制,签名后不可篡改。出事的时候,拿凭证哈希一查,谁授权的、授权了什么、什么时候授权的,一目了然。
二、DID + 可验证凭证的 AI 调用链
核心流程分四步。第一步,用户用自己的 DID 私钥签发一个授权凭证(VC),凭证里写清楚:允许 Agent 做什么操作、调用哪些模型、最多用多少 Token、有效期多长。第二步,Agent 携带这个 VC 去调用模型 API。第三步,API 网关在链上验证 VC 的签名有效性、有效期和权限范围。第四步,调用记录和 VC 哈希上链存证,形成不可篡改的审计轨迹。
这个架构的权衡点在于验证延迟和去中心化程度的平衡。完全去中心化(链上验证)安全性最高但延迟大,适合高合规要求场景。半去中心化(链下验证 + 定期上链存证)延迟可控,适合大多数生产场景。我们实际落地时选择了后者:验证走本地 DID Resolver 缓存,每 5 分钟批量上链存证,单次调用增加约 8ms 延迟。
sequenceDiagram participant U as 用户 DID participant A as AI Agent participant V as 凭证验证器 participant L as LLM API participant C as 链上存证 U->>A: 签发授权VC<br/>(范围: 读文件, 有效期: 1h) A->>V: 提交VC + 待执行操作 V->>V: 验证: 签名/有效期/权限范围 V->>C: 查询DID文档确认公钥 alt VC有效 且 操作在权限内 V-->>A: 验证通过 A->>L: 携带VC哈希发起推理 L->>L: 记录调用者DID L->>C: 调用记录+VC哈希上链 L-->>A: 返回结果 else VC无效 V-->>A: 拒绝 + 原因 end三、DID 集成的 Go 实现
下面是我们生产环境使用的 DID 授权框架核心实现。基于 Ed25519 签名,支持权限范围声明、调用次数限制和链上存证。
package didauth import ( "crypto/ed25519" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "time" ) // DID 标识符 type DID struct { Method string `json:"method"` // did:example ID string `json:"id"` // 唯一标识 } func (d DID) String() string { return fmt.Sprintf("did:%s:%s", d.Method, d.ID) } // VerifiableCredential 可验证凭证 type VerifiableCredential struct { Context []string `json:"@context"` Type []string `json:"type"` Issuer DID `json:"issuer"` // 签发者 Subject DID `json:"credentialSubject"` // 主体 IssuedAt time.Time `json:"issuanceDate"` ExpiresAt time.Time `json:"expirationDate"` // AI 特定的权限声明 Claims AIClaims `json:"credentialSubject.claims"` // 签名 Proof Proof `json:"proof"` } // AIClaims AI 相关权限声明 type AIClaims struct { // 允许的操作类型 AllowedActions []string `json:"allowed_actions"` // 允许的模型 AllowedModels []string `json:"allowed_models"` // 最大 Token 数 MaxTokens int `json:"max_tokens"` // 最大调用次数 MaxCalls int `json:"max_calls"` // 是否可以自动执行高危操作 AutoApprove bool `json:"auto_approve"` } // Proof 签名证明 type Proof struct { Type string `json:"type"` // Ed25519Signature2020 Created time.Time `json:"created"` Creator DID `json:"verificationMethod"` Signature string `json:"proofValue"` // 签名值(hex) } // DIDDocument DID 文档 type DIDDocument struct { ID DID `json:"id"` PublicKeyHex string `json:"publicKeyHex"` VerificationMethods []VerificationMethod `json:"verificationMethod"` } type VerificationMethod struct { ID string `json:"id"` Type string `json:"type"` // Ed25519VerificationKey2020 Controller DID `json:"controller"` PublicKeyHex string `json:"publicKeyHex"` } // CredentialVerifier 凭证验证器 type CredentialVerifier struct { // 实际项目中这里连接区块链或 DID Resolver didRegistry map[string]DIDDocument } // NewVerifier 创建验证器 func NewVerifier() *CredentialVerifier { return &CredentialVerifier{ didRegistry: make(map[string]DIDDocument), } } // RegisterDID 注册 DID func (v *CredentialVerifier) RegisterDID(doc DIDDocument) { v.didRegistry[doc.ID.String()] = doc } // VerifyCredential 验证凭证 func (v *CredentialVerifier) VerifyCredential(vc *VerifiableCredential) error { // 1. 验证过期 if time.Now().After(vc.ExpiresAt) { return fmt.Errorf("凭证已过期 (过期时间: %s)", vc.ExpiresAt.Format(time.RFC3339)) } // 2. 查找 DID 文档 doc, ok := v.didRegistry[vc.Issuer.String()] if !ok { return fmt.Errorf("未找到 DID 文档: %s", vc.Issuer.String()) } // 3. 验证签名 pubKeyBytes, err := hex.DecodeString(doc.PublicKeyHex) if err != nil { return fmt.Errorf("公钥解析失败: %w", err) } // 构造待签名数据 claimsData := v.canonicalize(vc.Claims) message := fmt.Sprintf("%s:%s:%s:%d:%d", vc.Issuer.String(), vc.Subject.String(), claimsData, vc.IssuedAt.Unix(), vc.ExpiresAt.Unix(), ) messageHash := sha256.Sum256([]byte(message)) signature, err := hex.DecodeString(vc.Proof.Signature) if err != nil { return fmt.Errorf("签名解析失败: %w", err) } if !ed25519.Verify(pubKeyBytes, messageHash[:], signature) { return fmt.Errorf("签名验证失败") } return nil } // VerifyAction 验证操作是否在授权范围内 func (v *CredentialVerifier) VerifyAction( vc *VerifiableCredential, action string, model string, tokens int, callCount int, ) error { claims := vc.Claims // 检查操作是否允许 if !contains(claims.AllowedActions, action) && !contains(claims.AllowedActions, "*") { return fmt.Errorf("操作 %s 不在授权范围内 (允许: %v)", action, claims.AllowedActions) } // 检查模型是否允许 if !contains(claims.AllowedModels, model) && !contains(claims.AllowedModels, "*") { return fmt.Errorf("模型 %s 不在授权范围内", model) } // 检查 Token 限制 if tokens > claims.MaxTokens { return fmt.Errorf("Token 数 %d 超过限制 %d", tokens, claims.MaxTokens) } // 检查调用次数 if callCount > claims.MaxCalls { return fmt.Errorf("调用次数 %d 超过限制 %d", callCount, claims.MaxCalls) } return nil } // IssueCredential 签发凭证 func IssueCredential( issuer DID, subject DID, claims AIClaims, validFor time.Duration, privateKey ed25519.PrivateKey, ) (*VerifiableCredential, error) { now := time.Now() vc := &VerifiableCredential{ Context: []string{"https://www.w3.org/2018/credentials/v1"}, Type: []string{"VerifiableCredential", "AIAuthorizationCredential"}, Issuer: issuer, Subject: subject, IssuedAt: now, ExpiresAt: now.Add(validFor), Claims: claims, } // 签名 claimsData := canonicalize(claims) message := fmt.Sprintf("%s:%s:%s:%d:%d", issuer.String(), subject.String(), claimsData, now.Unix(), vc.ExpiresAt.Unix(), ) messageHash := sha256.Sum256([]byte(message)) signature := ed25519.Sign(privateKey, messageHash[:]) vc.Proof = Proof{ Type: "Ed25519Signature2020", Created: now, Creator: issuer, Signature: hex.EncodeToString(signature), } return vc, nil } func canonicalize(claims AIClaims) string { data, _ := json.Marshal(claims) return string(data) } func contains(slice []string, item string) bool { for _, s := range slice { if s == item { return true } } return false } // ---- 日志记录 ---- type CallRecord struct { TraceID string `json:"trace_id"` DID string `json:"caller_did"` VC_Hash string `json:"vc_hash"` // 凭证哈希(链上可查) Action string `json:"action"` Model string `json:"model"` Tokens int `json:"tokens"` Timestamp time.Time `json:"timestamp"` ResultHash string `json:"result_hash"` } func RecordCall(vc *VerifiableCredential, record CallRecord) { vcBytes, _ := json.Marshal(vc) vcHash := sha256.Sum256(vcBytes) record.VC_Hash = hex.EncodeToString(vcHash[:]) record.Timestamp = time.Now() // 生产环境:写入区块链存证 data, _ := json.Marshal(record) fmt.Printf("[DID-CALL] %s\n", string(data)) }实际使用时的调用示例:
func main() { // 1. 生成 DID 密钥对 pubKey, privKey, _ := ed25519.GenerateKey(nil) userDID := DID{Method: "example", ID: "user-wanglei-001"} agentDID := DID{Method: "example", ID: "agent-code-reviewer"} // 2. 注册 DID 文档 verifier := NewVerifier() verifier.RegisterDID(DIDDocument{ ID: userDID, PublicKeyHex: hex.EncodeToString(pubKey), }) // 3. 用户签发授权凭证:允许 Agent 读代码,1小时有效 vc, err := IssueCredential( userDID, agentDID, AIClaims{ AllowedActions: []string{"read_file", "code_review"}, AllowedModels: []string{"gpt-4o", "gpt-4o-mini"}, MaxTokens: 50000, MaxCalls: 100, AutoApprove: false, // 高危操作必须人工确认 }, 1*time.Hour, privKey, ) if err != nil { panic(err) } // 4. Agent 执行操作时验证凭证 if err := verifier.VerifyCredential(vc); err != nil { fmt.Println("凭证验证失败:", err) return } if err := verifier.VerifyAction(vc, "read_file", "gpt-4o", 2000, 1); err != nil { fmt.Println("操作越权:", err) return } fmt.Println("授权验证通过,执行操作") RecordCall(vc, CallRecord{ TraceID: "trace-abc-123", DID: agentDID.String(), Action: "read_file", Model: "gpt-4o", Tokens: 2000, }) }四、DID 方案的工程代价
DID 增加每次调用的验证延迟约 10-50ms,取决于链上查询的响应速度。高频调用场景需要凭证缓存机制。我们踩过一个坑:初期每次调用都走链上验证,高峰期 QPS 到 500 时验证延迟飙到 200ms。后来改成本地缓存 DID 文档 + 5 分钟刷新一次,延迟降到 8ms。
DID 基础设施还不够成熟。生产环境依赖的 DID Resolver 需要高可用。目前可选的方案包括 ION(比特币侧链)、did:web(自托管)。我们选了 did:web,原因很简单:运维成本低,DNS + HTTPS 就够了,不需要额外维护区块链节点。但代价是去中心化程度弱,如果你的合规要求极高,ION 是更好的选择。
DID 方案的 ROI 需要分场景评估。对于内部工具且操作者明确的场景,OAuth 已经足够,加 DID 反而增加复杂度。对于涉及外部合作伙伴、多方审计、高合规要求的 AI 系统来说,DID 的价值就体现出来了。我们算过一笔账:一次合规审计事件如果追溯不到责任人,平均处置成本约 15 万人民币(含停机、人力、罚款)。而 DID 方案的全年维护成本约 3 万。只要一年能避免 1 次审计事故,ROI 就是正的。
不适合的场景:内部工具且操作者明确(OAuth 已足够);延迟敏感的高频交易系统(验证开销不可接受);不需要审计追踪的非敏感操作(过度设计)。
五、总结
DID + 可验证凭证将 AI 调用的身份粒度细化到操作级别。每次 Agent 行为都绑定签发者的 DID 签名,凭证包含权限范围、有效期和调用限制。验证通过后可选择将调用记录上链存证,形成不可篡改的审计轨迹。
落地建议:先在低频高价值场景试点(如生产环境部署审批、外部数据访问授权),验证流程跑通后再推广到高频场景。验证层用本地缓存 + 定期上链的模式平衡延迟和安全性。选择 did:web 起步,合规要求升级时迁移到 ION。
适用于需要明确责任归属的合规 AI 系统。不适用于内部低风险操作——过度设计比没有设计更危险。