music-assistant-server

1"""Authentication manager for Tidal integration."""
2
3import asyncio
4import base64
5import json
6import time
7from collections.abc import Callable
8from dataclasses import dataclass
9from typing import TYPE_CHECKING, Any, cast
10
11from music_assistant_models.errors import LoginFailed
12
13from music_assistant.helpers.app_vars import app_var
14
15from .constants import AUTH_SCOPE, AUTH_URL, SESSIONS_URL
16
17if TYPE_CHECKING:
18    from aiohttp import ClientResponse, ClientSession
19
20TOKEN_REFRESH_BUFFER = 60 * 7  # 7 minutes
21# Minimum time between two token refreshes, so that (concurrent) requests
22# hitting 401s cannot hammer the token endpoint with refresh calls.
23TOKEN_REFRESH_COOLDOWN = 30
24
25
26def _v2_client_credentials() -> tuple[str, str]:
27    """Return the (client_id, client_secret) of the Tidal v2 (device) client."""
28    return app_var("tidal_client_id_v2"), app_var("tidal_client_secret_v2")
29
30
31async def _read_json(response: ClientResponse, context: str) -> dict[str, Any]:
32    """
33    Parse an auth response body as JSON, raising LoginFailed on a non-JSON body.
34
35    :param response: The auth endpoint response to parse.
36    :param context: Short description of the request, used in the error message.
37    """
38    # A proxy/gateway error can carry an HTML body; content_type=None skips
39    # aiohttp's content-type check so such a failure surfaces as LoginFailed
40    # instead of a ContentTypeError escaping the login flow's error contract.
41    try:
42        return cast("dict[str, Any]", await response.json(content_type=None))
43    except (json.JSONDecodeError, UnicodeDecodeError) as err:
44        raise LoginFailed(f"{context} failed: HTTP {response.status} (non-JSON response)") from err
45
46
47def _basic_auth_headers(client_id: str, client_secret: str) -> dict[str, str]:
48    """Build the HTTP Basic auth + form headers for a token endpoint request."""
49    token = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
50    return {
51        "Authorization": f"Basic {token}",
52        "Content-Type": "application/x-www-form-urlencoded",
53    }
54
55
56@dataclass
57class TidalUser:
58    """Represent a Tidal user with their associated account information."""
59
60    user_id: str | None = None
61    country_code: str | None = None
62    session_id: str | None = None
63    profile_name: str | None = None
64    user_name: str | None = None
65    email: str | None = None
66
67
68class TidalAuthManager:
69    """Manager for Tidal authentication process."""
70
71    def __init__(
72        self,
73        http_session: ClientSession,
74        config_updater: Callable[[dict[str, Any]], None],
75        logger: Any,
76    ):
77        """Initialize Tidal auth manager."""
78        self.http_session = http_session
79        self.update_config = config_updater
80        self.logger = logger
81        self._auth_info: dict[str, Any] | None = None
82        self._refresh_lock = asyncio.Lock()
83        self._last_refresh: float = 0.0
84        self.user = TidalUser()
85
86    async def initialize(self, auth_data: str) -> bool:
87        """Initialize the auth manager with stored auth data."""
88        if not auth_data:
89            return False
90
91        # Parse stored auth data
92        try:
93            self._auth_info = json.loads(auth_data)
94        except json.JSONDecodeError as err:
95            self.logger.error("Invalid authentication data: %s", err)
96            return False
97
98        # Ensure we have a valid token
99        return await self.ensure_valid_token()
100
101    @property
102    def user_id(self) -> str | None:
103        """Return the current user ID."""
104        return self.user.user_id
105
106    @property
107    def country_code(self) -> str | None:
108        """Return the current country code."""
109        return self.user.country_code
110
111    @property
112    def session_id(self) -> str | None:
113        """Return the current session ID."""
114        return self.user.session_id
115
116    @property
117    def access_token(self) -> str | None:
118        """Return the current access token."""
119        return self._auth_info.get("access_token") if self._auth_info else None
120
121    async def ensure_valid_token(self) -> bool:
122        """Ensure we have a valid token, refresh if needed."""
123        if not self._auth_info:
124            return False
125
126        # Check if token is expired
127        expires_at = self._auth_info.get("expires_at", 0)
128        if expires_at > time.time() + TOKEN_REFRESH_BUFFER:
129            return True
130
131        # Need to refresh token
132        return await self.refresh_token()
133
134    async def refresh_token(self) -> bool:
135        """Refresh the auth token (single-flight, with a short cooldown)."""
136        async with self._refresh_lock:
137            # A refresh that just completed (e.g. by a concurrent request that
138            # hit the same 401) is considered good: don't hit the token
139            # endpoint again, let the caller retry with the current token.
140            if time.time() - self._last_refresh < TOKEN_REFRESH_COOLDOWN:
141                return True
142            if not await self._perform_refresh():
143                return False
144            self._last_refresh = time.time()
145            return True
146
147    async def update_user_info(self, user_info: dict[str, Any], session_id: str) -> None:
148        """Update user info from API response."""
149        # Update the TidalUser dataclass with values from API response
150        self.user = TidalUser(
151            user_id=user_info.get("id"),
152            country_code=user_info.get("countryCode"),
153            session_id=session_id,
154            profile_name=user_info.get("profileName"),
155            user_name=user_info.get("username"),
156        )
157
158    @staticmethod
159    async def start_device_login(http_session: ClientSession) -> dict[str, Any]:
160        """
161        Begin the Tidal device authorization flow.
162
163        Returns the device authorization response (device/user code, verification
164        URLs, poll interval and expiry) to show to the user and hand to
165        :meth:`poll_device_login`.
166
167        :param http_session: The shared aiohttp session to use for the request.
168        """
169        client_id, _ = _v2_client_credentials()
170        async with http_session.post(
171            f"{AUTH_URL}/device_authorization",
172            data={"client_id": client_id, "scope": AUTH_SCOPE},
173        ) as response:
174            device = await _read_json(response, "Device authorization")
175            if response.status != 200:
176                raise LoginFailed(f"Device authorization failed: {device}")
177        return device
178
179    @staticmethod
180    async def poll_device_login(
181        http_session: ClientSession, device: dict[str, Any]
182    ) -> dict[str, Any]:
183        """
184        Poll Tidal until the user approves the device code, returning the auth data.
185
186        Polls indefinitely at the server-advised interval; the caller bounds the wait
187        via the setup flow's step deadline (a fresh code is minted on expiry).
188
189        :param http_session: The shared aiohttp session to use for the requests.
190        :param device: The device authorization response from :meth:`start_device_login`.
191        """
192        client_id, client_secret = _v2_client_credentials()
193        headers = _basic_auth_headers(client_id, client_secret)
194        interval = int(device.get("interval", 2))
195        while True:
196            await asyncio.sleep(interval)
197            data = {
198                "client_id": client_id,
199                "device_code": device["deviceCode"],
200                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
201                "scope": AUTH_SCOPE,
202            }
203            async with http_session.post(
204                f"{AUTH_URL}/token", data=data, headers=headers
205            ) as response:
206                # A transient gateway error (5xx) must not abort a login the user
207                # may be mid-approval on; keep polling, the setup flow's step
208                # deadline bounds the wait.
209                if response.status >= 500:
210                    continue
211                token_data = await _read_json(response, "Device login")
212                if response.status == 200:
213                    return await TidalAuthManager._finalize_login(http_session, token_data)
214            # Anything other than the "keep waiting" signals is terminal. An
215            # expired_token is also non-terminal: Tidal's clock may declare the
216            # code expired moments before our step deadline does, and the setup
217            # flow re-mints a fresh code on that deadline, so keep polling
218            # rather than aborting a flow that is about to recover.
219            error = token_data.get("error")
220            if error == "slow_down":
221                # RFC 8628 section 3.5: increase the poll interval by 5 seconds.
222                interval += 5
223            elif error not in ("authorization_pending", "expired_token"):
224                raise LoginFailed(f"Device login failed: {token_data}")
225
226    async def _perform_refresh(self) -> bool:
227        """Perform the actual token refresh request."""
228        if not self._auth_info:
229            return False
230
231        refresh_token = self._auth_info.get("refresh_token")
232        if not refresh_token:
233            return False
234
235        # Always refresh against the v2 client: Tidal's token endpoint accepts
236        # refresh tokens issued to the previous client when presented with the
237        # new client credentials (live-verified with a pre-migration token).
238        client_id, client_secret = _v2_client_credentials()
239
240        data = {
241            "refresh_token": refresh_token,
242            "client_id": client_id,
243            "grant_type": "refresh_token",
244            "scope": AUTH_SCOPE,
245        }
246        headers = _basic_auth_headers(client_id, client_secret)
247
248        async with self.http_session.post(
249            f"{AUTH_URL}/token", data=data, headers=headers
250        ) as response:
251            if response.status != 200:
252                self.logger.error("Failed to refresh token: %s", await response.text())
253                return False
254
255            token_data = await response.json()
256
257            # Update auth info
258            self._auth_info["access_token"] = token_data["access_token"]
259            if "refresh_token" in token_data:
260                self._auth_info["refresh_token"] = token_data["refresh_token"]
261
262            # Update expiration
263            if "expires_in" in token_data:
264                self._auth_info["expires_at"] = time.time() + token_data["expires_in"]
265
266            # Store updated auth info
267            self.update_config(self._auth_info)
268
269            return True
270
271    @staticmethod
272    async def _finalize_login(
273        http_session: ClientSession, token_data: dict[str, Any]
274    ) -> dict[str, Any]:
275        """Validate device tokens, attach user/session info and an absolute expiry."""
276        if not token_data.get("access_token") or not token_data.get("refresh_token"):
277            raise LoginFailed("Failed to obtain authentication tokens from Tidal")
278
279        headers = {"Authorization": f"Bearer {token_data['access_token']}"}
280        async with http_session.get(SESSIONS_URL, headers=headers) as response:
281            if response.status != 200:
282                raise LoginFailed(f"Failed to get user info: {await response.text()}")
283            user_info = await response.json()
284
285        auth_data = {**token_data, **user_info}
286        auth_data["expires_at"] = time.time() + token_data.get("expires_in", 3600)
287        return auth_data
288