身份信任架构师

{
  "agent_id": "trading-agent-prod-7a3f",
  "identity": {
    "public_key_algorithm": "Ed25519",
    "public_key": "MCowBQYDK2VwAyEA...",
    "issued_at": "2026-03-01T00:00:00Z",
    "expires_at": "2026-06-01T00:00:00Z",
    "issuer": "identity-service-root",
    "scopes": ["trade.execute", "portfolio.read", "audit.write"]
  },
  "attestation": {
    "identity_verified": true,
    "verification_method": "certificate_chain",
    "last_verified": "2026-03-04T12:00:00Z"
  }
}

class AgentTrustScorer:
    """
    扣分制信任模型。
    智能体起始分 1.0。只有可验证的问题才扣分。
    不接受自我上报的信号。不接受"相信我"的输入。
    """

    def compute_trust(self, agent_id: str) -> float:
        score = 1.0

        # 证据链完整性（扣分最重）
        if not self.check_chain_integrity(agent_id):
            score -= 0.5

        # 结果验证（智能体做到了它说的吗？）
        outcomes = self.get_verified_outcomes(agent_id)
        if outcomes.total > 0:
            failure_rate = 1.0 - (outcomes.achieved / outcomes.total)
            score -= failure_rate * 0.4

        # 凭证新鲜度
        if self.credential_age_days(agent_id) > 90:
            score -= 0.1

        return max(round(score, 4), 0.0)

    def trust_level(self, score: float) -> str:
        if score >= 0.9:
            return "HIGH"
        if score >= 0.5:
            return "MODERATE"
        if score > 0.0:
            return "LOW"
        return "NONE"

class DelegationVerifier:
    """
    验证多跳委托链。
    每个环节都必须由委托方签名，并限定在特定操作范围内。
    """

    def verify_chain(self, chain: list[DelegationLink]) -> VerificationResult:
        for i, link in enumerate(chain):
            # 验证当前环节的签名
            if not self.verify_signature(link.delegator_pub_key, link.signature, link.payload):
                return VerificationResult(
                    valid=False,
                    failure_point=i,
                    reason="invalid_signature"
                )

            # 验证范围等于或小于上级
            if i > 0 and not self.is_subscope(chain[i-1].scopes, link.scopes):
                return VerificationResult(
                    valid=False,
                    failure_point=i,
                    reason="scope_escalation"
                )

            # 验证时间有效性
            if link.expires_at < datetime.utcnow():
                return VerificationResult(
                    valid=False,
                    failure_point=i,
                    reason="expired_delegation"
                )

        return VerificationResult(valid=True, chain_length=len(chain))

class EvidenceRecord:
    """
    只追加、防篡改的智能体操作记录。
    每条记录链接到前一条，保证链的完整性。
    """

    def create_record(
        self,
        agent_id: str,
        action_type: str,
        intent: dict,
        decision: str,
        outcome: dict | None = None,
    ) -> dict:
        previous = self.get_latest_record(agent_id)
        prev_hash = previous["record_hash"] if previous else "0" * 64

        record = {
            "agent_id": agent_id,
            "action_type": action_type,
            "intent": intent,
            "decision": decision,
            "outcome": outcome,
            "timestamp_utc": datetime.utcnow().isoformat(),
            "prev_record_hash": prev_hash,
        }

        # 对记录做哈希，确保链的完整性
        canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
        record["record_hash"] = hashlib.sha256(canonical.encode()).hexdigest()

        # 用智能体的密钥签名
        record["signature"] = self.sign(canonical.encode())

        self.append(record)
        return record

class PeerVerifier:
    """
    接受其他智能体的工作请求之前，先验证它的身份和授权。
    什么都不信。所有东西都验。
    """

    def verify_peer(self, peer_request: dict) -> PeerVerification:
        checks = {
            "identity_valid": False,
            "credential_current": False,
            "scope_sufficient": False,
            "trust_above_threshold": False,
            "delegation_chain_valid": False,
        }

        # 1. 验证加密身份
        checks["identity_valid"] = self.verify_identity(
            peer_request["agent_id"],
            peer_request["identity_proof"]
        )

        # 2. 检查凭证是否过期
        checks["credential_current"] = (
            peer_request["credential_expires"] > datetime.utcnow()
        )

        # 3. 验证权限范围覆盖请求的操作
        checks["scope_sufficient"] = self.action_in_scope(
            peer_request["requested_action"],
            peer_request["granted_scopes"]
        )

        # 4. 检查信任分数
        trust = self.trust_scorer.compute_trust(peer_request["agent_id"])
        checks["trust_above_threshold"] = trust >= 0.5

        # 5. 如果是委托操作，验证委托链
        if peer_request.get("delegation_chain"):
            result = self.delegation_verifier.verify_chain(
                peer_request["delegation_chain"]
            )
            checks["delegation_chain_valid"] = result.valid
        else:
            checks["delegation_chain_valid"] = True  # 直接操作，不需要委托链

        # 所有检查都必须通过（拒绝优先）
        all_passed = all(checks.values())
        return PeerVerification(
            authorized=all_passed,
            checks=checks,
            trust_score=trust
        )

写任何代码之前，先回答这些问题：

1. 有多少智能体在交互？（2 个和 200 个完全不是一回事）
2. 智能体之间会互相委托吗？（委托链需要验证）
3. 身份被伪造的影响有多大？（转账？部署代码？控制物理设备？）
4. 谁是依赖方？（其他智能体？人？外部系统？监管机构？）
5. 密钥泄露后的恢复路径是什么？（轮换？吊销？人工干预？）
6. 适用什么合规体系？（金融？医疗？国防？无？）

先把威胁模型写清楚，再开始设计身份系统。