ksjsb/tools/captcha_solver_http.py
2026-07-30 20:25:56 +08:00

546 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""纯 HTTP 快手滑块验证码求解器(无浏览器)。
设计依据out/captcha_net/js_08...deobf2.js 还原):
- POST /rest/zt/captcha/sliding/config body={captchaSession: <key>} -> {captchaSn, bgPicUrl, cutPicUrl, bgPicWidth=686, bgPicHeight=400, cutPicWidth=122, cutPicHeight=122, disX=24, disY=122, verifyUrl2, ...}
- GET bgPic/cutPic?captchaSn=...
- 缺口: ddddocr slide_match(cut, bg) -> target_x (686 原生像素空间)
- verify payload u = {captchaSn, bgDisWidth, bgDisHeight, cutDisWidth, cutDisHeight, relativeX, relativeY, trajectory, gpuInfo, captchaExtraParam}
- verifyParam = base64(Jose.$encrypt(utf8(JSON.stringify(u)), KEY)) KEY=c7b645db-...
- POST <verifyUrl2> body={verifyParam} -> {result, captchaToken|...}
坐标关键点: 配置响应处理把所有 *Img 字段乘 scaleRatio(=containerWidth/bgPicWidth)。
服务端校验的是比例(relativeX/bgDisWidth == gap/686),比例与 scale 无关,
故取 scaleRatio=1全原生坐标即可bgDisWidth=686, relativeX=target_x+offset, relativeY=disY。
relativeX/relativeY/trajectory 取自 Vue 的 sliderImgX/sliderImgY/trajectory原生空间
"""
from __future__ import annotations
import json
import math
import random
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
from urllib.parse import quote
ROOT = Path(__file__).resolve().parent.parent
JOSE_JS = ROOT / "tools" / "jose_encrypt.js"
JOSE_RAW_JS = ROOT / "tools" / "jose_encrypt_raw.js" # form-encode 明文加密(不做 JSON 包装)
NODE = sys.executable if Path(sys.executable).name.lower().startswith("node") else "node"
# 三坐标空间deobf2 JS + 浏览器 Jose.call hook 实测互证):
# NATIVE 图像 686×400ddddocr target_x、config bgPicWidth
# DISPLAY CSS 316×184浏览器 bgDisWidth/relativeXAPP WebView 控件布局常量config 不返回)
# PHYSICAL DPR=3 948×552浏览器 trajectory 终点 ~920
# native→physical = (DISPLAY_BG_W/NATIVE_BG_W)*DPR = (316/686)*3 = 1.382
NATIVE_BG_W = 686
DISPLAY_BG_W = 316 # 浏览器 hook 实测的 WebView 控件 CSS 宽
DISPLAY_BG_H = 184
DISPLAY_REL_Y = 70 # 浏览器 relativeY 实测
DPR = 3
NATIVE_TO_PHYS = (DISPLAY_BG_W / NATIVE_BG_W) * DPR # ≈1.382
# verifyParam 明文字段固定序deobf2 JS: .join("=")/.join("&") on encodeURIComponent
_VERIFY_FIELD_ORDER = (
"captchaSn", "bgDisWidth", "bgDisHeight",
"cutDisWidth", "cutDisHeight",
"relativeX", "relativeY", "trajectory",
"gpuInfo", "captchaExtraParam",
)
CAPTCHA_HOST = "https://captcha.zt.kuaishou.com"
CONFIG_URL = CAPTCHA_HOST + "/rest/zt/captcha/sliding/config"
BGPIC_URL = CAPTCHA_HOST + "/rest/zt/captcha/sliding/bgPic"
CUTPIC_URL = CAPTCHA_HOST + "/rest/zt/captcha/sliding/cutPic"
# 关键: error_url 里的整数 key 不能直接喂给 /sliding/config会 350004 session err
# APP 的 captcha.html 页面先走 JS 桥 /rest/wd/captcha/get用 {type,uri,key} 换出
# 真正的 protobuf captchaSession blobbase64前缀 "Cgp6dC5jYXB0Y2hh" = \n\x0a zt.captcha
# 再拿 blob 去 config。纯 HTTP 必须复刻这一步。
MINT_HOST = "https://app.m.kuaishou.com"
MINT_URL = MINT_HOST + "/rest/wd/captcha/get"
DEFAULT_CAPTCHA_TYPE = 7
DEFAULT_LOGIN_URI = "/rest/nebula/user/login/mobileVerifyCode"
KEY = "c7b645db-65e8-401f-b38c-4c07c5fff247"
# 真机 OnePlus PJZ110 / Adreno 830 的 WebView WebGL 指纹dumpsys SurfaceFlinger 实测)
GPU_INFO = {
"glRenderer": "Adreno (TM) 830",
"glVendor": "Qualcomm",
"unmaskRenderer": "Adreno (TM) 830",
"unmaskVendor": "Qualcomm",
}
MOBILE_UA = (
"Mozilla/5.0 (Linux; Android 16; PJZ110 Build/UKQ1.230917.001; wv) "
"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/126.0.6478.134 "
"Mobile Safari/537.36"
)
def jose_encrypt(payload: dict[str, Any]) -> str:
"""调用 tools/jose_encrypt.js 把 payload 加密成 base64 verifyParam。"""
proc = subprocess.run(
[NODE, str(JOSE_JS), json.dumps(payload, ensure_ascii=False)],
capture_output=True,
text=True,
timeout=60,
cwd=str(ROOT),
)
if proc.returncode != 0:
raise RuntimeError(f"jose_encrypt 进程失败 rc={proc.returncode}: {proc.stderr.strip()}")
out = proc.stdout.strip()
if out.startswith("ERR:") or not out:
raise RuntimeError(f"jose_encrypt 错误: {out or '(空)'}")
return out
def _enc(v: Any) -> str:
"""form-encode 单值dict/list 先 compact json.dumps再 encodeURIComponent。
与浏览器 encodeURIComponent 一致space→%20(非+)|→%7C,→%2C
"%22:→%3A{%7B。gpuInfo/captchaExtraParam 传 dict内层引号自动变 %22。
"""
s = (
json.dumps(v, ensure_ascii=False, separators=(",", ":"))
if isinstance(v, (dict, list))
else str(v)
)
return quote(s, safe="")
def jose_encrypt_form(fields: dict[str, Any]) -> str:
"""把 verify payload 按 10 字段固定序 form-encode 后喂 jose_encrypt_raw.js。
返回 base64 verifyParam。镜像 jose_encrypt(),但不做 JSON 包装——
明文是 `captchaSn=<enc>&bgDisWidth=<enc>&...&captchaExtraParam=<enc>`。
"""
missing = [k for k in _VERIFY_FIELD_ORDER if k not in fields]
if missing:
raise KeyError(f"verifyParam 缺字段: {missing}")
plaintext = "&".join(f"{k}={_enc(fields[k])}" for k in _VERIFY_FIELD_ORDER)
proc = subprocess.run(
[NODE, str(JOSE_RAW_JS), plaintext],
capture_output=True,
text=True,
timeout=60,
cwd=str(ROOT),
)
if proc.returncode != 0:
raise RuntimeError(f"jose_encrypt_raw rc={proc.returncode}: {proc.stderr.strip()}")
out = proc.stdout.strip()
if not out or out.startswith("ERR:"):
raise RuntimeError(f"jose_encrypt_raw 错误: {out or '(空)'}")
return out
def gap_target_x(bg_bytes: bytes, cut_bytes: bytes) -> int:
"""ddddocr 滑块缺口检测,返回缺口在 686 原生背景图里的 x。"""
import ddddocr
det = ddddocr.DdddOcr(det=False, ocr=False, show_ad=False)
res = det.slide_match(cut_bytes, bg_bytes)
tx = res.get("target_x")
if tx is None:
target = res.get("target") or [0]
tx = target[0] if target else 0
return int(tx)
def build_trajectory(
drag_distance: float,
*,
y: float = 0.0,
x0: float = 0.0,
seed: int | None = None,
) -> str:
"""拟人轨迹 -> "x|y|dt,x|y|dt,..."dt 相对首样本)。
x 从 x0 到 x0+drag_distance绝对指针 clientX 空间;浏览器实测终点 ~920 含按钮屏原点)。
余弦缓动 + 过冲回正 + 垂直抖动,对齐 captcha_assist._human_drag 的节奏。
"""
rnd = random.Random(seed)
steps = 44
total_ms = 820
peak = drag_distance + rnd.uniform(3.0, 9.0)
pts: list[list[float]] = []
t_cursor = rnd.randint(120, 260)
base = total_ms / steps
for i in range(1, steps + 1):
t = i / steps
ease = 0.5 * (1 - math.cos(math.pi * t))
x = x0 + peak * ease
yy = y + rnd.uniform(-2.0, 2.0)
t_cursor += int(base) + rnd.randint(0, 9)
pts.append([round(x, 2), round(yy, 2), t_cursor])
# 过冲后回正到 drag_distance
for j in range(1, 7):
t = j / 6
x = x0 + peak + (drag_distance - peak) * t
yy = y + rnd.uniform(-1.5, 1.5)
t_cursor += rnd.randint(14, 26)
pts.append([round(x, 2), round(yy, 2), t_cursor])
t_cursor += rnd.randint(90, 180)
pts.append([round(x0 + float(drag_distance), 2), round(y, 2), t_cursor])
base_t = pts[0][2]
return ",".join(f"{p[0]}|{p[1]}|{int(p[2] - base_t)}" for p in pts)
def captcha_extra_param() -> dict[str, Any]:
"""captchaExtraParam = merge(ua(), collectEnvInfo()) 的极简 Android 视图。
服务端对 captchaExtraParam 宽松(桌面 Chromium 也能 kSecretApiVerify result=1
故给出与设备身份自洽的 Android WebView 环境即可。
"""
return {
"ua": MOBILE_UA,
"language": "zh-cn",
"platform": "Linux armv8l",
"devicePixelRatio": 3,
"screenWidth": 1080,
"screenHeight": 2376,
"colorDepth": 24,
"timezone": -480,
"hardwareConcurrency": 8,
"deviceMemory": 12,
"touchSupport": "1",
"mod": "OnePlus(PJZ110)",
"sys": "ANDROID_16",
"appver": "14.5.50.11631",
"kpn": "NEBULA",
}
def _new_session():
from curl_cffi import requests as cffi_requests
# captcha.zt.kuaishou.com 服务于 APP 内 WebViewChrome/126 系 TLS
# 不需要登录链路的 OkHttp4 指纹。impersonate=chrome120 匹配 UA 的 JA3。
sess = cffi_requests.Session(impersonate="chrome120")
sess.headers.update(
{
"User-Agent": MOBILE_UA,
"Referer": "https://app.m.kuaishou.com/",
"Origin": "https://app.m.kuaishou.com",
"Accept-Language": "zh-CN,zh;q=0.9",
}
)
return sess
def _device_session(profile_path: str | None = None, verbose: bool = False):
"""带设备 cookie 的 sessionmint 需要 did/egid
复用 core.captcha_assist.sync_browser_cookies 把 build_captcha_browser_cookies(profile)
灌进 curl_cffi sessiondomain=.kuaishou.com 同时覆盖 captcha.zt 与 mint 主机 app.m。
默认加载 out/devices/device_001.json可复现
"""
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from core.captcha_assist import build_captcha_browser_cookies, sync_browser_cookies
from core.device_profile import DeviceProfileGenerator, load_device_profile
sess = _new_session()
default_profile = ROOT / "out" / "devices" / "device_001.json"
path = profile_path or str(default_profile)
profile = load_device_profile(path) if Path(path).is_file() else DeviceProfileGenerator().new_profile()
n = sync_browser_cookies(sess, build_captcha_browser_cookies(profile))
if verbose:
print(f"[solver] 注入 {n} 个设备 cookie did={getattr(profile, 'did', '?')}", flush=True)
return sess, profile
def _is_blob(s: str) -> bool:
"""protobuf captchaSession blob 的 base64 固定前缀field1="zt.captcha")。"""
return s.startswith("Cgp6dC5jYXB0Y2hh")
def _safe_json(resp: Any) -> tuple[dict | None, str]:
"""安全解析 JSON 响应。非 dict如耗尽 key 返字面 "result:501" 字符串)→ (None, raw)。"""
try:
obj = resp.json()
except Exception:
return None, (resp.text or "")[:300]
if not isinstance(obj, dict):
return None, repr(obj)[:300]
return obj, ""
def mint_captcha_session(
session: Any,
key: str,
*,
uri: str = DEFAULT_LOGIN_URI,
captchatype: int = DEFAULT_CAPTCHA_TYPE,
) -> tuple[str, str]:
"""把 error_url 的整数 key 换成 protobuf captchaSession blob。
复刻 APP captcha.html 页面的 `/rest/wd/captcha/get` 调用application/json
返回 (captchaSession_blob, config_url)。
"""
m = session.post(
MINT_URL,
json={"type": captchatype, "uri": uri, "key": key},
headers={"Content-Type": "application/json; charset=utf-8"},
timeout=15,
)
mj, raw = _safe_json(m)
if mj is None:
raise RuntimeError(f"mint 非 dict/无效响应: {raw}")
if mj.get("result") != 1 or not mj.get("data"):
raise RuntimeError(f"mint /rest/wd/captcha/get 失败 result={mj.get('result')} desc={mj.get('desc')}")
d_raw = mj["data"]
try:
d = json.loads(d_raw) if isinstance(d_raw, str) else d_raw
except Exception:
raise RuntimeError(f"mint data 不可解析: {repr(str(d_raw)[:200])}")
if not isinstance(d, dict) or not d.get("captchaSession"):
raise RuntimeError(f"mint 无 captchaSession: {repr(str(d)[:200])}")
return d["captchaSession"], d.get("url") or CONFIG_URL
def solve_captcha(
session: Any | None = None,
captcha_session: str = "",
*,
error_url: str = "",
uri: str = DEFAULT_LOGIN_URI,
captchatype: int = DEFAULT_CAPTCHA_TYPE,
offset: int = -48,
geom_space: str = "display",
traj_scale: float = NATIVE_TO_PHYS,
traj_y: float = 0.0,
traj_x_base: float = 0.0,
verbose: bool = True,
) -> dict[str, Any]:
"""对一个 captcha key/error_url/blob 跑完整纯 HTTP 流程,返回诊断 dict。
captcha_session 可为: error_url 里的整数 key、完整 error_url、或已 mint 出的 blob。
整数 key 会先经 /rest/wd/captcha/get 换成 blob复刻 APP captcha.html 的桥调用)。
session 可空(自建匿名 session正式接入登录时传入【与登录同一个】
curl_cffi session使 captcha token 与登录设备身份(did/egid cookies)绑定。
"""
own_session = session is None
if own_session:
session, _profile = _device_session(verbose=verbose)
if error_url:
from urllib.parse import parse_qs, urlsplit
qs = parse_qs(urlsplit(error_url).query)
if not captcha_session:
captcha_session = (qs.get("key") or qs.get("captchaSession") or [""])[0]
if qs.get("type"):
captchatype = int(qs["type"][0])
if qs.get("uri"):
uri = qs["uri"][0]
if not captcha_session:
return {"ok": False, "stage": "input", "error": "缺少 captchaSession/key"}
def log(msg: str) -> None:
if verbose:
print(f"[solver] {msg}", flush=True)
try:
# 0) 整数 key -> protobuf blob若已是 blob 则跳过)
if _is_blob(captcha_session):
cfg_url = CONFIG_URL
else:
log(f"mint key->blob type={captchatype} uri={uri}")
captcha_session, cfg_url = mint_captcha_session(
session, captcha_session, uri=uri, captchatype=captchatype
)
log(f"mint ok blob[:32]={captcha_session[:32]}...")
# 1) config
log(f"POST config captchaSession={captcha_session[:24]}...")
cfg_r = session.post(
cfg_url,
data={"captchaSession": captcha_session},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
cfg, cfg_raw = _safe_json(cfg_r)
if cfg is None:
return {"ok": False, "stage": "config", "status": cfg_r.status_code, "body": cfg_raw}
log(f"config result={cfg.get('result')} desc={cfg.get('desc')}")
if cfg.get("result") != 1:
return {"ok": False, "stage": "config", "raw": cfg}
sn = cfg["captchaSn"]
bg_w = int(cfg.get("bgPicWidth") or 686)
bg_h = int(cfg.get("bgPicHeight") or 400)
cut_w = int(cfg.get("cutPicWidth") or 122)
cut_h = int(cfg.get("cutPicHeight") or 122)
dis_x = int(cfg.get("disX") or 24)
dis_y = int(cfg.get("disY") or 122)
verify_url2 = cfg.get("verifyUrl2") or (CAPTCHA_HOST + "/rest/zt/captcha/sliding/kSecretApiVerify")
bg_url = (cfg.get("bgPicUrl") or BGPIC_URL) + f"?captchaSn={sn}"
cut_url = (cfg.get("cutPicUrl") or CUTPIC_URL) + f"?captchaSn={sn}"
# 2) bg/cut bytes
bg = session.get(bg_url, timeout=15).content
cut = session.get(cut_url, timeout=15).content
log(f"bg={len(bg)}B cut={len(cut)}B")
# 3) gap (原生 686 空间)
target_x = gap_target_x(bg, cut)
gap_native = target_x + offset # 缺口在原生图里的 x已含 ddddocr 右偏修正)
log(f"target_x={target_x} offset={offset} gap_native={gap_native} geom={geom_space} traj_scale={traj_scale:.3f}")
# 4) 几何空间 + trajectory 空间(浏览器实测:几何 DISPLAYtrajectory PHYSICAL混合空间
if geom_space == "display":
scale_d = DISPLAY_BG_W / NATIVE_BG_W # 0.4606
payload_bg_w, payload_bg_h = DISPLAY_BG_W, DISPLAY_BG_H
relative_x = round(gap_native * scale_d)
relative_y = DISPLAY_REL_Y
cut_dw, cut_dh = int(cut_w * scale_d), int(cut_h * scale_d)
else: # nativecontrol旧全原生行为
payload_bg_w, payload_bg_h = bg_w, bg_h
relative_x = gap_native
relative_y = dis_y
cut_dw, cut_dh = cut_w, cut_h
# trajectory 在物理像素空间drag_native × traj_scale默认 1.382
drag_native = gap_native - dis_x
drag_physical = drag_native * traj_scale
trajectory = build_trajectory(drag_physical, y=traj_y, x0=traj_x_base, seed=int(time.time()) & 0xFFFF)
# 5) captchaExtraParam = key1-39, 含拖拽相关传感器(合成, 与 trajectory 时间相关)
try:
from captcha_env import build_captcha_extra_param
except ImportError:
from tools.captcha_env import build_captcha_extra_param
extra_param = build_captcha_extra_param(
drag_physical=drag_physical,
slider_x=traj_x_base,
slider_y=float(dis_y) * (NATIVE_TO_PHYS if geom_space == "display" else 1.0) + 720.0,
duration_ms=820.0,
seed=int(time.time()) & 0xFFFF,
)
# 6) payload — gpuInfo/captchaExtraParam 传 dictjose_encrypt_form 内部 compact json + form-encode
payload = {
"captchaSn": sn,
"bgDisWidth": payload_bg_w,
"bgDisHeight": payload_bg_h,
"cutDisWidth": cut_dw,
"cutDisHeight": cut_dh,
"relativeX": relative_x,
"relativeY": relative_y,
"trajectory": trajectory,
"gpuInfo": GPU_INFO,
"captchaExtraParam": extra_param,
}
log(f"payload geom={geom_space} relX={relative_x} drag_native={drag_native} drag_phys={drag_physical:.1f}")
# 6) encryptform-encode 明文,非 JSON
verify_param = jose_encrypt_form(payload)
log(f"verifyParam len={len(verify_param)}")
# 7) verify (inner iframe: axios.post(url,{verifyParam},{Content-Type:application/json}))
verify_r = session.post(
verify_url2,
json={"verifyParam": verify_param},
headers={"Content-Type": "application/json"},
timeout=15,
)
vj, vj_raw = _safe_json(verify_r)
if vj is None:
vj = {"_text": vj_raw}
log(f"verify status={verify_r.status_code} body={str(vj)[:200]}")
# result==1 且含 captchaToken 才算成功
token = ""
if isinstance(vj, dict):
token = str(vj.get("captchaToken") or vj.get("token") or "")
return {
"ok": isinstance(vj, dict) and vj.get("result") == 1 and bool(token),
"stage": "verify",
"result": vj.get("result") if isinstance(vj, dict) else None,
"captcha_token": token,
"target_x": target_x,
"relativeX": relative_x,
"verifyParam": verify_param,
"raw": vj,
}
finally:
if own_session:
try:
session.close()
except Exception:
pass
def _default_session() -> str:
"""error_url 的整数 keydummy-code submit 免 SMS 回收),可反复 mint。"""
return "-7707052661950021982"
def sweep_solve(
key: str,
*,
session: Any | None = None,
offsets: tuple[int, ...] = (-60, -48, -36, -24, -12, 0),
traj_scales: tuple[float, ...] = (NATIVE_TO_PHYS, 1.0, 3.0),
geom_space: str = "display",
uri: str = DEFAULT_LOGIN_URI,
captchatype: int = DEFAULT_CAPTCHA_TYPE,
stop_on_success: bool = True,
verbose: bool = True,
) -> list[dict[str, Any]]:
"""对一个 key 扫描 (offset, traj_scale) 网格,找返回 result=1 的 cell350002 调参)。
一个设备 cookie session 复用所有 cell每次 solve_captcha 内部重 mint 新
captchaSessioncaptchaSn 是否单用未知,保守重 mintoffset 是 ddddocr 系统性
右偏的每图常量修正,会跨图泛化)。
"""
own_session = session is None
if own_session:
session, _profile = _device_session(verbose=verbose)
results: list[dict[str, Any]] = []
try:
for traj_scale in traj_scales:
for off in offsets:
res = solve_captcha(
session, key,
offset=off, traj_scale=traj_scale, geom_space=geom_space,
uri=uri, captchatype=captchatype, verbose=verbose,
)
cell = {
"offset": off, "traj_scale": round(traj_scale, 3), "geom": geom_space,
"ok": res.get("ok"), "result": res.get("result"),
"stage": res.get("stage"), "target_x": res.get("target_x"),
"token": (res.get("captcha_token") or "")[:16],
}
results.append(cell)
print(f"[sweep] off={off:+d} scale={traj_scale:.3f} geom={geom_space} "
f"-> result={cell['result']} stage={cell['stage']}", flush=True)
if cell["ok"] and stop_on_success:
print(f"[sweep] ✅ 命中 off={off:+d} scale={traj_scale:.3f}", flush=True)
return results
finally:
if own_session:
try:
session.close()
except Exception:
pass
return results
if __name__ == "__main__":
args = sys.argv[1:]
sweep = "--sweep" in args
positional = [a for a in args if a != "--sweep"]
key = positional[0] if positional else _default_session()
if sweep:
rows = sweep_solve(key, verbose=True)
print(json.dumps(rows, ensure_ascii=False, indent=2))
else:
res = solve_captcha(captcha_session=key, verbose=True)
print(json.dumps(res, ensure_ascii=False, indent=2))