> ## Documentation Index > Fetch the complete documentation index at: https://docs.profy.cn/llms.txt > Use this file to discover all available pages before exploring further. # Token 管理最佳实践 > 生产环境下的 OAuth Token 持久化、并发刷新、轮换安全与降级策略，构建可靠的多用户鉴权层 ## 你将构建什么 一个生产级的 Token 管理层——覆盖持久化存储选型、多用户隔离、并发刷新竞争、过期降级、主动撤销和安全存储，确保你的 SaaS 应用在 Token 轮换和服务重启后仍然正常工作。 ## Token 生命周期 ```mermaid theme={null} stateDiagram-v2 [*] --> Issued: exchangeCode() Issued --> Active: 存储 Token Active --> Active: API 调用（Access Token 有效） Active --> Refreshing: Access Token 过期（401） Refreshing --> Active: 刷新成功，新 Token 对已存储 Refreshing --> Expired: Refresh Token 失效 / 90 天到期 Active --> Revoked: 用户断开连接 / 主动撤销 Expired --> [*]: 引导用户重新授权 Revoked --> [*]: 清理存储 ``` | Token | 有效期 | 轮换规则 | | ------------- | --------- | ------------------------------- | | Access Token | 1 小时（JWT） | 每次刷新签发新的 | | Refresh Token | 90 天 | **每次使用即失效**，同时签发新 Refresh Token | ## 核心原则 违反以下任何一条，都会导致用户静默掉线——没有错误提示，只有突然的「请重新登录」。 **1. 刷新后必须持久化** — `onTokenRefresh` 回调是你保存新 Token 的唯一时机。错过它，旧 Refresh Token 已失效，新 Token 丢失，用户必须重新授权。 **2. 禁止纯内存存储** — 服务重启 = 所有用户 Token 丢失 = 全体重新登录。Token 必须落盘（DB / Redis / 文件）。 **3. 轮换必须原子化** — 每次 `refreshToken()` 调用后，旧的 Access Token 和 Refresh Token **同时失效**。保存新 Token 和废弃旧 Token 必须在同一事务中完成。 ## Step 1: 选择存储策略 | 方案 | 适用场景 | 优势 | 劣势 | | -------------------------------------------------------------- | ------------------- | ------------------ | ----------------- | | **DrizzleTokenStore** (TS) / **SQLAlchemyTokenStore** (Python) | 已有关系型 DB 的项目 | 零配置、事务安全、自带 schema | 需要 DB 连接 | | **Redis** | 高并发 SaaS、Session 管理 | 高性能、天然 TTL、原子操作 | 需要 Redis 实例、持久化配置 | | **自定义 DB**（Prisma / Mongoose / etc.） | 已有非 Drizzle ORM | 完全掌控 schema | 需要自行实现 CRUD | | **加密文件** | CLI 工具、单用户脚本 | 无外部依赖 | 不适合多用户、无并发控制 | 80% 的场景直接用 Contrib 的 DrizzleTokenStore 或 SQLAlchemyTokenStore 即可。只有特殊需求（Redis Session、自定义加密）才需要自行实现。 ### 使用 Contrib 存储 ```typescript TypeScript (Drizzle) theme={null} import { ProfyApp } from "@profy-ai/sdk"; import { DrizzleTokenStore } from "@profy-ai/sdk/contrib/drizzle"; import { drizzle } from "drizzle-orm/postgres-js"; import postgres from "postgres"; const db = drizzle(postgres(process.env.DATABASE_URL!)); const tokenStore = new DrizzleTokenStore(db); const profy = new ProfyApp({ clientId: process.env.PROFY_APP_ID!, clientSecret: process.env.PROFY_APP_SECRET!, onTokenRefresh: async (userId, newToken) => { await tokenStore.save(userId, newToken); }, }); ``` ```python Python (SQLAlchemy) theme={null} from profy_sdk import ProfyApp from profy_sdk.contrib.sqlalchemy import SQLAlchemyTokenStore from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine(os.environ["DATABASE_URL"]) token_store = SQLAlchemyTokenStore(engine) profy = ProfyApp( client_id=os.environ["PROFY_APP_ID"], client_secret=os.environ["PROFY_APP_SECRET"], on_token_refresh=lambda user_id, new_token: token_store.save(user_id, new_token), ) ``` ### 使用 Redis 自定义存储 ```typescript TypeScript theme={null} import { ProfyApp, type OAuthToken } from "@profy-ai/sdk"; import Redis from "ioredis"; const redis = new Redis(process.env.REDIS_URL!); const TOKEN_PREFIX = "profy:token:"; const TOKEN_TTL = 90 * 24 * 60 * 60; // 90 days async function saveToken(userId: string, token: OAuthToken) { await redis.set( `${TOKEN_PREFIX}${userId}`, JSON.stringify(token), "EX", TOKEN_TTL ); } async function getToken(userId: string): Promise { const raw = await redis.get(`${TOKEN_PREFIX}${userId}`); return raw ? JSON.parse(raw) : null; } async function deleteToken(userId: string) { await redis.del(`${TOKEN_PREFIX}${userId}`); } const profy = new ProfyApp({ clientId: process.env.PROFY_APP_ID!, clientSecret: process.env.PROFY_APP_SECRET!, onTokenRefresh: saveToken, }); ``` ```python Python theme={null} import json import os from redis.asyncio import Redis from profy_sdk import ProfyApp redis = Redis.from_url(os.environ["REDIS_URL"]) TOKEN_PREFIX = "profy:token:" TOKEN_TTL = 90 * 24 * 60 * 60 async def save_token(user_id: str, token: dict): await redis.set( f"{TOKEN_PREFIX}{user_id}", json.dumps(token), ex=TOKEN_TTL, ) async def get_token(user_id: str) -> dict | None: raw = await redis.get(f"{TOKEN_PREFIX}{user_id}") return json.loads(raw) if raw else None async def delete_token(user_id: str): await redis.delete(f"{TOKEN_PREFIX}{user_id}") profy = ProfyApp( client_id=os.environ["PROFY_APP_ID"], client_secret=os.environ["PROFY_APP_SECRET"], on_token_refresh=save_token, ) ``` ## Step 2: 多用户 Token 管理 SaaS 应用通常为每个用户维护独立的 Token。核心模式：**以 userId 为键，隔离读写**。 ```typescript TypeScript theme={null} import { isTokenExpired, AuthExpired } from "@profy-ai/sdk"; class TokenManager { constructor( private profy: ProfyApp, private store: { get: (userId: string) => Promise; save: (userId: string, token: OAuthToken) => Promise; delete: (userId: string) => Promise; } ) {} async getValidToken(userId: string): Promise { const stored = await this.store.get(userId); if (!stored) { throw new AuthExpired("No token found for user"); } if (!isTokenExpired(stored)) { return stored.accessToken; } const refreshed = await this.profy.refreshToken(stored.refreshToken); await this.store.save(userId, refreshed); return refreshed.accessToken; } async revokeUser(userId: string) { const stored = await this.store.get(userId); if (stored) { await this.profy.revokeToken(stored.accessToken); await this.store.delete(userId); } } } ``` ```python Python theme={null} from profy_sdk import ProfyApp, AuthExpired class TokenManager: def __init__(self, profy: ProfyApp, store): self.profy = profy self.store = store async def get_valid_token(self, user_id: str) -> str: stored = await self.store.get(user_id) if not stored: raise AuthExpired("No token found for user") if not stored.is_expired: return stored.access_token refreshed = await self.profy.refresh_token(stored.refresh_token) await self.store.save(user_id, refreshed) return refreshed.access_token async def revoke_user(self, user_id: str): stored = await self.store.get(user_id) if stored: await self.profy.revoke_token(stored.access_token) await self.store.delete(user_id) ``` ## Step 3: 并发刷新竞争 **问题**：两个请求同时发现 Access Token 过期，都尝试刷新。第一个成功，第二个因 Refresh Token 已轮换而收到 `invalid_grant`。 ```mermaid theme={null} sequenceDiagram participant R1 as 请求 A participant R2 as 请求 B participant App as 你的服务 participant Profy as Profy OAuth R1->>App: API 调用（Token 过期） R2->>App: API 调用（Token 过期） App->>Profy: refreshToken(old_refresh) [请求 A] App->>Profy: refreshToken(old_refresh) [请求 B] Profy->>App: ✅ 新 Token [请求 A] Profy->>App: ❌ invalid_grant [请求 B] ``` **解决方案**：用互斥锁保证同一用户同一时刻只有一个刷新操作。 ```typescript TypeScript theme={null} import { isTokenExpired } from "@profy-ai/sdk"; class SafeTokenManager { private refreshLocks = new Map>(); constructor( private profy: ProfyApp, private store: { get: (userId: string) => Promise; save: (userId: string, token: OAuthToken) => Promise; } ) {} async getValidToken(userId: string): Promise { const stored = await this.store.get(userId); if (!stored) throw new Error("No token found"); if (!isTokenExpired(stored)) { return stored.accessToken; } const refreshed = await this.refreshWithLock(userId, stored.refreshToken); return refreshed.accessToken; } private async refreshWithLock( userId: string, refreshToken: string ): Promise { const existing = this.refreshLocks.get(userId); if (existing) return existing; const promise = this.doRefresh(userId, refreshToken).finally(() => { this.refreshLocks.delete(userId); }); this.refreshLocks.set(userId, promise); return promise; } private async doRefresh( userId: string, refreshToken: string ): Promise { const refreshed = await this.profy.refreshToken(refreshToken); await this.store.save(userId, refreshed); return refreshed; } } ``` ```python Python theme={null} import asyncio from profy_sdk import ProfyApp class SafeTokenManager: def __init__(self, profy: ProfyApp, store): self.profy = profy self.store = store self._refresh_locks: dict[str, asyncio.Lock] = {} self._refresh_tasks: dict[str, asyncio.Task] = {} def _get_lock(self, user_id: str) -> asyncio.Lock: if user_id not in self._refresh_locks: self._refresh_locks[user_id] = asyncio.Lock() return self._refresh_locks[user_id] async def get_valid_token(self, user_id: str) -> str: stored = await self.store.get(user_id) if not stored: raise Exception("No token found") if not stored.is_expired: return stored.access_token return await self._refresh_with_lock(user_id, stored.refresh_token) async def _refresh_with_lock(self, user_id: str, refresh_token: str) -> str: lock = self._get_lock(user_id) async with lock: fresh = await self.store.get(user_id) if fresh and not fresh.is_expired: return fresh.access_token refreshed = await self.profy.refresh_token(refresh_token) await self.store.save(user_id, refreshed) return refreshed.access_token ``` Python 版本在获取锁后重新检查 Token 是否已被其他协程刷新（double-check pattern），避免重复刷新。TypeScript 版本通过共享同一个 Promise 实现相同效果。 对于多实例部署（多 Pod / 多进程），进程内的 `Map` / `dict` 锁不够。需要分布式锁： ```typescript Redis 分布式锁 theme={null} import Redis from "ioredis"; const redis = new Redis(process.env.REDIS_URL!); const LOCK_TTL = 10; // seconds async function refreshWithDistributedLock( userId: string, refreshToken: string ): Promise { const lockKey = `profy:refresh_lock:${userId}`; const acquired = await redis.set(lockKey, "1", "EX", LOCK_TTL, "NX"); if (!acquired) { await new Promise((r) => setTimeout(r, 1000)); const fresh = await getToken(userId); if (fresh && !isTokenExpired(fresh)) return fresh; throw new Error("Refresh in progress by another instance"); } try { const refreshed = await profy.refreshToken(refreshToken); await saveToken(userId, refreshed); return refreshed; } finally { await redis.del(lockKey); } } ``` ```python Redis 分布式锁 theme={null} import asyncio import redis.asyncio as aioredis redis_client = aioredis.from_url(os.environ["REDIS_URL"]) LOCK_TTL = 10 # seconds async def refresh_with_distributed_lock( user_id: str, refresh_token: str ) -> dict: lock_key = f"profy:refresh_lock:{user_id}" acquired = await redis_client.set(lock_key, "1", ex=LOCK_TTL, nx=True) if not acquired: await asyncio.sleep(1) fresh = await get_token(user_id) if fresh and not is_token_expired(fresh): return fresh raise Exception("Refresh in progress by another instance") try: refreshed = await profy.refresh_token(refresh_token) await save_token(user_id, refreshed) return refreshed finally: await redis_client.delete(lock_key) ``` ## Step 4: Token 过期降级 当 Refresh Token 也过期（90 天未使用）或被撤销时，SDK 抛出 `AuthExpired`。此时唯一的恢复路径是引导用户重新授权。 ```typescript TypeScript — API Route theme={null} import { AuthExpired } from "@profy-ai/sdk"; export async function POST(request: NextRequest) { try { const token = await tokenManager.getValidToken(userId); const result = await profy.reportEvent("generate", { token }); return NextResponse.json(result); } catch (error) { if (error instanceof AuthExpired) { await tokenManager.revokeUser(userId); return NextResponse.json( { error: "auth_expired", reauthorizeUrl: buildAuthUrl(), message: "授权已过期，请重新连接 Profy 账号", }, { status: 401 } ); } throw error; } } ``` ```python Python — FastAPI theme={null} from profy_sdk import AuthExpired from fastapi import HTTPException @app.post("/api/report") async def report_event(user_id: str, event_name: str): try: token = await token_manager.get_valid_token(user_id) result = await profy.report_event(event_name, token=token) return result except AuthExpired: await token_manager.revoke_user(user_id) raise HTTPException( status_code=401, detail={ "error": "auth_expired", "reauthorize_url": build_auth_url(), "message": "授权已过期，请重新连接 Profy 账号", }, ) ``` 前端处理 `auth_expired` 响应： ```tsx 前端降级组件 theme={null} "use client"; import { useState } from "react"; function ReauthorizePrompt({ authUrl }: { authUrl: string }) { return (

