这是安全领域最常见的混淆。一句话区分：
你是谁？
验证用户身份的过程。密码、指纹、OAuth 登录都属于认证。
英文词根: Greek authentikos — "genuine, principal"
你能做什么？
确定已认证用户的权限范围。RBAC、ABAC、权限检查都属于授权。
英文词根: Latin auctoritas — "authority, power"
认证之后，用户后续请求怎么证明"我已登录"？三种主流方案：
# Express + express-session (内存存储示例)
const session = require('express-session');
app.use(session({
secret: process.env.SESSION_SECRET,
resave: false,
saveUninitialized: false,
cookie: {
httpOnly: true, // JS 无法读取，防 XSS
secure: true, // 仅 HTTPS
sameSite: 'lax', // 防 CSRF
maxAge: 24 * 60 * 60 * 1000 // 24 小时
},
store: new RedisStore({ client: redisClient }) // 生产用 Redis
}));
app.post('/login', async (req, res) => {
const { email, password } = req.body;
const user = await User.findByEmail(email);
if (!user || !bcrypt.compareSync(password, user.passwordHash)) {
return res.status(401).json({ error: 'Invalid credentials' });
}
req.session.userId = user.id;
req.session.role = user.role;
res.json({ message: 'Logged in' });
});
// 强制登出 — 服务端直接销毁
app.post('/logout', (req, res) => {
req.session.destroy();
res.clearCookie('connect.sid');
res.json({ message: 'Logged out' });
});✅ 服务端完全控制，可随时吊销
✅ Cookie 自动携带，前端无需额外处理
✅ 不暴露用户信息在客户端
✅ 对传统 Web 应用最简单
❌ 需要服务端存储（内存/Redis/DB）
❌ 分布式部署需要共享 Session 存储
❌ 跨域/移动端不友好
❌ 水平扩展有状态瓶颈
JWT 由三段 Base64URL 编码组成：Header.Payload.Signature
# FastAPI + PyJWT 完整实现
import jwt
from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
JWT_SECRET = "your-256-bit-secret" # 生产环境用环境变量!
JWT_ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=15)
REFRESH_TOKEN_EXPIRE = timedelta(days=7)
security = HTTPBearer()
def create_access_token(data: dict) -> str:
"""签发短期 Access Token"""
to_encode = data.copy()
to_encode.update({
"exp": datetime.now(timezone.utc) + ACCESS_TOKEN_EXPIRE,
"iat": datetime.now(timezone.utc),
"type": "access"
})
return jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM)
def create_refresh_token(data: dict) -> str:
"""签发长期 Refresh Token"""
to_encode = data.copy()
to_encode.update({
"exp": datetime.now(timezone.utc) + REFRESH_TOKEN_EXPIRE,
"iat": datetime.now(timezone.utc),
"type": "refresh",
"jti": str(uuid4()) # 唯一 ID，用于吊销
})
return jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM)
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
) -> User:
"""从 JWT 提取当前用户"""
token = credentials.credentials
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
if payload.get("type") != "access":
raise HTTPException(401, "Invalid token type")
# 检查是否在黑名单中（可选，见下方吊销方案）
if await is_token_revoked(payload.get("jti")):
raise HTTPException(401, "Token revoked")
user = await User.get(payload["sub"])
if not user:
raise HTTPException(401, "User not found")
return user
except jwt.ExpiredSignatureError:
raise HTTPException(401, "Token expired")
except jwt.InvalidTokenError:
raise HTTPException(401, "Invalid token")// 前端 Token 管理 (React 示例)
class TokenManager {
private refreshPromise: Promise<string> | null = null;
async getValidToken(): Promise<string> {
const token = localStorage.getItem('access_token');
if (!token) throw new Error('Not authenticated');
// 检查是否即将过期（提前 30 秒刷新）
const payload = JSON.parse(atob(token.split('.')[1]));
if (payload.exp * 1000 - Date.now() < 30_000) {
return this.refreshAccessToken();
}
return token;
}
async refreshAccessToken(): Promise<string> {
// 防并发：多个请求同时发现 token 过期，只刷新一次
if (this.refreshPromise) return this.refreshPromise;
this.refreshPromise = (async () => {
try {
const refreshToken = localStorage.getItem('refresh_token');
const res = await fetch('/api/auth/refresh', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ refresh_token: refreshToken })
});
if (!res.ok) {
this.logout();
throw new Error('Refresh failed');
}
const data = await res.json();
localStorage.setItem('access_token', data.access_token);
if (data.refresh_token) {
localStorage.setItem('refresh_token', data.refresh_token);
}
return data.access_token;
} finally {
this.refreshPromise = null;
}
})();
return this.refreshPromise;
}
logout() {
localStorage.removeItem('access_token');
localStorage.removeItem('refresh_token');
window.location.href = '/login';
}
}✅ 无状态，服务端不需要存储
✅ 天然支持分布式/微服务
✅ 跨域友好，移动端/SPA 首选
✅ 可携带自定义 claims（角色、租户 ID）
❌ 签发后无法修改，吊销需要额外机制
❌ Token 体积大（Header+Payload+Sig）
❌ Payload 明文可读（不能放敏感信息）
❌ 续期策略复杂（双 Token 方案）
JWT 最大的争议就是"签了就不能撤"。实际生产中有几种方案：
| 方案 | 原理 | 适用场景 | 性能影响 |
|---|---|---|---|
| 短期 Access + 长期 Refresh | Access Token 15 分钟过期，Refresh Token 可服务端吊销 | 大多数 SaaS | 低 |
| Token 黑名单 (Redis) | 吊销时将 jti 写入 Redis，TTL = Token 剩余有效期 | 需要即时吊销 | 中 |
| Token 版本号 | User 表加 token_version 字段，签发时写入，验证时对比 | 全量吊销（改密码后） | 低 |
| OPA/Gatekeeper | 策略引擎实时判断 Token 是否有效 | 大型微服务 | 高 |
# Redis Token 黑名单实现 (Python)
import redis
redis_client = redis.Redis()
async def revoke_token(jti: str, exp: int) -> None:
"""将 Token 加入黑名单，TTL 设为剩余有效期"""
now = int(datetime.now(timezone.utc).timestamp())
ttl = max(exp - now, 0)
if ttl > 0:
await redis_client.setex(f"revoked:{jti}", ttl, "1")
async def is_token_revoked(jti: str | None) -> bool:
if not jti:
return False
return bool(await redis_client.exists(f"revoked:{jti}"))| 维度 | Session-Cookie | JWT (双 Token) | Opaque Token |
|---|---|---|---|
| 状态 | 有状态 | 无状态（+ 黑名单可选） | 有状态 |
| 存储 | Redis / DB | 客户端 | Redis / DB |
| 吊销 | 即时（删 Session） | 需要额外机制 | 即时（删记录） |
| 跨域 | 需额外配置 | 天然支持 | 需额外配置 |
| 体积 | 小（只传 Session ID） | 大（1-4KB） | 小（UUID） |
| 移动端 | 不友好 | 友好 | 友好 |
| 微服务 | 需共享存储 | 天然支持 | 需共享存储 |
| 安全敏感操作 | ✅ 最佳 | ⚠️ 需黑名单 | ✅ 好 |
OAuth2 是授权框架，OpenID Connect (OIDC) 是在 OAuth2 之上加了一层身份认证。90% 的场景你用的是 OIDC，只是习惯叫 OAuth2 登录。
| 模式 | 流程 | 安全性 | 适用场景 |
|---|---|---|---|
| Authorization Code | 重定向 → 授权码 → 换 Token | 高 | 服务端应用（最常用） |
| Authorization Code + PKCE | 加 code_verifier/challenge 防截获 | 最高 | SPA / 移动端（当前推荐） |
| Client Credentials | 直接用 client_id+secret 换 Token | 中 | 服务间调用（M2M） |
| Implicit (已废弃) | Token 直接返回在 URL fragment | 低 | ⚠️ 已被 PKCE 取代 |
| Resource Owner Password | 直接传用户名密码换 Token | 低 | ⚠️ 仅遗留系统 |
// Next.js App Router — Google OAuth2 + PKCE 登录
import { NextRequest, NextResponse } from 'next/server';
import crypto from 'crypto';
// 步骤 1: 发起 OAuth2 授权请求
export async function GET(req: NextRequest) {
const state = crypto.randomBytes(16).toString('hex');
const codeVerifier = crypto.randomBytes(32).toString('base64url');
const codeChallenge = crypto
.createHash('sha256')
.update(codeVerifier)
.digest('base64url');
// 存入 cookie（HttpOnly），后续回调时取出
const res = NextResponse.redirect(
`https://accounts.google.com/o/oauth2/v2/auth?` + new URLSearchParams({
client_id: process.env.GOOGLE_CLIENT_ID!,
redirect_uri: `${process.env.APP_URL}/api/auth/callback`,
response_type: 'code',
scope: 'openid email profile',
state,
code_challenge: codeChallenge,
code_challenge_method: 'S256',
})
);
res.cookies.set('oauth_state', state, { httpOnly: true, maxAge: 600 });
res.cookies.set('code_verifier', codeVerifier, { httpOnly: true, maxAge: 600 });
return res;
}
// 步骤 2: 处理回调，换 Token
export async function POST(req: NextRequest) {
const { code, state } = await req.json();
const savedState = req.cookies.get('oauth_state')?.value;
const codeVerifier = req.cookies.get('code_verifier')?.value;
if (state !== savedState) {
return NextResponse.json({ error: 'State mismatch' }, { status: 403 });
}
// 换 Token
const tokenRes = await fetch('https://oauth2.googleapis.com/token', {
method: 'POST',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({
code,
client_id: process.env.GOOGLE_CLIENT_ID!,
client_secret: process.env.GOOGLE_CLIENT_SECRET!,
redirect_uri: `${process.env.APP_URL}/api/auth/callback`,
grant_type: 'authorization_code',
code_verifier: codeVerifier!, // PKCE 关键
}),
});
const { id_token, access_token } = await tokenRes.json();
// 解码 ID Token (OIDC) 获取用户信息
const claims = JSON.parse(atob(id_token.split('.')[1]));
// { sub, email, name, picture, ... }
// 查找或创建本地用户
let user = await User.findByOAuth('google', claims.sub);
if (!user) {
user = await User.createFromOAuth({
provider: 'google',
providerId: claims.sub,
email: claims.email,
name: claims.name,
avatar: claims.picture,
});
}
// 签发自己的 JWT
const appToken = createAccessToken({ sub: user.id, role: user.role });
return NextResponse.json({ token: appToken, user });
}JWT 格式，包含用户身份信息（sub, email, name）。
用途：客户端知道"这是谁"
验证：必须验签
消费者：客户端应用
可以是 JWT 或 Opaque，用于调用资源服务器 API。
用途：访问受保护资源
验证：资源服务器验签或内省
消费者：资源服务器（API）
| 提供商 | 授权端点 | Token 端点 | Scope 示例 |
|---|---|---|---|
| accounts.google.com/o/oauth2/v2/auth | oauth2.googleapis.com/token | openid email profile | |
| GitHub | github.com/login/oauth/authorize | github.com/login/oauth/access_token | read:user user:email |
| Microsoft | login.microsoftonline.com/{tenant}/oauth2/v2.0/authorize | login.microsoftonline.com/{tenant}/oauth2/v2.0/token | openid profile User.Read |
| 微信 | open.weixin.qq.com/connect/qrconnect | api.weixin.qq.com/sns/oauth2/access_token | snsapi_login |
| Apple | appleid.apple.com/auth/authorize | appleid.apple.com/auth/token | email name |
企业客户的刚需：登录一次，访问所有子系统。实现 SSO 的核心协议：
企业级 SSO 的事实标准。XML-based，2005 年发布，至今仍是大型企业首选。
# Python SAML SP (Service Provider) 配置示例 (python3-saml)
# settings.json
{
"strict": true,
"debug": false,
"sp": {
"entityId": "https://your-app.com/metadata",
"assertionConsumerService": {
"url": "https://your-app.com/saml/acs",
"binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
},
"singleLogoutService": {
"url": "https://your-app.com/saml/sls",
"binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
},
"NameIDFormat": "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress",
"x509cert": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----",
"privateKey": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
},
"idp": {
"entityId": "https://idp.example.com/metadata",
"singleSignOnService": {
"url": "https://idp.example.com/sso",
"binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
},
"singleLogoutService": {
"url": "https://idp.example.com/slo",
"binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
},
"x509cert": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
}
}现代 SSO 首选协议。JSON-based，比 SAML 轻量，与 OAuth2 无缝衔接。
| 维度 | SAML 2.0 | OIDC |
|---|---|---|
| 数据格式 | XML | JSON (JWT) |
| 协议复杂度 | 高 | 中 |
| 浏览器 POST 绑定 | ✅ 原生支持 | 通过 redirect |
| API 友好 | ❌ 不适合 | ✅ 天然支持 |
| 企业接受度 | ✅ 极高 | 📈 快速增长 |
| 单点登出 | ✅ 原生支持 | ⚠️ 需要前端通道 |
| 移动端 | ❌ 不友好 | ✅ 友好 |
| 典型客户 | 金融/政府/大型企业 | 科技公司/SaaS |
认证解决"你是谁"，授权解决"你能做什么"。三种主流授权模型：
最经典、最广泛使用的授权模型。用户 → 角色 → 权限。
# FastAPI + SQLAlchemy RBAC 实现
from enum import Enum
from typing import Optional
from functools import wraps
class Role(str, Enum):
SUPER_ADMIN = "super_admin"
ADMIN = "admin"
EDITOR = "editor"
VIEWER = "viewer"
# 权限定义 (角色 → 权限集合)
ROLE_PERMISSIONS: dict[Role, set[str]] = {
Role.SUPER_ADMIN: {"*"}, # 所有权限
Role.ADMIN: {
"users.read", "users.write", "users.delete",
"projects.read", "projects.write", "projects.delete",
"settings.read", "settings.write",
},
Role.EDITOR: {
"projects.read", "projects.write",
"settings.read",
},
Role.VIEWER: {
"projects.read",
"settings.read",
},
}
def has_permission(user: User, permission: str) -> bool:
"""检查用户是否有某权限"""
perms = ROLE_PERMISSIONS.get(user.role, set())
return "*" in perms or permission in perms
def require_permission(permission: str):
"""装饰器：要求特定权限"""
def decorator(func):
@wraps(func)
async def wrapper(*args, current_user: User = Depends(get_current_user), **kwargs):
if not has_permission(current_user, permission):
raise HTTPException(
status_code=403,
detail=f"Requires permission: {permission}"
)
return await func(*args, current_user=current_user, **kwargs)
return wrapper
return decorator
# 使用
@app.get("/api/users")
@require_permission("users.read")
async def list_users(current_user: User = Depends(get_current_user)):
return await User.all()
@app.delete("/api/users/{user_id}")
@require_permission("users.delete")
async def delete_user(user_id: str, current_user: User = Depends(get_current_user)):
return await User.delete(user_id)✅ 简单直观，易于理解和实现
✅ 审计清晰：谁有什么角色
✅ 主流框架都有支持
✅ 适合 90% 的 SaaS 场景
❌ 角色爆炸：细粒度控制 → 角色组合爆炸
❌ 无法表达上下文关系（"文档作者可编辑"）
❌ 跨租户角色管理困难
❌ 动态权限变更需重新分配角色
根据用户属性、资源属性、环境属性、操作属性的组合做决策。更灵活，也更复杂。
# ABAC 策略引擎 (Python 示例)
from dataclasses import dataclass
from datetime import datetime
@dataclass
class AccessRequest:
"""访问请求上下文"""
user_attrs: dict # { role, department, clearance_level, ... }
resource_attrs: dict # { owner_id, classification, project_id, ... }
action: str # read, write, delete, share, ...
env_attrs: dict # { time, ip, device_type, ... }
class ABACEngine:
def __init__(self, policies: list[dict]):
self.policies = policies
def evaluate(self, request: AccessRequest) -> bool:
"""评估所有策略，返回是否允许"""
for policy in self.policies:
if self._matches(policy.condition, request):
return policy.effect == "allow"
return False # Default deny
def _matches(self, condition: dict, request: AccessRequest) -> bool:
"""递归评估条件表达式"""
if "and" in condition:
return all(self._matches(c, request) for c in condition["and"])
if "or" in condition:
return any(self._matches(c, request) for c in condition["or"])
if "not" in condition:
return not self._matches(condition["not"], request)
# 原子条件: { field, operator, value }
field_value = self._resolve_field(condition["field"], request)
return self._compare(field_value, condition["operator"], condition["value"])
# 策略定义示例
POLICIES = [
{
"name": "文档编辑权限",
"condition": {
"and": [
{"field": "action", "operator": "eq", "value": "write"},
{"or": [
{"field": "user.role", "operator": "eq", "value": "admin"},
{"field": "resource.owner_id", "operator": "eq", "field_ref": "user.id"},
]},
{"field": "resource.classification", "operator": "lte",
"field_ref": "user.clearance_level"},
{"field": "env.time", "operator": "in_range",
"value": ["09:00", "18:00"]},
]
},
"effect": "allow"
},
{
"name": "下班后只读",
"condition": {
"and": [
{"field": "env.time", "operator": "not_in_range",
"value": ["09:00", "18:00"]},
{"field": "action", "operator": "eq", "value": "read"},
]
},
"effect": "allow"
}
]Google Zanzibar 论文提出的模型。用关系元组定义权限："用户 U 是文档 D 的 owner"，"owner 可以 editor"，"editor 可以 viewer"。
# OpenFGA (开源 Zanzibar) 关系元组定义
# schema.fga
model
# 用户
type user
# 组织
type organization
relations
define owner: [user]
define admin: [user] or owner
define member: [user] or admin
define can_create_project: admin
define can_manage_billing: owner
# 项目
type project
relations
define organization: [organization]
define owner: [user] or owner from organization
define editor: [user] or owner
define viewer: [user] or editor
define can_edit: editor
define can_view: viewer
define can_delete: owner
# 文档
type document
relations
define project: [project]
define owner: [user] or owner from project
define editor: [user] or owner
define viewer: [user] or editor
define can_read: viewer
define can_write: editor
define can_delete: owner
define can_share: editor
# 使用示例 (TypeScript)
import { OpenFgaClient } from '@openfga/sdk';
const fga = new OpenFgaClient({ apiUrl: 'http://localhost:8080' });
// 写入关系元组
await fga.write({
writes: [
// Alice 是 organization:acme 的 owner
{ user: 'user:alice', relation: 'owner', object: 'organization:acme' },
// Bob 是 project:backend 的 editor
{ user: 'user:bob', relation: 'editor', object: 'project:backend' },
// project:backend 属于 organization:acme
{ user: 'organization:acme', relation: 'organization', object: 'project:backend' },
],
});
// 检查权限：Bob 能编辑 project:backend 吗？
const { allowed } = await fga.check({
user: 'user:bob',
relation: 'can_edit',
object: 'project:backend',
});
// → allowed: true (因为 Bob 是 editor，editor 可以 can_edit)
// 检查权限：Bob 能删除 project:backend 吗？
const { allowed: canDelete } = await fga.check({
user: 'user:bob',
relation: 'can_delete',
object: 'project:backend',
});
// → allowed: false (只有 owner 可以 can_delete，而且 owner 从 organization 继承)
// 但 Alice 是 organization:acme 的 owner，所以 project:backend 的 owner 从 organization 继承
// Alice 可以 can_delete| 维度 | RBAC | ABAC | ReBAC |
|---|---|---|---|
| 核心概念 | 角色 → 权限 | 属性 → 策略 | 关系 → 推导 |
| 灵活度 | ⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| 实现复杂度 | ⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
| 性能 | 🟢 快 | 🟡 中等 | 🟡 需要缓存 |
| 上下文感知 | ❌ | ✅ | ✅ |
| 层级继承 | ⚠️ 需自实现 | ⚠️ 需策略 | ✅ 原生支持 |
| 典型场景 | 后台管理 | 合规/安全 | 文档协作/B2B |
| 代表产品 | 自建 | OPA/Cedar | OpenFGA/SpiceDB |
# Python 密码处理最佳实践
from passlib.context import CryptContext
# 推荐：argon2 为主，bcrypt 兼容
pwd_context = CryptContext(
schemes=["argon2", "bcrypt"], # 优先 argon2，兼容 bcrypt
deprecated="auto", # 自动迁移旧哈希到新算法
argon2__memory_cost=65536, # 64 MB 内存
argon2__time_cost=3, # 3 轮迭代
argon2__parallelism=4, # 4 线程
bcrypt__rounds=12, # bcrypt 12 轮
)
def hash_password(password: str) -> str:
"""哈希密码 — 注册时调用"""
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
"""验证密码 — 登录时调用"""
return pwd_context.verify(plain, hashed)
def needs_rehash(hashed: str) -> bool:
"""检查是否需要重新哈希 — 登录成功后调用"""
return pwd_context.needs_update(hashed)
# 使用示例
@app.post("/register")
async def register(email: str, password: str):
hashed = hash_password(password)
user = await User.create(email=email, password_hash=hashed)
return {"id": user.id}
@app.post("/login")
async def login(email: str, password: str):
user = await User.findByEmail(email)
if not user or not verify_password(password, user.password_hash):
raise HTTPException(401, "Invalid credentials")
# 自动升级旧哈希（如从 bcrypt → argon2）
if needs_rehash(user.password_hash):
user.password_hash = hash_password(password)
await user.save()
return {"token": create_access_token({"sub": user.id})}| 算法 | 抗 GPU | 抗 ASIC | 内存消耗 | 推荐度 |
|---|---|---|---|---|
| bcrypt | ✅ | ⚠️ | 低 (~4KB) | ⭐⭐⭐ 仍可用 |
| scrypt | ✅✅ | ✅ | 可调 | ⭐⭐⭐ 好 |
| argon2id | ✅✅ | ✅✅ | 可调 (推荐 64MB+) | ⭐⭐⭐⭐⭐ 最佳 |
| MD5/SHA* | ❌ | ❌ | 无 | 🚫 绝对不要 |
认证因素的三个维度：你知道的（密码）、你拥有的（手机/Key）、你是什么（指纹）。MFA = 至少两个维度。
最常见的 MFA 方案，Google Authenticator / Authy 等 App 都支持。
# Python TOTP 实现
import pyotp
import qrcode
import io
import base64
class TOTPService:
@staticmethod
def generate_secret() -> str:
"""为用户生成 TOTP 密钥"""
return pyotp.random_base32()
@staticmethod
def get_provisioning_uri(secret: str, email: str, issuer: str = "MyApp") -> str:
"""生成 otpauth:// URI"""
totp = pyotp.TOTP(secret)
return totp.provisioning_uri(name=email, issuer_name=issuer)
@staticmethod
def generate_qr_code(secret: str, email: str) -> str:
"""生成 QR Code 的 base64 图片"""
uri = TOTPService.get_provisioning_uri(secret, email)
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(uri)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buf = io.BytesIO()
img.save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode()
@staticmethod
def verify(secret: str, code: str, valid_window: int = 1) -> bool:
"""验证 TOTP 码 — valid_window=1 允许前后 30 秒"""
totp = pyotp.TOTP(secret)
return totp.verify(code, valid_window=valid_window)
# 注册 MFA 流程
@app.post("/api/mfa/setup")
async def setup_mfa(current_user: User = Depends(get_current_user)):
secret = TOTPService.generate_secret()
# 先存为 pending 状态，验证通过才激活
await current_user.update(mfa_secret_pending=secret)
qr_base64 = TOTPService.generate_qr_code(secret, current_user.email)
return {"qr_code": qr_base64, "secret": secret}
@app.post("/api/mfa/verify-setup")
async def verify_mfa_setup(
code: str,
current_user: User = Depends(get_current_user)
):
secret = current_user.mfa_secret_pending
if not secret or not TOTPService.verify(secret, code):
raise HTTPException(400, "Invalid code")
# 激活 MFA
await current_user.update(
mfa_secret=secret,
mfa_secret_pending=None,
mfa_enabled=True
)
# 生成恢复码
recovery_codes = [secrets.token_hex(4) for _ in range(8)]
await current_user.update(recovery_codes=recovery_codes)
return {"recovery_codes": recovery_codes}
# 登录时验证 MFA
@app.post("/api/auth/login-step2")
async def login_mfa(email: str, code: str):
user = await User.findByEmail(email)
if not user or not user.mfa_enabled:
raise HTTPException(400, "MFA not enabled")
# 先验证 TOTP
if TOTPService.verify(user.mfa_secret, code):
return {"token": create_access_token({"sub": user.id})}
# 再验证恢复码（一次性）
if code in user.recovery_codes:
user.recovery_codes.remove(code)
await user.save()
return {"token": create_access_token({"sub": user.id})}
raise HTTPException(401, "Invalid MFA code")FIDO2 标准，基于公钥加密，无需密码。苹果/Google/微软都在推 Passkeys。
// 前端 WebAuthn 注册 (浏览器 API)
async function registerPasskey(userId: string) {
// 1. 从服务端获取注册选项
const options = await fetch('/api/webauthn/register-options', {
method: 'POST',
body: JSON.stringify({ userId }),
}).then(r => r.json());
// 2. 浏览器弹出安全密钥/指纹对话框
const credential = await navigator.credentials.create({
publicKey: {
...options,
challenge: Uint8Array.from(atob(options.challenge), c => c.charCodeAt(0)),
user: {
...options.user,
id: Uint8Array.from(atob(options.user.id), c => c.charCodeAt(0)),
},
excludeCredentials: options.excludeCredentials?.map(c => ({
...c,
id: Uint8Array.from(atob(c.id), c => c.charCodeAt(0)),
})),
},
});
// 3. 将凭证发送到服务端验证存储
await fetch('/api/webauthn/register-verify', {
method: 'POST',
body: JSON.stringify({
id: credential.id,
rawId: btoa(String.fromCharCode(...new Uint8Array(credential.rawId))),
response: {
attestationObject: btoa(String.fromCharCode(
...new Uint8Array(credential.response.attestationObject)
)),
clientDataJSON: btoa(String.fromCharCode(
...new Uint8Array(credential.response.clientDataJSON)
)),
},
type: credential.type,
}),
});
}| 方案 | 安全性 | 易用性 | 成本 | 钓鱼攻击 |
|---|---|---|---|---|
| SMS OTP | ⚠️ 低 | ⭐⭐⭐⭐⭐ | 💰 每条短信 | ❌ 可被 SIM Swap |
| Email OTP | ⚠️ 中 | ⭐⭐⭐⭐ | 🟢 低 | ❌ 可被钓鱼 |
| TOTP App | ✅ 高 | ⭐⭐⭐ | 🟢 低 | ❌ 可被钓鱼 |
| WebAuthn/Passkeys | ✅✅ 最高 | ⭐⭐⭐⭐ | 🟢 低 | ✅ 抗钓鱼 |
| 硬件 Key (YubiKey) | ✅✅ 最高 | ⭐⭐ | 💰 $25-70/个 | ✅ 抗钓鱼 |
链接 → 多租户架构 深入讨论数据隔离，这里聚焦认证授权层面的多租户问题。
# FastAPI 多租户中间件
from starlette.middleware.base import BaseHTTPMiddleware
class TenantMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
tenant_id = self._resolve_tenant(request)
if not tenant_id:
return JSONResponse(
status_code=400,
content={"error": "Tenant not identified"}
)
# 注入到请求上下文
request.state.tenant_id = tenant_id
# 验证用户是否属于该租户
user = getattr(request.state, 'user', None)
if user and not self._user_in_tenant(user, tenant_id):
return JSONResponse(
status_code=403,
content={"error": "Access denied to this tenant"}
)
response = await call_next(request)
return response
def _resolve_tenant(self, request) -> str | None:
"""三种租户识别方式，按优先级尝试"""
# 1. 子域名: tenant1.myapp.com
host = request.headers.get("host", "")
if "." in host and not host.startswith("www."):
subdomain = host.split(".")[0]
if subdomain in TENANT_CACHE:
return subdomain
# 2. 自定义域名: app.tenant.com
if host in DOMAIN_TENANT_MAP:
return DOMAIN_TENANT_MAP[host]
# 3. Header: X-Tenant-ID (API 调用)
tenant_header = request.headers.get("X-Tenant-ID")
if tenant_header:
return tenant_header
# 4. URL 路径: /t/{tenant_id}/... (最不推荐)
# ...
return None
def _user_in_tenant(self, user: User, tenant_id: str) -> bool:
return tenant_id in user.tenant_ids(user_id, tenant_id) → role 映射，而不是 user_id → role。
# 租户感知的 RBAC
class TenantRole:
"""用户在某租户的角色"""
user_id: str
tenant_id: str
role: Role
# 数据模型
class User(Base):
id: str
email: str
tenant_roles: list[TenantRole] # 一对多
class TenantRole(Base):
user_id: str
tenant_id: str
role: Role
__table_args__ = (
UniqueConstraint('user_id', 'tenant_id'), # 每租户一个角色
)
# 查询当前租户角色
def get_tenant_role(user: User, tenant_id: str) -> Role | None:
for tr in user.tenant_roles:
if tr.tenant_id == tenant_id:
return tr.role
return None
# 权限检查装饰器
def require_tenant_permission(permission: str):
def decorator(func):
@wraps(func)
async def wrapper(
*args,
current_user: User = Depends(get_current_user),
tenant_id: str = Depends(get_tenant_id),
**kwargs
):
role = get_tenant_role(current_user, tenant_id)
if not role or not has_permission_by_role(role, permission):
raise HTTPException(403, "Insufficient permissions")
return await func(*args, current_user=current_user, tenant_id=tenant_id, **kwargs)
return wrapper
return decorator认证系统是 SaaS 中最容易低估复杂度的模块。自建还是用现成服务？
| 维度 | 自建 | Clerk | Auth0 | Supabase Auth | WorkOS |
|---|---|---|---|---|---|
| 定价模式 | 服务器成本 | MAU 计费 | MAU 计费 | 免费额度大 | MAU 计费 |
| 免费额度 | N/A | 10K MAU | 7.5K MAU | 50K MAU | 1M MAU |
| 社交登录 | 🔧 自接 | ✅ 20+ | ✅ 30+ | ✅ 10+ | ✅ 20+ |
| SAML SSO | 🔧 自建 | 💰 付费 | 💰 付费 | ❌ | ✅ 免费包含 |
| MFA | 🔧 自建 | ✅ | ✅ | ✅ TOTP | ✅ |
| Passkeys | 🔧 自建 | ✅ | ✅ | ✅ | ✅ |
| 多租户 | 🔧 自建 | ✅ Organizations | ✅ | ⚠️ 基础 | ✅ Organizations |
| 自定义 UI | ✅ 完全控制 | ✅ 高度可定制 | ⚠️ 有限 | ✅ 自建 UI | ⚠️ 有限 |
| 数据所有权 | ✅ 完全拥有 | ⚠️ Clerk 持有 | ⚠️ Auth0 持有 | ✅ 自托管可拥有 | ⚠️ WorkOS 持有 |
| 实现时间 | 2-8 周 | 1-2 天 | 1-3 天 | 1-2 天 | 1-3 天 |
你的 SaaS 需要认证授权 →
│
├─ MVP / 早期阶段？
│ └─ ✅ Clerk 或 Supabase Auth（最快上线）
│
├─ 需要企业 SSO (SAML)？
│ └─ ✅ WorkOS（免费包含 SAML）或 Auth0 Enterprise
│
├─ 需要完全数据控制？
│ └─ Supabase Auth (可自托管) 或 自建
│
├─ 已有 Next.js 项目？
│ └─ ✅ Clerk（Next.js 深度集成）
│
└─ 特殊安全/合规要求？
└─ ✅ 自建（完全控制审计、加密、存储）| 攻击类型 | 原理 | 防护措施 |
|---|---|---|
| 🔒 CSRF | 恶意网站冒充用户发请求 | SameSite Cookie + CSRF Token + Origin 检查 |
| 🔓 XSS → Token 盗取 | 注入脚本读取 localStorage | HttpOnly Cookie + CSP + 输入过滤 |
| 🔑 暴力破解 | 自动化尝试密码 | 指数退避 + CAPTCHA + 账户锁定 |
| 📧 凭证填充 | 用泄露的账号密码批量尝试 | HIBP 检查 + 异地登录告警 + MFA |
| 🎣 钓鱼攻击 | 伪造登录页窃取凭证 | WebAuthn/Passkeys（抗钓鱼） |
| 🔄 中间人攻击 | 截获通信 | HTTPS (HSTS) + Certificate Pinning |
| ⏱ 时序攻击 | 通过响应时间推断信息 | 恒定时间比较函数 |
| 🎲 JWT 算法混淆 | 将 RS256 改为 HS256 用公钥做密钥 | 强制指定算法 + 库级别防护 |
# 关键安全防护代码
# 1. 恒定时间比较 (防时序攻击)
import hmac
def safe_compare(a: str, b: str) -> bool:
"""恒定时间字符串比较，防止时序攻击"""
return hmac.compare_digest(a.encode(), b.encode())
# 2. 登录限流 (指数退避)
from datetime import datetime, timedelta
async def check_login_rate_limit(email: str, ip: str) -> None:
key = f"login_attempts:{email}:{ip}"
attempts = await redis.incr(key)
if attempts == 1:
await redis.expire(key, 3600) # 1 小时窗口
if attempts > 5:
# 指数退避：5次后每次等 2^n 秒
wait_seconds = min(2 ** (attempts - 5), 3600)
last_attempt = await redis.get(f"last_attempt:{email}")
if last_attempt:
elapsed = datetime.now() - datetime.fromisoformat(last_attempt)
if elapsed.total_seconds() < wait_seconds:
raise HTTPException(
429,
f"Too many attempts. Retry after {wait_seconds}s"
)
await redis.set(f"last_attempt:{email}", datetime.now().isoformat())
# 3. CSP Header (防 XSS)
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
class CSPMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response = await call_next(request)
response.headers["Content-Security-Policy"] = (
"default-src 'self'; "
"script-src 'self' 'nonce-{nonce}'; "
"style-src 'self' 'unsafe-inline'; "
"frame-ancestors 'none'; "
"base-uri 'self'"
)
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response链接回 Agent 架构研究，Agent 系统的认证有特殊挑战：
Agent 代用户执行操作，需要：
# Agent Token 方案
# 给 Agent 签发受限 Token，而非直接使用用户 Token
def create_agent_token(user_id: str, agent_id: str, permissions: list[str]) -> str:
"""为 Agent 签发受限 Token"""
payload = {
"sub": user_id, # 代表的用户
"agent_id": agent_id, # 执行的 Agent
"permissions": permissions, # Agent 被授权的操作列表
"type": "agent", # 区分人类 Token 和 Agent Token
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
"jti": str(uuid4()), # 可吊销
}
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
# Agent Token 权限检查
async def check_agent_permission(token_payload: dict, required: str) -> bool:
if token_payload.get("type") == "agent":
# Agent Token 只能访问明确授权的权限
allowed = token_payload.get("permissions", [])
return required in allowed or "*" in allowed
else:
# 人类 Token 走正常 RBAC
return has_permission_by_role(token_payload["role"], required)在 OpenClaw 的研究中，它实现了完整的 Policy Pipeline：
这是 Agent 认证授权的最佳实践参考——最小权限 + 人工审批 + 完整审计。