215 lines
7.7 KiB
Python
215 lines
7.7 KiB
Python
"""The full 402 -> sign -> re-present path, offline against a mocked seller."""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from hermes_nano_x402.client import Response, X402NanoClient
|
|
from hermes_nano_x402.guard import PaymentGuard
|
|
from hermes_nano_x402.quote import PaymentRefused
|
|
|
|
from helpers import (
|
|
ONE_CALL_RAW,
|
|
OTHER_PAY_TO,
|
|
PAY_TO,
|
|
URL,
|
|
ScriptedTransport,
|
|
FakeSigner,
|
|
challenge,
|
|
decode_signature,
|
|
header,
|
|
quote_doc,
|
|
served,
|
|
)
|
|
|
|
|
|
CLIENT_KW = {"max_quote_age_seconds", "work_provider", "quote_retries", "timeout", "max_redirects", "clock"}
|
|
|
|
|
|
def make(transport, signer=None, **kwargs):
|
|
signer = signer or FakeSigner()
|
|
client_kwargs = {k: v for k, v in kwargs.items() if k in CLIENT_KW}
|
|
guard_kwargs = dict(
|
|
allowed_pay_to=(PAY_TO,),
|
|
allowed_origins=("https://pursekeeper.dev",),
|
|
per_call_cap_xno=Decimal("0.01"),
|
|
total_cap_xno=Decimal("0.05"),
|
|
)
|
|
guard_kwargs.update({k: v for k, v in kwargs.items() if k not in CLIENT_KW})
|
|
guard = PaymentGuard(**guard_kwargs)
|
|
return X402NanoClient(signer=signer, guard=guard, transport=transport, **client_kwargs), signer, guard
|
|
|
|
|
|
def test_happy_path_one_block_paid_and_served():
|
|
transport = ScriptedTransport(challenge(), served())
|
|
client, signer, guard = make(transport)
|
|
result = client.get(URL)
|
|
|
|
assert result.served and result.paid
|
|
assert result.amount_xno == "0.001"
|
|
assert len(signer.calls) == 1
|
|
call = signer.calls[0]
|
|
assert call["link"] == PAY_TO
|
|
assert call["balance_raw"] == 10**33 - ONE_CALL_RAW
|
|
assert call["work"] == "0" # extra.workOptional = true -> the seller computes it
|
|
|
|
assert transport.requests[0]["headers"].get("payment-signature") is None
|
|
sig = decode_signature(transport.requests[1]["headers"])
|
|
assert sig["x402Version"] == 2
|
|
assert sig["payload"]["block"]["link"] == PAY_TO
|
|
assert guard.spent_xno == Decimal("0.001")
|
|
|
|
|
|
def test_second_invocation_reuses_signature_and_never_signs_again():
|
|
transport = ScriptedTransport(challenge(), served(), served())
|
|
client, signer, _ = make(transport)
|
|
first = client.get(URL)
|
|
second = client.get(URL)
|
|
|
|
assert len(signer.calls) == 1, "a repeat call must never sign a second block"
|
|
assert second.reused_signature
|
|
assert len(transport.signed_requests) == 2
|
|
same = decode_signature(transport.signed_requests[0]["headers"]) == decode_signature(
|
|
transport.signed_requests[1]["headers"]
|
|
)
|
|
assert same
|
|
assert first.block_hash == second.block_hash
|
|
|
|
|
|
def test_non_402_response_signs_nothing():
|
|
transport = ScriptedTransport(Response(200, {}, b"free resource", URL))
|
|
client, signer, guard = make(transport)
|
|
result = client.get(URL)
|
|
assert result.served and not result.paid
|
|
assert signer.calls == []
|
|
assert guard.spent_xno == 0
|
|
|
|
|
|
def test_cross_origin_redirect_refused_before_paying():
|
|
redirect = Response(302, {"Location": "https://evil.example/x"}, b"", URL)
|
|
transport = ScriptedTransport(redirect, challenge(), served())
|
|
client, signer, guard = make(transport)
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get(URL)
|
|
assert exc.value.code == "redirect_refused"
|
|
assert signer.calls == []
|
|
assert guard.spent_xno == 0
|
|
assert len(transport.requests) == 1, "nothing may be sent to the redirect target"
|
|
|
|
|
|
def test_same_origin_redirect_is_followed_then_paid():
|
|
canonical = "https://pursekeeper.dev/v1/echo"
|
|
transport = ScriptedTransport(
|
|
Response(301, {"Location": canonical}, b"", URL),
|
|
Response(402, header(quote_doc(resource_url=canonical)), b"pay", canonical),
|
|
served(b'{"echo":"hello"}'),
|
|
)
|
|
client, signer, _ = make(transport)
|
|
result = client.get(URL)
|
|
assert result.served and result.paid
|
|
assert transport.requests[-1]["url"] == canonical
|
|
assert len(signer.calls) == 1
|
|
|
|
|
|
def test_no_paid_retry_after_non_2xx():
|
|
transport = ScriptedTransport(challenge(), Response(500, {}, b"boom", URL))
|
|
client, signer, _ = make(transport)
|
|
result = client.get(URL)
|
|
assert result.paid and result.status == 500
|
|
assert len(signer.calls) == 1
|
|
assert len(transport.requests) == 2, "no automatic paid retry"
|
|
|
|
|
|
def test_402_without_payment_required_header_refused():
|
|
transport = ScriptedTransport(Response(402, {}, b"pay up", URL))
|
|
client, signer, _ = make(transport)
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get(URL)
|
|
assert exc.value.code == "malformed_quote"
|
|
assert signer.calls == []
|
|
|
|
|
|
def test_unlisted_payto_refused_before_signing():
|
|
transport = ScriptedTransport(challenge(pay_to=OTHER_PAY_TO), served())
|
|
client, signer, _ = make(transport)
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get(URL)
|
|
assert exc.value.code == "payto_not_allowed"
|
|
assert signer.calls == []
|
|
assert len(transport.requests) == 1
|
|
|
|
|
|
def test_amount_above_cap_refused_before_signing():
|
|
transport = ScriptedTransport(challenge(amount=str(ONE_CALL_RAW * 50)), served())
|
|
client, signer, _ = make(transport)
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get(URL)
|
|
assert exc.value.code == "amount_above_cap"
|
|
assert signer.calls == []
|
|
|
|
|
|
def test_cumulative_cap_across_two_calls():
|
|
url2 = "https://pursekeeper.dev/v1/hash"
|
|
transport = ScriptedTransport(challenge(), served(), challenge(resource_url=url2), served())
|
|
client, signer, guard = make(transport, total_cap_xno=Decimal("0.0015"))
|
|
client.get(URL)
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.post(url2, body="{}")
|
|
assert exc.value.code == "cumulative_cap"
|
|
assert len(signer.calls) == 1
|
|
|
|
|
|
def test_work_required_without_provider_refused():
|
|
transport = ScriptedTransport(challenge(work_optional=False), served())
|
|
client, signer, _ = make(transport)
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get(URL)
|
|
assert exc.value.code == "work_required"
|
|
assert signer.calls == []
|
|
|
|
|
|
def test_work_required_with_provider_attaches_work():
|
|
transport = ScriptedTransport(challenge(work_optional=False), served())
|
|
client, signer, _ = make(transport, work_provider=lambda root: "1234567890abcdef")
|
|
result = client.get(URL)
|
|
assert result.paid
|
|
assert signer.calls[0]["work"] == "1234567890abcdef"
|
|
|
|
|
|
def test_expired_quote_is_refetched_once_then_used():
|
|
stale = Response(402, header(quote_doc(max_timeout_seconds=1)), b"pay", URL)
|
|
fresh = Response(402, header(quote_doc(max_timeout_seconds=60)), b"pay", URL)
|
|
transport = ScriptedTransport(stale, fresh, served())
|
|
|
|
# A clock that jumps past the 1s quote's maxTimeoutSeconds between the probe and the check.
|
|
ticks = iter([0.0, 100.0, 100.0, 100.0, 100.0])
|
|
clock = lambda: next(ticks) # noqa: E731
|
|
client, signer, _ = make(transport)
|
|
client.clock = clock
|
|
client.guard.clock = clock
|
|
|
|
result = client.get(URL)
|
|
assert result.served and result.paid
|
|
assert len(transport.requests) == 3 # probe, re-probe with a fresh quote, paid presentation
|
|
assert len(signer.calls) == 1
|
|
|
|
|
|
def test_insufficient_balance_refused():
|
|
transport = ScriptedTransport(challenge(), served())
|
|
client, signer, _ = make(transport, signer=FakeSigner(balance=1))
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get(URL)
|
|
assert exc.value.code == "insufficient_balance"
|
|
assert signer.calls == []
|
|
|
|
|
|
def test_signed_request_does_not_follow_redirects():
|
|
redirect_after_pay = Response(307, {"Location": "https://evil.example/next"}, b"", URL)
|
|
transport = ScriptedTransport(challenge(), redirect_after_pay)
|
|
client, signer, _ = make(transport)
|
|
result = client.get(URL)
|
|
assert result.paid and result.status == 307
|
|
assert len(signer.calls) == 1
|
|
assert all("evil.example" not in r["url"] for r in transport.requests)
|