ksjsb/core/device_profile.py
2026-07-30 20:25:56 +08:00

569 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
import hashlib
import json
import random
import re
import time
import uuid
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
from .device_id import egid_from_seed_material, is_valid_oaid, oaid_from_seed_material
ANDROID_ID_RE = re.compile(r"^[0-9a-f]{16}$")
ANDROID_PREFIX_RE = re.compile(r"^ANDROID_[0-9a-f]{16}$")
EGID_RE = re.compile(r"^$|^DFP[0-9A-F]{61}$")
@dataclass
class DfpRuntimeHints:
did_gt: str = ""
total_memory_bytes: str = ""
build_fingerprint: str = ""
build_product: str = ""
k4_native: str = ""
storage_available_bytes: int = 0
k51_native: str = ""
k84_native: str = ""
res_soc: str = ""
boot_id: str = ""
grdi: str = ""
ipv6_map: str = ""
lpss: str = ""
keeper_seed: str = ""
du: str = ""
sted_cache_json: str = ""
persisted_cache_m: str = ""
manus: str = ""
gaid: str = ""
oaid: str = ""
wifi_mac: str = ""
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "DfpRuntimeHints":
data = data or {}
return cls(
did_gt=str(data.get("did_gt", "")),
total_memory_bytes=str(data.get("total_memory_bytes", "")),
build_fingerprint=str(data.get("build_fingerprint", "")),
build_product=str(data.get("build_product", "")),
k4_native=str(data.get("k4_native", "")),
storage_available_bytes=int(data.get("storage_available_bytes", 0)),
k51_native=str(data.get("k51_native", "")),
k84_native=str(data.get("k84_native", "")),
res_soc=str(data.get("res_soc", "")),
boot_id=str(data.get("boot_id", "")),
grdi=str(data.get("grdi", "")),
ipv6_map=str(data.get("ipv6_map", "")),
lpss=str(data.get("lpss", "")),
keeper_seed=str(data.get("keeper_seed", "")),
du=str(data.get("du", "")),
sted_cache_json=str(data.get("sted_cache_json", data.get("cache_m", ""))),
persisted_cache_m=str(data.get("persisted_cache_m", "")),
manus=str(data.get("manus", "")),
gaid=str(data.get("gaid", "")),
oaid=str(data.get("oaid", "")),
wifi_mac=str(data.get("wifi_mac", "")),
)
@dataclass
class DeviceProfile:
android_id: str
local_did: str
did: str
o_did: str
rdid: str
g_rdi2: str
cdid_tag: int = 0
egid: str = ""
install_time_ms: int = 0
cold_launch_time_ms: int = 0
sid: str = ""
package_name: str = "com.kuaishou.nebula"
app_version: str = "14.5.50.11631"
android_release: str = "16"
manufacturer: str = "OnePlus"
brand: str = "OPPO"
model: str = "PJZ110"
build_id: str = "BP2A.250605.015"
build_display: str = "PJZ110_16.0.8.301(CN01)"
build_product: str = "OP5D0DL1"
build_fingerprint: str = ""
build_tags: str = "release-keys"
build_type: str = "user"
board_platform: str = "sun"
soc_name: str = "Qualcomm Snapdragon 8750"
max_memory: int = 256
device_bit: str = "4"
screen_width: int = 1080
screen_height: int = 2376
status_bar_height: int = 120
screen_density: float = 3.0
screen_xdpi: str = "386.36618"
screen_ydpi: str = "381.96454"
total_memory_mb: int = 15107
country_code: str = "cn"
isp: str = "CUCC"
runtime_hints: DfpRuntimeHints = field(default_factory=DfpRuntimeHints)
def __post_init__(self) -> None:
self.validate()
def validate(self) -> None:
if not ANDROID_ID_RE.fullmatch(self.android_id):
raise ValueError(f"invalid android_id: {self.android_id!r}")
for name in ("local_did", "did", "o_did", "rdid"):
value = getattr(self, name)
if not ANDROID_PREFIX_RE.fullmatch(value):
raise ValueError(f"invalid {name}: {value!r}")
if self.o_did != f"ANDROID_{self.android_id}":
raise ValueError("o_did must equal ANDROID_<android_id>")
expected_rdid = hashlib.md5(self.g_rdi2.encode("utf-8")).hexdigest()[16:32]
if self.rdid != f"ANDROID_{expected_rdid}":
raise ValueError("rdid must equal ANDROID_<md5(g_rdi2)[16:32]>")
if not isinstance(self.cdid_tag, int) or self.cdid_tag < 0:
raise ValueError(f"invalid cdid_tag: {self.cdid_tag!r}")
if not EGID_RE.fullmatch(self.egid):
raise ValueError(f"invalid egid: {self.egid!r}")
if self.runtime_hints.oaid and not is_valid_oaid(self.runtime_hints.oaid):
raise ValueError(f"invalid oaid: {self.runtime_hints.oaid!r}")
if self.screen_width <= 0 or self.screen_height <= 0:
raise ValueError("screen size must be positive")
if self.total_memory_mb <= 0:
raise ValueError("total_memory_mb must be positive")
if self.max_memory <= 0:
raise ValueError("max_memory must be positive")
if not self.board_platform:
raise ValueError("board_platform must not be empty")
if not self.soc_name:
raise ValueError("soc_name must not be empty")
def apply_cloud_identity(self, did: str, cdid_tag: int, egid: str = "") -> None:
self.did = did
self.cdid_tag = cdid_tag
if egid:
self.egid = egid
self.sync_egid_cache()
self.validate()
def sync_egid_cache(self) -> None:
"""让 k112/STED JSON 与公开 EGID 保持同一设备身份。"""
if not self.egid:
return
kuaishou_egid = egid_from_seed_material(["KUAISHOU", self.egid, self.android_id])
self.runtime_hints.sted_cache_json = json.dumps(
{"NEBULA": self.egid, "KUAISHOU": kuaishou_egid},
separators=(",", ":"),
)
self.refresh_persisted_cache_m()
def refresh_persisted_cache_m(self) -> None:
"""刷新 rq0.d.cache_m即 c.s(Context) 的本地硬件摘要。"""
from .dfp_cache import build_persisted_cache_m
self.runtime_hints.persisted_cache_m = build_persisted_cache_m(self)
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "DeviceProfile":
return cls(
android_id=str(data["android_id"]),
local_did=str(data["local_did"]),
did=str(data["did"]),
o_did=str(data["o_did"]),
rdid=str(data["rdid"]),
g_rdi2=str(data["g_rdi2"]),
cdid_tag=int(data.get("cdid_tag", 0)),
egid=str(data.get("egid", "")),
install_time_ms=int(data.get("install_time_ms", 0)),
cold_launch_time_ms=int(data.get("cold_launch_time_ms", 0)),
sid=str(data.get("sid", "")),
package_name=str(data.get("package_name", "com.kuaishou.nebula")),
app_version=str(data.get("app_version", "14.5.50.11631")),
android_release=str(data.get("android_release", "16")),
manufacturer=str(data.get("manufacturer", "OnePlus")),
brand=str(data.get("brand", "OPPO")),
model=str(data.get("model", "PJZ110")),
build_id=str(data.get("build_id", "BP2A.250605.015")),
build_display=str(data.get("build_display", "PJZ110_16.0.8.301(CN01)")),
build_product=str(data.get("build_product", "OP5D0DL1")),
build_fingerprint=str(data.get("build_fingerprint", "")),
build_tags=str(data.get("build_tags", "release-keys")),
build_type=str(data.get("build_type", "user")),
board_platform=str(data.get("board_platform", "sun")),
soc_name=str(data.get("soc_name", "Qualcomm Snapdragon 8750")),
max_memory=int(data.get("max_memory", 256)),
device_bit=str(data.get("device_bit", "4")),
screen_width=int(data.get("screen_width", 1080)),
screen_height=int(data.get("screen_height", 2376)),
status_bar_height=int(data.get("status_bar_height", 120)),
screen_density=float(data.get("screen_density", 3.0)),
screen_xdpi=str(data.get("screen_xdpi", "386.36618")),
screen_ydpi=str(data.get("screen_ydpi", "381.96454")),
total_memory_mb=int(data.get("total_memory_mb", 15107)),
country_code=str(data.get("country_code", "cn")),
isp=str(data.get("isp", "CUCC")),
runtime_hints=DfpRuntimeHints.from_dict(data.get("runtime_hints")),
)
@property
def screen_metrics(self) -> str:
content_height = self.screen_height - self.status_bar_height - 48
return (
f"[{self.screen_density:.1f},{self.screen_width},{content_height},"
f"{self.screen_density:.1f},{self.screen_xdpi},{self.screen_ydpi}]"
)
def to_env(self) -> str:
lines = [
f"KS_ANDROID_ID={self.android_id}",
f"KS_DID={self.did}",
f"KS_LOCAL_DID={self.local_did}",
f"KS_ODID={self.o_did}",
f"KS_RDID={self.rdid}",
f"KS_GRDI2={self.g_rdi2}",
f"KS_CDID_TAG={self.cdid_tag}",
f"KS_EGID={self.egid}",
f"KS_INSTALL_TIME_MS={self.install_time_ms}",
f"KS_COLD_LAUNCH_TIME_MS={self.cold_launch_time_ms}",
f"KS_SID={self.sid}",
f"KS_PACKAGE_NAME={self.package_name}",
f"KS_APPVER={self.app_version}",
f"KS_ANDROID_RELEASE={self.android_release}",
f"KS_MANUFACTURER={self.manufacturer}",
f"KS_BRAND={self.brand}",
f"KS_MODEL={self.model}",
f"KS_BUILD_ID={self.build_id}",
f"KS_BUILD_DISPLAY={self.build_display}",
f"KS_BUILD_PRODUCT={self.build_product}",
f"KS_BUILD_FINGERPRINT={self.build_fingerprint}",
f"KS_BUILD_TAGS={self.build_tags}",
f"KS_BUILD_TYPE={self.build_type}",
f"KS_BOARD_PLATFORM={self.board_platform}",
f"KS_SOC_NAME={self.soc_name}",
f"KS_MAX_MEMORY={self.max_memory}",
f"KS_DEVICE_BIT={self.device_bit}",
f"KS_SCREEN_WIDTH={self.screen_width}",
f"KS_SCREEN_HEIGHT={self.screen_height}",
f"KS_STATUS_BAR_HEIGHT={self.status_bar_height}",
f"KS_SCREEN_DENSITY={self.screen_density:.1f}",
f"KS_SCREEN_XDPI={self.screen_xdpi}",
f"KS_SCREEN_YDPI={self.screen_ydpi}",
f"KS_TOTAL_MEMORY_MB={self.total_memory_mb}",
f"KS_COUNTRY_CODE={self.country_code}",
f"KS_ISP={self.isp}",
f"KS_DID_GT={self.runtime_hints.did_gt}",
f"KS_TOTAL_MEMORY_BYTES={self.runtime_hints.total_memory_bytes}",
f"KS_K4={self.runtime_hints.k4_native}",
f"KS_K20={self.runtime_hints.storage_available_bytes}",
f"KS_K51={self.runtime_hints.k51_native}",
f"KS_K84={self.runtime_hints.k84_native}",
f"KS_RESSOC={self.runtime_hints.res_soc}",
f"KS_BOOT_ID={self.runtime_hints.boot_id}",
f"KS_K105={self.runtime_hints.grdi}",
f"KS_GRDI={self.runtime_hints.grdi}",
f"KS_IPV6_MAP={self.runtime_hints.ipv6_map}",
f"KS_LPSS={self.runtime_hints.lpss}",
f"KS_KEEPER_SEED={self.runtime_hints.keeper_seed}",
f"KS_DU={self.runtime_hints.du}",
f"KS_STED={self.runtime_hints.sted_cache_json}",
f"KS_DFP_CACHE_M={self.runtime_hints.persisted_cache_m}",
f"KS_MANUS={self.runtime_hints.manus}",
f"KS_GAID={self.runtime_hints.gaid}",
f"KS_OAID={self.runtime_hints.oaid}",
f"KS_WIFI_MAC={self.runtime_hints.wifi_mac}",
]
return "\n".join(lines) + "\n"
class DeviceProfileGenerator:
def __init__(self, seed: int | None = None) -> None:
self._random = random.Random(seed)
def new_profile(self) -> DeviceProfile:
now_ms = int(time.time() * 1000)
install_time_ms = now_ms - self._random.randint(10_000, 600_000)
android_id = self._hex16()
local_did = f"ANDROID_{self._hex16()}"
g_rdi2 = self._g_rdi2()
rdid_suffix = hashlib.md5(g_rdi2.encode("utf-8")).hexdigest()[16:32]
hardware = self._hardware_profile()
total_memory_bytes = str(
hardware.get("total_memory_bytes", int(hardware["total_memory_mb"]) * 1024 * 1024)
)
hardware.pop("total_memory_bytes", None)
sid = str(uuid.UUID(int=self._random.getrandbits(128)))
egid = self._egid(android_id, local_did, g_rdi2, install_time_ms, now_ms, sid, hardware)
rdid = f"ANDROID_{rdid_suffix}"
oaid = oaid_from_seed_material([android_id, local_did, rdid, g_rdi2, sid, egid])
profile = DeviceProfile(
android_id=android_id,
local_did=local_did,
did=local_did,
o_did=f"ANDROID_{android_id}",
rdid=rdid,
g_rdi2=g_rdi2,
egid=egid,
install_time_ms=install_time_ms,
cold_launch_time_ms=now_ms,
sid=sid,
runtime_hints=self._runtime_hints(
egid,
did_gt=str(install_time_ms + self._random.randint(1_000, 60_000)),
total_memory_bytes=total_memory_bytes,
build_fingerprint=str(hardware.get("build_fingerprint", "")),
build_product=str(hardware.get("build_product", "")),
oaid=oaid,
),
**hardware,
)
profile.sync_egid_cache()
return profile
def _hex16(self) -> str:
return f"{self._random.getrandbits(64):016x}"
def _random_mac(self) -> str:
"""随机本地管理 MAC首字节 bit1=1 locally administered, bit0=0 unicast"""
b = [0x02 | (self._random.getrandbits(6) << 2)]
b += [self._random.getrandbits(8) for _ in range(5)]
return ":".join(f"{x:02x}" for x in b)
def _g_rdi2(self) -> str:
parts = []
for _ in range(5):
left = self._random.choice((7, 8, 9)) * 100_000_000 + self._random.randint(0, 999_999)
right = self._random.choice((4741, 8641))
parts.append(f"{left}::{right}")
return "|".join(parts)
def _egid(
self,
android_id: str,
local_did: str,
g_rdi2: str,
install_time_ms: int,
cold_launch_time_ms: int,
sid: str,
hardware: dict[str, Any],
) -> str:
return egid_from_seed_material(
[
android_id,
local_did,
g_rdi2,
install_time_ms,
cold_launch_time_ms,
sid,
hardware.get("manufacturer", ""),
hardware.get("brand", ""),
hardware.get("model", ""),
hardware.get("build_id", ""),
hardware.get("screen_width", ""),
hardware.get("screen_height", ""),
hardware.get("total_memory_mb", ""),
]
)
def _hardware_profile(self) -> dict[str, Any]:
templates = [
{
"manufacturer": "OnePlus",
"brand": "OPPO",
"model": "PJZ110",
"build_id": "BP2A.250605.015",
"build_display": "PJZ110_16.0.8.301(CN01)",
"build_product": "OP5D0DL1",
"build_fingerprint": "OnePlus/PJZ110/OP5D0DL1:16/BP2A.250605.015/V.4e5c566-2a38f4c-2a4ca91:user/release-keys",
"board_platform": "sun",
"soc_name": "Qualcomm Snapdragon 8750",
"max_memory": 256,
"device_bit": "4",
"screen_width": 1080,
"screen_height": 2376,
"status_bar_height": 120,
"screen_density": 3.0,
"screen_xdpi": "386.36618",
"screen_ydpi": "381.96454",
"total_memory_mb": 15107,
"total_memory_bytes": 15841333248,
},
{
"manufacturer": "OPPO",
"brand": "OPPO",
"model": "PKB110",
"build_id": "BP1A.250305.019",
"build_display": "PKB110_15.0.1.601(CN01)",
"build_product": "PKB110",
"build_fingerprint": "OPPO/PKB110/PKB110:15/BP1A.250305.019/PKB110_15.0.1.601(CN01):user/release-keys",
"board_platform": "pineapple",
"soc_name": "Qualcomm Snapdragon 8 Gen 3",
"max_memory": 256,
"device_bit": "4",
"screen_width": 1080,
"screen_height": 2412,
"status_bar_height": 120,
"screen_density": 3.0,
"screen_xdpi": "394.215",
"screen_ydpi": "392.781",
"total_memory_mb": 12288,
"total_memory_bytes": 12884901888,
},
{
"manufacturer": "vivo",
"brand": "vivo",
"model": "V2408A",
"build_id": "BP1A.250205.007",
"build_display": "V2408A_A_15.1.9.6.W10",
"build_product": "V2408A",
"build_fingerprint": "vivo/V2408A/V2408A:15/BP1A.250205.007/V2408A_A_15.1.9.6.W10:user/release-keys",
"board_platform": "dimensity9400",
"soc_name": "MediaTek Dimensity 9400",
"max_memory": 256,
"device_bit": "4",
"screen_width": 1260,
"screen_height": 2800,
"status_bar_height": 132,
"screen_density": 3.0,
"screen_xdpi": "450.0",
"screen_ydpi": "450.0",
"total_memory_mb": 16384,
"total_memory_bytes": 17179869184,
},
]
selected = dict(self._random.choice(templates))
selected["isp"] = self._random.choice(["CUCC", "CTCC", "CMCC"])
return selected
def _runtime_hints(
self,
egid: str = "",
*,
did_gt: str = "",
total_memory_bytes: str = "",
build_fingerprint: str = "",
build_product: str = "",
oaid: str = "",
) -> DfpRuntimeHints:
storage_gb = self._random.randint(180, 460)
grdi_parts = []
for _ in range(5):
left = self._random.choice((5, 6, 7, 8, 9)) * 100_000_000 + self._random.randint(0, 99_999_999)
right = self._random.choice((3841, 4741, 5317, 8641))
grdi_parts.append(f"{left}::{right}")
ipv6_map = json.dumps(
self._ipv6_map(),
separators=(",", ":"),
)
manus = json.dumps(
{
"5": {
"1": "KWE_N",
"2": "KWE_N",
"3": "KWE_N",
"7": str(int(time.time() * 1000)),
"8": "KWE_N",
"10": "KWE_N",
}
},
separators=(",", ":"),
)
return DfpRuntimeHints(
did_gt=did_gt,
total_memory_bytes=total_memory_bytes,
build_fingerprint=build_fingerprint,
build_product=build_product,
k4_native=str(self._random.randint(1_000_000_000, 4_200_000_000)),
storage_available_bytes=storage_gb * 1024 * 1024 * 1024,
k51_native=self._hex16(),
k84_native=self._hex16(),
res_soc=f"soc-{self._hex16()}",
boot_id=str(uuid.UUID(int=self._random.getrandbits(128))),
grdi="|".join(grdi_parts),
ipv6_map=ipv6_map,
lpss=f"lp-{self._hex16()}",
keeper_seed=str(self._random.getrandbits(63)),
du=f"2@{self._hex16()}{self._hex16()}",
sted_cache_json=json.dumps({"NEBULA": egid}, separators=(",", ":")) if egid else "",
manus=manus,
gaid=str(uuid.UUID(int=self._random.getrandbits(128))),
oaid=oaid or (self._hex16() + self._hex16() + self._hex16() + self._hex16()).upper(),
wifi_mac=self._random_mac(),
)
def _ipv6_group(self) -> str:
return f"{self._random.getrandbits(16):x}"
def _ipv6_addr(self, prefix: str = "2408") -> str:
return (
f"{prefix}:{self._ipv6_group()}:{self._ipv6_group()}:{self._ipv6_group()}:"
f"{self._ipv6_group()}:{self._ipv6_group()}:{self._ipv6_group()}:{self._ipv6_group()}"
)
def _link_local(self, iface: str) -> str:
return f"fe80::{self._ipv6_group()}:{self._ipv6_group()}:fe{self._ipv6_group()[:2]}:{self._ipv6_group()}%{iface}"
def _ipv6_map(self) -> dict[str, str]:
ifaces = [
"rmnet_data1",
"",
"",
"wlan0",
"",
"",
"tun0",
"rmnet_data3",
"ifb0",
"",
"r_rmnet_data0",
"rmnet_data4",
"rmnet_data2",
"ifb1",
"dummy0",
"",
"ifb2",
"vgate0",
]
values: dict[str, str] = {"0": self._ipv6_addr()}
for index, iface in enumerate(ifaces, 1):
if iface:
values[str(index)] = self._link_local(iface)
else:
values[str(index)] = self._ipv6_addr(prefix=self._random.choice(["2408", "fd42", "2a00"]))
return values
def save_device_profile(profile: DeviceProfile, path: str | Path) -> None:
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(
json.dumps(profile.to_dict(), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
def load_device_profile(path: str | Path) -> DeviceProfile:
source = Path(path)
try:
data = json.loads(source.read_text(encoding="utf-8"))
except Exception as exc:
raise ValueError(f"failed to load device profile: {source}") from exc
if not isinstance(data, dict):
raise ValueError(f"device profile must be a JSON object: {source}")
return DeviceProfile.from_dict(data)