music-assistant-server

1"""Authentication manager for Music Assistant webserver."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import hashlib
8import logging
9import secrets
10from collections.abc import Callable, Mapping
11from datetime import datetime, timedelta
12from sqlite3 import IntegrityError, OperationalError
13from typing import TYPE_CHECKING, Any, cast
14
15import jwt as pyjwt
16from music_assistant_models.auth import (
17    AuthProviderType,
18    AuthToken,
19    Scope,
20    User,
21    UserAuthProvider,
22    UserRole,
23)
24from music_assistant_models.errors import (
25    AuthenticationRequired,
26    InsufficientPermissions,
27    InvalidDataError,
28)
29
30from music_assistant.constants import DB_TABLE_PLAYLOG, HOMEASSISTANT_SYSTEM_USER, MASS_LOGGER_NAME
31from music_assistant.controllers.webserver.helpers.auth_middleware import (
32    ROLE_SCOPES,
33    get_current_client_id,
34    get_current_peer_address,
35    get_current_token,
36    get_current_user,
37    has_scope,
38)
39from music_assistant.controllers.webserver.helpers.auth_providers import (
40    AuthResult,
41    BuiltinLoginProvider,
42    HomeAssistantOAuthProvider,
43    HomeAssistantProviderConfig,
44    LoginProvider,
45    LoginRateLimiter,
46    normalize_username,
47)
48from music_assistant.helpers.api import api_command
49from music_assistant.helpers.database import DatabaseConnection
50from music_assistant.helpers.datetime import utc
51from music_assistant.helpers.json import json_dumps, json_loads
52from music_assistant.helpers.jwt_auth import JWTHelper
53
54if TYPE_CHECKING:
55    from music_assistant.controllers.webserver import WebserverController
56    from music_assistant.providers.hass import HomeAssistantProvider
57
58LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.auth")
59
60# Database schema version
61DB_SCHEMA_VERSION = 5
62
63# Token expiration constants (in days)
64TOKEN_SHORT_LIVED_EXPIRATION = 30  # Short-lived tokens (auto-renewing on use)
65TOKEN_LONG_LIVED_EXPIRATION = 365  # Long-lived tokens (1 year, no auto-renewal)
66# Max days a sliding short-lived session may live from creation before re-auth.
67TOKEN_ABSOLUTE_MAX_EXPIRATION = 90
68TOKEN_GUEST_EXPIRATION = 1  # Guest sessions: short fixed lifetime, no renewal
69# Days before the absolute cap at which the HA integration token is rotated
70HA_TOKEN_ROTATION_MARGIN = 7
71# Minimum age of a token's stored last_used_at before token activity is persisted again
72TOKEN_ACTIVITY_PERSIST_INTERVAL = timedelta(hours=1)
73
74HA_TOKEN_SETTING_KEY = "ha_integration_token"
75HA_TOKEN_NAME = "Home Assistant Integration"
76
77# Join code constants (short codes for QR/link-based login)
78JOIN_CODE_LENGTH = 12
79JOIN_CODE_CHARSET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"  # No I/O/0/1 for readability
80JOIN_CODE_DEFAULT_EXPIRY_HOURS = 8
81# Failed exchanges are throttled per calling websocket connection, so one guest fumbling a
82# stale QR code cannot lock out every other guest at a party. Callers that reach the API
83# without a connection identity (the JSON RPC endpoint, in-process callers) share one bucket.
84JOIN_CODE_ANONYMOUS_RATE_LIMIT_KEY = "no-connection"
85# Second, server-wide bucket that backstops the per-connection buckets, since a client can
86# start a new connection (and thus a new bucket) at will. The join code itself is what makes
87# guessing infeasible (12 chars over a 32 symbol alphabet is ~2^60, valid for hours), so this
88# ceiling is deliberately far above any plausible party-scale burst of legitimate failures.
89JOIN_CODE_GLOBAL_RATE_LIMIT_KEY = "all-connections"
90JOIN_CODE_GLOBAL_FAILURE_CEILING = 1000
91JOIN_CODE_GLOBAL_COOLDOWN_SECONDS = 60
92
93
94class AuthenticationManager:
95    """Manager for authentication and user management (part of webserver controller)."""
96
97    def __init__(self, webserver: WebserverController) -> None:
98        """
99        Initialize the authentication manager.
100
101        :param webserver: WebserverController instance.
102        """
103        self.webserver = webserver
104        self.mass = webserver.mass
105        self.database: DatabaseConnection = None  # type: ignore[assignment]
106        self.login_providers: dict[str, LoginProvider] = {}
107        self.logger = LOGGER
108        self._has_users: bool = False
109        self.jwt_helper: JWTHelper = None  # type: ignore[assignment]
110        self._join_code_rate_limiter = LoginRateLimiter(subject="client")
111        self._join_code_global_rate_limiter = LoginRateLimiter(
112            delay_tiers=((JOIN_CODE_GLOBAL_FAILURE_CEILING, JOIN_CODE_GLOBAL_COOLDOWN_SECONDS),),
113            warn_threshold=JOIN_CODE_GLOBAL_FAILURE_CEILING,
114            alert_threshold=JOIN_CODE_GLOBAL_FAILURE_CEILING * 2,
115            subject="join_codes",
116        )
117        # Stops concurrent exchanges from passing the rate limit check before failures land
118        self._join_code_exchange_lock = asyncio.Lock()
119        self._access_revoked_callbacks: list[Callable[[User], None]] = []
120
121    async def setup(self) -> None:
122        """Initialize the authentication manager."""
123        # Setup database
124        db_path = self.mass.storage_path + "/auth.db"
125        self.database = DatabaseConnection(db_path)
126        await self.database.setup()
127
128        # Create database schema and handle migrations
129        await self._setup_database()
130
131        # Initialize JWT helper with secret key
132        jwt_secret = await self._get_or_create_jwt_secret()
133        self.jwt_helper = JWTHelper(jwt_secret)
134
135        # Setup login providers
136        await self._setup_login_providers()
137
138        self._has_users = await self._has_non_system_users()
139
140        # migrate the Home Assistant system user of pre-existing installs to the service role
141        await self._migrate_system_user_role()
142
143        self._schedule_join_code_cleanup()
144
145        self.logger.info(
146            "Authentication manager initialized (providers=%d)", len(self.login_providers)
147        )
148
149    async def close(self) -> None:
150        """Cleanup on exit."""
151        if self.database:
152            await self.database.close()
153
154    @property
155    def has_users(self) -> bool:
156        """Check if any users exist in the system."""
157        return self._has_users
158
159    async def authenticate_with_credentials(
160        self, provider_id: str, credentials: dict[str, Any]
161    ) -> AuthResult:
162        """
163        Authenticate a user with credentials.
164
165        :param provider_id: The login provider ID.
166        :param credentials: Provider-specific credentials.
167        """
168        provider = self.login_providers.get(provider_id)
169        if not provider:
170            return AuthResult(success=False, error="Invalid provider")
171
172        return await provider.authenticate(credentials)
173
174    async def authenticate_with_token(self, token: str) -> User | None:
175        """
176        Authenticate a user with an access token (JWT or legacy).
177
178        Supports both JWT tokens and legacy hash-based tokens for backward compatibility.
179
180        :param token: The access token (JWT or legacy hash token).
181        """
182        # Try to decode as JWT first
183        try:
184            payload = self.jwt_helper.decode_token(token, verify_exp=True)
185            token_id = payload.get("jti")
186            token_user_id = payload.get("sub")
187
188            if not token_id or not token_user_id:
189                return None
190
191            token_row = await self.database.get_row("auth_tokens", {"token_id": token_id})
192            if not token_row:
193                return None
194
195            # Database is source of truth for token metadata, not the (immutable) JWT payload.
196            # A payload/row mismatch means a tampered or stale token: reject rather than trust it.
197            if token_user_id != token_row["user_id"]:
198                return None
199            is_long_lived = bool(token_row["is_long_lived"])
200
201            # Database expiration is source of truth
202            if token_row["expires_at"]:
203                db_expires_at = datetime.fromisoformat(token_row["expires_at"])
204                if utc() > db_expires_at:
205                    await self.database.delete("auth_tokens", {"token_id": token_id})
206                    return None
207
208            user = await self.get_user(token_row["user_id"])
209            if not user:
210                return None
211
212            updates = await self._refresh_token_expiration(token_row, user, is_long_lived)
213            if updates is None:
214                return None
215            if updates:
216                await self.database.update("auth_tokens", {"token_id": token_id}, updates)
217
218            return user
219
220        except pyjwt.ExpiredSignatureError:
221            if token_id := self.jwt_helper.get_token_id(token):
222                await self.database.delete("auth_tokens", {"token_id": token_id})
223            return None
224        except pyjwt.InvalidTokenError:
225            self.logger.debug("Token is not a valid JWT, trying legacy hash lookup")
226        except Exception as err:
227            self.logger.debug("Error decoding JWT token: %s, trying legacy hash lookup", err)
228
229        # Fallback to legacy hash-based token lookup
230        token_hash = hashlib.sha256(token.encode()).hexdigest()
231        token_row = await self.database.get_row("auth_tokens", {"token_hash": token_hash})
232        if not token_row:
233            return None
234
235        # Check if token is expired
236        if token_row["expires_at"]:
237            expires_at = datetime.fromisoformat(token_row["expires_at"])
238            if utc() > expires_at:
239                # Token expired, delete it
240                await self.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
241                return None
242
243        user = await self.get_user(token_row["user_id"])
244        if not user:
245            return None
246
247        is_long_lived = bool(token_row["is_long_lived"])
248        legacy_updates = await self._refresh_token_expiration(token_row, user, is_long_lived)
249        if legacy_updates is None:
250            return None
251        if legacy_updates:
252            await self.database.update(
253                "auth_tokens", {"token_id": token_row["token_id"]}, legacy_updates
254            )
255
256        return user
257
258    async def get_token_id_from_token(self, token: str) -> str | None:
259        """
260        Get token_id from a token string (for tracking revocation).
261
262        :param token: The access token (JWT or legacy hash token).
263        :return: The token_id or None if token not found.
264        """
265        # Try to extract from JWT first
266        if token_id := self.jwt_helper.get_token_id(token):
267            return token_id
268
269        # Fallback: Hash-based lookup for legacy tokens
270        token_hash = hashlib.sha256(token.encode()).hexdigest()
271        token_row = await self.database.get_row("auth_tokens", {"token_hash": token_hash})
272        if not token_row:
273            return None
274        return str(token_row["token_id"])
275
276    @api_command("auth/user", required_scope=Scope.USERS_READ)
277    async def get_user(self, user_id: str) -> User | None:
278        """
279        Get user by ID (requires the users.read scope).
280
281        :param user_id: The user ID.
282        :return: User object or None if not found.
283        """
284        user_row = await self.database.get_row("users", {"user_id": user_id})
285        if not user_row or not user_row["enabled"]:
286            return None
287
288        return User(
289            user_id=user_row["user_id"],
290            username=user_row["username"],
291            role=user_row["role"],
292            enabled=bool(user_row["enabled"]),
293            created_at=datetime.fromisoformat(user_row["created_at"]),
294            display_name=user_row["display_name"],
295            avatar_url=user_row["avatar_url"],
296            preferences=json_loads(user_row["preferences"]),
297            player_filter=json_loads(user_row["player_filter"]),
298            provider_filter=json_loads(user_row["provider_filter"]),
299        )
300
301    async def get_user_by_username(self, username: str) -> User | None:
302        """
303        Get user by username.
304
305        :param username: The username.
306        :return: User object or None if not found.
307        """
308        username = normalize_username(username)
309
310        user_row = await self.database.get_row("users", {"username": username})
311        if not user_row:
312            return None
313
314        return await self.get_user(user_row["user_id"])
315
316    async def get_user_by_provider_link(
317        self, provider_type: AuthProviderType, provider_user_id: str
318    ) -> User | None:
319        """
320        Get user by their provider link.
321
322        :param provider_type: The auth provider type.
323        :param provider_user_id: The user ID from the provider.
324        """
325        link_row = await self.database.get_row(
326            "user_auth_providers",
327            {
328                "provider_type": provider_type.value,
329                "provider_user_id": provider_user_id,
330            },
331        )
332        if not link_row:
333            return None
334
335        return await self.get_user(link_row["user_id"])
336
337    async def create_user(
338        self,
339        username: str,
340        role: UserRole = UserRole.USER,
341        display_name: str | None = None,
342        avatar_url: str | None = None,
343        preferences: dict[str, Any] | None = None,
344        player_filter: list[str] | None = None,
345        provider_filter: list[str] | None = None,
346    ) -> User:
347        """
348        Create a new user.
349
350        :param username: The username.
351        :param role: The user role (default: USER).
352        :param display_name: Optional display name.
353        :param avatar_url: Optional avatar URL.
354        :param preferences: Optional user preferences dict.
355        :param player_filter: Optional list of player IDs user has access to.
356        :param provider_filter: Optional list of provider instance IDs user has access to.
357        """
358        normalized_username = normalize_username(username)
359
360        # Check if this is the first non-system user
361        is_first_user = not await self._has_non_system_users()
362
363        user_id = secrets.token_urlsafe(32)
364        created_at = utc()
365        if preferences is None:
366            preferences = {}
367        if player_filter is None:
368            player_filter = []
369        if provider_filter is None:
370            provider_filter = []
371
372        user_data = {
373            "user_id": user_id,
374            "username": normalized_username,
375            "role": role.value,
376            "enabled": True,
377            "created_at": created_at.isoformat(),
378            "display_name": display_name,
379            "avatar_url": avatar_url,
380            "preferences": json_dumps(preferences),
381            "player_filter": json_dumps(player_filter),
382            "provider_filter": json_dumps(provider_filter),
383        }
384
385        await self.database.insert("users", user_data)
386
387        user = User(
388            user_id=user_id,
389            username=normalized_username,
390            role=role,
391            enabled=True,
392            created_at=created_at,
393            display_name=display_name,
394            avatar_url=avatar_url,
395            preferences=preferences,
396            player_filter=player_filter,
397            provider_filter=provider_filter,
398        )
399
400        # If this is the first non-system user, migrate playlog entries to them
401        if is_first_user and normalized_username != HOMEASSISTANT_SYSTEM_USER:
402            self._has_users = True
403            await self._migrate_playlog_to_first_user(user_id)
404
405        return user
406
407    async def get_homeassistant_system_user(self) -> User:
408        """
409        Get or create the Home Assistant system user.
410
411        This is a special system user created automatically for Home Assistant integration.
412        It bypasses normal authentication but is restricted to the ingress webserver.
413
414        :return: The Home Assistant system user.
415        """
416        username = HOMEASSISTANT_SYSTEM_USER
417        display_name = "Home Assistant Integration"
418        role = UserRole.SERVICE
419
420        normalized_username = normalize_username(username)
421
422        # Try to find existing user by username
423        user_row = await self.database.get_row("users", {"username": normalized_username})
424        if user_row:
425            # Use get_user to ensure preferences are parsed correctly
426            user = await self.get_user(user_row["user_id"])
427            assert user is not None  # User exists in DB, so get_user must return it
428            return user
429
430        # Create new system user
431        user = await self.create_user(
432            username=username,
433            role=role,
434            display_name=display_name,
435        )
436        self.logger.debug("Created Home Assistant system user: %s (role: %s)", username, role.value)
437        return user
438
439    async def get_homeassistant_system_user_token(self) -> str:
440        """
441        Get the auth token to announce to the Home Assistant integration.
442
443        Returns the same (still valid) token on repeated calls so re-announcing it via
444        Supervisor discovery is idempotent for the HA integration. A replacement is only
445        minted when the current token is missing, expired or revoked, or shortly before
446        it reaches its absolute lifetime cap - allowing seamless rotation as HA reloads
447        with the newly announced token while the old one is still accepted.
448
449        :return: Authentication token for the Home Assistant system user.
450        """
451        system_user = await self.get_homeassistant_system_user()
452
453        # Keep the plain token in settings for re-announcing; the jwt_secret next to it can mint any token anyway
454        if token_row := await self.database.get_row("settings", {"key": HA_TOKEN_SETTING_KEY}):
455            token = str(token_row["value"])
456            if await self._can_reuse_ha_integration_token(token, system_user):
457                return token
458
459        # A superseded token stays valid until expiry, so HA keeps working until it reloads
460        token = await self.create_token(
461            user=system_user,
462            name=HA_TOKEN_NAME,
463            is_long_lived=False,
464        )
465        await self.database.insert_or_replace(
466            "settings",
467            {"key": HA_TOKEN_SETTING_KEY, "value": token, "type": "string"},
468        )
469        now = utc()
470        for old_row in await self.database.get_rows(
471            "auth_tokens", {"user_id": system_user.user_id, "name": HA_TOKEN_NAME}
472        ):
473            if old_row["expires_at"] and datetime.fromisoformat(old_row["expires_at"]) <= now:
474                await self.database.delete("auth_tokens", {"token_id": old_row["token_id"]})
475        await self.database.commit()
476        return token
477
478    async def link_user_to_provider(
479        self,
480        user: User,
481        provider_type: AuthProviderType,
482        provider_user_id: str,
483    ) -> UserAuthProvider:
484        """
485        Link a user to an authentication provider.
486
487        If a link already exists for this provider/provider_user_id, returns the existing link.
488
489        :param user: The user to link.
490        :param provider_type: The provider type.
491        :param provider_user_id: The user ID from the provider (e.g., password hash, OAuth ID).
492        """
493        # Check if a link already exists for this provider/provider_user_id
494        existing_link = await self.database.get_row(
495            "user_auth_providers",
496            {
497                "provider_type": provider_type.value,
498                "provider_user_id": provider_user_id,
499            },
500        )
501
502        if existing_link:
503            # Link already exists - return it
504            return UserAuthProvider(
505                link_id=existing_link["link_id"],
506                user_id=existing_link["user_id"],
507                provider_type=AuthProviderType(existing_link["provider_type"]),
508                provider_user_id=existing_link["provider_user_id"],
509                created_at=datetime.fromisoformat(existing_link["created_at"]),
510            )
511
512        # Create new link
513        link_id = secrets.token_urlsafe(32)
514        created_at = utc()
515        link_data = {
516            "link_id": link_id,
517            "user_id": user.user_id,
518            "provider_type": provider_type.value,
519            "provider_user_id": provider_user_id,
520            "created_at": created_at.isoformat(),
521        }
522
523        await self.database.insert("user_auth_providers", link_data)
524
525        return UserAuthProvider(
526            link_id=link_id,
527            user_id=user.user_id,
528            provider_type=provider_type,
529            provider_user_id=provider_user_id,
530            created_at=created_at,
531        )
532
533    async def update_user(
534        self,
535        user: User,
536        username: str | None = None,
537        display_name: str | None = None,
538        avatar_url: str | None = None,
539    ) -> User:
540        """
541        Update a user's profile information.
542
543        :param user: The user to update.
544        :param username: New username (optional).
545        :param display_name: New display name (optional).
546        :param avatar_url: New avatar URL (optional).
547        """
548        updates = {}
549        if username is not None:
550            # Normalize username for case-insensitive authentication
551            updates["username"] = normalize_username(username)
552        if display_name is not None:
553            updates["display_name"] = display_name
554        if avatar_url is not None:
555            updates["avatar_url"] = avatar_url
556
557        if updates:
558            await self.database.update("users", {"user_id": user.user_id}, updates)
559
560        # Return updated user
561        updated_user = await self.get_user(user.user_id)
562        assert updated_user is not None  # User exists, so get_user must return it
563        return updated_user
564
565    async def update_user_preferences(
566        self,
567        user: User,
568        preferences: dict[str, Any],
569    ) -> User:
570        """
571        Update a user's preferences.
572
573        :param user: The user to update.
574        :param preferences: New preferences dict (completely replaces existing preferences).
575        """
576        # Verify user exists
577        current_user = await self.get_user(user.user_id)
578        if not current_user:
579            raise ValueError(f"User {user.user_id} not found")
580
581        # Update database with new preferences (complete replacement)
582        await self.database.update(
583            "users",
584            {"user_id": user.user_id},
585            {"preferences": json_dumps(preferences)},
586        )
587
588        # Return updated user
589        updated_user = await self.get_user(user.user_id)
590        assert updated_user is not None  # User exists, so get_user must return it
591        return updated_user
592
593    async def update_provider_link(
594        self,
595        user: User,
596        provider_type: AuthProviderType,
597        provider_user_id: str,
598    ) -> None:
599        """
600        Update a user's provider link (e.g., change password).
601
602        :param user: The user.
603        :param provider_type: The provider type.
604        :param provider_user_id: The new provider user ID (e.g., new password hash).
605        """
606        # Find existing link
607        link_row = await self.database.get_row(
608            "user_auth_providers",
609            {
610                "user_id": user.user_id,
611                "provider_type": provider_type.value,
612            },
613        )
614
615        if link_row:
616            # Update existing link
617            await self.database.update(
618                "user_auth_providers",
619                {"link_id": link_row["link_id"]},
620                {"provider_user_id": provider_user_id},
621            )
622        else:
623            # Create new link
624            await self.link_user_to_provider(user, provider_type, provider_user_id)
625
626    async def create_token(self, user: User, name: str, is_long_lived: bool = False) -> str:
627        """
628        Create a new JWT access token for a user.
629
630        :param user: The user to create the token for.
631        :param name: A name/description for the token (e.g., device name).
632        :param is_long_lived: Whether this is a long-lived token (default: False).
633            Short-lived tokens (False): Auto-renewing on use, expire after 30 days of inactivity,
634            capped at an absolute maximum lifetime from creation (see TOKEN_ABSOLUTE_MAX_EXPIRATION).
635            Tokens for guest users get a short fixed lifetime instead and never renew.
636            Long-lived tokens (True): No auto-renewal, expire after 1 year.
637        :return: JWT token string.
638        """
639        # Generate unique token ID
640        token_id = secrets.token_urlsafe(32)
641
642        # Calculate expiration based on token type
643        created_at = utc()
644        if is_long_lived:
645            # Long-lived tokens expire after 1 year (no auto-renewal)
646            expires_at = created_at + timedelta(days=TOKEN_LONG_LIVED_EXPIRATION)
647            jwt_expires_at = expires_at
648        elif user.role == UserRole.GUEST:
649            expires_at = created_at + timedelta(days=TOKEN_GUEST_EXPIRATION)
650            jwt_expires_at = expires_at
651        else:
652            # Short-lived tokens expire after 30 days (with auto-renewal on use)
653            expires_at = created_at + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
654            # The exp claim must carry the absolute cap, or it would cut off sliding renewals
655            jwt_expires_at = created_at + timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION)
656
657        # Generate JWT token
658        token = self.jwt_helper.encode_token(
659            user=user,
660            token_id=token_id,
661            token_name=name,
662            expires_at=jwt_expires_at,
663            is_long_lived=is_long_lived,
664        )
665
666        # Store token hash in database for revocation checking
667        token_hash = hashlib.sha256(token.encode()).hexdigest()
668        token_data = {
669            "token_id": token_id,
670            "user_id": user.user_id,
671            "token_hash": token_hash,
672            "name": name,
673            "created_at": created_at.isoformat(),
674            "expires_at": expires_at.isoformat(),
675            "is_long_lived": 1 if is_long_lived else 0,
676        }
677        await self.database.insert("auth_tokens", token_data)
678
679        return token
680
681    @api_command("auth/token/revoke")
682    async def revoke_token(self, token_id: str) -> None:
683        """
684        Revoke an auth token.
685
686        :param token_id: The token ID to revoke.
687        """
688        user = get_current_user()
689        if not user:
690            raise AuthenticationRequired("Not authenticated")
691
692        token_row = await self.database.get_row("auth_tokens", {"token_id": token_id})
693        if not token_row:
694            raise InvalidDataError("Token not found")
695
696        # Check permissions - users can only revoke their own tokens
697        # unless they hold the users.manage scope
698        if token_row["user_id"] != user.user_id and not has_scope(user, Scope.USERS_MANAGE):
699            raise InsufficientPermissions("You can only revoke your own tokens")
700
701        await self.database.delete("auth_tokens", {"token_id": token_id})
702
703        # Disconnect any WebSocket connections using this token
704        self.webserver.disconnect_websockets_for_token(token_id)
705
706        self.logger.info(
707            "Token revoked by user '%s' (token_id=%s)",
708            user.username,
709            token_id,
710        )
711
712    def subscribe_user_access_revoked(self, callback: Callable[[User], None]) -> Callable[[], None]:
713        """
714        Subscribe to a user's access being withdrawn.
715
716        Fires on deliberate access withdrawal: bulk token revocation
717        (revoke_tokens_for_user), account disable, and account deletion. Revoking a
718        single token (e.g. a logout) does not fire it, so credentials bound to the
719        account survive a plain logout.
720
721        :param callback: Called with the affected user.
722        :return: Callable that removes the subscription.
723        """
724        self._access_revoked_callbacks.append(callback)
725
726        def _unsubscribe() -> None:
727            with contextlib.suppress(ValueError):
728                self._access_revoked_callbacks.remove(callback)
729
730        return _unsubscribe
731
732    async def revoke_tokens_for_user(self, user: User) -> int:
733        """
734        Revoke all auth tokens for a user.
735
736        This is an internal method for programmatic use (e.g., when disabling guest access).
737        Unlike revoke_token(), this does not require an authenticated user context.
738
739        :param user: The user whose tokens should be revoked.
740        :return: Number of tokens revoked.
741        """
742        token_rows = await self.database.get_rows("auth_tokens", {"user_id": user.user_id})
743
744        # Disconnect any WebSocket connections using these tokens
745        for token_row in token_rows:
746            self.webserver.disconnect_websockets_for_token(token_row["token_id"])
747
748        if token_rows:
749            # Delete all tokens in one go
750            await self.database.execute(
751                "DELETE FROM auth_tokens WHERE user_id = :user_id",
752                {"user_id": user.user_id},
753            )
754            await self.database.commit()
755            self.logger.info("Revoked %d token(s) for user '%s'", len(token_rows), user.username)
756
757        # Notify even with no tokens left: subscribers may hold credentials tied to
758        # this user's access that must be withdrawn regardless.
759        self._notify_user_access_revoked(user)
760
761        return len(token_rows)
762
763    @api_command("auth/tokens")
764    async def get_user_tokens(self, user_id: str | None = None) -> list[AuthToken]:
765        """
766        Get current user's auth tokens or another user's tokens (admin only).
767
768        The last_used_at timestamp is persisted at most once per hour, so it may lag
769        actual token usage by up to an hour.
770
771        :param user_id: Optional user ID to get tokens for (admin only).
772        :return: List of auth tokens.
773        """
774        current_user = get_current_user()
775        if not current_user:
776            return []
777
778        # If user_id is provided and different from current user,
779        # require the users.manage scope
780        if user_id and user_id != current_user.user_id:
781            if not has_scope(current_user, Scope.USERS_MANAGE):
782                return []
783            target_user = await self.get_user(user_id)
784            if not target_user:
785                return []
786        else:
787            target_user = current_user
788
789        token_rows = await self.database.get_rows(
790            "auth_tokens", {"user_id": target_user.user_id}, limit=100
791        )
792        return [AuthToken.from_dict(dict(row)) for row in token_rows]
793
794    @api_command("auth/users", required_scope=Scope.USERS_READ)
795    async def list_users(self) -> list[User]:
796        """
797        Get all users (requires the users.read scope).
798
799        System users are excluded from the list.
800
801        :return: List of user objects.
802        """
803        user_rows = await self.database.get_rows("users", limit=1000)
804        users = []
805        for row in user_rows:
806            # Skip system users
807            if row["username"] == HOMEASSISTANT_SYSTEM_USER:
808                continue
809            users.append(
810                User(
811                    user_id=row["user_id"],
812                    username=row["username"],
813                    role=row["role"],
814                    enabled=bool(row["enabled"]),
815                    created_at=datetime.fromisoformat(row["created_at"]),
816                    display_name=row["display_name"],
817                    avatar_url=row["avatar_url"],
818                    preferences=json_loads(row["preferences"]),
819                    player_filter=json_loads(row["player_filter"]),
820                    provider_filter=json_loads(row["provider_filter"]),
821                )
822            )
823        return users
824
825    async def update_user_role(self, user_id: str, new_role: UserRole, admin_user: User) -> bool:
826        """
827        Update a user's role (requires the users.manage scope).
828
829        :param user_id: The user ID to update.
830        :param new_role: The new role to assign.
831        :param admin_user: The user performing the action.
832        """
833        if not has_scope(admin_user, Scope.USERS_MANAGE):
834            return False
835
836        user_row = await self.database.get_row("users", {"user_id": user_id})
837        if not user_row:
838            return False
839
840        old_role = user_row["role"]
841        await self.database.update(
842            "users",
843            {"user_id": user_id},
844            {"role": new_role.value},
845        )
846        self.logger.info(
847            "User role changed: '%s' from '%s' to '%s' by admin '%s'",
848            user_row["username"],
849            old_role,
850            new_role.value,
851            admin_user.username,
852        )
853        return True
854
855    @api_command("auth/user/enable", required_scope=Scope.USERS_MANAGE)
856    async def enable_user(self, user_id: str) -> None:
857        """
858        Enable user account (admin only).
859
860        :param user_id: The user ID.
861        """
862        await self.database.update(
863            "users",
864            {"user_id": user_id},
865            {"enabled": 1},
866        )
867        self.logger.info("User account enabled (user_id=%s)", user_id)
868
869    @api_command("auth/user/disable", required_scope=Scope.USERS_MANAGE)
870    async def disable_user(self, user_id: str) -> None:
871        """
872        Disable user account (admin only).
873
874        :param user_id: The user ID.
875        """
876        admin_user = get_current_user()
877        if not admin_user:
878            raise AuthenticationRequired("Not authenticated")
879
880        # Cannot disable yourself
881        if user_id == admin_user.user_id:
882            raise InvalidDataError("Cannot disable your own account")
883
884        # Look up the user before disabling (get_user hides disabled accounts)
885        user_row = await self.database.get_row("users", {"user_id": user_id})
886        if not user_row:
887            raise InvalidDataError("User not found")
888
889        await self.database.update(
890            "users",
891            {"user_id": user_id},
892            {"enabled": 0},
893        )
894
895        # Disconnect all WebSocket connections for this user
896        self.webserver.disconnect_websockets_for_user(user_id)
897
898        # A disabled account's tokens stop authenticating, so credentials bound to its
899        # access must be withdrawn with them (they return on the next login after enable).
900        self._notify_user_access_revoked(
901            User(user_id=user_row["user_id"], username=user_row["username"], role=user_row["role"])
902        )
903
904        self.logger.info("User account disabled (user_id=%s)", user_id)
905
906    async def get_login_providers(self) -> list[dict[str, Any]]:
907        """Get list of available login providers (dynamically checks for HA provider)."""
908        # Sync HA OAuth provider with HA provider availability
909        await self._sync_ha_oauth_provider()
910
911        providers = []
912        for provider_id, provider in self.login_providers.items():
913            providers.append(
914                {
915                    "provider_id": provider_id,
916                    "provider_type": provider.provider_type.value,
917                    "requires_redirect": provider.requires_redirect,
918                }
919            )
920        return providers
921
922    @api_command("auth/login", authenticated=False)
923    async def login(
924        self,
925        username: str | None = None,
926        password: str | None = None,
927        provider_id: str = "builtin",
928        device_name: str | None = None,
929        **extra_credentials: Any,
930    ) -> dict[str, Any]:
931        """
932        Authenticate user with credentials via WebSocket.
933
934        This command allows clients to authenticate over the WebSocket connection
935        using username/password or other provider-specific credentials.
936
937        :param username: Username for authentication (for builtin provider).
938        :param password: Password for authentication (for builtin provider).
939        :param provider_id: The login provider ID (defaults to "builtin").
940        :param device_name: Optional device name for the token (e.g., "iPhone 15", "Desktop PC").
941        :param extra_credentials: Additional provider-specific credentials.
942        :return: Authentication result with access token if successful.
943        """
944        # Build credentials dict from parameters
945        credentials: dict[str, Any] = {}
946        if username is not None:
947            credentials["username"] = username
948        if password is not None:
949            credentials["password"] = password
950        credentials.update(extra_credentials)
951
952        auth_result = await self.authenticate_with_credentials(provider_id, credentials)
953
954        if not auth_result.success:
955            self.logger.warning(
956                "Login failed for username '%s' via provider '%s'",
957                username or "<not provided>",
958                provider_id,
959            )
960            return {
961                "success": False,
962                "error": auth_result.error or "Authentication failed",
963            }
964
965        if not auth_result.user:
966            return {
967                "success": False,
968                "error": "Authentication failed: no user returned",
969            }
970
971        # Create short-lived access token with device name if provided
972        token_name = device_name or f"WebSocket Session - {auth_result.user.username}"
973        token = await self.create_token(
974            auth_result.user,
975            is_long_lived=False,
976            name=token_name,
977        )
978
979        self.logger.info(
980            "User '%s' logged in via provider '%s'",
981            auth_result.user.username,
982            provider_id,
983        )
984
985        return {
986            "success": True,
987            "access_token": token,
988            "user": {
989                "user_id": auth_result.user.user_id,
990                "username": auth_result.user.username,
991                "display_name": auth_result.user.display_name,
992                "role": auth_result.user.role,
993            },
994        }
995
996    @api_command("auth/providers", authenticated=False)
997    async def get_providers(self) -> list[dict[str, Any]]:
998        """
999        Get list of available authentication providers.
1000
1001        Returns information about all available login providers including
1002        whether they require OAuth redirect flow.
1003        """
1004        return await self.get_login_providers()
1005
1006    @api_command("auth/authorization_url", authenticated=False)
1007    async def get_auth_url(
1008        self,
1009        provider_id: str,
1010        return_url: str | None = None,
1011    ) -> dict[str, str | None]:
1012        """
1013        Get OAuth authorization URL for authentication.
1014
1015        For OAuth providers (like Home Assistant), this returns the URL that
1016        the user should visit in their browser to authorize the application.
1017
1018        :param provider_id: The provider ID (e.g., "hass").
1019        :param return_url: URL to redirect to after OAuth completes.
1020        :return: Dictionary with authorization_url.
1021        """
1022        auth_url = await self.get_authorization_url(provider_id, return_url)
1023        if not auth_url:
1024            return {
1025                "authorization_url": None,
1026                "error": "Provider does not support OAuth or does not exist",
1027            }
1028
1029        return {
1030            "authorization_url": auth_url,
1031        }
1032
1033    async def get_authorization_url(
1034        self, provider_id: str, return_url: str | None = None
1035    ) -> str | None:
1036        """
1037        Get OAuth authorization URL for a provider.
1038
1039        :param provider_id: The provider ID.
1040        :param return_url: Optional URL to redirect to after successful login.
1041        """
1042        provider = self.login_providers.get(provider_id)
1043        if not provider or not provider.requires_redirect:
1044            return None
1045
1046        # Build callback redirect_uri
1047        redirect_uri = f"{self.webserver.base_url}/auth/callback?provider_id={provider_id}"
1048        return await provider.get_authorization_url(redirect_uri, return_url)
1049
1050    async def handle_oauth_callback(
1051        self, provider_id: str, code: str, state: str, redirect_uri: str
1052    ) -> AuthResult:
1053        """
1054        Handle OAuth callback.
1055
1056        :param provider_id: The provider ID.
1057        :param code: OAuth authorization code.
1058        :param state: OAuth state parameter.
1059        :param redirect_uri: The callback URL.
1060        """
1061        provider = self.login_providers.get(provider_id)
1062        if not provider:
1063            return AuthResult(success=False, error="Invalid provider")
1064
1065        return await provider.handle_oauth_callback(code, state, redirect_uri)
1066
1067    @api_command("auth/token/create")
1068    async def create_long_lived_token(self, name: str, user_id: str | None = None) -> str:
1069        """
1070        Create a new long-lived access token for current user or another user (admin only).
1071
1072        Long-lived tokens are intended for external integrations and API access.
1073        They expire after 1 year and do NOT auto-renew on use.
1074
1075        Short-lived tokens (for regular user sessions) are only created during login
1076        and auto-renew on each use (sliding 30-day expiration window).
1077
1078        Long-lived tokens cannot be created for guest accounts.
1079
1080        :param name: The name/description for the token (e.g., "Home Assistant", "Mobile App").
1081        :param user_id: Optional user ID to create token for (admin only).
1082        :return: The created token string.
1083        """
1084        current_user = get_current_user()
1085        if not current_user:
1086            raise AuthenticationRequired("Not authenticated")
1087
1088        # If user_id is provided and different from current user,
1089        # require the users.manage scope
1090        if user_id and user_id != current_user.user_id:
1091            if not has_scope(current_user, Scope.USERS_MANAGE):
1092                raise InsufficientPermissions(
1093                    "The users.manage scope is required to create tokens for other users"
1094                )
1095            target_user = await self.get_user(user_id)
1096            if not target_user:
1097                raise InvalidDataError("User not found")
1098        else:
1099            target_user = current_user
1100
1101        # Guest access is temporary by design, deny tokens that would outlive it
1102        if target_user.role == UserRole.GUEST:
1103            raise InsufficientPermissions("Long-lived tokens cannot be created for guest accounts")
1104
1105        # Create a long-lived token (only long-lived tokens can be created via this command)
1106        token = await self.create_token(target_user, name, is_long_lived=True)
1107        self.logger.info("Created long-lived token '%s' for user '%s'", name, target_user.username)
1108        return token
1109
1110    @api_command("auth/user/create", required_scope=Scope.USERS_MANAGE)
1111    async def create_user_with_api(
1112        self,
1113        username: str,
1114        password: str,
1115        role: str = "user",
1116        display_name: str | None = None,
1117        avatar_url: str | None = None,
1118        player_filter: list[str] | None = None,
1119        provider_filter: list[str] | None = None,
1120    ) -> User:
1121        """
1122        Create a new user with built-in authentication (admin only).
1123
1124        :param username: The username (minimum 2 characters).
1125        :param password: The password (minimum 8 characters).
1126        :param role: User role - "admin" or "user" (default: "user").
1127        :param display_name: Optional display name.
1128        :param avatar_url: Optional avatar URL.
1129        :param player_filter: Optional list of player IDs user has access to.
1130        :param provider_filter: Optional list of provider instance IDs user has access to.
1131        :return: Created user object.
1132        """
1133        # Validation
1134        if not username or len(username) < 2:
1135            raise InvalidDataError("Username must be at least 2 characters")
1136
1137        if not password or len(password) < 8:
1138            raise InvalidDataError("Password must be at least 8 characters")
1139
1140        # Validate role
1141        try:
1142            user_role = UserRole(role)
1143        except ValueError as err:
1144            raise InvalidDataError("Invalid role. Must be 'admin' or 'user'") from err
1145
1146        # Get built-in provider
1147        builtin_provider = self.login_providers.get("builtin")
1148        if not builtin_provider or not isinstance(builtin_provider, BuiltinLoginProvider):
1149            raise InvalidDataError("Built-in auth provider not available")
1150
1151        # Create user with password
1152        user = await builtin_provider.create_user_with_password(
1153            username,
1154            password,
1155            role=user_role,
1156            player_filter=player_filter,
1157            provider_filter=provider_filter,
1158        )
1159
1160        # Update optional fields if provided
1161        if display_name or avatar_url:
1162            updated_user = await self.update_user(
1163                user, display_name=display_name, avatar_url=avatar_url
1164            )
1165            if updated_user:
1166                user = updated_user
1167
1168        self.logger.info("User created by admin: %s (role: %s)", username, role)
1169        return user
1170
1171    @api_command("auth/user/delete", required_scope=Scope.USERS_MANAGE)
1172    async def delete_user(self, user_id: str) -> None:
1173        """
1174        Delete user account (admin only).
1175
1176        :param user_id: The user ID.
1177        """
1178        admin_user = get_current_user()
1179        if not admin_user:
1180            raise AuthenticationRequired("Not authenticated")
1181
1182        # Don't allow deleting yourself
1183        if user_id == admin_user.user_id:
1184            raise InvalidDataError("Cannot delete your own account")
1185
1186        # Look up the username before deleting
1187        user_row = await self.database.get_row("users", {"user_id": user_id})
1188        if not user_row:
1189            raise InvalidDataError("User not found")
1190
1191        # Delete user from database
1192        await self.database.delete("users", {"user_id": user_id})
1193        await self.database.commit()
1194
1195        # Disconnect all WebSocket connections for this user
1196        self.webserver.disconnect_websockets_for_user(user_id)
1197
1198        # Deletion cascades the user's tokens away, so it must announce the access
1199        # withdrawal itself for credentials bound to this user.
1200        self._notify_user_access_revoked(
1201            User(user_id=user_row["user_id"], username=user_row["username"], role=user_row["role"])
1202        )
1203
1204        self.logger.info(
1205            "User '%s' deleted by admin '%s'",
1206            user_row["username"],
1207            admin_user.username,
1208        )
1209
1210    @api_command("auth/me")
1211    async def get_current_user_info(self) -> User:
1212        """Get current authenticated user information."""
1213        current_user_obj = get_current_user()
1214        if not current_user_obj:
1215            raise AuthenticationRequired("Not authenticated")
1216        return current_user_obj
1217
1218    @api_command("auth/scopes")
1219    async def get_role_scopes(self) -> dict[str, list[str]]:
1220        """Get the scopes granted to each of the builtin user roles."""
1221        return {
1222            str(role): sorted(str(scope) for scope in scopes)
1223            for role, scopes in ROLE_SCOPES.items()
1224        }
1225
1226    async def update_user_filters(
1227        self,
1228        target_user: User,
1229        player_filter: list[str] | None,
1230        provider_filter: list[str] | None,
1231    ) -> User:
1232        """Update user player and provider filters (helper method)."""
1233        updates = {}
1234        if player_filter is not None:
1235            updates["player_filter"] = json_dumps(player_filter)
1236        if provider_filter is not None:
1237            updates["provider_filter"] = json_dumps(provider_filter)
1238
1239        if updates:
1240            await self.database.update("users", {"user_id": target_user.user_id}, updates)
1241            # Refresh target user to get updated filters
1242            refreshed_user = await self.get_user(target_user.user_id)
1243            if not refreshed_user:
1244                raise InvalidDataError("Failed to refresh user after filter update")
1245            return refreshed_user
1246        return target_user
1247
1248    @api_command("auth/user/update")
1249    async def update_user_profile(
1250        self,
1251        user_id: str | None = None,
1252        username: str | None = None,
1253        display_name: str | None = None,
1254        avatar_url: str | None = None,
1255        password: str | None = None,
1256        role: str | None = None,
1257        preferences: dict[str, Any] | None = None,
1258        player_filter: list[str] | None = None,
1259        provider_filter: list[str] | None = None,
1260    ) -> User:
1261        """
1262        Update user profile information.
1263
1264        Users can update their own profile. Admins can update any user including role and password.
1265
1266        :param user_id: User ID to update (optional, defaults to current user).
1267        :param username: New username (optional).
1268        :param display_name: New display name (optional).
1269        :param avatar_url: New avatar URL (optional).
1270        :param password: New password (optional, minimum 8 characters).
1271        :param role: New role - "admin" or "user" (optional, set by admin only).
1272        :param preferences: User preferences dict (completely replaces existing, optional).
1273        :param player_filter: List of player IDs user has access to (set by admin only, optional).
1274        :param provider_filter: List of provider instance IDs user has access to (set by admin only, optional).
1275        :return: Updated user object.
1276        """
1277        current_user_obj = get_current_user()
1278        if not current_user_obj:
1279            raise AuthenticationRequired("Not authenticated")
1280
1281        # Determine target user
1282        may_manage_users = has_scope(current_user_obj, Scope.USERS_MANAGE)
1283        if user_id and user_id != current_user_obj.user_id:
1284            # Updating another user - requires the users.manage scope
1285            if not may_manage_users:
1286                raise InsufficientPermissions(
1287                    "The users.manage scope is required to update other users"
1288                )
1289            target_user = await self.get_user(user_id)
1290            if not target_user:
1291                raise InvalidDataError("User not found")
1292        else:
1293            # Updating own profile
1294            target_user = current_user_obj
1295
1296        # Update role (requires the users.manage scope)
1297        if role:
1298            if not may_manage_users:
1299                raise InsufficientPermissions(
1300                    "The users.manage scope is required to update user roles"
1301                )
1302
1303            try:
1304                new_role = UserRole(role)
1305            except ValueError as err:
1306                raise InvalidDataError("Invalid role. Must be 'admin' or 'user'") from err
1307
1308            success = await self.update_user_role(target_user.user_id, new_role, current_user_obj)
1309            if not success:
1310                raise InvalidDataError("Failed to update role")
1311
1312            # Refresh target user to get updated role
1313            refreshed_user = await self.get_user(target_user.user_id)
1314            if not refreshed_user:
1315                raise InvalidDataError("Failed to refresh user after role update")
1316            target_user = refreshed_user
1317
1318        # Update basic profile fields
1319        if username or display_name or avatar_url:
1320            updated_user = await self.update_user(
1321                target_user,
1322                username=username,
1323                display_name=display_name,
1324                avatar_url=avatar_url,
1325            )
1326            if not updated_user:
1327                raise InvalidDataError("Failed to update user profile")
1328            target_user = updated_user
1329
1330        # Update preferences if provided
1331        if preferences is not None:
1332            target_user = await self.update_user_preferences(target_user, preferences)
1333
1334        # Update player_filter and provider_filter (requires the users.manage scope)
1335        if player_filter is not None or provider_filter is not None:
1336            if not may_manage_users:
1337                raise InsufficientPermissions(
1338                    "The users.manage scope is required to update player/provider filters"
1339                )
1340            target_user = await self.update_user_filters(
1341                target_user, player_filter, provider_filter
1342            )
1343
1344        # Update password if provided
1345        if password:
1346            await self._update_profile_password(
1347                target_user, password, may_manage_users, current_user_obj
1348            )
1349
1350        return target_user
1351
1352    @api_command("auth/logout")
1353    async def logout(self) -> None:
1354        """Logout current user by revoking the current token."""
1355        user = get_current_user()
1356        if not user:
1357            raise AuthenticationRequired("Not authenticated")
1358
1359        # Get current token from context
1360        token = get_current_token()
1361        if not token:
1362            raise InvalidDataError("No token in context")
1363
1364        # Find and revoke the token
1365        token_hash = hashlib.sha256(token.encode()).hexdigest()
1366        token_row = await self.database.get_row("auth_tokens", {"token_hash": token_hash})
1367        if token_row:
1368            await self.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
1369
1370            # Disconnect any WebSocket connections using this token
1371            self.webserver.disconnect_websockets_for_token(token_row["token_id"])
1372
1373        self.logger.info("User '%s' logged out", user.username)
1374
1375    @api_command("auth/user/providers")
1376    async def get_my_providers(self) -> list[dict[str, Any]]:
1377        """
1378        Get current user's linked authentication providers.
1379
1380        :return: List of provider links.
1381        """
1382        user = get_current_user()
1383        if not user:
1384            return []
1385
1386        # Get provider links from database
1387        rows = await self.database.get_rows("user_auth_providers", {"user_id": user.user_id})
1388        providers = [UserAuthProvider.from_dict(dict(row)) for row in rows]
1389        return [p.to_dict() for p in providers]
1390
1391    @api_command("auth/user/unlink_provider", required_scope=Scope.USERS_MANAGE)
1392    async def unlink_provider(self, user_id: str, provider_type: str) -> bool:
1393        """
1394        Unlink authentication provider from user (admin only).
1395
1396        :param user_id: The user ID.
1397        :param provider_type: Provider type to unlink.
1398        :return: True if successful.
1399        """
1400        await self.database.delete(
1401            "user_auth_providers", {"user_id": user_id, "provider_type": provider_type}
1402        )
1403        await self.database.commit()
1404
1405        self.logger.info(
1406            "Auth provider '%s' unlinked from user (user_id=%s)",
1407            provider_type,
1408            user_id,
1409        )
1410        return True
1411
1412    # ==================== Join Code Methods ====================
1413
1414    async def generate_join_code(
1415        self,
1416        user: User,
1417        expires_in_hours: int = JOIN_CODE_DEFAULT_EXPIRY_HOURS,
1418        max_uses: int = 1,
1419        device_name: str = "Short Code Login",
1420    ) -> tuple[str, datetime]:
1421        """
1422        Generate a short join code for link/QR-based login.
1423
1424        This creates a short alphanumeric code that can be exchanged for a JWT token.
1425        Used for features like the party provider guest access, device pairing,
1426        or other short-code authentication flows.
1427
1428        :param user: The guest user that tokens created from this code will belong to.
1429        :param expires_in_hours: Hours until code expires (default: 8).
1430        :param max_uses: Maximum number of uses (0 = unlimited).
1431        :param device_name: Device name for tokens created with this code.
1432        :return: Tuple of (code, expires_at datetime).
1433        """
1434        if expires_in_hours <= 0:
1435            raise ValueError("expires_in_hours must be positive")
1436        if max_uses < 0:
1437            raise ValueError("max_uses must be non-negative (0 = unlimited)")
1438        if user.role != UserRole.GUEST:
1439            raise ValueError("Join codes can only be generated for guest accounts")
1440
1441        now = utc()
1442        expires_at = now + timedelta(hours=expires_in_hours)
1443
1444        for _ in range(3):  # Try up to 3 times to avoid code collisions
1445            code = "".join(secrets.choice(JOIN_CODE_CHARSET) for _ in range(JOIN_CODE_LENGTH))
1446            code_data = {
1447                "code_id": secrets.token_urlsafe(32),
1448                "code": code,
1449                "user_id": user.user_id,
1450                "created_at": now.isoformat(),
1451                "expires_at": expires_at.isoformat(),
1452                "max_uses": max_uses,
1453                "use_count": 0,
1454                "device_name": device_name,
1455            }
1456            try:
1457                await self.database.insert("join_codes", code_data)
1458                await self.database.commit()
1459                self.logger.info(
1460                    "Join code generated for user %s (expires: %s, max_uses: %s)",
1461                    user.username,
1462                    expires_at,
1463                    max_uses,
1464                )
1465                return code, expires_at
1466            except IntegrityError:
1467                self.logger.warning("Join code collision, retrying...")
1468                continue
1469
1470        raise RuntimeError("Failed to generate a unique join code after 3 attempts")
1471
1472    async def revoke_join_codes(self, user: User) -> int:
1473        """
1474        Revoke all join codes for a user.
1475
1476        :param user: The user whose join codes should be revoked.
1477        :return: Number of codes revoked.
1478        """
1479        cursor = await self.database.execute(
1480            "DELETE FROM join_codes WHERE user_id = :user_id",
1481            {"user_id": user.user_id},
1482        )
1483        await self.database.commit()
1484
1485        count = int(cursor.rowcount)
1486        if count > 0:
1487            self.logger.info("Revoked %d join code(s) for user %s", count, user.username)
1488        return count
1489
1490    async def get_active_join_code(self, user: User) -> str | None:
1491        """
1492        Get the most recently created, non-expired join code for a user.
1493
1494        :param user: The user to look up codes for.
1495        :return: The join code string if found, None otherwise.
1496        """
1497        now = utc()
1498        cursor = await self.database.execute(
1499            """
1500            SELECT code FROM join_codes
1501            WHERE user_id = :user_id
1502            AND expires_at > :now
1503            AND (max_uses = 0 OR use_count < max_uses)
1504            ORDER BY created_at DESC
1505            LIMIT 1
1506            """,
1507            {"user_id": user.user_id, "now": now.isoformat()},
1508        )
1509        row = await cursor.fetchone()
1510        return str(row["code"]) if row else None
1511
1512    async def get_join_code_expiry(self, code: str, user: User | None = None) -> datetime | None:
1513        """
1514        Get the expiry datetime for an active join code.
1515
1516        :param code: The join code to look up.
1517        :param user: Optional user that must own the join code.
1518        :return: The expiry datetime if the code is active, None otherwise.
1519        """
1520        query = """
1521            SELECT expires_at FROM join_codes
1522            WHERE code = :code
1523            AND expires_at > :now
1524            AND (max_uses = 0 OR use_count < max_uses)
1525            """
1526        params: dict[str, Any] = {"code": code.upper(), "now": utc().isoformat()}
1527        if user is not None:
1528            query += "AND user_id = :user_id "
1529            params["user_id"] = user.user_id
1530        cursor = await self.database.execute(query + "LIMIT 1", params)
1531        row = await cursor.fetchone()
1532        return datetime.fromisoformat(str(row["expires_at"])) if row else None
1533
1534    @api_command("auth/join_code/exchange", authenticated=False)
1535    async def exchange_join_code(self, code: str) -> dict[str, Any]:
1536        """
1537        Exchange a join code for an access token (public API).
1538
1539        This is the public API endpoint for short-code authentication.
1540        Clients call this with a code (e.g., from QR scan or link) to receive a JWT token.
1541
1542        :param code: The short join code.
1543        :return: Authentication result with access token if successful.
1544        """
1545        rate_limit_key, key_is_exclusive = _join_code_rate_limit_key()
1546        async with self._join_code_exchange_lock:
1547            if throttled := await self._check_join_code_rate_limit(rate_limit_key):
1548                return throttled
1549
1550            token = await self._exchange_join_code(code)
1551
1552            if not token:
1553                await self._join_code_rate_limiter.record_failed_attempt(rate_limit_key)
1554                await self._join_code_global_rate_limiter.record_failed_attempt(
1555                    JOIN_CODE_GLOBAL_RATE_LIMIT_KEY
1556                )
1557                return {
1558                    "success": False,
1559                    "error": "Invalid or expired join code",
1560                }
1561
1562            # A bucket is only cleared when it belongs to one caller alone, so presenting a
1563            # valid code never lifts the throttle for anyone else.
1564            if key_is_exclusive:
1565                await self._join_code_rate_limiter.clear_attempts(rate_limit_key)
1566
1567        # Decode token to get user info
1568        try:
1569            payload = self.jwt_helper.decode_token(token)
1570            return {
1571                "success": True,
1572                "access_token": token,
1573                "user": {
1574                    "user_id": payload.get("sub"),
1575                    "username": payload.get("username"),
1576                    "role": payload.get("role"),
1577                },
1578            }
1579        except pyjwt.InvalidTokenError:
1580            return {
1581                "success": False,
1582                "error": "Failed to create access token",
1583            }
1584
1585    @api_command("auth/join_codes", required_scope=Scope.USERS_MANAGE)
1586    async def list_join_codes(self, user_id: str | None = None) -> list[dict[str, Any]]:
1587        """
1588        List join codes, optionally filtered by user (admin only).
1589
1590        :param user_id: Optional user ID to filter codes for.
1591        :return: List of join code records.
1592        """
1593        filter_args = {"user_id": user_id} if user_id else None
1594        rows = await self.database.get_rows("join_codes", filter_args, limit=100)
1595        return [dict(row) for row in rows]
1596
1597    @api_command("auth/join_code/revoke", required_scope=Scope.USERS_MANAGE)
1598    async def revoke_join_code(self, code_id: str) -> None:
1599        """
1600        Revoke a specific join code (admin only).
1601
1602        :param code_id: The code ID to revoke.
1603        """
1604        code_row = await self.database.get_row("join_codes", {"code_id": code_id})
1605        if not code_row:
1606            raise InvalidDataError("Join code not found")
1607
1608        await self.database.delete("join_codes", {"code_id": code_id})
1609        await self.database.commit()
1610        self.logger.info("Join code revoked (code_id=%s)", code_id)
1611
1612    async def _setup_database(self) -> None:
1613        """Set up database schema and handle migrations."""
1614        # Always create tables if they don't exist
1615        await self._create_database_tables()
1616
1617        # Check current schema version
1618        try:
1619            if db_row := await self.database.get_row("settings", {"key": "schema_version"}):
1620                prev_version = int(db_row["value"])
1621            else:
1622                prev_version = DB_SCHEMA_VERSION
1623        except KeyError, ValueError, Exception:
1624            # settings table doesn't exist yet or other error
1625            prev_version = 0
1626
1627        # Perform migration if needed
1628        if prev_version < DB_SCHEMA_VERSION:
1629            self.logger.warning(
1630                "Performing database migration from schema version %s to %s",
1631                prev_version,
1632                DB_SCHEMA_VERSION,
1633            )
1634            await self._migrate_database(prev_version)
1635
1636        # Store current schema version
1637        await self.database.insert_or_replace(
1638            "settings",
1639            {"key": "schema_version", "value": str(DB_SCHEMA_VERSION), "type": "int"},
1640        )
1641
1642        # Create indexes
1643        await self._create_database_indexes()
1644        await self.database.commit()
1645
1646    async def _create_database_tables(self) -> None:
1647        """Create database tables."""
1648        # Settings table (for schema version and other settings)
1649        await self.database.execute(
1650            """
1651            CREATE TABLE IF NOT EXISTS settings (
1652                key TEXT PRIMARY KEY,
1653                value TEXT,
1654                type TEXT
1655            )
1656            """
1657        )
1658        # Users table
1659        await self.database.execute(
1660            """
1661            CREATE TABLE IF NOT EXISTS users (
1662                user_id TEXT PRIMARY KEY,
1663                username TEXT NOT NULL UNIQUE,
1664                role TEXT NOT NULL,
1665                enabled INTEGER NOT NULL DEFAULT 1,
1666                created_at TEXT NOT NULL,
1667                display_name TEXT,
1668                avatar_url TEXT,
1669                preferences json NOT NULL DEFAULT '{}',
1670                player_filter json NOT NULL DEFAULT '[]',
1671                provider_filter json NOT NULL DEFAULT '[]'
1672            )
1673            """
1674        )
1675        # User auth provider links (many-to-many)
1676        await self.database.execute(
1677            """
1678            CREATE TABLE IF NOT EXISTS user_auth_providers (
1679                link_id TEXT PRIMARY KEY,
1680                user_id TEXT NOT NULL,
1681                provider_type TEXT NOT NULL,
1682                provider_user_id TEXT NOT NULL,
1683                created_at TEXT NOT NULL,
1684                UNIQUE(provider_type, provider_user_id),
1685                FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
1686            )
1687            """
1688        )
1689        # Auth tokens table
1690        await self.database.execute(
1691            """
1692            CREATE TABLE IF NOT EXISTS auth_tokens (
1693                token_id TEXT PRIMARY KEY,
1694                user_id TEXT NOT NULL,
1695                token_hash TEXT NOT NULL UNIQUE,
1696                name TEXT NOT NULL,
1697                created_at TEXT NOT NULL,
1698                expires_at TEXT,
1699                last_used_at TEXT,
1700                is_long_lived INTEGER NOT NULL DEFAULT 0,
1701                FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
1702            )
1703            """
1704        )
1705        # Join codes table (for short code to JWT exchange, used by providers like party)
1706        await self.database.execute(
1707            """
1708            CREATE TABLE IF NOT EXISTS join_codes (
1709                code_id TEXT PRIMARY KEY,
1710                code TEXT NOT NULL UNIQUE,
1711                user_id TEXT NOT NULL,
1712                created_at TEXT NOT NULL,
1713                expires_at TEXT NOT NULL,
1714                max_uses INTEGER DEFAULT 0,
1715                use_count INTEGER DEFAULT 0,
1716                last_used_at TEXT,
1717                device_name TEXT,
1718                FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
1719            )
1720            """
1721        )
1722        await self.database.commit()
1723
1724    async def _create_database_indexes(self) -> None:
1725        """Create database indexes."""
1726        await self.database.execute(
1727            "CREATE INDEX IF NOT EXISTS idx_user_auth_providers_user "
1728            "ON user_auth_providers(user_id)"
1729        )
1730        await self.database.execute(
1731            "CREATE INDEX IF NOT EXISTS idx_user_auth_providers_provider "
1732            "ON user_auth_providers(provider_type, provider_user_id)"
1733        )
1734        await self.database.execute(
1735            "CREATE INDEX IF NOT EXISTS idx_tokens_user ON auth_tokens(user_id)"
1736        )
1737        await self.database.execute(
1738            "CREATE INDEX IF NOT EXISTS idx_tokens_hash ON auth_tokens(token_hash)"
1739        )
1740        await self.database.execute(
1741            "CREATE INDEX IF NOT EXISTS idx_join_codes_user ON join_codes(user_id)"
1742        )
1743
1744    async def _migrate_database(self, from_version: int) -> None:
1745        """
1746        Perform database migration.
1747
1748        :param from_version: The schema version to migrate from.
1749        """
1750        self.logger.info(
1751            "Migrating auth database from version %s to %s", from_version, DB_SCHEMA_VERSION
1752        )
1753        # Migration to version 2: Recreate tables due to password salt breaking change
1754        if from_version < 2:
1755            # Drop all auth-related tables
1756            await self.database.execute("DROP TABLE IF EXISTS auth_tokens")
1757            await self.database.execute("DROP TABLE IF EXISTS user_auth_providers")
1758            await self.database.execute("DROP TABLE IF EXISTS users")
1759            await self.database.commit()
1760
1761            # Recreate tables with current schema
1762            await self._create_database_tables()
1763
1764        # Migration to version 3: Add player_filter and provider_filter columns
1765        if from_version < 3:
1766            with contextlib.suppress(OperationalError):
1767                # Column(s) may already exist
1768                await self.database.execute(
1769                    "ALTER TABLE users ADD COLUMN player_filter json NOT NULL DEFAULT '[]'"
1770                )
1771                await self.database.execute(
1772                    "ALTER TABLE users ADD COLUMN provider_filter json NOT NULL DEFAULT '[]'"
1773                )
1774            await self.database.commit()
1775
1776        # Migration to version 4: Make usernames case-insensitive by converting to lowercase
1777        if from_version < 4:
1778            await self.database.execute("UPDATE users SET username = LOWER(username)")
1779            await self.database.commit()
1780
1781        # Migration to version 5: Add join codes table
1782        if from_version < 5:
1783            await self.database.execute(
1784                """
1785                CREATE TABLE IF NOT EXISTS join_codes (
1786                    code_id TEXT PRIMARY KEY,
1787                    code TEXT NOT NULL UNIQUE,
1788                    user_id TEXT NOT NULL,
1789                    created_at TEXT NOT NULL,
1790                    expires_at TEXT NOT NULL,
1791                    max_uses INTEGER DEFAULT 0,
1792                    use_count INTEGER DEFAULT 0,
1793                    last_used_at TEXT,
1794                    device_name TEXT,
1795                    FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE
1796                )
1797                """
1798            )
1799            await self.database.commit()
1800
1801    async def _get_or_create_jwt_secret(self) -> str:
1802        """
1803        Get or create JWT secret key from database.
1804
1805        :return: JWT secret key for signing tokens.
1806        """
1807        # Try to get existing secret
1808        if secret_row := await self.database.get_row("settings", {"key": "jwt_secret"}):
1809            return str(secret_row["value"])
1810
1811        # Generate new secret
1812        jwt_secret = JWTHelper.generate_secret_key()
1813
1814        # Store in database
1815        await self.database.insert_or_replace(
1816            "settings",
1817            {"key": "jwt_secret", "value": jwt_secret, "type": "string"},
1818        )
1819        await self.database.commit()
1820
1821        self.logger.info("Generated new JWT secret key")
1822        return jwt_secret
1823
1824    async def _setup_login_providers(self) -> None:
1825        """Set up available login providers based on configuration."""
1826        # Always enable built-in provider
1827        self.login_providers["builtin"] = BuiltinLoginProvider(self.mass, "builtin", {})
1828
1829        # Home Assistant OAuth provider
1830        # Automatically enabled if HA provider (plugin) is configured
1831        ha_provider = None
1832        for provider in self.mass.providers:
1833            if provider.domain == "hass" and provider.available:
1834                ha_provider = provider
1835                break
1836
1837        if ha_provider:
1838            ha_provider = cast("HomeAssistantProvider", ha_provider)
1839            ha_url = ha_provider.url
1840            if not ha_url:
1841                self.logger.warning(
1842                    "Home Assistant provider has no URL configured, "
1843                    "Home Assistant OAuth login is not available"
1844                )
1845                return
1846            ha_config: HomeAssistantProviderConfig = {"ha_url": ha_url}
1847            self.login_providers["homeassistant"] = HomeAssistantOAuthProvider(
1848                self.mass, "homeassistant", ha_config
1849            )
1850            self.logger.info(
1851                "Home Assistant OAuth provider enabled (using URL from HA provider: %s)",
1852                ha_url,
1853            )
1854
1855    async def _sync_ha_oauth_provider(self) -> None:
1856        """
1857        Sync HA OAuth provider with HA provider availability (dynamic check).
1858
1859        Adds the provider if HA is available, removes it if HA is not available.
1860        """
1861        # Find HA provider
1862        ha_provider = None
1863        for provider in self.mass.providers:
1864            if provider.domain == "hass" and provider.available:
1865                ha_provider = provider
1866                break
1867
1868        if ha_provider:
1869            # HA provider exists and is available - ensure OAuth provider is registered
1870            if "homeassistant" not in self.login_providers:
1871                ha_provider = cast("HomeAssistantProvider", ha_provider)
1872                ha_url = ha_provider.url
1873                if not ha_url:
1874                    # missing URL must never break the login providers endpoint,
1875                    # simply leave the HA OAuth provider unregistered
1876                    self.logger.debug(
1877                        "Home Assistant provider has no URL configured, "
1878                        "Home Assistant OAuth login is not available"
1879                    )
1880                    return
1881                ha_config: HomeAssistantProviderConfig = {"ha_url": ha_url}
1882                self.login_providers["homeassistant"] = HomeAssistantOAuthProvider(
1883                    self.mass, "homeassistant", ha_config
1884                )
1885                self.logger.info(
1886                    "Home Assistant OAuth provider dynamically enabled (using URL: %s)",
1887                    ha_url,
1888                )
1889        # HA provider not available - remove OAuth provider if present
1890        elif "homeassistant" in self.login_providers:
1891            del self.login_providers["homeassistant"]
1892            self.logger.info("Home Assistant OAuth provider removed (HA provider not available)")
1893
1894    async def _has_non_system_users(self) -> bool:
1895        """Check if any non-system users exist."""
1896        user_rows = await self.database.get_rows("users", limit=10)
1897        return any(row["username"] != HOMEASSISTANT_SYSTEM_USER for row in user_rows)
1898
1899    async def _migrate_system_user_role(self) -> None:
1900        """Migrate the Home Assistant system user of pre-existing installs to the service role."""
1901        user_row = await self.database.get_row(
1902            "users", {"username": normalize_username(HOMEASSISTANT_SYSTEM_USER)}
1903        )
1904        if user_row and user_row["role"] != UserRole.SERVICE.value:
1905            await self.database.update(
1906                "users", {"user_id": user_row["user_id"]}, {"role": UserRole.SERVICE.value}
1907            )
1908            self.logger.info(
1909                "Updated Home Assistant system user role to %s", UserRole.SERVICE.value
1910            )
1911
1912    async def _migrate_playlog_to_first_user(self, user_id: str) -> None:
1913        """
1914        Migrate all existing playlog entries to the first user.
1915
1916        This is called automatically when the first non-system user is created.
1917        All existing playlog entries (which have NULL userid) will be updated
1918        to belong to this first user.
1919
1920        :param user_id: The user ID of the first user.
1921        """
1922        try:
1923            # Update all playlog entries with NULL userid to this user
1924            await self.mass.music.database.execute(
1925                f"UPDATE {DB_TABLE_PLAYLOG} SET userid = :userid WHERE userid IS NULL",
1926                {"userid": user_id},
1927            )
1928            await self.mass.music.database.commit()
1929            self.logger.info("Migrated existing playlog entries to first user: %s", user_id)
1930        except Exception as err:
1931            self.logger.warning("Failed to migrate playlog entries: %s", err)
1932
1933    async def _update_profile_password(
1934        self,
1935        target_user: User,
1936        password: str,
1937        is_admin_update: bool,
1938        current_user: User,
1939    ) -> None:
1940        """Update user password (helper method)."""
1941        if len(password) < 8:
1942            raise InvalidDataError("Password must be at least 8 characters")
1943
1944        builtin_provider = self.login_providers.get("builtin")
1945        if not builtin_provider or not isinstance(builtin_provider, BuiltinLoginProvider):
1946            raise InvalidDataError("Built-in auth not available")
1947
1948        # Update password (used for both admin resets and user password changes)
1949        await builtin_provider.reset_password(target_user, password)
1950
1951        if is_admin_update:
1952            self.logger.info(
1953                "Password reset for user %s by admin %s",
1954                target_user.username,
1955                current_user.username,
1956            )
1957        else:
1958            self.logger.info("Password changed for user %s", target_user.username)
1959
1960    async def _check_join_code_rate_limit(self, key: str) -> dict[str, Any] | None:
1961        """
1962        Check the join code exchange throttles that apply to the calling client.
1963
1964        :param key: Rate limit key identifying the calling client.
1965        :return: The error result to return to the caller, or None if the attempt may proceed.
1966        """
1967        limiters = (
1968            ("client", self._join_code_rate_limiter, key),
1969            ("server", self._join_code_global_rate_limiter, JOIN_CODE_GLOBAL_RATE_LIMIT_KEY),
1970        )
1971        for scope, limiter, limiter_key in limiters:
1972            allowed, remaining_delay = await limiter.check_rate_limit(limiter_key)
1973            if allowed:
1974                continue
1975            # The attempted code is deliberately absent here: it has not been checked yet,
1976            # so it may well be a valid one. Each failure that filled the bucket already
1977            # logged its own (rejected, and therefore unusable) code.
1978            self.logger.warning(
1979                "Join code exchange throttled by the %s limit "
1980                "(client=%s, client_failures=%d, server_failures=%d). "
1981                "%d seconds remaining.",
1982                scope,
1983                key,
1984                self._join_code_rate_limiter.get_attempt_count(key),
1985                self._join_code_global_rate_limiter.get_attempt_count(
1986                    JOIN_CODE_GLOBAL_RATE_LIMIT_KEY
1987                ),
1988                remaining_delay,
1989            )
1990            return {
1991                "success": False,
1992                "error": (
1993                    f"Too many failed attempts. Please try again in {remaining_delay} seconds."
1994                ),
1995            }
1996        return None
1997
1998    async def _exchange_join_code(self, code: str) -> str | None:
1999        """
2000        Exchange a join code for a JWT access token.
2001
2002        The token is created for the user associated with the join code.
2003
2004        :param code: The short join code.
2005        :return: JWT token string if valid, None otherwise.
2006        """
2007        now = utc()
2008
2009        cursor = await self.database.execute(
2010            """
2011            UPDATE join_codes
2012            SET use_count = use_count + 1,
2013                last_used_at = :now
2014            WHERE code = :code
2015            AND expires_at > :now
2016            AND (max_uses = 0 OR use_count < max_uses)
2017            RETURNING user_id, device_name
2018            """,
2019            {"now": now.isoformat(), "code": code.upper()},
2020        )
2021        row = await cursor.fetchone()
2022        await self.database.commit()
2023
2024        if not row:
2025            self.logger.warning(
2026                "Join code exchange rejected (client=%s, code=%s)",
2027                get_current_client_id() or JOIN_CODE_ANONYMOUS_RATE_LIMIT_KEY,
2028                _mask_join_code(code),
2029            )
2030            return None
2031
2032        user = await self.get_user(row["user_id"])
2033        if not user:
2034            self.logger.error(
2035                "User not found for join code despite FK constraint (user_id=%s)", row["user_id"]
2036            )
2037            return None
2038
2039        device_name = row["device_name"] or "Short Code Login"
2040        token = await self.create_token(
2041            user,
2042            device_name,
2043            is_long_lived=False,
2044        )
2045
2046        self.logger.info(
2047            "Join code exchanged for token (user=%s)",
2048            user.username,
2049        )
2050        return token
2051
2052    async def _cleanup_expired_join_codes(self) -> None:
2053        """Delete expired and exhausted join codes from the database."""
2054        now = utc()
2055        cursor = await self.database.execute(
2056            """
2057            DELETE FROM join_codes
2058            WHERE expires_at < :now
2059               OR (max_uses > 0 AND use_count >= max_uses)
2060            """,
2061            {"now": now.isoformat()},
2062        )
2063        await self.database.commit()
2064        count = int(cursor.rowcount)
2065        if count > 0:
2066            self.logger.debug("Cleaned up %d expired/exhausted join code(s)", count)
2067
2068    def _schedule_join_code_cleanup(self) -> None:
2069        """Schedule periodic cleanup of expired join codes."""
2070        self.mass.create_task(self._cleanup_expired_join_codes())
2071        self.mass.call_later(86400, self._schedule_join_code_cleanup)
2072
2073    async def _refresh_token_expiration(
2074        self, token_row: Mapping[str, Any], user: User, is_long_lived: bool
2075    ) -> dict[str, str] | None:
2076        """
2077        Build the on-use column updates for a token, enforcing the absolute lifetime cap.
2078
2079        :param token_row: The auth_tokens row for the token being used.
2080        :param user: The user owning the token.
2081        :param is_long_lived: Whether the token is long-lived.
2082        :return: Column updates to apply (empty when the stored activity timestamp is
2083            still fresh, so callers can skip the write), or None if the token exceeded
2084            its max lifetime (in which case the token row is deleted).
2085        """
2086        now = utc()
2087
2088        if not is_long_lived:
2089            created_at = datetime.fromisoformat(token_row["created_at"])
2090            if now > created_at + timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION):
2091                await self.database.delete("auth_tokens", {"token_id": token_row["token_id"]})
2092                return None
2093
2094        # The HTTP API authenticates on every request, so persisting activity per use
2095        # would cost an UPDATE+commit (an fsync) per request. Skip the write while the
2096        # stored timestamp is fresh; last_used_at and the sliding expiration then lag
2097        # by at most this interval, which is negligible against the 30-day idle window.
2098        if last_used_at := token_row["last_used_at"]:
2099            if now - datetime.fromisoformat(last_used_at) < TOKEN_ACTIVITY_PERSIST_INTERVAL:
2100                return {}
2101
2102        updates = {"last_used_at": now.isoformat()}
2103        if not is_long_lived and user.role != UserRole.GUEST:
2104            # Short-lived token: extend expiration on each use (sliding window)
2105            new_expires_at = now + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
2106            updates["expires_at"] = new_expires_at.isoformat()
2107
2108        return updates
2109
2110    async def _can_reuse_ha_integration_token(self, token: str, system_user: User) -> bool:
2111        """Check whether the stored HA integration token is valid and not yet due for rotation."""
2112        token_id = self.jwt_helper.get_token_id(token)
2113        if not token_id:
2114            return False
2115        token_row = await self.database.get_row("auth_tokens", {"token_id": token_id})
2116        if not token_row or token_row["user_id"] != system_user.user_id:
2117            return False
2118        now = utc()
2119        if token_row["expires_at"] and datetime.fromisoformat(token_row["expires_at"]) <= now:
2120            return False
2121        created_at = datetime.fromisoformat(token_row["created_at"])
2122        rotate_after = created_at + timedelta(
2123            days=TOKEN_ABSOLUTE_MAX_EXPIRATION - HA_TOKEN_ROTATION_MARGIN
2124        )
2125        return now < rotate_after
2126
2127    def _notify_user_access_revoked(self, user: User) -> None:
2128        """Dispatch an access withdrawal to subscribers, isolating them from each other."""
2129        for callback in list(self._access_revoked_callbacks):
2130            self.mass.loop.call_soon(callback, user)
2131
2132
2133def _join_code_rate_limit_key() -> tuple[str, bool]:
2134    """
2135    Work out which bucket the calling client's failed join code exchanges belong to.
2136
2137    :return: The rate limit key, and whether that key identifies a single caller
2138        exclusively (a shared key must never be cleared on a successful exchange).
2139    """
2140    if client_id := get_current_client_id():
2141        return client_id, True
2142    if peer_address := get_current_peer_address():
2143        return f"peer:{peer_address}", False
2144    return JOIN_CODE_ANONYMOUS_RATE_LIMIT_KEY, False
2145
2146
2147def _mask_join_code(code: str) -> str:
2148    """
2149    Mask a join code so support logs can correlate attempts without exposing a usable code.
2150
2151    :param code: The join code as supplied by the client.
2152    :return: The code with everything past its prefix replaced by asterisks.
2153    """
2154    normalized = code.upper()
2155    return normalized[:4] + "*" * max(len(normalized) - 4, 0)
2156