基于Auth Token/JWT为DRF API实现MFA的架构咨询

你的这个基于令牌的MFA架构思路非常靠谱，完全适配DRF的场景！我来帮你把这个方案细化得更落地，贴合Django和DRF的生态，同时兼顾安全性和可维护性：

1. 数据模型设计 #

首先需要两个核心模型来支撑你的流程：一个存储待执行的请求，一个关联MFA验证令牌和验证码。这里要注意敏感数据的加密存储：

from django.db import models
from django.contrib.auth import get_user_model
from django.utils import timezone
from django.contrib.auth.hashers import make_password, check_password

User = get_user_model()

class PendingRequest(models.Model):
    STATUS_CHOICES = (
        ('PENDING', '待验证'),
        ('EXECUTED', '已执行'),
        ('EXPIRED', '已过期'),
    )
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    method = models.CharField(max_length=10)  # GET/POST/PUT等
    path = models.CharField(max_length=255)   # 比如/api/orders/1/
    query_params = models.JSONField(null=True, blank=True)  # URL参数
    body = models.JSONField(null=True, blank=True)          # 请求体
    created_at = models.DateTimeField(auto_now_add=True)
    expires_at = models.DateTimeField()  # 建议设15分钟过期
    status = models.CharField(max_length=10, choices=STATUS_CHOICES, default='PENDING')

    def is_expired(self):
        return timezone.now() > self.expires_at

class MFAToken(models.Model):
    STATUS_CHOICES = (
        ('UNVERIFIED', '未验证'),
        ('VERIFIED', '已验证'),
        ('EXPIRED', '已过期'),
    )
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    pending_request = models.OneToOneField(PendingRequest, on_delete=models.CASCADE)
    token = models.CharField(max_length=64, unique=True)  # 用于关联请求的唯一令牌
    mfa_code = models.CharField(max_length=20)  # 哈希存储的验证码，绝对不能明文！
    sent_via = models.CharField(max_length=20, default='SMS')
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='UNVERIFIED')
    expires_at = models.DateTimeField()  # 验证码建议设10分钟过期

    def is_expired(self):
        return timezone.now() > self.expires_at

    def verify_code(self, code):
        return check_password(code, self.mfa_code)

2. DRF权限拦截逻辑 #

用DRF自定义权限类来拦截受MFA保护的请求，判断用户是否已完成验证，未验证则生成待处理请求和MFA令牌，触发短信发送：

from rest_framework.permissions import BasePermission
from rest_framework.exceptions import PermissionDenied
from django.utils.crypto import get_random_string
from django.utils import timezone
from .models import PendingRequest, MFAToken
from .utils import send_sms  # 你自己实现的短信发送工具

class MFARequiredPermission(BasePermission):
    def has_permission(self, request, view):
        # 先确保用户已通过JWT/Token基础认证
        if not request.user.is_authenticated:
            return False

        # 检查请求是否携带MFA验证凭据
        mfa_token = request.headers.get('X-MFA-Token')
        mfa_code = request.headers.get('X-MFA-Code')

        if mfa_token and mfa_code:
            # 验证MFA令牌和验证码
            try:
                mfa_record = MFAToken.objects.get(token=mfa_token, user=request.user, status='UNVERIFIED')
                if mfa_record.is_expired():
                    mfa_record.status = 'EXPIRED'
                    mfa_record.save()
                    raise PermissionDenied("MFA令牌已过期，请重新发起请求")
                
                if mfa_record.verify_code(mfa_code):
                    mfa_record.status = 'VERIFIED'
                    mfa_record.save()
                    # 标记请求已通过MFA，后续流程可直接执行
                    request.mfa_passed = True
                    request.pending_request = mfa_record.pending_request
                    return True
                else:
                    raise PermissionDenied("MFA验证码错误")
            except MFAToken.DoesNotExist:
                raise PermissionDenied("无效的MFA令牌")
        else:
            # 未携带MFA凭据，生成待处理请求和验证令牌
            pending_req = PendingRequest.objects.create(
                user=request.user,
                method=request.method,
                path=request.path,
                query_params=dict(request.query_params),
                body=request.data if request.method in ['POST', 'PUT', 'PATCH'] else None,
                expires_at=timezone.now() + timezone.timedelta(minutes=15)
            )

            # 生成唯一令牌和6位数字验证码
            mfa_token_str = get_random_string(64)
            mfa_code_str = get_random_string(6, allowed_chars='0123456789')
            
            MFAToken.objects.create(
                user=request.user,
                pending_request=pending_req,
                token=mfa_token_str,
                mfa_code=make_password(mfa_code_str),
                expires_at=timezone.now() + timezone.timedelta(minutes=10)
            )

            # 发送短信验证码
            send_sms(request.user.phone_number, f"你的MFA验证码是：{mfa_code_str}，10分钟内有效")

            # 返回需要验证的提示和令牌
            raise PermissionDenied({
                "detail": "请完成多因素认证",
                "mfa_token": mfa_token_str
            })

