diff --git a/python/x402/mcp/server_async.py b/python/x402/mcp/server_async.py index 6d0d86b67c..3d509ba790 100644 --- a/python/x402/mcp/server_async.py +++ b/python/x402/mcp/server_async.py @@ -272,6 +272,14 @@ async def wrapped_handler(args: dict[str, Any], extra: dict[str, Any]) -> MCPToo resource_server, tool_name, config, str(e) ) + if not settle_result.success: + return await _create_settlement_failed_result_async( + resource_server, + tool_name, + config, + settle_result.error_reason or "Unknown settlement failure", + ) + # Run onAfterSettlement hook if present if config.hooks and config.hooks.on_after_settlement: settlement_context = SettlementContext( diff --git a/python/x402/mcp/server_sync.py b/python/x402/mcp/server_sync.py index 7c133b69b8..45917ee8ea 100644 --- a/python/x402/mcp/server_sync.py +++ b/python/x402/mcp/server_sync.py @@ -158,6 +158,14 @@ def wrapped_handler(args: dict[str, Any], extra: dict[str, Any]) -> MCPToolResul resource_server, tool_name, config, str(e) ) + if not settle_result.success: + return _create_settlement_failed_result_sync( + resource_server, + tool_name, + config, + settle_result.error_reason or "Unknown settlement failure", + ) + if config.hooks and config.hooks.on_after_settlement: settlement_context = SettlementContext( tool_name=tool_name, diff --git a/python/x402/mcp/tests/test_server.py b/python/x402/mcp/tests/test_server.py index b8f7fbf803..6b2ee38033 100644 --- a/python/x402/mcp/tests/test_server.py +++ b/python/x402/mcp/tests/test_server.py @@ -59,7 +59,9 @@ def find_matching_requirements(self, available, payload): return req return None - def _create_payment_required_response_real(self, accepts, resource_info, error_msg): + def _create_payment_required_response_real( + self, accepts, resource_info, error_msg, extensions=None + ): """Real implementation of create payment required response.""" from x402.schemas import PaymentRequired @@ -352,6 +354,78 @@ def handler(args, context): assert "settlement" in str(result.content).lower() or result.structured_content is not None +def test_create_payment_wrapper_settlement_returns_failure(): + """Replay attack regression: settle returning success=False must withhold tool output.""" + from x402.mcp.types import MCP_PAYMENT_RESPONSE_META_KEY + + server = MockResourceServer() + server.settle_payment = Mock( + return_value=SettleResponse( + success=False, + error_reason="AuthorizationAlreadyUsed", + transaction="", + network="eip155:84532", + ) + ) + + after_settlement_calls = [] + + def on_after_settlement(ctx): + after_settlement_calls.append(ctx) + + config = PaymentWrapperConfig( + accepts=[ + PaymentRequirements( + scheme="exact", + network="eip155:84532", + amount="1000", + asset="USDC", + pay_to="0xrecipient", + max_timeout_seconds=300, + ) + ], + hooks=PaymentWrapperHooks(on_after_settlement=on_after_settlement), + ) + + paid = create_payment_wrapper(server, config) + + def handler(args, context): + return {"content": [{"type": "text", "text": "should-not-be-returned"}]} + + wrapped = paid(handler) + payload = PaymentPayload( + x402_version=2, + accepted={ + "scheme": "exact", + "network": "eip155:84532", + "amount": "1000", + "asset": "USDC", + "pay_to": "0xrecipient", + "max_timeout_seconds": 300, + }, + payload={"signature": "0x123"}, + ) + result = wrapped( + {"test": "value"}, + { + "_meta": { + "x402/payment": ( + payload.model_dump() if hasattr(payload, "model_dump") else payload + ) + } + }, + ) + + assert result.is_error is True + assert "should-not-be-returned" not in str(result.content) + assert "should-not-be-returned" not in str(result.structured_content) + assert result.structured_content is not None + payment_response = result.structured_content[MCP_PAYMENT_RESPONSE_META_KEY] + assert payment_response["success"] is False + assert "AuthorizationAlreadyUsed" in result.structured_content["error"] + assert after_settlement_calls == [] + + def test_create_payment_wrapper_handler_error_no_settlement(): """Test that settlement is NOT called when handler returns an error.""" server = MockResourceServer() diff --git a/python/x402/mcp/tests/test_server_async.py b/python/x402/mcp/tests/test_server_async.py index 6279dec6db..431c4a3197 100644 --- a/python/x402/mcp/tests/test_server_async.py +++ b/python/x402/mcp/tests/test_server_async.py @@ -48,7 +48,9 @@ def find_matching_requirements(self, available, payload): return req return None - async def _create_payment_required_response_real(self, accepts, resource_info, error_msg): + async def _create_payment_required_response_real( + self, accepts, resource_info, error_msg, extensions=None + ): """Real implementation of create payment required response.""" from x402.schemas import PaymentRequired @@ -379,6 +381,79 @@ async def handler(args, context): assert "settlement" in str(result.content).lower() or result.structured_content is not None +@pytest.mark.asyncio +async def test_create_payment_wrapper_async_settlement_returns_failure(): + """Replay attack regression: settle returning success=False must withhold tool output.""" + from x402.mcp.types import MCP_PAYMENT_RESPONSE_META_KEY + + server = MockAsyncResourceServer() + server.settle_payment = AsyncMock( + return_value=SettleResponse( + success=False, + error_reason="AuthorizationAlreadyUsed", + transaction="", + network="eip155:84532", + ) + ) + + after_settlement_calls = [] + + async def on_after_settlement(ctx): + after_settlement_calls.append(ctx) + + config = PaymentWrapperConfig( + accepts=[ + PaymentRequirements( + scheme="exact", + network="eip155:84532", + amount="1000", + asset="USDC", + pay_to="0xrecipient", + max_timeout_seconds=300, + ) + ], + hooks=PaymentWrapperHooks(on_after_settlement=on_after_settlement), + ) + + paid = create_payment_wrapper(server, config) + + async def handler(args, context): + return {"content": [{"type": "text", "text": "should-not-be-returned"}]} + + wrapped = paid(handler) + payload = PaymentPayload( + x402_version=2, + accepted={ + "scheme": "exact", + "network": "eip155:84532", + "amount": "1000", + "asset": "USDC", + "pay_to": "0xrecipient", + "max_timeout_seconds": 300, + }, + payload={"signature": "0x123"}, + ) + result = await wrapped( + {"test": "value"}, + { + "_meta": { + "x402/payment": ( + payload.model_dump() if hasattr(payload, "model_dump") else payload + ) + } + }, + ) + + assert result.is_error is True + assert "should-not-be-returned" not in str(result.content) + assert "should-not-be-returned" not in str(result.structured_content) + assert result.structured_content is not None + payment_response = result.structured_content[MCP_PAYMENT_RESPONSE_META_KEY] + assert payment_response["success"] is False + assert "AuthorizationAlreadyUsed" in result.structured_content["error"] + assert after_settlement_calls == [] + + @pytest.mark.asyncio async def test_create_payment_wrapper_async_handler_error_no_settlement(): """Test that settlement is NOT called when async handler returns an error."""