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