music-assistant-server

1"""
2Controller that manages the builtin webserver that hosts the api and frontend.
3
4Unlike the streamserver (which is as simple and unprotected as possible),
5this webserver allows for more fine grained configuration to better secure it.
6"""
7
8from __future__ import annotations
9
10import asyncio
11import hashlib
12import html
13import inspect
14import os
15import urllib.parse
16from collections.abc import Awaitable, Callable
17from concurrent import futures
18from contextlib import aclosing
19from functools import partial
20from typing import TYPE_CHECKING, Any, Final, cast
21
22import aiofiles
23from aiohttp import web
24from mashumaro.exceptions import MissingField
25from music_assistant_frontend import where as locate_frontend
26from music_assistant_models.api import CommandMessage
27from music_assistant_models.auth import UserRole
28from music_assistant_models.config_entries import (
29    ConfigActionResult,
30    ConfigEntry,
31    ConfigValueOption,
32)
33from music_assistant_models.enums import ConfigEntryType
34from music_assistant_models.errors import (
35    InsufficientPermissions,
36    InvalidDataError,
37    UserNotFoundError,
38)
39from music_assistant_models.media_items.metadata import IMAGE_PROXY_ID_RESOLVER
40from music_assistant_models.translations import TRANSLATION_RESOLVER
41
42from music_assistant.constants import (
43    CONF_AUTH_ALLOW_SELF_REGISTRATION,
44    CONF_BIND_IP,
45    CONF_BIND_PORT,
46    CONF_VALUE_AUTO,
47    DEFAULT_HOST,
48    INGRESS_SERVER_PORT,
49    RESOURCES_DIR,
50    SENDSPIN_SERVER_PORT,
51    VERBOSE_LOG_LEVEL,
52    WILDCARD_BIND_IPS,
53)
54from music_assistant.controllers.webserver.helpers.ssl import (
55    create_server_ssl_context,
56    format_certificate_info,
57    verify_ssl_certificate,
58)
59from music_assistant.helpers.api import parse_arguments
60from music_assistant.helpers.json import json_dumps, json_loads
61from music_assistant.helpers.redirect_validation import (
62    build_code_redirect_url,
63    is_allowed_redirect_url,
64)
65from music_assistant.helpers.util import (
66    format_ip_for_url,
67    get_ip_addresses,
68    get_publish_ip_candidates,
69)
70from music_assistant.helpers.webserver import Webserver
71from music_assistant.models.core_controller import CoreController
72
73from .api_docs import generate_commands_json, generate_openapi_spec, generate_schemas_json
74from .auth import AuthenticationManager
75from .helpers.auth_middleware import (
76    get_authenticated_user,
77    has_scope,
78    is_request_from_ingress,
79    resolve_command_impersonation,
80    set_current_peer_address,
81    set_current_token,
82    set_current_user,
83    set_impersonated_user,
84)
85from .helpers.auth_providers import BuiltinLoginProvider, get_ha_user_role
86from .remote_access import RemoteAccessManager
87from .sendspin_proxy import SendspinProxyHandler
88from .websocket_client import WebsocketClientHandler
89
90if TYPE_CHECKING:
91    from music_assistant_models.config_entries import CoreConfig
92
93    from music_assistant import MusicAssistant
94    from music_assistant.helpers.api import APICommandHandler
95
96DEFAULT_SERVER_PORT = 8095
97CONF_BASE_URL = "base_url"
98CONF_ENABLE_SSL = "enable_ssl"
99CONF_SSL_CERTIFICATE = "ssl_certificate"
100CONF_SSL_PRIVATE_KEY = "ssl_private_key"
101CONF_ACTION_VERIFY_SSL = "verify_ssl"
102MAX_PENDING_MSG = 512
103CANCELLATION_ERRORS: Final = (asyncio.CancelledError, futures.CancelledError)
104
105
106def _get_publish_addresses(
107    bind_ip: str | None, publish_ip: str, publish_candidates: tuple[str, ...]
108) -> list[str]:
109    """
110    Return the IP addresses the webserver should publish/advertise.
111
112    :param bind_ip: The configured bind IP (None or a wildcard means all interfaces).
113    :param publish_ip: The resolved primary publish IP.
114    :param publish_candidates: Host addresses reachable from the local network, ranked.
115    """
116    addresses = [publish_ip]
117    if bind_ip and bind_ip not in WILDCARD_BIND_IPS:
118        return addresses
119    # bound to all interfaces: also publish the primary address of the other
120    # IP family (if any) so both IPv4-only and IPv6-only clients can connect
121    publish_is_ipv6 = ":" in publish_ip
122    for ip in publish_candidates:
123        if (":" in ip) != publish_is_ipv6:
124            addresses.append(ip)
125            break
126    return addresses
127
128
129def _get_internal_connect_ip(bind_ip: str | None, publish_ip: str) -> str:
130    """
131    Return the IP address to reach a server running on this host.
132
133    :param bind_ip: The server's configured bind IP (None or a wildcard means all interfaces).
134    :param publish_ip: The server's resolved publish IP.
135    """
136    if bind_ip and bind_ip not in WILDCARD_BIND_IPS:
137        # bound to one specific interface, so loopback would not reach the server
138        return bind_ip
139    # Use IPv6 loopback if publish_ip is IPv6 (indicates IPv6-only host)
140    return "::1" if ":" in publish_ip else "127.0.0.1"
141
142
143def _locale_from_request(request: web.Request) -> str | None:
144    """
145    Determine the UI locale for an HTTP request from the standard ``Accept-Language`` header.
146
147    Returns None when the header is absent, so the server falls back to the English source.
148
149    :param request: The aiohttp request.
150    """
151    header = request.headers.get("Accept-Language")
152    if not header:
153        return None
154    # take the first/highest-priority tag, dropping any quality factor ("nl-NL,nl;q=0.9" -> "nl-NL")
155    locale = header.split(",", 1)[0].split(";", 1)[0].strip()
156    return locale or None
157
158
159class WebserverController(CoreController):
160    """Core Controller that manages the builtin webserver that hosts the api and frontend."""
161
162    domain: str = "webserver"
163
164    def __init__(self, mass: MusicAssistant) -> None:
165        """Initialize instance."""
166        super().__init__(mass)
167        self._server = Webserver(self.logger, enable_dynamic_routes=True)
168        self.register_dynamic_route = self._server.register_dynamic_route
169        self.unregister_dynamic_route = self._server.unregister_dynamic_route
170        self.clients: set[WebsocketClientHandler] = set()
171        # the URL that the "auto" base_url setting resolves to, detected at setup
172        self._auto_base_url: str = ""
173        # whether SSL is switched on in the config, resolved at setup
174        self._ssl_configured: bool = False
175        # whether the webserver actually serves TLS, resolved at setup
176        self._ssl_active: bool = False
177        self.bind_ip: str | None = None
178        self.publish_addresses: list[str] = []
179        self.manifest.name = "Web Server (frontend and api)"
180        self.manifest.description = (
181            "The built-in webserver that hosts the Music Assistant Websockets API and frontend"
182        )
183        self.manifest.icon = "web-box"
184        self.auth = AuthenticationManager(self)
185        self.remote_access = RemoteAccessManager(self)
186        self._sendspin_proxy = SendspinProxyHandler(self)
187
188    @property
189    def base_url(self) -> str:
190        """Return the base_url for the webserver."""
191        config = getattr(self, "config", None)
192        if config is None:
193            return ""
194        base_url = str(config.get_value(CONF_BASE_URL) or CONF_VALUE_AUTO)
195        if base_url == CONF_VALUE_AUTO:
196            return self._auto_base_url
197        return base_url.removesuffix("/")
198
199    @property
200    def internal_base_url(self) -> str:
201        """Return the URL to reach this webserver's own API from this host."""
202        # the advertised address is not necessarily dialable here: a configured base URL
203        # routes out through DNS and a reverse proxy just to come back in, and a published
204        # IP need not exist on this host at all (e.g. a container or NAT setup), so derive
205        # the address from what the webserver actually binds to
206        connect_ip = _get_internal_connect_ip(self.bind_ip, self.publish_ip)
207        protocol = "https" if self._ssl_active else "http"
208        return f"{protocol}://{format_ip_for_url(connect_ip)}:{self.publish_port}"
209
210    @property
211    def internal_sendspin_url(self) -> str:
212        """Return the URL to reach the in-process Sendspin server from this host."""
213        # the advertised address is not necessarily dialable here (e.g. a container or
214        # NAT setup), so derive the address from what the Sendspin server actually binds to
215        connect_ip = _get_internal_connect_ip(
216            self.mass.streams.bind_ip, str(self.mass.streams.publish_ip)
217        )
218        return f"ws://{format_ip_for_url(connect_ip)}:{SENDSPIN_SERVER_PORT}/sendspin"
219
220    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
221        """Return all Config Entries for this core module (if any)."""
222        return await self._build_config_entries()
223
224    async def handle_config_action(
225        self, action: str
226    ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
227        """Handle a one-shot action button press and report its outcome."""
228        if action == CONF_ACTION_VERIFY_SSL:
229            # the certificate/key are read from the stored config, so they must be saved
230            # before verifying - the action no longer receives the (unsaved) form values
231            cert_info = await verify_ssl_certificate(
232                str(self.get_config_value(CONF_SSL_CERTIFICATE, "")),
233                str(self.get_config_value(CONF_SSL_PRIVATE_KEY, "")),
234            )
235            if not cert_info.is_valid:
236                # a result only ever reports success, so an unusable certificate must raise
237                raise InvalidDataError(
238                    f"Certificate verification failed: {cert_info.error_message}",
239                    translation_key="ssl_verification_failed",
240                    translation_args=[cert_info.error_message or ""],
241                    translation_owner=self.translation_owner,
242                )
243            return ConfigActionResult(message=format_certificate_info(cert_info))
244        return await super().handle_config_action(action)
245
246    async def setup(self, config: CoreConfig) -> None:  # noqa: PLR0915
247        """Async initialize of module."""
248        self.config = config
249        # work out all routes
250        routes: list[tuple[str, str, Callable[[web.Request], Awaitable[web.StreamResponse]]]] = []
251        # frontend routes
252        frontend_dir = locate_frontend()
253        for filename in next(os.walk(frontend_dir))[2]:
254            if filename.endswith(".py"):
255                continue
256            filepath = os.path.join(frontend_dir, filename)
257            handler = partial(self._server.serve_static, filepath)
258            routes.append(("GET", f"/{filename}", handler))
259        # add index (with onboarding check)
260        self._index_path = os.path.join(frontend_dir, "index.html")
261        routes.append(("GET", "/", self._handle_index))
262        routes.append(("HEAD", "/", self._handle_index))
263        # add logo
264        logo_path = str(RESOURCES_DIR.joinpath("logo.png"))
265        handler = partial(self._server.serve_static, logo_path)
266        routes.append(("GET", "/logo.png", handler))
267        # add common CSS for HTML resources
268        common_css_path = str(RESOURCES_DIR.joinpath("common.css"))
269        handler = partial(self._server.serve_static, common_css_path)
270        routes.append(("GET", "/resources/common.css", handler))
271        # add info
272        routes.append(("GET", "/info", self._handle_server_info))
273        routes.append(("OPTIONS", "/info", self._handle_cors_preflight))
274        # add websocket api
275        routes.append(("GET", "/ws", self._handle_ws_client))
276        # the canonical /imageproxy/<image_id> form is registered as a dynamic
277        # route on the webserver by MetaDataController.post_setup()
278        # also host the audio preview service
279        routes.append(("GET", "/preview", self.serve_preview_stream))
280        # add jsonrpc api
281        routes.append(("POST", "/api", self._handle_jsonrpc_api_command))
282        # add api documentation
283        routes.append(("GET", "/api-docs", self._handle_api_intro))
284        routes.append(("GET", "/api-docs/", self._handle_api_intro))
285        routes.append(("GET", "/api-docs/commands", self._handle_commands_reference))
286        routes.append(("GET", "/api-docs/commands/", self._handle_commands_reference))
287        routes.append(("GET", "/api-docs/commands.json", self._handle_commands_json))
288        routes.append(("GET", "/api-docs/schemas", self._handle_schemas_reference))
289        routes.append(("GET", "/api-docs/schemas/", self._handle_schemas_reference))
290        routes.append(("GET", "/api-docs/schemas.json", self._handle_schemas_json))
291        routes.append(("GET", "/api-docs/openapi.json", self._handle_openapi_spec))
292        routes.append(("GET", "/api-docs/swagger", self._handle_swagger_ui))
293        routes.append(("GET", "/api-docs/swagger/", self._handle_swagger_ui))
294        # add authentication routes
295        routes.append(("GET", "/login", self._handle_login_page))
296        routes.append(("POST", "/auth/login", self._handle_auth_login))
297        routes.append(("OPTIONS", "/auth/login", self._handle_cors_preflight))
298        routes.append(("POST", "/auth/logout", self._handle_auth_logout))
299        routes.append(("GET", "/auth/me", self._handle_auth_me))
300        routes.append(("PATCH", "/auth/me", self._handle_auth_me_update))
301        routes.append(("GET", "/auth/providers", self._handle_auth_providers))
302        routes.append(("GET", "/auth/authorize", self._handle_auth_authorize))
303        routes.append(("GET", "/auth/callback", self._handle_auth_callback))
304        # add first-time setup routes
305        routes.append(("GET", "/setup", self._handle_setup_page))
306        routes.append(("POST", "/setup", self._handle_setup))
307        # add sendspin proxy route (authenticated WebSocket proxy to internal sendspin server)
308        routes.append(("GET", "/sendspin", self._sendspin_proxy.handle_sendspin_proxy))
309        await self.auth.setup()
310        # start the webserver
311        if self.mass.running_as_hass_addon:
312            # if we're running on the HA supervisor we start an additional TCP site
313            # on the internal ("172.30.32.") IP for the HA ingress proxy - that address
314            # lives on a docker bridge, so it needs the unfiltered adapter list
315            all_ip_addresses = await get_ip_addresses(include_ipv6=True)
316            ingress_host = next(
317                (x for x in all_ip_addresses if x.startswith("172.30.32.")), all_ip_addresses[0]
318            )
319            ingress_tcp_site_params = (ingress_host, INGRESS_SERVER_PORT)
320        else:
321            ingress_tcp_site_params = None
322        port_value = config.get_value(CONF_BIND_PORT)
323        assert isinstance(port_value, int)
324        self.publish_port = port_value
325        bind_ip = cast("str | None", config.get_value(CONF_BIND_IP))
326        # Create SSL context if SSL is enabled
327        ssl_context = None
328        self._ssl_configured = bool(config.get_value(CONF_ENABLE_SSL, False))
329        if self._ssl_configured:
330            ssl_context = await create_server_ssl_context(
331                str(config.get_value(CONF_SSL_CERTIFICATE) or ""),
332                str(config.get_value(CONF_SSL_PRIVATE_KEY) or ""),
333                logger=self.logger,
334            )
335        # a missing or invalid certificate falls back to plain HTTP, so every URL we hand
336        # out must follow the context that was actually created, not the configured value
337        self._ssl_active = ssl_context is not None
338        protocol = "https" if self._ssl_active else "http"
339        publish_candidates = await get_publish_ip_candidates(include_ipv6=True)
340        self._resolve_publish_state(bind_ip, publish_candidates, protocol)
341
342        await self._server.setup(
343            bind_ip=bind_ip,
344            bind_port=self.publish_port,
345            static_routes=routes,
346            # add assets subdir as static_content
347            static_content=("/assets", os.path.join(frontend_dir, "assets"), "assets"),
348            ingress_tcp_site_params=ingress_tcp_site_params,
349            # Add mass object to app for use by the auth helpers
350            app_state={"mass": self.mass},
351            ssl_context=ssl_context,
352        )
353        # adopt what the server actually bound to: a configured port of 0 is only resolved
354        # by the OS at bind time and an unavailable bind IP falls back to all interfaces
355        self.publish_port = cast("int", self._server.port)
356        self._resolve_publish_state(self._server.bind_ip, publish_candidates, protocol)
357        base_url = self.base_url
358        # print a big fat message in the log where the webserver is running
359        # because this is a common source of issues for people with more complex setups
360        if not self.auth.has_users:
361            self.logger.warning(
362                "\n\n################################################################################\n"
363                "###                           SETUP REQUIRED                                 ###\n"
364                "################################################################################\n"
365                "\n"
366                "Music Assistant is running in setup mode.\n"
367                "Please complete the setup by visiting:\n"
368                "\n"
369                "    %s/setup\n"
370                "\n"
371                "################################################################################\n",
372                base_url,
373            )
374        else:
375            self.logger.info(
376                "\n"
377                "################################################################################\n"
378                "\n"
379                "Webserver available on: %s\n"
380                "\n"
381                "If this address is incorrect, see the documentation on how to configure\n"
382                "the Webserver in Settings --> System --> Webserver\n"
383                "\n"
384                "################################################################################\n",
385                base_url,
386            )
387
388        # Setup remote access after webserver is running
389        await self.remote_access.setup()
390
391    async def close(self) -> None:
392        """Cleanup on exit."""
393        await self.remote_access.close()
394        for client in set(self.clients):
395            await client.disconnect()
396        await self._server.close()
397        await self.auth.close()
398
399    def register_websocket_client(self, client: WebsocketClientHandler) -> None:
400        """Register a WebSocket client for tracking."""
401        self.clients.add(client)
402
403    def unregister_websocket_client(self, client: WebsocketClientHandler) -> None:
404        """Unregister a WebSocket client."""
405        self.clients.discard(client)
406
407    def disconnect_websockets_for_token(self, token_id: str) -> None:
408        """Disconnect all WebSocket clients using a specific token."""
409        for client in list(self.clients):
410            if hasattr(client, "_token_id") and client._token_id == token_id:
411                username = (
412                    client._authenticated_user.username if client._authenticated_user else "unknown"
413                )
414                self.logger.warning(
415                    "Disconnecting WebSocket client due to token revocation: %s",
416                    username,
417                )
418                client._cancel()
419
420    def disconnect_websockets_for_user(self, user_id: str) -> None:
421        """Disconnect all WebSocket clients for a specific user."""
422        for client in list(self.clients):
423            if (
424                hasattr(client, "_authenticated_user")
425                and client._authenticated_user
426                and client._authenticated_user.user_id == user_id
427            ):
428                self.logger.warning(
429                    "Disconnecting WebSocket client due to user action: %s",
430                    client._authenticated_user.username,
431                )
432                client._cancel()
433
434    def update_active_user_filters(
435        self,
436        user_id: str,
437        player_filter: list[str] | None = None,
438        provider_filter: list[str] | None = None,
439    ) -> None:
440        """
441        Apply updated access filters to the live sessions of a user.
442
443        Call this after the filters of a user were changed in the database, so the
444        change takes effect right away instead of only on the next connection.
445
446        :param user_id: ID of the user whose sessions must be updated.
447        :param player_filter: The new player filter, or None to leave it untouched.
448        :param provider_filter: The new provider filter, or None to leave it untouched.
449        """
450        for client in list(self.clients):
451            user = client._authenticated_user
452            if user is None or user.user_id != user_id:
453                continue
454            # updated in place: the connection's context holds this very object
455            if player_filter is not None:
456                user.player_filter[:] = player_filter
457            if provider_filter is not None:
458                user.provider_filter[:] = provider_filter
459            self.logger.debug("Updated the access filters of a live session of %s", user.username)
460
461    def set_sendspin_player_for_token(self, token: str, player_id: str) -> None:
462        """
463        Set the sendspin player_id on the websocket clients holding the given token.
464
465        This is called by the sendspin proxy when a client connects, allowing
466        the player controller to auto-whitelist the player for that session.
467        Party guests all share one guest account, so the token (one per guest
468        device) decides which sessions (all tabs of that browser) a web player
469        belongs to, not the user.
470
471        :param token: The access token the sendspin proxy authenticated with.
472        :param player_id: The sendspin player ID to set.
473        """
474        for client in list(self.clients):
475            if client._current_token != token:
476                continue
477            client._sendspin_player_id = player_id
478            self.logger.debug(
479                "Set sendspin player %s for websocket client of user %s",
480                player_id,
481                client._authenticated_user.username if client._authenticated_user else "unknown",
482            )
483
484    def set_sendspin_player_for_webrtc_session(self, session_id: str, player_id: str) -> None:
485        """
486        Set the sendspin player_id on a websocket client for a WebRTC session.
487
488        This is called by the WebRTC gateway when it extracts the client_id from
489        the sendspin auth message, allowing auto-whitelisting of the player.
490
491        :param session_id: The WebRTC session ID.
492        :param player_id: The sendspin player ID to set.
493        """
494        for client in list(self.clients):
495            if client._webrtc_session_id == session_id:
496                client._sendspin_player_id = player_id
497                username = (
498                    client._authenticated_user.username
499                    if client._authenticated_user
500                    else "unauthenticated"
501                )
502                self.logger.debug(
503                    "Set sendspin player %s for WebRTC session %s (user: %s)",
504                    player_id,
505                    session_id,
506                    username,
507                )
508                return
509
510    async def serve_preview_stream(self, request: web.Request) -> web.StreamResponse:
511        """Serve short preview sample."""
512        provider_instance_id_or_domain = request.query["provider"]
513        item_id = urllib.parse.unquote(request.query["item_id"])
514        resp = web.StreamResponse(status=200, reason="OK", headers={"Content-Type": "audio/aac"})
515        await resp.prepare(request)
516        preview_stream = self.mass.streams.get_preview_stream(
517            provider_instance_id_or_domain, item_id
518        )
519        # aclosing guarantees the preview stream (and the ffmpeg process behind it)
520        # is torn down immediately when the client disconnects, instead of lingering
521        # until garbage collection finalizes the abandoned generator.
522        async with aclosing(preview_stream):
523            async for chunk in preview_stream:
524                await resp.write(chunk)
525        return resp
526
527    def _resolve_publish_state(
528        self, bind_ip: str | None, publish_candidates: tuple[str, ...], protocol: str
529    ) -> None:
530        """
531        Resolve the addresses and base URL to advertise for the given bind address.
532
533        Reads ``self.publish_port``, so set that first.
534
535        :param bind_ip: Address the webserver binds to (None or a wildcard means all interfaces).
536        :param publish_candidates: Host addresses reachable from the local network, ranked.
537        :param protocol: URL scheme the webserver serves.
538        """
539        self.bind_ip = bind_ip
540        if bind_ip and bind_ip not in WILDCARD_BIND_IPS:
541            self.publish_ip = bind_ip
542        else:
543            self.publish_ip = publish_candidates[0]
544        self.publish_addresses = _get_publish_addresses(
545            bind_ip, self.publish_ip, publish_candidates
546        )
547        self._auto_base_url = (
548            f"{protocol}://{format_ip_for_url(self.publish_ip)}:{self.publish_port}"
549        )
550
551    async def _build_config_entries(self) -> tuple[ConfigEntry, ...]:
552        """Build this module's config entries."""
553        ip_addresses = await get_ip_addresses(include_ipv6=True)
554        return (
555            ConfigEntry(
556                key=CONF_AUTH_ALLOW_SELF_REGISTRATION,
557                type=ConfigEntryType.BOOLEAN,
558                default_value=True,
559                hidden=not any(provider.domain == "hass" for provider in self.mass.providers),
560                requires_reload=False,
561            ),
562            ConfigEntry(
563                key=CONF_BASE_URL,
564                type=ConfigEntryType.STRING,
565                default_value=CONF_VALUE_AUTO,
566                requires_reload=False,
567            ),
568            ConfigEntry(
569                key=CONF_BIND_PORT,
570                type=ConfigEntryType.INTEGER,
571                default_value=DEFAULT_SERVER_PORT,
572                requires_reload=True,
573            ),
574            # the two alerts are mutually exclusive: the generic one while SSL is switched off,
575            # and the SSL specific one when a certificate failed to load and left the webserver
576            # on plain HTTP
577            ConfigEntry(
578                key="webserver_warn",
579                type=ConfigEntryType.ALERT,
580                required=False,
581                hidden=self._ssl_configured,
582                depends_on=CONF_ENABLE_SSL,
583                depends_on_value=False,
584            ),
585            ConfigEntry(
586                key="ssl_inactive_warn",
587                type=ConfigEntryType.ALERT,
588                required=False,
589                hidden=not self._ssl_configured or self._ssl_active,
590                depends_on=CONF_ENABLE_SSL,
591            ),
592            ConfigEntry(
593                key=CONF_ENABLE_SSL,
594                type=ConfigEntryType.BOOLEAN,
595                default_value=False,
596                requires_reload=True,
597            ),
598            ConfigEntry(
599                key=CONF_SSL_CERTIFICATE,
600                type=ConfigEntryType.STRING,
601                required=False,
602                depends_on=CONF_ENABLE_SSL,
603                requires_reload=True,
604            ),
605            ConfigEntry(
606                key=CONF_SSL_PRIVATE_KEY,
607                type=ConfigEntryType.SECURE_STRING,
608                required=False,
609                depends_on=CONF_ENABLE_SSL,
610                requires_reload=True,
611            ),
612            ConfigEntry(
613                key=CONF_ACTION_VERIFY_SSL,
614                type=ConfigEntryType.ACTION,
615                action=CONF_ACTION_VERIFY_SSL,
616                depends_on=CONF_ENABLE_SSL,
617                required=False,
618            ),
619            ConfigEntry(
620                key=CONF_BIND_IP,
621                type=ConfigEntryType.STRING,
622                default_value=DEFAULT_HOST,
623                options=[ConfigValueOption(x, title=x) for x in {DEFAULT_HOST, *ip_addresses}],
624                category="generic",
625                advanced=True,
626                requires_reload=True,
627            ),
628        )
629
630    async def _handle_cors_preflight(self, request: web.Request) -> web.Response:
631        """Handle CORS preflight OPTIONS request."""
632        return web.Response(
633            status=200,
634            headers={
635                "Access-Control-Allow-Origin": "*",
636                "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
637                "Access-Control-Allow-Headers": "Content-Type, Authorization",
638                "Access-Control-Max-Age": "86400",  # Cache preflight for 24 hours
639            },
640        )
641
642    async def _handle_server_info(self, request: web.Request) -> web.Response:
643        """Handle request for server info."""
644        server_info = self.mass.get_server_info()
645        # Add CORS headers to allow frontend to call from any origin
646        return web.json_response(
647            server_info.to_dict(),
648            headers={
649                "Access-Control-Allow-Origin": "*",
650                "Access-Control-Allow-Methods": "GET, OPTIONS",
651                "Access-Control-Allow-Headers": "Content-Type, Authorization",
652            },
653        )
654
655    async def _handle_ws_client(self, request: web.Request) -> web.WebSocketResponse:
656        connection = WebsocketClientHandler(self, request)
657        if lang := request.headers.get("Accept-Language"):
658            self.mass.metadata.set_default_preferred_language(lang.split(",")[0])
659        try:
660            self.clients.add(connection)
661            return await connection.handle_client()
662        finally:
663            self.clients.discard(connection)
664
665    async def _handle_jsonrpc_api_command(self, request: web.Request) -> web.Response:
666        """Handle incoming JSON RPC API command."""
667        # These requests carry no connection identity, so the peer address is all an
668        # unauthenticated handler has to tell one caller apart from another.
669        set_current_peer_address(request.remote)
670        # Fail early if we don't have any users yet
671        if not self.auth.has_users:
672            return web.Response(status=503, text="Setup required")
673        if not request.can_read_body:
674            return web.Response(status=400, text="Body required")
675        cmd_data = await request.read()
676        self.logger.log(VERBOSE_LOG_LEVEL, "Received on JSONRPC API: %s", cmd_data)
677        try:
678            command_msg = CommandMessage.from_json(cmd_data)
679        except ValueError:
680            error = f"Invalid JSON: {cmd_data.decode()}"
681            self.logger.error("Unhandled JSONRPC API error: %s", error)
682            return web.Response(status=400, text=error)
683        except MissingField as e:
684            # be forgiving if message_id is missing
685            cmd_data_dict = json_loads(cmd_data)
686            if e.field_name == "message_id" and "command" in cmd_data_dict:
687                cmd_data_dict["message_id"] = "unknown"
688                command_msg = CommandMessage.from_dict(cmd_data_dict)
689            else:
690                error = f"Missing field in JSON: {e.field_name}"
691                self.logger.error("Unhandled JSONRPC API error: %s", error)
692                return web.Response(status=400, text="Invalid JSON: missing required field")
693
694        # work out handler for the given path/command
695        handler = self.mass.command_handlers.get(command_msg.command)
696        if handler is None:
697            error = f"Invalid Command: {command_msg.command}"
698            self.logger.error("Unhandled JSONRPC API error: %s", error)
699            return web.Response(status=400, text=error)
700
701        # Check authentication if required
702        if error_response := await self._authenticate_api_command(request, handler):
703            return error_response
704
705        try:
706            # handle the optional impersonation argument for impersonation-enabled commands
707            if handler.allow_impersonation and command_msg.args:
708                if impersonation_user := await resolve_command_impersonation(
709                    self.mass, command_msg.args
710                ):
711                    set_impersonated_user(impersonation_user)
712            args = parse_arguments(handler.signature, handler.type_hints, command_msg.args)
713            result: Any = handler.target(**args)
714            if hasattr(result, "__anext__"):
715                # handle async generator (for really large listings)
716                result = [item async for item in result]
717            elif inspect.iscoroutine(result):
718                result = await result
719            # Determine the UI locale for this request from the HTTP headers and warm it up
720            # so localized strings can be injected during dict serialization without disk I/O.
721            locale = _locale_from_request(request)
722            await self.mass.translations.ensure_locale_loaded(locale)
723            return self._localized_json_response(result, locale)
724        except InsufficientPermissions as e:
725            return web.Response(status=403, text=str(e))
726        except (InvalidDataError, UserNotFoundError) as e:
727            return web.Response(status=400, text=str(e))
728        except Exception as e:
729            # Return clean error message without stacktrace
730            error_type = type(e).__name__
731            error_msg = str(e)
732            error = f"{error_type}: {error_msg}"
733            self.logger.exception("Error executing command %s: %s", command_msg.command, error)
734            return web.Response(status=500, text="Internal server error")
735
736    async def _authenticate_api_command(
737        self, request: web.Request, handler: APICommandHandler
738    ) -> web.Response | None:
739        """
740        Authenticate the request and check the handler's required scope.
741
742        Sets the authenticated user in context and returns an error response
743        if authentication or the scope check failed, None otherwise.
744        """
745        if not (handler.authenticated or handler.required_scope):
746            return None
747        try:
748            user = await get_authenticated_user(request)
749        except Exception as e:
750            self.logger.exception("Authentication error: %s", e)
751            return web.Response(
752                status=401,
753                text="Authentication failed",
754                headers={"WWW-Authenticate": 'Bearer realm="Music Assistant"'},
755            )
756
757        if not user:
758            return web.Response(
759                status=401,
760                text="Authentication required",
761                headers={"WWW-Authenticate": 'Bearer realm="Music Assistant"'},
762            )
763
764        # Set user and token in context and check the required scope
765        set_current_user(user)
766        auth_header = request.headers.get("Authorization", "")
767        if auth_header.lower().startswith("bearer "):
768            set_current_token(auth_header[7:])
769        if handler.required_scope and not has_scope(user, handler.required_scope):
770            return web.Response(
771                status=403,
772                text=f"This command requires the {handler.required_scope} scope",
773            )
774        return None
775
776    def _localized_json_response(self, result: Any, locale: str | None) -> web.Response:
777        """
778        Serialize a command result to a JSON response with the per-request resolvers bound.
779
780        Sets the image-proxy resolver (for ``proxy_id`` injection) and the translation
781        resolver (to localize human-readable fields) for the given locale during dict
782        serialization, then resets them.
783        """
784        token = IMAGE_PROXY_ID_RESOLVER.set(self.mass.metadata.compute_image_id)
785        token_loc = TRANSLATION_RESOLVER.set(
786            partial(self.mass.translations.get_translation, locale=locale)
787        )
788        try:
789            return web.json_response(result, dumps=json_dumps)
790        finally:
791            IMAGE_PROXY_ID_RESOLVER.reset(token)
792            TRANSLATION_RESOLVER.reset(token_loc)
793
794    async def _handle_api_intro(self, request: web.Request) -> web.Response:
795        """Handle request for API introduction/documentation page."""
796        intro_html_path = str(RESOURCES_DIR.joinpath("api_docs.html"))
797        # Read the template
798        async with aiofiles.open(intro_html_path) as f:
799            html_content = await f.read()
800
801        # Replace placeholders (escape values to prevent XSS)
802        html_content = html_content.replace("{VERSION}", html.escape(self.mass.version))
803        html_content = html_content.replace("{BASE_URL}", html.escape(self.base_url))
804        html_content = html_content.replace("{SERVER_HOST}", html.escape(request.host))
805
806        return web.Response(text=html_content, content_type="text/html")
807
808    async def _handle_openapi_spec(self, request: web.Request) -> web.Response:
809        """Handle request for OpenAPI specification (generated on-the-fly)."""
810        spec = generate_openapi_spec(
811            self.mass.command_handlers, server_url=self.base_url, version=self.mass.version
812        )
813        return web.json_response(spec)
814
815    async def _handle_commands_reference(self, request: web.Request) -> web.FileResponse:
816        """Handle request for commands reference page."""
817        commands_html_path = str(RESOURCES_DIR.joinpath("commands_reference.html"))
818        return await self._server.serve_static(commands_html_path, request)
819
820    async def _handle_commands_json(self, request: web.Request) -> web.Response:
821        """Handle request for commands JSON data (generated on-the-fly)."""
822        commands_data = generate_commands_json(self.mass.command_handlers)
823        return web.json_response(commands_data)
824
825    async def _handle_schemas_reference(self, request: web.Request) -> web.FileResponse:
826        """Handle request for schemas reference page."""
827        schemas_html_path = str(RESOURCES_DIR.joinpath("schemas_reference.html"))
828        return await self._server.serve_static(schemas_html_path, request)
829
830    async def _handle_schemas_json(self, request: web.Request) -> web.Response:
831        """Handle request for schemas JSON data (generated on-the-fly)."""
832        schemas_data = generate_schemas_json(self.mass.command_handlers)
833        return web.json_response(schemas_data)
834
835    async def _handle_swagger_ui(self, request: web.Request) -> web.FileResponse:
836        """Handle request for Swagger UI."""
837        swagger_html_path = str(RESOURCES_DIR.joinpath("swagger_ui.html"))
838        return await self._server.serve_static(swagger_html_path, request)
839
840    async def _render_error_page(self, error_message: str, status: int = 403) -> web.Response:
841        """
842        Render a user-friendly error page with the given message.
843
844        :param error_message: The error message to display to the user.
845        :param status: HTTP status code for the response.
846        """
847        error_html_path = str(RESOURCES_DIR.joinpath("error.html"))
848        async with aiofiles.open(error_html_path) as f:
849            html_content = await f.read()
850        # Replace placeholder with the actual error message (escape to prevent XSS)
851        html_content = html_content.replace("{{ERROR_MESSAGE}}", html.escape(error_message))
852        return web.Response(text=html_content, content_type="text/html", status=status)
853
854    async def _handle_index(self, request: web.Request) -> web.StreamResponse:
855        """Handle request for index page (Vue frontend)."""
856        is_ingress_request = is_request_from_ingress(request)
857
858        if (not self.auth.has_users or not self.mass.config.onboard_done) and is_ingress_request:
859            # a non-admin user tries to access the index via HA ingress
860            # while we're not yet onboarded, prevent that as it leads to a bad UX
861            ingress_user_id = request.headers.get("X-Remote-User-ID", "")
862            role = await get_ha_user_role(self.mass, ingress_user_id)
863            if role != UserRole.ADMIN:
864                return await self._render_error_page(
865                    "Administrator permissions are required to complete the initial setup. "
866                    "Please ask a Home Assistant administrator to complete the setup first."
867                )
868            # NOTE: For ingress admin user,
869            # we allow access to index, user will be auto created and then forwarded to the
870            # frontend (which will take care of onboarding)
871
872        if not self.auth.has_users and not is_ingress_request:
873            # non ingress request and no users yet, redirect to setup
874            return web.Response(status=302, headers={"Location": "setup"})
875
876        # Serve the Vue frontend index.html
877        return await self._server.serve_static(self._index_path, request)
878
879    async def _handle_login_page(self, request: web.Request) -> web.Response:
880        """Handle request for login page (external client OAuth callback scenario)."""
881        if not self.auth.has_users:
882            # not yet onboarded (no first admin user exists), redirect to setup
883            return_url = request.query.get("return_url", "")
884            device_name = request.query.get("device_name", "")
885            setup_url = (
886                f"/setup?return_url={return_url}&device_name={device_name}"
887                if return_url
888                else "/setup"
889            )
890            return web.Response(status=302, headers={"Location": setup_url})
891        # Serve login page for external clients
892        login_html_path = str(RESOURCES_DIR.joinpath("login.html"))
893        async with aiofiles.open(login_html_path) as f:
894            html_content = await f.read()
895        return web.Response(text=html_content, content_type="text/html")
896
897    async def _handle_auth_login(self, request: web.Request) -> web.Response:
898        """Handle login request."""
899        # Block until onboarding is complete
900        if not self.auth.has_users:
901            return web.json_response(
902                {"success": False, "error": "Setup required"},
903                status=403,
904                headers={
905                    "Access-Control-Allow-Origin": "*",
906                    "Access-Control-Allow-Methods": "POST, OPTIONS",
907                    "Access-Control-Allow-Headers": "Content-Type, Authorization",
908                },
909            )
910
911        try:
912            if not request.can_read_body:
913                return web.Response(status=400, text="Body required")
914
915            body = await request.json()
916            provider_id = body.get("provider_id", "builtin")  # Default to built-in provider
917            credentials = body.get("credentials", {})
918            return_url = body.get("return_url")  # Optional return URL for redirect after login
919
920            # Authenticate with provider
921            auth_result = await self.auth.authenticate_with_credentials(provider_id, credentials)
922
923            if not auth_result.success or not auth_result.user:
924                return web.json_response(
925                    {"success": False, "error": auth_result.error},
926                    status=401,
927                    headers={
928                        "Access-Control-Allow-Origin": "*",
929                        "Access-Control-Allow-Methods": "POST, OPTIONS",
930                        "Access-Control-Allow-Headers": "Content-Type, Authorization",
931                    },
932                )
933
934            # Create token for user
935            device_name = body.get(
936                "device_name", f"{request.headers.get('User-Agent', 'Unknown')[:50]}"
937            )
938            token = await self.auth.create_token(auth_result.user, device_name)
939
940            # Prepare response data
941            response_data = {
942                "success": True,
943                "token": token,
944                "user": auth_result.user.to_dict(),
945            }
946
947            # If return_url provided, append code parameter and return as redirect_to
948            if return_url:
949                # SECURITY FIX (GHSA-j369-4c4w-7qmq): only forward the token to trusted
950                # destinations. is_allowed_redirect_url returns (True, "external") for any
951                # unknown external URL, so checking is_valid alone would still leak the JWT.
952                # Unlike _handle_auth_authorize/_handle_auth_callback, this endpoint appends
953                # the token immediately with no consent step, so "external" must be rejected.
954                _, category = is_allowed_redirect_url(return_url, request, self.base_url)
955                if category != "trusted":
956                    return web.Response(status=400, text="Invalid return_url")
957
958                redirect_url = build_code_redirect_url(return_url, token)
959
960                response_data["redirect_to"] = redirect_url
961                self.logger.debug(
962                    "Login successful, returning redirect_to: %s",
963                    redirect_url.replace(token, "***TOKEN***"),
964                )
965
966            # Add CORS headers to allow login from any origin
967            return web.json_response(
968                response_data,
969                headers={
970                    "Access-Control-Allow-Origin": "*",
971                    "Access-Control-Allow-Methods": "POST, OPTIONS",
972                    "Access-Control-Allow-Headers": "Content-Type, Authorization",
973                },
974            )
975        except Exception:
976            self.logger.exception("Error during login")
977            return web.json_response(
978                {"success": False, "error": "Login failed"},
979                status=500,
980                headers={
981                    "Access-Control-Allow-Origin": "*",
982                    "Access-Control-Allow-Methods": "POST, OPTIONS",
983                    "Access-Control-Allow-Headers": "Content-Type, Authorization",
984                },
985            )
986
987    async def _handle_auth_logout(self, request: web.Request) -> web.Response:
988        """Handle logout request."""
989        user = await get_authenticated_user(request)
990        if not user:
991            return web.Response(status=401, text="Not authenticated")
992
993        # Get token from request
994        auth_header = request.headers.get("Authorization", "")
995        if auth_header.startswith("Bearer "):
996            token = auth_header[7:]
997            # Find and revoke the token
998            token_hash = hashlib.sha256(token.encode()).hexdigest()
999            token_row = await self.auth.database.get_row("auth_tokens", {"token_hash": token_hash})
1000            if token_row:
1001                await self.auth.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
1002
1003        return web.json_response({"success": True})
1004
1005    async def _handle_auth_me(self, request: web.Request) -> web.Response:
1006        """Handle request for current user information."""
1007        user = await get_authenticated_user(request)
1008        if not user:
1009            return web.Response(status=401, text="Not authenticated")
1010
1011        return web.json_response(user.to_dict())
1012
1013    async def _handle_auth_me_update(self, request: web.Request) -> web.Response:
1014        """Handle request to update current user's profile."""
1015        user = await get_authenticated_user(request)
1016        if not user:
1017            return web.Response(status=401, text="Not authenticated")
1018
1019        try:
1020            if not request.can_read_body:
1021                return web.Response(status=400, text="Body required")
1022
1023            body = await request.json()
1024            username = body.get("username")
1025            display_name = body.get("display_name")
1026            avatar_url = body.get("avatar_url")
1027
1028            # Update user
1029            updated_user = await self.auth.update_user(
1030                user,
1031                username=username,
1032                display_name=display_name,
1033                avatar_url=avatar_url,
1034            )
1035
1036            return web.json_response({"success": True, "user": updated_user.to_dict()})
1037        except Exception:
1038            self.logger.exception("Error updating user profile")
1039            return web.json_response(
1040                {"success": False, "error": "Failed to update profile"}, status=500
1041            )
1042
1043    async def _handle_auth_providers(self, request: web.Request) -> web.Response:
1044        """Handle request for available login providers."""
1045        try:
1046            providers = await self.auth.get_login_providers()
1047            return web.json_response(providers)
1048        except Exception:
1049            self.logger.exception("Error getting auth providers")
1050            return web.json_response({"error": "Failed to get auth providers"}, status=500)
1051
1052    async def _handle_auth_authorize(self, request: web.Request) -> web.Response:
1053        """Handle OAuth authorization request."""
1054        try:
1055            provider_id = request.query.get("provider_id")
1056            return_url = request.query.get("return_url")
1057
1058            self.logger.debug(
1059                "OAuth authorize request: provider_id=%s, return_url=%s", provider_id, return_url
1060            )
1061
1062            if not provider_id:
1063                return web.Response(status=400, text="provider_id required")
1064
1065            # Validate return_url if provided
1066            if return_url:
1067                is_valid, _ = is_allowed_redirect_url(return_url, request, self.base_url)
1068                if not is_valid:
1069                    return web.Response(status=400, text="Invalid return_url")
1070
1071            auth_url = await self.auth.get_authorization_url(provider_id, return_url)
1072            if not auth_url:
1073                return web.Response(
1074                    status=400, text="Provider does not support OAuth or is not configured"
1075                )
1076
1077            return web.json_response({"authorization_url": auth_url})
1078        except Exception:
1079            self.logger.exception("Error during OAuth authorization")
1080            return web.json_response({"error": "Authorization failed"}, status=500)
1081
1082    async def _handle_auth_callback(self, request: web.Request) -> web.Response:
1083        """Handle OAuth callback."""
1084        try:
1085            code = request.query.get("code")
1086            state = request.query.get("state")
1087            provider_id = request.query.get("provider_id")
1088
1089            if not code or not state or not provider_id:
1090                return web.Response(status=400, text="code, state, and provider_id required")
1091
1092            redirect_uri = f"{self.base_url}/auth/callback?provider_id={provider_id}"
1093            auth_result = await self.auth.handle_oauth_callback(
1094                provider_id, code, state, redirect_uri
1095            )
1096
1097            if not auth_result.success or not auth_result.user:
1098                # Return error page
1099                error_html = f"""
1100                <html>
1101                <body>
1102                    <h1>Authentication Failed</h1>
1103                    <p>{html.escape(auth_result.error or "Unknown error")}</p>
1104                    <a href="/login">Back to Login</a>
1105                </body>
1106                </html>
1107                """
1108                return web.Response(text=error_html, content_type="text/html", status=400)
1109
1110            # Create token
1111            device_name = f"OAuth ({provider_id})"
1112            token = await self.auth.create_token(auth_result.user, device_name)
1113
1114            # Determine redirect URL (use return_url from OAuth flow or default to root)
1115            final_redirect_url = auth_result.return_url or "/"
1116            requires_consent = False
1117
1118            # Validate redirect URL for security
1119            if auth_result.return_url:
1120                is_valid, category = is_allowed_redirect_url(
1121                    auth_result.return_url, request, self.base_url
1122                )
1123                if not is_valid:
1124                    self.logger.warning("Invalid return_url blocked: %s", auth_result.return_url)
1125                    final_redirect_url = "/"
1126                elif category == "external":
1127                    # External domain - require user consent
1128                    requires_consent = True
1129            final_redirect_url = build_code_redirect_url(final_redirect_url, token)
1130
1131            # Load OAuth callback success page template and inject token and redirect URL
1132            oauth_callback_html_path = str(RESOURCES_DIR.joinpath("oauth_callback.html"))
1133            async with aiofiles.open(oauth_callback_html_path) as f:
1134                success_html = await f.read()
1135
1136            # Replace the redirect last so its untrusted contents cannot match another placeholder.
1137            success_html = success_html.replace(
1138                "{REQUIRES_CONSENT}", "true" if requires_consent else "false"
1139            )
1140            success_html = success_html.replace("{TOKEN}", _serialize_script_value(token))
1141            success_html = success_html.replace(
1142                "{REDIRECT_URL}", _serialize_script_value(final_redirect_url)
1143            )
1144
1145            return web.Response(text=success_html, content_type="text/html")
1146        except Exception:
1147            self.logger.exception("Error during OAuth callback")
1148            error_html = """
1149            <html>
1150            <body>
1151                <h1>Authentication Failed</h1>
1152                <p>An error occurred during authentication</p>
1153                <a href="/login">Back to Login</a>
1154            </body>
1155            </html>
1156            """
1157            return web.Response(text=error_html, content_type="text/html", status=500)
1158
1159    async def _handle_setup_page(self, request: web.Request) -> web.Response:
1160        """Handle request for first-time setup page."""
1161        # Setup forwards the admin token here with no consent step, so require a trusted destination.
1162        return_url = request.query.get("return_url")
1163        if return_url:
1164            _, category = is_allowed_redirect_url(return_url, request, self.base_url)
1165            if category != "trusted":
1166                return web.Response(status=400, text="Invalid return_url")
1167
1168        if self.auth.has_users:
1169            # this should not happen, but guard anyways
1170            return await self._render_error_page("Setup has already been completed.")
1171
1172        setup_html_path = str(RESOURCES_DIR.joinpath("setup.html"))
1173        async with aiofiles.open(setup_html_path) as f:
1174            html_content = await f.read()
1175
1176        return web.Response(text=html_content, content_type="text/html")
1177
1178    async def _handle_setup(self, request: web.Request) -> web.Response:
1179        """Handle first-time setup request to create admin user (non-ingress only)."""
1180        if self.auth.has_users:
1181            return web.json_response(
1182                {"success": False, "error": "Setup already completed"}, status=400
1183            )
1184
1185        if not request.can_read_body:
1186            return web.Response(status=400, text="Body required")
1187
1188        body = await request.json()
1189        username = body.get("username", "").strip()
1190        password = body.get("password", "")
1191
1192        # Validation
1193        if not username or len(username) < 2:
1194            return web.json_response(
1195                {"success": False, "error": "Username must be at least 2 characters"}, status=400
1196            )
1197
1198        if not password or len(password) < 8:
1199            return web.json_response(
1200                {"success": False, "error": "Password must be at least 8 characters"}, status=400
1201            )
1202
1203        try:
1204            builtin_provider = self.auth.login_providers.get("builtin")
1205            if not builtin_provider:
1206                return web.json_response(
1207                    {"success": False, "error": "Built-in auth provider not available"},
1208                    status=500,
1209                )
1210
1211            if not isinstance(builtin_provider, BuiltinLoginProvider):
1212                return web.json_response(
1213                    {"success": False, "error": "Built-in provider configuration error"},
1214                    status=500,
1215                )
1216
1217            # Create admin user with password
1218            user = await builtin_provider.create_user_with_password(
1219                username, password, role=UserRole.ADMIN
1220            )
1221
1222            # Create token for the new admin
1223            device_name = body.get(
1224                "device_name", f"Setup ({request.headers.get('User-Agent', 'Unknown')[:50]})"
1225            )
1226            token = await self.auth.create_token(user, device_name)
1227
1228            self.logger.info("First admin user created: %s", username)
1229
1230            # Return token - frontend will complete onboarding via config/onboard_complete
1231            response_data: dict[str, Any] = {
1232                "success": True,
1233                "token": token,
1234                "user": user.to_dict(),
1235            }
1236
1237            # Only forward the token to a trusted destination (no consent step here).
1238            return_url = body.get("return_url")
1239            if return_url and isinstance(return_url, str):
1240                _, category = is_allowed_redirect_url(return_url, request, self.base_url)
1241                if category == "trusted":
1242                    response_data["redirect_to"] = build_code_redirect_url(
1243                        return_url, token, {"onboard": "true"}
1244                    )
1245                else:
1246                    self.logger.warning("Ignoring untrusted setup return_url: %s", return_url)
1247
1248            return web.json_response(response_data)
1249
1250        except Exception as e:
1251            self.logger.exception("Error during setup")
1252            return web.json_response(
1253                {"success": False, "error": f"Setup failed: {e!s}"}, status=500
1254            )
1255
1256
1257def _serialize_script_value(value: str) -> str:
1258    """Serialize a string for use inside an HTML script element."""
1259    return (
1260        json_dumps(value)
1261        .replace("&", "\\u0026")
1262        .replace("<", "\\u003c")
1263        .replace(">", "\\u003e")
1264        .replace("\u2028", "\\u2028")
1265        .replace("\u2029", "\\u2029")
1266    )
1267