hello_js_reverse_skill/tcaptcha/one_ms_adapter.py

# -*- coding: utf-8 -*-
"""
毫秒 (1ms.run) 业务适配器 —— 腾讯天御点选验证码凭证获取

协议来源: `/Users/yml/Desktop/抓包/毫秒/点选.har`

从抓包可还原的链路 (本适配器已全部实现):

    第一步  毫秒取验证码参数
            POST https://1ms.run/api/v1/captcha/params
            body {"scene": "checkin"}          (也可以带 identity_context)
            -> {"provider":"tencent","params":{"captchaAppId":195672344},
                "sessionUuid":"01a12a04-e5e7-726b-b00c-2f54b2e1c167"}

    第二步  腾讯天御 cap_union_new 全套纯协议解题
            (prehandle -> PoW -> tdc.js -> getcapbysig -> verify)
            -> {"ticket": "tr03...", "randstr": "@..."}

说明与边界
----------
* `identity_context` 在抓包中是页面内加密上下文 (emctx1.*), 实测 **不传同样可以** 拿到
  `captchaAppId`, 因此本适配器不依赖它; 若后续业务要求, 可通过参数显式传入。
* 抓包中 **不包含** 最终的签到提交请求 (`/api/v1/...` 业务接口), 因此本适配器只负责
  产出 `ticket` + `randstr`, 并给出一个可配置的 `submit_business()` 便于对接真实业务
  接口 (端点与字段需按实际业务抓包填写, 严禁凭猜测写死)。
"""

import logging
from dataclasses import dataclass
from typing import Dict, Optional, Tuple

import httpx

from tcaptcha.client import KIND_TEXT_CLICK, TencentCaptchaClient

logger = logging.getLogger("one_ms_adapter")

ONE_MS_BASE = "https://1ms.run"
DEFAULT_SCENE = "checkin"
DEFAULT_REFERER = "https://1ms.run/console/personal/benefits/check-in"
DEFAULT_USER_AGENT = ("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")


@dataclass
class CaptchaParams:
    provider: str
    app_id: str
    session_uuid: str
    enabled: bool


