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