feat(collect): 纯离线生成打通 1ms 签到文字点选 —— errorCode=0 收口

做了什么:
- 新增 tcaptcha/tdc_values.py: 37 个语义名的取值生成器(静态画像 / 会话量 /
  每次采集量三层)+ build_plain 四段组装 + generate_collect 纯离线出包
- crypto: 新增 generate_telemetry(mode="python_values");轨迹生成统一到
  trajectory_points()(首点绝对坐标 + 增量 + 固定收尾 [1,1,12])
- crypto.split_collect_segments 重写: pad 只有 1 个空格时旧 ` {2,}` 判据会漏段,
  且 UA 串里的普通空格会假阳性 —— 改为「模 24 算术 + 不落在 JSON 字符串内部 + 结构校验」
- tools/tdc_modules: trajectory 指纹原先指向只做 JSON.stringify 的模块,改为
  mouseMoveEvent/touchStartEvent(真轨迹模块),37/37 不变
- client 默认 telemetry_mode 改为 python_values
- 新增 scripts/verify/{align_session,audit_names,verify_values,verify_offline,send_test}.py
  与 tools/{cdp_collect,cdp_dump_scripts,tdc_profile}.py

为什么:
- 账本 #40「谁在发 HTTP 是 12 的剩余自变量」被推翻。同 aid/同客户端/同求解器、
  答案都对,只换 collect 构造方式: 本次会话构建生成 -> 0 (4/4),
  旧模板回填 -> 12 (2/2)。真因是 collect 必须与本次会话构建同源。

怎么验证:
- .venv/bin/python -m tcaptcha.selftest  全绿(含新增四段切分与纯离线闭环回归)
- scripts/verify/verify_values.py        4 份真机明文 152/152 逐槽全等
- scripts/verify/verify_offline.py       逐槽全等 + 分段几何一致
- scripts/verify/send_test.py --mode python_values  1ms 签到页文字点选 errorCode=0 x4
This commit is contained in:
yml2213
2026-10-12 03:38:18 +08:00
parent 80a122245a
commit 51dff870cf
18 changed files with 2102 additions and 91 deletions
+93
View File
@@ -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")
+142
View File
@@ -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())
+75
View File
@@ -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))
+138
View File
@@ -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())
+105
View File
@@ -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())
+89
View File
@@ -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())
+104
View File
@@ -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())