diff --git a/streamrip/client/qobuz.py b/streamrip/client/qobuz.py index 734e2b82..d28488af 100644 --- a/streamrip/client/qobuz.py +++ b/streamrip/client/qobuz.py @@ -12,7 +12,6 @@ from ..config import Config from ..exceptions import ( AuthenticationError, - IneligibleError, InvalidAppIdError, InvalidAppSecretError, MissingCredentialsError, @@ -152,6 +151,10 @@ def __init__(self, config: Config): config.session.downloads.requests_per_minute, ) self.secret: Optional[str] = None + # True for a free account (no streaming subscription) that can still + # download content it has *purchased* from the Qobuz download store. + # When set, file-url requests use intent=download instead of stream. + self.download_only: bool = False async def login(self): self.session = await self.get_session( @@ -204,8 +207,20 @@ async def login(self): logger.debug("Logged in to Qobuz") + # An empty credential.parameters means the account has no active + # streaming subscription. Such a (free) account cannot stream, but it + # CAN still download albums it has purchased from the Qobuz download + # store. Rather than refusing outright, flag the client as + # download-only so _request_file_url() requests intent=download. + # Genuine auth failures (401/400) are already handled above; this only + # reclassifies the free-but-owns-content case and never swallows them. if not resp["user"]["credential"]["parameters"]: - raise IneligibleError("Free accounts are not eligible to download tracks.") + self.download_only = True + logger.warning( + "Free Qobuz account detected (no streaming subscription): " + "streaming is unavailable. Only purchased download-store " + "content can be downloaded (intent=download)." + ) uat = resp["user_auth_token"] self.session.headers.update({"X-User-Auth-Token": uat}) @@ -328,8 +343,27 @@ async def get_downloadable(self, item: str, quality: int) -> Downloadable: if stream_url is None: restrictions = resp_json["restrictions"] if restrictions: + code = restrictions[0]["code"] + # Purchased (download-only) content is sold in exactly one + # format and Qobuz offers NO automatic fallback: requesting a + # higher tier than the purchased one fails with + # FormatRestrictedByFormatAvailability. Clamp by retrying one + # quality tier down until we hit the format the account owns. + if ( + self.download_only + and quality > 1 + and code == "FormatRestrictedByFormatAvailability" + ): + logger.warning( + "Quality %d unavailable for purchased track %s; " + "retrying one tier down at quality %d.", + quality, + item, + quality - 1, + ) + return await self.get_downloadable(item, quality - 1) # Turn CamelCase code into a readable sentence - words = re.findall(r"([A-Z][a-z]+)", restrictions[0]["code"]) + words = re.findall(r"([A-Z][a-z]+)", code) raise NonStreamableError( words[0] + " " + " ".join(map(str.lower, words[1:])) + ".", ) @@ -426,7 +460,11 @@ async def _request_file_url( ) -> tuple[int, dict]: quality = self.get_quality(quality) unix_ts = time.time() - r_sig = f"trackgetFileUrlformat_id{quality}intentstreamtrack_id{track_id}{unix_ts}{secret}" + # Owned-only (free) accounts must request intent=download; streaming + # accounts use intent=stream. The signed preimage and the params dict + # MUST agree on the value or Qobuz rejects the request with HTTP 400. + intent = "download" if self.download_only else "stream" + r_sig = f"trackgetFileUrlformat_id{quality}intent{intent}track_id{track_id}{unix_ts}{secret}" logger.debug("Raw request signature: %s", r_sig) r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest() logger.debug("Hashed request signature: %s", r_sig_hashed) @@ -435,7 +473,7 @@ async def _request_file_url( "request_sig": r_sig_hashed, "track_id": track_id, "format_id": quality, - "intent": "stream", + "intent": intent, } return await self._api_request("track/getFileUrl", params) diff --git a/tests/test_qobuz_download_only.py b/tests/test_qobuz_download_only.py new file mode 100644 index 00000000..bf0fba51 --- /dev/null +++ b/tests/test_qobuz_download_only.py @@ -0,0 +1,158 @@ +"""Unit tests for the conditional download-store (free/purchased account) path. + +A free Qobuz account (no streaming subscription) returns an empty +``credential.parameters`` on login. It cannot stream, but it can still download +albums it has *purchased*. These tests verify that: + +* a subscriber (non-empty parameters) is NOT flagged download-only and keeps + using ``intent=stream`` (no streaming regression), +* a free account (empty parameters) is flagged ``download_only`` without + raising, and +* ``_request_file_url`` selects the matching ``intent`` in BOTH the signed + request-signature preimage and the params dict. + +All network calls are mocked; these run without Qobuz credentials. +""" + +import hashlib +from unittest.mock import AsyncMock + +from util import arun + +from streamrip.client.qobuz import QobuzClient +from streamrip.config import Config + + +class _FakeSession: + """Minimal stand-in for the aiohttp session used during login().""" + + def __init__(self): + self.headers = {} + + async def close(self): + pass + + +def _make_client() -> QobuzClient: + config = Config.defaults() + c = config.session.qobuz + c.email_or_userid = "13103092" + c.password_or_token = "fake-token" + c.use_auth_token = True + # Pre-seed app_id/secrets so login() skips the spoofer/network fetch. + c.app_id = "123456789" + c.secrets = ["fakesecret"] + return QobuzClient(config) + + +def _login_resp(parameters): + return { + "user": {"credential": {"parameters": parameters}}, + "user_auth_token": "fake-uat", + } + + +def _run_login(monkeypatch, parameters) -> QobuzClient: + client = _make_client() + monkeypatch.setattr(client, "get_session", AsyncMock(return_value=_FakeSession())) + monkeypatch.setattr( + client, + "_api_request", + AsyncMock(return_value=(200, _login_resp(parameters))), + ) + monkeypatch.setattr( + client, "_get_valid_secret", AsyncMock(return_value="fakesecret") + ) + arun(client.login()) + return client + + +def test_subscriber_login_not_download_only(monkeypatch): + """Non-empty credential.parameters -> NOT download_only, does not raise.""" + client = _run_login(monkeypatch, {"lossy_streaming": True, "hires_streaming": True}) + assert client.download_only is False + assert client.logged_in is True + + +def test_free_account_login_sets_download_only(monkeypatch): + """Empty credential.parameters -> download_only=True, does NOT raise.""" + client = _run_login(monkeypatch, []) + assert client.download_only is True + # Login still succeeds (no IneligibleError); the account can download + # purchased content. + assert client.logged_in is True + + +def _capture_file_url_params(monkeypatch, download_only: bool) -> dict: + client = _make_client() + client.download_only = download_only + captured: dict = {} + + async def fake_api(epoint, params): + captured["epoint"] = epoint + captured["params"] = params + return (200, {}) + + monkeypatch.setattr(client, "_api_request", fake_api) + arun(client._request_file_url("19512574", 3, "abc123secret")) + return captured + + +def test_request_file_url_download_intent_when_download_only(monkeypatch): + """download_only=True -> intent=download in BOTH params and signed preimage.""" + secret = "abc123secret" + track_id = "19512574" + quality = 3 + client = _make_client() + client.download_only = True + captured: dict = {} + + async def fake_api(epoint, params): + captured["params"] = params + return (200, {}) + + monkeypatch.setattr(client, "_api_request", fake_api) + arun(client._request_file_url(track_id, quality, secret)) + + params = captured["params"] + # 1. params dict uses intent=download + assert params["intent"] == "download" + # 2. the signed preimage used intent=download too (reconstruct + md5 match) + format_id = QobuzClient.get_quality(quality) + expected_preimage = ( + f"trackgetFileUrlformat_id{format_id}intentdownload" + f"track_id{track_id}{params['request_ts']}{secret}" + ) + assert ( + hashlib.md5(expected_preimage.encode("utf-8")).hexdigest() + == params["request_sig"] + ) + + +def test_request_file_url_stream_intent_when_subscriber(monkeypatch): + """download_only=False -> intent=stream in BOTH params and signed preimage.""" + secret = "abc123secret" + track_id = "19512574" + quality = 3 + client = _make_client() + client.download_only = False + captured: dict = {} + + async def fake_api(epoint, params): + captured["params"] = params + return (200, {}) + + monkeypatch.setattr(client, "_api_request", fake_api) + arun(client._request_file_url(track_id, quality, secret)) + + params = captured["params"] + assert params["intent"] == "stream" + format_id = QobuzClient.get_quality(quality) + expected_preimage = ( + f"trackgetFileUrlformat_id{format_id}intentstream" + f"track_id{track_id}{params['request_ts']}{secret}" + ) + assert ( + hashlib.md5(expected_preimage.encode("utf-8")).hexdigest() + == params["request_sig"] + )