yml2213
fe1415508a
tdc_program 新增 symbolic_assignments(): 对小程序做线性符号求值, 把字节码还原成
`r6 = c.navigator.userAgent` 这样的可读表达式（覆盖字面串/BIND_THIS/MAKE_(REG_)PROP_REF/
GET_GLOBAL_PROP/NEW_FUNC/CALL_*/STORE_REG/LOAD_REG_IMM/REVERSE_TOP2/DUP/算术）。
tdc_layout 新增:
* slot_value_expr(): 精确规则 —— 采集器模块把值存进某寄存器, `get` 闭包只返回它
（caps 的父寄存器就是取值寄存器）; 含"模块里唯一闭包"的兜底
* slot_semantics(): 合并"工厂赋值链 + require 委派 + 内部闭包"三条线索
* 输出改成叶子语义 + 真机明文并排核对
实测确认（线上构建 + 同会话解出的真机 collect）:
* 清单模块 = registry[7], 构造方式是 `r6[r6.length] = require(N)` 追加
-> **require 顺序即 cd 槽位顺序**（37 个 require + 1 个字面量槽 = 38, 与真机完全吻合）
* 语义已精确命中: cd[3] <- c.navigator.userAgent, cd[5] <- c.navigator.languages
* 仍有若干槽的语义标签未对齐（取值寄存器规则需再收口）, 已在账本记录
265 lines
11 KiB
Python
265 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
tdc_layout.py —— 静态导出 `collect` 明文的 **cd 槽位表**（纯算, 不跑 JS, 不跑环境）。
|
||
|
||
背景（这条链是 2026-10-11 排查 errorCode=12 时打通的, 见 har/1ms_dianxuan/PROGRESS.md #31/#32）:
|
||
真机成功会话的 collect 明文解出来后, 发现 **5 份同日样本的 `cd` 槽位数(38)、取值集合、`sd`
|
||
全都一样, 但槽位顺序 5/5 互不相同** —— 也就是"哪些字段排在哪一槽"是**按构建随机化**的。
|
||
仓库里的 success_collect_template.b64 是**某一次**的顺序, 拿它发的 collect 会被服务端
|
||
按"本次构建的槽位表"读成自相矛盾的环境 -> errorCode=12。
|
||
|
||
槽位序从哪来（程序结构, 见 tools/tdc_program.py）:
|
||
1. 顶层 @0 先把 62 个**模块工厂**按 id 填进一个数组 (registry[id] = function(module, exports, require));
|
||
2. 其中一个模块（下称"清单模块"）返回一个**数组**, 它的构造方式是一串
|
||
r6[r6.length] = require(N)
|
||
也就是 **`cd` 的槽位顺序 = 这串 require 的顺序**;
|
||
3. 另一个模块负责把这些采集器按 `.on` / `.reset` / `.get` 分到三个数组, 再另有一个模块
|
||
遍历"有 .get() 的那些", 依次取 `get()` 的值拼成 `{"cd":[...],"sd":{...}}`。
|
||
|
||
所以本工具做的事:
|
||
* 找清单模块（= push(require(N)) 次数最多的那个模块）;
|
||
* 读出它的 require 序列 -> 槽位表 [(槽位号, 模块 id, 工厂 entry), ...];
|
||
* 把每个槽位子模块（含其内部闭包）里的**环境属性字面量**列出来, 作为该槽的语义线索;
|
||
* 可选: 读一份已解密的真机 collect 明文, 与槽位表并排打印, 人工/程序核对。
|
||
|
||
用法:
|
||
# 只看槽位表
|
||
.venv/bin/python tools/tdc_layout.py tcaptcha/core_js/tdc_sample.js
|
||
|
||
# 和真机明文并排核对 (明文是怎么来的: 同会话 tdc.js 密钥解密 verify 的 collect)
|
||
.venv/bin/python tools/tdc_layout.py <tdc.js> --plain /tmp/real_collect_plain.txt
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
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))
|
||
if str(ROOT / "tools") not in sys.path:
|
||
sys.path.insert(0, str(ROOT / "tools"))
|
||
|
||
import tdc_program as tp # noqa: E402
|
||
|
||
# 采集器用来读环境的属性名（用于给槽位打语义标签, 不是判定依据）
|
||
ENV_KEYS = [
|
||
"userAgent", "platform", "languages", "language", "hardwareConcurrency", "deviceMemory",
|
||
"colorDepth", "pixelDepth", "availHeight", "availWidth", "deviceXDPI", "logicalXDPI",
|
||
"fontSmoothingEnabled", "appVersion", "appName", "vendor", "cookieEnabled", "timezone",
|
||
"getTimezoneOffset", "toDataURL", "getParameter", "getExtension", "TCaptchaReferrer",
|
||
"webdriver", "referrer", "href", "width", "height", "innerWidth", "innerHeight",
|
||
"devicePixelRatio", "characterSet", "MAX_TEXTURE_SIZE", "WEBGL_debug_renderer_info",
|
||
]
|
||
|
||
|
||
def _fold_str_before(code: List[dict], end: int, span: int = 40) -> Optional[str]:
|
||
"""从 end 往前折叠一个字面串（PUSH_EMPTY_STR + 连续 PUSH_STR_CHAR）。"""
|
||
chars: List[str] = []
|
||
i = end
|
||
while i >= 0 and i > end - span and code[i]["name"] == "PUSH_STR_CHAR":
|
||
chars.append(chr(code[i]["operands"][0] & 0xFF))
|
||
i -= 1
|
||
if i >= 0 and code[i]["name"] == "PUSH_EMPTY_STR":
|
||
return "".join(reversed(chars))
|
||
return None
|
||
|
||
|
||
def slot_value_expr(p: tp.Program, entry: int) -> Optional[str]:
|
||
"""**精确**取一个槽位模块"这一槽的值是什么"。
|
||
|
||
规则（实测形状）: 采集器模块最后做
|
||
r3.exports.get = function () { ... return <被捕获的那个寄存器> ... }
|
||
也就是 `get` 闭包**只返回它捕获的寄存器**。所以:
|
||
1. 在工厂里找到被赋给属性 `get` 的那个 CREATE_CLOSURE;
|
||
2. 从它的 caps 里取出**父寄存器号**;
|
||
3. 该寄存器在工厂里的赋值表达式, 就是这一槽的语义。
|
||
例: 模块 18 -> caps=[r3=r6], 工厂里 `r6 = c.navigator.userAgent`。
|
||
"""
|
||
code = p.function_code(entry)
|
||
assigns = dict(p.symbolic_assignments(entry, limit=60))
|
||
for k, inst in enumerate(code):
|
||
if inst["name"] != "CREATE_CLOSURE":
|
||
continue
|
||
# 形状: <"get" 字面量>; REVERSE_TOP2; JUMP -> @site; @site: CREATE_CLOSURE
|
||
j = k - 1
|
||
if j >= 0 and code[j]["name"] == "JUMP":
|
||
j -= 1
|
||
if _fold_str_before(code, j) != "get":
|
||
continue
|
||
site = next((c for c in p.closures if c.pc == inst["pc"]), None)
|
||
if site is None:
|
||
continue
|
||
for _child, parent in site.caps: # caps 是 (子寄存器, 父寄存器)
|
||
expr = assigns.get(f"r{parent}")
|
||
if expr:
|
||
return expr
|
||
# 兜底（覆盖更广）: 采集器模块里通常**只有一个闭包**，它捕获的那个父寄存器就是取值寄存器
|
||
for c in p.closures:
|
||
if p.owner(c.pc) != entry or len(c.caps) != 1:
|
||
continue
|
||
_child, parent = c.caps[0]
|
||
expr = assigns.get(f"r{parent}")
|
||
if expr:
|
||
return expr
|
||
return None
|
||
|
||
|
||
def slot_semantics(p: tp.Program, reg: Dict[int, int], mid: int, depth: int = 2) -> List[str]:
|
||
"""还原某个槽位子模块"到底取什么值"。
|
||
|
||
三条线索合并（都来自 tdc_program 的符号求值, 不跑 JS）:
|
||
1. 模块工厂自身的赋值链, 例如 `r6 = c.navigator.userAgent`;
|
||
2. **委派**: 有些模块只是 `r6 = r5(N)`（require 另一个模块再导出），要跟进去;
|
||
3. **`get` 闭包内部**: canvas/WebGL 这类是延迟取值, 真正读环境在它内部创建的闭包里。
|
||
"""
|
||
exprs: List[str] = []
|
||
seen: set = set()
|
||
|
||
def visit(m: int, d: int) -> None:
|
||
if m in seen or d < 0 or m not in reg:
|
||
return
|
||
seen.add(m)
|
||
entry = reg[m]
|
||
for _r, v in p.symbolic_assignments(entry, limit=14):
|
||
v = v.strip()
|
||
if not v:
|
||
continue
|
||
exprs.append(v)
|
||
mm = re.match(r"r\d+\((\d+)\)$", v) # 纯委派 r6 = r5(N)
|
||
if mm:
|
||
visit(int(mm.group(1)), d - 1)
|
||
for c in p.closures: # 该模块内部创建的闭包（含 get 实现）
|
||
if p.owner(c.pc) != entry:
|
||
continue
|
||
for _r, v in p.symbolic_assignments(c.target, limit=10):
|
||
if v.strip():
|
||
exprs.append(v.strip())
|
||
mm = re.match(r"r\d+\((\d+)\)$", v.strip())
|
||
if mm:
|
||
visit(int(mm.group(1)), d - 1)
|
||
|
||
visit(mid, depth)
|
||
uniq, out = set(), []
|
||
for e in exprs:
|
||
if e not in uniq:
|
||
uniq.add(e)
|
||
out.append(e)
|
||
return out
|
||
|
||
|
||
def find_manifest_module(p: tp.Program) -> Tuple[int, int, List[int]]:
|
||
"""找"清单模块": 体内 `push(require(N))` 次数最多者。
|
||
|
||
返回 (模块 id, 工厂 entry, require 序列)。
|
||
"""
|
||
reg = dict(p.module_registry(0))
|
||
best = (0, -1, [])
|
||
for mid, entry in reg.items():
|
||
code = p.function_code(entry)
|
||
seq: List[int] = []
|
||
for k, inst in enumerate(code):
|
||
if inst["name"] != "PUSH_IMM" or not inst["operands"]:
|
||
continue
|
||
if k >= 1 and code[k - 1]["name"] == "LOAD_REG_IMM" \
|
||
and k + 1 < len(code) and code[k + 1]["name"] == "CALL_FUNC":
|
||
seq.append(int(inst["operands"][0]))
|
||
if len(seq) > len(best[2]):
|
||
best = (mid, entry, seq)
|
||
return best
|
||
|
||
|
||
def transitive_strings(p: tp.Program, entry: int, depth: int = 3) -> List[str]:
|
||
"""模块自身 + 它内部创建的闭包（递归 depth 层）里的字面串。"""
|
||
seen, out = set(), []
|
||
frontier = {entry}
|
||
for _ in range(depth):
|
||
nxt = set()
|
||
for e in frontier:
|
||
if e in seen:
|
||
continue
|
||
seen.add(e)
|
||
for pc, s in p.strings:
|
||
if p.owner(pc) == e:
|
||
out.append(s)
|
||
for c in p.closures:
|
||
if p.owner(c.pc) == e:
|
||
nxt.add(c.target)
|
||
frontier = nxt
|
||
return out
|
||
|
||
|
||
def env_hints(strings: List[str]) -> List[str]:
|
||
hits = []
|
||
for s in strings:
|
||
for k in ENV_KEYS:
|
||
if k.lower() in s.lower() and k not in hits:
|
||
hits.append(k)
|
||
return hits
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="静态导出 collect 的 cd 槽位表")
|
||
ap.add_argument("path")
|
||
ap.add_argument("--plain", default=None, help="已解密的真机 collect 明文 (json 文本)")
|
||
ap.add_argument("--json", default=None, help="把槽位表写成 JSON")
|
||
args = ap.parse_args()
|
||
|
||
p = tp.load(args.path)
|
||
reg = dict(p.module_registry(0))
|
||
mid, entry, seq = find_manifest_module(p)
|
||
print(f"文件 : {args.path}")
|
||
print(f"模块总数 : {len(reg)}")
|
||
print(f"清单模块 : registry[{mid}] = @{entry}")
|
||
print(f"槽位序列长度 : {len(seq)} (真机 cd 槽数 = 38, 差的部分是字面量槽)")
|
||
print(f"require 序列 : {seq}")
|
||
|
||
plain_cd: Optional[list] = None
|
||
if args.plain:
|
||
txt = open(args.plain, encoding="latin1").read().rstrip("\x00").strip()
|
||
obj = json.loads(txt)
|
||
plain_cd = obj.get("cd")
|
||
print(f"真机明文 cd 槽数: {len(plain_cd)}")
|
||
|
||
table = []
|
||
print("\n槽位 模块id 工厂entry 取值语义(静态还原) 真机值")
|
||
print("-" * 112)
|
||
for i, m in enumerate(seq):
|
||
tgt = reg.get(m)
|
||
expr = slot_value_expr(p, tgt) if tgt else None
|
||
# 有些模块只是"再 require 一层"（r6 = r5(M)），跟进一层
|
||
mm = re.match(r"r\d+\((\d+)\)$", (expr or "").strip())
|
||
if mm and int(mm.group(1)) in reg:
|
||
sub = slot_value_expr(p, reg[int(mm.group(1))])
|
||
if sub:
|
||
expr = f"{sub} [via require({mm.group(1)})]"
|
||
if not expr:
|
||
expr = "; ".join(slot_semantics(p, reg, m)[:2]) if tgt else ""
|
||
val = ""
|
||
if plain_cd is not None and i < len(plain_cd):
|
||
v = repr(plain_cd[i])
|
||
val = v if len(v) <= 30 else v[:30] + "..."
|
||
sem = expr or ""
|
||
if len(sem) > 44:
|
||
sem = sem[:44] + "..."
|
||
table.append({"slot": i, "module": m, "entry": tgt, "expr": expr})
|
||
print(f" {i:2d} {m:4d} @{tgt:<7} {sem:<46} {val}")
|
||
|
||
if plain_cd is not None:
|
||
print(f"\n(真机 cd 里还有 {len(plain_cd) - len(seq)} 个槽不来自 require —— "
|
||
f"通常是字面量, 例如 cd[0] 的空串)")
|
||
if args.json:
|
||
Path(args.json).write_text(json.dumps({"manifestModule": mid, "entry": entry,
|
||
"slots": table}, ensure_ascii=False, indent=1),
|
||
encoding="utf-8")
|
||
print(f"槽位表 -> {args.json}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|