From 0f7be381c89f1e0763d179041ac3d0bf2b3eecaa Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:12:45 -0700 Subject: [PATCH 1/2] feat: share runtime across async HTTP clients --- .changelog/runtime-async-http.md | 5 + README.md | 16 + pyproject.toml | 1 + src/mpp/client/_http.py | 564 ++++++++++++++++ src/mpp/client/transport.py | 449 ++++++------- src/mpp/errors.py | 22 + src/mpp/runtime.py | 96 ++- tests/test_client.py | 1065 +++++++++++++++++++++++++++++- tests/test_errors.py | 12 + tests/test_runtime.py | 38 ++ 10 files changed, 1999 insertions(+), 269 deletions(-) create mode 100644 .changelog/runtime-async-http.md create mode 100644 src/mpp/client/_http.py diff --git a/.changelog/runtime-async-http.md b/.changelog/runtime-async-http.md new file mode 100644 index 0000000..de71453 --- /dev/null +++ b/.changelog/runtime-async-http.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Let asynchronous HTTP clients share a `PaymentRuntime` with origin and uncertain-outcome safeguards. diff --git a/README.md b/README.md index 3b5a7ac..35e4362 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,22 @@ async with PaymentRuntime([method]) as runtime: The runtime borrows its methods and runs them on the caller's event loop. +The same runtime can power asynchronous HTTP clients while limiting which +origins may receive payment credentials: + +```python +async with PaymentRuntime( + [method], + allowed_origins=["https://api.example.com"], +) as runtime: + async with Client(runtime=runtime) as client: + response = await client.get("https://api.example.com/paid") +``` + +If a credential is sent but its outcome cannot be confirmed, matching attempts +raise `mpp.errors.PaymentOutcomeUnknownError`. Reconcile them externally before +calling `runtime.reset_unknown_outcomes(reconciled=True)`. + ## Examples | Example | Description | diff --git a/pyproject.toml b/pyproject.toml index 7596fbc..edb03e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,7 @@ description = "Python SDK for the Machine Payments Protocol (MPP)" readme = "README.md" requires-python = ">=3.11" dependencies = [ + "anyio>=4,<5", "httpx>=0.27", ] authors = [ diff --git a/src/mpp/client/_http.py b/src/mpp/client/_http.py new file mode 100644 index 0000000..65ba53e --- /dev/null +++ b/src/mpp/client/_http.py @@ -0,0 +1,564 @@ +"""Private HTTP payment primitives shared by HTTPX transports.""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass +from datetime import UTC, datetime +from http.cookies import CookieError, SimpleCookie +from typing import TYPE_CHECKING, Any +from urllib.parse import quote + +import httpx + +from mpp import Challenge, Credential +from mpp._parsing import ParseError +from mpp.errors import PaymentOutcomeUnknownError +from mpp.events import ClientPaymentFailedPayload + +if TYPE_CHECKING: + from collections.abc import Sequence + + from mpp.runtime import Method, PaymentRuntime + +_COOKIE_ESCAPE = re.compile(r"%[0-9a-fA-F]{2}") +_PAYMENT_MARKER = "mpp.payment_attempt" +_PAYMENT_SENT = "mpp.payment_sent" +# Unknown outcomes cannot be safely evicted; block payments before retention is unbounded. +_MAX_UNRECONCILED_OUTCOMES = 1024 + + +@dataclass(slots=True) +class _Reconciliation: + """Shared reset token for markers retained on detached requests.""" + + reconciled: bool = False + + +@dataclass(frozen=True, slots=True) +class _UnknownOutcome: + """Retained sent payment that must be reconciled before retry.""" + + challenge: Challenge + credential: Credential | None + cause: BaseException + request: httpx.Request + reconciliation: _Reconciliation + + +@dataclass(eq=False, slots=True) +class _HttpPaymentAttempt: + ledger: _HttpPaymentLedger + keys: tuple[str, str] + challenge: Challenge + request: httpx.Request + credential: Credential | None = None + retry_request: httpx.Request | None = None + sent: bool = False + completed: bool = False + unknown_outcome: _UnknownOutcome | None = None + + def mark_sent(self, request: httpx.Request) -> None: + self.ledger.mark_sent(self, request) + + def unknown(self, cause: BaseException) -> _UnknownOutcome: + return self.ledger.mark_unknown(self, cause) + + def complete(self) -> None: + self.ledger.complete(self) + + def discard(self) -> None: + self.ledger.discard(self) + + +class _HttpPaymentLedger: + """Fail closed around concurrent or uncertain HTTP payment attempts.""" + + def __init__(self) -> None: + self._entries: dict[str, _HttpPaymentAttempt | _UnknownOutcome] = {} + self._unreconciled_count = 0 + self._circuit: _UnknownOutcome | None = None + self._reconciliation = _Reconciliation() + + def begin(self, challenge: Challenge, request: httpx.Request) -> _HttpPaymentAttempt: + marker = request.extensions.get(_PAYMENT_MARKER) + if isinstance(marker, _HttpPaymentAttempt): + raise _outcome_error(marker) + if isinstance(marker, _UnknownOutcome): + if not marker.reconciliation.reconciled: + raise _outcome_error(marker) + request.extensions.pop(_PAYMENT_MARKER) + if self._circuit is not None: + raise _outcome_error(self._circuit) + + challenge_key, operation_key, idempotent = _attempt_keys(challenge, request) + existing = self._entries.get(challenge_key) + operation = self._entries.get(operation_key) + if existing is None and (idempotent or isinstance(operation, _UnknownOutcome)): + existing = operation + if existing is not None: + raise _outcome_error(existing) + + attempt = _HttpPaymentAttempt( + self, + (challenge_key, operation_key), + challenge, + request, + ) + self._entries[challenge_key] = attempt + if idempotent: + self._entries[operation_key] = attempt + return attempt + + def mark_sent(self, attempt: _HttpPaymentAttempt, request: httpx.Request) -> None: + if self._circuit is not None: + self.discard(attempt) + raise _outcome_error(self._circuit) + if isinstance(operation := self._entries.get(attempt.keys[1]), _UnknownOutcome): + self.discard(attempt) + raise _outcome_error(operation) + attempt.sent = True + attempt.retry_request = request + for current in (attempt.request, request): + current.extensions[_PAYMENT_MARKER] = attempt + current.extensions[_PAYMENT_SENT] = id(current) + + def mark_unknown( + self, + attempt: _HttpPaymentAttempt, + cause: BaseException, + ) -> _UnknownOutcome: + if attempt.unknown_outcome is not None: + return attempt.unknown_outcome + + outcome = _UnknownOutcome( + challenge=attempt.challenge, + credential=attempt.credential, + cause=_compact_cause(cause), + request=httpx.Request(attempt.request.method, attempt.request.url), + reconciliation=self._reconciliation, + ) + attempt.unknown_outcome = outcome + self._remove(attempt) + for request in (attempt.request, attempt.retry_request): + if request is not None: + request.extensions[_PAYMENT_MARKER] = outcome + + if self._circuit is not None: + return outcome + for key in attempt.keys: + self._entries[key] = outcome + self._unreconciled_count += 1 + if self._unreconciled_count >= _MAX_UNRECONCILED_OUTCOMES: + self._circuit = outcome + return outcome + + def complete(self, attempt: _HttpPaymentAttempt) -> None: + if attempt.completed or attempt.unknown_outcome is not None: + return + attempt.completed = True + self._remove(attempt) + for request in (attempt.request, attempt.retry_request): + if request is not None and request.extensions.get(_PAYMENT_MARKER) is attempt: + request.extensions.pop(_PAYMENT_MARKER, None) + + def discard(self, attempt: _HttpPaymentAttempt) -> None: + if not attempt.sent: + self.complete(attempt) + + def reset(self, *, reconciled: bool) -> None: + if not reconciled: + raise ValueError("Unknown payment outcomes must be externally reconciled before reset") + self._reconciliation.reconciled = True + self._reconciliation = _Reconciliation() + self._entries = { + key: entry + for key, entry in self._entries.items() + if isinstance(entry, _HttpPaymentAttempt) + } + self._unreconciled_count = 0 + self._circuit = None + + def _remove(self, attempt: _HttpPaymentAttempt) -> None: + for key in attempt.keys: + if self._entries.get(key) is attempt: + self._entries.pop(key) + + +@dataclass(slots=True) +class _HttpPayment: + challenges: list[Challenge] + challenge: Challenge + method: Method + request: httpx.Request + response: httpx.Response + credential: Credential | None = None + + def event_payload(self, response: httpx.Response | None = None) -> dict[str, Any]: + payload: dict[str, Any] = { + "challenge": self.challenge, + "challenges": self.challenges, + "method": self.method, + "request": self.request, + "response": response if response is not None else self.response, + } + if self.credential is not None: + payload["credential"] = self.credential + return payload + + def failed( + self, + error: Exception, + *, + response: httpx.Response | None = None, + ) -> ClientPaymentFailedPayload: + credential = self.credential + if isinstance(error, PaymentOutcomeUnknownError): + credential = error.credential + return _failed_payload( + challenge=self.challenge, + challenges=self.challenges, + credential=credential, + error=error, + method=self.method, + request=self.request, + response=response if response is not None else self.response, + ) + + def unknown(self, cause: BaseException) -> PaymentOutcomeUnknownError: + return PaymentOutcomeUnknownError( + self.challenge, + cause, + credential=self.credential, + request=self.request, + ) + + def retry_request(self, authorization: str) -> httpx.Request: + headers = httpx.Headers(self.request.headers) + headers["authorization"] = authorization + retry = _copy_request(self.request, headers=headers) + _apply_response_cookies(self.response, self.request, retry) + return retry + + +class _AllowedOrigins: + def __init__(self, allowed: Sequence[str] | None) -> None: + self._allow_all = allowed is None + self._origins = { + origin for value in allowed or () if (origin := _origin(str(value))) is not None + } + + def allows(self, url: httpx.URL) -> bool: + return self._allow_all or _httpx_origin(url) in self._origins + + +def _payment_challenges(response: httpx.Response) -> tuple[list[Challenge], ParseError | None]: + challenges: list[Challenge] = [] + parse_error: ParseError | None = None + for header in response.headers.get_list("www-authenticate"): + for value in _authentication_challenges(header): + if not value.lower().startswith("payment "): + continue + try: + challenges.append(Challenge.from_www_authenticate(value)) + except ParseError as error: + parse_error = error + return challenges, parse_error + + +def _authentication_challenges(header: str) -> list[str]: + """Split a WWW-Authenticate field without splitting auth-param lists.""" + challenges: list[str] = [] + start = 0 + quoted = escaped = False + token_chars = frozenset("!#$%&'*+-.^_`|~") + for index, character in enumerate(header): + if quoted: + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == '"': + quoted = False + continue + if character == '"': + quoted = True + continue + if character != ",": + continue + + next_index = index + 1 + while next_index < len(header) and header[next_index] in " \t": + next_index += 1 + token_end = next_index + while token_end < len(header) and ( + header[token_end].isalnum() or header[token_end] in token_chars + ): + token_end += 1 + after_token = token_end + while after_token < len(header) and header[after_token] in " \t": + after_token += 1 + if ( + token_end == next_index + or (after_token < len(header) and header[after_token] == "=") + or (token_end < len(header) and header[token_end] not in " \t") + ): + continue + + if challenge := header[start:index].strip(): + challenges.append(challenge) + start = next_index + if challenge := header[start:].strip(): + challenges.append(challenge) + return challenges + + +def _challenge_is_expired(challenge: Challenge) -> bool: + if challenge.expires is None: + return False + if not challenge.expires: + return True + try: + expires = datetime.fromisoformat(challenge.expires.replace("Z", "+00:00")) + return expires.tzinfo is None or expires.utcoffset() is None or expires < datetime.now(UTC) + except (OverflowError, TypeError, ValueError): + return True + + +def _match_http_challenge( + runtime: PaymentRuntime, + challenges: list[Challenge], +) -> tuple[Challenge | None, Method | None]: + try: + return runtime.match_challenge( + sorted(challenges, key=_challenge_is_expired), + prefer_method_order=False, + ) + except ValueError: + return None, None + + +def _failed_payload( + *, + challenge: Challenge | None, + challenges: list[Challenge], + credential: Credential | None | object, + error: Exception, + method: Method | None, + request: httpx.Request, + response: httpx.Response, +) -> ClientPaymentFailedPayload: + return { + "challenge": challenge, + "challenges": challenges, + "credential": credential if isinstance(credential, Credential) else None, + "error": error, + "method": method, + "request": request, + "response": response, + } + + +def _copy_request( + request: httpx.Request, + *, + headers: httpx.Headers | None = None, +) -> httpx.Request: + return httpx.Request( + request.method, + request.url, + headers=request.headers if headers is None else headers, + content=request.content, + extensions=dict(request.extensions), + ) + + +def _response_request( + response: httpx.Response, + fallback: httpx.Request, +) -> httpx.Request: + try: + return response.request + except RuntimeError: + response.request = fallback + return fallback + + +def _apply_response_cookies( + response: httpx.Response, + source_request: httpx.Request, + target_request: httpx.Request, +) -> None: + """Apply hidden 402 cookies before HTTPX's outer client can observe them.""" + headers = response.headers.get_list("set-cookie") + if not headers: + return + + _response_request(response, source_request) + cookies = httpx.Cookies() + cookies.extract_cookies(response) + cookie_request = httpx.Request(target_request.method, target_request.url) + cookies.set_cookie_header(cookie_request) + replacements = [ + part.strip() for part in cookie_request.headers.get("cookie", "").split(";") if part.strip() + ] + names = {part.split("=", 1)[0].strip() for part in replacements} + + for header in headers: + parsed = SimpleCookie() + try: + parsed.load(header) + except CookieError: + continue + for name, cookie in parsed.items(): + if ( + name not in names + and _cookie_applies(cookie, target_request.url) + and _cookie_replaced(response, name, cookie, target_request.url) + ): + names.add(name) + + existing = [ + part.strip() for part in target_request.headers.get("cookie", "").split(";") if part.strip() + ] + retained = [part for part in existing if part.split("=", 1)[0].strip() not in names] + if merged := "; ".join((*replacements, *retained)): + target_request.headers["cookie"] = merged + else: + target_request.headers.pop("cookie", None) + + +def _propagate_response_cookies(source: httpx.Response, target: httpx.Response) -> None: + """Expose hidden 402 cookies to the outer HTTPX client's cookie jar.""" + if source is target: + return + values = source.headers.get_list("set-cookie") + if not values: + return + target_values = target.headers.get_list("set-cookie") + target.headers.pop("set-cookie", None) + target.headers.update([("set-cookie", value) for value in (*values, *target_values)]) + + +def _cookie_applies(cookie: Any, url: httpx.URL) -> bool: + host = url.raw_host.decode("ascii").casefold() + domain = cookie["domain"].lstrip(".").casefold() + if domain and host != domain and not host.endswith(f".{domain}"): + return False + if cookie["secure"] and url.scheme.casefold() != "https": + return False + request_path = _cookie_request_path(url) + cookie_path = _cookie_path(cookie, url) + return request_path == cookie_path or ( + request_path.startswith(cookie_path) + and (cookie_path.endswith("/") or request_path[len(cookie_path) :].startswith("/")) + ) + + +def _cookie_replaced( + response: httpx.Response, + name: str, + cookie: Any, + url: httpx.URL, +) -> bool: + explicit_domain = cookie["domain"].lstrip(".").casefold() + domain = explicit_domain or url.raw_host.decode("ascii").casefold() + path = _cookie_path(cookie, url) + sentinel = "mpp-existing-cookie" + probe = httpx.Cookies() + probe.set(name, sentinel, domain=f".{domain}" if explicit_domain else domain, path=path) + probe.extract_cookies(response) + return not any( + item.name == name + and item.domain.lstrip(".").casefold() == domain + and item.path == path + and item.value == sentinel + for item in probe.jar + ) + + +def _cookie_path(cookie: Any, url: httpx.URL) -> str: + if (path := cookie["path"]).startswith("/"): + return path + request_path = _cookie_request_path(url) + last_slash = request_path.rfind("/") + return "/" if last_slash <= 0 else request_path[:last_slash] + + +def _cookie_request_path(url: httpx.URL) -> str: + path = url.raw_path.partition(b"?")[0].decode("ascii") + path = quote(path, safe="%/;:@&=+$,!~*'()") + return _COOKIE_ESCAPE.sub(lambda match: match[0].upper(), path) or "/" + + +async def _close_response(response: httpx.Response) -> None: + try: + await response.aclose() + except BaseException: + pass + + +def _attempt_keys(challenge: Challenge, request: httpx.Request) -> tuple[str, str, bool]: + origin = repr(_httpx_origin(request.url)) + challenge_key = _digest("challenge", origin, challenge.id) + if idempotency_key := request.headers.get("idempotency-key"): + idempotent = True + operation_key = _digest( + "idempotency", + request.method, + str(request.url).split("#", 1)[0], + idempotency_key, + ) + else: + idempotent = False + operation_key = _digest( + "request", + request.method, + str(request.url).split("#", 1)[0], + hashlib.sha256(request.content).hexdigest(), + ) + return challenge_key, operation_key, idempotent + + +def _digest(*parts: str) -> str: + return hashlib.sha256("\0".join(parts).encode()).hexdigest() + + +def _outcome_error(entry: _HttpPaymentAttempt | _UnknownOutcome) -> PaymentOutcomeUnknownError: + cause = ( + entry.cause + if isinstance(entry, _UnknownOutcome) + else RuntimeError("A matching payment attempt is already in progress") + ) + return PaymentOutcomeUnknownError( + entry.challenge, + cause, + credential=entry.credential, + request=entry.request, + ) + + +def _compact_cause(cause: BaseException) -> BaseException: + try: + compact = type(cause)(str(cause)) + except BaseException: + compact = RuntimeError(f"{type(cause).__name__}: {cause}") + compact.__traceback__ = compact.__cause__ = compact.__context__ = None + return compact + + +def _origin(value: str) -> tuple[str, str, int | None] | None: + try: + url = httpx.URL(value) + except (httpx.InvalidURL, TypeError, UnicodeError): + return None + return _httpx_origin(url) if url.scheme and url.raw_host else None + + +def _httpx_origin(url: httpx.URL) -> tuple[str, str, int | None]: + scheme = url.scheme.casefold() + port = url.port + if port == {"http": 80, "https": 443}.get(scheme): + port = None + return scheme, url.raw_host.decode("ascii").casefold(), port diff --git a/src/mpp/client/transport.py b/src/mpp/client/transport.py index c95b987..697a0cb 100644 --- a/src/mpp/client/transport.py +++ b/src/mpp/client/transport.py @@ -10,25 +10,32 @@ from __future__ import annotations import logging -from datetime import UTC, datetime from typing import TYPE_CHECKING, Any import httpx -from mpp import Challenge, Credential -from mpp._parsing import ParseError -from mpp.errors import PaymentError +from mpp.client._http import ( + _PAYMENT_SENT, + _challenge_is_expired, + _close_response, + _failed_payload, + _HttpPayment, + _match_http_challenge, + _payment_challenges, + _propagate_response_cookies, + _response_request, +) +from mpp.errors import PaymentError, PaymentOutcomeUnknownError from mpp.events import ( CHALLENGE_RECEIVED, CREDENTIAL_CREATED, PAYMENT_FAILED, PAYMENT_RESPONSE, - ClientPaymentFailedPayload, EventDispatcher, EventHandler, Unsubscribe, ) -from mpp.runtime import Method +from mpp.runtime import Method, PaymentRuntime logger = logging.getLogger(__name__) @@ -36,28 +43,27 @@ from collections.abc import Sequence -def _client_payment_failed_payload( - *, - challenge: Challenge | None, - challenges: list[Challenge], - credential: Credential | None, - error: Exception, - method: Method | None, - request: httpx.Request, - response: httpx.Response, -) -> ClientPaymentFailedPayload: - return { - "challenge": challenge, - "challenges": challenges, - "credential": credential, - "error": error, - "method": method, - "request": request, - "response": response, - } - - -class PaymentTransport(httpx.AsyncBaseTransport): +class _EventHandlers: + _events: EventDispatcher + + def on(self, name: str, handler: EventHandler) -> Unsubscribe: + """Register a client payment event handler.""" + return self._events.on(name, handler) + + def on_challenge_received(self, handler: EventHandler) -> Unsubscribe: + return self.on(CHALLENGE_RECEIVED, handler) + + def on_credential_created(self, handler: EventHandler) -> Unsubscribe: + return self.on(CREDENTIAL_CREATED, handler) + + def on_payment_response(self, handler: EventHandler) -> Unsubscribe: + return self.on(PAYMENT_RESPONSE, handler) + + def on_payment_failed(self, handler: EventHandler) -> Unsubscribe: + return self.on(PAYMENT_FAILED, handler) + + +class PaymentTransport(_EventHandlers, httpx.AsyncBaseTransport): """httpx transport that handles 402 Payment Required responses. Wraps an inner transport and automatically: @@ -78,219 +84,197 @@ class PaymentTransport(httpx.AsyncBaseTransport): def __init__( self, - methods: Sequence[Method], + methods: Sequence[Method] | None = None, inner: httpx.AsyncBaseTransport | None = None, events: EventDispatcher | None = None, + *, + runtime: PaymentRuntime | None = None, ) -> None: - self._methods = {m.name: m for m in methods} + self._owns_runtime = runtime is None + if runtime is not None: + if methods is not None or events is not None: + raise ValueError("Pass either methods/events or runtime, not both") + self._runtime = runtime + else: + if methods is None: + raise ValueError("Pass methods or runtime") + self._runtime = PaymentRuntime(methods, events=events) self._inner = inner or httpx.AsyncHTTPTransport() - self._events = events or EventDispatcher() + self._events = self._runtime.events - def on(self, name: str, handler: EventHandler) -> Unsubscribe: - """Register a client payment event handler.""" - return self._events.on(name, handler) + async def _fail(self, payment: _HttpPayment, error: Exception, **details: Any) -> None: + await self._runtime.emit_event(PAYMENT_FAILED, payment.failed(error, **details)) - def on_challenge_received(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for selected payment challenges.""" - return self.on(CHALLENGE_RECEIVED, handler) - - def on_credential_created(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for created credentials.""" - return self.on(CREDENTIAL_CREATED, handler) - - def on_payment_response(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for successful paid retry responses.""" - return self.on(PAYMENT_RESPONSE, handler) - - def on_payment_failed(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for failed automatic payment handling.""" - return self.on(PAYMENT_FAILED, handler) + async def _unknown( + self, + payment: _HttpPayment, + cause: BaseException, + response: httpx.Response | None = None, + ) -> PaymentOutcomeUnknownError: + error = payment.unknown(cause) + await self._fail(payment, error, response=response) + return error async def handle_async_request(self, request: httpx.Request) -> httpx.Response: """Handle request, automatically retrying on 402 with credentials.""" - # Async-generator bodies (content=async_gen()) produce an AsyncByteStream - # that is not also a SyncByteStream. They cannot be safely buffered for - # replay: the generator may be infinite, already partially consumed, or - # tied to a one-shot I/O source. Reject early so callers get a clear - # message instead of a silent empty body on the paid retry. - if isinstance(request.stream, httpx.AsyncByteStream) and not isinstance( - request.stream, httpx.SyncByteStream + if not ( + isinstance(request.stream, httpx.AsyncByteStream) + and not isinstance(request.stream, httpx.SyncByteStream) ): - raise PaymentError( - "Streaming request bodies (async generators) are not supported " - "through the payment retry flow. Use a buffered body " - "(bytes, str, files=, or data=) instead." - ) - - # Buffer the request body before the first dispatch so it can be replayed - # on a paid 402 retry. Handles bytes bodies and multipart/files= bodies. - # After aread() the stream is replaced with a replayable ByteStream. - await request.aread() - + await request.aread() response = await self._inner.handle_async_request(request) - if response.status_code != 402: return response - - await response.aread() - - # Handle multiple WWW-Authenticate headers (per RFC 9110) - www_auth_headers = response.headers.get_list("www-authenticate") - - challenges: list[Challenge] = [] - parse_error: ParseError | None = None - for header in www_auth_headers: - if not header.lower().startswith("payment "): - continue - try: - parsed = Challenge.from_www_authenticate(header) - except ParseError as error: - parse_error = error - continue - challenges.append(parsed) - - challenge = None - matched_method = None - for parsed in challenges: - if parsed.method in self._methods: - challenge = parsed - matched_method = self._methods[parsed.method] - break - - if not challenge or not matched_method: - if parse_error is not None or challenges: - # Surface parse/method-selection failures to observers while - # preserving the original 402 response for the caller. - await self._events.emit( - PAYMENT_FAILED, - _client_payment_failed_payload( - challenge=None, - challenges=challenges, - credential=None, - error=parse_error - or ValueError("No compatible payment method for challenges"), - method=None, - request=request, - response=response, - ), - ) + request = _response_request(response, request) + if not self._runtime.allows_http_payment(request.url): + return response + payment_source = request.extensions.get(_PAYMENT_SENT) + if isinstance(payment_source, int) and payment_source != id(request): return response - # Check expiry before paying (client-side guardrail) - if challenge.expires: + challenges, parse_error = _payment_challenges(response) + challenge = method = None + if challenges: try: - expires_dt = datetime.fromisoformat(challenge.expires.replace("Z", "+00:00")) - if expires_dt < datetime.now(UTC): - logger.warning("Challenge expired at %s, not paying", challenge.expires) - await self._events.emit( + self._runtime.start() + challenge, method = _match_http_challenge(self._runtime, challenges) + except BaseException: + await _close_response(response) + raise + if challenge is None or method is None: + if parse_error is not None or challenges: + try: + await self._runtime.emit_event( PAYMENT_FAILED, - _client_payment_failed_payload( - challenge=challenge, + _failed_payload( + challenge=None, challenges=challenges, credential=None, - error=ValueError(f"Challenge expired at {challenge.expires}"), - method=matched_method, + error=parse_error + or ValueError("No compatible payment method for challenges"), + method=None, request=request, response=response, ), ) - return response - except ValueError: - pass # If we can't parse, let server validate + except BaseException: + await _close_response(response) + raise + return response + + payment = _HttpPayment(challenges, challenge, method, request, response) + if _challenge_is_expired(challenge): + logger.warning("Challenge expired at %s, not paying", challenge.expires) + try: + await self._fail(payment, ValueError(f"Challenge expired at {challenge.expires}")) + except BaseException: + await _close_response(response) + raise + return response try: - # challenge.received is the one client event that can override the - # default credential creation path by returning a Credential. - event_credential = await self._events.emit( - CHALLENGE_RECEIVED, - { - "challenge": challenge, - "challenges": challenges, - "method": matched_method, - "request": request, - "response": response, - }, - first_result=True, - ) - credential = ( - event_credential - if isinstance(event_credential, Credential) - else await matched_method.create_credential(challenge) - ) - await self._events.emit( - CREDENTIAL_CREATED, - { - "challenge": challenge, - "credential": credential, - "method": matched_method, - "request": request, - "response": response, - }, - ) - auth_header = credential.to_authorization() - except Exception as error: - await self._events.emit( - PAYMENT_FAILED, - _client_payment_failed_payload( - challenge=challenge, - challenges=challenges, - credential=None, - error=error, - method=matched_method, - request=request, - response=response, - ), + await request.aread() + except httpx.StreamConsumed as cause: + error = PaymentError( + "Streaming request bodies cannot be replayed after a payment challenge. " + "Use a buffered body for paid requests." ) + try: + await self._fail(payment, error) + finally: + await _close_response(response) + raise error from cause + except BaseException: + await _close_response(response) raise - headers = httpx.Headers(request.headers) - headers["Authorization"] = auth_header - - retry_request = httpx.Request( - method=request.method, - url=request.url, - headers=headers, - content=request.content, - extensions=request.extensions, - ) - try: - payment_response = await self._inner.handle_async_request(retry_request) - except Exception as error: - await self._events.emit( - PAYMENT_FAILED, - _client_payment_failed_payload( - challenge=challenge, - challenges=challenges, - credential=credential, - error=error, - method=matched_method, - request=request, - response=response, - ), - ) + await response.aread() + await response.aclose() + except BaseException: + await _close_response(response) raise - if payment_response.is_success: - await self._events.emit( - PAYMENT_RESPONSE, - { - "challenge": challenge, - "credential": credential, - "method": matched_method, - "request": request, - "response": payment_response, - }, - ) + with self._runtime._paid_operation(): + try: + attempt = self._runtime._begin_http_payment(challenge, request) + except PaymentOutcomeUnknownError as error: + await self._fail(payment, error) + raise - return payment_response + try: + credential = await self._runtime.create_credential( + challenge, + method, + event_payload=payment.event_payload(), + ) + authorization = credential.to_authorization() + payment.credential = credential + attempt.credential = credential + retry_request = payment.retry_request(authorization) + except BaseException as error: + attempt.discard() + if isinstance(error, Exception): + await self._fail(payment, error) + raise + + try: + attempt.mark_sent(retry_request) + payment_response = await self._inner.handle_async_request(retry_request) + except BaseException as cause: + if not attempt.sent: + attempt.discard() + if isinstance(cause, Exception): + await self._fail(payment, cause) + raise + outcome = attempt.unknown(cause) + if not isinstance(cause, Exception): + raise + error = await self._unknown(payment, outcome.cause) + raise error from cause + + try: + if payment_response.status_code == 402: + cause = RuntimeError( + "Server returned another payment challenge after receiving a credential" + ) + attempt.unknown(cause) + error = await self._unknown(payment, cause, response=payment_response) + raise error from cause + + if payment_response.status_code >= 400: + cause = RuntimeError( + f"Credentialed request returned HTTP {payment_response.status_code}" + ) + attempt.unknown(cause) + await self._unknown(payment, cause, response=payment_response) + else: + attempt.complete() + + _response_request(payment_response, retry_request) + _propagate_response_cookies(response, payment_response) + if payment_response.is_success: + await self._runtime.emit_event( + PAYMENT_RESPONSE, + payment.event_payload(payment_response), + ) + return payment_response + except BaseException as error: + if attempt.sent and not attempt.completed and attempt.unknown_outcome is None: + attempt.unknown(error) + await _close_response(payment_response) + raise async def aclose(self) -> None: - """Close the inner transport.""" - await self._inner.aclose() + """Close the inner transport and an implicitly created runtime.""" + try: + await self._inner.aclose() + finally: + if self._owns_runtime: + await self._runtime.aclose() -class Client: +class Client(_EventHandlers): """HTTP client with automatic payment handling. Example: @@ -298,29 +282,15 @@ class Client: response = await client.get("https://api.example.com/resource") """ - def __init__(self, methods: Sequence[Method]) -> None: - self._transport = PaymentTransport(methods) + def __init__( + self, + methods: Sequence[Method] | None = None, + *, + runtime: PaymentRuntime | None = None, + ) -> None: + self._transport = PaymentTransport(methods=methods, runtime=runtime) self._client = httpx.AsyncClient(transport=self._transport) - - def on(self, name: str, handler: EventHandler) -> Unsubscribe: - """Register a client payment event handler.""" - return self._transport.on(name, handler) - - def on_challenge_received(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for selected payment challenges.""" - return self.on(CHALLENGE_RECEIVED, handler) - - def on_credential_created(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for created credentials.""" - return self.on(CREDENTIAL_CREATED, handler) - - def on_payment_response(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for successful paid retry responses.""" - return self.on(PAYMENT_RESPONSE, handler) - - def on_payment_failed(self, handler: EventHandler) -> Unsubscribe: - """Register a handler for failed automatic payment handling.""" - return self.on(PAYMENT_FAILED, handler) + self._events = self._transport._events async def __aenter__(self) -> Client: await self._client.__aenter__() @@ -359,7 +329,8 @@ async def request( method: str, url: str, *, - methods: Sequence[Method], + methods: Sequence[Method] | None = None, + runtime: PaymentRuntime | None = None, **kwargs: Any, ) -> httpx.Response: """Send an HTTP request with automatic payment handling. @@ -374,15 +345,27 @@ async def request( methods=[tempo(...)], ) """ - async with Client(methods) as client: + async with Client(methods, runtime=runtime) as client: return await client.request(method, url, **kwargs) -async def get(url: str, *, methods: Sequence[Method], **kwargs: Any) -> httpx.Response: +async def get( + url: str, + *, + methods: Sequence[Method] | None = None, + runtime: PaymentRuntime | None = None, + **kwargs: Any, +) -> httpx.Response: """Send a GET request with automatic payment handling.""" - return await request("GET", url, methods=methods, **kwargs) + return await request("GET", url, methods=methods, runtime=runtime, **kwargs) -async def post(url: str, *, methods: Sequence[Method], **kwargs: Any) -> httpx.Response: +async def post( + url: str, + *, + methods: Sequence[Method] | None = None, + runtime: PaymentRuntime | None = None, + **kwargs: Any, +) -> httpx.Response: """Send a POST request with automatic payment handling.""" - return await request("POST", url, methods=methods, **kwargs) + return await request("POST", url, methods=methods, runtime=runtime, **kwargs) diff --git a/src/mpp/errors.py b/src/mpp/errors.py index 7ba85c9..6ea0542 100644 --- a/src/mpp/errors.py +++ b/src/mpp/errors.py @@ -71,6 +71,28 @@ def to_problem_details(self, challenge_id: str | None = None) -> dict[str, Any]: return details +class PaymentOutcomeUnknownError(PaymentError, RuntimeError): + """A credential was sent, but the payment result could not be confirmed.""" + + def __init__( + self, + challenge: Any, + cause: BaseException, + *, + credential: Any | None = None, + request: Any | None = None, + ) -> None: + self.challenge = challenge + self.cause = cause + self.credential = credential + self.request = request + challenge_id = getattr(challenge, "id", "unknown") + super().__init__( + "Payment outcome is unknown after sending a credential " + f"for challenge {challenge_id}. Do not blindly retry." + ) + + class PaymentRequiredError(PaymentError): """No credential was provided but payment is required.""" diff --git a/src/mpp/runtime.py b/src/mpp/runtime.py index c82d0f0..3b9c9e2 100644 --- a/src/mpp/runtime.py +++ b/src/mpp/runtime.py @@ -3,7 +3,12 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from typing import Any, Protocol, Self, runtime_checkable +from contextlib import contextmanager +from contextvars import ContextVar +from typing import TYPE_CHECKING, Any, Protocol, Self, runtime_checkable + +import httpx +from anyio import get_current_task from mpp import Challenge, Credential from mpp.events import ( @@ -13,6 +18,25 @@ EventPayload, ) +if TYPE_CHECKING: + from mpp.client import PaymentTransport + from mpp.client._http import _HttpPaymentAttempt + +_PAID_RUNTIMES: ContextVar[tuple[tuple[object, int], ...]] = ContextVar( + "mpp_paid_runtimes", + default=(), +) + + +def _scope_active(key: object) -> bool: + try: + owner = get_current_task().id + except RuntimeError: + return False + return any( + scope_key is key and scope_owner == owner for scope_key, scope_owner in _PAID_RUNTIMES.get() + ) + @runtime_checkable class Method(Protocol): @@ -36,18 +60,26 @@ def __init__( methods: Sequence[Method] = (), *, events: EventDispatcher | None = None, + allowed_origins: Sequence[str] | None = None, ) -> None: + from mpp.client._http import _AllowedOrigins, _HttpPaymentLedger + self.methods = tuple(methods) for method in self.methods: if not _is_method(method): raise TypeError("methods must contain payment Methods") _method_intents(method) self.events = events or EventDispatcher() + self._allowed_origins = _AllowedOrigins(allowed_origins) + self._http = _HttpPaymentLedger() self._closed = False + self._closing = False + self._paid_operations = 0 + self._scope_key = object() def start(self) -> Self: """Open the runtime, or return it if already open.""" - if self._closed: + if self._closed or (self._closing and not _scope_active(self._scope_key)): raise RuntimeError("PaymentRuntime is closed") return self @@ -57,11 +89,15 @@ def __enter__(self) -> Self: def __exit__(self, *_args: Any) -> None: self.close() - async def __aenter__(self) -> Self: + async def astart(self) -> Self: + """Asynchronously open the runtime.""" return self.start() + async def __aenter__(self) -> Self: + return await self.astart() + async def __aexit__(self, *_args: Any) -> None: - self.close() + await self.aclose() def match_challenge( self, @@ -134,9 +170,59 @@ async def emit_event(self, name: str, payload: EventPayload) -> Any: self.start() return await self.events.emit(name, payload) + def payment_transport( + self, + inner: httpx.AsyncBaseTransport | None = None, + ) -> PaymentTransport: + """Create an asynchronous HTTPX transport backed by this runtime.""" + from mpp.client import PaymentTransport + + return PaymentTransport(inner=inner, runtime=self) + + def allows_http_payment(self, url: httpx.URL) -> bool: + """Return whether credentials may be created for an HTTP origin.""" + return self._allowed_origins.allows(url) + + def reset_unknown_outcomes(self, *, reconciled: bool) -> None: + """Allow new payments after retained uncertain outcomes were reconciled.""" + self._http.reset(reconciled=reconciled) + + def _begin_http_payment( + self, + challenge: Challenge, + request: httpx.Request, + ) -> _HttpPaymentAttempt: + self.start() + return self._http.begin(challenge, request) + + @contextmanager + def _paid_operation(self): + """Keep a committed payment flow alive if close is requested.""" + if _scope_active(self._scope_key): + yield + return + self.start() + # Context variables propagate to child tasks; bind this lease to its owner. + scope = (self._scope_key, get_current_task().id) + token = _PAID_RUNTIMES.set((*_PAID_RUNTIMES.get(), scope)) + self._paid_operations += 1 + try: + yield + finally: + self._paid_operations -= 1 + _PAID_RUNTIMES.reset(token) + if self._closing and not self._paid_operations: + self._closed = True + def close(self) -> None: """Prevent new runtime operations.""" - self._closed = True + self._closing = True + if not self._paid_operations: + self._closed = True + + async def aclose(self) -> None: + """Asynchronously close the runtime.""" + self.close() def _is_method(value: Any) -> bool: diff --git a/tests/test_client.py b/tests/test_client.py index c13de90..a0662c3 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,5 +1,7 @@ """Tests for client-side transport.""" +import asyncio +from typing import Any, cast from unittest.mock import AsyncMock import httpx @@ -8,6 +10,8 @@ from mpp import Challenge, Credential from mpp.client import Client, PaymentTransport, get, post, request +from mpp.errors import PaymentError, PaymentOutcomeUnknownError +from mpp.runtime import PaymentRuntime from tests import make_credential @@ -43,6 +47,52 @@ async def aclose(self) -> None: pass +class ConsumingTransport(MockTransport): + def __init__(self, responses: list[httpx.Response]) -> None: + super().__init__(responses) + self.bodies: list[bytes] = [] + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + stream = cast(httpx.AsyncByteStream, request.stream) + self.bodies.append(b"".join([chunk async for chunk in stream])) + response = self.responses[self._index] + self._index += 1 + return response + + +class TrackingStream(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes], *, broken: bool = False) -> None: + self.chunks = chunks + self.broken = broken + self.started = False + self.closed = False + + async def __aiter__(self): + self.started = True + if self.broken: + raise httpx.ReadError("body lost") + for chunk in self.chunks: + yield chunk + + async def aclose(self) -> None: + self.closed = True + + +def payment_required(identifier: str = "test-id", **kwargs: Any) -> httpx.Response: + challenge = Challenge( + id=identifier, + method="tempo", + intent="charge", + request={"amount": "1000"}, + **kwargs, + ) + return httpx.Response( + 402, + headers={"www-authenticate": challenge.to_www_authenticate("example.com")}, + ) + + class TestPaymentTransport: @pytest.mark.asyncio async def test_passes_through_non_402(self) -> None: @@ -67,7 +117,7 @@ async def test_handles_402_with_matching_method(self) -> None: ) www_auth = challenge.to_www_authenticate("example.com") - inner = MockTransport( + inner = ConsumingTransport( [ httpx.Response(402, headers={"www-authenticate": www_auth}), httpx.Response(200, content=b'{"data": "ok"}'), @@ -104,7 +154,7 @@ async def test_paid_retry_replays_request_body(self) -> None: ) www_auth = challenge.to_www_authenticate("example.com") - inner = MockTransport( + inner = ConsumingTransport( [ httpx.Response(402, headers={"www-authenticate": www_auth}), httpx.Response(200, content=b'{"data": "ok"}'), @@ -132,7 +182,7 @@ async def test_paid_retry_replays_multipart_body(self) -> None: ) www_auth = challenge.to_www_authenticate("example.com") - inner = MockTransport( + inner = ConsumingTransport( [ httpx.Response(402, headers={"www-authenticate": www_auth}), httpx.Response(200, content=b'{"data": "ok"}'), @@ -149,8 +199,7 @@ async def test_paid_retry_replays_multipart_body(self) -> None: await transport.handle_async_request(request) assert len(inner.requests) == 2 - initial_body = inner.requests[0].content - retry_body = inner.requests[1].content + initial_body, retry_body = inner.bodies assert retry_body == initial_body assert b"hello from file" in retry_body assert b"report.txt" in retry_body @@ -214,29 +263,8 @@ async def test_paid_retry_replays_patch_body(self) -> None: assert retry_request.headers["content-length"] == str(len(body)) @pytest.mark.asyncio - async def test_paid_retry_raises_for_streaming_body(self) -> None: - """Should raise PaymentError upfront for async generator (streaming) bodies. - - Streaming bodies cannot be reliably buffered and replayed: the generator - may be infinite, already partially consumed, or tied to a one-shot source. - A descriptive error is better than a silent empty body on the paid retry. - """ - from mpp.errors import PaymentError - - challenge = Challenge( - id="test-id", - method="tempo", - intent="charge", - request={"amount": "1000"}, - ) - www_auth = challenge.to_www_authenticate("example.com") - - inner = MockTransport( - [ - httpx.Response(402, headers={"www-authenticate": www_auth}), - httpx.Response(200, content=b'{"data": "ok"}'), - ] - ) + async def test_paid_retry_raises_for_consumed_streaming_body(self) -> None: + inner = ConsumingTransport([payment_required()]) transport = PaymentTransport(methods=[MockMethod()], inner=inner) async def async_body_gen(): @@ -247,7 +275,55 @@ async def async_body_gen(): with pytest.raises(PaymentError, match="Streaming request bodies"): await transport.handle_async_request(request) - assert len(inner.requests) == 0 + assert inner.bodies == [b"chunk1chunk2"] + assert len(inner.requests) == 1 + + @pytest.mark.asyncio + async def test_free_streaming_body_passes_through(self) -> None: + inner = ConsumingTransport([httpx.Response(200, content=b"free")]) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + async def body(): + yield b"streamed" + + response = await transport.handle_async_request( + httpx.Request("POST", "https://example.com", content=body()) + ) + + assert response.content == b"free" + assert inner.bodies == [b"streamed"] + + @pytest.mark.asyncio + async def test_free_stream_is_not_read_by_wrapper(self) -> None: + stream = TrackingStream([b"streamed"]) + transport = PaymentTransport( + methods=[MockMethod()], + inner=MockTransport([httpx.Response(200, content=b"free")]), + ) + + response = await transport.handle_async_request( + httpx.Request("POST", "https://example.com", content=stream) + ) + + assert response.content == b"free" + assert not stream.started + + @pytest.mark.asyncio + async def test_duplicate_method_names_still_match_intent(self) -> None: + class SubscriptionMethod(MockMethod): + intents = {"subscription": None} + + first = SubscriptionMethod() + second = MockMethod() + transport = PaymentTransport( + methods=[first, second], + inner=MockTransport([payment_required(), httpx.Response(200)]), + ) + + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + first.create_credential.assert_not_awaited() + second.create_credential.assert_awaited_once() @pytest.mark.asyncio async def test_emits_client_payment_events(self) -> None: @@ -585,10 +661,937 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: ) ) - with pytest.raises(RuntimeError, match="network failed"): + with pytest.raises(PaymentOutcomeUnknownError, match="Do not blindly retry") as raised: await transport.handle_async_request(httpx.Request("GET", "https://example.com")) - assert events == ["failed:test-id:RuntimeError"] + assert isinstance(raised.value.__cause__, RuntimeError) + assert events == ["failed:test-id:PaymentOutcomeUnknownError"] + + +class TestRuntimePaymentTransport: + @pytest.mark.asyncio + async def test_explicit_runtime_shares_events_and_survives_transport_close(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method]) + events: list[str] = [] + runtime.events.on("*", lambda event: events.append(event.name)) + transport = runtime.payment_transport( + MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + ) + + response = await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + await transport.aclose() + await runtime.emit_event("custom", {}) + + assert response.content == b"paid" + assert events == [ + "challenge.received", + "credential.created", + "payment.response", + "custom", + ] + + @pytest.mark.asyncio + async def test_implicit_runtime_preserves_method_caller_loop(self) -> None: + caller_loop = asyncio.get_running_loop() + loops: list[asyncio.AbstractEventLoop] = [] + + class LoopMethod(MockMethod): + async def create(self, challenge: Challenge) -> Credential: + loops.append(asyncio.get_running_loop()) + return make_credential({}, challenge_id=challenge.id) + + method = LoopMethod() + method.create_credential.side_effect = method.create + transport = PaymentTransport( + methods=[method], + inner=MockTransport([payment_required(), httpx.Response(200, content=b"paid")]), + ) + + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert loops == [caller_loop] + + @pytest.mark.asyncio + async def test_origin_policy_is_exact_and_normalizes_default_ports(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method], allowed_origins=["https://allowed.test:443"]) + blocked = MockTransport([payment_required()]) + allowed = MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + + blocked_response = await runtime.payment_transport(blocked).handle_async_request( + httpx.Request("GET", "http://allowed.test/resource") + ) + allowed_response = await runtime.payment_transport(allowed).handle_async_request( + httpx.Request("GET", "https://allowed.test/resource") + ) + + assert blocked_response.status_code == 402 + assert len(blocked.requests) == 1 + assert allowed_response.status_code == 200 + assert method.create_credential.await_count == 1 + + @pytest.mark.asyncio + async def test_invalid_origin_entries_fail_closed(self) -> None: + runtime = PaymentRuntime([MockMethod()], allowed_origins=["/relative", "not a url"]) + inner = MockTransport([payment_required()]) + + response = await runtime.payment_transport(inner).handle_async_request( + httpx.Request("GET", "https://example.com") + ) + + assert response.status_code == 402 + assert len(inner.requests) == 1 + + def test_rejects_ambiguous_or_missing_configuration(self) -> None: + runtime = PaymentRuntime() + with pytest.raises(ValueError, match="methods or runtime"): + PaymentTransport() + with pytest.raises(ValueError, match="either methods/events or runtime"): + PaymentTransport(methods=[], runtime=runtime) + with pytest.raises(ValueError, match="either methods/events or runtime"): + PaymentTransport(events=runtime.events, runtime=runtime) + + @pytest.mark.asyncio + async def test_owned_transport_closes_only_its_runtime(self) -> None: + owned = PaymentTransport(methods=[], inner=MockTransport([])) + owned_runtime = owned._runtime + borrowed_runtime = PaymentRuntime() + borrowed = borrowed_runtime.payment_transport(MockTransport([])) + + await owned.aclose() + await borrowed.aclose() + + with pytest.raises(RuntimeError, match="closed"): + await owned_runtime.astart() + assert await borrowed_runtime.astart() is borrowed_runtime + + @pytest.mark.asyncio + async def test_close_requested_during_payment_is_deferred_until_retry_finishes(self) -> None: + runtime = PaymentRuntime([MockMethod()]) + events: list[str] = [] + runtime.events.on("credential.created", lambda _payload: runtime.close()) + runtime.events.on("payment.response", lambda _payload: events.append("response")) + transport = runtime.payment_transport( + MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + ) + + response = await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert response.status_code == 200 + assert events == ["response"] + with pytest.raises(RuntimeError, match="closed"): + runtime.start() + + @pytest.mark.asyncio + async def test_closing_runtime_rejects_unrelated_tasks(self) -> None: + retry_started = asyncio.Event() + retry_release = asyncio.Event() + calls = 0 + + async def handler(_request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + if calls == 1: + return payment_required() + retry_started.set() + await retry_release.wait() + return httpx.Response(200, content=b"paid") + + runtime = PaymentRuntime([MockMethod()]) + transport = runtime.payment_transport(httpx.MockTransport(handler)) + request_task = asyncio.create_task( + transport.handle_async_request(httpx.Request("GET", "https://example.com")) + ) + await retry_started.wait() + runtime.close() + + with pytest.raises(RuntimeError, match="closed"): + await runtime.emit_event("outside", {}) + retry_release.set() + + assert (await request_task).status_code == 200 + + @pytest.mark.asyncio + async def test_close_rejects_child_that_has_not_acquired_its_own_lease(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method]) + child_inner = MockTransport([payment_required("child")]) + child_transport = runtime.payment_transport(child_inner) + child: asyncio.Task[httpx.Response] | None = None + + def close_after_spawning_child(payload: dict[str, Any]) -> None: + nonlocal child + if payload["challenge"].id != "parent": + return + child = asyncio.create_task( + child_transport.handle_async_request( + httpx.Request("GET", "https://example.com/child") + ) + ) + runtime.close() + + runtime.events.on("credential.created", close_after_spawning_child) + parent_transport = runtime.payment_transport( + MockTransport([payment_required("parent"), httpx.Response(200, content=b"parent")]) + ) + + parent = await parent_transport.handle_async_request( + httpx.Request("GET", "https://example.com/parent") + ) + + assert child is not None + with pytest.raises(RuntimeError, match="closed"): + await child + assert parent.content == b"parent" + assert len(child_inner.requests) == 1 + assert method.create_credential.await_count == 1 + + @pytest.mark.asyncio + async def test_close_waits_for_child_with_its_own_lease(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method]) + child_retry_started = asyncio.Event() + release_child = asyncio.Event() + child_calls = 0 + + async def child_handler(request: httpx.Request) -> httpx.Response: + nonlocal child_calls + child_calls += 1 + if "authorization" not in request.headers: + return payment_required("child") + child_retry_started.set() + await release_child.wait() + return httpx.Response(200, content=b"child") + + child_transport = runtime.payment_transport(httpx.MockTransport(child_handler)) + child: asyncio.Task[httpx.Response] | None = None + + async def close_after_child_is_leased(payload: dict[str, Any]) -> None: + nonlocal child + if payload["challenge"].id != "parent": + return + child = asyncio.create_task( + child_transport.handle_async_request( + httpx.Request("GET", "https://example.com/child") + ) + ) + await child_retry_started.wait() + runtime.close() + release_child.set() + assert (await child).content == b"child" + + runtime.events.on("credential.created", close_after_child_is_leased) + parent_transport = runtime.payment_transport( + MockTransport([payment_required("parent"), httpx.Response(200, content=b"parent")]) + ) + + parent = await parent_transport.handle_async_request( + httpx.Request("GET", "https://example.com/parent") + ) + + assert child is not None + assert parent.content == b"parent" + assert child_calls == 2 + assert method.create_credential.await_count == 2 + with pytest.raises(RuntimeError, match="closed"): + runtime.start() + + +class TestHttpPaymentSafety: + @pytest.mark.asyncio + async def test_request_read_abort_closes_challenge_response(self) -> None: + class Abort(BaseException): + pass + + class AbortStream(httpx.AsyncByteStream): + async def __aiter__(self): + raise Abort + yield b"" # pragma: no cover + + async def aclose(self) -> None: + pass + + stream = TrackingStream([b"challenge"]) + challenge = httpx.Response( + 402, + headers=payment_required().headers, + stream=stream, + ) + transport = PaymentTransport( + methods=[MockMethod()], + inner=MockTransport([challenge]), + ) + + with pytest.raises(Abort): + await transport.handle_async_request( + httpx.Request( + "POST", + "https://example.com", + content=AbortStream(), + ) + ) + + assert stream.closed + + @pytest.mark.asyncio + async def test_close_racing_with_begin_keeps_payment_attempt_leased( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + runtime = PaymentRuntime([MockMethod()]) + begin = runtime._begin_http_payment + + def begin_then_close(challenge: Challenge, request: httpx.Request) -> Any: + attempt = begin(challenge, request) + runtime.close() + return attempt + + monkeypatch.setattr(runtime, "_begin_http_payment", begin_then_close) + inner = MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) + + response = await runtime.payment_transport(inner).handle_async_request( + httpx.Request("GET", "https://example.com") + ) + + assert response.status_code == 200 + assert all("mpp.payment_attempt" not in request.extensions for request in inner.requests) + with pytest.raises(RuntimeError, match="closed"): + await runtime.emit_event("after", {}) + + @pytest.mark.asyncio + async def test_retry_construction_failure_discards_unsent_attempt( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + from mpp.client._http import _HttpPayment + + def fail_retry(_payment: _HttpPayment, _authorization: str) -> httpx.Request: + raise RuntimeError("retry construction failed") + + monkeypatch.setattr(_HttpPayment, "retry_request", fail_retry) + method = MockMethod() + runtime = PaymentRuntime([method]) + failed: list[dict[str, Any]] = [] + runtime.events.on("payment.failed", failed.append) + inner = MockTransport([payment_required(), payment_required()]) + transport = runtime.payment_transport(inner) + + for _ in range(2): + with pytest.raises(RuntimeError, match="retry construction failed"): + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert len(inner.requests) == 2 + assert method.create_credential.await_count == 2 + assert len(failed) == 2 + + @pytest.mark.asyncio + async def test_response_processing_failure_keeps_known_outcome( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + def fail_cookies(_source: httpx.Response, _target: httpx.Response) -> None: + raise RuntimeError("cookie propagation failed") + + monkeypatch.setattr( + "mpp.client.transport._propagate_response_cookies", + fail_cookies, + ) + paid_stream = TrackingStream([b"paid"]) + method = MockMethod() + transport = PaymentRuntime([method]).payment_transport( + MockTransport( + [ + payment_required(), + httpx.Response(200, stream=paid_stream), + payment_required(), + httpx.Response(200, content=b"again"), + ] + ) + ) + + with pytest.raises(RuntimeError, match="cookie propagation failed"): + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + monkeypatch.undo() + response = await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert paid_stream.closed + assert response.content == b"again" + assert method.create_credential.await_count == 2 + + @pytest.mark.parametrize("race", ["operation", "circuit"]) + @pytest.mark.asyncio + async def test_send_boundary_unknown_emits_payment_failed( + self, + monkeypatch: pytest.MonkeyPatch, + race: str, + ) -> None: + if race == "circuit": + monkeypatch.setattr("mpp.client._http._MAX_UNRECONCILED_OUTCOMES", 1) + runtime = PaymentRuntime([MockMethod()]) + request = httpx.Request("GET", "https://example.com") + + def trip_guard(_payload: dict[str, Any]) -> None: + blockers = ( + [("blocker", request.url)] + if race == "operation" + else [("blocker", httpx.URL("https://example.com/other"))] + ) + for identifier, url in blockers: + blocker_request = httpx.Request("GET", url) + blocker = runtime._begin_http_payment( + Challenge( + id=identifier, + method="tempo", + intent="charge", + request={}, + ), + blocker_request, + ) + blocker.credential = make_credential( + {"retained": True}, + challenge_id=identifier, + ) + blocker.mark_sent(blocker_request) + blocker.unknown(TimeoutError("response lost")) + + failed: list[dict[str, Any]] = [] + runtime.events.on("credential.created", trip_guard) + runtime.events.on("payment.failed", failed.append) + inner = MockTransport([payment_required()]) + + with pytest.raises(PaymentOutcomeUnknownError): + await runtime.payment_transport(inner).handle_async_request(request) + + assert len(inner.requests) == 1 + assert isinstance(failed[0]["error"], PaymentOutcomeUnknownError) + assert failed[0]["challenge"].id == "test-id" + assert failed[0]["credential"] == failed[0]["error"].credential + assert failed[0]["credential"].payload == {"retained": True} + + @pytest.mark.parametrize("idempotency_key", [None, "same-operation"]) + @pytest.mark.asyncio + async def test_concurrent_requests_only_share_active_operation_with_idempotency_key( + self, + idempotency_key: str | None, + ) -> None: + first_retry = asyncio.Event() + second_retry = asyncio.Event() + release = asyncio.Event() + challenge_calls = paid_calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal challenge_calls, paid_calls + if "authorization" not in request.headers: + challenge_calls += 1 + return payment_required(f"challenge-{challenge_calls}") + paid_calls += 1 + (first_retry if paid_calls == 1 else second_retry).set() + await release.wait() + return httpx.Response(200, content=b"paid") + + transport = PaymentRuntime([MockMethod()]).payment_transport(httpx.MockTransport(handler)) + headers = {"idempotency-key": idempotency_key} if idempotency_key else {} + + def request() -> httpx.Request: + return httpx.Request( + "POST", + "https://example.com/resource", + headers=headers, + content=b"same body", + ) + + first = asyncio.create_task(transport.handle_async_request(request())) + await first_retry.wait() + second = asyncio.create_task(transport.handle_async_request(request())) + + try: + if idempotency_key: + with pytest.raises(PaymentOutcomeUnknownError): + await second + assert paid_calls == 1 + else: + await asyncio.wait_for(second_retry.wait(), timeout=1) + finally: + release.set() + + assert (await first).status_code == 200 + if not idempotency_key: + assert (await second).status_code == 200 + + @pytest.mark.asyncio + async def test_unknown_operation_blocks_a_new_challenge_for_same_request(self) -> None: + challenge_calls = paid_calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal challenge_calls, paid_calls + if "authorization" not in request.headers: + challenge_calls += 1 + return payment_required(f"challenge-{challenge_calls}") + paid_calls += 1 + raise httpx.WriteError("response lost") + + method = MockMethod() + transport = PaymentRuntime([method]).payment_transport(httpx.MockTransport(handler)) + + def request() -> httpx.Request: + return httpx.Request( + "POST", + "https://example.com/resource", + content=b"same body", + ) + + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request(request()) + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request(request()) + + assert challenge_calls == 2 + assert paid_calls == 1 + method.create_credential.assert_awaited_once() + + @pytest.mark.asyncio + async def test_challenge_cookies_apply_to_retry_and_reach_caller(self) -> None: + initial = httpx.Response( + 402, + headers=[ + ("www-authenticate", payment_required().headers["www-authenticate"]), + ("set-cookie", "session=new; Path=/"), + ("set-cookie", "nonce=one; Path=/"), + ], + ) + paid = httpx.Response(200, headers={"set-cookie": "paid=yes; Path=/"}, content=b"ok") + inner = MockTransport([initial, paid]) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + response = await transport.handle_async_request( + httpx.Request( + "GET", + "https://example.com/resource", + headers={"cookie": "session=old; keep=yes"}, + ) + ) + + retry_cookie = inner.requests[1].headers["cookie"] + assert "session=new" in retry_cookie + assert "nonce=one" in retry_cookie + assert "keep=yes" in retry_cookie + assert "session=old" not in retry_cookie + assert response.headers.get_list("set-cookie") == [ + "session=new; Path=/", + "nonce=one; Path=/", + "paid=yes; Path=/", + ] + + @pytest.mark.asyncio + async def test_combined_authentication_header_finds_payment_challenge(self) -> None: + payment = Challenge( + id="combined", + method="tempo", + intent="charge", + request={"amount": "1000", "currency": "USD"}, + ).to_www_authenticate("example.com") + inner = MockTransport( + [ + httpx.Response( + 402, + headers={"www-authenticate": f'Bearer realm="api", {payment}'}, + ), + httpx.Response(200, content=b"paid"), + ] + ) + + response = await PaymentTransport( + methods=[MockMethod()], + inner=inner, + ).handle_async_request(httpx.Request("GET", "https://example.com")) + + assert response.status_code == 200 + assert len(inner.requests) == 2 + + @pytest.mark.asyncio + async def test_valid_offer_is_preferred_over_an_expired_offer(self) -> None: + expired = Challenge( + id="expired", + method="tempo", + intent="charge", + request={}, + expires="2020-01-01T00:00:00Z", + ) + current = Challenge(id="current", method="tempo", intent="charge", request={}) + inner = MockTransport( + [ + httpx.Response( + 402, + headers=[ + ("www-authenticate", expired.to_www_authenticate("example.com")), + ("www-authenticate", current.to_www_authenticate("example.com")), + ], + ), + httpx.Response(200, content=b"paid"), + ] + ) + method = MockMethod() + + await PaymentTransport(methods=[method], inner=inner).handle_async_request( + httpx.Request("GET", "https://example.com") + ) + + assert method.create_credential.await_args is not None + assert method.create_credential.await_args.args[0].id == "current" + + @pytest.mark.asyncio + async def test_http_matching_rejects_wrong_intent(self) -> None: + challenge = Challenge( + id="subscription", + method="tempo", + intent="subscription", + request={}, + ) + inner = MockTransport( + [ + httpx.Response( + 402, + headers={"www-authenticate": challenge.to_www_authenticate("example.com")}, + ), + httpx.Response(200, content=b"paid"), + ] + ) + + method = MockMethod() + response = await PaymentTransport( + methods=[method], + inner=inner, + ).handle_async_request(httpx.Request("GET", "https://example.com")) + + assert response.status_code == 402 + method.create_credential.assert_not_awaited() + + @pytest.mark.parametrize("blocked", ["missing", "disallowed"]) + @pytest.mark.asyncio + async def test_nonpayable_402_body_stays_lazy(self, blocked: str) -> None: + stream = TrackingStream([b"explanation"]) + response = ( + httpx.Response(402, stream=stream) + if blocked == "missing" + else httpx.Response( + 402, + headers=payment_required().headers, + stream=stream, + ) + ) + runtime = PaymentRuntime( + [MockMethod()], + allowed_origins=[] if blocked == "disallowed" else None, + ) + + result = await runtime.payment_transport(MockTransport([response])).handle_async_request( + httpx.Request("GET", "https://example.com") + ) + + assert result is response + assert not stream.started + assert not stream.closed + + @pytest.mark.asyncio + async def test_redirected_challenge_uses_response_request_everywhere(self) -> None: + original = httpx.Request("GET", "https://start.test") + challenged = httpx.Request("POST", "https://paid.test/resource", content=b"body") + challenge_response = payment_required() + challenge_response.request = challenged + inner = MockTransport([challenge_response, httpx.Response(200, content=b"paid")]) + runtime = PaymentRuntime([MockMethod()], allowed_origins=["https://paid.test"]) + failed: list[dict[str, Any]] = [] + runtime.events.on("payment.failed", failed.append) + + response = await runtime.payment_transport(inner).handle_async_request(original) + + assert response.status_code == 200 + assert inner.requests[1].method == "POST" + assert inner.requests[1].url == challenged.url + assert inner.requests[1].content == b"body" + + @pytest.mark.asyncio + async def test_redirect_cannot_trigger_a_second_payment(self) -> None: + requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path == "/start": + if "authorization" not in request.headers: + return payment_required("first") + return httpx.Response(302, headers={"location": "/next"}) + return payment_required("second") + + method = MockMethod() + transport = PaymentRuntime([method]).payment_transport(httpx.MockTransport(handler)) + async with httpx.AsyncClient(transport=transport, follow_redirects=True) as client: + response = await client.get("https://example.com/start") + + assert response.status_code == 402 + assert [request.url.path for request in requests] == ["/start", "/start", "/next"] + method.create_credential.assert_awaited_once() + + @pytest.mark.asyncio + async def test_unmatched_failure_abort_closes_challenge_response(self) -> None: + class Abort(BaseException): + pass + + stream = TrackingStream([b"challenge"]) + response = httpx.Response( + 402, + headers=payment_required().headers, + stream=stream, + ) + runtime = PaymentRuntime([]) + runtime.events.on("payment.failed", lambda _payload: (_ for _ in ()).throw(Abort())) + + with pytest.raises(Abort): + await runtime.payment_transport(MockTransport([response])).handle_async_request( + httpx.Request("GET", "https://example.com") + ) + + assert stream.closed + + @pytest.mark.asyncio + async def test_matching_failure_closes_challenge_response(self) -> None: + class BrokenRuntime(PaymentRuntime): + def match_challenge(self, *args: Any, **kwargs: Any): + raise RuntimeError("match failed") + + stream = TrackingStream([b"challenge"]) + response = httpx.Response(402, headers=payment_required().headers, stream=stream) + transport = BrokenRuntime([MockMethod()]).payment_transport(MockTransport([response])) + + with pytest.raises(RuntimeError, match="match failed"): + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert stream.closed + + @pytest.mark.asyncio + async def test_paid_response_event_abort_closes_response(self) -> None: + class Abort(BaseException): + pass + + stream = TrackingStream([b"paid"]) + transport = PaymentTransport( + methods=[MockMethod()], + inner=MockTransport([payment_required(), httpx.Response(200, stream=stream)]), + ) + transport.on_payment_response(lambda _payload: (_ for _ in ()).throw(Abort())) + + with pytest.raises(Abort): + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert stream.closed + + @pytest.mark.parametrize("terminal", ["complete", "error", "close"]) + @pytest.mark.asyncio + async def test_success_status_completes_payment_before_body(self, terminal: str) -> None: + stream = TrackingStream([b"paid"], broken=terminal == "error") + inner = MockTransport( + [ + payment_required(), + httpx.Response(200, stream=stream), + payment_required(), + httpx.Response(200, content=b"again"), + ] + ) + method = MockMethod() + runtime = PaymentRuntime([method]) + transport = runtime.payment_transport(inner) + request = httpx.Request("GET", "https://example.com") + response = await transport.handle_async_request(request) + + if terminal == "complete": + assert await response.aread() == b"paid" + again = await transport.handle_async_request( + httpx.Request("GET", "https://example.com") + ) + assert again.status_code == 200 + assert method.create_credential.await_count == 2 + return + if terminal == "error": + with pytest.raises(httpx.ReadError): + await response.aread() + else: + await response.aclose() + + again = await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + assert again.status_code == 200 + assert method.create_credential.await_count == 2 + + @pytest.mark.asyncio + async def test_repeated_challenge_after_credential_is_unknown(self) -> None: + method = MockMethod() + transport = PaymentTransport( + methods=[method], + inner=MockTransport([payment_required(), payment_required()]), + ) + + with pytest.raises(PaymentOutcomeUnknownError, match="Do not blindly retry"): + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + method.create_credential.assert_awaited_once() + + @pytest.mark.asyncio + async def test_error_response_retains_unknown_outcome(self) -> None: + runtime = PaymentRuntime([MockMethod()]) + inner = MockTransport( + [ + payment_required(), + httpx.Response(503, content=b"unavailable"), + payment_required(), + ] + ) + transport = runtime.payment_transport(inner) + failed: list[dict[str, Any]] = [] + runtime.events.on("payment.failed", failed.append) + + response = await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request(httpx.Request("GET", "https://example.com")) + + assert response.status_code == 503 + assert isinstance(failed[0]["error"], PaymentOutcomeUnknownError) + + @pytest.mark.asyncio + async def test_reset_allows_reusing_the_same_request(self) -> None: + paid_calls = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal paid_calls + if "authorization" not in request.headers: + return payment_required(f"challenge-{paid_calls}") + paid_calls += 1 + if paid_calls == 1: + raise httpx.WriteError("response lost") + return httpx.Response(200, content=b"paid") + + runtime = PaymentRuntime([MockMethod()]) + transport = runtime.payment_transport(httpx.MockTransport(handler)) + request = httpx.Request( + "POST", + "https://example.com/resource", + content=b"same body", + ) + + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request(request) + runtime.reset_unknown_outcomes(reconciled=True) + + response = await transport.handle_async_request(request) + + assert response.content == b"paid" + assert paid_calls == 2 + + @pytest.mark.asyncio + async def test_only_originating_runtime_can_reconcile_request_marker(self) -> None: + first_method = MockMethod() + first_runtime = PaymentRuntime([first_method]) + first_runtime.reset_unknown_outcomes(reconciled=True) + + async def lose_paid_response(request: httpx.Request) -> httpx.Response: + if "authorization" in request.headers: + raise httpx.WriteError("response lost") + return payment_required("first") + + request = httpx.Request( + "POST", + "https://example.com/resource", + content=b"same body", + ) + with pytest.raises(PaymentOutcomeUnknownError): + await first_runtime.payment_transport( + httpx.MockTransport(lose_paid_response) + ).handle_async_request(request) + + second_method = MockMethod() + second_inner = MockTransport( + [ + payment_required("second"), + payment_required("second"), + httpx.Response(200, content=b"paid"), + ] + ) + second_transport = PaymentRuntime([second_method]).payment_transport(second_inner) + + with pytest.raises(PaymentOutcomeUnknownError): + await second_transport.handle_async_request(request) + second_method.create_credential.assert_not_awaited() + + first_runtime.reset_unknown_outcomes(reconciled=True) + response = await second_transport.handle_async_request(request) + + assert response.content == b"paid" + second_method.create_credential.assert_awaited_once() + + @pytest.mark.asyncio + async def test_bounded_unknown_outcomes_trip_and_reset_circuit( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.setattr("mpp.client._http._MAX_UNRECONCILED_OUTCOMES", 1) + + class FailingPaidTransport(httpx.AsyncBaseTransport): + def __init__(self) -> None: + self.requests: list[httpx.Request] = [] + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + if "authorization" in request.headers: + raise httpx.WriteError("response lost") + return payment_required(request.url.path) + + method = MockMethod() + runtime = PaymentRuntime([method]) + inner = FailingPaidTransport() + transport = runtime.payment_transport(inner) + + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request( + httpx.Request("POST", "https://example.com/one", content="one") + ) + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request( + httpx.Request("POST", "https://example.com/two", content="two") + ) + assert method.create_credential.await_count == 1 + + with pytest.raises(ValueError, match="externally reconciled"): + runtime.reset_unknown_outcomes(reconciled=False) + runtime.reset_unknown_outcomes(reconciled=True) + with pytest.raises(PaymentOutcomeUnknownError): + await transport.handle_async_request( + httpx.Request("POST", "https://example.com/two", content="two") + ) + assert method.create_credential.await_count == 2 + + +@pytest.mark.parametrize( + ("set_cookie", "url", "expected"), + [ + ("session=; Max-Age=0; Path=/", "https://example.com/public", False), + ("session=; Max-Age=0; Path=/admin", "https://example.com/public", True), + ("session=new; Secure; Path=/", "http://example.com/public", True), + ("session=new; Domain=other.test; Path=/", "https://example.com/public", True), + ], +) +def test_challenge_cookie_scope( + set_cookie: str, + url: str, + expected: bool, +) -> None: + from mpp.client._http import _apply_response_cookies + + source = httpx.Request("GET", url) + response = httpx.Response(402, headers={"set-cookie": set_cookie}, request=source) + retry = httpx.Request("GET", url, headers={"cookie": "session=old; keep=yes"}) + + _apply_response_cookies(response, source, retry) + + assert ("session=old" in retry.headers["cookie"]) is expected + assert "keep=yes" in retry.headers["cookie"] class TestClient: diff --git a/tests/test_errors.py b/tests/test_errors.py index ad1e75b..6a7be78 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -12,6 +12,7 @@ PaymentExpiredError, PaymentInsufficientError, PaymentMethodUnsupportedError, + PaymentOutcomeUnknownError, PaymentRequiredError, VerificationFailedError, ) @@ -105,6 +106,17 @@ def test_auto_title(self, cls: type, expected_title: str) -> None: class TestSubclassInstantiation: + def test_unknown_outcome_retains_reconciliation_context(self) -> None: + challenge = type("Challenge", (), {"id": "challenge-1"})() + cause = TimeoutError("response lost") + request = object() + error = PaymentOutcomeUnknownError(challenge, cause, credential="proof", request=request) + + assert "Do not blindly retry" in str(error) + assert error.cause is cause + assert error.credential == "proof" + assert error.request is request + def test_payment_required_with_args(self) -> None: err = PaymentRequiredError(realm="api.example.com", description="Monthly quota") assert "api.example.com" in str(err) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 0a482ca..31ef0f1 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio +import gc +import weakref from types import MappingProxyType from typing import Any @@ -136,6 +138,16 @@ async def test_async_context_closes_runtime() -> None: await runtime.emit_event("test", {}) +async def test_async_lifecycle_aliases() -> None: + runtime = PaymentRuntime() + + assert await runtime.astart() is runtime + await runtime.aclose() + + with pytest.raises(RuntimeError, match="closed"): + await runtime.astart() + + async def test_borrowed_methods_are_not_entered_or_closed() -> None: events: list[str] = [] @@ -154,6 +166,32 @@ async def __aexit__(self, *_args: Any) -> None: assert events == [] +async def test_child_scope_does_not_retain_completed_parent_task() -> None: + runtime = PaymentRuntime() + release = asyncio.Event() + parent_ref: weakref.ReferenceType[asyncio.Task[Any]] | None = None + + async def parent() -> asyncio.Task[bool]: + nonlocal parent_ref + with runtime._paid_operation(): + child = asyncio.create_task(release.wait()) + task = asyncio.current_task() + assert task is not None + parent_ref = weakref.ref(task) + return child + + parent_task = asyncio.create_task(parent()) + child = await parent_task + del parent_task + await asyncio.sleep(0) + gc.collect() + + assert parent_ref is not None + assert parent_ref() is None + release.set() + await child + + def test_matching_prefers_method_order_by_default() -> None: class StripeMethod(MockMethod): name = "stripe" From bafcc2dbb11af1d1c8b502b71945ec330d884b12 Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:43:31 -0400 Subject: [PATCH 2/2] fix: tighten async payment safeguards --- pyproject.toml | 1 - src/mpp/client/_http.py | 5 ++-- src/mpp/client/transport.py | 6 ++--- src/mpp/runtime.py | 46 +++++++++++++++---------------------- tests/test_client.py | 40 ++++++++++++++++++++++++++++++-- tests/test_runtime.py | 28 ---------------------- 6 files changed, 62 insertions(+), 64 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index edb03e8..7596fbc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,6 @@ description = "Python SDK for the Machine Payments Protocol (MPP)" readme = "README.md" requires-python = ">=3.11" dependencies = [ - "anyio>=4,<5", "httpx>=0.27", ] authors = [ diff --git a/src/mpp/client/_http.py b/src/mpp/client/_http.py index 65ba53e..df61231 100644 --- a/src/mpp/client/_http.py +++ b/src/mpp/client/_http.py @@ -245,9 +245,8 @@ def retry_request(self, authorization: str) -> httpx.Request: class _AllowedOrigins: def __init__(self, allowed: Sequence[str] | None) -> None: self._allow_all = allowed is None - self._origins = { - origin for value in allowed or () if (origin := _origin(str(value))) is not None - } + values = (allowed,) if isinstance(allowed, str) else allowed or () + self._origins = {origin for value in values if (origin := _origin(value)) is not None} def allows(self, url: httpx.URL) -> bool: return self._allow_all or _httpx_origin(url) in self._origins diff --git a/src/mpp/client/transport.py b/src/mpp/client/transport.py index 697a0cb..a56fe68 100644 --- a/src/mpp/client/transport.py +++ b/src/mpp/client/transport.py @@ -103,7 +103,7 @@ def __init__( self._events = self._runtime.events async def _fail(self, payment: _HttpPayment, error: Exception, **details: Any) -> None: - await self._runtime.emit_event(PAYMENT_FAILED, payment.failed(error, **details)) + await self._runtime._emit_event(PAYMENT_FAILED, payment.failed(error, **details)) async def _unknown( self, @@ -203,7 +203,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: raise try: - credential = await self._runtime.create_credential( + credential = await self._runtime._create_credential( challenge, method, event_payload=payment.event_payload(), @@ -254,7 +254,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: _response_request(payment_response, retry_request) _propagate_response_cookies(response, payment_response) if payment_response.is_success: - await self._runtime.emit_event( + await self._runtime._emit_event( PAYMENT_RESPONSE, payment.event_payload(payment_response), ) diff --git a/src/mpp/runtime.py b/src/mpp/runtime.py index 3b9c9e2..b3b02d5 100644 --- a/src/mpp/runtime.py +++ b/src/mpp/runtime.py @@ -4,11 +4,9 @@ from collections.abc import Mapping, Sequence from contextlib import contextmanager -from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Protocol, Self, runtime_checkable import httpx -from anyio import get_current_task from mpp import Challenge, Credential from mpp.events import ( @@ -22,21 +20,6 @@ from mpp.client import PaymentTransport from mpp.client._http import _HttpPaymentAttempt -_PAID_RUNTIMES: ContextVar[tuple[tuple[object, int], ...]] = ContextVar( - "mpp_paid_runtimes", - default=(), -) - - -def _scope_active(key: object) -> bool: - try: - owner = get_current_task().id - except RuntimeError: - return False - return any( - scope_key is key and scope_owner == owner for scope_key, scope_owner in _PAID_RUNTIMES.get() - ) - @runtime_checkable class Method(Protocol): @@ -75,11 +58,10 @@ def __init__( self._closed = False self._closing = False self._paid_operations = 0 - self._scope_key = object() def start(self) -> Self: """Open the runtime, or return it if already open.""" - if self._closed or (self._closing and not _scope_active(self._scope_key)): + if self._closed or self._closing: raise RuntimeError("PaymentRuntime is closed") return self @@ -135,6 +117,21 @@ async def create_credential( ) -> Credential: """Create a credential and emit its lifecycle events.""" self.start() + return await self._create_credential( + challenge, + method, + allow_name_only=allow_name_only, + event_payload=event_payload, + ) + + async def _create_credential( + self, + challenge: Challenge, + method: Method, + *, + allow_name_only: bool = False, + event_payload: dict[str, Any] | None = None, + ) -> Credential: if not any(candidate is method for candidate in self.methods): raise ValueError("Method is not installed in this PaymentRuntime") if challenge.method != method.name or ( @@ -168,6 +165,9 @@ async def create_credential( async def emit_event(self, name: str, payload: EventPayload) -> Any: """Emit an event on the caller's event loop.""" self.start() + return await self._emit_event(name, payload) + + async def _emit_event(self, name: str, payload: EventPayload) -> Any: return await self.events.emit(name, payload) def payment_transport( @@ -192,25 +192,17 @@ def _begin_http_payment( challenge: Challenge, request: httpx.Request, ) -> _HttpPaymentAttempt: - self.start() return self._http.begin(challenge, request) @contextmanager def _paid_operation(self): """Keep a committed payment flow alive if close is requested.""" - if _scope_active(self._scope_key): - yield - return self.start() - # Context variables propagate to child tasks; bind this lease to its owner. - scope = (self._scope_key, get_current_task().id) - token = _PAID_RUNTIMES.set((*_PAID_RUNTIMES.get(), scope)) self._paid_operations += 1 try: yield finally: self._paid_operations -= 1 - _PAID_RUNTIMES.reset(token) if self._closing and not self._paid_operations: self._closed = True diff --git a/tests/test_client.py b/tests/test_client.py index a0662c3..8d70eec 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -712,10 +712,17 @@ async def create(self, challenge: Challenge) -> Credential: assert loops == [caller_loop] + @pytest.mark.parametrize( + "allowed_origins", + [["https://allowed.test:443"], "https://allowed.test:443"], + ) @pytest.mark.asyncio - async def test_origin_policy_is_exact_and_normalizes_default_ports(self) -> None: + async def test_origin_policy_is_exact_and_normalizes_default_ports( + self, + allowed_origins: list[str] | str, + ) -> None: method = MockMethod() - runtime = PaymentRuntime([method], allowed_origins=["https://allowed.test:443"]) + runtime = PaymentRuntime([method], allowed_origins=allowed_origins) blocked = MockTransport([payment_required()]) allowed = MockTransport([payment_required(), httpx.Response(200, content=b"paid")]) @@ -847,6 +854,35 @@ def close_after_spawning_child(payload: dict[str, Any]) -> None: assert len(child_inner.requests) == 1 assert method.create_credential.await_count == 1 + @pytest.mark.asyncio + async def test_close_rejects_nested_payment_in_same_task(self) -> None: + method = MockMethod() + runtime = PaymentRuntime([method]) + nested_inner = MockTransport([payment_required("nested")]) + nested = runtime.payment_transport(nested_inner) + + async def close_then_pay(payload: dict[str, Any]) -> None: + if payload["challenge"].id != "parent": + return + runtime.close() + with pytest.raises(RuntimeError, match="closed"): + await nested.handle_async_request( + httpx.Request("GET", "https://example.com/nested") + ) + + runtime.events.on("credential.created", close_then_pay) + parent = runtime.payment_transport( + MockTransport([payment_required("parent"), httpx.Response(200, content=b"parent")]) + ) + + response = await parent.handle_async_request( + httpx.Request("GET", "https://example.com/parent") + ) + + assert response.content == b"parent" + assert len(nested_inner.requests) == 1 + assert method.create_credential.await_count == 1 + @pytest.mark.asyncio async def test_close_waits_for_child_with_its_own_lease(self) -> None: method = MockMethod() diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 31ef0f1..9fe88c2 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -3,8 +3,6 @@ from __future__ import annotations import asyncio -import gc -import weakref from types import MappingProxyType from typing import Any @@ -166,32 +164,6 @@ async def __aexit__(self, *_args: Any) -> None: assert events == [] -async def test_child_scope_does_not_retain_completed_parent_task() -> None: - runtime = PaymentRuntime() - release = asyncio.Event() - parent_ref: weakref.ReferenceType[asyncio.Task[Any]] | None = None - - async def parent() -> asyncio.Task[bool]: - nonlocal parent_ref - with runtime._paid_operation(): - child = asyncio.create_task(release.wait()) - task = asyncio.current_task() - assert task is not None - parent_ref = weakref.ref(task) - return child - - parent_task = asyncio.create_task(parent()) - child = await parent_task - del parent_task - await asyncio.sleep(0) - gc.collect() - - assert parent_ref is not None - assert parent_ref() is None - release.set() - await child - - def test_matching_prefers_method_order_by_default() -> None: class StripeMethod(MockMethod): name = "stripe"