224 lines
8.1 KiB
Python
224 lines
8.1 KiB
Python
"""Common views for application Account"""
|
|
import json
|
|
|
|
from rest_framework import status
|
|
from braces.views import CsrfExemptMixin
|
|
from django.conf import settings
|
|
from django.utils.translation import gettext_lazy as _
|
|
from oauth2_provider.oauth2_backends import OAuthLibCore
|
|
from oauth2_provider.settings import oauth2_settings
|
|
from oauth2_provider.views.mixins import OAuthLibMixin
|
|
from rest_framework import permissions
|
|
from rest_framework.generics import GenericAPIView
|
|
from rest_framework.views import APIView
|
|
from rest_framework.response import Response
|
|
from rest_framework_social_oauth2.oauth2_backends import KeepRequestCore
|
|
from rest_framework_social_oauth2.oauth2_endpoints import SocialTokenServer
|
|
from account import models as account_models
|
|
from authorization.models import Application
|
|
from authorization.serializers import common as serializers
|
|
from utils import exceptions as utils_exceptions
|
|
|
|
|
|
# Mixins
|
|
class BaseViewMixin(GenericAPIView):
|
|
"""BaseMixin for classic auth views"""
|
|
def get_client_id(self, source) -> str:
|
|
"""Get application client id"""
|
|
qs = Application.objects.by_source(source=source)
|
|
if qs.exists():
|
|
return qs.first().client_id
|
|
else:
|
|
raise utils_exceptions.SerivceError(data={
|
|
'detail': _('Application is not found')})
|
|
|
|
def get_client_secret(self, source) -> str:
|
|
"""Get application client id"""
|
|
if source == Application.MOBILE:
|
|
qs = Application.objects.by_source(source=source)
|
|
if qs.exists:
|
|
return qs.first().client_secret
|
|
else:
|
|
raise utils_exceptions.SerivceError(data={
|
|
'detail': _('Not found an application with this source')})
|
|
|
|
|
|
class OAuth2ViewMixin(CsrfExemptMixin, OAuthLibMixin, BaseViewMixin):
|
|
"""Basic mixin for OAuth2 views"""
|
|
server_class = oauth2_settings.OAUTH2_SERVER_CLASS
|
|
validator_class = oauth2_settings.OAUTH2_VALIDATOR_CLASS
|
|
oauthlib_backend_class = OAuthLibCore
|
|
|
|
def prepare_request_data(self, validated_data: dict) -> dict:
|
|
"""Preparing request data"""
|
|
source = validated_data.get('source')
|
|
# Set OAuth2 request parameters
|
|
_request_data = {
|
|
'client_id': self.get_client_id(source)
|
|
}
|
|
# Fill client secret parameter by platform
|
|
if validated_data.get('source') == Application.MOBILE:
|
|
_request_data['client_secret'] = self.get_client_secret(source)
|
|
# Fill token parameter if transfer
|
|
if validated_data.get('token'):
|
|
_request_data['token'] = validated_data.get('token')
|
|
if _request_data:
|
|
return _request_data
|
|
else:
|
|
raise utils_exceptions.ServiceError()
|
|
|
|
|
|
# Login
|
|
class LoginByUsernameView(OAuth2ViewMixin, GenericAPIView):
|
|
"""
|
|
Implements an endpoint to provide access tokens
|
|
|
|
The endpoint is used in the following flows:
|
|
|
|
* Authorization code
|
|
* Password
|
|
* Client credentials
|
|
"""
|
|
permission_classes = (permissions.AllowAny,)
|
|
serializer_class = serializers.LoginSerializer
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
# Preparing request data
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
request_data = self.prepare_request_data(serializer.validated_data)
|
|
request_data.update({
|
|
'grant_type': 'password',
|
|
'username': serializer.validated_data.get('username'),
|
|
'password': serializer.validated_data.get('password'),
|
|
})
|
|
# Use the rest framework `.data` to fake the post body of the django request.
|
|
request._request.POST = request._request.POST.copy()
|
|
for key, value in request_data.items():
|
|
request._request.POST[key] = value
|
|
|
|
url, headers, body, status = self.create_token_response(request._request)
|
|
response = Response(data=json.loads(body), status=status)
|
|
|
|
for k, v in headers.items():
|
|
response[k] = v
|
|
return response
|
|
|
|
|
|
class LoginByEmailView(LoginByUsernameView):
|
|
"""
|
|
Implements an endpoint to provide access tokens
|
|
|
|
The endpoint is used in the following flows:
|
|
|
|
* Authorization code
|
|
* Password
|
|
* Client credentials
|
|
"""
|
|
serializer_class = serializers.LoginByEmailSerializer
|
|
|
|
|
|
# Logout
|
|
class LogoutView(GenericAPIView):
|
|
"""Logout view"""
|
|
permission_classes = (permissions.IsAuthenticated, )
|
|
serializer_class = serializers.LogoutSerializer
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Method to logout user by deleting access tokens by source"""
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
# Setup attributes
|
|
source = serializer.validated_data.get('source')
|
|
request.user.revoke_refresh_tokens(source=source)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
class RevokeTokenView(OAuth2ViewMixin, GenericAPIView):
|
|
"""
|
|
Implements an endpoint to revoke access or refresh tokens
|
|
"""
|
|
permission_classes = (permissions.AllowAny,)
|
|
serializer_class = serializers.OAuth2Serialzier
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
# Use the rest framework `.data` to fake the post body of the django request.
|
|
# Preparing request data
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
request_data = self.prepare_request_data(serializer.validated_data)
|
|
# Use the rest framework `.data` to fake the post body of the django request.
|
|
request._request.POST = request._request.POST.copy()
|
|
for key, value in request_data.items():
|
|
request._request.POST[key] = value
|
|
|
|
url, headers, body, status = self.create_revocation_response(request._request)
|
|
response = Response(data=json.loads(body) if body else '', status=status if body else 204)
|
|
|
|
for k, v in headers.items():
|
|
response[k] = v
|
|
return response
|
|
|
|
|
|
class SocialSignUpView(OAuth2ViewMixin, GenericAPIView):
|
|
"""
|
|
Implements an endpoint to convert a provider token to an access token
|
|
|
|
The endpoint is used in the following flows:
|
|
|
|
* Authorization code
|
|
* Client credentials
|
|
"""
|
|
server_class = SocialTokenServer
|
|
oauthlib_backend_class = KeepRequestCore
|
|
permission_classes = (permissions.AllowAny,)
|
|
serializer_class = serializers.OAuth2Serialzier
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Override POST method"""
|
|
# Preparing request data
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
request_data = self.prepare_request_data(serializer.validated_data)
|
|
request_data.update({
|
|
'grant_type': settings.OAUTH2_SOCIAL_AUTH_GRANT_TYPE,
|
|
'backend': settings.OAUTH2_SOCIAL_AUTH_BACKEND_NAME
|
|
})
|
|
# Use the rest framework `.data` to fake the post body of the django request.
|
|
request._request.POST = request._request.POST.copy()
|
|
for key, value in request_data.items():
|
|
request._request.POST[key] = value
|
|
|
|
url, headers, body, status = self.create_token_response(request._request)
|
|
response = Response(data=json.loads(body), status=status)
|
|
|
|
for k, v in headers.items():
|
|
response[k] = v
|
|
return response
|
|
|
|
|
|
class TokenView(CsrfExemptMixin, OAuthLibMixin, GenericAPIView):
|
|
"""
|
|
Implements an endpoint to provide access tokens
|
|
|
|
The endpoint is used in the following flows:
|
|
|
|
* Authorization code
|
|
* Password
|
|
* Client credentials
|
|
"""
|
|
permission_classes = (permissions.AllowAny,)
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
# Use the rest framework `.data` to fake the post body of the django request.
|
|
request._request.POST = request._request.POST.copy()
|
|
for key, value in request.data.items():
|
|
request._request.POST[key] = value
|
|
|
|
url, headers, body, status = self.create_token_response(request._request)
|
|
response = Response(data=json.loads(body), status=status)
|
|
|
|
for k, v in headers.items():
|
|
response[k] = v
|
|
return response
|