Files
djangoadmin-wiki/DjangoAdmin_Flask_EleVue/zh/volume3/security.md
T
2026-09-13 20:29:19 +08:00

548 lines
24 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 安全架构设计
本章详细阐述系统的安全架构设计,围绕认证、授权、防护、审计四大安全领域展开,旨在帮助开发者全面理解系统的安全保障体系,确保在开发与部署过程中遵循安全规范,有效防范各类安全风险。
**认证(Authentication)**:系统采用基于 JWT(JSON Web Token)的无状态认证机制,用户通过用户名密码登录后获取 Token,后续请求通过 `Authorization: Bearer <token>` 头传递,由 `login_required` 中间件统一拦截并验证 Token 的有效性(签名校验、有效期检查、Redis 黑名单过滤)。同时,系统内置登录失败锁定策略,连续失败 5 次即锁定账号 10 分钟,有效防御暴力破解攻击。
**授权(Authorization)**:基于 RBAC(基于角色的访问控制)模型设计,权限控制粒度细化至接口级。用户通过角色继承权限节点(如 `sys:position:add`),由 `@check_permission` 装饰器在路由层统一拦截校验,权限数据采用 Redis 缓存加速查询,管理员(`userId == 1`)自动放行。权限变更时,相关缓存主动失效,确保权限更新的实时性。
**防护(Protection)**:系统从多维度构建安全防护体系。参数校验通过 WTForms Schema 自动拦截非法输入,防止 SQL 注入和 XSS 攻击;敏感密码采用 bcrypt 加盐哈希存储,即使数据库泄露也无法还原明文;演示模式下 `@check_demo` 装饰器统一拦截写操作,保护测试数据不被污染;跨域配置精确控制可信域名,防止 CSRF 攻击。
**审计(Audit)**:系统提供完备的操作审计能力。操作日志中间件自动记录所有写操作(POST/PUT/DELETE/PATCH)的请求路径、参数、操作人、IP 地址及执行结果,存入 `flask_operation_log` 表;登录日志独立记录每次登录尝试(含 IP、User-Agent、登录状态),便于安全事件追溯与异常行为分析;定时任务每次执行记录至 `flask_job_log` 表,确保后台任务的执行过程可审计。
:::tip 安全设计原则
纵深防御:认证、授权、参数校验多层防护,单一防线失效时仍有其他机制兜底
最小权限:用户仅能访问已授权接口,角色权限按需分配,遵循"够用即可"原则
数据不落地:敏感信息(Token、密码)不在日志中明文记录,密码哈希不可逆
安全默认:框架默认配置为安全模式(如 DEBUG 关闭),如需放宽需显式配置
:::
## 安全架构总览
```
┌─────────────────────────────────────────────────────────────────────┐
│ 安全防御层次 │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 网络层防护 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ CORS │ │ XSS 防护 │ │ 上传体积限制 │ │ │
│ │ │ 跨域控制 │ │ 滑动窗口 │ │ 输入清洗 │ │ DoS 防御 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 认证层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ JWT 认证 │ │ 验证码 │ │ 登录锁定 │ │ Token 黑名单 │ │ │
│ │ │ Bearer │ │ 图形验证 │ │ 失败计数 │ │ 登出失效 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 授权层 │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ │
│ │ │ RBAC 权限 │ │ 演示模式保护 │ │ 超级管理员跳过 │ │ │
│ │ │ check_ │ │ check_demo │ │ user_id=1 bypass │ │ │
│ │ │ permission │ │ │ │ │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 数据层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ 密码加密 │ │ 参数校验 │ │ SQL 注入 │ │ 操作日志 │ │ │
│ │ │ bcrypt │ │ WTForms │ │ ORM 防护 │ │ 审计追踪 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
```
## 一、认证机制
### JWT 令牌认证
:::tip JWT 工作流程
用户登录成功后,服务端签发 JWT Token 返回给前端。前端将 Token 存储在本地,后续每次请求通过 `Authorization: Bearer <token>` 头携带。服务端验证 Token 签名和有效期,解析出用户信息。
:::
1. 工作流程时序图
```
[前端] [后端] [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
# 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令牌
参数:
payload (dict): JWT载荷数据,通常包含用户ID、用户名等信息
timeout (int): 令牌过期时间(分钟),默认使用DEFAULT_TIMEOUT_MINUTES
返回:
str: 生成的JWT令牌字符串
"""
try:
if timeout is None:
timeout = DEFAULT_TIMEOUT_MINUTES
payload_copy = payload.copy()
current_time = datetime.datetime.now(tz=datetime.timezone.utc)
payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout)
payload_copy['iat'] = current_time
headers = {
"typ": "JWT",
"alg": JWT_ALGORITHM
}
token = jwt.encode(
payload=payload_copy,
key=JWT_SALT,
algorithm=JWT_ALGORITHM,
headers=headers
)
logger.debug(f"生成JWT令牌成功,过期时间:{timeout}分钟")
return token
except Exception as e:
logger.error(f"生成JWT令牌失败: {str(e)}")
raise
```
#### Token 验证
```python
# src/middleware/auth.py
def login_required():
"""登录验证中间件:提取并校验 Token,校验用户有效,写入 g 对象"""
from flask import request, g
success, access_token, msg = get_access_token()
if not success:
return R.failed(msg=msg, code=401)
if is_token_blacklisted(access_token):
return R.failed(msg="登录已注销", code=401)
result = parse_payload(access_token)
if result['code'] != 0:
return R.failed(msg=result['msg'], code=401)
data = result['data']
user_id = int(data.get('userId', 0))
# 用户有效状态校验
from src.modules.system.user.repository import UserRepository
user = UserRepository.get_by_id(user_id)
if not user or user.status != 1:
return R.failed(msg="用户已被禁用或删除,请联系管理员", code=401)
g.user_id = user_id
g.username = data.get('username', '')
g.realname = data.get('realname', '')
```
#### Token 黑名单
用户登出时,将当前 Token 加入 Redis 黑名单,TTL 等于 Token 剩余有效期:
```python
# src/modules/auth/router.py
@auth_blue.route('/logout', methods=['GET'])
def logout():
"""用户退出登录"""
success, access_token, _ = get_access_token()
if success and access_token:
add_token_to_blacklist(access_token)
# 记录退出登录日志
username = get_username()
realname = get_realname()
record_login_log(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
def login(data):
redis = get_redis()
# 检查锁定状态
lock_key = f"login:lock:{data['username']}"
is_locked = redis.get(lock_key)
if is_locked:
return R.failed("账号已锁定,请10分钟后再试")
# 验证用户名密码
user = UserRepository.first_by({'username': data['username']})
if not user or not verify_password(data['password'], user.password):
# 原子递增失败计数
fail_key = f"login:fail:{data['username']}"
count = incr_with_expire(redis, fail_key, 300)
if count >= 5:
redis.set(lock_key, 1, ex=600) # 锁定 10 分钟
return R.failed(f"用户名或密码错误,剩余{5-count}次机会")
# 登录成功,清除失败计数
redis.delete(f"login:fail:{data['username']}")
```
### 密码安全
```python
# src/core/password.py
import secrets
import bcrypt
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)
### check_permission 装饰器
```python
# src/core/permission.py
from functools import wraps
from flask import request
from src.core import response as R
from src.core.security import get_user_id
def check_permission(permission: str):
"""节点权限鉴权装饰器,userId==1(管理员)直接放行"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
user_id = get_user_id()
if user_id == 1:
return func(*args, **kwargs)
from src.modules.system.menu.service import get_permissions_list
permission_list = get_permissions_list(user_id)
if permission not in permission_list:
return R.failed("权限不足")
return func(*args, **kwargs)
return wrapper
return decorator
```
### 权限缓存
```python
# src/modules/system/menu/service.py
def get_permissions_list(user_id: int) -> list:
"""获取用户权限列表(走 Redis 缓存)"""
cache_key = f"perm:user:{user_id}"
redis = get_redis()
cached = redis.get(cache_key)
if cached:
return json.loads(cached)
# 从数据库查询
permissions = query_user_permissions(user_id)
redis.set(cache_key, json.dumps(permissions), ex=3600)
return permissions
def invalidate_perm_cache(user_id: int):
"""失效用户权限缓存(角色/菜单变更时调用)"""
from src.core.redis_client import get_redis
redis = get_redis()
redis.delete(f"perm:user:{user_id}")
```
### check_demo 装饰器
```python
# src/middleware/demo.py
from functools import wraps
from flask import current_app
from src.core import response as R
def check_demo(func):
"""演示模式拦截装饰器,放在 @check_permission 之后、路由函数之前"""
@wraps(func)
def wrapper(*args, **kwargs):
if current_app.config.get('FLASK_DEMO'):
return R.failed("演示环境,暂无操作权限")
return func(*args, **kwargs)
return wrapper
```
- 保护所有写操作(add / update / delete / status / batchDelete / import)
- 在 `check_permission` 之后应用
- `FLASK_DEMO=True` 时,所有写操作返回失败
## 三、网络层防护
### CORS 跨域控制
```python
# src/app.py - create_app()
from flask_cors import CORS
from config import CORS_ALLOWED_ORIGINS
if CORS_ALLOWED_ORIGINS:
allowed_origins = [origin.strip() for origin in CORS_ALLOWED_ORIGINS.split(',') if origin.strip()]
CORS(app, resources={r"/*": {"origins": allowed_origins}}, supports_credentials=True)
else:
CORS(app, resources={r"/*": {"origins": "*"}}, supports_credentials=False)
```
- 生产环境应配置具体的 `origins`,避免使用 `*`
- 开发环境可配置为 `http://localhost:8001`
### XSS 防护
```python
# src/common/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/modules/upload/service.py
from config import UPLOAD_MAX_SIZE_MB
max_size_bytes = UPLOAD_MAX_SIZE_MB * 1024 * 1024
if file_size > max_size_bytes:
raise BizError(f"文件大小超过限制,最大允许 {UPLOAD_MAX_SIZE_MB}MB")
```
- 在业务层校验文件大小,通过 `UPLOAD_MAX_SIZE_MB` 环境变量配置上限
- 前端配合 `el-upload` 组件的 `beforeUpload` 钩子做前置校验
## 四、数据层防护
### SQL 注入防护
:::tip ORM 天然防注入
SQLAlchemy ORM 使用参数化查询(Parameterized Query),所有用户输入通过参数绑定传递,不直接拼接 SQL 字符串,从根本上防止 SQL 注入。
:::
```python
# SQLAlchemy 参数化查询示例
UserRepository.first_by({'username': 'admin'})
# 实际执行:SELECT * FROM flask_user WHERE username = :username AND is_delete = 0
# 参数绑定:{'username': 'admin'}
```
### 参数校验
```python
# WTForms 声明式校验
class PositionForm(BaseForm):
name = StringField('岗位名称', validators=[
DataRequired(message='岗位名称不能为空'),
Length(min=1, max=150, message='岗位名称长度必须在1-150个字符之间')
])
status = IntegerField('岗位状态', validators=[
DataRequired(message='岗位状态不能为空'),
NumberRange(min=1, max=2, message='岗位状态必须是1或2')
])
sort = IntegerField('岗位排序', validators=[
DataRequired(message='岗位排序不能为空'),
NumberRange(min=0, max=99999, message='岗位排序必须在0-99999之间')
])
```
- 所有请求参数通过 WTForms 声明式校验
- 自动拒绝不符合约束的请求
- 校验错误统一返回格式
### 敏感字段保护
- 密码使用 bcrypt 哈希存储,不可逆
- `to_dict()` 序列化时自动排除 `is_delete`、`password`、`salt`
- JWT 密钥通过环境变量配置,不硬编码
## 五、审计追踪
### 操作日志
```python
# src/middleware/operation_log.py
from flask import request, g
def operation_log():
if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
# 记录操作日志
log_data = {
'username': g.username,
'module': extract_module(request.path),
'action': request.method,
'url': request.url,
'params': get_request_params(request),
'ip': get_client_ip(),
'status': 1 if is_success else 2,
'duration': int(duration * 1000),
}
save_operation_log(log_data)
```
自动记录所有写操作的日志,包括:
- 操作人、操作模块、操作动作
- 请求 URL、请求参数
- 操作 IP、操作时间
- 响应状态、执行耗时
### 登录日志
每次登录/登出自动记录到 `flask_login_log` 表,包括登录账号、IP、浏览器、操作系统、登录状态等。
## 安全配置清单
| 配置项 | 环境变量 | 默认值 | 说明 |
|:---|:---|:---|:---|
| JWT 密钥 | `JWT_SALT` | 必须配置 | HS256 签名密钥 |
| JWT 过期时间 | `JWT_EXPIRE_MINUTES` | 120 | Token 有效期(分钟) |
| 密码错误锁定 | — | 5 次/10 分钟 | 连续失败后锁定 |
| 验证码有效期 | — | 120 秒 | 验证码 Redis TTL |
| CORS 源 | `CORS_ALLOWED_ORIGINS` | * | 允许的跨域源 |
| 上传限制 | `UPLOAD_MAX_SIZE_MB` | 10MB | 最大上传体积 |
| 演示模式 | `FLASK_DEMO` | False | 锁定写操作 |
| 数据库密码 | `DB_PASSWORD` | — | 数据库连接密码 |
| Redis 密码 | `REDIS_PASSWORD` | — | Redis 连接密码 |
:::warning 生产环境安全检查
1. `JWT_SALT` 必须使用强随机字符串,不能使用默认值
2. `CORS_ALLOWED_ORIGINS` 必须配置具体的域名,不能使用 `*`
3. `FLASK_DEMO` 在生产环境应为 `False`
4. 数据库和 Redis 必须设置密码
5. 启用 HTTPS(由 Nginx/负载均衡器处理)
:::
## 总结
项目的安全架构涵盖认证(JWT + 验证码 + 登录锁定 + Token 黑名单)、授权(RBAC + check_permission + check_demo)、防护(CORS + XSS + 上传限制 + SQL 注入防护)、审计(操作日志 + 登录日志)四大领域。通过纵深防御策略,在网络层、认证层、授权层、数据层分别设置安全屏障,确保系统在面对各类安全威胁时具备足够的防护能力。生产部署时需特别关注 JWT 密钥强度、CORS 配置、HTTPS 启用等关键安全项。