type: 收窄虎牙验证与设备注册类型

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)
-

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(