99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
"""Sessions GoodDrive: permanent until logout, account removal or password change.
|
|
|
|
Access JWTs remain short-lived. Refresh secrets are opaque random values hashed
|
|
in MySQL, not stateless JWTs that silently expire after seven days.
|
|
"""
|
|
import hashlib
|
|
import secrets
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
from rest_framework import status
|
|
from rest_framework.decorators import api_view, authentication_classes, permission_classes, throttle_classes
|
|
from rest_framework.permissions import AllowAny
|
|
from rest_framework.response import Response
|
|
from rest_framework.throttling import AnonRateThrottle
|
|
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
|
|
from rest_framework_simplejwt.tokens import AccessToken, RefreshToken
|
|
from rest_framework_simplejwt.exceptions import TokenError
|
|
|
|
from .models import PersistentLoginSession
|
|
|
|
|
|
def _digest(value):
|
|
return hashlib.sha256(value.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def _new_session(user):
|
|
secret = secrets.token_urlsafe(48)
|
|
session = PersistentLoginSession.objects.create(
|
|
user=user, token_hash=_digest(secret), auth_hash=user.get_session_auth_hash()
|
|
)
|
|
return session, secret
|
|
|
|
|
|
def _issue_access(session):
|
|
access = AccessToken.for_user(session.user)
|
|
access['sid'] = session.pk
|
|
return str(access)
|
|
|
|
|
|
def _valid_session(secret):
|
|
if not isinstance(secret, str) or len(secret) > 2048 or not secret:
|
|
return None
|
|
try:
|
|
session = PersistentLoginSession.objects.select_related('user').get(token_hash=_digest(secret), revoked_at__isnull=True)
|
|
except PersistentLoginSession.DoesNotExist:
|
|
return None
|
|
user = session.user
|
|
if not user.is_active or not secrets.compare_digest(session.auth_hash, user.get_session_auth_hash()):
|
|
return None
|
|
return session
|
|
|
|
|
|
@api_view(['POST'])
|
|
@authentication_classes([])
|
|
@permission_classes([AllowAny])
|
|
@throttle_classes([AnonRateThrottle])
|
|
def login(request):
|
|
serializer = TokenObtainPairSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
session, secret = _new_session(serializer.user)
|
|
return Response({'access': _issue_access(session), 'refresh': secret})
|
|
|
|
|
|
@api_view(['POST'])
|
|
@authentication_classes([])
|
|
@permission_classes([AllowAny])
|
|
@throttle_classes([AnonRateThrottle])
|
|
def refresh(request):
|
|
secret = request.data.get('refresh')
|
|
session = _valid_session(secret)
|
|
if session:
|
|
PersistentLoginSession.objects.filter(pk=session.pk).update(last_used_at=timezone.now())
|
|
return Response({'access': _issue_access(session)})
|
|
|
|
# One-time compatibility for browsers that still hold an old SimpleJWT
|
|
# refresh token in sessionStorage after deploying this update.
|
|
if isinstance(secret, str) and secret.count('.') == 2:
|
|
try:
|
|
legacy = RefreshToken(secret)
|
|
user_id = legacy['user_id']
|
|
user = get_user_model().objects.filter(pk=user_id, is_active=True).first()
|
|
if user:
|
|
session, new_secret = _new_session(user)
|
|
return Response({'access': _issue_access(session), 'refresh': new_secret})
|
|
except (TokenError, KeyError, ValueError, TypeError):
|
|
pass
|
|
return Response({'detail': 'Session expirée ou révoquée.'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
|
|
@api_view(['POST'])
|
|
@authentication_classes([])
|
|
@permission_classes([AllowAny])
|
|
def logout(request):
|
|
secret = request.data.get('refresh')
|
|
if isinstance(secret, str) and 0 < len(secret) <= 2048:
|
|
PersistentLoginSession.objects.filter(token_hash=_digest(secret)).update(revoked_at=timezone.now())
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|