from music_assistant.constants import HOMEASSISTANT_SYSTEM_USER
-from .auth_providers import get_ha_user_role
+from .auth_providers import get_ha_user_details, get_ha_user_role
if TYPE_CHECKING:
from music_assistant import MusicAssistant
if not user:
user = await mass.webserver.auth.get_user_by_username(ingress_username)
if not user:
+ # New user - fetch details from HA
+ ha_username, ha_display_name, avatar_url = await get_ha_user_details(
+ mass, ingress_user_id
+ )
role = await get_ha_user_role(mass, ingress_user_id)
user = await mass.webserver.auth.create_user(
- username=ingress_username,
+ username=ha_username or ingress_username,
role=role,
- display_name=ingress_display_name,
+ display_name=ha_display_name or ingress_display_name,
+ avatar_url=avatar_url,
)
# Link to Home Assistant provider (or create the link if user already existed)
user, AuthProviderType.HOME_ASSISTANT, ingress_user_id
)
+ # Update user with HA details if missing (HA is source of truth)
+ if not user.display_name or not user.avatar_url:
+ _, ha_display_name, avatar_url = await get_ha_user_details(mass, ingress_user_id)
+ update_display_name = ha_display_name if not user.display_name else None
+ update_avatar_url = avatar_url if not user.avatar_url else None
+ if update_display_name or update_avatar_url:
+ user = await mass.webserver.auth.update_user(
+ user,
+ display_name=update_display_name,
+ avatar_url=update_avatar_url,
+ )
+
# Store in request context
request[USER_CONTEXT_KEY] = user
return user
from music_assistant.providers.hass import HomeAssistantProvider
+LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.auth")
+
+
def normalize_username(username: str) -> str:
"""
Normalize username to lowercase for case-insensitive comparison.
return username.strip().lower()
-LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.auth")
+async def get_ha_user_details(
+ mass: MusicAssistant, ha_user_id: str
+) -> tuple[str | None, str | None, str | None]:
+ """
+ Get user username, display name and avatar URL from Home Assistant.
+
+ Uses the existing HA provider connection (which has admin access) to fetch
+ user details from config/auth/list and the person entity.
+
+ :param mass: MusicAssistant instance.
+ :param ha_user_id: Home Assistant user ID.
+ :return: Tuple of (username, display_name, avatar_url) or all None if not found.
+ """
+ hass_prov = mass.get_provider("hass")
+ if not hass_prov or not hass_prov.available:
+ return None, None, None
+
+ hass_prov = cast("HomeAssistantProvider", hass_prov)
+ return await hass_prov.get_user_details(ha_user_id)
async def get_ha_user_role(
),
)
- async def _fetch_ha_user_via_websocket(
- self, ha_url: str, access_token: str
- ) -> tuple[str | None, str | None, str | None]:
+ async def _fetch_ha_user_id_via_websocket(self, ha_url: str, access_token: str) -> str | None:
"""
- Fetch user information from Home Assistant via WebSocket.
+ Fetch the HA user ID from Home Assistant via WebSocket using OAuth token.
:param ha_url: Home Assistant URL.
:param access_token: Access token for WebSocket authentication.
- :return: Tuple of (user_id, username, display_name) or (None, None, None) if fetch fails.
+ :return: The HA user ID or None if fetch fails.
"""
ws_url = get_websocket_url(ha_url)
try:
# Use context manager to automatically handle connect/disconnect
async with HomeAssistantClient(ws_url, access_token, self.mass.http_session) as client:
- # Use the auth/current_user command to get user details
+ # Use the auth/current_user command to get user ID
result = await client.send_command("auth/current_user")
-
- if result:
- # Extract user_id, username and display name from response
- user_id = result.get("id")
- username = result.get("name") or result.get("username")
- display_name = result.get("name")
- if user_id and username:
- return user_id, username, display_name
-
- self.logger.warning("auth/current_user returned no user data")
- return None, None, None
-
+ if result and (user_id := result.get("id")):
+ return str(user_id)
+ self.logger.warning("auth/current_user returned no user data or missing id")
+ return None
except BaseHassClientError as ws_error:
self.logger.error("Failed to fetch HA user via WebSocket: %s", ws_error)
- return None, None, None
+ return None
async def _get_or_create_user(
- self, username: str, display_name: str | None, ha_user_id: str
+ self,
+ username: str,
+ display_name: str | None,
+ ha_user_id: str,
+ avatar_url: str | None = None,
) -> User | None:
"""
Get or create a user for Home Assistant OAuth authentication.
+ Updates existing users with display_name and avatar_url from HA on each OAuth login
+ (HA is considered the source of truth for these fields).
+
:param username: Username from Home Assistant.
:param display_name: Display name from Home Assistant.
:param ha_user_id: Home Assistant user ID.
+ :param avatar_url: Avatar URL from Home Assistant person entity.
:return: User object or None if creation failed.
"""
# Check if user already linked to HA
AuthProviderType.HOME_ASSISTANT, ha_user_id
)
if user:
+ # Update user with HA details if available (HA is source of truth)
+ if display_name or avatar_url:
+ user = await self.auth_manager.update_user(
+ user,
+ display_name=display_name,
+ avatar_url=avatar_url,
+ )
return user
username = normalize_username(username)
await self.auth_manager.link_user_to_provider(
existing_user, AuthProviderType.HOME_ASSISTANT, ha_user_id
)
+
+ # Update user with HA details if available (HA is source of truth)
+ if display_name or avatar_url:
+ existing_user = await self.auth_manager.update_user(
+ existing_user,
+ display_name=display_name,
+ avatar_url=avatar_url,
+ )
+
return existing_user
# New HA user - check if self-registration allowed
username=username,
role=role,
display_name=display_name or username,
+ avatar_url=avatar_url,
)
# Link to Home Assistant
if not access_token:
return AuthResult(success=False, error="No access token received from HA")
- # Fetch user information from HA via WebSocket (includes the real user ID)
- ha_user_id, username, display_name = await self._fetch_ha_user_via_websocket(
- ha_url, access_token
- )
-
- # If we couldn't get user info from WebSocket, fail authentication
- if not ha_user_id or not username:
+ # Get the HA user ID from the OAuth token via WebSocket
+ ha_user_id = await self._fetch_ha_user_id_via_websocket(ha_url, access_token)
+ if not ha_user_id:
return AuthResult(
success=False,
- error="Failed to get user info from Home Assistant",
+ error="Failed to get user ID from Home Assistant",
)
+ # Get username, display name and avatar from HA provider (has admin access)
+ username, display_name, avatar_url = await get_ha_user_details(self.mass, ha_user_id)
+
+ # Fall back to HA user ID as username if not found
+ if not username:
+ self.logger.warning("Could not get username from HA, using user ID as fallback")
+ username = ha_user_id
+
# Get or create user
- user = await self._get_or_create_user(username, display_name, ha_user_id)
+ user = await self._get_or_create_user(username, display_name, ha_user_id, avatar_url)
if not user:
return AuthResult(
from music_assistant.helpers.api import APICommandHandler, parse_arguments
from .helpers.auth_middleware import is_request_from_ingress, set_current_token, set_current_user
-from .helpers.auth_providers import get_ha_user_role
+from .helpers.auth_providers import get_ha_user_details, get_ha_user_role
if TYPE_CHECKING:
from music_assistant_models.event import MassEvent
user = await self.webserver.auth.get_user_by_username(ingress_username)
if not user:
+ # New user - fetch details from HA
+ ha_username, ha_display_name, avatar_url = await get_ha_user_details(
+ self.mass, ingress_user_id
+ )
# Auto-create user for Ingress (they're already authenticated by HA)
- # Determine role based on HA admin status
role = await get_ha_user_role(self.mass, ingress_user_id)
user = await self.webserver.auth.create_user(
- username=ingress_username,
+ username=ha_username or ingress_username,
role=role,
- display_name=ingress_display_name,
+ display_name=ha_display_name or ingress_display_name,
+ avatar_url=avatar_url,
)
# Link to Home Assistant provider (or create the link if user already existed)
user, AuthProviderType.HOME_ASSISTANT, ingress_user_id
)
+ # Update user with HA details if missing (HA is source of truth)
+ if not user.display_name or not user.avatar_url:
+ _, ha_display_name, avatar_url = await get_ha_user_details(
+ self.mass, ingress_user_id
+ )
+ update_display_name = ha_display_name if not user.display_name else None
+ update_avatar_url = avatar_url if not user.avatar_url else None
+ if update_display_name or update_avatar_url:
+ user = await self.webserver.auth.update_user(
+ user,
+ display_name=update_display_name,
+ avatar_url=update_avatar_url,
+ )
+
self._authenticated_user = user
self._logger.debug("Ingress user authenticated: %s", user.username)
else:
if player_control.supports_mute and entity_platform == "media_player":
player_control.volume_muted = attributes.get("volume_muted")
self.mass.players.update_player_control(entity_id)
+
+ async def get_user_details(self, ha_user_id: str) -> tuple[str | None, str | None, str | None]:
+ """
+ Get user username, display name and avatar URL from Home Assistant.
+
+ Looks up the user in config/auth/list for username, and the person entity
+ for display name and picture URL.
+
+ :param ha_user_id: Home Assistant user ID.
+ :return: Tuple of (username, display_name, avatar_url) or all None if not found.
+ """
+ try:
+ username: str | None = None
+ display_name: str | None = None
+ avatar_url: str | None = None
+
+ # Get username from config/auth/list (admin endpoint, we have admin access)
+ try:
+ users = await self.hass.send_command("config/auth/list")
+ for user in users or []:
+ if user.get("id") == ha_user_id:
+ username = user.get("username")
+ # Also get name as fallback display name
+ if not display_name:
+ display_name = user.get("name")
+ break
+ except Exception as err:
+ self.logger.debug("Failed to get HA user list: %s", err)
+
+ # Get external URL for building avatar URL
+ ha_url: str | None = None
+ try:
+ network_urls = await self.hass.send_command("network/url")
+ if network_urls:
+ ha_url = network_urls.get("external") or network_urls.get("internal")
+ except Exception as err:
+ self.logger.debug("Failed to get HA network URLs: %s", err)
+
+ # Find person linked to this HA user ID for display name and avatar
+ try:
+ persons = await self.hass.send_command("person/list")
+ # person/list returns {storage: [...], config: [...]}
+ all_persons = (persons.get("storage") or []) + (persons.get("config") or [])
+ for person in all_persons:
+ if person.get("user_id") == ha_user_id:
+ # Person name takes priority for display name
+ if person_name := person.get("name"):
+ display_name = person_name
+ if (person_picture := person.get("picture")) and ha_url:
+ avatar_url = f"{ha_url.rstrip('/')}{person_picture}"
+ break
+ except Exception as err:
+ self.logger.debug("Failed to get HA person details: %s", err)
+
+ return username, display_name, avatar_url
+ except Exception as err:
+ self.logger.debug("Failed to get HA user details: %s", err)
+ return None, None, None
projects / music-assistant-server.git / commitdiff