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