class OneMsAdapter:
    """1ms.run 验证码凭证适配器"""

    def __init__(self, cookie: Optional[str] = None, user_agent: str = DEFAULT_USER_AGENT,
                 timeout: float = 15.0, proxy: Optional[str] = None,
                 proxy_api: Optional[str] = None, telemetry_mode: str = "python_keyed",
                 captcha_cookies: Optional[Dict[str, str]] = None):
        self.user_agent = user_agent
        self.timeout = timeout
        # 出口代理 (可选)。注意: 实测换出口 IP 并不能解决 errorCode=12 (见 ANALYSIS.md §7)
        self.proxy = proxy
        self.proxy_api = proxy_api
        self.telemetry_mode = telemetry_mode
        # 可选: 透传到 captcha 域的腾讯云登录 cookie (账本 §7.5 的 H3 实验入口)
        self.captcha_cookies = captcha_cookies
        headers = {
            "User-Agent": user_agent,
            "Origin": ONE_MS_BASE,
            "Referer": DEFAULT_REFERER,
            "Accept": "application/json, text/plain, */*",
            "Content-Type": "application/json",
        }
        if cookie:
            headers["Cookie"] = cookie
        self.http = httpx.Client(base_url=ONE_MS_BASE, headers=headers, timeout=timeout,
                                 trust_env=False, follow_redirects=True)

    # ---------------------------------------------------------- 第一步
    def fetch_captcha_params(self, scene: str = DEFAULT_SCENE,
                             identity_context: str = "") -> CaptchaParams:
        """POST /api/v1/captcha/params —— 取当前场景的验证码配置"""
        payload: Dict[str, str] = {"scene": scene}
        if identity_context:
            payload["identity_context"] = identity_context
        resp = self.http.post("/api/v1/captcha/params", json=payload)
        data = resp.json()
        if data.get("code") != 0:
            raise RuntimeError(f"取验证码参数失败: {data}")
        info = data.get("data") or {}
        params = info.get("params") or {}
        return CaptchaParams(
            provider=info.get("provider", ""),
            app_id=str(params.get("captchaAppId", "")),
            session_uuid=info.get("sessionUuid", ""),
            enabled=bool(info.get("enabled")),
        )

    # ---------------------------------------------------------- 第二步
    def get_captcha_ticket(self, scene: str = DEFAULT_SCENE,
                           max_retries: int = 3,
                           max_proxy_rotations: int = 3) -> Tuple[Optional[str], Optional[str], Dict]:
        """
        取得可提交给毫秒业务接口的 (ticket, randstr)。

        仅在**网络/代理异常**时换出口 IP 重试 (最多 max_proxy_rotations 次)。
        业务层失败 (errorCode 12 / 50) **不换 IP** —— 实测换 IP 对 errorCode=12
        无效 (见 har/1ms_dianxuan/ANALYSIS.md §7)。
        """
        params = self.fetch_captcha_params(scene=scene)
        logger.info("毫秒验证码参数: provider=%s appId=%s enabled=%s sessionUuid=%s",
                    params.provider, params.app_id, params.enabled, params.session_uuid)
        if not params.enabled or not params.app_id:
            return None, None, {"error": "该场景未启用验证码"}
        if params.provider != "tencent":
            return None, None, {"error": f"暂只支持 tencent provider, 实际为 {params.provider}"}

        proxy = self.proxy
        if not proxy and self.proxy_api:
            from tcaptcha.proxy import fetch_proxy

            proxy = fetch_proxy(self.proxy_api)
            logger.info("从代理池初始化出口代理: %s", proxy)

        last: Dict = {}
        for rotation in range(max(1, max_proxy_rotations)):
            client = TencentCaptchaClient(
                aid=params.app_id,
                entry_url=DEFAULT_REFERER,
                referer=DEFAULT_REFERER,
                wb="1",
                accept_kinds=(KIND_TEXT_CLICK,),
                proxy=proxy,
                telemetry_mode=self.telemetry_mode,
                cookies=self.captcha_cookies,
            )
            connectivity_error = False
            try:
                result = client.solve(max_retries=max_retries, max_refreshes=8)
            except Exception as e:
                logger.error("客户端解题发生异常 (可能代理失效或超时): %s", e)
                result = {"success": False, "error": str(e)}
                connectivity_error = True
            finally:
                client.close()

            last = result
            if result.get("success"):
                return result.get("ticket") or None, result.get("randstr") or None, result

            # 业务层失败 (坐标错 / 风控 12) 与出口 IP 无关: 不换 IP, 直接返回
            if not connectivity_error or not self.proxy_api:
                break

            from tcaptcha.proxy import fetch_proxy
            new_proxy = fetch_proxy(self.proxy_api)
            if new_proxy:
                proxy = new_proxy
                logger.warning("代理不可用, 换出口 IP 重试 (%d/%d): %s",
                               rotation + 1, max_proxy_rotations, proxy)
            else:
                logger.warning("未能从代理 API 取到新代理，终止重试")
                break
        return None, None, last

    # ---------------------------------------------------------- 业务提交
    def submit_business(self, path: str, payload: Dict, method: str = "POST") -> Dict:
        """
        用 ticket/randstr 提交真实业务请求。

        注意: 点选.har 中 **没有** 最终的签到业务请求, 因此这里不预设端点与字段,
        由调用方按实际业务抓包传入 (通常是 {"vticket": ticket, "vrandstr": randstr, ...})。
        """
        if method.upper() == "POST":
            resp = self.http.post(path, json=payload)
        else:
            resp = self.http.get(path, params=payload)
        try:
            return resp.json()
        except Exception:
            return {"code": resp.status_code, "text": resp.text}

    def close(self):
        self.http.close()


if __name__ == "__main__":
    import argparse
    import json

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    ap = argparse.ArgumentParser(description="1ms.run 验证码票据获取")
    ap.add_argument("--scene", default=DEFAULT_SCENE)
    ap.add_argument("--cookie", default="")
    ap.add_argument("--proxy", default=None, help="出口代理, 如 http://1.2.3.4:8080")
    ap.add_argument("--proxy-api", default=None, help="代理提取 API (遇到 errorCode=12 自动换 IP)")
    args = ap.parse_args()

    adapter = OneMsAdapter(cookie=args.cookie or None, proxy=args.proxy,
                           proxy_api=args.proxy_api)
    try:
        ticket, randstr, raw = adapter.get_captcha_ticket(scene=args.scene)
    finally:
        adapter.close()
    print(json.dumps({"ticket": ticket, "randstr": randstr,
                      "error": raw.get("error", "")}, ensure_ascii=False, indent=2))