diff --git a/.changelog/tempo-sessions.md b/.changelog/tempo-sessions.md new file mode 100644 index 00000000..3f68106b --- /dev/null +++ b/.changelog/tempo-sessions.md @@ -0,0 +1,5 @@ +--- +pympp: minor +--- + +Added private-key Tempo sessions with persistent channels, top-ups, resumption, and sync or async SSE streaming. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 77721c02..b727a358 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -266,10 +266,24 @@ jobs: if: matrix.profile == 'tempo' run: | .smoke-venv/bin/python - <<'PY' - from mpp.methods.tempo import ChargeIntent, TempoAccount + from pytempo import TempoTransaction + + from mpp.methods.tempo import ( + ChargeIntent, + TempoAccount, + TempoSessionMethod, + tempo_session, + ) assert ChargeIntent.__name__ == "ChargeIntent" assert TempoAccount.__name__ == "TempoAccount" + assert hasattr(TempoTransaction, "encode_for_signing") + + method = tempo_session( + account=TempoAccount.from_key("0x" + "01" * 32), + max_deposit=1, + ) + assert isinstance(method, TempoSessionMethod) print("tempo install smoke test passed") PY diff --git a/src/mpp/methods/tempo/__init__.py b/src/mpp/methods/tempo/__init__.py index 2cb2b74a..0509347d 100644 --- a/src/mpp/methods/tempo/__init__.py +++ b/src/mpp/methods/tempo/__init__.py @@ -51,6 +51,11 @@ "get_transfers", ), "mpp.methods.tempo.schemas": ("Split",), + "mpp.methods.tempo.session": ( + "TIP20_CHANNEL_ESCROW", + "TempoSessionMethod", + "tempo_session", + ), } diff --git a/src/mpp/methods/tempo/__init__.pyi b/src/mpp/methods/tempo/__init__.pyi index 410b5dc5..ba22610e 100644 --- a/src/mpp/methods/tempo/__init__.pyi +++ b/src/mpp/methods/tempo/__init__.pyi @@ -15,6 +15,9 @@ from mpp.methods.tempo.intents import Transfer as _Transfer from mpp.methods.tempo.intents import ValidateSender as _ValidateSender from mpp.methods.tempo.intents import get_transfers as _get_transfers from mpp.methods.tempo.schemas import Split as _Split +from mpp.methods.tempo.session import TIP20_CHANNEL_ESCROW as _TIP20_CHANNEL_ESCROW +from mpp.methods.tempo.session import TempoSessionMethod as _TempoSessionMethod +from mpp.methods.tempo.session import tempo_session as _tempo_session CHAIN_ID = _CHAIN_ID ESCROW_CONTRACTS = _ESCROW_CONTRACTS @@ -33,3 +36,6 @@ Transfer = _Transfer ValidateSender = _ValidateSender get_transfers = _get_transfers Split = _Split +TIP20_CHANNEL_ESCROW = _TIP20_CHANNEL_ESCROW +TempoSessionMethod = _TempoSessionMethod +tempo_session = _tempo_session diff --git a/src/mpp/methods/tempo/_session_sse.py b/src/mpp/methods/tempo/_session_sse.py new file mode 100644 index 00000000..dada0ddf --- /dev/null +++ b/src/mpp/methods/tempo/_session_sse.py @@ -0,0 +1,274 @@ +"""SSE filtering for Tempo session payment control frames.""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator +from dataclasses import dataclass +from typing import Any, Literal, TypeAlias, cast + +import httpx + +ControlPayload: TypeAlias = dict[str, Any] +SyncControlHandler: TypeAlias = Callable[[ControlPayload], None] +AsyncControlHandler: TypeAlias = Callable[[ControlPayload], Awaitable[None]] +ControlKind: TypeAlias = Literal["payment-need-voucher", "payment-receipt"] + + +@dataclass(frozen=True, slots=True) +class _Frame: + raw: bytes + control: tuple[ControlKind, ControlPayload] | None = None + + +class _FrameDecoder: + def __init__(self) -> None: + self._buffer = bytearray() + + def feed(self, chunk: bytes) -> list[_Frame]: + self._buffer.extend(chunk) + return self._drain(complete=False) + + def finish(self) -> list[_Frame]: + frames = self._drain(complete=True) + if self._buffer: + frames.append(_Frame(bytes(self._buffer))) + self._buffer.clear() + return frames + + def _drain(self, *, complete: bool) -> list[_Frame]: + frames: list[_Frame] = [] + while separator := _find_separator(self._buffer, complete=complete): + index, length = separator + end = index + length + body = bytes(self._buffer[:index]) + raw = bytes(self._buffer[:end]) + del self._buffer[:end] + frames.append(_Frame(raw, _parse_control(body))) + return frames + + +class _SyncSseStream(httpx.SyncByteStream): + def __init__( + self, + response: httpx.Response, + on_need_voucher: SyncControlHandler, + on_receipt: SyncControlHandler, + ) -> None: + self._response = response + self._on_need_voucher = on_need_voucher + self._on_receipt = on_receipt + self._closed = False + + def __iter__(self) -> Iterator[bytes]: + decoder = _FrameDecoder() + try: + for chunk in self._response.iter_bytes(): + yield from self._handle(decoder.feed(chunk)) + yield from self._handle(decoder.finish()) + finally: + self.close() + + def close(self) -> None: + if not self._closed: + self._closed = True + self._response.close() + + def _handle(self, frames: list[_Frame]) -> Iterator[bytes]: + for frame in frames: + if frame.control is None: + yield frame.raw + continue + kind, payload = frame.control + if kind == "payment-need-voucher": + self._on_need_voucher(payload) + else: + self._on_receipt(payload) + + +class _AsyncSseStream(httpx.AsyncByteStream): + def __init__( + self, + response: httpx.Response, + on_need_voucher: AsyncControlHandler, + on_receipt: AsyncControlHandler, + ) -> None: + self._response = response + self._on_need_voucher = on_need_voucher + self._on_receipt = on_receipt + self._closed = False + + async def __aiter__(self) -> AsyncIterator[bytes]: + decoder = _FrameDecoder() + try: + async for chunk in self._response.aiter_bytes(): + async for raw in self._handle(decoder.feed(chunk)): + yield raw + async for raw in self._handle(decoder.finish()): + yield raw + finally: + await self.aclose() + + async def aclose(self) -> None: + if not self._closed: + self._closed = True + await self._response.aclose() + + async def _handle(self, frames: list[_Frame]) -> AsyncIterator[bytes]: + for frame in frames: + if frame.control is None: + yield frame.raw + continue + kind, payload = frame.control + if kind == "payment-need-voucher": + await self._on_need_voucher(payload) + else: + await self._on_receipt(payload) + + +def wrap_sync_sse_response( + response: httpx.Response, + *, + on_need_voucher: SyncControlHandler, + on_receipt: SyncControlHandler, +) -> httpx.Response: + """Wrap a sync SSE response and consume valid payment control frames inline.""" + if not isinstance(response.stream, httpx.SyncByteStream): + raise TypeError("Expected a synchronous response stream") + return _wrap_response( + response, + _SyncSseStream(response, on_need_voucher, on_receipt), + ) + + +def wrap_async_sse_response( + response: httpx.Response, + *, + on_need_voucher: AsyncControlHandler, + on_receipt: AsyncControlHandler, +) -> httpx.Response: + """Wrap an async SSE response and consume valid payment control frames inline.""" + if not isinstance(response.stream, httpx.AsyncByteStream): + raise TypeError("Expected an asynchronous response stream") + return _wrap_response( + response, + _AsyncSseStream(response, on_need_voucher, on_receipt), + ) + + +def _wrap_response( + response: httpx.Response, + stream: httpx.SyncByteStream | httpx.AsyncByteStream, +) -> httpx.Response: + headers = httpx.Headers(response.headers) + headers.pop("content-length", None) + headers.pop("content-encoding", None) + try: + request = response.request + except RuntimeError: + request = None + return httpx.Response( + response.status_code, + headers=headers, + stream=stream, + request=request, + extensions=dict(response.extensions), + history=list(response.history), + default_encoding=response.default_encoding, + ) + + +def _find_separator(buffer: bytearray, *, complete: bool) -> tuple[int, int] | None: + index = 0 + while index < len(buffer): + first = _line_ending_length(buffer, index, complete=complete) + if first: + second = _line_ending_length(buffer, index + first, complete=complete) + if second: + return index, first + second + index += first + else: + index += 1 + return None + + +def _line_ending_length(buffer: bytearray, index: int, *, complete: bool) -> int: + if index >= len(buffer): + return 0 + if buffer[index] == 0x0A: + return 1 + if buffer[index] != 0x0D: + return 0 + if index + 1 < len(buffer) and buffer[index + 1] == 0x0A: + return 2 + if index + 1 == len(buffer) and not complete: + return 0 + return 1 + + +def _parse_control(body: bytes) -> tuple[ControlKind, ControlPayload] | None: + try: + text = body.decode("utf-8") + except UnicodeDecodeError: + return None + + event = "message" + data: list[str] = [] + for line in text.replace("\r\n", "\n").replace("\r", "\n").split("\n"): + if line.startswith(":"): + continue + field, separator, value = line.partition(":") + if separator and value.startswith(" "): + value = value[1:] + if field == "event": + event = value + elif field == "data": + data.append(value) + + if event not in {"payment-need-voucher", "payment-receipt"} or not data: + return None + try: + value = json.loads("\n".join(data), parse_constant=_reject_json_constant) + except (ValueError, TypeError): + return None + if not isinstance(value, dict): + return None + payload = cast(ControlPayload, value) + if event == "payment-need-voucher" and _is_need_voucher(payload): + return event, payload + if event == "payment-receipt" and _is_receipt(payload): + return event, payload + return None + + +def _reject_json_constant(value: str) -> None: + raise ValueError(f"Invalid JSON constant: {value}") + + +def _is_need_voucher(value: ControlPayload) -> bool: + return all( + isinstance(value.get(field), str) + for field in ("channelId", "requiredCumulative", "acceptedCumulative", "deposit") + ) + + +def _is_receipt(value: ControlPayload) -> bool: + required_strings = ( + "timestamp", + "reference", + "challengeId", + "channelId", + "acceptedCumulative", + "spent", + ) + units = value.get("units") + return ( + value.get("method") == "tempo" + and value.get("intent") == "session" + and value.get("status") == "success" + and all(isinstance(value.get(field), str) for field in required_strings) + and ( + "units" not in value or isinstance(units, (int, float)) and not isinstance(units, bool) + ) + and ("txHash" not in value or isinstance(value.get("txHash"), str)) + ) diff --git a/src/mpp/methods/tempo/session.py b/src/mpp/methods/tempo/session.py new file mode 100644 index 00000000..2231e1db --- /dev/null +++ b/src/mpp/methods/tempo/session.py @@ -0,0 +1,1198 @@ +"""Client-side Tempo TIP-1034 session payments.""" + +from __future__ import annotations + +import asyncio +import json +import math +import os +import re +import threading +import time +from collections.abc import Sequence +from contextlib import asynccontextmanager, contextmanager +from contextvars import ContextVar +from dataclasses import asdict, dataclass, field +from datetime import datetime +from typing import TYPE_CHECKING, Any, Literal, cast + +import httpx + +from mpp import Challenge, Credential, MemoryStore, Receipt, Store +from mpp.methods.tempo._defaults import CHAIN_ID, RPC_URL +from mpp.methods.tempo._rpc import _rpc_call, get_tx_params +from mpp.methods.tempo._session_sse import ( + wrap_async_sse_response, + wrap_sync_sse_response, +) +from mpp.methods.tempo.client import ( + EXPIRING_NONCE_KEY, + FEE_PAYER_VALID_BEFORE_SECS, + TransactionError, +) +from mpp.methods.tempo.fee_payer_policy import get_policy + +if TYPE_CHECKING: + from mpp.methods.tempo.account import TempoAccount + from mpp.runtime import AsyncHttpResponseContext, SyncHttpResponseContext + + +TIP20_CHANNEL_ESCROW = "0x4d50500000000000000000000000000000000000" +NONCE_PRECOMPILE = "0x4e4f4e4345000000000000000000000000000000" +ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" +MAX_UINT96 = (1 << 96) - 1 +MAX_UINT64 = (1 << 64) - 1 +SESSION_GAS_LIMIT = 2_000_000 + +_ADDRESS_RE = re.compile(r"^0x[0-9a-fA-F]{40}$") +_BYTES32_RE = re.compile(r"^0x[0-9a-fA-F]{64}$") +_AMOUNT_RE = re.compile(r"^(?:0|[1-9][0-9]*)$") +_HELD: ContextVar[frozenset[tuple[int, str]]] = ContextVar( + "mpp_tempo_session_locks", default=frozenset() +) + + +@dataclass(frozen=True, slots=True) +class _Descriptor: + payer: str + payee: str + operator: str + token: str + salt: str + authorized_signer: str + expiring_nonce_hash: str + + @classmethod + def parse(cls, value: object) -> _Descriptor: + if not isinstance(value, dict): + raise ValueError("session descriptor must be an object") + return cls( + payer=_address(value.get("payer"), "descriptor.payer"), + payee=_address(value.get("payee"), "descriptor.payee"), + operator=_address(value.get("operator"), "descriptor.operator"), + token=_address(value.get("token"), "descriptor.token"), + salt=_bytes32(value.get("salt"), "descriptor.salt"), + authorized_signer=_address( + value.get("authorizedSigner"), "descriptor.authorizedSigner" + ), + expiring_nonce_hash=_bytes32( + value.get("expiringNonceHash"), "descriptor.expiringNonceHash" + ), + ) + + def wire(self) -> dict[str, str]: + return { + "payer": self.payer, + "payee": self.payee, + "operator": self.operator, + "token": self.token, + "salt": self.salt, + "authorizedSigner": self.authorized_signer, + "expiringNonceHash": self.expiring_nonce_hash, + } + + +@dataclass(slots=True) +class _Channel: + channel_id: str + descriptor: _Descriptor + escrow: str + chain_id: int + deposit: int + cumulative: int + accepted: int = 0 + status: Literal["pending", "open"] = "pending" + pending_transaction: str | None = None + pending_top_up: int | None = None + + def dump(self) -> str: + value = asdict(self) + value["descriptor"] = self.descriptor.wire() + for name in ("deposit", "cumulative", "accepted", "pending_top_up"): + if value[name] is None: + continue + value[name] = str(value[name]) + return json.dumps(value, separators=(",", ":"), sort_keys=True) + + @classmethod + def load(cls, raw: object) -> _Channel: + if isinstance(raw, bytes): + raw = raw.decode() + try: + value = json.loads(cast("str", raw)) + except (TypeError, ValueError) as error: + raise ValueError("invalid stored Tempo session channel") from error + if not isinstance(value, dict) or value.get("status") not in {"pending", "open"}: + raise ValueError("invalid stored Tempo session channel") + return cls( + channel_id=_bytes32(value.get("channel_id"), "channel_id"), + descriptor=_Descriptor.parse(value.get("descriptor")), + escrow=_address(value.get("escrow"), "escrow"), + chain_id=_chain_id(value.get("chain_id"), "chain_id"), + deposit=_amount(value.get("deposit"), "deposit"), + cumulative=_amount(value.get("cumulative"), "cumulative"), + accepted=_amount(value.get("accepted"), "accepted"), + status=cast("Literal['pending', 'open']", value["status"]), + pending_transaction=_transaction_hex(value.get("pending_transaction")), + pending_top_up=( + _amount(value["pending_top_up"], "pending_top_up") + if value.get("pending_top_up") is not None + else None + ), + ) + + +@dataclass(frozen=True, slots=True) +class _Request: + amount: int + payee: str + token: str + operator: str + escrow: str + chain_id: int + fee_payer: bool + suggested_deposit: int | None + min_voucher_delta: int + snapshot: dict[str, Any] | None + scope: str + + +@dataclass +class TempoSessionMethod: + """Client-only TIP-1034 v2 method for HTTP and SSE payments.""" + + account: TempoAccount + max_deposit: int + rpc_url: str = RPC_URL + chain_id: int = CHAIN_ID + escrow: str = TIP20_CHANNEL_ESCROW + channel_store: Store = field(default_factory=MemoryStore) + name: str = field(default="tempo", init=False) + _intents: dict[str, object] = field( + default_factory=lambda: {"session": object()}, init=False, repr=False + ) + _locks: dict[str, threading.Lock] = field(default_factory=dict, init=False, repr=False) + _locks_guard: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) + + @property + def intents(self) -> dict[str, object]: + return self._intents + + async def create_credential( + self, challenge: Challenge, *, context: object | None = None + ) -> Credential: + request = self._resolve(challenge) + async with self._locked(request.scope): + return await self._create(challenge, request, context) + + async def handle_async_http_response( + self, exchange: AsyncHttpResponseContext + ) -> httpx.Response: + request = self._resolve(exchange.challenge) + action = exchange.credential.payload.get("action") + if not exchange.response.is_success: + return exchange.response + async with self._locked(request.scope): + await exchange.run_async( + self._accept_response( + request, + exchange.response, + exchange.challenge, + exchange.credential.payload, + ) + ) + if action == "topUp" and exchange.refetch: + return await exchange.refetch() + if not _is_sse(exchange.response): + wants_sse = "text/event-stream" in exchange.request.headers.get("accept", "").lower() + if action == "open" and wants_sse and exchange.refetch: + return await exchange.refetch() + return exchange.response + + async def need_voucher(event: dict[str, Any]) -> None: + async with self._locked(request.scope): + await self._need_voucher_async(exchange, request, event) + + async def receipt(event: dict[str, Any]) -> None: + async with self._locked(request.scope): + await exchange.run_async(self._record_receipt(request, event, exchange.challenge)) + + return wrap_async_sse_response( + exchange.response, on_need_voucher=need_voucher, on_receipt=receipt + ) + + def handle_http_response(self, exchange: SyncHttpResponseContext) -> httpx.Response: + request = self._resolve(exchange.challenge) + action = exchange.credential.payload.get("action") + if not exchange.response.is_success: + return exchange.response + with self._locked_sync(request.scope): + exchange.run_sync( + self._accept_response( + request, + exchange.response, + exchange.challenge, + exchange.credential.payload, + ) + ) + if action == "topUp" and exchange.refetch: + return exchange.refetch() + if not _is_sse(exchange.response): + wants_sse = "text/event-stream" in exchange.request.headers.get("accept", "").lower() + if action == "open" and wants_sse and exchange.refetch: + return exchange.refetch() + return exchange.response + + def need_voucher(event: dict[str, Any]) -> None: + with self._locked_sync(request.scope): + self._need_voucher_sync(exchange, request, event) + + def receipt(event: dict[str, Any]) -> None: + with self._locked_sync(request.scope): + exchange.run_sync(self._record_receipt(request, event, exchange.challenge)) + + return wrap_sync_sse_response( + exchange.response, on_need_voucher=need_voucher, on_receipt=receipt + ) + + async def _create( + self, challenge: Challenge, request: _Request, context: object | None + ) -> Credential: + payload = ( + await self._management(request, context) + if context is not None + else await self._payment(request) + ) + return Credential( + challenge=challenge.to_echo(), + payload=payload, + source=f"did:pkh:eip155:{request.chain_id}:{self.account.address.lower()}", + ) + + async def _payment(self, request: _Request) -> dict[str, Any]: + channel = await self._load(request) + if channel is None and request.snapshot is not None: + channel = await self._recover(request) + if channel is None: + target = request.amount + _limit(target, self.max_deposit, "initial voucher") + deposit = min(max(target, request.suggested_deposit or target), self.max_deposit) + return await self._open(request, deposit, target) + + replacement = await self._replace_expired_pending(request, channel) + if replacement is not None: + return replacement + + accepted = channel.accepted + if request.snapshot is not None: + snapshot = self._snapshot(request, channel) + required = _amount(snapshot["requiredCumulative"], "requiredCumulative") + accepted = _amount(snapshot["acceptedCumulative"], "acceptedCumulative") + if accepted > channel.cumulative: + raise ValueError("server accepted cumulative exceeds local signed watermark") + await self._reconcile_deposit(channel, _amount(snapshot["deposit"], "deposit")) + if channel.status == "pending": + channel.status = "open" + channel.pending_transaction = None + channel.cumulative = max(channel.cumulative, accepted) + channel.accepted = max(channel.accepted, accepted) + accepted = channel.accepted + self._validate(channel, request) + await self._save(request, channel) + else: + if channel.status == "pending": + return self._open_payload(channel) + if channel.pending_top_up is not None: + return self._top_up_payload(channel) + required = channel.accepted + request.amount + + if channel.pending_top_up is not None: + return self._top_up_payload(channel) + + target = max(required, channel.cumulative, accepted + request.min_voucher_delta) + _limit(target, self.max_deposit, "voucher") + if target > channel.deposit: + return await self._top_up(request, channel, target - channel.deposit) + return await self._voucher(request, channel, target) + + async def _management(self, request: _Request, context: object) -> dict[str, Any]: + if not isinstance(context, dict): + raise ValueError("Tempo session context must be an object") + channel = await self._require(request) + if context.get("channelId") != channel.channel_id: + raise ValueError("session context references the wrong channel") + if channel.status != "open": + raise ValueError("Tempo session channel is not open") + action = context.get("action") + if action == "topUp": + additional = _amount(context.get("additionalDeposit"), "additionalDeposit") + if additional == 0: + raise ValueError("additionalDeposit must be greater than zero") + _limit(channel.deposit + additional, self.max_deposit, "channel deposit") + return await self._top_up(request, channel, additional) + if action == "voucher": + if channel.pending_top_up is not None: + raise ValueError("Tempo session top-up has not been confirmed") + target = max(channel.cumulative, _amount(context.get("cumulativeAmount"), "cumulative")) + _limit(target, min(self.max_deposit, channel.deposit), "voucher") + return await self._voucher(request, channel, target) + raise ValueError(f"Unsupported Tempo session action: {action!r}") + + async def _open(self, request: _Request, deposit: int, cumulative: int) -> dict[str, Any]: + if deposit == 0: + raise ValueError("Tempo session deposit must be greater than zero") + salt = "0x" + os.urandom(32).hex() + payer = self.account.address.lower() + call = _call( + "open(address,address,address,uint96,bytes32,address)", + ["address", "address", "address", "uint96", "bytes32", "address"], + [ + request.payee, + request.operator, + request.token, + deposit, + bytes.fromhex(salt[2:]), + payer, + ], + ) + transaction, unsigned = await self._transaction( + request, call, nonce_key=_session_nonce_key(salt) + ) + try: + preimage = unsigned.encode_for_signing() + except AttributeError as error: + raise RuntimeError( + "Tempo sessions require a pytempo release with encode_for_signing()" + ) from error + descriptor = _Descriptor( + payer=payer, + payee=request.payee, + operator=request.operator, + token=request.token, + salt=salt, + authorized_signer=payer, + expiring_nonce_hash="0x" + _keccak(preimage + bytes.fromhex(payer[2:])).hex(), + ) + channel_id = _channel_id(descriptor, request.escrow, request.chain_id) + channel = _Channel( + channel_id=channel_id, + descriptor=descriptor, + escrow=request.escrow, + chain_id=request.chain_id, + deposit=deposit, + cumulative=cumulative, + pending_transaction=transaction, + ) + await self._save(request, channel) + return self._open_payload(channel) + + async def _top_up( + self, request: _Request, channel: _Channel, additional: int + ) -> dict[str, Any]: + if channel.pending_top_up is not None: + return self._top_up_payload(channel) + transaction, _ = await self._transaction( + request, + _call( + "topUp((address,address,address,address,bytes32,address,bytes32),uint96)", + ["(address,address,address,address,bytes32,address,bytes32)", "uint96"], + [_descriptor_tuple(channel.descriptor), additional], + ), + nonce_key=_session_nonce_key(channel.descriptor.salt), + distinguish=True, + ) + channel.pending_transaction = transaction + channel.pending_top_up = additional + await self._save(request, channel) + return self._top_up_payload(channel) + + def _open_payload(self, channel: _Channel) -> dict[str, Any]: + transaction = channel.pending_transaction + if transaction is None: + raise ValueError("pending Tempo session open has no transaction") + return { + "action": "open", + "type": "transaction", + "channelId": channel.channel_id, + "transaction": transaction, + "descriptor": channel.descriptor.wire(), + "cumulativeAmount": str(channel.cumulative), + "signature": _voucher_signature( + self.account, + channel.channel_id, + channel.cumulative, + channel.escrow, + channel.chain_id, + ), + "authorizedSigner": channel.descriptor.authorized_signer, + } + + def _top_up_payload(self, channel: _Channel) -> dict[str, Any]: + transaction = channel.pending_transaction + additional = channel.pending_top_up + if transaction is None or additional is None: + raise ValueError("pending Tempo session top-up is incomplete") + return { + "action": "topUp", + "type": "transaction", + "channelId": channel.channel_id, + "transaction": transaction, + "descriptor": channel.descriptor.wire(), + "additionalDeposit": str(additional), + } + + async def _voucher( + self, request: _Request, channel: _Channel, cumulative: int + ) -> dict[str, Any]: + channel.cumulative = cumulative + await self._save(request, channel) + return { + "action": "voucher", + "channelId": channel.channel_id, + "descriptor": channel.descriptor.wire(), + "cumulativeAmount": str(cumulative), + "signature": _voucher_signature( + self.account, + channel.channel_id, + cumulative, + channel.escrow, + channel.chain_id, + ), + } + + async def _transaction( + self, + request: _Request, + data: bytes, + *, + nonce_key: int, + distinguish: bool = False, + ) -> tuple[str, Any]: + from pytempo import Call, TempoTransaction + + chain_id, _, gas_price = await get_tx_params(self.rpc_url, self.account.address) + if chain_id != request.chain_id: + raise TransactionError( + f"Chain ID mismatch: RPC returned {chain_id}, expected {request.chain_id}" + ) + nonce = 0 if request.fee_payer else await self._lane_nonce(nonce_key) + priority = gas_price + if request.fee_payer: + priority = min(gas_price, get_policy(chain_id).max_priority_fee_per_gas) + tx = TempoTransaction.create( + chain_id=chain_id, + gas_limit=SESSION_GAS_LIMIT, + max_fee_per_gas=gas_price, + max_priority_fee_per_gas=priority, + nonce=nonce, + nonce_key=EXPIRING_NONCE_KEY if request.fee_payer else nonce_key, + fee_token=None if request.fee_payer else request.token, + awaiting_fee_payer=request.fee_payer, + valid_before=( + int(time.time()) + FEE_PAYER_VALID_BEFORE_SECS if request.fee_payer else None + ), + valid_after=(_random_valid_after() if request.fee_payer and distinguish else None), + calls=(Call.create(to=request.escrow, value=0, data=data),), + ) + signed = tx.sign(self.account.private_key) + if request.fee_payer: + from mpp.methods.tempo.fee_payer_envelope import encode_fee_payer_envelope + + return "0x" + encode_fee_payer_envelope(signed).hex(), tx + return "0x" + signed.encode().hex(), tx + + async def _replace_expired_pending( + self, request: _Request, channel: _Channel + ) -> dict[str, Any] | None: + transaction = channel.pending_transaction + if transaction is None: + return None + valid_before = _transaction_valid_before(transaction) + if valid_before is None or valid_before > int(time.time()): + return None + chain_timestamp, block_number = await self._expiry_block() + if chain_timestamp < valid_before: + return None + + settled, deposit, closing = await self._channel_state(channel, block_number) + if ( + closing != 0 + or settled > channel.cumulative + or (deposit < channel.deposit and (channel.status == "open" or deposit != 0)) + ): + raise ValueError("expired Tempo session transaction cannot be safely reconciled") + _limit(deposit, self.max_deposit, "channel deposit") + + if channel.status == "pending": + if deposit == 0: + return await self._open(request, channel.deposit, channel.cumulative) + channel.status = "open" + channel.deposit = deposit + channel.pending_transaction = None + await self._save(request, channel) + return None + + additional = channel.pending_top_up + if additional is None: + raise ValueError("expired Tempo session top-up is incomplete") + target = channel.deposit + additional + channel.deposit = deposit + channel.pending_transaction = None + channel.pending_top_up = None + if deposit < target: + return await self._top_up(request, channel, target - deposit) + await self._save(request, channel) + return None + + async def _expiry_block(self) -> tuple[int, dict[str, object]]: + block = await _rpc_call(self.rpc_url, "eth_getBlockByNumber", ["finalized", False]) + if not isinstance(block, dict): + raise ValueError("invalid finalized block RPC response") + timestamp = _rpc_quantity(block.get("timestamp"), "finalized block timestamp") + block_hash = _bytes32(block.get("hash"), "finalized block hash") + return timestamp, {"blockHash": block_hash, "requireCanonical": True} + + async def _lane_nonce(self, nonce_key: int) -> int: + data = _call( + "getNonce(address,uint256)", + ["address", "uint256"], + [self.account.address, nonce_key], + ) + result = await _rpc_call( + self.rpc_url, + "eth_call", + [{"to": NONCE_PRECOMPILE, "data": "0x" + data.hex()}, "pending"], + ) + try: + nonce = int(result, 16) + except (TypeError, ValueError) as error: + raise TransactionError("Tempo nonce RPC returned invalid data") from error + if nonce > MAX_UINT64: + raise TransactionError("Tempo nonce RPC returned a value outside uint64 bounds") + return nonce + + async def _recover(self, request: _Request) -> _Channel: + snapshot = request.snapshot + assert snapshot is not None + accepted = _amount(snapshot.get("acceptedCumulative"), "acceptedCumulative") + channel = _Channel( + channel_id=_bytes32(snapshot.get("channelId"), "sessionSnapshot.channelId"), + descriptor=_Descriptor.parse(snapshot.get("descriptor")), + escrow=_address(snapshot.get("escrow"), "sessionSnapshot.escrow"), + chain_id=_chain_id(snapshot.get("chainId"), "sessionSnapshot.chainId"), + deposit=_amount(snapshot.get("deposit"), "sessionSnapshot.deposit"), + cumulative=accepted, + accepted=accepted, + status="open", + ) + self._snapshot(request, channel) + self._validate(channel, request) + settled, deposit, closing = await self._channel_state(channel) + if deposit == 0 or closing != 0 or settled > deposit: + raise ValueError("session channel is not reusable on chain") + _limit(deposit, self.max_deposit, "channel deposit") + if channel.cumulative > deposit: + raise ValueError("session snapshot exceeds on-chain channel deposit") + channel.deposit = deposit + channel.cumulative = max(channel.cumulative, settled) + await self._save(request, channel) + return channel + + async def _channel_state( + self, channel: _Channel, block: str | dict[str, object] = "latest" + ) -> tuple[int, int, int]: + from eth_abi.abi import decode + + data = _call( + "getChannelState(bytes32)", ["bytes32"], [bytes.fromhex(channel.channel_id[2:])] + ) + result = await _rpc_call( + self.rpc_url, + "eth_call", + [{"to": channel.escrow, "data": "0x" + data.hex()}, block], + ) + if not isinstance(result, str) or not result.startswith("0x"): + raise ValueError("invalid getChannelState RPC response") + try: + values = decode(["uint96", "uint96", "uint32"], bytes.fromhex(result[2:])) + except Exception as error: + raise ValueError("invalid getChannelState RPC response") from error + return cast("tuple[int, int, int]", tuple(map(int, values))) + + async def _reconcile_deposit(self, channel: _Channel, advertised: int) -> None: + if advertised == channel.deposit: + return + previous = channel.deposit + settled, deposit, closing = await self._channel_state(channel) + if closing != 0 or settled > channel.cumulative or deposit < channel.deposit: + raise ValueError("Tempo session deposit cannot be safely reconciled") + _limit(deposit, self.max_deposit, "channel deposit") + channel.deposit = deposit + if channel.pending_top_up is not None and deposit >= previous + channel.pending_top_up: + channel.pending_transaction = None + channel.pending_top_up = None + + async def _need_voucher_async( + self, exchange: AsyncHttpResponseContext, request: _Request, event: dict[str, Any] + ) -> None: + channel, target = await exchange.run_async(self._prepare_voucher(request, event)) + if target > channel.deposit: + additional = target - channel.deposit + credential = await exchange.create_credential( + { + "action": "topUp", + "channelId": channel.channel_id, + "additionalDeposit": str(additional), + } + ) + await self._post_async(exchange, request, credential) + credential = await exchange.create_credential( + { + "action": "voucher", + "channelId": channel.channel_id, + "cumulativeAmount": str(target), + } + ) + await self._post_async(exchange, request, credential) + + def _need_voucher_sync( + self, exchange: SyncHttpResponseContext, request: _Request, event: dict[str, Any] + ) -> None: + channel, target = exchange.run_sync(self._prepare_voucher(request, event)) + if target > channel.deposit: + additional = target - channel.deposit + credential = exchange.create_credential( + { + "action": "topUp", + "channelId": channel.channel_id, + "additionalDeposit": str(additional), + } + ) + self._post_sync(exchange, request, credential) + credential = exchange.create_credential( + { + "action": "voucher", + "channelId": channel.channel_id, + "cumulativeAmount": str(target), + } + ) + self._post_sync(exchange, request, credential) + + async def _prepare_voucher( + self, request: _Request, event: dict[str, Any] + ) -> tuple[_Channel, int]: + channel = await self._require(request) + if channel.pending_top_up is not None: + await self._replace_expired_pending(request, channel) + await self._reconcile_deposit(channel, _amount(event.get("deposit"), "deposit")) + self._validate(channel, request) + await self._save(request, channel) + return channel, self._voucher_target(request, channel, event) + + def _voucher_target(self, request: _Request, channel: _Channel, event: dict[str, Any]) -> int: + if event.get("channelId") != channel.channel_id: + raise ValueError("need-voucher event references the wrong channel") + required = _amount(event.get("requiredCumulative"), "requiredCumulative") + accepted = _amount(event.get("acceptedCumulative"), "acceptedCumulative") + if accepted > channel.cumulative or accepted > required: + raise ValueError("need-voucher event exceeds locally signed state") + accepted = max(accepted, channel.accepted) + target = max(required, channel.cumulative, accepted + request.min_voucher_delta) + _limit(target, self.max_deposit, "voucher") + return target + + async def _post_async( + self, exchange: AsyncHttpResponseContext, request: _Request, credential: Credential + ) -> None: + response = await exchange.send( + httpx.Request( + "POST", + exchange.request.url, + headers={"Authorization": credential.to_authorization()}, + ) + ) + try: + await response.aread() + if not response.is_success: + raise TransactionError( + f"Tempo session management POST failed with status {response.status_code}" + ) + await exchange.run_async( + self._accept_response(request, response, exchange.challenge, credential.payload) + ) + finally: + await response.aclose() + + def _post_sync( + self, exchange: SyncHttpResponseContext, request: _Request, credential: Credential + ) -> None: + response = exchange.send( + httpx.Request( + "POST", + exchange.request.url, + headers={"Authorization": credential.to_authorization()}, + ) + ) + try: + response.read() + if not response.is_success: + raise TransactionError( + f"Tempo session management POST failed with status {response.status_code}" + ) + exchange.run_sync( + self._accept_response(request, response, exchange.challenge, credential.payload) + ) + finally: + response.close() + + async def _accept_response( + self, + request: _Request, + response: httpx.Response, + challenge: Challenge, + payload: dict[str, Any], + ) -> None: + channel = await self._require(request) + self._apply_receipt_header(response, challenge, channel) + self._apply_accepted_response(channel, payload) + await self._save(request, channel) + + async def _record_receipt(self, request: _Request, value: object, challenge: Challenge) -> None: + channel = await self._require(request) + self._apply_receipt(value, challenge, channel) + await self._save(request, channel) + + def _apply_accepted_response(self, channel: _Channel, payload: dict[str, Any]) -> None: + action = payload.get("action") + if action == "topUp": + additional = _amount(payload.get("additionalDeposit"), "additionalDeposit") + if channel.pending_top_up == additional and channel.pending_transaction == payload.get( + "transaction" + ): + channel.deposit += additional + channel.pending_transaction = None + channel.pending_top_up = None + elif action in {"open", "voucher"}: + channel.status = "open" + channel.accepted = max( + channel.accepted, _amount(payload.get("cumulativeAmount"), "cumulativeAmount") + ) + if action == "open" and channel.pending_transaction == payload.get("transaction"): + channel.pending_transaction = None + + def _apply_receipt_header( + self, response: httpx.Response, challenge: Challenge, channel: _Channel + ) -> None: + header = response.headers.get("payment-receipt") + if header is None: + return + receipt = Receipt.from_payment_receipt(header) + value = { + "method": receipt.method, + "status": receipt.status, + "timestamp": receipt.timestamp.isoformat(), + "reference": receipt.reference, + **(receipt.extensions or {}), + } + self._apply_receipt(value, challenge, channel) + + def _apply_receipt(self, value: object, challenge: Challenge, channel: _Channel) -> None: + self._validate_receipt(value, challenge, channel) + assert isinstance(value, dict) + channel.accepted = max( + channel.accepted, _amount(value["acceptedCumulative"], "acceptedCumulative") + ) + channel.status = "open" + + def _validate_receipt(self, value: object, challenge: Challenge, channel: _Channel) -> None: + if not isinstance(value, dict): + raise ValueError("invalid Tempo session receipt") + if ( + value.get("method") != "tempo" + or value.get("intent") != "session" + or value.get("status") != "success" + ): + raise ValueError("invalid Tempo session receipt") + if ( + _bytes32(value.get("channelId"), "receipt.channelId") != channel.channel_id + or value.get("reference") != channel.channel_id + or value.get("challengeId") != challenge.id + ): + raise ValueError("Tempo session receipt references the wrong payment") + timestamp = value.get("timestamp") + if not isinstance(timestamp, str): + raise ValueError("invalid Tempo session receipt timestamp") + try: + datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + except ValueError as error: + raise ValueError("invalid Tempo session receipt timestamp") from error + accepted = _amount(value.get("acceptedCumulative"), "acceptedCumulative") + spent = _amount(value.get("spent"), "spent") + if spent > accepted or accepted > channel.cumulative: + raise ValueError("Tempo session receipt exceeds locally authorized spend") + units = value.get("units") + if units is not None and ( + not isinstance(units, (int, float)) + or isinstance(units, bool) + or not math.isfinite(units) + or units < 0 + ): + raise ValueError("invalid Tempo session receipt units") + if value.get("txHash") is not None: + _bytes32(value["txHash"], "receipt.txHash") + + async def _load(self, request: _Request) -> _Channel | None: + value = await self.channel_store.get("tempo:session:" + request.scope) + if value is None: + return None + channel = _Channel.load(value) + self._validate(channel, request) + return channel + + async def _require(self, request: _Request) -> _Channel: + channel = await self._load(request) + if channel is None: + raise ValueError("no local Tempo session channel available") + return channel + + async def _save(self, request: _Request, channel: _Channel) -> None: + await self.channel_store.put("tempo:session:" + request.scope, channel.dump()) + + def _validate(self, channel: _Channel, request: _Request) -> None: + descriptor = channel.descriptor + if ( + channel.chain_id != request.chain_id + or channel.escrow != request.escrow + or descriptor.payer != self.account.address.lower() + or descriptor.authorized_signer != self.account.address.lower() + or descriptor.payee != request.payee + or descriptor.operator != request.operator + or descriptor.token != request.token + ): + raise ValueError("stored Tempo session channel is outside this payment scope") + if channel.channel_id != _channel_id(descriptor, channel.escrow, channel.chain_id): + raise ValueError("stored Tempo session channelId does not match its descriptor") + _limit(channel.deposit, self.max_deposit, "stored channel deposit") + if channel.deposit == 0: + raise ValueError("stored Tempo session deposit must be greater than zero") + if not 0 <= channel.accepted <= channel.cumulative <= channel.deposit: + raise ValueError("stored Tempo session amounts are inconsistent") + if channel.status == "pending": + if channel.pending_transaction is None or channel.pending_top_up is not None: + raise ValueError("stored pending Tempo session open is incomplete") + elif channel.pending_top_up is None: + if channel.pending_transaction is not None: + raise ValueError("stored Tempo session pending state is inconsistent") + elif ( + channel.pending_transaction is None + or channel.pending_top_up == 0 + or channel.deposit + channel.pending_top_up > self.max_deposit + ): + raise ValueError("stored pending Tempo session top-up is inconsistent") + + def _snapshot(self, request: _Request, channel: _Channel) -> dict[str, Any]: + snapshot = request.snapshot + assert snapshot is not None + if ( + _bytes32(snapshot.get("channelId"), "sessionSnapshot.channelId") != channel.channel_id + or _address(snapshot.get("escrow"), "sessionSnapshot.escrow") != request.escrow + or _chain_id(snapshot.get("chainId"), "sessionSnapshot.chainId") != request.chain_id + or _Descriptor.parse(snapshot.get("descriptor")) != channel.descriptor + ): + raise ValueError("session snapshot does not match the active channel") + settled = _amount(snapshot.get("settled"), "settled") + spent = _amount(snapshot.get("spent"), "spent") + accepted = _amount(snapshot.get("acceptedCumulative"), "acceptedCumulative") + required = _amount(snapshot.get("requiredCumulative"), "requiredCumulative") + _amount(snapshot.get("deposit"), "deposit") + closing = snapshot.get("closeRequestedAt") + if closing is not None and _amount(closing, "closeRequestedAt") != 0: + raise ValueError("session snapshot channel is closing") + if settled > accepted or spent > accepted or accepted > required: + raise ValueError("session snapshot amounts are inconsistent") + return snapshot + + def _resolve(self, challenge: Challenge) -> _Request: + if challenge.method != "tempo" or challenge.intent != "session": + raise ValueError("TempoSessionMethod only handles tempo/session challenges") + raw = challenge.request + amount = _amount(raw.get("amount"), "amount") + token = _address(raw.get("currency"), "currency") + payee = _address(raw.get("recipient"), "recipient") + details = raw.get("methodDetails") + if not isinstance(details, dict) or details.get("sessionProtocol") != "v2": + raise ValueError("TempoSessionMethod requires methodDetails.sessionProtocol v2") + chain_id = _chain_id(details.get("chainId"), "methodDetails.chainId") + if chain_id != self.chain_id: + raise ValueError( + f"Challenge requests chain ID {chain_id}, " + f"but client is restricted to {self.chain_id}" + ) + escrow = _address(details.get("escrowContract"), "methodDetails.escrowContract") + if escrow != self.escrow: + raise ValueError("session challenge escrow is outside local policy") + operator = _address(details.get("operator", ZERO_ADDRESS), "methodDetails.operator") + fee_payer = details.get("feePayer", False) + if not isinstance(fee_payer, bool): + raise ValueError("methodDetails.feePayer must be a boolean") + suggested = raw.get("suggestedDeposit") + min_delta = details.get("minVoucherDelta") + snapshot = details.get("sessionSnapshot") + if snapshot is not None and not isinstance(snapshot, dict): + raise ValueError("methodDetails.sessionSnapshot must be an object") + payer = self.account.address.lower() + scope = ":".join((str(chain_id), escrow, payer, payee, operator, token, payer)) + return _Request( + amount=amount, + payee=payee, + token=token, + operator=operator, + escrow=escrow, + chain_id=chain_id, + fee_payer=fee_payer, + suggested_deposit=( + _amount(suggested, "suggestedDeposit") if suggested is not None else None + ), + min_voucher_delta=( + _amount(min_delta, "minVoucherDelta") if min_delta is not None else 0 + ), + snapshot=cast("dict[str, Any] | None", snapshot), + scope=scope, + ) + + def _lock(self, scope: str) -> threading.Lock: + with self._locks_guard: + return self._locks.setdefault(scope, threading.Lock()) + + @asynccontextmanager + async def _locked(self, scope: str): + key = (id(self), scope) + if key in _HELD.get(): + yield + return + lock = self._lock(scope) + acquire = asyncio.create_task(asyncio.to_thread(lock.acquire)) + try: + await asyncio.shield(acquire) + except BaseException: + await asyncio.shield(acquire) + lock.release() + raise + token = _HELD.set(_HELD.get() | {key}) + try: + yield + finally: + _HELD.reset(token) + lock.release() + + @contextmanager + def _locked_sync(self, scope: str): + key = (id(self), scope) + if key in _HELD.get(): + yield + return + lock = self._lock(scope) + lock.acquire() + token = _HELD.set(_HELD.get() | {key}) + try: + yield + finally: + _HELD.reset(token) + lock.release() + + +def tempo_session( + *, + account: TempoAccount, + max_deposit: int, + rpc_url: str = RPC_URL, + chain_id: int = CHAIN_ID, + escrow: str = TIP20_CHANNEL_ESCROW, + channel_store: Store | None = None, +) -> TempoSessionMethod: + """Create a private-key TIP-1034 session method. + + ``max_deposit`` is a required cap in raw token base units. The generic + ``mpp.Store`` defaults to process-local memory; multi-process callers need + external locking or atomic channel updates. + """ + if not isinstance(max_deposit, int) or isinstance(max_deposit, bool): + raise TypeError("max_deposit must be an integer in raw token base units") + _limit(max_deposit, MAX_UINT96, "max_deposit") + if max_deposit == 0: + raise ValueError("max_deposit must be greater than zero") + return TempoSessionMethod( + account=account, + max_deposit=max_deposit, + rpc_url=rpc_url, + chain_id=_chain_id(chain_id, "chain_id"), + escrow=_address(escrow, "escrow"), + channel_store=channel_store if channel_store is not None else MemoryStore(), + ) + + +def _descriptor_tuple(descriptor: _Descriptor) -> tuple[object, ...]: + return ( + descriptor.payer, + descriptor.payee, + descriptor.operator, + descriptor.token, + bytes.fromhex(descriptor.salt[2:]), + descriptor.authorized_signer, + bytes.fromhex(descriptor.expiring_nonce_hash[2:]), + ) + + +def _session_nonce_key(salt: str) -> int: + """Derive one reusable, non-reserved 2D nonce lane per channel.""" + value = bytes.fromhex(_bytes32(salt, "descriptor.salt")[2:]) + return int.from_bytes(b"\x01" + value[1:]) + + +def _random_valid_after() -> int: + """Distinguish otherwise-identical sponsored transactions with a past timestamp.""" + latest = int(time.time()) - 60 + return 1 + int.from_bytes(os.urandom(8)) % latest if latest > 0 else 0 + + +def _channel_id(descriptor: _Descriptor, escrow: str, chain_id: int) -> str: + from eth_abi.abi import encode + + encoded = encode( + [ + "address", + "address", + "address", + "address", + "bytes32", + "address", + "bytes32", + "address", + "uint256", + ], + [*_descriptor_tuple(descriptor), escrow, chain_id], + ) + return "0x" + _keccak(encoded).hex() + + +def _voucher_signature( + account: TempoAccount, channel_id: str, cumulative: int, escrow: str, chain_id: int +) -> str: + from eth_abi.abi import encode + + _uint96(cumulative, "cumulativeAmount") + domain = _keccak( + encode( + ["bytes32", "bytes32", "bytes32", "uint256", "address"], + [ + _keccak( + b"EIP712Domain(string name,string version,uint256 chainId," + b"address verifyingContract)" + ), + _keccak(b"TIP20 Channel Reserve"), + _keccak(b"1"), + chain_id, + escrow, + ], + ) + ) + voucher = _keccak( + encode( + ["bytes32", "bytes32", "uint96"], + [ + _keccak(b"Voucher(bytes32 channelId,uint96 cumulativeAmount)"), + bytes.fromhex(channel_id[2:]), + cumulative, + ], + ) + ) + return "0x" + account.sign_hash(_keccak(b"\x19\x01" + domain + voucher)).hex() + + +def _call(signature: str, types: Sequence[str], values: Sequence[object]) -> bytes: + from eth_abi.abi import encode + + return _keccak(signature.encode())[:4] + encode(types, values) + + +def _keccak(value: bytes) -> bytes: + from eth_hash.auto import keccak + + return cast("bytes", keccak(value)) + + +def _address(value: object, name: str) -> str: + if not isinstance(value, str) or _ADDRESS_RE.fullmatch(value) is None: + raise ValueError(f"{name} must be a 20-byte hex address") + return value.lower() + + +def _bytes32(value: object, name: str) -> str: + if not isinstance(value, str) or _BYTES32_RE.fullmatch(value) is None: + raise ValueError(f"{name} must be a 32-byte hex value") + return value.lower() + + +def _transaction_hex(value: object) -> str | None: + if value is None: + return None + if not isinstance(value, str) or not value.startswith("0x") or len(value) == 2: + raise ValueError("pending_transaction must be hex encoded") + try: + bytes.fromhex(value[2:]) + except ValueError as error: + raise ValueError("pending_transaction must be hex encoded") from error + return value.lower() + + +def _transaction_valid_before(transaction: str) -> int | None: + import rlp + + raw = bytes.fromhex(transaction[2:]) + try: + fields = rlp.decode(raw[1:]) + valid_before = fields[8] + except (IndexError, rlp.DecodingError) as error: + raise ValueError("invalid stored Tempo session transaction") from error + if not isinstance(fields, list) or raw[0] not in {0x76, 0x78}: + raise ValueError("invalid stored Tempo session transaction") + if not isinstance(valid_before, bytes): + raise ValueError("invalid stored Tempo session transaction") + return int.from_bytes(valid_before) if valid_before else None + + +def _amount(value: object, name: str) -> int: + if not isinstance(value, str) or _AMOUNT_RE.fullmatch(value) is None: + raise ValueError(f"{name} must be a canonical decimal string") + return _uint96(int(value), name) + + +def _uint96(value: int, name: str) -> int: + if not 0 <= value <= MAX_UINT96: + raise ValueError(f"{name} is outside uint96 bounds") + return value + + +def _chain_id(value: object, name: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise ValueError(f"{name} must be a positive integer") + return value + + +def _rpc_quantity(value: object, name: str) -> int: + if ( + not isinstance(value, str) + or re.fullmatch(r"0x(?:0|[1-9a-fA-F][0-9a-fA-F]*)", value) is None + ): + raise ValueError(f"invalid {name} RPC response") + return int(value, 16) + + +def _limit(value: int, limit: int, name: str) -> None: + _uint96(value, name) + if value > limit: + raise ValueError(f"{name} exceeds max_deposit") + + +def _is_sse(response: httpx.Response) -> bool: + return response.headers.get("content-type", "").lower().startswith("text/event-stream") diff --git a/tests/test_tempo_session.py b/tests/test_tempo_session.py new file mode 100644 index 00000000..12ea2683 --- /dev/null +++ b/tests/test_tempo_session.py @@ -0,0 +1,800 @@ +"""Focused client tests for Tempo TIP-1034 sessions.""" + +from __future__ import annotations + +import json +from typing import Any, cast + +import pytest +import rlp +from eth_abi.abi import decode, encode +from eth_hash.auto import keccak + +from mpp import Challenge, MemoryStore +from mpp.methods.tempo import ( + TIP20_CHANNEL_ESCROW, + TempoAccount, + TempoSessionMethod, + tempo_session, +) +from mpp.methods.tempo import session as session_module + +PRIVATE_KEY = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" +CHAIN_ID = 42431 +RPC_URL = "https://rpc.test" +TOKEN = "0x20c0000000000000000000000000000000000001" +PAYEE = "0x0000000000000000000000000000000000000002" +ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" +OPERATOR = "0x0000000000000000000000000000000000000004" +SALT = bytes.fromhex("11" * 32) +CHANNEL_ID = "0xf8e4ab2eca9ec42f2cb0478ba074a76fa803607cacae5e557171f6679924fcf9" +FINALIZED_HASH = "0x" + "ab" * 32 +PREIMAGE = ( + "76f9012582a5bf0202831e8480f8def8dc944d5050000000000000000000000000000000000080b8" + "c4edc53b000000000000000000000000000000000000000000000000000000000000000002000000" + "00000000000000000000000000000000000000000000000000000000000000000000000000000000" + "0020c000000000000000000000000000000000000100000000000000000000000000000000000000" + "00000000000000000000000032111111111111111111111111111111111111111111111111111111" + "1111111111000000000000000000000000fcad0b19bb29d4674531d6f115237e16afce377cc0a001" + + "11" * 31 + + "8080809420c000000000000000000000000000000000000180c0" +) + + +@pytest.fixture +def account() -> TempoAccount: + return TempoAccount.from_key(PRIVATE_KEY) + + +def challenge( + *, + amount: str = "10", + suggested_deposit: str | None = "50", + min_voucher_delta: str | None = "25", + snapshot: dict[str, Any] | None = None, + method: str = "tempo", + intent: str = "session", + protocol: str = "v2", + chain_id: int = CHAIN_ID, + escrow: str = TIP20_CHANNEL_ESCROW, + operator: str | None = None, + fee_payer: bool = False, +) -> Challenge: + details: dict[str, Any] = { + "sessionProtocol": protocol, + "chainId": chain_id, + "escrowContract": escrow, + } + if min_voucher_delta is not None: + details["minVoucherDelta"] = min_voucher_delta + if snapshot is not None: + details["sessionSnapshot"] = snapshot + if operator is not None: + details["operator"] = operator + if fee_payer: + details["feePayer"] = True + request: dict[str, Any] = { + "amount": amount, + "currency": TOKEN, + "recipient": PAYEE, + "methodDetails": details, + } + if suggested_deposit is not None: + request["suggestedDeposit"] = suggested_deposit + return Challenge( + id="challenge-1", + realm="example.com", + method=method, + intent=intent, + request=request, + ) + + +def method( + account: TempoAccount, + *, + max_deposit: int = 50, + store: MemoryStore | None = None, +) -> TempoSessionMethod: + return tempo_session( + account=account, + max_deposit=max_deposit, + rpc_url=RPC_URL, + chain_id=CHAIN_ID, + channel_store=store, + ) + + +def deterministic_transactions( + monkeypatch: pytest.MonkeyPatch, +) -> list[tuple[str, str]]: + calls: list[tuple[str, str]] = [] + + async def get_tx_params(rpc_url: str, address: str) -> tuple[int, int, int]: + calls.append((rpc_url, address)) + return CHAIN_ID, 7, 2 + + async def lane_nonce(self: TempoSessionMethod, nonce_key: int) -> int: + return 0 + + def urandom(length: int) -> bytes: + assert length == 32 + return SALT + + monkeypatch.setattr(session_module, "get_tx_params", get_tx_params) + monkeypatch.setattr(TempoSessionMethod, "_lane_nonce", lane_nonce) + monkeypatch.setattr(session_module.os, "urandom", urandom) + return calls + + +def deterministic_sponsored_transactions( + monkeypatch: pytest.MonkeyPatch, now: list[int] +) -> list[tuple[str, str]]: + calls = deterministic_transactions(monkeypatch) + random_value = 0 + + def urandom(length: int) -> bytes: + nonlocal random_value + if length == 32: + return SALT + assert length == 8 + random_value += 1 + return random_value.to_bytes(8) + + monkeypatch.setattr(session_module.os, "urandom", urandom) + monkeypatch.setattr(session_module.time, "time", lambda: now[0]) + return calls + + +def transaction_fields(payload: dict[str, Any]) -> list[Any]: + raw = bytes.fromhex(cast("str", payload["transaction"])[2:]) + return cast("list[Any]", rlp.decode(raw[1:])) + + +def test_public_factory(account: TempoAccount) -> None: + store = MemoryStore() + result = method(account, store=store) + + assert isinstance(result, TempoSessionMethod) + assert result.intents.keys() == {"session"} + assert result.channel_store is store + assert result.max_deposit == 50 + with pytest.raises(TypeError, match="max_deposit must be an integer"): + tempo_session(account=account, max_deposit=True) # type: ignore[arg-type] + with pytest.raises(ValueError, match="greater than zero"): + tempo_session(account=account, max_deposit=0) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("invalid", "message"), + [ + (challenge(method="stripe"), "only handles tempo/session"), + (challenge(intent="charge"), "only handles tempo/session"), + (challenge(protocol="v1"), "sessionProtocol v2"), + (challenge(chain_id=1), "restricted to 42431"), + ( + challenge(escrow="0x0000000000000000000000000000000000000003"), + "escrow is outside local policy", + ), + ], +) +async def test_challenge_policy(account: TempoAccount, invalid: Challenge, message: str) -> None: + with pytest.raises(ValueError, match=message): + await method(account).create_credential(invalid) + + +@pytest.mark.asyncio +async def test_open_accepts_nonzero_operator( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + deterministic_transactions(monkeypatch) + + payload = (await method(account).create_credential(challenge(operator=OPERATOR))).payload + call = cast("list[bytes]", transaction_fields(payload)[4][0]) + decoded = decode(["address", "address", "address", "uint96", "bytes32", "address"], call[2][4:]) + + assert cast("dict[str, str]", payload["descriptor"])["operator"] == OPERATOR + assert decoded[1] == OPERATOR + + +@pytest.mark.asyncio +async def test_open_rejects_zero_deposit(account: TempoAccount) -> None: + with pytest.raises(ValueError, match="deposit must be greater than zero"): + await method(account).create_credential( + challenge(amount="0", suggested_deposit=None, min_voucher_delta="0") + ) + + +@pytest.mark.asyncio +async def test_stored_channel_rejects_zero_deposit( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + deterministic_transactions(monkeypatch) + store = MemoryStore() + payment_method = method(account, store=store) + await payment_method.create_credential(challenge()) + key, raw = next(iter(store._data.items())) + value = json.loads(cast("str", raw)) + value.update(deposit="0", cumulative="0", accepted="0") + await store.put(key, json.dumps(value)) + + with pytest.raises(ValueError, match="stored Tempo session deposit must be greater than zero"): + await payment_method.create_credential(challenge()) + + +@pytest.mark.asyncio +async def test_open_payload_is_deterministic( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + rpc_calls = deterministic_transactions(monkeypatch) + + credential = await method(account).create_credential(challenge()) + payload = credential.payload + descriptor = cast("dict[str, str]", payload["descriptor"]) + + assert set(payload) == { + "action", + "type", + "channelId", + "transaction", + "descriptor", + "cumulativeAmount", + "signature", + "authorizedSigner", + } + assert payload["action"] == "open" + assert payload["type"] == "transaction" + assert payload["channelId"] == CHANNEL_ID + assert payload["cumulativeAmount"] == "10" + assert payload["signature"] == ( + "0xf6bd824d8f1a27b12d052aa2ce30564a35f2d53b506a9cd2260fdcf01f326494" + "59f288db3c39792f338bccea830456b325f6587f5aaf0a79a29a2bb0c879a6231c" + ) + assert descriptor == { + "payer": account.address.lower(), + "payee": PAYEE, + "operator": ZERO_ADDRESS, + "token": TOKEN, + "salt": "0x" + SALT.hex(), + "authorizedSigner": account.address.lower(), + "expiringNonceHash": ("0xd6d945c2cf976d050b069e8281c9272222ad96d23b82f4d87dc40266fe800315"), + } + assert payload["authorizedSigner"] == account.address.lower() + assert credential.source == f"did:pkh:eip155:{CHAIN_ID}:{account.address.lower()}" + + raw = bytes.fromhex(cast("str", payload["transaction"])[2:]) + fields = transaction_fields(payload) + call = cast("list[bytes]", fields[4][0]) + assert raw[0] == 0x76 + assert keccak(raw).hex() == "6fa21191db73f670fa853d94ea4e26297575668dae41b17ecdb15482b3684686" + assert [int.from_bytes(cast("bytes", value)) for value in fields[:4]] == [ + CHAIN_ID, + 2, + 2, + 2_000_000, + ] + assert int.from_bytes(cast("bytes", fields[6])) == int.from_bytes(b"\x01" + SALT[1:]) + assert fields[7] == b"" + assert call[0] == bytes.fromhex(TIP20_CHANNEL_ESCROW[2:]) + assert call[1] == b"" + assert call[2][:4].hex() == "edc53b00" + decoded = decode( + ["address", "address", "address", "uint96", "bytes32", "address"], + call[2][4:], + ) + assert decoded == (PAYEE, ZERO_ADDRESS, TOKEN, 50, SALT, account.address.lower()) + preimage = b"\x76" + rlp.encode(fields[:13]) + assert preimage.hex() == PREIMAGE + assert ( + "0x" + keccak(preimage + bytes.fromhex(account.address[2:])).hex() + == descriptor["expiringNonceHash"] + ) + assert rpc_calls == [(RPC_URL, account.address)] + + +@pytest.mark.asyncio +async def test_persisted_voucher_honors_delta_and_cap( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + deterministic_transactions(monkeypatch) + store = MemoryStore() + opened = await method(account, store=store).create_credential(challenge()) + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "50", + "acceptedCumulative": "10", + "spent": "10", + "settled": "0", + "requiredCumulative": "26", + } + restarted = method(account, store=store) + + voucher = await restarted.create_credential(challenge(amount="5", snapshot=snapshot)) + + assert voucher.payload["action"] == "voucher" + assert voucher.payload["channelId"] == opened.payload["channelId"] + assert voucher.payload["cumulativeAmount"] == "35" + snapshot["acceptedCumulative"] = "35" + snapshot["requiredCumulative"] = "36" + with pytest.raises(ValueError, match="voucher exceeds max_deposit"): + await restarted.create_credential( + challenge( + amount="1", + suggested_deposit=None, + min_voucher_delta="25", + snapshot=snapshot, + ) + ) + + +@pytest.mark.asyncio +async def test_snapshot_recovery_reads_channel_state( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + deterministic_transactions(monkeypatch) + opened = await method(account, max_deposit=100).create_credential( + challenge(suggested_deposit="100", min_voucher_delta="0") + ) + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "100", + "acceptedCumulative": "10", + "spent": "8", + "settled": "5", + "requiredCumulative": "20", + } + calls: list[tuple[str, str, list[object]]] = [] + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> str: + calls.append((rpc_url, rpc_method, params)) + return "0x" + encode(["uint96", "uint96", "uint32"], [5, 100, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + + voucher = await method(account, max_deposit=100, store=MemoryStore()).create_credential( + challenge( + amount="5", + suggested_deposit=None, + min_voucher_delta="0", + snapshot=snapshot, + ) + ) + + assert voucher.payload["action"] == "voucher" + assert voucher.payload["cumulativeAmount"] == "20" + assert calls == [ + ( + RPC_URL, + "eth_call", + [ + { + "to": TIP20_CHANNEL_ESCROW, + "data": "0xd18da8b1" + cast("str", opened.payload["channelId"])[2:], + }, + "latest", + ], + ) + ] + + async def oversized_deposit(rpc_url: str, rpc_method: str, params: list[object]) -> str: + return "0x" + encode(["uint96", "uint96", "uint32"], [5, 101, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", oversized_deposit) + with pytest.raises(ValueError, match="channel deposit exceeds max_deposit"): + await method(account, max_deposit=100, store=MemoryStore()).create_credential( + challenge( + amount="5", + suggested_deposit=None, + min_voucher_delta="0", + snapshot=snapshot, + ) + ) + + +@pytest.mark.asyncio +async def test_rejected_snapshot_recovery_does_not_poison_store( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + tx_calls = deterministic_transactions(monkeypatch) + opened = await method(account, max_deposit=100).create_credential( + challenge(suggested_deposit="100", min_voucher_delta="0") + ) + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "100", + "acceptedCumulative": "10", + "spent": "90", + "settled": "0", + "requiredCumulative": "20", + } + rpc_calls: list[str] = [] + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> str: + rpc_calls.append(rpc_method) + return "0x" + encode(["uint96", "uint96", "uint32"], [0, 100, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + store = MemoryStore() + payment_method = method(account, max_deposit=100, store=store) + + with pytest.raises(ValueError, match="snapshot amounts are inconsistent"): + await payment_method.create_credential( + challenge(amount="5", min_voucher_delta="0", snapshot=snapshot) + ) + + assert store._data == {} + assert rpc_calls == [] + fresh = await payment_method.create_credential( + challenge(amount="5", suggested_deposit="20", min_voucher_delta="0") + ) + assert fresh.payload["action"] == "open" + assert fresh.payload["cumulativeAmount"] == "5" + assert len(tx_calls) == 2 + + +@pytest.mark.asyncio +async def test_pending_open_and_voucher_are_retried_without_advancing( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + rpc_calls = deterministic_transactions(monkeypatch) + store = MemoryStore() + payment_method = method(account, store=store) + + opened = await payment_method.create_credential(challenge(min_voucher_delta="0")) + retried_open = await payment_method.create_credential(challenge(min_voucher_delta="0")) + + assert retried_open.payload == opened.payload + assert len(rpc_calls) == 1 + + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "50", + "acceptedCumulative": "10", + "spent": "10", + "settled": "0", + "requiredCumulative": "15", + } + voucher = await payment_method.create_credential( + challenge(amount="5", min_voucher_delta="0", snapshot=snapshot) + ) + retried_voucher = await payment_method.create_credential( + challenge(amount="5", min_voucher_delta="0") + ) + + assert voucher.payload["action"] == "voucher" + assert voucher.payload["cumulativeAmount"] == "15" + assert retried_voucher.payload == voucher.payload + + +@pytest.mark.asyncio +@pytest.mark.parametrize("onchain_deposit", [0, 50]) +async def test_expired_sponsored_open_reconciles_before_replacement( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch, onchain_deposit: int +) -> None: + now = [1_000] + chain_time = [1_024] + tx_calls = deterministic_sponsored_transactions(monkeypatch, now) + rpc_calls: list[str] = [] + state_blocks: list[object] = [] + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> object: + rpc_calls.append(rpc_method) + if rpc_method == "eth_getBlockByNumber": + assert params == ["finalized", False] + return {"hash": FINALIZED_HASH, "timestamp": hex(chain_time[0])} + state_blocks.append(params[1]) + return "0x" + encode(["uint96", "uint96", "uint32"], [0, onchain_deposit, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + payment_method = method(account) + request = challenge(fee_payer=True, min_voucher_delta="0") + opened = await payment_method.create_credential(request) + + now[0] = 1_024 + assert (await payment_method.create_credential(request)).payload == opened.payload + assert rpc_calls == [] + + now[0] = 1_025 + lagged = await payment_method.create_credential(request) + assert lagged.payload == opened.payload + assert rpc_calls == ["eth_getBlockByNumber"] + + chain_time[0] = 1_025 + replacement = await payment_method.create_credential(request) + + assert replacement.payload["action"] == ("voucher" if onchain_deposit else "open") + if onchain_deposit: + assert "transaction" not in replacement.payload + assert replacement.payload["channelId"] == opened.payload["channelId"] + else: + assert replacement.payload["transaction"] != opened.payload["transaction"] + assert replacement.payload["channelId"] != opened.payload["channelId"] + assert rpc_calls == ["eth_getBlockByNumber", "eth_getBlockByNumber", "eth_call"] + assert state_blocks == [{"blockHash": FINALIZED_HASH, "requireCanonical": True}] + assert len(tx_calls) == (1 if onchain_deposit else 2) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("block", [{"timestamp": "0x1"}, {"hash": "0x12", "timestamp": "0x1"}]) +async def test_expiry_block_requires_valid_hash( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch, block: dict[str, str] +) -> None: + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> object: + assert (rpc_method, params) == ("eth_getBlockByNumber", ["finalized", False]) + return block + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + + with pytest.raises(ValueError, match="finalized block hash"): + await method(account)._expiry_block() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("onchain_deposit", "reconnect_time"), [(20, 1_025), (25, 1_024), (25, 1_025)] +) +async def test_sse_reconnect_reconciles_pending_sponsored_top_up( + account: TempoAccount, + monkeypatch: pytest.MonkeyPatch, + onchain_deposit: int, + reconnect_time: int, +) -> None: + now = [1_000] + tx_calls = deterministic_sponsored_transactions(monkeypatch, now) + rpc_calls: list[str] = [] + state_blocks: list[object] = [] + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> object: + rpc_calls.append(rpc_method) + if rpc_method == "eth_getBlockByNumber": + assert params == ["finalized", False] + return {"hash": FINALIZED_HASH, "timestamp": hex(now[0])} + state_blocks.append(params[1]) + return "0x" + encode(["uint96", "uint96", "uint32"], [0, onchain_deposit, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + payment_method = method(account) + opened = await payment_method.create_credential( + challenge(fee_payer=True, suggested_deposit="20", min_voucher_delta="0") + ) + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "20", + "acceptedCumulative": "10", + "spent": "10", + "settled": "0", + "requiredCumulative": "25", + } + request = challenge( + amount="15", + fee_payer=True, + suggested_deposit=None, + min_voucher_delta="0", + snapshot=snapshot, + ) + top_up = await payment_method.create_credential(request) + + now[0] = 1_024 + retry = await payment_method.create_credential( + challenge(amount="15", fee_payer=True, suggested_deposit=None, min_voucher_delta="0") + ) + assert retry.payload == top_up.payload + assert rpc_calls == [] + + now[0] = reconnect_time + reconnect = challenge( + amount="15", fee_payer=True, suggested_deposit=None, min_voucher_delta="0" + ) + channel, target = await payment_method._prepare_voucher( + payment_method._resolve(reconnect), + { + "channelId": opened.payload["channelId"], + "deposit": str(onchain_deposit), + "requiredCumulative": "25", + "acceptedCumulative": "10", + }, + ) + action = "voucher" if onchain_deposit == 25 else "topUp" + context = {"action": action, "channelId": channel.channel_id} + if action == "voucher": + context["cumulativeAmount"] = str(target) + else: + context["additionalDeposit"] = str(target - channel.deposit) + replacement = await payment_method.create_credential(reconnect, context=context) + + assert replacement.payload["action"] == action + if onchain_deposit == 25: + assert "transaction" not in replacement.payload + else: + assert replacement.payload["additionalDeposit"] == "5" + assert replacement.payload["transaction"] != top_up.payload["transaction"] + expired = reconnect_time == 1_025 + assert rpc_calls == (["eth_getBlockByNumber", "eth_call"] if expired else ["eth_call"]) + assert state_blocks == ( + [{"blockHash": FINALIZED_HASH, "requireCanonical": True}] if expired else ["latest"] + ) + assert len(tx_calls) == (2 if onchain_deposit == 25 else 3) + + +@pytest.mark.asyncio +async def test_sponsored_top_ups_are_unique_within_one_second( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + now = [1_000] + deterministic_sponsored_transactions(monkeypatch, now) + + async def create_top_up() -> dict[str, Any]: + payment_method = method(account) + opened = await payment_method.create_credential( + challenge(fee_payer=True, suggested_deposit="20", min_voucher_delta="0") + ) + return ( + await payment_method.create_credential( + challenge( + amount="15", + fee_payer=True, + suggested_deposit=None, + min_voucher_delta="0", + snapshot={ + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "20", + "acceptedCumulative": "10", + "spent": "10", + "settled": "0", + "requiredCumulative": "25", + }, + ) + ) + ).payload + + first = await create_top_up() + second = await create_top_up() + first_fields = transaction_fields(first) + second_fields = transaction_fields(second) + + assert first_fields[8] == second_fields[8] + assert first_fields[9] != second_fields[9] + assert first_fields[9] and second_fields[9] + assert first["transaction"] != second["transaction"] + + +@pytest.mark.asyncio +async def test_channel_nonce_lane_reads_pending_state( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + calls: list[tuple[str, str, list[object]]] = [] + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> str: + calls.append((rpc_url, rpc_method, params)) + return "0x" + encode(["uint64"], [7]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + nonce_key = int.from_bytes(b"\x01" + SALT[1:]) + + assert await method(account)._lane_nonce(nonce_key) == 7 + data = keccak(b"getNonce(address,uint256)")[:4] + encode( + ["address", "uint256"], [account.address, nonce_key] + ) + assert calls == [ + ( + RPC_URL, + "eth_call", + [ + { + "to": session_module.NONCE_PRECOMPILE, + "data": "0x" + data.hex(), + }, + "pending", + ], + ) + ] + + +@pytest.mark.asyncio +async def test_advertised_deposit_rollback_uses_onchain_state( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + deterministic_transactions(monkeypatch) + payment_method = method(account) + opened = await payment_method.create_credential(challenge()) + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "0", + "acceptedCumulative": "10", + "spent": "10", + "settled": "0", + "requiredCumulative": "26", + } + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> str: + return "0x" + encode(["uint96", "uint96", "uint32"], [0, 50, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + + credential = await payment_method.create_credential(challenge(snapshot=snapshot)) + + assert credential.payload["action"] == "voucher" + assert credential.payload["cumulativeAmount"] == "35" + + +@pytest.mark.asyncio +async def test_required_above_deposit_selects_top_up( + account: TempoAccount, monkeypatch: pytest.MonkeyPatch +) -> None: + rpc_calls = deterministic_transactions(monkeypatch) + store = MemoryStore() + payment_method = method(account, store=store) + opened = await payment_method.create_credential( + challenge(suggested_deposit="20", min_voucher_delta="0") + ) + snapshot = { + "channelId": opened.payload["channelId"], + "descriptor": opened.payload["descriptor"], + "escrow": TIP20_CHANNEL_ESCROW, + "chainId": CHAIN_ID, + "deposit": "20", + "acceptedCumulative": "10", + "spent": "10", + "settled": "0", + "requiredCumulative": "25", + } + + top_up = await payment_method.create_credential( + challenge( + amount="15", + suggested_deposit=None, + min_voucher_delta="0", + snapshot=snapshot, + ) + ) + + assert top_up.payload["action"] == "topUp" + assert top_up.payload["channelId"] == opened.payload["channelId"] + assert top_up.payload["additionalDeposit"] == "5" + assert transaction_fields(top_up.payload)[4][0][2][:4].hex() == "dc48471e" + assert transaction_fields(top_up.payload)[6] == transaction_fields(opened.payload)[6] + + retried = await payment_method.create_credential( + challenge(amount="15", suggested_deposit=None, min_voucher_delta="0") + ) + assert retried.payload == top_up.payload + assert len(rpc_calls) == 2 + + async def rpc_call(rpc_url: str, rpc_method: str, params: list[object]) -> str: + return "0x" + encode(["uint96", "uint96", "uint32"], [0, 25, 0]).hex() + + monkeypatch.setattr(session_module, "_rpc_call", rpc_call) + snapshot["deposit"] = "25" + confirmed = await payment_method.create_credential( + challenge( + amount="15", + suggested_deposit=None, + min_voucher_delta="0", + snapshot=snapshot, + ) + ) + assert confirmed.payload["action"] == "voucher" + assert confirmed.payload["cumulativeAmount"] == "25" + assert len(rpc_calls) == 2 diff --git a/tests/test_tempo_session_http.py b/tests/test_tempo_session_http.py new file mode 100644 index 00000000..ffe72b05 --- /dev/null +++ b/tests/test_tempo_session_http.py @@ -0,0 +1,303 @@ +"""HTTP transport coverage for Tempo session SSE payments.""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Iterator +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from mpp import Challenge, Credential, MemoryStore, Receipt +from mpp.client import PaymentTransport, SyncPaymentTransport +from mpp.methods.tempo import TempoAccount, tempo_session +from mpp.runtime import PaymentRuntime + +PRIVATE_KEY = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" +TOKEN = "0x20c0000000000000000000000000000000000000" +PAYEE = "0x2222222222222222222222222222222222222222" +URL = "https://api.example.test/stream?topic=weather" + + +@pytest.fixture(autouse=True) +def mock_session_transactions() -> Iterator[None]: + with ( + patch( + "mpp.methods.tempo.session.get_tx_params", + new=AsyncMock(return_value=(4217, 1, 1)), + ), + patch( + "mpp.methods.tempo.session.TempoSessionMethod._lane_nonce", + new=AsyncMock(return_value=0), + ), + ): + yield + + +def _challenge() -> Challenge: + return Challenge( + id="session-challenge", + method="tempo", + intent="session", + realm="api.example.test", + request={ + "amount": "2", + "currency": TOKEN, + "recipient": PAYEE, + "suggestedDeposit": "5", + "methodDetails": { + "sessionProtocol": "v2", + "chainId": 4217, + "escrowContract": "0x4d50500000000000000000000000000000000000", + }, + }, + ) + + +def _receipt(channel_id: str, accepted: int, spent: int) -> dict[str, object]: + return { + "method": "tempo", + "intent": "session", + "status": "success", + "timestamp": "2026-07-20T00:00:00Z", + "reference": channel_id, + "challengeId": "session-challenge", + "channelId": channel_id, + "acceptedCumulative": str(accepted), + "spent": str(spent), + } + + +def _receipt_header(value: dict[str, object]) -> str: + extensions = dict(value) + for field in ("method", "status", "timestamp", "reference"): + extensions.pop(field) + return Receipt( + method="tempo", + status="success", + timestamp=datetime(2026, 7, 20, tzinfo=UTC), + reference=str(value["reference"]), + extensions=extensions, + ).to_payment_receipt() + + +class SessionServer: + def __init__(self) -> None: + self.challenge = _challenge() + self.actions: list[str] = [] + self.management_urls: list[str] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + authorization = request.headers.get("authorization") + if authorization is None: + return httpx.Response( + 402, + headers={ + "WWW-Authenticate": self.challenge.to_www_authenticate("api.example.test") + }, + ) + + payload = Credential.from_authorization(authorization).payload + action = str(payload["action"]) + self.actions.append(action) + channel_id = str(payload["channelId"]) + + if action == "open": + need = { + "channelId": channel_id, + "requiredCumulative": "8", + "acceptedCumulative": "2", + "deposit": "5", + } + body = ( + b"data: first\n\n" + + f"event: payment-need-voucher\ndata: {json.dumps(need)}\n\n".encode() + + b"data: second\n\n" + + ( + "event: payment-receipt\ndata: " + + json.dumps(_receipt(channel_id, 8, 8)) + + "\n\n" + ).encode() + ) + return httpx.Response( + 200, + headers={ + "content-type": "text/event-stream", + "content-length": str(len(body)), + "payment-receipt": _receipt_header(_receipt(channel_id, 2, 0)), + }, + stream=httpx.ByteStream(body), + ) + + self.management_urls.append(str(request.url)) + accepted = 2 if action == "topUp" else 8 + return httpx.Response( + 204, + headers={"payment-receipt": _receipt_header(_receipt(channel_id, accepted, 0))}, + ) + + +class ErrorSseServer: + def __init__(self) -> None: + self.challenge = _challenge() + self.payloads: list[dict[str, object]] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + authorization = request.headers.get("authorization") + if authorization is None: + return httpx.Response( + 402, + headers={ + "WWW-Authenticate": self.challenge.to_www_authenticate("api.example.test") + }, + ) + payload = Credential.from_authorization(authorization).payload + self.payloads.append(payload) + channel_id = str(payload["channelId"]) + need = { + "channelId": channel_id, + "requiredCumulative": "8", + "acceptedCumulative": "2", + "deposit": "5", + } + body = f"event: payment-need-voucher\ndata: {json.dumps(need)}\n\n".encode() + return httpx.Response( + 500, + headers={ + "content-type": "text/event-stream", + "payment-receipt": _receipt_header(_receipt(channel_id, 2, 0)), + }, + stream=httpx.ByteStream(body), + ) + + +class LoopBoundStore(MemoryStore): + def __init__(self) -> None: + super().__init__() + self.loop: asyncio.AbstractEventLoop | None = None + + def _check_loop(self) -> None: + loop = asyncio.get_running_loop() + if self.loop is None: + self.loop = loop + elif self.loop is not loop: + raise RuntimeError("store used from a different event loop") + + async def get(self, key: str) -> Any | None: + self._check_loop() + return await super().get(key) + + async def put(self, key: str, value: Any) -> None: + self._check_loop() + await super().put(key, value) + + +@pytest.mark.asyncio +async def test_async_transport_drives_session_sse_top_up_and_voucher() -> None: + server = SessionServer() + method = tempo_session( + account=TempoAccount.from_key(PRIVATE_KEY), + max_deposit=10, + rpc_url="https://rpc.test", + ) + transport = PaymentTransport(methods=[method], inner=httpx.MockTransport(server)) + + async with httpx.AsyncClient(transport=transport) as client: + response = await client.get(URL, headers={"Accept": "text/event-stream"}) + assert b"".join([chunk async for chunk in response.aiter_bytes()]) == ( + b"data: first\n\ndata: second\n\n" + ) + + assert server.actions == ["open", "topUp", "voucher"] + assert server.management_urls == [URL, URL] + assert "content-length" not in response.headers + + +@pytest.mark.asyncio +async def test_async_session_state_stays_on_runtime_loop() -> None: + server = SessionServer() + store = LoopBoundStore() + method = tempo_session( + account=TempoAccount.from_key(PRIVATE_KEY), + max_deposit=10, + rpc_url="https://rpc.test", + channel_store=store, + ) + runtime = PaymentRuntime([method]) + transport = PaymentTransport(runtime=runtime, inner=httpx.MockTransport(server)) + + try: + async with httpx.AsyncClient(transport=transport) as client: + response = await client.get(URL, headers={"Accept": "text/event-stream"}) + assert b"".join([chunk async for chunk in response.aiter_bytes()]) == ( + b"data: first\n\ndata: second\n\n" + ) + finally: + await runtime.aclose() + + assert store.loop is not asyncio.get_running_loop() + + +def test_sync_transport_drives_session_sse_top_up_and_voucher() -> None: + server = SessionServer() + method = tempo_session( + account=TempoAccount.from_key(PRIVATE_KEY), + max_deposit=10, + rpc_url="https://rpc.test", + ) + transport = SyncPaymentTransport(methods=[method], inner=httpx.MockTransport(server)) + + with httpx.Client(transport=transport) as client: + response = client.get(URL, headers={"Accept": "text/event-stream"}) + assert b"".join(response.iter_bytes()) == b"data: first\n\ndata: second\n\n" + + assert server.actions == ["open", "topUp", "voucher"] + assert server.management_urls == [URL, URL] + assert "content-length" not in response.headers + + +@pytest.mark.asyncio +async def test_async_error_sse_cannot_advance_session() -> None: + server = ErrorSseServer() + transport = PaymentTransport( + methods=[ + tempo_session( + account=TempoAccount.from_key(PRIVATE_KEY), + max_deposit=10, + rpc_url="https://rpc.test", + ) + ], + inner=httpx.MockTransport(server), + ) + + async with httpx.AsyncClient(transport=transport) as client: + assert (await client.get(URL)).status_code == 500 + assert (await client.get(URL)).status_code == 500 + + assert [payload["action"] for payload in server.payloads] == ["open", "open"] + assert server.payloads[0]["transaction"] == server.payloads[1]["transaction"] + + +def test_sync_error_sse_cannot_advance_session() -> None: + server = ErrorSseServer() + transport = SyncPaymentTransport( + methods=[ + tempo_session( + account=TempoAccount.from_key(PRIVATE_KEY), + max_deposit=10, + rpc_url="https://rpc.test", + ) + ], + inner=httpx.MockTransport(server), + ) + + with httpx.Client(transport=transport) as client: + assert client.get(URL).status_code == 500 + assert client.get(URL).status_code == 500 + + assert [payload["action"] for payload in server.payloads] == ["open", "open"] + assert server.payloads[0]["transaction"] == server.payloads[1]["transaction"] diff --git a/tests/test_tempo_session_sse.py b/tests/test_tempo_session_sse.py new file mode 100644 index 00000000..0b66e288 --- /dev/null +++ b/tests/test_tempo_session_sse.py @@ -0,0 +1,300 @@ +"""Tests for Tempo session SSE control-frame filtering.""" + +from __future__ import annotations + +import asyncio +import gzip +import json +from collections.abc import AsyncIterator, Iterator +from typing import Any + +import httpx +import pytest + +from mpp.methods.tempo._session_sse import ( + wrap_async_sse_response, + wrap_sync_sse_response, +) + +NEED_VOUCHER = { + "channelId": "0xchannel", + "requiredCumulative": "20", + "acceptedCumulative": "10", + "deposit": "100", +} +RECEIPT = { + "method": "tempo", + "intent": "session", + "status": "success", + "timestamp": "2025-01-01T00:00:00Z", + "reference": "0xchannel", + "challengeId": "challenge-1", + "channelId": "0xchannel", + "acceptedCumulative": "20", + "spent": "12", + "units": 3, +} + + +def _event(name: str, payload: object, newline: str = "\n") -> bytes: + return f"event: {name}{newline}data: {json.dumps(payload)}{newline}{newline}".encode() + + +class TrackingSyncStream(httpx.SyncByteStream): + def __init__(self, content: bytes) -> None: + self.chunks = [content[index : index + 1] for index in range(len(content))] + self.closed = False + + def __iter__(self) -> Iterator[bytes]: + yield from self.chunks + + def close(self) -> None: + self.closed = True + + +class TrackingAsyncStream(httpx.AsyncByteStream): + def __init__(self, content: bytes) -> None: + self.chunks = [content[index : index + 1] for index in range(len(content))] + self.closed = False + + async def __aiter__(self) -> AsyncIterator[bytes]: + for chunk in self.chunks: + yield chunk + + async def aclose(self) -> None: + self.closed = True + + +class BlockingAsyncStream(httpx.AsyncByteStream): + def __init__(self) -> None: + self.waiting = asyncio.Event() + self.closed = False + + async def __aiter__(self) -> AsyncIterator[bytes]: + yield b"data: first\n\n" + self.waiting.set() + await asyncio.Event().wait() + + async def aclose(self) -> None: + self.closed = True + + +def _mixed_stream() -> tuple[bytes, bytes]: + application = "event: custom\r\ndata: café\r\n\r\n".encode() + comment = b": keepalive\r\r" + malformed = b'event: payment-receipt\ndata: {"status":"success"}\n\n' + malformed_json = b"event: payment-need-voucher\ndata: {nope}\n\n" + unknown = _event("custom-payment", RECEIPT) + need = _event("payment-need-voucher", NEED_VOUCHER, "\r") + receipt = _event("payment-receipt", RECEIPT, "\r\n") + unterminated = b'event: payment-receipt\ndata: {"still":"application"}' + source = ( + application + need + comment + malformed + malformed_json + receipt + unknown + unterminated + ) + expected = application + comment + malformed + malformed_json + unknown + unterminated + return source, expected + + +def test_sync_wrapper_filters_controls_and_preserves_other_bytes() -> None: + content, expected = _mixed_stream() + stream = TrackingSyncStream(content) + request = httpx.Request("GET", "https://example.com/stream") + response = httpx.Response( + 201, + headers={"content-type": "text/event-stream", "content-length": str(len(content))}, + stream=stream, + request=request, + extensions={"reason_phrase": b"Created"}, + ) + controls: list[tuple[str, dict[str, Any]]] = [] + + wrapped = wrap_sync_sse_response( + response, + on_need_voucher=lambda value: controls.append(("need", value)), + on_receipt=lambda value: controls.append(("receipt", value)), + ) + + assert wrapped.status_code == 201 + assert wrapped.reason_phrase == "Created" + assert wrapped.request is request + assert wrapped.headers["content-type"] == "text/event-stream" + assert "content-length" not in wrapped.headers + assert b"".join(wrapped.iter_raw()) == expected + assert controls == [("need", NEED_VOUCHER), ("receipt", RECEIPT)] + assert stream.closed + + +def test_sync_wrapper_decodes_compressed_sse_before_filtering() -> None: + application = b"data: application\n\n" + compressed = gzip.compress(_event("payment-need-voucher", NEED_VOUCHER) + application) + stream = TrackingSyncStream(compressed) + controls: list[dict[str, Any]] = [] + wrapped = wrap_sync_sse_response( + httpx.Response( + 200, + headers={"content-encoding": "gzip", "content-length": str(len(compressed))}, + stream=stream, + ), + on_need_voucher=controls.append, + on_receipt=lambda _: None, + ) + + assert b"".join(wrapped.iter_raw()) == application + assert controls == [NEED_VOUCHER] + assert "content-encoding" not in wrapped.headers + assert "content-length" not in wrapped.headers + assert stream.closed + + +@pytest.mark.asyncio +async def test_async_wrapper_awaits_handlers_and_preserves_other_bytes() -> None: + need = _event("payment-need-voucher", NEED_VOUCHER) + application = b"data: after\n\n" + stream = TrackingAsyncStream(need + application) + response = httpx.Response(200, stream=stream) + started = asyncio.Event() + release = asyncio.Event() + + async def on_need_voucher(value: dict[str, Any]) -> None: + assert value == NEED_VOUCHER + started.set() + await release.wait() + + async def on_receipt(value: dict[str, Any]) -> None: + raise AssertionError(f"unexpected receipt: {value}") + + wrapped = wrap_async_sse_response( + response, + on_need_voucher=on_need_voucher, + on_receipt=on_receipt, + ) + iterator = wrapped.aiter_raw() + + async def read_next() -> bytes: + return await anext(iterator) + + next_frame = asyncio.create_task(read_next()) + + await started.wait() + assert not next_frame.done() + release.set() + assert await next_frame == application + with pytest.raises(StopAsyncIteration): + await anext(iterator) + assert stream.closed + + +@pytest.mark.asyncio +async def test_async_wrapper_handles_mixed_boundaries_and_split_utf8() -> None: + content, expected = _mixed_stream() + stream = TrackingAsyncStream(content) + controls: list[tuple[str, dict[str, Any]]] = [] + + async def on_need_voucher(value: dict[str, Any]) -> None: + controls.append(("need", value)) + + async def on_receipt(value: dict[str, Any]) -> None: + controls.append(("receipt", value)) + + wrapped = wrap_async_sse_response( + httpx.Response( + 200, + headers={"content-length": str(len(content))}, + stream=stream, + ), + on_need_voucher=on_need_voucher, + on_receipt=on_receipt, + ) + + assert b"".join([chunk async for chunk in wrapped.aiter_raw()]) == expected + assert controls == [("need", NEED_VOUCHER), ("receipt", RECEIPT)] + assert "content-length" not in wrapped.headers + assert stream.closed + + +@pytest.mark.asyncio +async def test_async_wrapper_decodes_compressed_sse_before_filtering() -> None: + application = b"data: application\n\n" + compressed = gzip.compress(_event("payment-receipt", RECEIPT) + application) + stream = TrackingAsyncStream(compressed) + controls: list[dict[str, Any]] = [] + + async def on_receipt(value: dict[str, Any]) -> None: + controls.append(value) + + async def ignore(_: dict[str, Any]) -> None: + return None + + wrapped = wrap_async_sse_response( + httpx.Response( + 200, + headers={"content-encoding": "gzip", "content-length": str(len(compressed))}, + stream=stream, + ), + on_need_voucher=ignore, + on_receipt=on_receipt, + ) + + assert b"".join([chunk async for chunk in wrapped.aiter_raw()]) == application + assert controls == [RECEIPT] + assert "content-encoding" not in wrapped.headers + assert "content-length" not in wrapped.headers + assert stream.closed + + +def test_sync_close_closes_source_without_consuming_it() -> None: + stream = TrackingSyncStream(b"data: pending\n\n") + wrapped = wrap_sync_sse_response( + httpx.Response(200, stream=stream), + on_need_voucher=lambda _: None, + on_receipt=lambda _: None, + ) + + wrapped.close() + + assert stream.closed + + +@pytest.mark.asyncio +async def test_async_close_closes_source_without_consuming_it() -> None: + stream = TrackingAsyncStream(b"data: pending\n\n") + + async def ignore(_: dict[str, Any]) -> None: + return None + + wrapped = wrap_async_sse_response( + httpx.Response(200, stream=stream), + on_need_voucher=ignore, + on_receipt=ignore, + ) + + await wrapped.aclose() + + assert stream.closed + + +@pytest.mark.asyncio +async def test_cancelling_iteration_closes_source() -> None: + stream = BlockingAsyncStream() + + async def ignore(_: dict[str, Any]) -> None: + return None + + wrapped = wrap_async_sse_response( + httpx.Response(200, stream=stream), + on_need_voucher=ignore, + on_receipt=ignore, + ) + iterator = wrapped.aiter_raw() + assert await anext(iterator) == b"data: first\n\n" + + async def read_next() -> bytes: + return await anext(iterator) + + task = asyncio.create_task(read_next()) + await stream.waiting.wait() + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + assert stream.closed