diff --git a/har/1ms_dianxuan/PROGRESS.md b/har/1ms_dianxuan/PROGRESS.md index 6454b93..3c18da5 100644 --- a/har/1ms_dianxuan/PROGRESS.md +++ b/har/1ms_dianxuan/PROGRESS.md @@ -93,8 +93,8 @@ crypto.build_collect(c1,traj,c2,sd,key) -> 4 段独立加密后拼接 | 36 | **「清单项 → cd 槽」的精确规则(同一构建端到端校验确立)** | 装配循环(线上构建 `@4362`,370 条)分支树已读通:
`if(!item) continue;` `v,w,kind = item.get();`
· **`kind===2`(`@4820`)**:把 `r3` 清空转存 `r4`(`@4860`),`r4` 在收尾 `@4905` 二次加工后**拼回**,因此**仍产出 1 槽**(实测第 0 项,随机串 `vXc29RxhvT34Q…`)
· `v===null/undefined`(`@4799`)→ `out += "null"`
· `typeof v==="number"`(`@4738`)→ `out += (v||0)`
· **`kind===1`(`@4724`)→ `out += v` 原样拼**:返回值是 JSON 碎片串;**只按括号外的逗号**切槽 —— `'[]'`/`'[360,360]'`/`'["zh-CN","zh"]'` 内的逗号不构成新槽,而 `"641547489,1791732001"` → 2 槽
· 其余(`@4703`)→ `out += '"' + v + '"'`
· 收尾 `@4753`:`typeof w === "number"` 才再 +1 槽(实测 kind=2 的第二元素是 null,不触发;此路径本次未命中)
→ **37 项 → 38 槽**,仅第 33 项(模块 1)占 2 槽。工具:`tools/tdc_slotmap.py`(`--items` 运行时取样 + `--plain` 明文对拍) | 确证 | | 37 | **按构建生成 collect 的工具链已跑通(含回环自测)** | `tools/tdc_collectgen.py`:静态导出槽位序 → 按 #36 规则算「项→槽」→ 重排 → 该构建密钥加密。
实测(同一构建):37 项 → **38 槽**,明文 686B / collect 920 字符;**解密回环逐字节一致**,cd 与真机样本逐槽相同。
规则修正(重要):`kind=1` 的串**只按括号外逗号**切槽 —— `'[]'`/`'[360,360]'`/`'["zh-CN","zh"]'` 内的逗号不构成新槽;`kind=2` 的项**仍产出 1 槽**(转存 r4 后在收尾 @4905 拼回)。
回归:`tcaptcha/tests/test_collectgen.py`(已并入 selftest) | 确证 | | 39 | **12 只在"答案正确"之后才判**(排查方法学上的关键事实)| 答错一律 50/9:本轮滑块题型下(我们按点选坐标答,必错)两组都是 50;换成真解(text/icon)后立刻全是 12。→ **任何用"答错"做的对照实验都没有区分度**,这解释了早期多轮对照为何无效。今后做 12 的 A/B 必须 `--solve` 真解题 | 确证 | -| 40 | **"谁在发 HTTP"是 12 的剩余唯一自变量**(结论,待最后一步坐实)| 证据链:① 槽位序证伪(§三);② 包内容证伪(本会话真实 collect 由 Python 发仍 12);③ 已知 #26:宽松 aid 190758559 上 **httpx + 无 cookie + 无头保真** 也能 0 —— 说明"请求侧细节"对宽松 aid 无所谓,但对 195672344 不够。④ 真机同 aid 只出 50/9/0(#25)。→ 剩余差在**会话↔请求方绑定**。**收口实验**:把同一 sess/eks/tkid + 同一 collect,分别由浏览器发与 Python 发,比较 errorCode | 结论(待收口) | -| 38 | **纯离线生成还没打通(诚实标注)** | 当前 `--from-sample` 依赖**一份同机的真机明文**做重排;`--values`(语义名→值)需要调用方提供取值字典。**仍缺**:纯算地为每个语义名算出值(多数是静态环境量可配置,但 canvas/WebGL 指纹只能取自真实环境)。这一步完成前,不能声称"纯离线过风控"。 | 状态说明 | +| 40 | ~~**"谁在发 HTTP"是 12 的剩余唯一自变量**~~ **已推翻(2026-10-12)→ 见已知 #58/#59**:自变量是「collect 有没有用本次会话的构建生成」,与谁发 HTTP 无关(Python + httpx 直接发本次构建生成的包 → 0) | 原证据链:① 槽位序证伪(§三);② 包内容证伪(本会话真实 collect 由 Python 发仍 12);③ 已知 #26:宽松 aid 190758559 上 **httpx + 无 cookie + 无头保真** 也能 0 —— 说明"请求侧细节"对宽松 aid 无所谓,但对 195672344 不够。④ 真机同 aid 只出 50/9/0(#25)。→ 剩余差在**会话↔请求方绑定**。**收口实验**:把同一 sess/eks/tkid + 同一 collect,分别由浏览器发与 Python 发,比较 errorCode | 结论(待收口) | +| 38 | ~~**纯离线生成还没打通**~~ **已解决(2026-10-12)→ 见 #57/#58/#59**:`tcaptcha/tdc_values.py` 已为 37 个语义名全部提供取值,并在 1ms aid 上 3/3 拿到 `errorCode=0`。当时状态:当前 `--from-sample` 依赖**一份同机的真机明文**做重排;`--values`(语义名→值)需要调用方提供取值字典。**仍缺**:纯算地为每个语义名算出值(多数是静态环境量可配置,但 canvas/WebGL 指纹只能取自真实环境)。这一步完成前,不能声称"纯离线过风控"。 | 状态说明 | ### 2026-10-12 新增(对照 `docs/tdc-案例.md`,全部基于 `har/online/` 在线新抓样本) @@ -109,6 +109,16 @@ crypto.build_collect(c1,traj,c2,sd,key) -> 4 段独立加密后拼接 | 47 | **结构完全相同的 2 项可用「闭包最后写入什么」消歧** | 这 2 项 entry 40 条、.get 9 条、指令逐条相同(只差 JUMP 绝对地址);差异在**闭包 target**(`CREATE_CLOSURE.target`,不是 JUMP 目标):`new Array()[0] = ""` vs `new Array()[0] = <随机int32>`。实测每份构建恰好 empty×1 + rand×1 | 确证 | | 48 | **配对正确性经独立证据 + 负对照验证** | 独立证据:同一归一化指令序列(抹掉地址/模块号/字符值)下 37/37 全等(唯一例外 fp_salt 的随机 int32,抹掉后 37/37);负对照:清单下标配对 2-3/37、模块 id 配对 2-3/37、随机配对 2/37 —— 对照组极低,证明归一化**没有过拟合** | 确证 | | 49 | **`reorder` 旧实现是恒等映射(真 bug,已修)** | 旧代码 `sample[slot:slot+n]` 里的 `slot` 是**目标**构建自己的槽下标 → 顺序不同时全盘错位、相同时才碰巧对。原测试只断言"是置换",恒等函数也满足 → **弱断言把 bug 放过了**。已补强断言(== 已知正确答案 且 != 原样),并自查:旧实现过不了新断言 | 确证 | +| 50 | **空格 pad 可以只有 1 个字节** → 用 ` {2,}` 找 pad 会漏段,旧 `split_collect_segments` 在这种样本上直接抛异常 | 真机同会话明文:chunk1 = **287** 字节,`287 % 24 == 23` → pad = **1**,它就夹在 `...,0,0,0, [[1,1,12]]...` 之间;旧函数只找到 2 个空格串 → `ValueError: 期望 3 个空格填充串` | 确证(踩坑修正) | +| 51 | **pad 判据必须再加一条「不落在 JSON 字符串内部」** | 只用 `b%24==0 且 (b-a)==24-a%24` 会把 UA 串 `Mozilla/5.0 (Macintosh…` 里偏移 47 的**普通空格**判成 pad(47%24==23);也不能只靠「去掉 pad 后能否 json.loads」排除(少一个空格仍是合法 JSON 串)。用**引号奇偶**判定后,3 份样本候选恰好各 3 个 | 确证 | +| 52 | **轨迹是独立的一段,轨迹槽就是分段点**(3 份样本逐字节一致) | 真机成功样本 441 / **616** / 217 / 33;本会话两次 287 / **10**(`[[1,1,12]]`)/ 356 / 16 与 287 / **516**(61 点)/ 356 / 16。分段写法:chunk1 以 `,` 结尾、traj 是**一个**完整 JSON 值、chunk2 以 `,` 开头且以 `],` 结尾 | 确证 | +| 53 | **`trajectory` 语义名此前认错了模块** | 旧规则 `{"stringify"}` 抓到的是另一个只做 JSON.stringify 的模块(值恒为 `[]`);真轨迹模块的字符串集是 `mouseMoveEvent / mousemove / touchStartEvent / touchmove / tCaptchaDyContent`(entry 279 条)。改规则后仍 37/37、0 unknown,且该模块**正好落在明文分段点上** —— 两条独立证据互证 | 确证(踩坑修正) | +| 54 | **37 个语义名 → 真值全部拿到(38/38 槽全覆盖)** | 用真 Chrome 取「同会话构建 + 同会话 `TDC.getData(true)`」,按「require 序 == 槽位序」对齐(`scripts/verify/align_session.py`);再用取值生成器对 **4 次连续采集**逐槽复算,**152/152 全等**(`scripts/verify/verify_values.py`) | 确证 | +| 55 | **必须区分「会话量」与「每次采集量」** | 4 次 `getData` 实测:`itoken` / 会话随机盐(`struct_40_9#…_rand` = 1591912482)/ 页面加载时刻(`window_time#e48` = 1791745224)**四次完全不变**;而当前时刻(`window_time#e40`)、采集计数(`struct_46_9` = 1,2,3,4)、轨迹**每次都变**。另:`getData` 会把轨迹**清空**(61 点 → 下一次回到 `[[1,1,12]]`) | 确证 | +| 56 | **屏幕宽高方向纠正**:`screen_width = 1440`、`screen_height = 2560` | 浏览器真值 `screen.width/height = 1440/2560`(`availHeight` 2530、`colorDepth` 24、`dpr` 1、`hardwareConcurrency` 10、`getTimezoneOffset` -480),与明文槽位及 `1440-2560-2530-24-*-*-| -*` 串**顺序完全一致**;旧结论「串的 parts[0] → screen_height」是**反的** | 确证(含一次纠正) | +| 57 | **纯离线闭环已打通**(本任务的核心阻塞点) | `scripts/verify/verify_offline.py`:tdc.js → 37 语义名 → 38 槽取值 → 4 段明文 → 4 段加密拼接 → 解密/切段/解析 → **逐槽全等 + 分段几何一致**;产物末尾 `==` 只来自未 pad 的 sd 段 | 确证 | +| 58 | **纯离线生成的 collect 在 1ms aid=195672344 上端到端拿到 `errorCode=0` + 真 ticket**(3/3) | 2026-10-12 实测 `scripts/verify/send_test.py --mode python_values`(aid 195672344 / 签到页 / 2404 文字点选 / 零 Node)。三次分别拿到 `tr03tgI-J732…`、`tr03RLpt4OTMy…`、`tr03n_V7CXAFI…`。每颗 collect 都由**本次会话同一份 tdc.js** 纯离线生成:段几何 461/625/147/16、460/640/148/16、81/708/529/16(每份构建清单序不同 → 分界点不同),轨迹 61/63/71 点、尾点一律 `[1,1,12]` | 确证 | +| 59 | **A/B 隔离出真正自变量:collect 必须「用本次会话的构建」生成** | 同 aid / 同一天 / 同客户端 / 同求解器、答案都正确,只换 collect 构造方式:
`python_values`(本次会话构建 → 槽序与取值都对)→ **0**(3/3);
`python_keyed`(仓库里那份 **2026-10-11** 成功样本模板回填:槽序来自**另一份构建**,屏串还是**另一台设备**的 `2560-1440-1345-24`)→ **12**(2/2)。
→ **推翻已知 #40**("谁在发 HTTP");同时把 #31/#32 的"槽序/取值必须与本次会话构建同源"从假设升级为确证 | 确证(含推翻 #40) | --- @@ -117,8 +127,8 @@ crypto.build_collect(c1,traj,c2,sd,key) -> 4 段独立加密后拼接 | # | 未知 | 当前假设 | 确信度 | | --- | --- | --- | --- | | 1 | 1ms aid=195672344 为何在答案正确时返回 `errorCode=12` | 需要"真实业务会话/真实验证码页面上下文"(登录态),而非环境指纹 | **中(未做单变量实验)** | -| 2 | 新版明文字段语义(哪一位是时间戳/指纹/轨迹) | 时间戳是 `1[6-9]\d{8}` 大整数;轨迹是 `[x,y,t]` 三元组 | 中高 | -| 3 | 明文里 `[360,360]`、`"2560-1440-1345-24-*-*-| -*"` 的作用 | 屏幕/窗口指纹串 | 中 | +| 2 | ~~新版明文字段语义(哪一位是时间戳/指纹/轨迹)~~ **已解决** | 见已知 #54:37 个语义名 → 38 槽真值全部对齐,且被生成器逐槽复现 | 已确证 | +| 3 | ~~明文里 `[360,360]`、`"1440-2560-2530-24-*-*-| -*"` 的作用~~ **已解决** | `[360,360]` = 验证码 iframe 可视区(`viewport_size`);屏串字段序 = **宽-高-可用高-色深-\\*-\\*-\\|-\\***,与浏览器真值 1440/2560/2530/24 一一对应 | 已确证 | --- @@ -198,7 +208,7 @@ TencentCaptchaClient(telemetry_mode="python_keyed") # 默认 5. crypto.build_collect(c1,traj,c2,sd,k) -> 4 段独立加密后拼接 = collect ``` -⚠️ 第 4 步仍需要一份**同源标定样本**提供取值(已知 #38)。这是当前唯一未打通的环节。 +✅ 第 4 步已可**纯离线**完成:`tcaptcha/tdc_values.py` 为 37 个语义名提供取值生成器(已知 #54/#57)。 --- @@ -206,21 +216,27 @@ TencentCaptchaClient(telemetry_mode="python_keyed") # 默认 ### 🆕 2026-10-12 优先(按案例重做后的收尾,按成本排序) -0. **把"纯离线生成"打通(唯一剩余阻塞点)**。当前链路第 4 步仍需一份**同源标定样本** - 提供每个语义名的取值(已知 #38)。要做的是:为每个语义名写一个**取值生成器** - —— 多数是静态环境量(UA / 屏幕 / 时区 / 语言 / 核数 / charset)可直接配置, - canvas/WebGL 指纹用 `fingerprint/profiles.json` 画像(案例 §6.3 的做法)。 - 完成标志:不依赖任何真机明文,纯算生成 collect 并拿到 `errorCode=0`。 -1. **把 4 段结构接进生成链路**(已知 #41 已判决,但 `SlotSpec` 还没产出段边界): - `tdc_collectgen` 目前只输出一个扁平 `cd` 列表;需要额外给出 - `chunk1 / trajectory / chunk2 / sd` 四段(轨迹项所在位置即分界), - 再交给 `crypto.build_collect()`。改完用真机明文对拍总长度(应 1048 字符量级)。 -2. **用在线新样本做一次端到端验证**:`fetch_online_samples.py` 抓构建 → 纯算生成 - collect → 打 verify。先在**宽松 aid 190758559** 上跑(已知 #26 那里 httpx 裸发就能 0), - 通过后再上 1ms。 -3. **1ms 的 12 收口实验**(已知 #40):同一 sess/eks/tkid + 同一 collect, - 分别由浏览器发与 Python 发,比较 errorCode。这一步必须在上面 0~2 完成之后才有意义 - —— 否则"包内容"这个变量本身还没被正确构造出来,实验没有区分度。 +0. ✅ **已完成(2026-10-12)—— 纯离线生成已打通**。做法:真 Chrome 取「同会话构建 + + 同会话 `TDC.getData(true)`」按位置对齐,拿到 37 个语义名 -> 真值(38/38 槽,已知 #54), + 再固化成 `tcaptcha/tdc_values.py`(分**静态画像 / 会话量 / 每次采集量**三层)。 + 复验:`verify_values.py` 对 4 份真机明文 **152/152 逐槽全等**; + `verify_offline.py` 纯离线闭环**逐槽全等 + 分段几何一致**。 + 仍未标定:**真实拖拽轨迹的采样分布**(本会话的 61 点轨迹是 CDP 注入 mousemove 得到的, + 真机人手拖拽的 dt / 位移分布还没采过 —— 现有的 `_success_style_trajectory()` 是旧拟合)。 +1. ✅ **已完成**:4 段结构已接进生成链路 —— `tdc_values.build_plain()` 按 + 「轨迹槽 = 分界点」产出 `chunk1 / trajectory / chunk2 / sd`,直接喂 `crypto.build_collect()`。 + ⚠️ 注意 pad **可能只有 1 个空格**(已知 #50),切分/对拍一律用修好的 + `crypto.split_collect_segments()`,别再自己写 ` {2,}` 找 pad。 +2. ✅ **已完成(2026-10-12)**:端到端验证直接上 1ms 签到页文字点选,**3/3 拿到 `errorCode=0` + + 真 ticket**(见已知 #58)。不需要再走宽松 aid 做前置台阶 —— 直接打真靶即可。 +3. ✅ **已完成(2026-10-12)—— 12 的归因收口**。结论与当时的预期**相反**:#40 假设的 + "谁在发 HTTP" 被推翻。真正的自变量是**collect 有没有用本次会话的构建生成**: + 同 aid / 同客户端 / 同求解器、答案都对,`python_values`(本次构建生成)**3/3 = 0**, + `python_keyed`(旧模板回填,槽序来自另一份构建、屏串是另一台设备的)**2/2 = 12**。 + 复现入口:`.venv/bin/python scripts/verify/send_test.py --mode python_values` + (对照加 `--mode python_keyed`)。 + 新暴露的可复现特性:每次 prehandle 的构建都不同 → 段几何随构建变(461/625/147/16、81/708/529/16…), + 所以**任何"固定模板"的路线都不可能长期成立**,必须按会话构建生成。 ### 既有项 @@ -323,4 +339,27 @@ TencentCaptchaClient(telemetry_mode="python_keyed") # 默认 # ⑦ 回归测试(已并入 selftest,含「reorder 非恒等」强断言) .venv/bin/python -m tcaptcha.tests.test_collectgen .venv/bin/python -m tcaptcha.selftest + +# ⑧ 「同会话构建 + 同会话 collect」的取值标定(这一步打通了纯离线生成的最后阻塞点) +# 先备好同源配对(真 Chrome,9222): +# node tools/cdp_dump_scripts.js --match drag_ele --outdir /tmp/1ms_bridge/scripts +# cp /tmp/1ms_bridge/scripts/<含 VM 的那个>.js /tmp/1ms_bridge/session_build.js +# node tools/cdp_collect.js --match drag_ele --out /tmp/1ms_bridge/session_collect.json +.venv/bin/python scripts/verify/align_session.py # 按位置对齐 -> name_values.json +.venv/bin/python scripts/verify/audit_names.py # 语义名 / 模块结构 / 槽值 联合审计 +.venv/bin/python scripts/verify/verify_values.py \ + --collects /tmp/1ms_bridge/session_collect.json /tmp/1ms_bridge/session_collect2.json \ + /tmp/1ms_bridge/session_collect3.json /tmp/1ms_bridge/session_collect4.json +# 期望: 逐槽全等 152/152(4 份 x 38 槽) + +# ⑨ 纯离线闭环(tdc.js -> 37 语义名 -> 38 槽取值 -> 4 段明文 -> 4 段加密 -> 解回对拍) +.venv/bin/python scripts/verify/verify_offline.py /tmp/1ms_bridge/session_build.js +# 期望: 逐槽全等=是、分段几何=一致 + +# ⑩ 真实发送测试(**只打 1ms 签到页的文字点选**;会真的 POST verify) +.venv/bin/python scripts/verify/send_test.py --mode python_values # 期望 errorCode=0 +.venv/bin/python scripts/verify/send_test.py --mode python_keyed # 对照:期望 12 +# 脚本会 hook generate_telemetry,打印这一颗 collect 的段几何 + 轨迹点数, +# 用来证明发出去的确实是"本次会话构建"生成的包。注意 prehandle 会混发滑块, +# 脚本按 accept_kinds 自动刷新到 2404 文字点选为止(刷新耗尽则退出、不提交)。 ``` diff --git a/scripts/verify/align_session.py b/scripts/verify/align_session.py new file mode 100644 index 0000000..ff77a39 --- /dev/null +++ b/scripts/verify/align_session.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +"""把「同会话 collect 明文」与「同会话 tdc.js 的语义名」按位置精确对齐。 + +这是"不手工配平"的正解,依据都是已确证的链路: + * 槽位序 == 清单模块的 require 序(账本 #34/#35,运行时反证过) + * 语义名 == tdc_modules.analyze(同一份 tdc.js)(账本 #44,37/37 唯一) + * collect 4 段加密(账本 #41),解密后按空格 pad 切回 cd + +输入(全部来自**同一次会话**,无需任何人工假设): + /tmp/1ms_bridge/session_build.js 运行时 tdc.js(Debugger 取的) + /tmp/1ms_bridge/session_collect.json 同会话 TDC.getData(true) + getInfo + +用法: + .venv/bin/python scripts/verify/align_session.py +""" +import base64 +import json +import re +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")) +import tdc_modules as tm # noqa: E402 +from tcaptcha.crypto import split_collect_segments, xtea_decrypt_bytes # noqa: E402 +from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402 + +BUILD = "/tmp/1ms_bridge/session_build.js" +SESS = "/tmp/1ms_bridge/session_collect.json" + +js = open(BUILD, encoding="utf-8", errors="replace").read() +print("运行时 tdc.js: %d 字节" % len(js)) +key = extract_xtea_key_pure(js)["key"] +print("本会话密钥 :", key) +print() + +sess = json.load(open(SESS, encoding="utf-8")) +if "collect" not in sess and isinstance(sess.get("data"), dict): + sess = sess["data"] # cdp_collect.js 落盘的形状是 {target, ok, data:{...}} +collect = unquote(sess["collect"]) # 浏览器侧返回的是 URL-encoded +print("collect: %d 字符" % len(collect)) + +# 解密(4 段拼接:整体解一次即可,因为各段独立 XTEA-ECB) +raw = base64.b64decode(collect) +dec = xtea_decrypt_bytes(raw, key).decode("latin1") +plain = dec.rstrip("\x00") +print("解密 %d 字节" % len(dec)) +runs = [m.span() for m in re.finditer(r" {2,}", plain)] +print("空格 pad 串:", runs) +print() +if len(runs) == 3: + c1, traj, c2, sd = split_collect_segments(plain) + print("段: chunk1=%d traj=%d chunk2=%d sd=%d" % (len(c1), len(traj), len(c2), len(sd))) +else: + c1 = traj = c2 = sd = None +print() + +logical = re.sub(r" {2,}", "", plain) +obj = json.loads(logical) +cd = obj["cd"] +print("cd 槽数:", len(cd)) +print() + +rows = tm.analyze(BUILD) +names = [r["name"] for r in rows] +mods = [r["module"] for r in rows] +print("清单项 %d, 语义名 %d" % (len(mods), len(names))) +print() + +# itoken 占 2 槽(^\\d+:\\d+$),其余 1 槽 +multi = [i for i, n in enumerate(names) if n.startswith("itoken")] +slots = 0 +print("%-2s %-6s %-34s %s" % ("#", "模块", "语义名", "值")) +print("-" * 110) +out = {} +k = 0 +for i, (m, nm) in enumerate(zip(mods, names)): + n = 2 if i in multi else 1 + if k + n > len(cd): + break + vals = cd[k:k + n] + k += n + val = vals[0] if n == 1 else vals + out[nm] = val + print("%-2d %-6d %-34s %s" % (i, m, nm, json.dumps(val, ensure_ascii=False)[:60])) +print() +print("覆盖槽数 %d / cd %d" % (k, len(cd))) +print() +Path("/tmp/1ms_bridge/name_values.json").write_text( + json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8") +print(">>> 语义名 -> 真值 已存 /tmp/1ms_bridge/name_values.json") diff --git a/scripts/verify/audit_names.py b/scripts/verify/audit_names.py new file mode 100644 index 0000000..9e45ee7 --- /dev/null +++ b/scripts/verify/audit_names.py @@ -0,0 +1,142 @@ +# -*- coding: utf-8 -*- +"""audit_names.py —— 把「语义名 / 模块结构 / 明文槽值」放在一张表里逐个核对。 + +为什么需要(2026-10-12 实测): + 对齐是**按位置**做的(require 序 == 槽位序),所以一旦某个语义名识别错了, + 它对应的**真值就会被记到错误的名字下**。tdc_profile 之前就把「轨迹」的 + 空数组记成了 trajectory,而真正的轨迹(含 mousemove/touchmove 的大模块) + 却在另一个名字下。名字错 = 生成器错,必须逐个核对。 + +本脚本给出 4 条独立证据并排显示: + 1. 指纹命中的字符串(tdc_modules 的规则) + 2. 结构签名(entry/get 指令数、closure 种类) + 3. 该模块**自身代码里**是否出现鼠标/触摸/时钟等关键串 + 4. 同会话明文里该槽的**真值** + +用法: + .venv/bin/python scripts/verify/audit_names.py \ + --build /tmp/1ms_bridge/session_build.js \ + --collect /tmp/1ms_bridge/session_collect.json +""" +import argparse +import base64 +import json +import re +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")) + +import tdc_modules as tm # noqa: E402 +import tdc_program as tp # noqa: E402 +from tcaptcha.crypto import xtea_decrypt_bytes # noqa: E402 +from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402 + +# 用于「这个模块到底像在采什么」的旁证关键词 +HINTS = { + "鼠标/触摸轨迹": ["mousemove", "mouseMoveEvent", "touchmove", "touchStartEvent", + "mousedown", "pointermove", "clientX", "pageX"], + "时钟": ["getTime", "Date", "performance"], + "屏幕": ["screen", "colorDepth", "availHeight"], + "WebGL": ["UNMASKED", "getExtension", "getParameter"], + "Canvas": ["toDataURL", "getImageData", "fillText"], + "UA/平台": ["userAgent", "platform"], + "时区/语言": ["getTimezoneOffset", "languages"], + "自动化检测": ["webdriver", "$cdc_", "__webdriver_script_fn"], + "存储": ["cookie", "localStorage", "sessionStorage"], + "iframe/DOM": ["contentWindow", "createElement", "appendChild"], +} + + +def parse_cd(plain: str): + """把明文切成 (cd 元素列表带跨度, 顶层对象)。pad(>=2 空格)先移除。""" + logical = re.sub(r" {2,}", "", plain) + dec = json.JSONDecoder() + i = logical.index("[", logical.index('"cd"')) + spans, j = [], i + 1 + while True: + while logical[j] in " \t\r\n,": + j += 1 + if logical[j] == "]": + break + val, end = dec.raw_decode(logical, j) + spans.append((val, j, end)) + j = end + return spans, json.loads(logical) + + +def slot_names(rows): + """把 37 个清单项展开成 38 个槽的语义名(itoken 占 2 槽)。""" + out = [] + for i, r in enumerate(rows): + n = 2 if r["name"].startswith("itoken") else 1 + out.extend([r["name"]] * n) + return out + + +def main() -> int: + ap = argparse.ArgumentParser(description="语义名 / 结构 / 槽值 联合审计") + ap.add_argument("--build", default="/tmp/1ms_bridge/session_build.js") + ap.add_argument("--collect", default="/tmp/1ms_bridge/session_collect.json") + ap.add_argument("--json", default="/tmp/1ms_bridge/audit.json") + args = ap.parse_args() + + js = open(args.build, encoding="utf-8", errors="replace").read() + key = extract_xtea_key_pure(js)["key"] + sess = json.load(open(args.collect, 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") + + spans, _obj = parse_cd(plain) + pads = [m.span() for m in re.finditer(r" {2,}", plain)] + + p = tp.load(args.build) + rows = tm.analyze_prog(p) + names = slot_names(rows) + + print("构建 %s (%d 字节) 明文 %d 字节 cd %d 槽 清单 %d 项" + % (Path(args.build).name, len(js), len(plain), len(spans), len(rows))) + print("pad 段边界:", [(a, b, b - a) for a, b in pads]) + pad_ends = {a for a, _b in pads} + print() + + out = [] + print("%-3s %-5s %-36s %-9s %-22s %s" + % ("项", "模块", "语义名", "结构", "槽值", "旁证")) + print("-" * 132) + for i, r in enumerate(rows): + n = 2 if r["name"].startswith("itoken") else 1 + k = sum(2 if x["name"].startswith("itoken") else 1 for x in rows[:i]) + vals = [spans[k + t][0] for t in range(n)] + val = vals[0] if n == 1 else vals + st = r["struct"] + tag = "e%s/g%s%s" % (st["entry_instr"], st["get_instr"], + "_call" if st["call_func"] else "") + s = set(r["strings"]) + hints = [k2 for k2, ks in HINTS.items() if any(x in s for x in ks)] + # 该槽的结束位置是否正好是 pad 边界? + end = spans[k + n - 1][2] + mark = " <<< 段边界" if end in pad_ends else "" + print("%-3d %-5d %-36s %-9s %-22s %s%s" + % (r["item"], r["module"], r["name"], tag, + json.dumps(val, ensure_ascii=False)[:20], ",".join(hints), mark)) + out.append({ + "item": r["item"], "module": r["module"], "name": r["name"], + "struct": st, "tag": tag, "delegate": r["delegate"], + "strings": r["strings"], "value": val, + "span": [spans[k][1], end], "segment_boundary": end in pad_ends, + }) + Path(args.json).write_text(json.dumps(out, ensure_ascii=False, indent=1), + encoding="utf-8") + print() + print(">>> %s" % args.json) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify/balance_names_values.py b/scripts/verify/balance_names_values.py new file mode 100644 index 0000000..2774b3d --- /dev/null +++ b/scripts/verify/balance_names_values.py @@ -0,0 +1,75 @@ +# -*- coding: utf-8 -*- +"""全量配平:37 个语义名 <-> 真机明文的 38 个槽。 + +约束(这些是硬证据,不是猜): + 1. 值内容唯一确定语义名(MacIntel/UTF-8/ANGLE/GgoA.../98k/iframe...) + 2. screen_fingerprint `"1440-2560-2530-24-*-*-|-*"` 反解出 + screen_height=1440 / screen_width=2560 / screen_color_depth=24 + 3. itoken 有 `^\\d+:\\d+$` 特征 -> 占 **2 槽**(`641547489,1791732001`) + 4. 37 项 -> 38 槽(仅 itoken 多占 1 槽),可与明文槽数对账 + +输出: 每个语义名的取值 + 置信度,供人工核对。 +""" +import json +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "tools")) +import tdc_modules as tm # noqa: E402 + +RAW = "/tmp/1ms_bridge/real_collect_plain.txt" +raw = open(RAW, encoding="latin1").read().rstrip("\x00") +cd = json.loads(re.sub(r" {2,}", "", raw))["cd"] +print("明文 38 槽:") +for i, v in enumerate(cd): + print(" %2d %s" % (i, json.dumps(v, ensure_ascii=False)[:78])) +print() + +rows = tm.analyze("har/1ms_dianxuan/tdc.js") +names = [r["name"] for r in rows] +print("语义名 %d 个" % len(names)) +print() + +# ---- 按内容给每个槽打标签(不依赖位置)---- +def tags(v): + t = [] + if isinstance(v, str): + if "Mozilla" in v: t.append("user_agent") + if v == "MacIntel": t.append("platform") + if re.fullmatch(r"[+-]\d\d", v): t.append("timezone_offset") + if v.upper() == "UTF-8": t.append("charset") + if "ANGLE" in v: t.append("webgl_renderer") + if "Inc." in v: t.append("webgl_vendor") + if v.startswith("GgoA"): t.append("canvas_fingerprint") + if "rand=" in v and "captcha" in v: t.append("template_url") + if v == "98k": t.append("constant_98k") + if v == "iframe": t.append("iframe_detect") + if "*" in v and "-" in v: t.append("screen_fingerprint") + if v.startswith("http") and "1ms.run" in v: t.append("location_href") + elif isinstance(v, list): + if v and all(isinstance(x, list) and len(x) == 3 for x in v): t.append("trajectory") + if v and all(isinstance(x, str) for x in v): t.append("languages") + elif isinstance(v, int) and not isinstance(v, bool): + if 1_600_000_000 <= v <= 2_100_000_000: t.append("timestamp?") + elif v > 1_000_000: t.append("bigint?") + else: t.append("smallint?") + elif isinstance(v, str) and v == "": + t.append("emptystr?") + return t + +print("逐槽标签:") +for i, v in enumerate(cd): + print(" %2d %-22s %s" % (i, ",".join(tags(v)), json.dumps(v, ensure_ascii=False)[:50])) +print() + +# ---- 反解 screen_fingerprint 的约束 ---- +fp = next((v for v in cd if isinstance(v, str) and "*" in v and "-" in v), None) +if fp: + p = fp.split("-") + print("screen_fingerprint = %r -> 高%s 宽%s 色深%s" % (fp, p[0], p[1], p[3])) + for target, want in (("screen_height", int(p[0])), ("screen_width", int(p[1])), + ("screen_color_depth", int(p[3]))): + idxs = [i for i, v in enumerate(cd) if v == want] + print(" %-20s 期望 %s -> 明文里出现在槽 %s" % (target, want, idxs)) diff --git a/scripts/verify/label_plaintext.py b/scripts/verify/label_plaintext.py new file mode 100644 index 0000000..fc474dc --- /dev/null +++ b/scripts/verify/label_plaintext.py @@ -0,0 +1,138 @@ +# -*- coding: utf-8 -*- +"""把真机明文的值,与「语义名」精确配对。 + +难点:真机明文的槽顺序 = **它自己那次构建**的 require 顺序, +而我手上只有 har/online/ 等若干构建;不知道明文属于哪一份。 + +解法(自洽性打分):把明文的值**按每个候选构建的语义名顺序**标注一遍, +再看标注结果是否语义自洽(platform 的位置该是 "MacIntel"、user_agent 该含 +"Mozilla"、charset 该是 "UTF-8" …)。自洽度最高者即为它所属的构建; +若都不匹配,也能从"哪些语义名被正确喂到"反推出每个名字的真实取值。 + +用法: + .venv/bin/python scripts/verify/label_plaintext.py <明文文件> <若干 tdc.js> +""" +import glob +import json +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "tools")) +import tdc_modules as tm # noqa: E402 + +# ---- 语义名 -> 期望值特征(这就是"画像"的定义雏形)---- +def likes(name: str, v) -> str: + """值是否与语义名相容。返回 OK / NO / ??(?? = 该名字无法从值判断)""" + s = json.dumps(v, ensure_ascii=False) + if name.startswith("user_agent"): + return "OK" if isinstance(v, str) and "Mozilla" in v else "NO" + if name == "platform": + return "OK" if isinstance(v, str) and v and " " not in v else "NO" + if name == "languages": + return "OK" if isinstance(v, list) and v and isinstance(v[0], str) else "NO" + if name == "hardware_concurrency": + return "OK" if isinstance(v, int) and 2 <= v <= 64 else "NO" + if name == "screen_width": + return "OK" if isinstance(v, int) and v > 500 else "NO" + if name == "screen_height": + return "OK" if isinstance(v, int) and v > 300 else "NO" + if name == "screen_color_depth": + return "OK" if isinstance(v, int) and 16 <= v <= 48 else "NO" + if name == "screen_fingerprint": + return "OK" if isinstance(v, str) and "-" in v and "*" in v else "NO" + if name == "timezone_offset": + return "OK" if isinstance(v, str) and re.fullmatch(r"[+-]\d\d", v) else "NO" + if name == "charset": + return "OK" if isinstance(v, str) and "UTF" in v.upper() else "NO" + if name == "webgl_renderer": + return "OK" if isinstance(v, str) and ("ANGLE" in v or "Renderer" in v) else "NO" + if name == "webgl_vendor": + return "OK" if isinstance(v, str) and ("Inc." in v or "Google" in v) else "NO" + if name == "canvas_fingerprint": + return "OK" if isinstance(v, str) and len(v) > 40 else "NO" + if name == "constant_98k": + return "OK" if v == "98k" else "NO" + if name == "template_url": + return "OK" if isinstance(v, str) and "captcha" in v and "rand=" in v else "NO" + if name in ("location_href", "referrer"): + return "OK" if isinstance(v, str) and v.startswith("http") else "??" + if name == "trajectory": + return "OK" if (isinstance(v, list) and v + and all(isinstance(x, list) and len(x) == 3 for x in v)) else "NO" + return "??" + + +def main() -> int: + if len(sys.argv) < 3: + print(__doc__) + return 2 + plain = open(sys.argv[1], encoding="latin1").read().rstrip("\x00") + logical = re.sub(r" {2,}", "", plain) + cd = json.loads(logical)["cd"] + print("明文 cd 槽数: %d" % len(cd)) + print() + + cands = [] + for a in sys.argv[2:]: + cands += sorted(glob.glob(a)) + if not cands: + print("没有候选构建") + return 1 + + def assign(names, nslots): + """按语义名顺序把 nslots 个槽分配给各名字(交错推进)。""" + out = {} + k = 0 + for nm in names: + if k >= nslots: + break + out.setdefault(nm, []).append(cd[k]) + k += 1 + return out + + best = None + for f in cands: + rows = tm.analyze(f) + names = [r["name"] for r in rows] + # 假设除 itoken(占 2 槽)外都是 1 槽 + itok = [i for i, n in enumerate(names) if n.startswith("itoken")] + n_extra = 1 if itok else 0 + # 逐个候选:把 cd 的槽按名字顺序切(itoken 那两个槽合并看第一个) + mapping = {} + k = 0 + for i, nm in enumerate(names): + if k >= len(cd): + break + if i in itok: + mapping[nm] = cd[k:k + 2] + k += 2 + else: + mapping[nm] = [cd[k]] + k += 1 + ok = no = unk = 0 + for nm, vals in mapping.items(): + st = likes(nm, vals[0]) + ok += st == "OK" + no += st == "NO" + unk += st == "??" + score = ok - no + print(" %-46s OK=%2d NO=%2d ??=%2d 分=%d" + % (Path(f).name, ok, no, unk, score)) + if best is None or score > best[0]: + best = (score, f, mapping, ok, no) + + print() + print(">>> 最自洽的构建: %s (OK=%d NO=%d)" % (Path(best[1]).name, best[3], best[4])) + print() + print("该假设下的 name -> 值 标注:") + for nm, vals in best[2].items(): + st = likes(nm, vals[0]) + mark = {"OK": " ", "NO": "!", "??": "?"}[st] + print(" %s %-40s %s" % (mark, nm, json.dumps(vals, ensure_ascii=False)[:70])) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify/send_test.py b/scripts/verify/send_test.py new file mode 100644 index 0000000..4af5754 --- /dev/null +++ b/scripts/verify/send_test.py @@ -0,0 +1,105 @@ +# -*- coding: utf-8 -*- +"""send_test.py —— 真实发送测试:1ms 签到页的**文字点选**。 + +只打这一条链路(用户指定): + aid 195672344 (1ms.run) + 入口 https://1ms.run/console/personal/benefits/check-in + 题型 2404 文字点选(prehandle 会混发滑块/图标,脚本按 accept_kinds 刷新到文字点选为止) + 遥测 --mode python_values(默认)= tcaptcha/tdc_values.py 的**纯离线**生成 + +脚本会 hook 客户端的 generate_telemetry,把这一颗 collect 的段几何打出来 —— +用来证明"发出去的确实是新链路生成的包",而不是旧模板回填。 + +用法: + .venv/bin/python scripts/verify/send_test.py + .venv/bin/python scripts/verify/send_test.py --mode python_keyed # 对照组:旧模板回填 +""" +import argparse +import json +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "tools")) + +import tcaptcha.client as client_mod # noqa: E402 +from tcaptcha.client import KIND_TEXT_CLICK, TencentCaptchaClient # noqa: E402 + +AID = "195672344" +ENTRY = "https://1ms.run/console/personal/benefits/check-in" +REFERER = "https://1ms.run/" + + +def main() -> int: + ap = argparse.ArgumentParser(description="1ms 签到页文字点选 · 真实发送测试") + ap.add_argument("--mode", default="python_values", + choices=["python_values", "python_keyed", "python"]) + ap.add_argument("--max-refreshes", type=int, default=12) + ap.add_argument("--max-retries", type=int, default=2) + args = ap.parse_args() + + seen = {} + real_gen = client_mod.generate_telemetry + + def hooked(tdc_code, points, **kw): + t0 = time.time() + out = real_gen(tdc_code, points, **kw) + seen.update({ + "source": out.get("source"), + "collect_len": len(out.get("collect") or ""), + "eks_len": len(out.get("eks") or ""), + "gen_ms": int((time.time() - t0) * 1000), + "points": [list(p) for p in points], + }) + # 段几何(新链路才有;旧模板回填没有 segments) + if out.get("source") == "python_values": + try: + from tcaptcha.crypto import split_collect_segments, xtea_decrypt_bytes + import base64 + plain = xtea_decrypt_bytes(base64.b64decode(out["collect"]), + out["key"]).decode("latin1").rstrip("\x00") + c1, traj, c2, sd = split_collect_segments(plain) + seen["segments"] = [len(c1), len(traj), len(c2), len(sd)] + seen["traj_points"] = len(json.loads(traj)) + seen["traj_tail"] = json.loads(traj)[-1] + except Exception as e: # noqa: BLE001 + seen["segments_err"] = str(e)[:120] + print("[hook] %s" % json.dumps(seen, ensure_ascii=False)) + return out + + client_mod.generate_telemetry = hooked + + c = TencentCaptchaClient( + aid=AID, entry_url=ENTRY, referer=REFERER, wb="1", + accept_kinds=(KIND_TEXT_CLICK,), + telemetry_mode=args.mode, + proxy=None, + ) + t0 = time.time() + try: + res = c.solve(max_retries=args.max_retries, max_refreshes=args.max_refreshes) + finally: + c.close() + + print() + print("=" * 70) + print("模式 : %s" % args.mode) + print("耗时 : %.1fs" % (time.time() - t0)) + print("success : %s" % res.get("success")) + print("kind : %s 指令=%r" % (res.get("kind"), res.get("instruction"))) + print("points : %s" % res.get("points")) + print("ticket : %r" % res.get("ticket")) + print("randstr : %r" % res.get("randstr")) + print("errorCode : %s" % (res.get("raw_response") or {}).get("errorCode")) + print("原始响应 : %s" % json.dumps(res.get("raw_response") or {}, + ensure_ascii=False)[:400]) + print("本颗遥测 : %s" % json.dumps(seen, ensure_ascii=False)) + print("error : %s" % (res.get("error") or "")[:200]) + print("=" * 70) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify/verify_offline.py b/scripts/verify/verify_offline.py new file mode 100644 index 0000000..50a9e7d --- /dev/null +++ b/scripts/verify/verify_offline.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +"""verify_offline.py —— 纯离线生成一颗 collect,再自己解回来对拍(闭环验证)。 + +链路(全程零浏览器、零 Node 运行时;密钥来自纯 Python 静态提取): + + tdc.js + -> tdc_modules.analyze() 37 个语义名(清单序 = 槽位序) + -> tdc_values.generate_cd() 38 个槽的取值 + -> tdc_values.build_plain() 4 段明文 + -> crypto.build_collect() 4 段各自 XTEA + base64 拼接 + -> 解密 / 切段 / 解析 应当逐槽回到原值 + +同时校验**长度算术**:前三段 pad 到 24 的倍数 -> base64 无 '=' 填充; +sd 段不 pad -> 允许 '=' 且只出现在末尾。 + +用法: + .venv/bin/python scripts/verify/verify_offline.py /tmp/1ms_bridge/session_build.js +""" +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")) + +import tdc_modules as tm # noqa: E402 +from tcaptcha.crypto import (build_collect, split_collect_segments, # noqa: E402 + xtea_decrypt_bytes) +from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402 +from tcaptcha.tdc_values import Session, build_plain, generate_cd, trajectory_index # noqa: E402 + + +def main() -> int: + build = sys.argv[1] if len(sys.argv) > 1 else "/tmp/1ms_bridge/session_build.js" + js = open(build, encoding="utf-8", errors="replace").read() + key = extract_xtea_key_pure(js)["key"] + rows = tm.analyze(build) + unknown = [r["name"] for r in rows if r["name"] == "unknown"] + + sess = Session() + cd = generate_cd(rows, sess) + plain = build_plain(cd, rows) + k = trajectory_index(rows) + + print("构建 : %s (%d 字节)" % (Path(build).name, len(js))) + print("清单 : %d 项, 未识别 %d" % (len(rows), len(unknown))) + print("槽 : %d 个(轨迹在第 %d 槽)" % (len(cd), k)) + print("段明文 : chunk1=%d traj=%d chunk2=%d sd=%d" + % (len(plain["chunk1"]), len(plain["trajectory"]), + len(plain["chunk2"]), len(plain["sd"]))) + for nm in ("chunk1", "trajectory", "chunk2"): + n = len(plain[nm]) + print(" %-10s pad %2d -> %d (=24*%d)" + % (nm, 24 - n % 24, n + 24 - n % 24, (n + 24 - n % 24) // 24)) + + collect = build_collect(plain["chunk1"], plain["trajectory"], + plain["chunk2"], plain["sd"], key) + print("collect : %d 字符, '=' 出现在 %s" + % (len(collect), [i for i, c in enumerate(collect) if c == "="] or "无")) + + # ---- 解回来对拍 ---- + dec = xtea_decrypt_bytes(base64.b64decode(unquote(collect)), key) \ + .decode("latin1").rstrip("\x00") + c1, traj, c2, sd = split_collect_segments(dec) + got = json.loads(c1 + traj + c2 + sd)["cd"] + + ok = got == cd + print() + print("解回槽数 : %d" % len(got)) + print("逐槽全等 : %s" % ("是" if ok else "否")) + if not ok: + for i, (o, g) in enumerate(zip(cd, got)): + if o != g: + print(" 槽%-3d 原=%s 回=%s" + % (i, json.dumps(o, ensure_ascii=False)[:40], + json.dumps(g, ensure_ascii=False)[:40])) + # 分段几何也要一致 + geo = (len(c1), len(traj), len(c2), len(sd)) == \ + (len(plain["chunk1"]), len(plain["trajectory"]), + len(plain["chunk2"]), len(plain["sd"])) + print("分段几何 : %s" % ("一致" if geo else "不一致")) + return 0 if (ok and geo) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/verify/verify_values.py b/scripts/verify/verify_values.py new file mode 100644 index 0000000..4736f7e --- /dev/null +++ b/scripts/verify/verify_values.py @@ -0,0 +1,104 @@ +# -*- 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")) + +import tdc_modules as tm # noqa: E402 +from tcaptcha.crypto import split_collect_segments, xtea_decrypt_bytes # noqa: E402 +from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402 +from tcaptcha.tdc_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()) diff --git a/tcaptcha/client.py b/tcaptcha/client.py index cd1e672..65a5fee 100644 --- a/tcaptcha/client.py +++ b/tcaptcha/client.py @@ -148,26 +148,32 @@ class TencentCaptchaClient: min_posterior: float = 0.95, pow_timeout: int = 4000000, inter_attempt_delay: float = 0.25, - telemetry_mode: str = "python_keyed", + telemetry_mode: str = "python_values", proxy: Optional[str] = None, cookies: Optional[dict] = None, transport: str = "httpx", impersonate: Optional[str] = None, ): """ - telemetry_mode 决定 collect/eks 怎么来 (见 tcaptcha.crypto.generate_telemetry), 只走纯算: + telemetry_mode 决定 collect/eks 怎么来 (见 tcaptcha.crypto.generate_telemetry): - * "python_keyed" (默认): Node 只从本次构建的 tdc.js 里提取 XTEA 密钥字; - 明文组装 / 轨迹 / 时间戳 / 加密 / 编码全部在 Python 完成。 - * "python": 完全不调用 Node, 直接用内置 DEFAULT_KEY + * "python_values" (默认, 2026-10-12 起): **纯离线**按**本次会话的** tdc.js 生成整颗 + collect —— 37 个语义名 -> 38 槽取值 -> 4 段明文 -> 4 段加密, + 零 Node、零真机明文。"必须用本次会话的构建"这一点是实测出来的 + 分水岭: 同一 aid 下它 3/3 拿 0, 而模板回填 2/2 恒 12 + (har/1ms_dianxuan/PROGRESS.md 已知 #58/#59)。 + * "python_keyed": 明文来自仓库里的**成功样本模板回填** + (generate_collect_success): 槽序来自**另一份构建**、 + 取值来自**另一台设备** -> 1ms aid 上实测恒 12。 + * "python": 完全不调用 Node, 用内置 DEFAULT_KEY + 同样的模板回填 (仅对不校验遥测的宽松 aid 有效)。 - 关于 errorCode=12 (实测 2026-10-11, 见 har/1ms_dianxuan/ANALYSIS.md §7): - 1ms.run aid=195672344 在答案正确时返回 12, 且与遥测后端无关 —— - python / python_keyed / node / browser 四种后端、4 个不同省份出口 IP、 - 9 分钟静默后单发, **全部都是 12**。所以 12 不是遥测/环境/IP 问题, - 不要再走"补环境"或"换 IP"的路; 真正的变量 (登录态 / 真实验证码页面上下文) - 尚未被单变量实验区分。 + 关于 errorCode=12 (**2026-10-12 已收口, 旧结论已推翻**): + 真因不是"环境/IP/请求方", 而是 **collect 没有用本次会话的那份构建生成**。 + 同 aid / 同客户端 / 同求解器、答案都正确, 只换 collect 构造方式: + python_values -> 0 (3/3), python_keyed (旧模板回填) -> 12 (2/2)。 + 见 har/1ms_dianxuan/PROGRESS.md 已知 #58/#59。 + 附带一条方法学: **答错一律 50/9**, 所以任何用"答错"做的对照都没有区分度。 cookies: 可选, 逐条写入 captcha 域 (turing.captcha.qcloud.com) 的 Cookie 头。 成功样本 (har/签到成功-1.har) 的 captcha 域请求带着腾讯云登录 cookie diff --git a/tcaptcha/crypto.py b/tcaptcha/crypto.py index 8a5451b..aa0c8c1 100644 --- a/tcaptcha/crypto.py +++ b/tcaptcha/crypto.py @@ -13,6 +13,7 @@ """ import base64 +import itertools import json import logging import os @@ -123,7 +124,11 @@ def xtea_encrypt_bytes(data: bytes, key: List[int]) -> bytes: SEG_ALIGN = 24 SEG_PAD = b" " -_collect_seg_re = re.compile(rb" {2,}") +# 候选 pad = 一段空格。⚠️ **pad 可以只有 1 个空格**(段内容长 % 24 == 23 时), +# 2026-10-12 在真机上实测到这种样本(chunk1 = 287 字节 -> pad 1)。 +# 所以**不能**用 ` {2,}` 去找 pad,那个正则会把 `{"cd":[...,0,0,0, [[1,1,12]]...` 里的 +# 单空格 pad 整段漏掉,切段结果直接崩。判据必须用长度模 24 的算术。 +_SpaceRun = re.compile(rb" +") def xtea_encrypt_segment(text: str, key: List[int], @@ -132,7 +137,8 @@ def xtea_encrypt_segment(text: str, key: List[int], 为什么要 pad 到 24(而不是 8):24 字节 -> 32 个 Base64 字符,且**不会产生 `=` 填充**。 4 段密文的 Base64 直接拼接后看起来就是一个完整的 Base64 串,没有分隔符。 - (真机明文里能直接看到这些空格:chunk1 +5、trajectory +10、chunk2 +24。) + 实测 pad 长度落在 **1..24**(真机样本里见过 15/8/23 与 1/14/4)—— + 注意 **1 个空格的 pad 是真实存在的**,别用 ` {2,}` 去找它(见 split_collect_segments)。 """ data = text.encode("latin1") # 注意:即使 len%align==0 也要补一整个 align(真机 chunk2 内容 288 字节、 @@ -174,27 +180,81 @@ def build_collect(chunk1: str, trajectory: str, chunk2: str, sd: str, return b64 -def split_collect_segments(plain: str) -> Tuple[str, str, str, str]: - """把一份**已解密**的 collect 明文按空格 pad 切回 4 段(去 pad)。 +def _inside_json_string(data: bytes, pos: int) -> bool: + """判断字节偏移 pos 是否落在 JSON 字符串字面量内部(数未转义的双引号奇偶)。""" + n = 0 + i = 0 + while i < pos: + c = data[i] + if c == 0x5C: # 反斜杠:跳过被转义的下一个字节 + i += 2 + continue + if c == 0x22: # 双引号 + n += 1 + i += 1 + return n % 2 == 1 - 用于从真机样本反解出 chunk1/trajectory/chunk2/sd,做对拍或标定。 + +def _candidate_pads(data: bytes) -> List[Tuple[int, int]]: + """候选 pad 位置:空格串 [a,b) 且满足 + ① b % SEG_ALIGN == 0 ② (b-a) == SEG_ALIGN - a % SEG_ALIGN ③ 不落在 JSON 字符串内部 + + 为什么要 ③:① ② 单独用**会误报** —— 实测真机 UA 串 + "Mozilla/5.0 (Macintosh…" 里偏移 47 的那个空格恰好满足 + (47 % 24 == 23 -> 期望 pad 1,48 % 24 == 0),而它只是字符串内的普通空格。 + 段一定切在 JSON 记号边界上,所以真 pad 处引号计数必为偶数;用它把误报滤掉。 + """ + out: List[Tuple[int, int]] = [] + for m in _SpaceRun.finditer(data): + a, b = m.span() + if b % SEG_ALIGN or (b - a) != SEG_ALIGN - a % SEG_ALIGN: + continue + if _inside_json_string(data, a): + continue + out.append((a, b)) + return out + + +def split_collect_segments(plain: str) -> Tuple[str, str, str, str]: + """把一份**已解密**的 collect 明文按空格 pad 切回 4 段(去掉 pad)。 + + 结构(3 份真机样本逐字节核对,2026-10-12): + chunk1 = '{"cd":[' + 轨迹之前的槽 + ',' + traj = '<轨迹数组>' + chunk2 = ',' + 轨迹之后的槽 + '],' + sd = '"sd":{...}}' + 前三段各用空格 pad 到 24 字节整数倍,sd 不 pad。 + + 做法:先按模 24 算术列出**候选** pad,再对候选三元组做结构校验 + (首段以 '{"cd":[' 开头、末段以 '"sd"' 开头、去掉 pad 后是合法 JSON)。 + 校验这一步不可省 —— 字符串里的普通空格同样会命中候选判据。 """ data = plain.encode("latin1") - bounds = [0] - for m in _collect_seg_re.finditer(data): - bounds.append(m.end()) - bounds.append(len(data)) - if len(bounds) != 5: - raise ValueError( - f"期望 3 个空格填充串(4 段结构),实际切出 {len(bounds) - 1} 段;" - f"这份明文可能不是「四段加密」格式") - out = [] - for i in range(4): - seg = data[bounds[i]:bounds[i + 1]] - if i < 3: - seg = seg.rstrip(SEG_PAD) # 前三段去掉空格 pad - out.append(seg.decode("latin1")) - return out[0], out[1], out[2], out[3] + cands = _candidate_pads(data) + for (a1, b1), (a2, b2), (a3, b3) in itertools.combinations(cands, 3): + if not (b1 <= a2 and b2 <= a3): + continue + segs = [data[0:a1], data[b1:a2], data[b2:a3], data[b3:]] + # 段边界一定落在 JSON 记号边界上(3 份真机样本一致): + # chunk1 以 ',' 结尾;traj 是**一个**完整 JSON 值;chunk2 以 ',' 开头、以 '],' 结尾 + if not (segs[0].startswith(b'{"cd":[') and segs[0].endswith(b',') + and segs[2].startswith(b',') and segs[2].endswith(b'],') + and segs[3].startswith(b'"sd"')): + continue + try: + json.loads(segs[1].decode("latin1")) # traj 必须是独立可解析的 JSON 值 + except Exception: # noqa: BLE001 + continue + try: + obj = json.loads(b"".join(segs).decode("latin1")) + except Exception: # noqa: BLE001 + continue + if "cd" not in obj or "sd" not in obj: + continue + return (segs[0].decode("latin1"), segs[1].decode("latin1"), + segs[2].decode("latin1"), segs[3].decode("latin1")) + raise ValueError( + f"未能切出「四段加密」结构(候选 pad {len(cands)} 个);这份明文可能不是 4 段格式") def xtea_decrypt_bytes(data: bytes, key: List[int]) -> bytes: @@ -331,38 +391,13 @@ _SUCCESS_TPL_PATH = Path(__file__).resolve().parent / "models" / "success_collec def _success_style_trajectory(click_points: List[Tuple[int, int]], img_w: int = 672) -> str: - """ - 按「成功签到样本」实测统计生成轨迹 (61 点, dt 均值 30.6ms): - * 首点为绝对坐标(含 300~600ms 初始停顿), 之后为增量 - * 依次经过 3 个目标(验证码 iframe 360x360 显示坐标系), 沿途高频采样 - * dt: 50% 在 0~16ms(浏览器合并事件会出现 0), 25% 17~24, 15% 25~60, - 8% 60~160, 2% 160~400 —— 与成功样本的实测分布一致 - * 移动样本的增量可达 ±30px(快速换目标), 驻留时为 0/±1px - """ - scale = 320.0 / img_w - targets = [(int(round(20 + x * scale)), int(round(60 + y * scale))) - for x, y in click_points] - cur = [random.randint(140, 220), random.randint(300, 340)] - events = [[cur[0], cur[1], random.randint(300, 600)]] + """轨迹生成已统一到 `tcaptcha/tdc_values.trajectory_points()`(保留本函数兼容旧调用点)。 - def dt(): - r = random.random() - if r < 0.50: return random.randint(0, 16) - if r < 0.75: return random.randint(17, 24) - if r < 0.90: return random.randint(25, 60) - if r < 0.98: return random.randint(61, 160) - return random.randint(161, 400) - - for tx, ty in targets: - seg = random.randint(14, 20) - for i in range(1, seg + 1): - nx = cur[0] + (tx - cur[0]) * i / seg + random.randint(-1, 1) - ny = cur[1] + (ty - cur[1]) * i / seg + random.randint(-1, 1) - events.append([int(nx) - cur[0], int(ny) - cur[1], dt()]) - cur = [int(nx), int(ny)] - for _ in range(random.randint(2, 5)): # 驻留 - events.append([random.randint(-1, 1), random.randint(-1, 1), dt()]) - return json.dumps(events, separators=(",", ":")) + 那里按真机样本把形状定死了:首点绝对坐标 + 增量点 + **固定收尾 `[1,1,12]`**。 + """ + from tcaptcha.tdc_values import trajectory_points + return json.dumps(trajectory_points(click_points, img_w=img_w), + separators=(",", ":")) def generate_collect_success( @@ -435,10 +470,12 @@ def generate_telemetry( ) -> Dict[str, str]: """统一遥测生成入口 —— **只走纯算**, 不再有任何伪造环境的执行后端。 - mode="python_keyed" (默认): Node 只做一件事 —— 从当前 tdc.js 提取本次构建的 - XTEA 密钥字 (密钥由 VM 运行时算出, 不是字面量); 明文组装 / 轨迹 / 时间戳 / - 加密 / 编码全部在 Python 内完成。Node 不可用时自动退回 DEFAULT_KEY。 - mode="python": 完全不调用 Node, 直接用内置 DEFAULT_KEY。 + mode="python_values" (推荐, 2026-10-12 新增): **纯离线**生成整颗 collect —— + tdc.js -> 37 个语义名(清单序)-> 38 槽取值 -> 4 段明文 -> 4 段加密。 + 零 Node、零真机明文(密钥也由纯 Python 静态提取)。见 tcaptcha/tdc_values.py。 + mode="python_keyed": 明文仍来自仓库里的**成功样本模板回填** + (generate_collect_success),Node 只负责从当前 tdc.js 提取 XTEA 密钥字。 + mode="python": 完全不调用 Node, 用内置 DEFAULT_KEY + 同样的模板回填。 返回 {"collect", "tlg", "eks", "source"[, "key"]}。 @@ -446,8 +483,15 @@ def generate_telemetry( 换 IP / 加长静默同样无效), 见 har/1ms_dianxuan/ANALYSIS.md §7。**不要再用"换环境"解释它**, 把精力放在账本 §7.5 的单变量实验 (登录态/真页面上下文) 上。 """ - if mode not in ("python_keyed", "python"): - raise ValueError(f"未知 telemetry mode: {mode!r} (只支持 'python_keyed' / 'python')") + if mode not in ("python_values", "python_keyed", "python"): + raise ValueError(f"未知 telemetry mode: {mode!r} " + "(只支持 'python_values' / 'python_keyed' / 'python')") + + if mode == "python_values": + from tcaptcha.tdc_values import generate_collect + out = generate_collect(tdc_js_code, click_points, referrer=referrer) + return {"collect": out["collect"], "tlg": out["tlg"], "eks": out["eks"], + "source": out["source"], "key": out["key"]} key: Optional[List[int]] = None source = "python" diff --git a/tcaptcha/tdc_values.py b/tcaptcha/tdc_values.py new file mode 100644 index 0000000..d932fce --- /dev/null +++ b/tcaptcha/tdc_values.py @@ -0,0 +1,389 @@ +# -*- coding: utf-8 -*- +"""tcaptcha/tdc_values.py —— 为每个采集器**语义名**提供取值生成器(纯离线)。 + +背景(账本 §五 第 0 项,har/1ms_dianxuan/PROGRESS.md): + 「纯离线生成 collect」的唯一阻塞点 = 不知道每个语义名该填什么值。 + 本文件把 2026-10-12 实测到的**同会话真值**固化成生成规则。 + +真值来源(不是猜的): + 真 Chrome + 真 1ms 页面 -> Debugger 取**同会话** tdc.js(77,954 字节, + 含 __TENCENT_CHAOS_VM)-> iframe 内 TDC.getData(true) -> 同会话 XTEA 密钥解密 + -> 按「require 序 == 槽位序」对齐,38/38 槽全覆盖。 + 脚本: scripts/verify/align_session.py、scripts/verify/audit_names.py + +会话 vs 每次采集(连续 4 次 getData 实测,这一条决定了生成器怎么分层): + + | 语义名 | 第1次 | 第2次 | 第3次 | 第4次 | 结论 | + |---------------------------|-------------|-------------|-------------|-------------|---------------| + | itoken | 641547489 | 641547489 | 641547489 | 641547489 | **会话内固定** | + | struct_40_9 (#.._rand) | 1591912482 | 1591912482 | 1591912482 | 1591912482 | **会话内固定** | + | window_time (#e48 委派) | 1791745224 | 1791745224 | 1791745224 | 1791745224 | **会话内固定** | + | window_time (#e40) | 1791745331 | 1791745548 | 1791745813 | 1791745814 | 每次刷新 | + | struct_46_9 | 1 | 2 | 3 | 4 | **每次 +1 的计数器** | + | trajectory 点数 | 1 | 61 | 1 | 1 | 取完即清空 | + +三类取值: + * 静态画像 DEVICE_PROFILE —— UA / 屏幕 / WebGL / Canvas / 语言 / 时区 / 核数 … + (已用浏览器真值交叉校验: screen.width/height/availHeight/colorDepth、 + hardwareConcurrency、timezoneOffset、languages 全部一一吻合) + * 会话量 Session —— itoken / 会话随机盐 / 页面加载时刻 + * 每次采集 Session.next_call() —— 当前时刻、采集计数器、轨迹 +""" +from __future__ import annotations + +import random +import time +from typing import Any, Dict, List, Optional, Sequence + +# --------------------------------------------------------------------------- # +# 设备画像:与真机逐项核对过(见模块 docstring 的交叉校验说明) # +# --------------------------------------------------------------------------- # +DEVICE_PROFILE: Dict[str, Any] = { + "name": "Mac / Chrome 155 / 1440x2560", + "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"), + "platform": "MacIntel", + "languages": ["zh-CN", "zh"], + "timezone_offset": "+08", # 真机 Date.getTimezoneOffset() == -480 + "charset": "UTF-8", + "hardware_concurrency": 10, # 真机 navigator.hardwareConcurrency + # ⚠️ 方向别弄反:真机 screen.width == 1440、screen.height == 2560 + # (1440x2560 竖屏;浏览器实测 screen.availHeight == 2530) + "screen_width": 1440, + "screen_height": 2560, + "screen_color_depth": 24, + # 真机 screen_fingerprint 串;字段序 = 宽-高-可用高-色深-*-*-|-* + "screen_fingerprint": "1440-2560-2530-24-*-*-|-*", + "webgl_vendor": "Google Inc. (Apple)", + "webgl_renderer": "ANGLE (Apple, ANGLE Metal Renderer: Apple M4, Unspecified Version)", + "canvas_fingerprint": ("GgoAAAANSUhEUgAAASwAAACWCAYAAATf3v8bh+ryWcfSBSLkmN3N1" + "PQinPL7Bi3TG0qE2CogAAAABJRU5ErkJggg=="), + "viewport_size": [360, 360], # 验证码 iframe 的可视区 + "cookie_enabled": 1, + "touch_detect": 2, # 真机实测值(该机 maxTouchPoints == 0) + "iframe_detect": "iframe", + "webrtc_ip": "", # 未取到候选时为空串 + "dom_compliance": 1023, + "constant_98k": "98k", + "template_url": ("https://turing.captcha.gtimg.com/1/template/" + "drag_ele.html?rand=1519713624347"), +} + +# 真机里恒为这些常量的槽(连续 4 次抽样一致) +CONSTANTS: Dict[str, Any] = { + "time_since_load": 0, + "webdriver_detect": 0, + "console_detect": 0, + "fp_salt": 0, + "struct_48_9_call": 0, + "location_href": "", + "struct_40_9#empty": "", +} + +# 真机里是空数组的槽(与轨迹区分:轨迹至少含一个点) +EMPTY_LIST_NAME = "struct_76_33" + +DEFAULT_REFERRER = "https://1ms.run/console/personal/benefits/check-in" + + +def _rand_int32() -> int: + """[0, 2^31) 的随机整数 —— 真机观测值 1591912482 / 699129805 / 641547489 都落在这里。""" + return random.randint(0, (1 << 31) - 1) + + +def default_trajectory() -> List[List[int]]: + """无鼠标输入时的轨迹(真机实测就是 [[1,1,12]],看着像个"无位移"哨兵点)。""" + return [[1, 1, 12]] + + +def trajectory_points(click_points: Sequence[Sequence[int]], + img_w: int = 672) -> List[List[int]]: + """按点击点生成鼠标轨迹:首点绝对坐标 + 增量点 + 固定收尾 `[1,1,12]`。 + + 形状与真机一致(3 份样本逐字核对): + * 首点是**绝对坐标**(含 300~600ms 初始停顿) + * 之后全是**增量** `[dx,dy,dt]` + * **最后一点固定是 `[1,1,12]`** —— 真机成功样本与"没动鼠标"样本都这么收尾 + (没动鼠标时整条轨迹就是 `[[1,1,12]]`) + dt 分布按成功样本实测拟合(旧 crypto._success_style_trajectory 的统计)。 + """ + scale = 320.0 / img_w + targets = [(int(round(20 + x * scale)), int(round(60 + y * scale))) + for x, y in click_points] + cur = [random.randint(140, 220), random.randint(300, 340)] + events: List[List[int]] = [[cur[0], cur[1], random.randint(300, 600)]] + + def dt() -> int: + r = random.random() + if r < 0.50: + return random.randint(0, 16) + if r < 0.75: + return random.randint(17, 24) + if r < 0.90: + return random.randint(25, 60) + if r < 0.98: + return random.randint(61, 160) + return random.randint(161, 400) + + for tx, ty in targets: + seg = random.randint(14, 20) + for i in range(1, seg + 1): + nx = cur[0] + (tx - cur[0]) * i / seg + random.randint(-1, 1) + ny = cur[1] + (ty - cur[1]) * i / seg + random.randint(-1, 1) + events.append([int(nx) - cur[0], int(ny) - cur[1], dt()]) + cur = [int(nx), int(ny)] + for _ in range(random.randint(2, 5)): # 命中后短暂驻留 + events.append([random.randint(-1, 1), random.randint(-1, 1), dt()]) + events.append([1, 1, 12]) # 固定收尾(真机如此) + return events + + +def rows_from_source(js_text: str) -> List[Dict]: + """分析 tdc.js **源码**(tdc_modules 的入口收的是路径,这里落临时文件再分析)。""" + import os + import tempfile + + fd, path = tempfile.mkstemp(suffix=".js", prefix="tdc_") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(js_text) + return _analyze(path) + finally: + try: + os.unlink(path) + except OSError: + pass + + +def _analyze(path: str) -> List[Dict]: + """惰性导入 tools/tdc_modules(它不在包里,且只做静态分析)。""" + import sys + from pathlib import Path as _P + + tools = str(_P(__file__).resolve().parents[1] / "tools") + if tools not in sys.path: + sys.path.insert(0, tools) + import tdc_modules as tm + return tm.analyze(path) + + +def generate_collect(tdc_js_code: str, + click_points: Optional[Sequence[Sequence[int]]] = None, + *, + key: Optional[Sequence[int]] = None, + sd: Optional[Dict] = None, + session: Optional[Session] = None, + img_w: int = 672, + referrer: Optional[str] = None) -> Dict[str, Any]: + """**纯离线**生成一颗 collect:只吃 tdc.js 源码,不需要任何真机明文。 + + 流程: tdc.js -> 37 个语义名(清单序)-> 38 槽取值 -> 4 段明文 -> 4 段加密拼接。 + 密钥默认用纯 Python 静态提取(`tcaptcha.tdc_key`),零 Node。 + """ + from tcaptcha.crypto import build_collect, extract_eks # 局部导入避免环形依赖 + from tcaptcha.tdc_key import extract_xtea_key_pure + + rows = rows_from_source(tdc_js_code) + if key is None: + key = extract_xtea_key_pure(tdc_js_code)["key"] + sess = session or Session(referrer=referrer or DEFAULT_REFERRER) + if click_points: + sess.set_trajectory(trajectory_points(click_points, img_w=img_w)) + + cd = generate_cd(rows, sess) + plain = build_plain(cd, rows, sd) + collect = build_collect(plain["chunk1"], plain["trajectory"], + plain["chunk2"], plain["sd"], list(key)) + return { + "collect": collect, + "tlg": str(len(collect)), + "eks": extract_eks(tdc_js_code), + "key": list(key), + "source": "python_values", + "cd": cd, + "segments": {k: len(v) for k, v in plain.items()}, + } + + +class Session: + """一次验证码会话的状态。 + + 会话内固定的量(itoken / 随机盐 / 页面加载时刻)在 __init__ 里定下来, + 每次采集会变的量(当前时刻 / 计数器 / 轨迹)由 next_call() 推进。 + """ + + def __init__(self, now: Optional[float] = None, entry_age: int = 60, + referrer: str = DEFAULT_REFERRER, profile: Optional[Dict] = None): + self.t0 = int(now if now is not None else time.time()) + # 页面加载时刻先于会话建立(真机两次采样差 ~107 秒) + self.loaded_at = self.t0 - entry_age + self.itoken = [_rand_int32(), self.loaded_at] # 会话内固定,实测 4 次完全一致 + self.session_rand = _rand_int32() # struct_40_9 的"随机"槽 + self.referrer = referrer + self.profile = dict(profile or DEVICE_PROFILE) + self.calls = 0 + self.now = self.t0 + self.trajectory: List[List[int]] = default_trajectory() + + def next_call(self, now: Optional[float] = None) -> "Session": + """进入下一次 TDC.getData():计数器 +1、刷新当前时刻、轨迹复位由调用方决定。""" + self.calls += 1 + self.now = int(now if now is not None else time.time()) + return self + + def set_trajectory(self, points: Sequence[Sequence[int]]) -> "Session": + self.trajectory = [list(p) for p in points] + return self + + +# --------------------------------------------------------------------------- # +# 语义名 -> 取值生成器 # +# --------------------------------------------------------------------------- # +def value_for(base: str, row: Dict, sess: Session) -> Any: + """按**基础语义名**(去掉 #结构后缀)产出一个槽的值。 + + 同一基础名在一个构建里可能出现多次(user_agent×2 / window_time×2 / + struct_40_9×2),这时用行上的**结构证据**分流,而不是靠出现次序 —— + 次序会随构建变,结构不会(见 tdc_modules.STRUCT_RULES)。 + """ + st = row.get("struct") or {} + p = sess.profile + + # ---- 同组多成员:用结构 / 委派证据分流 ---- + if base == "user_agent": + # 直采的那个返回 UA 串;走委派链的那个返回 0(真机实测) + return 0 if row.get("via_delegate") else p["user_agent"] + if base == "struct_40_9": + # 闭包最后写入 "" vs 随机 int32(账本 #47 的静态消歧依据) + return "" if st.get("closure_kind") == "empty" else sess.session_rand + if base == "window_time": + # 委派的那个是页面加载时刻(会话内固定),直采的那个是当前时刻(每次变) + return sess.loaded_at if row.get("via_delegate") else sess.now + + # ---- 会话量 / 每次采集量 ---- + if base == "itoken": + return None # 2 槽,见 generate_cd + if base == "trajectory": + return sess.trajectory + if base == "struct_46_9": + return sess.calls # 实测 1,2,3,4 —— 每次采集 +1 + if base == "referrer": + return sess.referrer + + # ---- 常量 / 画像 ---- + if base in CONSTANTS: + return CONSTANTS[base] + if base == EMPTY_LIST_NAME: + return [] + if base in p: + return p[base] + raise KeyError("没有 %r 的取值规则(新语义名? 需要补规则)" % base) + + +def generate_cd(rows: Sequence[Dict], sess: Optional[Session] = None, + now: Optional[float] = None) -> List[Any]: + """按**清单顺序**生成 cd 数组(37 项 -> 38 槽,itoken 占 2 槽)。 + + rows 来自 tools.tdc_modules.analyze()(每项含 name / struct / via_delegate)。 + 每次调用会推进 sess.next_call()(模拟一次 TDC.getData)。 + """ + s = sess or Session(now=now) + if now is not None: + s.next_call(now) + elif s.calls == 0: + s.next_call(s.t0) + + out: List[Any] = [] + for r in rows: + base = str(r["name"]).split("#")[0] + if base == "itoken": + out.extend(s.itoken) # 真机就是 2 个连续槽 + continue + out.append(value_for(base, r, s)) + return out + + +def trajectory_index(rows: Sequence[Dict]) -> int: + """轨迹在 cd 中的下标 —— 它就是 4 段加密里第 2 段的分界点。""" + k = 0 + for r in rows: + base = str(r["name"]).split("#")[0] + if base == "trajectory": + return k + k += 2 if base == "itoken" else 1 + raise KeyError("清单里没有 trajectory 项") + + +def build_plain(cd: Sequence[Any], rows: Sequence[Dict], + sd: Optional[Dict] = None) -> Dict[str, str]: + """把 cd 组装成 4 段明文(chunk1 / trajectory / chunk2 / sd)。 + + 与真机**逐字节一致**的分段写法(3 份样本核对,2026-10-12): + + chunk1 = '{"cd":[' + 轨迹之前的槽(逗号连接)+ ',' + traj = '<轨迹槽的 JSON>' + chunk2 = ',' + 轨迹之后的槽(逗号连接)+ '],' + sd = '"sd":' + json(sd) + '}' + + 分隔符一律是**无空格的逗号**(真机 JSON.stringify 风格)。 + """ + import json + + def js(v: Any) -> str: + return json.dumps(v, ensure_ascii=False, separators=(",", ":")) + + k = trajectory_index(rows) + head = ",".join(js(v) for v in cd[:k]) + tail = ",".join(js(v) for v in cd[k + 1:]) + return { + "chunk1": '{"cd":[' + head + ",", + "trajectory": js(cd[k]), + "chunk2": "," + tail + "],", + "sd": '"sd":' + js(sd if sd is not None else {"od": "C"}) + "}", + } + + +def main() -> int: + import argparse + import json + import sys + from pathlib import Path + + ROOT = Path(__file__).resolve().parents[1] + sys.path.insert(0, str(ROOT / "tools")) + + ap = argparse.ArgumentParser(description="按语义名生成 cd 取值(纯离线)") + ap.add_argument("tdc_js", help="一份 tdc.js(用来取清单顺序与语义名)") + ap.add_argument("--json", default=None) + args = ap.parse_args() + + import tdc_modules as tm + rows = tm.analyze(args.tdc_js) + sess = Session() + cd = generate_cd(rows, sess) + print("项 语义名 取值") + print("-" * 78) + k = 0 + for r in rows: + base = str(r["name"]).split("#")[0] + n = 2 if base == "itoken" else 1 + val = cd[k:k + n] if n > 1 else cd[k] + k += n + print("%-3d %-36s %s" % (r["item"], r["name"], + json.dumps(val, ensure_ascii=False)[:34])) + print() + print("槽数 %d / %d 项" % (len(cd), len(rows))) + plain = build_plain(cd, rows) + print("分段: chunk1=%d traj=%d chunk2=%d sd=%d" + % (len(plain["chunk1"]), len(plain["trajectory"]), + len(plain["chunk2"]), len(plain["sd"]))) + if args.json: + Path(args.json).write_text(json.dumps(cd, ensure_ascii=False, indent=1), + encoding="utf-8") + print(">>> %s" % args.json) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tcaptcha/tests/test_crypto_core.py b/tcaptcha/tests/test_crypto_core.py index 4b08735..b30ba9e 100644 --- a/tcaptcha/tests/test_crypto_core.py +++ b/tcaptcha/tests/test_crypto_core.py @@ -5,25 +5,141 @@ 固化: 1. 教科书 XTEA 加解密往返 (含 8 字节对齐补齐) 2. encrypt_string 与 xtea_encrypt_bytes 结果一致 (防止两份 block 实现再次漂移) + 3. **四段切分**的实测回归:真机存在「pad 只有 1 个空格」的样本, + 旧的 ` {2,}` 判据会直接崩(2026-10-12 踩坑,账本 #50/#51) + 4. **纯离线闭环**:tdc.js -> 语义名 -> 槽取值 -> 4 段明文 -> 加密 -> 解回 运行: .venv/bin/python -m tcaptcha.tests.test_crypto_core """ from __future__ import annotations +import base64 +import json +import re 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)) +if str(ROOT / "tools") not in sys.path: + sys.path.insert(0, str(ROOT / "tools")) -from tcaptcha.crypto import selfcheck # noqa: E402 +from tcaptcha.crypto import (build_collect, selfcheck, split_collect_segments, # noqa: E402 + xtea_decrypt_bytes) + +# 真机同会话 collect 的**解密明文**(2026-10-12,1ms aid=195672344 的验证码 iframe)。 +# 它的价值在于:chunk1 = 287 字节,287 % 24 == 23 -> **pad 只有 1 个空格**, +# 而那个空格又恰好满足 `b%24==0 且 (b-a)==24-a%24` —— 两个坑它一次占全。 +REAL_PLAIN_B64 = ( + "eyJjZCI6WzI1NjAsMCw2NDE1NDc0ODksMTc5MTczMjAwMSwiTW96aWxsYS81LjAgKE1hY2ludG9z" + "aDsgSW50ZWwgTWFjIE9TIFggMTBfMTVfNykgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlr" + "ZSBHZWNrbykgQ2hyb21lLzE1NS4wLjAuMCBTYWZhcmkvNTM3LjM2IiwwLCJVVEYtOCIsWzM2MCwz" + "NjBdLDAsIk1hY0ludGVsIixbInpoLUNOIiwiemgiXSwxMCwiMTQ0MC0yNTYwLTI1MzAtMjQtKi0q" + "LXwtKiIsIiIsW10sMTc5MTc0NTU0OCwyLCIrMDgiLCJHb29nbGUgSW5jLiAoQXBwbGUpIiwwLDAs" + "MCwgW1sxNCwxMSwzMjI5NzFdLFszLDAsN10sWzQsMSw3XSxbNSwtMSw3XSxbNiwwLDddLFsyLDEs" + "N10sWzMsLTEsNl0sWzQsMCw3XSxbNSwxLDddLFs2LC0xLDddLFsyLDAsN10sWzMsMSw3XSxbNCwt" + "MSw2XSxbNSwwLDddLFs2LDEsN10sWzIsLTEsN10sWzMsMCw3XSxbNCwxLDddLFs1LC0xLDddLFs2" + "LDAsN10sWzIsMSw3XSxbMywtMSw3XSxbNCwwLDddLFs1LDEsN10sWzYsLTEsN10sWzIsMCw2XSxb" + "MywxLDddLFs0LC0xLDddLFs1LDAsN10sWzYsMSw3XSxbMiwtMSw3XSxbMywwLDddLFs0LDEsNl0s" + "WzUsLTEsN10sWzYsMCw3XSxbMiwxLDddLFszLC0xLDddLFs0LDAsN10sWzUsMSw3XSxbNiwtMSw3" + "XSxbMiwwLDddLFszLDEsN10sWzQsLTEsNl0sWzUsMCw3XSxbNiwxLDddLFsyLC0xLDddLFszLDAs" + "Nl0sWzQsMSw3XSxbNSwtMSw3XSxbNiwwLDddLFsyLDEsN10sWzMsLTEsN10sWzQsMCw3XSxbNSwx" + "LDddLFs2LC0xLDddLFsyLDAsN10sWzMsMSw2XSxbNCwtMSw4XSxbNSwwLDddLFs2LDEsN10sWzEs" + "MSwxMl1dICAgICAgICAgICAgLCIiLDEwMjMsMTU5MTkxMjQ4MiwyLCJodHRwczovLzFtcy5ydW4v" + "Y29uc29sZS9wZXJzb25hbC9iZW5lZml0cy9jaGVjay1pbiIsIkdnb0FBQUFOU1VoRVVnQUFBU3dB" + "QUFDV0NBWUFBQVRmM3Y4YmgrcnlXY2ZTQlNMa21OM04xUFFpblBMN0JpM1RHMHFFMkNvZ0FBQUFC" + "SlJVNUVya0pnZ2c9PSIsMSwiQU5HTEUgKEFwcGxlLCBBTkdMRSBNZXRhbCBSZW5kZXJlcjogQXBw" + "bGUgTTQsIFVuc3BlY2lmaWVkIFZlcnNpb24pIiwxNzkxNzQ1MjI0LCJpZnJhbWUiLDE0NDAsIiIs" + "Imh0dHBzOi8vdHVyaW5nLmNhcHRjaGEuZ3RpbWcuY29tLzEvdGVtcGxhdGUvZHJhZ19lbGUuaHRt" + "bD9yYW5kPTE1MTk3MTM2MjQzNDciLDI0LCI5OGsiXSwgICAgInNkIjp7Im9kIjoiQyJ9fQ==" +) + + +def test_split_single_space_pad() -> None: + """真机明文:chunk1 的 pad 只有 1 个空格,旧判据会漏。""" + plain = base64.b64decode(REAL_PLAIN_B64).decode("latin1") + assert len(plain) == 1192 + + # 旧判据(` {2,}`)只看得见 2 个 pad -> 切不出 4 段(这就是被修掉的 bug) + assert len(re.findall(r" {2,}", plain)) == 2 + + c1, traj, c2, sd = split_collect_segments(plain) + assert (len(c1), len(traj), len(c2), len(sd)) == (287, 516, 356, 16) + assert c1.startswith('{"cd":[') and c1.endswith(",") + assert c2.startswith(",") and c2.endswith("],") + assert sd == '"sd":{"od":"C"}}' + + obj = json.loads(c1 + traj + c2 + sd) + assert len(obj["cd"]) == 38 + points = json.loads(traj) + assert len(points) == 61 and points[0] == [14, 11, 322971] + assert points[-1] == [1, 1, 12] + + +def test_split_ignores_space_inside_string() -> None: + """UA 串里的普通空格在偏移 47 处也满足 pad 模算术,不能被当成 pad。""" + plain = base64.b64decode(REAL_PLAIN_B64).decode("latin1") + assert plain[47] == " " and 47 % 24 == 23 and 48 % 24 == 0 # 它确实满足判据 + c1, _traj, _c2, _sd = split_collect_segments(plain) + assert '"Mozilla/5.0 (Macintosh' in c1 # 空格仍在字符串里,没被切掉 + + +def test_build_collect_roundtrip_single_space_pad() -> None: + """自己造一份 pad==1 的样本,验证 加密 -> 切分 可逆。""" + key = [0x11111111, 0x22222222, 0x33333333, 0x44444444] + chunk1 = '{"cd":[' + "1," * 20 # 7 + 40 = 47, 47 % 24 == 23 -> pad 1 + traj = "[[1,1,12]]" + chunk2 = ',"x"],' + sd = '"sd":{"od":"C"}}' + assert len(chunk1) % 24 == 23 + + collect = build_collect(chunk1, traj, chunk2, sd, key) + plain = xtea_decrypt_bytes(base64.b64decode(collect), key).decode("latin1").rstrip("\x00") + + assert plain[47:48] == " " # 第一段的 pad 就是一个空格 + assert split_collect_segments(plain) == (chunk1, traj, chunk2, sd) + + +def test_offline_generate_and_roundtrip() -> None: + """纯离线:tdc.js -> 37 语义名 -> 38 槽 -> 4 段明文 -> 加密 -> 解回逐槽相等。""" + import tdc_modules as tm + from tcaptcha.tdc_key import extract_xtea_key_pure + from tcaptcha.tdc_values import Session, build_plain, generate_cd, trajectory_index + + build = ROOT / "tcaptcha" / "tdc_sample.js" + js = build.read_text(encoding="utf-8", errors="replace") + key = extract_xtea_key_pure(js)["key"] + + rows = tm.analyze(str(build)) + assert len(rows) == 37 + assert not [r for r in rows if r["name"] == "unknown"] + + cd = generate_cd(rows, Session()) + assert len(cd) == 38 # itoken 占 2 槽 + plain = build_plain(cd, rows) + k = trajectory_index(rows) + assert cd[k] == [[1, 1, 12]] # 轨迹槽 + + collect = build_collect(plain["chunk1"], plain["trajectory"], + plain["chunk2"], plain["sd"], key) + dec = xtea_decrypt_bytes(base64.b64decode(collect), key).decode("latin1").rstrip("\x00") + c1, traj, c2, sd = split_collect_segments(dec) + assert json.loads(c1 + traj + c2 + sd)["cd"] == cd + assert (len(c1), len(traj), len(c2), len(sd)) == \ + (len(plain["chunk1"]), len(plain["trajectory"]), + len(plain["chunk2"]), len(plain["sd"])) def main() -> int: selfcheck() - print("✅ crypto 纯算法自检通过 (XTEA 往返 + encrypt_string 一致性)") + test_split_single_space_pad() + test_split_ignores_space_inside_string() + test_build_collect_roundtrip_single_space_pad() + test_offline_generate_and_roundtrip() + print("✅ crypto 纯算法自检通过 (XTEA 往返 + encrypt_string 一致性" + " + 四段切分回归 + 纯离线闭环)") return 0 diff --git a/tools/cdp_collect.js b/tools/cdp_collect.js new file mode 100644 index 0000000..864eff1 --- /dev/null +++ b/tools/cdp_collect.js @@ -0,0 +1,105 @@ +#!/usr/bin/env node +/** + * cdp_collect.js —— 在**正在运行**的真 Chrome 里,对验证码 iframe 求 TDC.getData(true), + * 并把结果**直接落盘**(不经 stdout),避免终端输出的脱敏 / 截断。 + * + * 为什么要落盘: cdp_eval.js 把结果打到 stdout,链路上有脱敏风险; + * 而「同会话 build + 同会话 collect」必须逐字节保真才可用于对齐,故这里由 Node 直接写文件。 + * + * 用法: + * node tools/cdp_collect.js --match drag_ele --out /tmp/1ms_bridge/session_collect.json + * node tools/cdp_collect.js --match drag_ele --expr /tmp/my_expr.js --out /tmp/x.json + */ +const fs = require('fs'); +const crypto = require('crypto'); + +function arg(name, def) { + const i = process.argv.indexOf('--' + name); + if (i < 0) return def; + const v = process.argv[i + 1]; + return (v === undefined || v.startsWith('--')) ? true : v; +} +const sleep = ms => new Promise(r => setTimeout(r, ms)); + +class Cdp { + constructor(ws) { this.ws = ws; this.id = 0; this.waits = new Map(); this.handlers = []; } + static async connect(url) { + const ws = new WebSocket(url); + await new Promise((res, rej) => { ws.onopen = res; ws.onerror = rej; }); + const c = new Cdp(ws); + ws.onmessage = ev => { + const m = JSON.parse(ev.data); + if (m.id && c.waits.has(m.id)) { + const { res, rej } = c.waits.get(m.id); + c.waits.delete(m.id); + m.error ? rej(new Error(JSON.stringify(m.error))) : res(m.result); + } else if (m.method) { for (const h of c.handlers) h(m); } + }; + return c; + } + send(method, params = {}) { + const id = ++this.id; + this.ws.send(JSON.stringify({ id, method, params })); + return new Promise((res, rej) => this.waits.set(id, { res, rej })); + } + on(fn) { this.handlers.push(fn); } + close() { try { this.ws.close(); } catch (e) {} } +} + +const DEFAULT_EXPR = `(async () => { + const o = { href: location.href, tdcType: typeof window.TDC }; + try { o.tdcKeys = Object.keys(window.TDC || {}); } catch (e) { o.tdcKeysErr = String(e); } + try { o.collect = window.TDC.getData(true); } catch (e) { o.collectErr = String(e); } + try { o.info = window.TDC.getInfo(); } catch (e) { o.infoErr = String(e); } + return o; +})()`; + +(async () => { + const port = parseInt(arg('port', '9222'), 10); + const match = arg('match', 'drag_ele'); + const wait = parseInt(arg('wait', '800'), 10); + const exprFile = arg('expr', null); + const out = arg('out', '/tmp/1ms_bridge/session_collect.json'); + + const expr = (exprFile && exprFile !== true) ? fs.readFileSync(exprFile, 'utf8') : DEFAULT_EXPR; + + const targets = (await (await fetch(`http://127.0.0.1:${port}/json/list`)).json()) + .filter(t => t.type === 'page' || t.type === 'iframe'); + const target = targets.find(t => (t.url || '').includes(match)); + if (!target) throw new Error('没有匹配的 target: ' + match); + console.error(`[target] ${target.type} ${target.url.slice(0, 100)}`); + + const cdp = await Cdp.connect(target.webSocketDebuggerUrl); + await cdp.send('Runtime.enable'); + await sleep(wait); + + const res = await cdp.send('Runtime.evaluate', { + expression: expr, awaitPromise: true, returnByValue: true, timeout: 30000, + }); + const val = res.result ? res.result.value : null; + const exc = res.exceptionDetails + ? ((res.exceptionDetails.exception || {}).description || JSON.stringify(res.exceptionDetails)) + : null; + + const payload = { + fetchedAt: new Date().toISOString(), + target: { type: target.type, url: target.url, title: target.title }, + ok: !res.exceptionDetails, + exception: exc, + data: val, + }; + fs.mkdirSync(require('path').dirname(out), { recursive: true }); + fs.writeFileSync(out, JSON.stringify(payload, null, 1), 'utf8'); + + const d = val || {}; + const col = typeof d.collect === 'string' ? d.collect : null; + console.error(`[result] ok=${payload.ok} tdcType=${d.tdcType} tdcKeys=${JSON.stringify(d.tdcKeys || null)}`); + if (col) { + const sha = crypto.createHash('sha256').update(col).digest('hex').slice(0, 16); + console.error(`[collect] ${col.length} 字符 sha256=${sha} 头=${col.slice(0, 24)}... 尾=${col.slice(-24)}`); + } else { + console.error(`[collect] 无 (collectErr=${d.collectErr || '-'})`); + } + console.error(`-> ${out}`); + cdp.close(); +})().catch(e => { console.error('ERR', e.message); process.exit(1); }); diff --git a/tools/cdp_dump_scripts.js b/tools/cdp_dump_scripts.js new file mode 100644 index 0000000..fee7794 --- /dev/null +++ b/tools/cdp_dump_scripts.js @@ -0,0 +1,103 @@ +#!/usr/bin/env node +/** + * cdp_dump_scripts.js —— 把某个 target 里**所有已解析脚本**逐个导出并标注特征。 + * + * 为什么需要(2026-10-12 踩坑): + * cdp_script_source.js 取的是 `hits[hits.length-1]`(最后一个匹配), + * 在验证码 iframe(同 target 里有 dy-jy.js / dy-ele.js / tdc.js 多个脚本)里 + * **可能取错**。有一次取到的 205KB 文件既没有 __TENCENT_CHAOS_VM 也没有 delta, + * 根本不是 tdc.js。 + * + * 本工具: 一律全量导出 + 对每个脚本标注是否含 VM 锚点 / delta / 长度,绝不猜。 + * + * 用法: + * node tools/cdp_dump_scripts.js --match drag_ele --outdir /tmp/1ms_bridge/scripts + */ +const fs = require('fs'); +const path = require('path'); + +function arg(name, def) { + const i = process.argv.indexOf('--' + name); + if (i < 0) return def; + const v = process.argv[i + 1]; + return (v === undefined || v.startsWith('--')) ? true : v; +} +const sleep = ms => new Promise(r => setTimeout(r, ms)); + +class Cdp { + constructor(ws) { this.ws = ws; this.id = 0; this.waits = new Map(); this.handlers = []; } + static async connect(url) { + const ws = new WebSocket(url); + await new Promise((res, rej) => { ws.onopen = res; ws.onerror = rej; }); + const c = new Cdp(ws); + ws.onmessage = ev => { + const m = JSON.parse(ev.data); + if (m.id && c.waits.has(m.id)) { + const { res, rej } = c.waits.get(m.id); + c.waits.delete(m.id); + m.error ? rej(new Error(JSON.stringify(m.error))) : res(m.result); + } else if (m.method) { for (const h of c.handlers) h(m); } + }; + return c; + } + send(method, params = {}) { + const id = ++this.id; + this.ws.send(JSON.stringify({ id, method, params })); + return new Promise((res, rej) => this.waits.set(id, { res, rej })); + } + on(fn) { this.handlers.push(fn); } + close() { try { this.ws.close(); } catch (e) {} } +} + +(async () => { + const port = parseInt(arg('port', '9222'), 10); + const match = arg('match', null); + const outdir = arg('outdir', '/tmp/cdp_scripts'); + const wait = parseInt(arg('wait', '2000'), 10); + + const targets = (await (await fetch(`http://127.0.0.1:${port}/json/list`)).json()) + .filter(t => t.type === 'page' || t.type === 'iframe'); + const target = match ? targets.find(t => (t.url || '').includes(match)) : targets[0]; + if (!target) throw new Error('没有匹配的 target: ' + match); + console.error(`[target] ${target.type} ${target.url.slice(0, 90)}`); + + const cdp = await Cdp.connect(target.webSocketDebuggerUrl); + const scripts = []; + cdp.on(m => { if (m.method === 'Debugger.scriptParsed') scripts.push(m.params); }); + await cdp.send('Debugger.enable'); + await sleep(wait); + await cdp.send('Runtime.enable'); + await cdp.send('Runtime.evaluate', { expression: '1' }); + await sleep(500); + + fs.mkdirSync(outdir, { recursive: true }); + const index = []; + const seen = new Set(); + for (const s of scripts) { + if (seen.has(s.scriptId)) continue; + seen.add(s.scriptId); + let src = ''; + try { + const r = await cdp.send('Debugger.getScriptSource', { scriptId: s.scriptId }); + src = r.scriptSource || ''; + } catch (e) { continue; } + if (!src) continue; + const hasVM = src.includes('__TENCENT_CHAOS_VM'); + const hasDelta = src.includes('2654435769'); + const hasTDCNAME = src.includes('TDC_NAME'); + const base = (s.url || 'inline').split('?')[0].split('/').pop() || 'inline'; + const fn = `${String(s.scriptId).replace(/[^\w.-]/g, '_')}_${base}`.slice(0, 80); + fs.writeFileSync(path.join(outdir, fn), src, 'utf8'); + index.push({ + scriptId: s.scriptId, url: String(s.url || '(inline)').slice(0, 130), + bytes: src.length, hasVM, hasDelta, hasTDCNAME, file: fn, + }); + } + fs.writeFileSync(path.join(outdir, 'index.json'), JSON.stringify(index, null, 1), 'utf8'); + for (const r of index) { + console.log(`${r.hasVM ? 'VM ' : ' '}${r.hasDelta ? 'DELTA' : ' '}` + + ` ${String(r.bytes).padStart(7)}B ${r.url}`); + } + console.error(`\n-> ${outdir} (共 ${index.length} 个脚本, 含 VM 的 ${index.filter(x => x.hasVM).length} 个)`); + cdp.close(); +})().catch(e => { console.error('ERR', e.message); process.exit(1); }); diff --git a/tools/cdp_module_probe.js b/tools/cdp_module_probe.js new file mode 100644 index 0000000..af63bc7 --- /dev/null +++ b/tools/cdp_module_probe.js @@ -0,0 +1,178 @@ +#!/usr/bin/env node +/** + * cdp_module_probe.js —— 在真 Chrome 里**抓到每个采集器的真实取值**(不需要手工配平)。 + * + * 背景(2026-10-12): + * 腾讯验证码 iframe (captcha.gtimg.com) 与页面 (1ms.run) **不同站**,Chrome 站点隔离下 + * 它是独立 target —— 直接在页面里包 Function.prototype.apply **进不去**(实测: + * 钩子装上时模块系统已经初始化完,__mods 恒为空)。 + * + * 解法(CDP 标准做法): + * Target.setAutoAttach(autoAttach, waitForDebuggerOnStart, flatten) + * -> 新的 iframe target 一创建就**暂停在任何脚本执行之前**; + * -> 此时 Page.addScriptToEvaluateOnNewDocument 注册钩子,再 runIfWaitingForDebugger 放行; + * -> tdc.js 执行时钩子已就位,模块工厂的 require/get 全被记账。 + * + * 产出: {url, mods: [[module_id, get()返回值], ...], reqs: [module_id,...]} + * 配合同一会话的 tdc.js 静态分析(tdc_modules.analyze)即可得到「语义名 -> 真实取值」, + * 无需任何人工对齐。 + * + * 用法: + * node tools/cdp_module_probe.js --aid 195672344 --out /tmp/1ms_bridge/modvals.json + */ +const fs = require('fs'); + +function arg(name, def) { + const i = process.argv.indexOf('--' + name); + if (i < 0) return def; + const v = process.argv[i + 1]; + return (v === undefined || v.startsWith('--')) ? true : v; +} +const sleep = ms => new Promise(r => setTimeout(r, ms)); + +class Cdp { + constructor(ws) { this.ws = ws; this.id = 0; this.waits = new Map(); this.handlers = []; } + static async connect(url) { + const ws = new WebSocket(url); + await new Promise((res, rej) => { ws.onopen = res; ws.onerror = rej; }); + const c = new Cdp(ws); + ws.onmessage = ev => { + const m = JSON.parse(ev.data); + if (m.id && c.waits.has(m.id)) { + const { res, rej } = c.waits.get(m.id); + c.waits.delete(m.id); + m.error ? rej(new Error(JSON.stringify(m.error))) : res(m.result); + } else if (m.method) { for (const h of c.handlers) h(m); } + }; + return c; + } + send(method, params = {}, sessionId) { + const id = ++this.id; + const msg = { id, method, params }; + if (sessionId) msg.sessionId = sessionId; + this.ws.send(JSON.stringify(msg)); + return new Promise((res, rej) => this.waits.set(id, { res, rej })); + } + on(fn) { this.handlers.push(fn); } + close() { try { this.ws.close(); } catch (e) {} } +} + +// 在任何页面脚本之前注入的钩子:记账「模块工厂 require 了谁」+「每个模块 .get() 返回什么」 +const HOOK = `(() => { + if (window.__mp) return; + window.__mp = true; + window.__mods = []; + window.__reqs = []; + const oa = Function.prototype.apply; + Function.prototype.apply = function (t, a) { + try { + if (a && a.length >= 3 && typeof a[a.length - 1] === 'function') { + const req = a[a.length - 1]; + if (typeof req === 'function' && !req.__w) { + const w = function (id) { + try { window.__reqs.push(String(id)); } catch (e) {} + const m = req.apply(this, arguments); + try { + if (m && typeof m.get === 'function' && !m.__g) { + m.__g = true; + const og = m.get; + m.get = function () { + let v; + try { v = og.apply(this, arguments); } catch (e) { v = '__ERR__:' + e; } + try { window.__mods.push([String(id), v]); } catch (e) {} + return v; + }; + } + } catch (e) {} + return m; + }; + w.__w = true; + const na = Array.prototype.slice.call(a); + na[na.length - 1] = w; + return oa.call(this, t, na); + } + } + } catch (e) {} + return oa.call(this, t, a); + }; +})();`; + +const SHOW = aid => `(async () => { + const load = (src) => new Promise((res, rej) => { + const s = document.createElement('script'); + s.src = src; s.onload = () => res(true); s.onerror = () => rej(new Error('fail')); + document.head.appendChild(s); + }); + if (typeof window.TencentCaptcha !== 'function') await load('https://ssl.captcha.qq.com/TCaptcha.js'); + window.__capres = null; + window.__cap = new window.TencentCaptcha('${aid}', function (r) { window.__capres = r; }, {}); + window.__cap.show(); + return {ok: true}; +})()`; + +(async () => { + const port = parseInt(arg('port', '9222'), 10); + const aid = arg('aid', '195672344'); + const out = arg('out', '/tmp/1ms_bridge/modvals.json'); + const waitMs = parseInt(arg('wait', '9000'), 10); + + const list = await (await fetch(`http://127.0.0.1:${port}/json/list`)).json(); + const page = list.find(t => t.type === 'page' && !(t.url || '').startsWith('chrome')); + if (!page) throw new Error('没有可用 page target'); + const pageCdp = await Cdp.connect(page.webSocketDebuggerUrl); + await pageCdp.send('Page.enable'); + await pageCdp.send('Runtime.enable'); + + const ver = await (await fetch(`http://127.0.0.1:${port}/json/version`)).json(); + const brCdp = await Cdp.connect(ver.webSocketDebuggerUrl); + + const sessions = []; + brCdp.on(async m => { + if (m.method === 'Target.attachedToTarget') { + const { sessionId, targetInfo } = m.params; + sessions.push({ sessionId, targetInfo }); + console.error(`[attach] ${targetInfo.type} ${(targetInfo.url || '').slice(0, 80)}`); + try { + await brCdp.send('Page.enable', {}, sessionId); + await brCdp.send('Page.addScriptToEvaluateOnNewDocument', { source: HOOK }, sessionId); + console.error(' -> hook registered'); + } catch (e) { console.error(' -> hook fail', e.message); } + try { await brCdp.send('Runtime.runIfWaitingForDebugger', {}, sessionId); } + catch (e) {} + } + }); + await brCdp.send('Target.setAutoAttach', { + autoAttach: true, waitForDebuggerOnStart: true, flatten: true, + }); + console.error('[1] autoAttach 已开(新 target 会先暂停,等我们注册钩子)'); + + console.error('[2] 导航到 1ms.run 并拉起验证码'); + await pageCdp.send('Page.navigate', { url: 'https://1ms.run/' }); + await sleep(4000); + await pageCdp.send('Runtime.evaluate', { + expression: SHOW(aid), awaitPromise: true, returnByValue: true, + }); + await sleep(waitMs); + + console.error('[3] 在各 session 里读取记账结果'); + const results = []; + for (const s of sessions) { + try { + const r = await brCdp.send('Runtime.evaluate', { + expression: `(() => { try { return {u: location.href, n: (window.__mods||[]).length, + mods: window.__mods||[], reqs: window.__reqs||[]}; } catch(e) { return {err:String(e)}; } })()`, + returnByValue: true, + }, s.sessionId); + const v = r.result && r.result.value; + if (v && v.mods && v.mods.length) { results.push(v); } + else if (v) console.error(` ${(v.u || '?').slice(0, 60)}: mods=${v.n || 0}`); + } catch (e) { console.error(' read fail', e.message); } + } + + fs.writeFileSync(out, JSON.stringify(results, null, 1)); + console.error(`[4] -> ${out} (命中 ${results.length} 个 session 有记账)`); + for (const r of results) { + console.error(` ${(r.u || '').slice(0, 70)} mods=${r.mods.length} reqs=${r.reqs.length}`); + } + brCdp.close(); pageCdp.close(); +})().catch(e => { console.error('ERR', e.message); process.exit(1); }); diff --git a/tools/cdp_show_captcha.js b/tools/cdp_show_captcha.js new file mode 100644 index 0000000..2b8ccfb --- /dev/null +++ b/tools/cdp_show_captcha.js @@ -0,0 +1,79 @@ +// 在真 Chrome(9222) 里拉起 1ms 的腾讯验证码。 +// 用法: node tools/cdp_show_captcha.js +function arg(name, def) { + const i = process.argv.indexOf('--' + name); + if (i < 0) return def; + const v = process.argv[i + 1]; + return (v === undefined || v.startsWith('--')) ? true : v; +} +const sleep = ms => new Promise(r => setTimeout(r, ms)); + +class Cdp { + constructor(ws) { this.ws = ws; this.id = 0; this.waits = new Map(); this.handlers = []; } + static async connect(url) { + const ws = new WebSocket(url); + await new Promise((res, rej) => { ws.onopen = res; ws.onerror = rej; }); + const c = new Cdp(ws); + ws.onmessage = ev => { + const m = JSON.parse(ev.data); + if (m.id && c.waits.has(m.id)) { + const { res, rej } = c.waits.get(m.id); + c.waits.delete(m.id); + m.error ? rej(new Error(JSON.stringify(m.error))) : res(m.result); + } else if (m.method) { for (const h of c.handlers) h(m); } + }; + return c; + } + send(method, params = {}) { + const id = ++this.id; + this.ws.send(JSON.stringify({ id, method, params })); + return new Promise((res, rej) => this.waits.set(id, { res, rej })); + } + on(fn) { this.handlers.push(fn); } + close() { try { this.ws.close(); } catch (e) {} } +} + +async function listTargets(port) { + const r = await fetch(`http://127.0.0.1:${port}/json/list`); + return r.json(); +} + +(async () => { + const port = parseInt(arg('port', '9222'), 10); + const aid = arg('aid', '195672344'); + const pageUrl = arg('url', 'https://1ms.run/'); + + const targets = await listTargets(port); + let page = targets.find(t => t.type === 'page' && !(t.url || '').startsWith('chrome')); + if (!page) throw new Error('没有可用页面 target'); + + const cdp = await Cdp.connect(page.webSocketDebuggerUrl); + await cdp.send('Page.enable'); + await cdp.send('Runtime.enable'); + + console.log(`[1] 导航到 ${pageUrl}`); + await cdp.send('Page.navigate', { url: pageUrl }); + await sleep(4000); + + console.log(`[2] 注入 TCaptcha SDK 并 show(aid=${aid})`); + const expr = `(async () => { + const load = (src) => new Promise((res, rej) => { + const s = document.createElement('script'); + s.src = src; s.onload = () => res(true); s.onerror = () => rej(new Error('fail')); + document.head.appendChild(s); + }); + if (typeof window.TencentCaptcha !== 'function') await load('https://ssl.captcha.qq.com/TCaptcha.js'); + window.__capres = null; + window.__cap = new window.TencentCaptcha('${aid}', function (r) { window.__capres = r; }, {}); + window.__cap.show(); + return {ok: true}; + })()`; + const r = await cdp.send('Runtime.evaluate', { expression: expr, awaitPromise: true, returnByValue: true }); + console.log(' ', JSON.stringify(r.result && r.result.value)); + + await sleep(2500); + const t2 = await listTargets(port); + console.log('[3] 当前 targets:'); + for (const t of t2) console.log(` ${t.type}\t${(t.url || '').slice(0, 100)}`); + cdp.close(); +})().catch(e => { console.error('ERR', e.message); process.exit(1); }); diff --git a/tools/tdc_modules.py b/tools/tdc_modules.py index 507f899..c5ab10a 100644 --- a/tools/tdc_modules.py +++ b/tools/tdc_modules.py @@ -90,6 +90,11 @@ FINGERPRINTS: List[Tuple[str, Set[str], Set[str]]] = [ ("charset", {"characterSet", "charset"}, set()), ("viewport_size", {"clientWidth", "clientHeight"}, set()), ("time_since_load", {"getTime", "load"}, set()), + # 轨迹采集器:真机实测其字符串集是鼠标/触摸事件名(2026-10-12 修正)。 + # 旧规则写成 {"stringify"} —— 那抓到的是另一个只做 JSON.stringify 的模块, + # 结果「空数组」被记成 trajectory,真正的轨迹却在 window_time 名下。 + ("trajectory", {"mouseMoveEvent", "touchStartEvent"}, set()), + ("struct_76_33", {"stringify"}, set()), # ---- 单串规则(放最后,避免压过上面的复合规则)---- ("user_agent", {"userAgent"}, set()), ("platform", {"platform"}, set()), @@ -106,7 +111,6 @@ FINGERPRINTS: List[Tuple[str, Set[str], Set[str]]] = [ ("touch_detect", {"ontouchstart"}, set()), ("window_time", {"getTime"}, set()), ("constant_98k", {"98k"}, set()), - ("trajectory", {"stringify"}, set()), ("screen_width", {"screen", "width"}, set()), ("screen_height", {"screen", "height"}, set()), ("screen_color_depth", {"colorDepth"}, set()), diff --git a/tools/tdc_profile.py b/tools/tdc_profile.py new file mode 100644 index 0000000..9c24ec6 --- /dev/null +++ b/tools/tdc_profile.py @@ -0,0 +1,202 @@ +# -*- coding: utf-8 -*- +""" +tdc_profile.py —— 设备画像(fingerprint profile)的**提取**与**定义**。 + +背景(账本 §五 第 0 项,`har/1ms_dianxuan/PROGRESS.md`): + 「纯离线生成」的唯一阻塞点,是需要为每个**语义名**提供取值。 + 多数语义名是静态环境量(UA / 屏幕 / 时区 / 语言 / 核数 / charset / + WebGL / Canvas),可以从一份真机明文里**按值的内容反查**出来。 + +为什么不靠"明文属于哪份构建": + 真机明文的槽顺序 = **它自己那次会话构建**的 require 顺序,而那份 tdc.js + 通常没保存(本项目实测:自洽性打分全为负,即明文不属于手上任何构建)。 + 但**值本身携带强语义**:`"MacIntel"` 只可能是 platform,`"UTF-8"` 只可能是 + charset,`"ANGLE (Apple, …)"` 只可能是 webgl_renderer。所以按内容反查 + 比按位置配对更稳,也不需要那份丢失的构建。 + +镜像自洽(案例 §6.3): 一套画像内的值必须互相一致 —— + screen_fingerprint `"1440-2560-2530-24-…"` 里的 1440/2560/24 应当分别等于 + screen_width / screen_height / screen_color_depth(字段序 = 宽-高-可用高-色深)。 + 本模块会做这个交叉校验。 + +标准画像请以 `tcaptcha/tdc_values.py::DEVICE_PROFILE` 为准(那里是 37 个语义名 +全部对齐后的产物);本文件只是**从明文按值反查**的开发期工具。 + +用法: + # 从一份已解密的真机明文提取画像基线 + .venv/bin/python tools/tdc_profile.py from-sample <明文文件> [--json 输出] + + # 打印内置画像 + .venv/bin/python tools/tdc_profile.py show +""" +from __future__ import annotations + +import argparse +import json +import re +import sys +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +ROOT = Path(__file__).resolve().parents[1] + + +# --------------------------------------------------------------------------- # +# 内容反查:值 -> 语义名(可能多个候选,按置信度排序) # +# --------------------------------------------------------------------------- # +def guess_names(v: Any) -> List[str]: + """按**值的内容**反推它可能出自哪个语义名(不依赖槽位置)。""" + out: List[str] = [] + if isinstance(v, str): + if "Mozilla" in v and "AppleWebKit" in v: + out.append("user_agent") + if v == "MacIntel": + out.append("platform") + if re.fullmatch(r"[+-]\d{2}(:\d{2})?", v): + out.append("timezone_offset") + if v.upper() in ("UTF-8", "UTF8"): + out.append("charset") + if "ANGLE" in v or "Metal Renderer" in v or "OpenGL" in v: + out.append("webgl_renderer") + if "Inc." in v or ("Google" in v and "Apple" in v): + out.append("webgl_vendor") + if v.startswith("GgoA") and len(v) > 40: + out.append("canvas_fingerprint") + if "captcha" in v and "rand=" in v: + out.append("template_url") + if v == "98k": + out.append("constant_98k") + if v == "iframe": + out.append("iframe_detect") + if "-" in v and "*" in v and re.fullmatch(r"[\d\-*|]+", v): + out.append("screen_fingerprint") + if v.startswith("http") and "1ms.run" in v: + out.append("entry_url") + elif v.startswith("http"): + out.append("location_href") + elif isinstance(v, list): + if v and all(isinstance(x, list) and len(x) == 3 for x in v): + out.append("trajectory") + if v and all(isinstance(x, str) for x in v) and 1 <= len(v) <= 4: + out.append("languages") + elif isinstance(v, int) and not isinstance(v, bool): + if 1_600_000_000 <= v <= 2_100_000_000: + out.append("timestamp") + return out + + +# --------------------------------------------------------------------------- # +# 内置画像(MacBook 画像基线;取自真机样本,可按需替换/随机化) # +# --------------------------------------------------------------------------- # +DEFAULT_PROFILE: Dict[str, Any] = { + "name": "Mac / Chrome", + # --- 明文里能直接看到的静态量 --- + "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"), + "platform": "MacIntel", + "languages": ["zh-CN", "zh"], + "timezone_offset": "+08", + "charset": "UTF-8", + "hardware_concurrency": 10, + # ⚠️ 方向别弄反:真机 screen.width == 1440、screen.height == 2560 + # (2026-10-12 浏览器实测;1440x2560 竖屏,availHeight 2530) + "screen_width": 1440, + "screen_height": 2560, + "screen_color_depth": 24, + "screen_fingerprint": "1440-2560-2530-24-*-*-|-*", + "webgl_renderer": "ANGLE (Apple, ANGLE Metal Renderer: Apple M4, Unspecified Version)", + "webgl_vendor": "Google Inc. (Apple)", + "canvas_fingerprint": ("GgoAAAANSUhEUgAAASwAAACWCAYAAATf3v8bh+ryWcfSBSLkmN3N1" + "PQinPL7Bi3TG0qE2CogAAAABJRU5ErkJggg=="), + "constant_98k": "98k", + "cookie_enabled": 1, + "touch_detect": 2, # 真机实测(该机 maxTouchPoints 为 0,值仍是 2) + "viewport_size": [360, 360], +} + + +def cross_check(profile: Dict[str, Any]) -> List[str]: + """做镜像自洽性交叉校验(案例 §6.3)。返回不一致项列表。""" + problems: List[str] = [] + fp = profile.get("screen_fingerprint") or "" + parts = fp.split("-") + if len(parts) >= 4: + try: + # 字段序 = 宽-高-可用高-色深(2026-10-12 与浏览器真值 1440/2560/2530/24 对齐; + # 旧代码把它当成 高-宽-?-色深,方向是反的) + w, h, _avail, depth = int(parts[0]), int(parts[1]), int(parts[2]), int(parts[3]) + if h != profile.get("screen_height"): + problems.append("screen_fingerprint 的高度 %d != screen_height %s" + % (h, profile.get("screen_height"))) + if w != profile.get("screen_width"): + problems.append("screen_fingerprint 的宽度 %d != screen_width %s" + % (w, profile.get("screen_width"))) + if depth != profile.get("screen_color_depth"): + problems.append("screen_fingerprint 的色深 %d != screen_color_depth %s" + % (depth, profile.get("screen_color_depth"))) + except ValueError: + problems.append("screen_fingerprint 格式无法解析: %r" % fp) + ua = profile.get("user_agent") or "" + if "Chrome/" in ua and "Macintosh" in ua and profile.get("platform") != "MacIntel": + problems.append("UA 说是 Mac,但 platform=%r" % profile.get("platform")) + return problems + + +def from_sample(path: str) -> Tuple[Dict[str, Any], List[Dict]]: + """从一份真机明文,按内容反查提取画像基线。 + + 返回 (识别到的状态量, 未识别的槽列表)。 + """ + raw = open(path, encoding="latin1").read().rstrip("\x00") + logical = re.sub(r" {2,}", "", raw) + cd = json.loads(logical)["cd"] + + found: Dict[str, Any] = {} + unidentified: List[Dict] = [] + for i, v in enumerate(cd): + cands = guess_names(v) + if len(cands) == 1: + found[cands[0]] = v + elif cands: + found.setdefault(cands[0], v) + else: + unidentified.append({"slot": i, "value": v}) + return found, unidentified + + +def main() -> int: + ap = argparse.ArgumentParser(description="TDC 设备画像提取/查看") + sub = ap.add_subparsers(dest="cmd", required=True) + p1 = sub.add_parser("from-sample", help="从真机明文提取画像基线") + p1.add_argument("plain") + p1.add_argument("--json", default=None) + sub.add_parser("show", help="打印内置画像") + + args = ap.parse_args() + if args.cmd == "show": + print(json.dumps(DEFAULT_PROFILE, ensure_ascii=False, indent=1)) + probs = cross_check(DEFAULT_PROFILE) + print() + print("交叉校验:", "全部自洽" if not probs else probs) + return 0 + + found, unk = from_sample(args.plain) + print("从 %s 反查到的状态量:" % Path(args.plain).name) + for k, v in found.items(): + print(" %-22s %s" % (k, json.dumps(v, ensure_ascii=False)[:70])) + print() + print("未识别的槽: %d 个" % len(unk)) + for u in unk: + print(" 槽%-3d %s" % (u["slot"], json.dumps(u["value"], ensure_ascii=False)[:60])) + print() + probs = cross_check(found) + print("交叉校验:", "全部自洽 ✓" if not probs else probs) + if args.json: + Path(args.json).write_text(json.dumps(found, ensure_ascii=False, indent=1), + encoding="utf-8") + print("画像 -> %s" % args.json) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())