feat(tcaptcha): 完成腾讯天御点选纯协议全自动化客户端与雨云业务适配器

tcaptcha/client.py

@@ -0,0 +1,327 @@
+"""
+腾讯天御点选验证码工业级纯协议客户端 (Tencent Turing Captcha Protocol Client)
+实现端到端全自动纯协议闭环：
+1. GET /cap_union_prehandle 获取初始会话与 PoW 题目
+2. 毫秒级求解 Proof of Work 哈希碰撞
+3. 并发拉取动态混淆的 tdc.js 脚本
+4. POST /cap_union_new_getsig 获取当前轮次题目与图片
+5. YOLOv8n + ResNet18 模型进行视觉检测、分类与全局非冲突点选排序
+6. Node.js 沙箱环境运行 tdc.js 并拟合生成拟人化贝塞尔轨迹密文 collect 与 eks
+7. POST /cap_union_new_verify 提交验证，获取最终 ticket 与 randstr
+"""
+
+import hashlib
+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
+
+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.solver import CaptchaSolver
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+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,
+        aid: str = DEFAULT_AID,
+        solver: Optional[CaptchaSolver] = None,
+        timeout: float = 15.0,
+        user_agent: Optional[str] = None,
+    ):
+        self.aid = aid
+        self.timeout = timeout
+        self.user_agent = user_agent or (
+            "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"
+        )
+        self.solver = solver or CaptchaSolver()
+
+        self.http_client = httpx.Client(
+            base_url=self.BASE_URL,
+            trust_env=False,  # 屏蔽环境代理干扰
+            headers={
+                "User-Agent": self.user_agent,
+                "Referer": "https://turing.captcha.qcloud.com/",
+                "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+            },
+            timeout=self.timeout,
+            follow_redirects=True,
+        )
+
+    def _solve_pow(self, prefix: str, target_md5: str) -> Tuple[str, int]:
+        """求解 Proof of Work 哈希碰撞"""
+        t0 = time.time()
+        for i in range(2000000):
+            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}")
+
+    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())
+        return {
+            "collect": data.get("collect", ""),
+            "tlg": str(data.get("tlg", 0)),
+            "eks": data.get("eks", ""),
+        }
+
+    def solve(self, max_retries: int = 2) -> Dict:
+        """
+        端到端全自动解题并提交验证
+        返回格式:
+        {
+            "success": bool,
+            "ticket": str,
+            "randstr": str,
+            "elapsed_ms": int,
+            "points": [(x1, y1), (x2, y2), (x3, y3)],
+            "raw_response": dict,
+            "error": str
+        }
+        """
+        start_time = time.time()
+        active_sess = None
+        tkid = None
+        tdc_code = None
+        pow_answer = None
+        pow_calc_time = None
+        last_raw = {}
+        last_error = "Max retries exceeded or verification rejected"
+
+        for attempt in range(max_retries):
+            try:
+                # 1. 预检请求 (Prehandle) 或在会话内刷新
+                cb_name = f"_aq_{random.randint(100000, 999999)}"
+                if not tkid:
+                    tkid = str(random.randint(100000000, 999999999))
+
+                if not active_sess:
+                    # 首次进入：执行 Prehandle
+                    ua_b64 = "TW96aWxsYS81LjAgKE1hY2ludG9zaDsgSW50ZWwgTWFjIE9TIFggMTBfMTVfNykgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlrZSBHZWNrbykgQ2hyb21lLzE1Mi4wLjAuMCBTYWZhcmkvNTM3LjM2"
+                    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",
+                        "lang": "zh-cn",
+                        "entry_url": "https://app.rainyun.com/account/reward/earn#",
+                        "elder_captcha": "0",
+                        "js": "/tcaptcha-frame.ba387dd1.js",
+                        "wb": "2",
+                        "tkid": tkid,
+                        "subsid": str(random.randint(1, 5)),
+                        "callback": cb_name,
+                    }
+
+                    pre_resp = self.http_client.get(
+                        "/cap_union_prehandle",
+                        params=params,
+                        headers={"Referer": "https://app.rainyun.com/"},
+                    )
+                    match = re.search(rf"{cb_name}\((.*)\)", pre_resp.text)
+                    if not match:
+                        logger.warning("Failed to parse prehandle callback, retrying...")
+                        continue
+
+                    pre_data = json.loads(match.group(1))
+                    initial_sess = pre_data.get("sess")
+                    comm_cfg = pre_data.get("data", {}).get("comm_captcha_cfg", {})
+
+                    # 2. 计算 PoW
+                    pow_cfg = comm_cfg.get("pow_cfg", {})
+                    pow_answer, pow_calc_time = self._solve_pow(pow_cfg["prefix"], pow_cfg["md5"])
+
+                    # 3. 拉取当前会话专属动态 tdc.js
+                    tdc_path = comm_cfg.get("tdc_path", "")
+                    tdc_resp = self.http_client.get(
+                        tdc_path,
+                        headers={"Referer": "https://turing.captcha.gtimg.com/"},
+                    )
+                    tdc_code = tdc_resp.text
+                    current_sess = initial_sess
+                else:
+                    # 会话内刷新：直接使用前一轮返回的 sess
+                    current_sess = active_sess
+
+                # 4. 获取题目信息与图片 URL (POST /cap_union_new_getsig)
+                sig_resp = self.http_client.post(
+                    "/cap_union_new_getsig",
+                    data={"sess": current_sess, "tkid": tkid},
+                    headers={
+                        "Content-Type": "application/x-www-form-urlencoded",
+                        "Origin": "https://turing.captcha.gtimg.com",
+                        "Referer": "https://turing.captcha.gtimg.com/",
+                    },
+                )
+                sig_json = sig_resp.json()
+                if str(sig_json.get("ret")) != "0":
+                    logger.warning(f"getsig returned non-zero ret: {sig_json}")
+                    active_sess = None
+                    continue
+
+                dyn_info = sig_json.get("data", {})
+                bg_url = dyn_info.get("bg_elem_cfg", {}).get("img_url", "")
+                prompt_url = bg_url.replace("img_index=1", "img_index=0")
+
+                # 提取图片 URL 中的最新有效 sess
+                sess_match = re.search(r"sess=([^&]+)", bg_url)
+                active_sess = sess_match.group(1) if sess_match else current_sess
+
+                # 5. 下载背景大图与提示图标图
+                bg_bytes = self.http_client.get(
+                    bg_url,
+                    headers={"Referer": "https://turing.captcha.gtimg.com/"},
+                ).content
+                prompt_bytes = self.http_client.get(
+                    prompt_url,
+                    headers={"Referer": "https://turing.captcha.gtimg.com/"},
+                ).content
+
+                # 6. AI 模型高精度求解
+                pts = self.solver.solve(bg_bytes, prompt_bytes)
+                if len(pts) < 3:
+                    logger.warning("Solver failed to find 3 valid points, retrying...")
+                    continue
+
+                # 7. 格式化答案 ans
+                ans_list = [
+                    {"elem_id": 1, "type": "DynAnswerType_POS", "data": f"{pts[0][0]},{pts[0][1]}"},
+                    {"elem_id": 2, "type": "DynAnswerType_POS", "data": f"{pts[1][0]},{pts[1][1]}"},
+                    {"elem_id": 3, "type": "DynAnswerType_POS", "data": f"{pts[2][0]},{pts[2][1]}"},
+                ]
+                ans_str = json.dumps(ans_list, separators=(",", ":"))
+
+                # 8. Node 沙箱环境生成真实轨迹 collect 与 eks
+                telemetry = self._generate_telemetry(tdc_code, pts)
+
+                # 9. 提交验证 POST /cap_union_new_verify
+                verify_body = {
+                    "ans": ans_str,
+                    "collect": telemetry["collect"],
+                    "tlg": telemetry["tlg"],
+                    "eks": telemetry["eks"],
+                    "sess": active_sess,
+                    "tkid": tkid,
+                    "pow_answer": pow_answer,
+                    "pow_calc_time": str(pow_calc_time),
+                }
+
+                verify_resp = self.http_client.post(
+                    "/cap_union_new_verify",
+                    data=verify_body,
+                    headers={
+                        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
+                        "Origin": "https://turing.captcha.gtimg.com",
+                        "Referer": "https://turing.captcha.gtimg.com/",
+                    },
+                )
+
+                v_json = verify_resp.json()
+                total_elapsed = int((time.time() - start_time) * 1000)
+
+                if str(v_json.get("errorCode")) == "0":
+                    return {
+                        "success": True,
+                        "ticket": v_json.get("ticket", ""),
+                        "randstr": v_json.get("randstr", ""),
+                        "elapsed_ms": total_elapsed,
+                        "points": pts,
+                        "raw_response": v_json,
+                        "error": "",
+                    }
+                else:
+                    err_code = str(v_json.get("errorCode"))
+                    err_msg = v_json.get("errMessage", "")
+                    if err_code == "12":
+                        last_error = f"触发腾讯接口频控限制 (errorCode: 12, f_limit)，建议稍候或更换IP"
+                    elif err_code == "50":
+                        last_error = f"验证坐标未命中或会话需刷新 (errorCode: 50)"
+                    else:
+                        last_error = f"验证失败 (errorCode: {err_code}, msg: {err_msg})"
+
+                    logger.warning(f"Attempt {attempt + 1} verify: {last_error}")
+                    last_raw = v_json
+                    # 如果接口返回了新的会话 sess，下一轮直接沿用该会话重试刷新
+                    new_sess = v_json.get("sess")
+                    if new_sess:
+                        active_sess = new_sess
+                    else:
+                        active_sess = None
+            except Exception as e:
+                last_error = f"请求异常: {e}"
+                logger.error(f"Attempt {attempt + 1} encountered exception: {e}")
+
+        total_elapsed = int((time.time() - start_time) * 1000)
+        return {
+            "success": False,
+            "ticket": "",
+            "randstr": "",
+            "elapsed_ms": total_elapsed,
+            "points": [],
+            "raw_response": last_raw,
+            "error": last_error,
+        }
+
+
+def main():
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Tencent Captcha Pure Protocol Solver Client")
+    parser.add_argument("--aid", default="2039519451", help="Captcha AppID (default: 2039519451 for Rainyun)")
+    args = parser.parse_args()
+
+    client = TencentCaptchaClient(aid=args.aid)
+    print(f"[*] 正在发起腾讯天御验证码纯协议识别 (AppID: {args.aid})...")
+    result = client.solve()
+
+    if result["success"]:
+        print("\n" + "=" * 50)
+        print("  🎉 验证码纯协议求解成功！")
+        print("=" * 50)
+        print(f"耗时: {result['elapsed_ms']} ms")
+        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()