add impersonated user

Author: fmunkes

music_assistant/controllers/music.py

@@ -14,7 +14,6 @@
 from itertools import zip_longest
 from typing import TYPE_CHECKING, Any, Final, cast
 
-from music_assistant_models.auth import UserRole
 from music_assistant_models.background_task import BackgroundTask, TaskMetadata, TaskSchedule
 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
 from music_assistant_models.enums import (
@@ -26,7 +25,6 @@
     TaskStatus,
 )
 from music_assistant_models.errors import (
-    InsufficientPermissions,
     InvalidDataError,
     InvalidProviderID,
     InvalidProviderURI,
@@ -80,7 +78,10 @@
     VACUUM_MIN_RECLAIM_RATIO,
 )
 from music_assistant.controllers.tasks.context import update_current_task_progress_text
-from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
+from music_assistant.controllers.webserver.helpers.auth_middleware import (
+    UseImpersonatedUser,
+    get_current_user,
+)
 from music_assistant.helpers.api import api_command
 from music_assistant.helpers.compare import compare_strings, compare_version, create_safe_string
 from music_assistant.helpers.database import UNSET, DatabaseConnection
@@ -3659,57 +3660,58 @@ async def get_item_by_name(
         username_or_user_id: str | None = None,
     ) -> MediaItemType | ItemMapping | None:
         """Try to find a media item (such as a playlist) by name."""
