"""Admin JWT:与 App 用户 token 完全隔离。 用独立 secret(settings.ADMIN_JWT_SECRET ≠ JWT_SECRET_KEY)+ payload typ="admin", 双重保证 App 用户的 access_token 无法当 admin token 用(secret 不同直接验签失败)。 admin 无 refresh:过期(默认 12h)重新登录,简单。 """ from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any import jwt from app.core.config import settings _ALGO = "HS256" class AdminTokenError(Exception): """admin token 解析/校验失败,api 层捕获后返回 401。""" def _now() -> datetime: return datetime.now(timezone.utc) def create_admin_token(*, admin_id: int, role: str) -> tuple[str, int]: """签发 admin access token,返回 (token, expires_in_seconds)。""" now = _now() expire = now + timedelta(minutes=settings.ADMIN_JWT_EXPIRE_MINUTES) payload: dict[str, Any] = { "sub": str(admin_id), "role": role, "typ": "admin", "iat": int(now.timestamp()), "exp": int(expire.timestamp()), } token = jwt.encode(payload, settings.ADMIN_JWT_SECRET, algorithm=_ALGO) return token, int((expire - now).total_seconds()) def decode_admin_token(token: str) -> dict[str, Any]: """解析校验 admin token。签名错/过期/typ 非 admin → AdminTokenError。""" try: payload = jwt.decode(token, settings.ADMIN_JWT_SECRET, algorithms=[_ALGO]) except jwt.ExpiredSignatureError as e: raise AdminTokenError("token expired") from e except jwt.InvalidTokenError as e: raise AdminTokenError(f"invalid token: {e}") from e if payload.get("typ") != "admin": raise AdminTokenError("not an admin token") if "sub" not in payload: raise AdminTokenError("token missing sub") return payload