music-assistant-server

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