yml/hello_js_reverse_skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d136fb7328f9b59518413ad15003e89fbf451463
hello_js_reverse_skill/tcaptcha/crypto.py
T
yml2213 1ef819b48d feat(纯算): 以成功样本明文为模板的纯 Python 出包管线 + 证据账本 v2 
做了什么
- 从「签到成功-1.har」提取该次会话的 tdc.js, 提取密钥解密出
  errorCode=0 那次 collect 的完整明文 (1360 字节), 存为
  tcaptcha/models/success_collect_template.b64
- 新增 generate_collect_success(): 以成功样本为结构模板, Python 只回填
  轨迹/当前时间戳/随机盐, 其余环境字段 (canvas/WebGL/屏幕/语言/sd.ft) 原样保留
- 新增 _success_style_trajectory(): 按成功样本实测统计生成轨迹
  (61 点, dt 均值 30.6ms, 50% 在 0~16ms, 途经 3 个目标, 驻留抖动)
  —— 修正了统计正则不允许负增量的 bug (真实 61 点, 此前误数成 18)
- 发现成功样本用 \x00 填充 (不是空格), sd 多一个 ft 字段
- telemetry_mode=python_keyed 默认走该路径; Node 仅提取每构建 XTEA 密钥
- ANALYSIS.md 第七节重写为 v2 证据账本, 显式撤回 4 个无证据结论

验证
- 纯 Python 密文 == 真实 tdc.js 密文 (逐字节)
- 回归: test_xtea_key 11 项、test_glyph_solver 8/8
- 端到端 1ms 实测: 仍 errorCode=12 (诚实记录, 未解决)

结论(见 ANALYSIS.md 7.5-7.7)
- 12 不是"检测到机器人": 同 IP 有成功样本、参数逐字段一致、
  真 Chrome 网络栈/真人点击/CDP 均排除
- 剩余假设: H1 需已登录业务会话(未用 auth_token 调 captcha/params 复测)
            H3 需有效 qcloud.com cookie(过期 skey 复测不能定论)
- 下一步: 用登录态 auth_token 调 /api/v1/captcha/params,
  对比返回是否多出鉴权字段 —— 这是唯一未做且成本低的关键实验
2026-10-11 19:12:23 +08:00

887 lines
36 KiB
Python
Raw Blame History