你的 Profy 授权已过期，需要重新连接才能继续使用。

重新连接 Profy
); } ``` ## Step 5: Token 撤销 用户主动断开连接（取消授权）或你需要清理不再使用的 Token 时，调用撤销接口。 ```typescript TypeScript theme={null} async function disconnectUser(userId: string) { const stored = await tokenStore.get(userId); if (!stored) return; await profy.revokeToken(stored.accessToken); await tokenStore.delete(userId); } ``` ```python Python theme={null} async def disconnect_user(user_id: str): stored = await token_store.get(user_id) if not stored: return await profy.revoke_token(stored.access_token) await token_store.delete(user_id) ``` 撤销后，该用户的 Access Token 和 Refresh Token **立即失效**。后续 API 调用会收到 401。确保在撤销前已完成所有进行中的请求。 撤销的典型触发场景： * 用户在你的 App 设置页点击「断开 Profy 连接」 * 用户删除账号 * 管理员批量清理不活跃授权 * 检测到异常访问模式 ## Step 6: 安全存储 ### 静态加密 Token 等同于用户凭证，在数据库中应加密存储： ```typescript TypeScript theme={null} import { createCipheriv, createDecipheriv, randomBytes } from "crypto"; const ALGORITHM = "aes-256-gcm"; const KEY = Buffer.from(process.env.TOKEN_ENCRYPTION_KEY!, "hex"); // 32 bytes function encrypt(plaintext: string): string { const iv = randomBytes(16); const cipher = createCipheriv(ALGORITHM, KEY, iv); const encrypted = Buffer.concat([ cipher.update(plaintext, "utf8"), cipher.final(), ]); const tag = cipher.getAuthTag(); return [iv.toString("hex"), encrypted.toString("hex"), tag.toString("hex")].join(":"); } function decrypt(ciphertext: string): string { const [ivHex, encHex, tagHex] = ciphertext.split(":"); const decipher = createDecipheriv( ALGORITHM, KEY, Buffer.from(ivHex, "hex") ); decipher.setAuthTag(Buffer.from(tagHex, "hex")); return decipher.update(encHex, "hex", "utf8") + decipher.final("utf8"); } ``` ```python Python theme={null} import os from cryptography.fernet import Fernet KEY = os.environ["TOKEN_ENCRYPTION_KEY"].encode() fernet = Fernet(KEY) def encrypt(plaintext: str) -> str: return fernet.encrypt(plaintext.encode()).decode() def decrypt(ciphertext: str) -> str: return fernet.decrypt(ciphertext.encode()).decode() ``` ### 安全清单 | 措施 | 说明 | | --------------- | ------------------------------------------------------------ | | 加密存储 | Token 在 DB / Redis 中以密文存储 | | 禁止日志输出 | 日志中不打印 Token 值（用 `userId` 或 Token 前 8 位替代） | | HTTPS Only | Token 传输仅通过 HTTPS | | HttpOnly Cookie | 如果用 cookie 传递 Token 引用，设置 `HttpOnly` + `Secure` + `SameSite` | | 密钥轮换 | 加密密钥定期轮换，支持多密钥解密（新密钥加密、新旧密钥均可解密） | | 最小权限 | OAuth scope 只申请必要权限（如 `events:write`） | ## 反模式 以下是生产环境中常见的 Token 管理错误，每一条都曾导致真实的线上事故。 ### 1. 前端 localStorage 存储 Token ```typescript theme={null} // ❌ 任何 XSS 攻击都能窃取 Token localStorage.setItem("profy_token", JSON.stringify(token)); ``` Token 应该只存在于服务端。前端通过 HttpOnly Cookie 或 Session ID 引用。 ### 2. 忽略 onTokenRefresh 回调 ```typescript theme={null} // ❌ SDK 自动刷新后新 Token 丢失 const profy = new ProfyApp({ clientId: "...", clientSecret: "...", // onTokenRefresh 未设置 → 新 Token 没有持久化 }); ``` 刷新后旧 Refresh Token 已失效。如果新 Token 没有保存，下次刷新会失败，用户被迫重新授权。 ### 3. 使用已轮换的 Refresh Token ```typescript theme={null} // ❌ 缓存了旧 Token，刷新后仍用旧的 const cachedToken = memoryCache.get(userId); // 可能是过时的 await profy.refreshToken(cachedToken.refreshToken); // invalid_grant ``` 每次从持久化存储读取最新 Token，不要依赖内存缓存。 ### 4. 不处理并发刷新 ```typescript theme={null} // ❌ 两个并发请求各自刷新 → 第二个一定失败 async function getToken(userId: string) { const stored = await db.getToken(userId); if (isTokenExpired(stored)) { return await profy.refreshToken(stored.refreshToken); // 无锁 } return stored; } ``` 参照 Step 3 的互斥锁模式。 ## 生产环境清单 上线前逐项确认： * [ ] `onTokenRefresh` 已配置且写入持久化存储 * [ ] Token 存储于服务端（DB / Redis），不在前端 * [ ] Token 在存储层加密 * [ ] 并发刷新有互斥锁保护 * [ ] `AuthExpired` 有降级处理（引导重新授权） * [ ] 用户断开连接时调用 `revokeToken()` 清理 * [ ] 日志不包含完整 Token 值 * [ ] OAuth scope 遵循最小权限 * [ ] 多实例部署使用分布式锁 * [ ] 加密密钥不硬编码在代码中 ## 下一步 从零构建完整的 OAuth + 计费 Next.js 应用 Python SDK + SQLAlchemy Token 持久化 接收并处理 Profy 平台推送的实时事件 双语言 API 详解与 Token 持久化