### Django Rest Framework Libs ### from rest_framework import viewsets, permissions, filters, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from rest_framework_simplejwt.authentication import JWTAuthentication from rest_framework_simplejwt.tokens import RefreshToken, AccessToken from django.contrib.auth import authenticate from rest_framework.parsers import MultiPartParser, FormParser, JSONParser ### Django Libs ### from django.conf import settings from django_filters import rest_framework as django_filters from django.contrib.auth.hashers import check_password import time from djongo.models import ObjectIdField # Add this import import os from django.core.files.base import ContentFile from django.core.files.storage import default_storage from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi ### App Libs ### from .models import * from .serializers import * from .permissions import * from accounts.connection import get_mongo_connection from .permissions import HasGroupOrPermission # Import custom permission from django.contrib.auth.models import Permission # class PermissionFilter(django_filters.FilterSet): # class Meta: # model = Permission # fields = { # 'id': ['in', 'exact'], # 'name': ['icontains', 'exact'], # 'codename': ['icontains', 'exact'], # 'content_type__app_label': ['icontains', 'exact'], # 'content_type__model': ['icontains', 'exact'], # } # class PermissionViewSet(viewsets.ReadOnlyModelViewSet): # permission_classes = [IsAuthenticated] # # authentication_classes = [JWTAuthentication] # queryset = Permission.objects.filter(content_type__app_label__in=['auth_app', 'accounts', 'panel']) # serializer_class = PermissionSerializer # filter_backends = [django_filters.DjangoFilterBackend, filters.OrderingFilter, filters.SearchFilter] # filterset_class = PermissionFilter # search_fields = ['name', 'codename', 'content_type__app_label', 'content_type__model'] # ordering_fields = ['id', 'name', 'codename'] # def list(self, request): # try: # permissions = self.filter_queryset(self.get_queryset()) # serializer = self.get_serializer(permissions, many=True) # if permissions: # return Response(serializer.data, status=status.HTTP_200_OK) # else: # return Response({'message': 'Not allowed'}, status=status.HTTP_401_UNAUTHORIZED) # except Exception as e: # response = {'message': 'Error listing items', 'Error': str(e)} # return Response(response, status=status.HTTP_400_BAD_REQUEST) # @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) # def create(self, request): # return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) # def retrieve(self, request, pk=None): # return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) # def update(self, request, pk=None): # return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) # def partial_update(self, request, pk=None): # return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) # @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) # def destroy(self, request, pk=None): # return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) ########################################## from django.contrib.auth import get_user_model from rest_framework import viewsets, permissions, status from rest_framework.response import Response from rest_framework.decorators import action from .serializers import UsuarioSerializer # Replace with your actual serializer User = get_user_model() class UserFilter(django_filters.FilterSet): class Meta: model = User fields = { 'id': ['in', 'exact'], 'email': ['icontains', 'exact'], 'first_name': ['icontains', 'exact'], 'last_name': ['icontains', 'exact'], 'telefone': ['icontains', 'exact'], 'bio': ['icontains', 'exact'], 'data_nascimento': ['exact'], 'is_active': ['exact'], } filter_overrides = { ObjectIdField: { 'filter_class': django_filters.CharFilter } } class UserViewSet(viewsets.ModelViewSet): queryset = User.objects.all() serializer_class = UsuarioSerializer permission_classes = [permissions.IsAuthenticated, HasGroupOrPermission] # Adjust permissions as needed authentication_classes = [JWTAuthentication] filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = UserFilter search_fields = '__all__' ordering_fields = '__all__' parser_classes = [MultiPartParser, FormParser] required_groups = [] # Required groups for this viewset # πŸ”₯ Define required permissions for each action def get_required_permissions(self): """ Dynamically generates the required permission based on the action and model name. Example: If action is "list" and model is "usuario" (from app "accounts"), it returns ('accounts', 'view_usuario'). """ action_permissions = { "list": "view", "retrieve": "view", "create": "add", "update": "change", "partial_update": "change", "destroy": "delete", } model = self.queryset.model model_name = model._meta.object_name.lower() app_label = model._meta.app_label # βœ… Get app_label dynamically action = action_permissions.get(self.action, "") codename = f"{action}_{model_name}" if action else "" # print(f"πŸ”Ή App Label: {app_label}") # print(f"πŸ”Ή Action: {self.action}") # print(f"πŸ”Ή Codename: {codename}") return app_label, codename # Returning ('app_label', codename) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""List all registers.""", operation_summary='List registers', operation_id='usuarios_list', tags=['usuarios'], responses={200: openapi.Response('Data successfully listed', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, ) def list(self, request): return super().list(request) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Retrieve register.""", operation_summary='Retrieve register', operation_id='usuarios_retrieve', tags=['usuarios'], responses={200: openapi.Response('Data successfully retrieved', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, ) def retrieve(self, request, pk=None): try: instance = self.queryset.get(id=ObjectId(pk)) serializer = self.get_serializer(instance) return Response(serializer.data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error retrieving user: {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Create register.""", operation_summary='Create register', operation_id='usuarios_create', tags=['usuarios'], responses={200: openapi.Response('Register successfully created', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, request_body=UserCreateSerializer, ) def create(self, request): return super().create(request) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') id_param = openapi.Parameter('id', openapi.IN_PATH, description="User ID", type=openapi.TYPE_STRING, required=True) manual_parameters=[jwt_token_param, id_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Update register.""", operation_summary='Update register', operation_id='usuarios_update', tags=['usuarios'], responses={200: openapi.Response('Data successfully updated', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, request_body=UserUpdateSerializer, ) def update(self, request, pk=None): try: instance = self.queryset.get(id=ObjectId(pk)) serializer = self.get_serializer(instance, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error updating user: {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') id_param = openapi.Parameter('id', openapi.IN_PATH, description="User ID", type=openapi.TYPE_STRING, required=True) manual_parameters=[jwt_token_param, id_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Partial Update register.""", operation_summary='Partial Update register', operation_id='usuarios_partial_update', tags=['usuarios'], responses={200: openapi.Response('Data successfully partial updated', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, request_body=UserUpdateSerializer, ) def partial_update(self, request, pk=None): try: instance = self.queryset.get(id=ObjectId(pk)) serializer = self.get_serializer(instance, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error partial updating user: {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters = [jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Delete a user.""", operation_summary='Delete a user', operation_id='usuarios_delete', tags=['usuarios'], responses={204: openapi.Response('No Content', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, ) def destroy(self, request, pk=None): try: instance = self.queryset.get(id=ObjectId(pk)) instance.delete() return Response(status=status.HTTP_204_NO_CONTENT) except Exception as e: return Response({'message': f'Error deleting user: {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters = [jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Retrieve my data.""", operation_summary='Retrive my data', operation_id='usuarios-self_retrieve', tags=['usuarios-self'], responses={200: openapi.Response('Data successfully retrieved', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, ) @action(detail=False, methods=['get'], url_path='me', name='Retrieve My Profile', permission_classes=[permissions.IsAuthenticated], authentication_classes=[JWTAuthentication]) def retrieve_me(self, request): """Retrieve the authenticated user's details.""" print(request.user) serializer = self.get_serializer(request.user) return Response(serializer.data, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Update my register.""", operation_summary='Update my register', operation_id='usuarios-self_update', tags=['usuarios-self'], responses={200: openapi.Response('Data successfully updated', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, request_body=UserUpdateSerializer, ) @action(detail=False, methods=['put'], url_path='me/update', name='Update My Profile', permission_classes=[permissions.IsAuthenticated], authentication_classes=[JWTAuthentication]) def update_me(self, request): """Update the authenticated user's details (full update).""" serializer = self.get_serializer(request.user, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) # @action(detail=False, methods=['patch'], url_path='me/partial-update', name='Partial Update My Profile', permission_classes=[permissions.IsAuthenticated], authentication_classes=[JWTAuthentication]) # def partial_update_me(self, request): # """Partially update the authenticated user's details.""" # serializer = self.get_serializer(request.user, data=request.data, partial=True) # serializer.is_valid(raise_exception=True) # serializer.save() # return Response(serializer.data, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Inactivate my specific register. Arguments: request -- The HTTP request object. Returns: A Response object 200. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register inactivated', 400: 'Error while inactivating register', }, request_body=UserActivateInactivateSerializer,\ operation_summary='Inactivate my specific register', operation_id='usuarios-self_inactivate', tags=['usuarios-self']) @action(detail=False, methods=['patch'], url_path='me/inactivate', name='Inactive My Profile', permission_classes=[IsAuthenticated], authentication_classes=[JWTAuthentication], parser_classes=[JSONParser]) def inactive_me(self, request): """Deactivate the authenticated user's account.""" request.user.is_active = False request.user.save() return Response({"message": "Account deactivated"}, status=status.HTTP_200_OK) # jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') # manual_parameters=[jwt_token_param] # @swagger_auto_schema(operation_description=""" # Activate my specific register. # Arguments: # request -- The HTTP request object. # Returns: # A Response 200. # """, manual_parameters=manual_parameters,\ # responses={ # 200: 'Register activated', # 400: 'Error while activating register', }, # request_body=UserActivateInactivateSerializer,\ # operation_summary='Activate my specific register', # operation_id='usuarios-self_activate', tags=['usuarios-self']) # @action(detail=False, methods=['patch'], url_path='me/activate', name='Activate My Profile', permission_classes=[IsAuthenticated], # authentication_classes=[JWTAuthentication], parser_classes=[JSONParser]) # def activate_me(self, request): # """Deactivate the authenticated user's account.""" # request.user.is_active = True # request.user.save() # return Response({"message": "Account activated"}, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters = [jwt_token_param] @swagger_auto_schema( manual_parameters= manual_parameters, operation_description="""Delete my user.""", operation_summary='Delete my user', operation_id='usuarios-self_delete', tags=['usuarios-self'], responses={204: openapi.Response('No Content', UsuarioSerializer), 400: openapi.Response('Error Generating Response', UsuarioSerializer)}, ) @action(detail=False, methods=['delete'], url_path='me/delete', name='Delete My Profile', permission_classes=[permissions.IsAuthenticated], authentication_classes=[JWTAuthentication]) def destroy_me(self, request): """Delete the authenticated user's account.""" request.user.delete() return Response(status=status.HTTP_204_NO_CONTENT) from .models import Cliente from .serializers import ClienteSerializer, ClienteAuthSerializer from .authentication import ClienteTokenAuthentication class ClienteFilter(django_filters.FilterSet): class Meta: model = Cliente fields = { 'id': ['in', 'exact'], 'nome': ['icontains', 'exact'], 'email': ['icontains', 'exact'], 'telefone': ['icontains', 'exact'], 'data_nascimento': ['exact'], 'is_active': ['exact'], } filter_overrides = { ObjectIdField: { 'filter_class': django_filters.CharFilter } } class ClientViewSet(viewsets.ModelViewSet): queryset = Cliente.objects.all() serializer_class = ClienteSerializer filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = ClienteFilter search_fields = '__all__' ordering_fields = '__all__' permission_classes = [permissions.IsAuthenticated, HasGroupOrPermission] # Adjust permissions as needed authorization_classes = [JWTAuthentication] parser_classes = [MultiPartParser, FormParser] required_groups = [] # Required groups for this viewset # πŸ”₯ Define required permissions for each action def get_required_permissions(self): """ Dynamically generates the required permission based on the action and model name. Example: If action is "list" and model is "usuario" (from app "accounts"), it returns ('accounts', 'view_usuario'). """ action_permissions = { "list": "view", "retrieve": "view", "create": "add", "update": "change", "partial_update": "change", "destroy": "delete", } model = self.queryset.model model_name = model._meta.object_name.lower() app_label = model._meta.app_label # βœ… Get app_label dynamically action = action_permissions.get(self.action, "") codename = f"{action}_{model_name}" if action else "" # print(f"πŸ”Ή App Label: {app_label}") # print(f"πŸ”Ή Action: {self.action}") # print(f"πŸ”Ή Codename: {codename}") return app_label, codename # Returning ('app_label', codename) def get_permissions(self): if self.action in ['create']: return [permissions.AllowAny()] return super().get_permissions() def get_authentication_classes(self): if self.action in ['create']: return [] return super().get_authentication_classes() jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""List all registers.""", manual_parameters=manual_parameters, operation_summary='List registers', operation_id='clientes_list', tags=['clientes'], responses={200: openapi.Response('Data successfully listed', ClienteSerializer), 400: openapi.Response('Error Generating Response', ClienteSerializer)}, ) def list(self, request): return super().list(request) # jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') # manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Create a new object. Arguments: request -- The HTTP request object. Returns: A Response object containing the object as JSON. """, # manual_parameters=manual_parameters,\ responses={ 200: 'Register updated', 400: 'Error while updating register', }, request_body=ClienteSerializer,\ operation_summary='Create a new register', operation_id='clientes_create', tags=['clientes']) def create(self, request): return super().create(request) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Retrieve a specific register.""", manual_parameters=manual_parameters, operation_summary='Retrieve a specific register', operation_id='clientes_retrieve', tags=['clientes'], responses={200: openapi.Response('Data successfully retrieved', ClienteSerializer), 400: openapi.Response('Error Generating Response', ClienteSerializer)} ) def retrieve(self, request, *args, **kwargs): try: instance = self.queryset.get(id=ObjectId(kwargs['pk'])) serializer = self.get_serializer(instance) return Response(serializer.data, status=status.HTTP_200_OK) except Exception as e: return Response({'message': f'Error retrieving item => {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') id_param = openapi.Parameter('id', openapi.IN_PATH, description="ID of the register", type=openapi.TYPE_STRING, required=True) manual_parameters=[jwt_token_param, id_param] @swagger_auto_schema(operation_description=""" Update a specific register. Arguments: request -- The HTTP request object. Returns: A Response object containing the updated object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register updated', 400: 'Error while updating register', }, request_body=ClienteUpdateSerializer,\ operation_summary='Update a specific register', operation_id='clientes_update', tags=['clientes']) def update(self, request, *args, **kwargs): try: instance = self.queryset.get(id=ObjectId(kwargs['pk'])) serializer = self.get_serializer(instance, data=request.data, partial=True) if serializer.is_valid(): serializer.save() serializer = self.serializer_class(self.queryset.get(id=ObjectId(kwargs['pk']))) return Response({'message':'Data successfully updated','results':serializer.data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') id_param = openapi.Parameter('id', openapi.IN_PATH, description="ID of the register", type=openapi.TYPE_STRING, required=True) @swagger_auto_schema(operation_description=""" Partial update a specific register. Arguments: request -- The HTTP request object. Returns: A Response object containing the updated object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register partial updated', 400: 'Error while partial updating register', }, request_body=ClienteUpdateSerializer,\ operation_summary='Partial update a specific register', operation_id='clientes_partial_update', tags=['clientes']) def partial_update(self, request, *args, **kwargs): try: instance = self.queryset.get(id=ObjectId(kwargs['pk'])) serializer = self.get_serializer(instance, data=request.data, partial=True) if serializer.is_valid(): serializer.save() serializer = self.serializer_class(self.queryset.get(id=ObjectId(kwargs['pk']))) return Response({'message':'Data successfully updated','results':serializer.data}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({'message': f'Error updating item => {e}'}, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Delete a specific register. Arguments: request -- The HTTP request object. Returns: A 204 No Content response. """, manual_parameters=manual_parameters,\ responses={ 204: 'No Content', 400: 'Error while deleting a register', }, request_body=ClienteSerializer,\ operation_summary='Delete a specific register', operation_id='clientes_delete', tags=['clientes']) def destroy(self, request, *args, **kwargs): return super().destroy(request, *args, **kwargs) def get_logged_cliente(self, request): """Retrieve the authenticated Cliente from JWT token.""" try: print(f"Authenticated User: {request.user}") # Debugging if not request.user or not hasattr(request.user, "id"): raise ValueError("request.user is not set correctly") cliente_instance = Cliente.objects.get(id=ObjectId(request.user.id)) if not cliente_instance: return None cliente_data = {field.name: getattr(cliente_instance, field.name) for field in cliente_instance._meta.fields} cliente_data["id"] = str(cliente_instance.id) if "data_nascimento" in cliente_data and isinstance(cliente_data["data_nascimento"], datetime): cliente_data["data_nascimento"] = cliente_data["data_nascimento"].date() return Cliente(**cliente_data) # Convert dict to Cliente instance except Exception as e: print(f"Error fetching logged-in Cliente: {e}") return None @swagger_auto_schema(operation_description=""" Give access to a specific client. Arguments: request -- The HTTP request object. Returns: A 200 OK response. Token is returned. """, responses={ 200: 'Client user can login', 400: 'Error while giving access to a client', }, request_body=ClienteAuthSerializer,\ operation_summary='Client user can login', operation_id='clientes-self_login', tags=['clientes-self']) @action( detail=False, methods=["post"], url_path="login", permission_classes=[permissions.AllowAny], authentication_classes=[], parser_classes=[JSONParser], ) def login(self, request): """Authenticate Cliente and ensure they have only one token at a time""" serializer = ClienteAuthSerializer(data=request.data) serializer.is_valid(raise_exception=True) user = serializer.validated_data["user"] # πŸ”₯ Connect to MongoDB db = get_mongo_connection() cliente_tokens = db["accounts_clientetoken"] # πŸ”₯ Delete any existing token for the Cliente delete_result = cliente_tokens.delete_many({"cliente_id": user.id}) if delete_result.deleted_count > 0: print(f"πŸ—‘οΈ Deleted {delete_result.deleted_count} existing token(s) for Cliente {user.id}") # βœ… Create and store JWT token = ClienteToken.create_token(user) return Response({"token": token.key}, status=status.HTTP_200_OK) @swagger_auto_schema(operation_description=""" Check if a client token is valid. Arguments: request -- The HTTP request object. Returns: A 200 OK response. """, responses={ 200: 'Token is valid', 400: 'Error while checking token', }, request_body=ClienteAuthCheckSerializer,\ operation_summary='Check if a client token is valid', operation_id='clientes-self_check', tags=['clientes-self']) @action( detail=False, methods=["post"], url_path="check", permission_classes=[permissions.AllowAny], # Allow public access parser_classes=[JSONParser], ) def check_token(self, request): """Check if a given Cliente token is valid and exists in MongoDB""" token = request.data.get("token", None) if not token: return Response({"error": "Token is required"}, status=status.HTTP_400_BAD_REQUEST) try: # Decode the JWT token decoded_token = AccessToken(token) cliente_id = decoded_token.get("sub") if not cliente_id: return Response({"error": "Invalid token structure"}, status=status.HTTP_401_UNAUTHORIZED) # Connect to MongoDB and check if the token exists db = get_mongo_connection() cliente_tokens = db["accounts_clientetoken"] token_exists = cliente_tokens.find_one({"key": token}) if not token_exists: return Response({"error": "Token not found"}, status=status.HTTP_401_UNAUTHORIZED) return Response({"message": "Valid token"}, status=status.HTTP_200_OK) except Exception as e: print(f"❌ Token check failed: {e}") return Response({"error": "Invalid or expired token"}, status=status.HTTP_401_UNAUTHORIZED) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Logout a specific client with token access. Arguments: request -- The HTTP request object. Returns: A 200 OK response. """, responses={ 200: 'Client loggeed out', 400: 'Error while logging out client', }, manual_parameters=manual_parameters, operation_summary='Logged in Client user can logout', request_body=ClienteAuthLogoutSerializer,\ operation_id='clientes-self_logout', tags=['clientes-self']) @action( detail=False, methods=["post"], url_path="logout", permission_classes=[permissions.IsAuthenticated], authentication_classes = [ClienteTokenAuthentication], parser_classes=[JSONParser], ) def logout(self, request): """Delete JWT token from MongoDB to log out Cliente.""" try: if not request.auth: return Response({"error": "No authentication token provided"}, status=status.HTTP_401_UNAUTHORIZED) token_key = request.auth.key # Get token key from request print(f"πŸ”Ή Logging out Cliente {request.user.id} with token {token_key}") # Delete the token from MongoDB deleted, _ = ClienteToken.objects.filter(key=token_key).delete() if deleted == 0: return Response({"error": "Token not found"}, status=status.HTTP_404_NOT_FOUND) return Response({"message": "Logged out successfully"}, status=status.HTTP_200_OK) except Exception as e: print(f"❌ Logout failed: {e}") return Response({"error": f"Logout failed: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) def get_logged_cliente(self, request): """Retrieve the authenticated Cliente from JWT token.""" try: print(f"Authenticated User: {request.user}") # Debugging if not request.user or not hasattr(request.user, "id"): raise ValueError("request.user is not set correctly") cliente_instance = Cliente.objects.get(id=ObjectId(request.user.id)) if not cliente_instance: return None cliente_data = {field.name: getattr(cliente_instance, field.name) for field in cliente_instance._meta.fields} cliente_data["id"] = str(cliente_instance.id) if "data_nascimento" in cliente_data and isinstance(cliente_data["data_nascimento"], datetime): cliente_data["data_nascimento"] = cliente_data["data_nascimento"].date() return Cliente(**cliente_data) # Convert dict to Cliente instance except Exception as e: print(f"Error fetching logged-in Cliente: {e}") return None jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Retrieve my data.""", manual_parameters=manual_parameters, operation_summary='Retrieve my data', operation_id='clientes-self_retrive', tags=['clientes-self'], responses={200: openapi.Response('Data successfully retrieved', ClienteSerializer), 400: openapi.Response('Error Generating Response', ClienteSerializer)}, ) @action( detail=False, methods=["get"], url_path="me", permission_classes=[IsAuthenticated], authentication_classes=[ClienteTokenAuthentication], parser_classes=[JSONParser], ) def retrieve_me(self, request): """Retrieve the authenticated Cliente's details.""" print(f"πŸ”Ή Authenticated User: {request.user}") # Debugging if not request.user or not hasattr(request.user, "id"): return Response({"message": "Cliente not found"}, status=status.HTTP_404_NOT_FOUND) cliente = self.get_logged_cliente(request) if not cliente: return Response({"message": "Cliente not found"}, status=status.HTTP_404_NOT_FOUND) serializer = ClienteSerializer(cliente) # βœ… No need for `data=request.data` in GET return Response(serializer.data, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Update my specific register. Arguments: request -- The HTTP request object. Returns: A Response object containing the updated object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register updated', 400: 'Error while updating register', }, request_body=ClienteSerializer,\ operation_summary='Update my specific register', operation_id='clientes-self_update', tags=['clientes-self']) @action( detail=False, methods=["put"], url_path="me/update", name="Update My Profile", permission_classes=[IsAuthenticated], authentication_classes=[ClienteTokenAuthentication] ) def update_me(self, request): """Update the authenticated user's details (full update).""" serializer = self.get_serializer(request.user, data=request.data, partial=True) serializer.is_valid(raise_exception=True) # Exclude password unless explicitly provided if "password" in serializer.validated_data: password = serializer.validated_data.pop("password") request.user.set_password(password) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Partial Update my specific register. Arguments: request -- The HTTP request object. Returns: A Response object containing the updated object as JSON. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register partial updated', 400: 'Error while partial updating register', }, request_body=ClienteSerializer,\ operation_summary='Partial Update my specific register', operation_id='clientes-self_partial_update', tags=['clientes-self']) @action(detail=False, methods=['patch'], url_path='me/partial-update', name='Partial Update My Profile', permission_classes=[IsAuthenticated], authentication_classes=[ClienteTokenAuthentication], parser_classes=[JSONParser]) def partial_update_me(self, request): """Partially update the authenticated user's details.""" serializer = self.get_serializer(request.user, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Inactivate my specific register. Arguments: request -- The HTTP request object. Returns: A Response object 200. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register inactivated', 400: 'Error while inactivating register', }, request_body=ClienteAuthLogoutSerializer,\ operation_summary='Inactivate my specific register', operation_id='clientes-self_inactivate', tags=['clientes-self']) @action(detail=False, methods=['patch'], url_path='me/inactivate', name='Inactive My Profile', permission_classes=[IsAuthenticated], authentication_classes=[ClienteTokenAuthentication], parser_classes=[JSONParser]) def inactive_me(self, request): """Deactivate the authenticated user's account.""" request.user.is_active = False request.user.save() return Response({"message": "Account deactivated"}, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Activate my specific register. Arguments: request -- The HTTP request object. Returns: A Response 200. """, manual_parameters=manual_parameters,\ responses={ 200: 'Register activated', 400: 'Error while activating register', }, request_body=ClienteAuthLogoutSerializer,\ operation_summary='Activate my specific register', operation_id='clientes-self_activate', tags=['clientes-self']) @action(detail=False, methods=['patch'], url_path='me/activate', name='Activate My Profile', permission_classes=[IsAuthenticated], authentication_classes=[ClienteTokenAuthentication], parser_classes=[JSONParser]) def activate_me(self, request): """Deactivate the authenticated user's account.""" request.user.is_active = True request.user.save() return Response({"message": "Account activated"}, status=status.HTTP_200_OK) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema(operation_description=""" Delete my specific register. Arguments: request -- The HTTP request object. Returns: A Response 204. """, manual_parameters=manual_parameters,\ responses={ 204: 'No Content', 400: 'Error while deleting register', }, request_body=ClienteAuthLogoutSerializer,\ operation_summary='Activate my specific register', operation_id='clientes-self_delete', tags=['clientes-self']) @action(detail=False, methods=['delete'], url_path='me/delete', name='Delete My Profile', permission_classes=[IsAuthenticated], authentication_classes=[ClienteTokenAuthentication], parser_classes=[JSONParser]) def destroy_me(self, request): """Delete the authenticated user's account.""" request.user.delete() return Response(status=status.HTTP_204_NO_CONTENT) from rest_framework import viewsets, status from rest_framework.response import Response from django.contrib.auth.models import Group from accounts.serializers import GroupSerializer from rest_framework import permissions class GroupFilter(django_filters.FilterSet): class Meta: model = Group fields = { 'id': ['in', 'exact'], 'name': ['icontains', 'exact'], } class GroupViewSet(viewsets.ModelViewSet): queryset = Group.objects.all() serializer_class = GroupSerializer permission_classes = [permissions.IsAuthenticated, HasGroupOrPermission] authentication_classes = [JWTAuthentication] filterset_class = GroupFilter filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] search_fields = '__all__' ordering_fields = '__all__' required_groups = [] # πŸ”₯ Define required permissions for each action def get_required_permissions(self): """ Dynamically generates the required permission based on the action and model name. Example: If action is "list" and model is "usuario" (from app "accounts"), it returns ('accounts', 'view_usuario'). """ action_permissions = { "list": "view", "retrieve": "view", "create": "add", "update": "change", "partial_update": "change", "destroy": "delete", } model = self.queryset.model model_name = model._meta.object_name.lower() app_label = model._meta.app_label # βœ… Get app_label dynamically action = action_permissions.get(self.action, "") codename = f"{action}_{model_name}" if action else "" # print(f"πŸ”Ή App Label: {app_label}") # print(f"πŸ”Ή Action: {self.action}") # print(f"πŸ”Ή Codename: {codename}") return app_label, codename # Returning ('app_label', codename) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""List data.""", manual_parameters=manual_parameters, operation_summary='List data', operation_id='niveis_list', tags=['niveis'], responses={200: openapi.Response('Data successfully listed', GroupSerializer), 400: openapi.Response('Error Generating Response', GroupSerializer)}, ) def list(self, request): return super().list(request) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Retrieve data.""", manual_parameters=manual_parameters, operation_summary='Retrieve data', operation_id='niveis_retrieve', tags=['niveis'], responses={200: openapi.Response('Data successfully retrieved', GroupSerializer), 400: openapi.Response('Error Generating Response', GroupSerializer)}, ) def retrieve(self, request, pk=None): return super().retrieve(request, pk) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Create a new register.""", manual_parameters=manual_parameters, operation_summary='Create a new register', operation_id='niveis_create', tags=['niveis'], responses={200: openapi.Response('Data successfully created', GroupSerializer), 400: openapi.Response('Error Generating Response', GroupSerializer)}, request_body=GroupSerializer ) def create(self, request, *args, **kwargs): """Handles creation of a new Group and registers permissions in `auth_group_permissions`.""" permission_ids = request.data.get("permissions", "").split(",") permission_ids = [int(perm_id.strip()) for perm_id in permission_ids if perm_id.strip().isdigit()] # Check if the group name already exists # Connect to DB to fetch data from auth group permissions db = get_mongo_connection() group_db = db.auth_group group_permissions_db = db.auth_group_permissions permission = db.auth_permission # Check if the group name already exists if group_db.find_one({"name": request.data["name"]}): return Response( {"message": "This group name already exists!"}, status=status.HTTP_400_BAD_REQUEST, ) # If permissions are provided, check if they exist if permission_ids: for perm_id in permission_ids: if not permission.find_one({"id": perm_id}): return Response( {"message": f"Permission with ID {perm_id} does not exist"}, status=status.HTTP_400_BAD_REQUEST, ) serializer = self.get_serializer(data=request.data) if serializer.is_valid(): group = serializer.save() group_id = group.id # Get the group ID print(f"Group created: {group}") # Register permissions using `AuthGroupPermissionSerializer` if permission_ids: try: #permissions_in_group_db = db.auth_group_permissions.find({"group_id": group_id}) # get last id field last_perm = group_permissions_db.find_one(sort=[("id", -1)]) new_id = (last_perm["id"] + 1) if last_perm else 1 print("Registering permissions...") for perm_id in permission_ids: perm_data = {"id": new_id, "group_id": group_id, "permission_id": perm_id} group_permissions_db.insert_one(perm_data) new_id += 1 except Exception as e: print(f"Error registering permissions: {e}") return Response( {"message": "Error registering permissions"}, status=status.HTTP_400_BAD_REQUEST, ) return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Update a register.""", manual_parameters=manual_parameters, operation_summary='Update a register', operation_id='niveis_update', tags=['niveis'], responses={200: openapi.Response('Data successfully updated', GroupSerializer), 400: openapi.Response('Error Generating Response', GroupSerializer)}, request_body=GroupUpdateSerializer ) def update(self, request, *args, **kwargs): """Handles updating the Group and its permissions.""" instance = self.get_object() #print(f"Instance ID: {instance.id}") # Update permissions permission_ids = request.data.get("permissions", "").split(",") permis # Connect to DB to fetch data from auth group permissionsds = [int(perm_id.strip()) for perm_id in permission_ids if perm_id.strip().isdigit()] #print(f"permission_ids: {permission_ids}") # Connect to DB to fetch data from auth group permissions db = get_mongo_connection() group_db = db.auth_group group_permissions_db = db.auth_group_permissions permission = db.auth_permission # Check if the group name already exists if 'name' in request.data and group_db.find_one({"name": request.data["name"], "id": {"$ne": instance.id}}): return Response( {"message": "This group name already exists!"}, status=status.HTTP_400_BAD_REQUEST, ) # If permissions are provided, check if they exist if permission_ids: for perm_id in permission_ids: if not permission.find_one({"id": perm_id}): return Response( {"message": f"Permission with ID {perm_id} does not exist"}, status=status.HTTP_400_BAD_REQUEST, ) serializer = self.get_serializer(instance, data=request.data, partial=True) # partial=False for full update if serializer.is_valid(): # Update group name group = serializer.save() group_id = group.id # Get the group ID try: # Remove old permissions group_permissions_db.delete_many({"group_id": group_id}) # get last id field last_perm = group_permissions_db.find_one(sort=[("id", -1)]) new_id = (last_perm["id"] + 1) if last_perm else 1 # print(f"new_id: {new_id}") # print(f"permission_ids: {permission_ids}") if permission_ids: print("Updating permissions...") for perm_id in permission_ids: #print(f"perm_id: {perm_id}") perm_data = {"id": new_id, "group_id": int(group_id), "permission_id": int(perm_id)} #print(f"perm_data: {perm_data}") group_permissions_db.insert_one(perm_data) new_id += 1 #print(f"new_id: {new_id}") except Exception as e: print(f"Error updating permissions: {e}") return Response( {"message": "Error updating permissions"}, status=status.HTTP_400_BAD_REQUEST, ) return Response(serializer.data) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Delete a register.""", manual_parameters=manual_parameters, operation_summary='Delete a register', operation_id='niveis_delete', tags=['niveis'], responses={204: openapi.Response('No Content', GroupSerializer), 400: openapi.Response('Error Generating Response', GroupSerializer)} ) def destroy(self, request, *args, **kwargs): instance = self.get_object() try: # Connect to DB to fetch data from auth group permissions db = get_mongo_connection() group_db = db.auth_group group_permissions_db = db.auth_group_permissions group_users_db = db.accounts_usuario_groups # Check if the group name already exists if group_db.find_one({"id": instance.id}): group_permissions_db.delete_many({"group_id": instance.id}) group_users_db.delete_many({"group_id": instance.id}) except Exception as e: print(f"Error deleting group: {e}") return Response( {"message": "Error deleting group"}, status=status.HTTP_400_BAD_REQUEST, ) return super().destroy(request, *args, **kwargs) #Method not allowed @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def partial_update(self, request, *args, **kwargs): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) class PermissionFilter(django_filters.FilterSet): class Meta: model = Permission fields = { 'id': ['in', 'exact'], 'name': ['icontains', 'exact'], 'codename': ['icontains', 'exact'], 'content_type__app_label': ['icontains', 'exact'], 'content_type__model': ['icontains', 'exact'], } class PermissionViewSet(viewsets.ModelViewSet): queryset = Permission.objects.all() serializer_class = PermissionSerializer permission_classes = [permissions.IsAuthenticated] authentication_classes = [JWTAuthentication] filterset_class = PermissionFilter filter_backends = [django_filters.DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] search_fields = '__all__' ordering_fields = '__all__' required_groups = [] # πŸ”₯ Define required permissions for each action def get_required_permissions(self): """ Dynamically generates the required permission based on the action and model name. Example: If action is "list" and model is "usuario" (from app "accounts"), it returns ('accounts', 'view_usuario'). """ action_permissions = { "list": "view", "retrieve": "view", "create": "add", "update": "change", "partial_update": "change", "destroy": "delete", } model = self.queryset.model model_name = model._meta.object_name.lower() app_label = model._meta.app_label # βœ… Get app_label dynamically action = action_permissions.get(self.action, "") codename = f"{action}_{model_name}" if action else "" # print(f"πŸ”Ή App Label: {app_label}") # print(f"πŸ”Ή Action: {self.action}") # print(f"πŸ”Ή Codename: {codename}") return app_label, codename # Returning ('app_label', codename) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""List data.""", manual_parameters=manual_parameters, operation_summary='List data', operation_id='permissoes_list', tags=['permissoes'], responses={200: openapi.Response('Data successfully listed', PermissionSerializer), 400: openapi.Response('Error Generating Response', PermissionSerializer)} ) def list(self, request, *args, **kwargs): return super().list(request) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def create(self, request): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def retrieve(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def update(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def partial_update(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(responses={405: 'Method not allowed'}, auto_schema=None) def destroy(self, request, pk=None): return Response({'message': 'Method not allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) from rest_framework_simplejwt.views import TokenObtainPairView from rest_framework_simplejwt.tokens import UntypedToken, RefreshToken from pymongo import MongoClient from django.conf import settings from rest_framework.response import Response from rest_framework import status class CustomTokenObtainPairView(TokenObtainPairView): """ πŸ”₯ Custom Token View to manually store access & refresh tokens """ def post(self, request, *args, **kwargs): response = super().post(request, *args, **kwargs) if response.status_code == 200: tokens = response.data access_token = tokens.get("access") refresh_token = tokens.get("refresh") # βœ… Decode tokens properly access_jti = UntypedToken(access_token)["jti"] # βœ… Use UntypedToken for Access refresh_jti = RefreshToken(refresh_token)["jti"] # βœ… Use RefreshToken for Refresh # Connect to DB to fetch data from auth group permissionsπŸ”Ή Access Token JTI: {access_jti}") print(f"πŸ”Ή Refresh Token JTI: {refresh_jti}") # βœ… Store in MongoDB db = get_mongo_connection() outstanding_tokens = db.token_blacklist_outstandingtoken if outstanding_tokens.find_one({"jti":refresh_jti}): outstanding_tokens.update_one({"jti":refresh_jti}, {"$set": {"access_jti": access_jti, "access_token": access_token}}) # Get the last id field # last_token = outstanding_tokens.find_one(sort=[("id", -1)]) # new_id = (last_token["id"] + 1) if last_token else 1 # # πŸ”Ή Store Access Token # if not outstanding_tokens.find_one({"jti": access_jti}): # outstanding_tokens.insert_one({ # "id":new_id, # "jti": access_jti, # "token_type": "access", # }) # # πŸ”Ή Store Refresh Token # if not outstanding_tokens.find_one({"jti": refresh_jti}): # outstanding_tokens.insert_one({ # "id":new_id + 1, # "jti": refresh_jti, # "token_type": "refresh", # }) return response from pymongo import MongoClient from django.conf import settings from rest_framework.response import Response from rest_framework import status from rest_framework_simplejwt.tokens import UntypedToken from rest_framework_simplejwt.views import TokenVerifyView class CustomTokenVerifyView(TokenVerifyView): def post(self, request, *args, **kwargs): """ πŸš€ Overrides default token verification to check MongoDB """ token = request.data.get("token") if not token: return Response({"error": "Token is required"}, status=status.HTTP_400_BAD_REQUEST) try: decoded_token = UntypedToken(token) jti = decoded_token["jti"] token_type = decoded_token["token_type"] print(f"πŸ”Ž Extracted JTI: {jti}") print(f"πŸ”Ž Token type: {token_type}") except Exception as e: return Response({"error": f"Invalid token: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST) # βœ… Check MongoDB for tokens # Connect to DB to fetch data from auth group permissions db = get_mongo_connection() outstanding_tokens = db.token_blacklist_outstandingtoken token_blacklist = db.token_blacklist_blacklistedtoken # # πŸ”Ž Log all outstanding tokens # print(f"πŸ“Œ Outstanding Tokens in DB: {list(outstanding_tokens.find({}, {'jti': 1, '_id': 0}))}") # πŸ”Ž Search for extracted JTI token_in_db = outstanding_tokens.find_one({"$or": [{"jti": jti}, {"access_jti": jti}]}) if not token_in_db: print("❌ Token not found in OutstandingToken collection.") return Response({"error": "Token not found"}, status=status.HTTP_401_UNAUTHORIZED) # βœ… Check if token is blacklisted is_blacklisted = token_blacklist.find_one({"token_id": token_in_db.get("id")}) is not None if is_blacklisted: return Response({"error": "This token has been blacklisted"}, status=status.HTTP_401_UNAUTHORIZED) return Response({"message": "Token is valid", "token_type": token_type}, status=status.HTTP_200_OK) from rest_framework_simplejwt.views import TokenBlacklistView from rest_framework_simplejwt.tokens import RefreshToken from rest_framework.response import Response from rest_framework import status from pymongo import MongoClient from django.conf import settings class CustomTokenBlacklistView(TokenBlacklistView): def post(self, request, *args, **kwargs): """ πŸš€ Override the default logout to avoid Djongo ORM issues """ token = request.data.get("refresh") # Extract refresh token from request if not token: return Response({"error": "Refresh token is required"}, status=status.HTTP_400_BAD_REQUEST) try: decoded_token = UntypedToken(token) jti = decoded_token["jti"] token_type = decoded_token["token_type"] print(f"πŸ”Ž Extracted JTI: {jti}") print(f"πŸ”Ž Token type: {token_type}") except Exception as e: return Response({"error": f"Invalid token: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST) # Step 3: Check if the token is blacklisted using raw MongoDB query try: # Connect to DB to fetch data from auth group permissions db = get_mongo_connection() # Check if token exists in OutstandingToken collection token_in_db = db.token_blacklist_outstandingtoken.find_one({"$or": [{"jti": jti}, {"access_jti": jti}]}) print(f"Token id in OutstandingToken: {token_in_db['id']}") if not token_in_db: print("❌ Old refresh token not found in database.") raise ValidationError("Invalid refresh token.") # Check if token is blacklisted is_blacklisted = db.token_blacklist_blacklistedtoken.find_one({"token_id": token_in_db['id']}) is not None if is_blacklisted: print("⚠️ Token is already blacklisted.") raise ValidationError("This token is already blacklisted.") else: print("βœ… Token is not blacklisted!.") except Exception as e: print(f"❌ Failed to check token blacklist: {e}") raise ValidationError("Error checking token blacklist.") # πŸ›  Manually check MongoDB instead of using Django ORM try: # Manually insert the token into MongoDB from datetime import datetime last_blacklisted_token = db.token_blacklist_blacklistedtoken.find_one(sort=[("id", -1)]) new_id = (last_blacklisted_token["id"] + 1) if last_blacklisted_token else 1 db.token_blacklist_blacklistedtoken.insert_one({"id":int(new_id),"token_id": token_in_db['id'], "jti": jti, "token":token_in_db['token'], "access_jti":token_in_db['access_jti'], "access_token":token_in_db["access_token"], "blacklisted_at": datetime.now()}) #Update the outstanding token to insert blacklisted at information db.token_blacklist_outstandingtoken.update_one({"id": token_in_db['id']}, {"$set": {"blacklist_id":new_id, "blacklisted_at": datetime.now()}}) print("βœ… Token successfully blacklisted.") return Response({"message": "Token successfully blacklisted"}, status=status.HTTP_200_OK) except Exception as e: return Response({"error": f"MongoDB Error to Insert into blacklisted token: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) from rest_framework_simplejwt.views import TokenRefreshView from .serializers import CustomTokenRefreshSerializer class CustomTokenRefreshView(TokenRefreshView): serializer_class = CustomTokenRefreshSerializer from rest_framework import viewsets, permissions from .models import UserActionLog from .serializers import UserActionLogSerializer from django_filters.rest_framework import DjangoFilterBackend class UserActionLogFilter(django_filters.FilterSet): id = django_filters.CharFilter(method='filter_by_id') user = django_filters.CharFilter(method='filter_by_user_id') class Meta: model = UserActionLog fields = { 'action': ['icontains', 'exact'], 'timestamp': ['exact','gte','lte'], 'path': ['icontains', 'exact'], } # πŸ”₯ Override ObjectIdField to use CharFilter instead filter_overrides = { UserActionLog._meta.get_field("id"): { "filter_class": django_filters.CharFilter, "extra": lambda f: {"lookup_expr": "exact"}, }, } def filter_by_id(self, queryset, name, value): """ Convert ObjectId to a valid query filter. """ try: object_id = ObjectId(value) return queryset.filter(id=object_id) except Exception: return queryset.none() def filter_by_user_id(self, queryset, name, value): """ Convert user ObjectId to a valid query filter. """ try: user = User.objects.get(id=ObjectId(value)) return queryset.filter(user=user) except User.DoesNotExist: return queryset.none() except Exception as e: print(f"❌ Filtering error: {e}") return queryset.none() class UserActionLogViewSet(viewsets.ModelViewSet): queryset = UserActionLog.objects.all().order_by('-timestamp') serializer_class = UserActionLogSerializer filter_backends = [DjangoFilterBackend, filters.OrderingFilter, filters.SearchFilter] filterset_class = UserActionLogFilter # Use the custom filter search_fields = ['action', 'path'] # Search support ordering_fields = ['timestamp', 'user', 'action'] # Allow ordering authentication_classes = [JWTAuthentication] permission_classes = [permissions.IsAuthenticated, HasGroupOrPermission] required_groups = [] # πŸ”₯ Define required permissions for each action def get_required_permissions(self): """ Dynamically generates the required permission based on the action and model name. Example: If action is "list" and model is "usuario" (from app "accounts"), it returns ('accounts', 'view_usuario'). """ action_permissions = { "list": "view", "retrieve": "view", "create": "add", "update": "change", "partial_update": "change", "destroy": "delete", } model = self.queryset.model model_name = model._meta.object_name.lower() app_label = model._meta.app_label # βœ… Get app_label dynamically action = action_permissions.get(self.action, "") codename = f"{action}_{model_name}" if action else "" # print(f"πŸ”Ή App Label: {app_label}") # print(f"πŸ”Ή Action: {self.action}") # print(f"πŸ”Ή Codename: {codename}") return app_label, codename # Returning ('app_label', codename) def get_queryset(self): queryset = super().get_queryset() # Additional custom filtering logic if needed return queryset jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""List data.""", manual_parameters=manual_parameters, operation_summary='List data', operation_id='acessos_list', tags=['acessos'], responses={200: openapi.Response('Data successfully listed', UserActionLogSerializer), 400: openapi.Response('Error Generating Response', UserActionLogSerializer)}, ) def list(self, request, *args, **kwargs): """ List all logs. """ return super().list(request, *args, **kwargs) # jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') # manual_parameters=[jwt_token_param] # @swagger_auto_schema( # operation_description="""Retrieve data.""", # manual_parameters=manual_parameters, # operation_summary='Retrieve data', # operation_id='logs_retrieve', tags=['logs'], # responses={200: openapi.Response('Data successfully retrieved', UserActionLogSerializer), # 400: openapi.Response('Error Generating Response', UserActionLogSerializer)}, # ) # def retrieve(self, request, pk=None): # """ # Retrieve a specific log. # """ # print(f"pk: {pk}") # pk = ObjectId(pk) # return super().retrieve(request, pk) jwt_token_param = openapi.Parameter('Authorization', openapi.IN_HEADER, description="JWT Token", type=openapi.TYPE_STRING, required=True, example='Bearer $access_token') manual_parameters=[jwt_token_param] @swagger_auto_schema( operation_description="""Retrieve my data.""", manual_parameters=manual_parameters, operation_summary='Retrieve my data', operation_id='acessos_self-list', tags=['acessos-self'], responses={200: openapi.Response('Data successfully listed', UserActionLogSerializer), 400: openapi.Response('Error Generating Response', UserActionLogSerializer)}, ) @action(detail=False, methods=['get'], url_path='my-logs', permission_classes=[permissions.IsAuthenticated]) def my_logs(self, request): """ Retrieve the authenticated user's logs. """ queryset = self.get_queryset().filter(user=request.user) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return Response(serializer.data, status=status.HTTP_200_OK) ### Not allowed methods @swagger_auto_schema(auto_schema=None) def retrieve(self,request,pk=None): return Response({"message": "Method not allowed"}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(auto_schema=None) def create(self, request): return Response({"message": "Method not allowed"}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(auto_schema=None) def update(self, request, *args, **kwargs): return Response({"message": "Method not allowed"}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(auto_schema=None) def partial_update(self, request, *args, **kwargs): return Response({"message": "Method not allowed"}, status=status.HTTP_405_METHOD_NOT_ALLOWED) @swagger_auto_schema(auto_schema=None) def destroy(self, request, *args, **kwargs): return Response({"message": "Method not allowed"}, status=status.HTTP_405_METHOD_NOT_ALLOWED) # from django.contrib.auth.models import PermissionsMixin # class AuthGroupPermissionViewSet(viewsets.ModelViewSet): # queryset = PermissionsMixin.objects.all() # serializer_class = AuthGroupPermissionSerializer # permission_classes = [permissions.IsAuthenticated] # authentication_classes = [JWTAuthentication] # filter_backends = [django_filters.DjangoFilterBackend, filters.OrderingFilter, filters.SearchFilter] # search_fields = ['group__name', 'permission__name'] # ordering_fields = ['id', 'group__name', 'permission__name'] # required_groups = [] # # πŸ”₯ Define required permissions for each action # def get_required_permissions(self): # """ # Dynamically generates the required permission based on the action and model name. # Example: If action is "list" and model is "usuario" (from app "accounts"), # it returns ('accounts', 'view_usuario'). # """ # action_permissions = { # "list": "view", # "retrieve": "view", # "create": "add", # "update": "change", # "partial_update": "change", # "destroy": "delete", # } # model = self.queryset.model # model_name = model._meta.object_name.lower() # app_label = model._meta.app_label # βœ… Get app_label dynamically # action = action_permissions.get(self.action, "") # codename = f"{action}_{model_name}" if action else "" # print(f"πŸ”Ή App Label: {app_label}") # print(f"πŸ”Ή Action: {self.action}") # print(f"πŸ”Ή Codename: {codename}") # return app_label, codename # Returning ('app_label', codename)