from .models import * from .utils import generate_jwt, check_jwt, blacklist_token, refresh_jwt from rest_framework.viewsets import ModelViewSet from rest_framework.decorators import action from .serializers import * from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework_simplejwt.authentication import JWTAuthentication from django.contrib.auth import authenticate from rest_framework_simplejwt.tokens import TokenError from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi ## serializer user class from accounts.serializers import UsuarioSerializer class AuthViewSet(ModelViewSet): ### allow any user to access the login and logout views queryset = JWTToken.objects.all() #permission_classes = [IsAuthenticated] serializer_class = JWTTokenSerializer ### bloclk get method @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def list(self, request): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def retrieve(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def create(self, request): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def update(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def partial_update(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def destroy(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405) email_param = openapi.Parameter('email', openapi.IN_PATH, description="User email", type=openapi.TYPE_STRING, required=True) password_param = openapi.Parameter('password', openapi.IN_PATH, description="User password", type=openapi.TYPE_STRING, required=True) manual_parameters = [email_param, password_param] @swagger_auto_schema( request_body=LoginSerializer, operation_description="""Login to get the access token and refresh token. **httpie Example**: http POST $http_domain/auth/register/login/ email='usuario1@teste.com' password='123456' """, operation_id='login', operation_summary='Login to the system', tags=['Authentication'], manual_parameters=manual_parameters, responses={200: openapi.Response('Login successful', LoginSerializer), 400: openapi.Response('Invalid credentials', LoginSerializer), 500: openapi.Response('Token generation failed', LoginSerializer)} ) @action(detail=False, methods=['post'], permission_classes=[AllowAny], url_path='login', url_name='login', name='Login') def login(self, request): serializer_class = LoginSerializer serializer = serializer_class(data=request.data) if serializer.is_valid(): email = serializer.validated_data.get('email') password = serializer.validated_data.get('password') #email = request.data.get('email') #password = request.data.get('password') # Authenticate the user usuario = authenticate(request, email=email, password=password) if usuario is not None: try: # Generate access and refresh tokens access_token, refresh_token = generate_jwt(usuario) return Response({ "message": "Login successful", "access_token": access_token, "refresh_token": refresh_token }, status=200) except TokenError: return Response({"error": "Token generation failed."}, status=500) else: return Response({"error": "Invalid credentials."}, status=400) else: return Response(serializer.errors, status=400) token_param = openapi.Parameter('token', openapi.IN_PATH, description="Token", type=openapi.TYPE_STRING, required=True) manual_parameters = [token_param] @swagger_auto_schema( request_body=LoginSerializer, operation_description="""Blacklist the token to logout the system. **httpie Example**: http $http_domain/auth/register/logout/ token=$token """, operation_id='logout', operation_summary='Logout from the system and blacklist the token', tags=['Authentication'], manual_parameters=manual_parameters, responses={200: openapi.Response('Token blacklisted successfully', LogoutSerializer), 400: openapi.Response('Invalid token', LogoutSerializer)} ) @action(detail=False, methods=['post'], permission_classes=[AllowAny], url_path='logout') def logout(self, request): serializer_class = LogoutSerializer token = request.data.get('token') if blacklist_token(token): return Response({'message': 'Token blacklisted successfully'}, status=200) return Response({'error': 'Invalid token'}, status=400) token_param = openapi.Parameter('token', openapi.IN_PATH, description="Token", type=openapi.TYPE_STRING, required=True) manual_parameters = [token_param] @swagger_auto_schema( request_body=LoginSerializer, operation_description="""Check if the token is valid or not. **httpie Example**: http $http_domain/auth/register/check/ token=$token """, operation_id='check', operation_summary='Check if the token is valid or not', tags=['Authentication'], manual_parameters=manual_parameters, responses={ 200: openapi.Response('Access granted', CheckSerializer), 400: openapi.Response('Invalid token', CheckSerializer) } ) @action(detail=False, methods=['post'], permission_classes=[AllowAny], url_path='check') def check(self, request): serializer_class = CheckSerializer token = request.data.get('token') payload = check_jwt(token) #print(f"DEBUG: payload {payload}") if payload: user_id = UsuarioSerializer(Usuario.objects.get(email=payload['email'])).data['id'] return Response({'message': 'Access granted', 'user': payload['email'], 'user_id': user_id, 'type': payload['token_type']}, status=200) return Response({'error': 'Invalid token'}, status=400) refresh_param = openapi.Parameter('refresh_token', openapi.IN_PATH, description="Refresh Token", type=openapi.TYPE_STRING, required=True) manual_parameters = [refresh_param] @swagger_auto_schema( request_body=LoginSerializer, operation_description="""Check if the token is valid or not. **httpie Example**: http POST $http_domain/auth/register/refresh/ refresh_token=$refresh """, operation_id='refresh', operation_summary='Refresh the access token using the refresh token', tags=['Authentication'], manual_parameters=manual_parameters, responses={ 200: openapi.Response('Token refreshed successfully', RefreshSerializer), 400: openapi.Response('Invalid token', RefreshSerializer) } ) @action(detail=False, methods=['post'], permission_classes=[AllowAny], url_path='refresh') def refresh(self, request): serializer_class = RefreshSerializer refresh_token = request.data.get('refresh_token') access_token = refresh_jwt(refresh_token) if access_token is not None: return Response({'message': 'Token refreshed successfully', 'access_token': access_token}, status=200) return Response({'error': 'Invalid token'}, status=400) # ### create viewset for testing the jwt token # class JustAnyDataViewSet(ModelViewSet): # queryset = JustAnyData.objects.all() # serializer_class = JustAnyDataSerializer # permission_classes = [IsAuthenticated] # authentication_classes = [JWTAuthentication] """ class UserAccess(models.Model): access_type_choices = ( ('login', 'Login'), ('logout', 'Logout'), ('check', 'Check'), ('refresh', 'Refresh'), ) id = models.ObjectIdField(db_column='_id', primary_key=True, editable=False) user = models.ArrayReferenceField(to=Usuario, default=None) user_email = models.CharField(max_length=255, default=None) date = models.DateTimeField(auto_now_add=True) access_type = models.CharField(max_length=255, choices=access_type_choices, default='login') def __str__(self): return f"{self.user_email} - {self.date}" class Meta: verbose_name = 'User Access' verbose_name_plural = 'User Access' app_label = 'auth_app' objects = models.DjongoManager() """ from accounts.views import CustomViewSet # from rest_framework import filters # from django_filters import rest_framework as django_filters # from rest_framework.pagination import PageNumberPagination # from url_filter.integrations.drf import DjangoFilterBackend # class SetPagination(PageNumberPagination): # page_query_param = 'page' # page_size_query_param = 'page_size' # page_size = 20 # max_page_size = 10000 # class AccessFilter(django_filters.FilterSet): # ### DEALING true or false field passing as string # blacklisted_at = django_filters.DateTimeFilter(field_name='blacklisted_at') # created_at = django_filters.DateTimeFilter(field_name='created_at') # expires = django_filters.DateTimeFilter(field_name='expires') # blacklisted = django_filters.BooleanFilter(field_name='blacklisted') # class Meta: # model = JWTToken # fields = { # 'blacklisted_at': ['exact'], # 'created_at': ['icontains'], # 'expires': ['icontains'], # 'user': ['exact'], # 'token_type': ['exact'], # } class UserAccessViewSet(CustomViewSet): # queryset = JWTToken.objects.all() # serializer_class = JWTTokenSerializer # permission_classes = [IsAuthenticated] # authentication_classes = [JWTAuthentication] # pagination_class = SetPagination # ordering = ['-created_at'] # ### use filtering from django rest framework # filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] # filterset_class = AccessFilter # ordering_fields = '__all__' # search_fields = '__all__' model = JWTToken queryset = JWTToken.objects.all() serializer_class = UserAccessSerializer exclude_fields = ['token','user__password','user__is_active','user__is_staff','user__is_superuser','user__last_login','user__nivel__permissions'] filter_fields = { 'blacklisted_at': ['exact', 'range_datetime', 'gte_datetime', 'lte_datetime', 'gt_datetime', 'lt_datetime'], 'blacklisted': ['exact_boolean'], 'created_at': ['exact_datetime', 'range_datetime', 'gte_datetime', 'lte_datetime', 'gt_datetime', 'lt_datetime'], 'expires': ['icontains', 'range_datetime', 'gte_datetime', 'lte_datetime', 'gt_datetime', 'lt_datetime'], 'user_id': ['exact_oid'], 'token_type': ['exact'], } @swagger_auto_schema( operation_description="""List all Niveis. **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/niveis/ """, operation_id='list', operation_summary='List all Niveis', tags=['Niveis'], responses={200: openapi.Response('Data successfully retrieved', UserAccessSerializer), 400: openapi.Response('Error Generating Response', UserAccessSerializer)} ) def list(self, request): return super().list(request) # @swagger_auto_schema( # operation_description="""Retrieve a Nivel. # **httpie Example**: http -A bearer -a $token -f GET $http_domain/accounts/niveis/66743218d8236dbea8742435/ # """, # operation_id='retrieve', # operation_summary='Retrieve a Nivel', # tags=['Niveis'], # responses={200: openapi.Response('Data successfully retrieved', UserAccessSerializer), # 400: openapi.Response('Error generating Response', UserAccessSerializer)} # ) # def retrieve(self, request, pk=None): # return super().retrieve(request, pk) ### Methods Not Allowed ### @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def retrieve(self, request): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def create(self, request): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def update(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def partial_update(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def destroy(self, request, pk=None): return Response({'error': 'Method not allowed'}, status=405)