""" 腾讯天御点选验证码纯协议客户端 (Tencent Turing Captcha Protocol Client) 覆盖腾讯天御 `cap_union_new` 协议族的完整纯协议闭环: 1. GET /cap_union_prehandle 获取会话 sess / PoW 题目 / 题型 / 首道题目 2. 本地求解 PoW (MD5 哈希碰撞) 3. GET 拉取当前会话动态混淆的 tdc.js, 提取会话密钥 eks 4. GET /cap_union_new_getcapbysig 下载 672x480 背景图 (img_index=1); 图标题型额外下载 img_index=0 提示图 5. 视觉求解: * 文字点选 (subcapclass 2404) -> tcaptcha.glyph_solver.TextClickSolver * 图标点选 (subcapclass 2408) -> tcaptcha.solver.CaptchaSolver (YOLO+ResNet18) 6. 生成行为遥测密文 collect (纯 Python 还原的自定义 XTEA + 贝塞尔轨迹) 7. POST /cap_union_new_verify 提交 ans/collect/eks/sess/pow_answer 8. 失败 → POST /cap_union_new_getsig 在同一会话内刷新下一道题并重试 题型识别不依赖 subcapclass 数字 (getsig 不返回该字段), 而是根据 instruction + bg_elem_cfg.click_cfg 推断, 详见 classify_challenge()。 """ import base64 import hashlib import json import logging import random import re import sys import time from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Sequence, 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)) from tcaptcha.crypto import extract_eks, generate_telemetry from tcaptcha.glyph_solver import TextClickSolver, parse_instruction logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger("tcaptcha_client") DEFAULT_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") # 题型 KIND_TEXT_CLICK = "text_click" # 2404 文字点选 KIND_ICON_CLICK = "icon_click" # 2408 图标点选 (需要提示图 img_index=0) KIND_SLIDER = "slider" # 2402 滑块拼图 (本客户端未实现) KIND_UNKNOWN = "unknown" @dataclass class Challenge: """一道待求解的验证码题目""" sess: str instruction: str = "" img_url: str = "" kind: str = KIND_UNKNOWN subcapclass: str = "" click_cfg: Optional[dict] = None raw: dict = field(default_factory=dict) @property def targets(self) -> List[str]: return parse_instruction(self.instruction) def classify_challenge(dyn_show_info: dict, subcapclass: str = "") -> str: """根据题目配置推断题型 (getsig 不返回 subcapclass, 只能靠结构判断)""" instruction = (dyn_show_info or {}).get("instruction", "") or "" click_cfg = ((dyn_show_info or {}).get("bg_elem_cfg") or {}).get("click_cfg") or {} data_types = click_cfg.get("data_type") or [] if "DynAnswerType_POS" in data_types or "DynAnswerType_POINT" in data_types: # 有点选坐标配置: 有目标文字 => 文字点选; 无目标文字 => 图标点选(提示走提示图) return KIND_TEXT_CLICK if parse_instruction(instruction) else KIND_ICON_CLICK if "滑块" in instruction or "拖动" in instruction or subcapclass == "2402": return KIND_SLIDER if subcapclass == "2404": return KIND_TEXT_CLICK if subcapclass == "2408": return KIND_ICON_CLICK return KIND_UNKNOWN class TencentCaptchaClient: BASE_URL = "https://turing.captcha.qcloud.com" DEFAULT_AID = "2039519451" def __init__( self, aid: str = DEFAULT_AID, solver=None, text_solver: Optional[TextClickSolver] = None, timeout: float = 15.0, user_agent: Optional[str] = None, entry_url: str = "https://app.rainyun.com/account/reward/earn#", referer: Optional[str] = None, wb: str = "2", js: str = "/tcaptcha-frame.ba387dd1.js", accept_kinds: Sequence[str] = (KIND_TEXT_CLICK, KIND_ICON_CLICK), min_posterior: float = 0.95, pow_timeout: int = 4000000, inter_attempt_delay: float = 0.25, telemetry_mode: str = "python_keyed", proxy: Optional[str] = None, ): """ telemetry_mode 决定 collect/eks 怎么来 (见 tcaptcha.crypto.generate_telemetry), 只走纯算: * "python_keyed" (默认): Node 只从本次构建的 tdc.js 里提取 XTEA 密钥字; 明文组装 / 轨迹 / 时间戳 / 加密 / 编码全部在 Python 完成。 * "python": 完全不调用 Node, 直接用内置 DEFAULT_KEY (仅对不校验遥测的宽松 aid 有效)。 关于 errorCode=12 (实测 2026-10-11, 见 har/1ms_dianxuan/ANALYSIS.md §7): 1ms.run aid=195672344 在答案正确时返回 12, 且与遥测后端无关 —— python / python_keyed / node / browser 四种后端、4 个不同省份出口 IP、 9 分钟静默后单发, **全部都是 12**。所以 12 不是遥测/环境/IP 问题, 不要再走"补环境"或"换 IP"的路; 真正的变量 (登录态 / 真实验证码页面上下文) 尚未被单变量实验区分。 """ self.aid = str(aid) self.timeout = timeout self.user_agent = user_agent or DEFAULT_UA self.entry_url = entry_url self.referer = referer or entry_url self.wb = str(wb) self.js = js self.accept_kinds = tuple(accept_kinds) self.min_posterior = min_posterior self.pow_timeout = pow_timeout # 抽题/投票之间的最小间隔 (降低接口压力; 与 errorCode=12 无因果关系) self.inter_attempt_delay = inter_attempt_delay # 遥测生成方式: 只走纯算 (见 tcaptcha.crypto.generate_telemetry) self.telemetry_mode = telemetry_mode # 出口代理 (可选)。注意: 实测换出口 IP 并不能解决 errorCode=12 self.proxy = proxy # 图标题型求解器 (惰性加载, 避免只跑文字题型时白加载 56MB 权重) self._icon_solver = solver self._text_solver = text_solver self.subsid = 1 self.http_client = httpx.Client( base_url=self.BASE_URL, trust_env=False, # 屏蔽环境代理干扰 proxy=self.proxy, headers={ "User-Agent": self.user_agent, "Referer": self.referer, "Accept-Language": "zh-CN,zh;q=0.9,fr;q=0.8,de;q=0.7,en;q=0.6", "sec-ch-ua": '"Google Chrome";v="155", "Chromium";v="155", "Not(A:Brand";v="24"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"macOS"', "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", }, timeout=self.timeout, follow_redirects=True, ) # --------------------------------------------------------------- helpers @property def text_solver(self) -> TextClickSolver: if self._text_solver is None: logger.info("初始化文字点选求解器 (ddddocr + PP-OCRv4 集成)...") self._text_solver = TextClickSolver() return self._text_solver @property def icon_solver(self): if self._icon_solver is None: from tcaptcha.solver import CaptchaSolver logger.info("初始化图标点选求解器 (YOLOv8 + ResNet18)...") self._icon_solver = CaptchaSolver() return self._icon_solver def _solve_pow(self, prefix: str, target_md5: str) -> Tuple[str, int]: """求解 Proof of Work 哈希碰撞""" t0 = time.time() for i in range(self.pow_timeout): candidate = f"{prefix}{i}" if hashlib.md5(candidate.encode("utf-8")).hexdigest() == target_md5: calc_time = max(1, round((time.time() - t0) * 1000)) return candidate, calc_time raise RuntimeError(f"PoW MD5 collision search failed for prefix {prefix}") # ------------------------------------------------------------- 1. prehandle def prehandle(self, tkid: str) -> Tuple[Challenge, dict, str]: """GET /cap_union_prehandle —— 返回 (题目, comm_captcha_cfg, sid)""" cb_name = f"_aq_{random.randint(100000, 999999)}" ua_b64 = base64.b64encode(self.user_agent.encode()).decode() self.subsid += 2 params = { "aid": self.aid, "protocol": "https", "accver": "1", "showtype": "popup", "ua": ua_b64, "noheader": "1", "fb": "1", "aged": "0", "enableAged": "0", "enableDarkMode": "0", "grayscale": "1", "clientype": "2", "cap_cd": "", "uid": "", "lang": "zh-cn", "entry_url": self.entry_url, "elder_captcha": "0", "js": self.js, "login_appid": "", "wb": self.wb, "tkid": tkid, "subsid": str(self.subsid), "callback": cb_name, "sess": "", } resp = self.http_client.get("/cap_union_prehandle", params=params, headers={"Referer": self.referer}) match = re.search(rf"{re.escape(cb_name)}\((.*)\)\s*$", resp.text, re.S) if not match: raise RuntimeError(f"prehandle 回调解析失败: {resp.text[:200]}") data = json.loads(match.group(1)) dyn = (data.get("data") or {}).get("dyn_show_info") or {} sub = str(data.get("subcapclass", "")) challenge = Challenge( sess=data.get("sess", ""), instruction=dyn.get("instruction", ""), img_url=((dyn.get("bg_elem_cfg") or {}).get("img_url") or ""), kind=classify_challenge(dyn, sub), subcapclass=sub, click_cfg=(dyn.get("bg_elem_cfg") or {}).get("click_cfg"), raw=dyn, ) return challenge, (data.get("data") or {}).get("comm_captcha_cfg", {}) or {}, data.get("sid", "") # --------------------------------------------------------------- 2. getsig def getsig(self, sess: str, tkid: str) -> Tuple[Challenge, str]: """POST /cap_union_new_getsig —— 在同一会话内刷新题目""" resp = self.http_client.post( "/cap_union_new_getsig", data={"sess": sess, "tkid": tkid}, headers={"Content-Type": "application/x-www-form-urlencoded", "Origin": "https://turing.captcha.gtimg.com", "Referer": "https://turing.captcha.gtimg.com/"}, ) payload = resp.json() if str(payload.get("ret")) != "0": raise RuntimeError(f"getsig 返回异常: {payload}") dyn = payload.get("data") or {} new_sess = payload.get("sess") or sess challenge = Challenge( sess=new_sess, instruction=dyn.get("instruction", ""), img_url=((dyn.get("bg_elem_cfg") or {}).get("img_url") or ""), kind=classify_challenge(dyn), click_cfg=(dyn.get("bg_elem_cfg") or {}).get("click_cfg"), raw=dyn, ) return challenge, new_sess # ------------------------------------------------------------ 3. download def fetch_tdc(self, tdc_path: str) -> str: resp = self.http_client.get(tdc_path, headers={"Referer": "https://turing.captcha.gtimg.com/"}) return resp.text def fetch_image(self, img_url: str, img_index: Optional[int] = None) -> bytes: if img_index is not None: img_url = re.sub(r"img_index=\d+", f"img_index={img_index}", img_url) resp = self.http_client.get( img_url, headers={"Referer": "https://turing.captcha.gtimg.com/", "Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8"}, ) return resp.content # ------------------------------------------------------------------ solve def solve(self, max_retries: int = 3, max_refreshes: int = 6) -> Dict: """ 端到端解题并提交验证。 返回: {success, ticket, randstr, elapsed_ms, points, kind, instruction, raw_response, error} """ start_time = time.time() tkid = str(random.randint(100000000, 999999999)) sess: Optional[str] = None eks: Optional[str] = None tdc_code: Optional[str] = None pow_answer: Optional[str] = None pow_calc_time: Optional[int] = None comm_cfg: Dict = {} challenge: Optional[Challenge] = None last_raw: Dict = {} last_error = "未完成任何一次验证提交" attempts = 0 refreshes = 0 prehandles = 0 while attempts < max_retries and refreshes < max_refreshes: try: # ---- 拿题: 首选 prehandle, 之后用 getsig 在同会话内刷新 round_start_time = time.time() if challenge is None: if sess: challenge, sess = self.getsig(sess, tkid) logger.info("getsig 刷新题目 -> kind=%s instruction=%r", challenge.kind, challenge.instruction) else: challenge, comm_cfg, _sid = self.prehandle(tkid) prehandles += 1 logger.info("prehandle -> subcapclass=%s kind=%s instruction=%r", challenge.subcapclass, challenge.kind, challenge.instruction) pow_cfg = comm_cfg.get("pow_cfg") or {} if pow_cfg.get("prefix") and pow_cfg.get("md5"): pow_answer, pow_calc_time = self._solve_pow(pow_cfg["prefix"], pow_cfg["md5"]) logger.info("PoW 命中 %s (%dms)", pow_answer, pow_calc_time) tdc_code = self.fetch_tdc(comm_cfg.get("tdc_path", "")) eks = extract_eks(tdc_code) if not eks: logger.warning("未从 tdc.js 中提取到 eks, 继续尝试") # 无论是预期的还是非预期题型，第一时间下载原图落盘到本地 scratch/captcha_debug/，辅助用户肉眼判断 bg_bytes: Optional[bytes] = None debug_dir = Path("scratch/captcha_debug") debug_dir.mkdir(parents=True, exist_ok=True) ts = time.strftime("%Y%m%d_%H%M%S") safe_inst = re.sub(r"[^\w\u4e00-\u9fff]+", "_", challenge.instruction).strip("_") or "no_inst" if challenge.img_url: try: bg_bytes = self.fetch_image(challenge.img_url, img_index=1) raw_path = debug_dir / f"{ts}_{challenge.kind}_{safe_inst}.jpg" raw_path.write_bytes(bg_bytes) logger.info("📸 [本地图片落盘] %s (题型: %s, 指令: %r)", raw_path, challenge.kind, challenge.instruction) except Exception as e: logger.warning("下载题目原图失败: %s", e) if challenge.kind not in self.accept_kinds: # 题型在会话内是固定的 (getsig 不会换题型), 必须重新 prehandle 抽题 # 严格模拟人类审题、发现不是目标题型、移动鼠标并点击换题的交互耗时 (2.8 ~ 4.2 秒，拒绝秒换) wait_switch = random.uniform(2.8, 4.2) logger.info("⏳ 题型 %s 不在支持范围 %s, 模拟人类思考与换题等待 %.2fs (已抽 %d 次)...", challenge.kind, list(self.accept_kinds), wait_switch, prehandles) refreshes += 1 challenge = None sess = None eks = None pow_answer = None time.sleep(wait_switch) continue # ---- 视觉求解 points: List[Tuple[int, int]] if challenge.kind == KIND_TEXT_CLICK: bg = bg_bytes if bg_bytes else self.fetch_image(challenge.img_url, img_index=1) result = self.text_solver.solve_ex(bg, challenge.instruction) logger.info("文字点选求解: %s", result.describe()) # 如果识别出坐标，生成带红圈和序号 (1, 2, 3) 的标注图便于肉眼核验 if result.points: try: import cv2 import numpy as np bgr = cv2.imdecode(np.frombuffer(bg, np.uint8), cv2.IMREAD_COLOR) for idx, pt in enumerate(result.points): cv2.circle(bgr, (int(pt[0]), int(pt[1])), 18, (0, 0, 255), 2) cv2.putText( bgr, str(idx + 1), (int(pt[0]) - 6, int(pt[1]) + 7), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2, ) anno_path = debug_dir / f"{ts}_{challenge.kind}_{safe_inst}_annotated.jpg" cv2.imwrite(str(anno_path), bgr) logger.info("📸 [标注图已落盘] %s", anno_path) except Exception as e: logger.warning("生成标注图失败: %s", e) if not result.ok: wait_refresh = random.uniform(2.5, 3.8) logger.warning("文字点选求解失败(未识别全目标字), 模拟人类找字无果换题等待 %.2fs", wait_refresh) refreshes += 1 challenge = None time.sleep(wait_refresh) continue if result.posterior < self.min_posterior or result.min_target_score < 0.2: wait_refresh = random.uniform(2.5, 3.8) logger.warning("求解置信度过低 (posterior=%.3f, min_score=%.3f), 模拟犹豫换题等待 %.2fs", result.posterior, result.min_target_score, wait_refresh) refreshes += 1 challenge = None time.sleep(wait_refresh) continue points = result.points else: # KIND_ICON_CLICK bg = bg_bytes if bg_bytes else self.fetch_image(challenge.img_url, img_index=1) prompt = self.fetch_image(challenge.img_url, img_index=0) if prompt: prompt_path = debug_dir / f"{ts}_icon_prompt_{safe_inst}.jpg" prompt_path.write_bytes(prompt) logger.info("📸 [提示图落盘] %s", prompt_path) else: wait_refresh = random.uniform(2.5, 3.8) logger.warning("图标题型未取到提示图, 等待 %.2fs 后刷新题目", wait_refresh) refreshes += 1 time.sleep(wait_refresh) continue points = self.icon_solver.solve(bg, prompt) if len(points) < 3: wait_refresh = random.uniform(2.5, 3.8) logger.warning("图标点选求解失败, 等待 %.2fs 后刷新题目", wait_refresh) refreshes += 1 time.sleep(wait_refresh) continue # ---- 模拟真实人类阅读审题、定位字形与逐个点击耗时 (5.5 ~ 7.5 秒，对齐抓包实测 7.41s) elapsed = time.time() - round_start_time target_duration = random.uniform(5.5, 7.5) if elapsed < target_duration: sleep_secs = target_duration - elapsed logger.info("模拟人类阅读审题与点击耗时: 等待 %.2fs (目标交互时长 %.2fs)", sleep_secs, target_duration) time.sleep(sleep_secs) # ---- 组包并提交 attempts += 1 ans_list = [{"elem_id": i + 1, "type": "DynAnswerType_POS", "data": f"{p[0]},{p[1]}"} for i, p in enumerate(points)] ans_str = json.dumps(ans_list, separators=(",", ":")) telemetry = generate_telemetry( tdc_code or "", points, aid=self.aid, user_agent=self.user_agent, referrer=self.referer, mode=self.telemetry_mode, ) logger.info("遥测密文来源=%s collect_len=%s eks_len=%s", telemetry.get("source"), len(telemetry["collect"]), len(telemetry.get("eks", ""))) verify_body = { "ans": ans_str, "collect": telemetry["collect"], "tlg": telemetry["tlg"], "eks": telemetry.get("eks") or eks or "", "sess": challenge.sess, "tkid": tkid, "pow_answer": pow_answer or "", "pow_calc_time": str(pow_calc_time or 1), } verify_resp = self.http_client.post( "/cap_union_new_verify", data=verify_body, headers={ "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "Accept": "application/json, text/javascript, */*; q=0.01", "Origin": "https://turing.captcha.gtimg.com", "Referer": "https://turing.captcha.gtimg.com/", }, ) v_json = verify_resp.json() last_raw = v_json elapsed = int((time.time() - start_time) * 1000) if str(v_json.get("errorCode")) == "0": logger.info("✅ 验证通过! randstr=%s", v_json.get("randstr")) return { "success": True, "ticket": v_json.get("ticket", ""), "randstr": v_json.get("randstr", ""), "elapsed_ms": elapsed, "points": points, "kind": challenge.kind, "instruction": challenge.instruction, "raw_response": v_json, "error": "", } err_code = str(v_json.get("errorCode")) if err_code == "12": last_error = ("服务端返回 errorCode=12 (腾讯天御风控评估未通过)。" "实测与遥测后端 / 出口 IP / 静默时长均无关 " "(见 har/1ms_dianxuan/ANALYSIS.md §7), 不要再尝试换环境或换 IP") logger.error(last_error) break if err_code in ("50", "9"): last_error = f"坐标未命中 (errorCode: {err_code})" else: last_error = (f"验证失败 (errorCode: {err_code}, " f"msg: {v_json.get('errMessage', '')})") logger.warning(last_error) # 服务端返回新 sess 时沿用, 否则回到 prehandle new_sess = v_json.get("sess") if new_sess: sess = new_sess challenge = None else: challenge = None sess = None eks = None pow_answer = None except Exception as exc: # 网络/解析异常: 丢弃当前会话重来 last_error = f"请求异常: {exc}" logger.error("第 %d 次尝试异常: %s", attempts + 1, exc) challenge = None sess = None eks = None pow_answer = None elapsed = int((time.time() - start_time) * 1000) return { "success": False, "ticket": "", "randstr": "", "elapsed_ms": elapsed, "points": [], "kind": challenge.kind if challenge else "", "instruction": challenge.instruction if challenge else "", "raw_response": last_raw, "error": last_error, } def close(self): self.http_client.close() def main(): import argparse parser = argparse.ArgumentParser(description="Tencent Captcha Pure Protocol Solver Client") parser.add_argument("--aid", default="195672344", help="验证码 AppID (默认毫秒 1ms.run)") parser.add_argument("--entry-url", default="https://1ms.run/console/personal/benefits/check-in") parser.add_argument("--wb", default="1") parser.add_argument("--retries", type=int, default=3) parser.add_argument("--telemetry", default="python_keyed", choices=["python_keyed", "python"], help="遥测方式: python_keyed(Node 仅提密钥) / python(纯 Python 固定密钥)") parser.add_argument("--proxy", default=None, help="出口代理, 如 http://1.2.3.4:8080") parser.add_argument("--proxy-api", default=None, help="代理提取 API (可选; 注意换 IP 并不能解决 errorCode=12)") args = parser.parse_args() proxy = args.proxy if not proxy and args.proxy_api: from tcaptcha.proxy import fetch_proxy proxy = fetch_proxy(args.proxy_api) print(f"[*] 从代理 API 提取到: {proxy}") client = TencentCaptchaClient(aid=args.aid, entry_url=args.entry_url, referer="https://1ms.run/", wb=args.wb, proxy=proxy, telemetry_mode=args.telemetry) print(f"[*] 正在发起腾讯天御验证码纯协议识别 (AppID: {args.aid}, proxy={proxy})...") result = client.solve(max_retries=args.retries) # 注意: errorCode=12 实测与出口 IP 无关 (ANALYSIS.md §7), 这里不再自动换 IP。 # 如需指定出口 IP, 请在启动时用 --proxy / --proxy-api。 if result["success"]: print("\n" + "=" * 50) print(" 🎉 验证码纯协议求解成功！") print("=" * 50) print(f"耗时: {result['elapsed_ms']} ms") print(f"题型: {result['kind']} 题目: {result['instruction']!r}") print(f"点击坐标: {result['points']}") print(f"randstr: {result['randstr']}") print(f"ticket: {result['ticket']}") print("=" * 50) else: print(f"\n❌ 验证失败: {result['error']}") sys.exit(1) if __name__ == "__main__": main()