如何在Django项目中实现基于Token认证的模拟授权服务?
实现模拟授权服务的Token签发与验证方案
以下是针对你的Django项目,实现模拟授权服务的具体步骤,完全贴合你现有项目结构进行修改:
1. 创建自定义模拟Token模型
为了模拟第三方授权服务的Token生命周期管理,我们创建独立的Token模型来存储Token信息、关联用户并记录有效期:
在security/models.py中添加:
from django.db import models from django.contrib.auth.models import User import uuid from datetime import timedelta from django.utils import timezone class MockAuthToken(models.Model): token = models.UUIDField(default=uuid.uuid4, editable=False, unique=True) user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='mock_tokens') created_at = models.DateTimeField(auto_now_add=True) expires_at = models.DateTimeField() def save(self, *args, **kwargs): # 默认设置Token有效期为7天 if not self.expires_at: self.expires_at = timezone.now() + timedelta(days=7) super().save(*args, **kwargs) def is_valid(self): return timezone.now() < self.expires_at def __str__(self): return f"{self.user.username} - {str(self.token)[:8]}"
执行迁移命令生效:
python manage.py makemigrations && python manage.py migrate
2. 实现模拟授权的Token签发端点
创建一个公开的视图,模拟第三方授权服务的登录Token获取接口,用户传入账号密码验证后返回有效Token:
在security/views.py中添加:
from rest_framework.views import APIView from django.contrib.auth import authenticate from .models import MockAuthToken from rest_framework.exceptions import AuthenticationFailed from rest_framework.response import Response from rest_framework import status from django.utils import timezone from datetime import timedelta class MockAuthTokenObtainView(APIView): permission_classes = [] # 允许未认证访问,模拟授权服务的公开端点 def post(self, request): username = request.data.get('username') password = request.data.get('password') if not username or not password: return Response({"error": "请提供用户名和密码"}, status=status.HTTP_400_BAD_REQUEST) user = authenticate(username=username, password=password) if not user: raise AuthenticationFailed("用户名或密码错误") # 生成或复用用户的有效Token:若已有Token未过期则返回,否则生成新Token token_obj, created = MockAuthToken.objects.get_or_create( user=user, defaults={'expires_at': timezone.now() + timedelta(days=7)} ) if not created and not token_obj.is_valid(): token_obj.delete() token_obj = MockAuthToken.objects.create(user=user) return Response({ "token": str(token_obj.token), "expires_at": token_obj.expires_at.isoformat(), "user_id": user.id, "username": user.username })
3. 配置模拟授权端点的URL
在项目根urls.py中添加Token签发的URL:
from security.views import MockAuthTokenObtainView urlpatterns = [ # ... 原有URL配置 path('mock-auth/token/', MockAuthTokenObtainView.as_view(), name='mock-auth-token'), ]
4. 实现自定义Token认证类
创建自定义认证逻辑,验证请求中的Token是否为模拟授权服务签发的有效Token:
新建security/authentication.py文件,添加:
from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from .models import MockAuthToken from django.utils.translation import gettext_lazy as _ class MockAuthTokenAuthentication(BaseAuthentication): keyword = 'Bearer' # 遵循标准Bearer Token格式 def authenticate(self, request): auth_header = request.META.get('HTTP_AUTHORIZATION') if not auth_header: return None try: keyword, token = auth_header.split() if keyword.lower() != self.keyword.lower(): raise AuthenticationFailed(_("不支持的认证方案")) except ValueError: raise AuthenticationFailed(_("无效的认证头格式")) try: token_obj = MockAuthToken.objects.get(token=token) except MockAuthToken.DoesNotExist: raise AuthenticationFailed(_("无效的Token")) if not token_obj.is_valid(): raise AuthenticationFailed(_("Token已过期")) return (token_obj.user, token_obj) def authenticate_header(self, request): return self.keyword
5. 配置DRF使用自定义认证类
修改settings.py中的DRF配置,替换默认的Token认证为我们的模拟授权认证:
REST_FRAMEWORK = { 'DEFAULT_AUTHENTICATION_CLASSES': [ 'security.authentication.MockAuthTokenAuthentication', # 自定义模拟认证类 # 'rest_framework.authentication.SessionAuthentication', # 可选保留会话认证 ], 'DEFAULT_PERMISSION_CLASSES': [ 'rest_framework.permissions.IsAuthenticatedOrReadOnly', ], }
6. 测试完整流程
步骤1:注册用户
curl -X POST http://localhost:8000/register/ \ -H "Content-Type: application/json" \ -d '{"username": "testuser", "password": "testpass123", "email": "test@example.com"}'
步骤2:获取模拟授权Token
curl -X POST http://localhost:8000/mock-auth/token/ \ -H "Content-Type: application/json" \ -d '{"username": "testuser", "password": "testpass123"}'
返回示例:
{ "token": "a1b2c3d4-5678-90ef-ghij-klmnopqrstuv", "expires_at": "2024-12-31T23:59:59.999999Z", "user_id": 1, "username": "testuser" }
步骤3:使用Token访问受保护API
curl -X GET http://localhost:8000/api/security-records/ \ -H "Authorization: Bearer a1b2c3d4-5678-90ef-ghij-klmnopqrstuv"
可选优化
- 添加客户端ID/密钥验证,更贴近真实第三方授权服务场景
- 实现Token刷新端点,允许用户用即将过期的Token获取新Token
- 增加Token黑名单功能,支持用户主动注销Token
内容的提问来源于stack exchange,提问作者Nimo Db
相关产品推荐
相关产品推荐

