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_short_lived_jwt_exp_carries_absolute_max(
1152    auth_manager: AuthenticationManager,
1153) -> None:
1154    """
1155    Test that a short-lived JWT's exp claim equals the absolute max lifetime.
1156
1157    The database expires_at enforces the sliding idle window; an exp claim shorter
1158    than the absolute max would cut off active sessions before renewal can happen.
1159
1160    :param auth_manager: AuthenticationManager instance.
1161    """
1162    user = await auth_manager.create_user(username="jwtexpuser", role=UserRole.USER)
1163    token = await auth_manager.create_token(user, "JWT Exp Test", is_long_lived=False)
1164
1165    token_hash = hashlib.sha256(token.encode()).hexdigest()
1166    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1167    assert token_row is not None
1168    created_at = datetime.fromisoformat(token_row["created_at"])
1169
1170    payload = auth_manager.jwt_helper.decode_token(token, verify_exp=False)
1171    expected = created_at + timedelta(days=TOKEN_ABSOLUTE_MAX_EXPIRATION)
1172    assert payload["exp"] == int(expected.timestamp())
1173
1174    # The database keeps the shorter sliding window as source of truth
1175    expires_at = datetime.fromisoformat(token_row["expires_at"])
1176    assert expires_at - created_at == timedelta(days=TOKEN_SHORT_LIVED_EXPIRATION)
1177
1178
1179async def test_guest_token_fixed_short_lifetime(auth_manager: AuthenticationManager) -> None:
1180    """
1181    Test that guest tokens get a short fixed lifetime and never renew on use.
1182
1183    :param auth_manager: AuthenticationManager instance.
1184    """
1185    user = await auth_manager.create_user(username="guestexpiry", role=UserRole.GUEST)
1186    token = await auth_manager.create_token(user, "Guest Session", is_long_lived=False)
1187
1188    token_hash = hashlib.sha256(token.encode()).hexdigest()
1189    token_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1190    assert token_row is not None
1191    created_at = datetime.fromisoformat(token_row["created_at"])
1192    expires_at = datetime.fromisoformat(token_row["expires_at"])
1193    assert expires_at - created_at == timedelta(days=TOKEN_GUEST_EXPIRATION)
1194
1195    # The JWT exp claim must match the fixed window, not the absolute max
1196    payload = auth_manager.jwt_helper.decode_token(token, verify_exp=False)
1197    assert payload["exp"] == int(expires_at.timestamp())
1198
1199    # Authenticating must not extend the expiration (no sliding window for guests)
1200    assert await auth_manager.authenticate_with_token(token) is not None
1201    updated_row = await auth_manager.database.get_row("auth_tokens", {"token_hash": token_hash})
1202    assert updated_row is not None
1203    assert updated_row["expires_at"] == token_row["expires_at"]
1204
1205
1206async def test_guest_cannot_create_long_lived_token(auth_manager: AuthenticationManager) -> None:
1207    """
1208    Test that a guest cannot create a long-lived token for their own account.
1209
1210    :param auth_manager: AuthenticationManager instance.
1211    """
1212    guest = await auth_manager.create_user(username="guesttoken", role=UserRole.GUEST)
1213    set_current_user(guest)
1214
1215    with pytest.raises(InsufficientPermissions):
1216        await auth_manager.create_long_lived_token("Guest Escalation")
1217
1218
1219async def test_no_long_lived_token_for_guest_account(auth_manager: AuthenticationManager) -> None:
1220    """
1221    Test that a long-lived token cannot be created for a guest account, even by an admin.
1222
1223    :param auth_manager: AuthenticationManager instance.
1224    """
1225    admin = await auth_manager.create_user(username="tokenadmin", role=UserRole.ADMIN)
1226    guest = await auth_manager.create_user(username="guesttarget", role=UserRole.GUEST)
1227    set_current_user(admin)
1228
1229    with pytest.raises(InsufficientPermissions):
1230        await auth_manager.create_long_lived_token("Guest Token", user_id=guest.user_id)
1231
1232
1233async def test_username_case_insensitive_creation(auth_manager: AuthenticationManager) -> None:
1234    """
1235    Test that usernames are normalized to lowercase on creation.
1236
1237    :param auth_manager: AuthenticationManager instance.
1238    """
1239    # Create user with mixed case username
1240    user = await auth_manager.create_user(
1241        username="TestUser",
1242        role=UserRole.USER,
1243        display_name="Test User",
1244    )
1245
1246    # Username should be stored in lowercase
1247    assert user.username == "testuser"
1248
1249
1250async def test_username_case_insensitive_duplicate_prevention(
1251    auth_manager: AuthenticationManager,
1252) -> None:
1253    """
1254    Test that duplicate usernames with different cases are prevented.
1255
1256    :param auth_manager: AuthenticationManager instance.
1257    """
1258    # Create user with lowercase username
1259    await auth_manager.create_user(username="admin", role=UserRole.USER)
1260
1261    # Try to create user with same username but different case should fail
1262    # (SQLite UNIQUE constraint violation)
1263    with pytest.raises(IntegrityError, match="UNIQUE constraint failed"):
1264        await auth_manager.create_user(username="Admin", role=UserRole.USER)
1265
1266
1267async def test_username_case_insensitive_login(auth_manager: AuthenticationManager) -> None:
1268    """
1269    Test that login works with any case variation of username.
1270
1271    :param auth_manager: AuthenticationManager instance.
1272    """
1273    builtin_provider = auth_manager.login_providers.get("builtin")
1274    assert builtin_provider is not None
1275    assert isinstance(builtin_provider, BuiltinLoginProvider)
1276
1277    # Create user with lowercase username
1278    await builtin_provider.create_user_with_password(
1279        username="testadmin",
1280        password="SecurePassword123",
1281        role=UserRole.ADMIN,
1282    )
1283
1284    # Test login with lowercase
1285    result = await auth_manager.authenticate_with_credentials(
1286        "builtin",
1287        {"username": "testadmin", "password": "SecurePassword123"},
1288    )
1289    assert result.success is True
1290    assert result.user is not None
1291    assert result.user.username == "testadmin"
1292
1293    # Test login with uppercase
1294    result = await auth_manager.authenticate_with_credentials(
1295        "builtin",
1296        {"username": "TESTADMIN", "password": "SecurePassword123"},
1297    )
1298    assert result.success is True
1299    assert result.user is not None
1300    assert result.user.username == "testadmin"
1301
1302    # Test login with mixed case
1303    result = await auth_manager.authenticate_with_credentials(
1304        "builtin",
1305        {"username": "TestAdmin", "password": "SecurePassword123"},
1306    )
1307    assert result.success is True
1308    assert result.user is not None
1309    assert result.user.username == "testadmin"
1310
1311
1312async def test_username_case_insensitive_lookup(auth_manager: AuthenticationManager) -> None:
1313    """
1314    Test that user lookup by username is case-insensitive.
1315
1316    :param auth_manager: AuthenticationManager instance.
1317    """
1318    # Create user with lowercase username
1319    created_user = await auth_manager.create_user(username="lookupuser", role=UserRole.USER)
1320
1321    # Lookup with lowercase
1322    user1 = await auth_manager.get_user_by_username("lookupuser")
1323    assert user1 is not None
1324    assert user1.user_id == created_user.user_id
1325
1326    # Lookup with uppercase
1327    user2 = await auth_manager.get_user_by_username("LOOKUPUSER")
1328    assert user2 is not None
1329    assert user2.user_id == created_user.user_id
1330
1331    # Lookup with mixed case
1332    user3 = await auth_manager.get_user_by_username("LookUpUser")
1333    assert user3 is not None
1334    assert user3.user_id == created_user.user_id
1335
1336
1337async def test_username_update_normalizes(auth_manager: AuthenticationManager) -> None:
1338    """
1339    Test that updating username normalizes it to lowercase.
1340
1341    :param auth_manager: AuthenticationManager instance.
1342    """
1343    user = await auth_manager.create_user(username="originaluser", role=UserRole.USER)
1344
1345    # Update username with mixed case
1346    updated_user = await auth_manager.update_user(user, username="UpdatedUser")
1347
1348    # Username should be normalized to lowercase
1349    assert updated_user is not None
1350    assert updated_user.username == "updateduser"
1351
1352
1353async def test_link_user_to_provider_idempotent(auth_manager: AuthenticationManager) -> None:
1354    """
1355    Test that linking user to provider is idempotent.
1356
1357    This tests the fix for the bug where re-linking a user would cause
1358    IntegrityError due to UNIQUE constraint on (provider_type, provider_user_id).
1359
1360    :param auth_manager: AuthenticationManager instance.
1361    """
1362    user = await auth_manager.create_user(username="hauser", role=UserRole.USER)
1363
1364    # Link user to Home Assistant provider for the first time
1365    link1 = await auth_manager.link_user_to_provider(
1366        user,
1367        AuthProviderType.HOME_ASSISTANT,
1368        "ha_user_456",
1369    )
1370
1371    assert link1 is not None
1372    assert link1.user_id == user.user_id
1373    assert link1.provider_type == AuthProviderType.HOME_ASSISTANT
1374    assert link1.provider_user_id == "ha_user_456"
1375
1376    # Linking the same user again should return existing link without error
1377    link2 = await auth_manager.link_user_to_provider(
1378        user,
1379        AuthProviderType.HOME_ASSISTANT,
1380        "ha_user_456",
1381    )
1382
1383    assert link2 is not None
1384    assert link2.link_id == link1.link_id  # Should be same link
1385    assert link2.user_id == user.user_id
1386    assert link2.provider_type == AuthProviderType.HOME_ASSISTANT
1387    assert link2.provider_user_id == "ha_user_456"
1388
1389
1390async def test_ingress_auth_existing_username(auth_manager: AuthenticationManager) -> None:
1391    """
1392    Test HA ingress auth when username exists but isn't linked to HA provider.
1393
1394    This tests the scenario where a user is created during setup, and then
1395    tries to login via HA ingress with the same username.
1396
1397    :param auth_manager: AuthenticationManager instance.
1398    """
1399    # Simulate user created during initial setup
1400    existing_user = await auth_manager.create_user(
1401        username="admin",
1402        role=UserRole.ADMIN,
1403        display_name="Admin User",
1404    )
1405
1406    # Now simulate HA ingress trying to auto-create a user with same username
1407    # This should find the existing user and link it instead of creating new one
1408    user = await auth_manager.get_user_by_username("admin")
1409    assert user is not None
1410    assert user.user_id == existing_user.user_id
1411
1412    # Link the existing user to HA provider (what ingress flow would do)
1413    link = await auth_manager.link_user_to_provider(
1414        user,
1415        AuthProviderType.HOME_ASSISTANT,
1416        "ha_admin_123",
1417    )
1418
1419    assert link is not None
1420    assert link.user_id == existing_user.user_id
1421
1422    # Verify we can retrieve user by provider link
1423    retrieved_user = await auth_manager.get_user_by_provider_link(
1424        AuthProviderType.HOME_ASSISTANT,
1425        "ha_admin_123",
1426    )
1427
1428    assert retrieved_user is not None
1429    assert retrieved_user.user_id == existing_user.user_id
1430    assert retrieved_user.username == "admin"
1431
1432
1433# ==================== Join Code Tests ====================
1434
1435
1436async def test_generate_join_code(auth_manager: AuthenticationManager) -> None:
1437    """
1438    Test generating a join code for a user.
1439
1440    :param auth_manager: AuthenticationManager instance.
1441    """
1442    user = await auth_manager.create_user(username="joincodeuser", role=UserRole.GUEST)
1443
1444    code, expires_at = await auth_manager.generate_join_code(
1445        user=user,
1446        expires_in_hours=24,
1447        max_uses=0,
1448        device_name="Test Device",
1449    )
1450
1451    assert code is not None
1452    assert len(code) == JOIN_CODE_LENGTH
1453    assert code.isalnum()
1454    assert expires_at is not None
1455    assert expires_at > utc()
1456
1457
1458async def test_get_join_code_expiry(auth_manager: AuthenticationManager) -> None:
1459    """
1460    Test looking up the expiry for a specific active join code.
1461
1462    :param auth_manager: AuthenticationManager instance.
1463    """
1464    user = await auth_manager.create_user(username="joinexpiryuser", role=UserRole.GUEST)
1465
1466    code, expires_at = await auth_manager.generate_join_code(
1467        user=user,
1468        expires_in_hours=24,
1469    )
1470
1471    assert await auth_manager.get_join_code_expiry(code, user) == expires_at
1472    assert await auth_manager.get_join_code_expiry(code.lower(), user) == expires_at
1473    assert await auth_manager.get_join_code_expiry("BADCODE", user) is None
1474
1475
1476async def test_get_join_code_expiry_requires_matching_user(
1477    auth_manager: AuthenticationManager,
1478) -> None:
1479    """
1480    Test that join code expiry lookup can be scoped to a specific user.
1481
1482    :param auth_manager: AuthenticationManager instance.
1483    """
1484    user = await auth_manager.create_user(username="joinexpiryowner", role=UserRole.GUEST)
1485    other_user = await auth_manager.create_user(
1486        username="joinexpiryother",
1487        role=UserRole.GUEST,
1488    )
1489
1490    code, expires_at = await auth_manager.generate_join_code(
1491        user=user,
1492        expires_in_hours=24,
1493    )
1494
1495    assert await auth_manager.get_join_code_expiry(code, user) == expires_at
1496    assert await auth_manager.get_join_code_expiry(code) == expires_at
1497    assert await auth_manager.get_join_code_expiry(code, other_user) is None
1498
1499
1500async def test_get_join_code_expiry_expired(auth_manager: AuthenticationManager) -> None:
1501    """
1502    Test that expired join codes have no active expiry.
1503
1504    :param auth_manager: AuthenticationManager instance.
1505    """
1506    user = await auth_manager.create_user(username="joinexpiryexpired", role=UserRole.GUEST)
1507
1508    code, _ = await auth_manager.generate_join_code(
1509        user=user,
1510        expires_in_hours=24,
1511    )
1512    code_row = await auth_manager.database.get_row("join_codes", {"code": code})
1513    assert code_row is not None
1514
1515    past_time = utc() - timedelta(hours=1)
1516    await auth_manager.database.update(
1517        "join_codes",
1518        {"code_id": code_row["code_id"]},
1519        {"expires_at": past_time.isoformat()},
1520    )
1521
1522    assert await auth_manager.get_join_code_expiry(code, user) is None
1523
1524
1525async def test_generate_join_code_non_guest_rejected(
1526    auth_manager: AuthenticationManager,
1527) -> None:
1528    """
1529    Test that generating a join code for non-guest users is rejected.
1530
1531    :param auth_manager: AuthenticationManager instance.
1532    """
1533    admin = await auth_manager.create_user(username="joinadmin", role=UserRole.ADMIN)
1534    user = await auth_manager.create_user(username="joinuser", role=UserRole.USER)
1535
1536    with pytest.raises(ValueError, match="guest accounts"):
1537        await auth_manager.generate_join_code(user=admin)
1538
1539    with pytest.raises(ValueError, match="guest accounts"):
1540        await auth_manager.generate_join_code(user=user)
1541
1542
1543async def test_exchange_join_code(auth_manager: AuthenticationManager) -> None:
1544    """
1545    Test exchanging a valid join code for a JWT token.
1546
1547    :param auth_manager: AuthenticationManager instance.
1548    """
1549    user = await auth_manager.create_user(username="exchangeuser", role=UserRole.GUEST)
1550
1551    code, _ = await auth_manager.generate_join_code(
1552        user=user,
1553        expires_in_hours=24,
1554        device_name="Exchange Test",
1555    )
1556
1557    # Exchange code for token
1558    token = await auth_manager._exchange_join_code(code)
1559
1560    assert token is not None
1561    assert len(token) > 0
1562
1563    # Verify token works for authentication
1564    authenticated_user = await auth_manager.authenticate_with_token(token)
1565    assert authenticated_user is not None
1566    assert authenticated_user.user_id == user.user_id
1567    assert authenticated_user.username == user.username
1568
1569
1570async def test_exchange_join_code_case_insensitive(auth_manager: AuthenticationManager) -> None:
1571    """
1572    Test that join codes are case-insensitive.
1573
1574    :param auth_manager: AuthenticationManager instance.
1575    """
1576    user = await auth_manager.create_user(username="caseuser", role=UserRole.GUEST)
1577
1578    code, _ = await auth_manager.generate_join_code(
1579        user=user,
1580        expires_in_hours=24,
1581    )
1582
1583    # Exchange with lowercase version
1584    token = await auth_manager._exchange_join_code(code.lower())
1585    assert token is not None
1586
1587    # Verify token works
1588    authenticated_user = await auth_manager.authenticate_with_token(token)
1589    assert authenticated_user is not None
1590    assert authenticated_user.user_id == user.user_id
1591
1592
1593async def test_exchange_join_code_invalid(auth_manager: AuthenticationManager) -> None:
1594    """
1595    Test that invalid join codes are rejected.
1596
1597    :param auth_manager: AuthenticationManager instance.
1598    """
1599    token = await auth_manager._exchange_join_code("INVALID")
1600    assert token is None
1601
1602
1603async def test_exchange_join_code_expired(auth_manager: AuthenticationManager) -> None:
1604    """
1605    Test that expired join codes are rejected.
1606
1607    :param auth_manager: AuthenticationManager instance.
1608    """
1609    user = await auth_manager.create_user(username="expiredcodeuser", role=UserRole.GUEST)
1610
1611    code, _ = await auth_manager.generate_join_code(
1612        user=user,
1613        expires_in_hours=24,
1614    )
1615
1616    # Manually expire the code by updating expires_at in database
1617    code_row = await auth_manager.database.get_row("join_codes", {"code": code})
1618    assert code_row is not None
1619
1620    past_time = utc() - timedelta(hours=1)
1621    await auth_manager.database.update(
1622        "join_codes",
1623        {"code_id": code_row["code_id"]},
1624        {"expires_at": past_time.isoformat()},
1625    )
1626
1627    # Try to exchange expired code
1628    token = await auth_manager._exchange_join_code(code)
1629    assert token is None
1630
1631
1632async def test_exchange_join_code_max_uses(auth_manager: AuthenticationManager) -> None:
1633    """
1634    Test that join codes respect max_uses limit.
1635
1636    :param auth_manager: AuthenticationManager instance.
1637    """
1638    user = await auth_manager.create_user(username="maxusesuser", role=UserRole.GUEST)
1639
1640    code, _ = await auth_manager.generate_join_code(
1641        user=user,
1642        expires_in_hours=24,
1643        max_uses=2,  # Only allow 2 uses
1644    )
1645
1646    # First use should succeed
1647    token1 = await auth_manager._exchange_join_code(code)
1648    assert token1 is not None
1649
1650    # Second use should succeed
1651    token2 = await auth_manager._exchange_join_code(code)
1652    assert token2 is not None
1653
1654    # Third use should fail (max_uses=2 exceeded)
1655    token3 = await auth_manager._exchange_join_code(code)
1656    assert token3 is None
1657
1658
1659async def test_exchange_join_code_unlimited_uses(auth_manager: AuthenticationManager) -> None:
1660    """
1661    Test that join codes with max_uses=0 have unlimited uses.
1662
1663    :param auth_manager: AuthenticationManager instance.
1664    """
1665    user = await auth_manager.create_user(username="unlimiteduser", role=UserRole.GUEST)
1666
1667    code, _ = await auth_manager.generate_join_code(
1668        user=user,
1669        expires_in_hours=24,
1670        max_uses=0,  # Unlimited
1671    )
1672
1673    # Should be able to use multiple times
1674    for _ in range(5):
1675        token = await auth_manager._exchange_join_code(code)
1676        assert token is not None
1677
1678
1679async def test_revoke_join_codes_for_user(auth_manager: AuthenticationManager) -> None:
1680    """
1681    Test revoking join codes for a specific user.
1682
1683    :param auth_manager: AuthenticationManager instance.
1684    """
1685    user1 = await auth_manager.create_user(username="revokeuser1", role=UserRole.GUEST)
1686    user2 = await auth_manager.create_user(username="revokeuser2", role=UserRole.GUEST)
1687
1688    # Create codes for both users
1689    code1, _ = await auth_manager.generate_join_code(user=user1)
1690    code2, _ = await auth_manager.generate_join_code(user=user2)
1691
1692    # Revoke codes for user1 only
1693    revoked_count = await auth_manager.revoke_join_codes(user1)
1694    assert revoked_count == 1
1695
1696    # User1's code should no longer work
1697    token1 = await auth_manager._exchange_join_code(code1)
1698    assert token1 is None
1699
1700    # User2's code should still work
1701    token2 = await auth_manager._exchange_join_code(code2)
1702    assert token2 is not None
1703
1704
1705async def test_authenticate_with_join_code_api(auth_manager: AuthenticationManager) -> None:
1706    """
1707    Test the public API endpoint for join code authentication.
1708
1709    :param auth_manager: AuthenticationManager instance.
1710    """
1711    user = await auth_manager.create_user(
1712        username="apijoincodeuser",
1713        role=UserRole.GUEST,
1714        display_name="API Guest",
1715    )
1716
1717    code, _ = await auth_manager.generate_join_code(
1718        user=user,
1719        expires_in_hours=24,
1720    )
1721
1722    # Call the API endpoint
1723    result = await auth_manager.exchange_join_code(code)
1724
1725    assert result["success"] is True
1726    assert "access_token" in result
1727    assert result["user"]["user_id"] == user.user_id
1728    assert result["user"]["username"] == user.username
1729    assert result["user"]["role"] == "guest"
1730
1731
1732async def test_authenticate_with_join_code_api_invalid(
1733    auth_manager: AuthenticationManager,
1734) -> None:
1735    """
1736    Test the API endpoint with invalid join code.
1737
1738    :param auth_manager: AuthenticationManager instance.
1739    """
1740    result = await auth_manager.exchange_join_code("BADCODE")
1741
1742    assert result["success"] is False
1743    assert "error" in result
1744    assert "access_token" not in result
1745
1746
1747async def test_list_join_codes(auth_manager: AuthenticationManager) -> None:
1748    """
1749    Test listing active join codes (admin only).
1750
1751    :param auth_manager: AuthenticationManager instance.
1752    """
1753    admin = await auth_manager.create_user(username="listcodesadmin", role=UserRole.ADMIN)
1754    guest1 = await auth_manager.create_user(username="listguest1", role=UserRole.GUEST)
1755    guest2 = await auth_manager.create_user(username="listguest2", role=UserRole.GUEST)
1756    set_current_user(admin)
1757
1758    # Create codes for both guests
1759    await auth_manager.generate_join_code(user=guest1)
1760    await auth_manager.generate_join_code(user=guest2)
1761
1762    # List all codes
1763    codes = await auth_manager.list_join_codes()
1764    assert len(codes) == 2
1765
1766    # List codes for specific user
1767    codes = await auth_manager.list_join_codes(user_id=guest1.user_id)
1768    assert len(codes) == 1
1769    assert codes[0]["user_id"] == guest1.user_id
1770
1771
1772async def test_revoke_join_code_api(auth_manager: AuthenticationManager) -> None:
1773    """
1774    Test revoking a specific join code by code_id (admin only).
1775
1776    :param auth_manager: AuthenticationManager instance.
1777    """
1778    admin = await auth_manager.create_user(username="revokecodeadmin", role=UserRole.ADMIN)
1779    guest = await auth_manager.create_user(username="revokeguest", role=UserRole.GUEST)
1780    set_current_user(admin)
1781
1782    code, _ = await auth_manager.generate_join_code(user=guest)
1783
1784    # Get the code_id from the database
1785    codes = await auth_manager.list_join_codes(user_id=guest.user_id)
1786    assert len(codes) == 1
1787    code_id = codes[0]["code_id"]
1788
1789    # Revoke the specific code
1790    await auth_manager.revoke_join_code(code_id)
1791
1792    # Code should no longer work
1793    token = await auth_manager._exchange_join_code(code)
1794    assert token is None
1795
1796    # List should be empty
1797    codes = await auth_manager.list_join_codes(user_id=guest.user_id)
1798    assert len(codes) == 0
1799
1800
1801async def test_revoke_join_code_api_not_found(auth_manager: AuthenticationManager) -> None:
1802    """
1803    Test revoking a non-existent join code raises error.
1804
1805    :param auth_manager: AuthenticationManager instance.
1806    """
1807    admin = await auth_manager.create_user(username="revokenotfound", role=UserRole.ADMIN)
1808    set_current_user(admin)
1809
1810    with pytest.raises(InvalidDataError, match="Join code not found"):
1811        await auth_manager.revoke_join_code("nonexistent-code-id")
1812
1813
1814async def test_impersonated_user_context_manager(auth_manager: AuthenticationManager) -> None:
1815    """Test the ImpersonatedUser context manager."""
1816    admin_user = await auth_manager.create_user(username="admin", role=UserRole.ADMIN)
1817    standard_user_a = await auth_manager.create_user(username="user_a", role=UserRole.USER)
1818    standard_user_b = await auth_manager.create_user(username="user_b", role=UserRole.USER)
1819    service_user = await auth_manager.create_user(username="service", role=UserRole.SERVICE)
1820
1821    # non-authenticated user must raise
1822    set_current_user(None)
1823    with pytest.raises(InsufficientPermissions):
1824        async with ImpersonatedUser(auth_manager.mass, "user_a"):
1825            ...
1826    # impersonation attempt without the users.impersonate scope must raise
1827    set_current_user(standard_user_a)
1828    with pytest.raises(InsufficientPermissions):
1829        async with ImpersonatedUser(auth_manager.mass, "admin"):
1830            ...
1831    # invalid username must raise
1832    set_current_user(admin_user)
1833    with pytest.raises(UserNotFoundError):
1834        async with ImpersonatedUser(auth_manager.mass, "wrong_username"):
1835            ...
1836
1837    # verify that a standard user may impersonate itself (by username or user_id)
1838    set_current_user(standard_user_a)
1839    set_impersonated_user(None)
1840    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1841        assert get_current_user() == standard_user_a
1842    async with ImpersonatedUser(auth_manager.mass, standard_user_a.user_id):
1843        assert get_current_user() == standard_user_a
1844    # passing None is a no-op which preserves any active impersonation
1845    set_impersonated_user(standard_user_b)
1846    async with ImpersonatedUser(auth_manager.mass, None):
1847        assert get_current_user() == standard_user_b
1848    assert get_current_user() == standard_user_b
1849
1850    # verify that an admin user may impersonate another user
1851    set_current_user(admin_user)
1852
1853    set_impersonated_user(None)  # non-nested use
1854    assert get_current_user() == admin_user
1855    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1856        assert get_current_user() == standard_user_a
1857    assert get_current_user() == admin_user
1858
1859    set_impersonated_user(standard_user_b)  # nested use
1860    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1861        assert get_current_user() == standard_user_a
1862    assert get_current_user() == standard_user_b
1863
1864    # verify that a service user may impersonate another user (users.impersonate scope)
1865    set_current_user(service_user)
1866    set_impersonated_user(None)
1867    assert has_scope(service_user, Scope.USERS_IMPERSONATE)
1868    async with ImpersonatedUser(auth_manager.mass, "user_a"):
1869        assert get_current_user() == standard_user_a
1870    assert get_current_user() == service_user
1871
1872
1873async def test_impersonated_user_anonymous_playback_is_noop(
1874    auth_manager: AuthenticationManager,
1875) -> None:
1876    """
1877    Verify an unauthenticated call without a username is a no-op.
1878
1879    Regression: play_media wraps every call in ImpersonatedUser, so protocol/hardware
1880    triggered playback (presets, Spotify Connect, ...) - which has no authenticated user
1881    and passes no username - must not raise.
1882    """
1883    set_current_user(None)
1884    set_impersonated_user(None)
1885    async with ImpersonatedUser(auth_manager.mass, None):
1886        assert get_current_user() is None
1887    assert get_current_user() is None
1888
1889    # an unauthenticated caller may still not impersonate another user
1890    with pytest.raises(InsufficientPermissions):
1891        async with ImpersonatedUser(auth_manager.mass, "user_a"):
1892            ...
1893
1894
1895async def test_join_code_length_at_least_12() -> None:
1896    """Verify join codes are long enough to resist brute force (security finding 7.3.2)."""
1897    assert JOIN_CODE_LENGTH >= 12
1898
1899
1900async def test_exchange_join_code_rate_limited(auth_manager: AuthenticationManager) -> None:
1901    """
1902    Verify repeated failed join code exchanges get throttled (security finding 7.3.2).
1903
1904    :param auth_manager: AuthenticationManager instance.
1905    """
1906    # Three failures trip the progressive delay threshold.
1907    for _ in range(3):
1908        result = await auth_manager.exchange_join_code("WRONGCODE123")
1909        assert result["success"] is False
1910
1911    # The next attempt must be rejected for rate limiting, not just "invalid".
1912    result = await auth_manager.exchange_join_code("WRONGCODE123")
1913    assert result["success"] is False
1914    assert "too many" in result["error"].lower()
1915
1916
1917async def test_exchange_join_code_rate_limit_concurrent_burst(
1918    auth_manager: AuthenticationManager,
1919) -> None:
1920    """
1921    Verify concurrent failed exchanges cannot race past the rate limiter.
1922
1923    Without serialization, parallel requests all pass the rate limit check
1924    before any of them records a failure, allowing brute-force bursts.
1925
1926    :param auth_manager: AuthenticationManager instance.
1927    """
1928    results = await asyncio.gather(
1929        *(auth_manager.exchange_join_code("WRONGCODE123") for _ in range(10))
1930    )
1931
1932    assert all(result["success"] is False for result in results)
1933    # Only the first 3 attempts may reach the actual code check; the rest must be throttled.
1934    invalid_count = sum(1 for result in results if "invalid" in result["error"].lower())
1935    throttled_count = sum(1 for result in results if "too many" in result["error"].lower())
1936    assert invalid_count == 3
1937    assert throttled_count == 7
1938
1939
1940async def test_exchange_join_code_success_does_not_reset_rate_limit(
1941    auth_manager: AuthenticationManager,
1942) -> None:
1943    """
1944    Verify a successful exchange does not clear the shared failed-attempt counter.
1945
1946    Callers without a connection identity share one bucket, so clearing it on success
1947    would let an attacker holding any valid code reset the counter for everyone.
1948
1949    :param auth_manager: AuthenticationManager instance.
1950    """
1951    user = await auth_manager.create_user(username="norstuser", role=UserRole.GUEST)
1952    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
1953
1954    # A couple of failures, still below the throttle threshold.
1955    for _ in range(2):
1956        assert (await auth_manager.exchange_join_code("NOPE12345678"))["success"] is False
1957
1958    # A valid exchange still succeeds but must not wipe the counter.
1959    assert (await auth_manager.exchange_join_code(code))["success"] is True
1960
1961    # The third failure trips the threshold, throttling further attempts.
1962    assert (await auth_manager.exchange_join_code("NOPE12345678"))["success"] is False
1963    result = await auth_manager.exchange_join_code(code)
1964    assert result["success"] is False
1965    assert "too many" in result["error"].lower()
1966
1967
1968async def test_exchange_join_code_rate_limit_is_per_connection(
1969    auth_manager: AuthenticationManager,
1970) -> None:
1971    """
1972    Verify one throttled client does not lock out the other clients.
1973
1974    At a party every guest exchanges a join code over their own connection, so a guest
1975    re-scanning an expired QR code must only ever throttle their own connection.
1976
1977    :param auth_manager: AuthenticationManager instance.
1978    """
1979    user = await auth_manager.create_user(username="partyguest", role=UserRole.GUEST)
1980    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
1981
1982    # One guest burns through the progressive delay threshold with a stale code.
1983    set_current_client_id("connection-a")
1984    for _ in range(3):
1985        assert (await auth_manager.exchange_join_code("EXPIRED12345"))["success"] is False
1986    result = await auth_manager.exchange_join_code("EXPIRED12345")
1987    assert "too many" in result["error"].lower()
1988
1989    # A second guest is unaffected, both for a bad code and for a valid one.
1990    set_current_client_id("connection-b")
1991    result = await auth_manager.exchange_join_code("EXPIRED12345")
1992    assert "invalid" in result["error"].lower()
1993    assert (await auth_manager.exchange_join_code(code))["success"] is True
1994
1995
1996async def test_exchange_join_code_success_clears_connection_rate_limit(
1997    auth_manager: AuthenticationManager,
1998) -> None:
1999    """
2000    Verify a successful exchange clears the counter of that connection only.
2001
2002    A per-connection counter can only be cleared by the connection that filled it, so a
2003    valid code holder cannot lift the throttle for anyone else.
2004
2005    :param auth_manager: AuthenticationManager instance.
2006    """
2007    user = await auth_manager.create_user(username="clearingguest", role=UserRole.GUEST)
2008    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
2009
2010    set_current_client_id("connection-c")
2011    for _ in range(2):
2012        assert (await auth_manager.exchange_join_code("MISTYPED1234"))["success"] is False
2013
2014    assert (await auth_manager.exchange_join_code(code))["success"] is True
2015
2016    # Without the reset, the fourth failure overall would trip the threshold.
2017    for _ in range(2):
2018        result = await auth_manager.exchange_join_code("MISTYPED1234")
2019        assert "invalid" in result["error"].lower()
2020
2021    # The shared bucket of connection-less callers is untouched by all of this.
2022    set_current_client_id(None)
2023    result = await auth_manager.exchange_join_code("MISTYPED1234")
2024    assert "invalid" in result["error"].lower()
2025
2026
2027async def test_exchange_join_code_global_ceiling_throttles_every_client(
2028    auth_manager: AuthenticationManager,
2029) -> None:
2030    """
2031    Verify the server-wide ceiling still backstops the per-connection counters.
2032
2033    A client can open a fresh connection (and thus a fresh counter) at will, so the
2034    ceiling is what bounds sustained abuse.
2035
2036    :param auth_manager: AuthenticationManager instance.
2037    """
2038    user = await auth_manager.create_user(username="ceilingguest", role=UserRole.GUEST)
2039    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
2040
2041    for _ in range(JOIN_CODE_GLOBAL_FAILURE_CEILING):
2042        await auth_manager._join_code_global_rate_limiter.record_failed_attempt(
2043            JOIN_CODE_GLOBAL_RATE_LIMIT_KEY
2044        )
2045
2046    # An untouched connection is throttled, even when it presents a valid code.
2047    set_current_client_id("connection-never-seen-before")
2048    result = await auth_manager.exchange_join_code(code)
2049    assert result["success"] is False
2050    assert "too many" in result["error"].lower()
2051
2052
2053async def test_exchange_join_code_rate_limit_falls_back_to_peer_address(
2054    auth_manager: AuthenticationManager,
2055) -> None:
2056    """
2057    Verify stateless API callers are told apart by the address they connect from.
2058
2059    The login page exchanges join codes over the JSON RPC endpoint, which carries no
2060    connection identity, so without this every such caller would share one bucket.
2061
2062    :param auth_manager: AuthenticationManager instance.
2063    """
2064    user = await auth_manager.create_user(username="httpguest", role=UserRole.GUEST)
2065    code, _ = await auth_manager.generate_join_code(user=user, expires_in_hours=24, max_uses=0)
2066
2067    set_current_peer_address("192.0.2.10")
2068    for _ in range(3):
2069        assert (await auth_manager.exchange_join_code("EXPIRED12345"))["success"] is False
2070    assert "too many" in (await auth_manager.exchange_join_code("EXPIRED12345"))["error"].lower()
2071
2072    # A caller from another address is unaffected.
2073    set_current_peer_address("192.0.2.11")
2074    assert (await auth_manager.exchange_join_code(code))["success"] is True
2075
2076    # An address can be shared by everyone behind a proxy, so success must not clear it.
2077    set_current_peer_address("192.0.2.10")
2078    assert "too many" in (await auth_manager.exchange_join_code(code))["error"].lower()
2079
2080    # A websocket client id always wins over the peer address.
2081    set_current_client_id("connection-e")
2082    assert (await auth_manager.exchange_join_code(code))["success"] is True
2083
2084
2085def test_mask_join_code() -> None:
2086    """Verify the log mask keeps a correlatable prefix but no usable code."""
2087    assert _mask_join_code("abcdefghjklm") == "ABCD********"
2088    assert _mask_join_code("abc") == "ABC"
2089
2090
2091async def test_exchange_join_code_logs_identify_the_caller(
2092    auth_manager: AuthenticationManager,
2093    caplog: pytest.LogCaptureFixture,
2094) -> None:
2095    """
2096    Verify a support log can tell a rejected code apart from a throttled client.
2097
2098    :param auth_manager: AuthenticationManager instance.
2099    :param caplog: Log capture fixture.
2100    """
2101    set_current_client_id("connection-d")
2102    for _ in range(3):
2103        await auth_manager.exchange_join_code("MISTYPED1234")
2104    rejections = [record.getMessage() for record in caplog.records if "rejected" in record.message]
2105    assert len(rejections) == 3
2106    assert "client=connection-d" in rejections[0]
2107    # the attempted code is only ever logged masked
2108    assert "code=MIST********" in rejections[0]
2109    assert "MISTYPED1234" not in rejections[0]
2110
2111    caplog.clear()
2112    await auth_manager.exchange_join_code("MISTYPED1234")
2113    throttles = [record.getMessage() for record in caplog.records if "throttled" in record.message]
2114    assert len(throttles) == 1
2115    assert "throttled by the client limit" in throttles[0]
2116    assert "client=connection-d" in throttles[0]
2117    assert "client_failures=3" in throttles[0]
2118    assert "server_failures=3" in throttles[0]
2119
2120
2121async def test_login_rate_limiter_default_tiers() -> None:
2122    """Verify the default progressive delays escalate with the failed attempt count."""
2123    limiter = LoginRateLimiter()
2124
2125    assert limiter.get_attempt_count("bob") == 0
2126    for expected_count, expected_delay in ((2, 0), (3, 30), (6, 60), (10, 120), (15, 300)):
2127        while limiter.get_attempt_count("bob") < expected_count:
2128            await limiter.record_failed_attempt("bob")
2129        assert limiter.get_delay("bob") == expected_delay
2130
2131    await limiter.clear_attempts("bob")
2132    assert limiter.get_attempt_count("bob") == 0
2133    assert limiter.get_delay("bob") == 0
2134
2135
2136async def test_login_rate_limiter_custom_tiers() -> None:
2137    """Verify a limiter can be configured with its own threshold and delay."""
2138    limiter = LoginRateLimiter(delay_tiers=((3, 60),))
2139
2140    for _ in range(2):
2141        await limiter.record_failed_attempt("key")
2142    assert await limiter.check_rate_limit("key") == (True, 0)
2143
2144    await limiter.record_failed_attempt("key")
2145    allowed, remaining_delay = await limiter.check_rate_limit("key")
2146    assert allowed is False
2147    assert 0 < remaining_delay <= 60
2148
2149
2150async def test_login_rate_limiter_drops_attempts_outside_window() -> None:
2151    """Verify attempts older than the tracking window stop counting."""
2152    limiter = LoginRateLimiter(tracking_window=timedelta(seconds=0))
2153
2154    await limiter.record_failed_attempt("key")
2155    assert limiter.get_attempt_count("key") == 0
2156
2157
2158async def test_login_rate_limiter_prunes_expired_keys() -> None:
2159    """
2160    Verify expired keys do not pile up when they are never used again.
2161
2162    Keys are one-off by nature (a connection that gives up, a made-up username), so
2163    without the sweep the bookkeeping would grow for as long as the server runs.
2164    """
2165    limiter = LoginRateLimiter(tracking_window=timedelta(seconds=0))
2166
2167    for index in range(PRUNE_THRESHOLD + 1):
2168        await limiter.record_failed_attempt(f"key{index}")
2169
2170    # Every attempt is already outside this limiter's window, so the sweep drops them all.
2171    assert len(limiter._failed_attempts) == 0
2172
2173    live = LoginRateLimiter()
2174    for index in range(PRUNE_THRESHOLD + 1):
2175        await live.record_failed_attempt(f"key{index}")
2176
2177    # Attempts inside the tracking window are never swept away.
2178    assert len(live._failed_attempts) == PRUNE_THRESHOLD + 1
2179
2180
2181async def test_resolve_command_impersonation(auth_manager: AuthenticationManager) -> None:
2182    """Test resolving the impersonation argument of an incoming API command."""
2183    admin_user = await auth_manager.create_user(username="admin", role=UserRole.ADMIN)
2184    standard_user = await auth_manager.create_user(username="user_a", role=UserRole.USER)
2185    set_current_user(admin_user)
2186    set_impersonated_user(None)
2187
2188    # no user argument present is a no-op and leaves other args untouched
2189    args: dict[str, object] = {"queue_id": "abc"}
2190    assert await resolve_command_impersonation(auth_manager.mass, args) is None
2191    assert args == {"queue_id": "abc"}
2192
2193    # an empty string is deliberately treated as "no impersonation requested"
2194    # (optional fields in automations/scripts commonly template to an empty string)
2195    args = {"queue_id": "abc", "user": ""}
2196    assert await resolve_command_impersonation(auth_manager.mass, args) is None
2197    assert args == {"queue_id": "abc"}
2198
2199    # the user argument is popped and resolved (by username)
2200    args = {"queue_id": "abc", "user": "user_a"}
2201    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2202    assert resolved == standard_user
2203    assert args == {"queue_id": "abc"}
2204
2205    # the user argument is also resolved by user_id
2206    args = {"user": standard_user.user_id}
2207    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2208    assert resolved == standard_user
2209
2210    # username is accepted as (deprecated) alias for user
2211    args = {"username": "user_a"}
2212    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2213    assert resolved == standard_user
2214    assert args == {}
2215
2216    # the dict form with the builtin provider is equivalent to the plain string form
2217    args = {"user": {"provider": "builtin", "user_id": "user_a"}}
2218    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2219    assert resolved == standard_user
2220
2221    # a caller without the users.impersonate scope may not impersonate another user
2222    set_current_user(standard_user)
2223    with pytest.raises(InsufficientPermissions):
2224        await resolve_command_impersonation(auth_manager.mass, {"user": "admin"})
2225
2226
2227async def test_resolve_command_impersonation_provider_link(
2228    auth_manager: AuthenticationManager,
2229) -> None:
2230    """Test resolving the dict form of the user argument by provider link."""
2231    service_user = await auth_manager.create_user(username="ha_service", role=UserRole.SERVICE)
2232    linked_user = await auth_manager.create_user(username="linked", role=UserRole.USER)
2233    await auth_manager.link_user_to_provider(
2234        linked_user, AuthProviderType.HOME_ASSISTANT, "ha-user-1"
2235    )
2236    set_current_user(service_user)
2237    set_impersonated_user(None)
2238
2239    # a linked HA user resolves to the mapped MA user and pops the argument
2240    args: dict[str, object] = {
2241        "queue_id": "abc",
2242        "user": {"provider": "homeassistant", "user_id": "ha-user-1"},
2243    }
2244    resolved = await resolve_command_impersonation(auth_manager.mass, args)
2245    assert resolved == linked_user
2246    assert args == {"queue_id": "abc"}
2247
2248    # an unknown user is an error by default (required defaults to true)...
2249    args = {"user": {"provider": "homeassistant", "user_id": "ha-user-unknown"}}
2250    with pytest.raises(UserNotFoundError):
2251        await resolve_command_impersonation(auth_manager.mass, args)
2252
2253    # ...but softly resolves to None (no impersonation) when not required
2254    args = {"user": {"provider": "homeassistant", "user_id": "ha-user-unknown", "required": False}}
2255    assert await resolve_command_impersonation(auth_manager.mass, args) is None
2256    assert args == {}
2257
2258    # required also applies to the builtin provider
2259    args = {"user": {"provider": "builtin", "user_id": "nobody", "required": False}}
2260    assert await resolve_command_impersonation(auth_manager.mass, args) is None
2261    args = {"user": {"provider": "builtin", "user_id": "nobody"}}
2262    with pytest.raises(UserNotFoundError):
2263        await resolve_command_impersonation(auth_manager.mass, args)
2264
2265    # a malformed dict form is rejected (unknown provider, missing user_id, non-bool required);
2266    # notably an unknown provider must not fall back to builtin (enum coercion default)
2267    for malformed in (
2268        {"provider": "nonsense", "user_id": "ha-user-1"},
2269        {"provider": None, "user_id": "ha-user-1"},
2270        {"user_id": "ha-user-1"},
2271        {"provider": "homeassistant"},
2272        {"provider": "homeassistant", "user_id": ""},
2273        {"provider": "homeassistant", "user_id": "ha-user-1", "required": "yes"},
2274    ):
2275        with pytest.raises(InvalidDataError):
2276            await resolve_command_impersonation(auth_manager.mass, {"user": malformed})
2277
2278    # an empty dict is treated as "no impersonation requested"
2279    assert await resolve_command_impersonation(auth_manager.mass, {"user": {}}) is None
2280
2281    # resolving another user by provider link requires the users.impersonate scope
2282    standard_user = await auth_manager.create_user(username="plain", role=UserRole.USER)
2283    set_current_user(standard_user)
2284    with pytest.raises(InsufficientPermissions):
2285        await resolve_command_impersonation(
2286            auth_manager.mass, {"user": {"provider": "homeassistant", "user_id": "ha-user-1"}}
2287        )
2288
2289
2290def test_has_scope() -> None:
2291    """Test the scope check for each of the builtin user roles."""
2292
2293    def _user(role: str) -> User:
2294        return User(user_id="abc123", username="testuser", role=role)
2295
2296    # admin has all scopes through the wildcard
2297    assert has_scope(_user(UserRole.ADMIN), Scope.CONFIG_CORE_WRITE)
2298    assert has_scope(_user(UserRole.ADMIN), Scope.LIBRARY_MANAGE)
2299    # regular user
2300    assert has_scope(_user(UserRole.USER), Scope.LIBRARY_WRITE)
2301    assert has_scope(_user(UserRole.USER), Scope.CONFIG_CORE_READ)
2302    assert not has_scope(_user(UserRole.USER), Scope.CONFIG_CORE_WRITE)
2303    assert not has_scope(_user(UserRole.USER), Scope.USERS_IMPERSONATE)
2304    # guest
2305    assert has_scope(_user(UserRole.GUEST), Scope.LIBRARY_READ)
2306    assert not has_scope(_user(UserRole.GUEST), Scope.LIBRARY_WRITE)
2307    assert not has_scope(_user(UserRole.GUEST), Scope.CONFIG_CORE_READ)
2308    # service
2309    assert has_scope(_user(UserRole.SERVICE), Scope.USERS_IMPERSONATE)
2310    assert has_scope(_user(UserRole.SERVICE), Scope.USERS_READ)
2311    assert has_scope(_user(UserRole.SERVICE), Scope.CONFIG_PLAYERS_WRITE)
2312    assert not has_scope(_user(UserRole.SERVICE), Scope.CONFIG_CORE_WRITE)
2313    # reading user accounts does not imply managing them
2314    assert not has_scope(_user(UserRole.SERVICE), Scope.USERS_MANAGE)
2315    # an unknown (custom) role id is fail-closed and grants no scopes at all
2316    assert not has_scope(_user("some_future_role"), Scope.LIBRARY_READ)
2317
2318
2319async def test_homeassistant_system_user_may_read_users(
2320    auth_manager: AuthenticationManager,
2321) -> None:
2322    """
2323    Test that the Home Assistant integration may list users to resolve the calling user.
2324
2325    :param auth_manager: AuthenticationManager instance.
2326    """
2327    system_user = await auth_manager.get_homeassistant_system_user()
2328    standard_user = await auth_manager.create_user(username="user_a", role=UserRole.USER)
2329    guest_user = await auth_manager.create_user(username="guest_a", role=UserRole.GUEST)
2330    for command in (AuthenticationManager.list_users, AuthenticationManager.get_user):
2331        assert getattr(command, "api_required_scope", None) is Scope.USERS_READ
2332    assert has_scope(system_user, Scope.USERS_READ)
2333    # reading user accounts remains off limits for regular users and guests
2334    assert not has_scope(standard_user, Scope.USERS_READ)
2335    assert not has_scope(guest_user, Scope.USERS_READ)
2336
2337
2338async def test_homeassistant_system_user_has_service_role(
2339    auth_manager: AuthenticationManager,
2340) -> None:
2341    """Test that the Home Assistant system user is created with the service role."""
2342    system_user = await auth_manager.get_homeassistant_system_user()
2343    assert system_user.role == UserRole.SERVICE
2344
2345    # a pre-existing system user with the old user role is migrated to service
2346    await auth_manager.database.update(
2347        "users", {"user_id": system_user.user_id}, {"role": UserRole.USER.value}
2348    )
2349    await auth_manager._migrate_system_user_role()
2350    migrated_user = await auth_manager.get_user(system_user.user_id)
2351    assert migrated_user is not None
2352    assert migrated_user.role == UserRole.SERVICE
2353