qq 也可以获取支付二维码成功微信支付了

2026-08-12 21:41:26 +08:00
parent 4c978b923b
commit d610933f87
7 changed files with 286 additions and 18 deletions
@@ -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", "")
@@ -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