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