feat(密钥): 纯 Python 静态提取 XTEA 密钥 —— 去掉最后一个 Node 依赖 (6/6 构建与神谕逐字一致)

做了什么
- 新增 tcaptcha/tdc_opcodes.py: 按构建从解释器源码推导 opcode 表。
  实测 opcode 编号**每份构建整体重排**(同一源码 pc0 的 SET_STACK_LEN 在 6 份构建里
  = 53/19/32/57/55/1, JUMP = 60/17/42/58/21/2); 旧固化表只对 core_js 有效 ——
  用它反汇编其它构建会在 pc0 撞上 RET_TRUE, 得到"空反汇编"的假象。
  做法: 定位主循环 X[Y[Z++]]() 的 handler 数组 -> JS 感知切分 -> handler 源码规范化
  (去空白 + 局部变量按首次出现重命名) 取指纹 -> 与内置 58 条 core_js 参考指纹匹配;
  数组下标即该构建的 opcode, argc 由语义名取 (与 handler 体内 x[y++] 次数实测一致)。
- 新增 tcaptcha/tdc_key.py: 纯 Python XTEA 密钥静态还原器 (零 Node / 零 pc 硬编码)。
  反汇编 -> 定位含 delta 的 XTEA 闭包 -> 用"两层捕获链"选出真密钥数组:
  字闭包捕获的 B 帧寄存器, 经"定义 B 的那个闭包"的 caps 映射到 G 帧后须 == XTEA 闭包
  捕获的寄存器; 借此排除脚本里并行计算的第二套诱饵字 -> 读 4 个 (源串, 偏移, 下标)
  -> 折入轮加盐。
- tools/tdc_disasm.py: disassemble_cfg/_decode_one 支持注入按构建推导的 opcode 表
  (names/arity/dynamic), 默认仍回落 core_js 表; 控制流 opcode 改为按语义名定位。
- tools/tdc_key_pure.py: CLI (-v 打印推导过程 / --opcodes 打印 opcode 表 / --json)。
- tcaptcha/crypto.py: extract_xtea_key 改为"纯 Python 主路径 + Node 插桩兜底"。
- 新增 tcaptcha/tests/test_tdc_key.py 并接入 selftest;
  新增入库 fixture tcaptcha/tests/fixtures/tdc_fresh_2026-10-11.js (现场新抓的 tdc.js)。

为什么
- AGENTS.md §2.4 分层交付 / §1.1: 去掉最后一个 Node 依赖, 提高并发与抗发版能力;
- §5.7 负结果也是资产: 显式撤回"opcode 数值跨构建固定"与"密钥只能靠 Node 插桩拿"两条旧结论。

怎么验证
- 6 份构建纯 Python 结果 == Node 插桩 extract_xtea_key, **逐字一致**:
  core_js / root / 1ms tdc.js / tdc_live / tdc_latest / 2026-10-11 现场新抓 tdc_fresh。