"""
Pure Python implementation of Tencent TDC Chaos VM (JSVMP) Cryptographic
and Telemetry generation algorithm.
Eliminates the Node.js sandbox dependency by replicating the 32-round custom
XTEA block cipher, human-like mouse movement trajectory generation, and
telemetry payload serialization.
"""
import base64
import json
import math
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"
TDC_BROWSER_PATH = Path(__file__).resolve().parent / "tdc_browser.js"
# TDC VM Cryptographic Constants
DELTA = 2654435769 # 0x9E3779B9
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/152.0.0.0 Safari/537.36"
)
DEFAULT_RENDERER = "Apple M1"
DEFAULT_VENDOR = "Apple"
DEFAULT_CANVAS_FP = (
"GgoAAAANSUhEUgAAAAEAAAABCAYAAAUhEUgAAAAEAAAABCAYAAAA"
"fFcSJAAAADhgGAWjR9awAAAABJRU5ErkJggg=="
)
def xtea_encrypt_block(v0: int, v1: int, key: List[int]) -> Tuple[int, int]:
"""
Tencent custom 32-round XTEA block cipher for a 64-bit block (v0, v1).
Includes specific additive offsets (2829068 and 328986) depending on round index.
"""
v0 &= 0xFFFFFFFF
v1 &= 0xFFFFFFFF
total_sum = 0
for _ in range(32):
# Step 1: Update v0
k_idx1 = total_sum & 3
shift1 = (((v1 << 4) ^ (v1 >> 5)) + v1) & 0xFFFFFFFF
if k_idx1 == 1:
k_term1 = (total_sum + key[1] + 2829068) & 0xFFFFFFFF
elif k_idx1 == 2:
k_term1 = (total_sum + key[2] + 328986) & 0xFFFFFFFF
else:
k_term1 = (total_sum + key[k_idx1]) & 0xFFFFFFFF
v0 = (v0 + (shift1 ^ k_term1)) & 0xFFFFFFFF
# Step 2: Update total_sum (standard XTEA delta increment)
total_sum = (total_sum + DELTA) & 0xFFFFFFFFFFFFFFFF
# Step 3: Update v1
k_idx2 = (total_sum >> 11) & 3
shift2 = (((v0 << 4) ^ (v0 >> 5)) + v0) & 0xFFFFFFFF
if k_idx2 == 1:
k_term2 = (total_sum + key[1] + 2829068) & 0xFFFFFFFF
elif k_idx2 == 2:
k_term2 = (total_sum + key[2] + 328986) & 0xFFFFFFFF
else:
k_term2 = (total_sum + key[k_idx2]) & 0xFFFFFFFF
v1 = (v1 + (shift2 ^ k_term2)) & 0xFFFFFFFF
return v0, v1
def encrypt_string(plaintext: str, key: List[int] = DEFAULT_KEY) -> str:
"""
Encrypts an ASCII/Latin1 plaintext string into Base64 using the custom XTEA cipher.
Input is padded with null bytes to an 8-byte boundary if needed.
"""
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 _bezier_interp(
p0: Tuple[float, float],
p1: Tuple[float, float],
steps: int,
) -> List[Tuple[int, int]]:
"""Generates cubic Bézier interpolation curve points with natural tremor."""
dx = p1[0] - p0[0]
dy = p1[1] - p0[1]
dist = math.hypot(dx, dy) or 1.0
perp_x = -dy / dist
perp_y = dx / dist
curvature = (random.random() - 0.5) * min(50.0, dist * 0.4)
cp1 = (
p0[0] + dx * 0.3 + perp_x * curvature,
p0[1] + dy * 0.3 + perp_y * curvature,
)
cp2 = (
p0[0] + dx * 0.7 - perp_x * curvature * 0.5,
p0[1] + dy * 0.7 - perp_y * curvature * 0.5,
)
pts = []
for i in range(steps + 1):
t = i / float(steps)
ease_t = t * t * (3.0 - 2.0 * t)
inv_t = 1.0 - ease_t
bx = (
inv_t**3 * p0[0]
+ 3 * inv_t**2 * ease_t * cp1[0]
+ 3 * inv_t * ease_t**2 * cp2[0]
+ ease_t**3 * p1[0]
)
by = (
inv_t**3 * p0[1]
+ 3 * inv_t**2 * ease_t * cp1[1]
+ 3 * inv_t * ease_t**2 * cp2[1]
+ ease_t**3 * p1[1]
)
tremor = 0.0 if i in (0, steps) else (random.random() - 0.5) * 1.5
pts.append((int(round(bx + tremor)), int(round(by + tremor))))
return pts
def human_dt_ms() -> int:
"""
真实 mousemove 的采样间隔 (毫秒)。
依据: 成功签到样本 (har/签到成功-1.har) 解密出的轨迹 dt 分布为
12~601ms, 均值 63.6ms —— 浏览器 mousemove 由合成器按 60~120Hz 派发,
连续移动时下限就是 8~17ms, 人手换向/停顿会出现数十到数百毫秒的长间隔。
之前这里用 random.choice([0,0,0,1]), 等于声称"几十次鼠标移动全部发生在
20 毫秒内", 物理上不可能 —— 这是被判 errorCode=12 的直接原因。
"""
r = random.random()
if r < 0.68:
return random.randint(13, 19) # 连续采样 (60~75Hz)
if r < 0.90:
return random.randint(20, 45) # 轻微停顿
if r < 0.98:
return random.randint(46, 120) # 明显停顿 (换向/瞄准)
return random.randint(121, 400) # 长停顿
def generate_trajectory_payload(
click_points: List[Tuple[int, int]],
target_len: int = 600,
) -> str:
"""
Generates realistic human mouse movement trajectory and serializes it
into the TDC relative-displacement JSON format:
[[x0, y0, 2], [dx1, dy1, dt1], [dx2, dy2, dt2], ...]
The string is padded with trailing spaces to target_len (multiple of 8).
"""
# Scale from original 672x480 coordinate space to 360x360 captcha iframe display space
scale = 320.0 / 672.0
mapped_targets = [
(int(round(20 + p[0] * scale)), int(round(60 + p[1] * scale)))
for p in click_points
]
# Starting position near entry area
current_pt = (
int(round(150 + (random.random() - 0.5) * 60)),
int(round(320 + (random.random() - 0.5) * 20)),
)
events: List[List[int]] = []
# 首个三元组的第三位是"进入验证码区域"的初始停顿 (成功样本实测 601ms)
events.append([current_pt[0], current_pt[1], random.randint(280, 820)])
last_x, last_y = current_pt[0], current_pt[1]
for target in mapped_targets:
# 真实样本 3 次点击只记到 18 个三元组 -> 每次点击 5~8 个采样点
steps = random.randint(5, 8)
curve = _bezier_interp((last_x, last_y), target, steps)
for pt in curve[1:]:
dx = pt[0] - last_x
dy = pt[1] - last_y
events.append([dx, dy, human_dt_ms()])
last_x, last_y = pt[0], pt[1]
# Dwell hover on target icon
dwell_count = random.randint(1, 2)
for _ in range(dwell_count):
dx = random.randint(-1, 1)
dy = random.randint(-1, 1)
events.append([dx, dy, human_dt_ms()])
last_x += dx
last_y += dy
import json
raw_json = json.dumps(events, separators=(",", ":"))
# 填充到 24 字节对齐 —— 24 是 8 (XTEA 分组) 与 3 (Base64 分组) 的最小公倍数,
# 这样每个分块单独 base64 后拼接不会产生 **块内 '=' 填充**。
# 实测真实 tdc.js 的 collect 整体只有末尾一个 '==' (见 点选.har),
# 块内出现 '=' 会导致服务端解析/解密失败。
desired_len = max(len(raw_json), target_len)
pad_needed = (24 - (desired_len % 24)) % 24
total_len = desired_len + pad_needed
return raw_json.ljust(total_len, " ")
def extract_eks(tdc_js_code: str) -> str:
"""Extracts session public key (`eks`) from TDC.js script header."""
match = re.search(r"window\.KQnlRHY\w+\s*=\s*'([^']+)'", tdc_js_code)
if match:
return match.group(1)
# Fallback to any base64 string assignment
match_fallback = re.search(r"window\.\w+\s*=\s*'([A-Za-z0-9+/=]{100,})'", tdc_js_code)
if match_fallback:
return match_fallback.group(1)
return ""
# --------------------------------------------------------------------------- #
# 标准 XTEA (32 轮, delta=0x9E3779B9, sum 取 mod 2^32) —— 等价于原 crypto.py 的
# 自定义形态 + 密钥折叠: K' = [k0, k1 + salt1, k2 + salt2, k3]
# 由此「算法」就是教科书 XTEA, 唯一需要从每份 tdc.js 里拿到的是那 4 个密钥字。
# --------------------------------------------------------------------------- #
XTEA_DELTA = 0x9E3779B9
CD_PRE_RAW = '{"cd":[[],"1920-1080-1055-24-*-*-|-*",0,0,"",'
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:
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)
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 位字)。
流程 (Node 只负责"跑出密钥", 算法全在 Python):
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)
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 填值, 与加密无关)
#
# 取值来源: /Users/yml/Desktop/抓包/毫秒/签到成功-1.har 里那次 errorCode=0 的 collect,
# 用该会话自己的 tdc.js 提取密钥解密后读出的真实字段。
# 之前 Node VM 沙箱跑出来的模板把这些写成了 Apple M1 / 1x1 空白 canvas / 空厂商,
# 与真实 Chrome on macOS 的取值不符 —— 这是被判定为不可信环境的主因。
# --------------------------------------------------------------------------- #
REAL_SCREEN_W = 2560
REAL_SCREEN_H = 1440
REAL_WINDOW_W = 1440
REAL_PLATFORM = "MacIntel"
# macOS 上真实 Chrome 报的是 ANGLE 包装串, 不是 "Apple M1"
REAL_GL_RENDERER = "ANGLE (Apple, ANGLE Metal Renderer: Apple M4, Unspecified Version)"
REAL_GL_VENDOR = "Google Inc. (Apple)"
# 真实 canvas 指纹 (成功样本原值, 368 bytes PNG)
REAL_CANVAS_FP = ("GgoAAAANSUhEUgAAASwAAACWCAYAAATf3v8bh+ryWcfSBSLkmN3N1PQinPL7Bi3"
"TG0qE2CogAAAABJRU5ErkJggg==")
REAL_UA = ("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")
_CANVAS_RE = re.compile(r'"[A-Za-z0-9+/]{40,}={0,2}"')
_RENDERER_RE = re.compile(r'"(?:ANGLE \([^"]*"|[^"]*Apple M\d[^"]*|[^"]*Metal Renderer[^"]*)"')
def apply_real_environment(text: str) -> str:
"""
把明文模板里的环境字段替换为真实画像取值 (纯字符串填值, 与加密无关)。
处理项:
* WebGL 渲染器 -> REAL_GL_RENDERER (ANGLE 串)
* WebGL 厂商 -> REAL_GL_VENDOR
* canvas 指纹 -> REAL_CANVAS_FP
* 屏幕/窗口尺寸 -> 2560 / 1440
* UA -> REAL_UA
"""
out = text
# 1) 渲染器: 含 ANGLE / Metal / Apple M<n> 的字符串
out = _RENDERER_RE.sub('"%s"' % REAL_GL_RENDERER, out, count=1)
# 2) 厂商: 只有 Apple / Google Inc. (Apple) 这类短串
out = re.sub(r'"(?:Apple|Google Inc\. \(Apple\)|Google Inc\.)"',
'"%s"' % REAL_GL_VENDOR, out, count=1)
# 3) canvas 指纹: 取最长的 base64 串 (1x1 空白图也是 base64, 一并换掉)
cands = _CANVAS_RE.findall(out)
if cands:
longest = max(cands, key=len)
out = out.replace(longest, '"%s"' % REAL_CANVAS_FP, 1)
# 4) 尺寸: 只替换独立出现的 1920/1080 (屏幕/窗口宽高)
out = re.sub(r'(?<![\d.])1920(?![\d.])', str(REAL_SCREEN_W), out)
out = re.sub(r'(?<![\d.])1080(?![\d.])', str(REAL_SCREEN_H), out)
out = re.sub(r'(?<![\d.])1440(?![\d.])', str(REAL_WINDOW_W), out)
# 5) UA
out = re.sub(r'"Mozilla/5\.0[^"]*"', '"%s"' % REAL_UA, out, count=1)
return out
# --------------------------------------------------------------------------- #
# 纯算主路径: 以「成功签到样本的明文」为结构模板 (服务端已认可的字段序与取值),
# Python 只回填随会话变化的量: 轨迹 / 当前时间戳 / 随机盐。
# 环境字段 (canvas / WebGL / 屏幕 / 语言 / sd.ft) 原样保留真实取值。
# Node 的唯一职责 = 提取当前构建的 XTEA 密钥。
# --------------------------------------------------------------------------- #
_SUCCESS_TPL_PATH = Path(__file__).resolve().parent / "models" / "success_collect_template.b64"
_SD_TAIL_RE = re.compile(r'"sd":\{[^}]*\}')
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",
) -> str:
"""成功样本模板 + Python 回填。加密/组包/编码全部纯 Python。"""
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 填充
return urllib.parse.quote_from_bytes(
base64.b64encode(xtea_encrypt_bytes(data, key)), 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 -> TDC URL 编码
加密与组包 100% 在 Python, Node 只提供密钥与一次性模板。
"""
text = template.decode("latin1") if isinstance(template, (bytes, bytearray)) else template
text = apply_real_environment(text)
traj = generate_trajectory_payload(click_points, target_len=0).rstrip()
m = _TRAJ_RE.search(text)
if not m:
raise RuntimeError("模板里未找到轨迹数组, 无法回填")
text = text[:m.start()] + traj + text[m.end():]
if refresh_time:
now = int(time.time())
text = _EPOCH_RE.sub(str(now), 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_collect_pure(
tdc_js_code: str,
click_points: List[Tuple[int, int]],
aid: str = "2039519451",
user_agent: str = DEFAULT_USER_AGENT,
renderer: str = DEFAULT_RENDERER,
vendor: str = DEFAULT_VENDOR,
canvas_fp: str = DEFAULT_CANVAS_FP,
node_bin: str = "node",
) -> Tuple[str, List[int]]:
"""
**纯 Python 出包入口**: Node 只负责「跑出密钥 + 一次性标定明文模板」,
轨迹生成 / 组包 / XTEA 加密 / 编码 全部由 Python 完成。
两次 Node 调用均按脚本内容缓存 (同一构建只付一次成本)。
"""
try:
template, key = recover_payload_template(tdc_js_code, node_bin=node_bin)
return generate_collect_from_template(template, click_points, key), key
except Exception:
# 模板不适用 (例如两段式旧结构) -> 退回旧版结构 + 同一把密钥
key = extract_xtea_key(tdc_js_code, node_bin=node_bin)
return generate_collect_keyed(click_points, key, aid=aid, user_agent=user_agent,
renderer=renderer, vendor=vendor,
canvas_fp=canvas_fp), key
def _collect_segments(click_points: List[Tuple[int, int]], aid: str, user_agent: str,
renderer: str, vendor: str, canvas_fp: str) -> List[bytes]:
"""组装 collect 的 4 个明文分块 (与真实 tdc.js 一致)"""
now_sec = int(time.time())
rand_salt = random.randint(100000000, 999999999)
cd_pre = CD_PRE_RAW.ljust(48, " ")
traj_payload = generate_trajectory_payload(click_points, target_len=600)
cd_post_inner = (
f',"",'
f'"undefined?rand=1519713624347",'
f"1,"
f'"{canvas_fp}",'
f"{now_sec},"
f"{rand_salt},"
f"{now_sec},"
f'"",'
f"0,"
f'"{user_agent}",'
f'"top",'
f'"MacIntel",'
f'"98k",'
f"40,"
f'"",'
f'"{renderer}",'
f"0,"
f'"{vendor}",'
f"{now_sec},"
f'"+08",'
f'["zh-CN","zh","en"],'
f"1422008362,"
f"2,"
f"3,"
f'"https://turing.captcha.qcloud.com/cap_union_new_show?aid={aid}",'
f"0,"
f"24,"
f"22,"
f"1920,"
f"1,"
f"[360,360],"
f"1080],"
)
cd_post = cd_post_inner.ljust(504, " ")
sd = '"sd":{"od":"C"}}'
return [s.encode("latin1") for s in (cd_pre, traj_payload, cd_post, sd)]
def _assemble_collect(segments: List[bytes], key: Optional[List[int]] = None) -> str:
"""4 段分别加密 + base64 + 拼接 + TDC URL 编码。
每段长度都是 3 的倍数 (48 / 24k / 504), 末段 16 字节 -> 只有末尾一个 '==',
与真实样本结构一致 (块内出现 '=' 会导致服务端解析失败)。
"""
if key is None:
enc = [encrypt_string(s.decode("latin1")) for s in segments]
else:
enc = [base64.b64encode(xtea_encrypt_bytes(s, key)).decode("ascii") for s in segments]
return "".join(enc).replace("+", "%2B").replace("/", "%2F").replace("=", "%3D")
def generate_collect_keyed(
click_points: List[Tuple[int, int]],
key: List[int],
aid: str = "2039519451",
user_agent: str = DEFAULT_USER_AGENT,
renderer: str = DEFAULT_RENDERER,
vendor: str = DEFAULT_VENDOR,
canvas_fp: str = DEFAULT_CANVAS_FP,
) -> str:
"""用「从 tdc.js 提取到的本次构建密钥」纯 Python 生成 collect (标准 XTEA)"""
segments = _collect_segments(click_points, aid, user_agent, renderer, vendor, canvas_fp)
return _assemble_collect(segments, key=key)
def generate_collect(
click_points: List[Tuple[int, int]],
aid: str = "2039519451",
user_agent: str = DEFAULT_USER_AGENT,
renderer: str = DEFAULT_RENDERER,
vendor: str = DEFAULT_VENDOR,
canvas_fp: str = DEFAULT_CANVAS_FP,
) -> str:
"""
用 crypto.py 内置的固定密钥 (DEFAULT_KEY + 两个历史加盐) 生成 collect。
等价于标准 XTEA + K' = [k0, k1+salt1, k2+salt2, k3]。
兼容性: 实测雨云 aid 直接接受 (errorCode=0), 说明服务端并不校验 collect 与 eks
的同源性; 需要严格场景请用 `generate_collect_keyed()` + `extract_xtea_key()`。
"""
segments = _collect_segments(click_points, aid, user_agent, renderer, vendor, canvas_fp)
return _assemble_collect(segments, key=None)
# --------------------------------------------------------------------------- #
# 真实 tdc.js 执行路径 (推荐): 在 Node VM 沙箱中跑当前会话的 tdc.js
# --------------------------------------------------------------------------- #
def generate_telemetry_node(
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/",
frame_size: Tuple[int, int] = (360, 360),
display_scale: Optional[float] = None,
node_bin: str = "node",
timeout: float = 60.0,
) -> Dict[str, str]:
"""
在 Node 沙箱里执行真实 tdc.js, 产出与浏览器一致的 collect / tlg / eks。
为什么必须走这条路
------------------
tdc.js 每次加载都会重新混淆, 且其 XTEA 会话密钥并非固定常量 —— 用
`har/1ms_dianxuan/live/tdc_live.js` 产出的真实 collect 用 DEFAULT_KEY 解密
得到的是随机字节 (见 ANALYSIS.md 验证记录)。因此纯 Python 硬编码密钥的实现
只对当初逆向的那一份 tdc.js 有效, 换会话即失效; 而 Node 沙箱直接执行服务端
下发的 tdc.js, 拿到的是服务端自己认可的密文。
返回: {"collect": str, "tlg": str, "eks": str, "source": "node"}
"""
script = TDC_WORKER_PATH
if not script.exists():
raise FileNotFoundError(f"缺少 tdc_worker.js: {script}")
options = {
"aid": str(aid),
"referrer": referrer,
"userAgent": user_agent,
"screenWidth": 1920,
"screenHeight": 1080,
"availHeight": 1055,
"frameWidth": frame_size[0],
"frameHeight": frame_size[1],
"offsetX": 0,
"offsetY": 0,
}
if display_scale is not None:
options["displayScale"] = float(display_scale)
cmd = [node_bin, str(script), "-", json.dumps([list(p) for p in click_points]),
json.dumps(options)]
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]}")
try:
payload = json.loads(out[-1])
except json.JSONDecodeError as exc:
raise RuntimeError(f"tdc_worker 输出无法解析: {out[-1][:300]}") from exc
if payload.get("error"):
raise RuntimeError(f"tdc_worker 执行 tdc.js 失败: {payload['error'][:400]}")
collect = payload.get("collect") or ""
if not collect:
raise RuntimeError("tdc_worker 未产出 collect")
return {
"collect": collect,
"tlg": str(payload.get("tlg") or len(collect)),
"eks": payload.get("eks") or "",
"source": "node",
}
def generate_telemetry_browser(
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/",
frame_size: Tuple[int, int] = (360, 360),
display_scale: Optional[float] = None,
chrome_path: Optional[str] = None,
headless: bool = True,
node_bin: str = "node",
timeout: float = 120.0,
) -> Dict[str, str]:
"""
用 **真实 Chrome** (CDP 驱动) 执行 tdc.js, 产出环境指纹真实度最高的 collect / eks。
与 `generate_telemetry_node` 的区别: 后者在 Node VM 里伪造 DOM/BOM/Canvas,
实测答题正确后仍会被服务端判定为环境异常 (errorCode 12); 本函数把 tdc.js 放进
真实浏览器, 用 Input.dispatchMouseEvent 派发可信鼠标事件, 再读回 TDC.getData()。
"""
script = TDC_BROWSER_PATH
if not script.exists():
raise FileNotFoundError(f"缺少 tdc_browser.js: {script}")
options = {
"aid": str(aid),
"referer": referrer,
"userAgent": user_agent,
"frameWidth": frame_size[0],
"frameHeight": frame_size[1],
"offsetX": 0,
"offsetY": 0,
"headless": headless,
}
if chrome_path:
options["chromePath"] = chrome_path
if display_scale is not None:
options["displayScale"] = float(display_scale)
cmd = [node_bin, str(script), "-", json.dumps([list(p) for p in click_points]),
json.dumps(options)]
proc = subprocess.run(cmd, input=tdc_js_code, capture_output=True, text=True, timeout=timeout)
out = (proc.stdout or "").strip().splitlines()
if not out:
raise RuntimeError(f"tdc_browser 无输出 (rc={proc.returncode}): {proc.stderr[:600]}")
try:
payload = json.loads(out[-1])
except json.JSONDecodeError as exc:
raise RuntimeError(f"tdc_browser 输出无法解析: {out[-1][:300]}") from exc
if payload.get("error"):
raise RuntimeError(f"真实 Chrome 执行 tdc.js 失败: {payload['error'][:600]}")
collect = payload.get("collect") or ""
if not collect:
raise RuntimeError("真实 Chrome 未产出 collect")
return {"collect": collect, "tlg": str(payload.get("tlg") or len(collect)),
"eks": payload.get("eks") or "", "source": "browser"}
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",
) -> Dict[str, str]:
"""
统一遥测生成入口。
mode="python" : 纯 Python 算法还原 (内置固定密钥; 默认, 无 Node 依赖, ~1ms)
mode="python_keyed": **Node 仅提取本次构建的 XTEA 密钥, 加密全部在 Python**
mode="node" : Node VM 沙箱执行服务端下发的真实 tdc.js (~0.1s)
mode="browser" : 真实 Chrome + CDP 执行真实 tdc.js (~50s)
mode="auto" : python -> node -> browser 逐级回退
为什么默认仍然是纯 Python
------------------------
实测(2026-10-11)证明服务端 **并不按 tdc.js 的会话密钥解密 collect**:
雨云 aid=2039519451 + 纯 Python(硬编码 DEFAULT_KEY) 的 collect + 正确答案
→ errorCode=0 并返回真实 ticket (tr03jvqlp225MjNBxjDj_-mnM1VTS8O9)
而 `DEFAULT_KEY` 只对仓库里那份 tdc_sample.js 有效(能解出明文
`{"cd":[[],"1920-1080..."`), 对当天实时抓的同一 aid 的脚本已失效 —— 说明
tdc.js 的密码学参数(4 个密钥字 + 2829068/328986 两个轮加盐)是**每次构建随机**的,
但服务端并不校验收到的 collect 是否与 eks 同源。
因此纯 Python 方案在协议层完全够用; 需要 node/browser 的场景是目标 aid 启用了
更严格的风控(例如 1ms.run 的 aid=195672344 在答案正确时返回 errorCode=12),
详见 har/1ms_dianxuan/ANALYSIS.md。
"""
chain = {"python": ["python"], "python_keyed": ["python_keyed"],
"node": ["node"], "browser": ["browser"],
"auto": ["python", "python_keyed", "node", "browser"],
"auto_keyed": ["python_keyed", "node", "python"]}.get(mode, ["python"])
errors = []
for kind in chain:
try:
if kind == "browser":
return generate_telemetry_browser(tdc_js_code, click_points, aid=aid,
user_agent=user_agent, referrer=referrer)
if kind == "node":
return generate_telemetry_node(tdc_js_code, click_points, aid=aid,
user_agent=user_agent, referrer=referrer)
if kind == "python_keyed":
# Node 只负责提取当前构建的 XTEA 密钥; 明文/加密/组包全部纯 Python
key = extract_xtea_key(tdc_js_code)
collect = generate_collect_success(click_points, key)
return {"collect": collect, "tlg": str(len(collect)),
"eks": extract_eks(tdc_js_code), "source": "python_keyed",
"key": key}
collect = generate_collect(click_points, aid=aid, user_agent=user_agent)
return {"collect": collect, "tlg": str(len(collect)),
"eks": extract_eks(tdc_js_code), "source": "python"}
except Exception as exc:
errors.append(f"{kind}: {exc}")
if mode != "auto":
raise
raise RuntimeError("所有遥测后端均失败: " + " | ".join(errors))
if __name__ == "__main__":
# Self-test unit verification
test4 = '"sd":{"od":"C"}}'
exp4 = "nshM2N4iPvX6JRNXi68v5w=="
act4 = encrypt_string(test4)
assert act4 == exp4, f"XTEA unit test failed: {act4} != {exp4}"
test2 = '{"cd":[[],"1920-1080-1055-24-*-*-|-*",0,0,"", '
exp2 = "9qJOPk+13StWMa2FQroCP6yba6nGw0BmJaarCCtGtLi0Wfg7WQyVXwMoFAjWcXAj"
act2 = encrypt_string(test2)
assert act2 == exp2, f"XTEA unit test failed: {act2} != {exp2}"
sample_pts = [(626, 253), (120, 120), (438, 33)]
sample_collect = generate_collect(sample_pts)
print(f"XTEA Unit Tests: PASSED")
print(f"Sample collect length: {len(sample_collect)}")
print(f"Sample collect prefix: {sample_collect[:80]}...")
Reference in New Issue View Git Blame Copy Permalink