music-assistant-server

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