From 01e9fe9963f5666411d9840e6cf0c634e95a6533 Mon Sep 17 00:00:00 2001 From: Jordan Rejaud Date: Thu, 25 Jun 2026 19:58:13 +0700 Subject: [PATCH] qobuz: support download-store (purchased) content on free accounts A Qobuz account with no streaming subscription returns an empty credential.parameters on login. Previously login() raised IneligibleError, so such an account could not download albums it had *purchased* from the Qobuz download store (#673, #918). Make the behavior conditional instead of regressing streaming for subscribers: - login() flags the client download_only=True (WARNING log) on empty credential.parameters instead of raising. Genuine 401/400 auth failures still raise. - _request_file_url() selects intent=download when download_only else intent=stream, in both the signed request_sig preimage and the params dict. - get_downloadable() clamps quality for purchased content: on a FormatRestrictedByFormatAvailability restriction it retries one tier down (Qobuz offers no automatic fallback for owned content). Adds unit tests covering both account types and both intent paths. This only enables downloading content the account already owns; Qobuz still gates getFileUrl on ownership server-side. Closes #673 Closes #918 Co-Authored-By: Claude Opus 4.8 (1M context) --- streamrip/client/qobuz.py | 48 ++++++++- tests/test_qobuz_download_only.py | 158 ++++++++++++++++++++++++++++++ 2 files changed, 201 insertions(+), 5 deletions(-) create mode 100644 tests/test_qobuz_download_only.py 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"] + )