Merge pull request #376 from MySecondLanguage/handled-entire-permission

Handled entire permission

Author: MySecondLanguage

nxtbn/admin_schema.py

@@ -7,13 +7,14 @@
 from nxtbn.product.admin_queries import ProductQuery
 from nxtbn.users.admin_mutation import AdminUserMutation
 from nxtbn.order.admin_queries import AdminOrderQuery
+from nxtbn.users.admin_queries import UserAdminQuery
 from nxtbn.warehouse.admin_queries import WarehouseQuery
 
 
 
 
 
-class Query(ProductQuery, AdminOrderQuery, AdminCoreQuery, WarehouseQuery, AdminCartQuery):
+class Query(ProductQuery, AdminOrderQuery, AdminCoreQuery, WarehouseQuery, AdminCartQuery, UserAdminQuery):
     pass
 
 class Mutation(AdminUserMutation, ProductMutation, CoreMutation):

nxtbn/cart/admin_query.py

@@ -3,7 +3,7 @@
 
 from nxtbn.cart.models import Cart
 from nxtbn.cart.admin_types import CartItemType, CartType
-from nxtbn.core.admin_permissions import check_user_permissions
+from nxtbn.core.admin_permissions import staff_required
 
 
 class AdminCartQuery(graphene.ObjectType):
@@ -13,19 +13,19 @@ class AdminCartQuery(graphene.ObjectType):
 
     items_in_cart = graphene.List(CartItemType, cart_id=graphene.ID(required=True))
 
+    @staff_required
     def resolve_carts(self, info, **kwargs):
-        check_user_permissions(info, any_staff=True)
         return Cart.objects.all()
 
+    @staff_required
     def resolve_cart_by_user(self, info, user_id):
-        check_user_permissions(info, any_staff=True)
         try:
             return Cart.objects.get(user_id=user_id)
         except Cart.DoesNotExist:
             return None
 
+    @staff_required
     def resolve_items_in_cart(self, info, cart_id):
-        check_user_permissions(info, any_staff=True)
         try:
             cart = Cart.objects.get(id=cart_id)
             return cart.items.all()

nxtbn/core/admin_mutation.py

@@ -2,7 +2,6 @@
 import graphene
 
 from nxtbn.core import CurrencyTypes
-from nxtbn.core.admin_permissions import check_user_permissions
 from nxtbn.core.admin_types import CurrencyExchangeType
 from nxtbn.core.models import CurrencyExchange
 from nxtbn.users import UserRole
@@ -23,7 +22,6 @@ class Arguments:
 
     @staticmethod
     def mutate(root, info, input):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN])
         # Validate base_currency
         base_currency = input.base_currency
         if base_currency != settings.BASE_CURRENCY:
@@ -51,7 +49,6 @@ class Arguments:
 
     @staticmethod
     def mutate(root, info, id, input):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN])
         try:
             currency_exchange = CurrencyExchange.objects.get(pk=id)
         except CurrencyExchange.DoesNotExist:
@@ -69,7 +66,6 @@ class Arguments:
 
     @staticmethod
     def mutate(root, info, id):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN])
         try:
             currency_exchange = CurrencyExchange.objects.get(pk=id)
         except CurrencyExchange.DoesNotExist:

nxtbn/core/admin_permissions.py

@@ -3,64 +3,109 @@
 
 from nxtbn.users import UserRole
 
-class NxtbnAdminPermission(BasePermission):
+
+import functools
+from graphql import GraphQLError
+class GranularPermission(BasePermission):
+    def get_permission_name(self, model_name, action):
+       
+        return f"{model_name}.{action}"
+
     def has_permission(self, request, view):
-        user = request.user
-        if not user.is_authenticated:
+        if not request.user.is_authenticated:
             return False
 
-        if user.is_superuser:
+        if request.user.is_superuser:
+            return True
+        
+        if request.user.method in SAFE_METHODS and request.user.is_staff: # Every staff can view
             return True
+      
+        model_name = view.queryset.model.__name__.lower()  # Get model name dynamically
+        action = view.action.required_perm
 
-        return False
+        permission_name = self.get_permission_name(model_name, action)
 
+        # Check if the user has the generated permission
+        return request.user.has_perm(permission_name)
+    
 
-class RoleBasedPermission(BasePermission):
-    """
-    Custom permission that grants or denies access based on the role and action.
-    """
-    def has_permission(self, request, view):
-        user = request.user
+class ModelPermissions(BasePermission):
 
