From 3f97dbe5256997d5ee2d555da0d6429e3b1b1fdd Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 30 Aug 2026 20:07:12 +0800 Subject: [PATCH] =?UTF-8?q?type:=20=E6=94=B6=E7=AA=84=E8=99=8E=E7=89=99?= =?UTF-8?q?=E9=AA=8C=E8=AF=81=E4=B8=8E=E8=AE=BE=E5=A4=87=E6=B3=A8=E5=86=8C?= =?UTF-8?q?=E7=B1=BB=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/huya/dfp_register.py | 60 +++++++---- core/huya/verification/solver.py | 164 ++++++++++++++++++++----------- 2 files changed, 152 insertions(+), 72 deletions(-) diff --git a/core/huya/dfp_register.py b/core/huya/dfp_register.py index 33afeb5..15a2405 100644 --- a/core/huya/dfp_register.py +++ b/core/huya/dfp_register.py @@ -23,6 +23,7 @@ - ``_random_triple`` 的 64hex 占位只是 cw JSON 模板形态 — 真实 dfpReport 指纹 JSON (R36 解密 12 样本) 无 hdid 字段 (顶层 Athena/deviceinfo/terminal/version)。 """ + from __future__ import annotations import base64 @@ -74,23 +75,33 @@ def _load_chain() -> dict[str, tuple[bytes, bytes]]: try: data = json.loads(CHAIN_FILE.read_text(encoding="utf-8")) return { - name: (base64.b64decode(item["req_b64"]), base64.b64decode(item["resp_b64"])) + name: ( + base64.b64decode(item["req_b64"]), + base64.b64decode(item["resp_b64"]), + ) for name, item in data.items() } except (OSError, ValueError, KeyError) as exc: raise DfpRegistrationError(f"注册链模板读取失败: {exc}") from exc -def _post(body: bytes, content_type: str = "application/octet-stream", - timeout: float = 20, proxies: Mapping[str, str] | None = None) -> bytes: +def _post( + body: bytes, + content_type: str = "application/octet-stream", + timeout: float = 20, + proxies: Mapping[str, str] | None = None, +) -> bytes: try: response = requests.post( WSAPI, data=body, - headers={"Content-Type": content_type, "User-Agent": UA, - "Accept-Encoding": "gzip"}, + headers={ + "Content-Type": content_type, + "User-Agent": UA, + "Accept-Encoding": "gzip", + }, timeout=timeout, - proxies=proxies, + proxies=dict(proxies) if proxies else None, ) response.raise_for_status() return response.content @@ -130,11 +141,16 @@ def _build_random_dfp_body() -> bytes: json_sec = os.urandom(CW_JSON_LEN) if len(seed) <= CW_JSON_LEN: mask = os.urandom(len(seed)) - json_sec = bytes(a ^ b for a, b in zip(seed, mask)) + json_sec[len(seed):] + json_sec = bytes(a ^ b for a, b in zip(seed, mask)) + json_sec[len(seed) :] cw = os.urandom(2) + json_sec + os.urandom(CW_COLL_LEN) + CW_TAIL if len(cw) != CW_LEN: raise DfpRegistrationError(f"dfpReport cw 长度异常: {len(cw)}") - body = struct.pack(">I", len(TAF_HEAD) + len(MAGIC) + len(cw) + 4) + TAF_HEAD + MAGIC + cw + body = ( + struct.pack(">I", len(TAF_HEAD) + len(MAGIC) + len(cw) + 4) + + TAF_HEAD + + MAGIC + + cw + ) if len(body) != 4226: raise DfpRegistrationError(f"dfpReport body 长度异常: {len(body)}") return body @@ -145,23 +161,34 @@ def _select_operator_request(template: bytes, fingerprint: str | None) -> bytes: old = b"02df398797432eadefcc12767119ad5e80999389" index = template.find(old) if index >= 0: - return template[:index] + fingerprint.encode("ascii") + template[index + len(old):] + return ( + template[:index] + + fingerprint.encode("ascii") + + template[index + len(old) :] + ) return template def _parse_response(data: bytes) -> tuple[str, str, str]: try: - t1 = re.search(rb"\x16\x20([0-9a-f]{32})", data).group(1).decode() - t2 = re.search(rb"\x26\xb4([A-Za-z0-9+/=]{180})", data).group(1).decode("latin1") - t5 = re.search(rb"\x56\x28([0-9a-f]{40})", data).group(1).decode() - except AttributeError as exc: + t1_match = re.search(rb"\x16\x20([0-9a-f]{32})", data) + t2_match = re.search(rb"\x26\xb4([A-Za-z0-9+/=]{180})", data) + t5_match = re.search(rb"\x56\x28([0-9a-f]{40})", data) + if not t1_match or not t2_match or not t5_match: + raise ValueError("missing response fields") + t1 = t1_match.group(1).decode() + t2 = t2_match.group(1).decode("latin1") + t5 = t5_match.group(1).decode() + except (AttributeError, ValueError) as exc: raise DfpRegistrationError("dfpReport 响应缺少 t1/t2/t5") from exc return t1, t2, t5 -def register_device(fingerprint: str | None = None, - proxies: Mapping[str, str] | None = None, - timeout: float = 20) -> tuple[str, str, str]: +def register_device( + fingerprint: str | None = None, + proxies: Mapping[str, str] | None = None, + timeout: float = 20, +) -> tuple[str, str, str]: """执行新注册链，返回 ``(t1, safedeviceid, device_id)``。""" chain = _load_chain() _post(chain["getDfpConfig"][0], timeout=timeout, proxies=proxies) @@ -169,4 +196,3 @@ def register_device(fingerprint: str | None = None, _post(select_request, "application/x-wup", timeout, proxies) response = _post(_build_random_dfp_body(), timeout=timeout, proxies=proxies) return _parse_response(response) - diff --git a/core/huya/verification/solver.py b/core/huya/verification/solver.py index dc6c6b9..5131a95 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 typing import Any, cast from urllib.parse import parse_qs, quote, urlparse import cv2 @@ -86,17 +87,19 @@ class HuyaVerificationSolver: self.session.cookies.update(self.cookie) if proxies: self.session.proxies = dict(proxies) - self.session.headers.update({ - "Accept-Language": "zh-CN,zh;q=0.9", - "Cache-Control": "no-cache", - "Connection": "keep-alive", - "Origin": "https://aq.huya.com", - "Pragma": "no-cache", - "Referer": "https://aq.huya.com/", - "User-Agent": self.ua, - "accept": "application/json, text/plain, */*", - "content-type": "application/json", - }) + self.session.headers.update( + { + "Accept-Language": "zh-CN,zh;q=0.9", + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "Origin": "https://aq.huya.com", + "Pragma": "no-cache", + "Referer": "https://aq.huya.com/", + "User-Agent": self.ua, + "accept": "application/json, text/plain, */*", + "content-type": "application/json", + } + ) self.timeout = timeout self.ocr = ocr or default_ocr() @@ -105,7 +108,7 @@ class HuyaVerificationSolver: self.code = "" self.url_param_map: dict[str, str] = {} self.strategy_url = "" - self.js_ctx = None + self.js_ctx: Any = None self.behavior_events: list[dict] = [] self.app_id = app_id self.page_url = page_url or self._auth_page_url() @@ -125,9 +128,13 @@ class HuyaVerificationSolver: return base64.b64decode(base64_string) @staticmethod - def _vertical_concat_align_left(bg_bytes: bytes, tip_bytes: bytes, image_format: str = "JPEG") -> bytes: + def _vertical_concat_align_left( + bg_bytes: bytes, tip_bytes: bytes, image_format: str = "JPEG" + ) -> bytes: bg = cv2.imdecode(np.frombuffer(bg_bytes, np.uint8), cv2.IMREAD_COLOR) tip = cv2.imdecode(np.frombuffer(tip_bytes, np.uint8), cv2.IMREAD_COLOR) + if bg is None or tip is None: + raise HuyaVerificationError("验证码图片解码失败") max_width = max(bg.shape[1], tip.shape[1]) def pad_right(img, target_width): @@ -162,7 +169,9 @@ class HuyaVerificationSolver: return image.crop(bbox), black_border_height @classmethod - def _composite_slider_images(cls, bg_bytes: bytes, tip_bytes: bytes, y_position: int) -> bytes: + def _composite_slider_images( + cls, bg_bytes: bytes, tip_bytes: bytes, y_position: int + ) -> bytes: bg = Image.open(io.BytesIO(bg_bytes)).convert("RGB") tip = Image.open(io.BytesIO(tip_bytes)).convert("RGBA") tip, black_border_height = cls._remove_black_borders(tip) @@ -174,8 +183,14 @@ class HuyaVerificationSolver: @staticmethod def _opencv_slider_x(target_bytes: bytes, background_bytes: bytes) -> int: - target = cv2.imdecode(np.frombuffer(target_bytes, np.uint8), cv2.IMREAD_ANYCOLOR) - background = cv2.imdecode(np.frombuffer(background_bytes, np.uint8), cv2.IMREAD_ANYCOLOR) + target = cv2.imdecode( + np.frombuffer(target_bytes, np.uint8), cv2.IMREAD_ANYCOLOR + ) + background = cv2.imdecode( + np.frombuffer(background_bytes, np.uint8), cv2.IMREAD_ANYCOLOR + ) + if target is None or background is None: + raise HuyaVerificationError("滑块图片解码失败") target = cv2.cvtColor(cv2.Canny(target, 100, 200), cv2.COLOR_GRAY2RGB) background = cv2.cvtColor(cv2.Canny(background, 100, 200), cv2.COLOR_GRAY2RGB) result = cv2.matchTemplate(background, target, cv2.TM_CCOEFF_NORMED) @@ -188,7 +203,9 @@ class HuyaVerificationSolver: try: return response.json() except json.JSONDecodeError as exc: - raise HuyaVerificationError(f"虎牙验证接口返回不是 JSON: {response.text[:200]}") from exc + raise HuyaVerificationError( + f"虎牙验证接口返回不是 JSON: {response.text[:200]}" + ) from exc def _compile_js(self, dx_js: str) -> None: js_data = """ @@ -205,7 +222,9 @@ class HuyaVerificationSolver: """ self.js_ctx = execjs.compile(js_data.replace("$$self_js$$", dx_js)) - def _build_common_payload(self, extra: dict, info_override: str | None = None) -> dict: + def _build_common_payload( + self, extra: dict, info_override: str | None = None + ) -> dict: payload = { "urlParamMap": self.url_param_map, "wupData": "", @@ -224,11 +243,17 @@ class HuyaVerificationSolver: """web 用 mouse 事件，App WebView 场景用 touch 事件。""" if not self.use_touch_events: return action - return {"mousedown": "touchstart", "mousemove": "touchmove", "mouseup": "touchend"}.get(action, action) + return { + "mousedown": "touchstart", + "mousemove": "touchmove", + "mouseup": "touchend", + }.get(action, action) def _encode_behavior(self) -> str: """编码浏览器行为，失败重试时会携带前一次点击记录。""" - return quote(json.dumps(self.behavior_events, separators=(",", ":")), safe="~()*!.'") + return quote( + json.dumps(self.behavior_events, separators=(",", ":")), safe="~()*!.'" + ) @staticmethod def _extract_strategy_url(challenge_payload: dict | str) -> str: @@ -294,7 +319,9 @@ class HuyaVerificationSolver: ) self.behavior_events = [] - def _restore_slider_background(self, bg_bytes: bytes, segment: str, huyapk: str) -> bytes: + def _restore_slider_background( + self, bg_bytes: bytes, segment: str, huyapk: str + ) -> bytes: recover_list = self.js_ctx.call("get_regoin_arr", segment, huyapk) image = Image.open(io.BytesIO(bg_bytes)).convert("RGB") width, height = image.size @@ -320,15 +347,21 @@ class HuyaVerificationSolver: logger.info("虎牙验证类型: 滑块") bg_bytes = self._decode_image(data.get("dtuStr")) tip_bytes = self._decode_image(data.get("puzzleStr")) - restored_bg = self._restore_slider_background(bg_bytes, segment, self.huyapk) - composite = self._composite_slider_images(restored_bg, tip_bytes, int(data.get("location_y") or 0)) + restored_bg = self._restore_slider_background( + bg_bytes, segment, self.huyapk + ) + composite = self._composite_slider_images( + restored_bg, tip_bytes, int(data.get("location_y") or 0) + ) slider_box = self.ocr.ocr_slider_box(composite) yolo_x = int(slider_box[0] - 13) 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) + logger.debug( + "虎牙滑块识别距离: yolo={}, opencv={}, 使用yolo", yolo_x, opencv_x + ) return yolo_x logger.info("虎牙验证类型: 点选") @@ -348,18 +381,20 @@ class HuyaVerificationSolver: "endTime": int(track_points[-1][-1]) - int(track_points[0][-1]), } act = self.js_ctx.call("get_act", act_data, self.huyapk) - self.behavior_events.append({ - "action": self._pointer_action("mousedown"), - "id": "", - "txt": "", - "clientX": track_points[0][0], - "clientY": track_points[0][1], - "clientWidth": 350, - "clientHeight": 268, - "isTrusted": True, - "timeStamp": random.randint(1800, 5500), - "nowTime": int(time.time() * 1000), - }) + self.behavior_events.append( + { + "action": self._pointer_action("mousedown"), + "id": "", + "txt": "", + "clientX": track_points[0][0], + "clientY": track_points[0][1], + "clientWidth": 350, + "clientHeight": 268, + "isTrusted": True, + "timeStamp": random.randint(1800, 5500), + "nowTime": int(time.time() * 1000), + } + ) payload = self._build_common_payload({"act": act, "ver": 5}) return self._submit_verify(url, payload) @@ -387,18 +422,20 @@ class HuyaVerificationSolver: txt_list = ["", "1", "1\\n2"] behavior_list = [] for index, item in enumerate(click_data): - behavior_list.append({ - "action": self._pointer_action("mousedown"), - "id": "", - "txt": txt_list[index] if index < len(txt_list) else "", - "clientX": item[0], - "clientY": item[1], - "clientWidth": 350, - "clientHeight": 268, - "isTrusted": True, - "timeStamp": random.randint(2000, 5000), - "nowTime": item[2], - }) + behavior_list.append( + { + "action": self._pointer_action("mousedown"), + "id": "", + "txt": txt_list[index] if index < len(txt_list) else "", + "clientX": item[0], + "clientY": item[1], + "clientWidth": 350, + "clientHeight": 268, + "isTrusted": True, + "timeStamp": random.randint(2000, 5000), + "nowTime": item[2], + } + ) self.behavior_events.extend(behavior_list) payload = self._build_common_payload({"act": act, "ver": 5}) @@ -426,8 +463,13 @@ class HuyaVerificationSolver: dx_js = "" for index in range(2): logger.debug("虎牙验证获取config3 ({}/2)", index + 1) - config = self._request_json("post", "https://udbrtt.huya.com/auth/client/config3", json=config_payload) - data = config.get("data") if isinstance(config.get("data"), dict) else {} + config = self._request_json( + "post", + "https://udbrtt.huya.com/auth/client/config3", + json=config_payload, + ) + raw_data = config.get("data") + data = cast(dict[str, Any], raw_data) if isinstance(raw_data, dict) else {} gurl = str(data.get("gurl") or "") vurl = str(data.get("vurl") or "") dx_js = str(data.get("js") or "") @@ -440,7 +482,10 @@ class HuyaVerificationSolver: len(dx_js), ) if not (gurl and vurl): - logger.warning("虎牙config3未下发验证会话, 完整响应: {}", json.dumps(config, ensure_ascii=False)[:1500]) + logger.warning( + "虎牙config3未下发验证会话, 完整响应: {}", + json.dumps(config, ensure_ascii=False)[:1500], + ) if gurl and vurl and dx_js: break if index == 0 and self.url_param_map.get("sceneType") != "4": @@ -451,7 +496,9 @@ class HuyaVerificationSolver: time.sleep(0.3) if not gurl or not vurl or not dx_js: - data_summary = {k: str(v)[:80] for k, v in (data or {}).items() if k != "js"} + data_summary = { + k: str(v)[:80] for k, v in (data or {}).items() if k != "js" + } raise HuyaNoSessionError( "config3未下发验证会话: " f"returnCode={config.get('returnCode')}, " @@ -461,12 +508,17 @@ class HuyaVerificationSolver: ) self._compile_js(dx_js) - point_payload = self._build_common_payload({"ver": 5}, info_override=self._csid() if self.use_touch_events else None) + point_payload = self._build_common_payload( + {"ver": 5}, info_override=self._csid() if self.use_touch_events else None + ) last_result: dict = {} for index in range(3): if index: time.sleep(0.4) - point_payload = self._build_common_payload({"ver": 5}, info_override=self._csid() if self.use_touch_events else None) + point_payload = self._build_common_payload( + {"ver": 5}, + info_override=self._csid() if self.use_touch_events else None, + ) logger.info("虎牙验证失败后重试: {}/3", index + 1) point = self._get_track_point(gurl, point_payload) @@ -476,7 +528,9 @@ class HuyaVerificationSolver: else: last_result = self._send_slider_verify(vurl, int(point)) - if isinstance(last_result, dict) and (last_result.get("authId") or last_result.get("nextAuthUrl")): + if isinstance(last_result, dict) and ( + last_result.get("authId") or last_result.get("nextAuthUrl") + ): return last_result logger.info(