/
/
/
1"""Authentication manager for Music Assistant webserver."""
2
3from __future__ import annotations
4
5import contextlib
6import hashlib
7import logging
8import secrets
9from datetime import datetime, timedelta
10from sqlite3 import OperationalError
11from typing import TYPE_CHECKING, Any
12
13import jwt as pyjwt
14from music_assistant_models.auth import (
15 AuthProviderType,
16 AuthToken,
17 User,
18 UserAuthProvider,
19 UserRole,
20)
21from music_assistant_models.errors import (
22 AuthenticationRequired,
23 InsufficientPermissions,
24 InvalidDataError,
25)
26
27from music_assistant.constants import (
28 CONF_AUTH_ALLOW_SELF_REGISTRATION,
29 DB_TABLE_PLAYLOG,
30 HOMEASSISTANT_SYSTEM_USER,
31 MASS_LOGGER_NAME,
32)
33from music_assistant.controllers.webserver.helpers.auth_middleware import (
34 get_current_token,
35 get_current_user,
36)
37from music_assistant.controllers.webserver.helpers.auth_providers import (
38 AuthResult,
39 BuiltinLoginProvider,
40 HomeAssistantOAuthProvider,
41 HomeAssistantProviderConfig,
42 LoginProvider,
43 LoginProviderConfig,
44 normalize_username,
45)
46from music_assistant.helpers.api import api_command
47from music_assistant.helpers.database import DatabaseConnection
48from music_assistant.helpers.datetime import utc
49from music_assistant.helpers.json import json_dumps, json_loads
50from music_assistant.helpers.jwt_auth import JWTHelper
51
52if TYPE_CHECKING:
53 from music_assistant.controllers.webserver import WebserverController
54
55LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.auth")
56
57# Database schema version
58DB_SCHEMA_VERSION = 4
59
60# Token expiration constants (in days)
61TOKEN_SHORT_LIVED_EXPIRATION = 30 # Short-lived tokens (auto-renewing on use)
62TOKEN_LONG_LIVED_EXPIRATION = 3650 # Long-lived tokens (10 years, no auto-renewal)
63
64
65class AuthenticationManager:
66 """Manager for authentication and user management (part of webserver controller)."""
67
68 def __init__(self, webserver: WebserverController) -> None:
69 """
70 Initialize the authentication manager.
71
72 :param webserver: WebserverController instance.
73 """
74 self.webserver = webserver
75 self.mass = webserver.mass
76 self.database: DatabaseConnection = None # type: ignore[assignment]
77 self.login_providers: dict[str, LoginProvider] = {}
78 self.logger = LOGGER
79 self._has_users: bool = False
80 self.jwt_helper: JWTHelper = None # type: ignore[assignment]
81
82 async def setup(self) -> None:
83 """Initialize the authentication manager."""
84 # Get auth settings from config
85 allow_self_registration = self.webserver.config.get_value(CONF_AUTH_ALLOW_SELF_REGISTRATION)
86 assert isinstance(allow_self_registration, bool)
87
88 # Setup database
89 db_path = self.mass.storage_path + "/auth.db"
90 self.database = DatabaseConnection(db_path)
91 await self.database.setup()
92
93 # Create database schema and handle migrations
94 await self._setup_database()
95
96 # Initialize JWT helper with secret key
97 jwt_secret = await self._get_or_create_jwt_secret()
98 self.jwt_helper = JWTHelper(jwt_secret)
99
100 # Setup login providers based on config
101 await self._setup_login_providers(allow_self_registration)
102
103 self._has_users = await self._has_non_system_users()
104
105 self.logger.info(
106 "Authentication manager initialized (providers=%d)", len(self.login_providers)
107 )
108
109 async def close(self) -> None:
110 """Cleanup on exit."""
111 if self.database:
112 await self.database.close()
113
114 @property
115 def has_users(self) -> bool:
116 """Check if any users exist in the system."""
117 return self._has_users
118
119 async def _setup_database(self) -> None:
120 """Set up database schema and handle migrations."""
121 # Always create tables if they don't exist
122 await self._create_database_tables()
123
124 # Check current schema version
125 try:
126 if db_row := await self.database.get_row("settings", {"key": "schema_version"}):
127 prev_version = int(db_row["value"])
128 else:
129 prev_version = DB_SCHEMA_VERSION
130 except (KeyError, ValueError, Exception):
131 # settings table doesn't exist yet or other error
132 prev_version = 0
133
134 # Perform migration if needed
135 if prev_version < DB_SCHEMA_VERSION:
136 self.logger.warning(
137 "Performing database migration from schema version %s to %s",
138 prev_version,
139 DB_SCHEMA_VERSION,
140 )
141 await self._migrate_database(prev_version)
142
143 # Store current schema version
144 await self.database.insert_or_replace(
145 "settings",
146 {"key": "schema_version", "value": str(DB_SCHEMA_VERSION), "type": "int"},
147 )
148
149 # Create indexes
150 await self._create_database_indexes()
151 await self.database.commit()
152
153 async def _create_database_tables(self) -> None:
154 """Create database tables."""
155 # Settings table (for schema version and other settings)
156 await self.database.execute(
157 """
158 CREATE TABLE IF NOT EXISTS settings (
159 key TEXT PRIMARY KEY,
160 value TEXT,
161 type TEXT
162 )
163 """
164 )
165 # Users table
166 await self.database.execute(
167 """
168 CREATE TABLE IF NOT EXISTS users (
169 user_id TEXT PRIMARY KEY,
170 username TEXT NOT NULL UNIQUE,
171 role TEXT NOT NULL,
172 enabled INTEGER NOT NULL DEFAULT 1,
173 created_at TEXT NOT NULL,
174 display_name TEXT,
175 avatar_url TEXT,
176 preferences json NOT NULL DEFAULT '{}',
177 player_filter json NOT NULL DEFAULT '[]',
178 provider_filter json NOT NULL DEFAULT '[]'
179 )
180 """
181 )
182 # User auth provider links (many-to-many)
183 await self.database.execute(
184 """
185 CREATE TABLE IF NOT EXISTS user_auth_providers (
186 link_id TEXT PRIMARY KEY,
187 user_id TEXT NOT NULL,
188 provider_type TEXT NOT NULL,
189 provider_user_id TEXT NOT NULL,
190 created_at TEXT NOT NULL,
191 UNIQUE(provider_type, provider_user_id),
192 FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
193 )
194 """
195 )
196 # Auth tokens table
197 await self.database.execute(
198 """
199 CREATE TABLE IF NOT EXISTS auth_tokens (
200 token_id TEXT PRIMARY KEY,
201 user_id TEXT NOT NULL,
202 token_hash TEXT NOT NULL UNIQUE,
203 name TEXT NOT NULL,
204 created_at TEXT NOT NULL,
205 expires_at TEXT,
206 last_used_at TEXT,
207 is_long_lived INTEGER NOT NULL DEFAULT 0,
208 FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
209 )
210 """
211 )
212 await self.database.commit()
213
214 async def _create_database_indexes(self) -> None:
215 """Create database indexes."""
216 await self.database.execute(
217 "CREATE INDEX IF NOT EXISTS idx_user_auth_providers_user "
218 "ON user_auth_providers(user_id)"
219 )
220 await self.database.execute(
221 "CREATE INDEX IF NOT EXISTS idx_user_auth_providers_provider "
222 "ON user_auth_providers(provider_type, provider_user_id)"
223 )
224 await self.database.execute(
225 "CREATE INDEX IF NOT EXISTS idx_tokens_user ON auth_tokens(user_id)"
226 )
227 await self.database.execute(
228 "CREATE INDEX IF NOT EXISTS idx_tokens_hash ON auth_tokens(token_hash)"
229 )
230
231 async def _migrate_database(self, from_version: int) -> None:
232 """Perform database migration.
233
234 :param from_version: The schema version to migrate from.
235 """
236 self.logger.info(
237 "Migrating auth database from version %s to %s", from_version, DB_SCHEMA_VERSION
238 )
239 # Migration to version 2: Recreate tables due to password salt breaking change
240 if from_version < 2:
241 # Drop all auth-related tables
242 await self.database.execute("DROP TABLE IF EXISTS auth_tokens")
243 await self.database.execute("DROP TABLE IF EXISTS user_auth_providers")
244 await self.database.execute("DROP TABLE IF EXISTS users")
245 await self.database.commit()
246
247 # Recreate tables with current schema
248 await self._create_database_tables()
249
250 # Migration to version 3: Add player_filter and provider_filter columns
251 if from_version < 3:
252 with contextlib.suppress(OperationalError):
253 # Column(s) may already exist
254 await self.database.execute(
255 "ALTER TABLE users ADD COLUMN player_filter json NOT NULL DEFAULT '[]'"
256 )
257 await self.database.execute(
258 "ALTER TABLE users ADD COLUMN provider_filter json NOT NULL DEFAULT '[]'"
259 )
260 await self.database.commit()
261
262 # Migration to version 4: Make usernames case-insensitive by converting to lowercase
263 if from_version < 4:
264 await self.database.execute("UPDATE users SET username = LOWER(username)")
265 await self.database.commit()
266
267 async def _get_or_create_jwt_secret(self) -> str:
268 """Get or create JWT secret key from database.
269
270 :return: JWT secret key for signing tokens.
271 """
272 # Try to get existing secret
273 if secret_row := await self.database.get_row("settings", {"key": "jwt_secret"}):
274 return str(secret_row["value"])
275
276 # Generate new secret
277 jwt_secret = JWTHelper.generate_secret_key()
278
279 # Store in database
280 await self.database.insert_or_replace(
281 "settings",
282 {"key": "jwt_secret", "value": jwt_secret, "type": "string"},
283 )
284 await self.database.commit()
285
286 self.logger.info("Generated new JWT secret key")
287 return jwt_secret
288
289 async def _setup_login_providers(self, allow_self_registration: bool) -> None:
290 """
291 Set up available login providers based on configuration.
292
293 :param allow_self_registration: Whether to allow self-registration via OAuth.
294 """
295 # Always enable built-in provider
296 builtin_config: LoginProviderConfig = {"allow_self_registration": False}
297 self.login_providers["builtin"] = BuiltinLoginProvider(self.mass, "builtin", builtin_config)
298
299 # Home Assistant OAuth provider
300 # Automatically enabled if HA provider (plugin) is configured
301 ha_provider = None
302 for provider in self.mass.providers:
303 if provider.domain == "hass" and provider.available:
304 ha_provider = provider
305 break
306
307 if ha_provider:
308 # Get URL from the HA provider config
309 ha_url = ha_provider.config.get_value("url")
310 assert isinstance(ha_url, str)
311 ha_config: HomeAssistantProviderConfig = {
312 "ha_url": ha_url,
313 "allow_self_registration": allow_self_registration,
314 }
315 self.login_providers["homeassistant"] = HomeAssistantOAuthProvider(
316 self.mass, "homeassistant", ha_config
317 )
318 self.logger.info(
319 "Home Assistant OAuth provider enabled (using URL from HA provider: %s)",
320 ha_url,
321 )
322
323 async def _sync_ha_oauth_provider(self) -> None:
324 """
325 Sync HA OAuth provider with HA provider availability (dynamic check).
326
327 Adds the provider if HA is available, removes it if HA is not available.
328 """
329 # Find HA provider
330 ha_provider = None
331 for provider in self.mass.providers:
332 if provider.domain == "hass" and provider.available:
333 ha_provider = provider
334 break
335
336 if ha_provider:
337 # HA provider exists and is available - ensure OAuth provider is registered
338 if "homeassistant" not in self.login_providers:
339 # Get allow_self_registration config
340 allow_self_registration = bool(
341 self.webserver.config.get_value(CONF_AUTH_ALLOW_SELF_REGISTRATION, True)
342 )
343
344 # Get URL from the HA provider config
345 ha_url = ha_provider.config.get_value("url")
346 assert isinstance(ha_url, str)
347 ha_config: HomeAssistantProviderConfig = {
348 "ha_url": ha_url,
349 "allow_self_registration": allow_self_registration,
350 }
351 self.login_providers["homeassistant"] = HomeAssistantOAuthProvider(
352 self.mass, "homeassistant", ha_config
353 )
354 self.logger.info(
355 "Home Assistant OAuth provider dynamically enabled (using URL: %s)",
356 ha_url,
357 )
358 # HA provider not available - remove OAuth provider if present
359 elif "homeassistant" in self.login_providers:
360 del self.login_providers["homeassistant"]
361 self.logger.info("Home Assistant OAuth provider removed (HA provider not available)")
362
363 async def authenticate_with_credentials(
364 self, provider_id: str, credentials: dict[str, Any]
365 ) -> AuthResult:
366 """
367 Authenticate a user with credentials.
368
369 :param provider_id: The login provider ID.
370 :param credentials: Provider-specific credentials.
371 """
372 provider = self.login_providers.get(provider_id)
373 if not provider:
374 return AuthResult(success=False, error="Invalid provider")
375
376 return await provider.authenticate(credentials)
377
378 async def authenticate_with_token(self, token: str) -> User | None:
379 """
380 Authenticate a user with an access token.
381
382 Supports both JWT tokens and legacy hash-based tokens for backward compatibility.
383
384 :param token: The access token (JWT or legacy hash token).
385 """
386 # Try to decode as JWT first
387 try:
388 payload = self.jwt_helper.decode_token(token, verify_exp=True)
389 token_id = payload.get("jti")
390 user_id = payload.get("sub")
391 is_long_lived = payload.get("is_long_lived", False)
392
393 if not token_id or not user_id:
394 return None
395
396 token_row = await self.database.get_row("auth_tokens", {"token_id": token_id})
397 if not token_row:
398 return None
399
400 # Database expiration is source of truth
401 if token_row["expires_at"]:
402 db_expires_at = datetime.fromisoformat(token_row["expires_at"])
403 if utc() > db_expires_at:
404 await self.database.delete("auth_tokens", {"token_id": token_id})
405 return None
406
407 # Update last used timestamp
408 now = utc()
409 updates = {"last_used_at": now.isoformat()}
410
411 if not is_long_lived:
412 # Short-lived token: extend expiration on each use (sliding window)
413 new_expires_at = now + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
414 updates["expires_at"] = new_expires_at.isoformat()
415
416 # Update database
417 await self.database.update(
418 "auth_tokens",
419 {"token_id": token_id},
420 updates,
421 )
422
423 return await self.get_user(user_id)
424
425 except pyjwt.ExpiredSignatureError:
426 if token_id := self.jwt_helper.get_token_id(token):
427 await self.database.delete("auth_tokens", {"token_id": token_id})
428 return None
429 except pyjwt.InvalidTokenError:
430 self.logger.debug("Token is not a valid JWT, trying legacy hash lookup")
431 except Exception as err:
432 self.logger.debug("Error decoding JWT token: %s, trying legacy hash lookup", err)
433
434 # Fallback to legacy hash-based token lookup
435 token_hash = hashlib.sha256(token.encode()).hexdigest()
436 token_row = await self.database.get_row("auth_tokens", {"token_hash": token_hash})
437 if not token_row:
438 return None
439
440 # Check if token is expired
441 if token_row["expires_at"]:
442 expires_at = datetime.fromisoformat(token_row["expires_at"])
443 if utc() > expires_at:
444 # Token expired, delete it
445 await self.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
446 return None
447
448 # Implement sliding expiration for short-lived tokens
449 is_long_lived = bool(token_row["is_long_lived"])
450 now = utc()
451 updates = {"last_used_at": now.isoformat()}
452
453 if not is_long_lived and token_row["expires_at"]:
454 # Short-lived token: extend expiration on each use (sliding window)
455 new_expires_at = now + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
456 updates["expires_at"] = new_expires_at.isoformat()
457
458 # Update last used timestamp and potentially expiration
459 await self.database.update(
460 "auth_tokens",
461 {"token_id": token_row["token_id"]},
462 updates,
463 )
464
465 # Get user
466 return await self.get_user(token_row["user_id"])
467
468 async def get_token_id_from_token(self, token: str) -> str | None:
469 """
470 Get token_id from a token string (for tracking revocation).
471
472 :param token: The access token (JWT or legacy hash token).
473 :return: The token_id or None if token not found.
474 """
475 # Try to extract from JWT first
476 if token_id := self.jwt_helper.get_token_id(token):
477 return token_id
478
479 # Fallback: Hash-based lookup for legacy tokens
480 token_hash = hashlib.sha256(token.encode()).hexdigest()
481 token_row = await self.database.get_row("auth_tokens", {"token_hash": token_hash})
482 if not token_row:
483 return None
484
485 return str(token_row["token_id"])
486
487 @api_command("auth/user", required_role="admin")
488 async def get_user(self, user_id: str) -> User | None:
489 """
490 Get user by ID (admin only).
491
492 :param user_id: The user ID.
493 :return: User object or None if not found.
494 """
495 user_row = await self.database.get_row("users", {"user_id": user_id})
496 if not user_row or not user_row["enabled"]:
497 return None
498
499 return User(
500 user_id=user_row["user_id"],
501 username=user_row["username"],
502 role=UserRole(user_row["role"]),
503 enabled=bool(user_row["enabled"]),
504 created_at=datetime.fromisoformat(user_row["created_at"]),
505 display_name=user_row["display_name"],
506 avatar_url=user_row["avatar_url"],
507 preferences=json_loads(user_row["preferences"]),
508 player_filter=json_loads(user_row["player_filter"]),
509 provider_filter=json_loads(user_row["provider_filter"]),
510 )
511
512 async def get_user_by_username(self, username: str) -> User | None:
513 """
514 Get user by username.
515
516 :param username: The username.
517 :return: User object or None if not found.
518 """
519 username = normalize_username(username)
520
521 user_row = await self.database.get_row("users", {"username": username})
522 if not user_row:
523 return None
524
525 return await self.get_user(user_row["user_id"])
526
527 async def get_user_by_provider_link(
528 self, provider_type: AuthProviderType, provider_user_id: str
529 ) -> User | None:
530 """
531 Get user by their provider link.
532
533 :param provider_type: The auth provider type.
534 :param provider_user_id: The user ID from the provider.
535 """
536 link_row = await self.database.get_row(
537 "user_auth_providers",
538 {
539 "provider_type": provider_type.value,
540 "provider_user_id": provider_user_id,
541 },
542 )
543 if not link_row:
544 return None
545
546 return await self.get_user(link_row["user_id"])
547
548 async def create_user(
549 self,
550 username: str,
551 role: UserRole = UserRole.USER,
552 display_name: str | None = None,
553 avatar_url: str | None = None,
554 preferences: dict[str, Any] | None = None,
555 player_filter: list[str] | None = None,
556 provider_filter: list[str] | None = None,
557 ) -> User:
558 """
559 Create a new user.
560
561 :param username: The username.
562 :param role: The user role (default: USER).
563 :param display_name: Optional display name.
564 :param avatar_url: Optional avatar URL.
565 :param preferences: Optional user preferences dict.
566 :param player_filter: Optional list of player IDs user has access to.
567 :param provider_filter: Optional list of provider instance IDs user has access to.
568 """
569 normalized_username = normalize_username(username)
570
571 # Check if this is the first non-system user
572 is_first_user = not await self._has_non_system_users()
573
574 user_id = secrets.token_urlsafe(32)
575 created_at = utc()
576 if preferences is None:
577 preferences = {}
578 if player_filter is None:
579 player_filter = []
580 if provider_filter is None:
581 provider_filter = []
582
583 user_data = {
584 "user_id": user_id,
585 "username": normalized_username,
586 "role": role.value,
587 "enabled": True,
588 "created_at": created_at.isoformat(),
589 "display_name": display_name,
590 "avatar_url": avatar_url,
591 "preferences": json_dumps(preferences),
592 "player_filter": json_dumps(player_filter),
593 "provider_filter": json_dumps(provider_filter),
594 }
595
596 await self.database.insert("users", user_data)
597
598 user = User(
599 user_id=user_id,
600 username=normalized_username,
601 role=role,
602 enabled=True,
603 created_at=created_at,
604 display_name=display_name,
605 avatar_url=avatar_url,
606 preferences=preferences,
607 player_filter=player_filter,
608 provider_filter=provider_filter,
609 )
610
611 # If this is the first non-system user, migrate playlog entries to them
612 if is_first_user and normalized_username != HOMEASSISTANT_SYSTEM_USER:
613 self._has_users = True
614 await self._migrate_playlog_to_first_user(user_id)
615
616 return user
617
618 async def _has_non_system_users(self) -> bool:
619 """Check if any non-system users exist."""
620 user_rows = await self.database.get_rows("users", limit=10)
621 return any(row["username"] != HOMEASSISTANT_SYSTEM_USER for row in user_rows)
622
623 async def _migrate_playlog_to_first_user(self, user_id: str) -> None:
624 """
625 Migrate all existing playlog entries to the first user.
626
627 This is called automatically when the first non-system user is created.
628 All existing playlog entries (which have NULL userid) will be updated
629 to belong to this first user.
630
631 :param user_id: The user ID of the first user.
632 """
633 try:
634 # Update all playlog entries with NULL userid to this user
635 await self.mass.music.database.execute(
636 f"UPDATE {DB_TABLE_PLAYLOG} SET userid = :userid WHERE userid IS NULL",
637 {"userid": user_id},
638 )
639 await self.mass.music.database.commit()
640 self.logger.info("Migrated existing playlog entries to first user: %s", user_id)
641 except Exception as err:
642 self.logger.warning("Failed to migrate playlog entries: %s", err)
643
644 async def get_homeassistant_system_user(self) -> User:
645 """
646 Get or create the Home Assistant system user.
647
648 This is a special system user created automatically for Home Assistant integration.
649 It bypasses normal authentication but is restricted to the ingress webserver.
650
651 :return: The Home Assistant system user.
652 """
653 username = HOMEASSISTANT_SYSTEM_USER
654 display_name = "Home Assistant Integration"
655 role = UserRole.USER
656
657 normalized_username = normalize_username(username)
658
659 # Try to find existing user by username
660 user_row = await self.database.get_row("users", {"username": normalized_username})
661 if user_row:
662 # Use get_user to ensure preferences are parsed correctly
663 user = await self.get_user(user_row["user_id"])
664 assert user is not None # User exists in DB, so get_user must return it
665 return user
666
667 # Create new system user
668 user = await self.create_user(
669 username=username,
670 role=role,
671 display_name=display_name,
672 )
673 self.logger.debug("Created Home Assistant system user: %s (role: %s)", username, role.value)
674 return user
675
676 async def get_homeassistant_system_user_token(self) -> str:
677 """
678 Get or create an auth token for the Home Assistant system user.
679
680 This method ensures only one active token exists for the HA integration.
681 If an old token exists, it is deleted and a new one is created.
682 The token auto-renews on use (expires after 30 days of inactivity).
683
684 :return: Authentication token for the Home Assistant system user.
685 """
686 token_name = "Home Assistant Integration"
687
688 # Get the system user
689 system_user = await self.get_homeassistant_system_user()
690
691 # Delete any existing tokens with this name to avoid accumulation
692 # We can't retrieve the plain token from the hash, so we always create a new one
693 existing_tokens = await self.database.get_rows(
694 "auth_tokens",
695 {"user_id": system_user.user_id, "name": token_name},
696 )
697 for token_row in existing_tokens:
698 await self.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
699
700 # Create a new token for the system user
701 return await self.create_token(
702 user=system_user,
703 name=token_name,
704 is_long_lived=False,
705 )
706
707 async def link_user_to_provider(
708 self,
709 user: User,
710 provider_type: AuthProviderType,
711 provider_user_id: str,
712 ) -> UserAuthProvider:
713 """
714 Link a user to an authentication provider.
715
716 If a link already exists for this provider/provider_user_id, returns the existing link.
717
718 :param user: The user to link.
719 :param provider_type: The provider type.
720 :param provider_user_id: The user ID from the provider (e.g., password hash, OAuth ID).
721 """
722 # Check if a link already exists for this provider/provider_user_id
723 existing_link = await self.database.get_row(
724 "user_auth_providers",
725 {
726 "provider_type": provider_type.value,
727 "provider_user_id": provider_user_id,
728 },
729 )
730
731 if existing_link:
732 # Link already exists - return it
733 return UserAuthProvider(
734 link_id=existing_link["link_id"],
735 user_id=existing_link["user_id"],
736 provider_type=AuthProviderType(existing_link["provider_type"]),
737 provider_user_id=existing_link["provider_user_id"],
738 created_at=datetime.fromisoformat(existing_link["created_at"]),
739 )
740
741 # Create new link
742 link_id = secrets.token_urlsafe(32)
743 created_at = utc()
744 link_data = {
745 "link_id": link_id,
746 "user_id": user.user_id,
747 "provider_type": provider_type.value,
748 "provider_user_id": provider_user_id,
749 "created_at": created_at.isoformat(),
750 }
751
752 await self.database.insert("user_auth_providers", link_data)
753
754 return UserAuthProvider(
755 link_id=link_id,
756 user_id=user.user_id,
757 provider_type=provider_type,
758 provider_user_id=provider_user_id,
759 created_at=created_at,
760 )
761
762 async def update_user(
763 self,
764 user: User,
765 username: str | None = None,
766 display_name: str | None = None,
767 avatar_url: str | None = None,
768 ) -> User:
769 """
770 Update a user's profile information.
771
772 :param user: The user to update.
773 :param username: New username (optional).
774 :param display_name: New display name (optional).
775 :param avatar_url: New avatar URL (optional).
776 """
777 updates = {}
778 if username is not None:
779 # Normalize username for case-insensitive authentication
780 updates["username"] = normalize_username(username)
781 if display_name is not None:
782 updates["display_name"] = display_name
783 if avatar_url is not None:
784 updates["avatar_url"] = avatar_url
785
786 if updates:
787 await self.database.update("users", {"user_id": user.user_id}, updates)
788
789 # Return updated user
790 updated_user = await self.get_user(user.user_id)
791 assert updated_user is not None # User exists, so get_user must return it
792 return updated_user
793
794 async def update_user_preferences(
795 self,
796 user: User,
797 preferences: dict[str, Any],
798 ) -> User:
799 """
800 Update a user's preferences.
801
802 :param user: The user to update.
803 :param preferences: New preferences dict (completely replaces existing preferences).
804 """
805 # Verify user exists
806 current_user = await self.get_user(user.user_id)
807 if not current_user:
808 raise ValueError(f"User {user.user_id} not found")
809
810 # Update database with new preferences (complete replacement)
811 await self.database.update(
812 "users",
813 {"user_id": user.user_id},
814 {"preferences": json_dumps(preferences)},
815 )
816
817 # Return updated user
818 updated_user = await self.get_user(user.user_id)
819 assert updated_user is not None # User exists, so get_user must return it
820 return updated_user
821
822 async def update_provider_link(
823 self,
824 user: User,
825 provider_type: AuthProviderType,
826 provider_user_id: str,
827 ) -> None:
828 """
829 Update a user's provider link (e.g., change password).
830
831 :param user: The user.
832 :param provider_type: The provider type.
833 :param provider_user_id: The new provider user ID (e.g., new password hash).
834 """
835 # Find existing link
836 link_row = await self.database.get_row(
837 "user_auth_providers",
838 {
839 "user_id": user.user_id,
840 "provider_type": provider_type.value,
841 },
842 )
843
844 if link_row:
845 # Update existing link
846 await self.database.update(
847 "user_auth_providers",
848 {"link_id": link_row["link_id"]},
849 {"provider_user_id": provider_user_id},
850 )
851 else:
852 # Create new link
853 await self.link_user_to_provider(user, provider_type, provider_user_id)
854
855 async def create_token(self, user: User, name: str, is_long_lived: bool = False) -> str:
856 """
857 Create a new access token for a user.
858
859 :param user: The user to create the token for.
860 :param name: A name/description for the token (e.g., device name).
861 :param is_long_lived: Whether this is a long-lived token (default: False).
862 Short-lived tokens (False): Auto-renewing on use, expire after 30 days of inactivity.
863 Long-lived tokens (True): No auto-renewal, expire after 10 years.
864 """
865 # Generate unique token ID
866 token_id = secrets.token_urlsafe(32)
867
868 # Calculate expiration based on token type
869 created_at = utc()
870 if is_long_lived:
871 # Long-lived tokens expire after 10 years (no auto-renewal)
872 expires_at = created_at + timedelta(days=TOKEN_LONG_LIVED_EXPIRATION)
873 else:
874 # Short-lived tokens expire after 30 days (with auto-renewal on use)
875 expires_at = created_at + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
876
877 # Generate JWT token
878 token = self.jwt_helper.encode_token(
879 user=user,
880 token_id=token_id,
881 token_name=name,
882 expires_at=expires_at,
883 is_long_lived=is_long_lived,
884 )
885
886 # Store token hash in database for revocation checking
887 token_hash = hashlib.sha256(token.encode()).hexdigest()
888 token_data = {
889 "token_id": token_id,
890 "user_id": user.user_id,
891 "token_hash": token_hash,
892 "name": name,
893 "created_at": created_at.isoformat(),
894 "expires_at": expires_at.isoformat(),
895 "is_long_lived": 1 if is_long_lived else 0,
896 }
897 await self.database.insert("auth_tokens", token_data)
898
899 return token
900
901 @api_command("auth/token/revoke")
902 async def revoke_token(self, token_id: str) -> None:
903 """
904 Revoke an auth token.
905
906 :param token_id: The token ID to revoke.
907 """
908 user = get_current_user()
909 if not user:
910 raise AuthenticationRequired("Not authenticated")
911
912 token_row = await self.database.get_row("auth_tokens", {"token_id": token_id})
913 if not token_row:
914 raise InvalidDataError("Token not found")
915
916 # Check permissions - users can only revoke their own tokens unless admin
917 if token_row["user_id"] != user.user_id and user.role != UserRole.ADMIN:
918 raise InsufficientPermissions("You can only revoke your own tokens")
919
920 await self.database.delete("auth_tokens", {"token_id": token_id})
921
922 # Disconnect any WebSocket connections using this token
923 self.webserver.disconnect_websockets_for_token(token_id)
924
925 @api_command("auth/tokens")
926 async def get_user_tokens(self, user_id: str | None = None) -> list[AuthToken]:
927 """
928 Get current user's auth tokens or another user's tokens (admin only).
929
930 :param user_id: Optional user ID to get tokens for (admin only).
931 :return: List of auth tokens.
932 """
933 current_user = get_current_user()
934 if not current_user:
935 return []
936
937 # If user_id is provided and different from current user, require admin
938 if user_id and user_id != current_user.user_id:
939 if current_user.role != UserRole.ADMIN:
940 return []
941 target_user = await self.get_user(user_id)
942 if not target_user:
943 return []
944 else:
945 target_user = current_user
946
947 token_rows = await self.database.get_rows(
948 "auth_tokens", {"user_id": target_user.user_id}, limit=100
949 )
950 return [AuthToken.from_dict(dict(row)) for row in token_rows]
951
952 @api_command("auth/users", required_role="admin")
953 async def list_users(self) -> list[User]:
954 """
955 Get all users (admin only).
956
957 System users are excluded from the list.
958
959 :return: List of user objects.
960 """
961 user_rows = await self.database.get_rows("users", limit=1000)
962 users = []
963 for row in user_rows:
964 # Skip system users
965 if row["username"] == HOMEASSISTANT_SYSTEM_USER:
966 continue
967 users.append(
968 User(
969 user_id=row["user_id"],
970 username=row["username"],
971 role=UserRole(row["role"]),
972 enabled=bool(row["enabled"]),
973 created_at=datetime.fromisoformat(row["created_at"]),
974 display_name=row["display_name"],
975 avatar_url=row["avatar_url"],
976 preferences=json_loads(row["preferences"]),
977 player_filter=json_loads(row["player_filter"]),
978 provider_filter=json_loads(row["provider_filter"]),
979 )
980 )
981 return users
982
983 async def update_user_role(self, user_id: str, new_role: UserRole, admin_user: User) -> bool:
984 """
985 Update a user's role (admin only).
986
987 :param user_id: The user ID to update.
988 :param new_role: The new role to assign.
989 :param admin_user: The admin user performing the action.
990 """
991 if admin_user.role != UserRole.ADMIN:
992 return False
993
994 user_row = await self.database.get_row("users", {"user_id": user_id})
995 if not user_row:
996 return False
997
998 await self.database.update(
999 "users",
1000 {"user_id": user_id},
1001 {"role": new_role.value},
1002 )
1003 return True
1004
1005 @api_command("auth/user/enable", required_role="admin")
1006 async def enable_user(self, user_id: str) -> None:
1007 """
1008 Enable user account (admin only).
1009
1010 :param user_id: The user ID.
1011 """
1012 await self.database.update(
1013 "users",
1014 {"user_id": user_id},
1015 {"enabled": 1},
1016 )
1017
1018 @api_command("auth/user/disable", required_role="admin")
1019 async def disable_user(self, user_id: str) -> None:
1020 """
1021 Disable user account (admin only).
1022
1023 :param user_id: The user ID.
1024 """
1025 admin_user = get_current_user()
1026 if not admin_user:
1027 raise AuthenticationRequired("Not authenticated")
1028
1029 # Cannot disable yourself
1030 if user_id == admin_user.user_id:
1031 raise InvalidDataError("Cannot disable your own account")
1032
1033 await self.database.update(
1034 "users",
1035 {"user_id": user_id},
1036 {"enabled": 0},
1037 )
1038
1039 # Disconnect all WebSocket connections for this user
1040 self.webserver.disconnect_websockets_for_user(user_id)
1041
1042 async def get_login_providers(self) -> list[dict[str, Any]]:
1043 """Get list of available login providers (dynamically checks for HA provider)."""
1044 # Sync HA OAuth provider with HA provider availability
1045 await self._sync_ha_oauth_provider()
1046
1047 providers = []
1048 for provider_id, provider in self.login_providers.items():
1049 providers.append(
1050 {
1051 "provider_id": provider_id,
1052 "provider_type": provider.provider_type.value,
1053 "requires_redirect": provider.requires_redirect,
1054 }
1055 )
1056 return providers
1057
1058 @api_command("auth/login", authenticated=False)
1059 async def login(
1060 self,
1061 username: str | None = None,
1062 password: str | None = None,
1063 provider_id: str = "builtin",
1064 device_name: str | None = None,
1065 **extra_credentials: Any,
1066 ) -> dict[str, Any]:
1067 """Authenticate user with credentials via WebSocket.
1068
1069 This command allows clients to authenticate over the WebSocket connection
1070 using username/password or other provider-specific credentials.
1071
1072 :param username: Username for authentication (for builtin provider).
1073 :param password: Password for authentication (for builtin provider).
1074 :param provider_id: The login provider ID (defaults to "builtin").
1075 :param device_name: Optional device name for the token (e.g., "iPhone 15", "Desktop PC").
1076 :param extra_credentials: Additional provider-specific credentials.
1077 :return: Authentication result with access token if successful.
1078 """
1079 # Build credentials dict from parameters
1080 credentials: dict[str, Any] = {}
1081 if username is not None:
1082 credentials["username"] = username
1083 if password is not None:
1084 credentials["password"] = password
1085 credentials.update(extra_credentials)
1086
1087 auth_result = await self.authenticate_with_credentials(provider_id, credentials)
1088
1089 if not auth_result.success:
1090 return {
1091 "success": False,
1092 "error": auth_result.error or "Authentication failed",
1093 }
1094
1095 if not auth_result.user:
1096 return {
1097 "success": False,
1098 "error": "Authentication failed: no user returned",
1099 }
1100
1101 # Create short-lived access token with device name if provided
1102 token_name = device_name or f"WebSocket Session - {auth_result.user.username}"
1103 token = await self.create_token(
1104 auth_result.user,
1105 is_long_lived=False,
1106 name=token_name,
1107 )
1108
1109 return {
1110 "success": True,
1111 "access_token": token,
1112 "user": {
1113 "user_id": auth_result.user.user_id,
1114 "username": auth_result.user.username,
1115 "display_name": auth_result.user.display_name,
1116 "role": auth_result.user.role.value,
1117 },
1118 }
1119
1120 @api_command("auth/providers", authenticated=False)
1121 async def get_providers(self) -> list[dict[str, Any]]:
1122 """Get list of available authentication providers.
1123
1124 Returns information about all available login providers including
1125 whether they require OAuth redirect flow.
1126 """
1127 return await self.get_login_providers()
1128
1129 @api_command("auth/authorization_url", authenticated=False)
1130 async def get_auth_url(
1131 self,
1132 provider_id: str,
1133 return_url: str | None = None,
1134 ) -> dict[str, str | None]:
1135 """Get OAuth authorization URL for authentication.
1136
1137 For OAuth providers (like Home Assistant), this returns the URL that
1138 the user should visit in their browser to authorize the application.
1139
1140 :param provider_id: The provider ID (e.g., "hass").
1141 :param return_url: URL to redirect to after OAuth completes.
1142 :return: Dictionary with authorization_url.
1143 """
1144 auth_url = await self.get_authorization_url(provider_id, return_url)
1145 if not auth_url:
1146 return {
1147 "authorization_url": None,
1148 "error": "Provider does not support OAuth or does not exist",
1149 }
1150
1151 return {
1152 "authorization_url": auth_url,
1153 }
1154
1155 async def get_authorization_url(
1156 self, provider_id: str, return_url: str | None = None
1157 ) -> str | None:
1158 """
1159 Get OAuth authorization URL for a provider.
1160
1161 :param provider_id: The provider ID.
1162 :param return_url: Optional URL to redirect to after successful login.
1163 """
1164 provider = self.login_providers.get(provider_id)
1165 if not provider or not provider.requires_redirect:
1166 return None
1167
1168 # Build callback redirect_uri
1169 redirect_uri = f"{self.webserver.base_url}/auth/callback?provider_id={provider_id}"
1170 return await provider.get_authorization_url(redirect_uri, return_url)
1171
1172 async def handle_oauth_callback(
1173 self, provider_id: str, code: str, state: str, redirect_uri: str
1174 ) -> AuthResult:
1175 """
1176 Handle OAuth callback.
1177
1178 :param provider_id: The provider ID.
1179 :param code: OAuth authorization code.
1180 :param state: OAuth state parameter.
1181 :param redirect_uri: The callback URL.
1182 """
1183 provider = self.login_providers.get(provider_id)
1184 if not provider:
1185 return AuthResult(success=False, error="Invalid provider")
1186
1187 return await provider.handle_oauth_callback(code, state, redirect_uri)
1188
1189 @api_command("auth/token/create")
1190 async def create_long_lived_token(self, name: str, user_id: str | None = None) -> str:
1191 """
1192 Create a new long-lived access token for current user or another user (admin only).
1193
1194 Long-lived tokens are intended for external integrations and API access.
1195 They expire after 10 years and do NOT auto-renew on use.
1196
1197 Short-lived tokens (for regular user sessions) are only created during login
1198 and auto-renew on each use (sliding 30-day expiration window).
1199
1200 :param name: The name/description for the token (e.g., "Home Assistant", "Mobile App").
1201 :param user_id: Optional user ID to create token for (admin only).
1202 :return: The created token string.
1203 """
1204 current_user = get_current_user()
1205 if not current_user:
1206 raise AuthenticationRequired("Not authenticated")
1207
1208 # If user_id is provided and different from current user, require admin
1209 if user_id and user_id != current_user.user_id:
1210 if current_user.role != UserRole.ADMIN:
1211 raise InsufficientPermissions(
1212 "Admin access required to create tokens for other users"
1213 )
1214 target_user = await self.get_user(user_id)
1215 if not target_user:
1216 raise InvalidDataError("User not found")
1217 else:
1218 target_user = current_user
1219
1220 # Create a long-lived token (only long-lived tokens can be created via this command)
1221 token = await self.create_token(target_user, name, is_long_lived=True)
1222 self.logger.info("Created long-lived token '%s' for user '%s'", name, target_user.username)
1223 return token
1224
1225 @api_command("auth/user/create", required_role="admin")
1226 async def create_user_with_api(
1227 self,
1228 username: str,
1229 password: str,
1230 role: str = "user",
1231 display_name: str | None = None,
1232 avatar_url: str | None = None,
1233 player_filter: list[str] | None = None,
1234 provider_filter: list[str] | None = None,
1235 ) -> User:
1236 """
1237 Create a new user with built-in authentication (admin only).
1238
1239 :param username: The username (minimum 2 characters).
1240 :param password: The password (minimum 8 characters).
1241 :param role: User role - "admin" or "user" (default: "user").
1242 :param display_name: Optional display name.
1243 :param avatar_url: Optional avatar URL.
1244 :param player_filter: Optional list of player IDs user has access to.
1245 :param provider_filter: Optional list of provider instance IDs user has access to.
1246 :return: Created user object.
1247 """
1248 # Validation
1249 if not username or len(username) < 2:
1250 raise InvalidDataError("Username must be at least 2 characters")
1251
1252 if not password or len(password) < 8:
1253 raise InvalidDataError("Password must be at least 8 characters")
1254
1255 # Validate role
1256 try:
1257 user_role = UserRole(role)
1258 except ValueError as err:
1259 raise InvalidDataError("Invalid role. Must be 'admin' or 'user'") from err
1260
1261 # Get built-in provider
1262 builtin_provider = self.login_providers.get("builtin")
1263 if not builtin_provider or not isinstance(builtin_provider, BuiltinLoginProvider):
1264 raise InvalidDataError("Built-in auth provider not available")
1265
1266 # Create user with password
1267 user = await builtin_provider.create_user_with_password(
1268 username,
1269 password,
1270 role=user_role,
1271 player_filter=player_filter,
1272 provider_filter=provider_filter,
1273 )
1274
1275 # Update optional fields if provided
1276 if display_name or avatar_url:
1277 updated_user = await self.update_user(
1278 user, display_name=display_name, avatar_url=avatar_url
1279 )
1280 if updated_user:
1281 user = updated_user
1282
1283 self.logger.info("User created by admin: %s (role: %s)", username, role)
1284 return user
1285
1286 @api_command("auth/user/delete", required_role="admin")
1287 async def delete_user(self, user_id: str) -> None:
1288 """
1289 Delete user account (admin only).
1290
1291 :param user_id: The user ID.
1292 """
1293 admin_user = get_current_user()
1294 if not admin_user:
1295 raise AuthenticationRequired("Not authenticated")
1296
1297 # Don't allow deleting yourself
1298 if user_id == admin_user.user_id:
1299 raise InvalidDataError("Cannot delete your own account")
1300
1301 # Delete user from database
1302 await self.database.delete("users", {"user_id": user_id})
1303 await self.database.commit()
1304
1305 # Disconnect all WebSocket connections for this user
1306 self.webserver.disconnect_websockets_for_user(user_id)
1307
1308 @api_command("auth/me")
1309 async def get_current_user_info(self) -> User:
1310 """Get current authenticated user information."""
1311 current_user_obj = get_current_user()
1312 if not current_user_obj:
1313 raise AuthenticationRequired("Not authenticated")
1314 return current_user_obj
1315
1316 async def _update_profile_password(
1317 self,
1318 target_user: User,
1319 password: str,
1320 is_admin_update: bool,
1321 current_user: User,
1322 ) -> None:
1323 """Update user password (helper method)."""
1324 if len(password) < 8:
1325 raise InvalidDataError("Password must be at least 8 characters")
1326
1327 builtin_provider = self.login_providers.get("builtin")
1328 if not builtin_provider or not isinstance(builtin_provider, BuiltinLoginProvider):
1329 raise InvalidDataError("Built-in auth not available")
1330
1331 # Update password (used for both admin resets and user password changes)
1332 await builtin_provider.reset_password(target_user, password)
1333
1334 if is_admin_update:
1335 self.logger.info(
1336 "Password reset for user %s by admin %s",
1337 target_user.username,
1338 current_user.username,
1339 )
1340 else:
1341 self.logger.info("Password changed for user %s", target_user.username)
1342
1343 async def update_user_filters(
1344 self,
1345 target_user: User,
1346 player_filter: list[str] | None,
1347 provider_filter: list[str] | None,
1348 ) -> User:
1349 """Update user player and provider filters (helper method)."""
1350 updates = {}
1351 if player_filter is not None:
1352 updates["player_filter"] = json_dumps(player_filter)
1353 if provider_filter is not None:
1354 updates["provider_filter"] = json_dumps(provider_filter)
1355
1356 if updates:
1357 await self.database.update("users", {"user_id": target_user.user_id}, updates)
1358 # Refresh target user to get updated filters
1359 refreshed_user = await self.get_user(target_user.user_id)
1360 if not refreshed_user:
1361 raise InvalidDataError("Failed to refresh user after filter update")
1362 return refreshed_user
1363 return target_user
1364
1365 @api_command("auth/user/update")
1366 async def update_user_profile(
1367 self,
1368 user_id: str | None = None,
1369 username: str | None = None,
1370 display_name: str | None = None,
1371 avatar_url: str | None = None,
1372 password: str | None = None,
1373 role: str | None = None,
1374 preferences: dict[str, Any] | None = None,
1375 player_filter: list[str] | None = None,
1376 provider_filter: list[str] | None = None,
1377 ) -> User:
1378 """
1379 Update user profile information.
1380
1381 Users can update their own profile. Admins can update any user including role and password.
1382
1383 :param user_id: User ID to update (optional, defaults to current user).
1384 :param username: New username (optional).
1385 :param display_name: New display name (optional).
1386 :param avatar_url: New avatar URL (optional).
1387 :param password: New password (optional, minimum 8 characters).
1388 :param role: New role - "admin" or "user" (optional, set by admin only).
1389 :param preferences: User preferences dict (completely replaces existing, optional).
1390 :param player_filter: List of player IDs user has access to (set by admin only, optional).
1391 :param provider_filter: List of provider instance IDs user has access to (set by admin only, optional).
1392 :return: Updated user object.
1393 """ # noqa: E501
1394 current_user_obj = get_current_user()
1395 if not current_user_obj:
1396 raise AuthenticationRequired("Not authenticated")
1397
1398 # Determine target user
1399 is_admin = current_user_obj.role == UserRole.ADMIN
1400 if user_id and user_id != current_user_obj.user_id:
1401 # Updating another user - requires admin
1402 if not is_admin:
1403 raise InsufficientPermissions("Admin access required")
1404 target_user = await self.get_user(user_id)
1405 if not target_user:
1406 raise InvalidDataError("User not found")
1407 else:
1408 # Updating own profile
1409 target_user = current_user_obj
1410
1411 # Update role (admin only)
1412 if role:
1413 if not is_admin:
1414 raise InsufficientPermissions("Only admins can update user roles")
1415
1416 try:
1417 new_role = UserRole(role)
1418 except ValueError as err:
1419 raise InvalidDataError("Invalid role. Must be 'admin' or 'user'") from err
1420
1421 success = await self.update_user_role(target_user.user_id, new_role, current_user_obj)
1422 if not success:
1423 raise InvalidDataError("Failed to update role")
1424
1425 # Refresh target user to get updated role
1426 refreshed_user = await self.get_user(target_user.user_id)
1427 if not refreshed_user:
1428 raise InvalidDataError("Failed to refresh user after role update")
1429 target_user = refreshed_user
1430
1431 # Update basic profile fields
1432 if username or display_name or avatar_url:
1433 updated_user = await self.update_user(
1434 target_user,
1435 username=username,
1436 display_name=display_name,
1437 avatar_url=avatar_url,
1438 )
1439 if not updated_user:
1440 raise InvalidDataError("Failed to update user profile")
1441 target_user = updated_user
1442
1443 # Update preferences if provided
1444 if preferences is not None:
1445 target_user = await self.update_user_preferences(target_user, preferences)
1446
1447 # Update player_filter and provider_filter (admin only)
1448 if player_filter is not None or provider_filter is not None:
1449 if not is_admin:
1450 raise InsufficientPermissions("Only admins can update player/provider filters")
1451 target_user = await self.update_user_filters(
1452 target_user, player_filter, provider_filter
1453 )
1454
1455 # Update password if provided
1456 if password:
1457 await self._update_profile_password(target_user, password, is_admin, current_user_obj)
1458
1459 return target_user
1460
1461 @api_command("auth/logout")
1462 async def logout(self) -> None:
1463 """Logout current user by revoking the current token."""
1464 user = get_current_user()
1465 if not user:
1466 raise AuthenticationRequired("Not authenticated")
1467
1468 # Get current token from context
1469 token = get_current_token()
1470 if not token:
1471 raise InvalidDataError("No token in context")
1472
1473 # Find and revoke the token
1474 token_hash = hashlib.sha256(token.encode()).hexdigest()
1475 token_row = await self.database.get_row("auth_tokens", {"token_hash": token_hash})
1476 if token_row:
1477 await self.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
1478
1479 # Disconnect any WebSocket connections using this token
1480 self.webserver.disconnect_websockets_for_token(token_row["token_id"])
1481
1482 @api_command("auth/user/providers")
1483 async def get_my_providers(self) -> list[dict[str, Any]]:
1484 """
1485 Get current user's linked authentication providers.
1486
1487 :return: List of provider links.
1488 """
1489 user = get_current_user()
1490 if not user:
1491 return []
1492
1493 # Get provider links from database
1494 rows = await self.database.get_rows("user_auth_providers", {"user_id": user.user_id})
1495 providers = [UserAuthProvider.from_dict(dict(row)) for row in rows]
1496 return [p.to_dict() for p in providers]
1497
1498 @api_command("auth/user/unlink_provider", required_role="admin")
1499 async def unlink_provider(self, user_id: str, provider_type: str) -> bool:
1500 """
1501 Unlink authentication provider from user (admin only).
1502
1503 :param user_id: The user ID.
1504 :param provider_type: Provider type to unlink.
1505 :return: True if successful.
1506 """
1507 await self.database.delete(
1508 "user_auth_providers", {"user_id": user_id, "provider_type": provider_type}
1509 )
1510 await self.database.commit()
1511 return True
1512