/
/
/
1"""Authentication provider base classes and implementations."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import logging
8import secrets
9from abc import ABC, abstractmethod
10from collections.abc import Sequence
11from dataclasses import dataclass
12from datetime import datetime, timedelta
13from typing import TYPE_CHECKING, Any, Final, TypedDict, cast
14from urllib.parse import urlparse
15
16from hass_client import HomeAssistantClient
17from hass_client.exceptions import BaseHassClientError
18from hass_client.utils import base_url, get_auth_url, get_token, get_websocket_url
19from music_assistant_models.auth import AuthProviderType, User, UserRole
20from music_assistant_models.errors import AuthenticationFailed
21
22from music_assistant.constants import CONF_AUTH_ALLOW_SELF_REGISTRATION, MASS_LOGGER_NAME
23from music_assistant.helpers.datetime import utc
24
25if TYPE_CHECKING:
26 from music_assistant import MusicAssistant
27 from music_assistant.controllers.webserver.auth import AuthenticationManager
28 from music_assistant.providers.hass import HomeAssistantProvider
29
30
31LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.auth")
32
33# Progressive (failed attempts, delay in seconds) tiers applied to a single rate limit key
34DEFAULT_DELAY_TIERS: Final[tuple[tuple[int, int], ...]] = ((3, 30), (6, 60), (10, 120), (15, 300))
35DEFAULT_TRACKING_WINDOW: Final = timedelta(minutes=30)
36# Tracked keys before expired ones are swept from the rate limiter's bookkeeping
37PRUNE_THRESHOLD: Final = 128
38
39
40def normalize_username(username: str) -> str:
41 """
42 Normalize username to lowercase for case-insensitive comparison.
43
44 :param username: The username to normalize.
45 :return: Normalized username (lowercase, stripped).
46 """
47 return username.strip().lower()
48
49
50async def get_ha_user_details(
51 mass: MusicAssistant, ha_user_id: str, wait_timeout: float = 10.0
52) -> tuple[str | None, str | None, str | None]:
53 """
54 Get user username, display name and avatar URL from Home Assistant.
55
56 Uses the existing HA provider connection (which has admin access) to fetch
57 user details from config/auth/list and the person entity.
58
59 :param mass: MusicAssistant instance.
60 :param ha_user_id: Home Assistant user ID.
61 :param wait_timeout: Maximum time to wait for HA provider to become available (default 10s).
62 :return: Tuple of (username, display_name, avatar_url) or all None if not found.
63 """
64 # Wait for the HA provider to become available using event-based signaling
65 try:
66 await asyncio.wait_for(mass.get_provider_ready_event("hass").wait(), timeout=wait_timeout)
67 except TimeoutError:
68 LOGGER.debug(
69 "HA provider not available after %.1fs, cannot fetch user details", wait_timeout
70 )
71 return None, None, None
72
73 hass_prov = mass.get_provider("hass")
74 if hass_prov is None or not hass_prov.available:
75 LOGGER.debug("HA provider not available, cannot fetch user details")
76 return None, None, None
77
78 hass_prov = cast("HomeAssistantProvider", hass_prov)
79 return await hass_prov.get_user_details(ha_user_id)
80
81
82async def get_ha_user_role(
83 mass: MusicAssistant, ha_user_id: str, wait_timeout: float = 10.0
84) -> UserRole:
85 """
86 Get user role based on Home Assistant admin status.
87
88 :param mass: MusicAssistant instance.
89 :param ha_user_id: The Home Assistant user ID to check.
90 :param wait_timeout: Maximum time to wait for HA provider to become available (default 10s).
91 """
92 try:
93 # Wait for the HA provider to become available using event-based signaling
94 try:
95 await asyncio.wait_for(
96 mass.get_provider_ready_event("hass").wait(), timeout=wait_timeout
97 )
98 except TimeoutError:
99 raise RuntimeError(f"Home Assistant provider not available after {wait_timeout}s")
100
101 hass_prov = mass.get_provider("hass")
102 if hass_prov is None or not hass_prov.available:
103 raise RuntimeError("Home Assistant provider not available")
104
105 if TYPE_CHECKING:
106 hass_prov = cast("HomeAssistantProvider", hass_prov)
107 # Query HA for user list to check admin status
108 result = await hass_prov.hass.send_command("config/auth/list")
109 if not result:
110 raise RuntimeError("Failed to retrieve user list from Home Assistant")
111 for ha_user in result:
112 if ha_user.get("id") == ha_user_id:
113 # User is admin if they have "system-admin" in their group_ids
114 group_ids = ha_user.get("group_ids", [])
115 if "system-admin" in group_ids:
116 LOGGER.debug("HA user %s is admin, granting ADMIN role", ha_user_id)
117 return UserRole.ADMIN
118 return UserRole.USER
119 raise RuntimeError(f"HA user ID {ha_user_id} not found in user list")
120 except Exception as err:
121 msg = f"Failed to check HA admin status: {err}"
122 raise AuthenticationFailed(msg) from err
123
124
125class LoginRateLimiter:
126 """Rate limiter for login attempts to prevent brute force attacks."""
127
128 def __init__(
129 self,
130 delay_tiers: Sequence[tuple[int, int]] = DEFAULT_DELAY_TIERS,
131 tracking_window: timedelta = DEFAULT_TRACKING_WINDOW,
132 warn_threshold: int = 10,
133 alert_threshold: int = 20,
134 subject: str = "username",
135 ) -> None:
136 """
137 Initialize the rate limiter.
138
139 :param delay_tiers: (failed attempts, delay in seconds) pairs in ascending order of
140 attempts. The highest tier whose attempt count is reached sets the delay.
141 :param tracking_window: How long a failed attempt keeps counting towards the tiers.
142 :param warn_threshold: Failed attempts for one key before suspicious activity is logged.
143 :param alert_threshold: Failed attempts for one key before a stronger warning is logged.
144 :param subject: What the keys of this limiter identify, used in log messages.
145 """
146 # Track failed attempts per key: {key: [timestamp1, timestamp2, ...]}
147 self._failed_attempts: dict[str, list[datetime]] = {}
148 self._delay_tiers = tuple(delay_tiers)
149 self._tracking_window = tracking_window
150 self._warn_threshold = warn_threshold
151 self._alert_threshold = alert_threshold
152 self._subject = subject
153 # Lock for thread-safe access to _failed_attempts
154 self._lock = asyncio.Lock()
155
156 def get_attempt_count(self, key: str) -> int:
157 """
158 Get the number of failed attempts for a key inside the tracking window.
159
160 :param key: The key to count failed attempts for.
161 :return: Number of failed attempts still being tracked.
162 """
163 cutoff_time = utc() - self._tracking_window
164 return sum(1 for timestamp in self._failed_attempts.get(key, ()) if timestamp > cutoff_time)
165
166 def get_delay(self, key: str) -> int:
167 """
168 Get the delay in seconds before the next attempt for a key is allowed.
169
170 :param key: The key attempting to authenticate.
171 :return: Delay in seconds (0 if no delay needed).
172 """
173 attempt_count = self.get_attempt_count(key)
174 delay = 0
175 for tier_attempts, tier_delay in self._delay_tiers:
176 if attempt_count >= tier_attempts:
177 delay = tier_delay
178 return delay
179
180 async def check_rate_limit(self, key: str) -> tuple[bool, int]:
181 """
182 Check if an attempt is allowed and apply delay if needed.
183
184 :param key: The key attempting to authenticate.
185 :return: Tuple of (allowed, delay_seconds). If not allowed, includes remaining delay.
186 """
187 async with self._lock:
188 self._cleanup_old_attempts(key)
189
190 if key not in self._failed_attempts or not self._failed_attempts[key]:
191 return True, 0
192
193 # Get the most recent failed attempt
194 last_attempt = self._failed_attempts[key][-1]
195 required_delay = self.get_delay(key)
196
197 if required_delay == 0:
198 return True, 0
199
200 # Calculate how much time has passed since last attempt
201 time_since_last = (utc() - last_attempt).total_seconds()
202
203 if time_since_last < required_delay:
204 # Still in cooldown period
205 remaining_delay = int(required_delay - time_since_last)
206 return False, remaining_delay
207
208 return True, 0
209
210 async def record_failed_attempt(self, key: str) -> None:
211 """
212 Record a failed attempt.
213
214 :param key: The key that failed to authenticate.
215 """
216 async with self._lock:
217 self._cleanup_old_attempts(key)
218
219 if key not in self._failed_attempts:
220 self._failed_attempts[key] = []
221
222 self._failed_attempts[key].append(utc())
223
224 # Log warning for suspicious activity
225 attempt_count = len(self._failed_attempts[key])
226 if attempt_count == self._warn_threshold:
227 LOGGER.warning(
228 "Suspicious activity: %d failed attempts (%s=%s)",
229 attempt_count,
230 self._subject,
231 key,
232 )
233 elif attempt_count == self._alert_threshold:
234 LOGGER.warning(
235 "High suspicious activity: %d failed attempts (%s=%s). "
236 "Manual intervention may be needed.",
237 attempt_count,
238 self._subject,
239 key,
240 )
241
242 # A key is only cleaned up when it is used again, and most keys (a one-off
243 # connection, a made-up username) never come back, so sweep the whole map once
244 # it grows past what any legitimate burst of callers produces.
245 if len(self._failed_attempts) > PRUNE_THRESHOLD:
246 for tracked_key in list(self._failed_attempts):
247 self._cleanup_old_attempts(tracked_key)
248
249 async def clear_attempts(self, key: str) -> None:
250 """
251 Clear failed attempts for a key (called after a successful attempt).
252
253 :param key: The key to clear.
254 """
255 async with self._lock:
256 if key in self._failed_attempts:
257 del self._failed_attempts[key]
258
259 def _cleanup_old_attempts(self, key: str) -> None:
260 """
261 Remove failed attempts outside the tracking window.
262
263 :param key: The key to clean up.
264 """
265 if key not in self._failed_attempts:
266 return
267
268 cutoff_time = utc() - self._tracking_window
269 self._failed_attempts[key] = [
270 timestamp for timestamp in self._failed_attempts[key] if timestamp > cutoff_time
271 ]
272
273 # Remove key if no attempts left
274 if not self._failed_attempts[key]:
275 del self._failed_attempts[key]
276
277
278class LoginProviderConfig(TypedDict, total=False):
279 """Base configuration for login providers."""
280
281
282class HomeAssistantProviderConfig(LoginProviderConfig):
283 """Configuration for Home Assistant OAuth provider."""
284
285 ha_url: str
286
287
288@dataclass
289class AuthResult:
290 """Result of an authentication attempt."""
291
292 success: bool
293 user: User | None = None
294 error: str | None = None
295 access_token: str | None = None
296 return_url: str | None = None
297
298
299class LoginProvider(ABC):
300 """Base class for login providers."""
301
302 def __init__(self, mass: MusicAssistant, provider_id: str, config: LoginProviderConfig) -> None:
303 """
304 Initialize login provider.
305
306 :param mass: MusicAssistant instance.
307 :param provider_id: Unique identifier for this provider instance.
308 :param config: Provider-specific configuration.
309 """
310 self.mass = mass
311 self.provider_id = provider_id
312 self.config = config
313 self.logger = LOGGER
314
315 @property
316 def allow_self_registration(self) -> bool:
317 """Return whether self-registration is allowed for this provider."""
318 return False
319
320 @property
321 def auth_manager(self) -> AuthenticationManager:
322 """Get auth manager from webserver."""
323 return self.mass.webserver.auth
324
325 @property
326 @abstractmethod
327 def provider_type(self) -> AuthProviderType:
328 """Return the provider type."""
329
330 @property
331 @abstractmethod
332 def requires_redirect(self) -> bool:
333 """Return True if this provider requires OAuth redirect."""
334
335 @abstractmethod
336 async def authenticate(self, credentials: dict[str, Any]) -> AuthResult:
337 """
338 Authenticate user with provided credentials.
339
340 :param credentials: Provider-specific credentials (username/password, OAuth code, etc).
341 """
342
343 async def get_authorization_url(
344 self, redirect_uri: str, return_url: str | None = None
345 ) -> str | None:
346 """
347 Get OAuth authorization URL if applicable.
348
349 :param redirect_uri: The callback URL for OAuth flow.
350 :param return_url: Optional URL to redirect to after successful login.
351 """
352 return None
353
354 async def handle_oauth_callback(self, code: str, state: str, redirect_uri: str) -> AuthResult:
355 """
356 Handle OAuth callback if applicable.
357
358 :param code: OAuth authorization code.
359 :param state: OAuth state parameter for CSRF protection.
360 :param redirect_uri: The callback URL.
361 """
362 return AuthResult(success=False, error="OAuth not supported by this provider")
363
364
365class BuiltinLoginProvider(LoginProvider):
366 """Built-in username/password login provider."""
367
368 def __init__(self, mass: MusicAssistant, provider_id: str, config: LoginProviderConfig) -> None:
369 """
370 Initialize built-in login provider.
371
372 :param mass: MusicAssistant instance.
373 :param provider_id: Unique identifier for this provider instance.
374 :param config: Provider-specific configuration.
375 """
376 super().__init__(mass, provider_id, config)
377 self._rate_limiter = LoginRateLimiter()
378
379 @property
380 def provider_type(self) -> AuthProviderType:
381 """Return the provider type."""
382 return AuthProviderType.BUILTIN
383
384 @property
385 def requires_redirect(self) -> bool:
386 """Return False - built-in provider doesn't need redirect."""
387 return False
388
389 async def authenticate(self, credentials: dict[str, Any]) -> AuthResult:
390 """
391 Authenticate user with username and password.
392
393 :param credentials: Dict containing 'username' and 'password'.
394 """
395 username = credentials.get("username")
396 password = credentials.get("password")
397
398 if not username or not password:
399 return AuthResult(success=False, error="Username and password required")
400
401 username = normalize_username(username)
402
403 # Check rate limit before attempting authentication
404 allowed, remaining_delay = await self._rate_limiter.check_rate_limit(username)
405 if not allowed:
406 self.logger.warning(
407 "Rate limit exceeded for username '%s'. %d seconds remaining.",
408 username,
409 remaining_delay,
410 )
411 return AuthResult(
412 success=False,
413 error=f"Too many failed attempts. Please try again in {remaining_delay} seconds.",
414 )
415
416 # First, look up user by username to get user_id
417 # This is needed to create the password hash with user_id in the salt
418 user_row = await self.auth_manager.database.get_row("users", {"username": username})
419 if not user_row:
420 # Record failed attempt even if username doesn't exist
421 # This prevents username enumeration timing attacks
422 await self._rate_limiter.record_failed_attempt(username)
423 return AuthResult(success=False, error="Invalid username or password")
424
425 user_id = user_row["user_id"]
426
427 # Hash the password using user_id for enhanced security
428 password_hash = self._hash_password(password, user_id)
429
430 # Verify the password by checking if provider link exists
431 user = await self.auth_manager.get_user_by_provider_link(
432 AuthProviderType.BUILTIN, password_hash
433 )
434
435 if not user:
436 # Record failed attempt
437 await self._rate_limiter.record_failed_attempt(username)
438 return AuthResult(success=False, error="Invalid username or password")
439
440 # Check if user is enabled
441 if not user.enabled:
442 # Record failed attempt for disabled accounts too
443 await self._rate_limiter.record_failed_attempt(username)
444 return AuthResult(success=False, error="User account is disabled")
445
446 # Successful login - clear any failed attempts
447 await self._rate_limiter.clear_attempts(username)
448 return AuthResult(success=True, user=user)
449
450 async def create_user_with_password(
451 self,
452 username: str,
453 password: str,
454 role: UserRole = UserRole.USER,
455 display_name: str | None = None,
456 player_filter: list[str] | None = None,
457 provider_filter: list[str] | None = None,
458 ) -> User:
459 """
460 Create a new built-in user with password.
461
462 :param username: The username.
463 :param password: The password (will be hashed).
464 :param role: The user role (default: USER).
465 :param display_name: Optional display name.
466 :param player_filter: Optional list of player IDs user has access to.
467 :param provider_filter: Optional list of provider instance IDs user has access to.
468 """
469 # Create the user
470 user = await self.auth_manager.create_user(
471 username=username,
472 role=role,
473 display_name=display_name,
474 player_filter=player_filter,
475 provider_filter=provider_filter,
476 )
477
478 # Hash password using user_id for enhanced security
479 password_hash = self._hash_password(password, user.user_id)
480 await self.auth_manager.link_user_to_provider(user, AuthProviderType.BUILTIN, password_hash)
481
482 return user
483
484 async def change_password(self, user: User, old_password: str, new_password: str) -> bool:
485 """
486 Change user password.
487
488 :param user: The user.
489 :param old_password: Current password for verification.
490 :param new_password: The new password.
491 """
492 # Verify old password first using user_id
493 old_password_hash = self._hash_password(old_password, user.user_id)
494 existing_user = await self.auth_manager.get_user_by_provider_link(
495 AuthProviderType.BUILTIN, old_password_hash
496 )
497
498 if not existing_user or existing_user.user_id != user.user_id:
499 return False
500
501 # Update password link with new hash using user_id
502 new_password_hash = self._hash_password(new_password, user.user_id)
503 await self.auth_manager.update_provider_link(
504 user, AuthProviderType.BUILTIN, new_password_hash
505 )
506
507 return True
508
509 async def reset_password(self, user: User, new_password: str) -> None:
510 """
511 Reset user password (admin only - no old password verification).
512
513 :param user: The user whose password to reset.
514 :param new_password: The new password.
515 """
516 # Hash new password using user_id and update provider link
517 new_password_hash = self._hash_password(new_password, user.user_id)
518 await self.auth_manager.update_provider_link(
519 user, AuthProviderType.BUILTIN, new_password_hash
520 )
521
522 def _hash_password(self, password: str, user_id: str) -> str:
523 """
524 Hash password with salt combining user ID and server ID.
525
526 :param password: Plain text password.
527 :param user_id: User ID to include in salt (random token for high entropy).
528 """
529 # Combine user_id (random) and server_id for maximum security
530 salt = f"{user_id}:{self.mass.server_id}"
531 return hashlib.pbkdf2_hmac(
532 "sha256", password.encode(), salt.encode(), iterations=100000
533 ).hex()
534
535
536class HomeAssistantOAuthProvider(LoginProvider):
537 """Home Assistant OAuth login provider."""
538
539 def __init__(self, mass: MusicAssistant, provider_id: str, config: LoginProviderConfig) -> None:
540 """
541 Initialize Home Assistant OAuth provider.
542
543 :param mass: MusicAssistant instance.
544 :param provider_id: Unique identifier for this provider instance.
545 :param config: Provider-specific configuration.
546 """
547 super().__init__(mass, provider_id, config)
548 # Store OAuth state -> return_url mapping to support concurrent sessions
549 self._oauth_sessions: dict[str, str | None] = {}
550
551 @property
552 def allow_self_registration(self) -> bool:
553 """Return whether self-registration is allowed, read dynamically from config."""
554 return bool(self.mass.webserver.config.get_value(CONF_AUTH_ALLOW_SELF_REGISTRATION))
555
556 @property
557 def provider_type(self) -> AuthProviderType:
558 """Return the provider type."""
559 return AuthProviderType.HOME_ASSISTANT
560
561 @property
562 def requires_redirect(self) -> bool:
563 """Return True - Home Assistant OAuth requires redirect."""
564 return True
565
566 async def authenticate(self, credentials: dict[str, Any]) -> AuthResult:
567 """
568 Not used for OAuth providers - use handle_oauth_callback instead.
569
570 :param credentials: Not used.
571 """
572 return AuthResult(success=False, error="Use OAuth flow for Home Assistant authentication")
573
574 async def get_authorization_url(
575 self, redirect_uri: str, return_url: str | None = None
576 ) -> str | None:
577 """
578 Get Home Assistant OAuth authorization URL using hass_client.
579
580 :param redirect_uri: The callback URL.
581 :param return_url: Optional URL to redirect to after successful login.
582 """
583 # Get the correct HA URL (external URL if running as add-on)
584 ha_url = await self._get_external_ha_url()
585 if not ha_url:
586 return None
587
588 # If HA URL is still the internal supervisor URL (no external_url in HA config),
589 # infer from redirect_uri (the URL user is accessing MA from)
590 if "supervisor" in ha_url.lower():
591 # Extract scheme and host from redirect_uri to build external HA URL
592 parsed = urlparse(redirect_uri)
593 # HA typically runs on port 8123, but use default ports for HTTPS (443) or HTTP (80)
594 if parsed.scheme == "https":
595 # HTTPS - use default port 443 (no port in URL)
596 inferred_ha_url = f"{parsed.scheme}://{parsed.hostname}"
597 else:
598 # HTTP - assume HA runs on default port 8123
599 inferred_ha_url = f"{parsed.scheme}://{parsed.hostname}:8123"
600
601 self.logger.debug(
602 "HA external_url not configured, inferring from callback URL: %s",
603 inferred_ha_url,
604 )
605 ha_url = inferred_ha_url
606
607 state = secrets.token_urlsafe(32)
608 # Store return_url keyed by state to support concurrent OAuth sessions
609 # This prevents race conditions when multiple users/sessions login simultaneously
610 self._oauth_sessions[state] = return_url
611
612 # Use base_url of callback as client_id (same as HA provider does)
613 client_id = base_url(redirect_uri)
614
615 # Use hass_client's get_auth_url utility
616 return cast(
617 "str",
618 get_auth_url(
619 ha_url,
620 redirect_uri,
621 client_id=client_id,
622 state=state,
623 ),
624 )
625
626 async def handle_oauth_callback(self, code: str, state: str, redirect_uri: str) -> AuthResult:
627 """
628 Handle Home Assistant OAuth callback using hass_client.
629
630 :param code: OAuth authorization code.
631 :param state: OAuth state parameter.
632 :param redirect_uri: The callback URL.
633 """
634 # Verify state and retrieve return_url from session
635 if state not in self._oauth_sessions:
636 return AuthResult(success=False, error="Invalid or expired state parameter")
637
638 # Retrieve and remove the return_url for this session (cleanup)
639 return_url = self._oauth_sessions.pop(state)
640
641 # Get the correct HA URL (external URL if running as add-on)
642 # This must be the same URL used in get_authorization_url
643 ha_url = await self._get_external_ha_url()
644 if not ha_url:
645 return AuthResult(success=False, error="Home Assistant URL not configured")
646
647 try:
648 # Use base_url of callback as client_id (same as HA provider does)
649 client_id = base_url(redirect_uri)
650
651 # Use hass_client's get_token utility - no client_secret needed!
652 try:
653 token_details = await get_token(ha_url, code, client_id=client_id)
654 except Exception as token_error:
655 self.logger.error(
656 "Failed to get token from HA: %s (client_id: %s, ha_url: %s)",
657 token_error,
658 client_id,
659 ha_url,
660 )
661 return AuthResult(
662 success=False, error=f"Failed to exchange OAuth code: {token_error}"
663 )
664
665 access_token = token_details.get("access_token")
666 if not access_token:
667 return AuthResult(success=False, error="No access token received from HA")
668
669 # Get the HA user ID from the OAuth token via WebSocket
670 ha_user_id = await self._fetch_ha_user_id_via_websocket(ha_url, access_token)
671 if not ha_user_id:
672 return AuthResult(
673 success=False,
674 error="Failed to get user ID from Home Assistant",
675 )
676
677 # Get username, display name and avatar from HA provider (has admin access)
678 username, display_name, avatar_url = await get_ha_user_details(self.mass, ha_user_id)
679
680 # Fall back to HA user ID as username if not found
681 if not username:
682 self.logger.warning("Could not get username from HA, using user ID as fallback")
683 username = ha_user_id
684
685 # Get or create user
686 user = await self._get_or_create_user(username, display_name, ha_user_id, avatar_url)
687
688 if not user:
689 return AuthResult(
690 success=False,
691 error="Self-registration is disabled. Please contact an administrator.",
692 )
693
694 return AuthResult(success=True, user=user, return_url=return_url)
695
696 except Exception as e:
697 self.logger.exception("Error during Home Assistant OAuth callback")
698 return AuthResult(success=False, error=str(e))
699
700 async def _get_external_ha_url(self) -> str | None:
701 """
702 Get the external URL for Home Assistant from the config API.
703
704 This is needed when MA runs as HA add-on and connects via internal docker network
705 (http://supervisor/api) but needs the external URL for OAuth redirects.
706
707 :return: External URL if available, otherwise None.
708 """
709 ha_url = (
710 cast("str", self.config.get("ha_url")).strip() if self.config.get("ha_url") else None
711 )
712 if not ha_url:
713 return None
714
715 # Check if we're using the internal supervisor URL
716 if "supervisor" not in ha_url.lower():
717 # Not using internal URL, return as-is
718 return ha_url
719
720 # We're using internal URL - try to get external URL from HA provider
721 ha_provider = self.mass.get_provider("hass")
722 if not ha_provider:
723 # No HA provider available, use configured URL
724 return ha_url
725
726 ha_provider = cast("HomeAssistantProvider", ha_provider)
727
728 try:
729 # Access the hass client from the provider
730 hass_client = ha_provider.hass
731 if not hass_client or not hass_client.connected:
732 return ha_url
733
734 # Get network URLs from Home Assistant using WebSocket API
735 # This command returns internal, external, and cloud URLs
736 network_urls = await hass_client.send_command("network/url")
737
738 if network_urls:
739 # Priority: external > cloud > internal
740 # External is the manually configured external URL
741 # Cloud is the Nabu Casa cloud URL
742 # Internal is the local network URL
743 external_url = network_urls.get("external")
744 cloud_url = network_urls.get("cloud")
745 internal_url = network_urls.get("internal")
746
747 # Use external URL first, then cloud, then internal
748 final_url = cast("str", external_url or cloud_url or internal_url).strip()
749 if final_url:
750 self.logger.debug(
751 "Using HA URL for OAuth: %s (from network/url, configured: %s)",
752 final_url,
753 ha_url,
754 )
755 return final_url
756 except Exception as err:
757 self.logger.warning("Failed to fetch HA network URLs: %s", err, exc_info=True)
758
759 # Fallback to configured URL
760 return ha_url
761
762 async def _fetch_ha_user_id_via_websocket(self, ha_url: str, access_token: str) -> str | None:
763 """
764 Fetch the HA user ID from Home Assistant via WebSocket using OAuth token.
765
766 :param ha_url: Home Assistant URL.
767 :param access_token: Access token for WebSocket authentication.
768 :return: The HA user ID or None if fetch fails.
769 """
770 ws_url = get_websocket_url(ha_url)
771
772 try:
773 # Use context manager to automatically handle connect/disconnect
774 async with HomeAssistantClient(ws_url, access_token, self.mass.http_session) as client:
775 # Use the auth/current_user command to get user ID
776 result = await client.send_command("auth/current_user")
777 if result and (user_id := result.get("id")):
778 return str(user_id)
779 self.logger.warning("auth/current_user returned no user data or missing id")
780 return None
781 except BaseHassClientError as ws_error:
782 self.logger.error("Failed to fetch HA user via WebSocket: %s", ws_error)
783 return None
784
785 async def _get_or_create_user(
786 self,
787 username: str,
788 display_name: str | None,
789 ha_user_id: str,
790 avatar_url: str | None = None,
791 ) -> User | None:
792 """
793 Get or create a user for Home Assistant OAuth authentication.
794
795 Updates existing users with display_name and avatar_url from HA on each OAuth login
796 (HA is considered the source of truth for these fields).
797
798 :param username: Username from Home Assistant.
799 :param display_name: Display name from Home Assistant.
800 :param ha_user_id: Home Assistant user ID.
801 :param avatar_url: Avatar URL from Home Assistant person entity.
802 :return: User object or None if creation failed.
803 """
804 # Check if user already linked to HA
805 user = await self.auth_manager.get_user_by_provider_link(
806 AuthProviderType.HOME_ASSISTANT, ha_user_id
807 )
808 if user:
809 # Update user with HA details if available (HA is source of truth)
810 if display_name or avatar_url:
811 user = await self.auth_manager.update_user(
812 user,
813 display_name=display_name,
814 avatar_url=avatar_url,
815 )
816 return user
817
818 username = normalize_username(username)
819
820 # Check if a user with this username already exists (from built-in provider)
821 user_row = await self.auth_manager.database.get_row("users", {"username": username})
822 if user_row:
823 # User exists with this username - link them to HA provider
824 user_dict = dict(user_row)
825 existing_user = User(
826 user_id=user_dict["user_id"],
827 username=user_dict["username"],
828 role=user_dict["role"],
829 enabled=bool(user_dict["enabled"]),
830 created_at=datetime.fromisoformat(user_dict["created_at"]),
831 display_name=user_dict["display_name"],
832 avatar_url=user_dict["avatar_url"],
833 )
834
835 # Link existing user to Home Assistant
836 await self.auth_manager.link_user_to_provider(
837 existing_user, AuthProviderType.HOME_ASSISTANT, ha_user_id
838 )
839
840 # Update user with HA details if available (HA is source of truth)
841 if display_name or avatar_url:
842 existing_user = await self.auth_manager.update_user(
843 existing_user,
844 display_name=display_name,
845 avatar_url=avatar_url,
846 )
847
848 return existing_user
849
850 # New HA user - check if self-registration allowed
851 if not self.allow_self_registration:
852 return None
853
854 # Determine role based on HA admin status
855 role = await get_ha_user_role(self.mass, ha_user_id)
856
857 # Create new user
858 user = await self.auth_manager.create_user(
859 username=username,
860 role=role,
861 display_name=display_name or username,
862 avatar_url=avatar_url,
863 )
864
865 # Link to Home Assistant
866 await self.auth_manager.link_user_to_provider(
867 user, AuthProviderType.HOME_ASSISTANT, ha_user_id
868 )
869
870 return user
871