From d0143af305c758e025076b17f09d6cd0e86fd683 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Mon, 17 Aug 2026 21:13:32 -0700 Subject: [PATCH] Update challenge params after payment if needed. Malformed / unexpected (non-JSON) responses will continue to be ignored. --- src/livepeer_gateway/remote_signer.py | 25 ++++- tests/test_live_payment_session.py | 153 +++++++++++++++++++++++++- 2 files changed, 170 insertions(+), 8 deletions(-) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index dd4554f..1761dd6 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -256,11 +256,12 @@ async def send_payment(self) -> None: Raises LivepeerHTTPError on error responses so callers can branch on the status code, and SkipPaymentCycle when the signer gates the cycle. + Malformed success responses are ignored without changing the challenge. """ if not self._signer_url: return - from .http import _post_empty + from .http import request_json payment = await self.get_payment() if not payment.seg_creds: @@ -271,7 +272,27 @@ async def send_payment(self) -> None: "Livepeer-Payment": payment.payment, "Livepeer-Segment": payment.seg_creds, } - await _post_empty(self._challenge.payment_url, headers=headers, timeout=5.0) + try: + data = await request_json( + self._challenge.payment_url, + method="POST", + headers=headers, + timeout=5.0, + ) + except LivepeerGatewayError as e: + if isinstance(e.__cause__, (UnicodeDecodeError, json.JSONDecodeError)): + return + raise + if not isinstance(data, dict): + return + + payment_params = data.get("payment_params") + if not isinstance(payment_params, str) or not payment_params: + return + self._challenge = replace( + self._challenge, + payment_params=payment_params, + ) async def run_payments(self) -> bool: """Keep a metered session funded until cancelled or the session ends. diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py index 35afbb5..3af41e4 100644 --- a/tests/test_live_payment_session.py +++ b/tests/test_live_payment_session.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import types from unittest import mock @@ -112,14 +113,107 @@ async def test_none_signer_exits_early(self) -> None: assert payment.payment == "" assert payment.seg_creds is None - async def test_send_payment_reuses_empty_post_helper(self) -> None: + async def test_send_payment_rotates_payment_params_for_next_payment(self) -> None: + signer_calls: list[dict[str, object]] = [] + payment_posts: list[tuple[str, str, dict[str, str]]] = [] + refreshed_params = iter(("fresh-payment-params-1", "fresh-payment-params-2")) + + async def _post_json( + url: str, + payload: dict[str, object], + *, + headers: dict[str, str] | None = None, + timeout: float = 5.0, + ) -> dict[str, object]: + del url, headers, timeout + signer_calls.append(dict(payload)) + sequence = len(signer_calls) + return { + "payment": f"payment-{sequence}", + "segCreds": f"segment-{sequence}", + "state": {"sequence": sequence}, + } + + async def _request_body( + url: str, + *, + method: str, + payload: dict[str, object] | None, + headers: dict[str, str], + timeout: float, + ) -> tuple[bytes, str]: + del payload, timeout + payment_posts.append((url, method, dict(headers))) + payment_params = next(refreshed_params) + return ( + json.dumps( + { + "payment_params": payment_params, + "orchestrator": "https://orch.example.com", + "manifest_id": "manifest-1", + "payment_url": _PAYMENT_URL, + } + ).encode(), + "application/json", + ) + session = LivePaymentSession( "https://signer.example.com", type="lv2v", - challenge=_challenge(), + challenge=_challenge(payment_params="initial-payment-params"), + app="live-video-to-video/scope", + max_price={"price": 10.12, "currency": "wei", "unit": "pixels"}, + ) + + with ( + mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json), + mock.patch("livepeer_gateway.http._request_body", side_effect=_request_body), + ): + await session.send_payment() + await session.send_payment() + + assert [call["orchestrator"] for call in signer_calls] == [ + "initial-payment-params", + "fresh-payment-params-1", + ] + assert signer_calls[1]["state"] == {"sequence": 1} + assert [call["app"] for call in signer_calls] == [ + "live-video-to-video/scope", + "live-video-to-video/scope", + ] + assert [call["maxPrice"] for call in signer_calls] == [ + {"price": 10.12, "currency": "wei", "unit": "pixels"}, + {"price": 10.12, "currency": "wei", "unit": "pixels"}, + ] + assert payment_posts == [ + ( + _PAYMENT_URL, + "POST", + { + "Livepeer-Payment": "payment-1", + "Livepeer-Segment": "segment-1", + }, + ), + ( + _PAYMENT_URL, + "POST", + { + "Livepeer-Payment": "payment-2", + "Livepeer-Segment": "segment-2", + }, + ), + ] + assert session._challenge == _challenge( + payment_params="fresh-payment-params-2" + ) + + async def test_send_payment_accepts_legacy_non_json_response(self) -> None: + session = LivePaymentSession( + "https://signer.example.com", + type="lv2v", + challenge=_challenge(payment_params="legacy-payment-params"), ) - post_empty = mock.AsyncMock() with ( mock.patch.object( session, @@ -128,15 +222,62 @@ async def test_send_payment_reuses_empty_post_helper(self) -> None: return_value=types.SimpleNamespace(payment="p", seg_creds="s") ), ), - mock.patch("livepeer_gateway.http._post_empty", post_empty), + mock.patch( + "livepeer_gateway.http._request_body", + new=mock.AsyncMock( + return_value=(b"legacy-protobuf", "application/octet-stream") + ), + ) as request_body, ): await session.send_payment() - post_empty.assert_awaited_once_with( + request_body.assert_awaited_once_with( _PAYMENT_URL, + method="POST", + payload=None, headers={"Livepeer-Payment": "p", "Livepeer-Segment": "s"}, timeout=5.0, ) + assert session._challenge == _challenge( + payment_params="legacy-payment-params" + ) + + @pytest.mark.parametrize( + "body", + [ + b"not-json", + b"[]", + b"{}", + b'{"payment_params":""}', + ], + ) + async def test_send_payment_ignores_invalid_json_response_without_mutation( + self, + body: bytes, + ) -> None: + initial_challenge = _challenge(payment_params="initial-payment-params") + session = LivePaymentSession( + "https://signer.example.com", + type="lv2v", + challenge=initial_challenge, + ) + + with ( + mock.patch.object( + session, + "get_payment", + new=mock.AsyncMock( + return_value=types.SimpleNamespace(payment="p", seg_creds="s") + ), + ), + mock.patch( + "livepeer_gateway.http._request_body", + new=mock.AsyncMock(return_value=(body, "application/json")), + ), + ): + await session.send_payment() + + assert session._challenge is initial_challenge async def test_send_payment_preserves_typed_http_error(self) -> None: session = LivePaymentSession( @@ -160,7 +301,7 @@ async def test_send_payment_preserves_typed_http_error(self) -> None: ), ), mock.patch( - "livepeer_gateway.http._post_empty", + "livepeer_gateway.http._request_body", new=mock.AsyncMock(side_effect=error), ), ):