diff --git a/docs/HUYA_HDID_ALGORITHM_GEN.md b/docs/HUYA_HDID_ALGORITHM_GEN.md index 68005dc..f5043ef 100644 --- a/docs/HUYA_HDID_ALGORITHM_GEN.md +++ b/docs/HUYA_HDID_ALGORITHM_GEN.md @@ -299,3 +299,21 @@ lib 内置大量模拟器检测串(qemu_pipe / mumuvmm / genymotion / windroye 全链路用同一份虚拟 profile 保持一致性,铸币=整套身份模板随机化。 - 真机/未决项:sGuid 签发是否依赖 qimei16/36、hdid/cdid/sdid 上报(采集器还是把设备信息上报 给 dfpReport 侧),这些在 §六 的注册链实测中一并验证。 + +### 11.6 doLaunch wire 已离线还原(2026-08-29 续) +- **JCE struct 规格**(dex writeTo 精确还原,全部位于 classes9/com/duowan/HUYA): + - `LiveLaunchReq`:t0=tId(UserId) t1=tLiveUB(LiveUserbase) t2=bSupportDomain(int16) + - `LiveUserbase`:t0=eSource(=2) t1=eType(=1) t2=tUAEx(LiveAppUAEx) + - `LiveAppUAEx`:t1=sIMEI t2=sAPN t3=sNetType t4=sDeviceId t5=sMId + - `UserId`(classes11):t0=lUid(int64) t1=sGuid t2=sToken t3=sHuYaUA t4=sCookie + t5=iTokenType t6=sDeviceInfo t7=sQIMEI + - `LiveLaunchRsp`:**t0=sGuid**(铸币目标) t1=iTime t2=vProxyList t3=eAccess t4=sClientIp +- **信封**:UniPacket(同密码登录);servant=`launch`(`@WupServant("launch")`),func=`doLaunch`; + sBuffer=map 键 `_wup_data`;4B 大端长度前缀。 +- **工具 `tools/huya_launch_mint.py`**:build_live_launch_wup(profile)→请求字节、 + parse_launch_rsp(bytes)→候选 sGuid(递归解 SIMPLE_LIST 载荷)、`--self-test`(回环)/ + `--dump`/`--live [url]`。默认虚拟 profile 见 DEFAULT_PROFILE(mid 16hex、imei 15 位、 + device_id 32hex、guid=""→服务端签发)。 +- **传输待 live 实测确认**:hyns/KiwiServant(a09)栈的最终 URL/头(先按 wup.huya.com 平铺, + 与服务端对拍后再校准);响应 sBuffer 可能是 map 假象(解析器已兼容 + bytes/struct 两种形态,sGuid_candidates 兜底)。 diff --git a/tools/huya_launch_mint.py b/tools/huya_launch_mint.py new file mode 100644 index 0000000..64f0f78 --- /dev/null +++ b/tools/huya_launch_mint.py @@ -0,0 +1,362 @@ +#!/usr/bin/env python3 +"""虎牙 live-launch doLaunch 复放器 —— GUID(sGuid) 铸币机原型 (PC 侧, 离线自检先行). + +背景 (docs/HUYA_HDID_ALGORITHM_GEN.md §11): + 32hex 登录帧 hdid = 服务端 doLaunch 下发的 sGuid (LiveLaunchRsp.tag0), + App 经 HalImpl.sGuidProperty → WupHelper.getGuid → HyDeviceProxy.setAppInfoId(pnc.a) + → NativeBridge.b(100) → native getGUID 缓存回写, 全程无本地公式. + 铸币 = 构造虚拟设备指纹的 doLaunch 请求 → 服务端签发 sGuid → 登录帧 t1.t0 用它. + +本工具: + - build_live_launch_wup(profile) : UniPacket 信封 + LiveLaunchReq JCE (规格来自 + classes9/com/duowan/HUYA/{LiveLaunchReq,LiveUserbase,LiveAppUAEx,LiveLaunchRsp} + + classes11 UserId, servant=@WupServant("launch") func=doLaunch) + - parse_launch_rsp(bytes) : 解 UniPacket → sBuffer → LiveLaunchRsp → sGuid + - --self-test : 编码→解码回环自检 (不联网) + - --dump : 打印请求体结构/hex (不联网) + - --live [url] : 真实发送 (默认 https://wup.huya.com) + +传输细节 (hyns/KiwiServant a09 栈) 未完全静态定案: 先按经典 UniPacket + wup.huya.com +平铺, live 测试确认服务端接受度后再校准; 字段/tag 均已按 dex 精确还原. + +用法: + python tools/huya_launch_mint.py --self-test + python tools/huya_launch_mint.py --dump --mid a1b2c3d4e5f60718 --imei 860000000000000 + python tools/huya_launch_mint.py --live [url] +""" +from __future__ import annotations + +import argparse +import struct +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent +if str(HERE) not in sys.path: + sys.path.insert(0, str(HERE)) + +from huya_wup_encoder import _Writer # noqa: E402 + +try: + from core.huya.taf_protocol import TafInputStream, TafType # noqa: E402 + HAVE_TAF = True +except Exception: # pragma: no cover + HAVE_TAF = False + +WUP_URL = "https://wup.huya.com" +SERVANT = "launch" +FUNC = "doLaunch" + +# --------------------------------------------------------------------------- +# JCE 编码 (写端) —— 规格来自 dex writeTo() +# --------------------------------------------------------------------------- + + +def _w_string_or_skip(w: _Writer, tag: int, v: str | None) -> None: + if v: + w.string(tag, v) + + +def encode_live_launch_req(profile: dict) -> bytes: + """LiveLaunchReq: t0=tId(UserId) t1=tLiveUB(LiveUserbase) t2=bSupportDomain(int16). + + profile 键: mid, imei, device_id, guid(当前sGuid, 新设备=空), huya_ua, + model, qimei, luid, apn, net_type + """ + w = _Writer() + w.struct_begin(0) # UserId tId + w.int64(0, int(profile.get("luid", 0))) + _w_string_or_skip(w, 1, profile.get("guid")) + _w_string_or_skip(w, 2, profile.get("token")) + _w_string_or_skip(w, 3, profile.get("huya_ua")) + _w_string_or_skip(w, 4, profile.get("cookie")) + w.int16(5, int(profile.get("i_token_type", 0))) + _w_string_or_skip(w, 6, profile.get("model")) + _w_string_or_skip(w, 7, profile.get("qimei")) + w.struct_end() + + w.struct_begin(1) # LiveUserbase: t0=eSource t1=eType t2=tUAEx + w.int16(0, int(profile.get("e_source", 2))) + w.int16(1, int(profile.get("e_type", 1))) + w.struct_begin(2) # LiveAppUAEx: t1=sIMEI t2=sAPN t3=sNetType t4=sDeviceId t5=sMId + _w_string_or_skip(w, 1, profile.get("imei")) + _w_string_or_skip(w, 2, profile.get("apn")) + _w_string_or_skip(w, 3, profile.get("net_type")) + _w_string_or_skip(w, 4, profile.get("device_id")) + _w_string_or_skip(w, 5, profile.get("mid")) + w.struct_end() + w.struct_end() + + w.int16(2, int(profile.get("b_support_domain", 1))) # bSupportDomain + return w.get() + + +def build_live_launch_wup(profile: dict, request_id: int | None = None) -> bytes: + """UniPacket 信封 (与密码登录同构): t1 version=3 ... t7 sBuffer=map.""" + if request_id is None: + request_id = int.from_bytes(__import__("os").urandom(4), "big") & 0x7FFFFFFF + req_jce = encode_live_launch_req(profile) + + sb = _Writer() + sb.map_begin(0, 1) + sb.string(0, "_wup_data") + sb.bytes(1, req_jce) + s_buffer = sb.get() + + w = _Writer() + w.int16(1, 3) # iVersion = 3 + w.int8(2, 0) # cPacketType + w.int8(3, 0) # iMessageType + w.int32(4, request_id) + w.string(5, SERVANT) + w.string(6, FUNC) + w.bytes(7, s_buffer) + w.int32(8, 0) # iTimeout + w.map_begin(9, 0) # context + w.map_begin(10, 0) # status + body = w.get() + return struct.pack(">I", 4 + len(body)) + body + + +# --------------------------------------------------------------------------- +# JCE 解码 (读端) +# --------------------------------------------------------------------------- + +def _payload(i: TafInputStream, dtype: int): + """读当前 head 之后的 payload (head 已由调用方 read_head 消费).""" + buf = i.buf + if dtype == TafType.ZERO: + return 0 + if dtype == TafType.INT8: + return struct.unpack("b", buf.read(1))[0] + if dtype == TafType.INT16: + return struct.unpack(">h", buf.read(2))[0] + if dtype == TafType.INT32: + return struct.unpack(">i", buf.read(4))[0] + if dtype == TafType.INT64: + return struct.unpack(">q", buf.read(8))[0] + if dtype == TafType.STRING1: + n = buf.read(1)[0] + return buf.read(n).decode("utf-8", "replace") + if dtype == TafType.STRING4: + n = struct.unpack(">I", buf.read(4))[0] + return buf.read(n).decode("utf-8", "replace") + if dtype == TafType.MAP or dtype == TafType.LIST: + raise ValueError(f"map/list 需单独处理 dtype=0x{dtype:02x}") + if dtype == TafType.SIMPLE_LIST: + # 元素类型 head(0,INT8) + 长度 head(0,intN)+value + 数据 + i.read_head() # 元素类型 + ltag, ldtype = i.read_head() + n = _payload(i, ldtype) + return buf.read(n) + raise ValueError(f"payload dtype=0x{dtype:02x} @pos={buf.tell()}") + + +def _field_count(i: TafInputStream) -> int: + """map/list 的条目数 (读长度 head + value).""" + tag, dtype = i.read_head() + return _payload(i, dtype) + + +def _read_struct(i: TafInputStream, depth: int = 0) -> dict: + """读一个匿名 struct 到 STRUCT_END, 返回 tag->value. map/list 内容做粗显.""" + out: dict[str, object] = {} + buf = i.buf + while True: + pos = buf.tell() + try: + tag, dtype = i.read_head() + except Exception: + break + if dtype == TafType.STRUCT_END: + break + if dtype in (TafType.MAP, TafType.LIST): + n = _field_count(i) + out[tag] = f"<{'map' if dtype == TafType.MAP else 'list'} {n} @{pos:#x}>" + elif dtype == TafType.STRUCT_BEGIN: + out[tag] = _read_struct(i, depth + 1) + else: + out[tag] = _payload(i, dtype) + return out + + +def parse_launch_rsp(resp: bytes) -> dict: + """解 UniPacket 响应: 顶层字段 + sBuffer map (key->bytes/struct), 捞候选 sGuid. + + 返回 {'header': {...}, 'sGuid_candidates': [...], 'structs': [...]}。 + """ + if not HAVE_TAF: + raise RuntimeError("缺少 core.huya.taf_protocol, 无法解码") + # WUP 帧带 4 字节大端长度前缀 (len = 4+body); 剥离后解析 + if len(resp) >= 4: + declared = struct.unpack(">I", resp[:4])[0] + if declared == len(resp): + resp = resp[4:] + i = TafInputStream(resp) + header: dict[str, object] = {} + sguid_candidates: list[str] = [] + structs: list[dict] = [] + buf = i.buf + while True: + try: + tag, dtype = i.read_head() + except Exception: + break + if dtype == TafType.STRUCT_END: + break + if dtype == TafType.MAP: + n = _field_count(i) + header[f"h{tag}"] = f"map<{n}>" + for _ in range(n): + ktag, kdtype = i.read_head() + key = _payload(i, kdtype) if kdtype in (TafType.STRING1, TafType.STRING4) else "?" + try: + vtag, vdtype = i.read_head() + except Exception: + break + if vdtype == TafType.SIMPLE_LIST: + blob = _payload(i, vdtype) + if isinstance(blob, bytes): + if len(blob) > 2: + try: + st = _read_struct(TafInputStream(blob)) + structs.append(st) + if 0 in st and isinstance(st[0], str): + sguid_candidates.append(st[0]) + except Exception: + pass + else: + header[f"h{tag}_map.{key}"] = blob.hex() + elif vdtype == TafType.STRUCT_BEGIN: + st = _read_struct(i) + structs.append(st) + if 0 in st and isinstance(st[0], str): + sguid_candidates.append(st[0]) + else: + header[f"h{tag}_map.{key}"] = _payload(i, vdtype) + elif dtype == TafType.LIST: + n = _field_count(i) + header[f"h{tag}"] = f"list<{n}>" + elif dtype == TafType.STRUCT_BEGIN: + structs.append(_read_struct(i)) + else: + payload = _payload(i, dtype) + header[f"h{tag}"] = payload + # SIMPLE_LIST/bytes 值 = 嵌套 WUP/JCE 载荷 → 递归捞 sGuid + if isinstance(payload, bytes) and len(payload) > 2: + try: + nested = parse_launch_rsp(payload) + except Exception: + nested = None + if nested: + structs.extend(nested["structs"]) + sguid_candidates.extend(nested["sGuid_candidates"]) + header[f"h{tag}_nested"] = nested["header"] + return {"header": header, "sGuid_candidates": sguid_candidates, "structs": structs} + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +DEFAULT_PROFILE = { + "mid": "1e8bdf7d4f7a01d3", + "imei": "860000000000000", + "device_id": "3b5c1a9f22e7d40c88a6f5b3012e947d", + "guid": "", + "huya_ua": "android&13.4.22&xxx&30", + "model": "M2102J2SC", + "qimei": "", + "luid": 0, + "apn": "", + "net_type": "", + "token": "", + "cookie": "", +} + + +def self_test() -> None: + """编码→解码回环: 生成请求 JCE + 信封, 再解析 (验证层次正确).""" + p = dict(DEFAULT_PROFILE) + p["guid"] = "0a7dfaa882938a6ab502511452142c57" + req_jce = encode_live_launch_req(p) + print(f"[self-test] LiveLaunchReq JCE: {len(req_jce)}B") + # 解析回环: 构造一个假响应 = 信封 + LiveLaunchRsp{guid} + rsp_jce_w = _Writer() + rsp_jce_w.string(0, "1a2b3c4d5e6f708192a3b4c5d6e7f809") + rsp_jce_w.int32(1, 1700000000) + rsp_jce_w.int32(3, 1) + rsp_jce_w.string(4, "1.2.3.4") + rsp_jce = rsp_jce_w.get() + parsed = parse_launch_rsp(rsp_jce) + print(f"[self-test] roundtrip parse -> {parsed}") + print("[self-test] OK") + + +def dump_req(profile: dict) -> None: + pkt = build_live_launch_wup(profile) + print(f"[dump] doLaunch UniPacket: {len(pkt)}B") + print(pkt.hex()) + print("[dump] field profile:", profile) + + +def live(profile: dict, url: str) -> None: + import requests + + pkt = build_live_launch_wup(profile) + print(f"[live] POST {url} body={len(pkt)}B") + r = requests.post( + url, + data=pkt, + headers={ + "Content-Type": "application/multipart-formdata; charset=UTF-8", + "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 11)", + "Accept-Encoding": "gzip", + }, + timeout=20, + ) + print(f"[live] HTTP {r.status_code} len={len(r.content)}") + if r.status_code != 200: + print(r.content[:300]) + return + parsed = parse_launch_rsp(r.content) + print("[live] parsed:", parsed) + guid = None + # 尝试多种可能路径拿 sGuid + for k, v in parsed.items(): + if isinstance(v, dict) and 0 in v and isinstance(v[0], str): + guid = v[0] + print("[live] sGuid =", guid) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--self-test", action="store_true") + ap.add_argument("--dump", action="store_true") + ap.add_argument("--live", nargs="?", const=WUP_URL, default=None) + ap.add_argument("--mid", default=None) + ap.add_argument("--imei", default=None) + ap.add_argument("--device-id", default=None) + ap.add_argument("--guid", default=None) + ap.add_argument("--model", default=None) + args = ap.parse_args() + + profile = dict(DEFAULT_PROFILE) + for k, v in (("mid", args.mid), ("imei", args.imei), + ("device_id", args.device_id), ("guid", args.guid), + ("model", args.model)): + if v: + profile[k] = v + + if args.self_test: + self_test() + elif args.dump: + dump_req(profile) + elif args.live: + live(profile, args.live) + else: + ap.print_help() + + +if __name__ == "__main__": + main() \ No newline at end of file