1317 lines
31 KiB
Markdown
1317 lines
31 KiB
Markdown
# JWT 管理文档
|
||
|
||
DocAuditAI 系统 JWT（JSON Web Token）管理完整指南
|
||
|
||
## 目录
|
||
|
||
1. [JWT 概述](#jwt-概述)
|
||
2. [JWT 生成机制](#jwt-生成机制)
|
||
3. [JWT 验证流程](#jwt-验证流程)
|
||
4. [JWT 配置说明](#jwt-配置说明)
|
||
5. [JWT 刷新机制](#jwt-刷新机制)
|
||
6. [JWT 吊销管理](#jwt-吊销管理)
|
||
7. [前端对接指南](#前端对接指南)
|
||
8. [安全最佳实践](#安全最佳实践)
|
||
9. [常见问题](#常见问题)
|
||
10. [故障排查](#故障排查)
|
||
|
||
---
|
||
|
||
## JWT 概述
|
||
|
||
### 什么是 JWT？
|
||
|
||
JWT（JSON Web Token）是一种开放标准（RFC 7519），用于在各方之间安全地传输信息。在 DocAuditAI 系统中，JWT 用于用户身份认证和授权。
|
||
|
||
### JWT 结构
|
||
|
||
JWT 由三部分组成，用 `.` 分隔：
|
||
|
||
```
|
||
Header.Payload.Signature
|
||
```
|
||
|
||
**示例**：
|
||
```
|
||
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjo1LCJ1c2VybmFtZSI6ImFkbWluIiwiZXhwIjoxNzMyMTY0MjA3fQ.Kz8fN3Y8Xp4QmW2vY6Jh9Lk0Rn1Tc3Uv5Wx6Yz7
|
||
```
|
||
|
||
#### 1. Header（头部）
|
||
|
||
包含 Token 类型和签名算法：
|
||
|
||
```json
|
||
{
|
||
"alg": "HS256",
|
||
"typ": "JWT"
|
||
}
|
||
```
|
||
|
||
#### 2. Payload（载荷）
|
||
|
||
包含用户信息和权限数据：
|
||
|
||
```json
|
||
{
|
||
"user_id": 5,
|
||
"username": "admin",
|
||
"user_role": "系统管理员",
|
||
"ou_id": "000",
|
||
"ou_name": "test",
|
||
"exp": 1732164207,
|
||
"iat": 1732077807,
|
||
"aud": "docauditai:client"
|
||
}
|
||
```
|
||
|
||
#### 3. Signature（签名）
|
||
|
||
使用密钥对头部和载荷进行签名，防止篡改：
|
||
|
||
```
|
||
HMACSHA256(
|
||
base64UrlEncode(header) + "." +
|
||
base64UrlEncode(payload),
|
||
JWT_SECRET
|
||
)
|
||
```
|
||
|
||
---
|
||
|
||
## JWT 生成机制
|
||
|
||
### 生成流程
|
||
|
||
```
|
||
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
|
||
│ 1. 用户登录 │ --> │ 2. 验证凭据 │ --> │ 3. 生成JWT │ --> │ 4. 返回Token │
|
||
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
|
||
```
|
||
|
||
### 代码实现
|
||
|
||
#### 1. 用户登录接口
|
||
|
||
**接口**: `POST /auth/login`
|
||
|
||
**请求体**:
|
||
```json
|
||
{
|
||
"username": "admin",
|
||
"password": "your_password"
|
||
}
|
||
```
|
||
|
||
**响应**:
|
||
```json
|
||
{
|
||
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
|
||
"token_type": "bearer",
|
||
"expires_in": 86400,
|
||
"user_info": {
|
||
"user_id": 5,
|
||
"username": "admin",
|
||
"roles": ["系统管理员"],
|
||
"ou_id": "000",
|
||
"ou_name": "test"
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2. JWT 生成代码
|
||
|
||
**位置**: `app/auth/jwt_handler.py`
|
||
|
||
```python
|
||
import jwt
|
||
from datetime import datetime, timedelta
|
||
import os
|
||
|
||
def create_access_token(user_data: dict) -> str:
|
||
"""
|
||
生成 JWT Access Token
|
||
|
||
Args:
|
||
user_data: 用户信息字典
|
||
|
||
Returns:
|
||
JWT Token 字符串
|
||
"""
|
||
# JWT 配置
|
||
JWT_SECRET = os.getenv('JWT_SECRET', 'your-secret-key')
|
||
JWT_ALGORITHM = os.getenv('JWT_ALGORITHM', 'HS256')
|
||
JWT_EXPIRATION_HOURS = int(os.getenv('JWT_EXPIRATION_HOURS', 24))
|
||
JWT_AUDIENCE = os.getenv('JWT_AUDIENCE', 'docauditai:client')
|
||
|
||
# 计算过期时间
|
||
now = datetime.utcnow()
|
||
exp = now + timedelta(hours=JWT_EXPIRATION_HOURS)
|
||
|
||
# 构建 Payload
|
||
payload = {
|
||
'user_id': user_data['user_id'],
|
||
'username': user_data['username'],
|
||
'user_role': user_data.get('user_role', ''),
|
||
'ou_id': user_data.get('ou_id', ''),
|
||
'ou_name': user_data.get('ou_name', ''),
|
||
'exp': int(exp.timestamp()),
|
||
'iat': int(now.timestamp()),
|
||
'aud': JWT_AUDIENCE
|
||
}
|
||
|
||
# 生成 JWT
|
||
token = jwt.encode(
|
||
payload,
|
||
JWT_SECRET,
|
||
algorithm=JWT_ALGORITHM
|
||
)
|
||
|
||
return token
|
||
```
|
||
|
||
#### 3. 登录接口完整示例
|
||
|
||
```python
|
||
from fastapi import APIRouter, HTTPException, status
|
||
from pydantic import BaseModel
|
||
|
||
router = APIRouter()
|
||
|
||
class LoginRequest(BaseModel):
|
||
username: str
|
||
password: str
|
||
|
||
@router.post("/login")
|
||
async def login(request: LoginRequest):
|
||
"""用户登录接口"""
|
||
|
||
# 1. 验证用户名密码（示例：从数据库查询）
|
||
user = await verify_credentials(request.username, request.password)
|
||
if not user:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="用户名或密码错误"
|
||
)
|
||
|
||
# 2. 获取用户角色和权限
|
||
user_roles = await get_user_roles(user['id'])
|
||
|
||
# 3. 构建用户数据
|
||
user_data = {
|
||
'user_id': user['id'],
|
||
'username': user['username'],
|
||
'user_role': user_roles[0] if user_roles else '',
|
||
'ou_id': user['ou_id'],
|
||
'ou_name': user['ou_name']
|
||
}
|
||
|
||
# 4. 生成 JWT
|
||
access_token = create_access_token(user_data)
|
||
|
||
# 5. 返回响应
|
||
return {
|
||
'access_token': access_token,
|
||
'token_type': 'bearer',
|
||
'expires_in': 86400, # 24小时（秒）
|
||
'user_info': user_data
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## JWT 验证流程
|
||
|
||
### 双层验证机制
|
||
|
||
DocAuditAI 采用**两层 JWT 验证机制**：
|
||
|
||
#### 1. 中间件层（快速验证）
|
||
|
||
**位置**: `app/middleware/jwt_auth.py`
|
||
|
||
**目的**: 快速验证 Token 存在性，解析基本用户信息
|
||
|
||
**流程**:
|
||
```python
|
||
async def jwt_auth_middleware(request: Request, call_next):
|
||
"""JWT 认证中间件（宽松模式）"""
|
||
|
||
# 白名单路径（无需认证）
|
||
if request.url.path in WHITE_LIST:
|
||
return await call_next(request)
|
||
|
||
# 获取 Token
|
||
auth_header = request.headers.get('Authorization')
|
||
if not auth_header or not auth_header.startswith('Bearer '):
|
||
return JSONResponse(
|
||
status_code=401,
|
||
content={'detail': '缺少认证令牌'}
|
||
)
|
||
|
||
token = auth_header.replace('Bearer ', '')
|
||
|
||
try:
|
||
# 解析 JWT（不验证签名，提升性能）
|
||
payload = jwt.decode(
|
||
token,
|
||
options={"verify_signature": False}
|
||
)
|
||
|
||
# 设置用户信息到 request.state
|
||
request.state.current_user = {
|
||
'user_id': payload.get('user_id'),
|
||
'username': payload.get('username'),
|
||
'user_role': payload.get('user_role'),
|
||
'ou_id': payload.get('ou_id')
|
||
}
|
||
|
||
except jwt.DecodeError:
|
||
return JSONResponse(
|
||
status_code=401,
|
||
content={'detail': '无效的令牌格式'}
|
||
)
|
||
|
||
return await call_next(request)
|
||
```
|
||
|
||
**特点**:
|
||
- ✅ 高性能：不验证签名，只解析 Token
|
||
- ✅ 覆盖所有请求
|
||
- ⚠️ 不验证 Token 是否被篡改
|
||
- ⚠️ 不验证 Token 是否过期
|
||
|
||
#### 2. 路由层（完整验证）
|
||
|
||
**位置**: `app/auth/jwt_handler.py` 中的 `verify_token()` 依赖函数
|
||
|
||
**目的**: 严格验证 Token 完整性和有效性
|
||
|
||
**流程**:
|
||
```python
|
||
from fastapi import Depends, HTTPException, status
|
||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||
import jwt
|
||
import os
|
||
|
||
security = HTTPBearer()
|
||
|
||
async def verify_token(
|
||
credentials: HTTPAuthorizationCredentials = Depends(security)
|
||
) -> dict:
|
||
"""
|
||
完整 JWT 验证（路由层）
|
||
|
||
验证内容：
|
||
1. JWT 签名验证
|
||
2. 过期时间验证
|
||
3. Audience 验证
|
||
4. 必需字段验证
|
||
|
||
Returns:
|
||
用户信息字典
|
||
|
||
Raises:
|
||
HTTPException: Token 无效、过期或缺少必需字段
|
||
"""
|
||
token = credentials.credentials
|
||
JWT_SECRET = os.getenv('JWT_SECRET', 'your-secret-key')
|
||
JWT_ALGORITHM = os.getenv('JWT_ALGORITHM', 'HS256')
|
||
JWT_AUDIENCE = os.getenv('JWT_AUDIENCE', 'docauditai:client')
|
||
|
||
try:
|
||
# 完整验证（包含签名、过期时间、audience）
|
||
payload = jwt.decode(
|
||
token,
|
||
JWT_SECRET,
|
||
algorithms=[JWT_ALGORITHM],
|
||
audience=JWT_AUDIENCE,
|
||
options={
|
||
"verify_signature": True,
|
||
"verify_exp": True,
|
||
"verify_aud": True
|
||
}
|
||
)
|
||
|
||
# 验证必需字段
|
||
required_fields = ['user_id', 'username', 'user_role', 'exp']
|
||
for field in required_fields:
|
||
if field not in payload:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail=f'Token 缺少必需字段: {field}'
|
||
)
|
||
|
||
return payload
|
||
|
||
except jwt.ExpiredSignatureError:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail='令牌已过期，请重新登录'
|
||
)
|
||
except jwt.InvalidAudienceError:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail='令牌受众验证失败'
|
||
)
|
||
except jwt.InvalidSignatureError:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail='令牌签名验证失败'
|
||
)
|
||
except jwt.DecodeError:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail='令牌格式错误'
|
||
)
|
||
```
|
||
|
||
**使用示例**:
|
||
```python
|
||
from fastapi import APIRouter, Depends
|
||
from app.auth.jwt_handler import verify_token
|
||
|
||
router = APIRouter()
|
||
|
||
@router.get("/protected/resource")
|
||
async def get_protected_resource(
|
||
current_user: dict = Depends(verify_token)
|
||
):
|
||
"""需要完整 JWT 验证的接口"""
|
||
return {
|
||
'message': '访问成功',
|
||
'user_id': current_user['user_id'],
|
||
'username': current_user['username']
|
||
}
|
||
```
|
||
|
||
**特点**:
|
||
- ✅ 严格验证：签名、过期时间、受众、必需字段
|
||
- ✅ 防篡改：验证 Token 签名
|
||
- ✅ 防重放：验证过期时间
|
||
- ⚠️ 性能较低：需要密钥解密和验证
|
||
|
||
### 验证流程对比表
|
||
|
||
| 验证内容 | 中间件层 | 路由层 |
|
||
|---------|---------|--------|
|
||
| Token 存在性 | ✅ | ✅ |
|
||
| Token 格式正确 | ✅ | ✅ |
|
||
| JWT 签名验证 | ❌ | ✅ |
|
||
| 过期时间验证 | ❌ | ✅ |
|
||
| Audience 验证 | ❌ | ✅ |
|
||
| 必需字段验证 | ❌ | ✅ |
|
||
| 性能 | 高 | 中 |
|
||
| 使用场景 | 所有请求 | 敏感操作 |
|
||
|
||
---
|
||
|
||
## JWT 配置说明
|
||
|
||
### 环境变量配置
|
||
|
||
**位置**: `config/env.{port}` （例如 `config/env.8073`）
|
||
|
||
```ini
|
||
# JWT 配置
|
||
JWT_SECRET=your-256-bit-secret-key-here
|
||
JWT_ALGORITHM=HS256
|
||
JWT_EXPIRATION_HOURS=24
|
||
JWT_AUDIENCE=docauditai:client
|
||
|
||
# PostgREST JWT 配置（不同于用户 JWT）
|
||
PGRST_JWT_SECRET=different-secret-for-postgrest
|
||
```
|
||
|
||
### 配置项说明
|
||
|
||
| 配置项 | 说明 | 默认值 | 推荐值 |
|
||
|--------|------|--------|--------|
|
||
| `JWT_SECRET` | JWT 签名密钥（256位） | `your-secret-key` | 随机生成的256位字符串 |
|
||
| `JWT_ALGORITHM` | JWT 签名算法 | `HS256` | `HS256`（对称）或 `RS256`（非对称） |
|
||
| `JWT_EXPIRATION_HOURS` | Token 有效期（小时） | `24` | `24`（生产环境） / `1`（测试环境） |
|
||
| `JWT_AUDIENCE` | JWT 受众标识 | `docauditai:client` | `docauditai:client` |
|
||
| `PGRST_JWT_SECRET` | PostgREST 内部 JWT 密钥 | - | 与 `JWT_SECRET` 不同的密钥 |
|
||
|
||
### 密钥生成示例
|
||
|
||
#### 方法1：Python 生成
|
||
|
||
```python
|
||
import secrets
|
||
|
||
# 生成 256 位（32 字节）密钥
|
||
jwt_secret = secrets.token_urlsafe(32)
|
||
print(f"JWT_SECRET={jwt_secret}")
|
||
```
|
||
|
||
#### 方法2：OpenSSL 生成
|
||
|
||
```bash
|
||
openssl rand -base64 32
|
||
```
|
||
|
||
#### 方法3：在线生成
|
||
|
||
访问：https://randomkeygen.com/ （选择 256-bit 密钥）
|
||
|
||
### 安全注意事项
|
||
|
||
⚠️ **重要**：
|
||
1. **JWT_SECRET 必须保密**，不要提交到代码仓库
|
||
2. **JWT_SECRET 和 PGRST_JWT_SECRET 必须不同**
|
||
3. **生产环境使用强随机密钥**（至少 256 位）
|
||
4. **定期轮换密钥**（建议每 90 天）
|
||
|
||
---
|
||
|
||
## JWT 刷新机制
|
||
|
||
### 刷新策略
|
||
|
||
DocAuditAI 支持两种 Token 刷新策略：
|
||
|
||
#### 策略1：Token 自动续期（推荐）
|
||
|
||
**原理**: Token 过期前 N 小时内访问任何接口，自动返回新 Token
|
||
|
||
**配置**:
|
||
```ini
|
||
JWT_AUTO_REFRESH_ENABLED=true
|
||
JWT_AUTO_REFRESH_THRESHOLD_HOURS=6
|
||
```
|
||
|
||
**响应头**:
|
||
```
|
||
X-New-Access-Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
|
||
```
|
||
|
||
**前端处理**:
|
||
```javascript
|
||
// axios 响应拦截器
|
||
apiClient.interceptors.response.use(
|
||
response => {
|
||
// 检查是否有新 Token
|
||
const newToken = response.headers['x-new-access-token'];
|
||
if (newToken) {
|
||
localStorage.setItem('access_token', newToken);
|
||
console.log('Token 已自动续期');
|
||
}
|
||
return response;
|
||
},
|
||
error => Promise.reject(error)
|
||
);
|
||
```
|
||
|
||
#### 策略2：Refresh Token 机制
|
||
|
||
**原理**: 登录时返回 Access Token + Refresh Token，Access Token 过期后使用 Refresh Token 获取新 Token
|
||
|
||
**登录响应**:
|
||
```json
|
||
{
|
||
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
|
||
"refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
|
||
"token_type": "bearer",
|
||
"expires_in": 86400
|
||
}
|
||
```
|
||
|
||
**刷新接口**:
|
||
|
||
**请求**: `POST /auth/refresh`
|
||
|
||
**请求体**:
|
||
```json
|
||
{
|
||
"refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
|
||
}
|
||
```
|
||
|
||
**响应**:
|
||
```json
|
||
{
|
||
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
|
||
"token_type": "bearer",
|
||
"expires_in": 86400
|
||
}
|
||
```
|
||
|
||
**前端实现**:
|
||
```javascript
|
||
async function refreshAccessToken() {
|
||
const refreshToken = localStorage.getItem('refresh_token');
|
||
if (!refreshToken) {
|
||
// 跳转登录页
|
||
window.location.href = '/login';
|
||
return null;
|
||
}
|
||
|
||
try {
|
||
const response = await fetch('/auth/refresh', {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify({ refresh_token: refreshToken })
|
||
});
|
||
|
||
if (response.ok) {
|
||
const data = await response.json();
|
||
localStorage.setItem('access_token', data.access_token);
|
||
return data.access_token;
|
||
} else {
|
||
// Refresh Token 也过期，跳转登录页
|
||
localStorage.removeItem('access_token');
|
||
localStorage.removeItem('refresh_token');
|
||
window.location.href = '/login';
|
||
return null;
|
||
}
|
||
} catch (error) {
|
||
console.error('Token 刷新失败:', error);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
// axios 响应拦截器（自动刷新）
|
||
apiClient.interceptors.response.use(
|
||
response => response,
|
||
async error => {
|
||
const originalRequest = error.config;
|
||
|
||
// 如果是 401 错误且未重试过
|
||
if (error.response?.status === 401 && !originalRequest._retry) {
|
||
originalRequest._retry = true;
|
||
|
||
// 刷新 Token
|
||
const newToken = await refreshAccessToken();
|
||
if (newToken) {
|
||
// 使用新 Token 重试原请求
|
||
originalRequest.headers.Authorization = `Bearer ${newToken}`;
|
||
return apiClient(originalRequest);
|
||
}
|
||
}
|
||
|
||
return Promise.reject(error);
|
||
}
|
||
);
|
||
```
|
||
|
||
---
|
||
|
||
## JWT 吊销管理
|
||
|
||
### 吊销场景
|
||
|
||
1. **用户主动登出**
|
||
2. **管理员强制下线用户**
|
||
3. **密码修改后旧 Token 失效**
|
||
4. **检测到异常登录**
|
||
|
||
### 吊销实现方式
|
||
|
||
#### 方式1：黑名单（推荐）
|
||
|
||
**原理**: 将吊销的 Token 存入 Redis 黑名单，验证时检查
|
||
|
||
**表结构**: `jwt_tokens` 表
|
||
|
||
```sql
|
||
CREATE TABLE jwt_tokens (
|
||
id SERIAL PRIMARY KEY,
|
||
user_id INTEGER NOT NULL,
|
||
token_jti VARCHAR(64) UNIQUE NOT NULL, -- JWT ID
|
||
token_type VARCHAR(20) DEFAULT 'access',
|
||
revoked BOOLEAN DEFAULT false,
|
||
revoked_at TIMESTAMP,
|
||
expires_at TIMESTAMP NOT NULL,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE INDEX idx_jwt_tokens_jti ON jwt_tokens(token_jti);
|
||
CREATE INDEX idx_jwt_tokens_user_id ON jwt_tokens(user_id);
|
||
```
|
||
|
||
**生成 Token 时记录**:
|
||
```python
|
||
import uuid
|
||
|
||
def create_access_token(user_data: dict) -> str:
|
||
# 生成唯一 JWT ID
|
||
jti = str(uuid.uuid4())
|
||
|
||
# Payload 中包含 jti
|
||
payload = {
|
||
'user_id': user_data['user_id'],
|
||
'username': user_data['username'],
|
||
'jti': jti, # 唯一标识
|
||
'exp': int(exp.timestamp()),
|
||
'iat': int(now.timestamp()),
|
||
'aud': JWT_AUDIENCE
|
||
}
|
||
|
||
# 生成 Token
|
||
token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
||
|
||
# 记录到数据库
|
||
await record_token(
|
||
user_id=user_data['user_id'],
|
||
token_jti=jti,
|
||
expires_at=exp
|
||
)
|
||
|
||
return token
|
||
```
|
||
|
||
**验证 Token 时检查黑名单**:
|
||
```python
|
||
async def verify_token(token: str) -> dict:
|
||
# 解码 Token
|
||
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
||
jti = payload.get('jti')
|
||
|
||
# 检查是否在黑名单
|
||
is_revoked = await check_token_revoked(jti)
|
||
if is_revoked:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail='令牌已被吊销'
|
||
)
|
||
|
||
return payload
|
||
|
||
async def check_token_revoked(jti: str) -> bool:
|
||
"""检查 Token 是否已吊销"""
|
||
# 先查 Redis 缓存
|
||
cache_key = f"jwt:revoked:{jti}"
|
||
is_revoked = await redis_client.get(cache_key)
|
||
if is_revoked is not None:
|
||
return is_revoked == 'true'
|
||
|
||
# 查数据库
|
||
query = "SELECT revoked FROM jwt_tokens WHERE token_jti = $1"
|
||
result = await db_pool.fetchrow(query, jti)
|
||
|
||
if result and result['revoked']:
|
||
# 缓存到 Redis（24小时）
|
||
await redis_client.setex(cache_key, 86400, 'true')
|
||
return True
|
||
|
||
return False
|
||
```
|
||
|
||
**吊销 Token 接口**:
|
||
|
||
**登出接口**: `POST /auth/logout`
|
||
|
||
```python
|
||
@router.post("/logout")
|
||
async def logout(current_user: dict = Depends(verify_token)):
|
||
"""用户登出"""
|
||
jti = current_user.get('jti')
|
||
|
||
# 吊销 Token
|
||
await revoke_token(jti)
|
||
|
||
return {'message': '登出成功'}
|
||
|
||
async def revoke_token(jti: str):
|
||
"""吊销指定 Token"""
|
||
query = """
|
||
UPDATE jwt_tokens
|
||
SET revoked = true, revoked_at = CURRENT_TIMESTAMP
|
||
WHERE token_jti = $1
|
||
"""
|
||
await db_pool.execute(query, jti)
|
||
|
||
# 同步到 Redis
|
||
cache_key = f"jwt:revoked:{jti}"
|
||
await redis_client.setex(cache_key, 86400, 'true')
|
||
```
|
||
|
||
**强制下线用户**:
|
||
|
||
```python
|
||
async def revoke_user_all_tokens(user_id: int):
|
||
"""吊销用户的所有 Token"""
|
||
query = """
|
||
UPDATE jwt_tokens
|
||
SET revoked = true, revoked_at = CURRENT_TIMESTAMP
|
||
WHERE user_id = $1 AND revoked = false
|
||
RETURNING token_jti
|
||
"""
|
||
results = await db_pool.fetch(query, user_id)
|
||
|
||
# 同步到 Redis
|
||
for row in results:
|
||
cache_key = f"jwt:revoked:{row['token_jti']}"
|
||
await redis_client.setex(cache_key, 86400, 'true')
|
||
```
|
||
|
||
#### 方式2：Token 版本号
|
||
|
||
**原理**: 为每个用户维护 Token 版本号，密码修改/强制下线时递增版本号
|
||
|
||
**用户表新增字段**:
|
||
```sql
|
||
ALTER TABLE sso_users ADD COLUMN token_version INTEGER DEFAULT 1;
|
||
```
|
||
|
||
**Payload 中包含版本号**:
|
||
```python
|
||
payload = {
|
||
'user_id': user_data['user_id'],
|
||
'username': user_data['username'],
|
||
'token_version': user_data['token_version'], # 版本号
|
||
'exp': int(exp.timestamp()),
|
||
'aud': JWT_AUDIENCE
|
||
}
|
||
```
|
||
|
||
**验证时检查版本号**:
|
||
```python
|
||
async def verify_token(token: str) -> dict:
|
||
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
||
|
||
# 获取用户当前版本号
|
||
current_version = await get_user_token_version(payload['user_id'])
|
||
|
||
# 检查版本号是否匹配
|
||
if payload.get('token_version', 0) != current_version:
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail='令牌版本已过期'
|
||
)
|
||
|
||
return payload
|
||
```
|
||
|
||
**密码修改后递增版本号**:
|
||
```python
|
||
async def change_password(user_id: int, new_password: str):
|
||
# 更新密码
|
||
await update_user_password(user_id, new_password)
|
||
|
||
# 递增 Token 版本号（使所有旧 Token 失效）
|
||
query = """
|
||
UPDATE sso_users
|
||
SET token_version = token_version + 1
|
||
WHERE id = $1
|
||
"""
|
||
await db_pool.execute(query, user_id)
|
||
```
|
||
|
||
---
|
||
|
||
## 前端对接指南
|
||
|
||
### 1. 登录流程
|
||
|
||
```javascript
|
||
// 登录函数
|
||
async function login(username, password) {
|
||
try {
|
||
const response = await fetch('/auth/login', {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify({ username, password })
|
||
});
|
||
|
||
if (response.ok) {
|
||
const data = await response.json();
|
||
|
||
// 存储 Token
|
||
localStorage.setItem('access_token', data.access_token);
|
||
localStorage.setItem('user_info', JSON.stringify(data.user_info));
|
||
|
||
// 可选：存储 Refresh Token
|
||
if (data.refresh_token) {
|
||
localStorage.setItem('refresh_token', data.refresh_token);
|
||
}
|
||
|
||
// 跳转首页
|
||
window.location.href = '/';
|
||
} else {
|
||
const error = await response.json();
|
||
alert(`登录失败: ${error.detail}`);
|
||
}
|
||
} catch (error) {
|
||
console.error('登录错误:', error);
|
||
alert('网络错误，请稍后重试');
|
||
}
|
||
}
|
||
```
|
||
|
||
### 2. 配置 Axios 拦截器
|
||
|
||
```javascript
|
||
import axios from 'axios';
|
||
|
||
// 创建 Axios 实例
|
||
const apiClient = axios.create({
|
||
baseURL: 'http://localhost:8000/api/v1',
|
||
headers: { 'Content-Type': 'application/json' }
|
||
});
|
||
|
||
// 请求拦截器：自动添加 Token
|
||
apiClient.interceptors.request.use(
|
||
config => {
|
||
const token = localStorage.getItem('access_token');
|
||
if (token) {
|
||
config.headers.Authorization = `Bearer ${token}`;
|
||
}
|
||
return config;
|
||
},
|
||
error => Promise.reject(error)
|
||
);
|
||
|
||
// 响应拦截器：处理 401 错误
|
||
apiClient.interceptors.response.use(
|
||
response => {
|
||
// 检查是否有新 Token（自动续期）
|
||
const newToken = response.headers['x-new-access-token'];
|
||
if (newToken) {
|
||
localStorage.setItem('access_token', newToken);
|
||
}
|
||
return response;
|
||
},
|
||
async error => {
|
||
const originalRequest = error.config;
|
||
|
||
// 401 错误处理
|
||
if (error.response?.status === 401) {
|
||
// 如果有 Refresh Token，尝试刷新
|
||
if (!originalRequest._retry && localStorage.getItem('refresh_token')) {
|
||
originalRequest._retry = true;
|
||
const newToken = await refreshAccessToken();
|
||
if (newToken) {
|
||
originalRequest.headers.Authorization = `Bearer ${newToken}`;
|
||
return apiClient(originalRequest);
|
||
}
|
||
}
|
||
|
||
// 否则跳转登录页
|
||
localStorage.clear();
|
||
window.location.href = '/login';
|
||
}
|
||
|
||
return Promise.reject(error);
|
||
}
|
||
);
|
||
|
||
export default apiClient;
|
||
```
|
||
|
||
### 3. 获取用户信息
|
||
|
||
```javascript
|
||
// 从 localStorage 获取
|
||
function getCurrentUser() {
|
||
const userInfo = localStorage.getItem('user_info');
|
||
return userInfo ? JSON.parse(userInfo) : null;
|
||
}
|
||
|
||
// 从服务器获取最新用户信息
|
||
async function fetchCurrentUser() {
|
||
try {
|
||
const response = await apiClient.get('/auth/me');
|
||
const userInfo = response.data;
|
||
localStorage.setItem('user_info', JSON.stringify(userInfo));
|
||
return userInfo;
|
||
} catch (error) {
|
||
console.error('获取用户信息失败:', error);
|
||
return null;
|
||
}
|
||
}
|
||
```
|
||
|
||
### 4. 登出流程
|
||
|
||
```javascript
|
||
async function logout() {
|
||
try {
|
||
// 调用登出接口（吊销 Token）
|
||
await apiClient.post('/auth/logout');
|
||
} catch (error) {
|
||
console.error('登出接口调用失败:', error);
|
||
} finally {
|
||
// 清除本地存储
|
||
localStorage.clear();
|
||
|
||
// 跳转登录页
|
||
window.location.href = '/login';
|
||
}
|
||
}
|
||
```
|
||
|
||
### 5. 路由守卫（Vue.js 示例）
|
||
|
||
```javascript
|
||
import { createRouter, createWebHistory } from 'vue-router';
|
||
|
||
const router = createRouter({
|
||
history: createWebHistory(),
|
||
routes: [
|
||
{ path: '/login', component: LoginPage },
|
||
{ path: '/', component: HomePage, meta: { requiresAuth: true } },
|
||
{ path: '/documents', component: DocumentList, meta: { requiresAuth: true } }
|
||
]
|
||
});
|
||
|
||
// 全局路由守卫
|
||
router.beforeEach((to, from, next) => {
|
||
const token = localStorage.getItem('access_token');
|
||
|
||
// 需要认证的路由
|
||
if (to.meta.requiresAuth && !token) {
|
||
next('/login');
|
||
return;
|
||
}
|
||
|
||
// 已登录用户访问登录页，跳转首页
|
||
if (to.path === '/login' && token) {
|
||
next('/');
|
||
return;
|
||
}
|
||
|
||
next();
|
||
});
|
||
|
||
export default router;
|
||
```
|
||
|
||
---
|
||
|
||
## 安全最佳实践
|
||
|
||
### 1. Token 存储
|
||
|
||
**推荐方式**: `localStorage` 或 `sessionStorage`
|
||
|
||
**不推荐**: Cookie（容易受到 CSRF 攻击）
|
||
|
||
**示例**:
|
||
```javascript
|
||
// 存储
|
||
localStorage.setItem('access_token', token);
|
||
|
||
// 读取
|
||
const token = localStorage.getItem('access_token');
|
||
|
||
// 删除
|
||
localStorage.removeItem('access_token');
|
||
```
|
||
|
||
### 2. HTTPS 传输
|
||
|
||
⚠️ **生产环境必须使用 HTTPS**
|
||
|
||
- Token 通过 HTTPS 加密传输
|
||
- 防止中间人攻击（MITM）
|
||
- 防止 Token 被窃取
|
||
|
||
### 3. Token 有效期
|
||
|
||
**推荐配置**:
|
||
- **生产环境**: 24 小时
|
||
- **测试环境**: 1-2 小时
|
||
- **开发环境**: 7 天（方便调试）
|
||
|
||
### 4. 密钥管理
|
||
|
||
✅ **正确做法**:
|
||
- 使用环境变量存储密钥
|
||
- 生产环境使用强随机密钥
|
||
- 定期轮换密钥（90 天）
|
||
- 使用密钥管理服务（AWS KMS、Azure Key Vault）
|
||
|
||
❌ **错误做法**:
|
||
- 硬编码密钥在代码中
|
||
- 将密钥提交到 Git
|
||
- 使用弱密钥（如 `secret`、`12345`）
|
||
|
||
### 5. Token 吊销
|
||
|
||
✅ **必须实现**:
|
||
- 用户登出时吊销 Token
|
||
- 密码修改后吊销所有旧 Token
|
||
- 检测到异常登录时吊销 Token
|
||
- 管理员强制下线用户
|
||
|
||
### 6. 防止 XSS 攻击
|
||
|
||
✅ **防护措施**:
|
||
- 前端对用户输入进行转义
|
||
- 使用 CSP（Content Security Policy）
|
||
- 避免在 URL 中传递 Token
|
||
- 使用 HttpOnly Cookie（如果使用 Cookie 存储）
|
||
|
||
### 7. 防止 CSRF 攻击
|
||
|
||
✅ **防护措施**:
|
||
- 使用 Bearer Token 而非 Cookie
|
||
- 验证 Referer 头
|
||
- 使用 CSRF Token
|
||
|
||
---
|
||
|
||
## 常见问题
|
||
|
||
### Q1: Token 过期后如何处理？
|
||
|
||
**A**: 使用 Refresh Token 机制或自动续期机制：
|
||
|
||
```javascript
|
||
// 响应拦截器
|
||
apiClient.interceptors.response.use(
|
||
response => response,
|
||
async error => {
|
||
if (error.response?.status === 401 && !error.config._retry) {
|
||
error.config._retry = true;
|
||
const newToken = await refreshAccessToken();
|
||
if (newToken) {
|
||
error.config.headers.Authorization = `Bearer ${newToken}`;
|
||
return apiClient(error.config);
|
||
}
|
||
}
|
||
return Promise.reject(error);
|
||
}
|
||
);
|
||
```
|
||
|
||
### Q2: 如何在多标签页间同步登录状态？
|
||
|
||
**A**: 使用 `storage` 事件监听 `localStorage` 变化：
|
||
|
||
```javascript
|
||
window.addEventListener('storage', (event) => {
|
||
if (event.key === 'access_token') {
|
||
if (!event.newValue) {
|
||
// Token 被删除，其他标签页也跳转登录
|
||
window.location.href = '/login';
|
||
} else {
|
||
// Token 被更新，其他标签页也更新
|
||
console.log('Token 已在其他标签页更新');
|
||
}
|
||
}
|
||
});
|
||
```
|
||
|
||
### Q3: JWT 和 Session 有什么区别？
|
||
|
||
**A**:
|
||
|
||
| 特性 | JWT | Session |
|
||
|------|-----|---------|
|
||
| 存储位置 | 客户端（localStorage） | 服务器（Redis/数据库） |
|
||
| 状态 | 无状态 | 有状态 |
|
||
| 可扩展性 | 高（无需共享状态） | 低（需要共享 Session） |
|
||
| 吊销难度 | 高（需要黑名单） | 低（直接删除 Session） |
|
||
| 性能 | 高（无需查询） | 低（需要查询 Session） |
|
||
|
||
### Q4: 如何实现"记住我"功能？
|
||
|
||
**A**: 使用长期有效的 Refresh Token：
|
||
|
||
```javascript
|
||
// 登录时勾选"记住我"
|
||
async function login(username, password, rememberMe) {
|
||
const response = await fetch('/auth/login', {
|
||
method: 'POST',
|
||
body: JSON.stringify({
|
||
username,
|
||
password,
|
||
remember_me: rememberMe // 传递"记住我"标志
|
||
})
|
||
});
|
||
|
||
const data = await response.json();
|
||
|
||
if (rememberMe) {
|
||
// 长期有效的 Refresh Token（30 天）
|
||
localStorage.setItem('refresh_token', data.refresh_token);
|
||
}
|
||
}
|
||
```
|
||
|
||
### Q5: 如何检测 Token 是否被盗用？
|
||
|
||
**A**: 记录 Token 使用的 IP 和 User-Agent，检测异常：
|
||
|
||
```python
|
||
@router.post("/login")
|
||
async def login(request: Request, login_data: LoginRequest):
|
||
# 获取客户端信息
|
||
client_ip = request.client.host
|
||
user_agent = request.headers.get('User-Agent', '')
|
||
|
||
# 生成 Token 时记录
|
||
await record_token_metadata(
|
||
jti=jti,
|
||
client_ip=client_ip,
|
||
user_agent=user_agent
|
||
)
|
||
```
|
||
|
||
```python
|
||
async def verify_token(request: Request, token: str):
|
||
# 检查 IP 和 User-Agent 是否匹配
|
||
current_ip = request.client.host
|
||
current_user_agent = request.headers.get('User-Agent', '')
|
||
|
||
metadata = await get_token_metadata(jti)
|
||
|
||
if metadata['client_ip'] != current_ip:
|
||
# IP 不匹配，可能被盗用
|
||
await log_security_event(
|
||
'Token used from different IP',
|
||
jti=jti,
|
||
original_ip=metadata['client_ip'],
|
||
current_ip=current_ip
|
||
)
|
||
# 可选：吊销 Token
|
||
await revoke_token(jti)
|
||
raise HTTPException(status_code=401, detail='Token 异常使用')
|
||
```
|
||
|
||
---
|
||
|
||
## 故障排查
|
||
|
||
### 问题1: Token 验证失败 "InvalidSignatureError"
|
||
|
||
**原因**: JWT_SECRET 不一致
|
||
|
||
**解决**:
|
||
1. 检查 `config/env.{port}` 文件中的 `JWT_SECRET`
|
||
2. 确保前端请求的后端实例使用相同的 `JWT_SECRET`
|
||
3. 重新登录获取新 Token
|
||
|
||
### 问题2: Token 过期 "ExpiredSignatureError"
|
||
|
||
**原因**: Token 超过有效期
|
||
|
||
**解决**:
|
||
1. 使用 Refresh Token 刷新
|
||
2. 重新登录
|
||
3. 检查服务器时间是否正确（`exp` 基于 UTC 时间）
|
||
|
||
### 问题3: Token 解码失败 "DecodeError"
|
||
|
||
**原因**: Token 格式错误或被篡改
|
||
|
||
**解决**:
|
||
1. 检查 Token 是否完整（三部分用 `.` 分隔）
|
||
2. 检查 Token 是否被 URL 编码（不应该编码）
|
||
3. 检查前端是否正确存储和传递 Token
|
||
|
||
### 问题4: 中间件不生效
|
||
|
||
**原因**: 中间件注册顺序错误或白名单配置错误
|
||
|
||
**解决**:
|
||
1. 检查 `core/app.py` 中间件注册顺序（JWT 中间件应在路由之前）
|
||
2. 检查白名单配置 `WHITE_LIST`
|
||
3. 检查路由路径是否匹配
|
||
|
||
### 问题5: PostgREST 返回 401
|
||
|
||
**原因**: PostgREST 使用的 JWT Secret 不同
|
||
|
||
**解决**:
|
||
1. 确认 `PGRST_JWT_SECRET` 配置
|
||
2. PostgREST 内部 JWT 由后端生成，前端不直接使用
|
||
3. 检查 PostgREST 代理是否正确转换 Token
|
||
|
||
### 调试技巧
|
||
|
||
#### 1. 解码 Token（不验证签名）
|
||
|
||
```python
|
||
import jwt
|
||
|
||
# 解码 Token（查看内容）
|
||
token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
|
||
payload = jwt.decode(token, options={"verify_signature": False})
|
||
print(payload)
|
||
```
|
||
|
||
```javascript
|
||
// 前端解码 Token
|
||
function decodeToken(token) {
|
||
const base64Url = token.split('.')[1];
|
||
const base64 = base64Url.replace(/-/g, '+').replace(/_/g, '/');
|
||
const jsonPayload = decodeURIComponent(atob(base64).split('').map(function(c) {
|
||
return '%' + ('00' + c.charCodeAt(0).toString(16)).slice(-2);
|
||
}).join(''));
|
||
|
||
return JSON.parse(jsonPayload);
|
||
}
|
||
|
||
const token = localStorage.getItem('access_token');
|
||
console.log(decodeToken(token));
|
||
```
|
||
|
||
#### 2. 检查 Token 是否过期
|
||
|
||
```javascript
|
||
function isTokenExpired(token) {
|
||
const payload = decodeToken(token);
|
||
const now = Math.floor(Date.now() / 1000); // 当前时间（秒）
|
||
return payload.exp < now;
|
||
}
|
||
|
||
if (isTokenExpired(token)) {
|
||
console.log('Token 已过期');
|
||
}
|
||
```
|
||
|
||
#### 3. 启用详细日志
|
||
|
||
```ini
|
||
# config/env.{port}
|
||
LOG_LEVEL=DEBUG
|
||
JWT_DEBUG_ENABLED=true
|
||
```
|
||
|
||
---
|
||
|
||
## 附录
|
||
|
||
### A. JWT Payload 字段完整说明
|
||
|
||
| 字段 | 类型 | 必需 | 说明 | 示例 |
|
||
|------|------|------|------|------|
|
||
| `user_id` | integer | ✅ | 用户ID | `5` |
|
||
| `username` | string | ✅ | 用户名 | `"admin"` |
|
||
| `user_role` | string | ✅ | 用户角色 | `"系统管理员"` |
|
||
| `ou_id` | string | ❌ | 组织单位ID | `"000"` |
|
||
| `ou_name` | string | ❌ | 组织单位名称 | `"test"` |
|
||
| `exp` | integer | ✅ | 过期时间（Unix 时间戳） | `1732164207` |
|
||
| `iat` | integer | ✅ | 签发时间（Unix 时间戳） | `1732077807` |
|
||
| `aud` | string | ✅ | 受众 | `"docauditai:client"` |
|
||
| `jti` | string | ❌ | JWT ID（用于吊销） | `"550e8400-e29b-41d4-a716-446655440000"` |
|
||
| `token_version` | integer | ❌ | Token 版本号 | `1` |
|
||
|
||
### B. HTTP 状态码说明
|
||
|
||
| 状态码 | 说明 | 场景 |
|
||
|--------|------|------|
|
||
| 200 | 成功 | 登录成功、Token 验证成功 |
|
||
| 401 | 未授权 | Token 缺失、无效、过期、被吊销 |
|
||
| 403 | 禁止访问 | 权限不足 |
|
||
| 422 | 验证错误 | 登录参数错误 |
|
||
|
||
### C. 错误响应示例
|
||
|
||
```json
|
||
{
|
||
"detail": "令牌已过期，请重新登录",
|
||
"error_code": "TOKEN_EXPIRED",
|
||
"timestamp": "2025-11-17T10:30:00Z"
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
**文档版本**: v1.0
|
||
**最后更新**: 2025-11-17
|
||
**维护者**: DocAuditAI Team
|