music-assistant-server

1"""Tests for webserver authentication and user management."""
2
3import asyncio
4import hashlib
5import logging
6import pathlib
7import threading
8from collections.abc import AsyncGenerator
9from datetime import datetime, timedelta
10from sqlite3 import IntegrityError
11from typing import Any
12
13import pytest
14from music_assistant_models.auth import AuthProviderType, Scope, User, UserRole
15from music_assistant_models.errors import InsufficientPermissions, InvalidDataError
16
17from music_assistant.constants import HOMEASSISTANT_SYSTEM_USER
18from music_assistant.controllers.config import ConfigController
19from music_assistant.controllers.webserver.auth import (
20    JOIN_CODE_GLOBAL_FAILURE_CEILING,
21    JOIN_CODE_GLOBAL_RATE_LIMIT_KEY,
22    JOIN_CODE_LENGTH,
23    TOKEN_ABSOLUTE_MAX_EXPIRATION,
24    TOKEN_ACTIVITY_PERSIST_INTERVAL,
25    TOKEN_GUEST_EXPIRATION,
26    TOKEN_LONG_LIVED_EXPIRATION,
27    TOKEN_SHORT_LIVED_EXPIRATION,
28    AuthenticationManager,
29    _mask_join_code,
30)
31from music_assistant.controllers.webserver.controller import WebserverController
32from music_assistant.controllers.webserver.helpers.auth_middleware import (
33    ImpersonatedUser,
34    get_current_user,
35    has_scope,
36    resolve_command_impersonation,
37    set_current_client_id,
38    set_current_peer_address,
39    set_current_token,
40    set_current_user,
41    set_impersonated_user,
42)
43from music_assistant.controllers.webserver.helpers.auth_providers import (
44    PRUNE_THRESHOLD,
45    BuiltinLoginProvider,
46    LoginRateLimiter,
47)
48from music_assistant.helpers.datetime import utc
49from music_assistant.mass import MusicAssistant
50
51
52@pytest.fixture
53async def mass_minimal(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
54    """
55    Create a minimal Music Assistant instance for auth testing without starting the webserver.
56
57    :param tmp_path: Temporary directory for test data.
58    """
59    storage_path = tmp_path / "data"
60    cache_path = tmp_path / "cache"
61    storage_path.mkdir(parents=True)
62    cache_path.mkdir(parents=True)
63
64    # Suppress aiosqlite debug logging
65    logging.getLogger("aiosqlite").level = logging.INFO
66
67    mass_instance = MusicAssistant(str(storage_path), str(cache_path))
68
69    # Initialize the minimum required for auth testing
70    mass_instance.loop = asyncio.get_running_loop()
71    # fixture runs on the event loop thread, like MusicAssistant.start()
72    mass_instance.loop_thread_id = threading.get_ident()
73
74    # Create config controller
75    mass_instance.config = ConfigController(mass_instance)
76    await mass_instance.config.setup()
77
78    # Create webserver controller (but don't start the actual server)
79    webserver = WebserverController(mass_instance)
80    mass_instance.webserver = webserver
81
82    # Get webserver config and manually set it (avoids starting the server)
83    webserver_config = await mass_instance.config.get_core_config("webserver")
84    webserver.config = webserver_config
85
86    # Setup auth manager only (not the full webserver with routes/sockets)
87    await webserver.auth.setup()
88
89    try:
90        yield mass_instance
91    finally:
92        # Cleanup
93        await webserver.auth.close()
94        await mass_instance.config.close()
95
96
97@pytest.fixture
98async def auth_manager(mass_minimal: MusicAssistant) -> AuthenticationManager:
99    """
100    Get authentication manager from mass instance.
101
102    :param mass_minimal: Minimal MusicAssistant instance.
103    """
104    return mass_minimal.webserver.auth
105
106
107async def test_auth_manager_initialization(auth_manager: AuthenticationManager) -> None:
108    """
109    Test that the authentication manager initializes correctly.
110
111    :param auth_manager: AuthenticationManager instance.
112    """
113    assert auth_manager is not None
114    assert auth_manager.database is not None
115    assert "builtin" in auth_manager.login_providers
116    assert isinstance(auth_manager.login_providers["builtin"], BuiltinLoginProvider)
117
118
119async def test_has_users_initially_empty(auth_manager: AuthenticationManager) -> None:
120    """
121    Test that has_users returns False when no users exist.
122
123    :param auth_manager: AuthenticationManager instance.
124    """
125    has_users = auth_manager.has_users
126    assert has_users is False
127
128
129async def test_create_user(auth_manager: AuthenticationManager) -> None:
130    """
131    Test creating a new user.
132
133    :param auth_manager: AuthenticationManager instance.
134    """
135    user = await auth_manager.create_user(
136        username="testuser",
137        role=UserRole.USER,
138        display_name="Test User",
139    )
140
141    assert user is not None
142    assert user.username == "testuser"
143    assert user.role == UserRole.USER
144    assert user.display_name == "Test User"
145    assert user.enabled is True
146    assert user.user_id is not None
147
148    # Verify user exists in database
149    has_users = auth_manager.has_users
150    assert has_users is True
151
152
153async def test_get_user(auth_manager: AuthenticationManager) -> None:
154    """
155    Test retrieving a user by ID.
156
157    :param auth_manager: AuthenticationManager instance.
158    """
159    # Create a user first
160    created_user = await auth_manager.create_user(username="getuser", role=UserRole.USER)
161
162    # Set current user for authorization (get_user requires admin role)
163    admin_user = await auth_manager.create_user(username="admin", role=UserRole.ADMIN)
164    set_current_user(admin_user)
165
166    # Retrieve the user
167    retrieved_user = await auth_manager.get_user(created_user.user_id)
168
169    assert retrieved_user is not None
170    assert retrieved_user.user_id == created_user.user_id
171    assert retrieved_user.username == created_user.username
172
173
174async def test_create_user_with_builtin_provider(auth_manager: AuthenticationManager) -> None:
175    """
176    Test creating a user with built-in authentication.
177
178    :param auth_manager: AuthenticationManager instance.
179    """
180    builtin_provider = auth_manager.login_providers.get("builtin")
181    assert builtin_provider is not None
182    assert isinstance(builtin_provider, BuiltinLoginProvider)
183
184    user = await builtin_provider.create_user_with_password(
185        username="testuser2",
186        password="testpassword123",
187        role=UserRole.USER,
188    )
189
190    assert user is not None
191    assert user.username == "testuser2"
192
193
194async def test_authenticate_with_password(auth_manager: AuthenticationManager) -> None:
195    """
196    Test authenticating with username and password.
197
198    :param auth_manager: AuthenticationManager instance.
199    """
200    builtin_provider = auth_manager.login_providers.get("builtin")
201    assert builtin_provider is not None
202    assert isinstance(builtin_provider, BuiltinLoginProvider)
203
204    # Create user with password
205    await builtin_provider.create_user_with_password(
206        username="authtest",
207        password="secure_password_123",
208        role=UserRole.USER,
209    )
210
211    # Test successful authentication
212    result = await auth_manager.authenticate_with_credentials(
213        "builtin",
214        {"username": "authtest", "password": "secure_password_123"},
215    )
216
217    assert result.success is True
218    assert result.user is not None
219    assert result.user.username == "authtest"
220    # Note: Built-in provider doesn't auto-generate access token on login,
221    # that's done by the web login flow. We just verify authentication succeeds.
222
223    # Test failed authentication with wrong password
224    result = await auth_manager.authenticate_with_credentials(
225        "builtin",
226        {"username": "authtest", "password": "wrong_password"},
227    )
228
229    assert result.success is False
230    assert result.user is None
231    assert result.error is not None
232
233
234async def test_create_token(auth_manager: AuthenticationManager) -> None:
235    """
236    Test creating access tokens.
237
238    :param auth_manager: AuthenticationManager instance.
239    """
240    user = await auth_manager.create_user(username="tokenuser", role=UserRole.USER)
241
242    # Create short-lived token
243    short_token = await auth_manager.create_token(user, "Test Device", is_long_lived=False)
244    assert short_token is not None
245    assert len(short_token) > 0
246
247    # Create long-lived token
248    long_token = await auth_manager.create_token(user, "API Key", is_long_lived=True)
249    assert long_token is not None
250    assert len(long_token) > 0
251    assert long_token != short_token
252
253
254async def test_authenticate_with_token(auth_manager: AuthenticationManager) -> None:
255    """
256    Test authenticating with an access token.
257
258    :param auth_manager: AuthenticationManager instance.
259    """
260    user = await auth_manager.create_user(username="tokenauth", role=UserRole.USER)
261    token = await auth_manager.create_token(user, "Test Token", is_long_lived=False)
262
263    # Authenticate with token
264    authenticated_user = await auth_manager.authenticate_with_token(token)
265
266    assert authenticated_user is not None
267    assert authenticated_user.user_id == user.user_id
268    assert authenticated_user.username == user.username
269
270
271async def test_token_expiration(auth_manager: AuthenticationManager) -> None:
272    """
273    Test that expired tokens are rejected.
274
275    :param auth_manager: AuthenticationManager instance.
276    """
277    user = await auth_manager.create_user(username="expireuser", role=UserRole.USER)
278    token = await auth_manager.create_token(user, "Expire Test", is_long_lived=False)
279
280    # Hash the token to look it up
281    token_hash = hashlib.sha256(token.encode()).hexdigest()
282    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
283    assert token_row is not None
284
285    # Manually expire the token by setting expires_at in the past
286    past_time = utc() - timedelta(days=1)
287    await auth_manager.database.update(
288        "auth_tokens",
289        {"token_id": token_row["token_id"]},
290        {"expires_at": past_time.isoformat()},
291    )
292
293    # Try to authenticate with expired token
294    authenticated_user = await auth_manager.authenticate_with_token(token)
295    assert authenticated_user is None
296
297
298async def test_update_user_profile(auth_manager: AuthenticationManager) -> None:
299    """
300    Test updating user profile information.
301
302    :param auth_manager: AuthenticationManager instance.
303    """
304    user = await auth_manager.create_user(
305        username="updateuser",
306        role=UserRole.USER,
307        display_name="Original Name",
308    )
309
310    # Update user profile
311    updated_user = await auth_manager.update_user(
312        user,
313        display_name="New Name",
314        avatar_url="https://example.com/avatar.jpg",
315    )
316
317    assert updated_user is not None
318    assert updated_user.display_name == "New Name"
319    assert updated_user.avatar_url == "https://example.com/avatar.jpg"
320    assert updated_user.username == user.username
321
322
323async def test_change_password(auth_manager: AuthenticationManager) -> None:
324    """
325    Test changing user password.
326
327    :param auth_manager: AuthenticationManager instance.
328    """
329    builtin_provider = auth_manager.login_providers.get("builtin")
330    assert builtin_provider is not None
331    assert isinstance(builtin_provider, BuiltinLoginProvider)
332
333    # Create user with password
334    user = await builtin_provider.create_user_with_password(
335        username="pwdchange",
336        password="old_password_123",
337        role=UserRole.USER,
338    )
339
340    # Change password
341    success = await builtin_provider.change_password(
342        user,
343        "old_password_123",
344        "new_password_456",
345    )
346    assert success is True
347
348    # Verify old password no longer works
349    result = await auth_manager.authenticate_with_credentials(
350        "builtin",
351        {"username": "pwdchange", "password": "old_password_123"},
352    )
353    assert result.success is False
354
355    # Verify new password works
356    result = await auth_manager.authenticate_with_credentials(
357        "builtin",
358        {"username": "pwdchange", "password": "new_password_456"},
359    )
360    assert result.success is True
361
362
363async def test_revoke_token(auth_manager: AuthenticationManager) -> None:
364    """
365    Test revoking an access token.
366
367    :param auth_manager: AuthenticationManager instance.
368    """
369    user = await auth_manager.create_user(username="revokeuser", role=UserRole.USER)
370    token = await auth_manager.create_token(user, "Revoke Test", is_long_lived=False)
371
372    # Set current user context for authorization
373    set_current_user(user)
374
375    # Get token_id
376    token_id = await auth_manager.get_token_id_from_token(token)
377    assert token_id is not None
378
379    # Token should work before revocation
380    authenticated_user = await auth_manager.authenticate_with_token(token)
381    assert authenticated_user is not None
382
383    # Revoke the token
384    await auth_manager.revoke_token(token_id)
385
386    # Token should not work after revocation
387    authenticated_user = await auth_manager.authenticate_with_token(token)
388    assert authenticated_user is None
389
390
391async def test_list_users(auth_manager: AuthenticationManager) -> None:
392    """
393    Test listing all users (requires the users.read scope).
394
395    :param auth_manager: AuthenticationManager instance.
396    """
397    # Create admin user and set as current
398    admin = await auth_manager.create_user(username="listadmin", role=UserRole.ADMIN)
399    set_current_user(admin)
400
401    # Create some test users
402    await auth_manager.create_user(username="user1", role=UserRole.USER)
403    await auth_manager.create_user(username="user2", role=UserRole.USER)
404
405    # List all users
406    users = await auth_manager.list_users()
407
408    # Should not include system users
409    usernames = [u.username for u in users]
410    assert "listadmin" in usernames
411    assert "user1" in usernames
412    assert "user2" in usernames
413
414
415async def test_disable_enable_user(auth_manager: AuthenticationManager) -> None:
416    """
417    Test disabling and enabling user accounts.
418
419    :param auth_manager: AuthenticationManager instance.
420    """
421    # Create admin and regular user
422    admin = await auth_manager.create_user(username="disableadmin", role=UserRole.ADMIN)
423    user = await auth_manager.create_user(username="disableuser", role=UserRole.USER)
424
425    # Set admin as current user
426    set_current_user(admin)
427
428    # Disable the user
429    await auth_manager.disable_user(user.user_id)
430
431    # Verify user is disabled
432    disabled_user = await auth_manager.get_user(user.user_id)
433    assert disabled_user is None  # get_user filters out disabled users
434
435    # Enable the user
436    await auth_manager.enable_user(user.user_id)
437
438    # Verify user is enabled
439    enabled_user = await auth_manager.get_user(user.user_id)
440    assert enabled_user is not None
441
442
443async def test_cannot_disable_own_account(auth_manager: AuthenticationManager) -> None:
444    """
445    Test that users cannot disable their own account.
446
447    :param auth_manager: AuthenticationManager instance.
448    """
449    admin = await auth_manager.create_user(username="selfadmin", role=UserRole.ADMIN)
450    set_current_user(admin)
451
452    # Try to disable own account
453    with pytest.raises(InvalidDataError, match="Cannot disable your own account"):
454        await auth_manager.disable_user(admin.user_id)
455
456
457async def test_user_preferences(auth_manager: AuthenticationManager) -> None:
458    """
459    Test updating user preferences.
460
461    :param auth_manager: AuthenticationManager instance.
462    """
463    user = await auth_manager.create_user(username="prefuser", role=UserRole.USER)
464
465    # Update preferences
466    preferences = {"theme": "dark", "language": "en"}
467    updated_user = await auth_manager.update_user_preferences(user, preferences)
468
469    assert updated_user is not None
470    assert updated_user.preferences == preferences
471
472
473async def test_link_user_to_provider(auth_manager: AuthenticationManager) -> None:
474    """
475    Test linking user to authentication provider.
476
477    :param auth_manager: AuthenticationManager instance.
478    """
479    user = await auth_manager.create_user(username="linkuser", role=UserRole.USER)
480
481    # Link to provider
482    link = await auth_manager.link_user_to_provider(
483        user,
484        AuthProviderType.HOME_ASSISTANT,
485        "ha_user_123",
486    )
487
488    assert link is not None
489    assert link.user_id == user.user_id
490    assert link.provider_type == AuthProviderType.HOME_ASSISTANT
491    assert link.provider_user_id == "ha_user_123"
492
493    # Retrieve user by provider link
494    retrieved_user = await auth_manager.get_user_by_provider_link(
495        AuthProviderType.HOME_ASSISTANT,
496        "ha_user_123",
497    )
498
499    assert retrieved_user is not None
500    assert retrieved_user.user_id == user.user_id
501
502
503async def test_homeassistant_system_user(auth_manager: AuthenticationManager) -> None:
504    """
505    Test Home Assistant system user creation.
506
507    :param auth_manager: AuthenticationManager instance.
508    """
509    # Get or create system user
510    system_user = await auth_manager.get_homeassistant_system_user()
511
512    assert system_user is not None
513    assert system_user.username == HOMEASSISTANT_SYSTEM_USER
514    assert system_user.display_name == "Home Assistant Integration"
515    assert system_user.role == UserRole.SERVICE
516
517    # Getting it again should return the same user
518    system_user2 = await auth_manager.get_homeassistant_system_user()
519    assert system_user2.user_id == system_user.user_id
520
521
522async def test_homeassistant_system_user_token_stable_across_restarts(
523    auth_manager: AuthenticationManager,
524) -> None:
525    """
526    Test that a valid Home Assistant integration token is reused on repeated announces.
527
528    The addon announces on every startup; re-minting each time would invalidate the
529    token the HA integration still holds (issue #158174). Repeated calls must return
530    the exact same token so the re-announce is idempotent.
531
532    :param auth_manager: AuthenticationManager instance.
533    """
534    token1 = await auth_manager.get_homeassistant_system_user_token()
535    assert token1 is not None
536
537    # A later startup (restart) returns the same token unchanged.
538    token2 = await auth_manager.get_homeassistant_system_user_token()
539    assert token2 == token1
540
541    user = await auth_manager.authenticate_with_token(token1)
542    assert user is not None
543    assert user.username == HOMEASSISTANT_SYSTEM_USER
544
545
546async def test_homeassistant_system_user_token_reissued_when_invalid(
547    auth_manager: AuthenticationManager,
548) -> None:
549    """
550    Test that a fresh token is minted once the existing one is revoked or gone.
551
552    :param auth_manager: AuthenticationManager instance.
553    """
554    token1 = await auth_manager.get_homeassistant_system_user_token()
555
556    # Drop the token row, as would happen if it expired or was revoked.
557    token_id = auth_manager.jwt_helper.get_token_id(token1)
558    await auth_manager.database.delete("auth_tokens", {"token_id": token_id})
559
560    token2 = await auth_manager.get_homeassistant_system_user_token()
561    assert token2 != token1
562    assert await auth_manager.authenticate_with_token(token2) is not None
563    assert await auth_manager.authenticate_with_token(token1) is None
564
565
566async def test_homeassistant_system_user_token_rotated_before_absolute_max(
567    auth_manager: AuthenticationManager,
568) -> None:
569    """
570    Test that the Home Assistant integration token is rotated before its absolute cap.
571
572    The integration cannot reauth while running as an addon, so the token must be
573    replaced (and re-announced) before the absolute lifetime cap silently strands
574    the integration (issue #171938). The superseded token must remain valid so the
575    integration keeps working until it reloads with the new one.
576
577    :param auth_manager: AuthenticationManager instance.
578    """
579    token1 = await auth_manager.get_homeassistant_system_user_token()
580    token_id = auth_manager.jwt_helper.get_token_id(token1)
581
582    # Age the token into the rotation window (close to the absolute cap, still valid).
583    now = utc()
584    created_at = now - timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION - 1)
585    await auth_manager.database.update(
586        "auth_tokens",
587        {"token_id": token_id},
588        {
589            "created_at": created_at.isoformat(),
590            "expires_at": (now + timedelta(days=1)).isoformat(),
591        },
592    )
593
594    # The next (periodic) announce must mint a replacement.
595    token2 = await auth_manager.get_homeassistant_system_user_token()
596    assert token2 != token1
597
598    # Both tokens work: the old one until it expires, the new one going forward.
599    assert await auth_manager.authenticate_with_token(token1) is not None
600    assert await auth_manager.authenticate_with_token(token2) is not None
601
602    # The new token is stable again on subsequent announces.
603    assert await auth_manager.get_homeassistant_system_user_token() == token2
604
605
606async def test_homeassistant_system_user_token_cleans_up_expired_rows(
607    auth_manager: AuthenticationManager,
608) -> None:
609    """
610    Test that expired Home Assistant integration token rows are removed on rotation.
611
612    :param auth_manager: AuthenticationManager instance.
613    """
614    token1 = await auth_manager.get_homeassistant_system_user_token()
615    token_id = auth_manager.jwt_helper.get_token_id(token1)
616
617    # Expire the token entirely so the next announce mints a replacement.
618    await auth_manager.database.update(
619        "auth_tokens",
620        {"token_id": token_id},
621        {"expires_at": (utc() - timedelta(days=1)).isoformat()},
622    )
623
624    token2 = await auth_manager.get_homeassistant_system_user_token()
625    assert token2 != token1
626    assert await auth_manager.database.get_row("auth_tokens", {"token_id": token_id}) is None
627
628
629async def test_update_user_role(auth_manager: AuthenticationManager) -> None:
630    """
631    Test updating user role (admin only).
632
633    :param auth_manager: AuthenticationManager instance.
634    """
635    admin = await auth_manager.create_user(username="roleadmin", role=UserRole.ADMIN)
636    user = await auth_manager.create_user(username="roleuser", role=UserRole.USER)
637
638    # Update role
639    success = await auth_manager.update_user_role(user.user_id, UserRole.ADMIN, admin)
640    assert success is True
641
642    # Verify role was updated
643    set_current_user(admin)
644    updated_user = await auth_manager.get_user(user.user_id)
645    assert updated_user is not None
646    assert updated_user.role == UserRole.ADMIN
647
648
649async def test_delete_user(auth_manager: AuthenticationManager) -> None:
650    """
651    Test deleting a user account.
652
653    :param auth_manager: AuthenticationManager instance.
654    """
655    admin = await auth_manager.create_user(username="deleteadmin", role=UserRole.ADMIN)
656    user = await auth_manager.create_user(username="deleteuser", role=UserRole.USER)
657
658    # Set admin as current user
659    set_current_user(admin)
660
661    # Delete the user
662    await auth_manager.delete_user(user.user_id)
663
664    # Verify user is deleted
665    deleted_user = await auth_manager.get_user(user.user_id)
666    assert deleted_user is None
667
668
669async def test_cannot_delete_own_account(auth_manager: AuthenticationManager) -> None:
670    """
671    Test that users cannot delete their own account.
672
673    :param auth_manager: AuthenticationManager instance.
674    """
675    admin = await auth_manager.create_user(username="selfdeleteadmin", role=UserRole.ADMIN)
676    set_current_user(admin)
677
678    # Try to delete own account
679    with pytest.raises(InvalidDataError, match="Cannot delete your own account"):
680        await auth_manager.delete_user(admin.user_id)
681
682
683async def test_get_user_tokens(auth_manager: AuthenticationManager) -> None:
684    """
685    Test getting user's tokens.
686
687    :param auth_manager: AuthenticationManager instance.
688    """
689    user = await auth_manager.create_user(username="tokensuser", role=UserRole.USER)
690    set_current_user(user)
691
692    # Create some tokens
693    await auth_manager.create_token(user, "Device 1", is_long_lived=False)
694    await auth_manager.create_token(user, "Device 2", is_long_lived=True)
695
696    # Get user tokens
697    tokens = await auth_manager.get_user_tokens(user.user_id)
698
699    assert len(tokens) == 2
700    token_names = [t.name for t in tokens]
701    assert "Device 1" in token_names
702    assert "Device 2" in token_names
703
704
705async def test_get_login_providers(auth_manager: AuthenticationManager) -> None:
706    """
707    Test getting available login providers.
708
709    :param auth_manager: AuthenticationManager instance.
710    """
711    providers = await auth_manager.get_login_providers()
712
713    assert len(providers) > 0
714    assert any(p["provider_id"] == "builtin" for p in providers)
715
716
717class _FakeHassProvider:
718    """Minimal stand-in for the Home Assistant provider."""
719
720    domain = "hass"
721    available = True
722
723    def __init__(self, url: str | None) -> None:
724        self._url = url
725
726    @property
727    def url(self) -> str | None:
728        """Return the configured Home Assistant URL, or None if not configured."""
729        return self._url
730
731
732async def test_get_login_providers_with_ha_provider(
733    auth_manager: AuthenticationManager, mass_minimal: MusicAssistant
734) -> None:
735    """
736    Test that the HA OAuth login provider is registered when the HA provider has a URL.
737
738    :param auth_manager: AuthenticationManager instance.
739    :param mass_minimal: Minimal MusicAssistant instance.
740    """
741    mass_minimal._providers["hass"] = _FakeHassProvider("http://homeassistant.local:8123")  # type: ignore[assignment]
742
743    providers = await auth_manager.get_login_providers()
744
745    assert any(p["provider_id"] == "homeassistant" for p in providers)
746
747
748async def test_get_login_providers_ha_provider_without_url(
749    auth_manager: AuthenticationManager, mass_minimal: MusicAssistant
750) -> None:
751    """
752    Test that a HA provider without a URL does not break the login providers endpoint.
753
754    Regression test for the HA provider storing its URL in setup data instead of
755    config: builtin login must remain available and the endpoint must not raise.
756
757    :param auth_manager: AuthenticationManager instance.
758    :param mass_minimal: Minimal MusicAssistant instance.
759    """
760    mass_minimal._providers["hass"] = _FakeHassProvider(None)  # type: ignore[assignment]
761
762    providers = await auth_manager.get_login_providers()
763
764    assert any(p["provider_id"] == "builtin" for p in providers)
765    assert not any(p["provider_id"] == "homeassistant" for p in providers)
766
767
768async def test_create_user_with_api(auth_manager: AuthenticationManager) -> None:
769    """
770    Test creating user via API command.
771
772    :param auth_manager: AuthenticationManager instance.
773    """
774    # Create admin user and set as current
775    admin = await auth_manager.create_user(username="apiadmin", role=UserRole.ADMIN)
776    set_current_user(admin)
777
778    # Create user via API
779    user = await auth_manager.create_user_with_api(
780        username="apiuser",
781        password="password123",
782        role="user",
783        display_name="API User",
784    )
785
786    assert user is not None
787    assert user.username == "apiuser"
788    assert user.role == UserRole.USER
789    assert user.display_name == "API User"
790
791
792async def test_create_user_api_validation(auth_manager: AuthenticationManager) -> None:
793    """
794    Test validation in create_user_with_api.
795
796    :param auth_manager: AuthenticationManager instance.
797    """
798    admin = await auth_manager.create_user(username="validadmin", role=UserRole.ADMIN)
799    set_current_user(admin)
800
801    # Test username too short
802    with pytest.raises(InvalidDataError, match="Username must be at least 2 characters"):
803        await auth_manager.create_user_with_api(
804            username="a",
805            password="password123",
806        )
807
808    # Test 2-character username is accepted (minimum allowed)
809    user_2char = await auth_manager.create_user_with_api(
810        username="ab",
811        password="password123",
812    )
813    assert user_2char.username == "ab"
814
815    # Test password too short
816    with pytest.raises(InvalidDataError, match="Password must be at least 8 characters"):
817        await auth_manager.create_user_with_api(
818            username="validuser",
819            password="short",
820        )
821
822
823async def test_logout(auth_manager: AuthenticationManager) -> None:
824    """
825    Test logout functionality.
826
827    :param auth_manager: AuthenticationManager instance.
828    """
829    user = await auth_manager.create_user(username="logoutuser", role=UserRole.USER)
830    token = await auth_manager.create_token(user, "Logout Test", is_long_lived=False)
831
832    # Set current user and token
833    set_current_user(user)
834    set_current_token(token)
835
836    # Token should work before logout
837    authenticated_user = await auth_manager.authenticate_with_token(token)
838    assert authenticated_user is not None
839
840    # Logout
841    await auth_manager.logout()
842
843    # Token should not work after logout
844    authenticated_user = await auth_manager.authenticate_with_token(token)
845    assert authenticated_user is None
846
847
848async def test_token_sliding_expiration(auth_manager: AuthenticationManager) -> None:
849    """
850    Test that short-lived tokens auto-renew on use.
851
852    :param auth_manager: AuthenticationManager instance.
853    """
854    user = await auth_manager.create_user(username="slideuser", role=UserRole.USER)
855    token = await auth_manager.create_token(user, "Slide Test", is_long_lived=False)
856
857    # Get initial expiration
858    token_hash = hashlib.sha256(token.encode()).hexdigest()
859    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
860    assert token_row is not None
861    initial_expires_at = token_row["expires_at"]
862
863    # Use the token (authenticate)
864    authenticated_user = await auth_manager.authenticate_with_token(token)
865    assert authenticated_user is not None
866
867    # Check that expiration was updated
868    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
869    assert token_row is not None
870    updated_expires_at = token_row["expires_at"]
871
872    # Expiration should have been extended
873    assert updated_expires_at != initial_expires_at
874
875
876async def test_long_lived_token_no_auto_renewal(auth_manager: AuthenticationManager) -> None:
877    """
878    Test that long-lived tokens do NOT auto-renew on use.
879
880    :param auth_manager: AuthenticationManager instance.
881    """
882    user = await auth_manager.create_user(username="longuser", role=UserRole.USER)
883    token = await auth_manager.create_token(user, "Long Test", is_long_lived=True)
884
885    # Get initial expiration
886    token_hash = hashlib.sha256(token.encode()).hexdigest()
887    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
888    assert token_row is not None
889    initial_expires_at = token_row["expires_at"]
890
891    # Use the token (authenticate)
892    authenticated_user = await auth_manager.authenticate_with_token(token)
893    assert authenticated_user is not None
894
895    # Check that expiration was NOT updated
896    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
897    assert token_row is not None
898    updated_expires_at = token_row["expires_at"]
899
900    # Expiration should remain the same for long-lived tokens
901    assert updated_expires_at == initial_expires_at
902
903
904async def test_token_activity_write_throttled(
905    auth_manager: AuthenticationManager, monkeypatch: pytest.MonkeyPatch
906) -> None:
907    """
908    Test that rapid authentications persist the token activity only once.
909
910    The HTTP API authenticates on every request; the activity timestamp must not be
911    written to the database again while the stored one is still fresh.
912
913    :param auth_manager: AuthenticationManager instance.
914    :param monkeypatch: Pytest monkeypatch fixture.
915    """
916    user = await auth_manager.create_user(username="throttleuser", role=UserRole.USER)
917    token = await auth_manager.create_token(user, "Throttle Test", is_long_lived=False)
918
919    token_update_count = 0
920    original_update = auth_manager.database.update
921
922    async def counting_update(table: str, match: dict[str, Any], values: dict[str, Any]) -> None:
923        nonlocal token_update_count
924        if table == "auth_tokens":
925            token_update_count += 1
926        await original_update(table, match, values)
927
928    monkeypatch.setattr(auth_manager.database, "update", counting_update)
929
930    # Two rapid authentications: only the first persists the activity timestamp.
931    assert await auth_manager.authenticate_with_token(token) is not None
932    assert await auth_manager.authenticate_with_token(token) is not None
933    assert token_update_count == 1
934
935
936async def test_token_activity_write_resumes_after_interval(
937    auth_manager: AuthenticationManager,
938) -> None:
939    """
940    Test that token activity is persisted again once the stored timestamp is stale.
941
942    :param auth_manager: AuthenticationManager instance.
943    """
944    user = await auth_manager.create_user(username="throttleresume", role=UserRole.USER)
945    token = await auth_manager.create_token(user, "Throttle Resume Test", is_long_lived=False)
946    assert await auth_manager.authenticate_with_token(token) is not None
947
948    # Age the stored activity timestamp (and sliding expiration) past the persist interval.
949    token_hash = hashlib.sha256(token.encode()).hexdigest()
950    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
951    assert token_row is not None
952    stale_time = utc() - TOKEN_ACTIVITY_PERSIST_INTERVAL - timedelta(minutes=5)
953    stale_expires = stale_time + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
954    await auth_manager.database.update(
955        "auth_tokens",
956        {"token_id": token_row["token_id"]},
957        {"last_used_at": stale_time.isoformat(), "expires_at": stale_expires.isoformat()},
958    )
959
960    assert await auth_manager.authenticate_with_token(token) is not None
961
962    # Both the activity timestamp and the sliding expiration must be persisted again.
963    updated_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
964    assert updated_row is not None
965    assert datetime.fromisoformat(updated_row["last_used_at"]) > stale_time
966    assert datetime.fromisoformat(updated_row["expires_at"]) > stale_expires
967
968
969async def test_revoked_token_rejected_within_throttle_window(
970    auth_manager: AuthenticationManager,
971) -> None:
972    """
973    Test that revocation takes effect immediately while the activity write is throttled.
974
975    :param auth_manager: AuthenticationManager instance.
976    """
977    user = await auth_manager.create_user(username="throttlerevoke", role=UserRole.USER)
978    token = await auth_manager.create_token(user, "Throttle Revoke Test", is_long_lived=False)
979    set_current_user(user)
980
981    # First use persists a fresh activity timestamp (entering the throttle window).
982    assert await auth_manager.authenticate_with_token(token) is not None
983
984    token_id = await auth_manager.get_token_id_from_token(token)
985    assert token_id is not None
986    await auth_manager.revoke_token(token_id)
987
988    # The throttle only affects the activity write, never the validation reads.
989    assert await auth_manager.authenticate_with_token(token) is None
990
991
992async def test_expired_token_rejected_despite_fresh_activity(
993    auth_manager: AuthenticationManager,
994) -> None:
995    """
996    Test that expiry validation is not affected by a fresh activity timestamp.
997
998    :param auth_manager: AuthenticationManager instance.
999    """
1000    user = await auth_manager.create_user(username="throttleexpired", role=UserRole.USER)
1001    token = await auth_manager.create_token(user, "Throttle Expired Test", is_long_lived=False)
1002
1003    token_hash = hashlib.sha256(token.encode()).hexdigest()
1004    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1005    assert token_row is not None
1006    await auth_manager.database.update(
1007        "auth_tokens",
1008        {"token_id": token_row["token_id"]},
1009        {
1010            "expires_at": (utc() - timedelta(days=1)).isoformat(),
1011            "last_used_at": utc().isoformat(),
1012        },
1013    )
1014
1015    assert await auth_manager.authenticate_with_token(token) is None
1016
1017
1018async def test_token_absolute_max_enforced_despite_fresh_activity(
1019    auth_manager: AuthenticationManager,
1020) -> None:
1021    """
1022    Test that the absolute lifetime cap is enforced even with a fresh activity timestamp.
1023
1024    :param auth_manager: AuthenticationManager instance.
1025    """
1026    user = await auth_manager.create_user(username="throttleabsmax", role=UserRole.USER)
1027    token = await auth_manager.create_token(user, "Throttle Abs Max Test", is_long_lived=False)
1028
1029    token_hash = hashlib.sha256(token.encode()).hexdigest()
1030    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1031    assert token_row is not None
1032    created_at = utc() - timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION + 1)
1033    future_expires = utc() + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
1034    await auth_manager.database.update(
1035        "auth_tokens",
1036        {"token_id": token_row["token_id"]},
1037        {
1038            "created_at": created_at.isoformat(),
1039            "expires_at": future_expires.isoformat(),
1040            "last_used_at": utc().isoformat(),
1041        },
1042    )
1043
1044    assert await auth_manager.authenticate_with_token(token) is None
1045    assert await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash}) is None
1046
1047
1048async def test_long_lived_token_default_is_one_year() -> None:
1049    """Test that the long-lived token default lifetime is 365 days."""
1050    assert TOKEN_LONG_LIVED_EXPIRATION == 365
1051
1052
1053async def test_token_absolute_max_lifetime(auth_manager: AuthenticationManager) -> None:
1054    """
1055    Test that a short-lived token past its absolute max lifetime cannot be renewed.
1056
1057    The sliding window keeps a session alive on use, but a token created longer than
1058    the absolute maximum ago must be rejected regardless of the sliding expiration.
1059
1060    :param auth_manager: AuthenticationManager instance.
1061    """
1062    user = await auth_manager.create_user(username="absmaxuser", role=UserRole.USER)
1063    token = await auth_manager.create_token(user, "Abs Max Test", is_long_lived=False)
1064
1065    token_hash = hashlib.sha256(token.encode()).hexdigest()
1066    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1067    assert token_row is not None
1068
1069    # Created past the absolute max but with a future sliding expires_at, so only the cap can reject it.
1070    created_at = utc() - timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION + 1)
1071    future_expires = utc() + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
1072    await auth_manager.database.update(
1073        "auth_tokens",
1074        {"token_id": token_row["token_id"]},
1075        {"created_at": created_at.isoformat(), "expires_at": future_expires.isoformat()},
1076    )
1077
1078    # Token must be rejected and the row deleted.
1079    authenticated_user = await auth_manager.authenticate_with_token(token)
1080    assert authenticated_user is None
1081
1082    deleted_row = await auth_manager.database.get_row(
1083        "auth_tokens", {"token_id": token_row["token_id"]}
1084    )
1085    assert deleted_row is None
1086
1087
1088async def test_legacy_token_absolute_max_lifetime(auth_manager: AuthenticationManager) -> None:
1089    """
1090    Test that the absolute max lifetime is also enforced on the legacy hash-token path.
1091
1092    Legacy (non-JWT) tokens authenticate via a hash lookup that shares the same cap logic,
1093    so a hash token created past the absolute maximum must be rejected and its row deleted.
1094
1095    :param auth_manager: AuthenticationManager instance.
1096    """
1097    user = await auth_manager.create_user(username="legacyabsmax", role=UserRole.USER)
1098
1099    # A non-JWT token string forces the legacy hash-based lookup path.
1100    raw_token = "legacy-hash-token-absmax"
1101    token_id = "legacy-absmax-token-id"
1102    # Created past the absolute max but with a future sliding expires_at, so only the cap can reject it.
1103    created_at = utc() - timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION + 1)
1104    future_expires = utc() + timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
1105    await auth_manager.database.insert(
1106        "auth_tokens",
1107        {
1108            "token_id": token_id,
1109            "user_id": user.user_id,
1110            "token_hash": hashlib.sha256(raw_token.encode()).hexdigest(),
1111            "name": "Legacy Abs Max Test",
1112            "created_at": created_at.isoformat(),
1113            "expires_at": future_expires.isoformat(),
1114            "is_long_lived": 0,
1115        },
1116    )
1117
1118    # Token must be rejected and the row deleted.
1119    assert await auth_manager.authenticate_with_token(raw_token) is None
1120    assert await auth_manager.database.get_row("auth_tokens", {"token_id": token_id}) is None
1121
1122
1123async def test_revoke_tokens_for_user_persists(auth_manager: AuthenticationManager) -> None:
1124    """
1125    Test that revoke_tokens_for_user commits so the tokens no longer authenticate.
1126
1127    :param auth_manager: AuthenticationManager instance.
1128    """
1129    user = await auth_manager.create_user(username="guestrevoke", role=UserRole.GUEST)
1130    token = await auth_manager.create_token(user, "Guest Token", is_long_lived=False)
1131
1132    # Token works before revocation
1133    assert await auth_manager.authenticate_with_token(token) is not None
1134
1135    revoked = await auth_manager.revoke_tokens_for_user(user)
1136    assert revoked == 1
1137
1138    # Reopen the raw connection to roll back any uncommitted tx: an uncommitted DELETE would resurrect the row.
1139    await auth_manager.database._db.close()
1140    await auth_manager.database.setup()
1141
1142    token_hash = hashlib.sha256(token.encode()).hexdigest()
1143    assert await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash}) is None
1144    assert await auth_manager.authenticate_with_token(token) is None
1145
1146
1147async def test_short_lived_jwt_exp_carries_absolute_max(
1148    auth_manager: AuthenticationManager,
1149) -> None:
1150    """
1151    Test that a short-lived JWT's exp claim equals the absolute max lifetime.
1152
1153    The database expires_at enforces the sliding idle window; an exp claim shorter
1154    than the absolute max would cut off active sessions before renewal can happen.
1155
1156    :param auth_manager: AuthenticationManager instance.
1157    """
1158    user = await auth_manager.create_user(username="jwtexpuser", role=UserRole.USER)
1159    token = await auth_manager.create_token(user, "JWT Exp Test", is_long_lived=False)
1160
1161    token_hash = hashlib.sha256(token.encode()).hexdigest()
1162    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1163    assert token_row is not None
1164    created_at = datetime.fromisoformat(token_row["created_at"])
1165
1166    payload = auth_manager.jwt_helper.decode_token(token, verify_exp=False)
1167    expected = created_at + timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION)
1168    assert payload["exp"] == int(expected.timestamp())
1169
1170    # The database keeps the shorter sliding window as source of truth
1171    expires_at = datetime.fromisoformat(token_row["expires_at"])
1172    assert expires_at - created_at == timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
1173
1174
1175async def test_guest_token_fixed_short_lifetime(auth_manager: AuthenticationManager) -> None:
1176    """
1177    Test that guest tokens get a short fixed lifetime and never renew on use.
1178
1179    :param auth_manager: AuthenticationManager instance.
1180    """
1181    user = await auth_manager.create_user(username="guestexpiry", role=UserRole.GUEST)
1182    token = await auth_manager.create_token(user, "Guest Session", is_long_lived=False)
1183
1184    token_hash = hashlib.sha256(token.encode()).hexdigest()
1185    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1186    assert token_row is not None
1187    created_at = datetime.fromisoformat(token_row["created_at"])
1188    expires_at = datetime.fromisoformat(token_row["expires_at"])
1189    assert expires_at - created_at == timedelta(days=TOKEN_GUEST_EXPIRATION)
1190
1191    # The JWT exp claim must match the fixed window, not the absolute max
1192    payload = auth_manager.jwt_helper.decode_token(token, verify_exp=False)
1193    assert payload["exp"] == int(expires_at.timestamp())
1194
1195    # Authenticating must not extend the expiration (no sliding window for guests)
1196    assert await auth_manager.authenticate_with_token(token) is not None
1197    updated_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1198    assert updated_row is not None
1199    assert updated_row["expires_at"] == token_row["expires_at"]
1200
1201
1202async def test_guest_cannot_create_long_lived_token(auth_manager: AuthenticationManager) -> None:
1203    """
1204    Test that a guest cannot create a long-lived token for their own account.
1205
1206    :param auth_manager: AuthenticationManager instance.
1207    """
1208    guest = await auth_manager.create_user(username="guesttoken", role=UserRole.GUEST)
1209    set_current_user(guest)
1210
1211    with pytest.raises(InsufficientPermissions):
1212        await auth_manager.create_long_lived_token("Guest Escalation")
1213
1214
1215async def test_no_long_lived_token_for_guest_account(auth_manager: AuthenticationManager) -> None:
1216    """
1217    Test that a long-lived token cannot be created for a guest account, even by an admin.
1218
1219    :param auth_manager: AuthenticationManager instance.
1220    """
1221    admin = await auth_manager.create_user(username="tokenadmin", role=UserRole.ADMIN)
1222    guest = await auth_manager.create_user(username="guesttarget", role=UserRole.GUEST)
1223    set_current_user(admin)
1224
1225    with pytest.raises(InsufficientPermissions):
1226        await auth_manager.create_long_lived_token("Guest Token", user_id=guest.user_id)
1227
1228
1229async def test_username_case_insensitive_creation(auth_manager: AuthenticationManager) -> None:
1230    """
1231    Test that usernames are normalized to lowercase on creation.
1232
1233    :param auth_manager: AuthenticationManager instance.
1234    """
1235    # Create user with mixed case username
1236    user = await auth_manager.create_user(
1237        username="TestUser",
1238        role=UserRole.USER,
1239        display_name="Test User",
1240    )
1241
1242    # Username should be stored in lowercase
1243    assert user.username == "testuser"
1244
1245
1246async def test_username_case_insensitive_duplicate_prevention(
1247    auth_manager: AuthenticationManager,
1248) -> None:
1249    """
1250    Test that duplicate usernames with different cases are prevented.
1251
1252    :param auth_manager: AuthenticationManager instance.
1253    """
1254    # Create user with lowercase username
1255    await auth_manager.create_user(username="admin", role=UserRole.USER)
1256
1257    # Try to create user with same username but different case should fail
1258    # (SQLite UNIQUE constraint violation)
1259    with pytest.raises(IntegrityError, match="UNIQUE constraint failed"):
1260        await auth_manager.create_user(username="Admin", role=UserRole.USER)
1261
1262
1263async def test_username_case_insensitive_login(auth_manager: AuthenticationManager) -> None:
1264    """
1265    Test that login works with any case variation of username.
1266
1267    :param auth_manager: AuthenticationManager instance.
1268    """
1269    builtin_provider = auth_manager.login_providers.get("builtin")
1270    assert builtin_provider is not None
1271    assert isinstance(builtin_provider, BuiltinLoginProvider)
1272
1273    # Create user with lowercase username
1274    await builtin_provider.create_user_with_password(
1275        username="testadmin",
1276        password="SecurePassword123",
1277        role=UserRole.ADMIN,
1278    )
1279
1280    # Test login with lowercase
1281    result = await auth_manager.authenticate_with_credentials(
1282        "builtin",
1283        {"username": "testadmin", "password": "SecurePassword123"},
1284    )
1285    assert result.success is True
1286    assert result.user is not None
1287    assert result.user.username == "testadmin"
1288
1289    # Test login with uppercase
1290    result = await auth_manager.authenticate_with_credentials(
1291        "builtin",
1292        {"username": "TESTADMIN", "password": "SecurePassword123"},
1293    )
1294    assert result.success is True
1295    assert result.user is not None
1296    assert result.user.username == "testadmin"
1297
1298    # Test login with mixed case
1299    result = await auth_manager.authenticate_with_credentials(
1300        "builtin",
1301        {"username": "TestAdmin", "password": "SecurePassword123"},
1302    )
1303    assert result.success is True
1304    assert result.user is not None
1305    assert result.user.username == "testadmin"
1306
1307
1308async def test_username_case_insensitive_lookup(auth_manager: AuthenticationManager) -> None:
1309    """
1310    Test that user lookup by username is case-insensitive.
1311
1312    :param auth_manager: AuthenticationManager instance.
1313    """
1314    # Create user with lowercase username
1315    created_user = await auth_manager.create_user(username="lookupuser", role=UserRole.USER)
1316
1317    # Lookup with lowercase
1318    user1 = await auth_manager.get_user_by_username("lookupuser")
1319    assert user1 is not None
1320    assert user1.user_id == created_user.user_id
1321
1322    # Lookup with uppercase
1323    user2 = await auth_manager.get_user_by_username("LOOKUPUSER")
1324    assert user2 is not None
1325    assert user2.user_id == created_user.user_id
1326
1327    # Lookup with mixed case
1328    user3 = await auth_manager.get_user_by_username("LookUpUser")
1329    assert user3 is not None
1330    assert user3.user_id == created_user.user_id
1331
1332
1333async def test_username_update_normalizes(auth_manager: AuthenticationManager) -> None:
1334    """
1335    Test that updating username normalizes it to lowercase.
1336
1337    :param auth_manager: AuthenticationManager instance.
1338    """
1339    user = await auth_manager.create_user(username="originaluser", role=UserRole.USER)
1340
1341    # Update username with mixed case
1342    updated_user = await auth_manager.update_user(user, username="UpdatedUser")
1343
1344    # Username should be normalized to lowercase
1345    assert updated_user is not None
1346    assert updated_user.username == "updateduser"
1347
1348
1349async def test_link_user_to_provider_idempotent(auth_manager: AuthenticationManager) -> None:
1350    """
1351    Test that linking user to provider is idempotent.
1352
1353    This tests the fix for the bug where re-linking a user would cause
1354    IntegrityError due to UNIQUE constraint on (provider_type, provider_user_id).
1355
1356    :param auth_manager: AuthenticationManager instance.
1357    """
1358    user = await auth_manager.create_user(username="hauser", role=UserRole.USER)
1359
1360    # Link user to Home Assistant provider for the first time
1361    link1 = await auth_manager.link_user_to_provider(
1362        user,
1363        AuthProviderType.HOME_ASSISTANT,
1364        "ha_user_456",
1365    )
1366
1367    assert link1 is not None
1368    assert link1.user_id == user.user_id
1369    assert link1.provider_type == AuthProviderType.HOME_ASSISTANT
1370    assert link1.provider_user_id == "ha_user_456"
1371
1372    # Linking the same user again should return existing link without error
1373    link2 = await auth_manager.link_user_to_provider(
1374        user,
1375        AuthProviderType.HOME_ASSISTANT,
1376        "ha_user_456",
1377    )
1378
1379    assert link2 is not None
1380    assert link2.link_id == link1.link_id  # Should be same link
1381    assert link2.user_id == user.user_id
1382    assert link2.provider_type == AuthProviderType.HOME_ASSISTANT
1383    assert link2.provider_user_id == "ha_user_456"
1384
1385
1386async def test_ingress_auth_existing_username(auth_manager: AuthenticationManager) -> None:
1387    """
1388    Test HA ingress auth when username exists but isn't linked to HA provider.
1389
1390    This tests the scenario where a user is created during setup, and then
1391    tries to login via HA ingress with the same username.
1392
1393    :param auth_manager: AuthenticationManager instance.
1394    """
1395    # Simulate user created during initial setup
1396    existing_user = await auth_manager.create_user(
1397        username="admin",
1398        role=UserRole.ADMIN,
1399        display_name="Admin User",
1400    )
1401
1402    # Now simulate HA ingress trying to auto-create a user with same username
1403    # This should find the existing user and link it instead of creating new one
1404    user = await auth_manager.get_user_by_username("admin")
1405    assert user is not None
1406    assert user.user_id == existing_user.user_id
1407
1408    # Link the existing user to HA provider (what ingress flow would do)
1409    link = await auth_manager.link_user_to_provider(
1410        user,
1411        AuthProviderType.HOME_ASSISTANT,
1412        "ha_admin_123",
1413    )
1414
1415    assert link is not None
1416    assert link.user_id == existing_user.user_id
1417
1418    # Verify we can retrieve user by provider link
1419    retrieved_user = await auth_manager.get_user_by_provider_link(
1420        AuthProviderType.HOME_ASSISTANT,
1421        "ha_admin_123",
1422    )
1423
1424    assert retrieved_user is not None
1425    assert retrieved_user.user_id == existing_user.user_id
1426    assert retrieved_user.username == "admin"
1427
1428
1429# ==================== Join Code Tests ====================
1430
1431
1432async def test_generate_join_code(auth_manager: AuthenticationManager) -> None:
1433    """
1434    Test generating a join code for a user.
1435
1436    :param auth_manager: AuthenticationManager instance.
1437    """
1438    user = await auth_manager.create_user(username="joincodeuser", role=UserRole.GUEST)
1439
1440    code, expires_at = await auth_manager.generate_join_code(
1441        user=user,
1442        expires_in_hours=24,
1443        max_uses=0,
1444        device_name="Test Device",
1445    )
1446
1447    assert code is not None
1448    assert len(code) == JOIN_CODE_LENGTH
1449    assert code.isalnum()
1450    assert expires_at is not None
1451    assert expires_at > utc()
1452
1453
1454async def test_get_join_code_expiry(auth_manager: AuthenticationManager) -> None:
1455    """
1456    Test looking up the expiry for a specific active join code.
1457
1458    :param auth_manager: AuthenticationManager instance.
1459    """
1460    user = await auth_manager.create_user(username="joinexpiryuser", role=UserRole.GUEST)
1461
1462    code, expires_at = await auth_manager.generate_join_code(
1463        user=user,
1464        expires_in_hours=24,
1465    )
1466
1467    assert await auth_manager.get_join_code_expiry(code, user) == expires_at
1468    assert await auth_manager.get_join_code_expiry(code.lower(), user) == expires_at
1469    assert await auth_manager.get_join_code_expiry("BADCODE", user) is None
1470
1471
1472async def test_get_join_code_expiry_requires_matching_user(
1473    auth_manager: AuthenticationManager,
1474) -> None:
1475    """
1476    Test that join code expiry lookup can be scoped to a specific user.
1477
1478    :param auth_manager: AuthenticationManager instance.
1479    """
1480    user = await auth_manager.create_user(username="joinexpiryowner", role=UserRole.GUEST)
1481    other_user = await auth_manager.create_user(
1482        username="joinexpiryother",
1483        role=UserRole.GUEST,
1484    )
1485
1486    code, expires_at = await auth_manager.generate_join_code(
1487        user=user,
1488        expires_in_hours=24,
1489    )
1490
1491    assert await auth_manager.get_join_code_expiry(code, user) == expires_at
1492    assert await auth_manager.get_join_code_expiry(code) == expires_at
1493    assert await auth_manager.get_join_code_expiry(code, other_user) is None
1494
1495
1496async def test_get_join_code_expiry_expired(auth_manager: AuthenticationManager) -> None:
1497    """
1498    Test that expired join codes have no active expiry.
1499
1500    :param auth_manager: AuthenticationManager instance.
1501    """
1502    user = await auth_manager.create_user(username="joinexpiryexpired", role=UserRole.GUEST)
1503
1504    code, _ = await auth_manager.generate_join_code(
1505        user=user,
1506        expires_in_hours=24,
1507    )
1508    code_row = await auth_manager.database.get_row("join_codes", {"code": code})
1509    assert code_row is not None
1510
1511    past_time = utc() - timedelta(hours=1)
1512    await auth_manager.database.update(
1513        "join_codes",
1514        {"code_id": code_row["code_id"]},
1515        {"expires_at": past_time.isoformat()},
1516    )
1517
1518    assert await auth_manager.get_join_code_expiry(code, user) is None
1519
1520
1521async def test_generate_join_code_non_guest_rejected(
1522    auth_manager: AuthenticationManager,
1523) -> None:
1524    """
1525    Test that generating a join code for non-guest users is rejected.
1526
1527    :param auth_manager: AuthenticationManager instance.
1528    """
1529    admin = await auth_manager.create_user(username="joinadmin", role=UserRole.ADMIN)
1530    user = await auth_manager.create_user(username="joinuser", role=UserRole.USER)
1531
1532    with pytest.raises(ValueError, match="guest accounts"):
1533        await auth_manager.generate_join_code(user=admin)
1534
1535    with pytest.raises(ValueError, match="guest accounts"):
1536        await auth_manager.generate_join_code(user=user)
1537
1538
1539async def test_exchange_join_code(auth_manager: AuthenticationManager) -> None:
1540    """
1541    Test exchanging a valid join code for a JWT token.
1542
1543    :param auth_manager: AuthenticationManager instance.
1544    """
1545    user = await auth_manager.create_user(username="exchangeuser", role=UserRole.GUEST)
1546
1547    code, _ = await auth_manager.generate_join_code(
1548        user=user,
1549        expires_in_hours=24,
1550        device_name="Exchange Test",
1551    )
1552
1553    # Exchange code for token
1554    token = await auth_manager._exchange_join_code(code)
1555
1556    assert token is not None
1557    assert len(token) > 0
1558
1559    # Verify token works for authentication
1560    authenticated_user = await auth_manager.authenticate_with_token(token)
1561    assert authenticated_user is not None
1562    assert authenticated_user.user_id == user.user_id
1563    assert authenticated_user.username == user.username
1564
1565
1566async def test_exchange_join_code_case_insensitive(auth_manager: AuthenticationManager) -> None:
1567    """
1568    Test that join codes are case-insensitive.
1569
1570    :param auth_manager: AuthenticationManager instance.
1571    """
1572    user = await auth_manager.create_user(username="caseuser", role=UserRole.GUEST)
1573
1574    code, _ = await auth_manager.generate_join_code(
1575        user=user,
1576        expires_in_hours=24,
1577    )
1578
1579    # Exchange with lowercase version
1580    token = await auth_manager._exchange_join_code(code.lower())
1581    assert token is not None
1582
1583    # Verify token works
1584    authenticated_user = await auth_manager.authenticate_with_token(token)
1585    assert authenticated_user is not None
1586    assert authenticated_user.user_id == user.user_id
1587
1588
1589async def test_exchange_join_code_invalid(auth_manager: AuthenticationManager) -> None:
1590    """
1591    Test that invalid join codes are rejected.
1592
1593    :param auth_manager: AuthenticationManager instance.
1594    """
1595    token = await auth_manager._exchange_join_code("INVALID")
1596    assert token is None
1597
1598
1599async def test_exchange_join_code_expired(auth_manager: AuthenticationManager) -> None:
1600    """
1601    Test that expired join codes are rejected.
1602
1603    :param auth_manager: AuthenticationManager instance.
1604    """
1605    user = await auth_manager.create_user(username="expiredcodeuser", role=UserRole.GUEST)
1606
1607    code, _ = await auth_manager.generate_join_code(
1608        user=user,
1609        expires_in_hours=24,
1610    )
1611
1612    # Manually expire the code by updating expires_at in database
1613    code_row = await auth_manager.database.get_row("join_codes", {"code": code})
1614    assert code_row is not None
1615
1616    past_time = utc() - timedelta(hours=1)
1617    await auth_manager.database.update(
1618        "join_codes",
1619        {"code_id": code_row["code_id"]},
1620        {"expires_at": past_time.isoformat()},
1621    )
1622
1623    # Try to exchange expired code
1624    token = await auth_manager._exchange_join_code(code)
1625    assert token is None
1626
1627
1628async def test_exchange_join_code_max_uses(auth_manager: AuthenticationManager) -> None:
1629    """
1630    Test that join codes respect max_uses limit.
1631
1632    :param auth_manager: AuthenticationManager instance.
1633    """
1634    user = await auth_manager.create_user(username="maxusesuser", role=UserRole.GUEST)
1635
1636    code, _ = await auth_manager.generate_join_code(
1637        user=user,
1638        expires_in_hours=24,
1639        max_uses=2,  # Only allow 2 uses
1640    )
1641
1642    # First use should succeed
1643    token1 = await auth_manager._exchange_join_code(code)
1644    assert token1 is not None
1645
1646    # Second use should succeed
1647    token2 = await auth_manager._exchange_join_code(code)
1648    assert token2 is not None
1649
1650    # Third use should fail (max_uses=2 exceeded)
1651    token3 = await auth_manager._exchange_join_code(code)
1652    assert token3 is None
1653
1654
1655async def test_exchange_join_code_unlimited_uses(auth_manager: AuthenticationManager) -> None:
1656    """
1657    Test that join codes with max_uses=0 have unlimited uses.
1658
1659    :param auth_manager: AuthenticationManager instance.
1660    """
1661    user = await auth_manager.create_user(username="unlimiteduser", role=UserRole.GUEST)
1662
1663    code, _ = await auth_manager.generate_join_code(
1664        user=user,
1665        expires_in_hours=24,
1666        max_uses=0,  # Unlimited
1667    )
1668
1669    # Should be able to use multiple times
1670    for _ in range(5):
1671        token = await auth_manager._exchange_join_code(code)
1672        assert token is not None
1673
1674
1675async def test_revoke_join_codes_for_user(auth_manager: AuthenticationManager) -> None:
1676    """
1677    Test revoking join codes for a specific user.
1678
1679    :param auth_manager: AuthenticationManager instance.
1680    """
1681    user1 = await auth_manager.create_user(username="revokeuser1", role=UserRole.GUEST)
1682    user2 = await auth_manager.create_user(username="revokeuser2", role=UserRole.GUEST)
1683
1684    # Create codes for both users
1685    code1, _ = await auth_manager.generate_join_code(user=user1)
1686    code2, _ = await auth_manager.generate_join_code(user=user2)
1687
1688    # Revoke codes for user1 only
1689    revoked_count = await auth_manager.revoke_join_codes(user1)
1690    assert revoked_count == 1
1691
1692    # User1's code should no longer work
1693    token1 = await auth_manager._exchange_join_code(code1)
1694    assert token1 is None
1695
1696    # User2's code should still work
1697    token2 = await auth_manager._exchange_join_code(code2)
1698    assert token2 is not None
1699
1700
1701async def test_authenticate_with_join_code_api(auth_manager: AuthenticationManager) -> None:
1702    """
1703    Test the public API endpoint for join code authentication.
1704
1705    :param auth_manager: AuthenticationManager instance.
1706    """
1707    user = await auth_manager.create_user(
1708        username="apijoincodeuser",
1709        role=UserRole.GUEST,
1710        display_name="API Guest",
1711    )
1712
1713    code, _ = await auth_manager.generate_join_code(
1714        user=user,
1715        expires_in_hours=24,
1716    )
1717
1718    # Call the API endpoint
1719    result = await auth_manager.exchange_join_code(code)
1720
1721    assert result["success"] is True
1722    assert "access_token" in result
1723    assert result["user"]["user_id"] == user.user_id
1724    assert result["user"]["username"] == user.username
1725    assert result["user"]["role"] == "guest"
1726
1727
1728async def test_authenticate_with_join_code_api_invalid(
1729    auth_manager: AuthenticationManager,
1730) -> None:
1731    """
1732    Test the API endpoint with invalid join code.
1733
1734    :param auth_manager: AuthenticationManager instance.
1735    """
1736    result = await auth_manager.exchange_join_code("BADCODE")
1737
1738    assert result["success"] is False
1739    assert "error" in result
1740    assert "access_token" not in result
1741
1742
1743async def test_list_join_codes(auth_manager: AuthenticationManager) -> None:
1744    """
1745    Test listing active join codes (admin only).
1746
1747    :param auth_manager: AuthenticationManager instance.
1748    """
1749    admin = await auth_manager.create_user(username="listcodesadmin", role=UserRole.ADMIN)
1750    guest1 = await auth_manager.create_user(username="listguest1", role=UserRole.GUEST)
1751    guest2 = await auth_manager.create_user(username="listguest2", role=UserRole.GUEST)
1752    set_current_user(admin)
1753
1754    # Create codes for both guests
1755    await auth_manager.generate_join_code(user=guest1)
1756    await auth_manager.generate_join_code(user=guest2)
1757
1758    # List all codes
1759    codes = await auth_manager.list_join_codes()
1760    assert len(codes) == 2
1761
1762    # List codes for specific user
1763    codes = await auth_manager.list_join_codes(user_id=guest1.user_id)
1764    assert len(codes) == 1
1765    assert codes[0]["user_id"] == guest1.user_id
1766
1767
1768async def test_revoke_join_code_api(auth_manager: AuthenticationManager) -> None:
1769    """
1770    Test revoking a specific join code by code_id (admin only).
1771
1772    :param auth_manager: AuthenticationManager instance.
1773    """
1774    admin = await auth_manager.create_user(username="revokecodeadmin", role=UserRole.ADMIN)
1775    guest = await auth_manager.create_user(username="revokeguest", role=UserRole.GUEST)
1776    set_current_user(admin)
1777
1778    code, _ = await auth_manager.generate_join_code(user=guest)
1779
1780    # Get the code_id from the database
1781    codes = await auth_manager.list_join_codes(user_id=guest.user_id)
1782    assert len(codes) == 1
1783    code_id = codes[0]["code_id"]
1784
1785    # Revoke the specific code
1786    await auth_manager.revoke_join_code(code_id)
1787
1788    # Code should no longer work
1789    token = await auth_manager._exchange_join_code(code)
1790    assert token is None
1791
1792    # List should be empty
1793    codes = await auth_manager.list_join_codes(user_id=guest.user_id)
1794    assert len(codes) == 0
1795
1796
1797async def test_revoke_join_code_api_not_found(auth_manager: AuthenticationManager) -> None:
1798    """
1799    Test revoking a non-existent join code raises error.
1800
1801    :param auth_manager: AuthenticationManager instance.
1802    """
1803    admin = await auth_manager.create_user(username="revokenotfound", role=UserRole.ADMIN)
1804    set_current_user(admin)
1805
1806    with pytest.raises(InvalidDataError, match="Join code not found"):
1807        await auth_manager.revoke_join_code("nonexistent-code-id")
1808
1809
1810async def test_impersonated_user_context_manager(auth_manager: AuthenticationManager) -> None:
1811    """Test the ImpersonatedUser context manager."""
1812    admin_user = await auth_manager.create_user(username="admin", role=UserRole.ADMIN)
1813    standard_user_a = await auth_manager.create_user(username="user_a", role=UserRole.USER)
1814    standard_user_b = await auth_manager.create_user(username="user_b", role=UserRole.USER)
1815    service_user = await auth_manager.create_user(username="service", role=UserRole.SERVICE)
1816
1817    # non-authenticated user must raise
1818    set_current_user(None)
1819    with pytest.raises(InsufficientPermissions):
1820        async with ImpersonatedUser(auth_manager.mass, "user_a"):
1821            ...
1822    # impersonation attempt without the users.impersonate scope must raise
1823    set_current_user(standard_user_a)
1824    with pytest.raises(InsufficientPermissions):
1825        async with ImpersonatedUser(auth_manager.mass, "admin"):
1826            ...
1827    # invalid username must raise
1828    set_current_user(admin_user)
1829    with pytest.raises(InvalidDataError):
1830        async with ImpersonatedUser(auth_manager.mass, "wrong_username"):
1831            ...
1832
1833    # verify that a standard user may impersonate itself (by username or user_id)
1834    set_current_user(standard_user_a)
1835    set_impersonated_user(None)
1836    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1837        assert get_current_user() == standard_user_a
1838    async with ImpersonatedUser(auth_manager.mass, standard_user_a.user_id):
1839        assert get_current_user() == standard_user_a
1840    # passing None is a no-op which preserves any active impersonation
1841    set_impersonated_user(standard_user_b)
1842    async with ImpersonatedUser(auth_manager.mass, None):
1843        assert get_current_user() == standard_user_b
1844    assert get_current_user() == standard_user_b
1845
1846    # verify that an admin user may impersonate another user
1847    set_current_user(admin_user)
1848
1849    set_impersonated_user(None)  # non-nested use
1850    assert get_current_user() == admin_user
1851    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1852        assert get_current_user() == standard_user_a
1853    assert get_current_user() == admin_user
1854
1855    set_impersonated_user(standard_user_b)  # nested use
1856    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1857        assert get_current_user() == standard_user_a
1858    assert get_current_user() == standard_user_b
1859
1860    # verify that a service user may impersonate another user (users.impersonate scope)
1861    set_current_user(service_user)
1862    set_impersonated_user(None)
1863    assert has_scope(service_user, Scope.USERS_IMPERSONATE)
1864    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1865        assert get_current_user() == standard_user_a
1866    assert get_current_user() == service_user
1867
1868
1869async def test_impersonated_user_anonymous_playback_is_noop(
1870    auth_manager: AuthenticationManager,
1871) -> None:
1872    """
1873    Verify an unauthenticated call without a username is a no-op.
1874
1875    Regression: play_media wraps every call in ImpersonatedUser, so protocol/hardware
1876    triggered playback (presets, Spotify Connect, ...) - which has no authenticated user
1877    and passes no username - must not raise.
1878    """
1879    set_current_user(None)
1880    set_impersonated_user(None)
1881    async with ImpersonatedUser(auth_manager.mass, None):
1882        assert get_current_user() is None
1883    assert get_current_user() is None
1884
1885    # an unauthenticated caller may still not impersonate another user
1886    with pytest.raises(InsufficientPermissions):
1887        async with ImpersonatedUser(auth_manager.mass, "user_a"):
1888            ...
1889
1890
1891async def test_join_code_length_at_least_12() -> None:
1892    """Verify join codes are long enough to resist brute force (security finding 7.3.2)."""
1893    assert JOIN_CODE_LENGTH >= 12
1894
1895
1896async def test_exchange_join_code_rate_limited(auth_manager: AuthenticationManager) -> None:
1897    """
1898    Verify repeated failed join code exchanges get throttled (security finding 7.3.2).
1899
1900    :param auth_manager: AuthenticationManager instance.
1901    """
1902    # Three failures trip the progressive delay threshold.
1903    for _ in range(3):
1904        result = await auth_manager.exchange_join_code("WRONGCODE123")
1905        assert result["success"] is False
1906
1907    # The next attempt must be rejected for rate limiting, not just "invalid".
1908    result = await auth_manager.exchange_join_code("WRONGCODE123")
1909    assert result["success"] is False
1910    assert "too many" in result["error"].lower()
1911
1912
1913async def test_exchange_join_code_rate_limit_concurrent_burst(
1914    auth_manager: AuthenticationManager,
1915) -> None:
1916    """
1917    Verify concurrent failed exchanges cannot race past the rate limiter.
1918
1919    Without serialization, parallel requests all pass the rate limit check
1920    before any of them records a failure, allowing brute-force bursts.
1921
1922    :param auth_manager: AuthenticationManager instance.
1923    """
1924    results = await asyncio.gather(
1925        *(auth_manager.exchange_join_code("WRONGCODE123") for _ in range(10))
1926    )
1927
1928    assert all(result["success"] is False for result in results)
1929    # Only the first 3 attempts may reach the actual code check; the rest must be throttled.
1930    invalid_count = sum(1 for result in results if "invalid" in result["error"].lower())
1931    throttled_count = sum(1 for result in results if "too many" in result["error"].lower())
1932    assert invalid_count == 3
1933    assert throttled_count == 7
1934
1935
1936async def test_exchange_join_code_success_does_not_reset_rate_limit(
1937    auth_manager: AuthenticationManager,
1938) -> None:
1939    """
1940    Verify a successful exchange does not clear the shared failed-attempt counter.
1941
1942    Callers without a connection identity share one bucket, so clearing it on success
1943    would let an attacker holding any valid code reset the counter for everyone.
1944
1945    :param auth_manager: AuthenticationManager instance.
1946    """
1947    user = await auth_manager.create_user(username="norstuser", role=UserRole.GUEST)
1948    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
1949
1950    # A couple of failures, still below the throttle threshold.
1951    for _ in range(2):
1952        assert (await auth_manager.exchange_join_code("NOPE12345678"))["success"] is False
1953
1954    # A valid exchange still succeeds but must not wipe the counter.
1955    assert (await auth_manager.exchange_join_code(code))["success"] is True
1956
1957    # The third failure trips the threshold, throttling further attempts.
1958    assert (await auth_manager.exchange_join_code("NOPE12345678"))["success"] is False
1959    result = await auth_manager.exchange_join_code(code)
1960    assert result["success"] is False
1961    assert "too many" in result["error"].lower()
1962
1963
1964async def test_exchange_join_code_rate_limit_is_per_connection(
1965    auth_manager: AuthenticationManager,
1966) -> None:
1967    """
1968    Verify one throttled client does not lock out the other clients.
1969
1970    At a party every guest exchanges a join code over their own connection, so a guest
1971    re-scanning an expired QR code must only ever throttle their own connection.
1972
1973    :param auth_manager: AuthenticationManager instance.
1974    """
1975    user = await auth_manager.create_user(username="partyguest", role=UserRole.GUEST)
1976    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
1977
1978    # One guest burns through the progressive delay threshold with a stale code.
1979    set_current_client_id("connection-a")
1980    for _ in range(3):
1981        assert (await auth_manager.exchange_join_code("EXPIRED12345"))["success"] is False
1982    result = await auth_manager.exchange_join_code("EXPIRED12345")
1983    assert "too many" in result["error"].lower()
1984
1985    # A second guest is unaffected, both for a bad code and for a valid one.
1986    set_current_client_id("connection-b")
1987    result = await auth_manager.exchange_join_code("EXPIRED12345")
1988    assert "invalid" in result["error"].lower()
1989    assert (await auth_manager.exchange_join_code(code))["success"] is True
1990
1991
1992async def test_exchange_join_code_success_clears_connection_rate_limit(
1993    auth_manager: AuthenticationManager,
1994) -> None:
1995    """
1996    Verify a successful exchange clears the counter of that connection only.
1997
1998    A per-connection counter can only be cleared by the connection that filled it, so a
1999    valid code holder cannot lift the throttle for anyone else.
2000
2001    :param auth_manager: AuthenticationManager instance.
2002    """
2003    user = await auth_manager.create_user(username="clearingguest", role=UserRole.GUEST)
2004    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
2005
2006    set_current_client_id("connection-c")
2007    for _ in range(2):
2008        assert (await auth_manager.exchange_join_code("MISTYPED1234"))["success"] is False
2009
2010    assert (await auth_manager.exchange_join_code(code))["success"] is True
2011
2012    # Without the reset, the fourth failure overall would trip the threshold.
2013    for _ in range(2):
2014        result = await auth_manager.exchange_join_code("MISTYPED1234")
2015        assert "invalid" in result["error"].lower()
2016
2017    # The shared bucket of connection-less callers is untouched by all of this.
2018    set_current_client_id(None)
2019    result = await auth_manager.exchange_join_code("MISTYPED1234")
2020    assert "invalid" in result["error"].lower()
2021
2022
2023async def test_exchange_join_code_global_ceiling_throttles_every_client(
2024    auth_manager: AuthenticationManager,
2025) -> None:
2026    """
2027    Verify the server-wide ceiling still backstops the per-connection counters.
2028
2029    A client can open a fresh connection (and thus a fresh counter) at will, so the
2030    ceiling is what bounds sustained abuse.
2031
2032    :param auth_manager: AuthenticationManager instance.
2033    """
2034    user = await auth_manager.create_user(username="ceilingguest", role=UserRole.GUEST)
2035    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
2036
2037    for _ in range(JOIN_CODE_GLOBAL_FAILURE_CEILING):
2038        await auth_manager._join_code_global_rate_limiter.record_failed_attempt(
2039            JOIN_CODE_GLOBAL_RATE_LIMIT_KEY
2040        )
2041
2042    # An untouched connection is throttled, even when it presents a valid code.
2043    set_current_client_id("connection-never-seen-before")
2044    result = await auth_manager.exchange_join_code(code)
2045    assert result["success"] is False
2046    assert "too many" in result["error"].lower()
2047
2048
2049async def test_exchange_join_code_rate_limit_falls_back_to_peer_address(
2050    auth_manager: AuthenticationManager,
2051) -> None:
2052    """
2053    Verify stateless API callers are told apart by the address they connect from.
2054
2055    The login page exchanges join codes over the JSON RPC endpoint, which carries no
2056    connection identity, so without this every such caller would share one bucket.
2057
2058    :param auth_manager: AuthenticationManager instance.
2059    """
2060    user = await auth_manager.create_user(username="httpguest", role=UserRole.GUEST)
2061    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
2062
2063    set_current_peer_address("192.0.2.10")
2064    for _ in range(3):
2065        assert (await auth_manager.exchange_join_code("EXPIRED12345"))["success"] is False
2066    assert "too many" in (await auth_manager.exchange_join_code("EXPIRED12345"))["error"].lower()
2067
2068    # A caller from another address is unaffected.
2069    set_current_peer_address("192.0.2.11")
2070    assert (await auth_manager.exchange_join_code(code))["success"] is True
2071
2072    # An address can be shared by everyone behind a proxy, so success must not clear it.
2073    set_current_peer_address("192.0.2.10")
2074    assert "too many" in (await auth_manager.exchange_join_code(code))["error"].lower()
2075
2076    # A websocket client id always wins over the peer address.
2077    set_current_client_id("connection-e")
2078    assert (await auth_manager.exchange_join_code(code))["success"] is True
2079
2080
2081def test_mask_join_code() -> None:
2082    """Verify the log mask keeps a correlatable prefix but no usable code."""
2083    assert _mask_join_code("abcdefghjklm") == "ABCD********"
2084    assert _mask_join_code("abc") == "ABC"
2085
2086
2087async def test_exchange_join_code_logs_identify_the_caller(
2088    auth_manager: AuthenticationManager,
2089    caplog: pytest.LogCaptureFixture,
2090) -> None:
2091    """
2092    Verify a support log can tell a rejected code apart from a throttled client.
2093
2094    :param auth_manager: AuthenticationManager instance.
2095    :param caplog: Log capture fixture.
2096    """
2097    set_current_client_id("connection-d")
2098    for _ in range(3):
2099        await auth_manager.exchange_join_code("MISTYPED1234")
2100    rejections = [record.getMessage() for record in caplog.records if "rejected" in record.message]
2101    assert len(rejections) == 3
2102    assert "client=connection-d" in rejections[0]
2103    # the attempted code is only ever logged masked
2104    assert "code=MIST********" in rejections[0]
2105    assert "MISTYPED1234" not in rejections[0]
2106
2107    caplog.clear()
2108    await auth_manager.exchange_join_code("MISTYPED1234")
2109    throttles = [record.getMessage() for record in caplog.records if "throttled" in record.message]
2110    assert len(throttles) == 1
2111    assert "throttled by the client limit" in throttles[0]
2112    assert "client=connection-d" in throttles[0]
2113    assert "client_failures=3" in throttles[0]
2114    assert "server_failures=3" in throttles[0]
2115
2116
2117async def test_login_rate_limiter_default_tiers() -> None:
2118    """Verify the default progressive delays escalate with the failed attempt count."""
2119    limiter = LoginRateLimiter()
2120
2121    assert limiter.get_attempt_count("bob") == 0
2122    for expected_count, expected_delay in ((2, 0), (3, 30), (6, 60), (10, 120), (15, 300)):
2123        while limiter.get_attempt_count("bob") < expected_count:
2124            await limiter.record_failed_attempt("bob")
2125        assert limiter.get_delay("bob") == expected_delay
2126
2127    await limiter.clear_attempts("bob")
2128    assert limiter.get_attempt_count("bob") == 0
2129    assert limiter.get_delay("bob") == 0
2130
2131
2132async def test_login_rate_limiter_custom_tiers() -> None:
2133    """Verify a limiter can be configured with its own threshold and delay."""
2134    limiter = LoginRateLimiter(delay_tiers=((3, 60),))
2135
2136    for _ in range(2):
2137        await limiter.record_failed_attempt("key")
2138    assert await limiter.check_rate_limit("key") == (True, 0)
2139
2140    await limiter.record_failed_attempt("key")
2141    allowed, remaining_delay = await limiter.check_rate_limit("key")
2142    assert allowed is False
2143    assert 0 < remaining_delay <= 60
2144
2145
2146async def test_login_rate_limiter_drops_attempts_outside_window() -> None:
2147    """Verify attempts older than the tracking window stop counting."""
2148    limiter = LoginRateLimiter(tracking_window=timedelta(seconds=0))
2149
2150    await limiter.record_failed_attempt("key")
2151    assert limiter.get_attempt_count("key") == 0
2152
2153
2154async def test_login_rate_limiter_prunes_expired_keys() -> None:
2155    """
2156    Verify expired keys do not pile up when they are never used again.
2157
2158    Keys are one-off by nature (a connection that gives up, a made-up username), so
2159    without the sweep the bookkeeping would grow for as long as the server runs.
2160    """
2161    limiter = LoginRateLimiter(tracking_window=timedelta(seconds=0))
2162
2163    for index in range(PRUNE_THRESHOLD + 1):
2164        await limiter.record_failed_attempt(f"key{index}")
2165
2166    # Every attempt is already outside this limiter's window, so the sweep drops them all.
2167    assert len(limiter._failed_attempts) == 0
2168
2169    live = LoginRateLimiter()
2170    for index in range(PRUNE_THRESHOLD + 1):
2171        await live.record_failed_attempt(f"key{index}")
2172
2173    # Attempts inside the tracking window are never swept away.
2174    assert len(live._failed_attempts) == PRUNE_THRESHOLD + 1
2175
2176
2177async def test_resolve_command_impersonation(auth_manager: AuthenticationManager) -> None:
2178    """Test resolving the impersonation argument of an incoming API command."""
2179    admin_user = await auth_manager.create_user(username="admin", role=UserRole.ADMIN)
2180    standard_user = await auth_manager.create_user(username="user_a", role=UserRole.USER)
2181    set_current_user(admin_user)
2182    set_impersonated_user(None)
2183
2184    # no user argument present is a no-op and leaves other args untouched
2185    args: dict[str, object] = {"queue_id": "abc"}
2186    assert await resolve_command_impersonation(auth_manager.mass, args) is None
2187    assert args == {"queue_id": "abc"}
2188
2189    # an empty string is deliberately treated as "no impersonation requested"
2190    # (optional fields in automations/scripts commonly template to an empty string)
2191    args = {"queue_id": "abc", "user": ""}
2192    assert await resolve_command_impersonation(auth_manager.mass, args) is None
2193    assert args == {"queue_id": "abc"}
2194
2195    # the user argument is popped and resolved (by username)
2196    args = {"queue_id": "abc", "user": "user_a"}
2197    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2198    assert resolved == standard_user
2199    assert args == {"queue_id": "abc"}
2200
2201    # the user argument is also resolved by user_id
2202    args = {"user": standard_user.user_id}
2203    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2204    assert resolved == standard_user
2205
2206    # username is accepted as (deprecated) alias for user
2207    args = {"username": "user_a"}
2208    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2209    assert resolved == standard_user
2210    assert args == {}
2211
2212    # a caller without the users.impersonate scope may not impersonate another user
2213    set_current_user(standard_user)
2214    with pytest.raises(InsufficientPermissions):
2215        await resolve_command_impersonation(auth_manager.mass, {"user": "admin"})
2216
2217
2218def test_has_scope() -> None:
2219    """Test the scope check for each of the builtin user roles."""
2220
2221    def _user(role: str) -> User:
2222        return User(user_id="abc123", username="testuser", role=role)
2223
2224    # admin has all scopes through the wildcard
2225    assert has_scope(_user(UserRole.ADMIN), Scope.CONFIG_CORE_WRITE)
2226    assert has_scope(_user(UserRole.ADMIN), Scope.LIBRARY_MANAGE)
2227    # regular user
2228    assert has_scope(_user(UserRole.USER), Scope.LIBRARY_WRITE)
2229    assert has_scope(_user(UserRole.USER), Scope.CONFIG_CORE_READ)
2230    assert not has_scope(_user(UserRole.USER), Scope.CONFIG_CORE_WRITE)
2231    assert not has_scope(_user(UserRole.USER), Scope.USERS_IMPERSONATE)
2232    # guest
2233    assert has_scope(_user(UserRole.GUEST), Scope.LIBRARY_READ)
2234    assert not has_scope(_user(UserRole.GUEST), Scope.LIBRARY_WRITE)
2235    assert not has_scope(_user(UserRole.GUEST), Scope.CONFIG_CORE_READ)
2236    # service
2237    assert has_scope(_user(UserRole.SERVICE), Scope.USERS_IMPERSONATE)
2238    assert has_scope(_user(UserRole.SERVICE), Scope.USERS_READ)
2239    assert has_scope(_user(UserRole.SERVICE), Scope.CONFIG_PLAYERS_WRITE)
2240    assert not has_scope(_user(UserRole.SERVICE), Scope.CONFIG_CORE_WRITE)
2241    # reading user accounts does not imply managing them
2242    assert not has_scope(_user(UserRole.SERVICE), Scope.USERS_MANAGE)
2243    # an unknown (custom) role id is fail-closed and grants no scopes at all
2244    assert not has_scope(_user("some_future_role"), Scope.LIBRARY_READ)
2245
2246
2247async def test_homeassistant_system_user_may_read_users(
2248    auth_manager: AuthenticationManager,
2249) -> None:
2250    """
2251    Test that the Home Assistant integration may list users to resolve the calling user.
2252
2253    :param auth_manager: AuthenticationManager instance.
2254    """
2255    system_user = await auth_manager.get_homeassistant_system_user()
2256    standard_user = await auth_manager.create_user(username="user_a", role=UserRole.USER)
2257    guest_user = await auth_manager.create_user(username="guest_a", role=UserRole.GUEST)
2258    for command in (AuthenticationManager.list_users, AuthenticationManager.get_user):
2259        assert getattr(command, "api_required_scope", None) is Scope.USERS_READ
2260    assert has_scope(system_user, Scope.USERS_READ)
2261    # reading user accounts remains off limits for regular users and guests
2262    assert not has_scope(standard_user, Scope.USERS_READ)
2263    assert not has_scope(guest_user, Scope.USERS_READ)
2264
2265
2266async def test_homeassistant_system_user_has_service_role(
2267    auth_manager: AuthenticationManager,
2268) -> None:
2269    """Test that the Home Assistant system user is created with the service role."""
2270    system_user = await auth_manager.get_homeassistant_system_user()
2271    assert system_user.role == UserRole.SERVICE
2272
2273    # a pre-existing system user with the old user role is migrated to service
2274    await auth_manager.database.update(
2275        "users", {"user_id": system_user.user_id}, {"role": UserRole.USER.value}
2276    )
2277    await auth_manager._migrate_system_user_role()
2278    migrated_user = await auth_manager.get_user(system_user.user_id)
2279    assert migrated_user is not None
2280    assert migrated_user.role == UserRole.SERVICE
2281