hello_js_reverse_skill/scripts/analyze_collect_layout.py

# -*- coding: utf-8 -*-
"""
collect 明文「字段布局」分析器 —— 回答一个关键问题:

    `collect` 解密后的 `cd` 数组, 槽位顺序到底是"按 tdc.js 版本固定"还是"按会话/构建随机"?

为什么要问这个 (2026-10-11 排查 errorCode=12):
    tcaptcha/models/success_collect_template.b64 存的是**明文**模板 (base64 封装),
    generate_collect_success() 只回填 轨迹 / entry_url / 几个时间戳, 其余字段原样保留。
    真机成功样本解密后发现: **同一批值, 顺序完全不同** —— 若服务端按"本次构建的槽位表"
    解释 cd, 那么我们用旧槽位序发出的 collect 会被读成"自相矛盾的环境" -> 12。

本脚本把多份样本 (HAR + CDP 抓包 + 仓库模板) 的 cd 槽位打成类型指纹并对比顺序。

用法:
    .venv/bin/python scripts/analyze_collect_layout.py                 # 默认样本集
    .venv/bin/python scripts/analyze_collect_layout.py --json /tmp/cap2.json
"""
from __future__ import annotations

import argparse
import base64
import json
import os
import re
import sys
import urllib.parse
from pathlib import Path
from typing import Dict, List, Optional, Tuple

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from tcaptcha.crypto import xtea_decrypt_bytes  # noqa: E402
from tcaptcha.tdc_key import extract_xtea_key_pure  # noqa: E402

HARS = [
    ("har/原始抓包/签到成功-1.har", "签到成功-1"),
    ("har/原始抓包/点选.har", "点选"),
    ("har/login_checkin.har", "login_checkin"),
]


def _decrypt_collect(collect_b64: str, tdc_js: str) -> Optional[dict]:
    key = extract_xtea_key_pure(tdc_js)["key"]
    pt = xtea_decrypt_bytes(base64.b64decode(collect_b64), key)
    text = pt.decode("latin1").rstrip("\x00").strip()
    try:
        return json.loads(text)
    except Exception:
        return None


def from_har(path: str) -> List[Tuple[str, dict]]:
    if not os.path.exists(path):
        return []
    data = json.load(open(path, encoding="utf-8"))
    tdcs: Dict[str, str] = {}
    sess2sid: Dict[str, str] = {}
    verifies: List[dict] = []
    for e in data["log"]["entries"]:
        url = e["request"]["url"]
        content = (e["response"].get("content") or {}).get("text") or ""
        if "tdc.js" in url and content:
            q = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
            tdcs[(q.get("app_data") or [""])[0]] = content
        elif "cap_union_prehandle" in url and content:
            m = re.search(r"^[^(]*\((.*)\)\s*$", content, re.S)
            if m:
                try:
                    j = json.loads(m.group(1))
                    sess2sid[j.get("sess", "")] = str(j.get("sid", ""))
                except Exception:
                    pass
        elif "cap_union_new_verify" in url:
            pd = (e["request"].get("postData") or {}).get("text") or ""
            if pd:
                verifies.append(dict(urllib.parse.parse_qsl(pd, keep_blank_values=True)))
    out = []
    for i, v in enumerate(verifies):
        js = tdcs.get(sess2sid.get(v.get("sess", ""), ""))
        if not js:
            continue
        j = _decrypt_collect(v["collect"], js)
        if j is not None:
            out.append((f"{Path(path).name}#{i}", j))
    return out


def from_cdp_capture(path: str) -> List[Tuple[str, dict]]:
    if not os.path.exists(path):
        return []
    d = json.load(open(path, encoding="utf-8"))
    recs = d.get("records") or []
    tdcs = [(r["url"], r.get("body") or "") for r in recs if "tdc.js" in r["url"] and r.get("body")]
    vers = [r for r in recs if "verify" in r["url"] and r.get("postData")]
    out = []
    for i, v in enumerate(vers):
        form = dict(urllib.parse.parse_qsl(v["postData"], keep_blank_values=True))
        sess = form.get("sess", "")
        # sess 前 2 段与 tdc.js 的 app_data(sid) 无直接映射, 这里按时间就近取**最后一份**构建
        js = tdcs[-1][1] if tdcs else ""
        if not js:
            continue
        j = _decrypt_collect(form["collect"], js)
        if j is not None:
            out.append((f"{Path(path).name}#{i}", j))
    return out


def slot_tags(cd: List) -> List[str]:
    tags = []
    for v in cd:
        if isinstance(v, str):
            if v.startswith("http"):
                tags.append("url")
            elif "Mozilla" in v:
                tags.append("UA")
            elif v.startswith("GgoAAA"):
                tags.append("png")
            elif re.fullmatch(r"[-+]\d\d", v):
                tags.append("tz")
            elif v == "":
                tags.append("empty")
            elif len(v) <= 6 and v.isupper():
                tags.append("enc:" + v)
            else:
                tags.append("str:" + v[:10])
        elif isinstance(v, list):
            if len(v) == 2 and all(isinstance(x, int) for x in v):
                tags.append("vec2")
            elif v and isinstance(v[0], list):
                tags.append("traj")
            else:
                tags.append("list%d" % len(v))
        elif isinstance(v, bool):
            tags.append("bool")
        else:
            tags.append("num")
    return tags


def main() -> int:
    ap = argparse.ArgumentParser(description="collect 明文字段布局分析")
    ap.add_argument("--json", action="append", default=[], help="额外加入的 cdp_capture JSON")
    args = ap.parse_args()

    samples: List[Tuple[str, dict]] = []
    for p in args.json:
        samples += from_cdp_capture(p)
    for p, tag in HARS:
        for name, j in from_har(str(ROOT / p)):
            samples.append((f"{tag}#{name.split('#')[-1]}", j))
    tpl_path = ROOT / "tcaptcha" / "models" / "success_collect_template.b64"
    if tpl_path.exists():
        tpl = json.loads(base64.b64decode(tpl_path.read_text(encoding="ascii"))
                         .decode("latin1").rstrip("\x00").strip())
        samples.append(("TEMPLATE(仓库在用)", tpl))

    if not samples:
        print("没有可用样本")
        return 1

    for name, j in samples:
        tags = slot_tags(j["cd"])
        print(f"\n[{name}]  len(cd)={len(j['cd'])}  sd={j.get('sd')}")
        print("   " + " ".join(f"{i}:{t}" for i, t in enumerate(tags)))

    print("\n=== 槽位指纹两两对比 (看顺序是否随会话变化) ===")
    sigs = [(n, slot_tags(j["cd"])) for n, j in samples]
    base_name, base = sigs[0]
    print(f"  基准 = {base_name}")
    for n, s in sigs[1:]:
        diff = [i for i in range(max(len(base), len(s)))
                if (base[i] if i < len(base) else None) != (s[i] if i < len(s) else None)]
        print(f"    {n:22s} 顺序相同={s == base}  不同槽位数={len(diff)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())