Add SSL Verification Options and Improved Certificate Handling

pyproject.toml

@@ -39,6 +39,7 @@ pytest-mock = "^3.11.1"
 pytest-asyncio = "^0.21.1"
 rich = "^13.6.0"
 click-help-colors = "^0.9.2"
+certifi = { version = "^2025.1.31", optional = true }
 
 [tool.poetry.urls]
 "Bug Reports" = "https://github.com/nathom/streamrip/issues"
@@ -89,3 +90,6 @@ skip-magic-trailing-comma = false
 
 # Like Black, automatically detect the appropriate line ending.
 line-ending = "auto"
+
+[tool.poetry.extras]
+ssl = ["certifi"]

streamrip/client/client.py

@@ -7,6 +7,7 @@
 import aiohttp
 import aiolimiter
 
+from ..utils.ssl_utils import get_aiohttp_connector_kwargs
 from .downloadable import Downloadable
 
 logger = logging.getLogger("streamrip")
@@ -49,10 +50,17 @@ def get_rate_limiter(
         )
 
     @staticmethod
-    async def get_session(headers: dict | None = None) -> aiohttp.ClientSession:
+    async def get_session(
+        headers: dict | None = None, verify_ssl: bool = True
+    ) -> aiohttp.ClientSession:
         if headers is None:
             headers = {}
+
+        # Get connector kwargs based on SSL verification setting
+        connector_kwargs = get_aiohttp_connector_kwargs(verify_ssl=verify_ssl)
+        connector = aiohttp.TCPConnector(**connector_kwargs)
+
         return aiohttp.ClientSession(
-            headers={"User-Agent": DEFAULT_USER_AGENT},
-            **headers,
+            headers={"User-Agent": DEFAULT_USER_AGENT} | headers,
+            connector=connector,
         )

streamrip/client/deezer.py

@@ -42,7 +42,9 @@ def __init__(self, config: Config):
 
     async def login(self):
         # Used for track downloads
-        self.session = await self.get_session()
+        self.session = await self.get_session(
+            verify_ssl=self.global_config.session.downloads.verify_ssl
+        )
         arl = self.config.arl
         if not arl:
             raise MissingCredentialsError

streamrip/client/qobuz.py

@@ -47,7 +47,7 @@
 class QobuzSpoofer:
     """Spoofs the information required to stream tracks from Qobuz."""
 
-    def __init__(self):
+    def __init__(self, verify_ssl: bool = True):
         """Create a Spoofer."""
         self.seed_timezone_regex = (
             r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.ut'
@@ -62,6 +62,7 @@ def __init__(self):
             r'production:{api:{appId:"(?P<app_id>\d{9})",appSecret:"(\w{32})'
         )
         self.session = None
+        self.verify_ssl = verify_ssl
 
     async def get_app_id_and_secrets(self) -> tuple[str, list[str]]:
         assert self.session is not None
@@ -125,7 +126,13 @@ async def get_app_id_and_secrets(self) -> tuple[str, list[str]]:
         return app_id, secrets_list
 
     async def __aenter__(self):
-        self.session = aiohttp.ClientSession()
+        from ..utils.ssl_utils import get_aiohttp_connector_kwargs
+
+        # For the spoofer, always use SSL verification
+        connector_kwargs = get_aiohttp_connector_kwargs(verify_ssl=True)
+        connector = aiohttp.TCPConnector(**connector_kwargs)
+
+        self.session = aiohttp.ClientSession(connector=connector)
         return self
 
     async def __aexit__(self, *_):
@@ -147,7 +154,15 @@ def __init__(self, config: Config):
         self.secret: Optional[str] = None
 
     async def login(self):
-        self.session = await self.get_session()
+        self.session = await self.get_session(
+            verify_ssl=self.config.session.downloads.verify_ssl
+        )
+        """User credentials require either a user token OR a user email & password.
+
+        A hash of the password is stored in self.config.qobuz.password_or_token.
+        This data as well as the app_id is passed to self._get_user_auth_token() to get
+        the actual credentials for the user.
+        """
         c = self.config.session.qobuz
         if not c.email_or_userid or not c.password_or_token:
             raise MissingCredentialsError
@@ -164,7 +179,7 @@ async def login(self):
             f.set_modified()
 
         self.session.headers.update({"X-App-Id": str(c.app_id)})
-        
+
         if c.use_auth_token:
             params = {
                 "user_id": c.email_or_userid,
@@ -379,7 +394,9 @@ async def _paginate(
         return pages
 
     async def _get_app_id_and_secrets(self) -> tuple[str, list[str]]:
-        async with QobuzSpoofer() as spoofer:
+        async with QobuzSpoofer(
+            verify_ssl=self.config.session.downloads.verify_ssl
+        ) as spoofer:
             return await spoofer.get_app_id_and_secrets()
 
     async def _test_secret(self, secret: str) -> Optional[str]:
@@ -393,8 +410,8 @@ async def _test_secret(self, secret: str) -> Optional[str]:
 
     async def _get_valid_secret(self, secrets: list[str]) -> str:
         results = await asyncio.gather(
-                *[self._test_secret(secret) for secret in secrets],
-                )
+            *[self._test_secret(secret) for secret in secrets],
+        )
         working_secrets = [r for r in results if r is not None]
         if len(working_secrets) == 0:
             raise InvalidAppSecretError(secrets)

streamrip/client/soundcloud.py

@@ -36,7 +36,9 @@ def __init__(self, config: Config):
         )
 
     async def login(self):
