Files
live-hub-py/services/yyb-worker/runtime/pyvm/goods.py
T

114 lines
3.2 KiB
Python

"""goods 侧(web_save / web_new_encrypt,goodsBiz CHAOS VM 116 opcode)高层 API。
与 mall 侧(pyvm/mall.py)同属 YYB 加密体系。goods 侧 E3 已验证:
会话态(xMidasOps 59640 + key16/key1) + 订单参数
→ build_plaintext 21 字段明文(528 字符)
→ a:8 变换(Te0-3 + key16 按块轮转)
→ webSave(33 块变换)
→ encrypt_msg(1056 hex)
"""
from __future__ import annotations
import json
from pathlib import Path
from .algorithm import generate_encrypt_msg_offline
REPLAY = Path(__file__).resolve().parent.parent / "replay"
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",
]
class GoodsSession:
"""goods 会话态:xMidasOps(59640,服务端生成)+ key16/key1 + args_template。"""
def __init__(
self,
xmidas_ops: list,
key16: list,
key1: list,
args_template_d: str = "",
xmidas_token: str = "",
):
self.xmidas_ops = xmidas_ops
self.key16 = key16
self.key1 = key1
self.args_template_d = args_template_d
self.xmidas_token = xmidas_token
self.validate()
def validate(self) -> None:
if len(self.xmidas_ops) != 59640:
raise ValueError(f"goods xMidasOps 应为 59640,实际 {len(self.xmidas_ops)}")
if len(self.key16) != 16 or len(self.key1) != 16:
raise ValueError("key16/key1 应为 16 字节")
@classmethod
def from_session_state(cls, path: str | Path) -> GoodsSession:
"""从 scripts/capture-session.mjs 生成的 session-state.json 加载。"""
d = json.loads(Path(path).read_text(encoding="utf-8"))
return cls(
d["xmidas_ops"],
d["key16"],
d["key1"],
d.get("args_template_d", ""),
d.get("xmidas_token", ""),
)
def to_json(self) -> dict:
return {
"xmidas_ops": self.xmidas_ops,
"key16": self.key16,
"key1": self.key1,
"args_template_d": self.args_template_d,
"xmidas_token": self.xmidas_token,
}
@classmethod
def from_json(cls, d: dict) -> GoodsSession:
return cls(
d["xmidas_ops"],
d["key16"],
d["key1"],
d.get("args_template_d", ""),
d.get("xmidas_token", ""),
)
def generate_encrypt_msg(session: GoodsSession, order: dict) -> str:
"""用会话态 + 订单参数生成 goods encrypt_msg(1056 hex)。"""
from .algorithm import decode_d
params = {k: order.get(k, "") for k in ORDER_FIELDS}
args_tpl = decode_d(session.args_template_d) if session.args_template_d else None
return generate_encrypt_msg_offline(
params,
order.get("fk_extend", ""),
order.get("ts", ""),
order.get("_rand", ""),
xmidas=session.xmidas_ops,
xmidas_token=session.xmidas_token,
args_template=args_tpl,
key16=session.key16,
key1=session.key1,
)