Add support for new Qobuz login flow

README.md

@@ -130,6 +130,40 @@ For more customization, see the config file
 rip config open
 ```
 
+### Qobuz login (token-based)
+
+Qobuz no longer supports the old direct email/password API login flow used by streamrip.  
+Streamrip now attempts to capture `user.id` and `user_auth_token` automatically in a managed browser session.
+
+Before using automatic capture, install Playwright browser runtime:
+
+```bash
+pip install playwright
+playwright install chromium
+```
+
+If automatic capture fails or times out, streamrip falls back to manual token input.
+
+In your config:
+
+```toml
+[qobuz]
+use_auth_token = true
+email_or_userid = "YOUR_QOBUZ_USER_ID"
+password_or_token = "YOUR_USER_AUTH_TOKEN"
+```
+
+To refresh an expired token:
+
+1. Run any Qobuz command again (streamrip will retry auto-capture)
+2. If needed, log in at `qobuz.com` or `play.qobuz.com`
+3. Open DevTools -> Network
+4. Filter requests by `user/login`
+5. Open a successful request response and copy:
+   - `user_auth_token`
+   - `user.id`
+6. Update `email_or_userid` and `password_or_token` in config
+
 If you're confused about anything, see the help pages. The main help pages can be accessed by typing `rip` by itself in the command line. The help pages for each command can be accessed with the `--help` flag. For example, to see the help page for the `url` command, type
 
 ```

pyproject.toml

@@ -38,6 +38,7 @@ aiolimiter = "^1.1.0"
 rich = "^13.6.0"
 click-help-colors = "^0.9.2"
 certifi = { version = "^2025.1.31", optional = true }
+playwright = "^1.58.0"
 
 [tool.poetry.urls]
 "Bug Reports" = "https://github.com/nathom/streamrip/issues"

streamrip/client/qobuz.py

@@ -193,14 +193,21 @@ async def login(self):
                 "app_id": str(c.app_id),
             }
 
-        logger.debug("Request params %s", params)
+        logger.debug("Request params %s", self._redact_auth_payload(params))
         status, resp = await self._api_request("user/login", params)
-        logger.debug("Login resp: %s", resp)
+        logger.debug("Login resp: %s", self._redact_auth_payload(resp))
 
         if status == 401:
-            raise AuthenticationError(f"Invalid credentials from params {params}")
+            if c.use_auth_token:
+                raise AuthenticationError(
+                    "Invalid Qobuz token or user id. The token may have expired; "
+                    "refresh user_auth_token from a logged-in browser session."
+                )
+            raise AuthenticationError("Invalid Qobuz credentials.")
         elif status == 400:
-            raise InvalidAppIdError(f"Invalid app id from params {params}")
+            raise InvalidAppIdError(
+                f"Invalid app id from params {self._redact_auth_payload(params)}"
+            )
 
         logger.debug("Logged in to Qobuz")
 
@@ -453,3 +460,11 @@ async def _api_request(self, epoint: str, params: dict) -> tuple[int, dict]:
     def get_quality(quality: int):
         quality_map = (5, 6, 7, 27)
         return quality_map[quality - 1]
+
+    @staticmethod
+    def _redact_auth_payload(payload: dict) -> dict:
+        redacted = dict(payload)
+        for key in ("password", "user_auth_token"):
+            if key in redacted and redacted[key]:
+                redacted[key] = "***REDACTED***"
+        return redacted

streamrip/config.toml

@@ -27,11 +27,11 @@ quality = 3
 # This will download booklet pdfs that are included with some albums
 download_booklets = true
 
-# Authenticate to Qobuz using auth token? Value can be true/false only
-use_auth_token = false
-# Enter your userid if the above use_auth_token is set to true, else enter your email
+# Qobuz web login now uses OAuth/reCAPTCHA. Use token auth from a logged-in browser session.
+use_auth_token = true
+# Enter your Qobuz user id (numeric string)
 email_or_userid = ""
-# Enter your auth token if the above use_auth_token is set to true, else enter the md5 hash of your plaintext password
+# Enter your Qobuz user_auth_token (JWT). Refresh it when it expires.
 password_or_token = ""
 # Do not change
 app_id = ""

streamrip/rip/main.py

