安全架构设计

本章详细阐述系统的安全架构设计，围绕认证、授权、防护、审计四大安全领域展开，旨在帮助开发者全面理解系统的安全保障体系，确保在开发与部署过程中遵循安全规范，有效防范各类安全风险。

认证（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 表，确保后台任务的执行过程可审计。

安全设计原则

纵深防御：认证、授权、参数校验多层防护，单一防线失效时仍有其他机制兜底

最小权限：用户仅能访问已授权接口，角色权限按需分配，遵循"够用即可"原则

数据不落地：敏感信息（Token、密码）不在日志中明文记录，密码哈希不可逆

安全默认：框架默认配置为安全模式（如 DEBUG 关闭），如需放宽需显式配置

安全架构总览

┌─────────────────────────────────────────────────────────────────────┐
│ 安全防御层次                                                        │
│                                                                     │
│ ┌───────────────────────────────────────────────────────────────┐  │
│ │ 网络层防护                                                    │  │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐     │  │
│ │ │ CORS     │ │ XSS 防护 │ │ 上传体积限制 │     │  │
│ │ │ 跨域控制 │ │ 滑动窗口 │ │ 输入清洗 │ │ DoS 防御     │     │  │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘     │  │
│ └───────────────────────────────────────────────────────────────┘  │
│                                                                     │
│ ┌───────────────────────────────────────────────────────────────┐  │
│ │ 认证层                                                        │  │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐     │  │
│ │ │ JWT 认证 │ │ 验证码   │ │ 登录锁定 │ │ Token 黑名单 │     │  │
│ │ │ Bearer   │ │ 图形验证 │ │ 失败计数 │ │ 登出失效     │     │  │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘     │  │
│ └───────────────────────────────────────────────────────────────┘  │
│                                                                     │
│ ┌───────────────────────────────────────────────────────────────┐  │
│ │ 授权层                                                        │  │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐   │  │
│ │ │ RBAC 权限    │ │ 演示模式保护 │ │ 超级管理员跳过       │   │  │
│ │ │ check_       │ │ check_demo   │ │ user_id=1 bypass     │   │  │
│ │ │ permission   │ │              │ │                      │   │  │
│ │ └──────────────┘ └──────────────┘ └──────────────────────┘   │  │
│ └───────────────────────────────────────────────────────────────┘  │
│                                                                     │
│ ┌───────────────────────────────────────────────────────────────┐  │
│ │ 数据层                                                        │  │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐     │  │
│ │ │ 密码加密 │ │ 参数校验 │ │ SQL 注入 │ │ 操作日志     │     │  │
│ │ │ bcrypt   │ │ WTForms  │ │ ORM 防护 │ │ 审计追踪     │     │  │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘     │  │
│ └───────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────┘

一、认证机制

JWT 令牌认证

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} ──────────────│                         │

流程说明

1. 前端提交用户名、密码、验证码及 UUID
2. 后端校验验证码，验证密码，生成 JWT 并缓存至 Redis
3. 前端后续请求携带 JWT，后端校验 Token 有效性并检查黑名单
4. 校验通过后返回业务数据

JWT 配置

# 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 签发

# 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 验证

# 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 剩余有效期：

# 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="注销成功")

登录验证码

验证码防暴力破解

每次登录前需先获取验证码，验证码存储在 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
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']}")

密码安全

# 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 装饰器

# 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

权限缓存

# 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 装饰器

# 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 跨域控制

# 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 防护

# 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

上传体积限制

# 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 注入防护

ORM 天然防注入

SQLAlchemy ORM 使用参数化查询（Parameterized Query），所有用户输入通过参数绑定传递，不直接拼接 SQL 字符串，从根本上防止 SQL 注入。

# SQLAlchemy 参数化查询示例
UserRepository.first_by({'username': 'admin'})
# 实际执行：SELECT * FROM flask_user WHERE username = :username AND is_delete = 0
# 参数绑定：{'username': 'admin'}

参数校验

# 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 密钥通过环境变量配置，不硬编码

五、审计追踪

操作日志

# 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 连接密码

生产环境安全检查

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 启用等关键安全项。