Token 管理最佳实践

你将构建什么

一个生产级的 Token 管理层——覆盖持久化存储选型、多用户隔离、并发刷新竞争、过期降级、主动撤销和安全存储，确保你的 SaaS 应用在 Token 轮换和服务重启后仍然正常工作。

Token 生命周期

Token	有效期	轮换规则
Access Token	1 小时（JWT RS256）	每次刷新签发新的
Refresh Token	90 天	每次使用即失效，同时签发新 Refresh Token

核心原则

违反以下任何一条，都会导致用户静默掉线——没有错误提示，只有突然的「请重新登录」。

1. 刷新后必须持久化 — report_event 的 on_refresh 回调（或手动调用 refresh() 后）是你保存新 Token 的唯一时机。错过它，旧 Refresh Token 已失效，新 Token 丢失，用户必须重新授权。 2. 禁止纯内存存储 — 服务重启 = 所有用户 Token 丢失 = 全体重新登录。Token 必须落盘（DB / Redis / 文件）。 3. 轮换必须原子化 — 每次 refresh() 调用后，旧的 Access Token 和 Refresh Token 同时失效。保存新 Token 和废弃旧 Token 必须在同一事务中完成。

Step 1: 选择存储策略

方案	适用场景	优势	劣势
关系型 DB（Drizzle / Prisma / SQLAlchemy）	已有数据库的项目	事务安全、自带 schema	需要 DB 连接
Redis	高并发 SaaS、Session 管理	高性能、天然 TTL、原子操作	需要 Redis 实例
加密文件	CLI 工具、单用户脚本	无外部依赖	不适合多用户、无并发控制

ProfyApp 不内置 Token 存储——它只负责 OAuth 握手和 API 调用。存储是你的应用的职责，这保证了最大灵活性。

初始化 ProfyApp + 存储

ProfyApp 只持有 App 身份，不存储用户 Token。Token 在每次 API 调用时传入，刷新后通过 on_refresh 回调持久化。

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;

