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