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