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

92 lines
3.4 KiB
Python

"""会话态模型:web_new_encrypt 离线生成所需的会话级输入。
F-2049/F-2051(E3):encrypt_msg 与当前会话绑定——
- xMidasOps: goods.shtml 页面内嵌伪随机表(59640 值,服务端每次生成,页面级)
- key16: goodsBiz VM 加载期 Math.random 前 16 次调用(页面级恒定)
- key1: 诱饵密钥(点击级,捕获即可)
服务端能验证 key 派生状态(随机 key 变体 ret:1099),因此新订单必须先捕获会话态。
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
DEFAULT_XMIDAS_TOKEN = "DE46DBA4754D42A6B66ADD4319FF80C144D9ED22ED73384C271793D9A9AA2FFBD6C104BE8D4A7F4ED2A8688FCB6F7540"
@dataclass
class SessionState:
"""一次 goods 页加载捕获的会话态。
xmidas_ops: list[int] 59640 项(页面级)
key16: list[int] 16 字节(VM 加载期 Math.random 前 16 次)
key1: list[int] 16 字节(诱饵)
xmidas_token: str
cookies: dict[str, str](提交 web_save 用,可选)
openid/openkey: str(提交 web_save 用,可选)
"""
xmidas_ops: list[int] = field(default_factory=list)
key16: list[int] = field(default_factory=list)
key1: list[int] = field(default_factory=list)
xmidas_token: str = DEFAULT_XMIDAS_TOKEN
args_template_d: str = "" # webSave 18 参深拷贝(deepcap PC 85091 的 C[2] 原始 D 编码,会话绑定)
cookies: dict[str, str] = field(default_factory=dict)
openid: str = ""
openkey: str = ""
source: str = "" # 捕获来源说明
def validate(self) -> list[str]:
errs: list[str] = []
if len(self.xmidas_ops) != 59640:
errs.append(f"xmidas_ops 长度 {len(self.xmidas_ops)} != 59640")
if len(self.key16) != 16:
errs.append(f"key16 长度 {len(self.key16)} != 16")
if len(self.key1) != 16:
errs.append(f"key1 长度 {len(self.key1)} != 16")
if not self.xmidas_token:
errs.append("xmidas_token 为空")
if not self.args_template_d:
errs.append("args_template_d 为空(需 deepcap PC 85091 的 C[2] 原始 D 编码)")
return errs
def to_json(self) -> dict:
return {
"xmidas_ops": self.xmidas_ops,
"key16": self.key16,
"key1": self.key1,
"xmidas_token": self.xmidas_token,
"args_template_d": self.args_template_d,
"cookies": self.cookies,
"openid": self.openid,
"openkey": self.openkey,
"source": self.source,
}
@classmethod
def from_json(cls, d: dict) -> "SessionState":
return cls(
xmidas_ops=list(d.get("xmidas_ops", [])),
key16=list(d.get("key16", [])),
key1=list(d.get("key1", [])),
xmidas_token=d.get("xmidas_token", DEFAULT_XMIDAS_TOKEN),
args_template_d=d.get("args_template_d", ""),
cookies=dict(d.get("cookies", {})),
openid=d.get("openid", ""),
openkey=d.get("openkey", ""),
source=d.get("source", ""),
)
def load_session(path: str | Path) -> SessionState:
"""从 JSON 加载会话态并校验。"""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"会话态文件不存在: {p}")
st = SessionState.from_json(json.loads(p.read_text(encoding="utf-8")))
errs = st.validate()
if errs:
raise ValueError("会话态校验失败:\n " + "\n ".join(errs))
return st