Transition from PyCryptodome to cryptography for better performance

environment.yml

@@ -4,15 +4,15 @@ channels:
   - microsoft
 dependencies:
   # CORE DEPENDENCIES
-  - levenshtein
+  - cryptography
   - mutagen
-  - numpy
-  - pycryptodome
   - requests
   # OPTIONAL DEPENDENCIES
   - ffmpeg
   - flask
-  - pytest-playwright
+  - levenshtein
+  - numpy
+  - playwright
   # DEVELOPMENT DEPENDENCIES
   - pytest
   - ruff

requirements.txt

@@ -1,6 +1,6 @@
 # CORE DEPENDENCIES
+cryptography
 mutagen
-pycryptodome
 requests
 
 # OPTIONAL DEPENDENCIES

requirements_minimal.txt

@@ -1,3 +1,3 @@
+cryptography
 mutagen
-pycryptodome
 requests

src/minim/audio.py

@@ -73,7 +73,7 @@ def _from_file(self) -> None:
                 value = ([sv for v in value for sv in getattr(v, base)]
                          if len(value) > 1 else getattr(value[0], base))
                 if list not in self._FIELDS_TYPES[field]:
-                    value = utility.multivalue_formatter(value, False, 
+                    value = utility.format_multivalue(value, False, 
                                                          primary=True)
                     if type(value) not in self._FIELDS_TYPES[field]:
                         try:
@@ -139,7 +139,7 @@ def write_metadata(self) -> None:
         for field, (frame, base, func) in self._FIELDS.items():
             value = getattr(self, field)
             if value:
-                value = utility.multivalue_formatter(
+                value = utility.format_multivalue(
                     value, self._multivalue, sep=self._sep
                 )
                 self._tags.add(
@@ -239,7 +239,7 @@ def _from_file(self) -> None:
             value = self._tags.get(key)
             if value:
                 if list not in self._FIELDS_TYPES[field]:
-                    value = utility.multivalue_formatter(value, False, 
+                    value = utility.format_multivalue(value, False, 
                                                          primary=True)
                     if type(value) not in self._FIELDS_TYPES[field]:
                         try:
@@ -319,7 +319,7 @@ def write_metadata(self) -> None:
         for field, (key, func) in (self._FIELDS | self._FIELDS_SPECIAL).items():
             value = getattr(self, field)
             if value:
-                value = utility.multivalue_formatter(
+                value = utility.format_multivalue(
                     value, self._multivalue, sep=self._sep
                 )
                 self._tags[key] = func(value) if func else value
@@ -1065,7 +1065,7 @@ def set_metadata_using_qobuz(
                 if album_feat_artist and "feat." not in self.album:
                     self.album += (" [feat. {}]" if "(" in self.album 
                                    else " (feat. {})").format(
-                        utility.multivalue_formatter(album_feat_artist, False)
+                        utility.format_multivalue(album_feat_artist, False)
                     )
             if data["album"]["version"]:
                 self.album += (
@@ -1139,7 +1139,7 @@ def set_metadata_using_qobuz(
                     and "feat." not in self.title:
                 self.title += (" [feat. {}]" if "(" in self.title 
                                else " (feat. {})").format(
-                    utility.multivalue_formatter(feat_artist, False)
+                    utility.format_multivalue(feat_artist, False)
                 )
             if data["version"]:
                 self.title += (" [{}]" if "(" in self.title 
@@ -1417,7 +1417,7 @@ def _from_file(self) -> None:
             value = self._tags.get(key)
             if value:
                 if list not in self._FIELDS_TYPES[field]:
-                    value = utility.multivalue_formatter(value, False, 
+                    value = utility.format_multivalue(value, False, 
                                                          primary=True)
                     if type(value) not in self._FIELDS_TYPES[field]:
                         try:
@@ -1451,7 +1451,7 @@ def _from_file(self) -> None:
             self.track_number = self.track_count = None
 
         if "covr" in self._tags:
-            self.artwork = utility.multivalue_formatter(self._tags.get("covr"), 
+            self.artwork = utility.format_multivalue(self._tags.get("covr"), 
                                                         False, primary=True)
             self._artwork_format = str(
                 self._IMAGE_FORMATS[self.artwork.imageformat]
@@ -1469,7 +1469,7 @@ def write_metadata(self) -> None:
         for field, key in self._FIELDS.items():
             value = getattr(self, field)
             if value:
-                value = utility.multivalue_formatter(
+                value = utility.format_multivalue(
                     value, self._multivalue, sep=self._sep
                 )
                 try:

src/minim/tidal.py

@@ -10,8 +10,7 @@
 
 from xml.dom import minidom
 
-from Crypto.Cipher import AES
-from Crypto.Util import Counter
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 
 from . import (
     base64, datetime, hashlib, json, logging, os, pathlib, re, requests, 
@@ -1981,8 +1980,6 @@ class PrivateAPI:
     """
 
     _FLOWS = {"pkce", "device_code"}
-    _MASTER_KEY = (b"P\x89SLC&\x98\xb7\xc6\xa3\n?P.\xb4\xc7a\xf8\xe5n"
-                   b"\x8cth\x13E\xfa?\xbah8\xef\x9e")
     _NAME = f"{__module__}.{__qualname__}"
 
     API_URL = "https://api.tidal.com"
@@ -6673,16 +6670,19 @@ def get_track_stream(
             codec = manifest["codecs"]
             with self.session.get(manifest["urls"][0]) as r:
                 stream = r.content
-            if manifest["encryptionType"] != "NONE":
-                d_key_id = base64.b64decode(manifest['keyId'])
-                d_id = AES.new(self._MASTER_KEY, AES.MODE_CBC,
-                               d_key_id[:16]).decrypt(d_key_id[16:])
-                d_key, d_nonce = d_id[:16], d_id[16:24]
-                stream = AES.new(
-                    d_key, AES.MODE_CTR,
-                    counter=Counter.new(64, prefix=d_nonce, 
-                                        initial_value=0)
-                ).decrypt(stream)
+            if manifest["encryptionType"] == "OLD_AES":
+                key_id = base64.b64decode(manifest["keyId"])
+                key_nonce = Cipher(
+                    algorithms.AES(b"P\x89SLC&\x98\xb7\xc6\xa3\n?P.\xb4\xc7"
+                                   b"a\xf8\xe5n\x8cth\x13E\xfa?\xbah8\xef\x9e"),
+                    modes.CBC(key_id[:16])
+                ).decryptor().update(key_id[16:])
+                stream = Cipher(
+                    algorithms.AES(key_nonce[:16]),
+                    modes.CTR(key_nonce[16:32])
+                ).decryptor().update(stream)
+            elif manifest["encryptionType"] != "NONE":
+                raise NotImplementedError("Unsupported encryption type.")
         return stream, codec
 
     def get_video_stream(

src/minim/utility.py

@@ -19,7 +19,7 @@
 except ModuleNotFoundError:
     FOUND_NUMPY = False
 
-__all__ = ["levenshtein_ratio", "multivalue_formatter"]
+__all__ = ["levenshtein_ratio", "format_multivalue"]
 
 def levenshtein_ratio(
         base: str, values: Union[str, list[str]]
@@ -57,7 +57,7 @@ def levenshtein_ratio(
         return np.fromiter(gen, dtype=float, count=len(values))
     return list(gen)
 
-def multivalue_formatter(
+def format_multivalue(
         value: Any, multivalue: bool, *, primary: bool = False,
         sep: Union[str, tuple[str]] = (", ", " & ")) -> Union[str, list[Any]]: