diff --git a/services/yyb-worker/runtime/main.py b/services/yyb-worker/runtime/main.py index 637f1f7..ec911aa 100644 --- a/services/yyb-worker/runtime/main.py +++ b/services/yyb-worker/runtime/main.py @@ -36,10 +36,12 @@ from pyvm.algorithm import build_plaintext, generate_encrypt_msg, generate_encry 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 REPLAY = ROOT / "replay" DEFAULT_APPID = "1450243039" DEFAULT_SAVE_URL = f"https://api.unipay.qq.com/v1/r/{DEFAULT_APPID}/web_save" +MALL_API_URL = "https://storeapi.pay.qq.com/api/CommonCallMpgo" 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", @@ -367,7 +369,7 @@ def cmd_mall_submit(args) -> int: 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)) + 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: @@ -552,7 +554,7 @@ def mall_getpaytoken(cookies: dict) -> tuple[list, str]: "login_check_param_json": json.dumps(login), }, } - url = "https://pagedooapi.pay.qq.com/api/CommonCallMpgo?t=" + str(int(time.time() * 1000)) + 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: @@ -690,7 +692,7 @@ def cmd_mall_auto(args) -> int: 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)) + 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/", @@ -722,6 +724,7 @@ def cmd_mall_auto(args) -> int: 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 diff --git a/services/yyb-worker/runtime/probe-jsdom-pay.mjs b/services/yyb-worker/runtime/probe-jsdom-pay.mjs new file mode 100644 index 0000000..c4d0deb --- /dev/null +++ b/services/yyb-worker/runtime/probe-jsdom-pay.mjs @@ -0,0 +1,46 @@ +import fs from 'node:fs'; +import { JSDOM, ResourceLoader } from 'jsdom'; +import { patchEnvironment } from './scripts/jsdom-patch-env.mjs'; + +const ROOT = '/Users/yml/codes/douyu_login_py'; +const out = ROOT + '/data/yyb-worker-dev/49684f85543d4fd1/jsdom-order'; +const goodsUrl = JSON.parse(fs.readFileSync(`${out}/device-fp.json`, 'utf8')).goods_url; + +const captured = []; +const requests = []; +const dom = new JSDOM(fs.readFileSync(`${out}/goods.html`, 'utf8'), { + url: goodsUrl, + referrer: 'https://z.iwan.yyb.qq.com/', + pretendToBeVisual: true, + runScripts: 'dangerously', + resources: new ResourceLoader(), + beforeParse(window) { + patchEnvironment(window); + const NativeXHR = window.XMLHttpRequest; + window.XMLHttpRequest = class extends NativeXHR { + open(method, url, async = true) { this.__m = method; this.__u = String(url); return super.open(method, url, async); } + send(body) { + const name = this.__u.includes('web_save') ? 'web_save' + : this.__u.includes('web_page_info') ? 'web_page_info' + : this.__u.includes('fp-behv') ? 'fp_behv' : 'other'; + requests.push({ name, url: this.__u.slice(0, 120), bodyLen: String(body || '').length }); + if (name === 'web_save') captured.push(String(body || '')); + this.readyState = 4; this.status = 200; + this.responseText = '{"ret":0,"msg":"","info":{}}'; + if (typeof this.onreadystatechange === 'function') this.onreadystatechange(); + if (typeof this.onload === 'function') this.onload(); + } + }; + }, +}); + +await new Promise(r => setTimeout(r, 15000)); +console.log('=== 页面发出的 XHR ==='); +for (const r of requests) console.log(' ', r.name, '|', r.url, '| bodyLen:', r.bodyLen); +console.log('=== web_save body 捕获:', captured.length ? 'YES len=' + captured[0].length : 'NO'); +console.log('=== 页面按钮 ==='); +const btns = dom.window.document.querySelectorAll('[class*="btn"],[class*="confirm"],[class*="submit"],button'); +for (const b of btns) { + const t = (b.textContent || '').trim().replace(/\s+/g, ' '); + if (t && t.length < 40 && b.offsetParent !== null) console.log(' btn:', t.slice(0, 30)); +} diff --git a/services/yyb-worker/runtime/pyvm/algorithm.py b/services/yyb-worker/runtime/pyvm/algorithm.py index bbc6e9c..09aa5f7 100644 --- a/services/yyb-worker/runtime/pyvm/algorithm.py +++ b/services/yyb-worker/runtime/pyvm/algorithm.py @@ -1,10 +1,12 @@ """encrypt_msg (web_new_encrypt) 生成器 — goodsBiz CHAOS VM 的纯 Python 移植 -算法背景(见 FINDINGS.md F-2038..F-2043): +算法背景(见 FINDINGS.md F-2038..F-2043,破解见 F-2052/2026-08-12): - goodsBiz.js 是腾讯 __TENCENT_CHAOS_VM 解释器(116 opcode,0..115)+ 178093 元素字节码。 - 加密核心 = webSave h@85091(18 参):表生成 → 检查 window.xMidasOps → 33 次块变换 (T 表 AES 类)→ 回调 h@69667(4 参)→ randGen h@85173(16 次重随机 key2)。 - - 输入:18 参(key1 诱饵 / key2 真实 / args[10]=528B 明文缓冲 / 常量表 / 回调)。 + - 输入:18 参(key1 诱饵 / key2 真实 / args[10]=16 字节对齐的明文缓冲 / 常量表 / 回调)。 + - key 派生已破解:key16 = Sbox[Te 链(key1)];服务端校验两者派生一致性, + 离线用随机 key16 + derive_key1_from_key16 反解 key1 即可通过(无需页面捕获 key)。 - 输出:C[3476] 528 字节密文(hex 1056 字符,头部 5574bea9... 固定前缀)。 本文件复刻解释器语义,可在无浏览器环境运行;两组独立真实向量均 1056/1056 逐字节一致。 @@ -660,6 +662,14 @@ class VM: except Exception: parts.append("??") self.out_hex = "".join(parts) + # 仅供离线排查 webSave 输出缓冲何时变化，生产路径默认不启用。 + if getattr(self, "capture_output_lengths", False): + v = js_index(C, 3476) + if isinstance(v, list): + length = len(v) + if length != getattr(self, "_last_output_length", None): + self._last_output_length = length + self.output_length_trace.append((u, op, length)) # ---- opcode dispatch ---- if op == 0: a, b, c = o[u+1], o[u+2], o[u+3]; u += 3 @@ -1461,15 +1471,18 @@ def derive_key1_from_key16(key16, te_tables=None, sbox=None): def generate_encrypt_msg_offline(params, fk_extend, ts, rand_val, key16=None, key1=None, args_template=None, cb_json=None, - random_seed=None, xmidas=None, xmidas_token=None): + random_seed=None, xmidas=None, xmidas_token=None, + diagnostics=None): """纯离线生成 encrypt_msg(A2 路线,2026-08-10 双向量 E3 闭环)。 params: 订单字段字典(18 字段,见 build_plaintext);fk_extend: tdrc_session 串 (最终编码形态 tdrc_session%3Dpay-...,不再二次编码);ts: 秒级时间戳; rand_val: _rand 值(页面实测 8 字母数字 + 3×\\x03(run A)或 7 字母数字 + 4×\\x04(run B),调用方自定即可); - key16/key1: 16 字节密钥,**不影响最终输出**(E3:双向量随机密钥 1056/1056 - 复现;webSave 回调内部重派生密钥,机制见 U-2030); + key16/key1: 16 字节密钥,**不影响最终输出密文字节**(E3:双向量随机密钥 1056/1056 + 复现;webSave 回调内部重派生密钥),但服务端会校验 key16 与 key1 的派生一致性 + (key16=Sbox[Te 链(key1)],F-2052),key1 必须由 derive_key1_from_key16 从 key16 + 反解,不能随意给定; args_template: 常量表模板(跨次恒定,缺省用 work/replay/e2e/ws-args.json)。 """ import random as _r @@ -1523,6 +1536,13 @@ def generate_encrypt_msg_offline(params, fk_extend, ts, rand_val, # 参数全部以 a:1 包装传入;key16 按全局块索引轮转:out[j] = word_byte[j] ^ key16[(4*blk+j)%16] pt = build_plaintext(params, fk_extend, ts, rand_val) pb = pt.encode("latin-1") + # goods VM 只逐个处理完整的 16 字节块，不会为尾块自动填充。 + # 未对齐时继续执行会静默丢弃尾部字段，必须在请求前终止。 + if len(pb) % 16: + raise ValueError( + f"webSave 明文必须为 16 字节对齐，当前 {len(pb)} 字节；" + "请用 _rand 补齐后再生成 encrypt_msg" + ) def _cont(*a): return None te_cap = json.loads(json.loads((replay / "e2e" / "multi-1.json").read_text().strip())) @@ -1562,7 +1582,18 @@ def generate_encrypt_msg_offline(params, fk_extend, ts, rand_val, vm.init_c = init_c vm.capture_at_return = True vm.out_hex = None + if diagnostics is not None: + vm.capture_output_lengths = True + vm.output_length_trace = [] + vm._last_output_length = None vm.run(web, []) if vm.out_hex is None: raise RuntimeError("webSave capture failed") + if diagnostics is not None: + diagnostics.update({ + "plaintext_length": len(pb), + "transform_output_length": len(outbuf), + "ciphertext_length": len(vm.out_arr), + "output_length_trace": vm.output_length_trace, + }) return vm.out_hex diff --git a/services/yyb-worker/runtime/pyvm/payment_errors.py b/services/yyb-worker/runtime/pyvm/payment_errors.py new file mode 100644 index 0000000..c7ef357 --- /dev/null +++ b/services/yyb-worker/runtime/pyvm/payment_errors.py @@ -0,0 +1,31 @@ +"""YYB 下单和付款失败响应的统一归类。""" +from __future__ import annotations + + +def describe_payment_failure(provider: str, phase: str, response: object) -> str: + """根据已保存的响应内容返回不会诱导连续重试的失败说明。""" + text = str(response) + is_qq = provider.lower() in {"qq", "qc"} + + if "1099-1007-0" in text or "ret=1099" in text: + if is_qq: + return ( + "QQ 支付请求被风控/限流暂时拦截（1099-1007-0），未生成微信付款码；" + "请停止连续重试，等待 30-60 分钟后再使用新订单尝试。" + ) + return "支付请求被暂时拦截（1099-1007-0），未生成微信付款码；请稍后使用新订单重试。" + + if "10007" in text: + if is_qq: + return ( + "QQ 商城下单被支付侧暂时拒绝（10007 系统繁忙），未创建可付款订单；" + "请停止连续重试，等待 30-60 分钟后再尝试。" + ) + return "商城下单被暂时拒绝（10007 系统繁忙），未创建可付款订单；请稍后重试。" + + if "10014" in text or "gamerole err" in text: + return "角色或区服校验失败（10014），未创建可付款订单；请重新选择角色和区服。" + + if phase == "order": + return "商城订单创建失败，未创建可付款订单；请查看任务日志中的业务码。" + return "微信付款码生成失败；请查看任务日志中的业务码。" diff --git a/services/yyb-worker/runtime/scripts/jsdom-pay.py b/services/yyb-worker/runtime/scripts/jsdom-pay.py index 53d2aa0..f682477 100644 --- a/services/yyb-worker/runtime/scripts/jsdom-pay.py +++ b/services/yyb-worker/runtime/scripts/jsdom-pay.py @@ -26,12 +26,14 @@ from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) -from pyvm.algorithm import derive_key1_from_key16, generate_encrypt_msg_offline # noqa: E402 +from pyvm.algorithm import build_plaintext, derive_key1_from_key16, generate_encrypt_msg_offline # noqa: E402 from pyvm.login_profile import midas_login_params # noqa: E402 +from pyvm.payment_errors import describe_payment_failure # noqa: E402 APPID = "1450243039" GOODS_URL = "https://pay.qq.com/midas/minipay_v2/views/cpay/goods.shtml" SAVE_URL = f"https://api.unipay.qq.com/v1/r/{APPID}/web_save" +PAGE_INFO_URL = f"https://api.unipay.qq.com/v1/r/{APPID}/web_page_info" FP_URL = "https://api.unipay.qq.com/cgi-bin/fp-behv.fcg" def load_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) @@ -131,7 +133,7 @@ def build_save_fields(order: dict[str, str], cookies: dict[str, str], web_token: if amount_fen <= 0: raise ValueError("充值金额必须为正数") login = midas_login_params(cookies) - return { + fields = { "token_id": order.get("token_id", ""), "transaction_id": order.get("transaction_id", ""), "out_trade_no": order.get("out_trade_no", ""), @@ -163,7 +165,6 @@ def build_save_fields(order: dict[str, str], cookies: dict[str, str], web_token: "pushtype": "NodeJS", "wx_order_interface": "1", "encrypt_msg": encrypt_msg, - "qq_appid": login["qq_appid"], "base_key_version": "web_1.0.6", "encrypt_way": "web_new_encrypt", "web_token": web_token, @@ -172,6 +173,10 @@ def build_save_fields(order: dict[str, str], cookies: dict[str, str], web_token: "t": str(int(time.time() * 1000)), "__refer": "https://z.iwan.yyb.qq.com/", } + # QQ 网页端会携带该字段；微信 OAuth 不携带，不能使用固定 appid 混入两条链路。 + if login["qq_appid"]: + fields["offerid_for_qq_appid"] = login["qq_appid"] + return fields def build_save_body(order: dict[str, str], cookies: dict[str, str], web_token: str, @@ -183,6 +188,71 @@ def build_save_body(order: dict[str, str], cookies: dict[str, str], web_token: s )) +def build_page_info_body(order: dict[str, str], cookies: dict[str, str], anti_token: str, + zone_id: str, pf: str) -> str: + """构造网页端 fp-behv 后的 web_page_info 表单。""" + login = midas_login_params(cookies) + fields = { + "token_id": order.get("token_id", ""), + "transaction_id": order.get("transaction_id", ""), + "out_trade_no": order.get("out_trade_no", ""), + "offer_type": order.get("offer_type", "0"), + "type": "bg", + "pf": pf, + "pfkey": "pfkey", + "from_h5": "1", + "pc_st": str(uuid.uuid4()).upper() + str(int(time.time() * 1000)), + "r": str(random.random()), + "openid": cookies.get("openid", ""), + "openkey": cookies.get("accesstoken", ""), + "session_id": login["session_id"], + "session_type": login["session_type"], + "sck": hashlib.md5((APPID + cookies.get("accesstoken", "")).encode()).hexdigest().upper(), + "anti_auto_script_token_id": anti_token, + "isusempaymode": "1", + "zoneid": zone_id, + "webversion": "minipayv2", + "from_https": "1", + "t": str(int(time.time() * 1000)), + "__refer": "https://z.iwan.yyb.qq.com/", + } + if login["qq_appid"]: + fields["offerid_for_qq_appid"] = login["qq_appid"] + return urllib.parse.urlencode(fields) + + +def make_encrypt_rand(params: dict[str, str], fk_extend: str, ts: str) -> str: + """生成与当前订单明文长度匹配的 _rand，确保 goods VM 不截断尾块。""" + prefix = "".join(random.choices(string.ascii_letters + string.digits, k=8)) + base_length = len(build_plaintext(params, fk_extend, ts, "").encode("latin-1")) + padding_length = (-(base_length + len(prefix))) % 16 + return prefix + ("\x03" * padding_length) + + +def save_web_save_request_meta(out_dir: Path, fields: dict[str, str], device_fp_length: int, + plaintext_length: int) -> None: + """保存可比对的脱敏请求形态，避免在证据目录落盘支付凭据。""" + sensitive = ("token_id", "transaction_id", "out_trade_no", "openid", "openkey", + "sck", "anti_auto_script_token_id", "web_token", "encrypt_msg") + fingerprints = { + name: { + "length": len(fields.get(name, "")), + "sha256_prefix": hashlib.sha256(fields.get(name, "").encode()).hexdigest()[:12], + } + for name in sensitive + } + meta = { + "field_names": sorted(fields), + "plaintext_length": plaintext_length, + "encrypt_msg_length": len(fields.get("encrypt_msg", "")), + "device_fp_length": device_fp_length, + "sensitive_field_fingerprints": fingerprints, + } + (out_dir / "web-save-request-meta.json").write_text( + json.dumps(meta, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", + ) + + def make_qr(sign: str, output: Path) -> None: try: import segno @@ -323,6 +393,7 @@ def main() -> int: return cmd_check_only(session_path, out_dir) if args.amount_fen <= 0: raise ValueError("充值金额必须为正数") + payment_pf = args.pf or "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android" session = load_json(session_path) cookies = dict(session.get("cookies", {})) @@ -366,13 +437,36 @@ def main() -> int: print("[jsdom-pay] 上报 fp-behv...") fp_response = request_bytes(fp.get("fp_url", FP_URL), cookies, fp["fp_body"].encode()) (out_dir / "fp-response.json").write_text(fp_response, encoding="utf-8") + try: + fp_json = json.loads(fp_response) + except json.JSONDecodeError: + fp_json = {} + if fp_json.get("ret") != 0: + print(f"[jsdom-pay] fp-behv 失败: ret={fp_json.get('ret')} {fp_json.get('msg', '')}") + return 1 + + print("[jsdom-pay] 拉取 web_page_info...") + page_info = request_bytes(PAGE_INFO_URL, cookies, build_page_info_body( + order, cookies, anti_token, args.zone_id, payment_pf, + ).encode()) + (out_dir / "web-page-info-response.json").write_text(page_info, encoding="utf-8") + try: + page_info_json = json.loads(page_info) + except json.JSONDecodeError: + page_info_json = {} + if page_info_json.get("ret") != 0: + print(f"[jsdom-pay] web_page_info 失败: ret={page_info_json.get('ret')} " + f"{page_info_json.get('msg', '')}") + print(f"[jsdom-pay] 证据目录: {out_dir}") + return 1 web_args = load_template_args() - payment_pf = args.pf or "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android" + # key 派生已破解(F-2052,2026-08-12):key16 可随机生成,key1 由反解器 + # derive_key1_from_key16 求解(key16=Sbox[Te 链(key1)]),满足服务端派生校验, + # 无需浏览器捕获 key(微信渠道纯 HTTP 已 ret:0)。 key16 = [random.randrange(256) for _ in range(16)] tables = [web_args[index][0] for index in (1, 2, 3, 4)] key1 = derive_key1_from_key16(key16, te_tables=tables, sbox=web_args[5][0]) - random_suffix = "".join(random.choices(string.ascii_letters + string.digits, k=8)) + "\x01" params = { "token_id": order.get("token_id", ""), "openid": cookies.get("openid", ""), @@ -394,14 +488,19 @@ def main() -> int: } params.update(midas_login_params(cookies)) now_seconds = str(int(time.time())) + fk_extend = "tdrc_session%3D" + fp["session_id"] + random_suffix = make_encrypt_rand(params, fk_extend, now_seconds) + plaintext_length = len(build_plaintext(params, fk_extend, now_seconds, random_suffix).encode("latin-1")) encrypt_msg = generate_encrypt_msg_offline( - params, "tdrc_session%3D" + fp["session_id"], now_seconds, random_suffix, + params, fk_extend, now_seconds, random_suffix, key16=key16, key1=key1, args_template=web_args, xmidas=xmidas, xmidas_token=web_token, ) - body = build_save_body( + fields = build_save_fields( order, cookies, web_token, anti_token, encrypt_msg, args.zone_id, payment_pf, args.amount_fen, ) + save_web_save_request_meta(out_dir, fields, int(fp.get("device_fp_length", 0)), plaintext_length) + body = urllib.parse.urlencode(fields) from pyvm.order_status import order_completion_states, get_official_orders # 尽量在发起支付前建立基线，缩小极快支付造成的检测窗口；失败不阻塞支付，付款码生成后重试。 @@ -421,7 +520,10 @@ def main() -> int: except json.JSONDecodeError: response_json = {} if response_json.get("ret") != 0: - print(f"[jsdom-pay] web_save 失败: ret={response_json.get('ret')} {response_json.get('msg', '')}") + detail = f"ret={response_json.get('ret')} err_code={response_json.get('err_code', '')} " \ + f"{response_json.get('msg', '')}" + print(f"[jsdom-pay] web_save 失败: {detail}") + print(f"[jsdom-pay] {describe_payment_failure(cookies.get('logintype', ''), 'payment', detail)}") print(f"[jsdom-pay] 证据目录: {out_dir}") return 1 sign = response_json.get("info", {}).get("channel_info", {}).get("sign", "") diff --git a/services/yyb-worker/runtime/scripts/yyb-worker.py b/services/yyb-worker/runtime/scripts/yyb-worker.py index b003a44..bd2780e 100644 --- a/services/yyb-worker/runtime/scripts/yyb-worker.py +++ b/services/yyb-worker/runtime/scripts/yyb-worker.py @@ -23,11 +23,21 @@ from pathlib import Path from urllib.parse import urlparse ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) DEFAULT_DATA = ROOT / "config" / "worker-jobs" WORKER_KEY = os.environ.get("YYB_WORKER_KEY", "") _jobs: dict[str, dict] = {} _lock = threading.Lock() +from pyvm.payment_errors import describe_payment_failure # noqa: E402 + + +def _payment_failure_message(job: dict, phase: str, fallback: str) -> str: + """从阶段日志提取业务码，向管理端返回准确且可操作的失败信息。""" + detail = "\n".join(job.get("logs", [])) + message = describe_payment_failure(str(job.get("provider", "")), phase, detail) + return message if message else fallback + def _load_selector(): path = ROOT / "scripts" / "select-peace-elite.py" @@ -161,7 +171,7 @@ def _payment_stage(job_id: str, command: list[str], phase: str, running_message: return False job["status"] = "failed" job["phase"] = phase - job["message"] = f"{failed_message}（退出码 {code}）" + job["message"] = _payment_failure_message(job, phase, failed_message) return False return True diff --git a/services/yyb-worker/runtime/tests/test_jsdom_pay.py b/services/yyb-worker/runtime/tests/test_jsdom_pay.py index 435a102..9193b6b 100644 --- a/services/yyb-worker/runtime/tests/test_jsdom_pay.py +++ b/services/yyb-worker/runtime/tests/test_jsdom_pay.py @@ -15,6 +15,8 @@ _MODULE = importlib.util.module_from_spec(_SPEC) sys.modules[_SPEC.name] = _MODULE _SPEC.loader.exec_module(_MODULE) +from pyvm.payment_errors import describe_payment_failure + class TestJsdomPay(unittest.TestCase): def test_build_save_body_has_current_payment_context(self): @@ -42,8 +44,51 @@ class TestJsdomPay(unittest.TestCase): fields = parse_qs(body, keep_blank_values=True) self.assertEqual(fields["session_id"], ["openid"]) self.assertEqual(fields["session_type"], ["kp_accesstoken"]) - self.assertEqual(fields["qq_appid"], ["102033112"]) self.assertEqual(fields["wx_appid"], [""]) + self.assertEqual(fields["offerid_for_qq_appid"], ["102033112"]) + self.assertNotIn("qq_appid", fields) + + def test_wechat_payment_does_not_include_qq_offer_field(self): + body = _MODULE.build_save_body( + {"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"}, + {"openid": "openid", "accesstoken": "access-token", "logintype": "WX"}, + "web-token", "anti-token", "cipher", "1", "pf", 100, + ) + fields = parse_qs(body, keep_blank_values=True) + self.assertNotIn("offerid_for_qq_appid", fields) + + def test_qq_page_info_matches_the_web_flow(self): + body = _MODULE.build_page_info_body( + {"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"}, + {"openid": "openid", "accesstoken": "access-token", "logintype": "QC", "appid": "102033112"}, + "anti-token", "1", "pf", + ) + fields = parse_qs(body, keep_blank_values=True) + self.assertEqual(fields["isusempaymode"], ["1"]) + self.assertEqual(fields["offerid_for_qq_appid"], ["102033112"]) + self.assertNotIn("pay_method", fields) + + def test_qq_risk_rejection_is_not_reported_as_qr_error(self): + message = describe_payment_failure("QC", "payment", "ret=1099 err_code=1099-1007-0") + self.assertIn("风控/限流", message) + self.assertIn("未生成微信付款码", message) + self.assertIn("停止连续重试", message) + + def test_game_role_error_is_classified_separately(self): + message = describe_payment_failure("QC", "order", "inner=10014 gamerole err") + self.assertIn("角色或区服校验失败", message) + self.assertIn("重新选择角色和区服", message) + + def test_encrypt_rand_aligns_plaintext_to_aes_block(self): + params = { + "token_id": "t", "openid": "o", "openkey": "k", "session_id": "openid", + "session_type": "kp_accesstoken", "zoneid": "1", "pay_method": "wechat", + "buy_quantity": "1", "from_h5": "1", "webversion": "minipayv2", + } + rand_value = _MODULE.make_encrypt_rand(params, "tdrc_session%3Dpay-test", "1775990000") + plaintext = _MODULE.build_plaintext(params, "tdrc_session%3Dpay-test", "1775990000", rand_value) + self.assertEqual(len(plaintext.encode("latin-1")) % 16, 0) + self.assertGreaterEqual(len(rand_value), 8) if __name__ == "__main__":