Files
live-hub-py/core/huya/verification/solver.py
T
yml2213 06a0a1e7d3 虎牙登录: 接入hydevice高信任设备指纹与App场景验证支持
- 新增 fingerprint/: node仿真Android WebView运行官方hydevice.js(含WASM加密), 产出与真机同构的高信任sdid
- 新增 device_fingerprint.py: py编排node runner获取sdid, 失败自动降级旧无指纹流程
- login.py: prepare_device优先高信任指纹; 增加App场景常量(appId=5008, appSign=md5(appId+verCode+appKey)[:8], 逆向自APK UdbNetHelper)
- solver: 支持appId/page_url/use_touch_events参数化(App模式); get3使用csid会话id; 识别qr_auth扫码(冷却)与dx_auth短信策略并抛专用异常
2026-08-23 17:32:34 +08:00

526 lines
21 KiB
Python

"""虎牙登录风控验证闭环。"""
from __future__ import annotations
import base64
import io
import json
import random
import time
import uuid
from collections.abc import Mapping
from http.cookies import SimpleCookie
from urllib.parse import parse_qs, quote, urlparse
import cv2
import execjs
import numpy as np
import requests
from PIL import Image
from loguru import logger
from .ocr import HuyaCaptchaOcr, default_ocr
from .track import format_track, generate_slide_track
DEFAULT_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
)
class HuyaVerificationError(RuntimeError):
"""虎牙风控验证失败。"""
class HuyaNoSessionError(HuyaVerificationError):
"""config3 未下发图形验证会话(策略不匹配或无需滑块)。"""
class HuyaQrAuthRequiredError(HuyaVerificationError):
"""风控策略为扫码验证(qr_auth),需已登录设备确认,无法自动化。"""
class HuyaSmsAuthRequiredError(HuyaVerificationError):
"""风控策略为短信验证(dx_auth),需接码/收码后提交。"""
def __init__(self, challenge_payload: dict | str, message: str = ""):
self.challenge_payload = challenge_payload
super().__init__(message or "虎牙风控要求短信验证")
def _cookie_mapping(cookie: Mapping[str, str] | str | None) -> dict[str, str]:
"""兼容 dict 和 Cookie 字符串。"""
if not cookie:
return {}
if isinstance(cookie, str):
parsed = SimpleCookie()
parsed.load(cookie)
return {key: morsel.value for key, morsel in parsed.items()}
return {str(key): str(value) for key, value in cookie.items()}
class HuyaVerificationSolver:
"""处理虎牙 returnCode=10030/10039 后的滑块或点选验证。"""
def __init__(
self,
cookie: Mapping[str, str] | str | None = None,
ua: str = DEFAULT_UA,
sdid: str = "",
session: requests.Session | None = None,
ocr: HuyaCaptchaOcr | None = None,
proxies: Mapping[str, str] | None = None,
timeout: tuple[float, float] = (8, 20),
app_id: str = "5002",
page_url: str | None = None,
use_touch_events: bool = False,
):
self.cookie = _cookie_mapping(cookie)
self.ua = ua or DEFAULT_UA
self.sdid = sdid or self.cookie.get("sdid", "")
self.session = session or requests.Session()
self.session.trust_env = False
if self.cookie:
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.timeout = timeout
self.ocr = ocr or default_ocr()
self.huyapk = ""
self.scene_type = ""
self.code = ""
self.url_param_map: dict[str, str] = {}
self.strategy_url = ""
self.js_ctx = None
self.behavior_events: list[dict] = []
self.app_id = app_id
self.page_url = page_url or self._auth_page_url()
self.use_touch_events = use_touch_events
@staticmethod
def _auth_page_url() -> str:
"""生成 auth 页面地址,形式参考易语言实现。"""
return f"https://aq.huya.com/p/udb_login.html?v={str(int(time.time() * 1000))[:8]}00000"
@staticmethod
def _decode_image(base64_string: str | None) -> bytes:
if not base64_string:
raise HuyaVerificationError("验证码图片为空")
if ";base64," in base64_string:
base64_string = base64_string.split(";base64,")[-1]
return base64.b64decode(base64_string)
@staticmethod
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)
max_width = max(bg.shape[1], tip.shape[1])
def pad_right(img, target_width):
height, width = img.shape[:2]
if width >= target_width:
return img
return cv2.copyMakeBorder(
img,
0,
0,
0,
target_width - width,
cv2.BORDER_CONSTANT,
value=(255, 255, 255),
)
concat_img = np.vstack([pad_right(tip, max_width), pad_right(bg, max_width)])
success, buffer = cv2.imencode(f".{image_format.lower()}", concat_img)
if not success:
raise HuyaVerificationError("点选验证码图片编码失败")
return buffer.tobytes()
@staticmethod
def _remove_black_borders(image: Image.Image) -> tuple[Image.Image, int]:
if image.mode != "RGBA":
image = image.convert("RGBA")
bbox = image.getbbox()
if not bbox:
return image, 0
cropped_height = bbox[3] - bbox[1]
black_border_height = image.height - cropped_height
return image.crop(bbox), black_border_height
@classmethod
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)
adjusted_y = y_position + int(black_border_height / 2) + 1
bg.paste(tip, (10, adjusted_y), tip)
output = io.BytesIO()
bg.save(output, format="PNG")
return output.getvalue()
@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.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)
_, _, _, max_loc = cv2.minMaxLoc(result)
return int(max_loc[0])
def _request_json(self, method: str, url: str, **kwargs) -> dict:
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
response.raise_for_status()
try:
return response.json()
except json.JSONDecodeError as exc:
raise HuyaVerificationError(f"虎牙验证接口返回不是 JSON: {response.text[:200]}") from exc
def _compile_js(self, dx_js: str) -> None:
js_data = """
window = globalThis;
$$self_js$$
function get_regoin_arr(segment, huyapk) {
return JSON.parse(atob(window.UdbCipher.decrypt(segment, huyapk))).ord
}
function get_act(act_data, huyapk) {
act_data = encodeURI(JSON.stringify(act_data))
return window.UdbCipher.encrypt(btoa(act_data), huyapk)
}
"""
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:
payload = {
"urlParamMap": self.url_param_map,
"wupData": "",
"page": self.page_url,
"behavior": self._encode_behavior(),
"info": info_override if info_override is not None else self.sdid,
}
payload.update(extra)
return {"appId": self.app_id, "data": payload}
def _csid(self) -> str:
"""App WebView 场景 get3 使用 csid_ 前缀会话 id。"""
return f"csid_{uuid.uuid4().hex}"
def _pointer_action(self, action: str) -> str:
"""web 用 mouse 事件,App WebView 场景用 touch 事件。"""
if not self.use_touch_events:
return 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="~()*!.'")
@staticmethod
def _extract_strategy_url(challenge_payload: dict | str) -> str:
"""兼容完整响应、data 子对象、data URL 字符串三种输入。"""
if isinstance(challenge_payload, str):
return challenge_payload
data = challenge_payload.get("data")
if isinstance(data, str) and data.startswith("http"):
return data
if isinstance(data, str):
try:
data = json.loads(data)
except json.JSONDecodeError:
data = {}
strategy_root = data if isinstance(data, dict) else challenge_payload
try:
return strategy_root["strategys"][0]["data"]
except (KeyError, IndexError, TypeError) as exc:
raise HuyaVerificationError("虎牙响应中没有 strategys[0].data") from exc
def _parse_strategy(self, challenge_payload: dict | str) -> None:
strategy_url = self._extract_strategy_url(challenge_payload)
query_params = parse_qs(urlparse(strategy_url).query)
self.url_param_map = {key: value[0] for key, value in query_params.items()}
self.scene_type = self.url_param_map.get("sceneType", "")
self.strategy_url = strategy_url
if not self.url_param_map.get("param"):
raise HuyaVerificationError("虎牙验证链接缺少 param")
logger.debug(
"虎牙验证参数: sceneType={}, mobile={}, param长度={}",
self.scene_type,
self.url_param_map.get("mobile", ""),
len(self.url_param_map.get("param", "")),
)
strategies = []
if isinstance(challenge_payload, dict):
strategies = (challenge_payload.get("data") or {}).get("strategys") or []
if strategies:
try:
logger.warning(
"虎牙登录风控strategys完整结构: {}",
json.dumps(strategies, ensure_ascii=False)[:1200],
)
except Exception: # noqa: BLE001
pass
strategy_url_lower = strategy_url.lower()
is_qr_auth = "qr_auth" in strategy_url_lower or any(
isinstance(s, dict) and str(s.get("strategy")) == "64" for s in strategies
)
if is_qr_auth and "dx_auth" not in strategy_url_lower:
raise HuyaQrAuthRequiredError(
"虎牙风控要求扫码验证(qr_auth/strategy=64),需已登录App设备确认,"
f"账号: {self.url_param_map.get('mobile', '?')}"
)
if "dx_auth" in strategy_url_lower:
raise HuyaSmsAuthRequiredError(
challenge_payload,
f"虎牙风控要求短信验证(dx_auth),手机号: {self.url_param_map.get('mobile', '?')}",
)
self.behavior_events = []
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
captcha = Image.new("RGB", (width, height))
for index, region_index in enumerate(recover_list):
region = image.crop((region_index * 26, 0, region_index * 26 + 26, 180))
captcha.paste(region, (index * 26, 0, index * 26 + 26, 180))
output = io.BytesIO()
captcha.save(output, format="JPEG")
return output.getvalue()
def _get_track_point(self, url: str, payload: dict):
logger.debug("虎牙验证获取图片: {}", url)
result = self._request_json("post", url, json=payload)
if result.get("returnCode") != 0 or not result.get("data", {}).get("code"):
raise HuyaVerificationError(f"获取验证码图片失败: {result}")
data = result["data"]
self.code = data["code"]
self.huyapk = data["huyapk"]
segment = data.get("segment")
if segment:
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))
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)
return yolo_x
logger.info("虎牙验证类型: 点选")
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]
def _send_slider_verify(self, url: str, point: int) -> dict:
track_points = generate_slide_track(point)
act_data = {
"point": int(point),
"travel": format_track(track_points),
"code": self.code,
"sceneType": self.scene_type,
"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),
})
payload = self._build_common_payload({"act": act, "ver": 5})
return self._submit_verify(url, payload)
def _send_click_verify(self, url: str, points: list[list[int]]) -> dict:
if not points:
raise HuyaVerificationError("点选坐标识别为空")
start_time = int(time.time() * 1000) - random.randint(5000, 10000)
click_data = []
current_time = start_time
for point in points:
current_time += random.randint(1000, 2000)
click_data.append([int(point[0]), int(point[1]), current_time])
interval_time_list = [item[-1] for item in click_data]
act_data = {
"clickData": ",".join(f"{x}_{y}" for x, y in points),
"intervalTime": "_".join(map(str, interval_time_list)),
"endTime": interval_time_list[-1] - interval_time_list[0],
"sceneType": self.scene_type,
"code": self.code,
}
act = self.js_ctx.call("get_act", act_data, self.huyapk)
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],
})
self.behavior_events.extend(behavior_list)
payload = self._build_common_payload({"act": act, "ver": 5})
return self._submit_verify(url, payload)
def _submit_verify(self, url: str, payload: dict) -> dict:
logger.debug("虎牙验证提交: {}", url)
body = json.dumps(payload, separators=(",", ":"))
result = self._request_json("post", url, params={"lock": "true"}, data=body)
data = result.get("data") or result
if isinstance(data, dict):
logger.debug(
"虎牙验证提交结果: authId={}, nextAuthUrl={}",
bool(data.get("authId")),
bool(data.get("nextAuthUrl")),
)
return data
def _solve_current_strategy(self) -> dict:
"""提交当前 strategy_url 对应的验证。"""
config_payload = self._build_common_payload({"cTuIndex": 0})
config = {}
gurl = ""
vurl = ""
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 {}
gurl = str(data.get("gurl") or "")
vurl = str(data.get("vurl") or "")
dx_js = str(data.get("js") or "")
logger.debug(
"虎牙config3返回: returnCode={}, desc={}, gurl={}, vurl={}, js长度={}",
config.get("returnCode"),
config.get("description") or config.get("message") or "",
bool(gurl),
bool(vurl),
len(dx_js),
)
if not (gurl and vurl):
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":
self.url_param_map["sceneType"] = "4"
self.scene_type = "4"
config_payload = self._build_common_payload({"cTuIndex": 0})
logger.debug("虎牙config3取参不完整,改用旧流程sceneType=4重试")
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"}
raise HuyaNoSessionError(
"config3未下发验证会话: "
f"returnCode={config.get('returnCode')}, "
f"description={config.get('description') or config.get('message') or ''}, "
f"gurl={bool(gurl)}, vurl={bool(vurl)}, js={bool(dx_js)}, "
f"data={json.dumps(data_summary, ensure_ascii=False)[:600]}"
)
self._compile_js(dx_js)
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)
logger.info("虎牙验证失败后重试: {}/3", index + 1)
point = self._get_track_point(gurl, point_payload)
logger.debug(f"虎牙验证识别结果: {point}")
if isinstance(point, list):
last_result = self._send_click_verify(vurl, point)
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")):
return last_result
logger.info(
"虎牙验证提交未通过: returnCode={}, message={} ({}/3)",
last_result.get("returnCode") if isinstance(last_result, dict) else "",
last_result.get("message") if isinstance(last_result, dict) else "",
index + 1,
)
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)
result = self._solve_current_strategy()
return result
def solve_huya_verification(
challenge_payload: dict | str,
cookie: Mapping[str, str] | str | None = None,
ua: str = DEFAULT_UA,
sdid: str = "",
session: requests.Session | None = None,
proxies: Mapping[str, str] | None = None,
app_id: str = "5002",
page_url: str | None = None,
use_touch_events: bool = False,
) -> dict:
"""函数式入口,便于登录流程直接调用。"""
solver = HuyaVerificationSolver(
cookie=cookie,
ua=ua,
sdid=sdid,
session=session,
proxies=proxies,
app_id=app_id,
page_url=page_url,
use_touch_events=use_touch_events,
)
return solver.solve(challenge_payload)