/
/
/
1"""Authentication helpers for HTTP requests and WebSocket connections."""
2
3from __future__ import annotations
4
5import logging
6from collections.abc import Mapping
7from contextvars import ContextVar
8from types import TracebackType
9from typing import TYPE_CHECKING, Any, Final, Self, cast
10
11from music_assistant_models.auth import AuthProviderType, Scope, User, UserRole
12from music_assistant_models.errors import (
13 InsufficientPermissions,
14 InvalidDataError,
15 UserNotFoundError,
16)
17
18from music_assistant.constants import HOMEASSISTANT_SYSTEM_USER, MASS_LOGGER_NAME, VERBOSE_LOG_LEVEL
19
20from .auth_providers import get_ha_user_details, get_ha_user_role
21
22LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.auth")
23
24if TYPE_CHECKING:
25 from aiohttp import web
26
27 from music_assistant import MusicAssistant
28
29# Context key for storing authenticated user in request
30USER_CONTEXT_KEY = "authenticated_user"
31
32_GUEST_SCOPES: Final[frozenset[Scope]] = frozenset(
33 {
34 Scope.LIBRARY_READ,
35 Scope.PLAYERS_READ,
36 Scope.PLAYERS_CONTROL,
37 Scope.QUEUES_READ,
38 Scope.QUEUES_CONTROL,
39 Scope.PROVIDERS_READ,
40 Scope.CONFIG_PLAYERS_READ,
41 }
42)
43_USER_SCOPES: Final[frozenset[Scope]] = _GUEST_SCOPES | {
44 Scope.LIBRARY_WRITE,
45 Scope.CONFIG_PROVIDERS_READ,
46 Scope.CONFIG_CORE_READ,
47 Scope.USERS_INVITE,
48 Scope.SYSTEM_READ,
49}
50
51# Scopes granted to each of the builtin user roles.
52# Roles are identified by their (string) role id to allow for custom roles in the future:
53# a role id not present in this mapping simply grants no scopes at all.
54ROLE_SCOPES: Final[Mapping[str, frozenset[Scope]]] = {
55 UserRole.ADMIN: frozenset({Scope.ALL}),
56 UserRole.USER: _USER_SCOPES,
57 UserRole.GUEST: _GUEST_SCOPES,
58 # service accounts (such as the Home Assistant integration) get
59 # slightly elevated rights over a regular user
60 UserRole.SERVICE: (
61 _USER_SCOPES | {Scope.CONFIG_PLAYERS_WRITE, Scope.USERS_READ, Scope.USERS_IMPERSONATE}
62 ),
63}
64
65# ContextVar for tracking current user and token across async calls
66current_user: ContextVar[User | None] = ContextVar("current_user", default=None)
67current_token: ContextVar[str | None] = ContextVar("current_token", default=None)
68# ContextVar to impersonate another user. Admin permissions required. Used in HA context.
69impersonated_user: ContextVar[User | None] = ContextVar("impersonated_user", default=None)
70# ContextVar for tracking the sendspin player associated with the current connection
71sendspin_player_id: ContextVar[str | None] = ContextVar("sendspin_player_id", default=None)
72# ContextVar for tracking the websocket client id associated with the current connection
73current_client_id: ContextVar[str | None] = ContextVar("current_client_id", default=None)
74# ContextVar for tracking the network address a stateless API request came from.
75# A reverse proxy or Home Assistant Ingress presents its own address for every client
76# behind it, so this identifies a caller far less precisely than a client id does.
77current_peer_address: ContextVar[str | None] = ContextVar("current_peer_address", default=None)
78
79
80async def get_authenticated_user(request: web.Request) -> User | None:
81 """
82 Get authenticated user from request.
83
84 :param request: The aiohttp request.
85 """
86 # Return the user resolved by an earlier call on this same request
87 if USER_CONTEXT_KEY in request:
88 return cast("User | None", request[USER_CONTEXT_KEY])
89
90 mass: MusicAssistant = request.app["mass"]
91
92 # Check for Home Assistant Ingress connections
93 if is_request_from_ingress(request):
94 ingress_user_id = request.headers.get("X-Remote-User-ID")
95 ingress_username = request.headers.get("X-Remote-User-Name")
96 ingress_display_name = request.headers.get("X-Remote-User-Display-Name")
97
98 # Require all Ingress headers to be present for security
99 if not (ingress_user_id and ingress_username):
100 return None
101
102 # Try to find existing user linked to this HA user ID
103 user = await mass.webserver.auth.get_user_by_provider_link(
104 AuthProviderType.HOME_ASSISTANT, ingress_user_id
105 )
106 if not user:
107 user = await mass.webserver.auth.get_user_by_username(ingress_username)
108 if not user:
109 # New user - fetch details from HA
110 ha_username, ha_display_name, avatar_url = await get_ha_user_details(
111 mass, ingress_user_id
112 )
113 role = await get_ha_user_role(mass, ingress_user_id)
114 user = await mass.webserver.auth.create_user(
115 username=ha_username or ingress_username,
116 role=role,
117 display_name=ha_display_name or ingress_display_name,
118 avatar_url=avatar_url,
119 )
120
121 # Link to Home Assistant provider (or create the link if user already existed)
122 await mass.webserver.auth.link_user_to_provider(
123 user, AuthProviderType.HOME_ASSISTANT, ingress_user_id
124 )
125
126 # Update user with HA details if available (HA is source of truth)
127 # Fall back to ingress headers if API lookup doesn't return values
128 _, ha_display_name, avatar_url = await get_ha_user_details(mass, ingress_user_id)
129 final_display_name = ha_display_name or ingress_display_name
130 LOGGER.log(
131 VERBOSE_LOG_LEVEL,
132 "Ingress auth for user %s: ha_display_name=%s, ingress_display_name=%s, "
133 "final_display_name=%s, avatar_url=%s",
134 user.username,
135 ha_display_name,
136 ingress_display_name,
137 final_display_name,
138 avatar_url,
139 )
140 if final_display_name or avatar_url:
141 user = await mass.webserver.auth.update_user(
142 user,
143 display_name=final_display_name,
144 avatar_url=avatar_url,
145 )
146 LOGGER.log(
147 VERBOSE_LOG_LEVEL,
148 "Updated user %s: display_name=%s, avatar_url=%s",
149 user.username,
150 user.display_name,
151 user.avatar_url,
152 )
153
154 # Store in request context
155 request[USER_CONTEXT_KEY] = user
156 return user
157
158 # Try to authenticate from Authorization header
159 auth_header = request.headers.get("Authorization")
160 if not auth_header:
161 return None
162
163 # Expected format: "Bearer <token>"
164 parts = auth_header.split(" ", 1)
165 if len(parts) != 2 or parts[0].lower() != "bearer":
166 return None
167
168 token = parts[1]
169
170 # Authenticate with token (works for both user tokens and API keys)
171 user = await mass.webserver.auth.authenticate_with_token(token)
172 if user:
173 # Security: Deny homeassistant system user on regular (non-Ingress) webserver
174 if not is_request_from_ingress(request) and user.username == HOMEASSISTANT_SYSTEM_USER:
175 # Reject system user on regular webserver (should only use Ingress server)
176 return None
177
178 # Store in request context
179 request[USER_CONTEXT_KEY] = user
180
181 return user
182
183
184def has_scope(user: User, scope: Scope) -> bool:
185 """
186 Check if the given user is granted the given scope (through its role).
187
188 :param user: The user to check.
189 :param scope: The scope required.
190 """
191 role_scopes = ROLE_SCOPES.get(user.role, frozenset())
192 return Scope.ALL in role_scopes or scope in role_scopes
193
194
195async def resolve_impersonated_user(
196 mass: MusicAssistant,
197 provider_type: AuthProviderType,
198 provider_user_id: str,
199 required: bool = True,
200) -> User | None:
201 """
202 Resolve and validate the user to impersonate for the current call.
203
204 A builtin user is looked up by user_id or username, users of other auth providers
205 by their provider link. The authenticated caller may always impersonate itself,
206 impersonating another user requires the users.impersonate scope.
207
208 :param mass: The MusicAssistant instance.
209 :param provider_type: The auth provider the user reference belongs to.
210 :param provider_user_id: The user's id at that provider.
211 :param required: Raise if the user cannot be found, instead of
212 resolving to None (no impersonation).
213 """
214 authenticated_user = current_user.get()
215 if authenticated_user is None:
216 raise InsufficientPermissions("Authentication is necessary to impersonate another user.")
217 if provider_type == AuthProviderType.BUILTIN:
218 # a builtin identity is the MA account itself: resolve directly instead of through
219 # the provider link table, whose builtin rows hold credentials (password hashes)
220 target_user = await mass.webserver.auth.get_user(provider_user_id)
221 if target_user is None:
222 target_user = await mass.webserver.auth.get_user_by_username(provider_user_id)
223 else:
224 target_user = await mass.webserver.auth.get_user_by_provider_link(
225 provider_type, provider_user_id
226 )
227 if target_user is None:
228 if not required:
229 return None
230 if provider_type == AuthProviderType.BUILTIN:
231 raise UserNotFoundError(
232 f"A user with user id or name {provider_user_id} is not available.",
233 translation_args=[provider_user_id],
234 )
235 raise UserNotFoundError(
236 f"A user linked to {provider_type.value} user id {provider_user_id} is not available.",
237 translation_args=[provider_user_id],
238 )
239 if target_user.user_id != authenticated_user.user_id and not has_scope(
240 authenticated_user, Scope.USERS_IMPERSONATE
241 ):
242 raise InsufficientPermissions(
243 "The users.impersonate scope is required to impersonate another user."
244 )
245 return target_user
246
247
248async def resolve_command_impersonation(mass: MusicAssistant, args: dict[str, Any]) -> User | None:
249 """
250 Pop and resolve the optional impersonation argument for an API command invocation.
251
252 The user argument is either a user_id/username string, or a dict referencing the
253 user by auth provider: {"provider": ..., "user_id": ..., "required": ...}.
254
255 Returns the user to impersonate for the command, or None if no
256 impersonation was requested.
257
258 :param mass: The MusicAssistant instance.
259 :param args: The (mutable) arguments dict of the incoming command.
260 """
261 user_arg = args.pop("user", None)
262 # username is accepted as (deprecated) alias for user
263 username_arg = args.pop("username", None)
264 # deliberately treat None and empty values as "no impersonation requested":
265 # optional fields in automations/scripts commonly template to an empty string
266 target = user_arg or username_arg
267 if not target:
268 return None
269 if isinstance(target, Mapping):
270 return await resolve_impersonated_user(mass, *_parse_provider_user_arg(target))
271 return await resolve_impersonated_user(mass, AuthProviderType.BUILTIN, str(target))
272
273
274def _parse_provider_user_arg(value: Mapping[str, Any]) -> tuple[AuthProviderType, str, bool]:
275 """Validate and unpack the dict form of the user impersonation argument."""
276 provider = value.get("provider")
277 user_id = value.get("user_id")
278 required = value.get("required", True)
279 # explicit membership check: AuthProviderType coerces unknown values to BUILTIN
280 if not isinstance(provider, str) or provider not in AuthProviderType:
281 raise InvalidDataError(f"Invalid auth provider type: {provider}")
282 if not isinstance(user_id, str) or not user_id:
283 raise InvalidDataError("A user_id is required to impersonate a user by auth provider.")
284 if not isinstance(required, bool):
285 raise InvalidDataError("The required field of the user argument must be a boolean.")
286 return AuthProviderType(provider), user_id, required
287
288
289def get_current_user() -> User | None:
290 """
291 Get the current authenticated user from context.
292
293 :return: The current user or None if not authenticated.
294 """
295 if impersonated_user := get_impersonated_user():
296 return impersonated_user
297 return current_user.get()
298
299
300def set_current_user(user: User | None) -> None:
301 """
302 Set the current authenticated user in context.
303
304 :param user: The user to set as current.
305 """
306 current_user.set(user)
307
308
309def get_impersonated_user() -> User | None:
310 """
311 Get the current impersonated user from context.
312
313 :return: The current impersonated user or None if not existing.
314 """
315 return impersonated_user.get()
316
317
318def set_impersonated_user(user: User | None) -> None:
319 """
320 Set the current impersonated user in context.
321
322 :param user: The user to set as impersonated.
323 """
324 impersonated_user.set(user)
325
326
327def get_current_token() -> str | None:
328 """
329 Get the current authentication token from context.
330
331 :return: The current token or None if not authenticated.
332 """
333 return current_token.get()
334
335
336def set_current_token(token: str | None) -> None:
337 """
338 Set the current authentication token in context.
339
340 :param token: The token to set as current.
341 """
342 current_token.set(token)
343
344
345def get_sendspin_player_id() -> str | None:
346 """
347 Get the sendspin player ID associated with the current connection.
348
349 :return: The sendspin player ID or None if not a sendspin connection.
350 """
351 return sendspin_player_id.get()
352
353
354def set_sendspin_player_id(player_id: str | None) -> None:
355 """
356 Set the sendspin player ID for the current connection.
357
358 :param player_id: The sendspin player ID to set.
359 """
360 sendspin_player_id.set(player_id)
361
362
363def get_current_client_id() -> str | None:
364 """
365 Get the websocket client id associated with the current connection.
366
367 :return: The client id, or None if not called from within a websocket command.
368 """
369 return current_client_id.get()
370
371
372def set_current_client_id(client_id: str | None) -> None:
373 """
374 Set the websocket client id for the current connection.
375
376 :param client_id: The client id to set.
377 """
378 current_client_id.set(client_id)
379
380
381def get_current_peer_address() -> str | None:
382 """
383 Get the network address the current stateless API request came from.
384
385 :return: The peer address, or None if the caller is not a stateless API request.
386 """
387 return current_peer_address.get()
388
389
390def set_current_peer_address(peer_address: str | None) -> None:
391 """
392 Set the network address for the current stateless API request.
393
394 :param peer_address: The peer address to set.
395 """
396 current_peer_address.set(peer_address)
397
398
399def is_request_from_ingress(request: web.Request) -> bool:
400 """
401 Check if request is coming from Home Assistant Ingress (internal network).
402
403 Security is enforced by socket-level verification (IP/port binding), not headers.
404 Only requests on the internal ingress TCP site (172.30.32.x:8094) are accepted.
405
406 :param request: The aiohttp request.
407 """
408 # Check if ingress site is configured in the app
409 ingress_site_params = request.app.get("ingress_site")
410 if not ingress_site_params:
411 # No ingress site configured, can't be an ingress request
412 return False
413
414 try:
415 # Security: Verify the request came through the ingress site by checking socket
416 # to prevent bypassing authentication on the regular webserver
417 transport = request.transport
418 if transport:
419 sockname = transport.get_extra_info("sockname")
420 if sockname and len(sockname) >= 2:
421 server_ip, server_port = sockname[0], sockname[1]
422 expected_ip, expected_port = ingress_site_params
423 # Request must match the ingress site's bind address and port
424 return bool(server_ip == expected_ip and server_port == expected_port)
425 except Exception: # noqa: S110
426 pass
427
428 return False
429
430
431class ImpersonatedUser:
432 """
433 Optional impersonated user context manager, for use by internal (server) code.
434
435 API commands should instead be registered with the allow_impersonation flag,
436 which handles impersonation centrally in the command dispatch.
437
438 Nested use possible: passing None for the user is a no-op which preserves
439 any impersonation already active in the current context.
440 """
441
442 def __init__(self, mass: MusicAssistant, user: str | None) -> None:
443 """
444 Initialize ImpersonatedUser.
445
446 :param mass: The MusicAssistant instance.
447 :param user: The user_id or username of the user to impersonate, or None for a no-op.
448 """
449 self.mass = mass
450 self.user = user
451 self.previous_impersonated_user = impersonated_user.get()
452
453 async def __aenter__(self) -> Self:
454 """Set the impersonated user if applicable."""
455 if self.user is None:
456 # no-op: nothing to impersonate (e.g. playback from a hardware button
457 # or an external protocol without a user context)
458 return self
459 set_impersonated_user(
460 await resolve_impersonated_user(self.mass, AuthProviderType.BUILTIN, self.user)
461 )
462 return self
463
464 async def __aexit__(
465 self,
466 exc_type: type[BaseException] | None,
467 exc_val: BaseException | None,
468 exc_tb: TracebackType | None,
469 ) -> bool | None:
470 """Unset the impersonated user."""
471 set_impersonated_user(self.previous_impersonated_user)
472 return None
473