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