live-hub-py/core/huya/device_fingerprint.py

"""虎牙设备指纹：通过 node 运行 hydevice.js 生成高信任 sdid。

流程与真机 App WebView 一致：
    df/token -> hydevice 采集+WASM 加密 -> df/collect -> sdid
"""

from __future__ import annotations

import shutil
import subprocess
import tempfile
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path

import requests
from loguru import logger

FINGERPRINT_DIR = Path(__file__).parent / "fingerprint"
RUNNER_JS = FINGERPRINT_DIR / "runner.js"
SDID_PREFIX = "__SDID__"

DEFAULT_TIMEOUT = (8, 40)


class HuyaFingerprintError(RuntimeError):
    """虎牙设备指纹生成失败。"""


@dataclass
class HuyaSdidResult:
    """sdid 获取结果。"""

    sdid: str = ""
    hdid: str = ""  # df/collect 同响应下发的 40hex 设备ID; 与 WUP 登录的 32hex hdid 非同一体系(见 docs/HUYA_APP_OVERVIEW.md §二)
    source: str = ""
    message: str = ""


HDID_PREFIX = "__HDID__"


def _run_node_runner(state_dir: Path, app_id: str, timeout: tuple[float, float]) -> tuple[str, str]:
    """调用 node runner，返回 (sdid, hdid)。"""
    node_bin = shutil.which("node")
    if not node_bin:
        raise HuyaFingerprintError("未找到 node 可执行文件（hydevice.js 需要 JS 引擎）")

    connect_timeout, read_timeout = timeout
    total_timeout = int(connect_timeout + read_timeout + 10)
    try:
        proc = subprocess.run(
            [node_bin, str(RUNNER_JS), str(state_dir), app_id],
            capture_output=True,
            text=True,
            timeout=total_timeout,
            cwd=str(FINGERPRINT_DIR),
        )
    except subprocess.TimeoutExpired as exc:
        raise HuyaFingerprintError(f"hydevice runner 超时({total_timeout}s)") from exc

    sdid, hdid = "", ""
    for line in (proc.stdout or "").splitlines():
        line = line.strip()
        if line.startswith(SDID_PREFIX) and len(line) > len(SDID_PREFIX) + 20:
            sdid = line[len(SDID_PREFIX):]
        if line.startswith(HDID_PREFIX) and len(line) > len(HDID_PREFIX) + 20:
            hdid = line[len(HDID_PREFIX):]
    if sdid:
        return sdid, hdid
    stderr_tail = (proc.stderr or "").strip().splitlines()
    detail = stderr_tail[-1][:120] if stderr_tail else f"exit={proc.returncode}"
    raise HuyaFingerprintError(f"hydevice runner 未返回 sdid: {detail}")


def _fallback_sdid(session: requests.Session, timeout) -> str:
    """旧版降级：仅 token+collect，无指纹（低信任）。"""
    token_url = "https://df.huya.com/web/df/token"
    collect_url = "https://df.huya.com/web/df/collect"
    token_res = session.post(
        token_url,
        json={"encryptVersion": "1.0.1", "fingerprintVersion": "1.2.45"},
        timeout=timeout,
    )
    token_res.raise_for_status()
    token = token_res.json().get("data", {}).get("token", "")
    if not token:
        raise HuyaFingerprintError(f"获取虎牙 df token 失败: {token_res.text[:120]}")
    collect_res = session.post(collect_url, json={"token": token}, timeout=timeout)
    collect_res.raise_for_status()
    return collect_res.json().get("data", {}).get("sdid", "")


def get_huya_sdid(
    app_id: str = "5008",
    state_dir: str | Path | None = None,
    session: requests.Session | None = None,
    proxies: Mapping | None = None,
    timeout: tuple[float, float] = DEFAULT_TIMEOUT,
    allow_fallback: bool = True,
) -> HuyaSdidResult:
    """获取虎牙高信任 sdid；node 链路失败时可选降级为低信任指纹。

    Args:
        app_id: 虎牙应用 id，App 场景 5008。
        state_dir: localStorage 持久化目录（保持设备一致性），None 用临时目录。
        allow_fallback: node 失败时是否降级为旧版无指纹方式。
    Returns:
        HuyaSdidResult(sdid, source, message)，source 为 fingerprint/fallback/none。
    """
    del session, proxies  # node runner 自行发请求；保留参数便于未来代理透传
    if state_dir is None:
        state_dir = Path(tempfile.gettempdir()) / "huya_fp_state"
    state_dir = Path(state_dir)
    try:
        sdid, hdid = _run_node_runner(state_dir, app_id, timeout)
        if sdid:
            logger.debug("虎牙设备指纹成功(node): sdid={}... hdid={}...",
                         sdid[:24], hdid[:10])
            return HuyaSdidResult(sdid=sdid, hdid=hdid, source="fingerprint")
    except HuyaFingerprintError as exc:
        logger.warning("虎牙 hydevice 指纹失败: {}", exc)
        if not allow_fallback:
            raise

    if allow_fallback:
        try:
            with requests.Session() as s:
                s.trust_env = False
                sdid = _fallback_sdid(s, timeout)
                if sdid:
                    logger.debug("虎牙设备指纹降级(无指纹): sdid={}...", sdid[:24])
                    return HuyaSdidResult(
                        sdid=sdid,
                        source="fallback",
                        message="低信任指纹(降级)",
                    )
        except Exception as exc:  # noqa: BLE001
            logger.warning("虎牙降级指纹也失败: {}", exc)

    return HuyaSdidResult(source="none", message="未能获取 sdid")