yml2213
06c4348163
做了什么
- 新增 tcaptcha/http_transport.py: curl_cffi(Chrome) 传输层壳, 对上层暴露 httpx.Client 同款
get/post/close; TencentCaptchaClient 增加 transport="httpx"|"curl" + impersonate 参数。
- client.py 按真实抓包(har/签到成功-1.har)逐 endpoint 保真请求头:
* prehandle/tdc.js/getcapbysig: Sec-Fetch-Dest=script/image + Mode=no-cors +
Storage-Access=active + captcha 域 cookie, Referer 为 site-origin / gtimg;
* getsig/verify: empty/cors + Origin gtimg, 且 **不带** cookie (SameSite=Lax 实测如此)。
cookie 由全局头改为按调用点附加 (此前 verify 也带, 与真实浏览器不符)。
- 新增 scripts/experiment_transport_ab.py: 只改传输层的单变量 A/B。
为什么 (AGENTS §2.1 / §5.7)
- 抓包里 1ms 的真机成功样本(errorCode=0)与真机失败样本(50/50/9)证明: 12 是"请求方身份"判定,
与答案/载荷无关 —— 需要把客户端与真机的每一处可测差异逐条单变量掉。
怎么验证 (2026-10-11 实测, 全部仍 12)
- 真实 cookie 值 (uin/skey/qcmainCSRFToken/intl) -> 12 (H3 用真值再次排除)
- curl_cffi impersonate=chrome (JA3/JA4 + HTTP/2 + 头序) + 真实 cookie -> 12 (传输层排除)
- curl_cffi + 逐 endpoint 头保真 (Sec-Fetch-*/cookie 位置) -> 12 (头保真排除)
- 我们的 collect 明文与 1ms 真实成功样本逐字节同构 (同 1353B/同字段序/同取值) -> 12 (内容排除)
- 反向: 用纯 Python 提取的密钥成功解密 1ms 真实成功样本的 collect -> 密钥链路对真机样本成立
- .venv/bin/python -m tcaptcha.selftest 全绿
账本 har/1ms_dianxuan/PROGRESS.md: 已知 #23~#25; 已排除表新增四行; 下一步改为 (a) 真机桥接保底 /
(b) identity_context 研究项。
664 lines
31 KiB
Python
664 lines
31 KiB
Python
"""
|
|
腾讯天御点选验证码纯协议客户端 (Tencent Turing Captcha Protocol Client)
|
|
|
|
覆盖腾讯天御 `cap_union_new` 协议族的完整纯协议闭环:
|
|
|
|
1. GET /cap_union_prehandle 获取会话 sess / PoW 题目 / 题型 / 首道题目
|
|
2. 本地求解 PoW (MD5 哈希碰撞)
|
|
3. GET <tdc_path> 拉取当前会话动态混淆的 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
|
|
from urllib.parse import urlsplit
|
|
|
|
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
|
|
|
|
|
|
def _site_origin(url: str) -> str:
|
|
"""取 `scheme://host[:port]/` —— 页内 <script>/<img> 资源的 Referer 是 site-only。"""
|
|
try:
|
|
parts = urlsplit(url)
|
|
if parts.scheme and parts.netloc:
|
|
return f"{parts.scheme}://{parts.netloc}/"
|
|
except Exception: # pragma: no cover
|
|
pass
|
|
return url
|
|
|
|
|
|
def _build_transport(transport: str, *, base_url: str, headers: Dict[str, str], timeout: float,
|
|
proxy: Optional[str], impersonate: Optional[str]):
|
|
"""
|
|
构造 HTTP 传输层。
|
|
|
|
* transport="httpx" (默认): 原行为 (HTTP/1.1 + OpenSSL)
|
|
* transport="curl" : curl_cffi 复刻 Chrome 网络指纹 (JA3/JA4 + HTTP/2 + 头序)
|
|
|
|
背景: 同一 aid / 同一 cookie / 同一参数下, 真实浏览器从不吃 errorCode=12, 而 httpx 会 ——
|
|
12 的自变量落在"传输指纹"上 (见 har/1ms_dianxuan/ANALYSIS.md §7.6)。
|
|
"""
|
|
transport = (transport or "httpx").lower()
|
|
if transport == "httpx":
|
|
return httpx.Client(
|
|
base_url=base_url,
|
|
trust_env=False, # 屏蔽环境代理干扰
|
|
proxy=proxy,
|
|
headers=headers,
|
|
timeout=timeout,
|
|
follow_redirects=True,
|
|
)
|
|
if transport == "curl":
|
|
from tcaptcha.http_transport import CurlCffiClient
|
|
|
|
return CurlCffiClient(base_url=base_url, headers=headers, timeout=timeout,
|
|
proxy=proxy, impersonate=impersonate or "chrome")
|
|
raise ValueError(f"未知 transport: {transport!r} (支持 'httpx' / 'curl')")
|
|
|
|
|
|
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,
|
|
cookies: Optional[dict] = None,
|
|
transport: str = "httpx",
|
|
impersonate: 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"的路; 真正的变量 (登录态 / 真实验证码页面上下文)
|
|
尚未被单变量实验区分。
|
|
|
|
cookies: 可选, 逐条写入 captcha 域 (turing.captcha.qcloud.com) 的 Cookie 头。
|
|
成功样本 (har/签到成功-1.har) 的 captcha 域请求带着腾讯云登录 cookie
|
|
(uin / skey / qcmainCSRFToken / intl), 而本客户端此前完全不带 ——
|
|
这是账本 §7.5 的 H3 单变量实验入口 (默认 None, 不改变原有行为)。
|
|
"""
|
|
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
|
|
|
|
# 可选: 透传 captcha 域的腾讯云登录 cookie (账本 §7.5 的 H3 单变量实验入口)。
|
|
# 注意: **不做成全局头** —— 真实浏览器只在 prehandle / tdc.js / getcapbysig
|
|
# (页面资源 GET) 上带它; getsig / verify (跨站 POST, SameSite=Lax) 上**不带**
|
|
# (har/签到成功-1.har 与 点选.har 实测)。这里保存, 由各调用点按需附加。
|
|
self._cookie_header = ("; ".join(f"{k}={v}" for k, v in cookies.items())
|
|
if cookies else None)
|
|
self.site_origin = _site_origin(self.referer)
|
|
self._gtimg_referer = "https://turing.captcha.gtimg.com/"
|
|
|
|
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",
|
|
}
|
|
|
|
self.http_client = (
|
|
_build_transport(transport, base_url=self.BASE_URL, headers=headers,
|
|
timeout=self.timeout, proxy=self.proxy, impersonate=impersonate)
|
|
)
|
|
|
|
# --------------------------------------------------------------- helpers
|
|
def _resource_headers(self, dest: str, referer: str) -> Dict[str, str]:
|
|
"""
|
|
页面资源 GET (prehandle / tdc.js / getcapbysig) 的**真实浏览器**请求头:
|
|
`Sec-Fetch-*` = cross-site + no-cors + Storage-Access: active, 且带 captcha 域 cookie。
|
|
(机器人常用 XHR 语义的 empty/cors 去拉这些资源 —— 这是可测的指纹差异。)
|
|
"""
|
|
h = {
|
|
"Referer": referer,
|
|
"Accept": "*/*",
|
|
"Sec-Fetch-Dest": dest,
|
|
"Sec-Fetch-Mode": "no-cors",
|
|
"Sec-Fetch-Site": "cross-site",
|
|
"Sec-Fetch-Storage-Access": "active",
|
|
}
|
|
if self._cookie_header:
|
|
h["Cookie"] = self._cookie_header
|
|
return h
|
|
|
|
@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=self._resource_headers("script", self.site_origin))
|
|
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=self._resource_headers("script", self._gtimg_referer))
|
|
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={**self._resource_headers("image", self._gtimg_referer),
|
|
"Accept": "image/jxl,image/avif,image/webp,image/apng,image/svg+xml,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()
|