hello_js_reverse_skill/scripts/verify/verify_values.py

# -*- coding: utf-8 -*-
"""verify_values.py —— 用**真机同会话明文**校验取值生成器（tcaptcha/tdc_values.py）。

做法（不手工配平，全部从观测值反推）:
  1. 解密 collect -> 切 4 段 -> 解析 cd
  2. 按清单顺序把槽绑到语义名（itoken 占 2 槽）
  3. 从观测值里读出「会话量」（itoken / 会话随机盐 / 页面加载时刻）
     与「每次采集量」（当前时刻 / 采集计数 / 轨迹），塞进 Session
  4. 让生成器重算一遍，逐槽对比

期望: 除「每次采集量」之外**必须逐槽全等**；而每次采集量被喂了观测值后也应全等
（等于证明生成器对这些量的建模是对的，而不是巧合）。

用法:
    .venv/bin/python scripts/verify/verify_values.py \
        --build /tmp/1ms_bridge/session_build.js \
        --collects /tmp/1ms_bridge/session_collect.json /tmp/1ms_bridge/session_collect2.json
"""
import argparse
import base64
import json
import sys
from pathlib import Path
from urllib.parse import unquote

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tools"))

from tcaptcha.telemetry import modules as tm                                     # noqa: E402
from tcaptcha.telemetry.crypto import split_collect_segments, xtea_decrypt_bytes   # noqa: E402
from tcaptcha.telemetry.key import extract_xtea_key_pure            # noqa: E402
from tcaptcha.telemetry.values import Session, generate_cd          # noqa: E402


def slot_names(rows):
    out = []
    for r in rows:
        base = str(r["name"]).split("#")[0]
        out.extend([r["name"]] * (2 if base == "itoken" else 1))
    return out


def decrypt_cd(path, key):
    sess = json.load(open(path, encoding="utf-8"))
    if "collect" not in sess and isinstance(sess.get("data"), dict):
        sess = sess["data"]
    plain = xtea_decrypt_bytes(base64.b64decode(unquote(sess["collect"])), key) \
        .decode("latin1").rstrip("\x00")
    c1, traj, c2, sd = split_collect_segments(plain)
    obj = json.loads(c1 + traj + c2 + sd)
    return obj["cd"], json.loads(traj), sd


def main() -> int:
    ap = argparse.ArgumentParser(description="校验语义名取值生成器")
    ap.add_argument("--build", default="/tmp/1ms_bridge/session_build.js")
    ap.add_argument("--collects", nargs="+", required=True)
    args = ap.parse_args()

    js = open(args.build, encoding="utf-8", errors="replace").read()
    key = extract_xtea_key_pure(js)["key"]
    rows = tm.analyze(args.build)
    names = slot_names(rows)
    bases = [n.split("#")[0] for n in names]

    total_ok = total = 0
    for path in args.collects:
        cd, traj, sd = decrypt_cd(path, key)
        slots = list(zip(names, cd))

        # 注意 itoken 占 2 槽且同名，所以用列表而不是 dict
        itoken = [v for n, v in slots if n == "itoken"]
        sess_rand = next(v for n, v in slots if n.startswith("struct_40_9#")
                         and not n.endswith("_empty"))
        loaded = next(v for n, v in slots
                      if n.startswith("window_time#") and n.endswith("call_empty"))
        now = next(v for n, v in slots
                   if n.startswith("window_time#") and not n.endswith("call_empty"))
        calls = next(v for n, v in slots if n == "struct_46_9")

        s = Session(now=loaded)
        s.itoken = list(itoken)
        s.session_rand = sess_rand
        s.loaded_at = loaded
        s.calls = calls - 1                       # generate_cd 会 +1
        s.set_trajectory(traj)
        gen = generate_cd(rows, s, now=now)

        bad = [(names[i], cd[i], gen[i]) for i in range(len(cd)) if cd[i] != gen[i]]
        total += len(cd)
        total_ok += len(cd) - len(bad)
        print("%-46s cd=%d 不一致=%d" % (Path(path).name, len(cd), len(bad)))
        for n, o, g in bad:
            print("     %-38s 观测=%s 生成=%s"
                  % (n, json.dumps(o, ensure_ascii=False)[:44],
                     json.dumps(g, ensure_ascii=False)[:44]))
    print()
    print("逐槽全等: %d/%d" % (total_ok, total))
    return 0 if total_ok == total else 1


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