Become a sponsor

系统从认证、授权、防护三个维度构建安全体系:
+------------------+
| 安全架构 |
+------------------+
|
+-- 认证层 (Authentication)
| +-- JWT Token 认证
| +-- Token 黑名单机制
| +-- 密码哈希存储
|
+-- 授权层 (Authorization)
| +-- RBAC 权限模型
| +-- 权限节点控制
| +-- 超级管理员机制
|
+-- 防护层 (Protection)
+-- CORS 跨域配置
+-- 演示模式保护
+-- SQL注入防护
+-- XSS防护用户登录成功后,系统生成 JWT Token 返回给前端:
# 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每次请求都需要验证 Token 的有效性:
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退出登录时将 Token 加入 Redis 黑名单,使 Token 立即失效:
_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) > 0def _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)用户 (User)
|
+-- 用户角色关联 (UserRole)
|
+-- 角色 (Role)
|
+-- 角色菜单关联 (RoleMenu)
|
+-- 菜单/权限 (Menu)
|
+-- 权限标识: sys:<module>:<action>每个操作对应一个权限节点,格式为 sys:<module>:<action>:
# 示例权限节点
"sys:example:page" # 案例分页查询
"sys:example:detail" # 案例详情查询
"sys:example:add" # 案例添加
"sys:example:update" # 案例更新
"sys:example:delete" # 案例删除
"sys:example:status" # 案例状态更新# 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)用户 ID 为 1 的用户被视为超级管理员,自动跳过所有权限检查:
# 超级管理员自动放行
if user_id and user_id != 1:
# 普通用户需要验证权限
permission_list = services.get_user_permissions(user_id)
# ...
else:
# 超级管理员直接放行
return True# 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',
)CORS 中间件必须放在最前面:
MIDDLEWARE = [
'corsheaders.middleware.CorsMiddleware', # 必须放在最前面
# ... 其他中间件
]通过环境变量 DJANGO_DEMO 控制演示模式:
# config/env.py
DJANGO_DEMO = (os.getenv('DJANGO_DEMO', 'True') == 'True')在演示模式下,所有写操作(添加、更新、删除)都会被拦截:
# 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| 操作 | HTTP方法 | 演示模式行为 |
|---|---|---|
| 添加 | POST | 拦截 |
| 更新 | PUT | 拦截 |
| 删除 | DELETE | 拦截 |
| 状态更新 | PUT | 拦截 |
| 查询 | GET | 放行 |
密码使用 Django 内置的密码哈希机制,支持多种哈希算法:
# 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)# 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'},
]Django ORM 自动使用参数化查询,防止 SQL 注入:
# ORM查询 - 自动参数化,安全
Example.objects.filter(name__contains=name)
Example.objects.filter(id=instance_id, is_delete=False)在使用原始 SQL 时,必须使用参数化查询:
# 正确 - 参数化查询
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}'由于系统使用 JWT 进行认证,CSRF 中间件已禁用:
MIDDLEWARE = [
# CSRF中间件已注释,使用JWT进行认证
# 'django.middleware.csrf.CsrfViewMiddleware',
]JWT Token 通过 Authorization 请求头传递,天然免疫 CSRF 攻击:
headers = {
"Authorization": "Bearer <token>"
}| 配置项 | 开发环境 | 生产环境 | 说明 |
|---|---|---|---|
DEBUG | True | False | 必须关闭调试模式 |
SECRET_KEY | 默认值 | 随机强密钥 | 必须修改 |
ALLOWED_HOSTS | ['*'] | 具体域名 | 限制访问来源 |
CORS_ORIGIN_ALLOW_ALL | True | False | 限制跨域来源 |
DJANGO_DEMO | True | False | 关闭演示模式 |
JWT_SALT | 默认值 | 随机强密钥 | 至少32字节 |
REDIS_PASSWORD | 123456 | 强密码 | Redis访问密码 |
DATABASE_PASSWORD | root | 强密码 | 数据库访问密码 |