238 lines
7.0 KiB
Python
238 lines
7.0 KiB
Python
"""KWSG 10400 `encData` and ZT envelope helpers."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from struct import unpack_from
|
||
import time
|
||
import zlib
|
||
|
||
|
||
KWSG_STATIC_AES_KEY = b"5FDxA9ngHE723pYw"
|
||
KWSG_STATIC_AES_IV = b"12345678kuaishou"
|
||
|
||
|
||
ZT_OUTER_CONFIGS = {
|
||
"95147564-9763-4413-a937-6f0e3c12caf1": {
|
||
"head8": bytes.fromhex("5a54eecde4d4ea61"),
|
||
"xor_key": b"M70gN2gdHXA34uIc",
|
||
},
|
||
"bbd910da-fda5-49e7-8667-f57200dac474": {
|
||
"head8": bytes.fromhex("5a54eecdf04c4dad"),
|
||
"xor_key": b"lpYKvL0Mz9bEtHXO",
|
||
},
|
||
"7e46b28a-8c93-4940-8238-4c60e64e3c81": {
|
||
"head8": bytes.fromhex("5a54ebcd594b0dae"),
|
||
"xor_key": b"fGqSL6alaNcUyV9W",
|
||
},
|
||
"5bbcf3cd-727b-48ab-b4b4-5f01e61ee9a5": {
|
||
"head8": bytes.fromhex("5a54eecdd3ce5ab6"),
|
||
"xor_key": b"lealm6bxeMABH3rQ",
|
||
},
|
||
}
|
||
|
||
|
||
KWSG_266FC_PERM = (
|
||
0, 5, 10, 15,
|
||
4, 9, 14, 3,
|
||
8, 13, 2, 7,
|
||
12, 1, 6, 11,
|
||
)
|
||
|
||
KWSG_10400_T1_LEN = 0x24000
|
||
KWSG_10400_T2_LEN = 0xA100
|
||
KWSG_10400_NONCE_XOR = 0xDDCC0DEF
|
||
KWSG_10400_DEFAULT_CFG9 = bytes.fromhex("00cf07009d9ec1b102")
|
||
|
||
|
||
def load_kwsg_10400_tables(
|
||
t1_path: str | Path = "bin/kwsg_10400_T1.bin",
|
||
t2_path: str | Path = "bin/kwsg_10400_T2.bin",
|
||
) -> tuple[bytes, bytes]:
|
||
"""加载 `0x266fc` 所需运行期 T1/T2 表。"""
|
||
t1 = Path(t1_path).read_bytes()
|
||
t2 = Path(t2_path).read_bytes()
|
||
if len(t1) < KWSG_10400_T1_LEN:
|
||
raise ValueError(f"T1 too short: {len(t1)}")
|
||
if len(t2) < KWSG_10400_T2_LEN:
|
||
raise ValueError(f"T2 too short: {len(t2)}")
|
||
return t1, t2
|
||
|
||
|
||
def _u32le(buf: bytes, off: int) -> int:
|
||
return unpack_from("<I", buf, off)[0]
|
||
|
||
|
||
def _kwsg_266fc_permute(state: bytearray) -> bytearray:
|
||
return bytearray(state[i] for i in KWSG_266FC_PERM)
|
||
|
||
|
||
def _kwsg_266fc_lane_core(state: bytearray, t1: bytes, round_base: int, lane: int) -> None:
|
||
p = lane * 4
|
||
table_base = round_base + lane * 0x1000
|
||
|
||
a = _u32le(t1, table_base + 0x000 + state[p + 0] * 4)
|
||
b = _u32le(t1, table_base + 0x400 + state[p + 1] * 4)
|
||
c = _u32le(t1, table_base + 0x800 + state[p + 2] * 4)
|
||
d = _u32le(t1, table_base + 0xC00 + state[p + 3] * 4)
|
||
w = a ^ b ^ c ^ d
|
||
|
||
state[p + 0] = (w >> 24) & 0xFF
|
||
state[p + 1] = (w >> 16) & 0xFF
|
||
state[p + 2] = (w >> 8) & 0xFF
|
||
state[p + 3] = w & 0xFF
|
||
|
||
|
||
def kwsg_266fc_block(block16: bytes, t1: bytes, t2: bytes) -> bytes:
|
||
"""移植 `libkwsgmain.so+0x266fc` 的 16-byte block transform。"""
|
||
if len(block16) != 16:
|
||
raise ValueError("block16 must be exactly 16 bytes")
|
||
|
||
state = bytearray(block16)
|
||
for round_idx in range(9):
|
||
state = _kwsg_266fc_permute(state)
|
||
round_base = round_idx * 0x4000
|
||
for lane in range(4):
|
||
_kwsg_266fc_lane_core(state, t1, round_base, lane)
|
||
|
||
state = _kwsg_266fc_permute(state)
|
||
for i in range(16):
|
||
state[i] = t2[0x9000 + i * 0x100 + state[i]]
|
||
return bytes(state)
|
||
|
||
|
||
def kwsg_10400_ecb_encrypt(payload: bytes, t1: bytes, t2: bytes) -> bytes:
|
||
"""移植 `0x27534` 包装层的 ECB-like + PKCS#7 加密输出。"""
|
||
pad = 16 - (len(payload) % 16)
|
||
padded = payload + bytes([pad]) * pad
|
||
return b"".join(
|
||
kwsg_266fc_block(padded[i:i + 16], t1, t2)
|
||
for i in range(0, len(padded), 16)
|
||
)
|
||
|
||
|
||
def kwsg_10400_nonce9(epoch_seconds: int | None = None) -> bytes:
|
||
"""生成 `0x11b08` 当前实测分支的 9-byte nonce 字段。"""
|
||
if epoch_seconds is None:
|
||
epoch_seconds = int(time.time())
|
||
value = (int(epoch_seconds) & 0xFFFFFFFF) ^ KWSG_10400_NONCE_XOR
|
||
return str(value).encode("ascii")[:9]
|
||
|
||
|
||
def zt_outer_wrap(inner: bytes, head8: bytes, xor_key: bytes) -> bytes:
|
||
"""`sub_0x122e4 mode 1`: `head8 + repeating_xor(inner)`."""
|
||
if len(head8) != 8:
|
||
raise ValueError("head8 must be 8 bytes")
|
||
if len(xor_key) != 16:
|
||
raise ValueError("xor_key must be 16 bytes")
|
||
body = bytes(b ^ xor_key[i & 0x0F] for i, b in enumerate(inner))
|
||
return head8 + body
|
||
|
||
|
||
def zt_outer_unwrap(raw: bytes, xor_key: bytes) -> bytes:
|
||
"""反解 `sub_0x122e4 mode 1` 外层 body，返回 inner ZT 数据。"""
|
||
if len(raw) < 8:
|
||
raise ValueError("raw too short")
|
||
if len(xor_key) != 16:
|
||
raise ValueError("xor_key must be 16 bytes")
|
||
body = raw[8:]
|
||
return bytes(b ^ xor_key[i & 0x0F] for i, b in enumerate(body))
|
||
|
||
|
||
def build_inner_zt_header(nonce9: bytes, cfg9: bytes, payload: bytes) -> bytes:
|
||
"""构造 `sub_0x11b08` 的 0x20-byte inner header。"""
|
||
if len(nonce9) != 9:
|
||
raise ValueError("nonce9 must be 9 bytes")
|
||
if len(cfg9) != 9:
|
||
raise ValueError("cfg9 must be 9 bytes")
|
||
return (
|
||
bytes.fromhex("dec0adde")
|
||
+ (0x20).to_bytes(2, "little")
|
||
+ nonce9
|
||
+ cfg9
|
||
+ (zlib.crc32(payload) & 0xFFFFFFFF).to_bytes(4, "little")
|
||
+ len(payload).to_bytes(4, "little")
|
||
)
|
||
|
||
|
||
def parse_inner_zt_header(inner: bytes) -> dict:
|
||
"""解析反 XOR 后的 inner ZT header。"""
|
||
if len(inner) < 0x20:
|
||
raise ValueError("inner too short")
|
||
return {
|
||
"magic": inner[:4],
|
||
"header_size": int.from_bytes(inner[4:6], "little"),
|
||
"nonce9": inner[6:15],
|
||
"cfg9": inner[15:24],
|
||
"crc32": int.from_bytes(inner[24:28], "little"),
|
||
"payload_len": int.from_bytes(inner[28:32], "little"),
|
||
"payload": inner[32:],
|
||
}
|
||
|
||
|
||
def derive_outer_xor_key(raw: bytes, inner_payload_first16: bytes) -> bytes:
|
||
"""由最终 raw 和 inner payload 前 16 字节反推 16-byte XOR key。"""
|
||
if len(raw) < 8 + 0x20 + 16:
|
||
raise ValueError("raw too short")
|
||
if len(inner_payload_first16) < 16:
|
||
raise ValueError("need 16 bytes of inner payload")
|
||
return bytes(raw[8 + 0x20 + i] ^ inner_payload_first16[i] for i in range(16))
|
||
|
||
|
||
def kwsg_10400_raw_with_inner_fields(
|
||
payload: bytes,
|
||
sdk_id: str,
|
||
nonce9: bytes,
|
||
cfg9: bytes,
|
||
t1: bytes,
|
||
t2: bytes,
|
||
) -> bytes:
|
||
"""生成当前已验证 `10400` raw 输出。"""
|
||
cfg = ZT_OUTER_CONFIGS[sdk_id]
|
||
encrypted = kwsg_10400_ecb_encrypt(payload, t1, t2)
|
||
inner = build_inner_zt_header(nonce9, cfg9, encrypted) + encrypted
|
||
return zt_outer_wrap(inner, cfg["head8"], cfg["xor_key"])
|
||
|
||
|
||
def kwsg_10400_raw(
|
||
payload: bytes,
|
||
sdk_id: str,
|
||
t1: bytes,
|
||
t2: bytes,
|
||
*,
|
||
epoch_seconds: int | None = None,
|
||
cfg9: bytes = KWSG_10400_DEFAULT_CFG9,
|
||
) -> bytes:
|
||
"""用当前已还原字段生成 `10400` raw。"""
|
||
return kwsg_10400_raw_with_inner_fields(
|
||
payload,
|
||
sdk_id,
|
||
kwsg_10400_nonce9(epoch_seconds),
|
||
cfg9,
|
||
t1,
|
||
t2,
|
||
)
|
||
|
||
|
||
__all__ = [
|
||
"KWSG_10400_DEFAULT_CFG9",
|
||
"KWSG_10400_NONCE_XOR",
|
||
"KWSG_10400_T1_LEN",
|
||
"KWSG_10400_T2_LEN",
|
||
"KWSG_266FC_PERM",
|
||
"KWSG_STATIC_AES_IV",
|
||
"KWSG_STATIC_AES_KEY",
|
||
"ZT_OUTER_CONFIGS",
|
||
"build_inner_zt_header",
|
||
"derive_outer_xor_key",
|
||
"kwsg_10400_ecb_encrypt",
|
||
"kwsg_10400_nonce9",
|
||
"kwsg_10400_raw",
|
||
"kwsg_10400_raw_with_inner_fields",
|
||
"kwsg_266fc_block",
|
||
"load_kwsg_10400_tables",
|
||
"parse_inner_zt_header",
|
||
"zt_outer_unwrap",
|
||
"zt_outer_wrap",
|
||
]
|