-        self.session = await self.get_session()
+        self.session = await self.get_session(
+            verify_ssl=self.global_config.session.downloads.verify_ssl
+        )
         client_id, app_version = self.config.client_id, self.config.app_version
         if not client_id or not app_version or not (await self._announce_success()):
             client_id, app_version = await self._refresh_tokens()

streamrip/client/tidal.py

@@ -50,7 +50,9 @@ def __init__(self, config: Config):
         )
 
     async def login(self):
-        self.session = await self.get_session()
+        self.session = await self.get_session(
+            verify_ssl=self.global_config.session.downloads.verify_ssl
+        )
         c = self.config
         if not c.access_token:
             raise Exception("Access token not found in config.")
@@ -74,7 +76,13 @@ async def get_metadata(self, item_id: str, media_type: str) -> dict:
         :type media_type: str
         :rtype: dict
         """
-        assert media_type in ("track", "playlist", "album", "artist"), media_type
+        assert media_type in (
+            "track",
+            "album",
+            "playlist",
+            "video",
+            "artist",
+        ), media_type
 
         url = f"{media_type}s/{item_id}"
         item = await self._api_request(url)
@@ -104,13 +112,18 @@ async def get_metadata(self, item_id: str, media_type: str) -> dict:
             item["albums"].extend(ep_resp["items"])
         elif media_type == "track":
             try:
-                resp = await self._api_request(f"tracks/{str(item_id)}/lyrics", base="https://listen.tidal.com/v1")
+                resp = await self._api_request(
+                    f"tracks/{item_id!s}/lyrics", base="https://listen.tidal.com/v1"
+                )
 
                 # Use unsynced lyrics for MP3, synced for others (FLAC, OPUS, etc)
-                if self.global_config.session.conversion.enabled and self.global_config.session.conversion.codec.upper() == "MP3":
-                    item["lyrics"] = resp.get("lyrics") or ''
+                if (
+                    self.global_config.session.conversion.enabled
+                    and self.global_config.session.conversion.codec.upper() == "MP3"
+                ):
+                    item["lyrics"] = resp.get("lyrics") or ""
                 else:
-                    item["lyrics"] = resp.get("subtitles") or resp.get("lyrics") or ''
+                    item["lyrics"] = resp.get("subtitles") or resp.get("lyrics") or ""
             except TypeError as e:
                 logger.warning(f"Failed to get lyrics for {item_id}: {e}")
 
@@ -153,7 +166,9 @@ async def get_downloadable(self, track_id: str, quality: int):
         except KeyError:
             raise Exception(resp["userMessage"])
         except JSONDecodeError:
-            logger.warning(f"Failed to get manifest for {track_id}. Retrying with lower quality.")
+            logger.warning(
+                f"Failed to get manifest for {track_id}. Retrying with lower quality."
+            )
             return await self.get_downloadable(track_id, quality - 1)
 
         logger.debug(manifest)

streamrip/config.py

@@ -198,6 +198,9 @@ class DownloadsConfig:
     # A value that is too high for your bandwidth may cause slowdowns
     max_connections: int
     requests_per_minute: int
+    # Verify SSL certificates for API connections
+    # Set to false if you encounter SSL certificate verification errors (not recommended)
+    verify_ssl: bool
 
 
 @dataclass(slots=True)

streamrip/config.toml

@@ -17,6 +17,9 @@ max_connections = 6
 # Max number of API requests per source to handle per minute
 # Set to -1 for no limit
 requests_per_minute = 60
+# Verify SSL certificates for API connections
+# Set to false if you encounter SSL certificate verification errors (not recommended)
+verify_ssl = true
 
 [qobuz]
 # 1: 320kbps MP3, 2: 16/44.1, 3: 24/<=96, 4: 24/>=96

streamrip/filepath_utils.py

@@ -10,7 +10,7 @@ def truncate_str(text: str) -> str:
     str_bytes = text.encode()
     str_bytes = str_bytes[:255]
     return str_bytes.decode(errors="ignore")
-    
+
 
 def clean_filename(fn: str, restrict: bool = False) -> str:
     path = truncate_str(str(sanitize_filename(fn)))

streamrip/media/playlist.py

@@ -24,6 +24,7 @@
     SearchResults,
     TrackMetadata,
 )
+from ..utils.ssl_utils import get_aiohttp_connector_kwargs
 from .artwork import download_artwork
 from .media import Media, Pending
 from .track import Track
@@ -350,7 +351,11 @@ async def fetch(session: aiohttp.ClientSession, url, **kwargs):
                 return await resp.text("utf-8")
 
         # Create new session so we're not bound by rate limit
-        async with aiohttp.ClientSession() as session:
+        verify_ssl = getattr(self.config.session.downloads, "verify_ssl", True)
+        connector_kwargs = get_aiohttp_connector_kwargs(verify_ssl=verify_ssl)
+        connector = aiohttp.TCPConnector(**connector_kwargs)
+
+        async with aiohttp.ClientSession(connector=connector) as session:
             page = await fetch(session, playlist_url)
             playlist_title_match = re_playlist_title_match.search(page)
             if playlist_title_match is None:

streamrip/metadata/album.py

@@ -77,7 +77,7 @@ def format_folder_path(self, formatter: str) -> str:
             "year": self.year,
             "container": self.info.container,
         }
-        
+
         return clean_filepath(formatter.format(**info))
 
     @classmethod

streamrip/metadata/track.py

@@ -212,7 +212,7 @@ def from_tidal(cls, album: AlbumMetadata, track) -> TrackMetadata:
             discnumber=discnumber,
             composer=None,
             isrc=isrc,
-            lyrics=lyrics
+            lyrics=lyrics,
         )
 
     @classmethod