diff --git a/scripts/experiment_relay_verify.py b/scripts/experiment_relay_verify.py new file mode 100644 index 0000000..69ee44e --- /dev/null +++ b/scripts/experiment_relay_verify.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +""" +决定性实验 A —— 把真机页面**组好但没发出**的 verify 包, 分别交给 Python / 浏览器 代发。 + +要回答的问题 (har/1ms_dianxuan/PROGRESS.md §五 1(a)): + 12 到底是"**谁在发这个 HTTP**"造成的, 还是"**发出去的包**"造成的? + +做法: + 1. 在真机页面的 **顶层页 + 验证码 iframe** 里 hook 掉 XMLHttpRequest.send / fetch, + 让腾讯 widget 照常组包, 但请求**不真正发出** (拦在 `cap_union_new_verify`)。 + 2. 用 CDP 真实鼠标把验证码答完 -> widget 组包 -> 被拦下, 我们拿到**原封不动的** body。 + 3. 把这个 body 发给 `cap_union_new_verify`: + --send-by python : httpx 按浏览器同款头发出 (实验组) + --send-by browser : 同 iframe 里 fetch 发出 (对照组: 证明这个包本身有效且未被消耗) + 4. 判读: + * 两组都 0 -> "谁在发"无关, 12 只剩"包内容"(H-LAYOUT) -> 下一步去取槽位表 + * python 12, 浏览器 0 -> 存在"会话↔请求方"绑定, 需换实验 A' + * 浏览器也非 0 -> 拦截本身污染了会话, 实验无效 + +用法: + .venv/bin/python scripts/experiment_relay_verify.py --send-by python + .venv/bin/python scripts/experiment_relay_verify.py --send-by browser +""" +from __future__ import annotations + +import argparse +import base64 +import importlib.util +import json +import subprocess +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import httpx # noqa: E402 + + +def _bridge(): + spec = importlib.util.spec_from_file_location("bridge_1ms", ROOT / "scripts" / "bridge_1ms_captcha.py") + mod = importlib.util.module_from_spec(spec) + sys.argv = ["experiment"] + spec.loader.exec_module(mod) + return mod + + +TRAP_JS = r"""(() => { + if (window.__vtrapInstalled) return {already: true}; + window.__vtrapInstalled = true; + window.__vtrap = null; + + const isVerify = (u) => typeof u === 'string' && u.indexOf('cap_union_new_verify') >= 0; + + const origOpen = XMLHttpRequest.prototype.open; + const origSend = XMLHttpRequest.prototype.send; + const origSRH = XMLHttpRequest.prototype.setRequestHeader; + + XMLHttpRequest.prototype.open = function (m, u) { + this.__vUrl = u; this.__vMethod = m; this.__vHdrs = {}; + return origOpen.apply(this, arguments); + }; + XMLHttpRequest.prototype.setRequestHeader = function (k, v) { + try { if (isVerify(this.__vUrl)) this.__vHdrs[k] = v; } catch (e) {} + return origSRH.apply(this, arguments); + }; + XMLHttpRequest.prototype.send = function (body) { + if (isVerify(this.__vUrl)) { + window.__vtrap = { via: 'xhr', url: String(this.__vUrl), method: this.__vMethod || 'POST', + headers: this.__vHdrs || {}, body: String(body), at: Date.now() }; + return; // 故意不发出 + } + return origSend.apply(this, arguments); + }; + + const origFetch = window.fetch; + if (typeof origFetch === 'function') { + window.fetch = function (input, init) { + const u = (typeof input === 'string') ? input : (input && input.url); + if (isVerify(u)) { + const body = (init && init.body) || ''; + window.__vtrap = { via: 'fetch', url: String(u), method: (init && init.method) || 'GET', + headers: (init && init.headers) || {}, body: String(body), at: Date.now() }; + return Promise.reject(new Error('__vtrap_blocked__')); + } + return origFetch.apply(this, arguments); + }; + } + return {installed: true}; +})()""" + +READ_TRAP = "(() => ({trap: window.__vtrap || null, installed: !!window.__vtrapInstalled}))()" + + +def send_by_python(url: str, body: str, headers: dict, want_origin: bool) -> dict: + h = { + "Content-Type": headers.get("Content-Type", + "application/x-www-form-urlencoded; charset=UTF-8"), + "Accept": headers.get("Accept", "application/json, text/javascript, */*; q=0.01"), + "Referer": "https://turing.captcha.gtimg.com/", + "User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/155.0.0.0 Safari/537.36"), + "sec-ch-ua-platform": '"macOS"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua": '"Google Chrome";v="155", "Chromium";v="155", "Not(A:Brand";v="24"', + } + for k, v in (headers or {}).items(): + h[k] = v + if want_origin: + h["Origin"] = "https://turing.captcha.gtimg.com" + with httpx.Client(timeout=20.0, trust_env=False) as c: + r = c.post(url, content=body.encode("utf-8"), headers=h) + return {"status": r.status_code, "text": r.text, "sent_headers": h} + + +def main() -> int: + ap = argparse.ArgumentParser(description="把真机组好的 verify 包交给 Python/浏览器 代发") + ap.add_argument("--send-by", choices=["python", "browser"], default="python") + ap.add_argument("--rounds", type=int, default=8, help="最多刷几轮去等一个文字点选题") + ap.add_argument("--origin", action="store_true", help="Python 代发时附带 Origin 头") + args = ap.parse_args() + + b = _bridge() + + # 1) 唤起 widget + ens = b.cdp_eval(b.EXPR_ENSURE, b.PAGE)["result"] + print(f"[1] 唤起 widget: {ens}") + time.sleep(2.0) + + # 2) 装陷阱 (顶层页 + iframe; 谁发都拦得住) + for ctx in (b.PAGE, b.IFRAME): + try: + print(f"[2] 装陷阱 {ctx}: {b.cdp_eval(TRAP_JS, ctx)['result']}") + except Exception as exc: + print(f"[2] 装陷阱 {ctx} 失败: {exc}") + + # 3) 跑桥接去答题 (回调必定超时 —— 因为 verify 被我们拦下了, 这正是预期) + print("[3] 让真机把题答完 (verify 会被拦下, 所以回调超时属预期) ...") + p = subprocess.run([sys.executable, str(ROOT / "scripts" / "bridge_1ms_captcha.py"), + "--only", "text", "--rounds", str(args.rounds), "--solves", "1"], + cwd=str(ROOT), capture_output=True, text=True, timeout=600) + tail = [ln for ln in p.stdout.splitlines() if ln.strip()][-8:] + print(" " + "\n ".join(tail)) + + # 4) 取包 + trap = None + for ctx in (b.IFRAME, b.PAGE): + try: + r = b.cdp_eval(READ_TRAP, ctx)["result"] + except Exception as exc: + print(f"[4] 读陷阱 {ctx} 失败: {exc}") + continue + print(f"[4] {ctx}: installed={r.get('installed')} trap={'有' if r.get('trap') else '无'}") + if r.get("trap"): + trap = r["trap"] + break + if not trap: + print("❌ 没拦到 verify 包 (可能本轮没抽到可解的题, 或 iframe 被重建)") + return 1 + body = trap["body"] + print(f"[4] 拦到 verify 包: via={trap['via']} url={trap['url'][:80]} body={len(body)}B") + print(f" headers={trap.get('headers')}") + + # 5) 代发 + if args.send_by == "python": + print("[5] Python(httpx) 代发同一包 ...") + res = send_by_python(trap["url"], body, trap.get("headers") or {}, args.origin) + else: + print("[5] 浏览器(同 iframe) 代发同一包 ...") + js = ("(async()=>{const body=atob('%s');" + "const r=await fetch('%s',{method:'POST',headers:" + "{'Content-Type':'application/x-www-form-urlencoded; charset=UTF-8'," + "'Accept':'application/json, text/javascript, */*; q=0.01'},body});" + "return {status:r.status, text: await r.text()};})()" + % (base64.b64encode(body.encode()).decode(), trap["url"])) + tmp = Path("/tmp/1ms_bridge/relay_browser.js") + tmp.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(js, encoding="utf-8") + res = b.cdp_eval(js, b.IFRAME)["result"] + + print(f"[6] 代发结果: status={res.get('status')}") + print(f" body={str(res.get('text'))[:400]}") + try: + code = json.loads(res.get("text") or "{}").get("errorCode") + except Exception: + code = None + print(f"\n>>> errorCode = {code} (sender={args.send_by})") + print(">>> 判读: 0 => '谁在发'无关, 12 只剩包内容(H-LAYOUT); 12 => 还需排查请求方绑定") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tcaptcha/VM_SPEC.md b/tcaptcha/VM_SPEC.md index bb7817b..d8111b9 100644 --- a/tcaptcha/VM_SPEC.md +++ b/tcaptcha/VM_SPEC.md @@ -264,3 +264,69 @@ E.push(function g() { 2. 与 `extract_xtea_key`（Node 神谕）逐字对齐后，替换掉最后一个 Node 依赖； 3. 把 lift 的栈模型补全（`SWAP/SET_STACK_LEN/REPLACE_TOP_IMM` 的表达式还原），消除残留的 `r`。 + +--- + +## 9. 程序结构层：模块系统 / 入口 / collect 装配（2026-10-11 补，工具 `tools/tdc_program.py`） + +§5/§6 曾把缺口写成「不是 VM 的问题，而是模块系统 + 宿主调用序列」。本节把它补成确定规格。 + +### 9.1 顶层就是「模块注册表装配」 + +``` +@0000: SET_STACK_LEN 3; ALLOC_REG 2; JUMP @1518 +@1518: r3 = CREATE_CLOSURE(target=@7, params=[p0->r3]) // VM 外层包装 +... +@1538: PUSH_IMM [0]; REVERSE_TOP2; JUMP -> @2085 +@2085: CREATE_CLOSURE(target=@1544, params=[p0->r3,p1->r4,p2->r5]); SET_PROP // registry[0] = 工厂@1544 +@2096: PUSH_IMM [1]; REVERSE_TOP2; JUMP -> @2914 +@2914: CREATE_CLOSURE(target=@2102, ...); SET_PROP // registry[1] = 工厂@2102 +... // 共 62 个 +``` + +即源码级的 `registry[id] = function (module, exports, require) { ... }`（CommonJS）。 +**模块工厂的形参固定为 (r3=module, r4=exports, r5=require)**；模块体内 `r5(N)` 就是 `require(N)`。 + +> `tools/tdc_program.py --modules` 直接导出这张表。 + +### 9.2 代码布局的坑（把工具改对了） + +- 函数体是**不连续**的：父函数的初始化序列与子函数体交错排布，父函数用 `JUMP` 跳过子函数体。 + 所以「按 pc 大小猜归属」全错（实测把父函数续段误判给夹在中间的兄弟函数）。 + 正确做法：**边界感知走法** —— 从入口走控制流，遇到**另一个函数入口**就停下。 +- 反汇编器曾漏掉终结符：`RET_TRUE`/`THROW` 之后**没有 fall-through**，否则控制流会穿过函数结尾 + 线性走下去，把后续所有代码都算进当前函数。已修（按语义名定位，不写死 opcode 数值）。 + +### 9.3 collect 的装配链（`{"cd":[...],"sd":{...}}` 是怎么出来的） + +1. **清单模块**：某个模块返回一个数组，构造方式是接连的 + + ``` + DUP?; PUSH_IMM_ARRAY [6]; "length" ... ; LOAD_REG_IMM [5]; PUSH_IMM [N]; CALL_FUNC [1]; SET_PROP + ``` + + 语义即 `arr[arr.length] = require(N)`。**这串 require 的顺序就是 `cd` 的槽位顺序。** + （core_js 那份：37 个 require + 1 个字面量槽；线上构建实测是**另一串** id。） +2. **分发模块**：一个闭包把清单按 `.on` / `.reset` / `.get` 分进三个数组 + （`caps=[r5=r9,r6=r12,r7=r13,r8=r14]`，即 出=r12/r13/r14）。 +3. **装配模块**：`r3 = '{"cd":['`，然后 + + ``` + for (r7 = 0; r7 < r15.length; r7++) { r8 = r15[r7]; v = r8.get(); ... 追加 r3 ... } + r3 += '],'; ... r4["cd"] = r3 + ``` + + 即**按数组顺序**调每个采集器 `.get()` 并序列化 → `cd` 槽位序 = 清单数组顺序。 + +### 9.4 为什么这件事决定 errorCode=12 + +- 槽位序**按构建随机化**（5 份同日样本 5/5 互不相同，见 `har/1ms_dianxuan/PROGRESS.md` #31）。 +- 仓库里的 `models/success_collect_template.b64` 是**某一次**的顺序 + （其指纹与 `签到成功-1.har` 那一次逐槽一致，#32）。 +- 拿旧顺序发的 collect，会被服务端按**本次构建**的槽位表解释成「自相矛盾的环境」（例如 + `cd[1]` 该是 `[360,360]` 却是 `"MacIntel"`）→ 风控判 12。 +- 这正是「换 IP / 换遥测后端 / 换传输指纹 / 换 cookie / 长静默全都无效」的原因： + 它们**都不动槽位序**。 + +> 因此纯算路线必须**按本次构建**导出槽位表（`tools/tdc_layout.py`）， +> 并用**本次 prehandle 的 sid** 去取那份 tdc.js（真机 URL 还多带 `appid=`）。 diff --git a/tcaptcha/core_js/tdc_sample.disasm.txt b/tcaptcha/core_js/tdc_sample.disasm.txt index e0eb7bd..259f653 100644 --- a/tcaptcha/core_js/tdc_sample.disasm.txt +++ b/tcaptcha/core_js/tdc_sample.disasm.txt @@ -3696,7 +3696,6 @@ @06336: CALL_METHOD [1] @06338: POP_TRY @06339: RET_TRUE -@06340: JUMP target=@6564 @06343: ALLOC_REG [9] @06345: PUSH_TRY catchTarget=@6274 catchVarReg=r9 @06348: PUSH_EMPTY_STR @@ -7430,7 +7429,6 @@ @12694: PUSH_IMM [0] @12696: POP_TRY @12697: RET_TRUE -@12698: JUMP target=@12789 @12701: ALLOC_REG [5] @12703: PUSH_TRY catchTarget=@12690 catchVarReg=r5 @12706: PUSH_IMM_ARRAY [3] @@ -7620,7 +7618,6 @@ @13033: PUSH_STR_CHAR [112] @13035: POP_TRY @13036: RET_TRUE -@13037: JUMP target=@13357 @13040: ALLOC_REG [6] @13042: PUSH_TRY catchTarget=@13024 catchVarReg=r6 @13045: PUSH_IMM_ARRAY [3] @@ -8358,7 +8355,6 @@ @14327: PUSH_TRUE @14328: POP_TRY @14329: RET_TRUE -@14330: JUMP target=@14518 @14333: ALLOC_REG [5] @14335: PUSH_TRY catchTarget=@14323 catchVarReg=r5 @14338: PUSH_IMM_ARRAY [3] @@ -9904,7 +9900,6 @@ @17076: PUSH_UNDEFINED @17077: POP_TRY @17078: RET_TRUE -@17079: JUMP target=@17101 @17082: ALLOC_REG [11] @17084: PUSH_TRY catchTarget=@17072 catchVarReg=r11 @17087: PUSH_IMM_ARRAY [8] @@ -10715,7 +10710,6 @@ @18571: PUSH_UNDEFINED @18572: POP_TRY @18573: RET_TRUE -@18574: JUMP target=@18892 @18577: ALLOC_REG [6] @18579: PUSH_TRY catchTarget=@18567 catchVarReg=r6 @18582: PUSH_IMM_ARRAY [3] @@ -15799,7 +15793,6 @@ @27198: PUSH_FALSE @27199: POP_TRY @27200: RET_TRUE -@27201: JUMP target=@27362 @27204: ALLOC_REG [5] @27206: PUSH_TRY catchTarget=@27194 catchVarReg=r5 @27209: PUSH_IMM_ARRAY [3] @@ -16195,7 +16188,6 @@ @27855: POP @27856: POP_TRY @27857: RET_TRUE -@27858: JUMP target=@27900 @27861: ALLOC_REG [5] @27863: PUSH_TRY catchTarget=@27829 catchVarReg=r5 @27866: PUSH_IMM_ARRAY [3] @@ -16262,7 +16254,6 @@ @27972: PUSH_FALSE @27973: POP_TRY @27974: RET_TRUE -@27975: JUMP target=@28477 @27978: ALLOC_REG [9] @27980: PUSH_TRY catchTarget=@27968 catchVarReg=r9 @27983: PUSH_IMM_ARRAY [3] @@ -16604,7 +16595,6 @@ @28575: PUSH_FALSE @28576: POP_TRY @28577: RET_TRUE -@28578: JUMP target=@29390 @28581: ALLOC_REG [12] @28583: PUSH_TRY catchTarget=@28571 catchVarReg=r12 @28586: PUSH_IMM_ARRAY [3] @@ -17157,7 +17147,6 @@ @29482: PUSH_FALSE @29483: POP_TRY @29484: RET_TRUE -@29485: JUMP target=@29919 @29488: ALLOC_REG [9] @29490: PUSH_TRY catchTarget=@29478 catchVarReg=r9 @29493: PUSH_IMM_ARRAY [3] @@ -17502,7 +17491,6 @@ @30087: PUSH_FALSE @30088: POP_TRY @30089: RET_TRUE -@30090: JUMP target=@30935 @30093: ALLOC_REG [10] @30095: PUSH_TRY catchTarget=@30083 catchVarReg=r10 @30098: PUSH_IMM_ARRAY [3] @@ -18345,7 +18333,6 @@ @31579: POP_TRY @31580: POP_TRY @31581: RET_TRUE -@31582: JUMP target=@31627 @31585: ALLOC_REG [10] @31587: PUSH_TRY catchTarget=@31574 catchVarReg=r10 @31590: PUSH_IMM_ARRAY [4] @@ -19626,7 +19613,6 @@ @33841: PUSH_EMPTY_STR @33842: POP_TRY @33843: RET_TRUE -@33844: JUMP target=@34228 @33847: ALLOC_REG [8] @33849: PUSH_TRY catchTarget=@33837 catchVarReg=r8 @33852: PUSH_IMM_ARRAY [3] @@ -24144,7 +24130,6 @@ @42082: POP @42083: PUSH_TRUE @42084: RET_TRUE -@42085: JUMP target=@42025 @42088: PUSH_FALSE @42089: RET_TRUE @42090: CREATE_CLOSURE target=@41870 caps=[r5=r3,r6=r4] params=[p0->r3] @@ -25676,7 +25661,6 @@ @44769: PUSH_STR_CHAR [114] @44771: NEW_METHOD [1] @44773: THROW -@44774: POP_TRY @44775: IS_EXCEPTION @44776: PUSH_UNDEFINED @44777: RET_TRUE diff --git a/tools/tdc_disasm.py b/tools/tdc_disasm.py index d3ece02..2eb3bf9 100644 --- a/tools/tdc_disasm.py +++ b/tools/tdc_disasm.py @@ -254,6 +254,10 @@ def disassemble_cfg(z: List[int], entry: int = 0, follow_closures: bool = True, op_jump_if = next((o for o, n in names.items() if n == "JUMP_IF_TRUE"), OP_JUMP_IF_TRUE) op_push_try = next((o for o, n in names.items() if n == "PUSH_TRY"), OP_PUSH_TRY) op_closure = next((o for o, n in names.items() if n == "CREATE_CLOSURE"), OP_CREATE_CLOSURE) + # 终结符: 返回/抛出之后**没有 fall-through**。 + # 漏掉这一步会让控制流"穿过函数结尾"线性走下去, 把所有后续代码都算进当前子程序 + # (实测会把父函数的初始化序列误判成子函数自己的代码)。 + op_terminators = {o for o, n in names.items() if n in ("RET_TRUE", "THROW")} visited: Dict[int, dict] = {} work = [entry] @@ -282,6 +286,8 @@ def disassemble_cfg(z: List[int], entry: int = 0, follow_closures: bool = True, # 操作数 [0] 是闭包体入口: 该段代码在闭包被调用时才执行, 但它是**代码**, # 必须作为遍历根加入, 否则大半个程序都算不到。 work.append(z[pc + 1]) + if op in op_terminators: # RET_TRUE / THROW: 无 fall-through + continue work.append(nxt) return [visited[k] for k in sorted(visited)] diff --git a/tools/tdc_layout.py b/tools/tdc_layout.py new file mode 100644 index 0000000..8b124f4 --- /dev/null +++ b/tools/tdc_layout.py @@ -0,0 +1,158 @@ +# -*- 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 --plain /tmp/real_collect_plain.txt +""" +from __future__ import annotations + +import argparse +import json +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 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("-" * 108) + for i, m in enumerate(seq): + tgt = reg.get(m) + strs = transitive_strings(p, tgt) if tgt else [] + hints = env_hints(strs) + val = "" + if plain_cd is not None and i < len(plain_cd): + v = repr(plain_cd[i]) + val = v if len(v) <= 34 else v[:34] + "..." + table.append({"slot": i, "module": m, "entry": tgt, "hints": hints}) + print(f" {i:2d} {m:4d} @{tgt:<7} {str(hints)[:34]:<34} {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()) diff --git a/tools/tdc_lift.py b/tools/tdc_lift.py index 69efc79..d6f2f49 100644 --- a/tools/tdc_lift.py +++ b/tools/tdc_lift.py @@ -118,7 +118,7 @@ class Lifter: succ = [ops[0]] elif op == OP_JUMP_IF_TRUE: succ = [ops[0], nxt] - elif op in (19, 49): # RET_TRUE / THROW + elif ins["name"] in ("RET_TRUE", "THROW"): # 按语义名, 不写死 opcode 数值 succ = [] else: succ = [nxt] diff --git a/tools/tdc_program.py b/tools/tdc_program.py new file mode 100644 index 0000000..5481453 --- /dev/null +++ b/tools/tdc_program.py @@ -0,0 +1,323 @@ +# -*- coding: utf-8 -*- +""" +tdc_program.py —— TDC Chaos VM 「程序结构层」分析器。 + +它补的是 VM_SPEC.md §5/§6 自己点出的缺口: + opcode 表 / 闭包调用约定 早已吃透, 但**模块系统 + 宿主调用序列**一直没被静态说明: + 顶层只"装配模块", 真正干活的代码在闭包被宿主调用时才跑。 + +本工具把这些做出确定规格: + * `closures` —— 全部 CREATE_CLOSURE 站点: (pc, target, caps[(dst,src)], params[reg]) + * `entries` —— 子程序入口集合 = {0} ∪ {所有 closure target} + * `strings` —— 从字节码折叠出的字面串 (PUSH_EMPTY_STR + 连续 PUSH_STR_CHAR) + * `owner(pc)` —— 某 pc 属于哪个子程序 (启发式: 覆盖它的、且 ≤pc 的最大入口) + * `call_graph` —— 子程序 → 它在自己体内创建的闭包 target (静态调用图的"创建边") + * `xfuncs` —— 同一 target 被多个站点创建的"共享体" (VM 里的复用函数) + +注意 (踩过的坑): + `_decode_one` 对 CREATE_CLOSURE 的 `operands` **不含 target**(target 只写在 note 里), + 必须自己按 `target = z[pc+1]` 取, 否则会把 cap 寄存器号当成 target。 + +用法: + .venv/bin/python tools/tdc_program.py tcaptcha/core_js/tdc_sample.js --summary + .venv/bin/python tools/tdc_program.py --seq --max 30 + .venv/bin/python tools/tdc_program.py --find getData + .venv/bin/python tools/tdc_program.py --at 1725 +""" +from __future__ import annotations + +import argparse +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Dict, List, Optional, Sequence, Tuple + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from tcaptcha.tdc_opcodes import derive_opcode_table # noqa: E402 +from tools.tdc_disasm import disassemble_cfg, extract_bytecode # noqa: E402 + +MIN_STR_LEN = 3 + + +@dataclass +class ClosureSite: + pc: int + target: int + caps: List[Tuple[int, int]] = field(default_factory=list) # (新闭包寄存器, 父闭包寄存器) + params: List[int] = field(default_factory=list) # 形参落到的寄存器号 + + +@dataclass +class Program: + path: str + z: List[int] + names: Dict[int, str] + arity: Dict[int, int] + dynamic: set + insts: Dict[int, dict] + closures: List[ClosureSite] + strings: List[Tuple[int, str]] + entries: List[int] + _owner: Dict[int, int] = field(default_factory=dict) + + # -------------------------------------------------------------- 基本查询 + def name(self, pc: int) -> str: + i = self.insts.get(pc) + return i["name"] if i else "?" + + def owner(self, pc: int) -> int: + """pc 属于哪个子程序入口。 + + 注意: 代码布局是「函数体之间夹着父函数的初始化序列 + 内联的子函数体」, + 所以**不能**用"覆盖它的最大入口 ≤pc"来判 (会把父函数续段误判给夹在中间的兄弟函数)。 + 这里用**边界感知走法**: 从入口开始走控制流, 一旦走到**另一个函数入口**就停下 + (函数只能被调用进入, 不会被穿过)。 + """ + if not self._owner: + self._build_owners() + return self._owner.get(pc, 0) + + def _next_pc(self, pc: int, inst: dict) -> int: + """下一条 fall-through 指令的 pc。CREATE_CLOSURE 是动态长度, 不能用固定 arity。""" + if inst["name"] == "CREATE_CLOSURE": + capn, parn = self.z[pc + 2], self.z[pc + 3] + return pc + 1 + 3 + 2 * capn + parn + return pc + 1 + self.arity.get(inst["op"], 0) + + def function_code(self, entry: int) -> List[dict]: + """边界感知地取某个子程序**自己的**指令序列。""" + starts = set(self.entries) + out: Dict[int, dict] = {} + work = [entry] + while work: + pc = work.pop() + if pc in out or pc >= len(self.z) or pc < 0: + continue + if pc != entry and pc in starts: # 进入别的函数体 -> 不跨越 + continue + inst = self.insts.get(pc) + if inst is None: + continue + out[pc] = inst + name = inst["name"] + ops = inst["operands"] or [] + if name == "JUMP": + work.append(ops[0]) + continue + if name == "JUMP_IF_TRUE": + work.append(ops[0]) + work.append(self._next_pc(pc, inst)) + continue + if name == "PUSH_TRY" and self.arity.get(inst["op"], 0) == 2: + work.append(ops[0]) + if name == "CREATE_CLOSURE": + work.append(self.z[pc + 1]) + if name in ("RET_TRUE", "THROW"): + continue + work.append(self._next_pc(pc, inst)) + return [out[k] for k in sorted(out)] + + def _build_owners(self) -> None: + for e in self.entries: + for inst in self.function_code(e): + self._owner.setdefault(inst["pc"], e) + + def closures_of(self, entry: int) -> List[ClosureSite]: + return [c for c in self.closures if self.owner(c.pc) == entry] + + def call_graph(self) -> Dict[int, List[int]]: + g: Dict[int, List[int]] = {} + for c in self.closures: + g.setdefault(self.owner(c.pc), []).append(c.target) + return {k: sorted(set(v)) for k, v in g.items()} + + def shared_bodies(self) -> Dict[int, List[int]]: + """target -> 创建它的站点列表 (长度>1 说明这些站点共用同一个代码体)。""" + m: Dict[int, List[int]] = {} + for c in self.closures: + m.setdefault(c.target, []).append(c.pc) + return {k: v for k, v in m.items() if len(v) > 1} + + def module_registry(self, entry: int = 0) -> List[Tuple[int, int]]: + """识别顶层「模块注册表装配」模式, 返回 [(模块 id, 工厂闭包 entry), ...]。 + + 模式 (源码级 `registry[id] = function(module, exports, require){...}` 的编译产物): + DUP; PUSH_IMM [id]; REVERSE_TOP2; JUMP -> @F; @F: CREATE_CLOSURE ...; SET_PROP + + 这是整个程序的骨架: 顶层只把 228 个模块工厂按 id 填进一个数组, 入口再 drive 它。 + """ + code = self.function_code(entry) + pairs: List[Tuple[int, int]] = [] + for k, inst in enumerate(code): + if inst["name"] != "CREATE_CLOSURE": + continue + for j in range(k - 1, max(-1, k - 4), -1): + if code[j]["name"] != "JUMP": + continue + ops = code[j]["operands"] or [] + if not ops or ops[0] != inst["pc"]: + continue + for m in range(j - 1, max(-1, j - 6), -1): + if code[m]["name"] == "PUSH_IMM" and code[m]["operands"]: + # 记的是**工厂函数体入口** (CREATE_CLOSURE 的 target), 不是创建站点 + pairs.append((int(code[m]["operands"][0]), self.z[inst["pc"] + 1])) + break + break + seen, out = set(), [] + for mid, tgt in sorted(pairs): + if mid in seen: + continue + seen.add(mid) + out.append((mid, tgt)) + return out + + def find_strings(self, pattern: str) -> List[Tuple[int, str, int]]: + """返回 (pc, 串, 所属子程序入口)。""" + rx = re.compile(pattern, re.I) + return [(pc, s, self.owner(pc)) for pc, s in self.strings if rx.search(s)] + + +def _fold_strings(insts: Dict[int, dict]) -> List[Tuple[int, str]]: + """PUSH_EMPTY_STR 后跟一串 PUSH_STR_CHAR -> 还原字面串 (tdc_disasm 不做这步)。""" + out: List[Tuple[int, str]] = [] + cur_pc: Optional[int] = None + buf: List[str] = [] + for pc in sorted(insts): + i = insts[pc] + n = i["name"] + if n == "PUSH_EMPTY_STR": + cur_pc, buf = pc, [] + continue + if n == "PUSH_STR_CHAR" and cur_pc is not None: + ops = i["operands"] or [] + if ops: + buf.append(chr(ops[0] & 0xFF)) + continue + if cur_pc is not None: + if len(buf) >= MIN_STR_LEN: + out.append((cur_pc, "".join(buf))) + cur_pc, buf = None, [] + if cur_pc is not None and len(buf) >= MIN_STR_LEN: + out.append((cur_pc, "".join(buf))) + return out + + +def load(js_path: str) -> Program: + js = open(js_path, encoding="utf-8", errors="replace").read() + z = extract_bytecode(js) + names, arity, dynamic, _diag = derive_opcode_table(js) + insts = {i["pc"]: i for i in + disassemble_cfg(z, 0, True, names=names, arity=arity, dynamic=dynamic) + if i["op"] is not None} + + closures: List[ClosureSite] = [] + for pc, i in insts.items(): + if i["name"] != "CREATE_CLOSURE": + continue + target, cap, par = z[pc + 1], z[pc + 2], z[pc + 3] + base = pc + 4 + caps = [(z[base + 2 * k], z[base + 2 * k + 1]) for k in range(cap)] + params = [z[base + 2 * cap + k] for k in range(par)] + closures.append(ClosureSite(pc=pc, target=target, caps=caps, params=params)) + + strings = _fold_strings(insts) + entries = sorted({0} | {c.target for c in closures}) + return Program(path=js_path, z=z, names=names, arity=arity, dynamic=dynamic, + insts=insts, closures=sorted(closures, key=lambda c: c.pc), + strings=strings, entries=entries) + + +# --------------------------------------------------------------------------- CLI +def main() -> int: + ap = argparse.ArgumentParser(description="TDC Chaos VM 程序结构分析") + ap.add_argument("path", nargs="?", default=str(ROOT / "tcaptcha/core_js/tdc_sample.js")) + ap.add_argument("--summary", action="store_true", help="总体规模") + ap.add_argument("--seq", action="store_true", help="按子程序列出闭包站点") + ap.add_argument("--max", type=int, default=40, help="--seq/--find 的截断条数") + ap.add_argument("--find", default=None, help="按正则找字面串") + ap.add_argument("--at", type=int, default=None, help="看某个 pc 属于谁") + ap.add_argument("--shared", action="store_true", help="列出被多个站点共享的闭包体") + ap.add_argument("--func", type=int, default=None, help="打印某个子程序**自己的**指令序列") + ap.add_argument("--asm-out", default=None, help="--func 时把反汇编写到该文件") + ap.add_argument("--modules", action="store_true", help="导出模块注册表 (id -> 工厂 entry)") + args = ap.parse_args() + + p = load(args.path) + shown = False + + if args.summary or not any([args.seq, args.find is not None, args.at is not None, + args.shared, args.func is not None]): + shown = True + print(f"file : {p.path}") + print(f"bytecode : {len(p.z)} ints") + print(f"reachable instrs: {len(p.insts)}") + print(f"closure sites : {len(p.closures)}") + print(f"subprograms : {len(p.entries)} (0 为顶层)") + print(f"distinct bodies : {len({c.target for c in p.closures})}") + print(f"literal strings : {len(p.strings)} (>= {MIN_STR_LEN} 字符)") + + if args.seq: + shown = True + g = p.call_graph() + print(f"\n--- 子程序与它们创建的闭包 (前 {args.max} 个入口) ---") + for e in p.entries[: args.max]: + kids = g.get(e, []) + st = [s for pc, s in p.strings if p.owner(pc) == e] + head = ", ".join(f"{s[:22]!r}" for s in st[:4]) + print(f" @{e:05d} -> {kids}") + if head: + print(f" 串: {head}") + + if args.find is not None: + shown = True + hits = p.find_strings(args.find) + print(f"\n--- 字面串匹配 {args.find!r}: {len(hits)} 条 ---") + for pc, s, own in hits[: args.max]: + print(f" @{pc:05d} [子程序 @{own:05d}] {s[:90]!r}") + + if args.at is not None: + shown = True + pc = args.at + print(f"\n--- pc @{pc} ---") + print(f" 指令 : {p.insts.get(pc)}") + print(f" 所属 : @{p.owner(pc)}") + + if args.func is not None: + shown = True + code = p.function_code(args.func) + lines = [] + for x in code: + extra = (" " + " ".join(x["note"])) if x["note"] else ( + (" " + str(x["operands"])) if x["operands"] else "") + lines.append(f"@{x['pc']:05d}: {x['name']:<18}{extra}") + print(f"\n--- 子程序 @{args.func} 自身指令: {len(lines)} 条 ---") + print("\n".join(lines[: args.max])) + if args.asm_out: + Path(args.asm_out).write_text("\n".join(lines), encoding="utf-8") + print(f" -> {args.asm_out}") + + if args.modules: + shown = True + reg = p.module_registry(0) + print(f"\n--- 模块注册表: {len(reg)} 个模块 (registry[id] = 工厂闭包) ---") + for mid, tgt in reg[: args.max]: + print(f" registry[{mid:3d}] = 工厂 @{tgt}") + if len(reg) > args.max: + print(f" ... 共 {len(reg)} 个") + + if args.shared: + shown = True + sh = p.shared_bodies() + print(f"\n--- 共享闭包体: {len(sh)} 个 target 被多处创建 ---") + for t, sites in sorted(sh.items(), key=lambda kv: -len(kv[1]))[: args.max]: + print(f" target @{t:05d} x{len(sites)} sites={sites[:8]}{'...' if len(sites) > 8 else ''}") + + return 0 if shown else 1 + + +if __name__ == "__main__": + raise SystemExit(main())