-        # Ensure user is authenticated
-        if not user.is_authenticated:
+    def has_permission(self, request, view):
+        if not request.user.is_authenticated:
             return False
 
-        if user.role == UserRole.ADMIN:
+        if request.user.is_superuser:
+            return True
+        
+        if request.user.method in SAFE_METHODS and request.user.is_staff: # Every staff can view
             return True
 
 
-        action = getattr(view, 'role_action', None) or getattr(view, 'action', None)
 
+        model_cls = getattr(view, 'queryset', None)
+        if model_cls is None:
+            return False
 
+        model_meta = model_cls.model._meta
 
-        # Check if action exists in the permissions for the user's role
-        role_permissions = view.ROLE_PERMISSIONS.get(user.role, set())
+        method_permissions_map = {
+            'GET': f'{model_meta.app_label}.view_{model_meta.model_name}',
+            'OPTIONS': f'{model_meta.app_label}.view_{model_meta.model_name}',
+            'HEAD': f'{model_meta.app_label}.view_{model_meta.model_name}',
+            'POST': f'{model_meta.app_label}.add_{model_meta.model_name}',
+            'PUT': f'{model_meta.app_label}.change_{model_meta.model_name}',
+            'PATCH': f'{model_meta.app_label}.change_{model_meta.model_name}',
+            'DELETE': f'{model_meta.app_label}.delete_{model_meta.model_name}',
+        }
 
-        if "all" in role_permissions:
-            return True
+        required_permission = method_permissions_map.get(request.method)
 
-        # Grant permission if the action is allowed for the user's role
-        if action in role_permissions:
-            return True
+        if required_permission is None:
+            return False
 
-        return False
-    
+        return request.user.has_perm(required_permission)
 
-def check_user_permissions(info, any_staff=False, allowed_roles=[]):
-    if not info.context.user.is_authenticated:
-        raise Exception("You must be logged in to perform this action")
-    
-    if not info.context.user.is_staff:
-        raise Exception("You must be a staff to perform this action")
-    
-    if info.context.user.is_superuser:
-        return True
-    
-    if any_staff:
-        return True
-    
-    if info.context.user.role not in allowed_roles:
-        raise Exception("You do not have permission to perform this action")
+
+
+
+def required_perm(code: str): # Used in graphql only
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(self, info, *args, **kwargs):
+            operation = info.operation.operation
+            user = info.context.user
+
+            if user.is_anonymous:
+                raise GraphQLError("Authentication required")
+
+            if operation == "query":
+                return func(self, info, *args, **kwargs)
+
+            
+            
+            if not user.has_perm(code):  # Check if user has the required permission
+                raise GraphQLError("Permission denied")  # Block unauthorized access
+
+            return func(self, info, *args, **kwargs)  # Call the actual resolver
+        
+        return wrapper
 
-    return True
+    return decorator
+
+
+def staff_required(func):
+    @functools.wraps(func)
+    def wrapper(self, info, *args, **kwargs):
+        user = info.context.user
+
+        if user.is_anonymous:
+            raise GraphQLError("Authentication required")
+
+        if not user.is_staff:  # Check if the user is a staff member
+            raise GraphQLError("Permission denied")  # Block access if the user is not staff
+
+        return func(self, info, *args, **kwargs)  # Call the actual resolver
+
+    return wrapper

nxtbn/core/admin_queries.py

@@ -3,7 +3,7 @@
 from graphql import GraphQLError
 
 from nxtbn.core import CurrencyTypes
-from nxtbn.core.admin_permissions import check_user_permissions
+from nxtbn.core.admin_permissions import staff_required
 from nxtbn.core.admin_types import AdminCurrencyTypesEnum, CurrencyExchangeType
 from nxtbn.core.models import CurrencyExchange
 from graphene_django.filter import DjangoFilterConnectionField
@@ -14,19 +14,18 @@ class AdminCoreQuery(graphene.ObjectType):
     currency_exchange = graphene.Field(CurrencyExchangeType, id=graphene.ID(required=True))
     allowed_currency_list = graphene.List(AdminCurrencyTypesEnum)
 
+    @staff_required
     def resolve_currency_exchanges(self, info, **kwargs):
-        check_user_permissions(info, any_staff=True)
         return CurrencyExchange.objects.all()
 
     def resolve_currency_exchange(self, info, id):
-        check_user_permissions(info, any_staff=True)
         try:
             return CurrencyExchange.objects.get(id=id)
         except CurrencyExchange.DoesNotExist:
             return None
 
