Skip to content

第四章 架构设计 - 安全架构

4.1 安全架构概述

系统从认证、授权、防护三个维度构建安全体系:

+------------------+
|   安全架构       |
+------------------+
        |
        +-- 认证层 (Authentication)
        |   +-- JWT Token 认证
        |   +-- Token 黑名单机制
        |   +-- 密码哈希存储
        |
        +-- 授权层 (Authorization)
        |   +-- RBAC 权限模型
        |   +-- 权限节点控制
        |   +-- 超级管理员机制
        |
        +-- 防护层 (Protection)
            +-- CORS 跨域配置
            +-- 演示模式保护
            +-- SQL注入防护
            +-- XSS防护

4.2 JWT 认证机制

4.2.1 Token 生成

用户登录成功后,系统生成 JWT Token 返回给前端:

python
# utils/jwt.py
import jwt
import datetime

JWT_SALT = os.environ.get('JWT_SALT', None)
JWT_ALGORITHM = 'HS256'
DEFAULT_TIMEOUT_MINUTES = 20

def create_token(payload, timeout=None):
    """生成JWT令牌"""
    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
    )
    return token

4.2.2 Token 验证

每次请求都需要验证 Token 的有效性:

python
def parse_payload(token):
    """解析验证JWT令牌"""
    result = {"code": 0, "data": None, "msg": "操作成功"}

    if not token:
        result['code'] = -1
        result['msg'] = "token不能为空"
        return result

    # 检查Token是否在黑名单中
    if is_token_blacklisted(token):
        result['code'] = -1
        result['msg'] = "token已失效,请重新登录"
        return result

    try:
        # 解密验证JWT令牌
        verified_payload = jwt.decode(
            token, JWT_SALT, algorithms=[JWT_ALGORITHM]
        )
        result['data'] = verified_payload
    except exceptions.ExpiredSignatureError:
        result['code'] = -1
        result['msg'] = "token已失效,请重新登录"
    except exceptions.DecodeError:
        result['code'] = -1
        result['msg'] = "token认证失败,无效的令牌格式"
    except exceptions.InvalidTokenError:
        result['code'] = -1
        result['msg'] = "非法的token,请检查令牌有效性"

    return result

4.2.3 Token 黑名单

退出登录时将 Token 加入 Redis 黑名单,使 Token 立即失效:

python
_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:'

