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