+    @staff_required
     def resolve_allowed_currency_list(self, info):
-        check_user_permissions(info, any_staff=True)
         allowed_currency_list = settings.ALLOWED_CURRENCIES
 
         if not allowed_currency_list:

nxtbn/core/api/dashboard/views.py

@@ -28,13 +28,15 @@
 from nxtbn.core import LanguageChoices
 from nxtbn.core.api.dashboard.serializers import InvoiceSettingsSerializer, SiteSettingsSerializer
 from nxtbn.core.models import InvoiceSettings, SiteSettings
+from nxtbn.users import UserRole
 
 
 
 class SiteSettingsView(generics.RetrieveUpdateAPIView):
     queryset = SiteSettings.objects.all()
     serializer_class = SiteSettingsSerializer
 
+
     def get_object(self):
         # Get the current site
         current_site = get_current_site(self.request)
@@ -50,6 +52,8 @@ class InvoiceSettingsView(generics.RetrieveUpdateAPIView):
     queryset = InvoiceSettings.objects.all()
     serializer_class = InvoiceSettingsSerializer
 
+
+
     def get_object(self):
         current_site = get_current_site(self.request)
         try:

nxtbn/core/enum_perms.py

@@ -0,0 +1,7 @@
+
+from django.db import models
+
+class PermissionsEnum(models.TextChoices):
+    CAN_APPROVE_ORDER = "can_approve_order"
+    CAN_CANCEL_ORDER = "can_cancel_order"
+    CAN_SHIP_ORDER = "can_ship_order"

nxtbn/discount/api/dashboard/views.py

@@ -9,6 +9,8 @@
 import django_filters
 from django_filters import rest_framework as filters
 
+from nxtbn.users import UserRole
+
 
 class PromocodeFilter(filters.FilterSet):
     username = filters.CharFilter(field_name='username', lookup_expr='icontains')
@@ -58,6 +60,7 @@ class PromoCodeUpdateRetrieveDeleteView(generics.RetrieveUpdateDestroyAPIView):
     serializer_class = PromoCodeCountedSerializer
     lookup_field = 'id'
 
+
 class AttachPromoCodeEntitiesAPIView(generics.CreateAPIView):
     serializer_class = AttachPromoCodeEntitiesSerializer
 
@@ -111,4 +114,3 @@ class PromoCodeUsageListAPIView(generics.ListAPIView):
     queryset = PromoCodeUsage.objects.all()
     serializer_class = PromoCodeUsageSerializer
     pagination_class = NxtbnPagination
-

nxtbn/filemanager/api/dashboard/views.py

@@ -10,7 +10,6 @@
     DocumentSerializer,
     ImageSerializer,
 )
-from nxtbn.core.admin_permissions import NxtbnAdminPermission
 from nxtbn.core.paginator import NxtbnPagination
 
 class ImageFilter(django_filters.FilterSet):
@@ -25,7 +24,6 @@ class ImageListView(generics.ListCreateAPIView):
     serializer_class = ImageSerializer
     queryset = Image.objects.all().order_by('-created_at')
     pagination_class = NxtbnPagination
-    permission_classes = (NxtbnAdminPermission,)
     filter_backends = [ DjangoFilterBackend,]
     filterset_class = ImageFilter
 
@@ -34,20 +32,17 @@ class ImageDetailView(generics.RetrieveUpdateDestroyAPIView):
     queryset = Image.objects.all()
     serializer_class = ImageSerializer
     pagination_class = NxtbnPagination
-    permission_classes = (NxtbnAdminPermission,)
     lookup_field = "id"
 
 
 class DocumentListView(generics.ListCreateAPIView):
     serializer_class = DocumentSerializer
     queryset = Document.objects.all()
-    permission_classes = (NxtbnAdminPermission,)
     pagination_class = NxtbnPagination
 
 
 class DocumentDetailView(generics.RetrieveUpdateDestroyAPIView):
     queryset = Document.objects.all()
     serializer_class = DocumentSerializer
     pagination_class = NxtbnPagination
-    permission_classes = (NxtbnAdminPermission,)
     lookup_field = "id"

nxtbn/invoice/api/dashboard/views.py

@@ -6,6 +6,7 @@
 
 from nxtbn.invoice.api.dashboard.serializers import OrderInvoiceSerializer
 from nxtbn.order.models import Order
+from nxtbn.users import UserRole
 
 
 class OrderInvoiceAPIView(generics.RetrieveAPIView):