860 lines
39 KiB
Python
860 lines
39 KiB
Python
#!/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 os
|
||
import re
|
||
import secrets
|
||
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_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
|
||
from pyvm.payment_errors import describe_payment_failure # noqa: E402
|
||
from pyvm.protocol import ( # noqa: E402
|
||
MALL_API_URL,
|
||
MALL_USER_AGENT,
|
||
GOODS_USER_AGENT,
|
||
PAY_APPID,
|
||
validate_goods_materials,
|
||
validate_mall_materials,
|
||
)
|
||
|
||
REPLAY = ROOT / "replay"
|
||
DEFAULT_APPID = PAY_APPID
|
||
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 _write_private_text(path: Path, content: str) -> None:
|
||
"""写入任务协议证据并限制为当前用户可读。"""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
os.chmod(path.parent, 0o700)
|
||
path.write_text(content, encoding="utf-8")
|
||
os.chmod(path, 0o600)
|
||
|
||
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 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:
|
||
"""校验当前受版本控制的 goods 协议材料，不依赖已删除的历史抓包。"""
|
||
args_path = REPLAY / "live" / "default" / "args-template.json"
|
||
args_template = json.loads(args_path.read_text(encoding="utf-8"))
|
||
validate_goods_materials(args_template)
|
||
required = ("bytecode.json", "constants.json", "xmidasops.json", "e2e/multi-1.json",
|
||
"e2e/ws-args.json", "deepcaps2/cap-69667.json")
|
||
missing = [name for name in required if not (REPLAY / name).exists()]
|
||
if missing:
|
||
print(f"❌ goods 协议材料缺失: {', '.join(missing)}")
|
||
return 1
|
||
print("✅ goods 协议材料校验通过（18 槽模板、Te/S-box、VM 文件均完整）")
|
||
print(" 历史逐字节黄金向量未随仓库保留；如需恢复，可放入 replay/golden 后由 CI 自动执行。")
|
||
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(secrets.SystemRandom().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": GOODS_USER_AGENT,
|
||
"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:
|
||
"""校验当前受版本控制的 mall VM 和固定槽模板。"""
|
||
fixed_path = REPLAY / "mall" / "transform-fixed.json"
|
||
bytecode_path = REPLAY / "mall" / "vm" / "bytecode-478657.json"
|
||
fixed = json.loads(fixed_path.read_text(encoding="utf-8"))
|
||
validate_mall_materials(fixed)
|
||
bytecode = json.loads(bytecode_path.read_text(encoding="utf-8"))
|
||
if not isinstance(bytecode, list) or not bytecode:
|
||
print("❌ mall VM 字节码为空或格式异常")
|
||
return 1
|
||
print("✅ mall 协议材料校验通过（固定槽模板与 VM 字节码完整）")
|
||
print(" 历史逐字节黄金向量未随仓库保留；不再引用不存在的 config/golden 文件。")
|
||
return 0
|
||
|
||
|
||
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 = MALL_API_URL + "?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": MALL_USER_AGENT,
|
||
},
|
||
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:
|
||
_write_private_text(out, raw)
|
||
print(f"✅ 提交完成（ret={ret}），响应已保存 → {out}")
|
||
return 0
|
||
# 失败:不覆盖上次成功响应,另存 .fail.json
|
||
fail_out = out.with_suffix(".fail.json")
|
||
_write_private_text(fail_out, raw)
|
||
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="校验当前 goods 协议材料")
|
||
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="校验当前 mall 协议材料")
|
||
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)。
|
||
"""
|
||
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] = [[secrets.randbelow(256) for _ in range(16)]]
|
||
ti[6] = [[secrets.randbelow(256) for _ in range(16)]]
|
||
ti[10] = [[secrets.randbelow(256) 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 = MALL_API_URL + "?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": MALL_USER_AGENT,
|
||
}, 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"))
|
||
validate_mall_materials(fixed)
|
||
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 = MALL_API_URL + "?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": MALL_USER_AGENT,
|
||
}, 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"
|
||
_write_private_text(out, raw)
|
||
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" {describe_payment_failure(str(cookies.get('logintype', '')), 'order', f'{ret} {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(secrets.SystemRandom().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": MALL_USER_AGENT,
|
||
"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())
|