343 lines
11 KiB
Python
343 lines
11 KiB
Python
"""纯 Python 复现 ``gdfp.gifshow.com/f/a/p`` 设备指纹上报端点。
|
||
|
||
该端点不是 ``passport_account_image`` 的生成源。登录票据由客户端
|
||
``Engine.pr`` 本地生成；本模块仅保留设备遥测协议研究和响应提取工具。
|
||
|
||
静态定论（``out/jadx/sources/com/kuaishou/weapon/ks/``）：
|
||
|
||
签名 ``h1.a(map)`` / ``h1.d(ctx)``
|
||
``sign = md5_hex(appkey + secretkey + timestamp)``
|
||
query = ``appkey=X&secretkey=Y×tamp=Z&sign=S``
|
||
|
||
appkey/secretkey ``t.a(ctx)`` / ``h1.a(ctx)``
|
||
优先读 ``wcfg``（``z0.i()``，``"appkey-secretkey"`` 拆分）；为空回退硬编码
|
||
默认值 ``appkey="16"``、``secretkey="62c80c436b7547a68a12774c67519836"``。
|
||
|
||
URL ``r1.a(payload, ctx)``
|
||
``{base}/f/a/p?{query}``，``x0.d`` = ``Base64("L2YvYS9w")`` = ``"/f/a/p"``。
|
||
|
||
Body ``r1.a`` 三参路径（注册上传，带 ``VIMG_`` 前缀）
|
||
``{"data": "VIMG_" + WeaponHI.b(payload)}``
|
||
其中 ``payload`` 是 JSON 字符串（见 ``build_default_payload``）。
|
||
|
||
响应 ``i.a`` 返回 JSON 字符串。尚无证据表明响应会签发登录使用的
|
||
``VIMG_<base64>$AI_<32hex>``；递归提取函数只用于兼容历史样本分析。
|
||
|
||
两个设备遥测字段仍待确认（代码内以 ``TODO(U1/U2)`` 标注）：
|
||
|
||
U1
|
||
payload 明文结构（``build_default_payload`` 的字段集/取值），候选来自
|
||
``t.a()`` / ``h1.b()``，需对照抓到的请求 body 校验。
|
||
|
||
U2
|
||
历史响应中若存在 ``a_y_q_z``，确认其字段路径和业务用途。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import urllib.parse
|
||
from dataclasses import dataclass
|
||
from typing import Any, Callable, Mapping
|
||
|
||
from .privacykit_encrypt import (
|
||
PASSPORT_ACCOUNT_IMAGE_AI_MARKER,
|
||
PASSPORT_ACCOUNT_IMAGE_PREFIX,
|
||
weaponhi_vimg_upload,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配置（可经环境变量覆盖，与 core/sms_login.py / core/dfp_client.py 风格一致）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DEFAULT_FAP_HOST = "https://gdfp.gifshow.com"
|
||
FAP_PATH = "/f/a/p"
|
||
|
||
# h1.java:52 硬编码默认值（t.a() 返回空时 APP 自身回退用）。
|
||
DEFAULT_APPKEY = "16"
|
||
DEFAULT_SECRETKEY = "62c80c436b7547a68a12774c67519836"
|
||
|
||
WEAPON_SDK_VERSION = "7.2.1"
|
||
|
||
# a_y_q_z 票据形态：VIMG_<base64>$AI_<32hex>。递归匹配兜底用。
|
||
_TICKET_RE = re.compile(
|
||
re.escape(PASSPORT_ACCOUNT_IMAGE_PREFIX)
|
||
+ r"[A-Za-z0-9+/]+={0,2}"
|
||
+ re.escape(PASSPORT_ACCOUNT_IMAGE_AI_MARKER)
|
||
+ r"[0-9a-f]{32}"
|
||
)
|
||
|
||
|
||
def _fap_host() -> str:
|
||
return os.environ.get("KS_FAP_HOST", DEFAULT_FAP_HOST).rstrip("/")
|
||
|
||
|
||
def _appkey() -> str:
|
||
return os.environ.get("KS_FAP_APPKEY", DEFAULT_APPKEY)
|
||
|
||
|
||
def _secretkey() -> str:
|
||
return os.environ.get("KS_FAP_SECRETKEY", DEFAULT_SECRETKEY)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 签名 / query（h1.java：纯 MD5，与 atlasSign / kws 无关）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def build_sign(appkey: str, secretkey: str, timestamp: int) -> str:
|
||
"""``md5_hex(appkey + secretkey + timestamp)``。
|
||
|
||
对照 ``v0.a(String)`` = ``MessageDigest("MD5").digest(str.getBytes())``。
|
||
"""
|
||
|
||
raw = f"{appkey}{secretkey}{timestamp}".encode("utf-8")
|
||
return hashlib.md5(raw).hexdigest()
|
||
|
||
|
||
def build_query(appkey: str, secretkey: str, timestamp: int) -> str:
|
||
"""``appkey=X&secretkey=Y×tamp=Z&sign=S``（``h1.d`` 顺序）。"""
|
||
|
||
sign = build_sign(appkey, secretkey, timestamp)
|
||
# 显式保序，避免依赖 dict/HashMap 迭代顺序。
|
||
pairs = [
|
||
("appkey", appkey),
|
||
("secretkey", secretkey),
|
||
("timestamp", str(timestamp)),
|
||
("sign", sign),
|
||
]
|
||
return urllib.parse.urlencode(pairs)
|
||
|
||
|
||
def build_fap_url(timestamp: int | None = None, *, host: str | None = None) -> str:
|
||
ts = int(timestamp if timestamp is not None else time.time())
|
||
host = host or _fap_host()
|
||
return f"{host}{FAP_PATH}?{build_query(_appkey(), _secretkey(), ts)}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Body：{"data": "VIMG_" + WeaponHI.b(payload_json)}（r1.a 三参路径）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def build_fap_body(payload: Mapping[str, Any] | str) -> bytes:
|
||
"""构造 ``{"data": "VIMG_" + WeaponHI.b(payload_json)}``。
|
||
|
||
``payload`` 可传 dict（自动 ``json.dumps``，``separators=(',', ':')`` 紧凑）
|
||
或已序列化的字符串。返回 UTF-8 JSON bytes。
|
||
"""
|
||
|
||
if isinstance(payload, str):
|
||
payload_str = payload
|
||
else:
|
||
# WeaponHI.b(str) 上游拿的是 str.getBytes()；JSON 紧凑序列化对齐 Java 侧 JSONObject。
|
||
payload_str = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
|
||
vimg = weaponhi_vimg_upload(payload_str)
|
||
return json.dumps({"data": vimg}, separators=(",", ":")).encode("utf-8")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# payload 明文（U1 候选，需抓证校验）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def build_default_payload(device_info: Mapping[str, Any] | None = None) -> dict[str, Any]:
|
||
"""构造 /f/a/p payload 候选结构。
|
||
|
||
候选字段来自 ``t.a()``（userId/platform/channel/mod/globalId/sysver/
|
||
rdid/did_tag/cdid_tag）与 ``h1.b(ctx)``（k/hp/hv/pver/platform/device_id/
|
||
sdkver/piv/sysver/mod）的并集，值从 ``device_info``（通常取自
|
||
``out/app_login_fields_latest.json`` 的 query_params 快照）映射。
|
||
|
||
# TODO(U1)：抓证后用真实请求 body 校验字段集/取值/嵌套结构，回填此函数。
|
||
"""
|
||
|
||
info = device_info or {}
|
||
get = lambda key, default="": str(info.get(key, default) or default or "")
|
||
|
||
return {
|
||
# t.a() 侧
|
||
"userId": "",
|
||
"platform": get("kpf", "ANDROID_PHONE"),
|
||
"channel": get("c", get("oc", "")),
|
||
"mod": get("mod", ""),
|
||
"globalId": "",
|
||
"sysver": get("sys", ""),
|
||
"rdid": get("rdid", ""),
|
||
"did_tag": get("did_tag", ""),
|
||
"cdid_tag": get("cdid_tag", ""),
|
||
# h1.b(ctx) 侧
|
||
"k": "",
|
||
"hp": get("kpn", "com.kuaishou.nebula"),
|
||
"hv": get("appver", ""),
|
||
"pver": "0.0.0",
|
||
"platform_h1": 1,
|
||
"device_id": get("did", get("oDid", "")),
|
||
"sdkver": WEAPON_SDK_VERSION,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 响应解析（U2，需抓证定位精确字段路径）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _iter_str_values(node: Any):
|
||
"""递归遍历解析后的 JSON，产出所有字符串值。"""
|
||
|
||
if isinstance(node, str):
|
||
yield node
|
||
elif isinstance(node, Mapping):
|
||
for v in node.values():
|
||
yield from _iter_str_values(v)
|
||
elif isinstance(node, (list, tuple)):
|
||
for v in node:
|
||
yield from _iter_str_values(v)
|
||
|
||
|
||
def extract_a_y_q_z(response: Any) -> str | None:
|
||
"""从 /f/a/p 响应里提取 ``a_y_q_z`` 完整票据。
|
||
|
||
# TODO(U2)：抓证后改为精确字段路径（如 ``response["a_y_q_z"]``）。
|
||
当前用递归匹配 ``VIMG_...$AI_...`` 形态兜底，能覆盖顶层/嵌套两种情况。
|
||
"""
|
||
|
||
if response is None:
|
||
return None
|
||
if isinstance(response, str):
|
||
# 响应是裸字符串时直接匹配；否则尝试 JSON 解析后递归。
|
||
m = _TICKET_RE.search(response)
|
||
if m:
|
||
return m.group(0)
|
||
try:
|
||
response = json.loads(response)
|
||
except Exception:
|
||
return None
|
||
for value in _iter_str_values(response):
|
||
m = _TICKET_RE.search(value)
|
||
if m:
|
||
return m.group(0)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP（参考 core/dfp_client.post_request 的 requests 范式）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FapResponse:
|
||
ok: bool
|
||
status_code: int
|
||
data: Any
|
||
text: str
|
||
error: str = ""
|
||
|
||
|
||
def _default_post() -> Callable[..., Any]:
|
||
import requests
|
||
|
||
return requests.post
|
||
|
||
|
||
def call_fap(
|
||
payload: Mapping[str, Any] | str,
|
||
*,
|
||
timestamp: int | None = None,
|
||
host: str | None = None,
|
||
headers: Mapping[str, str] | None = None,
|
||
timeout: int = 20,
|
||
post_func: Callable[..., Any] | None = None,
|
||
) -> FapResponse:
|
||
"""POST ``/f/a/p`` 并返回解析后的响应。
|
||
|
||
headers 可传 k1 的 c/d（``t.b()`` 设备串 / ``t.b(ctx)`` 加密设备头）。
|
||
默认不送；若服务端校验 header 则从抓证固化后传入。
|
||
"""
|
||
|
||
post = post_func or _default_post()
|
||
ts = int(timestamp if timestamp is not None else time.time())
|
||
url = build_fap_url(ts, host=host)
|
||
body = build_fap_body(payload)
|
||
|
||
hdrs = {"Content-Type": "application/json; charset=utf-8"}
|
||
if headers:
|
||
hdrs.update(dict(headers))
|
||
|
||
try:
|
||
resp = post(url, data=body, headers=hdrs, timeout=timeout)
|
||
except Exception as exc:
|
||
return FapResponse(ok=False, status_code=0, data=None, text=str(exc), error=exc.__class__.__name__)
|
||
|
||
text = getattr(resp, "text", "")
|
||
try:
|
||
data = resp.json()
|
||
except Exception:
|
||
try:
|
||
data = json.loads(text)
|
||
except Exception:
|
||
data = text
|
||
return FapResponse(
|
||
ok=bool(getattr(resp, "ok", False)),
|
||
status_code=int(getattr(resp, "status_code", 0)),
|
||
data=data,
|
||
text=text,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 顶层入口
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def fetch_fresh_a_y_q_z(
|
||
device_info: Mapping[str, Any] | None = None,
|
||
*,
|
||
payload: Mapping[str, Any] | str | None = None,
|
||
host: str | None = None,
|
||
headers: Mapping[str, str] | None = None,
|
||
timeout: int = 20,
|
||
post_func: Callable[..., Any] | None = None,
|
||
) -> str | None:
|
||
"""取新鲜 ``a_y_q_z`` 完整票据（``VIMG_...$AI_...``）。
|
||
|
||
默认用 ``build_default_payload(device_info)`` 组 payload；调用方可直接传
|
||
``payload``（dict 或已序列化字符串）覆盖。失败返回 None，由调用方决定回退。
|
||
"""
|
||
|
||
if payload is None:
|
||
payload = build_default_payload(device_info)
|
||
resp = call_fap(
|
||
payload,
|
||
host=host,
|
||
headers=headers,
|
||
timeout=timeout,
|
||
post_func=post_func,
|
||
)
|
||
if not resp.ok:
|
||
return None
|
||
return extract_a_y_q_z(resp.data)
|
||
|
||
|
||
__all__ = [
|
||
"DEFAULT_APPKEY",
|
||
"DEFAULT_FAP_HOST",
|
||
"DEFAULT_SECRETKEY",
|
||
"FAP_PATH",
|
||
"FapResponse",
|
||
"WEAPON_SDK_VERSION",
|
||
"build_default_payload",
|
||
"build_fap_body",
|
||
"build_fap_url",
|
||
"build_query",
|
||
"build_sign",
|
||
"call_fap",
|
||
"extract_a_y_q_z",
|
||
"fetch_fresh_a_y_q_z",
|
||
]
|