让自主 Agent 安全付款：用确定性信任门、会话预算与凭证隔离控制风险

from dataclasses import dataclass, field
from decimal import Decimal
from hashlib import sha256
from typing import Final

MIN_TRUST_SCORE: Final = 80
MAX_SINGLE_PAYMENT: Final = Decimal("2.00")


@dataclass(frozen=True)
class EndpointEvidence:
    identity_verified: bool
    success_rate: Decimal
    price_anomaly: bool
    dispute_count: int


def score_endpoint(evidence: EndpointEvidence) -> int:
    score = 50
    score += 25 if evidence.identity_verified else -30
    score += 20 if evidence.success_rate >= Decimal("0.99") else 0
    score -= 35 if evidence.price_anomaly else 0
    score -= min(evidence.dispute_count * 10, 30)
    return max(0, min(score, 100))


@dataclass
class SessionBudget:
    limit: Decimal
    spent: Decimal = Decimal("0")
    payment_ids: set[str] = field(default_factory=set)

    def reserve(self, amount: Decimal, payment_id: str) -> None:
        if payment_id in self.payment_ids:
            raise ValueError("duplicate payment_id")
        if amount <= 0 or amount > MAX_SINGLE_PAYMENT:
            raise ValueError("single-payment policy rejected the request")
        if self.spent + amount > self.limit:
            raise ValueError("session budget exceeded")

        # This mutation must be a database transaction in production.
        self.spent += amount
        self.payment_ids.add(payment_id)


def authorize_payment(
    endpoint: str,
    amount: Decimal,
    evidence: EndpointEvidence,
    budget: SessionBudget,
    request_id: str,
) -> dict[str, str]:
    trust_score = score_endpoint(evidence)
    if trust_score < MIN_TRUST_SCORE:
        raise PermissionError(f"endpoint rejected: trust_score={trust_score}")

    payment_id = sha256(
        f"{request_id}:{endpoint}:{amount}".encode("utf-8")
    ).hexdigest()
    budget.reserve(amount, payment_id)

    # Return a narrow payment intent. No wallet key or long-lived token is exposed.
    return {
        "endpoint": endpoint,
        "amount": str(amount),
        "currency": "USD",
        "payment_id": payment_id,
        "policy_version": "trust-gate-v1",
        "trust_score": str(trust_score),
    }


def execute_payment(intent: dict[str, str]) -> None:
    # Production assumption: call an isolated signer/payment service here.
    print("AUTHORIZED", intent)


if __name__ == "__main__":
    budget = SessionBudget(limit=Decimal("5.00"))
    evidence = EndpointEvidence(
        identity_verified=True,
        success_rate=Decimal("0.995"),
        price_anomaly=False,
        dispute_count=0,
    )
    intent = authorize_payment(
        endpoint="api.example.test/data",
        amount=Decimal("1.25"),
        evidence=evidence,
        budget=budget,
        request_id="agent-session-42-step-7",
    )
    execute_payment(intent)

python3 trust_gate.py