|
1
|
"""In-memory synthetic merchant that exposes a protocol-shaped x402 flow.""" |
|
2
|
|
|
3
|
from __future__ import annotations |
|
4
|
|
|
5
|
from datetime import UTC, datetime, timedelta |
|
6
|
from decimal import Decimal |
|
7
|
|
|
8
|
import httpx |
|
9
|
|
|
10
|
from .codec import encode_model |
|
11
|
from .errors import CommerceError |
|
12
|
from .models import ( |
|
13
|
PaymentRequired, |
|
14
|
PaymentRequirement, |
|
15
|
ResourceInfo, |
|
16
|
SupplierRiskReport, |
|
17
|
) |
|
18
|
from .payments import LocalPaymentProcessor |
|
19
|
|
|
20
|
PAYMENT_REQUIRED_HEADER = "PAYMENT-REQUIRED" |
|
21
|
PAYMENT_SIGNATURE_HEADER = "PAYMENT-SIGNATURE" |
|
22
|
PAYMENT_RESPONSE_HEADER = "PAYMENT-RESPONSE" |
|
23
|
|
|
24
|
RESOURCE_URL = "https://merchant.invalid/reports/SYNTH-SUPPLIER-RISK-001" |
|
25
|
|
|
26
|
|
|
27
|
class SyntheticMerchant: |
|
28
|
"""Serve one fictional paid report through an in-memory HTTP transport.""" |
|
29
|
|
|
30
|
def __init__( |
|
31
|
self, |
|
32
|
payment_processor: LocalPaymentProcessor, |
|
33
|
*, |
|
34
|
price: Decimal = Decimal("0.25"), |
|
35
|
now: datetime | None = None, |
|
36
|
malformed_challenge: bool = False, |
|
37
|
) -> None: |
|
38
|
self.payment_processor = payment_processor |
|
39
|
self.request_count = 0 |
|
40
|
self.fulfilled_count = 0 |
|
41
|
self.invalid_proof_count = 0 |
|
42
|
self.malformed_challenge = malformed_challenge |
|
43
|
issued_at = now or datetime.now(UTC) |
|
44
|
expires_at = issued_at + timedelta(minutes=5) |
|
45
|
amount = str(int(price * Decimal(1_000_000))) |
|
46
|
|
|
47
|
self.requirement = PaymentRequirement( |
|
48
|
amount=amount, |
|
49
|
asset="synthetic-usdc-base-sepolia", |
|
50
|
payTo="synthetic-merchant-wallet", |
|
51
|
maxTimeoutSeconds=300, |
|
52
|
extra={ |
|
53
|
"challengeId": "challenge-synth-report-001", |
|
54
|
"currency": "USDC", |
|
55
|
"decimals": 6, |
|
56
|
"merchantDomain": "merchant.invalid", |
|
57
|
"issuedAt": issued_at.isoformat(), |
|
58
|
"expiresAt": expires_at.isoformat(), |
|
59
|
"simulation": True, |
|
60
|
}, |
|
61
|
) |
|
62
|
self.payment_required = PaymentRequired( |
|
63
|
resource=ResourceInfo( |
|
64
|
url=RESOURCE_URL, |
|
65
|
description="Synthetic supplier-risk report", |
|
66
|
mimeType="application/json", |
|
67
|
), |
|
68
|
accepts=(self.requirement,), |
|
69
|
error="Payment is required for this synthetic report.", |
|
70
|
) |
|
71
|
self.report = SupplierRiskReport( |
|
72
|
report_id="SYNTH-SUPPLIER-RISK-001", |
|
73
|
supplier="Northstar Components", |
|
74
|
generated_from="synthetic data", |
|
75
|
signals=( |
|
76
|
"Delivery concentration increased in the last synthetic quarter.", |
|
77
|
"Two fictional certifications require manual verification.", |
|
78
|
), |
|
79
|
disclaimer=( |
|
80
|
"Training-only synthetic report; not purchasing, legal, " |
|
81
|
"compliance, or risk advice." |
|
82
|
), |
|
83
|
) |
|
84
|
|
|
85
|
def client(self) -> httpx.Client: |
|
86
|
return httpx.Client(transport=httpx.MockTransport(self._handle)) |
|
87
|
|
|
88
|
def _handle(self, request: httpx.Request) -> httpx.Response: |
|
89
|
self.request_count += 1 |
|
90
|
if request.method != "GET" or str(request.url) != RESOURCE_URL: |
|
91
|
return httpx.Response( |
|
92
|
404, |
|
93
|
json={"error": "synthetic_resource_not_found"}, |
|
94
|
) |
|
95
|
|
|
96
|
proof_header = request.headers.get(PAYMENT_SIGNATURE_HEADER) |
|
97
|
if proof_header is None: |
|
98
|
challenge = ( |
|
99
|
"not-base64-json" |
|
100
|
if self.malformed_challenge |
|
101
|
else encode_model(self.payment_required) |
|
102
|
) |
|
103
|
return httpx.Response( |
|
104
|
402, |
|
105
|
headers={PAYMENT_REQUIRED_HEADER: challenge}, |
|
106
|
json={"error": "payment_required"}, |
|
107
|
) |
|
108
|
|
|
109
|
try: |
|
110
|
settlement = self.payment_processor.verify( |
|
111
|
proof_header, |
|
112
|
self.requirement, |
|
113
|
) |
|
114
|
except CommerceError as exc: |
|
115
|
self.invalid_proof_count += 1 |
|
116
|
return httpx.Response( |
|
117
|
402, |
|
118
|
headers={PAYMENT_REQUIRED_HEADER: encode_model(self.payment_required)}, |
|
119
|
json={"error": exc.code}, |
|
120
|
) |
|
121
|
|
|
122
|
self.fulfilled_count += 1 |
|
123
|
return httpx.Response( |
|
124
|
200, |
|
125
|
headers={PAYMENT_RESPONSE_HEADER: encode_model(settlement)}, |
|
126
|
json=self.report.model_dump(mode="json"), |
|
127
|
) |