49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
from rest_framework_simplejwt.tokens import RefreshToken, AccessToken, Token, TokenError
|
|
from datetime import datetime, timedelta
|
|
from django.contrib.auth import get_user_model
|
|
from rest_framework import exceptions
|
|
|
|
from apps.user_auth.models import JwtToken as JwtTokenModel
|
|
|
|
|
|
|
|
CUserModel = get_user_model()
|
|
|
|
|
|
def get_tokens_for_user(user):
|
|
token = RefreshToken.for_user(user)
|
|
return {
|
|
'refresh': str(token),
|
|
'access': str(token.access_token),
|
|
}
|
|
|
|
|
|
def get_refresh_token(ref_token: str) -> dict:
|
|
|
|
try:
|
|
user_token = RefreshToken(token=ref_token)
|
|
except TokenError as exp:
|
|
JwtTokenModel.objects.filter(refresh=ref_token).delete()
|
|
raise exceptions.ValidationError({'error': str(exp)})
|
|
|
|
if not user_token:
|
|
raise exceptions.ValidationError({'error': 'Invalid Token in Verify'})
|
|
|
|
try:
|
|
jwt_token = JwtTokenModel.objects.get(refresh=ref_token)
|
|
except JwtTokenModel.DoesNotExist:
|
|
raise exceptions.ValidationError({'error': 'Invalid Token'})
|
|
|
|
if jwt_token.is_blocked:
|
|
raise exceptions.ValidationError({'error': 'Token is blocked'})
|
|
|
|
token = get_tokens_for_user(jwt_token.cuser)
|
|
JwtTokenModel.objects.create(
|
|
refresh=token['refresh'],
|
|
access=token['access'],
|
|
cuser=jwt_token.cuser,
|
|
)
|
|
# remove old token
|
|
jwt_token.delete()
|
|
return token
|