417 lines
11 KiB
Markdown
417 lines
11 KiB
Markdown
# 第四章 架构设计 - 安全架构
|
||
|
||
## 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 安全配置清单
|
||
|
||
### 生产环境检查项
|
||
|
||
| 配置项 | 开发环境 | 生产环境 | 说明 |
|
||
|--------|----------|----------|------|
|
||
| `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` | 强密码 | 数据库访问密码 |
|