from rest_framework.authentication import TokenAuthentication, get_authorization_header
from rest_framework.exceptions import AuthenticationFailed
class ExpiringTokenAuthentication(TokenAuthentication):
def authenticate_credentials(self, key):
try:
token = self.model.objects.get(key=key)
except self.model.DoesNotExist:
raise exceptions.AuthenticationFailed('Invalid token')
if not token.user.is_active:
raise exceptions.AuthenticationFailed('User inactive or deleted')
# This is required for the time comparison
utc_now = datetime.utcnow()
utc_now = utc_now.replace(tzinfo=pytz.utc)
if token.created < utc_now - timedelta(hours=24):
raise exceptions.AuthenticationFailed('Token has expired')
return token.user, token
还需要覆盖默认的 rest 框架登录视图,以便在完成登录时刷新令牌:
class ObtainExpiringAuthToken(ObtainAuthToken):
def post(self, request):
serializer = self.serializer_class(data=request.data)
if serializer.is_valid():
token, created = Token.objects.get_or_create(user=serializer.validated_data['user'])
if not created:
# update the created time of the token to keep it valid
token.created = datetime.datetime.utcnow()
token.save()
return Response({'token': token.key})
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
obtain_expiring_auth_token = ObtainExpiringAuthToken.as_view()
from django.utils import timezone
from rest_framework import status
from rest_framework.response import Response
from rest_framework.authtoken.models import Token
from rest_framework.authtoken.views import ObtainAuthToken
class ObtainExpiringAuthToken(ObtainAuthToken):
def post(self, request):
serializer = self.serializer_class(data=request.DATA)
if serializer.is_valid():
token, created = Token.objects.get_or_create(user=serializer.object['user'])
if not created:
# update the created time of the token to keep it valid
token.created = datetime.datetime.utcnow().replace(tzinfo=utc)
token.save()
return Response({'token': token.key})
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
authentication.py
from datetime import timedelta
from django.conf import settings
from django.utils import timezone
from rest_framework.authentication import TokenAuthentication
from rest_framework import exceptions
EXPIRE_HOURS = getattr(settings, 'REST_FRAMEWORK_TOKEN_EXPIRE_HOURS', 24)
class ExpiringTokenAuthentication(TokenAuthentication):
def authenticate_credentials(self, key):
try:
token = self.model.objects.get(key=key)
except self.model.DoesNotExist:
raise exceptions.AuthenticationFailed('Invalid token')
if not token.user.is_active:
raise exceptions.AuthenticationFailed('User inactive or deleted')
if token.created < timezone.now() - timedelta(hours=EXPIRE_HOURS):
raise exceptions.AuthenticationFailed('Token has expired')
return (token.user, token)
from datetime import timedelta
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from django.utils import timezone
from rest_framework.authentication import TokenAuthentication
from rest_framework import exceptions
EXPIRE_HOURS = getattr(settings, 'REST_FRAMEWORK_TOKEN_EXPIRE_HOURS', 24)
class ExpiringTokenAuthentication(TokenAuthentication):
def authenticate_credentials(self, key):
try:
token = self.get_model().objects.get(key=key)
except ObjectDoesNotExist:
raise exceptions.AuthenticationFailed('Invalid token')
if not token.user.is_active:
raise exceptions.AuthenticationFailed('User inactive or deleted')
if token.created < timezone.now() - timedelta(hours=EXPIRE_HOURS):
raise exceptions.AuthenticationFailed('Token has expired')
return token.user, token
视野,视野
import datetime
from pytz import utc
from rest_framework import status
from rest_framework.response import Response
from rest_framework.authtoken.models import Token
from rest_framework.authtoken.views import ObtainAuthToken
from rest_framework.authtoken.serializers import AuthTokenSerializer
class ObtainExpiringAuthToken(ObtainAuthToken):
def post(self, request, **kwargs):
serializer = AuthTokenSerializer(data=request.data)
if serializer.is_valid():
token, created = Token.objects.get_or_create(user=serializer.validated_data['user'])
if not created:
# update the created time of the token to keep it valid
token.created = datetime.datetime.utcnow().replace(tzinfo=utc)
token.save()
return Response({'token': token.key})
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
from rest_framework.authentication import TokenAuthentication
from datetime import timedelta
from datetime import datetime
import datetime as dtime
import pytz
class ExpiringTokenAuthentication(TokenAuthentication):
def authenticate_credentials(self, key):
model = self.get_model()
try:
token = model.objects.get(key=key)
except model.DoesNotExist:
raise exceptions.AuthenticationFailed('Invalid token')
if not token.user.is_active:
raise exceptions.AuthenticationFailed('User inactive or deleted')
# This is required for the time comparison
utc_now = datetime.now(dtime.timezone.utc)
utc_now = utc_now.replace(tzinfo=pytz.utc)
if token.created < utc_now - timedelta(hours=24):
raise exceptions.AuthenticationFailed('Token has expired')
return token.user, token
from rest_framework.authentication import TokenAuthentication
from datetime import timedelta
from datetime import datetime
import datetime as dtime
import pytz
class ExpiringTokenAuthentication(TokenAuthentication):
def authenticate_credentials(self, key):
model = self.get_model()
try:
token = model.objects.get(key=key)
except model.DoesNotExist:
raise exceptions.AuthenticationFailed('Invalid token')
if not token.user.is_active:
raise exceptions.AuthenticationFailed('User inactive or deleted')
# This is required for the time comparison
utc_now = datetime.now(dtime.timezone.utc)
utc_now = utc_now.replace(tzinfo=pytz.utc)
if token.created < utc_now - timedelta(minutes=15): # TOKEN WILL EXPIRE AFTER 15 MINUTES OF INACTIVITY
token.delete() # ADDED THIS LINE SO THAT EXPIRED TOKEN IS DELETED
raise exceptions.AuthenticationFailed('Token has expired')
else:
token.created = utc_now #THIS WILL SET THE token.created TO CURRENT TIME WITH EVERY REQUEST
token.save() #SAVE THE TOKEN
return token.user, token
不管是为移动客户端还是 web 客户端,在应用程序上设置过期机制都是一个很好的做法。有两种常见的解决方案:
系统过期令牌(特定时间后) ,用户必须重新登录才能获得新的有效令牌。
系统自动过期旧令牌(在特定时间之后)并用新令牌(更改令牌)替换旧令牌。
这两种解决方案的共同点是:
Setings.py 中的更改
DEFAULT_AUTHENTICATION_CLASSES = [
# you replace right path of 'ExpiringTokenAuthentication' class
'accounts.token_utils.ExpiringTokenAuthentication'
]
TOKEN_EXPIRED_AFTER_MINUTES = 300
创建 token _ utils. py
from django.conf import settings
from datetime import timedelta
from django.conf import settings
from django.utils import timezone
from rest_framework.authentication import TokenAuthentication
from rest_framework.authtoken.models import Token
from rest_framework.exceptions import AuthenticationFailed
def expires_in(token: Token):
elapsed_time = timezone.now() - token.created
return timedelta(minutes=settings.TOKEN_EXPIRED_AFTER_MINUTES) - elapsed_time
def is_token_expired(token):
return expires_in(token) < timedelta(seconds=0)
def handle_token_expired(token):
Token.objects.filter(key=token).delete()
class ExpiringTokenAuthentication(TokenAuthentication):
def authenticate_credentials(self, key):
try:
token = Token.objects.get(key=key)
except Token.DoesNotExist:
raise AuthenticationFailed("Invalid Token!")
if not token.user.is_active:
raise AuthenticationFailed("User inactive or deleted")
if is_token_expired(token):
handle_token_expired(token)
msg = "The token is expired!, user have to login again."
response = {"msg": msg}
raise AuthenticationFailed(response)
return token.user, token
如果使用选项2,请将这些行添加到 token _ utils. py
def handle_token_expired(token):
is_expired = is_token_expired(token)
if is_expired:
token.delete()
token = Token.objects.create(user = token.user)
return is_expired, token
class ExpiringTokenAuthentication(TokenAuthentication):
"""
when token is expired, it will be removed
and new one will be created
"""
def authenticate_credentials(self, key):
try:
token = Token.objects.get(key = key)
except Token.DoesNotExist:
raise AuthenticationFailed("Invalid Token")
if not token.user.is_active:
raise AuthenticationFailed("User is not active")
is_expired, token = handle_token_expired(token)
if is_expired:
raise AuthenticationFailed("The Token is expired")
return (token.user, token)