# -*- coding: utf-8 -*- # Author : Charley # Python : 3.12.10 # Date : 2026/08/12 """得卡 DECA App token 抓取脚本（root 直读本地存储，替代 mitmproxy 代理抓包）。 设备已 root（Magisk）且 得卡 App 已登录时，直接从 App 私有目录的 Jetpack DataStore 文件读出当前登录态的 accessToken / refreshToken， 算好过期时间，覆盖写入项目根目录 token.json，供 deca_sold_core.py 等复用。 与千岛版差异（都是 root 直读，只是存储格式不同）： 千岛 token 明文存 SharedPreferences xml，一条 grep 即可； 得卡 token 存 DataStore 的 protobuf 文件 deka_settings.preferences_pb 里， 藏在顶层 key `switch_accounts` 的 JSON 数组内（每个账号一条 userInfo）， 需按 protobuf 解出该 string，再解 JSON 取 accessToken 等字段。 App 登录 / 切号后该 pb 会实时更新，App 后台也会用 refreshToken 自动续期并回写， 所以本脚本读到的必然是「此刻有效」的 token。token.json 只需种一次，之后 deca_sold_core.ensure_token 会靠 refreshToken 自续期，refreshToken 失效时再跑本脚本。 用法： python get_token.py # 读设备 → 覆盖写 token.json → 打印脱敏摘要 """ import os import sys import json import time import base64 import subprocess # 得卡 App 包名（pm list packages 实测：有好卡 · 得卡） PACKAGE_NAME = "com.youhaoka.deka" # 登录态所在的 DataStore（Preferences，protobuf 二进制）文件 PB_PATH = f"/data/data/{PACKAGE_NAME}/files/datastore/deka_settings.preferences_pb" # token 藏身的顶层 DataStore key，value 是一段 JSON 数组（多账号切换列表） ACCOUNTS_KEY = "switch_accounts" # token.json 与本脚本同目录（deca_sold_core 从运行目录读它） TOKEN_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "token.json") def _read_varint(buf: bytes, i: int) -> tuple[int, int]: """从字节流指定位置解一个 protobuf varint。 Args: buf (bytes): 原始字节流。 i (int): 起始下标。 Returns: tuple[int, int]: (解出的整数值, 解析后的新下标)。 """ shift = 0 result = 0 while True: b = buf[i] i += 1 result |= (b & 0x7F) << shift if not b & 0x80: # 最高位为 0 表示 varint 结束 return result, i shift += 7 def _get_pref_string(raw: bytes, key: str) -> str | None: """从 DataStore Preferences protobuf 里取指定 key 的 string value。 DataStore 结构：顶层 map（field 1，每条是一个 entry message）， entry 内 field 1=key(string)、field 2=Value message；Value 里 field 5=string 值。 Args: raw (bytes): deka_settings.preferences_pb 的完整字节。 key (str): 要取的顶层 key 名，如 "switch_accounts"。 Returns: str | None: 该 key 的字符串值；未找到返回 None。 """ i = 0 while i < len(raw): tag, i = _read_varint(raw, i) field, wire = tag >> 3, tag & 7 if field == 1 and wire == 2: # 顶层 map entry（length-delimited） ln, i = _read_varint(raw, i) entry = raw[i:i + ln] i += ln j = 0 k = v = None while j < len(entry): t2, j = _read_varint(entry, j) f2, w2 = t2 >> 3, t2 & 7 if f2 == 1 and w2 == 2: # entry.key kl, j = _read_varint(entry, j) k = entry[j:j + kl].decode("utf-8", "replace") j += kl elif f2 == 2 and w2 == 2: # entry.value（Value message） vl, j = _read_varint(entry, j) vb = entry[j:j + vl] j += vl m = 0 while m < len(vb): # 在 Value 里找 field 5=string t3, m = _read_varint(vb, m) f3, w3 = t3 >> 3, t3 & 7 if f3 == 5 and w3 == 2: sl, m = _read_varint(vb, m) v = vb[m:m + sl].decode("utf-8", "replace") m += sl elif w3 == 0: # 跳过其它类型字段（bool/int 等） _, m = _read_varint(vb, m) elif w3 == 2: xl, m = _read_varint(vb, m) m += xl else: break else: break if k == key: return v elif wire == 0: # 跳过非 map 的 varint 字段 _, i = _read_varint(raw, i) elif wire == 2: # 跳过非 map 的 length-delimited 字段 ln, i = _read_varint(raw, i) i += ln else: break return None def _decode_jwt_exp(access: str) -> int | None: """从 JWT accessToken 的 payload 里解出 exp（过期 unix 秒）。 Args: access (str): JWT 字符串（header.payload.signature）。 Returns: int | None: exp 时间戳；解析失败返回 None。 """ try: payload = access.split(".")[1] payload += "=" * (-len(payload) % 4) # 补齐 base64url 填充 data = json.loads(base64.urlsafe_b64decode(payload)) return int(data["exp"]) if "exp" in data else None except Exception: return None def read_device_token() -> dict: """从已登录的 得卡 App 读取当前账号的 token 三元组。 通过 `adb exec-out su -c cat` 以 root 读 DataStore pb 文件（exec-out 保留二进制、 不做 CRLF 转换），解出 switch_accounts；多账号时取 tokenObtainedAtMs 最大者 （最近获取 = 当前登录账号）；exp 优先用 JWT 解，兜底 tokenObtainedAtMs+expiresIn。 Returns: dict: {"access": str, "refresh": str, "exp": int, "userId": ..., "phone": str}。 Raises: RuntimeError: adb 缺失 / 超时 / 未读到 pb / 未登录（无账号）时抛出。 """ cmd = ["adb", "exec-out", "su", "-c", f"cat {PB_PATH}"] try: result = subprocess.run(cmd, capture_output=True, timeout=20) except FileNotFoundError as e: raise RuntimeError(f"[{sys._getframe().f_code.co_name}] 未找到 adb，请确认已配置环境变量") from e except subprocess.TimeoutExpired as e: raise RuntimeError(f"[{sys._getframe().f_code.co_name}] adb 命令超时，请检查设备连接") from e raw = result.stdout or b"" if not raw: err = (result.stderr or b"").decode("utf-8", "replace").strip() or "未知错误" raise RuntimeError( f"[{sys._getframe().f_code.co_name}] 读取 pb 失败：{err}\n" f"排查：1) adb devices 是否有设备 2) 设备是否已 root 3) 得卡 App 是否已登录" ) accounts_json = _get_pref_string(raw, ACCOUNTS_KEY) if not accounts_json: raise RuntimeError( f"[{sys._getframe().f_code.co_name}] pb 里没有 {ACCOUNTS_KEY}，请确认 得卡 App 已登录" ) try: accounts = json.loads(accounts_json) except Exception as e: raise RuntimeError(f"[{sys._getframe().f_code.co_name}] {ACCOUNTS_KEY} 不是合法 JSON：{e}") from e if not accounts: raise RuntimeError(f"[{sys._getframe().f_code.co_name}] 账号列表为空，请在 得卡 App 登录") # 多账号时取 tokenObtainedAtMs 最大（最近登录/续期）的那个作为当前账号 def _obtained(acc: dict) -> int: """取账号 userInfo 里的 tokenObtainedAtMs，缺失按 0。""" return int((acc.get("userInfo") or {}).get("tokenObtainedAtMs") or 0) cur = max(accounts, key=_obtained) info = cur.get("userInfo") or {} access = info.get("accessToken") refresh = info.get("refreshToken") if not access or not refresh: raise RuntimeError(f"[{sys._getframe().f_code.co_name}] 当前账号缺 accessToken/refreshToken，请重登") exp = _decode_jwt_exp(access) if exp is None: # JWT 解不出 → 用获取时刻 + expiresIn obtained_s = _obtained(cur) // 1000 or int(time.time()) exp = obtained_s + int(info.get("expiresIn") or 900) return {"access": access, "refresh": refresh, "exp": exp, "userId": info.get("userId"), "phone": info.get("phone")} def write_token_json(tk: dict) -> None: """把 token 三元组覆盖写入 token.json（格式与旧抓包脚本一致，消费方零改动）。 Args: tk (dict): read_device_token 的返回值，至少含 access/refresh/exp。 """ payload = {"access": tk["access"], "refresh": tk["refresh"], "exp": tk["exp"]} with open(TOKEN_FILE, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False) def main() -> None: """读设备 token → 覆盖写 token.json → 打印脱敏摘要。""" tk = read_device_token() write_token_json(tk) acc, exp, now = tk["access"], tk["exp"], int(time.time()) left = exp - now print(f"[OK] token.json 已更新 -> {TOKEN_FILE}") print(f" userId : {tk.get('userId')} phone : {tk.get('phone')}") print(f" access : JWT 长{len(acc)} 头[{acc[:10]}] 尾[{acc[-6:]}]") print(f" exp : {exp} ({time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(exp))})" f" 剩余 {left} 秒 {'有效' if left > 0 else '已过期(App 侧会自动续期，稍后重跑)'}") if __name__ == "__main__": try: main() except RuntimeError as exc: print(exc) sys.exit(1)