如何在Django的SimpleJWT基础上实现条件式双因素认证？

以下是针对现有Django+SimpleJWT+React架构的具体改造步骤，核心逻辑是登录时验证当前设备/浏览器/IP与用户历史记录，不一致则触发邮件OTP验证：

1. 安装依赖 #

首先安装解析用户代理和生成OTP的工具库：

pip install django-user-agents pyotp

2. 核心工具函数：获取客户端信息 #

新增一个工具函数，用于从请求中提取IP、浏览器和设备信息：

from user_agents import parse
from django.conf import settings

def get_client_info(request):
    # 获取客户端IP（兼容代理场景）
    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
    ip = x_forwarded_for.split(',')[0] if x_forwarded_for else request.META.get('REMOTE_ADDR')
    
    # 解析User-Agent获取浏览器和设备
    user_agent = parse(request.META.get('HTTP_USER_AGENT', ''))
    browser = f"{user_agent.browser.family} {user_agent.browser.version_string}"
    device = user_agent.device.family or "Unknown Device"
    
    return ip, browser, device

3. 改造登录View：添加2FA触发逻辑 #

修改原有CookieTokenObtainPairView，在账号密码验证通过后，检查是否需要触发2FA：

from rest_framework.response import Response
from rest_framework import status
from django.core.cache import cache
from django.core.mail import send_mail
from rest_framework_simplejwt.views import TokenObtainPairView
from .serializers import CookieTokenObtainPairSerializer

class CookieTokenObtainPairView(TokenObtainPairView):
    serializer_class = CookieTokenObtainPairSerializer

    def post(self, request, *args, **kwargs):
        # 第一步：验证账号密码
        serializer = self.get_serializer(data=request.data)
        try:
            serializer.is_valid(raise_exception=True)
        except Exception as e:
            return Response({"detail": str(e)}, status=status.HTTP_401_UNAUTHORIZED)
        
        user = serializer.user
        current_ip, current_browser, current_device = get_client_info(request)
        
        # 检查是否需要2FA：对比历史记录与当前信息
        need_2fa = (
            user.last_login_location != current_ip
            or user.last_login_browser != current_browser
            or user.last_login_device != current_device
        )

        if not need_2fa:
            # 无需2FA：返回Token并更新用户历史记录
            response = super().post(request, *args, **kwargs)
            user.last_login_location = current_ip
            user.last_login_browser = current_browser
            user.last_login_device = current_device
            user.save(update_fields=['last_login_location', 'last_login_browser', 'last_login_device'])
            return self.finalize_response(request, response, *args, **kwargs)
        else:
            # 需要2FA：生成OTP并发送邮件
            otp = pyotp.TOTP(pyotp.random_base32(), interval=300).now()  # 5分钟有效期的6位OTP
            cache.set(f"2fa_otp_{user.id}", otp, timeout=300)  # 缓存OTP
            
            # 发送验证邮件
            send_mail(
                subject="登录验证码",
                message=f"你的登录验证码为：{otp}，5分钟内有效。",
                from_email=settings.DEFAULT_FROM_EMAIL,
                recipient_list=[user.mail],
                fail_silently=False
            )
            
            # 返回需要2FA的标识
            return Response({
                "requires_2fa": True,
                "user_id": user.id,
                "message": "检测到新设备/浏览器/IP，请输入邮件中的验证码"
            }, status=status.HTTP_202_ACCEPTED)

    def finalize_response(self, request, response, *args, **kwargs):
        if response.data.get('refresh'):
            cookie_max_age = 3600*24*14  # 修正为14天有效期
            response.set_cookie('refresh_token', response.data['refresh'], max_age=cookie_max_age, httponly=True)
            del response.data['refresh']
        return super().finalize_response(request, response, *args, **kwargs)

4. 新增OTP验证接口 #

创建专门的View处理OTP验证，验证通过后返回Token：

from rest_framework.views import APIView
from django.core.cache import cache
from rest_framework_simplejwt.tokens import RefreshToken
from .models import User

