yml2213
80a122245a
做的什么:
1. 判决并修正 collect 结构: 不是「整体一次加密」, 而是
chunk1 / trajectory / chunk2 / sd **四段独立加密**, 前 3 段
空格 pad 到 24 字节对齐、sd 不 pad, 4 段 base64 直接拼接。
2. 按案例 §5.3 实现字符串+结构指纹的模块识别器, 把 37 个采集器
全部识别且跨构建唯一。
3. 修 reorder 的恒等映射 bug, 并补强断言。
4. 在线抓取全新样本作为唯一数据源, 清理探索性脚本。
为什么:
* 旧「整体加密 + \x00 补齐」与真机数据矛盾 —— 导致多轮实验
建立在结构错误的 collect 上（这正是此前反复"乱碰"的根因）。
* 模块 id 每份构建重排, 不能当跨构建锚点; 但模块的**代码**
（字符串集合、结构）不变 —— 这是案例指明的不变量。
* reorder 旧实现用「目标槽下标」切「源数组」, 顺序不同即全盘错位;
原测试只断言"是置换", 恒等函数也满足, 所以 bug 被放过了。
怎么验证:
* 四段结构（3 条独立证据, 其中一条不需密钥）:
- 真机明文 3 处连续空格串 [+5,+10,+24], 且 pad == 24-len%24 全中
- 密文长度算术 480+96+416+56 = 1048 == 真实 1048 字符
- '=' 只出现在 1046/1048, 对应 sd 段 33B→补 7→40B
* 指纹: 4 份全新在线构建 37/37 识别、0 unknown、跨构建语义名集合
完全一致、单构建内 0 重名
* 配对正确性: 独立证据 37/37 一致; 负对照（清单下标 / 模块 id /
随机配对）均只有 2-3/37 —— 证明归一化没有过拟合
* 端到端重排 3/3 正确且非恒等
* selftest 全绿; 强断言自查: 旧实现过不了新断言
新增:
tools/tdc_modules.py 字符串+结构指纹识别（案例 §5.3）
scripts/fetch_online_samples.py 在线抓取全新 tdc.js
scripts/verify/*.py 7 个可复现的验证脚本
docs/tdc-案例.md 案例原文（看雪, 作者执着的猫）
修改:
tcaptcha/crypto.py +build_collect / split_collect_segments（四段）
tools/tdc_collectgen.py reorder 改为跨构建锚点对齐 + spec_from_seq
tools/tdc_layout.py +static_fingerprint / opcode_histogram
tools/tdc_program.py 修符号求值插值类型崩溃
tcaptcha/tests/test_collectgen.py 补「非恒等」强断言 + 在线新样本端到端
tcaptcha/client.py debug 目录改到系统临时目录（不再写仓库内 scratch/）
har/1ms_dianxuan/PROGRESS.md 新增已知 #41~#49, 显式修正被推翻的旧结论
.gitignore +scratch/ +har/online/
删除:
scratch/（含 10-11 遗留的一次性脚本, 已确认不可恢复后清理）
注意:
* 账本记录了一次自我纠错: 曾据案例 §5.5 误判 kind=2「不进 cd」,
实测（cd 38 槽、轨迹在第 10 槽）否定后已撤回。
* 仍未打通: 纯离线生成需要每个语义名的取值来源（已知 #38）。
486 lines
21 KiB
Python
486 lines
21 KiB
Python
"""
|
||
腾讯 TDC 遥测 (collect) 纯算法还原 —— 唯一路径: 明文模板回填 + 教科书 XTEA。
|
||
|
||
设计原则 (AGENTS.md §2.3 / §2.4):
|
||
* **只走纯算**: collect 的明文组装 / 轨迹 / 时间戳 / XTEA 加密 / 编码全部在 Python 内完成,
|
||
不再用伪造 DOM/WebGL/Canvas 的"补环境"沙箱去跑 tdc.js。
|
||
(踩过的坑见 har/1ms_dianxuan/ANALYSIS.md §7: 换 python/node/browser 后端对 errorCode=12
|
||
毫无差异, 说明"环境"不是自变量, 补环境路线已废弃。)
|
||
* **Node 的唯一用途**: 从服务端下发的 tdc.js 里提取本次构建的 4 个 XTEA 密钥字
|
||
(密钥由 VM 运行时算出, 不是字面量, 只能靠 opcode 插桩捕获)。Node 不参与明文生成。
|
||
* 明文结构以「真实成功样本」(models/success_collect_template.b64) 为模板,
|
||
Python 只回填随会话变化的量: 轨迹 / 当前时间戳 / 随机盐。
|
||
"""
|
||
|
||
import base64
|
||
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
import re
|
||
import struct
|
||
import subprocess
|
||
import time
|
||
import urllib.parse
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
M32 = 0xFFFFFFFF
|
||
|
||
TDC_WORKER_PATH = Path(__file__).resolve().parent / "tdc_worker.js"
|
||
|
||
logger = logging.getLogger("tcaptcha_crypto")
|
||
|
||
# 样本构建的 XTEA 密钥字 (未加盐)。真实样本的等价密钥 (含两个轮加盐) 见
|
||
# tests/test_xtea_key.py 的 SAMPLE_KEY = [1448563793, 1850559083, 1733513813, 1364613714]。
|
||
DEFAULT_KEY = [1448563793, 1847730015, 1733184827, 1364613714]
|
||
|
||
DEFAULT_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"
|
||
)
|
||
|
||
|
||
def encrypt_string(plaintext: str, key: List[int] = DEFAULT_KEY) -> str:
|
||
"""把 ASCII/Latin1 明文用教科书 XTEA 加密后转 Base64 (不足 8 字节用 \\x00 补齐)。
|
||
|
||
本模块只有一份 xtea_encrypt_block (教科书实现), 此函数即使用它。
|
||
"""
|
||
data = plaintext.encode("latin1")
|
||
pad_len = (8 - (len(data) % 8)) % 8
|
||
data += b"\x00" * pad_len
|
||
|
||
out = bytearray()
|
||
for i in range(0, len(data), 8):
|
||
v0, v1 = struct.unpack("<II", data[i : i + 8])
|
||
ev0, ev1 = xtea_encrypt_block(v0, v1, key)
|
||
out.extend(struct.pack("<II", ev0, ev1))
|
||
|
||
return base64.b64encode(out).decode("ascii")
|
||
|
||
|
||
def extract_eks(tdc_js_code: str) -> str:
|
||
"""从 tdc.js 中提取会话密钥信封 eks (一个较长的 Base64 字符串)。
|
||
|
||
注意: tdc.js 每次构建都会重命名全局变量 (见 AGENTS.md 附录 A#6), 因此**不能**写死变量名;
|
||
这里匹配任意 `window.<随机名> = '<base64>'`, 并取长度最长的候选。
|
||
"""
|
||
cands = re.findall(r"window\.[A-Za-z_$][\w$]*\s*=\s*'([A-Za-z0-9+/=]{100,})'", tdc_js_code)
|
||
if not cands:
|
||
return ""
|
||
return max(cands, key=len)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 标准 (教科书) XTEA: 32 轮, delta=0x9E3779B9, sum 取 mod 2^32。
|
||
# 这是本模块**唯一**的 XTEA 实现; 需要从每份 tdc.js 里拿到的是那 4 个密钥字。
|
||
# --------------------------------------------------------------------------- #
|
||
XTEA_DELTA = 0x9E3779B9
|
||
|
||
|
||
def xtea_encrypt_block(v0: int, v1: int, key: List[int]) -> Tuple[int, int]:
|
||
"""教科书 XTEA 加密 (32 轮), 全 32 位无符号运算"""
|
||
v0 &= M32
|
||
v1 &= M32
|
||
total = 0
|
||
for _ in range(32):
|
||
v0 = (v0 + (((((v1 << 4) & M32) ^ (v1 >> 5)) + v1) & M32
|
||
^ ((total + key[total & 3]) & M32))) & M32
|
||
total = (total + XTEA_DELTA) & M32
|
||
v1 = (v1 + (((((v0 << 4) & M32) ^ (v0 >> 5)) + v0) & M32
|
||
^ ((total + key[(total >> 11) & 3]) & M32))) & M32
|
||
return v0, v1
|
||
|
||
|
||
def xtea_decrypt_block(v0: int, v1: int, key: List[int]) -> Tuple[int, int]:
|
||
"""教科书 XTEA 解密 (32 轮), 用于校验提取到的密钥"""
|
||
v0 &= M32
|
||
v1 &= M32
|
||
total = (XTEA_DELTA * 32) & M32
|
||
for _ in range(32):
|
||
v1 = (v1 - (((((v0 << 4) & M32) ^ (v0 >> 5)) + v0) & M32
|
||
^ ((total + key[(total >> 11) & 3]) & M32))) & M32
|
||
total = (total - XTEA_DELTA) & M32
|
||
v0 = (v0 - (((((v1 << 4) & M32) ^ (v1 >> 5)) + v1) & M32
|
||
^ ((total + key[total & 3]) & M32))) & M32
|
||
return v0, v1
|
||
|
||
|
||
def xtea_encrypt_bytes(data: bytes, key: List[int]) -> bytes:
|
||
"""XTEA 加密一段字节；不足 8 字节用 \\x00 补齐（XTEA 块对齐所需）。"""
|
||
data = data + b"\x00" * ((8 - len(data) % 8) % 8)
|
||
out = bytearray()
|
||
for i in range(0, len(data), 8):
|
||
v0, v1 = struct.unpack("<II", data[i:i + 8])
|
||
e0, e1 = xtea_encrypt_block(v0, v1, key)
|
||
out += struct.pack("<II", e0, e1)
|
||
return bytes(out)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# collect 的「四段加密」结构（按 docs/tdc-案例.md §5.1，真机明文已判决） #
|
||
# --------------------------------------------------------------------------- #
|
||
SEG_ALIGN = 24
|
||
SEG_PAD = b" "
|
||
|
||
_collect_seg_re = re.compile(rb" {2,}")
|
||
|
||
|
||
def xtea_encrypt_segment(text: str, key: List[int],
|
||
align: int = SEG_ALIGN, pad: bytes = SEG_PAD) -> bytes:
|
||
"""加密 collect 的一个**数据段**：空格 pad 到 align 字节对齐后 XTEA + Base64。
|
||
|
||
为什么要 pad 到 24（而不是 8）：24 字节 -> 32 个 Base64 字符，且**不会产生 `=` 填充**。
|
||
4 段密文的 Base64 直接拼接后看起来就是一个完整的 Base64 串，没有分隔符。
|
||
（真机明文里能直接看到这些空格：chunk1 +5、trajectory +10、chunk2 +24。）
|
||
"""
|
||
data = text.encode("latin1")
|
||
# 注意：即使 len%align==0 也要补一整个 align（真机 chunk2 内容 288 字节、
|
||
# 288%24==0，却补了 24 个空格 -> pad 长度恒为 align - len%align，落在 1..align）
|
||
data += pad * (align - len(data) % align)
|
||
return base64.b64encode(xtea_encrypt_bytes(data, key))
|
||
|
||
|
||
def xtea_encrypt_segment_raw(text: str, key: List[int]) -> bytes:
|
||
"""加密 **sd 段**：不 pad 到 24，只由 XTEA 的 8 字节块补齐（真机实测如此）。"""
|
||
data = text.encode("latin1")
|
||
return base64.b64encode(xtea_encrypt_bytes(data, key))
|
||
|
||
|
||
def build_collect(chunk1: str, trajectory: str, chunk2: str, sd: str,
|
||
key: List[int], url_encode: bool = False) -> str:
|
||
"""按案例 §5.1 组装 collect —— **4 段密文直接拼接，无分隔符**。
|
||
|
||
collect = b64(xtea(chunk1 + 空格pad到24))
|
||
+ b64(xtea(traj + 空格pad到24))
|
||
+ b64(xtea(chunk2 + 空格pad到24))
|
||
+ b64(xtea(sd)) # sd 不 pad
|
||
|
||
段的内容边界（案例 §5.5，与真机明文逐字吻合）：
|
||
chunk1 = '{"cd":[' + 轨迹之前的指纹值 + ','
|
||
trajectory = '[[x,y,t],[dx,dy,dt],...,[1,1,12]]'
|
||
chunk2 = ',' + 轨迹之后的指纹值 + '],'
|
||
sd = '"sd":{"od":"C","ft":"..."}}'
|
||
"""
|
||
parts = [
|
||
xtea_encrypt_segment(chunk1, key),
|
||
xtea_encrypt_segment(trajectory, key),
|
||
xtea_encrypt_segment(chunk2, key),
|
||
xtea_encrypt_segment_raw(sd, key),
|
||
]
|
||
b64 = b"".join(parts).decode("ascii")
|
||
if url_encode:
|
||
return urllib.parse.quote(b64, safe="")
|
||
return b64
|
||
|
||
|
||
def split_collect_segments(plain: str) -> Tuple[str, str, str, str]:
|
||
"""把一份**已解密**的 collect 明文按空格 pad 切回 4 段（去 pad）。
|
||
|
||
用于从真机样本反解出 chunk1/trajectory/chunk2/sd，做对拍或标定。
|
||
"""
|
||
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]
|
||
|
||
|
||
def xtea_decrypt_bytes(data: bytes, key: List[int]) -> bytes:
|
||
out = bytearray()
|
||
for i in range(0, len(data) - len(data) % 8, 8):
|
||
v0, v1 = struct.unpack("<II", data[i:i + 8])
|
||
d0, d1 = xtea_decrypt_block(v0, v1, key)
|
||
out += struct.pack("<II", d0, d1)
|
||
return bytes(out)
|
||
|
||
|
||
_CAPTURE_CACHE: Dict[str, Tuple[List[List], bytes]] = {}
|
||
|
||
|
||
def _capture_key_words_cached(tdc_js_code: str, node_bin: str = "node"):
|
||
cached = _CAPTURE_CACHE.get(tdc_js_code)
|
||
if cached is None:
|
||
cached = _capture_key_words(tdc_js_code, node_bin=node_bin)
|
||
_CAPTURE_CACHE[tdc_js_code] = cached
|
||
return cached
|
||
|
||
|
||
def _capture_key_words(tdc_js_code: str, node_bin: str = "node",
|
||
timeout: float = 180.0) -> Tuple[List[List], bytes]:
|
||
"""调用 Node 插桩运行 tdc.js, 返回 (密钥候选, 该脚本产出的 collect 原始字节)"""
|
||
script = TDC_WORKER_PATH
|
||
if not script.exists():
|
||
raise FileNotFoundError(f"缺少 tdc_worker.js: {script}")
|
||
cmd = [node_bin, str(script), "-", "[[0,0],[0,0],[0,0]]",
|
||
json.dumps({"aid": "1", "captureKey": True})]
|
||
proc = subprocess.run(cmd, input=tdc_js_code, capture_output=True, text=True,
|
||
timeout=timeout, env={**os.environ, "NODE_OPTIONS": ""})
|
||
out = (proc.stdout or "").strip().splitlines()
|
||
if not out:
|
||
raise RuntimeError(f"tdc_worker 无输出 (rc={proc.returncode}): {proc.stderr[:400]}")
|
||
payload = json.loads(out[-1])
|
||
if payload.get("error"):
|
||
raise RuntimeError(f"tdc_worker 执行失败: {payload['error'][:400]}")
|
||
words = payload.get("key_words") or []
|
||
if not words:
|
||
raise RuntimeError("tdc_worker 未返回密钥候选 (captureKey 未生效?)")
|
||
raw = base64.b64decode(urllib.parse.unquote(payload["collect"]))
|
||
return words, raw
|
||
|
||
|
||
_KEY_CACHE: Dict[str, List[int]] = {}
|
||
|
||
|
||
def extract_xtea_key(tdc_js_code: str, node_bin: str = "node",
|
||
top_n: int = 8, use_cache: bool = True) -> List[int]:
|
||
"""
|
||
从服务端下发的 tdc.js 里提取本次构建的 XTEA 密钥 (4 个 32 位字)。
|
||
|
||
**主路径 = 纯 Python 静态还原** (`tcaptcha/tdc_key.py`):
|
||
反汇编字节码 -> 定位 XTEA 闭包 -> 回溯它捕获的密钥数组 -> 读出 4 个密钥字
|
||
闭包(源串 + 偏移 + 存取下标) -> 折入轮加盐。
|
||
零 Node、零 JS 运行时、零 pc 硬编码 (opcode 表按构建推导, 见 tdc_opcodes.py)。
|
||
|
||
**兜底路径 = Node 插桩** (密钥构造算法若被改动时仍可用):
|
||
1. Node 在 VM 上下文里把 ADD opcode 换成带日志版本;
|
||
XTEA 每轮的 (sum + K[j]) 是一次 ADD, 两个操作数必有一个等于 r*DELTA,
|
||
于是另一个操作数就是密钥字 —— 按频次排序得到无歧义候选。
|
||
2. Python 取候选 Top-N, 对 4! 个排列逐个用教科书 XTEA 解密
|
||
tdc.js 自己产出的 collect 前 48 字节, 与常量 cd 前缀比对 —— 命中即真密钥。
|
||
|
||
密钥按构建固定 (同一份脚本多次运行首块完全一致, 已实测), 故可缓存。
|
||
"""
|
||
import itertools
|
||
|
||
if use_cache:
|
||
cached = _KEY_CACHE.get(tdc_js_code)
|
||
if cached:
|
||
return list(cached)
|
||
|
||
# ---- 主路径: 纯 Python 静态还原 (零 Node) ----
|
||
try:
|
||
from .tdc_key import extract_xtea_key as _pure_extract
|
||
key = [int(k) & M32 for k in _pure_extract(tdc_js_code)]
|
||
if len(key) == 4:
|
||
if use_cache:
|
||
_KEY_CACHE[tdc_js_code] = key
|
||
return list(key)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("纯 Python 静态提取失败, 退回 Node 插桩: %s", exc)
|
||
|
||
words, raw = _capture_key_words_cached(tdc_js_code, node_bin=node_bin)
|
||
candidates = [int(k) & M32 for k, *_ in words[:top_n]]
|
||
# 验证用「内容特征」而非固定明文 (AGENTS.md §5.3 / 附录 A#6):
|
||
# 实测 tdc.js 至少两代明文布局 ——
|
||
# 旧版: 4 段 base64 拼接, 首段 = '{"cd":[[],"1920-1080-1055-24-*-*-|-*",0,0,"",'
|
||
# 新版: 单个 JSON, 首部 = '{"cd":[1,[],"Apple M1",24,...' / '{"cd":[1791712995,0,...'
|
||
# 字段序还随构建变化, 故只依赖跨版本稳定的结构特征: '{"cd":[' 开头 + 首块全可打印。
|
||
blocks = [struct.unpack("<II", raw[i:i + 8]) for i in range(0, 48, 8)]
|
||
|
||
for perm in itertools.permutations(candidates, 4):
|
||
key = list(perm)
|
||
head = b"".join(struct.pack("<II", *xtea_decrypt_block(v0, v1, key))
|
||
for v0, v1 in blocks)
|
||
if head[:7] == b'{"cd":[' and all(32 <= b < 127 for b in head[:48]):
|
||
if use_cache:
|
||
_KEY_CACHE[tdc_js_code] = key
|
||
return key
|
||
raise RuntimeError(
|
||
f"未能从 tdc.js 复原 XTEA 密钥 (候选 {len(candidates)} 个, 排列全试失败)")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# §5.8 标定式交付: 运行时取一次明文模板, 之后全部由 Python 回填
|
||
# (新版 tdc.js 的明文是单个 JSON, 且**字段序随构建变化** —— 实测两份同时期脚本
|
||
# 首部一个是 '{"cd":[1,[],"Apple M1",24,...', 另一个是 '{"cd":[1791712995,0,...',
|
||
# 所以不能硬编码模板, 只能每次标定一份。)
|
||
# --------------------------------------------------------------------------- #
|
||
_TRAJ_RE = re.compile(r"\[\[\d+,-?\d+,\d+\](?:,\[-?\d+,-?\d+,\d+\])+\]")
|
||
_EPOCH_RE = re.compile(r"(?<![\d.])(1[6-9]\d{8})(?![\d.])")
|
||
|
||
|
||
def recover_payload_template(tdc_js_code: str, node_bin: str = "node"):
|
||
"""
|
||
标定: 跑一次 tdc.js, 提取本次构建的 XTEA 密钥并解密出**明文模板**。
|
||
返回 (template_text, key)。
|
||
"""
|
||
key = extract_xtea_key(tdc_js_code, node_bin=node_bin)
|
||
_, raw = _capture_key_words_cached(tdc_js_code, node_bin=node_bin)
|
||
plain = xtea_decrypt_bytes(raw, key).rstrip(b"\x00")
|
||
return plain.decode("latin1"), key
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 纯算主路径: 以「真实成功样本的明文」为结构模板 (服务端已认可的字段序与取值),
|
||
# Python 只回填随会话变化的量: 轨迹 / 当前时间戳 / 随机盐。Node 只提供 XTEA 密钥。
|
||
# --------------------------------------------------------------------------- #
|
||
_SUCCESS_TPL_PATH = Path(__file__).resolve().parent / "models" / "success_collect_template.b64"
|
||
|
||
|
||
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)]]
|
||
|
||
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=(",", ":"))
|
||
|
||
|
||
def generate_collect_success(
|
||
click_points: List[Tuple[int, int]],
|
||
key: List[int],
|
||
entry_url: str = "https://1ms.run/console/personal/benefits/check-in",
|
||
raw_base64: bool = True,
|
||
) -> str:
|
||
"""成功样本模板 + Python 回填。加密/组包/编码全部纯 Python。
|
||
|
||
raw_base64=True 返回未 URL 编码的标准 Base64 (由 httpx 表单编码时单层编码,
|
||
与真实抓包一致); False 则预先做一次 percent-encoding。
|
||
"""
|
||
if not _SUCCESS_TPL_PATH.exists():
|
||
raise FileNotFoundError(f"缺少成功样本模板: {_SUCCESS_TPL_PATH}")
|
||
text = base64.b64decode(_SUCCESS_TPL_PATH.read_text(encoding="ascii")).decode("latin1")
|
||
text = text.replace(
|
||
'"https://1ms.run/console/personal/benefits/check-in"', '"%s"' % entry_url, 1)
|
||
text = _TRAJ_RE.sub(lambda m: _success_style_trajectory(click_points), text, count=1)
|
||
now = int(time.time())
|
||
text = text.replace("1791715501", str(now), 1) # 采集时刻
|
||
text = text.replace("1791715508", str(now + random.randint(4, 12)), 1)
|
||
text = text.replace("1791706326", str(now - random.randint(6000, 12000)), 1)
|
||
text = text.replace("699129805", str(random.randint(100000000, 999999999)), 1)
|
||
data = text.encode("latin1")
|
||
data += b"\x00" * ((8 - len(data) % 8) % 8) # 成功样本用 \x00 填充
|
||
b64 = base64.b64encode(xtea_encrypt_bytes(data, key)).decode("ascii")
|
||
if raw_base64:
|
||
return b64
|
||
return urllib.parse.quote(b64, safe="")
|
||
|
||
|
||
def generate_collect_from_template(
|
||
template: str,
|
||
click_points: List[Tuple[int, int]],
|
||
key: List[int],
|
||
refresh_time: bool = True,
|
||
) -> str:
|
||
"""用标定得到的明文模板 + Python 生成的轨迹回填出 collect (纯算)。
|
||
|
||
1. 用 Python 轨迹替换模板里的轨迹数组
|
||
2. 刷新模板里的秒级时间戳
|
||
3. 空格补齐到 8 字节倍数 -> XTEA 加密 -> base64 -> URL 编码
|
||
"""
|
||
text = template.decode("latin1") if isinstance(template, (bytes, bytearray)) else template
|
||
|
||
traj = _success_style_trajectory(click_points)
|
||
m = _TRAJ_RE.search(text)
|
||
if not m:
|
||
raise RuntimeError("模板里未找到轨迹数组, 无法回填")
|
||
text = text[:m.start()] + traj + text[m.end():]
|
||
|
||
if refresh_time:
|
||
text = _EPOCH_RE.sub(str(int(time.time())), text)
|
||
|
||
data = text.encode("latin1")
|
||
data += b" " * ((8 - len(data) % 8) % 8) # 与真实 tdc.js 一致: 空格补齐
|
||
return urllib.parse.quote_from_bytes(
|
||
base64.b64encode(xtea_encrypt_bytes(data, key)), safe="")
|
||
|
||
|
||
def generate_telemetry(
|
||
tdc_js_code: str,
|
||
click_points: List[Tuple[int, int]],
|
||
aid: str = "2039519451",
|
||
user_agent: str = DEFAULT_USER_AGENT,
|
||
referrer: str = "https://app.rainyun.com/",
|
||
mode: str = "python_keyed",
|
||
node_bin: str = "node",
|
||
) -> Dict[str, str]:
|
||
"""统一遥测生成入口 —— **只走纯算**, 不再有任何伪造环境的执行后端。
|
||
|
||
mode="python_keyed" (默认): Node 只做一件事 —— 从当前 tdc.js 提取本次构建的
|
||
XTEA 密钥字 (密钥由 VM 运行时算出, 不是字面量); 明文组装 / 轨迹 / 时间戳 /
|
||
加密 / 编码全部在 Python 内完成。Node 不可用时自动退回 DEFAULT_KEY。
|
||
mode="python": 完全不调用 Node, 直接用内置 DEFAULT_KEY。
|
||
|
||
返回 {"collect", "tlg", "eks", "source"[, "key"]}。
|
||
|
||
关于 errorCode=12: 实测与遥测后端无关 (python / python_keyed / node / browser 全部 12,
|
||
换 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')")
|
||
|
||
key: Optional[List[int]] = None
|
||
source = "python"
|
||
if mode == "python_keyed":
|
||
try:
|
||
key = extract_xtea_key(tdc_js_code, node_bin=node_bin)
|
||
source = "python_keyed"
|
||
except Exception as exc: # Node 缺失/插桩失败 -> 退回固定密钥
|
||
logger.warning("XTEA 密钥提取失败, 退回内置 DEFAULT_KEY: %s", exc)
|
||
if key is None:
|
||
key = list(DEFAULT_KEY)
|
||
|
||
collect = generate_collect_success(click_points, key, entry_url=referrer, raw_base64=True)
|
||
return {
|
||
"collect": collect,
|
||
"tlg": str(len(collect)),
|
||
"eks": extract_eks(tdc_js_code),
|
||
"source": source,
|
||
"key": key,
|
||
}
|
||
|
||
|
||
def selfcheck() -> None:
|
||
"""纯算法自检: XTEA 往返 + encrypt_string 一致性。"""
|
||
for chunk in (b"hello!!", bytes(range(32)), b"x" * 24):
|
||
dec = xtea_decrypt_bytes(xtea_encrypt_bytes(chunk, DEFAULT_KEY), DEFAULT_KEY)
|
||
assert dec[:len(chunk)] == chunk, f"XTEA 往返失败: {chunk!r}"
|
||
ref = base64.b64encode(
|
||
xtea_encrypt_bytes(b"abc\x00\x00\x00\x00\x00", DEFAULT_KEY)).decode("ascii")
|
||
assert encrypt_string("abc") == ref, "encrypt_string 与 xtea_encrypt_bytes 不一致"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
selfcheck()
|
||
print("crypto 自检通过: XTEA 往返 + encrypt_string 一致性")
|