music-assistant-server

1"""
2Sendspin WebSocket proxy handler for Music Assistant.
3
4This module provides an authenticated WebSocket proxy to the internal Sendspin server,
5allowing web clients to connect through the main webserver instead of requiring direct
6access to the Sendspin port.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import contextlib
13import json
14import logging
15from typing import TYPE_CHECKING
16
17import aiohttp
18from aiohttp import ClientConnectorError, WSMsgType, web
19
20from music_assistant.constants import MASS_LOGGER_NAME
21from music_assistant.controllers.webserver.helpers.auth_middleware import (
22    get_authenticated_user,
23    is_request_from_ingress,
24)
25
26if TYPE_CHECKING:
27    from music_assistant_models.auth import User
28
29    from music_assistant.controllers.webserver import WebserverController
30
31LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.sendspin_proxy")
32
33
34class SendspinProxyHandler:
35    """Handler for proxying WebSocket connections to the internal Sendspin server."""
36
37    def __init__(self, webserver: WebserverController) -> None:
38        """
39        Initialize the Sendspin proxy handler.
40
41        :param webserver: The webserver controller instance.
42        """
43        self.webserver = webserver
44        self.mass = webserver.mass
45        self.logger = LOGGER
46
47    async def handle_sendspin_proxy(self, request: web.Request) -> web.WebSocketResponse:
48        """
49        Handle incoming WebSocket connection and proxy to internal Sendspin server.
50
51        Authentication is required as the first message. The client must send:
52        {"type": "auth", "token": "<access_token>"}
53
54        After successful authentication, all messages are proxied bidirectionally.
55
56        :param request: The incoming HTTP request to upgrade to WebSocket.
57        :return: The WebSocket response.
58        """
59        wsock = web.WebSocketResponse(heartbeat=25)
60        await wsock.prepare(request)
61
62        self.logger.debug("Sendspin proxy connection from %s", request.remote)
63
64        # Check for ingress authentication (HA handles auth via headers)
65        if is_request_from_ingress(request):
66            user = await get_authenticated_user(request)
67            if not user:
68                self.logger.warning(
69                    "Ingress auth failed for sendspin proxy from %s", request.remote
70                )
71                await wsock.close(code=4001, message=b"Ingress authentication failed")
72                return wsock
73            self.logger.debug("Sendspin proxy authenticated via ingress: %s", user.username)
74        else:
75            # Regular auth via first message
76            try:
77                user = await self._authenticate(wsock)
78                if not user:
79                    return wsock
80            except TimeoutError:
81                self.logger.warning("Auth timeout for sendspin proxy from %s", request.remote)
82                await wsock.close(code=4001, message=b"Authentication timeout")
83                return wsock
84            except Exception:
85                self.logger.exception("Auth error for sendspin proxy")
86                await wsock.close(code=4001, message=b"Authentication error")
87                return wsock
88
89        # The internal Sendspin server may not be ready yet during startup
90        # (it starts in the provider load phase, after the webserver).
91        # Retry a few times with backoff to handle this race condition.
92        try:
93            internal_ws = None
94            for attempt in range(5):
95                try:
96                    internal_ws = await self.mass.http_session.ws_connect(
97                        self.webserver.internal_sendspin_url
98                    )
99                    break
100                except ClientConnectorError:
101                    if attempt < 4:
102                        await asyncio.sleep(0.5 * (attempt + 1))
103                        continue
104                    self.logger.exception("Failed to connect to internal Sendspin server")
105                    await wsock.close(code=1011, message=b"Internal server error")
106                    return wsock
107            if internal_ws is None:
108                raise RuntimeError("Retry loop exited without connecting or returning")
109        except Exception:
110            self.logger.exception("Failed to connect to internal Sendspin server")
111            await wsock.close(code=1011, message=b"Internal server error")
112            return wsock
113        self.logger.debug("Sendspin proxy authenticated and connected for %s", request.remote)
114
115        try:
116            await self._proxy_messages(wsock, internal_ws)
117        finally:
118            if not internal_ws.closed:
119                await internal_ws.close()
120            if not wsock.closed:
121                await wsock.close()
122
123        return wsock
124
125    async def _authenticate(self, wsock: web.WebSocketResponse) -> User | None:
126        """
127        Wait for and validate authentication message.
128
129        :param wsock: The client WebSocket connection.
130        :return: The authenticated user, or None if authentication failed.
131        """
132        async with asyncio.timeout(10):
133            msg = await wsock.receive()
134
135        if msg.type != WSMsgType.TEXT:
136            await wsock.close(code=4001, message=b"Expected text message for auth")
137            return None
138
139        try:
140            auth_data = json.loads(msg.data)
141        except json.JSONDecodeError:
142            await wsock.close(code=4001, message=b"Invalid JSON in auth message")
143            return None
144
145        if auth_data.get("type") != "auth":
146            await wsock.close(code=4001, message=b"First message must be auth")
147            return None
148
149        token = auth_data.get("token")
150        if not token:
151            await wsock.close(code=4001, message=b"Token required in auth message")
152            return None
153
154        user = await self.webserver.auth.authenticate_with_token(token)
155        if not user:
156            await wsock.close(code=4001, message=b"Invalid or expired token")
157            return None
158
159        # Set the sendspin player_id on the user's websocket client(s)
160        # This allows the player controller to auto-whitelist this (web)player
161        # without modifying the user's player_filter list
162        client_id = auth_data.get("client_id")
163        if client_id:
164            self.webserver.set_sendspin_player_for_user(user.user_id, client_id)
165            self.logger.debug("Registered sendspin player %s for user %s", client_id, user.username)
166
167        self.logger.debug("Sendspin proxy authenticated user: %s", user.username)
168        await wsock.send_str('{"type": "auth_ok"}')
169        return user
170
171    async def _proxy_messages(
172        self,
173        client_ws: web.WebSocketResponse,
174        internal_ws: aiohttp.ClientWebSocketResponse,
175    ) -> None:
176        """
177        Proxy messages bidirectionally between client and internal Sendspin server.
178
179        :param client_ws: The client WebSocket connection.
180        :param internal_ws: The internal Sendspin server WebSocket connection.
181        """
182        client_to_internal = asyncio.create_task(
183            self._forward_client_to_internal(client_ws, internal_ws)
184        )
185        internal_to_client = asyncio.create_task(
186            self._forward_internal_to_client(client_ws, internal_ws)
187        )
188
189        done, pending = await asyncio.wait(
190            [client_to_internal, internal_to_client],
191            return_when=asyncio.FIRST_COMPLETED,
192        )
193
194        for task in pending:
195            task.cancel()
196        peer_results = await asyncio.gather(*pending, return_exceptions=True)
197
198        # collect everything first so cleanup failures cannot mask the primary error
199        unexpected: list[BaseException] = []
200        for task in done:
201            with contextlib.suppress(asyncio.CancelledError):
202                if exc := task.exception():
203                    self._collect_proxy_exception(exc, unexpected)
204        for result in peer_results:
205            if isinstance(result, BaseException):
206                self._collect_proxy_exception(result, unexpected)
207        if not unexpected:
208            return
209        for extra in unexpected[1:]:
210            self.logger.warning(
211                "Additional Sendspin proxy error while forwarding: %s",
212                extra,
213            )
214        raise unexpected[0]
215
216    def _collect_proxy_exception(
217        self,
218        exc: BaseException,
219        unexpected: list[BaseException],
220    ) -> None:
221        """Log expected transport disconnects; collect anything else."""
222        if isinstance(exc, asyncio.CancelledError):
223            return
224        if isinstance(
225            exc,
226            (ConnectionError, aiohttp.ClientError, asyncio.IncompleteReadError, EOFError),
227        ):
228            self.logger.debug("Sendspin proxy connection closed while forwarding: %s", exc)
229            return
230        unexpected.append(exc)
231
232    async def _forward_client_to_internal(
233        self,
234        client_ws: web.WebSocketResponse,
235        internal_ws: aiohttp.ClientWebSocketResponse,
236    ) -> None:
237        """
238        Forward messages from client to internal Sendspin server.
239
240        :param client_ws: The client WebSocket connection.
241        :param internal_ws: The internal Sendspin server WebSocket connection.
242        """
243        async for msg in client_ws:
244            if msg.type == WSMsgType.TEXT:
245                await internal_ws.send_str(msg.data)
246            elif msg.type == WSMsgType.BINARY:
247                await internal_ws.send_bytes(msg.data)
248            elif msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSED, WSMsgType.ERROR):
249                if msg.type == WSMsgType.ERROR:
250                    self.logger.debug("Sendspin proxy client transport error: %s", msg.data)
251                break
252
253    async def _forward_internal_to_client(
254        self,
255        client_ws: web.WebSocketResponse,
256        internal_ws: aiohttp.ClientWebSocketResponse,
257    ) -> None:
258        """
259        Forward messages from internal Sendspin server to client.
260
261        :param client_ws: The client WebSocket connection.
262        :param internal_ws: The internal Sendspin server WebSocket connection.
263        """
264        async for msg in internal_ws:
265            if msg.type == WSMsgType.TEXT:
266                await client_ws.send_str(msg.data)
267            elif msg.type == WSMsgType.BINARY:
268                await client_ws.send_bytes(msg.data)
269            elif msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSED, WSMsgType.ERROR):
270                if msg.type == WSMsgType.ERROR:
271                    self.logger.debug("Sendspin proxy internal transport error: %s", msg.data)
272                break
273