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