Refactored entire permssion

Author: MySecondLanguage

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,24 +3,23 @@
 
 from nxtbn.users import UserRole
 
-class NxtbnAdminPermission(BasePermission):
-    def has_permission(self, request, view):
-        user = request.user
-        if not user.is_authenticated:
-            return False
-        
-        if user.is_superuser:
-            return True
-
-        return False
-
 
+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):
+        if not request.user.is_authenticated:
+            return False
+        
+        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
@@ -29,22 +28,84 @@ def has_permission(self, request, view):
 
         # Check if the user has the generated permission
         return request.user.has_perm(permission_name)
+    
 
+class ModelPermissions(BasePermission):
 
-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 has_permission(self, request, view):
+        if not request.user.is_authenticated:
+            return False
+        
+        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_cls = getattr(view, 'queryset', None)
+        if model_cls is None:
+            return False
+
+        model_meta = model_cls.model._meta
+
+        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}',
+        }
+
+        required_permission = method_permissions_map.get(request.method)
+
+        if required_permission is None:
+            return False
+
+        return request.user.has_perm(required_permission)
+
+
+
+
+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/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/order/admin_queries.py

@@ -1,7 +1,7 @@
 import graphene
 from graphene_django.filter import DjangoFilterConnectionField
 
-from nxtbn.core.admin_permissions import check_user_permissions
+from nxtbn.core.admin_permissions import staff_required
 from nxtbn.order.admin_types import OrderType
 from nxtbn.order.models import Address, Order
 from nxtbn.users import UserRole
@@ -11,31 +11,17 @@ class AdminOrderQuery(graphene.ObjectType):
     orders = DjangoFilterConnectionField(OrderType)
     order = graphene.Field(OrderType, id=graphene.Int(required=True))
 
-    # all_addresses = graphene.List(AddressGraphType)
-    # address = graphene.Field(AddressGraphType, id=graphene.Int(required=True))
+   
 
+    @staff_required
     def resolve_orders(self, info, **kwargs):
-        check_user_permissions(info, any_staff=True)
         return Order.objects.all()
 
+    @staff_required
     def resolve_order(self, info, id):
-        check_user_permissions(info, any_staff=True)
         try:
             order = Order.objects.get(id=id)
         except Order.DoesNotExist:
             raise Exception("Order not found")
 
-        return order
-
-    def resolve_addresses(self, info):
-        check_user_permissions(info, any_staff=True)
-        return Address.objects.all()
-
-    def resolve_address(self, info, id):
-        check_user_permissions(info, any_staff=True)
-        try:
-            address = Address.objects.get(id=id)
-        except Address.DoesNotExist:
-            raise Exception("Address not found")
-        
-        return address
+        return order

nxtbn/product/admin_mutations.py

@@ -1,5 +1,4 @@
 import graphene
-from nxtbn.core.admin_permissions import check_user_permissions
 from nxtbn.product.admin_types import CategoryTranslationType, CategoryType, CollectionTranslationType, ProductTagTranslationType, ProductTranslationType, ProductVariantTranslationType, SupplierTranslationType
 from nxtbn.product.models import Category, CategoryTranslation, CollectionTranslation, ProductTagTranslation, ProductTranslation, ProductVariantTranslation, SupplierTranslation
 from nxtbn.users import UserRole
@@ -20,7 +19,6 @@ class Arguments:
     category = graphene.Field(CategoryType)
 
     def mutate(self, info, id, input):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER])
         category = Category.objects.get(id=id)
         category.name = input.name
         category.description = input.description
@@ -48,7 +46,6 @@ class Arguments:
     product_translation = graphene.Field(ProductTranslationType)
 
     def mutate(self, info, base_product_id, lang_code, name, summary, description, meta_title, meta_description):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER, UserRole.TRANSLATOR])
         try:
             product_translation = ProductTranslation.objects.get(product_id=base_product_id, language_code=lang_code)
         except ProductTranslation.DoesNotExist:
@@ -76,7 +73,6 @@ class Arguments:
     category_translation = graphene.Field(CategoryTranslationType)
 
     def mutate(self, info, base_category_id, lang_code, name, description, meta_title, meta_description):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER, UserRole.TRANSLATOR])
         try:
             category_translation = CategoryTranslation.objects.get(category_id=base_category_id, language_code=lang_code)
         except CategoryTranslation.DoesNotExist:
@@ -103,7 +99,6 @@ class Arguments:
     supplier_translation = graphene.Field(SupplierTranslationType)
 
     def mutate(self, info, base_supplier_id, lang_code, name, description, meta_title, meta_description):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER, UserRole.TRANSLATOR])
         try:
             supplier_translation = SupplierTranslation.objects.get(supplier_id=base_supplier_id, language_code=lang_code)
         except SupplierTranslation.DoesNotExist:
@@ -127,7 +122,6 @@ class Arguments:
     product_variant_translation = graphene.Field(ProductVariantTranslationType)
 
     def mutate(self, info, base_product_variant_id, lang_code, name, description, meta_title, meta_description):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER, UserRole.TRANSLATOR])
         try:
             product_variant_translation = ProductVariantTranslation.objects.get(product_variant_id=base_product_variant_id, language_code=lang_code)
         except ProductVariantTranslation.DoesNotExist:
@@ -147,7 +141,6 @@ class Arguments:
     product_tag_translation = graphene.Field(ProductTagTranslationType)
 
     def mutate(self, info, base_product_tag_id, lang_code, name, description, meta_title, meta_description):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER, UserRole.TRANSLATOR])
         try:
             product_tag_translation = ProductTagTranslation.objects.get(product_tag_id=base_product_tag_id, language_code=lang_code)
         except ProductTagTranslation.DoesNotExist:
@@ -170,7 +163,6 @@ class Arguments:
     collection_translation = graphene.Field(CollectionTranslationType)
 
     def mutate(self, info, base_collection_id, lang_code, name, description, meta_title, meta_description):
-        check_user_permissions(info, allowed_roles=[UserRole.STORE_MANAGER, UserRole.ADMIN, UserRole.PRODUCT_MANAGER, UserRole.TRANSLATOR])
         try:
             collection_translation = CollectionTranslation.objects.get(collection_id=base_collection_id, language_code=lang_code)
         except CollectionTranslation.DoesNotExist: