yml2213
44109aa408
- JCE 规格: LiveLaunchReq{tId,tLiveUB,bSupportDomain}/LiveUserbase{eSource,eType,tUAEx}
LiveAppUAEx{sIMEI,sAPN,sNetType,sDeviceId,sMId}/UserId{lUid..sQIMEI}/LiveLaunchRsp{t0=sGuid}
(classes9/11 dex writeTo 精确还原)
- 信封: UniPacket, servant=launch (@WupServant), func=doLaunch, sBuffer key=_wup_data
- tools/huya_launch_mint.py: 请求构造 + 响应解析(sGuid_candidates 递归兜底) + self-test 回环 PASS
- 传输(hyns/KiwiServant)与响应形态留待 live 实测校准; 禁设备侧活动
362 lines
13 KiB
Python
362 lines
13 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)
|
|
|
|
传输细节 (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<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)
|
|
|
|
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() |