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