#!/usr/bin/env python3 """YYB web_save encrypt_msg 最终交付(纯 Python,无浏览器依赖)。 链路(F-2046..F-2052,E3 已验证): 会话态(xMidasOps+key16/key1) + 订单参数 → build_plaintext 21 字段明文(528 字符) → a:8 变换(标准 AES Te0-Te3,key16 按块轮转) → webSave(goodsBiz CHAOS VM 33 块变换) → encrypt_msg(1056 hex) → 提交 web_save → ret:0 + 微信支付 URL 用法: python3 main.py verify # 回归测试:双向量 + 离线 + live 复现 python3 main.py gen --session S --order O # 仅生成 encrypt_msg(不联网) python3 main.py submit --session S --order O [--appid A] [--body-tpl T] # 生成 + 提交真实 web_save(需授权登录态) python3 main.py sample-session # 用迁移的 live 捕获生成示例会话态 python3 main.py sample-order # 用迁移的 order7 生成示例订单 """ from __future__ import annotations import argparse import json import random import re import sys import time import urllib.request import uuid from pathlib import Path ROOT = Path(__file__).resolve().parent sys.path.insert(0, str(ROOT)) from pyvm.algorithm import build_plaintext, generate_encrypt_msg, generate_encrypt_msg_offline # noqa: E402 from pyvm.mall import MallSession, generate_encrypt_msg as mall_generate # noqa: E402 from pyvm.session import SessionState, load_session # noqa: E402 from pyvm.login_profile import midas_login_params # noqa: E402 REPLAY = ROOT / "replay" DEFAULT_APPID = "1450243039" DEFAULT_SAVE_URL = f"https://api.unipay.qq.com/v1/r/{DEFAULT_APPID}/web_save" ORDER_FIELDS = ["token_id", "openid", "openkey", "session_id", "session_type", "zoneid", "pay_method", "buy_quantity", "mb_pwd", "pay_id", "auth_key", "card_value", "accounttype", "provide_uin", "extend", "ts", "from_h5", "webversion"] # ---------------------------------------------------------------- 工具 def _load_cap(path: Path) -> dict: """加载 deepCap 捕获 JSON,兼容三种存档格式,返回顶层 dict(含 C 键)。 格式1: 直接 JSON 对象 {"u":85091,"C":[...]} 格式2: 双重编码 JSON 字符串 '"{\"u\":85091,...}"'(json.dumps(json.dumps(x))) 格式3: 手动转义去外层引号 {\"u\":85091,...}(replace('"','\\"') 后丢外层引号) """ raw = path.read_text().strip() d = None for attempt in ( lambda: json.loads(raw), lambda: json.loads(json.loads(raw)), lambda: json.loads(json.loads('"' + raw + '"')), ): try: d = attempt() if isinstance(d, dict) and "C" in d: return d except Exception: continue raise ValueError(f"无法解析 deepCap 文件: {path}(前 80 字符: {raw[:80]!r})") def _vector_of(body_path: Path) -> str: """从归档 body.json(json 双重编码)提取 encrypt_msg 期望值。""" body = json.loads(json.loads(body_path.read_text().strip())) bs = body["body"] i = bs.find("encrypt_msg=") return bs[i + len("encrypt_msg="): i + len("encrypt_msg=") + 1056] def load_order(path: str | Path) -> dict: p = Path(path) if not p.exists(): raise FileNotFoundError(f"订单文件不存在: {p}") return json.loads(p.read_text(encoding="utf-8")) def parse_plaintext(path: str | Path) -> dict: """解析归档明文:JSON dict 或 key=value&... 两种格式自动识别。""" raw = Path(path).read_text(encoding="utf-8").strip() try: d = json.loads(raw) if isinstance(d, dict): return {str(k): str(v) for k, v in d.items()} except Exception: pass fields: dict[str, str] = {} for kv in raw.split("&"): k, _, v = kv.partition("=") fields[k] = v return fields # ---------------------------------------------------------------- 命令 def cmd_verify(args=None) -> int: """回归测试:双向量(runA/runB)+ 离线(e2e/vector-b)+ live order7 复现。""" fails = 0 def check(name: str, got: str, expected: str) -> None: nonlocal fails ok = len(got) == 1056 and got == expected print(f" [{'✅' if ok else '❌'}] {name}: " f"{'1056/1056' if ok else f'失配 len={len(got)}'}") if not ok: fails += 1 print(f" got head: {got[:32]}...") print(f" exp head: {expected[:32]}...") print("== 1. 深拷贝 18 参向量(generate_encrypt_msg)==") for name, args_p, cb_p, body_p in [ ("runA", "deepcaps/cap-85091.json", "deepcaps2/cap-69667.json", "deepcaps/body.json"), ("runB", "deepcaps2/cap-85091.json", "deepcaps2/cap-69667.json", "deepcaps2/body.json"), ]: got = generate_encrypt_msg(str(REPLAY / args_p), str(REPLAY / cb_p)) check(name, got, _vector_of(REPLAY / body_p)) print("== 2. 离线生成(generate_encrypt_msg_offline,随机 key16/key1)==") for name, plain_p, body_p_or_vector in [ ("offline-runA(e2e)", "e2e/plaintext.json", "e2e/body.json"), ("offline-runB", "deepcaps2/plaintext.json", "vector-20260810b.json"), ]: fields = parse_plaintext(REPLAY / plain_p) params = {k: fields[k] for k in ORDER_FIELDS} got = generate_encrypt_msg_offline( params, fields["fk_extend"], fields["ts"], fields["_rand"]) if body_p_or_vector.endswith(".json") and "vector" in body_p_or_vector: expected = json.loads((REPLAY / body_p_or_vector).read_text())["encrypt_msg"] else: expected = _vector_of(REPLAY / body_p_or_vector) check(name, got, expected) print("== 3. live 会话态复现(order7,精确 key)==") fields = parse_plaintext(REPLAY / "live/order7-plaintext.txt") params = {k: fields[k] for k in ORDER_FIELDS} xmidas = json.loads((REPLAY / "live/order7-xmidasops.json").read_text()) cap = _load_cap(REPLAY / "live/caps7/cap-85091.json") from pyvm.algorithm import decode_d web_args = decode_d(cap["C"][2]) got = generate_encrypt_msg_offline( params, fields["fk_extend"], fields["ts"], fields["_rand"], xmidas=xmidas, args_template=web_args, key16=web_args[6][0], key1=web_args[0][0], ) expected = (REPLAY / "live/order7-page-encrypt.txt").read_text().strip() check("live-order7", got, expected) print("== 4. 更多 live 会话复现(order10 / order12,精确 key)==") for od in ["order10", "order12"]: d = REPLAY / "live" / od fields = parse_plaintext(d / "plaintext.txt") params = {k: fields[k] for k in ORDER_FIELDS} keys = json.loads((d / "keys.json").read_text()) xmidas = json.loads((d / "xmidasops.json").read_text()) web_args = decode_d(_load_cap(d / "caps" / "cap-85091.json")["C"][2]) got = generate_encrypt_msg_offline( params, fields["fk_extend"], fields["ts"], fields["_rand"], xmidas=xmidas, args_template=web_args, key16=keys["key16"], key1=keys["key1"], ) body = (d / "body.txt").read_text() m = re.search(r"encrypt_msg=([0-9a-f]{1056})", body) check(f"live-{od}", got, m.group(1) if m else "") print() if fails: print(f"❌ {fails} 项失败") return 1 print("✅ 全部通过(E3 双向量 + 离线 + live 复现 ×5)") return 0 def _gen_with_session(st: SessionState, order: dict) -> str: """用会话态 + 订单参数生成 encrypt_msg(args_template 会话绑定,必须用捕获值)。""" from pyvm.algorithm import decode_d params = {k: order.get(k, "") for k in ORDER_FIELDS} args_tpl = decode_d(st.args_template_d) return generate_encrypt_msg_offline( params, order.get("fk_extend", ""), order.get("ts", ""), order.get("_rand", ""), xmidas=st.xmidas_ops, xmidas_token=st.xmidas_token, args_template=args_tpl, key16=st.key16, key1=st.key1, ) def cmd_gen(args) -> int: st = load_session(args.session) order = load_order(args.order) params = {k: order.get(k, "") for k in ORDER_FIELDS} hex_msg = _gen_with_session(st, order) out = Path(args.output) if args.output else ROOT / "config" / "encrypt_msg.txt" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(hex_msg + "\n") print(f"encrypt_msg ({len(hex_msg)} hex) → {out}") print(hex_msg) return 0 def _build_body(order: dict, st: SessionState, offline_hex: str, body_tpl: Path) -> str: """在归档 body 模板上刷新订单字段与动态值。""" body = json.loads(json.loads(body_tpl.read_text()))["body"] # 刷新订单绑定字段(明文 18 字段中出现在 body 里的) for k in ORDER_FIELDS: if k in ("ts", "from_h5"): continue v = str(order.get(k, "")) body = re.sub(rf"{k}=[^&]*", f"{k}=" + v, body, count=1) # 动态值 new_pc = str(uuid.uuid4()).upper() + str(int(time.time() * 1000)) body = re.sub(r"pc_st=[^&]+", "pc_st=" + new_pc, body) body = re.sub(r"r=[0-9.]+", "r=" + str(random.random()), body) body = re.sub(r"&t=[0-9]+", "&t=" + str(int(time.time() * 1000)), body) body = re.sub(r"encrypt_msg=[0-9a-f]+", "encrypt_msg=" + offline_hex, body) return body def cmd_submit(args) -> int: st = load_session(args.session) order = load_order(args.order) if not st.cookies: print("⚠️ 会话态缺少 cookies —— 提交真实 web_save 需要登录态。") print(" 仅生成模式请用: python3 main.py gen --session ... --order ...") return 2 hex_msg = _gen_with_session(st, order) body_tpl = Path(args.body_tpl) if args.body_tpl else REPLAY / "e2e" / "body.json" body = _build_body(order, st, hex_msg, body_tpl) url = args.url or DEFAULT_SAVE_URL cookie_str = "; ".join(f"{k}={v}" for k, v in st.cookies.items()) req = urllib.request.Request( url, data=body.encode("utf-8"), headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", "Content-Type": "application/x-www-form-urlencoded", "Origin": "https://pay.qq.com", "Referer": "https://pay.qq.com/", "Cookie": cookie_str, }, method="POST", ) print(f"[submit] POST {url}") try: with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") except Exception as e: # noqa: BLE001 print(f"[submit] 网络错误: {e!r}") return 3 print(f"[submit] 响应: {raw[:600]}") try: js = json.loads(raw) ret = js.get("ret") if ret == 0: print("✅ web_save ret:0 —— 加密通过,支付流程继续") return 0 print(f"❌ web_save ret:{ret}({js.get('err_code', '')})—— 见 case 踩坑记录") return 1 except Exception: return 1 def cmd_sample_session(args) -> int: """用迁移的 live 捕获生成示例会话态(key16/key1/xmidas_ops)。""" from pyvm.algorithm import decode_d web_args = decode_d(_load_cap(REPLAY / "live/caps7/cap-85091.json")["C"][2]) cap7 = _load_cap(REPLAY / "live/caps7/cap-85091.json") st = SessionState( xmidas_ops=json.loads((REPLAY / "live/order7-xmidasops.json").read_text()), key16=list(web_args[6][0]), key1=list(web_args[0][0]), xmidas_token="DE46DBA4754D42A6B66ADD4319FF80C144D9ED22ED73384C271793D9A9AA2FFBD6C104BE8D4A7F4ED2A8688FCB6F7540", args_template_d=cap7["C"][2] if isinstance(cap7["C"][2], str) else json.dumps(cap7["C"][2], ensure_ascii=False), cookies={}, source="live/order7 (归档示例,cookies 为空)", ) out = Path(args.output) if args.output else ROOT / "config" / "session-state.json" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(st.to_json(), ensure_ascii=False, indent=2) + "\n") print(f"示例会话态 → {out}(仅含归档 key16/key1/xmidas_ops,cookies 需自行捕获)") return 0 def cmd_sample_order(args) -> int: fields = parse_plaintext(REPLAY / "live/order7-plaintext.txt") out = Path(args.output) if args.output else ROOT / "config" / "order.json" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(fields, ensure_ascii=False, indent=2) + "\n") print(f"示例订单 → {out}") return 0 # ---------------------------------------------------------------- main # ---------------------------------------------------------------- mall 命令 def cmd_mall_verify(args=None) -> int: """E3 黄金对验证:同会话 transform_input + xMidasOps → encrypt_msg 完全一致。""" g = json.loads((ROOT / "config/golden/golden-final.json").read_text(encoding="utf-8")) session = MallSession(g["transform_input"], g["xmidas_ops"]) hexout = mall_generate(session) expected = g["encrypt_msg_hex"] ok = hexout == expected print(f" [{'✅' if ok else '❌'}] mall 黄金对复现: {('完全一致 ' + hexout[:24] + '...') if ok else '失配'}") print(" xMidasOps:", len(session.xmidas_ops), "| encrypt_msg:", len(expected), "hex") return 0 if ok else 1 def cmd_mall_gen(args) -> int: """从 mall 会话态生成 encrypt_msg(纯 Python,E3)。""" d = json.loads(Path(args.session).read_text(encoding="utf-8")) session = MallSession(d["transform_input"], d["xmidas_ops"]) hexout = mall_generate(session) out = Path(args.output) if args.output else ROOT / "config" / "mall-encrypt_msg.txt" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(hexout + "\n") print(f"mall encrypt_msg ({len(hexout)} hex) → {out}") print(hexout[:64] + "...") return 0 def cmd_mall_sample(args) -> int: """用归档同会话黄金对生成示例 mall 会话态(config/mall-session.json)。""" g = json.loads((ROOT / "config/golden/golden-final.json").read_text(encoding="utf-8")) session = MallSession(g["transform_input"], g["xmidas_ops"]) out = Path(args.output) if args.output else ROOT / "config" / "mall-session.json" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(session.to_json(), ensure_ascii=False, indent=2) + "\n") print(f"示例 mall 会话态 → {out}(含 transform_input + xMidasOps 59620)") return 0 def cmd_mall_submit(args) -> int: """生成 encrypt_msg + 提交 PlaceOrder（纯 Python）。 会话态（transform_input + xMidasOps + cookies）来自浏览器采集； 订单模板（mall-order-template.json）来自采集的 PlaceOrder 请求体。 用 Python 生成的 encrypt_msg 替换模板中的原生值后提交。 """ d = json.loads(Path(args.session).read_text(encoding="utf-8")) session = MallSession(d["transform_input"], d["xmidas_ops"]) cookies = d.get("cookies", {}) if not cookies: print("⚠️ 会话态缺少 cookies（登录态）——采集脚本需在登录后保存 cookie") return 2 hex_msg = mall_generate(session) print(f"Python encrypt_msg: {hex_msg[:32]}...") tpl = json.loads(Path(args.order_template).read_text(encoding="utf-8")) body = tpl["body"] body_obj = json.loads(body) _refresh_login_check(body_obj, cookies) cpj = json.loads(body_obj["call_param"]["call_param_json"]) cpj["encrypt_msg"] = hex_msg body_obj["call_param"]["call_param_json"] = json.dumps(cpj, ensure_ascii=False) new_body = json.dumps(body_obj, ensure_ascii=False) url = "https://pagedooapi.pay.qq.com/api/CommonCallMpgo?t=" + str(int(time.time() * 1000)) cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items()) # midas 域 cookie 兜底(profile 导出常缺 midas_openid/midas_openkey,值为 openid/accesstoken) if "midas_openid" not in cookies and "openid" in cookies: cookie_str += "; midas_openid=" + cookies["openid"] if "midas_openkey" not in cookies and "accesstoken" in cookies: cookie_str += "; midas_openkey=" + cookies["accesstoken"] req = urllib.request.Request( url, data=new_body.encode("utf-8"), headers={ "Content-Type": "application/json", "Accept": "application/json, text/plain, */*", "Origin": "https://z.iwan.yyb.qq.com", "Referer": "https://z.iwan.yyb.qq.com/", "Cookie": cookie_str, "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36", }, method="POST", ) print(f"[submit] POST {url}") try: with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") except Exception as e: # noqa: BLE001 print(f"[submit] 网络错误: {e!r}") return 3 print(f"[submit] 响应: {raw[:800]}") out = Path(args.output) if args.output else ROOT / "config" / "mall-order-response.json" js = None ret = None try: js = json.loads(raw) if isinstance(js, dict): ret = js.get("ret", js.get("result_code", js.get("code"))) except Exception: pass ok = ret in (0, "0") if ok: out.write_text(raw, encoding="utf-8") print(f"✅ 提交完成（ret={ret}），响应已保存 → {out}") return 0 # 失败:不覆盖上次成功响应,另存 .fail.json fail_out = out.with_suffix(".fail.json") fail_out.write_text(raw, encoding="utf-8") info = "" if isinstance(js, dict): info = str(js.get("result_info") or js.get("msg") or js.get("err_code") or "") print(f"❌ 提交失败（ret={ret}）{info}") print(f" 失败响应 → {fail_out}(保留上次成功响应)") if ret in ("1018", 1018): print(" 原因: mall 登录态失效——请重新在浏览器登录并采集 mall 会话态") print(" (node scripts/capture-mall-data.mjs,或手动刷新 mall-session.json 的 cookies)") return 1 def cmd_mall_capture(args) -> int: """用浏览器捕获 mall 会话态(需授权登录,Node 脚本)。""" print("mall 会话态捕获(需在浏览器完成一次详情页购买):") print(" node scripts/capture-mall-session.mjs --url '<详情页URL>' --duration 600") print(" → config/golden-pair.frames.jsonl → python3 main.py mall sample") return 0 def main() -> int: ap = argparse.ArgumentParser(description="YYB 加密参数生成框架(goods web_save + mall PlaceOrder)") sub = ap.add_subparsers(dest="module", required=True) # ---- goods 组 ---- pg = sub.add_parser("goods", help="goods 侧(web_save / web_new_encrypt,CHAOS VM 116 opcode)") gsub = pg.add_subparsers(dest="cmd", required=True) gsub.add_parser("verify", help="E3 回归(双向量 + 离线 + live 复现)") p = gsub.add_parser("gen", help="仅生成 encrypt_msg(不联网)") p.add_argument("--session", required=True) p.add_argument("--order", required=True) p.add_argument("--output", default=None) p = gsub.add_parser("submit", help="生成 + 提交真实 web_save(需授权登录态)") p.add_argument("--session", required=True) p.add_argument("--order", required=True) p.add_argument("--appid", default=DEFAULT_APPID) p.add_argument("--url", default=None) p.add_argument("--body-tpl", default=None) p = gsub.add_parser("sample-session", help="用归档 live 捕获生成示例会话态") p.add_argument("--output", default=None) p = gsub.add_parser("sample-order", help="用归档 order7 生成示例订单") p.add_argument("--output", default=None) # ---- mall 组 ---- pm = sub.add_parser("mall", help="mall 侧(PlaceOrder,pagedoo VM 108 opcode)") msub = pm.add_subparsers(dest="cmd", required=True) msub.add_parser("verify", help="E3 黄金对验证(e377650 复现 encrypt_msg)") p = msub.add_parser("gen", help="从 mall 会话态生成 encrypt_msg(纯 Python)") p.add_argument("--session", required=True) p.add_argument("--output", default=None) p = msub.add_parser("sample-session", help="从捕获或归档生成 mall 会话态") p.add_argument("--frames", default=None, help="最新捕获的 frames.jsonl（优先）") p.add_argument("--output", default=None) p = msub.add_parser("submit", help="生成 encrypt_msg + 提交 PlaceOrder(纯 Python)") p.add_argument("--session", required=True, help="mall-session.json(含 transform_input/xmidas/cookies)") p.add_argument("--order-template", default=str(ROOT / "config" / "mall-order-template.json")) p.add_argument("--output", default=None) p = msub.add_parser("auto", help="仅凭 CK 全自动下单(纯HTTP GetPayToken + 纯Python encrypt_msg,无浏览器采集)") p.add_argument("--session", default=str(ROOT / "config" / "mall-session.json")) p.add_argument("--order-template", default=str(ROOT / "config" / "mall-order-template.json")) p.add_argument("--product-id", default=None, help="和平精英点券商品 ID；默认使用模板当前商品") p.add_argument("--offer-id", default=None, help="当前商品服务端 offer ID") p.add_argument("--quantity", type=int, default=None, help="当前点券商品购买份数（正整数）") p.add_argument("--role-id", default=None, help="游戏角色 ID；默认使用模板当前角色") p.add_argument("--role-name", default=None, help="游戏角色名称；默认使用模板当前角色") p.add_argument("--zone-id", default=None, help="游戏区服 ID；默认使用模板当前区服") p.add_argument("--zone-name", default=None, help="游戏区服名称") p.add_argument("--area", default=None, help="游戏大区 ID（QQ 平台与 zoneid 不同，来自角色查询的 partition_info）") p.add_argument("--partition", default=None, help="游戏分区 ID（QQ 平台可为空）") p.add_argument("--pf", default=None, help="支付平台标识；由角色选择器按 Android/iOS 写入") p.add_argument("--output", default=None) p = msub.add_parser("pay", help="mall 下单 -> goods web_save -> 微信支付二维码(纯 Python)") p.add_argument("--mall-response", default=str(ROOT / "config" / "mall-order-response.json"), help="PlaceOrder 响应(含 url_params/token)") p.add_argument("--goods-dir", default=str(ROOT / "replay" / "live" / "order10"), help="goods 会话态目录(plaintext/body/xmidasops/keys/cap/web-token)") p.add_argument("--session", default=str(ROOT / "config" / "mall-session.json"), help="mall 登录态(含 cookies)") p.add_argument("--appid", default=DEFAULT_APPID) p.add_argument("--output", default=None, help="二维码 PNG 输出路径") msub.add_parser("capture", help="用浏览器捕获 mall 会话态(Node 脚本)") args = ap.parse_args() try: if args.module == "goods": return {"verify": cmd_verify, "gen": cmd_gen, "submit": cmd_submit, "sample-session": cmd_sample_session, "sample-order": cmd_sample_order}[args.cmd](args) elif args.module == "mall": return {"verify": cmd_mall_verify, "gen": cmd_mall_gen, "submit": cmd_mall_submit, "sample-session": cmd_mall_sample, "pay": cmd_mall_pay, "auto": cmd_mall_auto, "capture": cmd_mall_capture}[args.cmd](args) return 2 except (FileNotFoundError, ValueError) as e: print(f"错误: {e}") return 2 def build_mall_transform(fixed: dict) -> list: """仅CK全自动:固定槽模板 + 随机 key16/槽6/明文缓冲 构造 transform_input。 E3 验证(2026-08-11):mall 服务端不校验 encrypt_msg 内容——随机 key16/槽6/ 明文缓冲(槽10)提交均 ret=0;仅需固定槽(Te/S-box/表)+ GetPayToken arrays。 ⚠️ 风险标记:随机 key/明文缓冲目前可用,但服务端可能后续校验加密内容, 或风控严格后拒绝随机加密(如 goods 侧随机 key 已被拒 1099)。若风控升级, `mall auto` 可能失效,需回退到真实会话态采集(capture-mall-data.mjs)。 """ import random ti = [None] * 18 for i in (1, 2, 3, 4, 5, 7, 8, 9, 11, 12, 13, 14, 15, 16, 17): ti[i] = json.loads(json.dumps(fixed[str(i)])) ti[0] = [[random.randint(0, 255) for _ in range(16)]] ti[6] = [[random.randint(0, 255) for _ in range(16)]] ti[10] = [[random.randint(0, 255) for _ in range(624)]] return ti def mall_getpaytoken(cookies: dict) -> tuple[list, str]: """GetPayToken(纯HTTP,仅需 cookies)→ (arrays 59620, pay_token)。 arrays = mall xMidasOps 来源(页面级);pay_token = mall web_token。 证据: evidence/getpaytoken-pure-http.json """ import urllib.request login = midas_login_params(cookies) login["offer_id"] = "800001492" body = { "acct_id": "1", "call_param": { "call_func": "GetPayToken", "call_param_json": json.dumps( {"version": "pagedoo-v2.0.0", "app_id": "202406061128117473047424", "content_id": "ct1755160919_GEOCGTMN"}), "call_type": "security_service", "login_check_param_json": json.dumps(login), }, } url = "https://pagedooapi.pay.qq.com/api/CommonCallMpgo?t=" + str(int(time.time() * 1000)) cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items()) for k, v in [("midas_openid", "openid"), ("midas_openkey", "accesstoken")]: if k not in cookies and v in cookies: cookie_str += f"; {k}=" + cookies[v] req = urllib.request.Request(url, data=json.dumps(body).encode("utf-8"), headers={ "Content-Type": "application/json", "Accept": "application/json, text/plain, */*", "Origin": "https://z.iwan.yyb.qq.com", "Referer": "https://z.iwan.yyb.qq.com/", "Cookie": cookie_str, "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36", }, method="POST") try: with urllib.request.urlopen(req, timeout=30) as r: raw = r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") js = json.loads(raw) cr = json.loads(js["data"]["call_reply"]) data = cr.get("data", {}) return data.get("arrays", []), data.get("pay_token", "") def _refresh_login_check(body_obj: dict, cookies: dict) -> None: """刷新 PlaceOrder 请求的 login_check_param_json(openid/openkey 等)为当前 CK。 模板(archived)中的 openid/openkey 是采集时的旧值,过期后必须用当前 mall-session cookies 替换,否则 pagedoo 返回 1018 login state check failed。 """ cp = body_obj.get("call_param", {}) try: lcp = json.loads(cp.get("login_check_param_json", "{}")) except Exception: # noqa: BLE001 lcp = {} lcp.update(midas_login_params(cookies)) if "offer_id" not in lcp: lcp["offer_id"] = "800001492" cp["login_check_param_json"] = json.dumps(lcp, ensure_ascii=False) def _apply_card_selection(payload: dict, args) -> None: """将用户显式指定的和平精英商品、角色和平台写入本次 PlaceOrder 载荷。 模板提供产品的服务端关联字段；这里只覆盖用户选择的商品、角色和平台字段， 不会修改磁盘上的模板，也不会猜测不同点券档位对应的 product_id。 """ quantity = getattr(args, "quantity", None) if quantity is not None and quantity <= 0: raise ValueError("--quantity 必须是正整数") products = payload.get("product_list") if not isinstance(products, list) or not products or not isinstance(products[0], dict): raise ValueError("订单模板缺少 product_list[0]") product = products[0] product_id = getattr(args, "product_id", None) if product_id: product["product_id"] = product_id metadata = payload.setdefault("metadata", {}) if isinstance(metadata, dict): metadata["productItemId"] = product_id offer_id = getattr(args, "offer_id", None) if offer_id: product["provide_offer_id"] = offer_id if quantity is not None: product["quantity"] = str(quantity) role_id = getattr(args, "role_id", None) if role_id: product["roleid"] = role_id payload["roleid"] = role_id role_name = getattr(args, "role_name", None) if role_name: product["rolename"] = role_name zone_id = getattr(args, "zone_id", None) if zone_id: product["zoneid"] = zone_id payload["zoneid"] = zone_id # QQ 平台的大区(area)与 zoneid 不同，来自角色查询的 partition_info； # 必须用真实 area，否则 PlaceOrder 校验角色失败(gamerole err)。 area = getattr(args, "area", None) if area: product["area"] = area elif zone_id: product["area"] = zone_id partition = getattr(args, "partition", None) if partition is not None: product["partition"] = partition zone_name = getattr(args, "zone_name", None) if zone_name: product["zonename"] = zone_name metadata = payload.setdefault("metadata", {}) if isinstance(metadata, dict): metadata["zone_name"] = zone_name metadata = payload.setdefault("metadata", {}) if isinstance(metadata, dict): metadata["trace_key"] = uuid.uuid4().hex pf = getattr(args, "pf", None) if pf: payload["pf"] = pf def cmd_mall_auto(args) -> int: """仅凭 CK 全自动下单(纯HTTP + 纯Python,无浏览器采集)。 链路:GetPayToken(纯HTTP)→ arrays+xMidasOps + pay_token(web_token) → 构造 transform_input(固定槽+随机key/明文缓冲) → 纯 Python 生成 encrypt_msg → PlaceOrder 提交 """ d = json.loads(Path(args.session).read_text(encoding="utf-8")) cookies = d.get("cookies", {}) if not cookies: print("⚠️ 缺少 cookies(登录态)") return 2 print("[auto] ① GetPayToken(纯HTTP)...") arrays, pay_token = mall_getpaytoken(cookies) if len(arrays) != 59620: print(f"❌ arrays 长度 {len(arrays)} != 59620") return 1 print(f"[auto] arrays {len(arrays)} | pay_token {pay_token[:16]}...") print("[auto] ② 构造 transform_input(固定槽 + 随机 key/明文缓冲)...") fixed = json.loads((ROOT / "replay/mall/transform-fixed.json").read_text(encoding="utf-8")) ti = build_mall_transform(fixed) print("[auto] ③ 纯 Python 生成 encrypt_msg...") session = MallSession(ti, arrays) hex_msg = mall_generate(session) print(f"[auto] encrypt_msg ({len(hex_msg)} hex): {hex_msg[:24]}...") print("[auto] ④ PlaceOrder 提交...") tpl = json.loads(Path(args.order_template).read_text(encoding="utf-8")) body_obj = json.loads(tpl["body"]) _refresh_login_check(body_obj, cookies) cpj = json.loads(body_obj["call_param"]["call_param_json"]) _apply_card_selection(cpj, args) cpj["encrypt_msg"] = hex_msg cpj["web_token"] = pay_token body_obj["call_param"]["call_param_json"] = json.dumps(cpj, ensure_ascii=False) new_body = json.dumps(body_obj, ensure_ascii=False) cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items()) for k, v in [("midas_openid", "openid"), ("midas_openkey", "accesstoken")]: if k not in cookies and v in cookies: cookie_str += f"; {k}=" + cookies[v] url = "https://pagedooapi.pay.qq.com/api/CommonCallMpgo?t=" + str(int(time.time() * 1000)) req = urllib.request.Request(url, data=new_body.encode("utf-8"), headers={ "Content-Type": "application/json", "Accept": "application/json, text/plain, */*", "Origin": "https://z.iwan.yyb.qq.com", "Referer": "https://z.iwan.yyb.qq.com/", "Cookie": cookie_str, "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36", }, method="POST") try: with urllib.request.urlopen(req, timeout=30) as r: raw = r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") out = Path(args.output) if args.output else ROOT / "config" / "mall-order-response.json" out.write_text(raw, encoding="utf-8") js = json.loads(raw) ret = js.get("result_code") try: call_reply = json.loads(js["data"]["call_reply"]) except (KeyError, TypeError, json.JSONDecodeError): call_reply = {} # PlaceOrder 业务失败(如 10014 gamerole err)时外层 result_code 仍可能为 0， # 必须检查 call_reply 的 result_code，避免把失败当成功后再崩溃。 inner_ret = call_reply.get("result_code") if ret == "0" and str(inner_ret) in ("0", "None", ""): data = call_reply.get("data") or {} token = str(data.get("token") or "")[:24] print(f"✅ 仅CK全自动下单成功! token={token or '(无 token)'}...") print(f" 响应已保存 → {out}") return 0 detail = js.get("result_info", "") or call_reply.get("result_info", "") or "" print(f"❌ 下单失败 ret={ret} inner={inner_ret} ({detail})") print(f" 响应已保存 → {out}") return 1 def cmd_mall_pay(args) -> int: """mall 下单 -> goods web_save -> 微信支付二维码(纯 Python,终极闭环)。 数据流: ① mall auto 下单(PlaceOrder 响应含 url_params -> goods 页) ② capture-goods-(auto|session).mjs 采集 goods 会话态 (xmidasops/key16/key1/args-template/web-token/body.txt) ③ 本命令:从捕获 body 取订单字段(web_token 绑定该订单), recover_plaintext_from_buffer 恢复页面真实明文(ts/fk_extend/_rand), 纯 Python 生成 encrypt_msg -> 构造 web_save body -> 提交 -> channel_info.sign(weixin://wxpay/bizpayurl?pr=...) E3(2026-08-11):同会话 4 变体全部 ret=0(页面body/精确纯py/新鲜ts/全流程), web_save 不消费订单,同一订单可重复提交。 """ # 1. goods 会话态 + 捕获 body(web_token 与订单绑定,同一次页面加载) gd = Path(args.goods_dir) if not gd.is_dir(): print(f"❌ goods 会话目录不存在: {gd}") return 2 body_tpl = (gd / "body.txt").read_text(encoding="utf-8") if not body_tpl: print("❌ 捕获目录缺少 body.txt(web_save 请求体)") return 2 body_fields = parse_plaintext(gd / "body.txt") token_id = body_fields.get("token_id", "") transaction_id = body_fields.get("transaction_id", "") out_trade_no = body_fields.get("out_trade_no", "") offer_type = body_fields.get("offer_type", "0") if not token_id or not out_trade_no: print(f"❌ 捕获 body 缺少 token 字段: token_id={token_id[:20]!r}") return 2 # mall 响应仅作核对(若换了新订单,web_token 不适用,须重采 goods 会话) resp_path = Path(args.mall_response) if resp_path.exists(): try: resp = json.loads(resp_path.read_text(encoding="utf-8")) cr = json.loads(resp["data"]["call_reply"]) up = cr["data"]["url_params"] from urllib.parse import parse_qs, urlparse q = parse_qs(urlparse(up).query) mall_tid = q.get("token_id", [""])[0] if mall_tid and mall_tid != token_id: print(f"⚠️ mall 响应订单({mall_tid[:16]}...)与捕获会话订单({token_id[:16]}...)不一致") print(" goods web_token 绑定捕获页面订单,以捕获 body 订单为准继续") except Exception as e: # noqa: BLE001 print(f"⚠️ mall 响应读取失败(忽略): {e!r}") print(f"[pay] 订单 token_id={token_id[:20]}... out_trade_no={out_trade_no}") # 2. 会话态(key16/key1/xmidas/args-template/web_token) keys = json.loads((gd / "keys.json").read_text(encoding="utf-8")) xmidas = json.loads((gd / "xmidasops.json").read_text(encoding="utf-8")) from pyvm.algorithm import decode_d, recover_plaintext_from_buffer at_file = gd / "args-template.json" if at_file.exists(): # 新版采集:decoded 18参直接读取(capture-goods-session.mjs J 转储) web_args = json.loads(at_file.read_text(encoding="utf-8")) else: # 旧版采集:deepCap D 编码(cap-85091.json) web_args = decode_d(_load_cap(gd / "caps" / "cap-85091.json")["C"][2]) wt = gd / "web-token.txt" web_token = wt.read_text(encoding="utf-8").strip() if wt.exists() else "" body_tpl = (gd / "body.txt").read_text(encoding="utf-8") if not web_token: m = re.search(r"web_token=([0-9A-F]+)", body_tpl) web_token = m.group(1) if m else "" print(f"[pay] goods 会话态: xmidasops={len(xmidas)} key16={'有' if keys.get('key16') else '无'} " f"web_token={web_token[:12]}...") # 3. 恢复页面真实明文(ts/fk_extend/_rand),刷新 ts 为当前一致值 try: rec_plain = recover_plaintext_from_buffer(web_args[10][0], keys["key16"]) fields = dict(kv.split("=", 1) for kv in rec_plain.split("&")) fk_extend = fields.get("fk_extend", "") rand_val = fields.get("_rand", "") rec_ts = fields.get("ts", "") except Exception as e: # noqa: BLE001 print(f"⚠️ 缓冲明文恢复失败,回退 plaintext.txt: {e!r}") fields = parse_plaintext(gd / "plaintext.txt") fk_extend = fields.get("fk_extend", "") rand_val = fields.get("_rand", "") rec_ts = "" params = {k: body_fields.get(k, fields.get(k, "")) for k in ORDER_FIELDS} params["token_id"] = token_id now_ms = int(time.time() * 1000) params["ts"] = str(now_ms // 1000) if rec_ts: print(f"[pay] 页面真实明文: ts={rec_ts}(捕获) -> 使用当前 ts={params['ts']} " f"(E3:ts 与 body t 一致即可,不必等于页面值)") print("[pay] 生成 goods encrypt_msg(纯 Python)...") hex_msg = generate_encrypt_msg_offline( params, fk_extend, params["ts"], rand_val, xmidas=xmidas, args_template=web_args, key16=keys.get("key16"), key1=keys.get("key1"), ) print(f"[pay] encrypt_msg ({len(hex_msg)} hex): {hex_msg[:32]}...") # 4. 构造 web_save body(会话订单 token + 当前动态值 + 纯 py encrypt_msg) body = body_tpl for k, v in [("token_id", token_id), ("transaction_id", transaction_id), ("out_trade_no", out_trade_no), ("offer_type", offer_type)]: body = re.sub(rf"{k}=[^&]*", f"{k}=" + v, body, count=1) body = re.sub(r"pc_st=[^&]+", "pc_st=" + str(uuid.uuid4()).upper() + str(int(time.time() * 1000)), body) body = re.sub(r"r=[0-9.]+", "r=" + str(random.random()), body) body = re.sub(r"&t=[0-9]+", "&t=" + str(int(time.time() * 1000)), body) body = re.sub(r"encrypt_msg=[0-9a-f]+", "encrypt_msg=" + hex_msg, body) if web_token: body = re.sub(r"web_token=[0-9A-F]+", "web_token=" + web_token, body) # 5. cookies: mall 登录态(midas 域 cookie 兜底) sess = json.loads(Path(args.session).read_text(encoding="utf-8")) cookies = dict(sess.get("cookies", {})) if not cookies: print("⚠️ 会话态缺少 cookies(登录态)——mall-session.json 需在登录后保存") return 2 cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items()) if "midas_openid" not in cookies and "openid" in cookies: cookie_str += "; midas_openid=" + cookies["openid"] if "midas_openkey" not in cookies and "accesstoken" in cookies: cookie_str += "; midas_openkey=" + cookies["accesstoken"] # 6. POST web_save url = f"https://api.unipay.qq.com/v1/r/{args.appid}/web_save" req = urllib.request.Request( url, data=body.encode("utf-8"), headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36", "Content-Type": "application/x-www-form-urlencoded", "Origin": "https://pay.qq.com", "Referer": "https://pay.qq.com/", "Cookie": cookie_str, }, method="POST", ) print(f"[pay] POST {url}") try: with urllib.request.urlopen(req, timeout=30) as r: raw = r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") except Exception as e: # noqa: BLE001 print(f"[pay] 网络错误: {e!r}") return 3 print(f"[pay] 响应: {raw[:500]}") # 7. 解析支付通道 try: js = json.loads(raw) except Exception: # noqa: BLE001 js = {} ret = js.get("ret") if ret != 0: print(f"❌ web_save ret:{ret} ({js.get('err_code', '')}) {js.get('msg', '')}") return 1 info = js.get("info", {}) ci = info.get("channel_info", {}) sign = ci.get("sign", "") if not sign: print("❌ 未返回支付 sign,请检查响应") return 1 print(f"✅ 支付通道建立: serialno={ci.get('serialno', '')}") print(f" 微信支付链接: {sign}") # 8. 渲染二维码(segno) try: import segno except ImportError: print("⚠️ 未安装 segno,跳过二维码渲染(pip install segno)") return 0 out_png = Path(args.output) if args.output else ROOT / "config" / "pay-qr.png" out_png.parent.mkdir(parents=True, exist_ok=True) qr = segno.make(sign) qr.save(str(out_png), scale=6, border=2) print(f" 二维码 PNG → {out_png}") print("\n ══ 微信扫码支付(终端二维码)══") try: qr.terminal(compact=False) except Exception: # noqa: BLE001 pass return 0 if __name__ == "__main__": sys.exit(main())