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