@@ -9,6 +9,7 @@
 from ..client import Client, DeezerClient, QobuzClient, SoundcloudClient, TidalClient
 from ..config import Config
 from ..console import console
+from ..exceptions import AuthenticationError
 from ..media import (
     Media,
     Pending,
@@ -147,7 +148,17 @@ async def get_logged_in_client(self, source: str):
             else:
                 with console.status(f"[cyan]Logging into {source}", spinner="dots"):
                     # Log into client using credentials from config
-                    await client.login()
+                    try:
+                        await client.login()
+                    except AuthenticationError:
+                        if source != "qobuz":
+                            raise
+                        console.print(
+                            "[yellow]Saved Qobuz token appears invalid or expired. "
+                            "Please provide a refreshed token.[/yellow]"
+                        )
+                        await prompter.prompt_and_login()
+                        prompter.save()
 
         assert client.logged_in
         return client

streamrip/rip/prompter.py

@@ -1,16 +1,16 @@
 import asyncio
-import hashlib
 import logging
 import time
 from abc import ABC, abstractmethod
 
 from click import launch
-from rich.prompt import Prompt
+from rich.prompt import Confirm, Prompt
 
 from ..client import Client, DeezerClient, QobuzClient, SoundcloudClient, TidalClient
 from ..config import Config
 from ..console import console
 from ..exceptions import AuthenticationError, MissingCredentialsError
+from .qobuz_token_capture import QobuzTokenCaptureError, capture_qobuz_auth_token
 
 logger = logging.getLogger("streamrip")
 
@@ -52,35 +52,62 @@ def has_creds(self) -> bool:
 
     async def prompt_and_login(self):
         if not self.has_creds():
-            self._prompt_creds_and_set_session_config()
+            await self._prompt_creds_and_set_session_config()
 
         while True:
             try:
                 await self.client.login()
                 break
             except AuthenticationError:
-                console.print("[yellow]Invalid credentials, try again.")
-                self._prompt_creds_and_set_session_config()
+                console.print(
+                    "[yellow]Invalid Qobuz token or user id. "
+                    "The token may have expired, please refresh it from your browser.[/yellow]"
+                )
+                await self._prompt_creds_and_set_session_config()
             except MissingCredentialsError:
-                self._prompt_creds_and_set_session_config()
+                await self._prompt_creds_and_set_session_config()
 
-    def _prompt_creds_and_set_session_config(self):
-        email = Prompt.ask("Enter your Qobuz email")
-        pwd_input = Prompt.ask("Enter your Qobuz password (invisible)", password=True)
+    async def _prompt_creds_and_set_session_config(self):
+        console.print(
+            "[cyan]Attempting automatic Qobuz token capture in a managed browser...[/cyan]"
+        )
+        try:
+            user_id, token = await capture_qobuz_auth_token(timeout_s=300)
+            self._set_session_qobuz_auth_token(user_id, token)
+            console.print(
+                f"[green]Credentials saved to config file at [bold cyan]{self.config.path}",
+            )
+            return
+        except QobuzTokenCaptureError as exc:
+            console.print(f"[yellow]{exc}[/yellow]")
 
-        pwd = hashlib.md5(pwd_input.encode("utf-8")).hexdigest()
+        console.print(
+            "[cyan]Qobuz now requires token-based login.[/cyan]\n"
+            "1) Log in at qobuz.com\n"
+            "2) Open browser DevTools -> Network\n"
+            "3) Find a successful [bold]user/login[/bold] request\n"
+            "4) Copy [bold]user_auth_token[/bold] and [bold]user id[/bold]"
+        )
+        if Confirm.ask("Open Qobuz login page in your browser now?", default=True):
+            launch("https://play.qobuz.com/login")
+
+        user_id = Prompt.ask("Enter your Qobuz user id")
+        token = Prompt.ask("Enter your Qobuz user_auth_token", password=True)
+        self._set_session_qobuz_auth_token(user_id, token)
         console.print(
             f"[green]Credentials saved to config file at [bold cyan]{self.config.path}",
         )
+
+    def _set_session_qobuz_auth_token(self, user_id: str, token: str):
         c = self.config.session.qobuz
-        c.use_auth_token = False
-        c.email_or_userid = email
-        c.password_or_token = pwd
+        c.use_auth_token = True
+        c.email_or_userid = user_id
+        c.password_or_token = token
 
     def save(self):
         c = self.config.session.qobuz
         cf = self.config.file.qobuz
-        cf.use_auth_token = False
+        cf.use_auth_token = True
         cf.email_or_userid = c.email_or_userid
         cf.password_or_token = c.password_or_token
         self.config.file.set_modified()

streamrip/rip/qobuz_token_capture.py

@@ -0,0 +1,110 @@
+import asyncio
+import logging
+import platform
+import time
+
+logger = logging.getLogger("streamrip")
+
+QOBUZ_LOGIN_PAGE = "https://play.qobuz.com/login"
+QOBUZ_LOGIN_API = "https://www.qobuz.com/api.json/0.2/user/login"
+
+
+class QobuzTokenCaptureError(Exception):
+    """Raised when automatic Qobuz token capture fails."""
+
+
+async def _capture_qobuz_auth_token_async(timeout_s: int = 300) -> tuple[str, str]:
+    """Capture user id and auth token from Qobuz web login traffic.
+
+    Returns:
+        Tuple of (user_id, user_auth_token)
+    """
+    try:
+        from playwright.async_api import async_playwright
+    except Exception as exc:  # pragma: no cover - import path only
+        raise QobuzTokenCaptureError(
+            "Automatic browser capture requires Playwright. "
+            "Install it and run `playwright install chromium`, or use manual token input."
+        ) from exc
+
+    result: dict[str, str] = {}
+
+    async def handle_response(response):
+        if response.url != QOBUZ_LOGIN_API:
+            return
+
+        try:
+            post_data = response.request.post_data or ""
+            if post_data and "extra=partner" not in post_data:
+                return
+            if response.status != 200:
+                return
+            payload = await response.json()
+        except Exception:
+            return
+
+        if not isinstance(payload, dict):
+            return
+
+        user = payload.get("user", {})
+        user_id = user.get("id")
+        token = payload.get("user_auth_token")
+        if user_id is None or not token:
+            return
+
+        result["user_id"] = str(user_id)
+        result["token"] = str(token)
+
+    try:
+        async with async_playwright() as playwright:
+            browser = await playwright.chromium.launch(headless=False)
+            context = await browser.new_context()
+            page = await context.new_page()
+            page.on("response", handle_response)
+            await page.goto(QOBUZ_LOGIN_PAGE, wait_until="domcontentloaded")
+
+            logger.info(
+                "Waiting for Qobuz login response in browser (timeout: %ss).",
+                timeout_s,
+            )
+            deadline = time.monotonic() + timeout_s
+            while "token" not in result and time.monotonic() < deadline:
+                await page.wait_for_timeout(250)
+
+            await context.close()
+            await browser.close()
+    except Exception as exc:
+        raise QobuzTokenCaptureError(
+            f"Automatic browser capture failed: {exc}"
+        ) from exc
+
+    if "token" not in result or "user_id" not in result:
+        raise QobuzTokenCaptureError(
+            "Could not detect a successful Qobuz user/login response. "
+            "Please complete login in the opened browser or use manual token input."
+        )
+
+    return result["user_id"], result["token"]
+
+
+def _capture_qobuz_auth_token_windows(timeout_s: int) -> tuple[str, str]:
+    """Run Playwright capture in an isolated Proactor loop on Windows."""
+    if not hasattr(asyncio, "ProactorEventLoop"):
+        raise QobuzTokenCaptureError(
+            "Windows Proactor event loop is unavailable; use manual token input."
+        )
+
+    loop = asyncio.ProactorEventLoop()  # type: ignore[attr-defined]
+    try:
+        asyncio.set_event_loop(loop)
+        return loop.run_until_complete(_capture_qobuz_auth_token_async(timeout_s))
+    finally:
+        loop.close()
+        asyncio.set_event_loop(None)
+
+
+async def capture_qobuz_auth_token(timeout_s: int = 300) -> tuple[str, str]:
+    """Capture user id and auth token from Qobuz web login traffic."""
+    if platform.system() == "Windows":
+        return await asyncio.to_thread(_capture_qobuz_auth_token_windows, timeout_s)
+    return await _capture_qobuz_auth_token_async(timeout_s)