music-assistant-server

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