145 lines
5.4 KiB
Python
145 lines
5.4 KiB
Python
"""快速诊断 keyconfig 刷新失败：直接打端点，打印真实响应字节/头。
|
||
|
||
keyconfig 失败发生在启动阶段（mobile/checker 之前），不依赖验证码，所以可以
|
||
脱离完整 CLI 单独复现。配合 core.sms_login 里 ``KS_DEBUG_KEYCONFIG`` 的 stderr
|
||
dump，一次运行就能看到服务端到底返回了什么。
|
||
|
||
用法：
|
||
uv run python -m tools.keyconfig_probe
|
||
# 复用真实设备画像（推荐，避免风控把随机 did 直接挡掉）：
|
||
uv run python -m tools.keyconfig_probe --device-profile out/sms_device_latest.json
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
import urllib.parse
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
if str(ROOT) not in sys.path:
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
# 触发 decode_keyconfig_payload 里的原始字节 dump
|
||
os.environ.setdefault("KS_DEBUG_KEYCONFIG", "1")
|
||
|
||
from core.device_profile import DeviceProfileGenerator, load_device_profile # noqa: E402
|
||
from core.http_transport import create_http_session # noqa: E402
|
||
from core.sms_login import ( # noqa: E402
|
||
DEFAULT_REGION_HOST,
|
||
REGION_FULL_CONFIG_HOST,
|
||
keyconfig_api_params,
|
||
refresh_region_ticket,
|
||
)
|
||
|
||
|
||
def _raw_fetch(host: str, profile: object) -> None:
|
||
"""绕过解析，直接 GET 一次看 status / 响应头 / 头部字节。"""
|
||
|
||
base = host.rstrip("/")
|
||
params = keyconfig_api_params(profile)
|
||
params.update(
|
||
{
|
||
"ud": "0",
|
||
"kcv": "6",
|
||
"keyconfig_state": "1",
|
||
"keyConfigVersion": "6",
|
||
"updatedKeyConfigKey": "base",
|
||
"diffInfo": "",
|
||
"ts": "0",
|
||
"apiInvokeTiming": "COLD_START",
|
||
"is_background": "0",
|
||
"cs": "false",
|
||
"language": "zh-cn",
|
||
}
|
||
)
|
||
from core.sig import sig as calc_sig
|
||
|
||
params["sig"] = calc_sig(params)
|
||
params["__NS_xfalcon"] = ""
|
||
url = f"{base}/rest/nebula/system/keyconfig?{urllib.parse.urlencode(params)}"
|
||
session = create_http_session("okhttp4-android10")
|
||
try:
|
||
resp = session.get(url, headers={"Accept-Encoding": "gzip"}, timeout=20)
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f" raw GET 失败: {exc.__class__.__name__}: {exc}")
|
||
return
|
||
status = int(getattr(resp, "status_code", 0) or 0)
|
||
content = getattr(resp, "content", None)
|
||
if content is None:
|
||
content = (getattr(resp, "text", "") or "").encode("utf-8", errors="ignore")
|
||
head_hex = content[:64].hex()
|
||
head_ascii = "".join(chr(b) if 32 <= b < 127 else "." for b in content[:64])
|
||
headers = dict(getattr(resp, "headers", {}) or {})
|
||
print(f" raw GET status={status} len={len(content)}")
|
||
print(f" raw GET head_hex={head_hex}")
|
||
print(f" raw GET head_ascii={head_ascii!r}")
|
||
interesting = {
|
||
k: v
|
||
for k, v in headers.items()
|
||
if k.lower() in {"content-type", "content-encoding", "content-length", "server", "location"}
|
||
}
|
||
print(f" raw GET headers={interesting}")
|
||
|
||
# 验证：gunzip 后的字节是明文 JSON 还是 XOR 0x2B JSON？
|
||
import gzip as _gzip
|
||
import json as _json
|
||
|
||
def _ascii(b: bytes) -> str:
|
||
return "".join(chr(x) if 32 <= x < 127 else "." for x in b[:80])
|
||
|
||
try:
|
||
gun = _gzip.decompress(content)
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f" gunzip 失败: {exc.__class__.__name__}: {exc}")
|
||
gun = b""
|
||
if gun:
|
||
print(f" gunzip(raw) head_ascii={_ascii(gun)!r}")
|
||
gun_xor = bytes(v ^ 0x2B for v in gun)
|
||
print(f" gunzip(raw)^0x2B head_ascii={_ascii(gun_xor)!r}")
|
||
for label, cand in (("gunzip", gun), ("gunzip^0x2B", gun_xor)):
|
||
try:
|
||
parsed = _json.loads(cand.decode("utf-8"))
|
||
print(f" [OK] {label} 解析为 JSON，顶层 keys={list(parsed.keys())[:8] if isinstance(parsed, dict) else type(parsed)}")
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f" [--] {label} 不是 JSON: {exc.__class__.__name__}")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="诊断 keyconfig 刷新失败")
|
||
parser.add_argument("--device-profile", default="", help="复用 new_device.py 保存的设备画像")
|
||
parser.add_argument("--region-host", default="", help="只测指定 keyconfig host")
|
||
args = parser.parse_args()
|
||
|
||
profile = (
|
||
load_device_profile(args.device_profile)
|
||
if args.device_profile
|
||
else DeviceProfileGenerator().new_profile()
|
||
)
|
||
print(f"[probe] did={profile.did} egid={profile.egid or '-'}")
|
||
session = create_http_session("okhttp4-android10")
|
||
hosts = [args.region_host] if args.region_host else [DEFAULT_REGION_HOST, REGION_FULL_CONFIG_HOST]
|
||
for host in hosts:
|
||
print(f"\n=== 经 refresh_region_ticket（curl okhttp4 TLS）host={host} ===")
|
||
region = refresh_region_ticket(
|
||
profile,
|
||
api_st="",
|
||
client_salt="",
|
||
user_id="",
|
||
base_url=host,
|
||
timeout=20,
|
||
get_func=session.get,
|
||
)
|
||
print(f" ok={region.ok} status={region.status} error={region.error}")
|
||
if region.ok:
|
||
print(f" ticket={region.ticket[:24]}... uid={region.uid or '0'} name={region.name or '-'}")
|
||
print(f"--- 同 host 原始 GET（看真实字节/头）---")
|
||
_raw_fetch(host, profile)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|