diff --git a/tcaptcha/client.py b/tcaptcha/client.py index 67b2b7d..571a3b9 100644 --- a/tcaptcha/client.py +++ b/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 生成真实行为轨迹遥测数据""" - points_arg = json.dumps([[int(p[0]), int(p[1])] for p in click_points]) - proc = subprocess.run( - ["node", str(self.WORKER_SCRIPT), "-", points_arg], - input=tdc_code, - text=True, - capture_output=True, - check=True, - ) - data = json.loads(proc.stdout.strip()) + """纯 Python 算法生成真实行为轨迹与遥测密文 collect 及 eks (100% 纯算免 Node.js)""" + eks = extract_eks(tdc_code) + collect = generate_collect(click_points, aid=self.aid, user_agent=self.user_agent) return { - "collect": data.get("collect", ""), - "tlg": str(data.get("tlg", 0)), - "eks": data.get("eks", ""), + "collect": collect, + "tlg": str(len(collect)), + "eks": eks, } def solve(self, max_retries: int = 2) -> Dict: diff --git a/tcaptcha/crypto.py b/tcaptcha/crypto.py new file mode 100644 index 0000000..4433e97 --- /dev/null +++ b/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(" 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]}...")