/
/
/
1"""
2Helpers for guest access to Music Assistant.
3
4Provides the shared building blocks for plugins that offer a guest experience
5(e.g. the party plugin): a dedicated guest user account, join codes and the
6join URL that guests open on their own device.
7"""
8
9from __future__ import annotations
10
11from typing import TYPE_CHECKING
12
13from music_assistant_models.auth import UserRole
14from music_assistant_models.errors import InvalidDataError
15
16if TYPE_CHECKING:
17 from music_assistant_models.auth import User
18
19 from music_assistant.mass import MusicAssistant
20
21DEFAULT_JOIN_CODE_EXPIRY_HOURS = 8
22
23
24async def get_or_create_guest_user(mass: MusicAssistant, username: str, display_name: str) -> User:
25 """
26 Get the guest user with the given username, creating it if needed.
27
28 :param mass: MusicAssistant instance.
29 :param username: Unique username for the guest account.
30 :param display_name: Human readable display name for the guest account.
31 :return: The (existing or newly created) guest User.
32 :raises InvalidDataError: If the username belongs to a non-guest account.
33 """
34 auth = mass.webserver.auth
35 if user := await auth.get_user_by_username(username):
36 # never hand out guest access on a higher privileged account
37 if user.role != UserRole.GUEST:
38 raise InvalidDataError(f"User {username} exists but is not a guest account")
39 return user
40 return await auth.create_user(
41 username=username,
42 role=UserRole.GUEST,
43 display_name=display_name,
44 )
45
46
47async def get_or_create_join_code(
48 mass: MusicAssistant,
49 user: User,
50 *,
51 expires_in_hours: int = DEFAULT_JOIN_CODE_EXPIRY_HOURS,
52 max_uses: int = 0,
53 device_name: str = "Guest",
54) -> str:
55 """
56 Get an active join code for the given guest user, creating one if needed.
57
58 :param mass: MusicAssistant instance.
59 :param user: The guest user the join code belongs to.
60 :param expires_in_hours: Hours until a newly created code expires.
61 :param max_uses: Maximum number of uses for a newly created code (0 = unlimited).
62 :param device_name: Device name for tokens created with the code.
63 :return: The active join code string.
64 """
65 auth = mass.webserver.auth
66 if existing_code := await auth.get_active_join_code(user):
67 return existing_code
68 code, _expires_at = await auth.generate_join_code(
69 user=user,
70 expires_in_hours=expires_in_hours,
71 max_uses=max_uses,
72 device_name=device_name,
73 )
74 return code
75
76
77def build_join_url(mass: MusicAssistant, code: str) -> str:
78 """
79 Build the URL guests open to join with the given join code.
80
81 When remote access is enabled, returns a URL that works from anywhere via
82 WebRTC. Otherwise, returns a local URL that only works on the same network.
83
84 :param mass: MusicAssistant instance.
85 :param code: The join code to embed in the URL.
86 :return: The guest join URL.
87 """
88 remote_access = mass.webserver.remote_access
89 if remote_access.is_enabled and remote_access.remote_id:
90 return f"https://app.music-assistant.io/?remote_id={remote_access.remote_id}&join={code}"
91 base_url = mass.webserver.base_url
92 assert base_url # for type-checker only
93 return f"{base_url}/?join={code}"
94
95
96async def revoke_guest_access(mass: MusicAssistant, username: str) -> tuple[int, int]:
97 """
98 Revoke all join codes and auth tokens for the guest user with the given username.
99
100 Active WebSocket connections of the guest user are disconnected so guests
101 are immediately logged out and can not reconnect.
102
103 :param mass: MusicAssistant instance.
104 :param username: Username of the guest account to revoke access for.
105 :return: Tuple of (number of join codes revoked, number of tokens revoked).
106 """
107 auth = mass.webserver.auth
108 if not (user := await auth.get_user_by_username(username)):
109 return (0, 0)
110 codes_revoked = await auth.revoke_join_codes(user)
111 tokens_revoked = await auth.revoke_tokens_for_user(user)
112 return (codes_revoked, tokens_revoked)
113