yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

强化支付协议校验与证据保护

Browse Source
This commit is contained in:
yml2213
2026-08-12 23:54:14 +08:00
parent 74a85c8328
commit 99520d9b9b
6 changed files with 292 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/yyb-worker/runtime/scripts/generate-devicefp-jsdom.mjs
+2
View File
@@ -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}`);
services/yyb-worker/runtime/scripts/jsdom-pay.py
+59 -29
View File
@@ -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:
services/yyb-worker/runtime/scripts/yyb-worker.py
+30
View File
@@ -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)