#Django安全最佳实践与部署
#课程目标
- 了解Django应用的安全威胁和防护措施
- 掌握Django的安全配置和最佳实践
- 学会部署Django应用到生产环境
- 了解性能优化和监控策略
#Django安全概述
#常见安全威胁
Django应用面临多种安全威胁,主要包括:
- SQL注入(SQL Injection)
- 跨站脚本攻击(XSS)
- 跨站请求伪造(CSRF)
- 点击劫持(Clickjacking)
- 会话固定攻击
- 开放重定向漏洞
#Django内置安全功能
- 自动防XSS:模板系统自动转义HTML
- CSRF保护:内置CSRF中间件
- SQL注入防护:ORM自动转义
- 点击劫持防护:X-Frame-Options头部
- 用户认证和权限系统
#安全配置最佳实践
#settings.py安全配置
# settings.py - 生产环境配置示例
import os
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
# 安全密钥配置
SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY')
# 调试模式 - 生产环境必须设为False
DEBUG = False
# 允许的主机
ALLOWED_HOSTS = [
'yourdomain.com',
'www.yourdomain.com',
'127.0.0.1',
'localhost',
]
# 安全相关设置
SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_SECONDS = 31536000 # 1年
SECURE_REDIRECT_EXEMPT = []
SECURE_SSL_REDIRECT = True # 如果使用HTTPS
SESSION_COOKIE_SECURE = True # HTTPS环境下
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Strict'
CSRF_COOKIE_SECURE = True # HTTPS环境下
CSRF_COOKIE_HTTPONLY = True
CSRF_COOKIE_SAMESITE = 'Strict'
# X-Frame-Options头部
X_FRAME_OPTIONS = 'DENY'
# 内容安全策略
SECURE_CSP_INCLUDE_NONCE_IN = ['script-src']
# 数据库配置 - 使用安全连接
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': os.environ.get('DB_NAME'),
'USER': os.environ.get('DB_USER'),
'PASSWORD': os.environ.get('DB_PASSWORD'),
'HOST': os.environ.get('DB_HOST'),
'PORT': '5432',
'OPTIONS': {
'sslmode': 'require', # 启用SSL
},
}
}#用户认证安全
# settings.py
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
'OPTIONS': {
'min_length': 12, # 最小长度
}
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
# 登录限制配置
from axes.backends import AxesStandaloneBackend
AUTHENTICATION_BACKENDS = [
'axes.backends.AxesStandaloneBackend',
'django.contrib.auth.backends.ModelBackend',
]
# Axes配置(防暴力破解)
AXES_FAILURE_LIMIT = 5
AXES_COOLOFF_TIME = 1 # 小时
AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP = True#输入验证和净化
#表单安全
# forms.py
from django import forms
from django.core.validators import RegexValidator
import html
class SafeContactForm(forms.Form):
# 使用内置验证器
name = forms.CharField(
max_length=100,
validators=[
RegexValidator(
regex=r'^[a-zA-Z\s]+$',
message='姓名只能包含字母和空格'
)
]
)
email = forms.EmailField()
# 防止XSS的自定义清理方法
def clean_message(self):
message = self.cleaned_data['message']
# 移除潜在的恶意HTML标签
from django.utils.html import strip_tags
clean_message = strip_tags(message)
# 或者使用更严格的净化
from django.utils.safestring import mark_safe
from bs4 import BeautifulSoup
soup = BeautifulSoup(message, 'html.parser')
# 只允许安全的标签
allowed_tags = ['p', 'br', 'strong', 'em', 'ul', 'ol', 'li']
for tag in soup.find_all():
if tag.name not in allowed_tags:
tag.unwrap()
return str(soup)
# 验证机器人防护
def clean(self):
cleaned_data = super().clean()
# 检查是否有异常的表单提交模式
message = cleaned_data.get('message', '')
name = cleaned_data.get('name', '')
# 检查是否包含垃圾邮件关键词
spam_keywords = ['viagra', 'casino', 'loan', 'credit']
message_lower = message.lower()
for keyword in spam_keywords:
if keyword in message_lower:
raise forms.ValidationError('消息包含不安全内容')
return cleaned_data#模型层安全
# models.py
from django.db import models
from django.core.validators import MinLengthValidator
import re
class Article(models.Model):
title = models.CharField(
max_length=200,
validators=[MinLengthValidator(5)]
)
content = models.TextField()
author = models.ForeignKey('auth.User', on_delete=models.CASCADE)
created_at = models.DateTimeField(auto_now_add=True)
def clean(self):
"""模型级别的验证"""
# 验证标题不包含恶意内容
malicious_patterns = [
r'<script.*?>.*?</script>', # JavaScript
r'javascript:', # JS协议
r'on\w+\s*=', # 事件处理器
]
for pattern in malicious_patterns:
if re.search(pattern, self.title, re.IGNORECASE):
raise ValidationError('标题包含不安全内容')
def save(self, *args, **kwargs):
"""保存前的安全处理"""
# 清理内容中的恶意代码
self.content = self._sanitize_content(self.content)
super().save(*args, **kwargs)
def _sanitize_content(self, content):
"""净化内容"""
import bleach
# 允许的标签和属性
allowed_tags = ['p', 'br', 'strong', 'em', 'u', 'ol', 'ul', 'li', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6']
allowed_attributes = {
'a': ['href', 'title'],
'img': ['src', 'alt', 'title'],
}
return bleach.clean(
content,
tags=allowed_tags,
attributes=allowed_attributes,
strip=True
)#CSRF和XSS防护
#CSRF防护
# views.py
from django.views.decorators.csrf import csrf_protect, csrf_exempt
from django.middleware.csrf import get_token
from django.http import JsonResponse
# 保护视图(默认行为)
@csrf_protect
def secure_form_view(request):
if request.method == 'POST':
# 处理表单数据
name = request.POST.get('name')
email = request.POST.get('email')
# ... 处理逻辑
return JsonResponse({'status': 'success'})
# 返回CSRF令牌给前端
csrf_token = get_token(request)
return render(request, 'form.html', {'csrf_token': csrf_token})
# AJAX请求的CSRF处理
def ajax_csrf_view(request):
if request.method == 'POST':
# 从请求头获取CSRF令牌
csrf_token = request.META.get('HTTP_X_CSRFTOKEN')
# Django会自动验证
# ... 处理逻辑
return JsonResponse({'status': 'success'})#XSS防护
# utils.py
from django.utils.html import escape, strip_tags
from django.utils.safestring import mark_safe
import bleach
def sanitize_user_input(input_text):
"""净化用户输入"""
# 基础转义
escaped = escape(input_text)
# 使用bleach进行更严格的净化
clean_text = bleach.clean(
escaped,
tags=[], # 不允许任何HTML标签
strip=True # 移除标签
)
return clean_text
def safe_render_markdown(markdown_text):
"""安全渲染Markdown"""
import markdown
from markdown.extensions import Extension
# 解析Markdown
html = markdown.markdown(markdown_text)
# 净化HTML
clean_html = bleach.clean(
html,
tags=['p', 'br', 'strong', 'em', 'ul', 'ol', 'li', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'pre', 'code'],
attributes={
'code': ['class'],
'pre': ['class'],
},
strip=True
)
return mark_safe(clean_html)#部署配置
#生产环境部署
# production_settings.py
from .base import *
import os
# 生产环境特定配置
DEBUG = False
ALLOWED_HOSTS = ['yourdomain.com', 'www.yourdomain.com']
# 数据库配置
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': os.environ.get('DB_NAME'),
'USER': os.environ.get('DB_USER'),
'PASSWORD': os.environ.get('DB_PASSWORD'),
'HOST': os.environ.get('DB_HOST'),
'PORT': '5432',
'CONN_MAX_AGE': 600, # 连接池
}
}
# 静态文件配置
STATIC_ROOT = '/var/www/static/'
STATICFILES_STORAGE = 'django.contrib.staticfiles.storage.ManifestStaticFilesStorage'
# 媒体文件配置
MEDIA_ROOT = '/var/www/media/'
MEDIA_URL = '/media/'
# 日志配置
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'handlers': {
'file': {
'level': 'INFO',
'class': 'logging.FileHandler',
'filename': '/var/log/django/info.log',
},
'error_file': {
'level': 'ERROR',
'class': 'logging.FileHandler',
'filename': '/var/log/django/error.log',
},
},
'loggers': {
'django': {
'handlers': ['file', 'error_file'],
'level': 'INFO',
'propagate': True,
},
},
}#WSGI配置
# wsgi.py - 生产环境配置
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.production_settings')
application = get_wsgi_application()
# 添加错误处理
def application_with_error_handling(environ, start_response):
try:
return application(environ, start_response)
except Exception:
# 记录错误
import logging
logger = logging.getLogger('django')
logger.exception('WSGI application error')
raise#Docker部署配置
# Dockerfile
FROM python:3.11-slim
# 设置工作目录
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
postgresql-client \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 复制requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 创建非root用户
RUN useradd --create-home --shell /bin/bash app
RUN chown -R app:app /app
USER app
# 收集静态文件
RUN python manage.py collectstatic --noinput
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "myproject.wsgi:application"]# docker-compose.yml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- DJANGO_SETTINGS_MODULE=myproject.production_settings
- DB_HOST=db
depends_on:
- db
volumes:
- static_volume:/app/staticfiles
- media_volume:/app/media
db:
image: postgres:15
environment:
- POSTGRES_DB=myproject
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
postgres_data:
static_volume:
media_volume:#性能优化
#缓存配置
# settings.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
}
}
}
# 视图中的缓存使用
from django.core.cache import cache
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator
from django.views.generic import ListView
@cache_page(60 * 15) # 缓存15分钟
def article_list_view(request):
articles = Article.objects.filter(status='published')
return render(request, 'articles/list.html', {'articles': articles})
class CachedArticleListView(ListView):
model = Article
template_name = 'articles/list.html'
context_object_name = 'articles'
@method_decorator(cache_page(60 * 15))
def dispatch(self, *args, **kwargs):
return super().dispatch(*args, **kwargs)
def get_queryset(self):
# 使用缓存的查询集
cache_key = 'published_articles'
articles = cache.get(cache_key)
if articles is None:
articles = Article.objects.filter(status='published').select_related('author')
cache.set(cache_key, articles, 60*15) # 缓存15分钟
return articles#数据库查询优化
# models.py - 优化查询
class ArticleQuerySet(models.QuerySet):
def published(self):
return self.filter(status='published')
def with_counts(self):
return self.annotate(
comment_count=Count('comments'),
like_count=Sum('likes')
)
def for_homepage(self):
return self.published().with_counts().select_related(
'author', 'category'
).prefetch_related(
'tags'
).order_by('-created_at')[:10]
class ArticleManager(models.Manager):
def get_queryset(self):
return ArticleQuerySet(self.model, using=self._db)
def published(self):
return self.get_queryset().published()
def for_homepage(self):
return self.get_queryset().for_homepage()
class Article(models.Model):
objects = ArticleManager()
# ... 其他字段
# views.py - 使用优化的查询
def homepage_view(request):
# 使用优化的查询集
articles = Article.objects.for_homepage()
return render(request, 'homepage.html', {'articles': articles})#监控和日志
#日志配置
# settings.py
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
'style': '{',
},
'simple': {
'format': '{levelname} {message}',
'style': '{',
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'verbose',
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': '/var/log/django/app.log',
'maxBytes': 1024*1024*15, # 15MB
'backupCount': 10,
'formatter': 'verbose',
},
'security_file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': '/var/log/django/security.log',
'maxBytes': 1024*1024*15,
'backupCount': 10,
'formatter': 'verbose',
},
},
'loggers': {
'django': {
'handlers': ['console', 'file'],
'level': 'INFO',
'propagate': False,
},
'django.security': {
'handlers': ['security_file'],
'level': 'INFO',
'propagate': False,
},
'myapp.security': {
'handlers': ['security_file'],
'level': 'INFO',
'propagate': False,
},
},
}#安全监控
# security_monitor.py
import logging
from django.contrib.auth.signals import user_logged_in, user_login_failed
from django.dispatch import receiver
from django.http import HttpRequest
logger = logging.getLogger('myapp.security')
@receiver(user_logged_in)
def log_user_login(sender, request, user, **kwargs):
"""记录用户登录"""
ip = get_client_ip(request)
user_agent = request.META.get('HTTP_USER_AGENT', '')
logger.info(
f'User login: {user.username} from {ip}, User-Agent: {user_agent}'
)
@receiver(user_login_failed)
def log_user_login_failed(sender, credentials, request, **kwargs):
"""记录登录失败"""
ip = get_client_ip(request)
username = credentials.get('username', 'unknown')
logger.warning(
f'Login failed: {username} from {ip}, Attempted path: {request.path}'
)
def get_client_ip(request: HttpRequest) -> str:
"""获取客户端IP"""
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0]
else:
ip = request.META.get('REMOTE_ADDR')
return ip
# 自定义中间件监控异常请求
class SecurityMonitoringMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
return response
def process_exception(self, request, exception):
"""记录异常"""
if 'sql' in str(exception).lower() or 'syntax' in str(exception).lower():
# 可能的SQL注入尝试
ip = get_client_ip(request)
logger.warning(f'Potential SQL injection from {ip}: {exception}')
return None#部署检查清单
#部署前检查
- DEBUG = False
- SECRET_KEY 安全存储
- ALLOWED_HOSTS 正确配置
- 数据库连接安全
- 静态文件正确配置
- SSL证书配置
- 安全头部设置
- 日志配置完成
- 备份策略就绪
- 监控系统就绪
#生产环境最佳实践
# gunicorn_config.py
import multiprocessing
# 服务器配置
bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "sync"
worker_connections = 1000
timeout = 30
keepalive = 2
# 日志配置
accesslog = "/var/log/gunicorn/access.log"
errorlog = "/var/log/gunicorn/error.log"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'
# 进程命名
proc_name = 'myproject'
# 预加载应用
preload_app = True#课程总结
本节课我们学习了Django应用的安全最佳实践和部署策略,涵盖了安全配置、输入验证、CSRF和XSS防护、生产环境部署、性能优化和监控等方面。安全是Web应用开发的重要组成部分,通过合理的配置和最佳实践,我们可以构建安全、可靠、高性能的Django应用。

