275 lines
8.1 KiB
Python
275 lines
8.1 KiB
Python
"""Nebula H5 `$encode` signInput builder and local JS-VM bridge.
|
||
|
||
前端真实逻辑在 `main-CZ3ZSK5w.js`：
|
||
|
||
1. 从 cookie 只取白名单设备字段；
|
||
2. 加入 `sigCatVer=1` 和接口 query/body；
|
||
3. 按 `key=value` 字符串字典序拼接；
|
||
4. 调 KsGuard/Yoda VM 的 `$encode` 生成 68hex `__NS_sig3`。
|
||
|
||
这里保留稳定的 signInput 组装规则和本地 VM runner fallback。正常主流程
|
||
已经在 `core.h5_sig3` 里用纯 Python 复现 `$encode` 的摘要字段。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import queue
|
||
import subprocess
|
||
import threading
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
H5_COOKIE_KEYS = (
|
||
"kpn",
|
||
"kpf",
|
||
"userId",
|
||
"did",
|
||
"c",
|
||
"appver",
|
||
"language",
|
||
"mod",
|
||
"did_tag",
|
||
"egid",
|
||
"oDid",
|
||
"androidApiLevel",
|
||
"newOc",
|
||
"browseType",
|
||
"socName",
|
||
"ftt",
|
||
"abi",
|
||
"userRecoBit",
|
||
"device_abi",
|
||
"grant_browse_type",
|
||
"iuid",
|
||
"rdid",
|
||
)
|
||
|
||
DEFAULT_H5_ENCODE_RUNNER = Path("core/h5_vendor_encode.mjs")
|
||
DEFAULT_H5_ENCODE_SERVER = Path("core/h5_vendor_encode_server.mjs")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class H5EncodeResult:
|
||
sign_input: str
|
||
sig3: str
|
||
c_info: Any = None
|
||
|
||
|
||
class H5JsBridgeEncoder:
|
||
"""长驻 Node `$encode` 进程，避免每次签名重复加载 VM chunk。"""
|
||
|
||
def __init__(
|
||
self,
|
||
server: str | Path = DEFAULT_H5_ENCODE_SERVER,
|
||
node_bin: str | None = None,
|
||
timeout: int = 30,
|
||
) -> None:
|
||
self.server = Path(server)
|
||
self.node = node_bin or os.environ.get("NODE_BIN") or "node"
|
||
self.timeout = timeout
|
||
self._proc: subprocess.Popen[str] | None = None
|
||
self._stdout_queue: queue.Queue[str] = queue.Queue()
|
||
self._lock = threading.Lock()
|
||
self._request_id = 0
|
||
|
||
def _start(self) -> None:
|
||
if self._proc is not None and self._proc.poll() is None:
|
||
return
|
||
if not self.server.exists():
|
||
raise FileNotFoundError(f"H5 encode server not found: {self.server}")
|
||
self._stdout_queue = queue.Queue()
|
||
self._proc = subprocess.Popen(
|
||
[self.node, str(self.server)],
|
||
stdin=subprocess.PIPE,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.DEVNULL,
|
||
text=True,
|
||
encoding="utf-8",
|
||
bufsize=1,
|
||
)
|
||
|
||
def read_stdout() -> None:
|
||
assert self._proc is not None
|
||
assert self._proc.stdout is not None
|
||
for line in self._proc.stdout:
|
||
self._stdout_queue.put(line)
|
||
|
||
thread = threading.Thread(target=read_stdout, name="H5JsBridgeEncoderStdout", daemon=True)
|
||
thread.start()
|
||
|
||
def close(self) -> None:
|
||
proc = self._proc
|
||
self._proc = None
|
||
if proc is None:
|
||
return
|
||
try:
|
||
if proc.stdin:
|
||
proc.stdin.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if proc.stdout:
|
||
proc.stdout.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
proc.wait(timeout=1)
|
||
except Exception:
|
||
try:
|
||
proc.kill()
|
||
except Exception:
|
||
pass
|
||
|
||
def encode(self, sign_input: str) -> H5EncodeResult:
|
||
with self._lock:
|
||
self._start()
|
||
assert self._proc is not None
|
||
assert self._proc.stdin is not None
|
||
self._request_id += 1
|
||
request_id = self._request_id
|
||
payload = json.dumps(
|
||
{"id": request_id, "signInput": sign_input},
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
)
|
||
try:
|
||
self._proc.stdin.write(payload + "\n")
|
||
self._proc.stdin.flush()
|
||
except Exception:
|
||
self.close()
|
||
raise
|
||
|
||
while True:
|
||
try:
|
||
line = self._stdout_queue.get(timeout=self.timeout)
|
||
except queue.Empty as exc:
|
||
self.close()
|
||
raise TimeoutError("H5 encode server timed out") from exc
|
||
try:
|
||
data = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if data.get("id") not in {request_id, None}:
|
||
continue
|
||
if not data.get("ok", True):
|
||
raise RuntimeError(f"H5 encode server failed: {data.get('error')}")
|
||
sig3 = str(data.get("result") or "")
|
||
if len(sig3) != 68:
|
||
raise RuntimeError(f"H5 encode server returned invalid sig3: {sig3!r}")
|
||
return H5EncodeResult(sign_input=sign_input, sig3=sig3, c_info=data.get("cInfo"))
|
||
|
||
|
||
def _js_json(value: Any) -> str:
|
||
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
|
||
def _h5_io(params: dict[str, Any]) -> str:
|
||
parts: list[str] = []
|
||
for key, value in params.items():
|
||
if key.startswith("__NS"):
|
||
continue
|
||
if isinstance(value, (dict, list)):
|
||
value = ""
|
||
parts.append(f"{key}={value}")
|
||
return "".join(sorted(parts))
|
||
|
||
|
||
def build_h5_sign_input(
|
||
cookie: dict[str, str],
|
||
query: dict[str, Any] | None = None,
|
||
body: str | dict[str, Any] | list[Any] | None = None,
|
||
method: str = "GET",
|
||
request_type: str = "json",
|
||
) -> str:
|
||
"""按前端 `Mc()/io()` 规则构造 `$encode` 的 secPlain。"""
|
||
selected = {key: cookie[key] for key in H5_COOKIE_KEYS if cookie.get(key)}
|
||
params: dict[str, Any] = {"sigCatVer": 1, **selected, **(query or {})}
|
||
method_lower = method.lower()
|
||
request_type_lower = request_type.lower()
|
||
|
||
if request_type_lower == "json":
|
||
if method_lower in {"get", "options", "head"}:
|
||
if isinstance(body, dict):
|
||
params.update(body)
|
||
return _h5_io(params)
|
||
if body is None:
|
||
body_text = ""
|
||
elif isinstance(body, str):
|
||
body_text = body
|
||
else:
|
||
body_text = _js_json(body)
|
||
return _h5_io(params) + body_text
|
||
|
||
if request_type_lower == "form" and isinstance(body, dict):
|
||
params.update(body)
|
||
return _h5_io(params)
|
||
|
||
|
||
def encode_h5_sig3_with_runner(
|
||
sign_input: str,
|
||
runner: str | Path = DEFAULT_H5_ENCODE_RUNNER,
|
||
node_bin: str | None = None,
|
||
timeout: int = 30,
|
||
) -> H5EncodeResult:
|
||
"""调用本地 JS-VM runner 生成 H5 68hex `__NS_sig3`。"""
|
||
runner_path = Path(runner)
|
||
if not runner_path.exists():
|
||
raise FileNotFoundError(f"H5 encode runner not found: {runner_path}")
|
||
|
||
node = node_bin or os.environ.get("NODE_BIN") or "node"
|
||
proc = subprocess.run(
|
||
[node, str(runner_path), "-"],
|
||
input=sign_input,
|
||
text=True,
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
check=False,
|
||
)
|
||
if proc.returncode != 0:
|
||
detail = (proc.stderr or proc.stdout or "").strip()
|
||
raise RuntimeError(f"H5 encode runner failed: {detail}")
|
||
data = json.loads(proc.stdout)
|
||
sig3 = str(data.get("result") or "")
|
||
if len(sig3) != 68:
|
||
raise RuntimeError(f"H5 encode runner returned invalid sig3: {sig3!r}")
|
||
return H5EncodeResult(sign_input=sign_input, sig3=sig3, c_info=data.get("cInfo"))
|
||
|
||
|
||
def h5_sig3_for_request(
|
||
cookie: dict[str, str],
|
||
query: dict[str, Any] | None = None,
|
||
body: str | dict[str, Any] | list[Any] | None = None,
|
||
method: str = "GET",
|
||
request_type: str = "json",
|
||
runner: str | Path = DEFAULT_H5_ENCODE_RUNNER,
|
||
node_bin: str | None = None,
|
||
timeout: int = 30,
|
||
encoder: H5JsBridgeEncoder | None = None,
|
||
) -> H5EncodeResult:
|
||
sign_input = build_h5_sign_input(
|
||
cookie,
|
||
query=query,
|
||
body=body,
|
||
method=method,
|
||
request_type=request_type,
|
||
)
|
||
if encoder is not None:
|
||
return encoder.encode(sign_input)
|
||
return encode_h5_sig3_with_runner(sign_input, runner=runner, node_bin=node_bin, timeout=timeout)
|
||
|
||
|
||
__all__ = [
|
||
"DEFAULT_H5_ENCODE_RUNNER",
|
||
"DEFAULT_H5_ENCODE_SERVER",
|
||
"H5_COOKIE_KEYS",
|
||
"H5EncodeResult",
|
||
"H5JsBridgeEncoder",
|
||
"build_h5_sign_input",
|
||
"encode_h5_sig3_with_runner",
|
||
"h5_sig3_for_request",
|
||
]
|