Merge pull request 'docs: 添加令牌桶限流模块设计文档' (#158) from feature/ratelimite into develop

Reviewed-on: http://8.161.227.145:3000/XEngineers/CamTalk/pulls/158
This commit was merged in pull request #158.
This commit is contained in:
2026-06-20 20:36:21 +08:00
5 changed files with 1160 additions and 6 deletions

View File

@@ -210,7 +210,7 @@ graph LR
| Models | 数据模型定义 |
| Migrations | 数据库版本化迁移,嵌入式 SQL 文件自动执行 |
| Model Router | 根据请求类型选择 AI 模型(待实现) |
| Rate Limiter | 令牌桶限流(待实现) |
| Rate Limiter | 令牌桶限流。详细设计见 [令牌桶限流设计](./13-令牌桶限流设计.md) |
## 前端组件
@@ -377,6 +377,19 @@ TieredManager
## 认证设计
采用 **JWT 双 token 轮转认证机制**,结合 bcrypt 密码哈希和 Refresh Token Rotation 安全策略。详细设计见 [鉴权体系设计](./12-鉴权体系设计.md)。
### 核心组件
| 组件 | 职责 |
|------|------|
| TokenManager | JWT 生成与验证(HS256 算法) |
| AuthService | 认证业务逻辑(注册/登录/刷新/登出) |
| AuthMiddleware | Gin 中间件,校验 access_token 并注入用户信息 |
| PasswordUtil | bcrypt 密码哈希(cost=10) |
### 认证流程
```mermaid
sequenceDiagram
participant C as 客户端
@@ -409,9 +422,45 @@ sequenceDiagram
G-->>C: {access_token, refresh_token}
```
**Token 策略**:access_token 15 分钟有效,refresh_token 7 天有效。refresh 时旧 token 失效(轮转),防止重放攻击。
### Token 策略
**WebSocket 认证**:连接地址 `ws://host/ws?token=<access_token>&conversation_id=<uuid>`。HTTP Upgrade 前校验 token,失败返回 401。
- **access_token**:15 分钟有效,用于 API 认证和 WebSocket 连接
- **refresh_token**:7 天有效,用于刷新 access_token
- **Refresh Token Rotation**:每次 refresh 都生成新的 token pair,旧 refresh_token 立即失效
- **复用检测**:检测到已删除的 refresh_token 被复用时,吊销该用户的所有 refresh_token
### 安全机制
1. **密码安全**:bcrypt 算法(cost=10),自动生成盐值,防彩虹表攻击
2. **Token 安全**:
- access_token 短有效期(15 分钟),降低泄露风险
- refresh_token 使用 SHA256 哈希存储,不存储原始 token
- Refresh Token Rotation 防重放攻击
- 复用检测 + 自动吊销机制
3. **传输安全**:HTTPS 强制,CORS 限制,HttpOnly Cookie 存储 refresh_token
4. **防攻击策略**:
- 防暴力破解:可选速率限制
- 防枚举攻击:统一错误信息
- 防 Token 泄露:复用检测 + 自动吊销
### WebSocket 认证
连接地址:`ws://host/ws?token=<access_token>&conversation_id=<uuid>`
- HTTP Upgrade 前校验 token
- 校验失败返回 401 Unauthorized
- 校验成功后,user_id 和 username 注入到连接上下文
### 配置
```yaml
auth:
jwt_secret: "" # JWT 签名密钥(必须通过 CAMTALK_AUTH_JWT_SECRET 环境变量设置)
access_ttl: 15 # access_token 有效期(分钟)
refresh_ttl: 10080 # refresh_token 有效期(分钟,7天)
```
> **安全要求**:`JWT_SECRET` **必须**通过环境变量设置,不能写入配置文件。生产环境使用 `openssl rand -hex 32` 生成随机密钥。
## 部署架构

View File

@@ -279,13 +279,27 @@ Client Server
#### 认证方式
需要认证的接口在请求头携带 JWT access token:
采用 **JWT 双 token 轮转认证机制**。详细设计见 [鉴权体系设计](./12-鉴权体系设计.md)。
**Token 类型**:
- **access_token**:短期令牌(15 分钟),用于 API 认证和 WebSocket 连接
- **refresh_token**:长期令牌(7 天),用于刷新 access_token
**请求头格式**:
```
Authorization: Bearer <access_token>
```
未认证或 token 过期时返回 `401 Unauthorized`。
**认证流程**:
1. 用户登录后获取 access_token + refresh_token
2. 请求受保护接口时携带 access_token
3. access_token 过期时,使用 refresh_token 刷新获取新的 token pair
4. refresh_token 采用轮转机制,每次刷新后旧 token 失效
**WebSocket 认证**:
- 连接地址:`ws://host/ws?token=<access_token>&conversation_id=<uuid>`
- HTTP Upgrade 前校验 token
- 校验失败返回 401 Unauthorized
#### 错误响应格式
@@ -380,7 +394,7 @@ interface LoginRequest {
| 400 | `INVALID_INPUT` | 请求参数缺失或格式错误 |
| 401 | `INVALID_CREDENTIALS` | 用户名或密码错误 |
#### 刷新 Token
#### 刷新 Token(Refresh Token Rotation)
```
POST /api/auth/refresh
@@ -397,12 +411,56 @@ interface RefreshRequest {
**成功响应** `200 OK`:同 `AuthResponse` 结构(返回新的 access_token + refresh_token,旧 refresh_token 失效——Token 轮转)。
**安全机制**:
- **Token 轮转**:每次 refresh 都会生成新的 token pair,旧 refresh_token 立即失效
- **复用检测**:如果检测到已删除的 refresh_token 被复用,立即吊销该用户的所有 refresh_token
- **强制重新登录**:吊销后,该用户所有设备都需要重新登录
**错误响应**:
| 状态码 | code | 场景 |
|--------|------|------|
| 401 | `INVALID_TOKEN` | refresh_token 无效或已过期 |
**前端集成示例**:
```typescript
// axios 响应拦截器
api.interceptors.response.use(
(response) => response,
async (error) => {
const originalRequest = error.config;
// 如果是 401 且不是 refresh 请求,尝试刷新 token
if (error.response?.status === 401 && !originalRequest._retry) {
originalRequest._retry = true;
try {
const refreshToken = getRefreshToken();
const response = await api.post('/api/auth/refresh', {
refresh_token: refreshToken,
});
const { access_token, refresh_token } = response.data;
setAccessToken(access_token);
setRefreshToken(refresh_token);
// 重试原始请求
originalRequest.headers.Authorization = `Bearer ${access_token}`;
return api(originalRequest);
} catch (refreshError) {
// 刷新失败,跳转登录页
clearTokens();
window.location.href = '/login';
return Promise.reject(refreshError);
}
}
return Promise.reject(error);
}
);
```
#### 登出
```

View File

@@ -0,0 +1,544 @@
# 鉴权体系设计
## 概述
CamTalk 采用 JWT 双 token 轮转认证机制,结合 bcrypt 密码哈希和 Refresh Token Rotation 安全策略,实现安全可靠的用户认证体系。
**设计原则**:
- **安全性**:access_token 短有效期(15 分钟),refresh_token 支持轮转防重放
- **可靠性**:Refresh Token Rotation 机制,检测复用时自动吊销用户所有令牌
- **可扩展性**:Repository 接口隔离存储层,支持内存和 PostgreSQL 双实现
## 整体架构
```mermaid
graph TB
subgraph Client["客户端"]
Browser["浏览器"]
end
subgraph AuthModule["Auth 模块"]
Service["AuthService<br/>Register / Login / Refresh / Logout"]
TokenMgr["TokenManager<br/>JWT 生成与验证"]
Middleware["AuthMiddleware<br/>Gin 中间件"]
Password["PasswordUtil<br/>bcrypt 哈希"]
end
subgraph Storage["存储层"]
UserRepo["UserRepository<br/>用户数据"]
TokenStore["RefreshToken 存储<br/>SHA256 哈希"]
end
Browser -->|"POST /api/auth/*"| Service
Service --> TokenMgr
Service --> Password
Service --> UserRepo
Service --> TokenStore
Middleware -->|"校验 access_token"| TokenMgr
Middleware -->|"写入 user_id/username"| GinContext["Gin Context"]
```
## 核心组件
### 1. JWT 令牌管理(TokenManager)
**文件位置**:`backend/internal/auth/jwt.go`
#### Claims 结构
```go
type Claims struct {
UserID string `json:"user_id"`
Username string `json:"username"`
TokenType string `json:"token_type"` // "access" | "refresh"
jwt.RegisteredClaims
}
```
**字段说明**:
- `UserID`:用户唯一标识(UUID)
- `Username`:用户名
- `TokenType`:令牌类型,用于区分 access 和 refresh token
- `RegisteredClaims`:JWT 标准声明(ExpiresAt, IssuedAt, Issuer, ID)
#### TokenManager 配置
```go
type TokenManager struct {
secret []byte // JWT 签名密钥(HS256)
accessTTL time.Duration // access_token 有效期(默认 15 分钟)
refreshTTL time.Duration // refresh_token 有效期(默认 7 天)
}
func NewTokenManager(secret string, accessTTL, refreshTTL time.Duration) *TokenManager
```
#### 令牌生成
```go
func (tm *TokenManager) GeneratePair(userID, username string) (access, refresh string, err error)
```
**生成逻辑**:
1. **access_token**:
- 签名算法:HS256
- 有效期:15 分钟
- 包含:UserID, Username, TokenType="access", ExpiresAt, IssuedAt, Issuer="camtalk"
2. **refresh_token**:
- 签名算法:HS256
- 有效期:7 天
- 包含:UserID, Username, TokenType="refresh", ID=UUID(用于 DB 关联), ExpiresAt, IssuedAt, Issuer="camtalk"
#### 令牌验证
```go
func (tm *TokenManager) ValidateAccess(tokenStr string) (*Claims, error)
func (tm *TokenManager) ValidateRefresh(tokenStr string) (*Claims, error)
```
**验证逻辑**:
1. 解析 JWT,验证签名算法为 HMAC
2. 验证签名是否有效
3. 验证令牌是否过期
4. 验证 TokenType 是否匹配(access 或 refresh)
5. 返回 Claims 或错误
#### Token 哈希
```go
func HashToken(token string) string
```
**用途**:对 refresh_token 做 SHA256 哈希后存储到数据库,避免直接存储原始 token。
### 2. 密码处理(PasswordUtil)
**文件位置**:`backend/internal/auth/password.go`
#### 密码哈希
```go
func HashPassword(password string) (string, error)
```
**实现**:
- 算法:bcrypt
- Cost:10(2^10 次迭代)
- 返回:base64 编码的哈希字符串
#### 密码验证
```go
func CheckPassword(hashedPassword, password string) error
```
**实现**:
- 使用 `bcrypt.CompareHashAndPassword` 验证
- 返回 nil 表示匹配,否则返回错误
### 3. 认证服务(AuthService)
**文件位置**:`backend/internal/auth/service.go`
#### 接口定义
```go
type Service interface {
Register(ctx context.Context, req RegisterRequest) (*AuthResponse, error)
Login(ctx context.Context, req LoginRequest) (*AuthResponse, error)
Refresh(ctx context.Context, req RefreshRequest) (*AuthResponse, error)
Logout(ctx context.Context, userID, refreshToken string) error
}
```
#### 注册流程(Register)
```go
func (s *authService) Register(ctx context.Context, req RegisterRequest) (*AuthResponse, error)
```
**流程**:
1. 检查用户名是否已存在(`FindByUsername`)
2. 如果存在,返回 `ErrUsernameTaken`
3. 使用 bcrypt 哈希密码(`HashPassword`)
4. 创建用户记录(`Create`)
5. 生成 access_token + refresh_token(`GeneratePair`)
6. 保存 refresh_token 的 SHA256 哈希到数据库(`SaveRefreshToken`)
7. 返回 `AuthResponse`
**错误处理**:
- `ErrUsernameTaken`:用户名已存在
- 数据库错误:透传底层错误
#### 登录流程(Login)
```go
func (s *authService) Login(ctx context.Context, req LoginRequest) (*AuthResponse, error)
```
**流程**:
1. 根据用户名查找用户(`FindByUsername`)
2. 如果用户不存在,返回 `ErrInvalidCredentials`
3. 验证密码(`CheckPassword`)
4. 如果密码错误,返回 `ErrInvalidCredentials`
5. 生成 access_token + refresh_token(`GeneratePair`)
6. 保存 refresh_token 的 SHA256 哈希到数据库(`SaveRefreshToken`)
7. 返回 `AuthResponse`
**错误处理**:
- `ErrInvalidCredentials`:用户名或密码错误(统一错误信息,防止枚举攻击)
#### 刷新令牌流程(Refresh)— Refresh Token Rotation
```go
func (s *authService) Refresh(ctx context.Context, req RefreshRequest) (*AuthResponse, error)
```
**流程**:
1. 验证 refresh_token 的签名和有效期(`ValidateRefresh`)
2. 计算 refresh_token 的 SHA256 哈希(`HashToken`)
3. 在数据库中查找该哈希(`FindRefreshToken`)
4. **如果哈希不存在**:
- JWT 校验已通过但 DB 中不存在 → token 已被 rotation 删除
- 这是 **token 复用行为**,属于安全风险
- 吊销该用户的所有 refresh_token(`DeleteUserRefreshTokens`)
- 返回 `ErrRefreshTokenUsed`
5. 验证 token 归属的用户与 claims 一致
6. 删除旧的 refresh_token 哈希(`DeleteRefreshToken`)
7. 生成新的 access_token + refresh_token(`GeneratePair`)
8. 保存新的 refresh_token 哈希到数据库(`SaveRefreshToken`)
9. 查询用户信息(`FindByID`)
10. 返回 `AuthResponse`
**安全机制**:
- **Token 轮转**:每次 refresh 都会生成新的 token pair,旧 refresh_token 立即失效
- **复用检测**:如果检测到已删除的 refresh_token 被复用,立即吊销该用户的所有 refresh_token
- **强制重新登录**:吊销后,该用户所有设备都需要重新登录
#### 登出流程(Logout)
```go
func (s *authService) Logout(ctx context.Context, userID, refreshToken string) error
```
**流程**:
1. 计算 refresh_token 的 SHA256 哈希(`HashToken`)
2. 从数据库删除该哈希(`DeleteRefreshToken`)
### 4. Gin 中间件(AuthMiddleware)
**文件位置**:`backend/internal/auth/middleware.go`
```go
func AuthMiddleware(tokenMgr *TokenManager) gin.HandlerFunc
```
**功能**:
1. 从请求头提取 `Authorization: Bearer <token>`
2. 验证 access_token(`ValidateAccess`)
3. 如果验证失败,返回 401 Unauthorized
4. 如果验证成功,将 `user_id` 和 `username` 写入 Gin Context
5. 调用 `c.Next()` 继续处理请求
**错误响应**:
```json
{
"code": "INVALID_TOKEN",
"message": "missing authorization header"
}
```
```json
{
"code": "INVALID_TOKEN",
"message": "invalid authorization format"
}
```
```json
{
"code": "INVALID_TOKEN",
"message": "invalid or expired token"
}
```
**Context Key**:
- `ContextKeyUserID = "user_id"`
- `ContextKeyUsername = "username"`
**使用示例**:
```go
// 在路由中使用中间件
authorized := r.Group("/api")
authorized.Use(auth.AuthMiddleware(tokenMgr))
{
authorized.GET("/conversations", handler.ListConversations)
authorized.POST("/conversations", handler.CreateConversation)
}
```
## 数据模型
### 用户表(users)
```sql
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
username VARCHAR(64) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
```
### Refresh Token 表(refresh_tokens)
```sql
CREATE TABLE refresh_tokens (
token_hash VARCHAR(64) PRIMARY KEY, -- SHA256 哈希
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
expires_at TIMESTAMP WITH TIME ZONE NOT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
CREATE INDEX idx_refresh_tokens_user_id ON refresh_tokens(user_id);
CREATE INDEX idx_refresh_tokens_expires_at ON refresh_tokens(expires_at);
```
## Repository 接口
### UserRepository
```go
type UserRepository interface {
// Create 创建用户,返回用户 ID
Create(ctx context.Context, username, passwordHash string) (string, error)
// FindByUsername 根据用户名查找用户
FindByUsername(ctx context.Context, username string) (*User, error)
// FindByID 根据 ID 查找用户
FindByID(ctx context.Context, id string) (*User, error)
// SaveRefreshToken 保存 refresh_token 哈希
SaveRefreshToken(ctx context.Context, userID, tokenHash string, expiresAt time.Time) error
// FindRefreshToken 根据 token 哈希查找用户 ID
FindRefreshToken(ctx context.Context, tokenHash string) (string, error)
// DeleteRefreshToken 删除指定的 refresh_token
DeleteRefreshToken(ctx context.Context, tokenHash string) error
// DeleteUserRefreshTokens 删除用户的所有 refresh_token(用于检测复用时吊销)
DeleteUserRefreshTokens(ctx context.Context, userID string) error
}
```
## 前端集成
### Token 存储
**推荐方案**:
- `access_token`:存储在内存中(JavaScript 变量)
- `refresh_token`:存储在 `httpOnly` Cookie 中(防止 XSS 攻击)
**备选方案**(开发环境):
- 两者都存储在 `localStorage`(便于调试,但存在 XSS 风险)
### 请求拦截器
```typescript
// axios 请求拦截器
api.interceptors.request.use((config) => {
const accessToken = getAccessToken();
if (accessToken) {
config.headers.Authorization = `Bearer ${accessToken}`;
}
return config;
});
// axios 响应拦截器
api.interceptors.response.use(
(response) => response,
async (error) => {
const originalRequest = error.config;
// 如果是 401 且不是 refresh 请求,尝试刷新 token
if (error.response?.status === 401 && !originalRequest._retry) {
originalRequest._retry = true;
try {
const refreshToken = getRefreshToken();
const response = await api.post('/api/auth/refresh', {
refresh_token: refreshToken,
});
const { access_token, refresh_token } = response.data;
setAccessToken(access_token);
setRefreshToken(refresh_token);
// 重试原始请求
originalRequest.headers.Authorization = `Bearer ${access_token}`;
return api(originalRequest);
} catch (refreshError) {
// 刷新失败,跳转登录页
clearTokens();
window.location.href = '/login';
return Promise.reject(refreshError);
}
}
return Promise.reject(error);
}
);
```
### WebSocket 认证
```typescript
// 建立 WebSocket 连接时传递 access_token
const wsUrl = `ws://${window.location.host}/ws?token=${accessToken}&conversation_id=${conversationId}`;
const ws = new WebSocket(wsUrl);
// 连接失败时(401),触发 token 刷新
ws.onerror = (error) => {
console.error('WebSocket connection failed');
// 可能需要刷新 token 后重连
};
```
## 安全考虑
### 1. 密码安全
- **bcrypt 算法**:使用 bcrypt 进行密码哈希,cost factor 为 10
- **盐值自动生成**:bcrypt 自动生成随机盐值,无需手动管理
- **防彩虹表**:每个密码的哈希值都不同,即使密码相同
### 2. Token 安全
- **短期 access_token**:15 分钟有效期,降低泄露风险
- **Refresh Token Rotation**:每次 refresh 都生成新 token,旧 token 立即失效
- **复用检测**:检测到已删除的 refresh_token 被复用时,吊销该用户的所有 token
- **SHA256 哈希存储**:数据库只存储 refresh_token 的哈希值,不存储原始 token
### 3. 传输安全
- **HTTPS 强制**:生产环境必须使用 HTTPS
- **CORS 限制**:配置 `AllowedOrigins` 限制允许的域名
- **HttpOnly Cookie**:refresh_token 存储在 httpOnly Cookie 中,防止 XSS 攻击
### 4. 防攻击策略
- **防暴力破解**:可选的速率限制(`RATE_LIMITED` 错误码)
- **防枚举攻击**:登录失败时统一返回 `INVALID_CREDENTIALS`,不区分用户名不存在还是密码错误
- **防重放攻击**:Refresh Token Rotation 确保每个 refresh_token 只能使用一次
- **防 Token 泄露**:检测到 token 复用时,立即吊销该用户的所有 token
## 配置说明
### 配置文件
```yaml
auth:
jwt_secret: "" # JWT 签名密钥(必须通过环境变量设置)
access_ttl: 15 # access_token 有效期(分钟)
refresh_ttl: 10080 # refresh_token 有效期(分钟,7天)
```
### 环境变量
| 环境变量 | 说明 | 示例 |
|---------|------|------|
| `CAMTALK_AUTH_JWT_SECRET` | JWT 签名密钥(必须) | `$(openssl rand -hex 32)` |
| `CAMTALK_AUTH_ACCESS_TTL` | access_token 有效期(分钟) | `15` |
| `CAMTALK_AUTH_REFRESH_TTL` | refresh_token 有效期(分钟) | `10080` |
**安全要求**:
- `JWT_SECRET` **必须**通过环境变量设置,不能写入配置文件
- 生产环境使用 `openssl rand -hex 32` 生成随机密钥
- 密钥长度建议至少 32 字节(256 位)
## 错误码
| 错误码 | HTTP 状态码 | 含义 | 客户端处理 |
|--------|-----------|------|-----------|
| `USERNAME_TAKEN` | 409 | 用户名已存在 | 提示换一个用户名 |
| `INVALID_CREDENTIALS` | 401 | 用户名或密码错误 | 提示检查输入 |
| `INVALID_TOKEN` | 401 | JWT 无效或已过期 | 尝试 refresh,失败则重新登录 |
## 测试用例
### 单元测试
**文件位置**:`backend/internal/auth/jwt_test.go`, `backend/internal/auth/service_test.go`
**测试覆盖**:
- Token 生成和验证
- Token 过期处理
- Refresh Token Rotation
- Token 复用检测和吊销
- 密码哈希和验证
- 边界条件和错误处理
### 集成测试
**测试场景**:
- 注册 → 登录 → 访问受保护资源
- Token 刷新流程
- Token 过期后自动刷新
- 并发刷新 token(竞态条件)
- Token 复用检测和吊销
## 监控指标
### 关键指标
- **登录成功率**:登录成功次数 / 登录总次数
- **Token 刷新率**:refresh 请求次数 / 总请求数
- **Token 复用检测**:检测到 token 复用的次数(安全事件)
- **认证延迟**:JWT 验证的平均耗时
### 告警规则
- **Token 复用检测**:任何 token 复用事件都应触发告警
- **异常登录失败率**:短时间内大量登录失败可能表示暴力破解攻击
- **Token 刷新失败率**:refresh 失败率突然上升可能表示系统问题
## 扩展点
### 1. 多设备管理
当前实现支持同一用户在多个设备上登录(每个设备独立的 refresh_token)。可以扩展为:
- 设备列表管理
- 单设备登录(踢出其他设备)
- 设备信任等级
### 2. OAuth 第三方登录
可以扩展 AuthService 支持 OAuth 2.0:
- Google、GitHub 等第三方登录
- 绑定/解绑第三方账号
- 统一的用户身份管理
### 3. 双因素认证(2FA)
可以扩展为:
- TOTP(基于时间的一次性密码)
- SMS 验证码
- 邮箱验证
### 4. 会话管理
可以扩展为:
- 活跃会话列表
- 远程登出其他会话
- 会话过期策略
## 参考资料
- [JWT 规范](https://tools.ietf.org/html/rfc7519)
- [bcrypt 算法](https://en.wikipedia.org/wiki/Bcrypt)
- [OWASP 认证备忘录](https://cheatsheetseries.owasp.org/cheatsheets/Authentication_Cheat_Sheet.html)
- [Refresh Token Rotation](https://auth0.com/blog/refresh-tokens-what-are-they-and-when-to-use-them/)

View File

@@ -0,0 +1,499 @@
# 令牌桶限流设计
## 概述
CamTalk 采用令牌桶(Token Bucket)算法实现按用户维度的速率限制,核心目标是**控制 AI 调用成本**,同时为 REST API 提供防暴力破解保护。
**设计原则**:
- **成本优先**:主要限流对象是 WebSocket `query` 消息(每次触发 STT + LLM + TTS 完整调用链)
- **用户隔离**:Per-user 维度限流,单用户超限不影响其他用户
- **弹性突发**:令牌桶允许合理的突发请求,优于固定窗口的滑动限流
- **存储适配**:内存 + Redis 双实现,单实例零依赖,多实例分布式一致
## 整体架构
```mermaid
graph TB
subgraph Entry["入口层"]
WS["WebSocket Handler<br/>query 消息"]
REST["REST API<br/>login / register"]
end
subgraph LimiterModule["Rate Limiter 模块"]
Interface["Limiter 接口<br/>Allow(userID) → (bool, retryAfter)"]
MemBucket["TokenBucket<br/>内存令牌桶"]
RedisBucket["RedisTokenBucket<br/>Redis 令牌桶(Lua 脚本)"]
Middleware["RateLimitMiddleware<br/>Gin 中间件"]
end
subgraph Storage["存储层"]
MemSync["sync.RWMutex<br/>进程内 map"]
Redis["Redis<br/>分布式计数"]
end
WS -->|"限流检查"| Interface
REST -->|"中间件"| Middleware
Middleware --> Interface
Interface --> MemBucket
Interface --> RedisBucket
MemBucket --> MemSync
RedisBucket --> Redis
```
## 令牌桶算法
### 原理
令牌桶以固定速率向桶中添加令牌,桶有最大容量上限。每次请求消耗一个令牌,桶空时拒绝请求。
```
桶容量(capacity) = 允许的突发请求数上限
填充速率(rate) = 每秒补充的令牌数
时间线示例(capacity=5, rate=0.2):
t=0s 桶满 5 令牌 → 用户连续发 5 个 query 全部通过
t=0s 桶空 → 第 6 个 query 被拒绝,retryAfter=5s
t=5s 桶补充 1 令牌 → 可再发 1 个 query
t=10s 桶补充 1 令牌 → 可再发 1 个 query
```
### 算法公式
```
elapsed = now - lastRefill
newTokens = elapsed * rate
currentTokens = min(capacity, lastTokens + newTokens)
if currentTokens >= 1:
currentTokens -= 1
allowed = true
else:
allowed = false
retryAfter = (1 - currentTokens) / rate
```
## 核心组件
### 1. Limiter 接口
**文件位置**:`backend/internal/ratelimit/limiter.go`
```go
// Limiter 速率限制器接口。
type Limiter interface {
// Allow 判断 key 是否允许执行一次操作。
// key 通常为 "userID:action" 格式。
// 返回 (allowed, retryAfter)。retryAfter 表示需要等待的时间。
Allow(ctx context.Context, key string) (bool, time.Duration)
}
```
**设计要点**:
- key 为字符串,不限定格式,由调用方决定维度(用户 ID、IP 地址等)
- 返回 `retryAfter` 供客户端/服务端设置 `Retry-After` header
- 接受 `context.Context` 支持超时和取消(Redis 实现需要)
### 2. 内存令牌桶(TokenBucket)
**文件位置**:`backend/internal/ratelimit/bucket.go`
```go
// TokenBucket 内存令牌桶,适用于单实例部署。
type TokenBucket struct {
capacity int // 桶容量
rate float64 // 每秒填充令牌数
tokens float64 // 当前令牌数
lastRefill time.Time // 上次填充时间
mu sync.Mutex
}
// Limiter 管理多个用户的令牌桶。
type Limiter struct {
buckets map[string]*TokenBucket
config Config
mu sync.RWMutex
stopOnce sync.Once
done chan struct{}
}
```
**并发安全**:
- 每个桶内部用 `sync.Mutex` 保护读写
- 桶 map 用 `sync.RWMutex` 保护(读多写少场景)
- 用户首次请求时惰性创建桶
**内存回收**:
- 后台 goroutine 定期扫描,清理超过 10 分钟无活动的桶
- 避免长期运行后内存泄漏
### 3. Redis 令牌桶(RedisTokenBucket)
**文件位置**:`backend/internal/ratelimit/redis_bucket.go`
使用 Redis Lua 脚本保证原子性,避免竞态条件:
```lua
-- KEYS[1] = 限流 key
-- ARGV[1] = capacity(桶容量)
-- ARGV[2] = rate(每秒填充数)
-- ARGV[3] = now(当前时间戳,秒,浮点)
-- ARGV[4] = ttl(key 过期时间,秒)
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local ttl = tonumber(ARGV[4])
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now
-- 计算新令牌
local elapsed = math.max(0, now - last_refill)
tokens = math.min(capacity, tokens + elapsed * rate)
local allowed = 0
local retry_after = 0
if tokens >= 1 then
tokens = tokens - 1
allowed = 1
else
retry_after = (1 - tokens) / rate
end
-- 回写状态
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, ttl)
return {allowed, tostring(retry_after)}
```
**设计要点**:
- 每个用户的限流状态存储为一个 Redis Hash(`tokens` + `last_refill`)
- TTL 自动过期,无需手动清理
- Lua 脚本保证"读取-计算-回写"原子执行
### 4. Gin 中间件
**文件位置**:`backend/internal/ratelimit/middleware.go`
```go
// RateLimitMiddleware 返回 Gin 中间件,按 key 维度限流。
// keyFunc 从请求中提取限流 key(如 IP、用户 ID)。
func RateLimitMiddleware(limiter Limiter, keyFunc func(*gin.Context) string) gin.HandlerFunc
```
**使用方式**:
```go
// 按 IP 限流(登录/注册,未登录用户无 userID)
loginGroup.POST("/login",
ratelimit.Middleware(limiter, func(c *gin.Context) string {
return c.ClientIP() + ":login"
}),
authHandler.Login,
)
// 按用户 ID 限流(已认证的 API)
authorized.POST("/conversations",
ratelimit.Middleware(limiter, func(c *gin.Context) string {
return c.GetString("user_id") + ":conversation"
}),
convHandler.Create,
)
```
**错误响应**:
REST API 返回 HTTP 429:
```json
{
"code": "RATE_LIMITED",
"message": "too many requests, retry after 5s"
}
```
同时设置 `Retry-After` header:
```
HTTP/1.1 429 Too Many Requests
Retry-After: 5
```
## 限流接入点
### WebSocket query 消息(核心)
在 `ws/handler.go` 的 `case "query"` 分支中,orchestrator 调用前检查:
```go
case "query":
// ... 解析消息 ...
// 限流检查
if limiter != nil {
allowed, retryAfter := limiter.Allow(ctx, userID+":query")
if !allowed {
errors.SendWSError(client, errors.CodeRateLimited, msg.RequestID,
fmt.Errorf("rate limited, retry after %s", retryAfter))
continue
}
}
// ... 继续处理 query ...
```
### REST API 登录/注册
在 `api/auth.go` 的路由注册中添加中间件:
```go
func (h *AuthHandler) RegisterRoutes(rg *gin.RouterGroup, limiter ratelimit.Limiter) {
auth := rg.Group("/auth")
if limiter != nil {
auth.POST("/register",
ratelimit.Middleware(limiter, ipKeyFunc("register")),
h.Register,
)
auth.POST("/login",
ratelimit.Middleware(limiter, ipKeyFunc("login")),
h.Login,
)
} else {
auth.POST("/register", h.Register)
auth.POST("/login", h.Login)
}
auth.POST("/refresh", h.Refresh)
auth.POST("/logout", h.Logout)
}
```
### 不限流的端点
| 端点 | 原因 |
|------|------|
| `ping` / `pong` | 心跳保活,无 AI 调用成本 |
| `config` | 配置更新,无 AI 调用成本 |
| `interrupt` | 中断请求,取消操作不应被限流 |
| `GET /api/health` | 健康检查,运维必需 |
| `POST /api/auth/refresh` | Token 刷新,限流会导致用户被迫重新登录 |
| `POST /api/auth/logout` | 登出,限流会导致用户无法正常退出 |
| `GET /api/conversations` | 查询列表,无 AI 调用成本 |
## 配置设计
### 配置文件
```yaml
# config.yaml 新增
ratelimit:
enabled: true
# WebSocket query 消息限流(核心,控制 AI 成本)
query:
capacity: 10 # 突发容量:允许连续发 10 个 query
rate: 0.2 # 填充速率:每 5 秒补充 1 个令牌
# REST API 登录限流(防暴力破解)
login:
capacity: 5 # 突发容量:允许连续 5 次登录尝试
rate: 0.1 # 填充速率:每 10 秒补充 1 次
# REST API 注册限流
register:
capacity: 3 # 突发容量:允许连续 3 次注册
rate: 0.05 # 填充速率:每 20 秒补充 1 次
```
### 配置结构体
```go
// config/config.go 新增
type RateLimitConfig struct {
Enabled bool `mapstructure:"enabled"`
Query BucketConfig `mapstructure:"query"`
Login BucketConfig `mapstructure:"login"`
Register BucketConfig `mapstructure:"register"`
}
type BucketConfig struct {
Capacity int `mapstructure:"capacity"` // 桶容量(突发上限)
Rate float64 `mapstructure:"rate"` // 每秒填充令牌数
}
```
### 默认值
```go
// setDefaults 新增
v.SetDefault("ratelimit.enabled", false)
v.SetDefault("ratelimit.query.capacity", 10)
v.SetDefault("ratelimit.query.rate", 0.2)
v.SetDefault("ratelimit.login.capacity", 5)
v.SetDefault("ratelimit.login.rate", 0.1)
v.SetDefault("ratelimit.register.capacity", 3)
v.SetDefault("ratelimit.register.rate", 0.05)
```
### 参数选择建议
| 场景 | capacity | rate | 含义 |
|------|----------|------|------|
| WebSocket query | 10 | 0.2 | 突发 10 个,之后每 5 秒 1 个 |
| 登录 | 5 | 0.1 | 突发 5 次,之后每 10 秒 1 次 |
| 注册 | 3 | 0.05 | 突发 3 次,之后每 20 秒 1 次 |
> **调参原则**:capacity 决定"能忍多少次突发",rate 决定"稳态下多久能再请求一次"。query 的 rate 建议根据 AI 调用成本和目标月预算反推。
## 依赖注入
### main.go 初始化
```go
// 初始化限流器
var limiter ratelimit.Limiter
if cfg.RateLimit.Enabled {
if rdb != nil {
// 多实例:使用 Redis 令牌桶
limiter = ratelimit.NewRedisLimiter(rdb, cfg.RateLimit)
logger.Log.Info("rate limiter initialized with Redis backend")
} else {
// 单实例:使用内存令牌桶
limiter = ratelimit.NewLimiter(cfg.RateLimit)
logger.Log.Info("rate limiter initialized with in-memory backend")
}
defer limiter.Stop()
}
```
### 注入到各模块
```go
// WebSocket Handler —— 新增 limiter 参数
r.GET("/ws", ws.ServeWS(sessionMgr, orch, cfg, tokenMgr, limiter))
// Auth REST —— 新增 limiter 参数
authHandler := api.NewAuthHandler(authService, tokenMgr)
authHandler.RegisterRoutes(apiGroup, limiter)
```
## 错误码
复用已有错误码 `RATE_LIMITED`(`backend/internal/errors/codes.go`):
| 传输层 | HTTP 状态码 | 错误格式 |
|--------|-----------|---------|
| REST API | 429 Too Many Requests | `{code: "RATE_LIMITED", message: "too many requests, retry after Xs"}` |
| WebSocket | — | `{type: "error", code: "RATE_LIMITED", request_id: "...", message: "..."}` |
## 文件结构
```
backend/internal/ratelimit/
├── limiter.go # Limiter 接口 + Config 类型定义
├── bucket.go # 内存令牌桶实现
├── bucket_test.go # 内存令牌桶单元测试
├── redis_bucket.go # Redis 令牌桶实现(Lua 脚本)
├── redis_bucket_test.go# Redis 令牌桶单元测试
└── middleware.go # Gin 中间件
```
## 测试用例
### 单元测试
**内存令牌桶**(`bucket_test.go`):
- 首次请求通过
- 连续消耗至桶空
- 桶空后拒绝,返回正确 retryAfter
- 等待后令牌补充,请求通过
- 并发安全性(多个 goroutine 同时 Allow)
- 桶容量边界(capacity=0, capacity=1)
- 填充速率边界(rate=0, rate 极大值)
- 不活跃桶的内存回收
**Redis 令牌桶**(`redis_bucket_test.go`):
- 与内存实现行为一致性
- Lua 脚本原子性
- key TTL 自动过期
- 并发安全性(多个客户端同时请求)
### 集成测试
- 限流关闭时不拦截请求
- 限流开启后,REST API 登录超限返回 429
- 限流开启后,WebSocket query 超限返回 `RATE_LIMITED` 错误
- 单实例内存限流 vs 多实例 Redis 限流行为一致
- 重启后内存限流重置,Redis 限流保持
## 扩展点
### 1. 多级限流
可扩展为多级限流策略:
```
全局限流(全用户共享) → 用户级限流(当前实现) → 端点级限流(不同 API 不同限制)
```
### 2. 动态调参
通过配置热更新或管理 API 动态调整限流参数,无需重启:
```go
// 预留接口
type DynamicLimiter interface {
Limiter
UpdateConfig(action string, cfg BucketConfig) error
}
```
### 3. 按用户等级差异化
不同用户等级使用不同的限流参数:
```yaml
ratelimit:
query:
capacity: 10 # 免费用户
rate: 0.2
query_premium:
capacity: 30 # 付费用户
rate: 1.0
```
### 4. 滑动窗口限流
令牌桶适合允许突发的场景。如果需要更平滑的限流,可增加滑动窗口实现:
```go
type SlidingWindowLimiter struct {
windowSize time.Duration
maxRequests int
}
```
### 5. 分布式全局限流
当前 Redis 实现是 Per-Instance 独立计数。如需全局精确限流,可改为 Redis 全局计数器(所有实例共享同一个 key)。
## 监控指标
### 关键指标
- **限流触发率**:被拒绝请求数 / 总请求数
- **各端点限流分布**:query / login / register 各自的触发率
- **等待时长分布**:retryAfter 的 P50/P99
- **桶状态**:各用户桶的平均令牌数(反映使用模式)
### 告警规则
- **限流触发率突增**:可能表示异常流量或攻击
- **单用户持续被限流**:可能表示客户端 bug(死循环请求)
## 参考资料
- [Token Bucket 算法](https://en.wikipedia.org/wiki/Token_bucket)
- [Redis Rate Limiting](https://redis.io/glossaries/rate-limiting/)
- [Cloudflare - How we built rate limiting capable of scaling to millions of domains](https://blog.cloudflare.com/counting-things-a-lot-of-different-things/)

View File

@@ -17,6 +17,8 @@ CamTalk 是一款多模态实时 AI 视觉对话助手。用户通过摄像头
| [09-技术名词解释](09-技术名词解释.md) | 前端/后端/AI 服务/Eino 框架技术名词简明解释 |
| [10-Eino重构方案](10-Eino重构方案.md) | Eino Graph 替换手写 goroutine 管道的设计方案 |
| [11-Eino框架技术文档](11-Eino框架技术文档.md) | Eino 框架在 CamTalk 中的使用指南(Graph、Lambda、Callback、State) |
| [12-鉴权体系设计](12-鉴权体系设计.md) | JWT 双 token 轮转认证、bcrypt 密码哈希、Refresh Token Rotation、安全机制 |
| [13-令牌桶限流设计](13-令牌桶限流设计.md) | 令牌桶限流算法、内存/Redis 双实现、Gin 中间件、WebSocket query 限流、配置设计 |
## 推荐阅读顺序
@@ -28,4 +30,6 @@ CamTalk 是一款多模态实时 AI 视觉对话助手。用户通过摄像头
5. **05~07** — 各技术领域的详细设计
6. **09-技术名词解释** — 遇到不熟悉的名词时查阅
7. **10~12** — Eino 重构相关(方案、框架文档、实施记录)
8. **12-鉴权体系设计** — 认证授权机制详细设计(JWT、bcrypt、Refresh Token Rotation)
9. **13-令牌桶限流设计** — 速率限制设计(令牌桶算法、成本控制、防暴力破解)