live-hub-py/tools/hy_har_parse.py

"""解析虎牙 App HAR：解码 WUP(TAF) 全字段 + dckey/check 分析。

用法:
  .venv/bin/python tools/hy_har_parse.py <file.har> [--req] [--resp]
"""
from __future__ import annotations

import base64
import json
import sys

sys.path.insert(0, ".")

from tools.taf_decode import (  # noqa: E402
    Taf, STRUCT_END, MAP, SIMPLE_LIST, decode_struct_bytes,
)

STR = (6, 7)  # STRING1, STRING4
SIMPLE = 13


def norm_payload(data: bytes) -> bytes:
    """去掉可选 4 字节 WUP 长度前缀。"""
    if len(data) < 4:
        return data
    declared = int.from_bytes(data[0:4], "big")
    if declared in (len(data), len(data) - 4):
        return data[4:]
    return data


def decode_wup_header(data: bytes):
    """返回 (servant, func, sBuffer_bytes)。"""
    t = Taf(norm_payload(data))
    servant = ""
    func = ""
    sbuf = b""
    while not t.eof():
        try:
            tag, dt = t.read_head()
        except EOFError:
            break
        if tag > 10:
            break
        if tag == 5 and dt in STR:
            servant = t.read_string_value(dt)
        elif tag == 6 and dt in STR:
            func = t.read_string_value(dt)
        elif tag == 7 and dt == SIMPLE:
            t.read_head()
            n = t.read_len()
            sbuf = t.d[t.p:t.p + n]
            t.p += n
        else:
            t.skip_value(dt)
    return servant, func, sbuf


def decode_sbuffer_map(sbuf: bytes):
    """sBuffer 是一个 Map<str, bytes>，返回 {key: bytes}。"""
    if not sbuf:
        return {}
    t = Taf(sbuf)
    out = {}
    try:
        tag, dt = t.read_head()
        if not (tag == 0 and dt == MAP):
            return {"<raw>": sbuf}
        n = t.read_len()
        for _ in range(min(n, 20)):
            _, kt = t.read_head()
            k = t.read_string_value(kt) if kt in STR else f"<k{kt}>"
            _, vt = t.read_head()
            if vt == SIMPLE:
                t.read_head()
                ln = t.read_len()
                out[k] = t.d[t.p:t.p + ln]; t.p += ln
            else:
                t.skip_value(vt)
    except Exception as e:
        out["<err>"] = str(e).encode()
    return out


def dump_har(path: str, show_req: bool = True, show_resp: bool = True):
    with open(path, encoding="utf-8", errors="ignore") as fp:
        h = json.load(fp)
    ents = h["log"]["entries"]

    targets = ("wup.huya.com", "statwup.huya.com", "wsapi.huya.com",
               "udbdf.huya.com", "udblgn.huya.com", "udbreg.huya.com")

    for i, e in enumerate(ents):
        req = e["request"]
        u = req["url"]
        if req["method"] != "POST" or not any(x in u for x in targets):
            continue
        pd = req.get("postData", {})
        txt = pd.get("text", "")
        mime = pd.get("mimeType", "")
        print("=" * 80)
        print(f"#{i} {u[:110]}")
        print(f"  mime={mime} textLen={len(txt)}")

        # ---- 请求 ----
        if txt:
            try:
                raw = base64.b64decode(txt)
            except Exception:
                raw = txt.encode("latin1")
            if len(raw) >= 6 and raw[4:6] == b"\x10\x03" and show_req:
                servant, func, sbuf = decode_wup_header(raw)
                print(f"  [REQ WUP] {servant}.{func}")
                for k, v in decode_sbuffer_map(sbuf).items():
                    print(f"    {k}: {json.dumps(decode_struct_bytes(v), ensure_ascii=False)[:700]}")
            else:
                # 加密 binary（如 dckey/check、statwup 应用层加密）
                print(f"  [REQ BINARY] {len(raw)}B hex={raw[:32].hex()}")
                analyze_binary(raw)

        # ---- 响应 ----
        resp = e.get("response", {})
        c = resp.get("content", {})
        rtext = c.get("text", "")
        if rtext and show_resp:
            if rtext.lstrip().startswith("{"):
                print(f"  [RESP JSON] {rtext[:220]}")
            else:
                try:
                    rraw = base64.b64decode(rtext)
                except Exception:
                    rraw = rtext.encode("latin1")
                if len(rraw) >= 6 and rraw[4:6] == b"\x10\x03":
                    servant, func, sbuf = decode_wup_header(rraw)
                    print(f"  [RESP WUP] {servant}.{func}")
                    for k, v in decode_sbuffer_map(sbuf).items():
                        print(f"    {k}: {json.dumps(decode_struct_bytes(v), ensure_ascii=False)[:700]}")
                else:
                    print(f"  [RESP BINARY] {len(rraw)}B hex={rraw[:32].hex()}")
        elif rtext:
            print(f"  [RESP raw] {rtext[:200]!r}")


def analyze_binary(raw: bytes):
    """对加密 body 做基础分析：长度/对齐/头部熵。"""
    print(f"    len={len(raw)}  mod16={len(raw) % 16}  mod64={len(raw) % 64}")
    # 头部是否像有固定结构（前几字节恒定）
    print(f"    head32={raw[:32].hex()}")
    print(f"    tail16={raw[-16:].hex()}")


if __name__ == "__main__":
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    path = args[0] if args else "-"
    show_req = "--req" not in sys.argv or True
    show_resp = "--resp" not in sys.argv or True
    dump_har(path)