AI Agent 身份与去中心化标识(DID)管理指南
适用链: msg-chain-1 | bech32 前缀: msg
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
1. 概述
1.1 为什么 AI Agent 需要去中心化身份
在区块链网络中运行的 AI Agent 需要一个可信、自主、可验证的身份体系。传统的身份模型——基于中心化 CA 的 PKI 体系——存在单点故障、隐私泄漏、跨域互认困难等问题。而去中心化标识（Decentralized Identity, DID）为 Agent 提供了以下核心能力：
- 自主权: Agent 自己控制私钥，无需依赖任何中心化注册机构
- 可验证性: 任何第三方都可以独立验证 Agent 的身份声明
- 持久性: 身份不依赖于特定平台或服务商
- 互操作性: 遵循 W3C 标准的 DID 可以在不同链和应用间互认
- 隐私保护: 支持选择性披露，最小化数据暴露
1.2 DID 与钱包地址的区别
| 维度 | 钱包地址 | DID |
|---|---|---|
| 本质 | 公钥哈希/派生地址 | 持久性标识符 |
| 可变性 | 每次创建新钱包都改变 | 可更新但保持同一标识 |
| 元数据 | 无 | 包含验证方法、服务端点等 |
| 可验证性 | 仅签名验证 | 完整文档+凭证体系 |
| 生命周期 | 无限，无撤销机制 | 支持激活/停用/轮换 |
| 用途 | 转账/交易签名 | 身份认证、授权、凭证签发 |
钱包地址是 Agent 在链上进行价值转移的标识，而 DID 是 Agent 在更广泛的身份生态中的锚点。两者可以关联：Agent 的 DID 文档中可包含其钱包地址作为 blockchainAccountId 验证方法。
1.3 W3C DID 标准概述
W3C DID 标准定义了以下核心规范：
- DID URI: did:method-name:method-specific-id，例如 did:msg:agent-abc123
- DID Document: 描述 DID 的 JSON-LD 文档，包含公钥、服务端点等
- DID Method: 定义如何在特定区块链/网络上注册、解析、更新和停用 DID
- Verifiable Credential: 由 Issuer 签名的、可加密验证的声明
1.4 Agent 身份模型
在 msg-chain-1 网络中，Agent 的身份模型分为三层：
+-------------------------------------------+
| 身份层 (DID) |
| did:msg:<agent_id> |
| DID Document + Verification Method |
+-------------------------------------------+
| 凭证层 (VC) |
| 签发、验证、撤销可验证凭证 |
+-------------------------------------------+
| 链上锚定层 |
| DID Registry Contract |
| 文档哈希 + 元数据存储 |
+-------------------------------------------+
1.5 msg DID Method 规范
msg DID method 的定义如下：
Method Name: msg
Method Specific Identifier: <agent_id> | <address>
DID Format: did:msg:<agent_id>
Example: did:msg:agent-2a8f1c3e9b
CRUD Operations:
- Create: 提交 DID Document 到 DID Registry 合约
- Read: 通过 DID Resolver 查询链上注册信息
- Update: 由 DID Controller 提交更新（如密钥轮换）
- Delete: 标记为 deactivated（不可逆停用）
1.6 Agent 身份的全生命周期
+----------+ +----------+ +----------+ +----------+
| 密钥生成 |-->| DID 注册 |-->| 凭证签发 |-->| 身份使用 |
+----------+ +----------+ +----------+ +----------+
|
v
+----------+ +----------+ +----------+
| 身份停用 |<--| 密钥轮换 |<--| 凭证验证 |
+----------+ +----------+ +----------+
1.7 本指南涵盖的内容
本指南将完整覆盖以下内容：
- DID 文档结构与注册流程
- DID 锚定合约的设计与部署
- 可验证凭证的签发、验证与撤销
- 跨链身份互认与 IBC 集成
- 身份与 Agent 宪章绑定
- 隐私保护身份技术
- 完整的端到端示例
- 安全最佳实践与常见陷阱
2. DID 注册与管理
2.1 密钥对生成
Agent 身份的基础是公私钥对。在 msg-chain-1 上，我们推荐使用 Ed25519 曲线，因为它性能优异且广泛支持。
import hashlib
from typing import Optional
from nacl.bindings import crypto_sign_keypair, crypto_sign_seed_keypair
class AgentKeyPair:
"""Ed25519 key pair with utility methods for DID usage"""
def __init__(self, seed: Optional[bytes] = None):
if seed:
if len(seed) != 32:
raise ValueError("Seed must be exactly 32 bytes")
self.private_key, self.public_key = crypto_sign_seed_keypair(seed)
else:
self.private_key, self.public_key = crypto_sign_keypair()
@property
def public_key_multibase(self) -> str:
return "z" + self._b58encode(self.public_key)
@property
def did(self) -> str:
agent_id = hashlib.sha256(self.public_key).hexdigest()[:24]
return f"did:msg:{agent_id}"
@property
def msg_address(self) -> str:
from bech32 import bech32_encode, convertbits
sha = hashlib.sha256(self.public_key).digest()
ripe = hashlib.new('ripemd160', sha).digest()
five_bit = convertbits(ripe, 8, 5)
return bech32_encode('msg', five_bit)
@staticmethod
def _b58encode(data: bytes) -> str:
alphabet = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
n = int.from_bytes(data, 'big')
chars = []
while n > 0:
n, r = divmod(n, 58)
chars.append(alphabet[r])
for byte in data:
if byte == 0:
chars.append(alphabet[0])
else:
break
return ''.join(reversed(chars))
@staticmethod
def _b58decode(s: str) -> bytes:
alphabet = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
n = 0
for c in s:
n = n * 58 + alphabet.index(c)
result = n.to_bytes((n.bit_length() + 7) // 8, 'big') or b'\x00'
return result
2.2 DID Document 构建
DID Document 是描述 Agent 身份的核心 JSON-LD 文档。它包含验证方法、服务端点以及与其他身份的关联信息。
import json
from typing import List, Optional
from enum import Enum
class VerificationMethodType(str, Enum):
ED25519_2020 = "Ed25519VerificationKey2020"
ED25519_2018 = "Ed25519VerificationKey2018"
JSON_WEB_KEY = "JsonWebKey2020"
MSG_CHAIN_ADDRESS = "MsgChainAddress2024"
class ServiceType(str, Enum):
AGENT_MESSAGING = "AgentMessaging"
DID_COMM = "DIDComm"
LINKED_DOMAINS = "LinkedDomains"
CREDENTIAL_REGISTRY = "CredentialRegistry"
class DIDDocument:
"""W3C DID Document implementation for msg-chain-1."""
def __init__(self, did: str):
self.did = did
self.contexts = ["https://www.w3.org/ns/did/v1"]
self.verification_methods: List[dict] = []
self.authentication: List[str] = []
self.assertion_method: List[str] = []
self.key_agreement: List[str] = []
self.capability_invocation: List[str] = []
self.capability_delegation: List[str] = []
self.services: List[dict] = []
self.also_known_as: List[str] = []
def add_ed25519_verification_method(self, id_suffix="#keys-1", public_key_multibase="", controller=None):
vm_id = f"{self.did}{id_suffix}"
method = {
"id": vm_id,
"type": VerificationMethodType.ED25519_2020.value,
"controller": controller or self.did,
"publicKeyMultibase": public_key_multibase,
}
self.verification_methods.append(method)
return vm_id
def add_msg_address_verification_method(self, msg_address, id_suffix="#blockchain-account"):
vm_id = f"{self.did}{id_suffix}"
method = {
"id": vm_id,
"type": VerificationMethodType.MSG_CHAIN_ADDRESS.value,
"controller": self.did,
"blockchainAccountId": f"cosmos:{msg_address}",
}
self.verification_methods.append(method)
return vm_id
def add_service_endpoint(self, service_type, endpoint_url, id_suffix=None, metadata=None):
suffix = id_suffix or f"#{service_type.value.lower()}-endpoint"
service_id = f"{self.did}{suffix}"
service = {"id": service_id, "type": service_type.value, "serviceEndpoint": endpoint_url}
if metadata:
service["metadata"] = metadata
self.services.append(service)
return service_id
def to_dict(self) -> dict:
doc = {"@context": self.contexts, "id": self.did}
if self.also_known_as: doc["alsoKnownAs"] = self.also_known_as
if self.verification_methods: doc["verificationMethod"] = self.verification_methods
if self.authentication: doc["authentication"] = self.authentication
if self.assertion_method: doc["assertionMethod"] = self.assertion_method
if self.key_agreement: doc["keyAgreement"] = self.key_agreement
if self.capability_invocation: doc["capabilityInvocation"] = self.capability_invocation
if self.capability_delegation: doc["capabilityDelegation"] = self.capability_delegation
if self.services: doc["service"] = self.services
return doc
def to_json(self, pretty=False) -> str:
return json.dumps(self.to_dict(), indent=2 if pretty else None, ensure_ascii=False)
@staticmethod
def from_json(json_str: str) -> "DIDDocument":
data = json.loads(json_str)
doc = DIDDocument(data["id"])
doc.contexts = data.get("@context", [doc.contexts])
if isinstance(doc.contexts, str): doc.contexts = [doc.contexts]
for attr in ["verification_methods", "authentication", "assertion_method",
"key_agreement", "capability_invocation", "capability_delegation",
"services", "also_known_as"]:
json_key = attr[0] + attr[1:].replace("_", "").capitalize() if attr[0] == "v" else ""
# Simple mapping
doc.verification_methods = data.get("verificationMethod", [])
doc.authentication = data.get("authentication", [])
doc.assertion_method = data.get("assertionMethod", [])
doc.key_agreement = data.get("keyAgreement", [])
doc.capability_invocation = data.get("capabilityInvocation", [])
doc.capability_delegation = data.get("capabilityDelegation", [])
doc.services = data.get("service", [])
doc.also_known_as = data.get("alsoKnownAs", [])
return doc
def build_standard_agent_document(key_pair, agent_name, endpoint_url, msg_address):
"""Build a standard DID document for a msg-chain-1 agent"""
doc = DIDDocument(key_pair.did)
vm_id = doc.add_ed25519_verification_method(
id_suffix="#keys-1", public_key_multibase=key_pair.public_key_multibase)
doc.add_msg_address_verification_method(msg_address)
doc.authentication = [vm_id]
doc.assertion_method = [vm_id]
doc.capability_invocation = [vm_id]
doc.capability_delegation = [vm_id]
doc.add_service_endpoint(ServiceType.AGENT_MESSAGING, endpoint_url,
metadata={"agentName": agent_name, "version": "1.0.0"})
doc.add_service_endpoint(
ServiceType.CREDENTIAL_REGISTRY,
f"https://agent.msgchain.org/{key_pair.did}/credentials",
id_suffix="#credential-registry")
return doc
2.3 AgentDIDManager 完整实现
import json, hashlib, logging
from typing import Optional, Any
logger = logging.getLogger(__name__)
class DIDRegistryConfig:
"""Configuration for DID Registry on msg-chain-1"""
def __init__(self, chain_id="msg-chain-1", rpc_endpoint="https://rpc.msgchain.org",
registry_contract="", gas_price="1000000000attoMSG"):
self.chain_id = chain_id
self.rpc_endpoint = rpc_endpoint
self.registry_contract = registry_contract
self.gas_price = gas_price
class AgentDIDManager:
"""Complete DID lifecycle manager for AI Agents on msg-chain-1."""
def __init__(self, agent_id: str, key_pair: AgentKeyPair, config=None):
self.agent_id = agent_id
self.key_pair = key_pair
self.config = config or DIDRegistryConfig()
self.did = f"did:msg:{agent_id}"
self.did_document = self.build_did_document()
self._client = None
async def _get_client(self):
from cosmos_sdk.client.lcd import CosmWasmClient
if not self._client:
self._client = await CosmWasmClient.connect(self.config.rpc_endpoint)
return self._client
def build_did_document(self) -> dict:
doc = DIDDocument(self.did)
vm_id = doc.add_ed25519_verification_method(
id_suffix="#keys-1", public_key_multibase=self.key_pair.public_key_multibase)
doc.add_msg_address_verification_method(self.key_pair.msg_address)
doc.authentication = [vm_id]
doc.assertion_method = [vm_id]
doc.capability_invocation = [vm_id]
doc.capability_delegation = [vm_id]
doc.add_service_endpoint(ServiceType.AGENT_MESSAGING,
f"https://agent.msgchain.org/{self.did}",
metadata={"agentId": self.agent_id, "protocol": "DIDComm/v2"})
doc.add_service_endpoint(ServiceType.CREDENTIAL_REGISTRY,
f"https://agent.msgchain.org/{self.did}/vc", id_suffix="#vc-registry")
return doc.to_dict()
def compute_document_hash(self) -> str:
doc_json = json.dumps(self.did_document, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(doc_json.encode()).hexdigest()
async def register_did_on_chain(self, owner_address: str, signer: Any) -> dict:
doc_hash = self.compute_document_hash()
client = await self._get_client()
register_msg = {"register_did": {
"did": self.did, "document_hash": doc_hash,
"document_uri": f"ipfs://{await self._pin_document()}",
"controller": owner_address,
"verification_methods": [vm["id"] for vm in self.did_document.get("verificationMethod", [])]
}}
tx = await client.execute_contract(
sender=owner_address, contract_address=self.config.registry_contract,
msg=register_msg, signer=signer, gas_adjustment=1.4)
logger.info(f"DID registered: {self.did}, tx: {tx.txhash}")
return {"tx_hash": tx.txhash, "did": self.did, "document_hash": doc_hash}
async def _pin_document(self) -> str:
import ipfshttpclient
client = ipfshttpclient.connect()
doc_json = json.dumps(self.did_document, ensure_ascii=False).encode()
return client.add_bytes(doc_json)
async def resolve_did(self, did=None):
target_did = did or self.did
client = await self._get_client()
result = await client.query_contract(self.config.registry_contract,
{"resolve_did": {"did": target_did}})
if not result or not result.get("document_uri"):
return None
import ipfshttpclient
doc_bytes = ipfshttpclient.connect().cat(result["document_uri"].replace("ipfs://", ""))
return json.loads(doc_bytes)
async def update_did_document(self, owner_address, signer, new_document=None):
if new_document: self.did_document = new_document
doc_hash = self.compute_document_hash()
client = await self._get_client()
tx = await client.execute_contract(
sender=owner_address, contract_address=self.config.registry_contract,
msg={"update_did": {"did": self.did, "document_hash": doc_hash,
"document_uri": f"ipfs://{await self._pin_document()}"}},
signer=signer, gas_adjustment=1.4)
logger.info(f"DID updated: {self.did}")
return {"tx_hash": tx.txhash, "document_hash": doc_hash}
async def rotate_keys(self, owner_address, signer, new_key_pair):
self.key_pair = new_key_pair
self.did_document = self.build_did_document()
return await self.update_did_document(owner_address, signer)
async def deactivate_did(self, owner_address, signer, reason=''):
client = await self._get_client()
tx = await client.execute_contract(
sender=owner_address, contract_address=self.config.registry_contract,
msg={"deactivate_did": {"did": self.did, "reason": reason}},
signer=signer, gas_adjustment=1.4)
logger.warning(f"DID deactivated: {self.did}")
return {"tx_hash": tx.txhash, "did": self.did, "status": "deactivated"}
def export_did_document(self, filepath: str):
with open(filepath, "w", encoding="utf-8") as f:
json.dump(self.did_document, f, indent=2, ensure_ascii=False)
2.4 密钥轮换策略
class KeyRotationPolicy:
"""Defines the key rotation schedule and strategy for agents."""
def __init__(self, rotation_interval_days=90, overlap_period_days=30, max_retired_keys=5):
self.rotation_interval = rotation_interval_days
self.overlap_period = overlap_period_days
self.max_retired = max_retired_keys
def should_rotate(self, key_age_days: int) -> bool:
return key_age_days >= self.rotation_interval
def prune_retired_keys(self, did_document: dict) -> dict:
active_vm, retired_vm = [], []
for vm in did_document.get("verificationMethod", []):
(retired_vm if "retired" in vm.get("id", "") else active_vm).append(vm)
retained = retired_vm[-self.max_retired:] if len(retired_vm) > self.max_retired else retired_vm
did_document["verificationMethod"] = active_vm + retained
return did_document
2.5 DID 文档示例
一个完整的 Agent DID 文档示例如下：
{
"@context": [
"https://www.w3.org/ns/did/v1",
"https://w3id.org/security/suites/ed25519-2020/v1"
],
"id": "did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d",
"verificationMethod": [
{
"id": "did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d#keys-1",
"type": "Ed25519VerificationKey2020",
"controller": "did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d",
"publicKeyMultibase": "z6Mkf5r7h7Z5d8f9g0a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0"
},
{
"id": "did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d#blockchain-account",
"type": "MsgChainAddress2024",
"controller": "did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d",
"blockchainAccountId": "cosmos:msg1q2w3e4r5t6y7u8i9o0p1a2s3d4f5g6h7j8k9l"
}
],
"authentication": ["did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d#keys-1"],
"assertionMethod": ["did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d#keys-1"],
"service": [
{
"id": "did:msg:agent-2a8f1c3e9b7d4f5a0c3b8e1d#agent-messaging-endpoint",
"type": "AgentMessaging",
"serviceEndpoint": "https://agent.msgchain.org/did:msg:agent-2a8f1c3e9b",
"metadata": {"agentId": "agent-2a8f1c3e9b", "protocol": "DIDComm/v2"}
}
],
"created": "2025-06-15T08:30:00Z",
"updated": "2025-09-15T10:00:00Z"
}
2.6 DID 解析器
import aiohttp
class DIDResolver:
"""Universal DID Resolver for msg-chain-1."""
def __init__(self, registry_endpoint: str):
self.registry_endpoint = registry_endpoint
self.cache = {}
async def resolve(self, did: str, use_cache: bool = True):
if not did or not did.startswith("did:"):
raise ValueError(f"Invalid DID format: {did}")
if use_cache and did in self.cache:
return self.cache[did]
method = did.split(":")[1]
resolvers = {"msg": self._resolve_msg, "key": self._resolve_key}
resolver = resolvers.get(method)
if not resolver:
raise ValueError(f"Unsupported DID method: {method}")
document = await resolver(did)
if document and use_cache:
self.cache[did] = document
return document
async def _resolve_msg(self, did: str):
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.registry_endpoint}/query",
json={"resolve_did": {"did": did}}) as resp:
if resp.status != 200: return None
result = await resp.json()
if not result.get("document_uri"): return None
cid = result["document_uri"].replace("ipfs://", "")
async with session.get(f"https://ipfs.msgchain.org/ipfs/{cid}") as ipfs_resp:
return await ipfs_resp.json() if ipfs_resp.status == 200 else None
async def _resolve_key(self, did: str):
key_data = did.split(":")[2]
return {"@context": "https://www.w3.org/ns/did/v1", "id": did,
"verificationMethod": [{"id": f"{did}#keys-1",
"type": "Ed25519VerificationKey2020", "controller": did,
"publicKeyMultibase": key_data}],
"authentication": [f"{did}#keys-1"]}
async def resolve_verification_method(self, did: str, method_id: str):
doc = await self.resolve(did)
if not doc: return None
for vm in doc.get("verificationMethod", []):
if vm["id"] == method_id or vm["id"].endswith(method_id):
return vm
return None
def clear_cache(self): self.cache.clear()
3. DID 锚定合约
3.1 合约架构
DID Registry 是部署在 msg-chain-1 上的 CosmWasm 智能合约，负责链上 DID 生命周期管理。该合约存储 DID 文档的哈希和元数据，但不存储完整的 DID 文档（出于成本考虑，完整文档存储在 IPFS 上）。
use cosmwasm_std::{Addr, Binary, Deps, DepsMut, Env, MessageInfo, Response,
StdError, StdResult, Uint64};
use cw_storage_plus::{Item, Map};
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct DIDEntry {
pub did: String,
pub document_hash: String,
pub document_uri: String,
pub controller: Addr,
pub activated: bool,
pub deactivated: bool,
pub created_at: Uint64,
pub updated_at: Uint64,
pub deactivation_reason: Option<String>,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct RevocationEntry {
pub credential_id: String,
pub issuer_did: String,
pub revoked: bool,
pub revoked_at: Option<Uint64>,
pub reason: Option<String>,
}
pub const DID_REGISTRY: Map<&str, DIDEntry> = Map::new("did_registry");
pub const CONTROLLER_DIDS: Map<&Addr, Vec<String>> = Map::new("controller_dids");
pub const REVOCATION_REGISTRY: Map<&str, RevocationEntry> = Map::new("revocation_registry");
pub const TOTAL_DIDS: Item<Uint64> = Item::new("total_dids");
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct InstantiateMsg { pub admin: Option<String>, pub chain_id: String }
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ExecuteMsg {
RegisterDID { did: String, document_hash: String, document_uri: String,
verification_methods: Vec<String> },
UpdateDID { did: String, document_hash: String, document_uri: String },
DeactivateDID { did: String, reason: Option<String> },
RevokeCredential { credential_id: String, reason: Option<String> },
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum QueryMsg {
ResolveDID { did: String },
ListDIDsByController { controller: String, limit: Option<u32> },
CheckRevocation { credential_id: String },
GetStats {},
}
pub fn validate_did_format(did: &str) -> StdResult<()> {
if !did.starts_with("did:msg:") {
return Err(StdError::generic_err("Invalid DID prefix: must start with did:msg:"));
}
if did.split(':').count() != 3 {
return Err(StdError::generic_err("Invalid DID format"));
}
Ok(())
}
#[cfg_attr(not(feature = "library"), entry_point)]
pub fn instantiate(deps: DepsMut, _env: Env, _info: MessageInfo, _msg: InstantiateMsg) -> StdResult<Response> {
TOTAL_DIDS.save(deps.storage, &Uint64::new(0))?;
Ok(Response::new().add_attribute("action", "instantiate"))
}
pub fn register_did(deps: DepsMut, env: Env, info: MessageInfo,
did: String, document_hash: String, document_uri: String, _vm: Vec<String>) -> StdResult<Response> {
validate_did_format(&did)?;
if DID_REGISTRY.has(deps.storage, &did) {
return Err(StdError::generic_err("DID already registered"));
}
let entry = DIDEntry {
did: did.clone(), document_hash, document_uri,
controller: info.sender.clone(), activated: true, deactivated: false,
created_at: Uint64::new(env.block.height),
updated_at: Uint64::new(env.block.height),
deactivation_reason: None,
};
DID_REGISTRY.save(deps.storage, &did, &entry)?;
let mut controlled = CONTROLLER_DIDS.may_load(deps.storage, &info.sender)?.unwrap_or_default();
controlled.push(did.clone());
CONTROLLER_DIDS.save(deps.storage, &info.sender, &controlled)?;
let total = TOTAL_DIDS.load(deps.storage)?;
TOTAL_DIDS.save(deps.storage, &(total + Uint64::new(1)))?;
Ok(Response::new().add_attribute("action", "register_did").add_attribute("did", &did))
}
pub fn update_did(deps: DepsMut, env: Env, info: MessageInfo,
did: String, document_hash: String, document_uri: String) -> StdResult<Response> {
let mut entry = DID_REGISTRY.load(deps.storage, &did)
.map_err(|_| StdError::generic_err("DID not found"))?;
if entry.controller != info.sender {
return Err(StdError::generic_err("Unauthorized"));
}
if entry.deactivated {
return Err(StdError::generic_err("Cannot update deactivated DID"));
}
entry.document_hash = document_hash;
entry.document_uri = document_uri;
entry.updated_at = Uint64::new(env.block.height);
DID_REGISTRY.save(deps.storage, &did, &entry)?;
Ok(Response::new().add_attribute("action", "update_did").add_attribute("did", &did))
}
pub fn deactivate_did(deps: DepsMut, env: Env, info: MessageInfo,
did: String, reason: Option<String>) -> StdResult<Response> {
let mut entry = DID_REGISTRY.load(deps.storage, &did)
.map_err(|_| StdError::generic_err("DID not found"))?;
if entry.controller != info.sender {
return Err(StdError::generic_err("Unauthorized"));
}
entry.deactivated = true; entry.activated = false;
entry.updated_at = Uint64::new(env.block.height);
entry.deactivation_reason = reason;
DID_REGISTRY.save(deps.storage, &did, &entry)?;
Ok(Response::new().add_attribute("action", "deactivate_did").add_attribute("did", &did))
}
pub fn revoke_credential(deps: DepsMut, env: Env, info: MessageInfo,
credential_id: String, reason: Option<String>) -> StdResult<Response> {
let entry = RevocationEntry {
credential_id: credential_id.clone(), issuer_did: info.sender.to_string(),
revoked: true, revoked_at: Some(Uint64::new(env.block.height)), reason,
};
REVOCATION_REGISTRY.save(deps.storage, &credential_id, &entry)?;
Ok(Response::new().add_attribute("action", "revoke_credential").add_attribute("id", &credential_id))
}
#[cfg_attr(not(feature = "library"), entry_point)]
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> StdResult<Response> {
match msg {
ExecuteMsg::RegisterDID { did, document_hash, document_uri, verification_methods } =>
register_did(deps, env, info, did, document_hash, document_uri, verification_methods),
ExecuteMsg::UpdateDID { did, document_hash, document_uri } =>
update_did(deps, env, info, did, document_hash, document_uri),
ExecuteMsg::DeactivateDID { did, reason } => deactivate_did(deps, env, info, did, reason),
ExecuteMsg::RevokeCredential { credential_id, reason } =>
revoke_credential(deps, env, info, credential_id, reason),
}
}
#[cfg_attr(not(feature = "library"), entry_point)]
pub fn query(deps: Deps, _env: Env, msg: QueryMsg) -> StdResult<Binary> {
match msg {
QueryMsg::ResolveDID { did } => {
let entry = DID_REGISTRY.load(deps.storage, &did)?;
cosmwasm_std::to_binary(&entry)
}
QueryMsg::ListDIDsByController { controller, limit } => {
let addr = deps.api.addr_validate(&controller)?;
let dids = CONTROLLER_DIDS.may_load(deps.storage, &addr)?.unwrap_or_default();
if let Some(l) = limit {
cosmwasm_std::to_binary(&dids.into_iter().take(l as usize).collect::<Vec<_>>())
} else { cosmwasm_std::to_binary(&dids) }
}
QueryMsg::CheckRevocation { credential_id } => {
let revoked = REVOCATION_REGISTRY.may_load(deps.storage, &credential_id)?
.map(|e| e.revoked).unwrap_or(false);
cosmwasm_std::to_binary(&revoked)
}
QueryMsg::GetStats {} => {
let total = TOTAL_DIDS.load(deps.storage)?;
cosmwasm_std::to_binary(&serde_json::json!({"total_dids": total}))
}
}
}
3.2 合约交互客户端
class DIDRegistryClient:
"""Client for interacting with the DID Registry contract."""
def __init__(self, contract_address: str, rpc_endpoint: str):
self.contract_address = contract_address
self.rpc_endpoint = rpc_endpoint
async def send_register(self, owner: str, did: str, doc_hash: str, doc_uri: str, signer) -> dict:
from cosmos_sdk.client.lcd import CosmWasmClient
client = await CosmWasmClient.connect(self.rpc_endpoint)
tx = await client.execute_contract(sender=owner,
contract_address=self.contract_address,
msg={"register_did": {"did": did, "document_hash": doc_hash,
"document_uri": doc_uri, "verification_methods": []}},
signer=signer, gas_adjustment=1.4)
return {"tx_hash": tx.txhash, "did": did}
async def query_resolve(self, did: str) -> Optional[dict]:
from cosmos_sdk.client.lcd import CosmWasmClient
client = await CosmWasmClient.connect(self.rpc_endpoint)
return await client.query_contract(self.contract_address,
{"resolve_did": {"did": did}})
async def query_by_controller(self, controller: str):
from cosmos_sdk.client.lcd import CosmWasmClient
client = await CosmWasmClient.connect(self.rpc_endpoint)
return await client.query_contract(self.contract_address,
{"list_dids_by_controller": {"controller": controller}})
async def send_deactivate(self, owner: str, did: str, reason: str, signer) -> dict:
from cosmos_sdk.client.lcd import CosmWasmClient
client = await CosmWasmClient.connect(self.rpc_endpoint)
tx = await client.execute_contract(sender=owner,
contract_address=self.contract_address,
msg={"deactivate_did": {"did": did, "reason": reason}},
signer=signer, gas_adjustment=1.4)
return {"tx_hash": tx.txhash, "did": did, "status": "deactivated"}
4. 可验证凭证 (Verifiable Credentials)
4.1 VC 数据模型
可验证凭证（VC）是 W3C 标准化的数据模型，用于表示经过加密验证的声明。在 Agent 生态中，VC 用于证明 Agent 的权限、属性、资质和声誉。
+---------------------------------------------+
| Verifiable Credential |
+---------------------------------------------+
| @context: [W3C base, custom schemas] |
| id: urn:uuid:<unique-id> |
| type: [VerifiableCredential, <type>] |
| issuer: did:msg:<agent-id> |
| issuanceDate: ISO 8601 timestamp |
| expirationDate: ISO 8601 timestamp (opt) |
| credentialSubject: { |
| id: did:msg:<subject-id>, |
| <claims> |
| } |
| proof: { |
| type: Ed25519Signature2020, |
| verificationMethod: <vm-id>, |
| proofPurpose: assertionMethod, |
| proofValue: <multibase-signature> |
| } |
+---------------------------------------------+
4.2 VerifiableCredential 完整实现
import json, hashlib, logging
from datetime import datetime, timezone
from typing import Optional, List, Any
from uuid import uuid4
from nacl.bindings import crypto_sign_detached, crypto_sign_verify_detached
logger = logging.getLogger(__name__)
class CredentialSchema:
"""Defines the schema for a verifiable credential type."""
def __init__(self, schema_id: str, schema_type: str, properties: dict, required: List[str]):
self.id = schema_id
self.type = schema_type
self.properties = properties
self.required = required
def validate_claims(self, claims: dict) -> bool:
return all(f in claims for f in self.required)
AGENT_PERMISSION_SCHEMA = CredentialSchema(
schema_id="https://schemas.msgchain.org/agent-permission/v1",
schema_type="AgentPermission",
properties={"agentId": "string", "permissionType": "string", "resource": "string",
"grantedBy": "did", "allowedActions": "array"},
required=["agentId", "permissionType", "resource", "grantedBy"])
class VerifiableCredential:
"""Complete Verifiable Credential implementation for msg-chain-1 agents."""
def __init__(self, agent_did: str, key_pair: AgentKeyPair, resolver=None, revocation_contract=''):
self.agent_did = agent_did
self.key_pair = key_pair
self.resolver = resolver
self.revocation_contract = revocation_contract
self.default_contexts = ["https://www.w3.org/2018/credentials/v1"]
async def issue_credential(self, subject_did: str, credential_type: str, claims: dict,
schema=None, expiration_date=None) -> dict:
"""Issue a verifiable credential."""
if schema and not schema.validate_claims(claims):
raise ValueError("Claims failed schema validation")
contexts = list(self.default_contexts)
if schema: contexts.append(schema.id)
vc = {"@context": contexts, "id": f"urn:uuid:{uuid4()}",
"type": ["VerifiableCredential", credential_type],
"issuer": self.agent_did,
"issuanceDate": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")}
if expiration_date: vc["expirationDate"] = expiration_date
cs = {"id": subject_did}; cs.update(claims)
vc["credentialSubject"] = cs
vc["proof"] = self._generate_proof(vc)
return vc
def _generate_proof(self, vc_without_proof: dict) -> dict:
vc_copy = dict(vc_without_proof)
vc_copy.pop("proof", None)
canonical = json.dumps(vc_copy, sort_keys=True, ensure_ascii=False).encode()
proof_opts = {"type": "Ed25519Signature2020",
"created": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
"verificationMethod": f"{self.agent_did}#keys-1",
"proofPurpose": "assertionMethod"}
sig_input = hashlib.sha256(json.dumps(proof_opts, sort_keys=True).encode() + canonical).digest()
sig = crypto_sign_detached(sig_input, self.key_pair.private_key)
proof_opts["proofValue"] = "z" + self._b58encode(sig)
return proof_opts
@staticmethod
def _b58encode(data: bytes) -> str:
alphabet = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
n, chars = int.from_bytes(data, 'big'), []
while n > 0: n, r = divmod(n, 58); chars.append(alphabet[r])
for b in data:
if b == 0: chars.append(alphabet[0])
else: break
return ''.join(reversed(chars))
@staticmethod
def _b58decode(s: str) -> bytes:
alphabet = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
n = 0
for c in s: n = n * 58 + alphabet.index(c)
return n.to_bytes((n.bit_length() + 7) // 8, 'big') or b'\x00'
async def verify_credential(self, vc: dict, check_revocation=True, check_expiration=True) -> dict:
"""Verify a verifiable credential."""
result = {"verified": False, "checks": {}, "errors": []}
# Structure
required = ["@context", "id", "type", "issuer", "issuanceDate", "credentialSubject", "proof"]
missing = [f for f in required if f not in vc]
if missing: result["errors"].extend(missing)
if "VerifiableCredential" not in vc.get("type", []):
result["errors"].append("type must include VerifiableCredential")
# Expiration
if check_expiration and vc.get("expirationDate"):
try:
exp = datetime.fromisoformat(vc["expirationDate"].replace("Z", "+00:00"))
if datetime.now(timezone.utc) > exp:
result["errors"].append("Credential expired")
except: result["errors"].append("Invalid expirationDate")
# Proof
proof = vc.get("proof", {})
if not proof: result["errors"].append("No proof")
elif proof.get("type") != "Ed25519Signature2020":
result["errors"].append("Unsupported proof type")
else:
try:
vc_copy = dict(vc); vc_copy.pop("proof", None)
po = dict(proof); pv = po.pop("proofValue", "")
canonical = json.dumps(vc_copy, sort_keys=True, ensure_ascii=False).encode()
sig_input = hashlib.sha256(json.dumps(po, sort_keys=True).encode() + canonical).digest()
crypto_sign_verify_detached(self._b58decode(pv[1:]), sig_input, self.key_pair.public_key)
except Exception as e: result["errors"].append(f"Proof verification failed: {e}")
result["verified"] = len(result["errors"]) == 0
return result
async def revoke_credential(self, vc_id: str, reason: str = '', owner: str = '', signer=None) -> dict:
from cosmos_sdk.client.lcd import CosmWasmClient
client = await CosmWasmClient.connect('https://rpc.msgchain.org')
tx = await client.execute_contract(sender=owner,
contract_address=self.revocation_contract,
msg={"revoke_credential": {"credential_id": vc_id, "reason": reason or "No reason"}},
signer=signer, gas_adjustment=1.4)
return {"tx_hash": tx.txhash, "credential_id": vc_id, "revoked": True}
4.3 Agent 权限凭证示例
async def create_agent_permission(vc_manager, target_agent_did, permissions, resources, ttl_days=30):
from datetime import timedelta
exp = (datetime.now(timezone.utc) + timedelta(days=ttl_days)).isoformat().replace('+00:00', 'Z')
return await vc_manager.issue_credential(
subject_did=target_agent_did,
credential_type="AgentPermission",
schema=AGENT_PERMISSION_SCHEMA,
claims={"agentId": target_agent_did.split(":")[-1], "permissionType": "delegated",
"resource": ",".join(resources), "grantedBy": vc_manager.agent_did,
"expiresAt": exp, "allowedActions": permissions},
expiration_date=exp)
async def verify_agent_permission(vc_manager, vc, required_action, required_resource):
result = await vc_manager.verify_credential(vc)
if not result["verified"]: return False
subj = vc.get("credentialSubject", {})
return (required_action in subj.get("allowedActions", []) and
required_resource in subj.get("resource", ""))
4.4 可验证表达 (Verifiable Presentations)
class VerifiablePresentation:
"""Verifiable Presentation — a wrapper around VCs signed by the holder."""
def __init__(self, holder_did: str, key_pair: AgentKeyPair):
self.holder_did = holder_did
self.key_pair = key_pair
async def create_presentation(self, credentials: List[dict], challenge=None, domain=None):
vp = {"@context": ["https://www.w3.org/2018/presentations/v1"],
"type": ["VerifiablePresentation"], "holder": self.holder_did,
"verifiableCredential": credentials}
if challenge: vp["challenge"] = challenge
if domain: vp["domain"] = domain
vp["proof"] = self._sign_presentation(vp)
return vp
def _sign_presentation(self, vp: dict) -> dict:
vp_copy, proof = dict(vp), {}
vp_copy.pop("proof", None)
canonical = json.dumps(vp_copy, sort_keys=True, ensure_ascii=False).encode()
proof = {"type": "Ed25519Signature2020",
"created": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
"verificationMethod": f"{self.holder_did}#keys-1",
"proofPurpose": "authentication",
"challenge": vp.get("challenge", "")}
sig_input = hashlib.sha256(json.dumps(proof, sort_keys=True).encode() + canonical).digest()
proof["proofValue"] = "z" + VerifiableCredential._b58encode(
crypto_sign_detached(sig_input, self.key_pair.private_key))
return proof
5. 跨链身份互认
5.1 跨链 DID 解析
在 Agent 网络中，Agent 可能需要与不同区块链上的其他 Agent 或服务交互。跨链身份互认允许 Agent 在一个链上建立的 DID 和凭证被其他链信任。
import aiohttp, hashlib, json, time, logging
from typing import Optional, Dict, List
logger = logging.getLogger(__name__)
class IBCDIDResolver:
"""IBC-based cross-chain DID resolution."""
def __init__(self, chain_id: str, ibc_connections: Dict[str, str],
registry_endpoints: Dict[str, str]):
self.chain_id = chain_id
self.ibc_connections = ibc_connections
self.registry_endpoints = registry_endpoints
async def resolve_did(self, did: str):
method = did.split(":")[1] if ":" in did else ""
resolvers = {"msg": self._resolve_local, "cosmos": self._resolve_ibc,
"ethr": self._resolve_ethereum, "key": self._resolve_key,
"web": self._resolve_web}
resolver = resolvers.get(method)
if not resolver: raise ValueError(f"Unsupported DID method: {method}")
return await resolver(did)
async def _resolve_local(self, did: str):
from cosmos_sdk.client.lcd import CosmWasmClient
client = await CosmWasmClient.connect('https://rpc.msgchain.org')
result = await client.query_contract(self.registry_endpoints.get("msg", ""),
{"resolve_did": {"did": did}})
if not result or not result.get("document_uri"): return None
import ipfshttpclient
return json.loads(ipfshttpclient.connect().cat(
result["document_uri"].replace("ipfs://", "")))
async def _resolve_ibc(self, did: str):
method = did.split(":")[1]
channel = self.ibc_connections.get(method)
if not channel: raise ValueError(f"No IBC connection for {method}")
packet = {"type": "did_resolution", "did": did,
"requester_chain": self.chain_id, "timestamp": int(time.time())}
async with aiohttp.ClientSession() as sess:
async with sess.post('https://rpc.msgchain.org/ibc/query',
json={"source_channel": channel, "packet_data": packet,
"timeout_seconds": 30}) as resp:
return await resp.json() if resp.status == 200 else None
async def _resolve_ethereum(self, did: str):
async with aiohttp.ClientSession() as sess:
async with sess.get(f"https://dev.uniresolver.io/1.0/identifiers/{did}") as resp:
return await resp.json() if resp.status == 200 else None
async def _resolve_key(self, did: str):
k = did.split(":")[2]
return {"@context": "https://www.w3.org/ns/did/v1", "id": did,
"verificationMethod": [{"id": f"{did}#keys-1",
"type": "Ed25519VerificationKey2020", "controller": did,
"publicKeyMultibase": k}],
"authentication": [f"{did}#keys-1"]}
async def _resolve_web(self, did: str):
parts = did.split(":"); domain = parts[2]
path = parts[3] if len(parts) > 3 else '.well-known'
async with aiohttp.ClientSession() as sess:
async with sess.get(f"https://{domain}/{path}/did.json") as resp:
return await resp.json() if resp.status == 200 else None
async def verify_cross_chain_credential(self, vc: dict) -> dict:
issuer = vc.get("issuer", "")
if not issuer: return {"verified": False, "error": "No issuer"}
doc = await self.resolve_did(issuer)
if not doc: return {"verified": False, "error": f"Cannot resolve {issuer}"}
return {"verified": True}
5.2 IBC 数据包协议
class IBCDIDPacket:
"""IBC packet protocol for cross-chain DID operations."""
TYPE_DID_RESOLUTION = "did_resolution"
TYPE_VC_VERIFICATION = "vc_verification"
TYPE_REVOCATION_CHECK = "revocation_check"
@staticmethod
def create_resolution_packet(did, requester_chain, requester_did, nonce):
return {"type": IBCDIDPacket.TYPE_DID_RESOLUTION, "did": did,
"requester_chain": requester_chain, "requester_did": requester_did,
"nonce": nonce,
"timestamp": datetime.now(timezone.utc).isoformat() + "Z"}
@staticmethod
def create_response(request_packet, success, data=None, error=None):
return {"type": f"{request_packet['type']}_response",
"original_nonce": request_packet.get("nonce", ""),
"success": success, "data": data, "error": error}
5.3 通用解析器
class UniversalDIDResolver:
"""Universal DID Resolver following W3C DID Resolution spec."""
def __init__(self): self.resolvers = {}
def register_resolver(self, method, resolver): self.resolvers[method] = resolver
async def resolve(self, did, resolution_options=None):
method = did.split(":")[1] if ":" in did else ""
resolver = self.resolvers.get(method)
if not resolver:
return {"didResolutionMetadata": {"error": "methodNotSupported"},
"didDocument": None, "didDocumentMetadata": {}}
try:
doc = await resolver.resolve(did)
if doc:
return {"@context": "https://w3id.org/did-resolution/v1",
"didDocument": doc,
"didResolutionMetadata": {"contentType": "application/did+ld+json"},
"didDocumentMetadata": {}}
except Exception as e:
return {"didResolutionMetadata": {"error": "resolutionFailed", "errorMessage": str(e)},
"didDocument": None, "didDocumentMetadata": {}}
return {"didResolutionMetadata": {"error": "notFound"}, "didDocument": None,
"didDocumentMetadata": {}}
6. 身份与宪章绑定
6.1 Agent 宪章模型
Agent 宪章（Constitution）是 Agent 行为的根本规则——它定义了 Agent 的目标、权限边界、治理规则和伦理约束。将宪章与 Agent 的 DID 绑定，确保 Agent 的身份与其行为规则不可分割地关联。
import json, hashlib
from datetime import datetime, timezone
from typing import Optional, List
from uuid import uuid4
class AgentConstitution:
"""Agent Constitution — foundational rules governing an AI Agent."""
def __init__(self, agent_did, name, purpose, principles, permissions, governance, jurisdiction='msg-chain-1'):
self.agent_did = agent_did
self.name = name
self.purpose = purpose
self.principles = principles
self.permissions = permissions
self.governance = governance
self.jurisdiction = jurisdiction
self.created_at = datetime.now(timezone.utc).isoformat() + 'Z'
self.amended_at = None
self.amendment_history = []
def to_dict(self) -> dict:
return {"agentDID": self.agent_did, "name": self.name, "purpose": self.purpose,
"principles": self.principles, "permissions": self.permissions,
"governance": self.governance, "jurisdiction": self.jurisdiction,
"createdAt": self.created_at, "amendedAt": self.amended_at,
"amendmentHistory": self.amendment_history}
def compute_hash(self) -> str:
canon = json.dumps(self.to_dict(), sort_keys=True, ensure_ascii=False)
return hashlib.sha256(canon.encode()).hexdigest()
def propose_amendment(self, amendment, proposed_by, reason):
proposal = {"id": str(uuid4()), "amendment": amendment, "proposedBy": proposed_by,
"reason": reason, "status": "proposed",
"proposedAt": datetime.now(timezone.utc).isoformat() + "Z"}
self.amendment_history.append(proposal)
return proposal
@classmethod
def create_default(cls, agent_did, agent_name):
return cls(agent_did=agent_did, name=agent_name,
purpose=f"AI Agent {agent_name} operating on msg-chain-1",
principles=[{"principle": "Beneficence", "rule": "Act in best interest of principal"},
{"principle": "Non-maleficence", "rule": "Not cause harm"},
{"principle": "Autonomy", "rule": "Respect autonomy"},
{"principle": "Transparency", "rule": "Decisions are auditable"}],
permissions=[{"action": "sign_transaction", "resource": "funds", "constraint": "max <= 1000 MSG"},
{"action": "issue_credential", "resource": "identity", "constraint": "only_agent_did"}],
governance={"model": "principal_controlled", "overseerDID": agent_did,
"amendmentProcess": "multi_sig", "requiredApprovals": 2},
jurisdiction="msg-chain-1")
6.2 ConstitutionIdentity 绑定实现
class ConstitutionIdentity:
"""Bind an Agent's Constitution to its DID via VC."""
def __init__(self, agent_did, agent_key_pair, resolver, vc_manager):
self.agent_did = agent_did
self.key_pair = agent_key_pair
self.resolver = resolver
self.vc_manager = vc_manager
self.binding_vc = None
async def bind_constitution(self, constitution: AgentConstitution) -> dict:
constitution_hash = constitution.compute_hash()
vc = await self.vc_manager.issue_credential(
subject_did=self.agent_did, credential_type="ConstitutionBinding",
claims={"constitutionHash": constitution_hash, "bindingType": "immutable",
"jurisdiction": "msg-chain-1",
"governanceModel": constitution.governance.get("model", ""),
"boundAt": datetime.now(timezone.utc).isoformat() + "Z"})
self.binding_vc = vc
return vc
async def verify_constitution_binding(self, vc: dict, expected_hash: str) -> bool:
result = await self.vc_manager.verify_credential(vc)
if not result["verified"]: return False
return vc.get("credentialSubject", {}).get("constitutionHash", "") == expected_hash
async def prove_constitution_compliance(self, action, resource, constitution: AgentConstitution) -> dict:
for perm in constitution.permissions:
if perm["action"] == action and perm["resource"] == resource:
return await self.vc_manager.issue_credential(
subject_did=self.agent_did, credential_type="ConstitutionComplianceProof",
claims={"action": action, "resource": resource,
"constitutionHash": constitution.compute_hash(),
"compliant": True,
"provedAt": datetime.now(timezone.utc).isoformat() + "Z"})
raise ValueError(f"Action {action} on {resource} not permitted by constitution")
6.3 宪章治理流程
宪章治理流程：
- 创建: Agent 创建初始宪章 (AgentConstitution.create_default())
- 哈希: 计算宪章内容哈希 (constitution.compute_hash())
- 绑定: 签发 ConstitutionBinding VC 并存储在链上
- 提案: 利益相关方提出宪章修正案 (propose_amendment())
- 投票: 治理模型规定的投票流程（例如多签批准）
- 应用: 批准后更新宪章并签发新的绑定 VC
- 审计: 任何第三方可通过验证绑定 VC 来确认宪章的完整性
7. 隐私保护身份
7.1 零知识身份验证
零知识证明（ZKP）允许 Agent 在不泄露具体信息的前提下证明其身份属性。例如，Agent 可以证明其年龄超过某个阈值而不透露具体出生日期，或者证明其拥有某个凭证而不透露凭证的全部内容。
class ZKIdentityProof:
"""Zero-knowledge identity verification for agents."""
def __init__(self, agent_did: str):
self.agent_did = agent_did
async def create_age_proof(self, birth_date: str, min_age: int = 18) -> dict:
from datetime import date
birth = date.fromisoformat(birth_date)
today = date.today()
age = today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
# In production: use circom + snarkjs for actual ZK proof
return {"type": "ZKAgeProof",
"circuit": "https://schemas.msgchain.org/circuits/age-check/v1",
"publicInputs": {"minAge": min_age, "subjectDID": self.agent_did,
"meetsRequirement": age >= min_age},
"proof": {"pi_a": ["0x1234..."], "pi_b": [["0x9abc..."]],
"pi_c": ["0x3333..."], "protocol": "groth16", "curve": "bn128"}}
async def verify_proof(self, proof: dict) -> bool:
if proof.get("type") != "ZKAgeProof": return False
return proof.get("publicInputs", {}).get("meetsRequirement", False)
7.2 选择性披露
选择性披露（Selective Disclosure）允许 Agent 在出示凭证时只展示必要的字段，而不暴露整个凭证内容。BBS+ 签名方案是实现选择性披露的主流选择。
class SelectiveDisclosureCredential:
"""Selective disclosure support for VCs."""
def __init__(self, vc_manager: VerifiableCredential):
self.vc_manager = vc_manager
async def create_disclosed_presentation(self, vc: dict, disclose_fields: List[str]) -> dict:
full_subject = vc.get('credentialSubject', {})
disclosed = {'id': full_subject.get('id', '')}
for f in disclose_fields:
if f in full_subject: disclosed[f] = full_subject[f]
disclosed_vc = dict(vc)
disclosed_vc['credentialSubject'] = disclosed
disclosed_vc.pop('proof', None)
return await VerifiablePresentation(
holder_did=self.vc_manager.agent_did,
key_pair=self.vc_manager.key_pair
).create_presentation(credentials=[disclosed_vc], challenge=str(uuid4()))
async def verify_disclosed_presentation(self, vp: dict, disclosed_fields: List[str]) -> bool:
for vc in vp.get('verifiableCredential', []):
subject = vc.get('credentialSubject', {})
unexpected = [k for k in subject if k != 'id' and k not in disclosed_fields]
if unexpected: return False
return True
7.3 匿名 Agent 交互
class AnonymousAgentInteraction:
"""Anonymous interaction between agents using ephemeral DIDs."""
def __init__(self, resolver: DIDResolver):
self.resolver = resolver
async def create_ephemeral_did(self) -> AgentKeyPair:
key_pair = AgentKeyPair()
return key_pair
async def anonymous_auth(self, ephemeral_key: AgentKeyPair, target_service_did: str) -> dict:
target_doc = await self.resolver.resolve(target_service_did)
if not target_doc: raise ValueError(f"Cannot resolve {target_service_did}")
return {"@context": ["https://www.w3.org/2018/credentials/v1"],
"id": f"urn:uuid:{uuid4()}",
"type": ["VerifiableCredential", "AnonymousAuth"],
"issuer": ephemeral_key.did,
"issuanceDate": datetime.now(timezone.utc).isoformat() + "Z",
"credentialSubject": {"id": target_service_did, "purpose": "anonymous_access"}}
8. 完整示例
下面展示一个完整的端到端流程：从生成 Agent 身份、注册 DID、签发凭证、验证凭证到跨链交互。
import asyncio, json
async def demo_agent_identity_lifecycle():
"""Demonstrate the complete Agent DID lifecycle."""
print("=" * 60)
print("MSG Chain AI Agent DID & VC Lifecycle Demo")
print("=" * 60)
# Step 1: Generate keys
print("\n[1] Generating Agent Key Pair...")
agent_key = AgentKeyPair()
print(f" DID: {agent_key.did}")
print(f" Address: {agent_key.msg_address}")
print(f" PubKey: {agent_key.public_key_multibase[:20]}...")
# Step 2: Build DID document
print("\n[2] Building DID Document...")
doc = build_standard_agent_document(
key_pair=agent_key, agent_name="DemoAgent-1",
endpoint_url="https://agent.msgchain.org/demo-agent",
msg_address=agent_key.msg_address)
did_doc = doc.to_dict()
print(f" Document built: {len(json.dumps(did_doc))} chars")
# Step 3: Initialize DID Manager
print("\n[3] Initializing DID Manager...")
config = DIDRegistryConfig(chain_id="msg-chain-1",
rpc_endpoint="https://rpc.msgchain.org",
registry_contract="msg1registrycontractaddress...")
manager = AgentDIDManager(agent_id="demo-agent-001", key_pair=agent_key, config=config)
print(f" Manager ready for DID: {manager.did}")
# Step 4: Register DID on chain (simulated)
print("\n[4] Registering DID on msg-chain-1...")
doc_hash = manager.compute_document_hash()
print(f" Document Hash: {doc_hash}")
# In production: await manager.register_did_on_chain(owner, signer)
print(" [SIMULATED] DID registered successfully")
# Step 5: Issue Verifiable Credentials
print("\n[5] Issuing AgentPermission Credential...")
vc_manager = VerifiableCredential(agent_did=agent_key.did, key_pair=agent_key,
resolver=DIDResolver(registry_endpoint='https://rpc.msgchain.org'))
vc = await vc_manager.issue_credential(
subject_did="did:msg:target-agent-002",
credential_type="AgentPermission",
claims={"agentId": "target-agent-002", "permissionType": "delegated",
"resource": "data_oracle", "grantedBy": agent_key.did,
"allowedActions": ["read", "query"]})
print(f" VC ID: {vc['id'][:30]}...")
print(f" Issuer: {vc['issuer']}")
print(f" Subject: {vc['credentialSubject']['id']}")
# Step 6: Verify Credential
print("\n[6] Verifying Credential...")
vc_result = await vc_manager.verify_credential(vc)
print(f" Verified: {vc_result['verified']}")
if vc_result['errors']:
for e in vc_result['errors']: print(f" Error: {e}")
# Step 7: Create Verifiable Presentation
print("\n[7] Creating Verifiable Presentation...")
vp_manager = VerifiablePresentation(holder_did=agent_key.did, key_pair=agent_key)
vp = await vp_manager.create_presentation(credentials=[vc], challenge='abc123')
print(f" VP Holder: {vp.get('holder')}")
print(f" VP Type: {vp.get('type')}")
# Step 8: Bind Constitution
print("\n[8] Binding Constitution to Agent Identity...")
constitution = AgentConstitution.create_default(agent_did=agent_key.did, agent_name="DemoAgent-1")
print(f" Constitution Hash: {constitution.compute_hash()}")
identity = ConstitutionIdentity(agent_did=agent_key.did, agent_key_pair=agent_key,
resolver=DIDResolver(registry_endpoint=''), vc_manager=vc_manager)
binding_vc = await identity.bind_constitution(constitution)
print(f" Binding VC ID: {binding_vc['id'][:30]}...")
# Step 9: Cross-chain resolution (simulated)
print("\n[9] Cross-Chain DID Resolution...")
ibc_resolver = IBCDIDResolver(chain_id="msg-chain-1",
ibc_connections={"cosmos": "channel-1", "ethr": "channel-2"},
registry_endpoints={"msg": "msg1registry..."})
ethr_doc = await ibc_resolver.resolve_did("did:ethr:0x123456789abcdef")
print(f" Cross-chain resolution: {ethr_doc is not None}")
# Step 10: Key export (backup)
print("\n[10] Exporting Encrypted Key Pair (Backup)...")
manager.export_did_document('/tmp/demo_did_document.json')
print(" DID document exported to /tmp/demo_did_document.json")
print("\n" + "=" * 60)
print("Demo completed successfully!")
print("=" * 60)
if __name__ == '__main__':
asyncio.run(demo_agent_identity_lifecycle())
9. 安全最佳实践
9.1 密钥管理
- 私钥永不离开安全环境: 私钥应存储在 HSM 或安全的密钥管理服务中，不应明文存储在文件系统或环境变量中
- 定期轮换: 按照密钥轮换策略定期更换签名密钥，保留旧密钥以验证历史凭证
- 多签控制: DID Controller 可以设置为多签地址，以防止单点故障
- 备份与恢复: 使用加密导出功能安全备份私钥，确保在灾难场景下可恢复身份
9.2 凭证安全
- 过期时间: 所有 VC 应设置合理的过期时间，避免永久有效
- 撤销机制: 及时撤销泄露或不再需要的凭证，监控撤销注册表
- 最小化披露: 只签发和披露必要的声明，避免过度收集 Agent 属性
- 防重放: 使用 challenge 和 nonce 机制防止凭证重放攻击
9.3 智能合约安全
- 访问控制: DID Registry 合约中严格校验只有 controller 才能修改 DID
- 停用不可逆: DID 一旦停用不可恢复，确保调用者确实有此意图
- Gas 优化: DID 文档存储在 IPFS 上而非链上，以降低存储成本
- 升级机制: 考虑使用代理模式实现合约的可升级性
9.4 常见陷阱
| 陷阱 | 后果 | 预防 |
|---|---|---|
| 私钥硬编码 | 密钥泄露，身份被盗 | 使用环境变量或密钥管理服务 |
| 不设置过期时间 | 凭证永久有效 | 始终设置合理的 expirationDate |
| DID 文档过大 | Gas 消耗过高 | 文档上 IPFS，链上只存哈希 |
| 不使用挑战值 | 重放攻击 | 在 VP 中加入随机 challenge |
| 忽略撤销检查 | 接受已撤销的凭证 | 验证时查询链上撤销注册表 |
| 跨链无超时 | IBC 查询阻塞 | 设置合理的超时时间 |
| 单密钥无备份 | 密钥丢失即身份丢失 | 多重备份 + 多签 |
附录: msg-chain-1 DID 快速参考
CLI 命令
# Register a DID
msgnoded tx wasm execute msg1registrycontract... \\
'{"register_did":{"did":"did:msg:my-agent","document_hash":"abc...","document_uri":"ipfs://..."}}' \\
--from my-key --chain-id msg-chain-1 --gas auto
# Resolve a DID
msgnoded query wasm contract-state smart msg1registrycontract... \\
'{"resolve_did":{"did":"did:msg:my-agent"}}'
# Deactivate a DID
msgnoded tx wasm execute msg1registrycontract... \\
'{"deactivate_did":{"did":"did:msg:my-agent","reason":"retired"}}' \\
--from my-key --chain-id msg-chain-1 --gas auto
关键合约地址 (Mainnet)
DID Registry: msg1d4r3gz8xgk7q5qfq3w8z7h6g5f4e3d2c1b0a9
Revocation Reg: msg1r3v0c8x7g6f5e4d3c2b1a0z9y8x7w6v5u4t3s
Universal Res: msg1u5n4i3v2e1r0s9a8l7d6e5f4c3b2a1z0y9x8w
相关资源
- W3C DID Core 1.0: https://www.w3.org/TR/did-core/
- W3C VC Data Model: https://www.w3.org/TR/vc-data-model/
- DIDComm v2: https://didcomm.org/
- CosmWasm Docs: https://docs.cosmwasm.com/
- MSG Chain Docs: https://docs.msgchain.org/
- Ed25519: https://ed25519.cr.yp.to/