def add_token_to_blacklist(token: str) -> bool:
    """将Token加入黑名单"""
    from django_redis import get_redis_connection
    redis = get_redis_connection("default")

    # 获取Token剩余有效期作为TTL
    expiration = get_token_expiration(token)
    if expiration is None:
        ttl_seconds = DEFAULT_TIMEOUT_MINUTES * 60
    else:
        ttl_seconds = int(
            (expiration - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds()
        )

    # 使用SHA256指纹作为Key
    fingerprint = hashlib.sha256(token.encode('utf-8')).hexdigest()
    key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
    redis.set(key, '1', ex=ttl_seconds)

    return True

def is_token_blacklisted(token: str) -> bool:
    """检查Token是否在黑名单中"""
    from django_redis import get_redis_connection
    redis = get_redis_connection("default")

    fingerprint = hashlib.sha256(token.encode('utf-8')).hexdigest()
    key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
    return redis.exists(key) > 0

4.2.4 JWT 密钥安全

python
def _validate_key_strength(key: str) -> None:
    """启动时校验JWT密钥强度"""
    if not key:
        raise RuntimeError(
            "JWT_SALT 未配置!请在 .env 中设置至少32字节的强随机密钥。"
        )

    key_bytes = len(key.encode('utf-8'))
    if key_bytes < 32:
        logger.warning(f"JWT密钥长度不足:{key_bytes} 字节,建议至少 32 字节。")

# 启动时校验密钥强度
_validate_key_strength(JWT_SALT)

4.3 RBAC 权限模型

4.3.1 权限模型设计

用户 (User)
    |
    +-- 用户角色关联 (UserRole)
        |
        +-- 角色 (Role)
            |
            +-- 角色菜单关联 (RoleMenu)
                |
                +-- 菜单/权限 (Menu)
                    |
                    +-- 权限标识: sys:<module>:<action>

4.3.2 权限节点格式

每个操作对应一个权限节点,格式为 sys:<module>:<action>

python
# 示例权限节点
"sys:example:page"      # 案例分页查询
"sys:example:detail"    # 案例详情查询
"sys:example:add"       # 案例添加
"sys:example:update"    # 案例更新
"sys:example:delete"    # 案例删除
"sys:example:status"    # 案例状态更新

4.3.3 权限验证实现

python
# middleware/permission_middleware.py
from django.contrib.auth.mixins import PermissionRequiredMixin
from utils import R
from utils.security import get_user_id

class PermissionRequired(PermissionRequiredMixin):
    """自定义权限控制混入类"""

    def has_permission(self):
        """检查用户是否拥有所需权限"""
        permissions = self.get_permission_required()
        user_id = get_user_id(self.request)

        # 超级管理员(ID=1)自动放行
        if user_id and user_id != 1:
            from application.menu import services
            permission_list = services.get_user_permissions(user_id)

            for permission in permissions:
                if permission not in permission_list:
                    return False

        return True

    def handle_no_permission(self):
        """无权限访问时的处理"""
        return R.failed("暂无操作权限", 401)

4.3.4 超级管理员机制

用户 ID 为 1 的用户被视为超级管理员,自动跳过所有权限检查:

python
# 超级管理员自动放行
if user_id and user_id != 1:
    # 普通用户需要验证权限
    permission_list = services.get_user_permissions(user_id)
    # ...
else:
    # 超级管理员直接放行
    return True

4.4 CORS 跨域配置

4.4.1 配置说明

python
# application/settings.py

# 允许跨域请求携带凭证
CORS_ALLOW_CREDENTIALS = True

# 允许所有域名跨域访问(开发环境)
CORS_ORIGIN_ALLOW_ALL = True

# 允许跨域的正则表达式匹配规则
CORS_ALLOWED_ORIGINS_REGEXES = [
    r'^http://.*?$',
]

# 允许跨域的HTTP请求方法
CORS_ALLOW_METHODS = (
    'DELETE', 'GET', 'OPTIONS', 'PATCH', 'POST', 'PUT', 'VIEW',
)

# 允许跨域的HTTP请求头
CORS_ALLOW_HEADERS = (
    'XMLHttpRequest', 'X_FILENAME', 'accept-encoding',
    'authorization',  # JWT认证头
    'content-type', 'dnt', 'origin', 'user-agent',
    'x-csrftoken', 'x-requested-with', 'Pragma',
)

4.4.2 中间件顺序

CORS 中间件必须放在最前面:

python
MIDDLEWARE = [
    'corsheaders.middleware.CorsMiddleware',  # 必须放在最前面
    # ... 其他中间件
]

4.5 演示模式保护

4.5.1 配置说明

通过环境变量 DJANGO_DEMO 控制演示模式:

python
# config/env.py
DJANGO_DEMO = (os.getenv('DJANGO_DEMO', 'True') == 'True')

4.5.2 演示模式限制

在演示模式下,所有写操作(添加、更新、删除)都会被拦截:

python
# application/example/views.py
from config.env import DJANGO_DEMO

class ExampleAddView(PermissionRequired, View):
    def post(self, request):
        # 演示环境禁止操作
        if DJANGO_DEMO:
            return R.failed("演示环境,暂无操作权限")

        result = services.add_example(request)
        return result

4.5.3 受保护的操作

操作HTTP方法演示模式行为
添加POST拦截
更新PUT拦截
删除DELETE拦截
状态更新PUT拦截
查询GET放行

4.6 密码安全

4.6.1 密码哈希存储

密码使用 Django 内置的密码哈希机制,支持多种哈希算法:

python
# utils/password.py
from django.contrib.auth.hashers import make_password, check_password

def hash_password(password):
    """哈希密码"""
    return make_password(password)

def verify_password(password, hashed_password):
    """验证密码"""
    return check_password(password, hashed_password)

4.6.2 密码验证器

python
# application/settings.py
AUTH_PASSWORD_VALIDATORS = [
    {'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
    {'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'},
    {'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
    {'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]

4.7 SQL 注入防护

4.7.1 Django ORM 防护

Django ORM 自动使用参数化查询,防止 SQL 注入:

python
# ORM查询 - 自动参数化,安全
Example.objects.filter(name__contains=name)
Example.objects.filter(id=instance_id, is_delete=False)

4.7.2 原始 SQL 注意事项

在使用原始 SQL 时,必须使用参数化查询:

python
# 正确 - 参数化查询
sql = 'SELECT m.* FROM django_menu AS m WHERE ur.user_id=%s'
menu_list = Menu.objects.raw(sql, [user_id])

# 错误 - 字符串拼接(存在SQL注入风险)
# sql = f'SELECT m.* FROM django_menu AS m WHERE ur.user_id={user_id}'

4.8 CSRF 防护

4.8.1 CSRF 禁用说明

由于系统使用 JWT 进行认证,CSRF 中间件已禁用:

python
MIDDLEWARE = [
    # CSRF中间件已注释,使用JWT进行认证
    # 'django.middleware.csrf.CsrfViewMiddleware',
]

4.8.2 安全替代方案

JWT Token 通过 Authorization 请求头传递,天然免疫 CSRF 攻击:

python
headers = {
    "Authorization": "Bearer <token>"
}

4.9 安全配置清单

生产环境检查项

配置项开发环境生产环境说明
DEBUGTrueFalse必须关闭调试模式
SECRET_KEY默认值随机强密钥必须修改
ALLOWED_HOSTS['*']具体域名限制访问来源
CORS_ORIGIN_ALLOW_ALLTrueFalse限制跨域来源
DJANGO_DEMOTrueFalse关闭演示模式
JWT_SALT默认值随机强密钥至少32字节
REDIS_PASSWORD123456强密码Redis访问密码
DATABASE_PASSWORDroot强密码数据库访问密码

小蚂蚁云团队 · 提供技术支持