yml2213
ccfa742e3b
- encode_live_launch_req: 双层 struct_begin(0)+struct_end 收尾 (服务器tag0 tId 期望STRUCT) - build 只发 tReq 单键 (对齐真机 launch 帧, 不再追加 platform/version 等) - parse_launch_rsp: gzip 解压 + \x06\x20(32hex) sGuid 可靠提取 - live 矩阵实证: 同指纹同 sGuid / 变异指纹新 sGuid / device_id 不驱动, mid 驱动 - docs §11.8: 破案链 + 实证表 + 铸币闭环 (mid→doLaunch→sGuid→登录链)
392 lines
14 KiB
Python
392 lines
14 KiB
Python
#!/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)
|
|
|
|
实测状态 (2026-08-29):
|
|
- 信封/路径/键均被服务端接受: servant/func 回声, UniAttribute 找到 tReq 键;
|
|
但 tReq 值被拒: "read 'struct' type mismatch, tag: 0, get type: 12" (恒定).
|
|
- 诊断: 值=0x0c 复现同错 (E:空值->require field; F:缺键->not found key) => 服务端在
|
|
解析值, 但首字段始终判为 ZERO. 疑点: a09 传输类型由服务端动态配置
|
|
(IFunctionTranspotModule.getTransportType("launch#doLaunch")), HTTP POST 路径
|
|
未必是 doLaunch 真渠道; wire 已逐字节对齐 App (见 docs §11.6), 待真机帧定案.
|
|
|
|
用法:
|
|
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 re
|
|
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()
|
|
# 外层 LiveLaunchReq struct (tag0) —— 缺它整体结构上移一层
|
|
w.struct_begin(0)
|
|
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
|
|
w.struct_end() # 外层 LiveLaunchReq
|
|
return w.get()
|
|
|
|
|
|
def build_live_launch_wup(profile: dict, request_id: int | None = None) -> bytes:
|
|
"""UniPacket 信封 (与密码登录同构): t1 version=3 ... t7 sBuffer=map<string,bytes>."""
|
|
if request_id is None:
|
|
request_id = int.from_bytes(__import__("os").urandom(4), "big") & 0x7FFFFFFF
|
|
req_jce = encode_live_launch_req(profile)
|
|
|
|
# 真机实测 (WG 全解密捕获): launch servant 的 map 只有 "tReq" 一个键,
|
|
# 不追加 platform/version/channel 等 (此前追加属过度拟合, 服务器仍拒值)
|
|
entries = [("tReq", ("bytes", req_jce))]
|
|
sb = _Writer()
|
|
sb.map_begin(0, len(entries))
|
|
for k, (kind, val) in entries:
|
|
sb.string(0, k)
|
|
if kind == "bytes":
|
|
sb.bytes(1, val)
|
|
else:
|
|
sb.string(1, val)
|
|
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, 无法解码")
|
|
# HTTP 直连响应可能 gzip 压缩 (1f 8b)
|
|
if resp[:2] == b"\x1f\x8b":
|
|
import gzip as _g
|
|
try:
|
|
resp = _g.decompress(resp)
|
|
except Exception:
|
|
pass
|
|
# 可靠路径: tRsp 结构内 tag0 = sGuid(STRING4 0x06, len 0x20=32hex)
|
|
_sguid_re = re.search(rb"\x06\x20([0-9a-f]{32})", resp)
|
|
# 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] = ([_sguid_re.group(1).decode()] if _sguid_re else [])
|
|
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": "",
|
|
"platform": "android",
|
|
"version": "13.4.22",
|
|
"channel": "xxx",
|
|
"yyuid": "0",
|
|
"uid": "0",
|
|
}
|
|
|
|
|
|
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() |