From 631cb134ec98c8e28716010637f9c8d475cd756d Mon Sep 17 00:00:00 2001 From: chirizcc Date: Fri, 27 Mar 2026 00:15:36 +0800 Subject: [PATCH] feat: add cookiecloud cookie source --- README.md | 72 ++++++++-- tests/test_cli.py | 110 +++++++++++++++ tests/test_cookies.py | 279 +++++++++++++++++++++++++++++++++++++++ xhs_cli/cli.py | 2 +- xhs_cli/commands/auth.py | 18 ++- xhs_cli/cookies.py | 192 ++++++++++++++++++++++++++- xhs_cli/exceptions.py | 15 ++- 7 files changed, 660 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 69a6168..ee16aed 100644 --- a/README.md +++ b/README.md @@ -150,15 +150,30 @@ xiaohongshu-cli supports multiple authentication methods: 1. **Saved cookies** — loads from `~/.xiaohongshu-cli/cookies.json` 2. **Browser cookies** — auto-detects installed browsers and extracts cookies (supports Chrome, Arc, Edge, Firefox, Safari, Brave, Chromium, Opera, Vivaldi, and more) -3. **QR code login** — browser-assisted login with terminal QR output (`xhs login --qrcode`) +3. **CookieCloud** — pulls cookies from a CookieCloud instance using environment variables only; useful when you keep your browser cookies synced remotely +4. **QR code login** — browser-assisted login with terminal QR output (`xhs login --qrcode`) -`xhs login` automatically tries all installed browsers and uses the first one with valid cookies. -Use `--cookie-source ` to specify a browser explicitly, or `--qrcode` for browser-assisted QR login. -Other authenticated commands automatically retry once with fresh browser cookies when the saved session has expired. +`xhs login` automatically tries installed browser providers and uses the first one with valid cookies. +Use `--cookie-source ` to specify a provider explicitly, such as a browser name or `cookiecloud`, or use `--qrcode` for browser-assisted QR login. +Other authenticated commands automatically retry once with fresh cookies from the selected source when the saved session has expired. + +CookieCloud uses these environment variables: + +```bash +export COOKIECLOUD_HOST="https://cookiecloud.example.com" +export COOKIECLOUD_UUID="your-uuid" +export COOKIECLOUD_PASSWORD="your-password" +export COOKIECLOUD_TIMEOUT="10" # optional +``` + +```bash +xhs login --cookie-source cookiecloud +xhs --cookie-source cookiecloud status +``` ### Cookie TTL -Saved cookies are valid for **7 days** by default. After that, the client automatically attempts to refresh from the browser. If browser extraction fails, the existing cookies are used with a warning. +Saved cookies are valid for **7 days** by default. After that, the client automatically attempts to refresh from the browser. If `cookiecloud` is the selected source, stale saved cookies can be refreshed from CookieCloud instead of the browser. If refresh fails, the existing cookies are used with a warning. ### Short-Index Navigation @@ -287,7 +302,23 @@ uv run ruff check . 1. Open any browser and visit https://www.xiaohongshu.com/ 2. Log in with your account -3. Run `xhs login` (auto-detects browser) or `xhs login --cookie-source ` +3. Run `xhs login` (auto-detects browser), `xhs login --cookie-source `, or `xhs login --cookie-source cookiecloud` if you use CookieCloud + +**Q: CookieCloud config is missing** + +Set `COOKIECLOUD_HOST`, `COOKIECLOUD_UUID`, and `COOKIECLOUD_PASSWORD`, then retry with `--cookie-source cookiecloud`. + +**Q: CookieCloud decryption failed** + +Check that the host, UUID, and password match your CookieCloud instance. A wrong password or mismatched UUID will prevent cookie decryption. + +**Q: CookieCloud returned no Xiaohongshu cookies** + +Confirm that your CookieCloud source contains Xiaohongshu cookies and that they have been synced from a logged-in browser session. + +**Q: CookieCloud returned cookies, but XHS still shows guest access** + +The fetched cookies may be expired, incomplete, or not tied to a logged-in Xiaohongshu session. Refresh the source, then rerun `xhs --cookie-source cookiecloud status`. **Q: `NeedVerifyError: Captcha required`** @@ -432,18 +463,37 @@ xiaohongshu-cli 支持多种认证方式： 1. **已保存 Cookie** — 从 `~/.xiaohongshu-cli/cookies.json` 加载 2. **浏览器 Cookie** — 自动检测已安装浏览器并提取（支持 Chrome、Arc、Edge、Firefox、Safari、Brave、Chromium、Opera、Vivaldi 等） -3. **二维码扫码登录** — browser-assisted 登录，终端显示二维码，用小红书 App 扫码（`xhs login --qrcode`） +3. **CookieCloud** — 通过环境变量从 CookieCloud 实例拉取 Cookie，适合把浏览器 Cookie 远程同步管理 +4. **二维码扫码登录** — browser-assisted 登录，终端显示二维码，用小红书 App 扫码（`xhs login --qrcode`） + +Cookie 保存后有效期 **7 天**，超时后自动尝试刷新。若当前选择的是 `cookiecloud`，过期的本地 Cookie 也可以从 CookieCloud 刷新，而不是只依赖浏览器提取。 + +`xhs login` 会自动尝试已安装的浏览器来源，并使用第一个有有效 Cookie 的来源。也可用 `--cookie-source ` 明确指定来源，例如某个浏览器名或 `cookiecloud`，或用 `--qrcode` 使用 browser-assisted 二维码登录。CookieCloud 需要通过环境变量配置： -Cookie 保存后有效期 **7 天**，超时后自动尝试从浏览器刷新。 +```bash +export COOKIECLOUD_HOST="https://cookiecloud.example.com" +export COOKIECLOUD_UUID="your-uuid" +export COOKIECLOUD_PASSWORD="your-password" +export COOKIECLOUD_TIMEOUT="10" # 可选 +``` + +```bash +xhs login --cookie-source cookiecloud +xhs --cookie-source cookiecloud status +``` -`xhs login` 会自动尝试所有已安装浏览器，使用第一个有有效 Cookie 的浏览器。也可用 `--cookie-source ` 指定浏览器，或 `--qrcode` 使用 browser-assisted 二维码登录。其他需认证命令在 session 过期时会自动重试一次。 +其他需认证命令在 session 过期时会自动重试一次。 ## 常见问题 -- `NoCookieError: No 'a1' cookie found` — 请先在任意浏览器打开 https://www.xiaohongshu.com/ 并登录，然后执行 `xhs login` +- `NoCookieError: No 'a1' cookie found` — 请先在任意浏览器打开 https://www.xiaohongshu.com/ 并登录，然后执行 `xhs login`，或配置 CookieCloud 后使用 `xhs login --cookie-source cookiecloud` +- `CookieCloud 配置缺失` — 请设置 `COOKIECLOUD_HOST`、`COOKIECLOUD_UUID`、`COOKIECLOUD_PASSWORD` 后重试 +- `CookieCloud 解密失败` — 检查 CookieCloud 地址、UUID 和密码是否与实例一致 +- `CookieCloud 没有小红书 Cookie` — 确认 CookieCloud 中确实同步了已登录的小红书 Cookie +- `CookieCloud 拉到了 Cookie，但仍然显示游客` — 说明 Cookie 可能已过期、内容不完整，或并非来自已登录的小红书会话；先刷新来源，再执行 `xhs --cookie-source cookiecloud status` - `NeedVerifyError` — 触发了验证码，请到浏览器中完成验证后重试 - `IpBlockedError` — IP 被限制，尝试切换网络（手机热点或 VPN） -- `SessionExpiredError` — Cookie 过期，执行 `xhs login` 刷新 +- `SessionExpiredError` — Cookie 过期，执行 `xhs login` 刷新；如果使用 CookieCloud，先刷新 CookieCloud 再重试 - 请求较慢是正常的 — 内置高斯随机延迟（~1-1.5s）是为了模拟人类浏览行为，避免触发风控 ## 作为 AI Agent Skill 使用 diff --git a/tests/test_cli.py b/tests/test_cli.py index f4bec77..5cf7870 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -44,6 +44,9 @@ def test_help(self): assert "xhs" in result.output assert "search" in result.output assert "read" in result.output + assert "browser-only" not in result.output.lower() + assert "cookiecloud" in result.output.lower() + assert "installed browsers" in result.output.lower() def test_search_help(self): result = runner.invoke(cli, ["search", "--help"]) @@ -57,11 +60,118 @@ def test_read_help(self): def test_login_help(self): result = runner.invoke(cli, ["login", "--help"]) assert result.exit_code == 0 + assert "browser-only" not in result.output.lower() + assert "cookiecloud" in result.output.lower() + assert "installed browsers" in result.output.lower() + + def test_help_mentions_cookie_source_as_provider(self): + result = runner.invoke(cli, ["--help"]) + assert result.exit_code == 0 + assert "browser-only" not in result.output.lower() + assert "cookiecloud" in result.output.lower() def test_status_help(self): result = runner.invoke(cli, ["status", "--help"]) assert result.exit_code == 0 + def test_status_uses_group_cookie_source_option(self, monkeypatch): + monkeypatch.setenv("OUTPUT", "auto") + monkeypatch.setattr( + "xhs_cli.commands.auth.run_client_action", + lambda ctx, action: {"nickname": "Alice", "red_id": "alice001"}, + ) + + result = runner.invoke(cli, ["--cookie-source", "cookiecloud", "status", "--yaml"]) + + assert result.exit_code == 0 + payload = yaml.safe_load(result.output) + assert payload["ok"] is True + assert payload["data"]["authenticated"] is True + assert payload["data"]["user"]["name"] == "Alice" + + def test_login_cookiecloud_yaml_succeeds_through_shared_path(self, monkeypatch): + monkeypatch.setenv("OUTPUT", "auto") + monkeypatch.setattr( + "xhs_cli.commands.auth.get_cookies", + lambda source, force_refresh=False: ("cookiecloud", {"a1": "x"}), + ) + + class FakeClient: + def __init__(self, cookies): + self.cookies = cookies + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def get_self_info(self): + return {"nickname": "Alice", "red_id": "alice001"} + + monkeypatch.setattr("xhs_cli.commands.auth.XhsClient", FakeClient) + monkeypatch.setattr( + "xhs_cli.commands.auth.normalize_xhs_user_payload", + lambda info: { + "guest": False, + "nickname": "Alice", + "red_id": "alice001", + "ip_location": "", + "desc": "", + "id": "u-1", + }, + ) + + result = runner.invoke(cli, ["login", "--cookie-source", "cookiecloud", "--yaml"]) + + assert result.exit_code == 0 + payload = yaml.safe_load(result.output) + assert payload["ok"] is True + assert payload["data"]["authenticated"] is True + assert payload["data"]["user"]["nickname"] == "Alice" + + def test_login_cookiecloud_invalid_session_message_is_source_aware(self, monkeypatch): + monkeypatch.setenv("OUTPUT", "auto") + monkeypatch.setattr( + "xhs_cli.commands.auth.get_cookies", + lambda source, force_refresh=False: ("cookiecloud", {"a1": "x"}), + ) + monkeypatch.setattr("xhs_cli.commands.auth.time.sleep", lambda _seconds: None) + + class FakeClient: + def __init__(self, cookies): + self.cookies = cookies + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def get_self_info(self): + return {"nickname": "Unknown", "red_id": ""} + + monkeypatch.setattr("xhs_cli.commands.auth.XhsClient", FakeClient) + monkeypatch.setattr( + "xhs_cli.commands.auth.normalize_xhs_user_payload", + lambda info: { + "guest": True, + "nickname": "Unknown", + "red_id": "", + "ip_location": "", + "desc": "", + "id": "", + }, + ) + + result = runner.invoke(cli, ["login", "--cookie-source", "cookiecloud", "--yaml"]) + + assert result.exit_code != 0 + payload = yaml.safe_load(result.output) + assert payload["ok"] is False + assert "Browser cookies were extracted" not in payload["error"]["message"] + assert "cookiecloud" in payload["error"]["message"] + def test_all_commands_registered(self): result = runner.invoke(cli, ["--help"]) commands_expected = [ diff --git a/tests/test_cookies.py b/tests/test_cookies.py index aae7636..778f060 100644 --- a/tests/test_cookies.py +++ b/tests/test_cookies.py @@ -1,15 +1,22 @@ """Unit tests for cookie management (no network required).""" +import base64 +import hashlib +import json import time +import httpx import pytest +from Crypto.Cipher import AES +from Crypto.Util.Padding import pad from xhs_cli.cookies import ( NOTE_CONTEXT_TTL_SECONDS, cache_note_context, clear_cookies, cookies_to_string, + extract_cookiecloud_cookies, get_cached_note_context, get_cached_xsec_token, get_cookies, @@ -21,6 +28,31 @@ save_cookies, save_note_index, ) +from xhs_cli.exceptions import NoCookieError, XhsApiError + + +def _evp_bytes_to_key(password: bytes, salt: bytes, key_len: int = 32, iv_len: int = 16) -> tuple[bytes, bytes]: + derived = b"" + block = b"" + while len(derived) < key_len + iv_len: + block = hashlib.md5(block + password + salt).digest() + derived += block + return derived[:key_len], derived[key_len : key_len + iv_len] + + +def _encrypt_cookiecloud_payload( + payload: dict[str, object], + *, + uuid: str = "uuid-1", + password: str = "secret-1", + salt: bytes = b"12345678", +) -> dict[str, str]: + passphrase = hashlib.md5(f"{uuid}-{password}".encode()).hexdigest()[:16].encode() + key, iv = _evp_bytes_to_key(passphrase, salt) + cipher = AES.new(key, AES.MODE_CBC, iv) + ciphertext = cipher.encrypt(pad(json.dumps(payload).encode(), AES.block_size)) + encrypted = base64.b64encode(b"Salted__" + salt + ciphertext).decode() + return {"encrypted": encrypted} @pytest.fixture @@ -110,6 +142,253 @@ def test_force_refresh_bypasses_saved_cookies(self, monkeypatch): assert cookies == {"a1": "fresh"} assert saved == [{"a1": "fresh"}] + def test_force_refresh_uses_cookiecloud_provider(self, monkeypatch): + monkeypatch.setattr("xhs_cli.cookies.load_saved_cookies", lambda: {"a1": "saved"}) + monkeypatch.setattr( + "xhs_cli.cookies.extract_cookiecloud_cookies", + lambda: ("cookiecloud", {"a1": "fresh-cookie"}), + ) + saved = [] + monkeypatch.setattr("xhs_cli.cookies.save_cookies", lambda cookies: saved.append(cookies)) + + browser, cookies = get_cookies("cookiecloud", force_refresh=True) + + assert browser == "cookiecloud" + assert cookies == {"a1": "fresh-cookie"} + assert saved == [{"a1": "fresh-cookie"}] + + def test_stale_saved_cookies_refresh_from_cookiecloud(self, monkeypatch): + now = 1_000_000.0 + monkeypatch.setattr( + "xhs_cli.cookies.load_saved_cookies", + lambda: {"a1": "stale-cookie", "saved_at": now - (8 * 86400)}, + ) + monkeypatch.setattr("xhs_cli.cookies.time.time", lambda: now) + monkeypatch.setattr( + "xhs_cli.cookies.extract_cookiecloud_cookies", + lambda: ("cookiecloud", {"a1": "fresh-cookie"}), + ) + saved = [] + monkeypatch.setattr("xhs_cli.cookies.save_cookies", lambda cookies: saved.append(cookies)) + + browser, cookies = get_cookies("cookiecloud") + + assert browser == "cookiecloud" + assert cookies == {"a1": "fresh-cookie"} + assert saved == [{"a1": "fresh-cookie"}] + + def test_force_refresh_cookiecloud_raises_when_no_usable_cookies(self, monkeypatch): + monkeypatch.setattr("xhs_cli.cookies.extract_cookiecloud_cookies", lambda: None) + + with pytest.raises(NoCookieError): + get_cookies("cookiecloud", force_refresh=True) + + +class TestCookieCloudExtraction: + def test_cookiecloud_requires_config(self, monkeypatch): + monkeypatch.delenv("COOKIECLOUD_HOST", raising=False) + monkeypatch.delenv("COOKIECLOUD_UUID", raising=False) + monkeypatch.delenv("COOKIECLOUD_PASSWORD", raising=False) + + with pytest.raises(XhsApiError, match="COOKIECLOUD_HOST"): + extract_cookiecloud_cookies() + + def test_extract_cookiecloud_cookies_returns_xhs_cookie_dict(self, monkeypatch): + monkeypatch.setenv("COOKIECLOUD_HOST", "https://cookiecloud.example.com") + monkeypatch.setenv("COOKIECLOUD_UUID", "uuid-1") + monkeypatch.setenv("COOKIECLOUD_PASSWORD", "secret-1") + payload = _encrypt_cookiecloud_payload( + { + "cookie_data": { + ".xiaohongshu.com": [ + {"name": "a1", "value": "a1-cookie"}, + {"name": "web_session", "value": "session-cookie"}, + ], + "xiaohongshu.com": [ + {"name": "web_session_sec", "value": "session-sec-cookie"}, + ], + ".example.com": [ + {"name": "ignore_me", "value": "1"}, + ], + } + } + ) + monkeypatch.setattr("xhs_cli.cookies._fetch_cookiecloud_payload", lambda config: payload) + + result = extract_cookiecloud_cookies() + + assert result == ( + "cookiecloud", + { + "a1": "a1-cookie", + "web_session": "session-cookie", + "web_session_sec": "session-sec-cookie", + }, + ) + + def test_extract_cookiecloud_cookies_returns_none_without_xhs_domain(self, monkeypatch): + monkeypatch.setenv("COOKIECLOUD_HOST", "https://cookiecloud.example.com") + monkeypatch.setenv("COOKIECLOUD_UUID", "uuid-1") + monkeypatch.setenv("COOKIECLOUD_PASSWORD", "secret-1") + payload = _encrypt_cookiecloud_payload( + { + "cookie_data": { + ".example.com": [ + {"name": "a1", "value": "other-cookie"}, + ] + } + } + ) + monkeypatch.setattr("xhs_cli.cookies._fetch_cookiecloud_payload", lambda config: payload) + + assert extract_cookiecloud_cookies() is None + + def test_extract_cookiecloud_cookies_returns_none_without_a1(self, monkeypatch): + monkeypatch.setenv("COOKIECLOUD_HOST", "https://cookiecloud.example.com") + monkeypatch.setenv("COOKIECLOUD_UUID", "uuid-1") + monkeypatch.setenv("COOKIECLOUD_PASSWORD", "secret-1") + payload = _encrypt_cookiecloud_payload( + { + "cookie_data": { + ".xiaohongshu.com": [ + {"name": "web_session", "value": "session-cookie"}, + ] + } + } + ) + monkeypatch.setattr("xhs_cli.cookies._fetch_cookiecloud_payload", lambda config: payload) + + assert extract_cookiecloud_cookies() is None + + def test_extract_cookiecloud_cookies_raises_on_decrypt_failure(self, monkeypatch): + monkeypatch.setenv("COOKIECLOUD_HOST", "https://cookiecloud.example.com") + monkeypatch.setenv("COOKIECLOUD_UUID", "uuid-1") + monkeypatch.setenv("COOKIECLOUD_PASSWORD", "secret-1") + monkeypatch.setattr( + "xhs_cli.cookies._fetch_cookiecloud_payload", + lambda config: {"encrypted": "not-valid-base64"}, + ) + + with pytest.raises(XhsApiError, match="decrypt"): + extract_cookiecloud_cookies() + + +class TestCookieCloudFetchErrors: + def test_fetch_cookiecloud_payload_timeout_raises_actionable_error(self, monkeypatch): + def fake_get(url, timeout, follow_redirects): + raise httpx.TimeoutException("timeout") + + monkeypatch.setattr("xhs_cli.cookies.httpx.get", fake_get) + + from xhs_cli.cookies import _fetch_cookiecloud_payload + + with pytest.raises(XhsApiError, match="timed out"): + _fetch_cookiecloud_payload( + {"host": "https://cookiecloud.example.com", "uuid": "uuid-1", "timeout": 10.0} + ) + + def test_fetch_cookiecloud_payload_http_status_error_raises_actionable_error(self, monkeypatch): + request = httpx.Request("GET", "https://cookiecloud.example.com/get/uuid-1") + response = httpx.Response(502, request=request) + + def fake_get(url, timeout, follow_redirects): + return response + + monkeypatch.setattr("xhs_cli.cookies.httpx.get", fake_get) + + from xhs_cli.cookies import _fetch_cookiecloud_payload + + with pytest.raises(XhsApiError, match="HTTP 502"): + _fetch_cookiecloud_payload( + {"host": "https://cookiecloud.example.com", "uuid": "uuid-1", "timeout": 10.0} + ) + + def test_fetch_cookiecloud_payload_http_error_raises_actionable_error(self, monkeypatch): + def fake_get(url, timeout, follow_redirects): + raise httpx.HTTPError("network down") + + monkeypatch.setattr("xhs_cli.cookies.httpx.get", fake_get) + + from xhs_cli.cookies import _fetch_cookiecloud_payload + + with pytest.raises(XhsApiError, match="request failed"): + _fetch_cookiecloud_payload( + {"host": "https://cookiecloud.example.com", "uuid": "uuid-1", "timeout": 10.0} + ) + + def test_fetch_cookiecloud_payload_invalid_json_raises_actionable_error(self, monkeypatch): + request = httpx.Request("GET", "https://cookiecloud.example.com/get/uuid-1") + response = httpx.Response(200, request=request, content=b"not-json") + + def fake_get(url, timeout, follow_redirects): + return response + + monkeypatch.setattr("xhs_cli.cookies.httpx.get", fake_get) + + from xhs_cli.cookies import _fetch_cookiecloud_payload + + with pytest.raises(XhsApiError, match="invalid JSON"): + _fetch_cookiecloud_payload( + {"host": "https://cookiecloud.example.com", "uuid": "uuid-1", "timeout": 10.0} + ) + + def test_fetch_cookiecloud_payload_invalid_shape_raises_actionable_error(self, monkeypatch): + request = httpx.Request("GET", "https://cookiecloud.example.com/get/uuid-1") + response = httpx.Response(200, request=request, json=["not", "a", "dict"]) + + def fake_get(url, timeout, follow_redirects): + return response + + monkeypatch.setattr("xhs_cli.cookies.httpx.get", fake_get) + + from xhs_cli.cookies import _fetch_cookiecloud_payload + + with pytest.raises(XhsApiError, match="invalid response payload"): + _fetch_cookiecloud_payload( + {"host": "https://cookiecloud.example.com", "uuid": "uuid-1", "timeout": 10.0} + ) + + def test_fetch_cookiecloud_payload_missing_encrypted_field_raises_actionable_error(self, monkeypatch): + request = httpx.Request("GET", "https://cookiecloud.example.com/get/uuid-1") + response = httpx.Response(200, request=request, json={"cookie_data": {}}) + + def fake_get(url, timeout, follow_redirects): + return response + + monkeypatch.setattr("xhs_cli.cookies.httpx.get", fake_get) + + from xhs_cli.cookies import _fetch_cookiecloud_payload + + with pytest.raises(XhsApiError, match="encrypted payload"): + _fetch_cookiecloud_payload( + {"host": "https://cookiecloud.example.com", "uuid": "uuid-1", "timeout": 10.0} + ) + + +class TestNoCookieError: + def test_cookiecloud_message_is_actionable(self): + message = str(NoCookieError("cookiecloud")) + + assert "CookieCloud" in message + assert "COOKIECLOUD_HOST" in message + assert "COOKIECLOUD_UUID" in message + assert "COOKIECLOUD_PASSWORD" in message + assert "xhs login --cookie-source cookiecloud" in message + + def test_named_browser_message_keeps_browser_guidance(self): + message = str(NoCookieError("chrome")) + + assert "in chrome" in message + assert "Open a browser" in message + assert "xhs login --cookie-source " in message + + def test_auto_message_keeps_browser_autodetect_guidance(self): + message = str(NoCookieError("auto")) + + assert "in any installed browser" in message + assert "Open a browser" in message + assert "xhs login --cookie-source " in message + class TestNoteContextCache: def test_cache_persists_token_and_source(self, tmp_config_dir): diff --git a/xhs_cli/cli.py b/xhs_cli/cli.py index a33a08e..80de993 100644 --- a/xhs_cli/cli.py +++ b/xhs_cli/cli.py @@ -54,7 +54,7 @@ def _fix_windows_encoding() -> None: type=str, default="auto", show_default=True, - help="Browser to read cookies from (auto = try all installed browsers)", + help="Cookie provider/source to read cookies from (auto = try installed browsers; use 'cookiecloud' explicitly)", ) @click.pass_context def cli(ctx, verbose: bool, cookie_source: str): diff --git a/xhs_cli/commands/auth.py b/xhs_cli/commands/auth.py index 5a50321..72939b7 100644 --- a/xhs_cli/commands/auth.py +++ b/xhs_cli/commands/auth.py @@ -52,7 +52,10 @@ def _print_status_summary(user: dict[str, object]) -> None: "--cookie-source", type=str, default=None, - help="Browser to read cookies from (default: auto-detect all installed browsers)", + help=( + "Cookie provider/source to read cookies from " + "(default: auto-detect installed browsers; use 'cookiecloud' explicitly)" + ), ) @structured_output_options @click.option("--qrcode", "use_qrcode", is_flag=True, default=False, @@ -94,13 +97,14 @@ def _login_with_qrcode() -> None: ) return - # Browser cookie extraction (default) + # Cookie provider/source extraction (default) if cookie_source is None: cookie_source = ctx.obj.get("cookie_source", "auto") if ctx.obj else "auto" - def _login_with_browser() -> None: - browser, cookies = get_cookies(cookie_source, force_refresh=True) - print_success(f"Cookies extracted from {browser}") + def _login_with_cookie_source() -> None: + source_name, cookies = get_cookies(cookie_source, force_refresh=True) + if not (as_json or as_yaml): + print_success(f"Cookies extracted from {source_name}") # Verify by fetching user info, retry once if session not yet propagated with XhsClient(cookies) as client: @@ -115,7 +119,7 @@ def _login_with_browser() -> None: if not _is_valid_login(user): raise XhsApiError( - "Browser cookies were extracted, but the session appears invalid " + f"Cookies were extracted from {source_name}, but the session appears invalid " "(guest or incomplete profile). Try: xhs login --qrcode" ) @@ -123,7 +127,7 @@ def _login_with_browser() -> None: _print_login_success(user) handle_errors( - _login_with_browser, + _login_with_cookie_source, as_json=as_json, as_yaml=as_yaml, prefix="Login verification failed", diff --git a/xhs_cli/cookies.py b/xhs_cli/cookies.py index 96665a4..d28277b 100644 --- a/xhs_cli/cookies.py +++ b/xhs_cli/cookies.py @@ -2,9 +2,12 @@ from __future__ import annotations +import base64 import functools +import hashlib import json import logging +import os import subprocess import sys import threading @@ -13,6 +16,10 @@ from pathlib import Path from typing import Any +import httpx +from Crypto.Cipher import AES +from Crypto.Util.Padding import unpad + from .constants import CONFIG_DIR_NAME, COOKIE_FILE, INDEX_CACHE_FILE, TOKEN_CACHE_FILE logger = logging.getLogger(__name__) @@ -20,6 +27,7 @@ # Cookie TTL: warn and attempt browser refresh after 7 days COOKIE_TTL_DAYS = 7 _COOKIE_TTL_SECONDS = COOKIE_TTL_DAYS * 86400 +COOKIECLOUD_TIMEOUT_SECONDS = 10.0 _TOKEN_CACHE_LOCK = threading.RLock() _TOKEN_CACHE_MEMORY: OrderedDict[str, dict[str, Any]] | None = None _TOKEN_CACHE_PATH: Path | None = None @@ -316,6 +324,173 @@ def get_cached_xsec_token(note_id: str) -> str: return get_cached_note_context(note_id).get("token", "") +def _load_cookiecloud_config() -> dict[str, str | float]: + from .exceptions import XhsApiError + + host = os.getenv("COOKIECLOUD_HOST", "").strip().rstrip("/") + uuid = os.getenv("COOKIECLOUD_UUID", "").strip() + password = os.getenv("COOKIECLOUD_PASSWORD", "").strip() + + missing = [ + name + for name, value in ( + ("COOKIECLOUD_HOST", host), + ("COOKIECLOUD_UUID", uuid), + ("COOKIECLOUD_PASSWORD", password), + ) + if not value + ] + if missing: + raise XhsApiError( + "CookieCloud requires environment variables: " + f"{', '.join(missing)}. Try: xhs login --cookie-source cookiecloud" + ) + + timeout = COOKIECLOUD_TIMEOUT_SECONDS + timeout_raw = os.getenv("COOKIECLOUD_TIMEOUT", "").strip() + if timeout_raw: + try: + timeout = float(timeout_raw) + except ValueError as exc: + raise XhsApiError("COOKIECLOUD_TIMEOUT must be a positive number of seconds.") from exc + if timeout <= 0: + raise XhsApiError("COOKIECLOUD_TIMEOUT must be a positive number of seconds.") + + return { + "host": host, + "uuid": uuid, + "password": password, + "timeout": timeout, + } + + +def _fetch_cookiecloud_payload(config: dict[str, str | float]) -> dict[str, Any]: + from .exceptions import XhsApiError + + url = f"{config['host']}/get/{config['uuid']}" + try: + response = httpx.get(url, timeout=float(config["timeout"]), follow_redirects=True) + response.raise_for_status() + except httpx.TimeoutException as exc: + raise XhsApiError("CookieCloud request timed out. Check COOKIECLOUD_HOST and try again.") from exc + except httpx.HTTPStatusError as exc: + raise XhsApiError( + f"CookieCloud request failed with HTTP {exc.response.status_code}. " + "Check COOKIECLOUD_HOST and try again." + ) from exc + except httpx.HTTPError as exc: + raise XhsApiError("CookieCloud request failed. Check COOKIECLOUD_HOST and try again.") from exc + + try: + payload = response.json() + except ValueError as exc: + raise XhsApiError("CookieCloud returned invalid JSON.") from exc + + if not isinstance(payload, dict): + raise XhsApiError("CookieCloud returned an invalid response payload.") + + encrypted = payload.get("encrypted") + if not isinstance(encrypted, str) or not encrypted.strip(): + raise XhsApiError("CookieCloud response did not include an encrypted payload.") + + return payload + + +def _evp_bytes_to_key(password: bytes, salt: bytes, *, key_len: int = 32, iv_len: int = 16) -> tuple[bytes, bytes]: + derived = b"" + block = b"" + while len(derived) < key_len + iv_len: + block = hashlib.md5(block + password + salt).digest() + derived += block + return derived[:key_len], derived[key_len : key_len + iv_len] + + +def _cookiecloud_passphrases(uuid: str, password: str) -> tuple[str, ...]: + candidates = [ + hashlib.md5(f"{uuid}-{password}".encode()).hexdigest()[:16], + hashlib.md5(f"{uuid}{password}".encode()).hexdigest()[:16], + ] + + unique: list[str] = [] + for candidate in candidates: + if candidate not in unique: + unique.append(candidate) + return tuple(unique) + + +def _decrypt_cookiecloud_payload(payload: dict[str, Any], uuid: str, password: str) -> dict[str, Any]: + from .exceptions import XhsApiError + + encrypted = payload.get("encrypted") + if not isinstance(encrypted, str) or not encrypted.strip(): + raise XhsApiError("CookieCloud response did not include an encrypted payload.") + + for passphrase in _cookiecloud_passphrases(uuid, password): + try: + raw_encrypted = base64.b64decode(encrypted, validate=True) + if ( + len(raw_encrypted) <= 16 + or len(raw_encrypted) % AES.block_size != 0 + or raw_encrypted[:8] != b"Salted__" + ): + continue + + salt = raw_encrypted[8:16] + ciphertext = raw_encrypted[16:] + key, iv = _evp_bytes_to_key(passphrase.encode(), salt) + cipher = AES.new(key, AES.MODE_CBC, iv) + decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size) + parsed = json.loads(decrypted.decode()) + except (ValueError, UnicodeDecodeError, json.JSONDecodeError): + continue + + if isinstance(parsed, dict): + return parsed + + raise XhsApiError( + "Failed to decrypt CookieCloud payload. Check COOKIECLOUD_UUID and COOKIECLOUD_PASSWORD." + ) + + +def _extract_xhs_cookiecloud_cookies(payload: dict[str, Any]) -> dict[str, str] | None: + from .exceptions import XhsApiError + + cookie_data = payload.get("cookie_data", payload) + if not isinstance(cookie_data, dict): + raise XhsApiError("CookieCloud payload did not include a valid cookie_data object.") + + cookies: dict[str, str] = {} + for domain in ("xiaohongshu.com", ".xiaohongshu.com"): + entries = cookie_data.get(domain, []) + if not isinstance(entries, list): + continue + for entry in entries: + if not isinstance(entry, dict): + continue + name = str(entry.get("name", "")).strip() + value = str(entry.get("value", "")).strip() + if name and value: + cookies[name] = value + + if not cookies.get("a1"): + return None + return cookies + + +def extract_cookiecloud_cookies() -> tuple[str, dict[str, str]] | None: + """Fetch, decrypt, and normalize CookieCloud cookies for xiaohongshu.com.""" + config = _load_cookiecloud_config() + payload = _fetch_cookiecloud_payload(config) + decrypted = _decrypt_cookiecloud_payload(payload, str(config["uuid"]), str(config["password"])) + cookies = _extract_xhs_cookiecloud_cookies(decrypted) + if not cookies: + logger.debug("No usable Xiaohongshu cookies found in CookieCloud payload") + return None + + logger.debug("Loaded XHS cookies from CookieCloud") + return "cookiecloud", cookies + + @functools.lru_cache(maxsize=1) def _available_browsers() -> tuple[str, ...]: """List all browser names supported by browser_cookie3 (cached).""" @@ -484,23 +659,30 @@ def get_cookies( """ Multi-strategy cookie acquisition with TTL-based auto-refresh. - Returns ``(browser_name, cookies)``. + Returns ``(source_name, cookies)``. 1. Load saved cookies (skip if stale > 7 days) - 2. Extract from browser (auto-detect if *cookie_source* is ``"auto"``) + 2. Extract from the selected cookie source 3. Raise error if all fail """ + def _extract_selected_cookies() -> tuple[str, dict[str, str]] | None: + if cookie_source == "cookiecloud": + return extract_cookiecloud_cookies() + return extract_browser_cookies(cookie_source) + # 1. Try saved cookies first if not force_refresh: saved = load_saved_cookies() if saved: + saved = dict(saved) saved_at = saved.pop("saved_at", 0) if saved_at and (time.time() - float(saved_at)) > _COOKIE_TTL_SECONDS: logger.info( - "Cookies older than %d days, attempting browser refresh", + "Cookies older than %d days, attempting refresh from %s", COOKIE_TTL_DAYS, + cookie_source, ) - result = extract_browser_cookies(cookie_source) + result = _extract_selected_cookies() if result: save_cookies(result[1]) return result @@ -513,7 +695,7 @@ def get_cookies( # 2. Try browser extraction from .exceptions import NoCookieError - result = extract_browser_cookies(cookie_source) + result = _extract_selected_cookies() if result: save_cookies(result[1]) return result diff --git a/xhs_cli/exceptions.py b/xhs_cli/exceptions.py index a2f2715..240d2b2 100644 --- a/xhs_cli/exceptions.py +++ b/xhs_cli/exceptions.py @@ -51,14 +51,21 @@ class NoCookieError(XhsApiError): """Raised when no valid cookies are found.""" def __init__(self, source: str, details: str = ""): - if source == "auto": + if source == "cookiecloud": + msg = "No usable Xiaohongshu cookies were found in CookieCloud." + elif source == "auto": msg = "No 'a1' cookie found for xiaohongshu.com in any installed browser." else: msg = f"No 'a1' cookie found for xiaohongshu.com in {source}." if details: msg += f"\n{details}" msg += "\n\nTroubleshooting:\n" - msg += " 1. Open a browser and visit https://www.xiaohongshu.com/\n" - msg += " 2. Make sure you are logged in\n" - msg += " 3. Try: xhs login --cookie-source " + if source == "cookiecloud": + msg += " 1. Check COOKIECLOUD_HOST, COOKIECLOUD_UUID, and COOKIECLOUD_PASSWORD\n" + msg += " 2. Confirm CookieCloud contains xiaohongshu.com cookies with a valid a1\n" + msg += " 3. Try: xhs login --cookie-source cookiecloud" + else: + msg += " 1. Open a browser and visit https://www.xiaohongshu.com/\n" + msg += " 2. Make sure you are logged in\n" + msg += " 3. Try: xhs login --cookie-source " super().__init__(msg)