From ff99475fab4e3f43d68316257dd6cd78404d0ef0 Mon Sep 17 00:00:00 2001 From: kuelteabmas Date: Sun, 18 May 2025 20:05:21 -0400 Subject: [PATCH] bugfix(spotify): Updated functionality to fetch access_token and client_id required for making requests * Removed deprecated session_data and its logic * Added TOTP for new _get_access_token_and_client_id method * Added some loggers --- app/__init__.py | 1 + app/providers/spotify.py | 96 ++++++++++++++++++++++++++++++++-------- 2 files changed, 78 insertions(+), 19 deletions(-) diff --git a/app/__init__.py b/app/__init__.py index a8e75bf..df04686 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -203,6 +203,7 @@ else: spotify_client = SpotifyClient() spotify_client.authenticate() +app.logger.info('spotify auth successful') from .registry import MusicProviderRegistry MusicProviderRegistry.register_provider(spotify_client) diff --git a/app/providers/spotify.py b/app/providers/spotify.py index a94aa0d..ac75a71 100644 --- a/app/providers/spotify.py +++ b/app/providers/spotify.py @@ -1,4 +1,3 @@ -from dataclasses import dataclass import os from app.providers.base import AccountAttributes, Album, Artist, BrowseCard, BrowseSection, Image, MusicProviderClient, Owner, Playlist, PlaylistTrack, Profile, Track, ExternalUrl, Category import base64 @@ -11,8 +10,38 @@ from typing import List, Dict, Optional from http.cookiejar import MozillaCookieJar import logging +from typing import Callable, Tuple +import hmac +import time +import hashlib + l = logging.getLogger(__name__) +_TOTP_SECRET = bytearray([53,53,48,55,49,52,53,56,53,51,52,56,55,52,57,57,53,57,50,50,52,56,54,51,48,51,50,57,51,52,55]) + +def generate_totp( + secret: bytes = _TOTP_SECRET, + algorithm: Callable[[], object] = hashlib.sha1, + digits: int = 6, + counter_factory: Callable[[], int] = lambda: int(time.time()) // 30, +) -> Tuple[str, int]: + counter = counter_factory() + hmac_result = hmac.new( + secret, counter.to_bytes(8, byteorder="big"), algorithm # type: ignore + ).digest() + + offset = hmac_result[-1] & 15 + truncated_value = ( + (hmac_result[offset] & 127) << 24 + | (hmac_result[offset + 1] & 255) << 16 + | (hmac_result[offset + 2] & 255) << 8 + | (hmac_result[offset + 3] & 255) + ) + return ( + str(truncated_value % (10**digits)).zfill(digits), + counter * 30_000, + ) # (30 * 1000) + class SpotifyClient(MusicProviderClient): """ Spotify implementation of the MusicProviderClient. @@ -23,9 +52,10 @@ class SpotifyClient(MusicProviderClient): def __init__(self, cookie_file: Optional[str] = None): self.base_url = "https://api-partner.spotify.com" - self.session_data = None self.app_server_config_data = None self.client_token = None + self.access_token = None + self.client_id = None self.cookies = None if cookie_file: self._load_cookies(cookie_file) @@ -52,16 +82,19 @@ class SpotifyClient(MusicProviderClient): """ if self.cookies: l.debug("Authenticating using cookies.") - self.session_data, self.app_server_config_data = self._fetch_session_data() - self.client_token = self._fetch_client_token() + self.app_server_config_data = self._fetch_app_server_config_data() else: l.debug("Authenticating without cookies.") - self.session_data, self.app_server_config_data = self._fetch_session_data(fetch_with_cookies=False) - self.client_token = self._fetch_client_token() + self.app_server_config_data = self._fetch_app_server_config_data(fetch_with_cookies=False) - def _fetch_session_data(self, fetch_with_cookies: bool = True): + self._get_access_token_and_client_id() + self.client_token = self._fetch_client_token() + + def _fetch_app_server_config_data(self, fetch_with_cookies: bool = True): """ - Fetch session data from Spotify. + Fetch app_server_config data from Spotify. + + We will use the correlationId from app_server_config_data for a successful _fetch_client_token call :param fetch_with_cookies: Whether to include cookies in the request. :return: Tuple containing session and config data. @@ -75,15 +108,14 @@ class SpotifyClient(MusicProviderClient): response = requests.get(url, headers=headers, cookies=cookies) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') - session_script = soup.find('script', {'id': 'session'}) app_server_config_script = soup.find('script', {'id': 'appServerConfig'}).text # decode JWT to obtain correlation required for obtain Bearer Token decoded_app_server_config_script = base64.b64decode(app_server_config_script) # base64 decode decoded_app_server_config_script = decoded_app_server_config_script.decode().strip("b'") # decode from byte object to string object - if session_script.string and decoded_app_server_config_script: - l.debug("fetched session and config scripts") - return json.loads(session_script.string, decoded_app_server_config_script) + if decoded_app_server_config_script: + l.debug("fetched app_server_config_script scripts") + return json.loads(decoded_app_server_config_script) else: - raise ValueError("Failed to fetch session or config data.") + raise ValueError("Failed to fetch app_server_config data.") def _fetch_client_token(self): """ @@ -101,7 +133,7 @@ class SpotifyClient(MusicProviderClient): payload = { "client_data": { "client_version": "1.2.52.404.gcb99a997", - "client_id": self.session_data.get("clientId", ""), + "client_id": self.client_id, "js_sdk_data": { "device_brand": "unknown", "device_model": "unknown", @@ -114,9 +146,36 @@ class SpotifyClient(MusicProviderClient): } response = requests.post(url, headers=headers, json=payload, cookies=self.cookies) response.raise_for_status() - l.debug("fetched granted_token") + l.debug("fetched client_token (granted_token)") return response.json().get("granted_token", "") + def _get_access_token_and_client_id(self, fetch_with_cookies: bool = True): + """ + Fetch the Access Token and Client ID by making GET call to open.spotify.com/get_access_token + """ + url = f'https://open.spotify.com/get_access_token' + headers = { + 'accept': 'application/json', + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36', + } + + totp, timestamp = generate_totp() + + query_params = { + "reason": "init", + "productType": "web-player", + "totp": totp, + "totpVer": 5, + "ts": timestamp, + } + cookies = self.cookies if fetch_with_cookies else None + + response = requests.get(url, params=query_params, headers=headers, cookies=self.cookies) + response.raise_for_status() + self.client_id = response.json().get("clientId", "") + self.access_token = response.json().get("accessToken", "") + l.debug("fetched access_token and client_id") + def _make_request(self, endpoint: str, params: dict = None) -> dict: """ Helper method to make authenticated requests to Spotify APIs. @@ -124,16 +183,15 @@ class SpotifyClient(MusicProviderClient): headers = { 'accept': 'application/json', 'app-platform': 'WebPlayer', - 'authorization': f'Bearer {self.session_data.get("accessToken", "")}', - 'client-token': self.client_token.get('token',''), + 'authorization': f'Bearer {self.access_token}', + 'client-token': self.client_token.get('token','') } l.debug(f"starting request: {self.base_url}/{endpoint}") response = requests.get(f"{self.base_url}/{endpoint}", headers=headers, params=params, cookies=self.cookies) # if the response is unauthorized, we need to reauthenticate if response.status_code == 401: l.debug("reauthenticating") - self.authenticate() - headers['authorization'] = f'Bearer {self.session_data.get("accessToken", "")}' + headers['authorization'] = f'Bearer {self.access_token}' headers['client-token'] = self.client_token.get('token','') response = requests.get(f"{self.base_url}/{endpoint}", headers=headers, params=params, cookies=self.cookies)