src/webui/device_tokens.py
Rama main · réplica de lectura del repositorio original.
"""已配对设备的独立 owner 凭据。
为什么不是「扫码换回访问密码」：访问密码在 STATE 里是 PBKDF2 哈希，服务端
自己也不知道明文，换不出来。更重要的是即便能换，把主密码分发到每台手机上
也是错的——一台设备丢失就只能整体改密码，其它设备跟着全掉线。
所以配对成功签发的是设备令牌：
- 与访问密码平级的 owner 凭据，同样走 ``Authorization: Bearer``；
- 高熵随机串，落盘只存 sha256 摘要（随机 token 不需要 KDF，而且它要在每个
请求上验证，慢哈希会直接变成每请求成本）；
- 可逐台吊销，吊销后该设备立即失去 owner 权限，不影响其它设备与主密码。
"""
from __future__ import annotations
import hashlib
import json
import logging
import secrets
import time
from datetime import datetime, timezone
from pathlib import Path
from aiohttp import web
logger = logging.getLogger("trpg")
MAX_DEVICES = 64
MAX_LABEL_CHARS = 40
# last_seen 每个请求都会变；按秒落盘会把设备清单写成热点文件，这里做节流。
LAST_SEEN_FLUSH_SECONDS = 60
def _now_iso() -> str:
return (
datetime.now(timezone.utc)
.isoformat(timespec="seconds")
.replace("+00:00", "Z")
)
def _digest(token: str) -> str:
return hashlib.sha256(token.strip().encode("utf-8")).hexdigest()
def sanitize_label(value: object) -> str:
text = " ".join(str(value or "").split())[:MAX_LABEL_CHARS]
return text
class InstallIdInUse(ValueError):
"""Another paired device is already bound to this install id."""
def normalize_install_id(value: object) -> str:
"""A client install id (canonical id) or ``""``; never a guess."""
from src.engine.world.contracts import CANONICAL_ID_PATTERN
text = value if isinstance(value, str) else ""
return text if CANONICAL_ID_PATTERN.fullmatch(text) else ""
class DeviceTokenStore:
"""已配对设备清单；令牌只以摘要形式落盘，明文只在签发响应里出现一次。"""
def __init__(self, data_dir: Path, max_devices: int = MAX_DEVICES) -> None:
self.path = Path(data_dir) / "paired_devices.json"
self.max_devices = max(1, int(max_devices))
self._devices: list[dict] = self._load()
self._last_flush = 0.0
# ---- 查询 ----
def entries(self) -> list[dict]:
"""给设置页用的清单；不含任何可用于鉴权的字段。"""
return [
{
"id": device["id"],
"label": device["label"],
"created_at": device["created_at"],
"last_seen_at": device["last_seen_at"],
"install_id": device.get("install_id", ""),
}
for device in sorted(
self._devices,
key=lambda item: item["created_at"],
reverse=True,
)
]
def verify(self, token: str) -> dict | None:
"""校验 Bearer 是否为有效设备令牌；命中时顺带记录活跃时间。"""
candidate = str(token or "").strip()
if not candidate:
return None
digest = _digest(candidate)
for device in self._devices:
if secrets.compare_digest(device["digest"], digest):
self._touch(device)
return device
return None
def has(self, device_id: str) -> bool:
"""Whether ``device_id`` is still paired (cheap: no digest work)."""
target = str(device_id or "")
return bool(target) and any(device["id"] == target for device in self._devices)
# ---- 变更 ----
def issue(self, label: str = "", install_id: str = "") -> tuple[str, dict]:
if self.install_id_holder(install_id):
raise InstallIdInUse(install_id)
token = secrets.token_urlsafe(32)
device = {
"id": secrets.token_hex(8),
"label": sanitize_label(label),
"digest": _digest(token),
"created_at": _now_iso(),
"last_seen_at": "",
"install_id": normalize_install_id(install_id),
}
self._devices.append(device)
# 超额时淘汰最久未活跃的设备，避免清单被无限撑大。
while len(self._devices) > self.max_devices:
self._devices.remove(
min(
self._devices,
key=lambda item: item["last_seen_at"] or item["created_at"],
)
)
self._save()
return token, device
# ---- 内容同步身份 ----
def bind_install_id(self, device_id: str, install_id: str) -> str:
"""Return the install id this device is bound to, binding it on first use.
A device paired before install ids existed has none; its first
declared content push binds the id it declares. Afterwards the device
may only declare that id. Returns ``""`` for an unknown device.
"""
target = str(device_id or "").strip()
wanted = normalize_install_id(install_id)
for device in self._devices:
if device["id"] != target:
continue
if not device.get("install_id") and wanted:
if self.install_id_holder(wanted, excluding=target):
raise InstallIdInUse(wanted)
device["install_id"] = wanted
self._save()
return str(device.get("install_id") or "")
return ""
def install_id_holder(self, install_id: str, *, excluding: str = "") -> str:
"""The paired device already bound to ``install_id`` (``""`` if none)."""
wanted = normalize_install_id(install_id)
if not wanted:
return ""
for device in self._devices:
if device["id"] != excluding and device.get("install_id") == wanted:
return str(device["id"])
return ""
def clear_install_id(self, device_id: str) -> bool:
"""Owner action: let the device bind a new install id (e.g. after a reinstall)."""
target = str(device_id or "").strip()
for device in self._devices:
if device["id"] == target:
device["install_id"] = ""
self._save()
return True
return False
def revoke(self, device_id: str) -> bool:
target = str(device_id or "").strip()
remaining = [device for device in self._devices if device["id"] != target]
if len(remaining) == len(self._devices):
return False
self._devices = remaining
self._save()
return True
def revoke_all(self) -> int:
count = len(self._devices)
self._devices = []
self._save()
return count
# ---- 持久化 ----
def _touch(self, device: dict) -> None:
device["last_seen_at"] = _now_iso()
now = time.monotonic()
if now - self._last_flush < LAST_SEEN_FLUSH_SECONDS:
return
self._last_flush = now
self._save()
def _load(self) -> list[dict]:
if not self.path.exists():
return []
try:
with self.path.open(encoding="utf-8") as file:
payload = json.load(file)
raw = payload.get("devices", []) if isinstance(payload, dict) else []
if not isinstance(raw, list):
raise ValueError("devices is not a list")
except (OSError, ValueError, json.JSONDecodeError, KeyError):
# 清单损坏时按「无已配对设备」继续：宁可让用户重新配对，
# 也不能把无法解析的内容当成有效凭据放行。
logger.warning("已配对设备清单损坏，按空清单继续", exc_info=True)
return []
devices: list[dict] = []
for entry in raw:
if not isinstance(entry, dict):
continue
digest = str(entry.get("digest") or "")
if len(digest) != 64:
continue
devices.append(
{
"id": str(entry.get("id") or secrets.token_hex(8)),
"label": sanitize_label(entry.get("label")),
"digest": digest,
"created_at": str(entry.get("created_at") or ""),
"last_seen_at": str(entry.get("last_seen_at") or ""),
"install_id": normalize_install_id(entry.get("install_id")),
}
)
return devices[-self.max_devices:]
def _save(self) -> None:
payload = json.dumps({"devices": self._devices}, ensure_ascii=False, indent=2)
temporary = self.path.with_suffix(self.path.suffix + ".tmp")
try:
self.path.parent.mkdir(parents=True, exist_ok=True)
temporary.write_text(payload, encoding="utf-8")
temporary.replace(self.path)
except OSError:
logger.exception("已配对设备清单写入失败")
DEVICE_TOKENS_KEY = web.AppKey("device_tokens", DeviceTokenStore)