/
/
/
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# Owner id prefixes, naming the kind of authorization a credential is bound to
24GUEST_OWNER_PREFIX = "guest-"
25USER_OWNER_PREFIX = "user-"
26
27
28async def get_or_create_guest_user(mass: MusicAssistant, username: str, display_name: str) -> User:
29 """
30 Get the guest user with the given username, creating it if needed.
31
32 :param mass: MusicAssistant instance.
33 :param username: Unique username for the guest account.
34 :param display_name: Human readable display name for the guest account.
35 :return: The (existing or newly created) guest User.
36 :raises InvalidDataError: If the username belongs to a non-guest account.
37 """
38 auth = mass.webserver.auth
39 if user := await auth.get_user_by_username(username):
40 # never hand out guest access on a higher privileged account
41 if user.role != UserRole.GUEST:
42 raise InvalidDataError(f"User {username} exists but is not a guest account")
43 return user
44 return await auth.create_user(
45 username=username,
46 role=UserRole.GUEST,
47 display_name=display_name,
48 )
49
50
51async def get_or_create_join_code(
52 mass: MusicAssistant,
53 user: User,
54 *,
55 expires_in_hours: int = DEFAULT_JOIN_CODE_EXPIRY_HOURS,
56 max_uses: int = 0,
57 device_name: str = "Guest",
58) -> str:
59 """
60 Get an active join code for the given guest user, creating one if needed.
61
62 :param mass: MusicAssistant instance.
63 :param user: The guest user the join code belongs to.
64 :param expires_in_hours: Hours until a newly created code expires.
65 :param max_uses: Maximum number of uses for a newly created code (0 = unlimited).
66 :param device_name: Device name for tokens created with the code.
67 :return: The active join code string.
68 """
69 auth = mass.webserver.auth
70 if existing_code := await auth.get_active_join_code(user):
71 return existing_code
72 code, _expires_at = await auth.generate_join_code(
73 user=user,
74 expires_in_hours=expires_in_hours,
75 max_uses=max_uses,
76 device_name=device_name,
77 )
78 return code
79
80
81def build_join_url(mass: MusicAssistant, code: str) -> str:
82 """
83 Build the URL guests open to join with the given join code.
84
85 When remote access is enabled, returns a URL that works from anywhere via
86 WebRTC. Otherwise, returns a local URL that only works on the same network.
87
88 :param mass: MusicAssistant instance.
89 :param code: The join code to embed in the URL.
90 :return: The guest join URL.
91 """
92 remote_access = mass.webserver.remote_access
93 if remote_access.is_enabled and remote_access.remote_id:
94 return f"https://app.music-assistant.io/?remote_id={remote_access.remote_id}&join={code}"
95 base_url = mass.webserver.base_url
96 assert base_url # for type-checker only
97 return f"{base_url}/?join={code}"
98
99
100def credential_owner(user: User) -> str:
101 """
102 Return the owner id that credentials minted by this user are bound to.
103
104 The prefix encodes the credential's lifetime: a guest's credentials end with
105 their session or access, a full user's with their account.
106
107 :param user: The authenticated user minting the credential.
108 """
109 guest_owner, user_owner = credential_owners_for_user_id(user.user_id)
110 return guest_owner if user.role == UserRole.GUEST else user_owner
111
112
113def credential_owners_for_user_id(user_id: str) -> tuple[str, str]:
114 """
115 Return every owner id credentials of the given user account may be bound to.
116
117 :param user_id: The user id whose credentials are looked up.
118 """
119 return (f"{GUEST_OWNER_PREFIX}{user_id}", f"{USER_OWNER_PREFIX}{user_id}")
120
121
122def credential_owner_user_id(owner: str) -> str | None:
123 """
124 Return the user account an owner id is bound to, or ``None`` for another owner kind.
125
126 :param owner: An owner id produced by credential_owner.
127 """
128 for prefix in (GUEST_OWNER_PREFIX, USER_OWNER_PREFIX):
129 if owner.startswith(prefix):
130 return owner.removeprefix(prefix)
131 return None
132
133
134def is_session_scoped_owner(owner: str) -> bool:
135 """
136 Return whether credentials bound to this owner end with the owner's session.
137
138 :param owner: An owner id produced by credential_owner.
139 """
140 return owner.startswith(GUEST_OWNER_PREFIX)
141
142
143async def revoke_guest_access(mass: MusicAssistant, username: str) -> tuple[int, int]:
144 """
145 Revoke all join codes and auth tokens for the guest user with the given username.
146
147 Active WebSocket connections of the guest user are disconnected so guests
148 are immediately logged out and can not reconnect.
149
150 :param mass: MusicAssistant instance.
151 :param username: Username of the guest account to revoke access for.
152 :return: Tuple of (number of join codes revoked, number of tokens revoked).
153 """
154 auth = mass.webserver.auth
155 if not (user := await auth.get_user_by_username(username)):
156 return (0, 0)
157 codes_revoked = await auth.revoke_join_codes(user)
158 tokens_revoked = await auth.revoke_tokens_for_user(user)
159 return (codes_revoked, tokens_revoked)
160