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