feat(诊断): 用真机复现解出 1ms 成功会话的 collect 明文 —— 归因改写为「cd 槽位序按会话随机」
结论 (已知 #29~#32, 推翻旧 #25/#27 与"collect 逐字节同构"这条否定实验):
* 同一个 tdc.js URL 每次请求返回不同构建 (4/4 md5 + XTEA 密钥全不同)
* 同会话同时抓 tdc.js + verify -> 解密出真机 collect 明文 (784B/38 槽, 可打印率 .991)
* 5 份同日样本 (真机 cap2 / 签到成功-1 / 点选 / login_checkin / 仓库模板) 全部 38 槽、
同一批取值、同一个 sd, 但槽位序 5/5 互不相同
* 仓库在用的模板槽位序 == 签到成功-1 那一次; generate_collect_success() 只回填
轨迹/时间戳/entry_url, 槽位序仍是那一次的
=> 头号可证伪假设 H-LAYOUT: 服务端按"本次会话构建"的槽位表解释 cd, 用别的会话的槽位序
会被读成自相矛盾的环境 -> 12 (答案错理应是 50/9)。这也解释了为何换 IP / 换遥测后端 /
换传输指纹 / 换 cookie / 长静默全都无效 —— 它们都不动槽位序。
新增:
* scripts/analyze_collect_layout.py —— collect 槽位序对比 (复现 #31)
* tcaptcha/slider_solver.py + test —— 滑块缺口求解 (ZNCC, 真机首轮命中 ret=0)
* tcaptcha/icon_click_solver.py + test —— 图文点选形状模板求解
* scripts/bridge_1ms_captcha.py —— 真机桥接三模式 (slider/icon/text) + --probe 题型侦察
* tools/cdp_eval.js / cdp_click.js / cdp_shot.js / cdp_net_capture.js / cdp_capture.js
* selftest 纳入 icon_click / slider 两个套件
账本: §一 修正 #25/#27、新增 #29~#32; §三 标注"collect 逐字节同构"否定实验有缺陷;
§五 重写 12 的归因与下一步 (决定性实验 A + 取槽位表两条路)
This commit is contained in:
yml2213
@@ -0,0 +1,182 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
collect 明文「字段布局」分析器 —— 回答一个关键问题:
|
||||
|
||||
`collect` 解密后的 `cd` 数组, 槽位顺序到底是"按 tdc.js 版本固定"还是"按会话/构建随机"?
|
||||
|
||||
为什么要问这个 (2026-10-11 排查 errorCode=12):
|
||||
tcaptcha/models/success_collect_template.b64 存的是**明文**模板 (base64 封装),
|
||||
generate_collect_success() 只回填 轨迹 / entry_url / 几个时间戳, 其余字段原样保留。
|
||||
真机成功样本解密后发现: **同一批值, 顺序完全不同** —— 若服务端按"本次构建的槽位表"
|
||||
解释 cd, 那么我们用旧槽位序发出的 collect 会被读成"自相矛盾的环境" -> 12。
|
||||
|
||||
本脚本把多份样本 (HAR + CDP 抓包 + 仓库模板) 的 cd 槽位打成类型指纹并对比顺序。
|
||||
|
||||
用法:
|
||||
.venv/bin/python scripts/analyze_collect_layout.py # 默认样本集
|
||||
.venv/bin/python scripts/analyze_collect_layout.py --json /tmp/cap2.json
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.parse
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from tcaptcha.crypto import xtea_decrypt_bytes # noqa: E402
|
||||
from tcaptcha.tdc_key import extract_xtea_key_pure # noqa: E402
|
||||
|
||||
HARS = [
|
||||
("har/原始抓包/签到成功-1.har", "签到成功-1"),
|
||||
("har/原始抓包/点选.har", "点选"),
|
||||
("har/login_checkin.har", "login_checkin"),
|
||||
]
|
||||
|
||||
|
||||
def _decrypt_collect(collect_b64: str, tdc_js: str) -> Optional[dict]:
|
||||
key = extract_xtea_key_pure(tdc_js)["key"]
|
||||
pt = xtea_decrypt_bytes(base64.b64decode(collect_b64), key)
|
||||
text = pt.decode("latin1").rstrip("\x00").strip()
|
||||
try:
|
||||
return json.loads(text)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def from_har(path: str) -> List[Tuple[str, dict]]:
|
||||
if not os.path.exists(path):
|
||||
return []
|
||||
data = json.load(open(path, encoding="utf-8"))
|
||||
tdcs: Dict[str, str] = {}
|
||||
sess2sid: Dict[str, str] = {}
|
||||
verifies: List[dict] = []
|
||||
for e in data["log"]["entries"]:
|
||||
url = e["request"]["url"]
|
||||
content = (e["response"].get("content") or {}).get("text") or ""
|
||||
if "tdc.js" in url and content:
|
||||
q = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
|
||||
tdcs[(q.get("app_data") or [""])[0]] = content
|
||||
elif "cap_union_prehandle" in url and content:
|
||||
m = re.search(r"^[^(]*\((.*)\)\s*$", content, re.S)
|
||||
if m:
|
||||
try:
|
||||
j = json.loads(m.group(1))
|
||||
sess2sid[j.get("sess", "")] = str(j.get("sid", ""))
|
||||
except Exception:
|
||||
pass
|
||||
elif "cap_union_new_verify" in url:
|
||||
pd = (e["request"].get("postData") or {}).get("text") or ""
|
||||
if pd:
|
||||
verifies.append(dict(urllib.parse.parse_qsl(pd, keep_blank_values=True)))
|
||||
out = []
|
||||
for i, v in enumerate(verifies):
|
||||
js = tdcs.get(sess2sid.get(v.get("sess", ""), ""))
|
||||
if not js:
|
||||
continue
|
||||
j = _decrypt_collect(v["collect"], js)
|
||||
if j is not None:
|
||||
out.append((f"{Path(path).name}#{i}", j))
|
||||
return out
|
||||
|
||||
|
||||
def from_cdp_capture(path: str) -> List[Tuple[str, dict]]:
|
||||
if not os.path.exists(path):
|
||||
return []
|
||||
d = json.load(open(path, encoding="utf-8"))
|
||||
recs = d.get("records") or []
|
||||
tdcs = [(r["url"], r.get("body") or "") for r in recs if "tdc.js" in r["url"] and r.get("body")]
|
||||
vers = [r for r in recs if "verify" in r["url"] and r.get("postData")]
|
||||
out = []
|
||||
for i, v in enumerate(vers):
|
||||
form = dict(urllib.parse.parse_qsl(v["postData"], keep_blank_values=True))
|
||||
sess = form.get("sess", "")
|
||||
# sess 前 2 段与 tdc.js 的 app_data(sid) 无直接映射, 这里按时间就近取**最后一份**构建
|
||||
js = tdcs[-1][1] if tdcs else ""
|
||||
if not js:
|
||||
continue
|
||||
j = _decrypt_collect(form["collect"], js)
|
||||
if j is not None:
|
||||
out.append((f"{Path(path).name}#{i}", j))
|
||||
return out
|
||||
|
||||
|
||||
def slot_tags(cd: List) -> List[str]:
|
||||
tags = []
|
||||
for v in cd:
|
||||
if isinstance(v, str):
|
||||
if v.startswith("http"):
|
||||
tags.append("url")
|
||||
elif "Mozilla" in v:
|
||||
tags.append("UA")
|
||||
elif v.startswith("GgoAAA"):
|
||||
tags.append("png")
|
||||
elif re.fullmatch(r"[-+]\d\d", v):
|
||||
tags.append("tz")
|
||||
elif v == "":
|
||||
tags.append("empty")
|
||||
elif len(v) <= 6 and v.isupper():
|
||||
tags.append("enc:" + v)
|
||||
else:
|
||||
tags.append("str:" + v[:10])
|
||||
elif isinstance(v, list):
|
||||
if len(v) == 2 and all(isinstance(x, int) for x in v):
|
||||
tags.append("vec2")
|
||||
elif v and isinstance(v[0], list):
|
||||
tags.append("traj")
|
||||
else:
|
||||
tags.append("list%d" % len(v))
|
||||
elif isinstance(v, bool):
|
||||
tags.append("bool")
|
||||
else:
|
||||
tags.append("num")
|
||||
return tags
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="collect 明文字段布局分析")
|
||||
ap.add_argument("--json", action="append", default=[], help="额外加入的 cdp_capture JSON")
|
||||
args = ap.parse_args()
|
||||
|
||||
samples: List[Tuple[str, dict]] = []
|
||||
for p in args.json:
|
||||
samples += from_cdp_capture(p)
|
||||
for p, tag in HARS:
|
||||
for name, j in from_har(str(ROOT / p)):
|
||||
samples.append((f"{tag}#{name.split('#')[-1]}", j))
|
||||
tpl_path = ROOT / "tcaptcha" / "models" / "success_collect_template.b64"
|
||||
if tpl_path.exists():
|
||||
tpl = json.loads(base64.b64decode(tpl_path.read_text(encoding="ascii"))
|
||||
.decode("latin1").rstrip("\x00").strip())
|
||||
samples.append(("TEMPLATE(仓库在用)", tpl))
|
||||
|
||||
if not samples:
|
||||
print("没有可用样本")
|
||||
return 1
|
||||
|
||||
for name, j in samples:
|
||||
tags = slot_tags(j["cd"])
|
||||
print(f"\n[{name}] len(cd)={len(j['cd'])} sd={j.get('sd')}")
|
||||
print(" " + " ".join(f"{i}:{t}" for i, t in enumerate(tags)))
|
||||
|
||||
print("\n=== 槽位指纹两两对比 (看顺序是否随会话变化) ===")
|
||||
sigs = [(n, slot_tags(j["cd"])) for n, j in samples]
|
||||
base_name, base = sigs[0]
|
||||
print(f" 基准 = {base_name}")
|
||||
for n, s in sigs[1:]:
|
||||
diff = [i for i in range(max(len(base), len(s)))
|
||||
if (base[i] if i < len(base) else None) != (s[i] if i < len(s) else None)]
|
||||
print(f" {n:22s} 顺序相同={s == base} 不同槽位数={len(diff)}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,390 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
1ms 验证码「真机桥接」驱动 —— 在**真实 Chrome 页面**里完成腾讯滑块/点选, 拿回 ticket。
|
||||
|
||||
为什么需要桥接 (har/1ms_dianxuan/PROGRESS.md 已知 #26/#27):
|
||||
aid=195672344 对该 aid 之外的会话恒返回 errorCode=12 (aid 级风控), 纯协议侧无解;
|
||||
但**真实浏览器页面**里能拿到 ec=0。所以这一层只负责"过验证码", 业务提交仍走纯 HTTPS。
|
||||
|
||||
它做什么 (每轮):
|
||||
1. 用 CDP 在真实页面里刷新腾讯验证码 widget (`__cap.refresh()`);
|
||||
2. 读 widget 内的题型/题面/几何 (滑块: slideBg + 拼图块 + 拖动柄; 点选: 提示图 + 背景图);
|
||||
3. 求解:
|
||||
* 滑块 -> tcaptcha.slider_solver.SliderSolver (ZNCC 找真缺口)
|
||||
* 图文点选 -> tcaptcha.icon_click_solver.IconClickSolver (形状模板 + 旋转 IoU)
|
||||
4. 用 CDP `Input.dispatchMouseEvent` 派发**真实(trusted)鼠标事件** (点击 / 拖动);
|
||||
5. 读回 widget 回调的 {ret, ticket, randstr}。
|
||||
|
||||
前置:
|
||||
Chrome 以 `--remote-debugging-port=9222` 启动且已登录 1ms;
|
||||
页面里已注入 widget 实例: `window.__cap = new TencentCaptcha(aid, cb, {}); __cap.show()`。
|
||||
|
||||
用法:
|
||||
.venv/bin/python scripts/bridge_1ms_captcha.py --rounds 12
|
||||
.venv/bin/python scripts/bridge_1ms_captcha.py --only slider --rounds 20
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import cv2 # noqa: E402
|
||||
|
||||
from tcaptcha.glyph_solver import TextClickSolver # noqa: E402
|
||||
from tcaptcha.icon_click_solver import IconClickSolver # noqa: E402
|
||||
from tcaptcha.slider_solver import SliderSolver, sprite_region # noqa: E402
|
||||
|
||||
NODE = "node"
|
||||
CDP_EVAL = str(ROOT / "tools" / "cdp_eval.js")
|
||||
CDP_CLICK = str(ROOT / "tools" / "cdp_click.js")
|
||||
TMP = Path("/tmp/1ms_bridge")
|
||||
|
||||
PAGE = "1ms.run"
|
||||
IFRAME = "drag_ele"
|
||||
|
||||
# 页面重载会丢掉注入的实例; 这个表达式负责"有就刷新, 没有就重建"
|
||||
EXPR_ENSURE = """(() => {
|
||||
try {
|
||||
if (window.__cap && typeof window.__cap.refresh === 'function') {
|
||||
window.__capres = null; window.__cap.refresh();
|
||||
return {mode: 'refresh'};
|
||||
}
|
||||
if (typeof window.TencentCaptcha !== 'function') return {err: 'TencentCaptcha 未加载'};
|
||||
window.__capres = null;
|
||||
window.__cap = new window.TencentCaptcha('%AID%',
|
||||
function (res) { window.__capres = res; }, {});
|
||||
window.__cap.show();
|
||||
return {mode: 'inject'};
|
||||
} catch (e) { return {err: String(e)}; }
|
||||
})()""".replace("%AID%", "195672344")
|
||||
|
||||
EXPR_RESULT = "(()=>({res: window.__capres||null}))()"
|
||||
|
||||
# 页面里可能有多个 tcaptcha iframe (隐藏的预加载 iframe 在 y=-1000000), 必须挑**可见**那个
|
||||
EXPR_FRAME = """(() => {
|
||||
const cands = [...document.querySelectorAll('iframe[id^="tcaptcha"], iframe[src*="captcha"]')];
|
||||
const f = cands.find(e => {
|
||||
const r = e.getBoundingClientRect();
|
||||
return r.width > 100 && r.height > 100 && r.top > -1000;
|
||||
});
|
||||
if (!f) return {found:false, count: cands.length};
|
||||
const r = f.getBoundingClientRect();
|
||||
return {found:true, x:r.x, y:r.y, w:r.width, h:r.height};
|
||||
})()"""
|
||||
|
||||
EXPR_GEO = """(() => {
|
||||
const R = el => { const r = el.getBoundingClientRect();
|
||||
return {x:r.x, y:r.y, w:r.width, h:r.height}; };
|
||||
const urlOf = el => { const bi = getComputedStyle(el).backgroundImage || '';
|
||||
const m = /url\\(["']?([^"')]+)["']?\\)/.exec(bi); return m ? m[1] : null; };
|
||||
const out = {
|
||||
instruction: ((document.getElementById('instructionText') || {}).textContent || '').trim(),
|
||||
mode: null,
|
||||
};
|
||||
// --- 图文点选: 提示条是 <img> 里 img_index=0, 背景是 img_index=1 ---
|
||||
const all = [...document.querySelectorAll('img')];
|
||||
if (/依次点击|点击/.test(out.instruction)) {
|
||||
const pr = all.map(i => i.src).find(s => (s || '').includes('img_index=0'));
|
||||
const bgs = [...document.querySelectorAll('*')].filter(e => {
|
||||
const bi = getComputedStyle(e).backgroundImage || '';
|
||||
return bi.includes('img_index=1') && e.getBoundingClientRect().width > 200;
|
||||
});
|
||||
const bgEl = bgs[0];
|
||||
// 可见的「确定」按钮: 逐层找叶子文本为 确定/确认 且尺寸非 0
|
||||
const btn = [...document.querySelectorAll('div,span,button,a')].find(e => {
|
||||
const t = (e.textContent || '').trim();
|
||||
const r = e.getBoundingClientRect();
|
||||
return (t === '确定' || t === '确认' || t === '提交') && r.width > 10 && r.height > 10;
|
||||
});
|
||||
out.mode = pr ? 'icon' : 'text'; // 有提示图 = 图文点选; 纯文字题面 = 文字点选
|
||||
out.promptUrl = pr || null;
|
||||
out.bgUrl = bgEl ? urlOf(bgEl) : null;
|
||||
out.bgRect = bgEl ? R(bgEl) : null;
|
||||
out.bgNatural = bgEl ? bgEl.getAttribute('style') : null;
|
||||
out.okBtn = btn ? R(btn) : null;
|
||||
return out;
|
||||
}
|
||||
// --- 滑块: slideBg 有 img_index=1 背景, 拼图块元素有 img_index=0 背景 ---
|
||||
const bgEl = document.getElementById('slideBg');
|
||||
if (bgEl && urlOf(bgEl) && bgEl.getBoundingClientRect().width > 150) {
|
||||
const st = getComputedStyle(bgEl);
|
||||
out.mode = 'slider';
|
||||
out.bgUrl = urlOf(bgEl);
|
||||
out.bgRect = R(bgEl);
|
||||
out.bgCssSize = st.backgroundSize;
|
||||
const pe = [...document.querySelectorAll('div')].find(e => {
|
||||
const r = e.getBoundingClientRect();
|
||||
return (getComputedStyle(e).backgroundImage || '').includes('img_index=0')
|
||||
&& r.height > 30 && r.height < 140 && r.width > 30 && r.width < 140;
|
||||
});
|
||||
if (pe) {
|
||||
const ps = getComputedStyle(pe);
|
||||
out.pieceRect = R(pe);
|
||||
out.pieceBgPos = ps.backgroundPosition;
|
||||
out.pieceBgSize = ps.backgroundSize;
|
||||
out.pieceUrl = urlOf(pe);
|
||||
}
|
||||
const he = document.querySelector('.tc-slider-normal') || document.querySelector('.tc-fg-item.tc-slider-normal');
|
||||
if (he) out.handleRect = R(he);
|
||||
return out;
|
||||
}
|
||||
return out;
|
||||
})()"""
|
||||
|
||||
|
||||
def cdp_eval(expr: str, match: str) -> dict:
|
||||
p = subprocess.run([NODE, CDP_EVAL, "--port", "9222", "--match", match],
|
||||
input=expr, capture_output=True, text=True, timeout=90)
|
||||
if p.returncode != 0:
|
||||
raise RuntimeError(f"cdp_eval 失败: {p.stderr.strip()[:300]}")
|
||||
return json.loads(p.stdout)
|
||||
|
||||
|
||||
def cdp_click(clicks) -> None:
|
||||
spec = ";".join("%d,%d" % (round(x), round(y)) for x, y in clicks)
|
||||
p = subprocess.run([NODE, CDP_CLICK, "--port", "9222", "--match", PAGE,
|
||||
"--clicks", spec], capture_output=True, text=True, timeout=180)
|
||||
if p.returncode != 0:
|
||||
raise RuntimeError(f"cdp_click 失败: {p.stderr.strip()[:300]}")
|
||||
|
||||
|
||||
def cdp_drag(x1: float, y1: float, x2: float, y2: float, steps: int = 34) -> None:
|
||||
spec = "%.2f,%.2f,%.2f,%.2f,%d" % (x1, y1, x2, y2, steps)
|
||||
p = subprocess.run([NODE, CDP_CLICK, "--port", "9222", "--match", PAGE,
|
||||
"--drag", spec], capture_output=True, text=True, timeout=180)
|
||||
if p.returncode != 0:
|
||||
raise RuntimeError(f"cdp_drag 失败: {p.stderr.strip()[:300]}")
|
||||
|
||||
|
||||
def download(url: str, path: Path) -> bool:
|
||||
p = subprocess.run(["curl", "-s", "--max-time", "30", "-o", str(path),
|
||||
"-H", "Referer: https://turing.captcha.gtimg.com/1/template/drag_ele.html",
|
||||
url], capture_output=True, text=True)
|
||||
ok = p.returncode == 0 and path.exists() and path.stat().st_size > 1000
|
||||
return ok
|
||||
|
||||
|
||||
def parse_px(value: str) -> tuple:
|
||||
"""把 '345.06px 313.69px' / '-70.8333px -247.917px' 解析成 (x, y)。"""
|
||||
nums = []
|
||||
for tok in str(value).replace("px", " ").split():
|
||||
try:
|
||||
nums.append(float(tok))
|
||||
except ValueError:
|
||||
pass
|
||||
return (nums[0], nums[1]) if len(nums) >= 2 else (0.0, 0.0)
|
||||
|
||||
|
||||
def solve_slider(geo: dict, tmp: Path, solver: SliderSolver):
|
||||
"""滑块: 返回 (from_xy, to_xy) —— 均为**顶层页面视口**坐标。"""
|
||||
bg_path, sp_path = tmp / "sl_img1", tmp / "sl_img0"
|
||||
if not download(geo["bgUrl"], bg_path) or not download(geo["pieceUrl"], sp_path):
|
||||
return None, "图片下载失败"
|
||||
bg = cv2.imread(str(bg_path))
|
||||
sprite = cv2.imread(str(sp_path), cv2.IMREAD_UNCHANGED)
|
||||
if bg is None or sprite is None:
|
||||
return None, "图片解码失败"
|
||||
|
||||
bg_rect, piece_rect = geo["bgRect"], geo["pieceRect"]
|
||||
scale = bg_rect["w"] / bg.shape[1] # 显示宽度 / 原图宽度
|
||||
piece_x_image = (piece_rect["x"] - bg_rect["x"]) / scale
|
||||
piece_y_image = (piece_rect["y"] - bg_rect["y"]) / scale
|
||||
|
||||
px, py = parse_px(geo["pieceBgPos"])
|
||||
sw, sh = parse_px(geo["pieceBgSize"])
|
||||
bx, by, bw, bh = sprite_region((px, py), (sw, sh),
|
||||
(piece_rect["w"], piece_rect["h"]), sprite.shape[:2])
|
||||
crop = sprite[by:by + bh, bx:bx + bw]
|
||||
if crop.size == 0:
|
||||
return None, f"精灵图裁剪越界 {(bx, by, bw, bh)} vs {sprite.shape}"
|
||||
|
||||
res = solver.solve(bg, crop, piece_x_image=piece_x_image, scale=scale,
|
||||
piece_y_image=piece_y_image)
|
||||
print(f" 滑块求解: ok={res.ok} {res.describe()}")
|
||||
if not res.ok:
|
||||
return None, "求解不可靠"
|
||||
cv2.imwrite(str(tmp / "slider_annot.png"), SliderSolver.annotate(bg, res, bw, bh))
|
||||
|
||||
handle = geo.get("handleRect")
|
||||
if not handle:
|
||||
return None, "未找到拖动柄"
|
||||
hx = handle["x"] + handle["w"] / 2.0
|
||||
hy = handle["y"] + handle["h"] / 2.0
|
||||
return (hx, hy, hx + res.dx_display, hy), None
|
||||
|
||||
|
||||
def solve_icon(geo: dict, tmp: Path, solver: IconClickSolver):
|
||||
"""图文点选: 返回 (clicks, ) —— 均为**iframe 内**坐标, 由调用方加偏移。"""
|
||||
pr_path, bg_path = tmp / "ic_prompt.jpg", tmp / "ic_bg.jpg"
|
||||
if not download(geo["promptUrl"], pr_path) or not download(geo["bgUrl"], bg_path):
|
||||
return None, "图片下载失败"
|
||||
pr = cv2.imread(str(pr_path))
|
||||
bg = cv2.imread(str(bg_path))
|
||||
if pr is None or bg is None:
|
||||
return None, "图片解码失败"
|
||||
res = solver.solve(bg, pr)
|
||||
print(f" 点选求解: ok={res.ok} {res.describe()}")
|
||||
if not res.ok:
|
||||
return None, "求解不可靠"
|
||||
cv2.imwrite(str(tmp / "icon_annot.png"), solver.annotate(bg, res))
|
||||
bg_rect = geo["bgRect"]
|
||||
sx, sy = bg_rect["w"] / bg.shape[1], bg_rect["h"] / bg.shape[0]
|
||||
clicks = [(bg_rect["x"] + x * sx, bg_rect["y"] + y * sy) for x, y in res.points]
|
||||
return clicks, None
|
||||
|
||||
|
||||
def solve_text(geo: dict, tmp: Path, solver: TextClickSolver):
|
||||
"""文字点选 (汉字 2404): 题面自带目标字, 返回 iframe 内点击坐标。"""
|
||||
bg_path = tmp / "tx_bg.jpg"
|
||||
if not download(geo["bgUrl"], bg_path):
|
||||
return None, "图片下载失败"
|
||||
bg = cv2.imread(str(bg_path))
|
||||
if bg is None:
|
||||
return None, "图片解码失败"
|
||||
res = solver.solve_ex(bg, geo["instruction"])
|
||||
print(f" 文字点选求解: ok={res.ok} {res.describe()}")
|
||||
print(f" points={res.points} targets={res.targets}")
|
||||
if not res.ok:
|
||||
return None, "求解不可靠"
|
||||
bg_rect = geo["bgRect"]
|
||||
sx, sy = bg_rect["w"] / bg.shape[1], bg_rect["h"] / bg.shape[0]
|
||||
clicks = [(bg_rect["x"] + x * sx, bg_rect["y"] + y * sy) for x, y in res.points]
|
||||
return clicks, None
|
||||
|
||||
|
||||
def probe(rounds: int) -> int:
|
||||
"""只刷新不点击, 统计目标站点实际下发的题型分布 (侦察用)。"""
|
||||
dist: dict = {}
|
||||
for rnd in range(1, rounds + 1):
|
||||
ens = cdp_eval(EXPR_ENSURE, PAGE)["result"]
|
||||
if ens.get("err"):
|
||||
print(f" 无法唤起 widget: {ens['err']}")
|
||||
return 2
|
||||
time.sleep(3.0)
|
||||
geo = cdp_eval(EXPR_GEO, IFRAME)["result"]
|
||||
mode = geo.get("mode") or "unknown"
|
||||
dist[mode] = dist.get(mode, 0) + 1
|
||||
print(f" [{rnd:02d}] {mode:8s} {geo.get('instruction')!r}")
|
||||
print(f"\n题型分布 ({rounds} 次刷新): {dist}")
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="1ms 验证码真机桥接 (CDP)")
|
||||
ap.add_argument("--aid", default="195672344")
|
||||
ap.add_argument("--rounds", type=int, default=12, help="最多尝试几轮 (1ms 滑块下发最多)")
|
||||
ap.add_argument("--solves", type=int, default=1, help="拿到几个 ticket 后停止 (用于统计成功率)")
|
||||
ap.add_argument("--probe", type=int, default=0, help="只刷新 N 次统计题型分布, 不点击")
|
||||
ap.add_argument("--only", choices=["slider", "icon", "text", "auto"], default="auto")
|
||||
args = ap.parse_args()
|
||||
|
||||
TMP.mkdir(parents=True, exist_ok=True)
|
||||
if args.probe:
|
||||
return probe(args.probe)
|
||||
|
||||
slider, icon = SliderSolver(), IconClickSolver()
|
||||
text_solver = None
|
||||
skipped = {"slider": 0, "icon": 0, "text": 0}
|
||||
stats = {"attempts": 0, "ok": 0, "fail": 0, "timeout": 0, "unsolved": 0, "skip": 0}
|
||||
tickets: list = []
|
||||
|
||||
for rnd in range(1, args.rounds + 1):
|
||||
if stats["ok"] >= args.solves:
|
||||
break
|
||||
print(f"\n=== 第 {rnd}/{args.rounds} 轮 (成功 {stats['ok']}/{args.solves}) ===")
|
||||
ens = cdp_eval(EXPR_ENSURE, PAGE)["result"]
|
||||
if ens.get("err"):
|
||||
print(f" 无法唤起 widget: {ens['err']}")
|
||||
return 2
|
||||
print(f" widget: {ens.get('mode')}")
|
||||
time.sleep(3.0)
|
||||
geo = cdp_eval(EXPR_GEO, IFRAME)["result"]
|
||||
mode = geo.get("mode")
|
||||
print(f" 题型={mode} 题面={geo.get('instruction')!r}")
|
||||
if mode is None:
|
||||
print(" 未就绪 -> 下一轮")
|
||||
continue
|
||||
if args.only != "auto" and mode != args.only:
|
||||
skipped[mode] += 1
|
||||
stats["skip"] += 1
|
||||
print(f" 跳过 (--only {args.only})")
|
||||
continue
|
||||
|
||||
fr = cdp_eval(EXPR_FRAME, PAGE)["result"]
|
||||
if not fr.get("found"):
|
||||
print(" 未找到可见的 captcha iframe -> 下一轮")
|
||||
continue
|
||||
ox, oy = fr["x"], fr["y"]
|
||||
|
||||
if mode == "slider":
|
||||
plan, err = solve_slider(geo, TMP, slider)
|
||||
if err:
|
||||
print(f" {err} -> 下一轮")
|
||||
stats["unsolved"] += 1
|
||||
continue
|
||||
x1, y1, x2, y2 = plan
|
||||
print(" 页面坐标拖动: (%.0f,%.0f) -> (%.0f,%.0f) dx=%.1f" % (x1, y1, x2, y2, x2 - x1))
|
||||
cdp_drag(ox + x1, oy + y1, ox + x2, oy + y2)
|
||||
else:
|
||||
if mode == "text":
|
||||
if text_solver is None:
|
||||
text_solver = TextClickSolver()
|
||||
clicks, err = solve_text(geo, TMP, text_solver)
|
||||
else:
|
||||
clicks, err = solve_icon(geo, TMP, icon)
|
||||
if err:
|
||||
print(f" {err} -> 下一轮")
|
||||
stats["unsolved"] += 1
|
||||
continue
|
||||
pts = [(ox + x, oy + y) for x, y in clicks]
|
||||
btn = geo.get("okBtn")
|
||||
if btn:
|
||||
pts.append((ox + btn["x"] + btn["w"] / 2, oy + btn["y"] + btn["h"] / 2))
|
||||
else:
|
||||
print(" ⚠️ 未找到可见的「确定」按钮")
|
||||
print(" 页面坐标点击: %s" % [("%d,%d" % (round(a), round(b))) for a, b in pts])
|
||||
cdp_click(pts)
|
||||
|
||||
stats["attempts"] += 1
|
||||
got = None
|
||||
for _ in range(25):
|
||||
time.sleep(0.6)
|
||||
got = cdp_eval(EXPR_RESULT, PAGE)["result"]["res"]
|
||||
if got:
|
||||
break
|
||||
if not got:
|
||||
print(" 等待回调超时 -> 下一轮")
|
||||
stats["timeout"] += 1
|
||||
continue
|
||||
print("\n >>> widget 回调: %s" % json.dumps(got, ensure_ascii=False)[:300])
|
||||
print(" >>> ret=%s ticket=%s randstr=%s" %
|
||||
(got.get("ret"), (got.get("ticket") or "")[:32], got.get("randstr")))
|
||||
if got.get("ret") == 0:
|
||||
stats["ok"] += 1
|
||||
tickets.append(got.get("ticket"))
|
||||
else:
|
||||
stats["fail"] += 1
|
||||
|
||||
print(f"\n{'=' * 62}")
|
||||
print("统计: %s" % stats)
|
||||
print("跳过: %s" % skipped)
|
||||
if tickets:
|
||||
print("最后 ticket: %s" % tickets[-1][:48])
|
||||
if stats["ok"] >= args.solves:
|
||||
print("✅ 全部达成 (%d/%d)" % (stats["ok"], args.solves))
|
||||
return 0
|
||||
print("❌ 未达成 (%d/%d)" % (stats["ok"], args.solves))
|
||||
return 1
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user