去中心化 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 系统。不适用于内部低风险操作——过度设计比没有设计更危险。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