diff --git a/services/yyb-worker/runtime/main.py b/services/yyb-worker/runtime/main.py index ec911aa..2e6d540 100644 --- a/services/yyb-worker/runtime/main.py +++ b/services/yyb-worker/runtime/main.py @@ -21,8 +21,9 @@ from __future__ import annotations import argparse import json -import random +import os import re +import secrets import sys import time import urllib.request @@ -32,16 +33,23 @@ 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.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 = "1450243039" +DEFAULT_APPID = PAY_APPID 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", @@ -50,6 +58,13 @@ ORDER_FIELDS = ["token_id", "openid", "openkey", "session_id", "session_type", " # ---------------------------------------------------------------- 工具 +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 键)。 @@ -73,14 +88,6 @@ def _load_cap(path: Path) -> dict: 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(): @@ -107,79 +114,18 @@ def parse_plaintext(path: str | Path) -> dict: # ---------------------------------------------------------------- 命令 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} 项失败") + """校验当前受版本控制的 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("✅ 全部通过(E3 双向量 + 离线 + live 复现 ×5)") + print("✅ goods 协议材料校验通过（18 槽模板、Te/S-box、VM 文件均完整）") + print(" 历史逐字节黄金向量未随仓库保留；如需恢复，可放入 replay/golden 后由 CI 自动执行。") return 0 @@ -221,7 +167,7 @@ def _build_body(order: dict, st: SessionState, offline_hex: str, body_tpl: Path) # 动态值 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"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 @@ -242,8 +188,7 @@ def cmd_submit(args) -> int: 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", + "User-Agent": GOODS_USER_AGENT, "Content-Type": "application/x-www-form-urlencoded", "Origin": "https://pay.qq.com", "Referer": "https://pay.qq.com/", @@ -308,15 +253,18 @@ def cmd_sample_order(args) -> int: # ---------------------------------------------------------------- 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 + """校验当前受版本控制的 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: @@ -384,8 +332,7 @@ def cmd_mall_submit(args) -> int: "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", + "User-Agent": MALL_USER_AGENT, }, method="POST", ) @@ -410,12 +357,12 @@ def cmd_mall_submit(args) -> int: pass ok = ret in (0, "0") if ok: - out.write_text(raw, encoding="utf-8") + _write_private_text(out, raw) print(f"✅ 提交完成（ret={ret}），响应已保存 → {out}") return 0 # 失败:不覆盖上次成功响应,另存 .fail.json fail_out = out.with_suffix(".fail.json") - fail_out.write_text(raw, encoding="utf-8") + _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 "") @@ -442,7 +389,7 @@ def main() -> int: # ---- 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 复现)") + 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) @@ -461,7 +408,7 @@ def main() -> int: # ---- 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)") + 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) @@ -524,13 +471,12 @@ def build_mall_transform(fixed: dict) -> list: 或风控严格后拒绝随机加密(如 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)]] + 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 @@ -563,8 +509,7 @@ def mall_getpaytoken(cookies: dict) -> tuple[list, str]: "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", + "User-Agent": MALL_USER_AGENT, }, method="POST") try: with urllib.request.urlopen(req, timeout=30) as r: @@ -673,6 +618,7 @@ def cmd_mall_auto(args) -> int: 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) @@ -697,8 +643,7 @@ def cmd_mall_auto(args) -> int: "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", + "User-Agent": MALL_USER_AGENT, }, method="POST") try: with urllib.request.urlopen(req, timeout=30) as r: @@ -706,7 +651,7 @@ def cmd_mall_auto(args) -> int: 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") + _write_private_text(out, raw) js = json.loads(raw) ret = js.get("result_code") try: @@ -832,7 +777,7 @@ def cmd_mall_pay(args) -> int: ("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"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: @@ -855,8 +800,7 @@ def cmd_mall_pay(args) -> int: 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", + "User-Agent": MALL_USER_AGENT, "Content-Type": "application/x-www-form-urlencoded", "Origin": "https://pay.qq.com", "Referer": "https://pay.qq.com/", diff --git a/services/yyb-worker/runtime/pyvm/protocol.py b/services/yyb-worker/runtime/pyvm/protocol.py new file mode 100644 index 0000000..4447f45 --- /dev/null +++ b/services/yyb-worker/runtime/pyvm/protocol.py @@ -0,0 +1,93 @@ +"""YYB 支付协议常量、材料校验与脱敏指纹。 + +这里仅收敛已由成功请求验证的固定值，不负责推断或修改协议字段。 +""" +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + + +PAY_APPID = "1450243039" +PAY_GOODS_URL = "https://pay.qq.com/midas/minipay_v2/views/cpay/goods.shtml" +PAY_SAVE_URL = f"https://api.unipay.qq.com/v1/r/{PAY_APPID}/web_save" +PAY_PAGE_INFO_URL = f"https://api.unipay.qq.com/v1/r/{PAY_APPID}/web_page_info" +PAY_FP_URL = "https://api.unipay.qq.com/cgi-bin/fp-behv.fcg" +MALL_API_URL = "https://storeapi.pay.qq.com/api/CommonCallMpgo" + +PAY_WEB_VERSION = "web_1.0.6" +PAY_WEBVERSION = "minipayv2" +DEFAULT_ORDER_PF = "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android" +OLD_ENCRYPT_OFFERIDS = ("1450007826", "1110165571") + +# 这两个 UA 分别来自已成功的 goods 和商城请求，不能因为版本不同而强行统一。 +GOODS_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" +) +MALL_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" +) + + +def _sha256_prefix_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest()[:12] + + +def _json_fingerprint(value: Any) -> str: + raw = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + return _sha256_prefix_bytes(raw) + + +def file_fingerprint(path: Path) -> str: + """返回协议文件的短哈希，用于日志比对，不暴露内容。""" + return _sha256_prefix_bytes(path.read_bytes()) + + +def validate_goods_materials(args_template: list, xmidas: list[int] | None = None) -> None: + """校验 goods webSave VM 所需静态表结构，发现升级时尽早失败。""" + if not isinstance(args_template, list) or len(args_template) != 18: + raise ValueError("goods args-template 必须是 18 槽数组") + for index in (0, 1, 2, 3, 4, 5, 6, 8, 9, 10, 11, 12, 13, 14, 15, 16): + if not isinstance(args_template[index], list) or not args_template[index]: + raise ValueError(f"goods args-template 槽 {index} 缺失") + expected_lengths = {0: 16, 1: 256, 2: 256, 3: 256, 4: 256, 5: 256, + 6: 16, 10: 528, 11: 1024, 12: 1024, 13: 256, + 14: 256, 15: 256, 16: 256} + for index, expected in expected_lengths.items(): + value = args_template[index][0] + if not isinstance(value, list) or len(value) != expected: + actual = len(value) if isinstance(value, list) else "非数组" + raise ValueError(f"goods args-template 槽 {index} 长度异常: {actual} != {expected}") + if xmidas is not None and len(xmidas) != 59640: + raise ValueError(f"goods xMidasOps 长度异常: {len(xmidas)} != 59640") + + +def validate_mall_materials(transform_fixed: dict[str, Any]) -> None: + """校验 mall 固定槽模板；动态槽仍由当前会话的 GetPayToken 填充。""" + if not isinstance(transform_fixed, dict): + raise ValueError("mall transform-fixed 必须是对象") + required = {str(index) for index in (1, 2, 3, 4, 5, 7, 8, 9, 11, 12, 13, 14, 15, 16, 17)} + missing = sorted(required - set(transform_fixed)) + if missing: + raise ValueError(f"mall transform-fixed 缺少槽: {', '.join(missing)}") + for index in required: + if not isinstance(transform_fixed[index], list): + raise ValueError(f"mall transform-fixed 槽 {index} 不是数组") + + +def goods_material_diagnostics(root: Path, args_template: list, xmidas: list[int]) -> dict[str, Any]: + """构造脱敏协议指纹，便于区分页面升级和服务端业务拒绝。""" + validate_goods_materials(args_template, xmidas) + replay = root / "replay" + return { + "goods_xmidas_ops_length": len(xmidas), + "args_template_slots": len(args_template), + "args_template_sha256_prefix": _json_fingerprint(args_template), + "bytecode_sha256_prefix": file_fingerprint(replay / "bytecode.json"), + "constants_sha256_prefix": file_fingerprint(replay / "constants.json"), + "web_version": PAY_WEB_VERSION, + } diff --git a/services/yyb-worker/runtime/scripts/generate-devicefp-jsdom.mjs b/services/yyb-worker/runtime/scripts/generate-devicefp-jsdom.mjs index 4c5fd4a..b8646e5 100644 --- a/services/yyb-worker/runtime/scripts/generate-devicefp-jsdom.mjs +++ b/services/yyb-worker/runtime/scripts/generate-devicefp-jsdom.mjs @@ -141,6 +141,8 @@ const result = { }; fs.mkdirSync(path.dirname(outputPath), { recursive: true }); fs.writeFileSync(outputPath, JSON.stringify(result, null, 2) + '\n'); +fs.chmodSync(path.dirname(outputPath), 0o700); +fs.chmodSync(outputPath, 0o600); dom.window.close(); console.log(`DeviceFP 已保存: ${outputPath}`); console.log(`SessionID: ${result.session_id}`); diff --git a/services/yyb-worker/runtime/scripts/jsdom-pay.py b/services/yyb-worker/runtime/scripts/jsdom-pay.py index 58f9301..e20144a 100644 --- a/services/yyb-worker/runtime/scripts/jsdom-pay.py +++ b/services/yyb-worker/runtime/scripts/jsdom-pay.py @@ -11,8 +11,8 @@ import argparse import hashlib import json import os -import random import re +import secrets import string import subprocess import sys @@ -29,12 +29,41 @@ sys.path.insert(0, str(ROOT)) 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 +from pyvm.protocol import ( # noqa: E402 + DEFAULT_ORDER_PF, + GOODS_USER_AGENT, + PAY_APPID, + PAY_FP_URL, + PAY_GOODS_URL, + PAY_PAGE_INFO_URL, + PAY_SAVE_URL, + PAY_WEB_VERSION, + PAY_WEBVERSION, + goods_material_diagnostics, + validate_goods_materials, +) -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" +APPID = PAY_APPID +GOODS_URL = PAY_GOODS_URL +SAVE_URL = PAY_SAVE_URL +PAGE_INFO_URL = PAY_PAGE_INFO_URL +FP_URL = PAY_FP_URL + + +def make_private_directory(path: Path) -> None: + """创建私有证据目录；已有目录也收紧权限。""" + path.mkdir(parents=True, exist_ok=True) + os.chmod(path, 0o700) + + +def write_private_text(path: Path, content: str) -> None: + """写入含订单或会话材料的原始证据，权限固定为 0600。""" + path.write_text(content, encoding="utf-8") + os.chmod(path, 0o600) + + +def write_private_json(path: Path, value: dict) -> None: + write_private_text(path, json.dumps(value, ensure_ascii=False, indent=2) + "\n") def load_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) @@ -60,8 +89,7 @@ def cookie_header(cookies: dict[str, str]) -> str: def request_bytes(url: str, cookies: dict[str, str], body: bytes | None = None) -> str: 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", + "User-Agent": GOODS_USER_AGENT, "Cookie": cookie_header(cookies), "Referer": "https://pay.qq.com/", } @@ -143,7 +171,7 @@ def build_save_fields(order: dict[str, str], cookies: dict[str, str], web_token: "pfkey": "pfkey", "from_h5": "1", "pc_st": str(uuid.uuid4()).upper() + str(int(time.time() * 1000)), - "r": str(random.random()), + "r": str(secrets.SystemRandom().random()), "openid": cookies.get("openid", ""), "openkey": cookies.get("accesstoken", ""), "session_id": login["session_id"], @@ -165,10 +193,10 @@ def build_save_fields(order: dict[str, str], cookies: dict[str, str], web_token: "pushtype": "NodeJS", "wx_order_interface": "1", "encrypt_msg": encrypt_msg, - "base_key_version": "web_1.0.6", + "base_key_version": PAY_WEB_VERSION, "encrypt_way": "web_new_encrypt", "web_token": web_token, - "webversion": "minipayv2", + "webversion": PAY_WEBVERSION, "from_https": "1", "t": str(int(time.time() * 1000)), "__refer": "https://z.iwan.yyb.qq.com/", @@ -202,7 +230,7 @@ def build_page_info_body(order: dict[str, str], cookies: dict[str, str], anti_to "pfkey": "pfkey", "from_h5": "1", "pc_st": str(uuid.uuid4()).upper() + str(int(time.time() * 1000)), - "r": str(random.random()), + "r": str(secrets.SystemRandom().random()), "openid": cookies.get("openid", ""), "openkey": cookies.get("accesstoken", ""), "session_id": login["session_id"], @@ -211,7 +239,7 @@ def build_page_info_body(order: dict[str, str], cookies: dict[str, str], anti_to "anti_auto_script_token_id": anti_token, "isusempaymode": "1", "zoneid": zone_id, - "webversion": "minipayv2", + "webversion": PAY_WEBVERSION, "from_https": "1", "t": str(int(time.time() * 1000)), "__refer": "https://z.iwan.yyb.qq.com/", @@ -224,7 +252,7 @@ def build_page_info_body(order: dict[str, str], cookies: dict[str, str], anti_to def make_encrypt_rand(params: dict[str, str], fk_extend: str, ts: str, is_qq_login: bool) -> str: """按登录渠道生成页面已验证形态的 _rand，不能仅按长度替换控制字节。""" - prefix = "".join(random.choices(string.ascii_letters + string.digits, k=8)) + prefix = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(8)) # 微信历史成功请求固定使用 8 个随机字符加 \x01。_rand 位于加密明文中， # 末控制字节是协议内容，不能为了对齐擅自替换为 QQ 使用的 \x03。 if not is_qq_login: @@ -255,9 +283,7 @@ def save_web_save_request_meta(out_dir: Path, fields: dict[str, str], device_fp_ "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", - ) + write_private_json(out_dir / "web-save-request-meta.json", meta) def make_qr(sign: str, output: Path) -> None: @@ -265,8 +291,9 @@ def make_qr(sign: str, output: Path) -> None: import segno except ImportError as exc: raise RuntimeError("缺少 segno；请安装后重新执行: python3 -m pip install segno") from exc - output.parent.mkdir(parents=True, exist_ok=True) + make_private_directory(output.parent) segno.make(sign).save(str(output), scale=6, border=2) + os.chmod(output, 0o600) def node_environment() -> dict[str, str]: @@ -286,7 +313,7 @@ def save_payment_status(path: Path, document: dict, baseline: dict[str, bool], m "listed_order_ids": sorted(order_ids(document)), "matched_completion": completion_summary(matched) if matched else None, } - path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_private_json(path, record) def save_payment_meta(path: Path, order: dict[str, str], baseline: dict[str, bool], @@ -313,7 +340,7 @@ def save_payment_meta(path: Path, order: dict[str, str], baseline: dict[str, boo "baseline_order_ids": sorted(order_ids(document)), "qr_created_at": int(time.time()), } - path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_private_json(path, record) def _match_finished_by_identifiers(listed: list[dict], identifiers: dict[str, str]) -> dict | None: @@ -396,11 +423,11 @@ def main() -> int: stamp = time.strftime("%Y%m%d-%H%M%S") out_dir = Path(args.out_dir) if args.out_dir else ROOT / "config" / f"jsdom-order-{stamp}" if args.check_only: - out_dir.mkdir(parents=True, exist_ok=True) + make_private_directory(out_dir) 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" + payment_pf = args.pf or DEFAULT_ORDER_PF session = load_json(session_path) cookies = dict(session.get("cookies", {})) @@ -418,13 +445,13 @@ def main() -> int: stamp = time.strftime("%Y%m%d-%H%M%S") out_dir = Path(args.out_dir) if args.out_dir else ROOT / "config" / f"jsdom-order-{stamp}" - out_dir.mkdir(parents=True, exist_ok=True) + make_private_directory(out_dir) url = goods_page_url(cookies, order, args.zone_id, args.pf) print(f"[jsdom-pay] 订单: {order['token_id'][:20]}...") print("[jsdom-pay] 拉取同订单 goods 页面...") html = request_bytes(url, cookies) html_path = out_dir / "goods.html" - html_path.write_text(html, encoding="utf-8") + write_private_text(html_path, html) xmidas, web_token, anti_token = extract_goods_state(html) print(f"[jsdom-pay] goods: xMidasOps={len(xmidas)} web_token={web_token[:12]}...") @@ -443,7 +470,7 @@ 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") + write_private_text(out_dir / "fp-response.json", fp_response) try: fp_json = json.loads(fp_response) except json.JSONDecodeError: @@ -458,7 +485,7 @@ def main() -> int: 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") + write_private_text(out_dir / "web-page-info-response.json", page_info) try: page_info_json = json.loads(page_info) except json.JSONDecodeError: @@ -470,10 +497,13 @@ def main() -> int: return 1 web_args = load_template_args() + validate_goods_materials(web_args, xmidas) + write_private_json(out_dir / "protocol-materials.json", goods_material_diagnostics(ROOT, web_args, xmidas)) + print("[jsdom-pay] 协议材料校验通过") # 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)] + key16 = [secrets.randbelow(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]) params = { @@ -493,7 +523,7 @@ def main() -> int: "provide_uin": "", "extend": "", "from_h5": "1", - "webversion": "minipayv2", + "webversion": PAY_WEBVERSION, } params.update(midas_login_params(cookies)) now_seconds = str(int(time.time())) @@ -525,7 +555,7 @@ def main() -> int: print(f"[jsdom-pay] 支付前建立基线失败({exc})，将在付款码生成后重试") print("[jsdom-pay] 提交 web_save...") raw = request_bytes(SAVE_URL, cookies, body.encode()) - (out_dir / "web-save-response.json").write_text(raw, encoding="utf-8") + write_private_text(out_dir / "web-save-response.json", raw) try: response_json = json.loads(raw) except json.JSONDecodeError: diff --git a/services/yyb-worker/runtime/scripts/yyb-worker.py b/services/yyb-worker/runtime/scripts/yyb-worker.py index 2843ee6..0d2b498 100644 --- a/services/yyb-worker/runtime/scripts/yyb-worker.py +++ b/services/yyb-worker/runtime/scripts/yyb-worker.py @@ -13,6 +13,7 @@ import importlib.util import json import os import re +import shutil import subprocess import sys import threading @@ -25,6 +26,7 @@ from urllib.parse import urlparse ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) DEFAULT_DATA = ROOT / "config" / "worker-jobs" +DEFAULT_EVIDENCE_TTL_HOURS = 72 WORKER_KEY = os.environ.get("YYB_WORKER_KEY", "") _jobs: dict[str, dict] = {} _lock = threading.Lock() @@ -53,6 +55,25 @@ def _job_dir(job_id: str) -> Path: return Path(_jobs[job_id]["directory"]) +def _cleanup_expired_jobs(data_dir: Path, ttl_hours: int) -> int: + """清理超过保留期的任务证据；只处理 data_dir 的一级任务目录。""" + if ttl_hours <= 0 or not data_dir.exists(): + return 0 + deadline = time.time() - ttl_hours * 3600 + removed = 0 + for directory in data_dir.iterdir(): + if not directory.is_dir() or directory.is_symlink(): + continue + try: + if directory.stat().st_mtime >= deadline: + continue + shutil.rmtree(directory) + removed += 1 + except OSError as exc: + print(f"无法清理过期任务证据 {directory.name}: {exc}", flush=True) + return removed + + def _safe_log(job: dict, line: str) -> None: # Do not persist cookies, payment URI, or long opaque tokens in the worker API. clean = re.sub(r"weixin://\S+", "[付款链接已隐藏]", line) @@ -389,6 +410,7 @@ class Handler(BaseHTTPRequestHandler): job_id = uuid.uuid4().hex[:16] directory = DEFAULT_DATA / job_id directory.mkdir(parents=True, exist_ok=True) + os.chmod(directory, 0o700) _jobs[job_id] = {"job_id": job_id, "status": "created", "phase": "login", "logs": [], "directory": str(directory), "created_at": int(time.time())} return self._json(201, _public_job(job_id)) @@ -565,6 +587,8 @@ def main() -> int: parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8810) parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA) + parser.add_argument("--evidence-ttl-hours", type=int, default=DEFAULT_EVIDENCE_TTL_HOURS, + help="任务原始证据保留时长；0 表示不自动清理") parser.add_argument("--key", default=None, help="HTTP Bearer 鉴权密钥；默认读取 YYB_WORKER_KEY") args = parser.parse_args() # 强制绝对路径：子进程以 ROOT 为 cwd，相对路径会让任务文件写到错误位置。 @@ -573,7 +597,13 @@ def main() -> int: WORKER_KEY = args.key if args.host not in {"127.0.0.1", "localhost", "::1"} and not WORKER_KEY: parser.error("监听非本机地址时必须设置 --key 或 YYB_WORKER_KEY") + if args.evidence_ttl_hours < 0: + parser.error("--evidence-ttl-hours 不能小于 0") DEFAULT_DATA.mkdir(parents=True, exist_ok=True) + os.chmod(DEFAULT_DATA, 0o700) + removed = _cleanup_expired_jobs(DEFAULT_DATA, args.evidence_ttl_hours) + if removed: + print(f"已清理 {removed} 个过期任务证据（保留期 {args.evidence_ttl_hours} 小时）", flush=True) restored = _restore_jobs(DEFAULT_DATA) if restored: print(f"已从任务目录恢复 {restored} 个支付任务", flush=True) diff --git a/services/yyb-worker/runtime/tests/test_jsdom_pay.py b/services/yyb-worker/runtime/tests/test_jsdom_pay.py index f34fb19..2c76422 100644 --- a/services/yyb-worker/runtime/tests/test_jsdom_pay.py +++ b/services/yyb-worker/runtime/tests/test_jsdom_pay.py @@ -16,6 +16,8 @@ sys.modules[_SPEC.name] = _MODULE _SPEC.loader.exec_module(_MODULE) from pyvm.payment_errors import describe_payment_failure +from pyvm.algorithm import derive_key1_from_key16, generate_encrypt_msg_offline +from pyvm.protocol import goods_material_diagnostics, validate_goods_materials class TestJsdomPay(unittest.TestCase): @@ -100,6 +102,54 @@ class TestJsdomPay(unittest.TestCase): self.assertEqual(len(rand_value), 9) self.assertEqual(rand_value[-1], "\x01") + def test_goods_protocol_template_is_compatible(self): + template = _MODULE.load_template_args() + validate_goods_materials(template, [0] * 59640) + diagnostics = goods_material_diagnostics(_MODULE.ROOT, template, [0] * 59640) + self.assertEqual(diagnostics["goods_xmidas_ops_length"], 59640) + self.assertEqual(diagnostics["args_template_slots"], 18) + self.assertEqual(len(diagnostics["bytecode_sha256_prefix"]), 12) + + def test_private_evidence_writer_restricts_permissions(self): + import tempfile + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "evidence.json" + _MODULE.write_private_text(path, "sensitive") + self.assertEqual(path.read_text(encoding="utf-8"), "sensitive") + self.assertEqual(path.stat().st_mode & 0o777, 0o600) + + def test_goods_vm_generates_expected_full_length_ciphertext(self): + """528B 明文应走完 33 个块，输出 1056 个 hex 字符。""" + import json + + template = _MODULE.load_template_args() + xmidas = json.loads((_MODULE.ROOT / "replay" / "xmidasops.json").read_text(encoding="utf-8")) + params = { + "token_id": "token", "openid": "openid", "openkey": "key", "session_id": "openid", + "session_type": "kp_accesstoken", "zoneid": "1", "pay_method": "wechat", + "buy_quantity": "1", "mb_pwd": "", "pay_id": "", "auth_key": "", "card_value": "", + "accounttype": "", "provide_uin": "", "extend": "", "from_h5": "1", + "webversion": "minipayv2", + } + timestamp = "1775990000" + fk_extend = "tdrc_session%3Dpay-test" + for length in range(512): + params["extend"] = "x" * length + rand_value = _MODULE.make_encrypt_rand(params, fk_extend, timestamp, True) + plaintext = _MODULE.build_plaintext(params, fk_extend, timestamp, rand_value) + if len(plaintext.encode("latin-1")) == 528: + break + else: + self.fail("无法构造 528B goods 测试明文") + key16 = list(range(16)) + key1 = derive_key1_from_key16(key16, [template[i][0] for i in (1, 2, 3, 4)], template[5][0]) + ciphertext = generate_encrypt_msg_offline( + params, fk_extend, timestamp, rand_value, key16=key16, key1=key1, + args_template=template, xmidas=xmidas, xmidas_token="A" * 96, + ) + self.assertEqual(len(ciphertext), 1056) + if __name__ == "__main__": unittest.main()