const app = new ProfyApp({
  clientId: process.env.PROFY_APP_ID!,
  clientSecret: process.env.PROFY_APP_SECRET!,
});

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<OAuthToken | null> {
  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}`);
}

import json
import os
from redis.asyncio import Redis
from profy import ProfyApp, OAuthToken

redis = Redis.from_url(os.environ["REDIS_URL"])
TOKEN_PREFIX = "profy:token:"
TOKEN_TTL = 90 * 24 * 60 * 60

app = ProfyApp(
    client_id=os.environ["PROFY_APP_ID"],
    client_secret=os.environ["PROFY_APP_SECRET"],
)

async def save_token(user_id: str, token: OAuthToken):
    data = {
        "access_token": token.access_token,
        "refresh_token": token.refresh_token,
        "token_type": token.token_type,
        "expires_at": token.expires_at,
        "scope": token.scope,
    }
    await redis.set(f"{TOKEN_PREFIX}{user_id}", json.dumps(data), ex=TOKEN_TTL)

async def get_token(user_id: str) -> OAuthToken | None:
    raw = await redis.get(f"{TOKEN_PREFIX}{user_id}")
    if not raw:
        return None
    data = json.loads(raw)
    return OAuthToken(**data)

async def delete_token(user_id: str):
    await redis.delete(f"{TOKEN_PREFIX}{user_id}")

Step 2: 多用户 Token 管理

SaaS 应用通常为每个用户维护独立的 Token。核心模式：以 userId 为键，隔离读写。

import { ProfyApp, isTokenExpired, AuthenticationError, type OAuthToken } from "@profy-ai/sdk";

class TokenManager {
  constructor(
    private app: ProfyApp,
    private store: {
      get: (userId: string) => Promise<OAuthToken | null>;
      save: (userId: string, token: OAuthToken) => Promise<void>;
      delete: (userId: string) => Promise<void>;
    }
  ) {}

  async getValidToken(userId: string): Promise<OAuthToken> {
    const stored = await this.store.get(userId);
    if (!stored) {
      throw new AuthenticationError(401, { error: "No token found for user" });
    }

    if (!isTokenExpired(stored)) {
      return stored;
    }

    const refreshed = await this.app.refresh(stored.refreshToken);
    await this.store.save(userId, refreshed);
    return refreshed;
  }

  async revokeUser(userId: string) {
    const stored = await this.store.get(userId);
    if (stored) {
      await this.app.revoke(stored.refreshToken);
      await this.store.delete(userId);
    }
  }
}

from profy import ProfyApp, AuthenticationError, OAuthToken

class TokenManager:
    def __init__(self, app: ProfyApp, store):
        self.app = app
        self.store = store

    async def get_valid_token(self, user_id: str) -> OAuthToken:
        stored = await self.store.get(user_id)
        if not stored:
            raise AuthenticationError(401, {"error": "No token found for user"})

        if not stored.is_expired():
            return stored

        refreshed = await self.app.refresh(stored.refresh_token)
        await self.store.save(user_id, refreshed)
        return refreshed

    async def revoke_user(self, user_id: str):
        stored = await self.store.get(user_id)
        if stored:
            await self.app.revoke(stored.refresh_token)
            await self.store.delete(user_id)

revoke() 接收的是 refresh token（不是 access token）。access token 是无状态 JWT，无法撤销；revoke 使 refresh token 失效从而阻止后续刷新。

Step 3: 并发刷新竞争

问题：两个请求同时发现 Access Token 过期，都尝试刷新。第一个成功，第二个因 Refresh Token 已轮换而收到 invalid_grant。 解决方案：用互斥锁保证同一用户同一时刻只有一个刷新操作。

import { ProfyApp, isTokenExpired, type OAuthToken } from "@profy-ai/sdk";

class SafeTokenManager {
  private refreshLocks = new Map<string, Promise<OAuthToken>>();

  constructor(
    private app: ProfyApp,
    private store: {
      get: (userId: string) => Promise<OAuthToken | null>;
      save: (userId: string, token: OAuthToken) => Promise<void>;
    }
  ) {}

  async getValidToken(userId: string): Promise<OAuthToken> {
    const stored = await this.store.get(userId);
    if (!stored) throw new Error("No token found");

    if (!isTokenExpired(stored)) return stored;

    return this.refreshWithLock(userId, stored.refreshToken);
  }

  private async refreshWithLock(userId: string, refreshToken: string): Promise<OAuthToken> {
    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<OAuthToken> {
    const refreshed = await this.app.refresh(refreshToken);
    await this.store.save(userId, refreshed);
    return refreshed;
  }
}

import asyncio
from profy import ProfyApp, OAuthToken

class SafeTokenManager:
    def __init__(self, app: ProfyApp, store):
        self.app = app
        self.store = store
        self._refresh_locks: dict[str, asyncio.Lock] = {}

    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) -> OAuthToken:
        stored = await self.store.get(user_id)
        if not stored:
            raise Exception("No token found")

        if not stored.is_expired():
            return stored

        return await self._refresh_with_lock(user_id, stored.refresh_token)

    async def _refresh_with_lock(self, user_id: str, refresh_token: str) -> OAuthToken:
        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

            refreshed = await self.app.refresh(refresh_token)
            await self.store.save(user_id, refreshed)
            return refreshed

Python 版本在获取锁后重新检查 Token 是否已被其他协程刷新（double-check pattern）。TypeScript 版本通过共享同一个 Promise 实现相同效果。

对于多实例部署（多 Pod / 多进程），进程内的 Map / dict 锁不够。需要分布式锁：

import Redis from "ioredis";
import { isTokenExpired, type OAuthToken } from "@profy-ai/sdk";

const redis = new Redis(process.env.REDIS_URL!);
const LOCK_TTL = 10;

async function refreshWithDistributedLock(userId: string, refreshToken: string): Promise<OAuthToken> {
  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 app.refresh(refreshToken);
    await saveToken(userId, refreshed);
    return refreshed;
  } finally {
    await redis.del(lockKey);
  }
}

import asyncio
import redis.asyncio as aioredis
from profy import OAuthToken

redis_client = aioredis.from_url(os.environ["REDIS_URL"])
LOCK_TTL = 10

async def refresh_with_distributed_lock(user_id: str, refresh_token: str) -> OAuthToken:
    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 fresh.is_expired():
            return fresh
        raise Exception("Refresh in progress by another instance")

    try:
        refreshed = await app.refresh(refresh_token)
        await save_token(user_id, refreshed)
        return refreshed
    finally:
        await redis_client.delete(lock_key)

Step 4: Token 过期降级

当 Refresh Token 也过期（90 天未使用）或被撤销时，refresh() 抛出 AuthenticationError（401）。此时唯一的恢复路径是引导用户重新授权。

import { AuthenticationError } from "@profy-ai/sdk";

export async function POST(request: NextRequest) {
  try {
    const token = await tokenManager.getValidToken(userId);
    const result = await app.reportEvent("generate", {
      token,
      onRefresh: (rotated) => tokenManager.store.save(userId, rotated),
    });
    return NextResponse.json(result);
  } catch (error) {
    if (error instanceof AuthenticationError) {
      await tokenManager.revokeUser(userId);

      return NextResponse.json(
        {
          error: "auth_expired",
          reauthorizeUrl: app.authorizationUrl({ redirectUri: REDIRECT_URI }),
          message: "授权已过期，请重新连接 Profy 账号",
        },
        { status: 401 }
      );
    }
    throw error;
  }
}

from profy import AuthenticationError
from fastapi import HTTPException

@router.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 app.report_event(
            event_name,
            token=token,
            on_refresh=lambda rotated: save_token(user_id, rotated),
        )
        return result
    except AuthenticationError:
        await token_manager.revoke_user(user_id)

        raise HTTPException(
            status_code=401,
            detail={
                "error": "auth_expired",
                "reauthorize_url": app.authorization_url(redirect_uri=REDIRECT_URI),
                "message": "授权已过期，请重新连接 Profy 账号",
            },
        )

前端处理 auth_expired 响应：

前端降级组件

"use client";

function ReauthorizePrompt({ authUrl }: { authUrl: string }) {
  return (
    <div className="rounded-lg border border-amber-200 bg-amber-50 p-4">
      <p className="text-sm text-amber-800">
        你的 Profy 授权已过期，需要重新连接才能继续使用。
      </p>
      <a
        href={authUrl}
        className="mt-2 inline-block rounded bg-amber-600 px-4 py-2 text-sm text-white"
      >
        重新连接 Profy
      </a>
    </div>
  );
}

Step 5: Token 撤销

用户主动断开连接（取消授权）或你需要清理不再使用的 Token 时，调用撤销接口。

async function disconnectUser(userId: string) {
  const stored = await getToken(userId);
  if (!stored) return;

  await app.revoke(stored.refreshToken);
  await deleteToken(userId);
}

async def disconnect_user(user_id: str):
    stored = await get_token(user_id)
    if not stored:
        return

    await app.revoke(stored.refresh_token)
    await delete_token(user_id)

revoke() 接收 refresh token。撤销后，该用户的 Access Token（1 小时内自然过期）和 Refresh Token 立即失效。后续 API 调用会收到 401。确保在撤销前已完成所有进行中的请求。

撤销的典型触发场景：

用户在你的 App 设置页点击「断开 Profy 连接」
用户删除账号
管理员批量清理不活跃授权
检测到异常访问模式

Step 6: 安全存储

静态加密

Token 等同于用户凭证，在数据库中应加密存储：

import { createCipheriv, createDecipheriv, randomBytes } from "crypto";

const ALGORITHM = "aes-256-gcm";
const KEY = Buffer.from(process.env.TOKEN_ENCRYPTION_KEY!, "hex");

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");
}

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

// ❌ 任何 XSS 攻击都能窃取 Token
localStorage.setItem("profy_token", JSON.stringify(token));

Token 应该只存在于服务端。前端通过 HttpOnly Cookie 或 Session ID 引用。

2. 忽略 on_refresh / 手动刷新后不持久化

// ❌ 刷新后新 Token 丢失
const refreshed = await app.refresh(oldToken.refreshToken);
// 没有 saveToken(userId, refreshed) → 下次刷新 invalid_grant

刷新后旧 Refresh Token 已失效。如果新 Token 没有保存，下次刷新会失败，用户被迫重新授权。

3. 使用已轮换的 Refresh Token

// ❌ 缓存了旧 Token，刷新后仍用旧的
const cachedToken = memoryCache.get(userId);
await app.refresh(cachedToken.refreshToken); // invalid_grant

每次从持久化存储读取最新 Token，不要依赖内存缓存。

4. 不处理并发刷新

// ❌ 两个并发请求各自刷新 → 第二个一定失败
async function getToken(userId: string) {
  const stored = await db.getToken(userId);
  if (isTokenExpired(stored)) {
    return await app.refresh(stored.refreshToken); // 无锁
  }
  return stored;
}

参照 Step 3 的互斥锁模式。

生产环境清单

上线前逐项确认：

下一步

按次计费实战

使用 ProfyApp report_event 实现按次扣费

Next.js 全栈集成

从零构建完整的 OAuth + 计费 Next.js 应用

Python FastAPI 集成

Python SDK + Token 持久化

SDK 完整指南

双语言 API 详解

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

Token 管理最佳实践

你将构建什么

Token 生命周期

核心原则

Step 1: 选择存储策略

初始化 ProfyApp + 存储

Step 2: 多用户 Token 管理

Step 3: 并发刷新竞争

Step 4: Token 过期降级

Step 5: Token 撤销

Step 6: 安全存储

静态加密

安全清单

反模式

1. 前端 localStorage 存储 Token

2. 忽略 on_refresh / 手动刷新后不持久化

3. 使用已轮换的 Refresh Token

4. 不处理并发刷新

生产环境清单

下一步

按次计费实战

Next.js 全栈集成

Python FastAPI 集成

SDK 完整指南

​你将构建什么

​Token 生命周期

​核心原则

​Step 1: 选择存储策略

​初始化 ProfyApp + 存储

​Step 2: 多用户 Token 管理

​Step 3: 并发刷新竞争

​Step 4: Token 过期降级

​Step 5: Token 撤销

​Step 6: 安全存储

​静态加密

​安全清单

​反模式

​1. 前端 localStorage 存储 Token

​2. 忽略 on_refresh / 手动刷新后不持久化

​3. 使用已轮换的 Refresh Token

​4. 不处理并发刷新

​生产环境清单

​下一步

按次计费实战

Next.js 全栈集成

Python FastAPI 集成

SDK 完整指南

你将构建什么

Token 生命周期

核心原则

Step 1: 选择存储策略

初始化 ProfyApp + 存储

Step 2: 多用户 Token 管理

Step 3: 并发刷新竞争

Step 4: Token 过期降级

Step 5: Token 撤销

Step 6: 安全存储

静态加密

安全清单

反模式

1. 前端 localStorage 存储 Token

2. 忽略 on_refresh / 手动刷新后不持久化

3. 使用已轮换的 Refresh Token

4. 不处理并发刷新

生产环境清单

下一步