music-assistant-server

1"""
2Music Assistant WebRTC Gateway.
3
4This module provides WebRTC-based remote access to Music Assistant instances.
5It connects to a signaling server and handles incoming WebRTC connections,
6bridging them to the local WebSocket API.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import base64
13import contextlib
14import json
15import logging
16from collections.abc import Awaitable, Callable
17from dataclasses import dataclass
18from typing import TYPE_CHECKING, Any, ClassVar
19from urllib.parse import urlparse
20
21import aiohttp
22from aiolibdatachannel import (
23    ConnectionClosedError,
24    IceServer,
25    LogLevel,
26    PeerConnection,
27    RTCConfiguration,
28    RTCError,
29    RTCState,
30    StateChangeEvent,
31    install_python_logger,
32)
33
34from music_assistant.constants import MASS_LOGGER_NAME, SENDSPIN_SERVER_PORT, VERBOSE_LOG_LEVEL
35
36if TYPE_CHECKING:
37    from aiolibdatachannel import DataChannel
38
39LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.remote_access")
40
41# Max concurrent proxied (image) fetches, so a burst of album-art requests over the
42# data channel stays bounded and never blocks the API message pump (see #4889).
43HTTP_PROXY_CONCURRENCY = 6
44
45# Chunk ma-api messages larger than this; libdatachannel caps data-channel messages at 256 KiB.
46MA_API_CHUNK_SIZE = 64 * 1024
47
48DEFAULT_SENDSPIN_URL = f"ws://localhost:{SENDSPIN_SERVER_PORT}/sendspin"
49
50
51@dataclass
52class WebRTCSession:
53    """Represents an active WebRTC session with a remote client."""
54
55    session_id: str
56    pc: PeerConnection
57    # Main API channel (ma-api) - bridges to local MA WebSocket API
58    data_channel: DataChannel | None = None
59    local_ws: aiohttp.ClientWebSocketResponse | None = None
60    # Sendspin channel - bridges to internal sendspin server
61    sendspin_channel: DataChannel | None = None
62    sendspin_ws: aiohttp.ClientWebSocketResponse | None = None
63    sendspin_player_id: str | None = None  # Extracted from first sendspin auth message
64
65
66class WebRTCGateway:
67    """
68    WebRTC Gateway for Music Assistant Remote Access.
69
70    This gateway:
71    1. Connects to a signaling server
72    2. Registers with a unique Remote ID
73    3. Handles incoming WebRTC connections from remote PWA clients
74    4. Bridges WebRTC DataChannel messages to the local WebSocket API
75    """
76
77    # Close code 4000 means this connection was replaced by a new one from the same server
78    # In that case, we should not reconnect as another connection is now active
79    CLOSE_CODE_REPLACED = 4000
80
81    # Default ICE servers (public STUN only - used as fallback)
82    DEFAULT_ICE_SERVERS: ClassVar[list[dict[str, Any]]] = [
83        {"urls": "stun:stun.home-assistant.io:3478"},
84        {"urls": "stun:stun.l.google.com:19302"},
85        {"urls": "stun:stun1.l.google.com:19302"},
86        {"urls": "stun:stun.cloudflare.com:3478"},
87    ]
88
89    def __init__(
90        self,
91        http_session: aiohttp.ClientSession,
92        remote_id: str,
93        cert_pem: str,
94        key_pem: str,
95        signaling_url: str = "wss://signaling.music-assistant.io/ws",
96        local_ws_url: str = "ws://localhost:8095/ws",
97        sendspin_url: str = DEFAULT_SENDSPIN_URL,
98        ice_servers: list[dict[str, Any]] | None = None,
99        ice_servers_callback: Callable[[], Awaitable[list[dict[str, Any]]]] | None = None,
100        set_sendspin_player_callback: Callable[[str, str], None] | None = None,
101    ) -> None:
102        """
103        Initialize the WebRTC Gateway.
104
105        :param http_session: Shared aiohttp ClientSession for HTTP/WebSocket connections.
106        :param remote_id: Remote ID for this server instance.
107        :param cert_pem: Persistent DTLS certificate (PEM), enabling client-side pinning.
108        :param key_pem: Private key (PEM) matching the DTLS certificate.
109        :param signaling_url: WebSocket URL of the signaling server.
110        :param local_ws_url: Same-host WebSocket URL of the Music Assistant API to bridge to.
111        :param sendspin_url: Internal Sendspin WebSocket URL to bridge to.
112        :param ice_servers: List of ICE server configurations (used at registration time).
113        :param ice_servers_callback: Optional callback to fetch fresh ICE servers for each session.
114        :param set_sendspin_player_callback: Callback to set sendspin player for a session.
115        """
116        self.http_session = http_session
117        self.signaling_url = signaling_url
118        self.local_ws_url = local_ws_url
119        self.sendspin_url = sendspin_url
120        self._remote_id = remote_id
121        self._cert_pem = cert_pem
122        self._key_pem = key_pem
123        self.logger = LOGGER
124        self._ice_servers_callback = ice_servers_callback
125        self._set_sendspin_player_callback = set_sendspin_player_callback
126
127        # Static ICE servers used at registration time (relayed to clients via signaling server)
128        self.ice_servers = ice_servers or self.DEFAULT_ICE_SERVERS
129
130        self.sessions: dict[str, WebRTCSession] = {}
131        self._background_tasks: set[asyncio.Task[None]] = set()
132        self._signaling_ws: aiohttp.ClientWebSocketResponse | None = None
133        self._running = False
134        self._reconnect_delay = 10  # Wait 10 seconds before reconnecting
135        self._max_reconnect_delay = 300  # Max 5 minutes between reconnects
136        self._current_reconnect_delay = 10
137        self._run_task: asyncio.Task[None] | None = None
138        self._is_connected = False
139        self._connecting = False
140        # gateway-wide by design: normally a single remote client is connected
141        self._http_proxy_semaphore = asyncio.Semaphore(HTTP_PROXY_CONCURRENCY)
142        self._chunk_group_seq = 0
143
144    @property
145    def is_running(self) -> bool:
146        """Return whether the gateway is running."""
147        return self._running
148
149    @property
150    def is_connected(self) -> bool:
151        """Return whether the gateway is connected to the signaling server."""
152        return self._is_connected
153
154    async def start(self) -> None:
155        """Start the WebRTC Gateway."""
156        if self._running:
157            self.logger.warning("WebRTC Gateway already running, skipping start")
158            return
159        # Failing candidates and permissions is how ICE converges: full chatter only at VERBOSE
160        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
161            rtc_log_level = LogLevel.VERBOSE
162        elif self.logger.isEnabledFor(logging.DEBUG):
163            rtc_log_level = LogLevel.WARNING
164            self.logger.addFilter(_BENIGN_NATIVE_NOISE_FILTER)
165        else:
166            rtc_log_level = LogLevel.ERROR
167        install_python_logger(self.logger, level=rtc_log_level)
168        self.logger.info("Starting WebRTC Gateway")
169        self.logger.debug("Signaling URL: %s", self.signaling_url)
170        self.logger.debug("Local WS URL: %s", self.local_ws_url)
171        self._running = True
172        self._run_task = asyncio.create_task(self._run())
173        self.logger.debug("WebRTC Gateway start task created")
174
175    async def stop(self) -> None:
176        """Stop the WebRTC Gateway."""
177        self.logger.info("Stopping WebRTC Gateway")
178        self.logger.removeFilter(_BENIGN_NATIVE_NOISE_FILTER)
179        self._running = False
180
181        # Close all sessions
182        for session_id in list(self.sessions.keys()):
183            await self._close_session(session_id)
184
185        # Close signaling connection gracefully
186        if self._signaling_ws and not self._signaling_ws.closed:
187            try:
188                await self._signaling_ws.close()
189            except Exception:
190                self.logger.debug("Error closing signaling WebSocket", exc_info=True)
191
192        # Cancel run task and wait for it to finish
193        if self._run_task and not self._run_task.done():
194            self._run_task.cancel()
195            with contextlib.suppress(asyncio.CancelledError):
196                await self._run_task
197
198        # Wait briefly for any in-progress connection to notice _running=False
199        if self._connecting:
200            await asyncio.sleep(0.1)
201
202        self._signaling_ws = None
203        self._connecting = False
204
205    async def _get_fresh_ice_servers(self) -> list[dict[str, Any]]:
206        """Get fresh ICE servers for a new WebRTC session."""
207        if self._ice_servers_callback:
208            try:
209                fresh_servers = await self._ice_servers_callback()
210                if fresh_servers:
211                    return fresh_servers
212            except Exception:
213                self.logger.exception("Failed to fetch fresh ICE servers, using cached servers")
214        return self.ice_servers
215
216    async def _run(self) -> None:
217        """Run the main loop with reconnection logic."""
218        self.logger.debug("WebRTC Gateway _run() loop starting")
219        while self._running:
220            should_reconnect = True
221            try:
222                should_reconnect = await self._connect_to_signaling()
223                # Connection closed gracefully or with error
224                self._is_connected = False
225                if self._running and should_reconnect:
226                    self.logger.warning(
227                        "Signaling server connection lost. Reconnecting in %ss...",
228                        self._current_reconnect_delay,
229                    )
230            except Exception:
231                self._is_connected = False
232                self.logger.exception("Signaling connection error")
233                if self._running:
234                    self.logger.info(
235                        "Reconnecting to signaling server in %ss",
236                        self._current_reconnect_delay,
237                    )
238
239            if self._running and should_reconnect:
240                await asyncio.sleep(self._current_reconnect_delay)
241                # Exponential backoff with max limit
242                self._current_reconnect_delay = min(
243                    self._current_reconnect_delay * 2, self._max_reconnect_delay
244                )
245            elif not should_reconnect:
246                # Connection was replaced by another instance, stop the run loop
247                self.logger.info("Connection replaced, stopping reconnection attempts")
248                self._running = False
249                break
250
251    async def _connect_to_signaling(self) -> bool:
252        """Connect to the signaling server."""
253        if self._connecting:
254            self.logger.warning("Already connecting to signaling server, skipping")
255            return False  # Don't trigger another reconnect cycle
256        self._connecting = True
257        close_code: int | None = None
258        self.logger.info("Connecting to signaling server: %s", self.signaling_url)
259        try:
260            self._signaling_ws = await self.http_session.ws_connect(
261                self.signaling_url,
262                heartbeat=35.0,  # Send ping every 35s (slightly above server's 30s interval)
263            )
264            # Check if we were stopped while connecting
265            if not self._running:
266                self.logger.debug("Gateway stopped during connection, closing WebSocket")
267                await self._signaling_ws.close()
268                self._signaling_ws = None
269                self._connecting = False
270                return False
271            self.logger.debug("WebSocket connection established, id=%s", id(self._signaling_ws))
272            self.logger.debug("Sending registration")
273            await self._register()
274            self._current_reconnect_delay = self._reconnect_delay
275            self.logger.debug("Registration sent, waiting for confirmation...")
276
277            # Run message loop and get close code
278            close_code = await self._signaling_message_loop(self._signaling_ws)
279
280            # Get close code from WebSocket if not already set from CLOSE message
281            if close_code is None:
282                close_code = self._signaling_ws.close_code
283            ws_exception = self._signaling_ws.exception()
284            self.logger.debug(
285                "Message loop exited - WebSocket closed: %s, close_code: %s, exception: %s",
286                self._signaling_ws.closed,
287                close_code,
288                ws_exception,
289            )
290        except TimeoutError:
291            self.logger.error("Timeout connecting to signaling server")
292        except aiohttp.ClientError as err:
293            self.logger.error("Failed to connect to signaling server: %s", err)
294        except Exception:
295            self.logger.exception("Unexpected error in signaling connection")
296        finally:
297            self._is_connected = False
298            self._connecting = False
299            self._signaling_ws = None
300
301        # Check if this connection was replaced by another one
302        if close_code == self.CLOSE_CODE_REPLACED:
303            self.logger.info("Connection was replaced by another instance - not reconnecting")
304            return False
305
306        return True
307
308    async def _signaling_message_loop(self, ws: aiohttp.ClientWebSocketResponse) -> int | None:
309        """Process messages from the signaling WebSocket."""
310        close_code: int | None = None
311        self.logger.debug("Entering message loop")
312        async for msg in ws:
313            if msg.type == aiohttp.WSMsgType.TEXT:
314                try:
315                    await self._handle_signaling_message(json.loads(msg.data))
316                except Exception:
317                    self.logger.exception("Error handling signaling message")
318            elif msg.type == aiohttp.WSMsgType.PING:
319                self.logger.log(VERBOSE_LOG_LEVEL, "Received WebSocket PING")
320            elif msg.type == aiohttp.WSMsgType.PONG:
321                self.logger.log(VERBOSE_LOG_LEVEL, "Received WebSocket PONG")
322            elif msg.type == aiohttp.WSMsgType.CLOSE:
323                close_code = msg.data
324                self.logger.warning(
325                    "Signaling server sent close frame: code=%s, reason=%s",
326                    msg.data,
327                    msg.extra,
328                )
329                break
330            elif msg.type == aiohttp.WSMsgType.CLOSED:
331                self.logger.warning("Signaling server closed connection")
332                break
333            elif msg.type == aiohttp.WSMsgType.ERROR:
334                self.logger.error("WebSocket error: %s", ws.exception())
335                break
336            else:
337                self.logger.warning("Unexpected WebSocket message type: %s", msg.type)
338        return close_code
339
340    async def _register(self) -> None:
341        """Register with the signaling server."""
342        if self._signaling_ws:
343            await self._signaling_ws.send_json(
344                {
345                    "type": "register-server",
346                    "remoteId": self._remote_id,
347                    "iceServers": self.ice_servers,
348                }
349            )
350
351    async def _handle_signaling_message(self, message: dict[str, Any]) -> None:
352        """Handle incoming signaling messages."""
353        msg_type = message.get("type")
354
355        if msg_type in ("ping", "pong"):
356            # Ignore JSON-level ping/pong messages - we use WebSocket protocol-level heartbeat
357            # The signaling server still sends these for backward compatibility with older clients
358            pass
359        elif msg_type == "registered":
360            self._is_connected = True
361            self.logger.info("Registered with signaling server")
362        elif msg_type == "error":
363            error_msg = message.get("error") or message.get("message", "Unknown error")
364            self.logger.error("Signaling server error: %s", error_msg)
365        elif msg_type == "client-connected":
366            session_id = message.get("sessionId")
367            if session_id:
368                await self._create_session(session_id)
369                # Send session-ready with fresh ICE servers for the client
370                fresh_ice_servers = await self._get_fresh_ice_servers()
371                if self._signaling_ws:
372                    await self._signaling_ws.send_json(
373                        {
374                            "type": "session-ready",
375                            "sessionId": session_id,
376                            "iceServers": fresh_ice_servers,
377                        }
378                    )
379        elif msg_type == "client-disconnected":
380            session_id = message.get("sessionId")
381            if session_id:
382                await self._close_session(session_id)
383        elif msg_type == "offer":
384            session_id = message.get("sessionId")
385            offer_data = message.get("data")
386            if session_id and offer_data:
387                await self._handle_offer(session_id, offer_data)
388        elif msg_type == "ice-candidate":
389            session_id = message.get("sessionId")
390            candidate_data = message.get("data")
391            if session_id and candidate_data:
392                await self._handle_ice_candidate(session_id, candidate_data)
393
394    async def _create_session(self, session_id: str) -> None:
395        """Create a new WebRTC session."""
396        session_ice_servers = await self._get_fresh_ice_servers()
397        config = RTCConfiguration(
398            ice_servers=self._build_ice_servers(session_ice_servers),
399            certificate_pem=self._cert_pem,
400            key_pem=self._key_pem,
401        )
402        pc = PeerConnection(config)
403        session = WebRTCSession(session_id=session_id, pc=pc)
404        self.sessions[session_id] = session
405
406        # PC-owned tasks are cancelled and awaited by pc.aclose() during teardown
407        pc.spawn_task(self._monitor_state(session))
408        pc.spawn_task(self._accept_channels(session))
409        pc.spawn_task(self._forward_local_candidates(session))
410
411    async def _handle_offer(self, session_id: str, offer: dict[str, Any]) -> None:
412        """Handle incoming WebRTC offer."""
413        session = self.sessions.get(session_id)
414        if not session:
415            return
416        pc = session.pc
417
418        if pc.closed:
419            return
420
421        sdp = offer.get("sdp")
422        sdp_type = offer.get("type")
423        if not sdp or not sdp_type:
424            self.logger.error("Invalid offer data: missing sdp or type")
425            return
426
427        try:
428            await pc.set_remote_description(str(sdp), "offer")
429
430            if session_id not in self.sessions or pc.closed:
431                return
432
433            # Trickle ICE: set_local_description returns as soon as the SDP is ready
434            # (candidate-less), so we answer immediately instead of blocking on
435            # create_answer() until ICE gathering completes (~20s+ with slow STUN/TURN).
436            # Local candidates are streamed separately by _forward_local_candidates.
437            answer = await pc.set_local_description("answer")
438
439            if session_id not in self.sessions or pc.closed:
440                return
441
442            if self._signaling_ws:
443                await self._signaling_ws.send_json(
444                    {
445                        "type": "answer",
446                        "sessionId": session_id,
447                        "data": {
448                            "sdp": answer.sdp,
449                            "type": answer.type,
450                        },
451                    }
452                )
453        except Exception:
454            self.logger.exception("Error handling offer for session %s", session_id)
455            # Clean up the session on error
456            await self._close_session(session_id)
457
458    async def _handle_ice_candidate(self, session_id: str, candidate: dict[str, Any]) -> None:
459        """Handle incoming ICE candidate."""
460        session = self.sessions.get(session_id)
461        if not session or not candidate:
462            return
463
464        pc = session.pc
465        if pc.closed:
466            return
467
468        candidate_str = candidate.get("candidate")
469        sdp_mid = candidate.get("sdpMid")
470
471        if not candidate_str:
472            return
473
474        # libdatachannel accepts both the browser's "candidate:..."-prefixed form and the
475        # bare form, so the string is forwarded as-is. add_remote_candidate buffers the
476        # candidate internally until the remote description is set.
477        mid = str(sdp_mid) if sdp_mid else ""
478        try:
479            await pc.add_remote_candidate(candidate_str, mid)
480        except Exception:
481            self.logger.exception("Failed to add ICE candidate for session %s", session_id)
482
483    async def _handle_http_proxy_request(
484        self, session: WebRTCSession, request_data: dict[str, Any]
485    ) -> None:
486        """Handle HTTP proxy request from remote client."""
487        request_id = request_data.get("id")
488        method = request_data.get("method", "GET")
489        path = request_data.get("path", "/")
490        headers = request_data.get("headers", {})
491
492        # Build local HTTP URL from the WebSocket URL.
493        # Handle both ws:// and wss:// schemes.
494        parsed = urlparse(self.local_ws_url)
495        http_scheme = "https" if parsed.scheme == "wss" else "http"
496        # Keep `path` as a URI path so an `@`/`//` prefix can't repoint the host (SSRF).
497        local_http_url = f"{http_scheme}://{parsed.netloc}/{path.lstrip('/')}"
498
499        self.logger.debug("HTTP proxy request: %s %s", method, local_http_url)
500
501        async with self._http_proxy_semaphore:
502            try:
503                # Use shared HTTP session for this request
504                # this dial never leaves the host: TLS verification would fail on the bind
505                # address, and an unfollowed redirect cannot take the unverified dial off-host
506                async with self.http_session.request(
507                    method, local_http_url, headers=headers, ssl=False, allow_redirects=False
508                ) as response:
509                    body = await response.read()
510                    await self._send_http_proxy_response(
511                        session, request_id, response.status, dict(response.headers), body
512                    )
513            except Exception as err:
514                self.logger.exception("Error handling HTTP proxy request")
515                await self._send_http_proxy_response(
516                    session, request_id, 500, {"Content-Type": "text/plain"}, str(err).encode()
517                )
518
519    async def _send_http_proxy_response(
520        self,
521        session: WebRTCSession,
522        request_id: str | None,
523        status: int,
524        headers: dict[str, str],
525        body: bytes,
526    ) -> None:
527        """Send an HTTP-proxy response over the ma-api channel."""
528        await self._send_ma_api(
529            session.data_channel,
530            json.dumps(
531                {
532                    "type": "http-proxy-response",
533                    "id": request_id,
534                    "status": status,
535                    "headers": headers,
536                    "body": body.hex(),
537                }
538            ),
539        )
540
541    async def _send_ma_api(self, channel: DataChannel | None, text: str) -> None:
542        """Send a text message on the ma-api channel, chunking it if it exceeds the size limit."""
543        data = text.encode()
544        if len(data) <= MA_API_CHUNK_SIZE:
545            await self._send_on_channel(channel, text)
546            return
547
548        # libdatachannel enforces the 256 KiB message limit, so oversized messages are split
549        # into base64 frames the client reassembles by group id (base64 keeps each frame's size
550        # predictable regardless of JSON escaping / unicode).
551        self._chunk_group_seq += 1
552        group_id = self._chunk_group_seq
553        count = (len(data) + MA_API_CHUNK_SIZE - 1) // MA_API_CHUNK_SIZE
554        for seq in range(count):
555            if channel is None or not channel.is_open:
556                return
557            piece = data[seq * MA_API_CHUNK_SIZE : (seq + 1) * MA_API_CHUNK_SIZE]
558            await self._send_on_channel(
559                channel,
560                json.dumps(
561                    {
562                        "type": "__chunk__",
563                        "id": group_id,
564                        "seq": seq,
565                        "count": count,
566                        "b64": base64.b64encode(piece).decode(),
567                    }
568                ),
569            )
570
571    async def _close_session(self, session_id: str) -> None:
572        """Close a WebRTC session."""
573        session = self.sessions.pop(session_id, None)
574        if not session:
575            return
576
577        # Close the local bridges first so no more data is fed through the channels
578        for local_ws in (session.local_ws, session.sendspin_ws):
579            if local_ws is not None and not local_ws.closed:
580                with contextlib.suppress(Exception):
581                    await local_ws.close()
582        session.local_ws = None
583        session.sendspin_ws = None
584
585        # aclose tears down all PC-owned pumps and every data channel; safe here because
586        # _close_session is never invoked from within a PC-owned task
587        await session.pc.aclose()
588
589    # ---- Session pumps (PC-owned tasks) --------------------------------------
590
591    async def _monitor_state(self, session: WebRTCSession) -> None:
592        """Close the session when its PeerConnection reports a failed state."""
593        async for event in session.pc.events():
594            if isinstance(event, StateChangeEvent) and event.state == RTCState.FAILED:
595                self._schedule_close(session.session_id)
596                return
597
598    async def _forward_local_candidates(self, session: WebRTCSession) -> None:
599        """Stream locally-gathered ICE candidates to the remote client (trickle ICE)."""
600        # The iterator ends on gathering-complete or when the PC closes.
601        async for candidate in session.pc.ice_candidates():
602            if session.session_id not in self.sessions or not self._signaling_ws:
603                return
604            await self._signaling_ws.send_json(
605                {
606                    "type": "ice-candidate",
607                    "sessionId": session.session_id,
608                    "data": {
609                        "candidate": candidate.candidate,
610                        "sdpMid": candidate.mid,
611                    },
612                }
613            )
614
615    async def _accept_channels(self, session: WebRTCSession) -> None:
616        """Accept incoming data channels and start their bridges."""
617        async for channel in session.pc.incoming_data_channels():
618            # The browser opens "ma-api" (default) and "sendspin" (by label); route each
619            if channel.label == "sendspin":
620                session.sendspin_channel = channel
621                session.pc.spawn_task(self._bridge_sendspin(session, channel))
622            else:
623                session.data_channel = channel
624                session.pc.spawn_task(self._bridge_ma_api(session, channel))
625
626    async def _bridge_ma_api(self, session: WebRTCSession, channel: DataChannel) -> None:
627        """Bridge the ma-api data channel to the local WebSocket API."""
628        # wait for the channel to open first, so the local WS's initial server_info (sent
629        # immediately on connect) isn't dropped by _send_on_channel while it's still opening
630        try:
631            await channel.wait_open()
632        except RTCError:
633            self._schedule_close(session.session_id)
634            return
635        try:
636            # Include session_id in URL so server can track WebRTC sessions
637            ws_url = f"{self.local_ws_url}?webrtc_session_id={session.session_id}"
638            # TLS verification would fail on the bind address and adds nothing to a dial
639            # that never leaves this host
640            session.local_ws = await self.http_session.ws_connect(ws_url, ssl=False)
641        except Exception:
642            self.logger.exception("Failed to connect to local WebSocket %s", self.local_ws_url)
643            channel.close()
644            self._schedule_close(session.session_id)
645            return
646
647        # from_local runs as its own PC-owned pump; this task drives channel -> local
648        session.pc.spawn_task(self._ma_api_from_local(session, channel))
649        await self._ma_api_to_local(session, channel)
650        # channel -> local loop ended (remote channel closed): tear down the session
651        self._schedule_close(session.session_id)
652
653    async def _ma_api_to_local(self, session: WebRTCSession, channel: DataChannel) -> None:
654        """Forward messages from the ma-api data channel to the local WebSocket."""
655        try:
656            async for message in channel:
657                if isinstance(message, str):
658                    # Check if this is an HTTP proxy request
659                    try:
660                        msg_data = json.loads(message)
661                        if (
662                            isinstance(msg_data, dict)
663                            and msg_data.get("type") == "http-proxy-request"
664                        ):
665                            # Handle off the receive loop so a slow image fetch never
666                            # blocks API messages or the next image (bounded by the
667                            # gateway-wide semaphore; see #4889).
668                            session.pc.spawn_task(
669                                self._handle_http_proxy_request(session, msg_data)
670                            )
671                            continue
672                    except json.JSONDecodeError, ValueError:
673                        pass
674
675                if session.local_ws and not session.local_ws.closed:
676                    if isinstance(message, bytes):
677                        await session.local_ws.send_bytes(message)
678                    else:
679                        await session.local_ws.send_str(message)
680        except ConnectionClosedError:
681            self.logger.debug("ma-api channel closed for session %s", session.session_id)
682        except asyncio.CancelledError:
683            raise
684        except Exception:
685            self.logger.exception("Error forwarding to local WebSocket")
686
687    async def _ma_api_from_local(self, session: WebRTCSession, channel: DataChannel) -> None:
688        """Forward messages from the local WebSocket to the ma-api data channel."""
689        local_ws = session.local_ws
690        if local_ws is None:
691            return
692        try:
693            async for msg in local_ws:
694                if msg.type == aiohttp.WSMsgType.TEXT:
695                    await self._send_ma_api(channel, msg.data)
696                elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED):
697                    break
698        except asyncio.CancelledError:
699            raise
700        except Exception:
701            self.logger.exception("Error forwarding from local WebSocket")
702        # the local WS closed: the ma-api session is unusable, so tear it down instead of
703        # leaving the client an open channel that silently drops messages
704        self._schedule_close(session.session_id)
705
706    async def _bridge_sendspin(self, session: WebRTCSession, channel: DataChannel) -> None:
707        """Bridge the sendspin data channel to the internal sendspin server."""
708        try:
709            session.sendspin_ws = await self.http_session.ws_connect(self.sendspin_url)
710            self.logger.debug("Sendspin channel connected for session %s", session.session_id)
711        except Exception:
712            self.logger.exception(
713                "Failed to connect sendspin channel to internal server for session %s",
714                session.session_id,
715            )
716            channel.close()
717            return
718
719        # from_local runs as its own PC-owned pump; this task drives channel -> internal
720        session.pc.spawn_task(self._sendspin_from_local(session, channel))
721        await self._sendspin_to_local(session, channel)
722        # channel -> internal loop ended (remote channel closed): tear down only this bridge
723        await self._close_sendspin_bridge(session)
724
725    async def _sendspin_to_local(self, session: WebRTCSession, channel: DataChannel) -> None:
726        """Forward messages from the sendspin data channel to the internal sendspin server."""
727        first_message = True
728        try:
729            async for message in channel:
730                # Check only the first message for client_id extraction
731                if first_message:
732                    first_message = False
733                    if isinstance(message, str):
734                        self._try_extract_sendspin_client_id(session, message)
735
736                if session.sendspin_ws and not session.sendspin_ws.closed:
737                    if isinstance(message, bytes):
738                        await session.sendspin_ws.send_bytes(message)
739                    else:
740                        await session.sendspin_ws.send_str(message)
741        except ConnectionClosedError:
742            self.logger.debug("Sendspin channel closed for session %s", session.session_id)
743        except asyncio.CancelledError:
744            raise
745        except Exception:
746            self.logger.exception("Error forwarding sendspin to local")
747
748    async def _sendspin_from_local(self, session: WebRTCSession, channel: DataChannel) -> None:
749        """Forward messages from the internal sendspin server to the sendspin data channel."""
750        sendspin_ws = session.sendspin_ws
751        if sendspin_ws is None:
752            return
753        try:
754            async for msg in sendspin_ws:
755                if msg.type in {aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY}:
756                    await self._send_on_channel(channel, msg.data)
757                elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED):
758                    break
759        except asyncio.CancelledError:
760            raise
761        except Exception:
762            self.logger.exception("Error forwarding sendspin from local")
763        # the internal sendspin WS closed: close only this bridge, leaving the ma-api session up
764        await self._close_sendspin_bridge(session)
765
766    async def _close_sendspin_bridge(self, session: WebRTCSession) -> None:
767        """Close the internal Sendspin bridge for a WebRTC session."""
768        sendspin_ws = session.sendspin_ws
769        session.sendspin_ws = None
770        if sendspin_ws is not None and not sendspin_ws.closed:
771            with contextlib.suppress(Exception):
772                await sendspin_ws.close()
773        channel = session.sendspin_channel
774        session.sendspin_channel = None
775        if channel is not None and not channel.closed:
776            with contextlib.suppress(Exception):
777                await channel.aclose()
778
779    # ---- Helpers -------------------------------------------------------------
780
781    def _build_ice_servers(self, servers: list[dict[str, Any]]) -> list[IceServer]:
782        """Build IceServer entries (one per url) for our own peer connection."""
783        ice_servers: list[IceServer] = []
784        skipped: list[str] = []
785        for server in servers:
786            urls = server.get("urls")
787            username = server.get("username")
788            credential = server.get("credential")
789            url_list = [urls] if isinstance(urls, str) else (urls or [])
790            for url in url_list:
791                if not _is_usable_ice_url(url):
792                    skipped.append(url)
793                    continue
794                ice_servers.append(IceServer(url=url, username=username, credential=credential))
795        if skipped:
796            self.logger.debug("Skipping ICE server urls unusable by libjuice: %s", skipped)
797        return ice_servers
798
799    def _schedule_close(self, session_id: str) -> None:
800        """Schedule session teardown on a gateway-owned task."""
801        # Run outside the PC-owned pump that triggered it so pc.aclose() (which the pump
802        # is awaited by) does not deadlock on itself
803        if session_id not in self.sessions:
804            return
805        task = asyncio.create_task(self._close_session(session_id))
806        self._background_tasks.add(task)
807        task.add_done_callback(self._background_tasks.discard)
808
809    async def _send_on_channel(self, channel: DataChannel | None, data: str | bytes) -> None:
810        """Send data on a data channel if it is currently open."""
811        if channel is None or not channel.is_open:
812            return
813        try:
814            await channel.send(data)
815        except ConnectionClosedError:
816            pass
817        except RTCError as err:
818            # a single failed send (e.g. an over-limit message) must not tear down the pump
819            self.logger.warning("Dropping %d-byte data channel message: %s", len(data), err)
820
821    def _try_extract_sendspin_client_id(self, session: WebRTCSession, message: str) -> None:
822        """Try to extract client_id from sendspin auth message and set on websocket client."""
823        try:
824            data = json.loads(message)
825            if data.get("type") != "auth":
826                return  # Not an auth message
827
828            # This is an auth message - extract client_id if present
829            if client_id := data.get("client_id"):
830                session.sendspin_player_id = client_id
831                self.logger.debug(
832                    "Extracted sendspin player %s for session %s",
833                    client_id,
834                    session.session_id,
835                )
836                # Use callback to set sendspin player on the websocket client
837                if self._set_sendspin_player_callback:
838                    self._set_sendspin_player_callback(session.session_id, client_id)
839        except json.JSONDecodeError, TypeError:
840            pass  # Not valid JSON, ignore
841
842
843class _BenignNativeNoiseFilter(logging.Filter):
844    """Drops known-benign native libdatachannel log lines."""
845
846    def filter(self, record: logging.LogRecord) -> bool:
847        """Return whether this log record is worth keeping."""
848        # Cloudflare omits the ERROR-CODE attribute, so libjuice warns on a benign refusal
849        return "TURN CreatePermission error response, code=0" not in record.getMessage()
850
851
852_BENIGN_NATIVE_NOISE_FILTER = _BenignNativeNoiseFilter()
853
854
855def _is_usable_ice_url(url: str) -> bool:
856    """
857    Return whether libdatachannel's ICE backend (libjuice) can use this ICE server url.
858
859    :param url: ICE server url, e.g. ``turn:turn.example.com:3478?transport=tcp``.
860    """
861    scheme, _, remainder = url.partition(":")
862    scheme = scheme.lower()
863    if scheme == "stun":
864        return True
865    if scheme not in ("turn", "turns"):
866        return False
867    # like rtc::IceServer url parsing, the transport parameter wins over the scheme
868    query = remainder.partition("?")[2].lower()
869    if "transport=udp" in query:
870        return True
871    return scheme == "turn" and not ("transport=tcp" in query or "transport=tls" in query)
872