Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions tests/test_cookies.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -21,6 +23,7 @@
save_cookies,
save_note_index,
)
from xhs_cli.exceptions import BrowserCookieDecryptionError


@pytest.fixture
Expand Down Expand Up @@ -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):
Expand Down
89 changes: 72 additions & 17 deletions xhs_cli/cookies.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import functools
import json
import logging
import re
import subprocess
import sys
import threading
Expand All @@ -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
Expand Down Expand Up @@ -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"):
Expand Down Expand Up @@ -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"]

Expand All @@ -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
Expand All @@ -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


Expand Down
17 changes: 17 additions & 0 deletions xhs_cli/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 <browser>"
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")