| 1 | """Deterministic application authorization for synthetic paid resources.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from datetime import UTC, datetime |
| 6 | from decimal import Decimal |
| 7 | |
| 8 | from .models import ( |
| 9 | ApprovalGrant, |
| 10 | AuthorizationDecision, |
| 11 | CommercePolicy, |
| 12 | PaymentReceipt, |
| 13 | PaymentRequired, |
| 14 | PurchaseRequest, |
| 15 | ) |
| 16 | |
| 17 | |
| 18 | class PolicyEngine: |
| 19 | """Authorize purchases independently of model reasoning.""" |
| 20 | |
| 21 | def __init__(self, policy: CommercePolicy) -> None: |
| 22 | self.policy = policy |
| 23 | self._spent_by_idempotency: dict[str, Decimal] = {} |
| 24 | |
| 25 | @property |
| 26 | def spent(self) -> Decimal: |
| 27 | return sum(self._spent_by_idempotency.values(), start=Decimal(0)) |
| 28 | |
| 29 | def preflight( |
| 30 | self, |
| 31 | request: PurchaseRequest, |
| 32 | *, |
| 33 | now: datetime | None = None, |
| 34 | ) -> AuthorizationDecision: |
| 35 | """Reject unsafe destinations and purposes before any HTTP request.""" |
| 36 | |
| 37 | now = now or datetime.now(UTC) |
| 38 | if request.resource_url.scheme != "https": |
| 39 | return self._deny("https_required", "Paid resources must use HTTPS.") |
| 40 | if request.resource_url.host not in self.policy.allowed_merchants: |
| 41 | return self._deny( |
| 42 | "merchant_not_allowed", |
| 43 | "The merchant is not in the application allowlist.", |
| 44 | ) |
| 45 | if request.purpose not in self.policy.allowed_purposes: |
| 46 | return self._deny( |
| 47 | "purpose_not_allowed", |
| 48 | "The requested purchase purpose is not allowed.", |
| 49 | ) |
| 50 | if now >= self.policy.session_expires_at: |
| 51 | return self._deny( |
| 52 | "session_expired", |
| 53 | "The application payment session has expired.", |
| 54 | ) |
| 55 | return AuthorizationDecision( |
| 56 | allowed=True, |
| 57 | code="preflight_allowed", |
| 58 | reason="The request destination and purpose satisfy preflight policy.", |
| 59 | requires_human_approval=False, |
| 60 | ) |
| 61 | |
| 62 | def authorize( |
| 63 | self, |
| 64 | request: PurchaseRequest, |
| 65 | required: PaymentRequired, |
| 66 | *, |
| 67 | approval: ApprovalGrant | None, |
| 68 | now: datetime | None = None, |
| 69 | ) -> AuthorizationDecision: |
| 70 | now = now or datetime.now(UTC) |
| 71 | requirement = required.accepts[0] |
| 72 | host = request.resource_url.host |
| 73 | amount = requirement.decimal_amount |
| 74 | |
| 75 | preflight = self.preflight(request, now=now) |
| 76 | if not preflight.allowed: |
| 77 | return preflight |
| 78 | if required.resource.url != request.resource_url: |
| 79 | return self._deny( |
| 80 | "resource_mismatch", |
| 81 | "The payment challenge refers to a different resource.", |
| 82 | ) |
| 83 | if requirement.merchant_domain != host: |
| 84 | return self._deny( |
| 85 | "merchant_mismatch", |
| 86 | "The payment challenge merchant does not match the URL.", |
| 87 | ) |
| 88 | return self.authorize_challenge( |
| 89 | request, |
| 90 | merchant_domain=requirement.merchant_domain, |
| 91 | network=requirement.network, |
| 92 | currency=requirement.currency, |
| 93 | amount=amount, |
| 94 | approval=approval, |
| 95 | expires_at=requirement.expires_at, |
| 96 | now=now, |
| 97 | ) |
| 98 | |
| 99 | def authorize_challenge( |
| 100 | self, |
| 101 | request: PurchaseRequest, |
| 102 | *, |
| 103 | merchant_domain: str, |
| 104 | network: str, |
| 105 | currency: str, |
| 106 | amount: Decimal, |
| 107 | approval: ApprovalGrant | None, |
| 108 | expires_at: datetime | None = None, |
| 109 | now: datetime | None = None, |
| 110 | ) -> AuthorizationDecision: |
| 111 | """Authorize normalized challenge facts from any x402 transport.""" |
| 112 | |
| 113 | now = now or datetime.now(UTC) |
| 114 | host = request.resource_url.host |
| 115 | preflight = self.preflight(request, now=now) |
| 116 | if not preflight.allowed: |
| 117 | return preflight |
| 118 | if merchant_domain != host: |
| 119 | return self._deny( |
| 120 | "merchant_mismatch", |
| 121 | "The payment challenge merchant does not match the URL.", |
| 122 | ) |
| 123 | if network != self.policy.network: |
| 124 | return self._deny( |
| 125 | "network_not_allowed", |
| 126 | "The payment network is outside the configured policy.", |
| 127 | ) |
| 128 | if currency != self.policy.currency: |
| 129 | return self._deny( |
| 130 | "currency_not_allowed", |
| 131 | "The payment currency is outside the configured policy.", |
| 132 | ) |
| 133 | if expires_at is not None and now >= expires_at: |
| 134 | return self._deny( |
| 135 | "challenge_expired", |
| 136 | "The merchant payment challenge has expired.", |
| 137 | ) |
| 138 | if amount > self.policy.per_request_limit: |
| 139 | return self._deny( |
| 140 | "request_limit_exceeded", |
| 141 | "The amount exceeds the per-request spending limit.", |
| 142 | ) |
| 143 | new_spend = ( |
| 144 | Decimal(0) |
| 145 | if request.idempotency_key in self._spent_by_idempotency |
| 146 | else amount |
| 147 | ) |
| 148 | if self.spent + new_spend > self.policy.per_run_limit: |
| 149 | return self._deny( |
| 150 | "run_limit_exceeded", |
| 151 | "The amount exceeds the remaining run budget.", |
| 152 | ) |
| 153 | |
| 154 | requires_approval = amount > self.policy.approval_threshold |
| 155 | if requires_approval: |
| 156 | approval_error = self._validate_approval( |
| 157 | request, |
| 158 | approval, |
| 159 | amount=amount, |
| 160 | now=now, |
| 161 | ) |
| 162 | if approval_error is not None: |
| 163 | return approval_error |
| 164 | |
| 165 | return AuthorizationDecision( |
| 166 | allowed=True, |
| 167 | code="authorized", |
| 168 | reason="The request satisfies application-owned commerce policy.", |
| 169 | requires_human_approval=requires_approval, |
| 170 | approved_amount=amount, |
| 171 | ) |
| 172 | |
| 173 | def record_spend(self, idempotency_key: str, amount: Decimal) -> None: |
| 174 | """Record one authorized amount without inventing receipt fields.""" |
| 175 | |
| 176 | self._spent_by_idempotency.setdefault(idempotency_key, amount) |
| 177 | |
| 178 | def record(self, receipt: PaymentReceipt) -> None: |
| 179 | self.record_spend(receipt.idempotency_key, receipt.amount) |
| 180 | |
| 181 | def _validate_approval( |
| 182 | self, |
| 183 | request: PurchaseRequest, |
| 184 | approval: ApprovalGrant | None, |
| 185 | *, |
| 186 | amount: Decimal, |
| 187 | now: datetime, |
| 188 | ) -> AuthorizationDecision | None: |
| 189 | if approval is None: |
| 190 | return self._deny( |
| 191 | "human_approval_required", |
| 192 | "A human approval grant is required above the threshold.", |
| 193 | requires_human_approval=True, |
| 194 | ) |
| 195 | if now >= approval.expires_at: |
| 196 | return self._deny( |
| 197 | "approval_expired", |
| 198 | "The human approval grant has expired.", |
| 199 | requires_human_approval=True, |
| 200 | ) |
| 201 | if ( |
| 202 | approval.request_id != request.request_id |
| 203 | or approval.resource_url != request.resource_url |
| 204 | or approval.purpose != request.purpose |
| 205 | ): |
| 206 | return self._deny( |
| 207 | "approval_scope_mismatch", |
| 208 | "The human approval grant is not bound to this request.", |
| 209 | requires_human_approval=True, |
| 210 | ) |
| 211 | if approval.currency != self.policy.currency: |
| 212 | return self._deny( |
| 213 | "approval_currency_mismatch", |
| 214 | "The human approval grant uses a different currency.", |
| 215 | requires_human_approval=True, |
| 216 | ) |
| 217 | if amount > approval.maximum_amount: |
| 218 | return self._deny( |
| 219 | "approval_amount_exceeded", |
| 220 | "The amount exceeds the human-approved maximum.", |
| 221 | requires_human_approval=True, |
| 222 | ) |
| 223 | return None |
| 224 | |
| 225 | @staticmethod |
| 226 | def _deny( |
| 227 | code: str, |
| 228 | reason: str, |
| 229 | *, |
| 230 | requires_human_approval: bool = False, |
| 231 | ) -> AuthorizationDecision: |
| 232 | return AuthorizationDecision( |
| 233 | allowed=False, |
| 234 | code=code, |
| 235 | reason=reason, |
| 236 | requires_human_approval=requires_human_approval, |
| 237 | ) |