feat(tcaptcha): 完成腾讯天御TDC虚拟机纯算法逆向还原与100%纯Python独立客户端

tcaptcha/client.py

@@ -15,18 +15,18 @@ import json
 import logging
 import random
 import re
-import subprocess
 import sys
 import time
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 
+import httpx
+
 PROJECT_ROOT = Path(__file__).resolve().parent.parent
 if str(PROJECT_ROOT) not in sys.path:
     sys.path.insert(0, str(PROJECT_ROOT))
 
-import httpx
+from tcaptcha.crypto import extract_eks, generate_collect
-
 from tcaptcha.solver import CaptchaSolver
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
@@ -36,7 +36,6 @@ logger = logging.getLogger("tcaptcha_client")
 class TencentCaptchaClient:
     BASE_URL = "https://turing.captcha.qcloud.com"
     DEFAULT_AID = "2039519451"
-    WORKER_SCRIPT = PROJECT_ROOT / "tcaptcha/tdc_worker.js"
 
     def __init__(
         self,
@@ -76,20 +75,13 @@ class TencentCaptchaClient:
         raise RuntimeError(f"PoW MD5 collision search failed for prefix {prefix}")
 
     def _generate_telemetry(self, tdc_code: str, click_points: List[Tuple[int, int]]) -> Dict[str, str]:
-        """通过 Node.js 沙箱运行 tdc.js 生成真实行为轨迹遥测数据"""
+        """纯 Python 算法生成真实行为轨迹与遥测密文 collect 及 eks (100% 纯算免 Node.js)"""
-        points_arg = json.dumps([[int(p[0]), int(p[1])] for p in click_points])
+        eks = extract_eks(tdc_code)
-        proc = subprocess.run(
+        collect = generate_collect(click_points, aid=self.aid, user_agent=self.user_agent)
-            ["node", str(self.WORKER_SCRIPT), "-", points_arg],
-            input=tdc_code,
-            text=True,
-            capture_output=True,
-            check=True,
-        )
-        data = json.loads(proc.stdout.strip())
         return {
-            "collect": data.get("collect", ""),
+            "collect": collect,
-            "tlg": str(data.get("tlg", 0)),
+            "tlg": str(len(collect)),
-            "eks": data.get("eks", ""),
+            "eks": eks,
         }
 
     def solve(self, max_retries: int = 2) -> Dict:

tcaptcha/crypto.py

@@ -0,0 +1,312 @@
+"""
+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 math
+import random
+import re
+import struct
+import time
+from typing import List, Tuple
+
+# 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 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]] = []
+    events.append([current_pt[0], current_pt[1], 2])  # 2 represents mouse movement event
+
+    last_x, last_y = current_pt[0], current_pt[1]
+
+    for target in mapped_targets:
+        steps = random.randint(16, 24)
+        curve = _bezier_interp((last_x, last_y), target, steps)
+
+        for pt in curve[1:]:
+            dx = pt[0] - last_x
+            dy = pt[1] - last_y
+            dt = random.choice([0, 0, 0, 1])
+            events.append([dx, dy, dt])
+            last_x, last_y = pt[0], pt[1]
+
+        # Dwell hover on target icon
+        dwell_count = random.randint(3, 5)
+        for _ in range(dwell_count):
+            dx = random.randint(-1, 1)
+            dy = random.randint(-1, 1)
+            events.append([dx, dy, random.choice([0, 1])])
+            last_x += dx
+            last_y += dy
+
+    import json
+    raw_json = json.dumps(events, separators=(",", ":"))
+
+    # Pad with spaces to multiple of 8 bytes (matching target_len, e.g. 600)
+    desired_len = max(len(raw_json), target_len)
+    pad_needed = (8 - (desired_len % 8)) % 8
+    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 ""
+
+
+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:
+    """
+    Generates the complete encrypted telemetry payload (`collect`) in 100% pure Python.
+
+    Assembles:
+      1. cd prefix (48 bytes, space padded)
+      2. mouse trajectory (space padded to 8-byte multiple)
+      3. cd post-body (504 bytes, space padded)
+      4. sd suffix (16 bytes)
+    Encrypts each chunk with the 32-round custom XTEA algorithm, concatenates
+    the Base64 results, and applies TDC URL encoding (%2B, %2F, %3D).
+    """
+    now_sec = int(time.time())
+    rand_salt = random.randint(100000000, 999999999)
+
+    # 1. cd prefix (48 bytes)
+    cd_pre_raw = '{"cd":[[],"1920-1080-1055-24-*-*-|-*",0,0,"",'
+    cd_pre = cd_pre_raw.ljust(48, " ")
+
+    # 2. Mouse trajectory (600 bytes)
+    traj_payload = generate_trajectory_payload(click_points, target_len=600)
+
+    # 3. cd post-body (504 bytes)
+    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, " ")
+
+    # 4. sd suffix (16 bytes)
+    sd = '"sd":{"od":"C"}}'
+
+    # Encrypt all 4 parts
+    enc_cd_pre = encrypt_string(cd_pre)
+    enc_traj = encrypt_string(traj_payload)
+    enc_cd_post = encrypt_string(cd_post)
+    enc_sd = encrypt_string(sd)
+
+    combined_b64 = enc_cd_pre + enc_traj + enc_cd_post + enc_sd
+
+    # TDC URL encoding
+    collect = (
+        combined_b64
+        .replace("+", "%2B")
+        .replace("/", "%2F")
+        .replace("=", "%3D")
+    )
+    return collect
+
+
+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]}...")