Django CSRF保护机制详解与实践
1. Django CSRF防御实现
Django内置通过中间件 django.middleware.csrf.CsrfViewMiddleware 来抵御跨站请求伪造攻击。
表单提交方式
<!-- 在表单中嵌入CSRF令牌 -->
<form method="post" action="/submit/">
{% csrf_token %}
<input type="text" name="username">
<button type="submit">提交</button>
</form>
AJAX提交方式
方法一：将令牌放入请求数据中
$.ajax({
url: '/csrf_test/',
type: 'post',
data: {
'username': $('[name="username"]').val(),
'csrfmiddlewaretoken': $('[name="csrfmiddlewaretoken"]').val()
},
success: function(response) {
console.log('请求成功', response);
},
error: function(error) {
console.log('请求失败', error);
}
});
方法二：使用模板变量直接赋值
$.ajax({
url: '/csrf_test/',
type: 'post',
data: {
'username': $('[name="username"]').val(),
'csrfmiddlewaretoken': '{{ csrf_token }}'
}
});
方法三：将令牌放入请求头
$.ajax({
url: '/csrf_test/',
type: 'post',
headers: {'X-CSRFToken': '{{ csrf_token }}'},
data: {
'username': $('[name="username"]').val()
}
});
2. 全局启用，局部禁用CSRF
from django.views.decorators.csrf import csrf_exempt
@csrf_exempt
def csrf_test(request):
if request.method == 'GET':
return render(request, 'csrf_test.html')
else:
username = request.POST.get('username')
password = request.POST.get('password')
print(f'用户名: {username}, 密码: {password}')
return HttpResponse('登录成功')
3. 全局禁用，局部启用CSRF
from django.views.decorators.csrf import csrf_protect
@csrf_protect
def csrf_test(request):
if request.method == 'GET':
return render(request, 'csrf_test.html')
else:
username = request.POST.get('username')
password = request.POST.get('password')
print(f'用户名: {username}, 密码: {password}')
return HttpResponse('登录成功')
4. 在URL配置中应用CSRF装饰器
from django.urls import path
from django.views.decorators.csrf import csrf_exempt
from . import views
urlpatterns = [
path('csrf_test/', csrf_exempt(views.csrf_test)),
]
5. 完整代码示例
登录认证中间件
# middlewares.py
from django.utils.deprecation import MiddlewareMixin
from django.shortcuts import render, HttpResponse, redirect
from django.urls import reverse
class AuthenticationMiddleware(MiddlewareMixin):
"""登录认证中间件：未登录用户只能访问login和home页面，其他页面需要登录"""
def process_request(self, request):
# 定义白名单URL
allow_urls = ['/login/']
# 检查用户是否已登录
is_logged_in = request.session.get('is_login')
current_path = request.get_full_path()
# 未登录且不在白名单中 - 重定向到登录页并记录原始路径
if not is_logged_in and request.path not in allow_urls:
login_url = reverse('login')
return redirect(f'{login_url}?returnUrl={current_path}')
# 已登录但访问登录页 - 提示先注销
elif is_logged_in and request.path in allow_urls:
return HttpResponse('您已登录，请先注销再重新登录')
# views.py
from django.shortcuts import render, redirect, HttpResponse
from . import models
def login(request):
if request.method == 'GET':
return render(request, 'login.html')
else:
username = request.POST.get('username')
password = request.POST.get('password')
user = models.User.objects.filter(name=username, password=password).first()
if user:
request.session['is_login'] = True
redirect_url = request.GET.get('returnUrl')
if redirect_url:
return redirect(redirect_url)
return redirect('login')
else:
return HttpResponse('用户名或密码错误')
统一处理前端不同编码格式的数据
# middlewares.py
import json
class DataConversionMiddleware(MiddlewareMixin):
"""将不同编码格式的前端数据统一转换为request.data"""
def process_request(self, request):
request.data = {}
content_type = request.META.get('CONTENT_TYPE', '')
# GET请求 - URL编码
if request.method == 'GET' and 'application/x-www-form-urlencoded' in content_type:
for key in request.GET:
request.data[key] = request.GET[key]
# POST请求 - URL编码
elif request.method == 'POST' and 'application/x-www-form-urlencoded' in content_type:
for key in request.POST:
request.data[key] = request.POST[key]
# POST请求 - multipart/form-data (含文件上传)
elif request.method == 'POST' and 'multipart/form-data' in content_type:
for key in request.POST:
request.data[key] = request.POST[key]
# POST请求 - JSON编码
elif request.method == 'POST' and 'application/json' in content_type:
try:
post_data = json.loads(request.body)
for key in post_data:
request.data[key] = post_data[key]
except json.JSONDecodeError:
pass
实现IP访问频率限制（同一IP每分钟最多访问5次）
import time
class RateLimitMiddleware(MiddlewareMixin):
"""同一IP地址每分钟限制访问次数"""
MAX_VISITS = 5
TIME_WINDOW = 60 # 秒
def process_request(self, request):
# 获取客户端IP
client_ip = request.META.get('REMOTE_ADDR', 'unknown')
# 获取该IP的访问记录
visit_times = request.session.get(client_ip, [])
current_time = time.time()
if len(visit_times) < self.MAX_VISITS:
# 访问次数未达到上限，记录本次访问时间
visit_times.append(current_time)
request.session[client_ip] = visit_times
else:
# 检查最早一次访问是否超过时间窗口
if current_time - visit_times[0] < self.TIME_WINDOW:
return HttpResponse('访问过于频繁，请稍后再试')
else:
# 滚动更新访问时间记录
visit_times[0], visit_times[1], visit_times[2], visit_times[3], visit_times[4] = \
visit_times[1], visit_times[2], visit_times[3], visit_times[4], current_time
request.session[client_ip] = visit_times
