From 70d7eac476804fc1808c9e68da9924749df569ce Mon Sep 17 00:00:00 2001 From: Berserk Agent Date: Sun, 21 Jun 2026 01:12:14 +0000 Subject: [PATCH] fix(auth): surface browser cookie decryption failures --- tests/test_cookies.py | 20 ++++++++++ xhs_cli/cookies.py | 89 ++++++++++++++++++++++++++++++++++--------- xhs_cli/exceptions.py | 17 +++++++++ 3 files changed, 109 insertions(+), 17 deletions(-) diff --git a/tests/test_cookies.py b/tests/test_cookies.py index aae7636..f4778dd 100644 --- a/tests/test_cookies.py +++ b/tests/test_cookies.py @@ -7,9 +7,11 @@ from xhs_cli.cookies import ( NOTE_CONTEXT_TTL_SECONDS, + CookieDecryptionError, cache_note_context, clear_cookies, cookies_to_string, + extract_browser_cookies, get_cached_note_context, get_cached_xsec_token, get_cookies, @@ -21,6 +23,7 @@ save_cookies, save_note_index, ) +from xhs_cli.exceptions import BrowserCookieDecryptionError @pytest.fixture @@ -110,6 +113,23 @@ def test_force_refresh_bypasses_saved_cookies(self, monkeypatch): assert cookies == {"a1": "fresh"} assert saved == [{"a1": "fresh"}] + def test_decryption_failure_is_not_reported_as_missing_cookie(self, monkeypatch): + monkeypatch.setattr( + "xhs_cli.cookies._extract_in_process", + lambda source: (_ for _ in ()).throw( + CookieDecryptionError(source, "Unable to get key for cookie decryption") + ), + ) + monkeypatch.setattr("xhs_cli.cookies._extract_via_subprocess", lambda source: None) + + with pytest.raises(BrowserCookieDecryptionError) as exc_info: + extract_browser_cookies("chrome") + + message = str(exc_info.value) + assert "Cookie decryption failed" in message + assert "Unable to get key for cookie decryption" in message + assert "xhs login --qrcode" in message + class TestNoteContextCache: def test_cache_persists_token_and_source(self, tmp_config_dir): diff --git a/xhs_cli/cookies.py b/xhs_cli/cookies.py index 96665a4..346cd71 100644 --- a/xhs_cli/cookies.py +++ b/xhs_cli/cookies.py @@ -5,6 +5,7 @@ import functools import json import logging +import re import subprocess import sys import threading @@ -17,6 +18,30 @@ logger = logging.getLogger(__name__) +_COOKIE_DECRYPTION_ERROR_PATTERNS = ( + re.compile(r"unable to get key for cookie decryption", re.IGNORECASE), + re.compile(r"decrypt", re.IGNORECASE), +) + + +class CookieExtractionError(Exception): + """Internal signal for browser cookie extraction errors.""" + + def __init__(self, source: str, message: str): + super().__init__(message) + self.source = source + self.message = message + + +class CookieDecryptionError(CookieExtractionError): + """Internal signal for cookie decryption failures.""" + + +def _is_cookie_decryption_error(message: str) -> bool: + """Return True when browser_cookie3 failed while decrypting cookies.""" + return any(pattern.search(message) for pattern in _COOKIE_DECRYPTION_ERROR_PATTERNS) + + # Cookie TTL: warn and attempt browser refresh after 7 days COOKIE_TTL_DAYS = 7 _COOKIE_TTL_SECONDS = COOKIE_TTL_DAYS * 86400 @@ -361,8 +386,11 @@ def _extract_in_process(source: str) -> dict[str, str] | None: try: jar = loader(domain_name=".xiaohongshu.com") except Exception as exc: - logger.debug("%s in-process extraction failed: %s", source, exc) - return None + message = str(exc) + logger.debug("%s in-process extraction failed: %s", source, message) + if _is_cookie_decryption_error(message): + raise CookieDecryptionError(source, message) from exc + raise CookieExtractionError(source, message) from exc cookies = {cookie.name: cookie.value for cookie in jar if "xiaohongshu.com" in (cookie.domain or "")} if cookies.get("a1"): @@ -414,8 +442,11 @@ def _extract_via_subprocess(source: str) -> dict[str, str] | None: data = json.loads(result.stdout.strip()) if "error" in data: - logger.debug("Cookie extraction error: %s", data["error"]) - return None + message = str(data["error"]) + logger.debug("Cookie extraction error: %s", message) + if _is_cookie_decryption_error(message): + raise CookieDecryptionError(source, message) + raise CookieExtractionError(source, message) return data["cookies"] @@ -437,12 +468,20 @@ def extract_browser_cookies(source: str = "auto") -> tuple[str, dict[str, str]] Returns ``(browser_name, cookies)`` on success, or ``None``. """ if source != "auto": - cookies = _extract_in_process(source) - if cookies: - return source, cookies - cookies = _extract_via_subprocess(source) - if cookies: - return source, cookies + extraction_errors: list[CookieExtractionError] = [] + for extractor in (_extract_in_process, _extract_via_subprocess): + try: + cookies = extractor(source) + except CookieExtractionError as exc: + extraction_errors.append(exc) + continue + if cookies: + return source, cookies + for exc in extraction_errors: + if isinstance(exc, CookieDecryptionError): + from .exceptions import BrowserCookieDecryptionError + + raise BrowserCookieDecryptionError(source, exc.message) from exc return None # Auto-detect: try all available browsers @@ -454,27 +493,43 @@ def extract_browser_cookies(source: str = "auto") -> tuple[str, dict[str, str]] from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait - def _try_browser(browser: str) -> tuple[str, dict[str, str]] | None: + def _try_browser(browser: str) -> tuple[str, dict[str, str]] | CookieExtractionError | None: logger.debug("Auto-detect: trying %s …", browser) - cookies = _extract_in_process(browser) - if cookies: - return browser, cookies - cookies = _extract_via_subprocess(browser) - if cookies: - return browser, cookies + extraction_errors: list[CookieExtractionError] = [] + for extractor in (_extract_in_process, _extract_via_subprocess): + try: + cookies = extractor(browser) + except CookieExtractionError as exc: + extraction_errors.append(exc) + continue + if cookies: + return browser, cookies + for exc in extraction_errors: + if isinstance(exc, CookieDecryptionError): + return exc return None + decryption_errors: list[CookieDecryptionError] = [] with ThreadPoolExecutor(max_workers=min(4, len(browsers) or 1)) as pool: pending = {pool.submit(_try_browser, browser) for browser in browsers} while pending: done, pending = wait(pending, return_when=FIRST_COMPLETED) for future in done: result = future.result() + if isinstance(result, CookieDecryptionError): + decryption_errors.append(result) + continue if result: for rest in pending: rest.cancel() return result + if decryption_errors: + from .exceptions import BrowserCookieDecryptionError + + details = "; ".join(f"{exc.source}: {exc.message}" for exc in decryption_errors) + raise BrowserCookieDecryptionError(source, details) + return None diff --git a/xhs_cli/exceptions.py b/xhs_cli/exceptions.py index a2f2715..4fea120 100644 --- a/xhs_cli/exceptions.py +++ b/xhs_cli/exceptions.py @@ -62,3 +62,20 @@ def __init__(self, source: str, details: str = ""): msg += " 2. Make sure you are logged in\n" msg += " 3. Try: xhs login --cookie-source " super().__init__(msg) + + +class BrowserCookieDecryptionError(NoCookieError): + """Raised when browser cookies cannot be decrypted by browser_cookie3.""" + + def __init__(self, source: str, reason: str = ""): + if source == "auto": + msg = "Cookie decryption failed while reading browser cookies." + else: + msg = f"Cookie decryption failed while reading {source} cookies." + if reason: + msg += f"\nReason: {reason}" + msg += "\n\nTroubleshooting:\n" + msg += " 1. Chrome/Edge 127+ on Windows may use App-Bound Encryption, " + msg += "which blocks third-party cookie readers\n" + msg += " 2. Use QR code login instead: xhs login --qrcode" + XhsApiError.__init__(self, msg, code="not_authenticated")