music-assistant-server

1"""Authentication manager for Tidal integration."""
2
3import asyncio
4import json
5import random
6import time
7import urllib
8from collections.abc import Callable
9from dataclasses import dataclass
10from typing import TYPE_CHECKING, Any
11
12import pkce
13from music_assistant_models.errors import LoginFailed
14
15from music_assistant.helpers.app_vars import app_var
16
17from .constants import AUTH_URL, LOGIN_URL, REDIRECT_URI, SESSIONS_URL
18
19if TYPE_CHECKING:
20    from aiohttp import ClientSession
21
22TOKEN_REFRESH_BUFFER = 60 * 7  # 7 minutes
23# Minimum time between two token refreshes, so that (concurrent) requests
24# hitting 401s cannot hammer the token endpoint with refresh calls.
25TOKEN_REFRESH_COOLDOWN = 30
26
27
28@dataclass
29class TidalUser:
30    """Represent a Tidal user with their associated account information."""
31
32    user_id: str | None = None
33    country_code: str | None = None
34    session_id: str | None = None
35    profile_name: str | None = None
36    user_name: str | None = None
37    email: str | None = None
38
39
40class TidalAuthManager:
41    """Manager for Tidal authentication process."""
42
43    def __init__(
44        self,
45        http_session: ClientSession,
46        config_updater: Callable[[dict[str, Any]], None],
47        logger: Any,
48    ):
49        """Initialize Tidal auth manager."""
50        self.http_session = http_session
51        self.update_config = config_updater
52        self.logger = logger
53        self._auth_info: dict[str, Any] | None = None
54        self._refresh_lock = asyncio.Lock()
55        self._last_refresh: float = 0.0
56        self.user = TidalUser()
57
58    async def initialize(self, auth_data: str) -> bool:
59        """Initialize the auth manager with stored auth data."""
60        if not auth_data:
61            return False
62
63        # Parse stored auth data
64        try:
65            self._auth_info = json.loads(auth_data)
66        except json.JSONDecodeError as err:
67            self.logger.error("Invalid authentication data: %s", err)
68            return False
69
70        # Ensure we have a valid token
71        return await self.ensure_valid_token()
72
73    @property
74    def user_id(self) -> str | None:
75        """Return the current user ID."""
76        return self.user.user_id
77
78    @property
79    def country_code(self) -> str | None:
80        """Return the current country code."""
81        return self.user.country_code
82
83    @property
84    def session_id(self) -> str | None:
85        """Return the current session ID."""
86        return self.user.session_id
87
88    @property
89    def access_token(self) -> str | None:
90        """Return the current access token."""
91        return self._auth_info.get("access_token") if self._auth_info else None
92
93    async def ensure_valid_token(self) -> bool:
94        """Ensure we have a valid token, refresh if needed."""
95        if not self._auth_info:
96            return False
97
98        # Check if token is expired
99        expires_at = self._auth_info.get("expires_at", 0)
100        if expires_at > time.time() + TOKEN_REFRESH_BUFFER:
101            return True
102
103        # Need to refresh token
104        return await self.refresh_token()
105
106    async def refresh_token(self) -> bool:
107        """Refresh the auth token (single-flight, with a short cooldown)."""
108        async with self._refresh_lock:
109            # A refresh that just completed (e.g. by a concurrent request that
110            # hit the same 401) is considered good: don't hit the token
111            # endpoint again, let the caller retry with the current token.
112            if time.time() - self._last_refresh < TOKEN_REFRESH_COOLDOWN:
113                return True
114            if not await self._perform_refresh():
115                return False
116            self._last_refresh = time.time()
117            return True
118
119    async def update_user_info(self, user_info: dict[str, Any], session_id: str) -> None:
120        """Update user info from API response."""
121        # Update the TidalUser dataclass with values from API response
122        self.user = TidalUser(
123            user_id=user_info.get("id"),
124            country_code=user_info.get("countryCode"),
125            session_id=session_id,
126            profile_name=user_info.get("profileName"),
127            user_name=user_info.get("username"),
128        )
129
130    @staticmethod
131    def build_pkce_login(quality: str) -> tuple[str, dict[str, Any]]:
132        """
133        Build a fresh PKCE login: the Tidal authorize URL and its matching auth params.
134
135        The returned auth_params dict carries the (single-use) PKCE code_verifier and must
136        be handed back to :meth:`process_pkce_login` to exchange the pasted redirect URL.
137
138        :param quality: The audio quality carried through to the token exchange.
139        """
140        code_verifier, code_challenge = pkce.generate_pkce_pair()
141        client_unique_key = format(random.getrandbits(64), "02x")
142        auth_params = {
143            "code_verifier": code_verifier,
144            "client_unique_key": client_unique_key,
145            "client_id": app_var("tidal_client_id"),
146            "client_secret": app_var("tidal_client_secret"),
147            "quality": quality,
148        }
149        params = {
150            "response_type": "code",
151            "redirect_uri": REDIRECT_URI,
152            "client_id": auth_params["client_id"],
153            "lang": "EN",
154            "appMode": "android",
155            "client_unique_key": client_unique_key,
156            "code_challenge": code_challenge,
157            "code_challenge_method": "S256",
158            "restrict_signup": "true",
159        }
160        authorize_url = f"{LOGIN_URL}?{urllib.parse.urlencode(params)}"
161        return authorize_url, auth_params
162
163    @staticmethod
164    async def process_pkce_login(
165        http_session: ClientSession, base64_auth_params: str, redirect_url: str
166    ) -> dict[str, Any]:
167        """Process TIDAL authentication with PKCE flow."""
168        # Parse the stored auth parameters
169        try:
170            auth_params = json.loads(base64_auth_params)
171        except json.JSONDecodeError as err:
172            raise LoginFailed("Invalid authentication data") from err
173
174        # Extract required parameters
175        code_verifier = auth_params.get("code_verifier")
176        client_unique_key = auth_params.get("client_unique_key")
177        client_secret = auth_params.get("client_secret")
178        client_id = auth_params.get("client_id")
179        quality = auth_params.get("quality")
180
181        if not code_verifier or not client_unique_key:
182            raise LoginFailed("Missing required authentication parameters")
183
184        # Extract the authorization code from the redirect URL
185        parsed_url = urllib.parse.urlparse(redirect_url)
186        query_params = urllib.parse.parse_qs(parsed_url.query)
187        code = query_params.get("code", [""])[0]
188
189        if not code:
190            raise LoginFailed("No authorization code found in redirect URL")
191
192        # Prepare the token exchange request
193        token_url = f"{AUTH_URL}/token"
194        data = {
195            "code": code,
196            "client_id": client_id,
197            "grant_type": "authorization_code",
198            "redirect_uri": REDIRECT_URI,
199            "scope": "r_usr w_usr w_sub",
200            "code_verifier": code_verifier,
201            "client_unique_key": client_unique_key,
202        }
203
204        headers = {"Content-Type": "application/x-www-form-urlencoded"}
205
206        # Make the token exchange request
207        async with http_session.post(token_url, data=data, headers=headers) as response:
208            if response.status != 200:
209                error_text = await response.text()
210                raise LoginFailed(f"Token exchange failed: {error_text}")
211
212            token_data = await response.json()
213
214        # Validate we have authentication data
215        if not token_data.get("access_token") or not token_data.get("refresh_token"):
216            raise LoginFailed("Failed to obtain authentication tokens from Tidal")
217
218        # Get user information using the new token
219        headers = {"Authorization": f"Bearer {token_data['access_token']}"}
220
221        # Again use mass.http_session
222        async with http_session.get(SESSIONS_URL, headers=headers) as response:
223            if response.status != 200:
224                error_text = await response.text()
225                raise LoginFailed(f"Failed to get user info: {error_text}")
226
227            user_info = await response.json()
228
229        # Combine token and user info, add expiration time
230        auth_data = {**token_data, **user_info}
231
232        # Add standard fields used by TidalProvider
233        auth_data["expires_at"] = time.time() + token_data.get("expires_in", 3600)
234        auth_data["quality"] = quality
235        auth_data["client_id"] = client_id
236        auth_data["client_secret"] = client_secret
237
238        return auth_data
239
240    async def _perform_refresh(self) -> bool:
241        """Perform the actual token refresh request."""
242        if not self._auth_info:
243            return False
244
245        refresh_token = self._auth_info.get("refresh_token")
246        if not refresh_token:
247            return False
248
249        client_id = self._auth_info.get("client_id", app_var("tidal_client_id"))
250
251        data = {
252            "refresh_token": refresh_token,
253            "client_id": client_id,
254            "grant_type": "refresh_token",
255            "scope": "r_usr w_usr w_sub",
256        }
257
258        headers = {"Content-Type": "application/x-www-form-urlencoded"}
259
260        async with self.http_session.post(
261            f"{AUTH_URL}/token", data=data, headers=headers
262        ) as response:
263            if response.status != 200:
264                self.logger.error("Failed to refresh token: %s", await response.text())
265                return False
266
267            token_data = await response.json()
268
269            # Update auth info
270            self._auth_info["access_token"] = token_data["access_token"]
271            if "refresh_token" in token_data:
272                self._auth_info["refresh_token"] = token_data["refresh_token"]
273
274            # Update expiration
275            if "expires_in" in token_data:
276                self._auth_info["expires_at"] = time.time() + token_data["expires_in"]
277
278            # Store updated auth info
279            self.update_config(self._auth_info)
280
281            return True
282