Become a sponsor

本章详细阐述系统的安全架构设计,围绕认证、授权、防护、审计四大安全领域展开,旨在帮助开发者全面理解系统的安全保障体系,确保在开发与部署过程中遵循安全规范,有效防范各类安全风险。
认证(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 表,确保后台任务的执行过程可审计。
安全设计原则
纵深防御:认证、授权、限流、参数校验多层防护,单一防线失效时仍有其他机制兜底
最小权限:用户仅能访问已授权接口,角色权限按需分配,遵循"够用即可"原则
数据不落地:敏感信息(Token、密码)不在日志中明文记录,密码哈希不可逆
安全默认:框架默认配置为安全模式(如 DEBUG 关闭、限流开启),如需放宽需显式配置
┌─────────────────────────────────────────────────────────────────────┐
│ 安全防御层次 │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 网络层防护 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ CORS │ │ 限流 │ │ XSS 防护 │ │ 上传体积限制 │ │ │
│ │ │ 跨域控制 │ │ 滑动窗口 │ │ 输入清洗 │ │ DoS 防御 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 认证层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ JWT 认证 │ │ 验证码 │ │ 登录锁定 │ │ Token 黑名单 │ │ │
│ │ │ Bearer │ │ 图形验证 │ │ 失败计数 │ │ 登出失效 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 授权层 │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ │
│ │ │ RBAC 权限 │ │ 演示模式保护 │ │ 超级管理员跳过 │ │ │
│ │ │ permission_ │ │ check_demo │ │ user_id=1 bypass │ │ │
│ │ │ required │ │ │ │ │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 数据层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ 密码加密 │ │ 参数校验 │ │ SQL 注入 │ │ 操作日志 │ │ │
│ │ │ bcrypt │ │ Pydantic │ │ ORM 防护 │ │ 审计追踪 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘JWT 工作流程
用户登录成功后,服务端签发 JWT Token 返回给前端。前端将 Token 存储在本地,后续每次请求通过 Authorization: Bearer <token> 头携带。服务端验证 Token 签名和有效期,解析出用户信息。

[前端] [后端] [Redis]
│ │ │
│── POST /login ─────────→│ │
│ {username, pwd, │ │
│ captcha, uuid} │ │
│ │── 验证码校验 (uuid) ──→│
│ │←─ captcha_text ────────│
│ │ │
│ │── 密码校验 │
│ │── 签发JWT │
│ │── 存储Token ──────────→│
│←─ {token} ─────────────│ │
│ │ │
│── GET /api/v1/xxx ────→│ │
│ Authorization: │ │
│ Bearer <token> │ │
│ │── 验证JWT │
│ │── 检查黑名单 ─────────→│
│ │←─ exists ─────────────│
│←─ {data} ──────────────│ │流程说明
# 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" # 签名算法# 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# 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 加入 Redis 黑名单,TTL 等于 Token 剩余有效期:
# 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="注销成功")验证码防暴力破解
每次登录前需先获取验证码,验证码存储在 Redis 中(TTL 120 秒),登录时校验验证码正确性。验证码使用 UUID 作为 key,避免被猜测。
# 获取验证码
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 分钟:
# 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}")# 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'))用户(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)# 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同步/异步双通道
装饰器根据被装饰函数的类型自动选择校验通道:
async def 端点 → _check_cached 走 Redis 缓存(异步)def 端点 → _check 直查 DB(同步,在 FastAPI 线程池中执行)这样避免了在同步端点中 asyncio.run() 新建事件循环导致的跨 loop 复用 Redis 连接问题。
# 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}")
)# 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_wrapperpermission_required 之后应用FASTAPI_DEMO=True 时,所有写操作返回失败# 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"]# 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 启用# 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)bleach 库白名单过滤危险标签和属性src 属性只允许合法 URL# 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 提前拒绝超大请求体ORM 天然防注入
SQLAlchemy ORM 使用参数化查询(Parameterized Query),所有用户输入通过参数绑定传递,不直接拼接 SQL 字符串,从根本上防止 SQL 注入。
# SQLAlchemy 参数化查询示例
user_repo.get_one(username="admin")
# 实际执行:SELECT * FROM fastapi_user WHERE username = :username AND is_delete = 0
# 参数绑定:{'username': 'admin'}# 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()to_dict() 序列化时自动排除 is_delete、password、salt# 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自动记录所有写操作的日志,包括:
每次登录/登出自动记录到 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 连接密码 |
生产环境安全检查
JWT_SALT 必须使用强随机字符串,不能使用默认值CORS_ORIGINS 必须配置具体的域名,不能使用 *FASTAPI_DEMO 在生产环境应为 False的安全架构涵盖认证(JWT + 验证码 + 登录锁定 + Token 黑名单)、授权(RBAC + permission_required + check_demo)、防护(CORS + 限流 + XSS + 上传限制 + SQL 注入防护)、审计(操作日志 + 登录日志)四大领域。通过纵深防御策略,在网络层、认证层、授权层、数据层分别设置安全屏障,确保系统在面对各类安全威胁时具备足够的防护能力。生产部署时需特别关注 JWT 密钥强度、CORS 配置、HTTPS 启用等关键安全项。