test(1ms): 逐条单变量排除 errorCode=12 的已知自变量 (cookie / 传输指纹 / 头保真 / collect)

做了什么
- 新增 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 研究项。
This commit is contained in:
yml2213
2026-10-11 23:01:24 +08:00
parent 2e2bbc876a
commit 06c4348163
6 changed files with 442 additions and 43 deletions
+78 -15
View File
@@ -32,6 +32,7 @@ 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:
@@ -87,6 +88,46 @@ def classify_challenge(dyn_show_info: dict, subcapclass: str = "") -> str:
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"
@@ -109,6 +150,8 @@ class TencentCaptchaClient:
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), 只走纯算:
@@ -152,6 +195,15 @@ class TencentCaptchaClient:
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,
@@ -163,20 +215,31 @@ class TencentCaptchaClient:
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "cross-site",
}
# 可选: 透传 captcha 域的腾讯云登录 cookie (账本 §7.5 的 H3 单变量实验入口)
if cookies:
headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in cookies.items())
self.http_client = httpx.Client(
base_url=self.BASE_URL,
trust_env=False, # 屏蔽环境代理干扰
proxy=self.proxy,
headers=headers,
timeout=self.timeout,
follow_redirects=True,
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:
@@ -236,7 +299,7 @@ class TencentCaptchaClient:
"sess": "",
}
resp = self.http_client.get("/cap_union_prehandle", params=params,
headers={"Referer": self.referer})
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]}")
@@ -281,8 +344,8 @@ class TencentCaptchaClient:
# ------------------------------------------------------------ 3. download
def fetch_tdc(self, tdc_path: str) -> str:
resp = self.http_client.get(tdc_path,
headers={"Referer": "https://turing.captcha.gtimg.com/"})
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:
@@ -290,8 +353,8 @@ class TencentCaptchaClient:
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"},
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
+78
View File
@@ -0,0 +1,78 @@
# -*- coding: utf-8 -*-
"""
HTTP 传输层可选实现 —— 为 `TencentCaptchaClient` 提供"浏览器级"传输指纹。
背景 (2026-10-11 实测, 见 har/1ms_dianxuan/ANALYSIS.md §7.6):
同一 aid (1ms 195672344)、同一套腾讯云登录 cookie、同一 prehandle 参数下,
**真实浏览器**对 cap_union_new_verify 只会得到 50 / 9 / 0 (从不 12);
而 httpx 客户端(HTTP/1.1 + OpenSSL, 头序与浏览器不同) 永远得到 12。
即 12 的自变量不在"载荷内容"(答案已对、密钥已用真机密钥), 而在"谁在发这个 HTTP"。
本模块用 curl_cffi 复刻 Chrome 的网络指纹(**JA3/JA4 + HTTP/2 + 头序 + 头大小写**),
对上层暴露与 `httpx.Client` 相同的最小接口 (get / post / close), 便于单变量对照。
用法::
TencentCaptchaClient(..., transport="curl", impersonate="chrome")
依赖: `curl_cffi` (uv add curl_cffi)。
"""
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
logger = logging.getLogger("tcaptcha_http")
try: # 依赖可选: 未安装时只影响 transport="curl"
from curl_cffi import requests as _cr
except Exception: # pragma: no cover
_cr = None
class CurlCffiClient:
"""`httpx.Client` 的最小兼容壳 —— 走 curl_cffi 的浏览器指纹。"""
def __init__(self, base_url: str, headers: Dict[str, str], timeout: float = 15.0,
proxy: Optional[str] = None, impersonate: str = "chrome",
follow_redirects: bool = True) -> None:
if _cr is None:
raise RuntimeError("curl_cffi 未安装: 先执行 `uv add curl_cffi`")
self._base = base_url.rstrip("/")
self._headers = {k: v for k, v in headers.items() if v is not None}
self._timeout = timeout
self._proxy = proxy
self._follow = follow_redirects
self._session = _cr.Session(impersonate=impersonate)
logger.debug("CurlCffiClient 就绪 (impersonate=%s)", impersonate)
# ------------------------------------------------------------------ 兼容面
def _url(self, path: str) -> str:
return path if str(path).startswith("http") else self._base + str(path)
def _proxy_kw(self) -> Dict[str, Any]:
if not self._proxy:
return {}
return {"proxies": {"http": self._proxy, "https": self._proxy}}
def _merge(self, extra: Optional[Dict[str, str]]) -> Dict[str, str]:
out = dict(self._headers)
if extra:
out.update({k: v for k, v in extra.items() if v is not None})
return out
def get(self, path: str, **kw: Any):
headers = self._merge(kw.pop("headers", None))
return self._session.get(self._url(path), headers=headers, timeout=self._timeout,
allow_redirects=self._follow, **self._proxy_kw(), **kw)
def post(self, path: str, **kw: Any):
headers = self._merge(kw.pop("headers", None))
return self._session.post(self._url(path), headers=headers, timeout=self._timeout,
allow_redirects=self._follow, **self._proxy_kw(), **kw)
def close(self) -> None:
try:
self._session.close()
except Exception: # pragma: no cover
pass