721 lines
30 KiB
Markdown
721 lines
30 KiB
Markdown
# 安全架构设计
|
||
|
||
本章详细阐述系统的安全架构设计，围绕认证、授权、防护、审计四大安全领域展开，旨在帮助开发者全面理解系统的安全保障体系，确保在开发与部署过程中遵循安全规范，有效防范各类安全风险。
|
||
|
||
**认证（Authentication）**：系统采用基于 JWT（JSON Web Token）的无状态认证机制，用户通过用户名密码登录后获取 Token，后续请求通过 `Authorization: Bearer <token>` 头传递，由 `login_required` 依赖统一拦截并验证 Token 的有效性（签名校验、有效期检查、Redis 黑名单过滤）。同时，系统内置登录失败锁定策略，连续失败 5 次即锁定账号 10 分钟，有效防御暴力破解攻击。
|
||
|
||
**授权（Authorization）**：基于 RBAC（基于角色的访问控制）模型设计，权限控制粒度细化至接口级。用户通过角色继承权限节点（如 `sys:position:add`），由 `@permission_required` 装饰器在路由层统一拦截校验，权限数据采用 Redis 缓存加速查询，管理员（`userId == 1`）自动放行。权限变更时，相关缓存主动失效，确保权限更新的实时性。
|
||
|
||
**防护（Protection）**：系统从多维度构建安全防护体系。限流中间件基于 Redis 滑动窗口算法，有效防御 CC 攻击和接口滥用；参数校验通过 Pydantic Schema 自动拦截非法输入，防止 SQL 注入和 XSS 攻击；敏感密码采用 bcrypt 加盐哈希存储，即使数据库泄露也无法还原明文；演示模式下 `@check_demo` 装饰器统一拦截写操作，保护测试数据不被污染；跨域配置精确控制可信域名，防止 CSRF 攻击。
|
||
|
||
**审计（Audit）**：系统提供完备的操作审计能力。操作日志中间件自动记录所有写操作（POST/PUT/DELETE/PATCH）的请求路径、参数、操作人、IP 地址及执行结果，存入 `fastapi_operation_log` 表；登录日志独立记录每次登录尝试（含 IP、User-Agent、登录状态），便于安全事件追溯与异常行为分析；定时任务每次执行记录至 `fastapi_job_log` 表，确保后台任务的执行过程可审计。
|
||
|
||
:::tip 安全设计原则
|
||
纵深防御：认证、授权、限流、参数校验多层防护，单一防线失效时仍有其他机制兜底
|
||
|
||
最小权限：用户仅能访问已授权接口，角色权限按需分配，遵循"够用即可"原则
|
||
|
||
数据不落地：敏感信息（Token、密码）不在日志中明文记录，密码哈希不可逆
|
||
|
||
安全默认：框架默认配置为安全模式（如 DEBUG 关闭、限流开启），如需放宽需显式配置
|
||
:::
|
||
|
||
## 安全架构总览
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 安全防御层次 │
|
||
│ │
|
||
│ ┌───────────────────────────────────────────────────────────────┐ │
|
||
│ │ 网络层防护 │ │
|
||
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
|
||
│ │ │ CORS │ │ 限流 │ │ XSS 防护 │ │ 上传体积限制 │ │ │
|
||
│ │ │ 跨域控制 │ │ 滑动窗口 │ │ 输入清洗 │ │ DoS 防御 │ │ │
|
||
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
|
||
│ └───────────────────────────────────────────────────────────────┘ │
|
||
│ │
|
||
│ ┌───────────────────────────────────────────────────────────────┐ │
|
||
│ │ 认证层 │ │
|
||
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
|
||
│ │ │ JWT 认证 │ │ 验证码 │ │ 登录锁定 │ │ Token 黑名单 │ │ │
|
||
│ │ │ Bearer │ │ 图形验证 │ │ 失败计数 │ │ 登出失效 │ │ │
|
||
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
|
||
│ └───────────────────────────────────────────────────────────────┘ │
|
||
│ │
|
||
│ ┌───────────────────────────────────────────────────────────────┐ │
|
||
│ │ 授权层 │ │
|
||
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ │
|
||
│ │ │ RBAC 权限 │ │ 演示模式保护 │ │ 超级管理员跳过 │ │ │
|
||
│ │ │ permission_ │ │ check_demo │ │ user_id=1 bypass │ │ │
|
||
│ │ │ required │ │ │ │ │ │ │
|
||
│ │ └──────────────┘ └──────────────┘ └──────────────────────┘ │ │
|
||
│ └───────────────────────────────────────────────────────────────┘ │
|
||
│ │
|
||
│ ┌───────────────────────────────────────────────────────────────┐ │
|
||
│ │ 数据层 │ │
|
||
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
|
||
│ │ │ 密码加密 │ │ 参数校验 │ │ SQL 注入 │ │ 操作日志 │ │ │
|
||
│ │ │ bcrypt │ │ Pydantic │ │ ORM 防护 │ │ 审计追踪 │ │ │
|
||
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
|
||
│ └───────────────────────────────────────────────────────────────┘ │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
## 一、认证机制
|
||
|
||
### JWT 令牌认证
|
||
|
||
:::tip JWT 工作流程
|
||
用户登录成功后，服务端签发 JWT Token 返回给前端。前端将 Token 存储在本地，后续每次请求通过 `Authorization: Bearer <token>` 头携带。服务端验证 Token 签名和有效期，解析出用户信息。
|
||
:::
|
||
|
||
1. 工作流程时序图
|
||
|
||
|
||
|
||
|
||
2. 纯文本树状时序图
|
||
|
||
```
|
||
[前端] [后端] [Redis]
|
||
│ │ │
|
||
│── POST /login ─────────→│ │
|
||
│ {username, pwd, │ │
|
||
│ captcha, uuid} │ │
|
||
│ │── 验证码校验 (uuid) ──→│
|
||
│ │←─ captcha_text ────────│
|
||
│ │ │
|
||
│ │── 密码校验 │
|
||
│ │── 签发JWT │
|
||
│ │── 存储Token ──────────→│
|
||
│←─ {token} ─────────────│ │
|
||
│ │ │
|
||
│── GET /api/v1/xxx ────→│ │
|
||
│ Authorization: │ │
|
||
│ Bearer <token> │ │
|
||
│ │── 验证JWT │
|
||
│ │── 检查黑名单 ─────────→│
|
||
│ │←─ exists ─────────────│
|
||
│←─ {data} ──────────────│ │
|
||
```
|
||
|
||
:::tip 流程说明
|
||
1. 前端提交用户名、密码、验证码及 UUID
|
||
2. 后端校验验证码，验证密码，生成 JWT 并缓存至 Redis
|
||
3. 前端后续请求携带 JWT，后端校验 Token 有效性并检查黑名单
|
||
4. 校验通过后返回业务数据
|
||
:::
|
||
|
||
#### JWT 配置
|
||
|
||
```python
|
||
# src/core/config/auth.py
|
||
# ============================================================
|
||
# JWT 配置
|
||
# ============================================================
|
||
# JWT签名密钥（必须通过环境变量设置，不能使用默认值上线）
|
||
JWT_SALT = os.getenv('JWT_SALT', '')
|
||
# JWT令牌过期时间（分钟）
|
||
JWT_EXPIRE_MINUTES = int(os.getenv('JWT_EXPIRE_MINUTES', '20'))
|
||
JWT_ALGORITHM = "HS256" # 签名算法
|
||
```
|
||
|
||
#### Token 签发
|
||
|
||
```python
|
||
# src/core/jwt.py
|
||
def create_token(payload, timeout=None):
|
||
"""
|
||
生成JWT令牌
|
||
=============================================================
|
||
|
||
根据传入的载荷数据生成JWT访问令牌，自动添加过期时间。
|
||
使用强密钥确保令牌安全性，符合RFC 7518安全规范。
|
||
|
||
参数:
|
||
payload (dict): JWT载荷数据，通常包含用户ID、用户名等信息
|
||
timeout (int): 令牌过期时间（分钟），默认使用DEFAULT_TIMEOUT_MINUTES
|
||
|
||
返回:
|
||
str: 生成的JWT令牌字符串
|
||
|
||
使用示例:
|
||
>>> payload = {'userId': 1001, 'username': 'admin'}
|
||
>>> token = create_token(payload, timeout=30)
|
||
>>> print(token)
|
||
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...
|
||
|
||
注意事项:
|
||
1. 密钥长度必须至少32字节，否则会发出警告
|
||
2. 生产环境建议通过环境变量设置JWT_SALT
|
||
3. 令牌过期后需要重新登录获取新令牌
|
||
"""
|
||
try:
|
||
# 设置过期时间
|
||
# 说明：在载荷中添加exp字段，指定令牌的过期时间点
|
||
# 计算方式：当前UTC时间 + 指定的分钟数
|
||
if timeout is None:
|
||
timeout = DEFAULT_TIMEOUT_MINUTES
|
||
|
||
# 创建payload副本，避免修改原始数据
|
||
payload_copy = payload.copy()
|
||
|
||
# 添加过期时间（exp）和签发时间（iat）
|
||
# exp: 过期时间，超过此时间令牌无效
|
||
# iat: 签发时间，用于令牌生命周期计算
|
||
current_time = datetime.datetime.now(tz=datetime.timezone.utc)
|
||
payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout)
|
||
payload_copy['iat'] = current_time
|
||
|
||
# 声明类型和加密算法
|
||
# 说明：定义JWT头部信息，包含令牌类型和加密算法
|
||
headers = {
|
||
"typ": "JWT", # 令牌类型
|
||
"alg": JWT_ALGORITHM # 加密算法
|
||
}
|
||
|
||
# 生成JWT令牌
|
||
# 函数：jwt.encode()
|
||
# 参数说明：
|
||
# payload: JWT载荷数据
|
||
# key: 加密密钥（盐）
|
||
# algorithm: 加密算法（HS256）
|
||
# headers: JWT头部信息
|
||
token = jwt.encode(
|
||
payload=payload_copy,
|
||
key=JWT_SALT,
|
||
algorithm=JWT_ALGORITHM,
|
||
headers=headers
|
||
)
|
||
|
||
# 记录日志（可选，生产环境建议使用DEBUG级别）
|
||
logger.debug(f"生成JWT令牌成功，过期时间：{timeout}分钟")
|
||
|
||
return token
|
||
|
||
except Exception as e:
|
||
# 捕获并记录生成令牌时的异常
|
||
logger.error(f"生成JWT令牌失败: {str(e)}")
|
||
raise
|
||
```
|
||
|
||
#### Token 验证
|
||
|
||
```python
|
||
# src/api/deps.py
|
||
# ============================================================
|
||
# 登录认证依赖
|
||
# ============================================================
|
||
async def login_required(request: Request):
|
||
"""登录验证依赖：提取并校验 Token，滑动窗口续签，校验用户有效，写入 request.state"""
|
||
success, access_token, msg = get_access_token(request)
|
||
if not success:
|
||
raise AuthorizationException(code=401, msg=msg)
|
||
|
||
if await is_token_blacklisted(access_token):
|
||
raise AuthorizationException(code=401, msg="登录已注销")
|
||
|
||
result = parse_payload(access_token)
|
||
if result['code'] != 0:
|
||
raise AuthorizationException(code=401, msg=result['msg'])
|
||
|
||
data = result['data']
|
||
user_id = int(data.get('userId', 0))
|
||
|
||
# 用户有效状态校验：优先读 Redis 缓存（状态变更/删除时已即时失效），
|
||
# 未命中再回源 DB 并回填缓存，避免每请求查库；Redis 不可用时缓存读失败自动回源 DB
|
||
from utils.perm_cache import get_cached_user_active, set_cached_user_active
|
||
active = await get_cached_user_active(user_id)
|
||
if active is None:
|
||
from modules.system.user.repository import user_repo
|
||
user = await asyncio.to_thread(user_repo.get_by_id, user_id)
|
||
active = bool(user and user.status == 1)
|
||
await set_cached_user_active(user_id, active)
|
||
|
||
if not active:
|
||
raise AuthorizationException(code=401, msg="用户已被禁用或删除，请联系管理员")
|
||
|
||
request.state.user_id = user_id
|
||
request.state.username = data.get('username', '')
|
||
request.state.realname = data.get('realname', '')
|
||
|
||
# 滑动窗口续签：token 生命过半时自动签发新 token，通过响应头返回
|
||
await _maybe_refresh_token(request, access_token, data)
|
||
```
|
||
|
||
#### Token 黑名单
|
||
|
||
用户登出时，将当前 Token 加入 Redis 黑名单，TTL 等于 Token 剩余有效期：
|
||
|
||
```python
|
||
# src/api/v1/endpoints/auth.py
|
||
# ============================================================
|
||
# 退出登录
|
||
# ============================================================
|
||
@router.get('/logout', summary='退出登录', dependencies=[Depends(login_required)])
|
||
async def logout(request: Request):
|
||
"""
|
||
用户退出登录
|
||
将当前访问令牌加入黑名单，并记录退出日志
|
||
:param request: FastAPI请求对象
|
||
:return: 退出结果响应
|
||
"""
|
||
# 将当前令牌加入黑名单
|
||
success, access_token, _ = get_access_token(request)
|
||
if success and access_token:
|
||
await add_token_to_blacklist(access_token)
|
||
|
||
# 记录退出登录日志
|
||
username = get_username(request)
|
||
realname = get_realname(request)
|
||
await asyncio.to_thread(login_log.record_login_log, request, username, 2, 0, 0, method='v1.logout',
|
||
request_method=request.method, create_user=realname, result="退出成功")
|
||
return R.ok(msg="注销成功")
|
||
```
|
||
|
||
### 登录验证码
|
||
|
||
:::tip 验证码防暴力破解
|
||
每次登录前需先获取验证码，验证码存储在 Redis 中（TTL 120 秒），登录时校验验证码正确性。验证码使用 UUID 作为 key，避免被猜测。
|
||
:::
|
||
|
||
```python
|
||
# 获取验证码
|
||
GET /api/v1/captcha
|
||
Response: { "key": "xxx", "captcha": "data:image/png;base64,iVBORw0KG..." }
|
||
|
||
# 登录时校验
|
||
POST /api/v1/login
|
||
Body: { "username": "admin", "password": "123456", "code": "a3Xb", "key": "xxx" }
|
||
```
|
||
|
||
### 登录失败锁定
|
||
|
||
连续 5 次登录失败后，账号自动锁定 10 分钟：
|
||
|
||
```python
|
||
# src/modules/auth/service.py
|
||
async def login(request, data):
|
||
redis = get_redis()
|
||
|
||
# 检查锁定状态
|
||
lock_key = f"login:lock:{data.username}"
|
||
is_locked = await redis.get(lock_key)
|
||
if is_locked:
|
||
return R.failed("账号已锁定，请10分钟后再试")
|
||
|
||
# 验证用户名密码
|
||
user = await asyncio.to_thread(user_repo.get_one, username=data.username)
|
||
if not user or not verify_password(data.password, user.password):
|
||
# 原子递增失败计数
|
||
fail_key = f"login:fail:{data.username}"
|
||
count = await incr_with_expire(redis, fail_key, 300)
|
||
if count >= 5:
|
||
await redis.set(lock_key, 1, ex=600) # 锁定 10 分钟
|
||
return R.failed(f"用户名或密码错误，剩余{5-count}次机会")
|
||
|
||
# 登录成功，清除失败计数
|
||
await redis.delete(f"login:fail:{data.username}")
|
||
```
|
||
|
||
### 密码安全
|
||
|
||
```python
|
||
# src/core/password.py
|
||
# +======================================================================
|
||
# | 模块: 密码加密工具
|
||
# | 说明: 基于 bcrypt 的双重加盐方案
|
||
# +======================================================================
|
||
|
||
"""密码加密工具：基于 bcrypt 的双重加盐方案。"""
|
||
|
||
import secrets
|
||
|
||
import bcrypt
|
||
|
||
# bcrypt 工作因子，值越大计算越慢越安全，默认12（约250ms）
|
||
BCRYPT_ROUNDS = 12
|
||
|
||
|
||
# ============================================================
|
||
# 密码工具函数
|
||
# ============================================================
|
||
def generate_salt(length: int = 10) -> str:
|
||
"""生成随机盐值"""
|
||
return secrets.token_urlsafe(length)[:length]
|
||
|
||
|
||
def encrypt_password(password: str, salt: str) -> str:
|
||
"""使用 bcrypt 加密密码（外部 salt 双重加盐）"""
|
||
combined = (password + salt).encode('utf-8')
|
||
hashed = bcrypt.hashpw(combined, bcrypt.gensalt(rounds=BCRYPT_ROUNDS))
|
||
return hashed.decode('utf-8')
|
||
|
||
|
||
def verify_password(password: str, salt: str, hashed: str) -> bool:
|
||
"""验证密码是否正确"""
|
||
combined = (password + salt).encode('utf-8')
|
||
return bcrypt.checkpw(combined, hashed.encode('utf-8'))
|
||
```
|
||
|
||
- 使用 bcrypt 算法，每次哈希自动生成随机盐
|
||
- 密码不可逆，数据库泄露后无法还原明文
|
||
- 默认密码在首次登录时强制修改
|
||
|
||
## 二、授权机制
|
||
|
||
### RBAC 权限模型
|
||
|
||
```
|
||
用户(User) ──N:M──→ 角色(Role) ──N:M──→ 菜单/权限(Menu)
|
||
│ │ │
|
||
│ user_role │ role_menu │ permission
|
||
│ 关联表 │ 关联表 │ 权限标识
|
||
└────────────────────┘ │
|
||
▼
|
||
sys:user:add
|
||
sys:user:page
|
||
sys:role:update
|
||
...
|
||
```
|
||
|
||
权限标识格式：`sys:{module}:{action}`
|
||
|
||
- `sys`：系统前缀
|
||
- `{module}`：模块名（user / role / menu / position / level 等）
|
||
- `{action}`：操作名（page / list / detail / add / update / delete / status / batchDelete / export / import）
|
||
|
||
### permission_required 装饰器
|
||
|
||
```python
|
||
# src/core/access_decorators.py
|
||
# ============================================================
|
||
# 辅助函数
|
||
# ============================================================
|
||
def _extract_request(*args, **kwargs):
|
||
"""从函数参数中提取 Request 对象"""
|
||
req = kwargs.get('request')
|
||
if req is not None:
|
||
return req
|
||
for arg in args:
|
||
if isinstance(arg, Request):
|
||
return arg
|
||
return None
|
||
|
||
|
||
# ============================================================
|
||
# 节点权限鉴权装饰器
|
||
# ============================================================
|
||
def permission_required(permission: str):
|
||
"""节点权限鉴权装饰器，userId==1（管理员）直接放行"""
|
||
|
||
def decorator(func):
|
||
def _check(request):
|
||
"""同步权限校验（直查DB，无Redis依赖，供线程池/回退场景使用）"""
|
||
userId = get_user_id(request)
|
||
if userId == 1:
|
||
return None
|
||
# 懒加载业务 service，避免 core → 业务模块顶层依赖
|
||
from modules.system.menu import service as menu
|
||
permission_list = menu.get_permissions_list_sync(userId)
|
||
if permission not in permission_list:
|
||
return R.failed("权限不足")
|
||
return None
|
||
|
||
async def _check_cached(request):
|
||
"""异步权限校验（走 Redis 缓存，未命中才回源DB）"""
|
||
userId = get_user_id(request)
|
||
if userId == 1:
|
||
return None
|
||
# 懒加载业务 service，避免 core → 业务模块顶层依赖
|
||
from modules.system.menu import service as menu
|
||
permission_list = await menu.get_permissions_list(userId)
|
||
if permission not in permission_list:
|
||
return R.failed("权限不足")
|
||
return None
|
||
|
||
if asyncio.iscoroutinefunction(func):
|
||
@wraps(func)
|
||
async def async_wrapper(*args, **kwargs):
|
||
request = _extract_request(*args, **kwargs)
|
||
denied = await _check_cached(request)
|
||
if denied:
|
||
return denied
|
||
return await func(*args, **kwargs)
|
||
|
||
return async_wrapper
|
||
|
||
@wraps(func)
|
||
def sync_wrapper(*args, **kwargs):
|
||
request = _extract_request(*args, **kwargs)
|
||
# 同步端点由 FastAPI 在线程池中执行，直接走同步直查 DB 校验；
|
||
# 不再每请求 asyncio.run 新建事件循环（避免跨 loop 复用 Redis 连接
|
||
# 被静默吞掉、缓存形同虚设的性能与正确性问题）
|
||
denied = _check(request)
|
||
if denied:
|
||
return denied
|
||
return func(*args, **kwargs)
|
||
|
||
return sync_wrapper
|
||
|
||
return decorator
|
||
```
|
||
|
||
:::tip 同步/异步双通道
|
||
装饰器根据被装饰函数的类型自动选择校验通道：
|
||
- `async def` 端点 → `_check_cached` 走 Redis 缓存（异步）
|
||
- `def` 端点 → `_check` 直查 DB（同步，在 FastAPI 线程池中执行）
|
||
|
||
这样避免了在同步端点中 `asyncio.run()` 新建事件循环导致的跨 loop 复用 Redis 连接问题。
|
||
:::
|
||
|
||
### 权限缓存
|
||
|
||
```python
|
||
# src/modules/system/menu/service.py
|
||
async def get_permissions_list(user_id: int) -> list:
|
||
"""获取用户权限列表（异步，走 Redis 缓存）"""
|
||
cache_key = f"perm:user:{user_id}"
|
||
redis = get_redis()
|
||
cached = await redis.get(cache_key)
|
||
if cached:
|
||
return json.loads(cached)
|
||
# 从数据库查询
|
||
permissions = query_user_permissions(user_id)
|
||
await redis.set(cache_key, json.dumps(permissions), ex=3600)
|
||
return permissions
|
||
|
||
|
||
def get_permissions_list_sync(user_id: int) -> list:
|
||
"""获取用户权限列表（同步，直查DB，供线程池场景使用）"""
|
||
return query_user_permissions(user_id)
|
||
|
||
|
||
def invalidate_perm_cache(user_id: int):
|
||
"""失效用户权限缓存（角色/菜单变更时调用）"""
|
||
from core.redis_client import get_redis
|
||
import asyncio
|
||
|
||
redis = get_redis()
|
||
asyncio.get_event_loop().run_until_complete(
|
||
redis.delete(f"perm:user:{user_id}")
|
||
)
|
||
```
|
||
|
||
### check_demo 装饰器
|
||
|
||
```python
|
||
# src/core/access_decorators.py
|
||
def check_demo(func):
|
||
"""演示模式拦截装饰器，放在 @permission_required 之后、路由函数之前"""
|
||
if asyncio.iscoroutinefunction(func):
|
||
@wraps(func)
|
||
async def async_wrapper(*args, **kwargs):
|
||
if FASTAPI_DEMO:
|
||
return R.failed("演示环境，暂无操作权限")
|
||
return await func(*args, **kwargs)
|
||
|
||
return async_wrapper
|
||
|
||
@wraps(func)
|
||
def sync_wrapper(*args, **kwargs):
|
||
if FASTAPI_DEMO:
|
||
return R.failed("演示环境，暂无操作权限")
|
||
return func(*args, **kwargs)
|
||
|
||
return sync_wrapper
|
||
```
|
||
|
||
- 保护所有写操作（add / update / delete / status / batchDelete / import）
|
||
- 在 `permission_required` 之后应用
|
||
- 根据被装饰函数类型自动选择同步/异步包装
|
||
- `FASTAPI_DEMO=True` 时，所有写操作返回失败
|
||
|
||
## 三、网络层防护
|
||
|
||
### CORS 跨域控制
|
||
|
||
```python
|
||
# src/middleware/cors.py
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
def register_cors(app):
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=CORS_ORIGINS, # 允许的源
|
||
allow_credentials=True, # 允许 Cookie
|
||
allow_methods=["*"], # 允许的 HTTP 方法
|
||
allow_headers=["*"], # 允许的请求头
|
||
)
|
||
```
|
||
|
||
- 生产环境应配置具体的 `allow_origins`，避免使用 `*`
|
||
- 开发环境可配置为 `["http://localhost:3000"]`
|
||
|
||
### 请求限流
|
||
|
||
```python
|
||
# src/middleware/rate_limit.py
|
||
async def rate_limit_middleware(request: Request, call_next):
|
||
# 基于客户端 IP 的滑动窗口限流
|
||
client_ip = get_client_ip(request)
|
||
count = await sliding_window_incr(redis, current_key, prev_key, ...)
|
||
if count > RATE_LIMIT_LIMIT:
|
||
return JSONResponse(
|
||
status_code=200,
|
||
content={"code": 1, "msg": "请求过于频繁，请稍后再试"}
|
||
)
|
||
return await call_next(request)
|
||
```
|
||
|
||
- 默认关闭，通过 `RATE_LIMIT_ENABLED=True` 启用
|
||
- 使用 Redis 滑动窗口计数器，消除固定窗口边界 2x 突发
|
||
- Redis 不可用时降级放行（fail-open）
|
||
- CORS 预检请求不计数
|
||
|
||
### XSS 防护
|
||
|
||
```python
|
||
# src/utils/rich_text.py
|
||
import bleach
|
||
|
||
def sanitize_html(content: str) -> str:
|
||
"""XSS 清洗：移除危险标签和属性"""
|
||
allowed_tags = [
|
||
'p', 'br', 'strong', 'em', 'u', 'ol', 'ul', 'li',
|
||
'h1', 'h2', 'h3', 'h4', 'h5', 'h6',
|
||
'a', 'img', 'table', 'tr', 'td', 'th', 'tbody', 'thead'
|
||
]
|
||
allowed_attrs = {
|
||
'a': ['href', 'title'],
|
||
'img': ['src', 'alt', 'width', 'height'],
|
||
}
|
||
return bleach.clean(content, tags=allowed_tags, attributes=allowed_attrs)
|
||
```
|
||
|
||
- 富文本内容入库前进行 XSS 清洗
|
||
- 使用 `bleach` 库白名单过滤危险标签和属性
|
||
- 图片 `src` 属性只允许合法 URL
|
||
|
||
### 上传体积限制
|
||
|
||
```python
|
||
# src/middleware/upload_size.py
|
||
async def upload_size_limit_middleware(request: Request, call_next):
|
||
content_length = request.headers.get('content-length')
|
||
if content_length and int(content_length) > MAX_UPLOAD_SIZE:
|
||
return JSONResponse(
|
||
status_code=200,
|
||
content={"code": 1, "msg": "上传文件超出大小限制"}
|
||
)
|
||
return await call_next(request)
|
||
```
|
||
|
||
- 按 `Content-Length` 提前拒绝超大请求体
|
||
- 避免超大 multipart 落临时盘后再由业务层校验
|
||
|
||
## 四、数据层防护
|
||
|
||
### SQL 注入防护
|
||
|
||
:::tip ORM 天然防注入
|
||
SQLAlchemy ORM 使用参数化查询（Parameterized Query），所有用户输入通过参数绑定传递，不直接拼接 SQL 字符串，从根本上防止 SQL 注入。
|
||
:::
|
||
|
||
```python
|
||
# SQLAlchemy 参数化查询示例
|
||
user_repo.get_one(username="admin")
|
||
# 实际执行：SELECT * FROM fastapi_user WHERE username = :username AND is_delete = 0
|
||
# 参数绑定：{'username': 'admin'}
|
||
```
|
||
|
||
### 参数校验
|
||
|
||
```python
|
||
# Pydantic 声明式校验
|
||
class PositionForm(BaseForm):
|
||
name: str = Field(..., min_length=1, max_length=150)
|
||
status: int = Field(..., ge=1, le=2)
|
||
sort: int = Field(..., ge=0, le=99999)
|
||
|
||
@field_validator('name')
|
||
@classmethod
|
||
def validate_name(cls, v):
|
||
if not v.strip():
|
||
raise ValueError('岗位名称不能为空')
|
||
return v.strip()
|
||
```
|
||
|
||
- 所有请求参数通过 Pydantic 声明式校验
|
||
- 自动拒绝不符合约束的请求
|
||
- 校验错误统一返回格式
|
||
|
||
### 敏感字段保护
|
||
|
||
- 密码使用 bcrypt 哈希存储，不可逆
|
||
- `to_dict()` 序列化时自动排除 `is_delete`、`password`、`salt`
|
||
- JWT 密钥通过环境变量配置，不硬编码
|
||
|
||
## 五、审计追踪
|
||
|
||
### 操作日志
|
||
|
||
```python
|
||
# src/middleware/operation_log.py
|
||
async def operation_log_middleware(request: Request, call_next):
|
||
response = await call_next(request)
|
||
if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
|
||
# 记录操作日志
|
||
log_data = {
|
||
'username': request.state.username,
|
||
'module': extract_module(request.url.path),
|
||
'action': request.method,
|
||
'url': str(request.url),
|
||
'params': await get_request_params(request),
|
||
'ip': get_client_ip(request),
|
||
'status': 1 if is_success else 2,
|
||
'duration': int(duration * 1000),
|
||
}
|
||
save_operation_log(log_data)
|
||
return response
|
||
```
|
||
|
||
自动记录所有写操作的日志，包括：
|
||
- 操作人、操作模块、操作动作
|
||
- 请求 URL、请求参数
|
||
- 操作 IP、操作时间
|
||
- 响应状态、执行耗时
|
||
|
||
### 登录日志
|
||
|
||
每次登录/登出自动记录到 `fastapi_login_log` 表，包括登录账号、IP、浏览器、操作系统、登录状态等。
|
||
|
||
## 安全配置清单
|
||
|
||
| 配置项 | 环境变量 | 默认值 | 说明 |
|
||
|:---|:---|:---|:---|
|
||
| JWT 密钥 | `JWT_SALT` | 必须配置 | HS256 签名密钥 |
|
||
| JWT 过期时间 | `JWT_EXPIRE_MINUTES` | 120 | Token 有效期（分钟） |
|
||
| 密码错误锁定 | — | 5 次/10 分钟 | 连续失败后锁定 |
|
||
| 验证码有效期 | — | 120 秒 | 验证码 Redis TTL |
|
||
| 限流开关 | `RATE_LIMIT_ENABLED` | False | 滑动窗口限流 |
|
||
| 限流阈值 | `RATE_LIMIT_LIMIT` | 100 | 每窗口最大请求数 |
|
||
| 限流窗口 | `RATE_LIMIT_WINDOW_SECONDS` | 60 | 窗口时长（秒） |
|
||
| CORS 源 | `CORS_ORIGINS` | * | 允许的跨域源 |
|
||
| 上传限制 | `MAX_UPLOAD_SIZE` | 10MB | 最大上传体积 |
|
||
| 演示模式 | `FASTAPI_DEMO` | False | 锁定写操作 |
|
||
| 数据库密码 | `DB_PASSWORD` | — | 数据库连接密码 |
|
||
| Redis 密码 | `REDIS_PASSWORD` | — | Redis 连接密码 |
|
||
|
||
:::warning 生产环境安全检查
|
||
1. `JWT_SALT` 必须使用强随机字符串，不能使用默认值
|
||
2. `CORS_ORIGINS` 必须配置具体的域名，不能使用 `*`
|
||
3. `FASTAPI_DEMO` 在生产环境应为 `False`
|
||
4. 数据库和 Redis 必须设置密码
|
||
5. 启用 HTTPS（由 Nginx/负载均衡器处理）
|
||
:::
|
||
|
||
## 总结
|
||
|
||
的安全架构涵盖认证（JWT + 验证码 + 登录锁定 + Token 黑名单）、授权（RBAC + permission_required + check_demo）、防护（CORS + 限流 + XSS + 上传限制 + SQL 注入防护）、审计（操作日志 + 登录日志）四大领域。通过纵深防御策略，在网络层、认证层、授权层、数据层分别设置安全屏障，确保系统在面对各类安全威胁时具备足够的防护能力。生产部署时需特别关注 JWT 密钥强度、CORS 配置、HTTPS 启用等关键安全项。
|