From 053a3418644516d2aa6c49878905a4ccb1e45ca0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E9=87=91=E9=BE=99?= Date: Mon, 22 Jun 2026 12:09:22 +0800 Subject: [PATCH] =?UTF-8?q?fix(captcha):=20API=20=E6=89=93=E7=A0=81?= =?UTF-8?q?=E6=B3=A8=E5=85=A5=20solution=20userAgent=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=20reCAPTCHA=20UA=20=E4=B8=8D=E4=B8=80=E8=87=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit YesCaptcha/CapMonster/EzCaptcha/CapSolver 等 API 打码返回的 solution 包含 gRecaptchaResponse 与 userAgent 两个字段。 _get_api_captcha_token 之前只返回 gRecaptchaResponse, 后续 Flow API 请求继续用 _generate_user_agent 生成的 macOS UA, 与打码时的 Windows UA 不一致, 触发 PUBLIC_ERROR_UNUSUAL_ACTIVITY: reCAPTCHA evaluation failed。 将 userAgent 合并到 fingerprint context, 让 _make_request 自动 沿用打码 UA, sec-ch-ua-platform 由 fallback 逻辑按 UA 自动推断 (Windows → "Windows", mac → "macOS" 等)。 新增 tests/test_api_captcha_fingerprint.py 覆盖该路径。 Co-Authored-By: Claude Opus 4.6 --- src/services/flow_client.py | 37 +++++++++-- tests/test_api_captcha_fingerprint.py | 90 +++++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 4 deletions(-) create mode 100644 tests/test_api_captcha_fingerprint.py diff --git a/src/services/flow_client.py b/src/services/flow_client.py index 615d0fd0..2746d453 100644 --- a/src/services/flow_client.py +++ b/src/services/flow_client.py @@ -2993,20 +2993,46 @@ async def _get_recaptcha_token( self._set_request_fingerprint(None) else: self._set_request_fingerprint(None) - token = await self._get_api_captcha_token(captcha_method, project_id, action) + api_result = await self._get_api_captcha_token(captcha_method, project_id, action) + if api_result is None: + return None, None + token, captcha_user_agent = api_result + # 把打码服务返回的 userAgent 合并到 fingerprint, 让 Flow API 提交请求沿用同一 UA。 + # 否则 Google reCAPTCHA V3 评估会因 UA 不一致判定 UNUSUAL_ACTIVITY 并返回 + # "reCAPTCHA evaluation failed"。其他 Client Hint (sec-ch-ua-platform 等) + # 由 _make_request 根据 UA 自动推断 (Windows → "Windows", etc.)。 + if captcha_user_agent: + existing_fp = self._request_fingerprint_ctx.get() + merged_fp = dict(existing_fp) if isinstance(existing_fp, dict) else {} + merged_fp["user_agent"] = captcha_user_agent + self._set_request_fingerprint(merged_fp) + debug_logger.log_info( + f"[reCAPTCHA {captcha_method}] 已将打码 UA 注入请求指纹: " + f"{captcha_user_agent[:80]}" + ) return token, None else: debug_logger.log_info(f"[reCAPTCHA] 未知的打码方式: {captcha_method}") self._set_request_fingerprint(None) return None, None - async def _get_api_captcha_token(self, method: str, project_id: str, action: str = "IMAGE_GENERATION") -> Optional[str]: + async def _get_api_captcha_token( + self, + method: str, + project_id: str, + action: str = "IMAGE_GENERATION" + ) -> Optional[tuple[str, str]]: """通用API打码服务 - + Args: method: 打码服务类型 project_id: 项目ID action: reCAPTCHA action类型 (IMAGE_GENERATION 或 VIDEO_GENERATION) + + Returns: + (gRecaptchaResponse, userAgent) 元组, 或 None (失败时)。 + 返回 userAgent 是为了让后续 Flow API 请求沿用打码时的 UA, + 避免 reCAPTCHA V3 评估因 UA 不一致判定 UNUSUAL_ACTIVITY。 """ # 获取配置 if method == "yescaptcha": @@ -3110,7 +3136,10 @@ async def _get_api_captcha_token(self, method: str, project_id: str, action: str response = solution.get('gRecaptchaResponse') if response: debug_logger.log_info(f"[reCAPTCHA {method}] Token获取成功") - return response + # 同时返回 userAgent, 调用方会注入到 fingerprint context, + # 保证 Flow API 提交请求时使用与打码一致的 UA。 + user_agent = solution.get('userAgent') + return response, user_agent await asyncio.sleep(3) diff --git a/tests/test_api_captcha_fingerprint.py b/tests/test_api_captcha_fingerprint.py new file mode 100644 index 00000000..400f310a --- /dev/null +++ b/tests/test_api_captcha_fingerprint.py @@ -0,0 +1,90 @@ +"""Verify API-mode captcha injects solution user_agent into the request fingerprint. + +背景: YesCaptcha / CapMonster / EzCaptcha / CapSolver 返回的 solution 包含 +gRecaptchaResponse 与 userAgent。Google reCAPTCHA V3 评估会校验 token 与 +提交请求的 User-Agent 一致性, 因此调用 Flow API 时必须沿用打码服务返回的 +UA, 否则服务端判定 UNUSUAL_ACTIVITY 并返回 reCAPTCHA evaluation failed。 +""" + +import unittest +from unittest.mock import patch, AsyncMock, MagicMock + +from src.services.flow_client import FlowClient + + +class _FakeProxyManager: + async def get_request_proxy_url(self): + return None + + +class _FakeAsyncSession: + """模拟 curl_cffi 的 AsyncSession: createTask 返回 taskId, getTaskResult 返回 ready。""" + + def __init__(self): + self._calls = 0 + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return False + + async def post(self, *args, **kwargs): + self._calls += 1 + response = MagicMock() + if self._calls == 1: + # createTask + response.status_code = 200 + response.json.return_value = { + "errorId": 0, + "taskId": "tid-xyz", + } + else: + # getTaskResult + response.status_code = 200 + response.json.return_value = { + "errorId": 0, + "status": "ready", + "solution": { + "gRecaptchaResponse": "token-abc", + "userAgent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/147.0.0.0 Safari/537.36" + ), + }, + } + return response + + +class ApiCaptchaFingerprintTests(unittest.IsolatedAsyncioTestCase): + async def test_api_captcha_returns_token_and_user_agent(self): + """_get_api_captcha_token 必须返回 (token, userAgent) 元组。""" + flow = FlowClient.__new__(FlowClient) + flow.proxy_manager = _FakeProxyManager() + fake_session = _FakeAsyncSession() + + with patch("src.services.flow_client.AsyncSession", lambda *a, **kw: fake_session), \ + patch("src.services.flow_client.config") as cfg, \ + patch("asyncio.sleep", new=AsyncMock()): + cfg.yescaptcha_api_key = "key" + cfg.yescaptcha_base_url = "https://api.yescaptcha.com" + cfg.yescaptcha_task_type = "RecaptchaV3TaskProxylessM1" + cfg.debug_enabled = False + + result = await flow._get_api_captcha_token( + method="yescaptcha", + project_id="proj-1", + action="IMAGE_GENERATION", + ) + + self.assertIsNotNone(result, "函数不应返回 None, 因为我们 mock 了 ready 状态") + self.assertIsInstance(result, tuple, "_get_api_captcha_token 应返回 (token, userAgent) 元组") + token, user_agent = result + self.assertEqual(token, "token-abc") + self.assertIn("Windows", user_agent, "userAgent 应当来自打码服务 solution, 包含 Windows") + self.assertIn("Chrome/147", user_agent) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file