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