- opcode 表 6/6 构建 58/58 handler 精确匹配, 0 未识别。
- .venv/bin/python -m tcaptcha.selftest 全绿 (含既有逐字节对拍 test_xtea_key)。
- 把 node_bin 指向不存在的程序时 generate_telemetry 仍产出 collect (证明已零 Node)。
This commit is contained in:
yml2213
2026-10-11 22:49:23 +08:00
parent 9286f77661
commit 2e2bbc876a
11 changed files with 1058 additions and 30 deletions
+28 -9
View File
@@ -1,9 +1,10 @@
# 反 VM 密钥还原 & 纯算交付 进度文档
> 目标: **算法 100% 纯 Python** —— collect 的明文组装 / 轨迹 / 时间戳 / XTEA 加密 / 编码
> 全部在 Python 内完成; Node 只保留**一个**用途: 从当前 tdc.js 提取本次构建的 XTEA 密钥字。
> 全部在 Python 内完成; ~~Node 只保留一个用途: 从当前 tdc.js 提取本次构建的 XTEA 密钥字~~
> **2026-10-11 起 Node 连提取都不再需要**(纯 Python 静态还原,见已知 #19~#22;Node 仅作兜底)。
> 依据: AGENTS.md §2.3(拒绝补环境) / §2.4(分层交付) / §3.3(字节码 VM) / §4.3(运行时捕获) / §5.8(标定式交付)。
> 最后更新: 2026-10-11 (纯算收敛: 移除全部补环境后端)
> 最后更新: 2026-10-11 (纯 Python XTEA 密钥静态提取落地 + opcode 表按构建推导)
---
@@ -48,6 +49,10 @@
| 16 | **纯 Python 可完整复算密钥,与 `extract_xtea_key` 逐字一致** | `base=[1817655617,1381395306,1245924197,1767660346]` → 折盐 → `[1818510928,1381395306,1245924197,1768384072]` == Node 结果 | 确证 |
| 17 | 密钥链路的运行时位置(神谕实测) | 逐字首次出现 PC:7740/7526/7895/8852(均 reg9);数组组装 @8886(reg7);进 XTEA 闭包 @9030 `CALL_FUNC[2]`;入口 @9109 | 确证 |
| 18 | `extract_xtea_key`(Node)本身正确 | 其密钥把 probe 的真实 collect 解成 `{"cd":[0,10,"top",[360,360],"MacIntel",1,"undefined?rand…` | 确证 |
| 19 | **opcode 编号每份构建整体重排**(不是只有变量名/加盐在变) | 同一源码的 pc0 `SET_STACK_LEN` 在 6 份构建里 = 53/19/32/57/55/1;`JUMP` = 60/17/42/58/21/2;core 表去解其它构建会在 pc0 就 `RET_TRUE` 假象 | 确证 |
| 20 | **opcode 表可从该构建解释器源码推导**:主循环 `X[Y[Z++]]()` 的 `X` 就是 handler 数组,下标即 opcode;handler 规范化指纹可跨构建精确匹配 | 6/6 份构建 **58/58 handler 精确匹配**、0 未识别(`tcaptcha/tdc_opcodes.py`)| 确证 |
| 21 | **密钥字闭包只写"被 XTEA 闭包捕获的那个数组"**;脚本同时算了两套 4 字(真 + 诱饵),靠**两层捕获链**区分:字闭包捕获的 B 帧寄存器 `w_src`,经"定义 B 的那个闭包"的 caps 映射到 G 帧后须 == XTEA 闭包捕获的寄存器 | core/root/1ms×3/实时新抓 全部选对;诱饵组按同规则被排除 | 确证 |
| 22 | **纯 Python 静态提取 XTEA 密钥成功,零 Node** | `tcaptcha/tdc_key.py`:6/6 份构建与 Node 插桩 `extract_xtea_key` **逐字一致**;含 2026-10-11 现场新抓的 `tdc.js`(`har/1ms_dianxuan/live/tdc_fresh.js`,77967B,密钥 `[1397244740,1799507557,1399740484,1516463189]`)| 确证 |
---
@@ -76,6 +81,8 @@
| **"补环境"能解决 1ms 的 12** | 同上三条;环境不是自变量 |
| **captcha 域带腾讯云登录 cookie 能解决 12 (H3)** | 单变量 A/B:带/不带均 12(`scripts/experiment_h3_cookie_ab.py`,2026-10-11)|
| 轨迹 dt 太小是 12 的"直接原因" | 换成真实 ms 分布后仍 12(E6) |
| **opcode 数值是跨构建固定的 VM 常量** | 6 份构建实测 pc0 的 `SET_STACK_LEN`/`JUMP` 数值全不同(见已知 #19);`tools/tdc_disasm.py` 里固化的表**只对 core_js 有效**,已改为"可注入按构建推导的表" |
| **"密钥字必须靠 Node 插桩才能拿到"** | `tcaptcha/tdc_key.py` 纯 Python 零 Node 复算,6/6 构建与 Node 逐字一致(已知 #22)→ Node 降级为兜底 |
---
@@ -83,23 +90,31 @@
```
TencentCaptchaClient(telemetry_mode="python_keyed") # 默认
├─ extract_xtea_key(tdc.js) Node 插桩 ADD opcode, 仅提取 4 个密钥字 (不可用时退回 DEFAULT_KEY)
├─ extract_xtea_key(tdc.js) **纯 Python 静态还原** (tcaptcha/tdc_key.py, 零 Node)
│ ├─ tdc_opcodes.py 按构建从解释器源码推导 opcode 表 (编号每份都重排)
│ ├─ tools/tdc_disasm.py 可达性 CFG 反汇编 (支持注入按构建的表)
│ ├─ 定位 XTEA 闭包 (delta 常数) -> 两层捕获链选出真密钥数组
│ └─ 读出 4 个字 (源串+偏移+下标) + 折入轮加盐
│ (Node 插桩版本保留为**兜底**: 纯 Python 失败时自动回落)
├─ generate_collect_success() 模板回填: 轨迹 / 当前时间戳 / 随机盐 → XTEA → Base64 (单层 URL 编码)
└─ extract_eks(tdc.js) 通用正则取最长 Base64 候选 (不再写死变量名)
```
* 纯 Python 提取器: `tcaptcha/tdc_key.py` (算法) + `tcaptcha/tdc_opcodes.py` (opcode 表推导)
* CLI: `tools/tdc_key_pure.py [tdc.js] [-v] [--opcodes]`
* 模板: `tcaptcha/models/success_collect_template.b64`(真实成功样本明文)
* 加密: `tcaptcha/crypto.py` 中**唯一**的教科书 XTEA
* 已从代码移除: `apply_real_environment`、4 段拼接旧结构、node/browser/probe 后端、`human_dt_ms`
* Node 现存用途: 仅 `tdc_worker.js` 兜底提取密钥(主路径已不需要)
---
## 五、下一步(可被下一轮直接执行)
0. **静态提取 XTEA 密钥(去掉最后一个 Node 依赖)**:密钥字是 ASCII 打包串、且不是字面量,
需对**正确反汇编**(`tools/tdc_disasm.py`)做一个小的**逐闭包符号执行/数据流**:
从 `CREATE_CLOSURE caps=[r7=r8]` 反查父闭包 r8 的赋值表达式并常量折叠。
寄存器是「每闭包一份、经 `PUSH_IMM_ARRAY <idx>` 引用间接访问」,所以需要真正的数据流而非模式匹配。
0. ~~**静态提取 XTEA 密钥(去掉最后一个 Node 依赖)**~~ —— **已完成**(已知 #19~#22):
`tcaptcha/tdc_key.py` 纯 Python 零 Node,6/6 构建与 Node 逐字一致。
残留增强项(非阻塞):① 若服务端改了 handler 语义,`tdc_opcodes.derive_opcode_table`
会抛"指纹未识别",此时应更新参考指纹而非改定位逻辑;② 抗"多套诱饵数组 >2 层捕获"的
兜底(当前两层已覆盖实测全部 6 份)。
1. **拿下 1ms 的 12**:做账本 §7.5 的**单变量实验** ——
(a) 给 captcha 域带上成功样本里的 `uin/skey/qcmainCSRFToken/intl` cookie(已做,H3 已排除);
(b) 登录态真页面触发 vs 未登录触发。二选一即可闭合。
@@ -113,7 +128,11 @@ TencentCaptchaClient(telemetry_mode="python_keyed") # 默认
```bash
.venv/bin/python -m tcaptcha.tests.test_crypto_core # XTEA 往返 + encrypt_string 一致性
.venv/bin/python -m tcaptcha.tests.test_xtea_key # XTEA + 密钥提取 (逐字节对拍)
.venv/bin/python -m tcaptcha.tests.test_xtea_key # XTEA + 密钥提取 (逐字节对拍, Node 参与)
.venv/bin/python -m tcaptcha.tests.test_tdc_key # **纯 Python** 密钥提取 (6 份构建, 零 Node)
.venv/bin/python -m tcaptcha.tests.test_glyph_solver # 文字点选求解 (8 张真实样本)
.venv/bin/python -m tcaptcha.selftest # 全部
.venv/bin/python tools/tdc_key_pure.py -v har/1ms_dianxuan/live/tdc_fresh.js # 单份构建 + 推导过程
.venv/bin/python tools/tdc_key_pure.py --opcodes tcaptcha/tdc_sample.js # 打印按构建推导的 opcode 表
```
+8 -1
View File
@@ -9,7 +9,9 @@
| 工具 | 路径 | 核心功能 | 解决什么问题 |
| :--- | :--- | :--- | :--- |
| **Webpack 解包索引器** | [`tools/unpack_webpack.js`](file:///Users/yml/reverse/hello_js_reverse_skill/tools/unpack_webpack.js) | 将 230KB 单文件 bundle 拆解为 122 个独立模块,建立加签/风控关键词索引 | 圈定加签核心,剔除 80% 开源库与 UI 噪音 (L1 解构) |
| **Chaos VM 字节码反汇编器 (纯 Python)** | [`tools/tdc_disasm.py`](file:///Users/yml/reverse/hello_js_reverse_skill/tools/tdc_disasm.py) | 纯 Python 提取 45,636 字节码 + 可达性 CFG 反汇编(26,168 条可达指令,**0 处非法 opcode**);opcode 表由 VM 源码 AST 推导 | 彻底透明化 TDC 字节码调度器 (L2/L3 解构) |
| **Chaos VM 字节码反汇编器 (纯 Python)** | [`tools/tdc_disasm.py`](file:///Users/yml/reverse/hello_js_reverse_skill/tools/tdc_disasm.py) | 纯 Python 提取 45,636 字节码 + 可达性 CFG 反汇编(26,168 条可达指令,**0 处非法 opcode**);opcode 表可**按构建注入**(见下) | 彻底透明化 TDC 字节码调度器 (L2/L3 解构) |
| **按构建推导 opcode 表** | [`tcaptcha/tdc_opcodes.py`](file:///Users/yml/reverse/hello_js_reverse_skill/tcaptcha/tdc_opcodes.py) | 从解释器主循环 `X[Y[Z++]]()` 取出 handler 数组 → 解析 → 规范化指纹 → 跨构建匹配出**该构建的 opcode 表** | **opcode 编号每份构建整体重排**,写死即换脚本全错(6/6 构建 58/58 handler 匹配) |
| **纯 Python XTEA 密钥静态提取** | [`tcaptcha/tdc_key.py`](file:///Users/yml/reverse/hello_js_reverse_skill/tcaptcha/tdc_key.py) + [`tools/tdc_key_pure.py`](file:///Users/yml/reverse/hello_js_reverse_skill/tools/tdc_key_pure.py) | 反汇编 → 定位 XTEA 闭包(delta) → 两层捕获链选出真密钥数组 → 读 4 个字 + 折盐 | **零 Node** 拿到该构建 XTEA 密钥;6/6 构建与 Node 插桩逐字一致 |
| **深度 Proxy 语义探针** | [`tools/tdc_probe.js`](file:///Users/yml/reverse/hello_js_reverse_skill/tools/tdc_probe.js) | 零 Headless/CDP 依赖,通过深度 Proxy 沙箱执行 `tdc.js`,拦截并记录全部 95 项环境读取 | 毫秒级生成与当前会话时间戳和密钥 100% 自洽的动态 `collect` 与 `eks` (L4 数据层) |
---
@@ -114,4 +116,9 @@ plain = xtea_decrypt_bytes(raw, key).rstrip(b"\\x00")
print("成功解密探针生成的遥测载荷,长度:", len(plain))
print("明文前 150 字节:", plain[:150].decode("latin1"))
'
# 5. 纯 Python XTEA 密钥提取 (零 Node) + 跨构建回归
.venv/bin/python tools/tdc_key_pure.py -v har/1ms_dianxuan/live/tdc_fresh.js
.venv/bin/python -m tcaptcha.tests.test_tdc_key # 6 份构建 (含实时新抓) 全部断言
.venv/bin/python tools/tdc_key_pure.py --opcodes tcaptcha/tdc_sample.js # 打印该构建的 opcode 表
```
+29 -3
View File
@@ -51,10 +51,36 @@ function __TENCENT_CHAOS_VM(H, z, c, E, I, D, U, C) {
## 3. 完整 opcode 表(70 槽 / 58 handler / 12 空洞)
`argc` = handler 体内 `z[H++]` 次数(= 固定操作数字节数);`57` 是动态长度(见 §4)。
**空洞 opcode**:`7, 16, 21, 24, 27, 40, 43, 44, 50, 58, 63, 67` —— 它们**永远不会**出现在可执行字节码里;一旦在反汇编里看到空洞 opcode,说明**解码失步**(旧线性反汇编器就是这么坏的)。
`argc` = handler 体内 `z[H++]` 次数(= 固定操作数字节数);动态长度的那个是 `CREATE_CLOSURE`(见 §4)。
| opcode | argc | handler 源码(VM 内 G 数组第 i 项) |
> ⚠️ **本节表格是"core_js 那一份构建"的编号,不能当跨构建常量用(重要修正,见 §3.1)。**
> 下面的 `opcode` 列只对 `tcaptcha/core_js/tdc_sample.js` 成立。
> **空洞 opcode**:`7, 16, 21, 24, 27, 40, 43, 44, 50, 58, 63, 67` —— 它们**永远不会**出现在可执行字节码里;
> 一旦在反汇编里看到空洞 opcode,说明**解码失步**(旧线性反汇编器就是这么坏的)。
### 3.1 opcode 编号**每份构建都会整体重排**(实测,5+1 份构建)
> **推翻早期结论**:早期把上面的 opcode 数值当成 VM 常量("`RET_TRUE`=19、`SET_STACK_LEN`=53…")。
> 实测是错的:**同一份源码编译出的字节码,pc0 处的 `SET_STACK_LEN` 在 6 份构建里分别是
> 53 / 19 / 32 / 57 / 55 / 1**,`JUMP` 分别是 60 / 17 / 42 / 58 / 21 / 2。
> 也就是说 **opcode → 语义 的映射表每份脚本都重新洗牌**;写死数值 = 换一份脚本全错
> (这也是为什么早先用 core 表去反汇编其它构建会得到"`RET_TRUE` 在 pc0 ⇒ 反汇编为空"的假象)。
**正确做法**:不在字节码上死磕,而是**从该构建自己的解释器源码里推导 opcode 表** ——
见 `tcaptcha/tdc_opcodes.py`:
1. 定位解释器主循环 `X[Y[Z++]]()`,取出**操作码处理器数组变量** `X`;
2. 解析该数组字面量(JS 感知的括号/字符串/正则切分)得到 70 个槽位(含空洞);
3. 把每个 handler 源码**规范化**(去空白 + 局部变量按首次出现重命名 `v1,v2,…`,
保留 JS 关键字/内建名与属性名)得到与命名无关的"指纹";
4. 用内置的 58 条参考指纹(取自 core_js,即本表)反查每个槽位的语义名 ——
**数组下标即该构建的 opcode**;`argc` 直接按语义名取(与 handler 体内 `x[y++]` 次数实测一致)。
> 交叉验证:6/6 份构建全部 **58/58 handler 精确匹配**、0 未识别。
> 因此 `tools/tdc_disasm.py` / `tools/tdc_lift.py` 现在都支持**注入按构建推导的 opcode 表**
> (`disassemble_cfg(z, names=…, arity=…, dynamic=…)`);不传才回落到本表(core_js)。
| opcode | argc | handler 源码(core_js 的 VM 内 G 数组第 i 项) |
| --- | --- | --- |
| 0 | 1 | `function () { E[E.length - 1] += String.fromCharCode(z[H++]); }` |
| 1 | 0 | `function () { return !!w; }` |
+17 -1
View File
@@ -167,7 +167,12 @@ def extract_xtea_key(tdc_js_code: str, node_bin: str = "node",
"""
从服务端下发的 tdc.js 里提取本次构建的 XTEA 密钥 (4 个 32 位字)。
流程 (Node 只负责"跑出密钥", 算法全在 Python):
**主路径 = 纯 Python 静态还原** (`tcaptcha/tdc_key.py`):
反汇编字节码 -> 定位 XTEA 闭包 -> 回溯它捕获的密钥数组 -> 读出 4 个密钥字
闭包(源串 + 偏移 + 存取下标) -> 折入轮加盐。
零 Node、零 JS 运行时、零 pc 硬编码 (opcode 表按构建推导, 见 tdc_opcodes.py)。
**兜底路径 = Node 插桩** (密钥构造算法若被改动时仍可用):
1. Node 在 VM 上下文里把 ADD opcode 换成带日志版本;
XTEA 每轮的 (sum + K[j]) 是一次 ADD, 两个操作数必有一个等于 r*DELTA,
于是另一个操作数就是密钥字 —— 按频次排序得到无歧义候选。
@@ -183,6 +188,17 @@ def extract_xtea_key(tdc_js_code: str, node_bin: str = "node",
if cached:
return list(cached)
# ---- 主路径: 纯 Python 静态还原 (零 Node) ----
try:
from .tdc_key import extract_xtea_key as _pure_extract
key = [int(k) & M32 for k in _pure_extract(tdc_js_code)]
if len(key) == 4:
if use_cache:
_KEY_CACHE[tdc_js_code] = key
return list(key)
except Exception as exc: # noqa: BLE001
logger.warning("纯 Python 静态提取失败, 退回 Node 插桩: %s", exc)
words, raw = _capture_key_words_cached(tdc_js_code, node_bin=node_bin)
candidates = [int(k) & M32 for k, *_ in words[:top_n]]
# 验证用「内容特征」而非固定明文 (AGENTS.md §5.3 / 附录 A#6):
+1
View File
@@ -15,6 +15,7 @@ SUITES = [
"tcaptcha.tests.test_crypto_core",
"tcaptcha.tests.test_glyph_solver",
"tcaptcha.tests.test_xtea_key",
"tcaptcha.tests.test_tdc_key",
]
+441
View File
@@ -0,0 +1,441 @@
# -*- coding: utf-8 -*-
"""
腾讯天御 TDC 混沌 VM (Chaos VM) —— XTEA 密钥的**纯 Python** 静态还原器。
为什么需要它
------------
此前要拿到某一版 tdc.js 的 XTEA 密钥, 只能把 tdc.js 丢进 Node 插桩跑一遍
(ADDU opcode 吐密钥字)。密钥**不是字面量**, 而是 VM 运行时由字节码算出来的,
所以必须"读指令流"而不是"搜字符串"。
本模块完全离线(零 Node、零 JS 运行时), 基于**正确反汇编**(tools/tdc_disasm.py)做
**模式 + 数据流**定位, 复刻密钥构造算法。全部结论都有运行时神谕(tools/tdc_probe.js)
逐条对照过, 见 har/1ms_dianxuan/PROGRESS.md。
算法 (已证: 纯 Python 结果 == Node extract_xtea_key 逐字一致)
------------------------------------------------------------
1. **4 个密钥字**: 每个字由一个专门的闭包算出:
word_i = Σ_{k=0..3} ( (ord(src_i[k]) + off_i) & 0xFF ) << (8*k)
源串 `src_i` 是紧跟在该闭包 `CREATE_CLOSURE` 之后的 4 个 `PUSH_STR_CHAR`
字面量; 偏移 `off_i` 是打包点的立即数; 存入下标 `i` 由闭包尾部的
`PUSH_IMM_ARRAY / PUSH_IMM [i] / ... / SET_PROP` 给出。
2. **哪些闭包是真密钥**: 密钥数组会被 **XTEA 闭包捕获**。
XTEA 轮里出现 `delta = 0x9E3779B9`; 含 delta 的那个闭包体所属的
`CREATE_CLOSURE` 即 XTEA 闭包, 其 `caps=[r7=r<K>]` 的 `<K>` 就是密钥数组所在的
父帧寄存器。**只有 caps 右侧等于该寄存器的密钥字闭包**才是真密钥
(同一份脚本往往还并行算了第二个"诱饵"数组, 用不同寄存器, 必须排除)。
3. **轮加盐**: XTEA 轮内 `key[(sum>>>11)&3]` 查表之后会**条件性**加上一个立即数
(salt): `GET_PROP / PUSH_IMM <salt> / ADD / ADD / XOR`。加给哪个下标由守卫条件
`idx == ((sum>>>11)&3)` / `idx == (sum & 3)` 里的 `idx` 常量决定。
有效密钥 = base 字按盐的下标位置相加。
所有定位都是**模式 + 立即数**, 不写死任何 pc (每份构建都会重排)。
"""
from __future__ import annotations
import os
import re
import sys
from typing import Dict, List, Optional, Tuple
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _ROOT not in sys.path: # 允许在任意 cwd 下 import tools.*
sys.path.insert(0, _ROOT)
from tools.tdc_disasm import extract_bytecode, disassemble_cfg # noqa: E402
from tcaptcha.tdc_opcodes import derive_opcode_table # noqa: E402
M32 = 0xFFFFFFFF
XTEA_DELTA = 0x9E3779B9 # 2654435769, XTEA 黄金比常数
KEY_LEN = 4
# 走"下标表达式"时要跳过的算子 (它们构成 `(sum & 3)` / `(sum>>>11)&3`)
_SKIP_OPS = {
"BITWISE_AND", "SHR_UNSIGNED", "LOAD_REG_IMM", "MAKE_REG_PROP_REF",
"GET_PROP", "SWAP", "STORE_REG", "ADD", "DUP", "LOAD_REG_INDIRECT",
"PUSH_IMM_ARRAY", "POP",
}
# 这些算子紧跟在"掩码 PUSH_IMM 3"之前, 说明那个 3 是掩码而不是下标
_MASK_NEIGHBOURS = {"BITWISE_AND", "LOAD_REG_IMM", "SHR_UNSIGNED"}
class _Dis:
"""对反汇编结果做少量索引封装, 便于按 pc 前后游走。"""
def __init__(self, js_text: str) -> None:
self.z = extract_bytecode(js_text)
# opcode 编号每份构建都重排 -> 必须先按构建把语义表推导出来
names, arity, dynamic, diag = derive_opcode_table(js_text)
self.names, self.arity, self.dynamic, self.diag = names, arity, dynamic, diag
self.insts = disassemble_cfg(self.z, names=names, arity=arity, dynamic=dynamic)
self.by_pc: Dict[int, dict] = {i["pc"]: i for i in self.insts}
self.pcs: List[int] = sorted(self.by_pc)
self._idx: Dict[int, int] = {pc: n for n, pc in enumerate(self.pcs)}
def op(self, pc: Optional[int]) -> Optional[str]:
i = self.by_pc.get(pc) if pc is not None else None
return i["name"] if i else None
def ops(self, pc: Optional[int]) -> List[int]:
i = self.by_pc.get(pc) if pc is not None else None
return list(i["operands"]) if i else []
def note(self, pc: int) -> str:
return " ".join(self.by_pc[pc].get("note") or [])
def nxt(self, pc: Optional[int]) -> Optional[int]:
n = self._idx.get(pc) if pc is not None else None
return self.pcs[n + 1] if n is not None and n + 1 < len(self.pcs) else None
def prev(self, pc: Optional[int]) -> Optional[int]:
n = self._idx.get(pc) if pc is not None else None
return self.pcs[n - 1] if n is not None and n > 0 else None
def in_range(self, lo: int, hi: int) -> List[int]:
import bisect
a = bisect.bisect_left(self.pcs, lo)
b = bisect.bisect_right(self.pcs, hi)
return self.pcs[a:b]
# --------------------------------------------------------------------------- #
# 1. 闭包清单 / 密钥字打包点
# --------------------------------------------------------------------------- #
def _closures(d: _Dis) -> List[dict]:
"""所有 CREATE_CLOSURE -> {pc, target, caps:{dst:src}, str_after}。"""
out: List[dict] = []
for i in d.insts:
if i["name"] != "CREATE_CLOSURE":
continue
pc = i["pc"]
note = " ".join(i["note"] or [])
m = re.search(r"target=@(\d+)", note)
if not m:
continue
caps: Dict[int, int] = {}
caps_part = re.search(r"caps=\[([^\]]*)\]", note)
if caps_part:
for dst, src in re.findall(r"r(\d+)=r(\d+)", caps_part.group(1)):
caps[int(dst)] = int(src)
out.append({"pc": pc, "target": int(m.group(1)), "caps": caps,
"str_after": _string_after(d, pc)})
return out
def _string_after(d: _Dis, pc: int) -> Optional[str]:
"""CREATE_CLOSURE 紧随其后的 `PUSH_EMPTY_STR + PUSH_STR_CHAR*` 字面量。"""
p = d.nxt(pc)
if p is None or d.op(p) != "PUSH_EMPTY_STR":
return None
chars: List[str] = []
q = d.nxt(p)
while q is not None and d.op(q) == "PUSH_STR_CHAR":
chars.append(chr(d.ops(q)[0]))
q = d.nxt(q)
return "".join(chars)
def _pack_sites(d: _Dis, lo: int, hi: int) -> List[dict]:
"""
在 [lo,hi] 里找"密钥字打包点":
... CALL_METHOD[1] / PUSH_IMM c / SUB (off = -c)
或 ... CALL_METHOD[1] / PUSH_IMM 0 / PUSH_IMM c / SUB / SUB (off = +c)
/ PUSH_IMM 8 / LOAD_REG_IMM / MUL / SHL / BITWISE_OR
返回 [{or_pc, off}]。
"""
res: List[dict] = []
for pc in d.in_range(lo, hi):
if d.op(pc) != "PUSH_IMM" or d.ops(pc) != [8]:
continue
a, b, c, e = d.nxt(pc), None, None, None
b = d.nxt(a)
c = d.nxt(b)
e = d.nxt(c)
if not (d.op(a) == "LOAD_REG_IMM" and d.op(b) == "MUL"
and d.op(c) == "SHL" and d.op(e) == "BITWISE_OR"):
continue
off = _decode_offset(d, pc)
if off is not None:
res.append({"or_pc": e, "imm8_pc": pc, "off": off})
return res
def _decode_offset(d: _Dis, imm8_pc: int) -> Optional[int]:
"""解出紧跟 CALL_METHOD 之后的偏移立即数 (含两种编码)。"""
p1 = d.prev(imm8_pc) # 最后一个 SUB
if d.op(p1) != "SUB":
return None
p2 = d.prev(p1)
if d.op(p2) == "PUSH_IMM": # X - c => off = -c
return -d.ops(p2)[0]
if d.op(p2) == "SUB": # X - (0 - c) => off = +c
p3 = d.prev(p2)
p4 = d.prev(p3)
if d.op(p3) == "PUSH_IMM" and d.op(p4) == "PUSH_IMM" and d.ops(p4) == [0]:
return d.ops(p3)[0]
return None
def _store_index(d: _Dis, lo: int, hi: int) -> Optional[int]:
"""闭包尾部的 `PUSH_IMM_ARRAY / PUSH_IMM [i] / MAKE_REG_PROP_REF / LOAD_REG_IMM / SET_PROP`。"""
hits: List[int] = []
for pc in d.in_range(lo, hi):
if d.op(pc) != "PUSH_IMM_ARRAY":
continue
a, b, c, e = d.nxt(pc), None, None, None
b = d.nxt(a)
c = d.nxt(b)
e = d.nxt(c)
if (d.op(a) == "PUSH_IMM" and d.op(b) == "MAKE_REG_PROP_REF"
and d.op(c) == "LOAD_REG_IMM" and d.op(e) == "SET_PROP"):
hits.append(d.ops(a)[0])
return hits[-1] if hits else None
# --------------------------------------------------------------------------- #
# 2. XTEA 闭包定位 + 轮加盐
# --------------------------------------------------------------------------- #
def _find_xtea_closure(d: _Dis, closures: List[dict]) -> Optional[dict]:
"""含 XTEA 常数(delta)的那个闭包体所属的 CREATE_CLOSURE。"""
delta_pcs = [pc for pc in d.pcs
if d.op(pc) == "PUSH_IMM" and d.ops(pc)[0] in (XTEA_DELTA, XTEA_DELTA * 32)]
# 取"最内层"(区间最小)的闭包体 —— 外层 init 函数的体也会横跨 delta,
# 但它不捕获密钥数组。
best: Optional[dict] = None
best_end: Optional[int] = None
for cl in closures:
end = _body_end(d, cl["pc"])
if end is None or end < cl["target"]:
continue
if any(cl["target"] <= pc <= end for pc in delta_pcs):
if best is None or best_end is None or (end - cl["target"]) < (best_end - best["target"]):
best, best_end = cl, end
return best
def _largest_reg(regs: Dict[int, int]) -> Optional[int]:
return max(regs.values()) if regs else None
def _caps_single(caps: Dict[int, int]) -> Optional[Tuple[int, int]]:
"""字/XTEA 闭包只捕获一个数组: 返回 (dst, src)。"""
if len(caps) != 1:
return None
(dst, src), = caps.items()
return dst, src
def _defining_closure(d: _Dis, closures: List[dict], pc: int) -> Optional[dict]:
"""含给定 pc 的**最内层**闭包(即创建 pc 处那条 CREATE_CLOSURE 的那个函数)。"""
best: Optional[dict] = None
best_span: Optional[int] = None
for c in closures:
e = _body_end(d, c["pc"])
if e is None or e < c["target"]:
continue
if c["target"] <= pc <= e:
span = e - c["target"]
if best_span is None or span < best_span:
best, best_span = c, span
return best
def _body_end(d: _Dis, cl_pc: int) -> Optional[int]:
"""
闭包体结束 pc (该体的 RET_TRUE)。约定: 闭包被创建后要么紧跟其体尾的 RET_TRUE,
要么中间夹一条 `PUSH_IMM_ARRAY`(闭包被塞进数组时)。
"""
e = d.prev(cl_pc)
if e is not None and d.op(e) == "PUSH_IMM_ARRAY":
e = d.prev(e)
return e if (e is not None and d.op(e) == "RET_TRUE") else None
def _guard_index(d: _Dis, eq_pc: int) -> Optional[int]:
"""
守卫 `idx == ((sum & 3) | ((sum>>>11)&3))` 里的常量 idx。
从 EQ_LOOSE 反向游走, 跳过下标表达式自身的算子, 第一个落在 0..3 的立即数即 idx。
掩码 3 靠"它的后继是 BITWISE_AND / LOAD_REG_IMM / SHR_UNSIGNED"识别并跳过。
"""
pc = d.prev(eq_pc)
last: Optional[str] = None
for _ in range(24):
if pc is None:
return None
nm, ops = d.op(pc), d.ops(pc)
if nm in ("PUSH_IMM", "REPLACE_TOP_IMM") and ops:
v = ops[0]
if 0 <= v <= 3:
if not (nm == "PUSH_IMM" and v == 3 and last in _MASK_NEIGHBOURS):
return v
elif nm not in _SKIP_OPS:
# 出现意料之外的算子: 继续向前找, 但记录上下文
pass
last = nm
pc = d.prev(pc)
return None
def _salts(d: _Dis, lo: int, hi: int) -> List[Tuple[int, int]]:
"""
XTEA 轮内的条件加盐: `GET_PROP / PUSH_IMM <salt> / ADD / ADD / XOR`。
返回 [(idx, salt)] (按 (idx,salt) 去重)。
"""
salt_pcs: List[int] = []
for pc in d.in_range(lo, hi):
if d.op(pc) != "PUSH_IMM":
continue
if d.op(d.prev(pc)) != "GET_PROP" or d.op(d.nxt(pc)) != "ADD":
continue
salt_pcs.append(pc)
if not salt_pcs:
return []
# 收集所有 (JUMP_IF_TRUE, EQ_LOOSE) 守卫; block(T) 到 T 之后第一条无条件 JUMP
guards: List[Tuple[int, int, Optional[int]]] = [] # (target, jump_pc, idx)
for pc in d.in_range(lo, hi):
if d.op(pc) != "JUMP_IF_TRUE":
continue
tgt_ops = d.ops(pc)
if not tgt_ops:
continue
tgt = tgt_ops[0]
eq = d.prev(pc)
if d.op(eq) != "EQ_LOOSE":
continue
idx = _guard_index(d, eq)
if idx is not None:
guards.append((tgt, pc, idx))
def block_end(tgt: int) -> int:
p = tgt
while p is not None and p <= hi:
if d.op(p) == "JUMP":
return p
p = d.nxt(p)
return hi
out: List[Tuple[int, int]] = []
for spc in salt_pcs:
salt = d.ops(spc)[0]
best: Optional[Tuple[int, int]] = None # (target, idx)
for tgt, _j, idx in guards:
if tgt <= spc < block_end(tgt):
if best is None or tgt > best[0]:
best = (tgt, idx)
if best is not None:
out.append((best[1], salt))
# 去重保序
seen = set()
uniq: List[Tuple[int, int]] = []
for x in out:
if x not in seen:
seen.add(x)
uniq.append(x)
return uniq
# --------------------------------------------------------------------------- #
# 3. 主入口
# --------------------------------------------------------------------------- #
def word_from_chars(src: str, off: int) -> int:
"""word = Σ ((ord(c) + off) & 0xFF) << (8*k) (小端打包)"""
w = 0
for k, ch in enumerate(src):
w |= ((ord(ch) + off) & 0xFF) << (8 * k)
return w & M32
def extract_xtea_key_pure(js_text: str, verbose: bool = False):
"""
纯 Python 从 tdc.js 里还原本次构建的 XTEA 有效密钥 (4 个 32 位字)。
返回 dict:
base : 4 个 base 字 (未折盐)
salts : [(idx, salt), ...]
key : 有效密钥
words : [(idx, word, src, off), ...] (便于诊断)
"""
d = _Dis(js_text)
closures = _closures(d)
xtea = _find_xtea_closure(d, closures)
if xtea is None:
raise ValueError("未找到 XTEA 闭包 (字节码里没有 delta 常数?)")
xs = _caps_single(xtea["caps"])
if xs is None:
raise ValueError("XTEA 闭包没有唯一捕获的数组, 无法确定密钥数组")
xtea_src = xs[1] # XTEA 所在帧里, 密钥数组所在的寄存器
# 密钥数组是"被 XTEA 闭包捕获"的那一个。字闭包与 XTEA 闭包往往不在同一帧:
# 父帧 G: r<R> = 密钥数组
# ├─ 字闭包 W: caps=[r7 = 本帧 r<Wsrc>] (Wsrc ∈ {R, 诱饵寄存器})
# └─ XTEA 闭包 : caps=[r7 = 本帧 r<xtea_src>]
# 所以要把 W 的捕获穿一层: B = 定义 W 的那个闭包(B 帧 -> G 帧), 检查
# B.caps[Wsrc] == xtea_src -> 该字闭包写的就是真密钥数组。
words: Dict[int, Tuple[int, str, int]] = {}
picked: List[str] = []
for cl in closures:
if cl["str_after"] is None:
continue
cs = _caps_single(cl["caps"])
if cs is None:
continue
w_src = cs[1] # 该字闭包捕获的、定义帧里的寄存器
definer = _defining_closure(d, closures, cl["pc"])
if definer is not None and definer["caps"]:
real = definer["caps"].get(w_src) == xtea_src
else: # 同帧: 直接比
real = (w_src == xtea_src)
if not real:
continue
end = _body_end(d, cl["pc"])
if end is None or end < cl["target"]:
continue
packs = _pack_sites(d, cl["target"], end)
if not packs:
continue
idx = _store_index(d, cl["target"], end)
if idx is None or not (0 <= idx < KEY_LEN):
continue
src, off = cl["str_after"], packs[-1]["off"]
words[idx] = (word_from_chars(src, off), src, off)
picked.append("%s(idx=%d,off=%+d)" % (src, idx, off))
if len(words) != KEY_LEN:
raise ValueError(
"密钥字不完整: 只定位到 %s (期望 0..3); XTEA 密钥帧寄存器 r%s"
% (sorted(words), xtea_src))
base = [words[i][0] for i in range(KEY_LEN)]
salts = _salts(d, xtea["target"], d.prev(xtea["pc"]) or xtea["target"])
key = list(base)
for idx, salt in salts:
key[idx] = (key[idx] + salt) & M32
report = {
"base": base,
"salts": salts,
"key": key,
"words": [(i, words[i][0], words[i][1], words[i][2]) for i in range(KEY_LEN)],
"xtea_closure_pc": xtea["pc"],
"key_reg": xtea_src,
}
if verbose:
print("[tdc_key_pure] XTEA closure @%d key array = r%s" % (xtea["pc"], xtea_src))
for i, w, s, o in report["words"]:
print(" word[%d] <- %-6r off=%+d = %d" % (i, s, o, w))
for idx, salt in salts:
print(" salt +%d -> key[%d]" % (salt, idx))
print(" base = %s" % base)
print(" key = %s" % key)
return report
def extract_xtea_key(js_text: str) -> List[int]:
"""便捷入口: 只取有效密钥 (与 crypto.extract_xtea_key 同口径)。"""
return list(extract_xtea_key_pure(js_text)["key"])
+320
View File
@@ -0,0 +1,320 @@
# -*- coding: utf-8 -*-
"""
腾讯天御 TDC 混沌 VM (Chaos VM) —— **按构建推导操作码表**。
为什么需要它
------------
`tools/tdc_disasm.py` 里固化的 opcode 表取自 **core_js 那一份构建**。
实测(5 份构建交叉验证): **opcode 编号每份构建都会整体重排** ——
同一份源码编译出的字节码, pc0 处的 `SET_STACK_LEN` 在 core_js 是 53,
在 root 样本是 19, 在 1ms 实时样本分别是 18 / 10 / 60。
所以写死 opcode 表 = 换一份脚本就全错。
本模块的做法:
1. 从 JS 源码里定位解释器主循环 `X[Y[Z++]]()` -> 取出**操作码处理器数组变量** `X`;
2. 解析该数组字面量(JS 感知的括号/字符串/正则切分)得到 70 个槽位(含空洞);
3. 把每个 handler 的源码**规范化**(去空白 + 局部变量按首次出现重命名为 v1,v2,...,
保留 JS 关键字/内建名与属性名)得到与变量命名无关的"指纹";
4. 用内置的 58 条参考指纹(取自 core_js = `tools/tdc_disasm.OP_NAMES` 那套语义)
反查每个槽位的语义名 —— 数组下标即该构建的 opcode。
5. 操作数个数按语义名从参考表取 (与"handler 体内 `x[y++]` 读取次数"实测一致)。
这样, 只要服务端还是这台 VM(handler 语义相同), 无论 opcode 怎么重排都能对上。
```
经过: 5/5 本构建全部 58/58 handler 精确匹配 (har/1ms_dianxuan/PROGRESS.md)。
```
"""
from __future__ import annotations
import os
import re
import sys
from typing import Dict, List, Optional, Tuple
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
from tools.tdc_disasm import ARITY as _CORE_ARITY, DYNAMIC_ARITY as _CORE_DYNAMIC, OP_NAMES as _CORE_NAMES # noqa: E402
# --------------------------------------------------------------------------- #
# 参考指纹: core_js 构建的 58 个 handler 规范化源码 -> 语义名 (与 OP_NAMES 一致)
# --------------------------------------------------------------------------- #
_REF_BLOB = r"""PUSH_STR_CHAR function(){v1[v1.length-1]+=String.fromCharCode(v2[v3++])}
IS_EXCEPTION function(){return!!v1}
DUP function(){v1.push(v1[v1.length-1])}
XOR function(){v1[v1.length-2]=v1[v1.length-2]^v1.pop()}
EQ_STRICT function(){v1[v1.length-2]=v1[v1.length-2]===v1.pop()}
MOD function(){v1[v1.length-2]=v1[v1.length-2]%v1.pop()}
CALL_FUNC function(){varv1=v2[v3++],v4=v1?v5.slice(-v1):[];v5.length-=v1,v5.push(v5.pop().apply(v6,v4))}
GET_GLOBAL_PROP function(){v1[v1.length-1]=v2[v1[v1.length-1]]}
MAKE_PROP_REF function(){varv1=v2.pop(),v3=v2.pop();v2.push([v3[0][v3[1]],v1])}
TYPEOF function(){v1.push(typeofv1.pop())}
DELETE function(){varv1=v2.pop();v2.push(deletev1[0][v1[1]])}
LOGICAL_NOT function(){v1.push(!v1.pop())}
EQ_LOOSE function(){v1[v1.length-2]=v1[v1.length-2]==v1.pop()}
SET_PROP function(){varv1=v2[v2.length-2];v1[0][v1[1]]=v2[v2.length-1]}
GREATER_THAN function(){v1[v1.length-2]=v1[v1.length-2]>v1.pop()}
SWAP function(){varv1=v2[v3++],v4=v5[v5.length-2-v1];v5[v5.length-2-v1]=v5.pop(),v5.push(v4)}
IN function(){v1[v1.length-2]=v1[v1.length-2]inv1.pop()}
RET_TRUE function(){return!0}
GREATER_EQ function(){v1[v1.length-2]=v1[v1.length-2]>=v1.pop()}
PUSH_IMM function(){v1.push(v2[v3++])}
MUL function(){v1[v1.length-2]=v1[v1.length-2]*v1.pop()}
LOAD_REG_INDIRECT function(){v1.push(v1[v1.pop()[0]][0])}
ITER_NEXT function(){v1[v1.length-1].length?v1.push(v1[v1.length-1].shift(),!0):v1.push(undefined,!1)}
BIND_THIS function(){v1.push([v2,v1.pop()])}
PUSH_NULL function(){v1.push(null)}
BITWISE_AND function(){v1[v1.length-2]=v1[v1.length-2]&v1.pop()}
PUSH_IMM_ARRAY function(){v1.push([v2[v3++]])}
MAKE_REG_PROP_REF function(){varv1=v2.pop();v2.push([v2[v2.pop()][0],v1])}
SUB function(){v1[v1.length-2]=v1[v1.length-2]-v1.pop()}
CALL_METHOD function(){varv1=v2[v3++],v4=v1?v5.slice(-v1):[],v1=(v5.length-=v1,v5.pop());v5.push(v1[0][v1[1]].apply(v1[0],v4))}
SHR function(){v1[v1.length-2]=v1[v1.length-2]>>v1.pop()}
SHL function(){v1[v1.length-2]=v1[v1.length-2]<<v1.pop()}
JUMP_IF_TRUE function(){varv1=v2[v3++];v4[v4.length-1]&&(v3=v1)}
FOR_IN_KEYS function(){varv1,v2=[];for(v1inv3.pop())v2.push(v1);v3.push(v2)}
ADD function(){v1[v1.length-2]=v1[v1.length-2]+v1.pop()}
NEW_METHOD function(){varv1=v2[v3++],v4=v1?v5.slice(-v1):[],v1=(v5.length-=v1,v4.unshift(null),v5.pop());v5.push(v6(v1[0][v1[1]],v4))}
PUSH_UNDEFINED function(){v1.push(undefined)}
PUSH_FALSE function(){v1.push(!1)}
STORE_REG function(){v1[v1[v1.length-2][0]][0]=v1[v1.length-1]}
ALLOC_REG function(){varv1=v2[v3++];v4[v1]=v4[v1]===undefined?[]:v4[v1]}
POP_TRY function(){v1.pop()}
THROW function(){throwv1[v1.length-1]}
POP function(){v1.pop()}
DIV function(){v1[v1.length-2]=v1[v1.length-2]/v1.pop()}
SET_STACK_LEN function(){v1.length=v2[v3++]}
REVERSE_TOP2 function(){v1.push([v1.pop(),v1.pop()].reverse())}
PUSH_EMPTY_STR function(){v1.push("")}
PUSH_TRY function(){v1.push([v2[v3++],v4.length,v2[v3++]])}
CREATE_CLOSURE function(){for(varv1=v2[v3++],v4=[],v5=v2[v3++],v6=v2[v3++],v7=[],v8=0;v8<v5;v8++)v4[v2[v3++]]=v9[v2[v3++]];for(v8=0;v8<v6;v8++)v7[v8]=v2[v3++];v9.push(functionv10(){varv5=v4.slice(0);v5[0]=[this],v5[1]=[arguments],v5[2]=[v10];for(varv6=0;v6<v7.length&&v6<arguments.length;v6++)0<v7[v6]&&(v5[v7[v6]]=[arguments[v6]]);returnv11(v1,v2,v12,v5,v13,v14,v15,v16)})}
GET_PROP function(){varv1=v2.pop();v2.push(v1[0][v1[1]])}
JUMP function(){v1=v2[v1++]}
NEW_FUNC function(){varv1=v2[v3++],v4=v1?v5.slice(-v1):[];v5.length-=v1,v4.unshift(null),v5.push(v6(v5.pop(),v4))}
BITWISE_OR function(){v1[v1.length-2]=v1[v1.length-2]|v1.pop()}
REPLACE_TOP_IMM function(){v1[v1.length-1]=v2[v3++]}
SHR_UNSIGNED function(){v1[v1.length-2]=v1[v1.length-2]>>>v1.pop()}
PUSH_TRUE function(){v1.push(!0)}
LOAD_REG_IMM function(){v1.push(v1[v2[v3++]][0])}
CLEAR_EXCEPTION function(){v1=null}"""
def _reference() -> Dict[str, str]:
out: Dict[str, str] = {}
for line in _REF_BLOB.splitlines():
if not line.strip():
continue
name, _, fp = line.partition("\t")
out[fp] = name
return out
_REF_FP2NAME = _reference()
# 语义名 -> 操作数个数 / 是否动态长度
_NAME_ARITY: Dict[str, int] = {}
for _op, _nm in _CORE_NAMES.items():
_NAME_ARITY[_nm] = _CORE_ARITY.get(_op, 0)
_NAME_DYNAMIC = {_CORE_NAMES[o] for o in _CORE_DYNAMIC}
# --------------------------------------------------------------------------- #
# JS 感知的工具
# --------------------------------------------------------------------------- #
_JS_KEEP = set("""
function var return new if else for while do typeof instanceof in of delete void this
null true false undefined break continue switch case default throw try catch finally
Object String Number Boolean Array Function RegExp Date Math JSON Error TypeError
Symbol parseInt parseFloat isNaN encodeURIComponent decodeURIComponent
Uint8Array Uint8ClampedArray Infinity NaN arguments
""".split())
_ID_RE = re.compile(r"[A-Za-z_$][\w$]*")
_PREV_REGEX_OK = set("(,=[{;:!&|?+-*%~^<>")
_OPERAND_READ_RE = re.compile(r"[A-Za-z_$][\w$]*\s*\[\s*[A-Za-z_$][\w$]*\s*\+\+\s*\]")
def _skip_js_string(js: str, i: int) -> int:
q = js[i]
i += 1
while i < len(js):
c = js[i]
if c == "\\":
i += 2
continue
if c == q:
return i + 1
i += 1
return i
def _skip_js_regex(js: str, i: int) -> int:
i += 1
while i < len(js):
c = js[i]
if c == "\\":
i += 2
continue
if c == "[":
i += 1
while i < len(js) and js[i] != "]":
if js[i] == "\\":
i += 1
i += 1
i += 1
continue
if c == "/":
i += 1
while i < len(js) and js[i].isalpha():
i += 1
return i
i += 1
return i
def split_js_array(js: str, lb: int) -> Tuple[List[str], int]:
"""把从 `lb`(指向 '[')起的数组字面量切成顶层元素字符串。"""
assert js[lb] == "["
i, depth, start, out = lb + 1, 0, lb + 1, []
while i < len(js):
c = js[i]
if c in "\"'`":
i = _skip_js_string(js, i)
continue
if c == "/":
j = i - 1
while j >= 0 and js[j].isspace():
j -= 1
if j < 0 or js[j] in _PREV_REGEX_OK:
i = _skip_js_regex(js, i)
continue
i += 1
continue
if c in "([{":
depth += 1
i += 1
continue
if c in ")]}":
if c == "]":
if depth == 0:
out.append(js[start:i])
return out, i
depth -= 1
else:
depth -= 1
i += 1
continue
if c == "," and depth == 0:
out.append(js[start:i])
start = i + 1
i += 1
continue
i += 1
raise ValueError("数组字面量未闭合")
def handler_array(js: str) -> List[str]:
"""
定位解释器主循环 `ARR[BC[PC++]]()`, 返回 `ARR` 的字面量元素(每个 = 一个 handler 源码)。
含空洞的槽位返回空串。
"""
m = re.search(r"([A-Za-z_$][\w$]*)\s*\[\s*[A-Za-z_$][\w$]*\s*\[\s*[A-Za-z_$][\w$]*\s*\+\+\s*\]\s*\]\s*\(\s*\)", js)
if not m:
raise ValueError("未找到 VM 解释器主循环 (X[Y[Z++]]())")
arr = m.group(1)
best: Optional[List[str]] = None
for mm in re.finditer(re.escape(arr) + r"\s*=\s*\[", js):
lb = js.index("[", mm.start())
try:
els, _ = split_js_array(js, lb)
except Exception:
continue
ne = sum(1 for e in els if e.strip())
if best is None or ne > sum(1 for e in best if e.strip()):
best = els
if best is None:
raise ValueError("未找到操作码处理器数组字面量 %s=[...]" % arr)
return best
def canonicalize(src: str) -> str:
"""把 handler 源码规范化成与变量命名无关的指纹。"""
out: List[str] = []
mp: Dict[str, str] = {}
n = 0
i = 0
prev = ""
while i < len(src):
c = src[i]
if c in "\"'`":
j = _skip_js_string(src, i)
out.append(src[i:j])
i, prev = j, "'"
continue
m = _ID_RE.match(src, i)
if m:
tok = m.group(0)
if prev == "." or tok in _JS_KEEP:
out.append(tok)
else:
if tok not in mp:
n += 1
mp[tok] = "v%d" % n
out.append(mp[tok])
i, prev = m.end(), "i"
continue
if not c.isspace():
out.append(c)
prev = c
i += 1
return "".join(out)
# --------------------------------------------------------------------------- #
# 对外: 按构建推导操作码表
# --------------------------------------------------------------------------- #
def derive_opcode_table(js: str) -> Tuple[Dict[int, str], Dict[int, int], set, Dict[str, str]]:
"""
返回 (names, arity, dynamic, diag):
names : {opcode: 语义名}
arity : {opcode: 固定操作数个数} (动态长度的不在其中)
dynamic : {opcode} 需要读操作数才能定长
diag : 诊断信息
"""
els = handler_array(js)
names: Dict[int, str] = {}
arity: Dict[int, int] = {}
dynamic: set = set()
unknown: List[int] = []
slots: List[Tuple[int, str]] = []
for op, src in enumerate(els):
if not src.strip():
continue
fp = canonicalize(src)
nm = _REF_FP2NAME.get(fp)
if nm is None:
unknown.append(op)
continue
names[op] = nm
if nm in _NAME_DYNAMIC:
dynamic.add(op)
else:
arity[op] = _NAME_ARITY.get(nm, 0)
slots.append((op, nm))
diag = {
"handler_count": len(names),
"unknown_slots": unknown,
"table": {op: nm for op, nm in slots},
}
if unknown:
raise ValueError("有 handler 指纹未能识别(服务端改了 VM 语义?): %s" % unknown[:10])
if len(names) < 40:
raise ValueError("推导出的 opcode 表过小(%d), 疑似解析失败" % len(names))
return names, arity, dynamic, diag
if __name__ == "__main__":
_js = open(sys.argv[1] if len(sys.argv) > 1 else "tcaptcha/core_js/tdc_sample.js",
encoding="utf-8", errors="replace").read()
_names, _arity, _dyn, _diag = derive_opcode_table(_js)
print("handlers: %d dynamic: %s" % (_diag["handler_count"], sorted(_dyn)))
for _op in sorted(_names):
print(" %3d %-18s arity=%s" % (_op, _names[_op], "dyn" if _op in _dyn else _arity.get(_op, 0)))
File diff suppressed because one or more lines are too long
+108
View File
@@ -0,0 +1,108 @@
# -*- coding: utf-8 -*-
"""
纯 Python XTEA 密钥提取 回归测试 (AGENTS.md §5.9)
固化的性质:
1. **跨构建通用**: opcode 编号每份构建整体重排, 但仍能按构建推导出正确语义表;
2. **纯 Python == Node 神谕**: 6 份构建(含 1 份新抓的实时 tdc.js)的期望密钥
全部由 Node `extract_xtea_key` 逐字对照确认过, 此处固化为断言 (不再依赖 Node 跑测试);
3. **定位不写死 pc**: 断言覆盖 core_js / root / 1ms(旧+两份实时) / 实时新抓 共 6 份
不同构建, 任一份 opcode 重排或寄存器分配变化都会立刻暴露。
运行:
.venv/bin/python -m tcaptcha.tests.test_tdc_key
"""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402
from tcaptcha.tdc_opcodes import derive_opcode_table # noqa: E402
# (相对路径, 期望有效密钥, 是否必须存在)
# 期望值来自 Node 插桩 extract_xtea_key, 逐字一致。
# 入库样本 (core_js / root / 实时新抓 fixture) 为必需门禁; har/** 被 .gitignore 忽略,
# 本地有这些样本时会一并回归, 缺少时自动跳过。
FIXTURES = [
("tcaptcha/core_js/tdc_sample.js", [1818510928, 1381395306, 1245924197, 1768384072], True),
("tcaptcha/tdc_sample.js", [1448563793, 1850559083, 1733513813, 1364613714], True),
# 2026-10-11 现场从 turing.captcha.qcloud.com 下发的 tdc.js (入库固化, 防"仅本地可验")
("tcaptcha/tests/fixtures/tdc_fresh_2026-10-11.js",
[1397244740, 1799507557, 1399740484, 1516463189], True),
("har/1ms_dianxuan/tdc.js", [1448106085, 1766670948, 1819173482, 1298358636], False),
("har/1ms_dianxuan/live/tdc_live.js", [1314475846, 1500397933, 1751410275, 1716217423], False),
("har/1ms_dianxuan/live/tdc_latest.js", [1800883800, 1850828643, 1197042522, 1800823125], False),
]
_fails: list = []
def check(name: str, cond: bool, detail: str = "") -> None:
print(f" [{'PASS' if cond else 'FAIL'}] {name}" + (f" {detail}" if detail else ""))
if not cond:
_fails.append(name)
def main() -> int:
print("1) 按构建推导 opcode 表 (58 个 handler, 无未识别)")
tables = {}
skipped = []
for rel, _expect, required in FIXTURES:
p = ROOT / rel
if not p.exists():
if required:
check(f"fixture {rel}", False, "缺失")
else:
skipped.append(rel)
continue
names, arity, dynamic, diag = derive_opcode_table(p.read_text(encoding="utf-8", errors="replace"))
tables[rel] = names
check(f"opcode 表 {Path(rel).name}",
diag["handler_count"] == 58 and not diag["unknown_slots"],
f"handlers={diag['handler_count']} unknown={diag['unknown_slots']}")
if skipped:
print(f" (跳过本地未提供的样本 {len(skipped)} 份: {[Path(s).name for s in skipped]})")
print("\n2) opcode 编号确实每份构建都不同 (证明没有依赖写死的数值)")
jump_ops = {rel: next(o for o, n in tabs.items() if n == "JUMP") for rel, tabs in tables.items()}
if len(jump_ops) >= 2:
check("JUMP 的 opcode 跨构建不统一", len(set(jump_ops.values())) >= 2, str(jump_ops))
print("\n3) 纯 Python 提取 == Node 神谕 (逐字)")
for rel, expect, required in FIXTURES:
p = ROOT / rel
if not p.exists():
continue
js = p.read_text(encoding="utf-8", errors="replace")
try:
rep = extract_xtea_key_pure(js)
check(Path(rel).name, rep["key"] == expect, f"{rep['key']}")
except Exception as exc: # noqa: BLE001
check(Path(rel).name, False, f"{type(exc).__name__}: {exc}")
print("\n4) 密钥构造自洽: base + 折盐 == key")
for rel, expect, required in FIXTURES:
p = ROOT / rel
if not p.exists():
continue
rep = extract_xtea_key_pure(p.read_text(encoding="utf-8", errors="replace"))
folded = list(rep["base"])
for idx, salt in rep["salts"]:
folded[idx] = (folded[idx] + salt) & 0xFFFFFFFF
check(f"折盐 {Path(rel).name}", folded == rep["key"] == expect)
print()
if _fails:
print(f"❌ test_tdc_key 失败 {len(_fails)} 项: {_fails}")
return 1
print("✅ test_tdc_key 全部通过")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+49 -16
View File
@@ -19,6 +19,11 @@
opcode 表来自对 VM 源码 AST 的推导(`G = [ ... ]` 中第 i 个 handler 即 opcode i,
操作数个数 = handler 体内 `z[H++]` 的出现次数),已固化在下方。
> ⚠️ **固化的表只对 `tcaptcha/core_js/tdc_sample.js` 有效。** 实测 **opcode 编号每份构建都整体重排**
> (同一源码 pc0 的 `SET_STACK_LEN` 在 6 份构建里是 53/19/32/57/55/1)。
> 处理其它构建请先用 `tcaptcha/tdc_opcodes.py: derive_opcode_table(js)` 推导该构建的表,再
> `disassemble_cfg(z, names=…, arity=…, dynamic=…)` 注入。详见 VM_SPEC.md §3.1。
用法:
.venv/bin/python tools/tdc_disasm.py [tdc.js] # 默认 tcaptcha/core_js/tdc_sample.js
"""
@@ -178,21 +183,32 @@ def extract_bytecode(js_text: str) -> List[int]:
# --------------------------------------------------------------------------- #
# 2. 可达性驱动的 CFG 反汇编 (只在可执行地址解码)
# --------------------------------------------------------------------------- #
def _decode_one(z: List[int], pc: int) -> Tuple[Optional[int], List[int], Optional[List[str]], int]:
def _decode_one(z: List[int], pc: int, names: Optional[Dict[int, str]] = None,
arity: Optional[Dict[int, int]] = None,
dynamic: Optional[set] = None) -> Tuple[Optional[int], List[int], Optional[List[str]], int]:
"""解码 pc 处一条指令 -> (opcode, operands, note, next_fallthrough_pc)。
返回值 (None,...) 表示该地址无法构成合法指令。
names/arity/dynamic 允许注入**按构建推导**的操作码表 (见 tcaptcha/tdc_opcodes.py);
不传则用本模块固化的 core_js 表。
"""
if names is None:
names = OP_NAMES
if arity is None:
arity = ARITY
if dynamic is None:
dynamic = DYNAMIC_ARITY
if pc >= len(z):
return None, [], None, pc
op = z[pc]
if op > 69 or op not in OP_NAMES: # 空洞 opcode / 越界 -> 非法
if op > 69 or op not in names: # 空洞 opcode / 越界 -> 非法
return None, [], None, pc + 1
pos = pc + 1
operands: List[int] = []
note: Optional[List[str]] = None
if op in DYNAMIC_ARITY:
if op in dynamic:
# 57: target, capCount, paramCount, [reg, reg]*capCount, [reg]*paramCount
if pos + 3 > len(z):
return None, [], None, pos
@@ -207,47 +223,64 @@ def _decode_one(z: List[int], pc: int) -> Tuple[Optional[int], List[int], Option
params = [f"p{i}->r{operands[2*cap+i]}" for i in range(param)]
note = [f"target=@{target}", f"caps=[{','.join(caps)}]", f"params=[{','.join(params)}]"]
else:
n = ARITY.get(op, 0)
n = arity.get(op, 0)
if pos + n > len(z):
return None, [], None, pos
operands = z[pos:pos + n]
pos += n
if op in (OP_JUMP, OP_JUMP_IF_TRUE) and operands:
if names[op] in ("JUMP", "JUMP_IF_TRUE") and operands:
note = [f"target=@{operands[0]}"]
elif op == OP_PUSH_TRY and len(operands) == 2:
elif names[op] == "PUSH_TRY" and len(operands) == 2:
note = [f"catchTarget=@{operands[0]}", f"catchVarReg=r{operands[1]}"]
return op, operands, note, pos
def disassemble_cfg(z: List[int], entry: int = 0, follow_closures: bool = True) -> List[dict]:
"""从 entry 出发按可达性反汇编; 返回按 pc 排序的指令列表。"""
def disassemble_cfg(z: List[int], entry: int = 0, follow_closures: bool = True,
names: Optional[Dict[int, str]] = None,
arity: Optional[Dict[int, int]] = None,
dynamic: Optional[set] = None) -> List[dict]:
"""从 entry 出发按可达性反汇编; 返回按 pc 排序的指令列表。
names/arity/dynamic 可注入按构建推导的操作码表 (默认 = 本模块固化的 core_js 表)。
"""
if names is None:
names = OP_NAMES
if arity is None:
arity = ARITY
if dynamic is None:
dynamic = DYNAMIC_ARITY
# 控制流相关 opcode 按"语义名"定位, 不写死数值 (opcode 每份构建都重排)
op_jump = next((o for o, n in names.items() if n == "JUMP"), OP_JUMP)
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)
visited: Dict[int, dict] = {}
work = [entry]
while work:
pc = work.pop()
if pc in visited or pc >= len(z) or pc < 0:
continue
op, operands, note, nxt = _decode_one(z, pc)
op, operands, note, nxt = _decode_one(z, pc, names, arity, dynamic)
if op is None:
# 非法地址 (落在数据区): 记录但不继续
visited[pc] = {"pc": pc, "op": None, "name": "DATA/INVALID", "operands": [], "note": None}
continue
visited[pc] = {"pc": pc, "op": op, "name": OP_NAMES[op], "operands": operands, "note": note}
if op == OP_JUMP:
visited[pc] = {"pc": pc, "op": op, "name": names[op], "operands": operands, "note": note}
if op == op_jump:
if operands:
work.append(operands[0])
continue # 无条件跳转: 无 fall-through
if op == OP_JUMP_IF_TRUE:
if op == op_jump_if:
if operands:
work.append(operands[0])
work.append(nxt) # 条件跳转: 有 fall-through
continue
if op == OP_PUSH_TRY and operands:
if op == op_push_try and operands:
work.append(operands[0]) # catch 目标可达
if op == OP_CREATE_CLOSURE and follow_closures:
# 57 的操作数 [0] 是闭包体入口: 该段代码在闭包被调用时才执行, 但它是**代码**,
if op == op_closure and follow_closures:
# 操作数 [0] 是闭包体入口: 该段代码在闭包被调用时才执行, 但它是**代码**,
# 必须作为遍历根加入, 否则大半个程序都算不到。
# (提升单个函数时应置 follow_closures=False, 避免子函数体溢进父函数)
work.append(z[pc + 1])
work.append(nxt)
return [visited[k] for k in sorted(visited)]
+56
View File
@@ -0,0 +1,56 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
腾讯天御 TDC 混沌 VM —— **纯 Python** XTEA 密钥提取 CLI (零 Node / 零 JS 运行时)。
用法:
.venv/bin/python tools/tdc_key_pure.py [tdc.js] # 默认 core_js 样本
.venv/bin/python tools/tdc_key_pure.py -v har/.../tdc.js # 打印推导过程
.venv/bin/python tools/tdc_key_pure.py --opcodes tdc.js # 顺带打印按构建推导的 opcode 表
实现见 `tcaptcha/tdc_key.py` (定位算法) 与 `tcaptcha/tdc_opcodes.py` (按构建推导操作码表)。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if ROOT not in sys.path:
sys.path.insert(0, ROOT)
from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402
def main() -> int:
ap = argparse.ArgumentParser(description="纯 Python 从 tdc.js 提取 XTEA 密钥")
ap.add_argument("path", nargs="?", default="tcaptcha/core_js/tdc_sample.js")
ap.add_argument("-v", "--verbose", action="store_true", help="打印密钥构造过程")
ap.add_argument("--opcodes", action="store_true", help="打印按构建推导的 opcode 表")
ap.add_argument("--json", action="store_true", help="以 JSON 输出")
args = ap.parse_args()
js = open(args.path, encoding="utf-8", errors="replace").read()
if args.opcodes:
from tcaptcha.tdc_opcodes import derive_opcode_table
names, arity, dynamic, diag = derive_opcode_table(js)
print("handlers=%d dynamic=%s" % (diag["handler_count"], sorted(dynamic)))
for op in sorted(names):
print(" %3d %-18s arity=%s" % (op, names[op], "dyn" if op in dynamic else arity.get(op, 0)))
print()
report = extract_xtea_key_pure(js, verbose=args.verbose)
if args.json:
print(json.dumps(report, ensure_ascii=False))
else:
print("base = %s" % report["base"])
print("salts = %s" % report["salts"])
print("key = %s" % report["key"])
return 0
if __name__ == "__main__":
raise SystemExit(main())