然后在需要保护的视图/视图集里添加这个权限类：

from rest_framework.viewsets import ModelViewSet
from .permissions import MFARequiredPermission
from .serializers import OrderSerializer
from .models import Order

class OrderViewSet(ModelViewSet):
    queryset = Order.objects.all()
    serializer_class = OrderSerializer
    permission_classes = [MFARequiredPermission]

3. MFA验证与请求执行端点 #

提供一个专门的验证端点，用户提交令牌和验证码后，执行之前保存的请求并返回结果：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.exceptions import PermissionDenied
from django.test.client import RequestFactory
from django.urls import resolve
from .models import MFAToken, PendingRequest

class MFAVerifyView(APIView):
    def post(self, request):
        mfa_token = request.data.get('mfa_token')
        mfa_code = request.data.get('mfa_code')
        
        if not mfa_token or not mfa_code:
            raise PermissionDenied("请提供MFA令牌和验证码")
        
        try:
            mfa_record = MFAToken.objects.get(token=mfa_token, user=request.user, status='UNVERIFIED')
            if mfa_record.is_expired():
                mfa_record.status = 'EXPIRED'
                mfa_record.save()
                raise PermissionDenied("MFA令牌已过期，请重新发起请求")
            
            if not mfa_record.verify_code(mfa_code):
                raise PermissionDenied("验证码错误")
            
            # 标记验证通过
            mfa_record.status = 'VERIFIED'
            mfa_record.save()
            pending_req = mfa_record.pending_request

            if pending_req.is_expired():
                pending_req.status = 'EXPIRED'
                pending_req.save()
                raise PermissionDenied("原请求已过期，请重新发起")
            
            # 模拟原请求并执行
            factory = RequestFactory()
            simulated_request = factory.generic(
                method=pending_req.method,
                path=pending_req.path,
                data=pending_req.body,
                content_type='application/json'
            )
            simulated_request.user = request.user

            # 解析并执行原视图
            view_func, args, kwargs = resolve(pending_req.path)
            response = view_func(simulated_request, *args, **kwargs)

            # 更新请求状态
            pending_req.status = 'EXECUTED'
            pending_req.save()

            # 返回原请求的响应结果
            return Response(response.data if hasattr(response, 'data') else response.content.decode())
        
        except MFAToken.DoesNotExist:
            raise PermissionDenied("无效的MFA令牌")

4. 安全性与优化建议 #

• 定时清理过期数据：用Django Celery Beat定时删除EXPIRED状态的PendingRequest和MFAToken，避免数据库冗余。
• 短信发送重试：添加短信发送失败的重试机制，或者给用户提供“重新发送验证码”的接口。
• 幂等性处理：对于POST/PUT等非幂等请求，确保PendingRequest执行后不会被重复触发（通过status字段判断）。
• 日志记录：记录所有MFA相关操作（请求创建、验证码发送、验证结果、请求执行），方便排查问题。
• 可选轻量化方案：如果不需要每次请求都验证MFA，可以改为“会话式”：用户验证MFA后，生成一个带mfa_verified=true标记的JWT，后续用这个JWT访问受保护API，这种方式不需要存储请求，更轻量，但安全性略低于每次验证的模式。

内容的提问来源于stack exchange，提问作者Mahammad Adil Azeem