-        # Future todo: enhance this method with AI capabilities to allow typos and
-        # natural language.
-        searchname = name.lower()
-        library_functions = [
-            x
-            for x in (
-                self.playlists.library_items,
-                self.radio.library_items,
-                self.tracks.library_items,
-                self.albums.library_items,
-                self.artists.library_items,
-                self.audiobooks.library_items,
-                self.podcasts.library_items,
-            )
-            if not media_type or media_type.value.lower() in x.__name__
-        ]
-        # prefer (exact) lookup in the library by name
-        for func in library_functions:
-            result = await func(search=searchname, username_or_user_id=username_or_user_id)
-            for item in result:
-                # handle optional artist filter
-                if (
-                    artist
-                    and (artists := getattr(item, "artists", None))
-                    and not any(x for x in artists if x.name.lower() == artist.lower())
-                ):
-                    continue
-                # handle optional album filter
-                if (
-                    album
-                    and (item_album := getattr(item, "album", None))
-                    and item_album.name.lower() != album.lower()
-                ):
-                    continue
-                if searchname == item.name.lower():
-                    return item
-        # nothing found in the library, fallback to global search
-        search_name = name
-        if album and artist:
-            search_name = f"{artist} - {album} - {name}"
-        elif album:
-            search_name = f"{album} - {name}"
-        elif artist:
-            search_name = f"{artist} - {name}"
-        search_results = await self.search(
-            search_query=search_name,
-            media_types=[media_type]
-            if media_type and media_type != MediaType.UNKNOWN
-            else MediaType.ALL,
-            limit=8,
-        )
+        async with UseImpersonatedUser(self.mass, username_or_user_id):
+            # Future todo: enhance this method with AI capabilities to allow typos and
+            # natural language.
+            searchname = name.lower()
+            library_functions = [
+                x
+                for x in (
+                    self.playlists.library_items,
+                    self.radio.library_items,
+                    self.tracks.library_items,
+                    self.albums.library_items,
+                    self.artists.library_items,
+                    self.audiobooks.library_items,
+                    self.podcasts.library_items,
+                )
+                if not media_type or media_type.value.lower() in x.__name__
+            ]
+            # prefer (exact) lookup in the library by name
+            for func in library_functions:
+                result = await func(search=searchname, username_or_user_id=username_or_user_id)
+                for item in result:
+                    # handle optional artist filter
+                    if (
+                        artist
+                        and (artists := getattr(item, "artists", None))
+                        and not any(x for x in artists if x.name.lower() == artist.lower())
+                    ):
+                        continue
+                    # handle optional album filter
+                    if (
+                        album
+                        and (item_album := getattr(item, "album", None))
+                        and item_album.name.lower() != album.lower()
+                    ):
+                        continue
+                    if searchname == item.name.lower():
+                        return item
+            # nothing found in the library, fallback to global search
+            search_name = name
+            if album and artist:
+                search_name = f"{artist} - {album} - {name}"
+            elif album:
+                search_name = f"{album} - {name}"
+            elif artist:
+                search_name = f"{artist} - {name}"
+            search_results = await self.search(
+                search_query=search_name,
+                media_types=[media_type]
+                if media_type and media_type != MediaType.UNKNOWN
+                else MediaType.ALL,
+                limit=8,
+            )
         for results in (
             search_results.tracks,
             search_results.albums,
@@ -3730,81 +3732,46 @@ async def verify_item_uri(self, uri: str, username_or_user_id: str | None = None
 
         If username_or_user_id is specified, verifies additionally, if this user may access this item. This requires the requesting (i.e. authorized user) to be able to access this item as well.
         """
-        user: User | None = None
-        if username_or_user_id:
-            # below raises if permissions are insufficient
-            user = await self.mass.music.get_requested_user_if_authorized(username_or_user_id)
-
-        try:
-            media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
-        except (InvalidProviderURI, InvalidProviderID):
-            return False
+        async with UseImpersonatedUser(self.mass, username_or_user_id):
+            user = get_current_user()
 
-        # fast return for a provider uri which is not part of a user with a provider filter
-        if (
-            provider_instance_id_or_domain != "library"
-            and user
-            and user.provider_filter
-            and provider_instance_id_or_domain not in user.provider_filter
-        ):
-            return False
+            try:
+                media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
+            except (InvalidProviderURI, InvalidProviderID):
+                return False
 
-        # verify that item itself exists
-        try:
-            item = await self.get_item(
-                media_type=media_type,
-                item_id=item_id,
-                provider_instance_id_or_domain=provider_instance_id_or_domain,
-                allow_update_metadata=False,  # no need trigger more methods
-            )
-        except MediaNotFoundError:
-            return False
+            # fast return for a provider uri which is not part of a user with a provider filter
+            if (
+                provider_instance_id_or_domain != "library"
+                and user
+                and user.provider_filter
+                and provider_instance_id_or_domain not in user.provider_filter
+            ):
+                return False
 
-        # non library item handling for users with no filter, or no user at all
-        if (
-            provider_instance_id_or_domain != "library"
-            or not user
-            or (user and not user.provider_filter)
-            or isinstance(item, BrowseFolder)
-        ):
-            return True
+            # verify that item itself exists
+            try:
+                item = await self.get_item(
+                    media_type=media_type,
+                    item_id=item_id,
+                    provider_instance_id_or_domain=provider_instance_id_or_domain,
+                    allow_update_metadata=False,  # no need trigger more methods
+                )
+            except MediaNotFoundError:
+                return False
 
-        # library item handling for users with provider filter
-        for provider_mapping in item.provider_mappings:
-            if provider_mapping.provider_instance in user.provider_filter:
+            # non library item handling for users with no filter, or no user at all
+            if (
+                provider_instance_id_or_domain != "library"
+                or not user
+                or (user and not user.provider_filter)
+                or isinstance(item, BrowseFolder)
+            ):
                 return True
 
-        return False
-
-    async def get_requested_user_if_authorized(self, username_or_user_id: str) -> User:
-        """Return requested user if authenticated user may access all music providers of this user.
-
-        Raises InsufficientPermissions otherwise.
-        """
-        requested_user = await self.mass.webserver.auth.get_user_by_id_or_name(username_or_user_id)
-        if not requested_user:
-            raise InvalidDataError(
-                f"A user with user id or name {username_or_user_id} is not available."
-            )
-
-        authenticated_user = get_current_user()
-        if not authenticated_user:
-            raise InsufficientPermissions("Only an authenticated user may request another user.")
+            # library item handling for users with provider filter
+            for provider_mapping in item.provider_mappings:
+                if provider_mapping.provider_instance in user.provider_filter:
+                    return True
 
-        if requested_user == authenticated_user:
-            return requested_user
-
-        if authenticated_user.role == UserRole.ADMIN or not authenticated_user.provider_filter:
-            # If no provider filter is set, a user is allowed to access any provider.
-            return requested_user
-
-        if not requested_user.provider_filter or not set(requested_user.provider_filter).issubset(
-            authenticated_user.provider_filter
-        ):
-            # Requested user may access any provider, but we excluded that the authenticated user can do the
-            # same already
-            raise InsufficientPermissions(
-                f"The authenticated user {authenticated_user.display_name} lacks permission to access all music providers accessible to {requested_user.display_name}."
-            )
-
-        return requested_user
+        return False

music_assistant/controllers/player_queues.py

@@ -79,6 +79,7 @@
 from music_assistant.controllers.players.constants import PlayerLockPurpose
 from music_assistant.controllers.streams.audio_buffer import AudioBuffer
 from music_assistant.controllers.webserver.helpers.auth_middleware import (
+    UseImpersonatedUser,
     get_current_user,
     set_current_user,
 )
@@ -550,10 +551,11 @@ async def play_media(
         self._check_player_permission(queue_id)
         if not self.get(queue_id):
             raise PlayerUnavailableError(f"Queue {queue_id} is not available")
-        # Lock is acquired by the @handle_play_action decorator on the internal handler
-        await self._handle_play_media(
-            queue_id, media, option, radio_mode, start_item, username, sort_by
-        )
+        async with UseImpersonatedUser(self.mass, username):
+            # Lock is acquired by the @handle_play_action decorator on the internal handler
+            await self._handle_play_media(
+                queue_id, media, option, radio_mode, start_item, username, sort_by
+            )
 
     @api_command("player_queues/move_item")
     def move_item(self, queue_id: str, queue_item_id: str, pos_shift: int = 1) -> None:

music_assistant/controllers/webserver/helpers/auth_middleware.py

@@ -4,10 +4,12 @@
 
 import logging
 from contextvars import ContextVar
-from typing import TYPE_CHECKING, Any, cast
+from types import TracebackType
+from typing import TYPE_CHECKING, Any, Self, cast
 
 from aiohttp import web
 from music_assistant_models.auth import AuthProviderType, User, UserRole
+from music_assistant_models.errors import InsufficientPermissions, InvalidDataError
 
 from music_assistant.constants import HOMEASSISTANT_SYSTEM_USER, MASS_LOGGER_NAME, VERBOSE_LOG_LEVEL
 
@@ -24,6 +26,8 @@
 # ContextVar for tracking current user and token across async calls
 current_user: ContextVar[User | None] = ContextVar("current_user", default=None)
 current_token: ContextVar[str | None] = ContextVar("current_token", default=None)
+# ContextVar to impersonate another user as and admin user. Used in HA context.
+impersonated_user: ContextVar[User | None] = ContextVar("impersonated_user", default=None)
 # ContextVar for tracking the sendspin player associated with the current connection
 sendspin_player_id: ContextVar[str | None] = ContextVar("sendspin_player_id", default=None)
 
@@ -162,6 +166,8 @@ def get_current_user() -> User | None:
 
     :return: The current user or None if not authenticated.
     """
+    if impersonated_user := get_impersonated_user():
+        return impersonated_user
     return current_user.get()
 
 
@@ -174,6 +180,24 @@ def set_current_user(user: User | None) -> None:
     current_user.set(user)
 
 
+def get_impersonated_user() -> User | None:
+    """
+    Get the current impersonated user from context.
+
+    :return: The current impersonated user or None if not existing.
+    """
+    return impersonated_user.get()
+
+
+def set_impersonated_user(user: User | None) -> None:
+    """
+    Set the current impersonated user in context.
+
+    :param user: The user to set as impersonated.
+    """
+    impersonated_user.set(user)
+
+
 def get_current_token() -> str | None:
     """
     Get the current authentication token from context.
@@ -278,3 +302,46 @@ async def auth_middleware(request: web.Request, handler: Any) -> web.StreamRespo
     # Let the handler decide if authentication is required
     # The handler will call require_authentication() if needed
     return cast("web.StreamResponse", await handler(request))
+
+
+class UseImpersonatedUser:
+    """Use impersonated user context manager.
+
+    If username_or_user_id is None this class does nothing. Otherwise, only an Admin
+    may impersonate another user. This is used for calls from HA (e.g. play_media on
+    the player queues controller).
+    """
+
+    def __init__(self, mass: MusicAssistant, username_or_user_id: str | None) -> None:
+        """Init."""
+        self.mass = mass
+        self.username_or_user_id = username_or_user_id
+        if (
+            username_or_user_id is not None
+            and (authenticated_user := get_current_user())
+            and authenticated_user.role != UserRole.ADMIN
+        ):
+            raise InsufficientPermissions("Can only impersonate another user as Admin.")
+
+    async def __aenter__(self) -> Self:
+        """Set the impersonated if given."""
+        if self.username_or_user_id is None:
+            return self
+        if impersonated_user := await self.mass.webserver.auth.get_user_by_id_or_name(
+            self.username_or_user_id
+        ):
+            set_impersonated_user(impersonated_user)
+            return self
+        raise InvalidDataError(
+            f"A user with user id or name {self.username_or_user_id} is not available."
+        )
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None:
+        """Unset the impersonated user."""
+        set_impersonated_user(None)
+        return None