Merge pull request #1 from cynthia2006/main

Flexible secrets URL & CDRM Integration fixes

README.md

@@ -22,13 +22,13 @@ A command-line app for downloading songs, podcasts and videos from Spotify.
 - **FFmpeg** on your system PATH. Use one of the recommended builds:
   - **Windows**: [AnimMouse's FFmpeg Builds](https://github.com/AnimMouse/ffmpeg-stable-autobuild/releases).
   - **Linux**: [John Van Sickle's FFmpeg Builds](https://johnvansickle.com/ffmpeg/).
-- A **.wvd file**.
-  - A .wvd file contains the Widevine keys from a device and is required to decrypt music videos and songs in AAC. The easiest method of obtaining one is using KeyDive, which extracts it from an Android device. Detailed instructions can be found here: https://github.com/hyugogirubato/KeyDive. **.wvd files extracted from emulated devices may not work**.
+- **(Optional)** A **.wvd file**.
+  - A `.wvd` file contains the Widevine keys from a device and is required to decrypt music videos and songs in AAC. The easiest method of obtaining one is using KeyDive, which extracts it from an Android device. Detailed instructions can be found here: https://github.com/hyugogirubato/KeyDive. **.wvd files extracted from emulated devices may not work**.
 
 #### Notes
 
 - **Some users have reported that Spotify suspended their accounts after using Votify**. Use it at your own risk.
-- The .wvd file is not required if you plan on only downloading podcasts and can be skipped by enabling the `disable_wvd` option.
+- The .wvd file is not required, but expect latency otherwise.
 - FFmpeg is not required if you plan on only downloading podcasts in Vorbis, but it's needed for downloading podcasts in AAC.
 
 ### Optional dependencies
@@ -51,7 +51,8 @@ The following tools are optional but required for specific features. Add them to
 2. Set up the cookies file.
    - Move the cookies file to the directory where you'll run Votify and rename it to `cookies.txt`.
    - Alternatively, specify the path to the cookies file using command-line arguments or the config file.
-3. Set up the .wvd file.
+3. **(Optional**) Set up the .wvd file.
+   
    - Move the .wvd file file to the directory where you'll run Votify and rename it to `device.wvd`.
    - Alternatively, specify the path to the .wvd file using command-line arguments or the config file.
 

votify/cli.py

@@ -227,6 +227,12 @@ def load_config_file(
     default=downloader_sig.parameters["packager_path"].default,
     help="Path to Shaka Packager binary.",
 )
+@click.option(
+    "--spotify-secrets-url",
+    type=str,
+    default="https://code.thetadev.de/ThetaDev/spotify-secrets/raw/branch/main/secrets/secretDict.json",
+    help="Spotify secrets for TOTP generation"
+)
 @click.option(
     "--template-folder-album",
     type=str,
@@ -395,6 +401,7 @@ def main(
     mp4box_path: str,
     mp4decrypt_path: str,
     packager_path: str,
+    spotify_secrets_url: str,
     template_folder_album: str,
     template_folder_compilation: str,
     template_file_single_disc: str,
@@ -428,7 +435,7 @@ def main(
     cookies_path = prompt_path(True, cookies_path, "Cookies file")
 
     logger.info("Starting Votify")
-    spotify_api = SpotifyApi.from_cookies_file(cookies_path)
+    spotify_api = SpotifyApi.from_cookies_file(cookies_path, secrets_url=spotify_secrets_url)
 
     downloader = Downloader(
         spotify_api,

votify/downloader.py

@@ -140,8 +140,6 @@ class Downloader:
     def set_cdm(self) -> None:
         if self.wvd_path.exists():
             self.cdm = Cdm.from_device(Device.load(self.wvd_path))
-        else:
-            self.cdm = None
 
     def get_url_info(self, url: str) -> UrlInfo:
         url_regex_result = re.search(self.URL_RE, url)
@@ -735,8 +733,8 @@ class Downloader:
         media_type: str,
     ) -> tuple[str, str]:
         try:
-            if self.cdm == None:
-                cmd = self.spotify_api.wvd_cdrm(pssh)
+            if self.cdm is None:
+                cmd = self.spotify_api.extract_keys_with_cdrm(pssh, media_type)
                 key_id, decryption_key = cmd.split(':')
             else:
                 pssh = PSSH(pssh)

votify/spotify_api.py

@@ -47,16 +47,24 @@ class SpotifyApi:
     CLIENT_TOKEN_URL = "https://clienttoken.spotify.com/v1/clienttoken"
 
     def __init__(
-        self,
+        self, *,
+        secrets_url: str,
         sp_dc: str | None = None,
         use_device_flow: bool = False,
     ) -> None:
+        self.session = requests.Session()
+
+        secrets = self.session.get(secrets_url)
+        check_response(secrets)
+        totp_version, secrets_ciphertext = max(secrets.json().items(), key=lambda item: int(item[0]))
+
+        self.totp = TOTP(version=totp_version, ciphertext=secrets_ciphertext)
         self.sp_dc = sp_dc
         self.use_device_flow = use_device_flow
         self._set_session()
 
     @classmethod
-    def from_cookies_file(cls, cookies_path: Path) -> SpotifyApi:
+    def from_cookies_file(cls, cookies_path: Path, **kwargs) -> SpotifyApi:
         cookies = MozillaCookieJar(cookies_path)
         cookies.load(ignore_discard=True, ignore_expires=True)
         parse_cookie = lambda name: next(
@@ -73,11 +81,9 @@ class SpotifyApi:
                 '"sp_dc" cookie not found in cookies. '
                 "Make sure you have exported the cookies from the Spotify homepage and are logged in."
             )
-        return cls(sp_dc=sp_dc)
+        return cls(sp_dc=sp_dc, **kwargs)
 
     def _set_session(self) -> None:
-        self.totp = TOTP()
-        self.session = requests.Session()
         self._setup_session_headers()
         self._setup_authorization()
         self._setup_user_profile()
@@ -235,14 +241,15 @@ class SpotifyApi:
         check_response(response)
         return response.json()
 
-    def wvd_cdrm(self, pssh):
+    def extract_keys_with_cdrm(self, pssh, media_type):
         cmd = self.session.post('https://cdrm-project.com/api/decrypt',
-        headers={'Content-Type': 'application/json'},
-        json={
-              'pssh': pssh,
-              'licurl': 'https://gue1-spclient.spotify.com/widevine-license/v1/audio/license',
-              'headers': str(self.session.headers)
-        }).json()
+            headers={'Accept': 'application/json',
+                     'Content-Type': 'application/json'},
+            json={
+                  'pssh': pssh,
+                  'licurl': self.WIDEVINE_LICENSE_API_URL.format(type=media_type),
+                  'headers': str(self.session.headers)
+            }).json()
         return cmd['message']
 
     def get_track(self, track_id: str) -> dict:

votify/totp.py

@@ -1,39 +1,26 @@
-from __future__ import annotations
-
 import hashlib
 import hmac
-import math
-import requests
+
+from typing import Collection
+
 
 class TOTP:
-    SPOTIFY_SECRETS_JSON = "https://code.thetadev.de/ThetaDev/spotify-secrets/raw/branch/main/secrets/secretDict.json"
     PERIOD = 30
     DIGITS = 6
 
-    def __init__(self) -> None:
-        self._setup()
-
-    def _setup(self) -> None:
-        version, secret_cipher_bytes = self.get_latest_secret()
+    def __init__(self, *, version: int, ciphertext: Collection[int]) -> None:
         self.version = version
-        self.secret = self.derive_secret_number(secret_cipher_bytes).encode()
+        self.secret = self.derive(ciphertext)
 
-    def derive_secret_number(self, secret_cipher_bytes: list[int]) -> str:
-        transformed = [
-            byte ^ ((i % 33) + 9) for i, byte in enumerate(secret_cipher_bytes)
-        ]
-        return "".join(str(n) for n in transformed)
-
-    def get_latest_secret(self) -> tuple[int, list[int]]:
-        response = requests.get(self.SPOTIFY_SECRETS_JSON)
-        response.raise_for_status()
-        secrets = response.json()
-        latest_version = max(int(v) for v in secrets.keys())
-        return latest_version, secrets[str(latest_version)]
+    @staticmethod
+    def derive(ciphertext: Collection[int]) -> bytes:
+        return ''.join(
+            str(byte ^ ((i % 33) + 9)) for i, byte in enumerate(ciphertext)
+        ).encode('ascii')
 
     def generate(self, timestamp: int) -> str:
-        counter = math.floor(timestamp / 1000 / self.PERIOD)
-        counter_bytes = counter.to_bytes(8, byteorder="big")
+        counter = int(timestamp) // 1000 // self.PERIOD
+        counter_bytes = counter.to_bytes(8)
 
         h = hmac.new(self.secret, counter_bytes, hashlib.sha1)
         hmac_result = h.digest()