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