class Verify2FAView(APIView):
    def post(self, request):
        user_id = request.data.get('user_id')
        otp = request.data.get('otp')
        
        if not user_id or not otp:
            return Response({"error": "用户ID和验证码不能为空"}, status=status.HTTP_400_BAD_REQUEST)
        
        try:
            user = User.objects.get(id=user_id)
        except User.DoesNotExist:
            return Response({"error": "用户不存在"}, status=status.HTTP_404_NOT_FOUND)
        
        # 验证OTP
        cached_otp = cache.get(f"2fa_otp_{user.id}")
        if not cached_otp or cached_otp != otp:
            return Response({"error": "验证码无效或已过期"}, status=status.HTTP_400_BAD_REQUEST)
        
        # 验证成功：清理缓存、更新用户记录、生成Token
        cache.delete(f"2fa_otp_{user.id}")
        current_ip, current_browser, current_device = get_client_info(request)
        user.last_login_location = current_ip
        user.last_login_browser = current_browser
        user.last_login_device = current_device
        user.save(update_fields=['last_login_location', 'last_login_browser', 'last_login_device'])
        
        refresh = RefreshToken.for_user(user)
        response = Response({
            'access': str(refresh.access_token),
            'mail': user.mail
        })
        response.set_cookie('refresh_token', str(refresh), max_age=3600*24*14, httponly=True)
        
        return response

5. 更新URL配置 #

添加OTP验证接口的路由：

from django.urls import path
from .views import CookieTokenObtainPairView, Verify2FAView
from rest_framework_simplejwt.views import (
    TokenRefreshView, TokenVerifyView, TokenBlacklistView
)

urlpatterns = [
    path('auth/token/', CookieTokenObtainPairView.as_view()),
    path('auth/token/verify-2fa/', Verify2FAView.as_view()),  # 新增2FA验证接口
    path('auth/token/refresh/', TokenRefreshView.as_view()),
    path('auth/token/verify/', TokenVerifyView.as_view()),
    path('auth/token/blacklist/', TokenBlacklistView.as_view()),
]

6. React前端适配 #

修改登录组件，处理两种响应状态（直接登录/需要OTP验证）：

import axios from 'axios';
import { useState } from 'react';

const LoginForm = () => {
  const [email, setEmail] = useState('');
  const [password, setPassword] = useState('');
  const [otp, setOtp] = useState('');
  const [showOtpForm, setShowOtpForm] = useState(false);
  const [userId, setUserId] = useState(null);
  const [message, setMessage] = useState('');

  const handleLogin = async (e) => {
    e.preventDefault();
    try {
      const res = await axios.post('/auth/token/', {
        username: email,  // 与你的用户认证字段对应，若用邮箱作为username则直接传
        password: password
      });
      if (res.data.requires_2fa) {
        setShowOtpForm(true);
        setUserId(res.data.user_id);
        setMessage(res.data.message);
      } else {
        localStorage.setItem('access_token', res.data.access);
        window.location.href = '/dashboard';
      }
    } catch (err) {
      setMessage(err.response?.data?.detail || '登录失败');
    }
  };

  const handleOtpVerify = async (e) => {
    e.preventDefault();
    try {
      const res = await axios.post('/auth/token/verify-2fa/', {
        user_id: userId,
        otp: otp
      });
      localStorage.setItem('access_token', res.data.access);
      window.location.href = '/dashboard';
    } catch (err) {
      setMessage(err.response?.data?.error || '验证码错误');
    }
  };

  return (
    <div className="login-container">
      {!showOtpForm ? (
        <form onSubmit={handleLogin}>
          <input
            type="email"
            value={email}
            onChange={(e) => setEmail(e.target.value)}
            placeholder="邮箱"
            required
          />
          <input
            type="password"
            value={password}
            onChange={(e) => setPassword(e.target.value)}
            placeholder="密码"
            required
          />
          <button type="submit">登录</button>
          {message && <p className="error">{message}</p>}
        </form>
      ) : (
        <form onSubmit={handleOtpVerify}>
          <p>{message}</p>
          <input
            type="text"
            value={otp}
            onChange={(e) => setOtp(e.target.value)}
            placeholder="输入6位验证码"
            maxLength={6}
            required
          />
          <button type="submit">验证</button>
          {message && <p className="error">{message}</p>}
        </form>
      )}
    </div>
  );
};

export default LoginForm;

关键注意事项 #

• 确保Django邮件配置正确：在settings.py中设置EMAIL_BACKEND、DEFAULT_FROM_EMAIL、EMAIL_HOST等参数
• 生产环境建议使用Redis作为缓存后端（替代默认内存缓存），避免OTP丢失
• 首次登录时用户历史字段为空，会自动触发2FA，符合安全逻辑
• 处理代理场景时，需在Nginx等反向代理中配置X-Forwarded-For头，并在Djangosettings.py中启用USE_X_FORWARDED_HOST = True

内容的提问来源于stack exchange，提问作者Anonymous