From 4f9a6864ebfbdaa068d7595ca62042d1ebb52b48 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 6 Sep 2026 13:02:37 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=AA=8C=E8=AF=81=E7=A0=81?= =?UTF-8?q?=E5=B1=95=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 5 + core/huya/app_login.py | 44 +++- core/huya/verification/capture.py | 377 +++++++++++++++++++++++++++++ core/huya/verification/solver.py | 165 +++++++++++-- scripts/capture_huya_captcha.py | 189 +++++++++++++++ tests/test_huya_captcha_capture.py | 209 ++++++++++++++++ web/backend/routers/huya.py | 13 +- 7 files changed, 972 insertions(+), 30 deletions(-) create mode 100644 core/huya/verification/capture.py create mode 100644 scripts/capture_huya_captcha.py create mode 100644 tests/test_huya_captcha_capture.py diff --git a/.gitignore b/.gitignore index 253c04d..5270e02 100644 --- a/.gitignore +++ b/.gitignore @@ -28,9 +28,13 @@ evidence/ scripts/* !scripts/migrate_sqlite_to_mysql.py !scripts/install-git-hooks.sh +!scripts/capture_huya_captcha.py # ── 逆向工具集（frida hook、unidbg hydev、虎牙协议复刻脚本等） tools/ +# 验证码抓取数据（本地分析，不入库） +data/captcha_captures/ + # Node 依赖 **/node_modules/ web/frontend/dist/ @@ -41,3 +45,4 @@ apks/ work/ analysis/ +artifacts/ diff --git a/core/huya/app_login.py b/core/huya/app_login.py index b58a640..75b9789 100644 --- a/core/huya/app_login.py +++ b/core/huya/app_login.py @@ -90,6 +90,14 @@ class HuyaAppQrAuthRequiredError(HuyaAppLoginError): """风控要求扫码验证 (qr_auth)。""" +class HuyaAppSmsAuthRequiredError(HuyaAppLoginError): + """风控要求短信验证 (sms_auth / sms_auth2)。""" + + def __init__(self, message: str = "", mobile: str = ""): + self.mobile = mobile + super().__init__(message or f"虎牙风控要求短信验证，手机号: {mobile or '?'}") + + def _golden_session_assets() -> tuple[dict, str, str]: """生成本次登录的会话元数据和行为轨迹。""" from .wup_encoder import make_trace_id, make_user_action @@ -202,7 +210,7 @@ def parse_risk_url(resp: bytes) -> str | None: urls.append(u) if not urls: return None - pt = [u for u in urls if "pt_auth" in u] + pt = [u for u in urls if "pt_auth" in u or "dx_auth" in u] return (pt or urls)[0] @@ -221,10 +229,12 @@ def solve_safe_auth( proxies=None, max_retry: int = 3, device_info: Mapping[str, object] | None = None, + account: str = "", ) -> dict: - """解 safe_auth 风控滑块。""" + """解 safe_auth 风控滑块/点选。""" from .verification.solver import ( HuyaQrAuthRequiredError, + HuyaSmsAuthRequiredError, HuyaVerificationSolver, ) @@ -243,11 +253,15 @@ def solve_safe_auth( app_id=app_id, page_url=risk_url, use_touch_events=True, + account=account, ) try: result = solver.solve(risk_url) except HuyaQrAuthRequiredError: raise + except HuyaSmsAuthRequiredError as exc: + mobile = getattr(exc, "mobile", "") + raise HuyaAppSmsAuthRequiredError(str(exc), mobile=mobile) from exc except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 last_err = exc logger.warning(f"[safe_auth] 第 {attempt + 1} 次过验异常: {exc}") @@ -255,9 +269,9 @@ def solve_safe_auth( continue auth_id = str((result or {}).get("authId") or "") if auth_id: - logger.info(f"[safe_auth] 滑块通过 authId={auth_id[:24]}...") + logger.info(f"[safe_auth] 验证码过验成功 authId={auth_id[:24]}...") return result - raise HuyaAppLoginError(f"safe_auth 滑块过验失败: {last_err or '未返回 authId'}") + raise HuyaAppLoginError(f"safe_auth 验证码过验失败: {last_err or '未返回 authId'}") def parse_real_uid(resp: bytes) -> int: @@ -326,15 +340,19 @@ def login_cred_with_flow( kind = ( "pt_auth(滑块)" if "pt_auth" in risk_url - else ("qr_auth(扫码)" if "qr_auth" in risk_url else "未知") + else ( + "dx_auth(点选)" + if "dx_auth" in risk_url + else ("qr_auth(扫码)" if "qr_auth" in risk_url else "未知") + ) ) logger.info(f"[huya-app] 第 {rnd + 1} 轮触发安全验证: {kind}") if "qr_auth" in risk_url: raise HuyaAppQrAuthRequiredError( "该账号 App 渠道要求扫码验证(qr_auth)，请先在手机虎牙 App 上正常登录一次建立设备信任。" ) - solve_safe_auth(risk_url, proxies=proxies, device_info=dev) - logger.info("[huya-app] safe_auth 滑块过验成功，重发 WUP 登录...") + solve_safe_auth(risk_url, proxies=proxies, device_info=dev, account=account) + logger.info(f"[huya-app] safe_auth ({kind}) 过验成功，重发 WUP 登录...") continue markers = _response_markers(resp) detail = f"，服务端字段: {markers}" if markers else "" @@ -633,12 +651,24 @@ class HuyaAppPasswordLogin: include_device_token=True, ) except HuyaAppQrAuthRequiredError as exc: + logger.warning(f"[huya-app] 账号 {acct} 登录受阻: {exc}") return HuyaLoginResult( success=False, message=str(exc), code="QR_AUTH_REQUIRED", ) + except HuyaAppSmsAuthRequiredError as exc: + logger.warning( + f"[huya-app] 账号 {acct} 登录中断：滑块与点选已全部通过，" + f"虎牙下发了手机短信二次验证 ({exc.mobile or '手机号见验证链接'})，需人工接码" + ) + return HuyaLoginResult( + success=False, + message=str(exc), + code="SMS_AUTH_REQUIRED", + ) except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 + logger.warning(f"[huya-app] 账号 {acct} App 登录异常: {exc}") return HuyaLoginResult( success=False, message=f"App 登录凭证获取失败: {exc}", diff --git a/core/huya/verification/capture.py b/core/huya/verification/capture.py new file mode 100644 index 0000000..9f9dcd3 --- /dev/null +++ b/core/huya/verification/capture.py @@ -0,0 +1,377 @@ +"""虎牙验证码全要素数据捕获与落盘模块。 + +每次触发验证码（滑块 pt_auth / 点选 dx_auth）时，将图片、提示语、识别结果、 +请求元数据及提交响应完整落盘保存到 data/captcha_captures/，便于分析与排查。 +""" + +from __future__ import annotations + +import datetime as dt +import json +from pathlib import Path +from typing import Any + +import cv2 +import numpy as np +from loguru import logger + +PROJECT_ROOT = Path(__file__).resolve().parents[3] +DEFAULT_CAPTURE_DIR = PROJECT_ROOT / "data" / "captcha_captures" + + +class CaptchaCaptureRecord: + """单次验证码捕获会话。""" + + def __init__( + self, + auth_type: str, + account: str = "", + page_url: str = "", + scene_type: str = "", + output_base_dir: Path | None = None, + ): + self.auth_type = auth_type + self.account = account + self.page_url = page_url + self.scene_type = scene_type + self.output_base_dir = Path(output_base_dir or DEFAULT_CAPTURE_DIR) + self.created_at = dt.datetime.now(dt.timezone(dt.timedelta(hours=8))) + + self.session_id = self.created_at.strftime("%Y%m%d_%H%M%S") + self.target_dir: Path | None = None + + self.meta: dict[str, Any] = { + "timestamp": self.created_at.isoformat(), + "account": account, + "auth_type": auth_type, + "auth_name": ( + "点选验证 (dx_auth)" + if "dx" in auth_type + else ( + "短信二次验证 (sms_auth)" + if "sms" in auth_type + else ( + "扫码验证 (qr_auth)" + if "qr" in auth_type + else "滑块验证 (pt_auth)" + ) + ) + ), + "page_url": page_url, + "scene_type": scene_type, + "prompts": {}, + "images": {}, + "solution": {}, + "submit_result": {}, + } + + def _ensure_target_dir(self, code: str = "") -> Path: + if self.target_dir is None: + short_code = code[:8] if code else "sample" + dir_name = f"{self.session_id}_{self.auth_type}_{short_code}" + self.target_dir = self.output_base_dir / dir_name + self.target_dir.mkdir(parents=True, exist_ok=True) + return self.target_dir + + def capture_slider_data( + self, + *, + code: str, + dtu_bytes: bytes, + puzzle_bytes: bytes, + restored_bg_bytes: bytes | None = None, + composite_bytes: bytes | None = None, + point_x: int | None = None, + raw_data: dict[str, Any] | None = None, + ) -> Path: + """捕获滑块验证码的所有图片和元数据。""" + folder = self._ensure_target_dir(code) + + # 1. 提示语与服务端描述 + raw = raw_data or {} + self.meta["code"] = code + self.meta["prompts"] = { + "title": "安全验证 (滑块拼图)", + "default_instruction": "拖动下方滑块完成拼图", + "location_y": raw.get("location_y"), + "description": raw.get("description") or raw.get("message") or "", + "tip": raw.get("tip") or "", + } + + # 2. 保存原始与合成图片 + images_info = {} + if dtu_bytes: + (folder / "bg_scrambled.jpg").write_bytes(dtu_bytes) + images_info["bg_scrambled"] = "bg_scrambled.jpg" + + if puzzle_bytes: + (folder / "puzzle.png").write_bytes(puzzle_bytes) + images_info["puzzle"] = "puzzle.png" + + if restored_bg_bytes: + (folder / "bg_restored.jpg").write_bytes(restored_bg_bytes) + images_info["bg_restored"] = "bg_restored.jpg" + + if composite_bytes: + (folder / "composite.png").write_bytes(composite_bytes) + images_info["composite"] = "composite.png" + + # 3. 绘制识别标注图（在还原后的背景图上标注缺口 X 坐标与 Y 坐标） + if restored_bg_bytes and point_x is not None: + try: + bg_arr = cv2.imdecode( + np.frombuffer(restored_bg_bytes, np.uint8), cv2.IMREAD_COLOR + ) + if bg_arr is not None: + loc_y = int(raw.get("location_y") or 0) + cv2.rectangle( + bg_arr, + (point_x, loc_y), + (point_x + 50, loc_y + 50), + (0, 0, 255), + 2, + ) + cv2.putText( + bg_arr, + f"X={point_x}", + (max(0, point_x - 10), max(20, loc_y - 10)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.6, + (0, 0, 255), + 2, + ) + cv2.imwrite(str(folder / "annotated.jpg"), bg_arr) + images_info["annotated"] = "annotated.jpg" + except Exception as exc: # noqa: BLE001 + logger.debug(f"[captcha_capture] 标注滑块图异常: {exc}") + + self.meta["images"] = images_info + if point_x is not None: + self.meta["solution"] = {"type": "slider", "distance_x": point_x} + + if raw_data: + clean_raw = { + k: v for k, v in raw_data.items() if not str(v).startswith("data:image") + } + (folder / "raw_data.json").write_text( + json.dumps(clean_raw, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + self._flush_meta() + self._log_summary() + return folder + + def capture_click_data( + self, + *, + code: str, + dtu_bytes: bytes, + hzi_bytes: bytes, + points: list[list[int]] | None = None, + raw_data: dict[str, Any] | None = None, + ) -> Path: + """捕获文字/图标点选验证码的所有图片和元数据。""" + folder = self._ensure_target_dir(code) + + raw = raw_data or {} + self.meta["code"] = code + question_text = str(raw.get("question") or "") + if not question_text: + question_text = "按顺序点击图中对应内容" + word_size = raw.get("wordSize") + self.meta["prompts"] = { + "title": str(raw.get("title") or "安全验证 (点选验证 dx_auth)"), + "question": question_text, + "word_size": word_size, + "tu_ming": raw.get("tuMing"), + "description": raw.get("description") or raw.get("message") or "", + } + + images_info = {} + if dtu_bytes: + (folder / "bg_main.jpg").write_bytes(dtu_bytes) + images_info["bg_main"] = "bg_main.jpg" + + if hzi_bytes: + (folder / "tip_text.png").write_bytes(hzi_bytes) + images_info["tip_text"] = "tip_text.png" + + # 绘制点选标注预览图 + if dtu_bytes and points: + try: + bg_arr = cv2.imdecode( + np.frombuffer(dtu_bytes, np.uint8), cv2.IMREAD_COLOR + ) + if bg_arr is not None: + for idx, (px, py) in enumerate(points): + # 绘制点击圆圈 + cv2.circle(bg_arr, (int(px), int(py)), 14, (0, 0, 255), 2) + cv2.circle(bg_arr, (int(px), int(py)), 4, (0, 255, 0), -1) + # 标注点击序号 1, 2, 3... + cv2.putText( + bg_arr, + str(idx + 1), + (int(px) - 5, int(py) - 18), + cv2.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 0, 255), + 2, + ) + cv2.imwrite(str(folder / "annotated.jpg"), bg_arr) + images_info["annotated"] = "annotated.jpg" + except Exception as exc: # noqa: BLE001 + logger.debug(f"[captcha_capture] 标注点选图异常: {exc}") + + self.meta["images"] = images_info + if points is not None: + self.meta["solution"] = {"type": "click", "points": points} + + if raw_data: + clean_raw = { + k: v for k, v in raw_data.items() if not str(v).startswith("data:image") + } + (folder / "raw_data.json").write_text( + json.dumps(clean_raw, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + self._flush_meta() + self._log_summary() + return folder + + def capture_sms_auth_data( + self, + *, + mobile: str = "", + page_url: str = "", + scene_type: str = "", + step: int = 1, + raw_params: dict[str, Any] | None = None, + ) -> Path: + """捕获短信风控验证拦截数据并落盘。""" + folder = self._ensure_target_dir(code="sms_auth") + self.meta["auth_type"] = "sms_auth" + self.meta["auth_name"] = "短信二次验证 (sms_auth)" + self.meta["page_url"] = page_url + self.meta["scene_type"] = scene_type + self.meta["mobile"] = mobile + self.meta["step"] = step + self.meta["prompts"] = { + "title": "安全验证 (短信二次验证 sms_auth)", + "instruction": f"前置图形验证码已通过，虎牙触发第 {step} 级短信安全验证，需向手机 {mobile} 发送验证码", + "mobile": mobile, + } + self.meta["submit_result"] = { + "status": "sms_auth_required", + "mobile": mobile, + "nextAuthUrl": page_url, + } + if raw_params: + (folder / "params.json").write_text( + json.dumps(raw_params, ensure_ascii=False, indent=2), encoding="utf-8" + ) + self._flush_meta() + + logger.warning( + f"\n┌── [虎牙风控拦截捕获] ──────────────────────────────────────────\n" + f"│ 类型: 短信二次验证 (sms_auth) (sceneType: {self.scene_type or scene_type or '4'})\n" + f"│ 账号: {self.account or '未指定'}\n" + f"│ 阶段: 第 {step} 阶段（前置滑块/点选已成功通过）\n" + f"│ 手机: {mobile or '未知'}\n" + f"│ 提示: 虎牙官方下发短信接码页面，协议自动化无法跳过，需人工接码\n" + f"│ 页面: {page_url[:80]}...\n" + f"│ 存储目录: {self.target_dir}\n" + f"└─────────────────────────────────────────────────────────────" + ) + return folder + + def capture_qr_auth_data( + self, + *, + mobile: str = "", + page_url: str = "", + scene_type: str = "", + step: int = 1, + raw_params: dict[str, Any] | None = None, + ) -> Path: + """捕获扫码风控验证拦截数据并落盘。""" + folder = self._ensure_target_dir(code="qr_auth") + self.meta["auth_type"] = "qr_auth" + self.meta["auth_name"] = "扫码二次验证 (qr_auth)" + self.meta["page_url"] = page_url + self.meta["scene_type"] = scene_type + self.meta["mobile"] = mobile + self.meta["step"] = step + self.meta["prompts"] = { + "title": "安全验证 (扫码二次验证 qr_auth)", + "instruction": f"虎牙触发第 {step} 级扫码安全验证，需在手机 App 上确认", + "mobile": mobile, + } + self.meta["submit_result"] = { + "status": "qr_auth_required", + "nextAuthUrl": page_url, + } + if raw_params: + (folder / "params.json").write_text( + json.dumps(raw_params, ensure_ascii=False, indent=2), encoding="utf-8" + ) + self._flush_meta() + + logger.warning( + f"\n┌── [虎牙风控拦截捕获] ──────────────────────────────────────────\n" + f"│ 类型: 扫码二次验证 (qr_auth) (sceneType: {self.scene_type or scene_type or '4'})\n" + f"│ 账号: {self.account or '未指定'}\n" + f"│ 阶段: 第 {step} 阶段\n" + f"│ 提示: 虎牙要求在手机 App 上扫码确认或使用常用设备登录\n" + f"│ 页面: {page_url[:80]}...\n" + f"│ 存储目录: {self.target_dir}\n" + f"└─────────────────────────────────────────────────────────────" + ) + return folder + + def record_submit_result(self, result: dict[str, Any]) -> None: + """记录验证码提交验证结果（authId、是否通过、是否流转下一阶段等）。""" + has_auth = bool(result.get("authId")) + has_next = bool(result.get("nextAuthUrl")) + status = ( + "auth_success" + if has_auth + else ("passed_to_next_auth" if has_next else "failed") + ) + self.meta["submit_result"] = { + "status": status, + "success": has_auth or has_next, + "authId": result.get("authId"), + "nextAuthUrl": result.get("nextAuthUrl"), + "returnCode": result.get("returnCode"), + "message": result.get("message") or result.get("description") or "", + } + self._flush_meta() + + def _flush_meta(self) -> None: + if self.target_dir: + (self.target_dir / "meta.json").write_text( + json.dumps(self.meta, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + def _log_summary(self) -> None: + prompts = self.meta.get("prompts") or {} + title = prompts.get("title") or "安全验证" + question = ( + prompts.get("question") + or prompts.get("default_instruction") + or "按顺序点击" + ) + word_size = prompts.get("word_size") + word_info = f", 目标字数: {word_size}" if word_size else "" + img_list = ", ".join(self.meta.get("images", {}).values()) + + logger.info( + f"\n┌── [虎牙验证码捕获] ──────────────────────────────────────────\n" + f"│ 类型: {self.meta.get('auth_name')} (sceneType: {self.scene_type or '默认'})\n" + f"│ 账号: {self.account or '未指定'}\n" + f"│ 标题: {title}\n" + f"│ 提示语: {question}{word_info}\n" + f"│ 存储目录: {self.target_dir}\n" + f"│ 保存图片: {img_list}\n" + f"└─────────────────────────────────────────────────────────────" + ) diff --git a/core/huya/verification/solver.py b/core/huya/verification/solver.py index c8a75f5..e281892 100644 --- a/core/huya/verification/solver.py +++ b/core/huya/verification/solver.py @@ -10,6 +10,7 @@ import time import uuid from collections.abc import Mapping from http.cookies import SimpleCookie +from pathlib import Path from typing import Any, cast from urllib.parse import parse_qs, quote, urlparse @@ -20,6 +21,7 @@ import requests from loguru import logger from PIL import Image +from .capture import CaptchaCaptureRecord from .ocr import HuyaCaptchaOcr, default_ocr from .track import format_track, generate_slide_track @@ -43,11 +45,14 @@ class HuyaQrAuthRequiredError(HuyaVerificationError): class HuyaSmsAuthRequiredError(HuyaVerificationError): - """风控策略为短信验证（dx_auth），需接码/收码后提交。""" + """风控策略为短信验证（sms_auth/sms_auth2），需接码/收码后提交。""" - def __init__(self, challenge_payload: dict | str, message: str = ""): + def __init__( + self, challenge_payload: dict | str, message: str = "", mobile: str = "" + ): self.challenge_payload = challenge_payload - super().__init__(message or "虎牙风控要求短信验证") + self.mobile = mobile + super().__init__(message or f"虎牙风控要求短信验证，手机号: {mobile or '?'}") def _cookie_mapping(cookie: Mapping[str, str] | str | None) -> dict[str, str]: @@ -76,6 +81,9 @@ class HuyaVerificationSolver: app_id: str = "5002", page_url: str | None = None, use_touch_events: bool = False, + account: str = "", + capture_dir: str | Path | None = None, + enable_capture: bool = True, ): self.cookie = _cookie_mapping(cookie) self.ua = ua or DEFAULT_UA @@ -102,6 +110,12 @@ class HuyaVerificationSolver: self.timeout = timeout self.ocr = ocr or default_ocr() + self.account = account + self.capture_dir = capture_dir + self.enable_capture = enable_capture + self.current_capture: CaptchaCaptureRecord | None = None + self.auth_type = "" + self.huyapk = "" self.scene_type = "" self.code = "" @@ -275,7 +289,7 @@ class HuyaVerificationSolver: except (KeyError, IndexError, TypeError) as exc: raise HuyaVerificationError("虎牙响应中没有 strategys[0].data") from exc - def _parse_strategy(self, challenge_payload: dict | str) -> None: + def _parse_strategy(self, challenge_payload: dict | str, step: int = 1) -> None: strategy_url = self._extract_strategy_url(challenge_payload) query_params = parse_qs(urlparse(strategy_url).query) @@ -306,15 +320,71 @@ class HuyaVerificationSolver: # 实测(2026-08-25): strategy=64 时 pt_auth.html 是滑块、qr_auth.html 才是扫码， # 旧的 `strategy==64 → 扫码` 硬编码会把可自动化的滑块误杀成死路。 is_qr_auth = "qr_auth" in strategy_url_lower - if is_qr_auth and "dx_auth" not in strategy_url_lower: + account_id = self.account or self.url_param_map.get("mobile", "") + if is_qr_auth: + if self.enable_capture: + cap = CaptchaCaptureRecord( + auth_type="qr_auth", + account=account_id, + page_url=self.strategy_url, + scene_type=self.scene_type, + output_base_dir=Path(self.capture_dir) + if self.capture_dir + else None, + ) + cap.capture_qr_auth_data( + mobile=self.url_param_map.get("mobile", "?"), + page_url=self.strategy_url, + scene_type=self.scene_type, + step=step, + raw_params=self.url_param_map, + ) + logger.warning( + "[solver] 虎牙风控在第 {} 步要求扫码验证 (qr_auth)，需 App 确认，账号: {}", + step, + self.url_param_map.get("mobile", "?"), + ) raise HuyaQrAuthRequiredError( "虎牙风控要求扫码验证(qr_auth)，需已登录App设备确认，" f"账号: {self.url_param_map.get('mobile', '?')}" ) - if "dx_auth" in strategy_url_lower: + if "sms_auth" in strategy_url_lower: + mobile = self.url_param_map.get("mobile", "?") + if self.enable_capture: + cap = CaptchaCaptureRecord( + auth_type="sms_auth", + account=account_id, + page_url=self.strategy_url, + scene_type=self.scene_type, + output_base_dir=Path(self.capture_dir) + if self.capture_dir + else None, + ) + cap.capture_sms_auth_data( + mobile=mobile, + page_url=self.strategy_url, + scene_type=self.scene_type, + step=step, + raw_params=self.url_param_map, + ) + logger.warning( + "[solver] 虎牙风控在第 {} 步要求短信验证 (sms_auth)，手机号: {}", + step, + mobile, + ) raise HuyaSmsAuthRequiredError( challenge_payload, - f"虎牙风控要求短信验证(dx_auth)，手机号: {self.url_param_map.get('mobile', '?')}", + f"虎牙风控要求短信验证(sms_auth)，手机号: {mobile}", + mobile=mobile, + ) + self.auth_type = "dx_auth" if "dx_auth" in strategy_url_lower else "pt_auth" + if self.enable_capture: + self.current_capture = CaptchaCaptureRecord( + auth_type=self.auth_type, + account=account_id, + page_url=self.strategy_url, + scene_type=self.scene_type, + output_base_dir=Path(self.capture_dir) if self.capture_dir else None, ) self.behavior_events = [] @@ -357,18 +427,49 @@ class HuyaVerificationSolver: opencv_x = self._opencv_slider_x(tip_bytes, restored_bg) if abs(opencv_x - yolo_x) < 10: logger.debug("虎牙滑块识别距离: yolo={}, opencv={}", yolo_x, opencv_x) - return int((opencv_x + yolo_x) / 2) - logger.debug( - "虎牙滑块识别距离: yolo={}, opencv={}, 使用yolo", yolo_x, opencv_x - ) - return yolo_x + final_x = int((opencv_x + yolo_x) / 2) + else: + logger.debug( + "虎牙滑块识别距离: yolo={}, opencv={}, 使用yolo", yolo_x, opencv_x + ) + final_x = yolo_x - logger.info("虎牙验证类型: 点选") + if self.current_capture: + try: + self.current_capture.capture_slider_data( + code=self.code, + dtu_bytes=bg_bytes, + puzzle_bytes=tip_bytes, + restored_bg_bytes=restored_bg, + composite_bytes=composite, + point_x=final_x, + raw_data=data, + ) + except Exception as exc: # noqa: BLE001 + logger.debug(f"[solver] 记录滑块捕获数据异常: {exc}") + + return final_x + + logger.info("虎牙验证类型: 点选 (dx_auth)") bg_bytes = self._decode_image(data.get("dTuStr")) tip_bytes = self._decode_image(data.get("hZiStr")) merged = self._vertical_concat_align_left(bg_bytes, tip_bytes) points = self.ocr.ocr_click_points(merged) - return [[int(x), int(y) - 44] for x, y in points] + final_points = [[int(x), int(y) - 44] for x, y in points] + + if self.current_capture: + try: + self.current_capture.capture_click_data( + code=self.code, + dtu_bytes=bg_bytes, + hzi_bytes=tip_bytes, + points=final_points, + raw_data=data, + ) + except Exception as exc: # noqa: BLE001 + logger.debug(f"[solver] 记录点选捕获数据异常: {exc}") + + return final_points def _send_slider_verify(self, url: str, point: int) -> dict: track_points = generate_slide_track(point) @@ -451,6 +552,11 @@ class HuyaVerificationSolver: bool(data.get("authId")), bool(data.get("nextAuthUrl")), ) + if self.current_capture: + try: + self.current_capture.record_submit_result(data) + except Exception as exc: # noqa: BLE001 + logger.debug(f"[solver] 记录提交结果异常: {exc}") return data def _solve_current_strategy(self) -> dict: @@ -542,15 +648,24 @@ class HuyaVerificationSolver: return last_result def solve(self, challenge_payload: dict | str) -> dict: - """提交虎牙风控验证，返回 vurl 响应中的 data。""" - self._parse_strategy(challenge_payload) - result = self._solve_current_strategy() - - next_auth_url = result.get("nextAuthUrl") if isinstance(result, dict) else None - if next_auth_url: - logger.info("虎牙滑块后继续点选验证") - self._parse_strategy(next_auth_url) + """提交虎牙风控验证，支持多级风控流转 (如 pt_auth -> dx_auth -> ...)。""" + current_challenge: dict | str = challenge_payload + result: dict = {} + for step in range(5): + self._parse_strategy(current_challenge, step=step + 1) result = self._solve_current_strategy() + if not isinstance(result, dict): + break + if result.get("authId"): + logger.info(f"虎牙验证在第 {step + 1} 步成功获取 authId") + break + next_auth_url = result.get("nextAuthUrl") + if not next_auth_url: + break + logger.info( + f"虎牙验证第 {step + 1} 步完成，进入下一阶段验证: {next_auth_url[:80]}..." + ) + current_challenge = next_auth_url return result @@ -564,6 +679,9 @@ def solve_huya_verification( app_id: str = "5002", page_url: str | None = None, use_touch_events: bool = False, + account: str = "", + capture_dir: str | Path | None = None, + enable_capture: bool = True, ) -> dict: """函数式入口，便于登录流程直接调用。""" solver = HuyaVerificationSolver( @@ -575,5 +693,8 @@ def solve_huya_verification( app_id=app_id, page_url=page_url, use_touch_events=use_touch_events, + account=account, + capture_dir=capture_dir, + enable_capture=enable_capture, ) return solver.solve(challenge_payload) diff --git a/scripts/capture_huya_captcha.py b/scripts/capture_huya_captcha.py new file mode 100644 index 0000000..6e4694e --- /dev/null +++ b/scripts/capture_huya_captcha.py @@ -0,0 +1,189 @@ +"""虎牙验证码数据采集与分析脚本。 + +功能： +1. 【URL 采集模式】直接传入 safe_auth URL（dx_auth 点选 / pt_auth 滑块）， + 拉取验证码图片、提示语与元数据并落盘保存； +2. 【登录探测模式】输入账号密码触发风控，捕获验证码全套图片与提示语。 + +用法示例： + # 模式一：直接解析已有风控 URL + python scripts/capture_huya_captcha.py url "https://aq.huya.com/p/safe_auth/dx_auth.html?param=..." + + # 模式二：仅拉取图片和提示语，不向服务端提交答案 + python scripts/capture_huya_captcha.py url "https://aq.huya.com/p/safe_auth/dx_auth.html?param=..." --dump-only + + # 模式三：登录触发捕获 + python scripts/capture_huya_captcha.py login <账号> <密码> --channel app +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +from loguru import logger + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from core.huya.verification.capture import DEFAULT_CAPTURE_DIR +from core.huya.verification.solver import ( + HuyaQrAuthRequiredError, + HuyaSmsAuthRequiredError, + HuyaVerificationSolver, +) + + +def capture_from_url( + url: str, + *, + dump_only: bool = False, + account: str = "", + proxies: dict | None = None, +) -> None: + """从 safe_auth URL 抓取并保存验证码数据。""" + logger.info(f"开始从 URL 采集验证码: {url[:100]}...") + q = parse_qs(urlparse(url).query) + app_id = q.get("appId", ["5002"])[0] + + solver = HuyaVerificationSolver( + app_id=app_id, + page_url=url, + account=account, + proxies=proxies, + enable_capture=True, + ) + solver._parse_strategy(url) + + if dump_only: + # 只拉取 config3 和 gurl，不提交 vurl + config_payload = solver._build_common_payload({"cTuIndex": 0}) + config = solver._request_json( + "post", + "https://udbrtt.huya.com/auth/client/config3", + json=config_payload, + ) + data = config.get("data") or {} + gurl = str(data.get("gurl") or "") + dx_js = str(data.get("js") or "") + if not gurl: + logger.error(f"config3 未下发 gurl: {config}") + return + if dx_js: + solver._compile_js(dx_js) + point_payload = solver._build_common_payload({"ver": 5}) + point = solver._get_track_point(gurl, point_payload) + logger.info(f"【仅采集模式】验证码解析完成，识别结果: {point}") + if solver.current_capture and solver.current_capture.target_dir: + logger.info(f"验证码数据已成功保存至: {solver.current_capture.target_dir}") + return + + # 完整求解并记录提交结果 + try: + result = solver.solve(url) + logger.info(f"验证码提交完成: authId={result.get('authId')}") + except HuyaQrAuthRequiredError as exc: + logger.warning(f"该 URL 属于扫码验证: {exc}") + except HuyaSmsAuthRequiredError as exc: + logger.warning(f"该 URL 触发短信二次验证: {exc}") + except Exception as exc: # noqa: BLE001 + logger.error(f"验证码处理异常: {exc}") + + +def capture_from_login( + account: str, + password: str, + channel: str = "app", + proxies: dict | None = None, +) -> None: + """通过尝试登录触发风控并自动落盘验证码。""" + logger.info(f"尝试登录账号 {account} (渠道: {channel}) 探测验证码...") + if channel == "app": + from core.huya.app_login import login_huya_app_password + + res = login_huya_app_password(account, password, proxies=proxies) + logger.info( + f"登录结果: success={res.success}, code={res.code}, msg={res.message}" + ) + else: + from core.huya.login import login_huya_password + + res = login_huya_password(account, password, proxies=proxies) + logger.info( + f"登录结果: success={res.success}, code={res.code}, msg={res.message}" + ) + + +def list_recent_captures() -> None: + """列出最近采集到的验证码。""" + if not DEFAULT_CAPTURE_DIR.exists(): + print("暂无捕获数据目录: ", DEFAULT_CAPTURE_DIR) + return + folders = sorted( + DEFAULT_CAPTURE_DIR.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True + ) + if not folders: + print("当前尚未捕获任何验证码数据。") + return + print(f"\n找到 {len(folders)} 条捕获记录 (位于 {DEFAULT_CAPTURE_DIR}):\n") + for f in folders[:10]: + meta_file = f / "meta.json" + if meta_file.exists(): + try: + meta = json.loads(meta_file.read_text(encoding="utf-8")) + auth_name = meta.get("auth_name", "未知") + prompts = meta.get("prompts", {}) + question = ( + prompts.get("question") or prompts.get("default_instruction") or "" + ) + files = [p.name for p in f.glob("*") if p.name != "meta.json"] + print(f"📁 {f.name}") + print(f" 类型: {auth_name} | 提示语: {question}") + print(f" 文件: {', '.join(files)}") + except Exception: # noqa: BLE001 + print(f"📁 {f.name}") + else: + print(f"📁 {f.name}") + + +def main() -> int: + parser = argparse.ArgumentParser(description="虎牙验证码采集与测试工具") + subparsers = parser.add_subparsers(dest="cmd") + + url_parser = subparsers.add_parser("url", help="从 safe_auth URL 抓取并落盘验证码") + url_parser.add_argument("url", help="safe_auth 完整 URL") + url_parser.add_argument("--account", default="", help="关联账号名 (可选)") + url_parser.add_argument( + "--dump-only", + action="store_true", + help="仅下载图片与提示语，不向服务器提交答案", + ) + + login_parser = subparsers.add_parser("login", help="通过密码登录触发验证码") + login_parser.add_argument("account", help="虎牙账号或手机号") + login_parser.add_argument("password", help="密码") + login_parser.add_argument( + "--channel", choices=["app", "web"], default="app", help="登录渠道 (默认 app)" + ) + + subparsers.add_parser("list", help="查看最近采集到的验证码记录") + + args = parser.parse_args() + if args.cmd == "url": + capture_from_url(args.url, dump_only=args.dump_only, account=args.account) + elif args.cmd == "login": + capture_from_login(args.account, args.password, channel=args.channel) + elif args.cmd == "list": + list_recent_captures() + else: + parser.print_help() + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_huya_captcha_capture.py b/tests/test_huya_captcha_capture.py new file mode 100644 index 0000000..56ea6ea --- /dev/null +++ b/tests/test_huya_captcha_capture.py @@ -0,0 +1,209 @@ +"""虎牙 dx_auth 与验证码捕获落盘功能测试。""" + +from __future__ import annotations + +import io +import json +from pathlib import Path + +from PIL import Image + +from core.huya.verification.capture import CaptchaCaptureRecord +from core.huya.verification.solver import ( + HuyaQrAuthRequiredError, + HuyaVerificationSolver, +) + + +def test_dx_auth_strategy_recognized_as_click_not_sms(tmp_path: Path): + """验证 dx_auth 不再被误杀为短信异常，而是识别为 dx_auth。""" + url = ( + "https://aq.huya.com/p/safe_auth/dx_auth.html?" + "param=test_param&mobile=13800000000&sceneType=16&appId=5002" + ) + solver = HuyaVerificationSolver( + page_url=url, + account="13800000000", + capture_dir=tmp_path, + enable_capture=True, + ) + + # 不抛出 HuyaSmsAuthRequiredError + solver._parse_strategy(url) + assert solver.auth_type == "dx_auth" + assert solver.current_capture is not None + assert solver.current_capture.meta["auth_type"] == "dx_auth" + assert solver.current_capture.meta["account"] == "13800000000" + + +def test_qr_auth_still_raises_qr_error(): + """验证真正的 qr_auth 仍然正确抛出 HuyaQrAuthRequiredError。""" + url = "https://aq.huya.com/p/safe_auth/qr_auth.html?param=test_param&mobile=13800000000" + solver = HuyaVerificationSolver(page_url=url) + try: + solver._parse_strategy(url) + except HuyaQrAuthRequiredError: + pass + else: + raise AssertionError("应该抛出 HuyaQrAuthRequiredError") + + +def test_click_captcha_capture_saves_files_and_prompts(tmp_path: Path): + """验证点选验证码捕获时正确落盘图片、提示语与元数据。""" + record = CaptchaCaptureRecord( + auth_type="dx_auth", + account="test_user", + page_url="https://aq.huya.com/p/safe_auth/dx_auth.html?param=123", + scene_type="16", + output_base_dir=tmp_path, + ) + + img = Image.new("RGB", (100, 100), color="blue") + buf = io.BytesIO() + img.save(buf, format="JPEG") + fake_img_bytes = buf.getvalue() + + tip_img = Image.new("RGBA", (80, 40), color="white") + tip_buf = io.BytesIO() + tip_img.save(tip_buf, format="PNG") + fake_tip_bytes = tip_buf.getvalue() + + folder = record.capture_click_data( + code="code123456", + dtu_bytes=fake_img_bytes, + hzi_bytes=fake_tip_bytes, + points=[[30, 40], [60, 70]], + raw_data={ + "question": "按顺序点击：虎、牙", + "wordSize": 2, + "tuMing": 1, + "description": "安全验证", + }, + ) + + assert folder.exists() + assert (folder / "bg_main.jpg").exists() + assert (folder / "tip_text.png").exists() + assert (folder / "annotated.jpg").exists() + assert (folder / "meta.json").exists() + + meta = json.loads((folder / "meta.json").read_text(encoding="utf-8")) + assert meta["auth_type"] == "dx_auth" + assert meta["prompts"]["question"] == "按顺序点击：虎、牙" + assert meta["prompts"]["word_size"] == 2 + assert meta["solution"]["points"] == [[30, 40], [60, 70]] + + +def test_slider_captcha_capture_saves_files_and_prompts(tmp_path: Path): + """验证滑块验证码捕获时正确落盘背景、拼图与标注图。""" + record = CaptchaCaptureRecord( + auth_type="pt_auth", + account="test_slider_user", + page_url="https://aq.huya.com/p/safe_auth/pt_auth.html?param=123", + scene_type="4", + output_base_dir=tmp_path, + ) + + img = Image.new("RGB", (260, 180), color="green") + buf = io.BytesIO() + img.save(buf, format="JPEG") + fake_bg_bytes = buf.getvalue() + + puzzle = Image.new("RGBA", (50, 50), color="red") + p_buf = io.BytesIO() + puzzle.save(p_buf, format="PNG") + fake_puzzle_bytes = p_buf.getvalue() + + folder = record.capture_slider_data( + code="slider_code_888", + dtu_bytes=fake_bg_bytes, + puzzle_bytes=fake_puzzle_bytes, + restored_bg_bytes=fake_bg_bytes, + point_x=120, + raw_data={"location_y": 60, "tip": "拖动滑块完成拼图"}, + ) + + assert folder.exists() + assert (folder / "bg_scrambled.jpg").exists() + assert (folder / "puzzle.png").exists() + assert (folder / "bg_restored.jpg").exists() + assert (folder / "annotated.jpg").exists() + + meta = json.loads((folder / "meta.json").read_text(encoding="utf-8")) + assert meta["auth_type"] == "pt_auth" + assert meta["solution"]["distance_x"] == 120 + assert meta["prompts"]["location_y"] == 60 + + +def test_submit_result_recorded_in_meta(tmp_path: Path): + """验证验证码提交结果正确更新进 meta.json。""" + record = CaptchaCaptureRecord( + auth_type="dx_auth", + account="test_user", + output_base_dir=tmp_path, + ) + + img = Image.new("RGB", (10, 10)) + buf = io.BytesIO() + img.save(buf, format="JPEG") + folder = record.capture_click_data( + code="c1", dtu_bytes=buf.getvalue(), hzi_bytes=b"" + ) + + record.record_submit_result({"authId": "auth_token_xyz999", "returnCode": 0}) + meta = json.loads((folder / "meta.json").read_text(encoding="utf-8")) + assert meta["submit_result"]["success"] is True + assert meta["submit_result"]["authId"] == "auth_token_xyz999" + + +def test_sms_auth_capture_saves_meta_and_params(tmp_path: Path): + """验证遇到短信风控拦截时正确落盘元数据与参数。""" + record = CaptchaCaptureRecord( + auth_type="sms_auth", + account="hy_test_sms", + page_url="https://aq.huya.com/m/safe_auth/sms_auth2.html?param=xyz&mobile=0018888", + scene_type="4", + output_base_dir=tmp_path, + ) + folder = record.capture_sms_auth_data( + mobile="0018888", + page_url="https://aq.huya.com/m/safe_auth/sms_auth2.html?param=xyz&mobile=0018888", + scene_type="4", + step=3, + raw_params={"sceneType": "4", "mobile": "0018888", "param": "xyz"}, + ) + assert folder.exists() + assert (folder / "meta.json").exists() + assert (folder / "params.json").exists() + + meta = json.loads((folder / "meta.json").read_text(encoding="utf-8")) + assert meta["auth_type"] == "sms_auth" + assert meta["step"] == 3 + assert meta["mobile"] == "0018888" + assert meta["submit_result"]["status"] == "sms_auth_required" + + +def test_qr_auth_capture_saves_meta_and_params(tmp_path: Path): + """验证遇到扫码风控拦截时正确落盘元数据与参数。""" + record = CaptchaCaptureRecord( + auth_type="qr_auth", + account="hy_test_qr", + page_url="https://aq.huya.com/m/safe_auth/qr_auth.html?param=abc", + scene_type="4", + output_base_dir=tmp_path, + ) + folder = record.capture_qr_auth_data( + mobile="13800138000", + page_url="https://aq.huya.com/m/safe_auth/qr_auth.html?param=abc", + scene_type="4", + step=2, + raw_params={"sceneType": "4", "mobile": "13800138000"}, + ) + assert folder.exists() + assert (folder / "meta.json").exists() + assert (folder / "params.json").exists() + + meta = json.loads((folder / "meta.json").read_text(encoding="utf-8")) + assert meta["auth_type"] == "qr_auth" + assert meta["step"] == 2 + assert meta["submit_result"]["status"] == "qr_auth_required" diff --git a/web/backend/routers/huya.py b/web/backend/routers/huya.py index 01c94c0..3b41835 100644 --- a/web/backend/routers/huya.py +++ b/web/backend/routers/huya.py @@ -8,6 +8,8 @@ import threading from datetime import UTC, datetime from typing import cast +from loguru import logger + from fastapi import ( APIRouter, Depends, @@ -772,8 +774,14 @@ def app_password_login_account( ) from exc if not result.success or not result.cookie: + logger.warning( + f"[huya-app] 账号 {req.username} 密码登录未完成: {result.message} (code={result.code})" + ) raise HTTPException( - status_code=502, detail=result.message or "虎牙 App 密码登录失败" + status_code=400 + if result.code in ("SMS_AUTH_REQUIRED", "QR_AUTH_REQUIRED") + else 502, + detail=result.message or "虎牙 App 密码登录失败", ) try: @@ -1269,6 +1277,9 @@ def app_password_login_selected_accounts( account.updated_at = datetime.now(UTC) db.commit() failed_count += 1 + logger.warning( + f"[huya-app] 账号 {username} 批量登录未完成: {result.message} (code={result.code})" + ) results.append( { "line": account.id,