diff --git a/core/permissions.py b/core/permissions.py index a189e7c3..2bb31dc8 100644 --- a/core/permissions.py +++ b/core/permissions.py @@ -14,18 +14,6 @@ class IsOwnerOrReadOnly(permissions.BasePermission): class EvibesPermission(permissions.BasePermission): - """ - Custom permission class for EvibesViewSet endpoints. - - - 'create' may be explicitly allowed via view.additional['create'] == 'ALLOW'. - - Certain actions are scoped to the request.user’s own objects. - - Standard model perms ('add', 'view', 'change', 'delete') are enforced for all other actions, - including for staff users. - - Publicly visible models allow anonymous list/retrieve. - - If an instance or queryset has a "user" attribute, ensure that the request.user is the same, - unless the user is an admin with the required django permission. - """ - ACTION_PERM_MAP = { "retrieve": "view", "list": "view", @@ -43,6 +31,8 @@ class EvibesPermission(permissions.BasePermission): "current", "add_order_product", "remove_order_product", + "bulk_add_order_products", + "bulk_remove_order_products", "add_wishlist_product", "remove_wishlist_product", "bulk_add_wishlist_products", @@ -51,8 +41,7 @@ class EvibesPermission(permissions.BasePermission): } def has_permission(self, request, view): - action = str(getattr(view, "action", None)) - + action = view.action model = view.queryset.model app_label = model._meta.app_label model_name = model._meta.model_name @@ -67,10 +56,8 @@ class EvibesPermission(permissions.BasePermission): return True perm_prefix = self.ACTION_PERM_MAP.get(action) - if perm_prefix: - codename = f"{perm_prefix}_{model_name}" - if request.user.has_perm(f"{app_label}.{codename}"): - return True + if perm_prefix and request.user.has_perm(f"{app_label}.{perm_prefix}_{model_name}"): + return True return bool(action in ("list", "retrieve") and getattr(model, "is_publicly_visible", False)) @@ -81,33 +68,17 @@ class EvibesPermission(permissions.BasePermission): if hasattr(obj, "user"): if obj.user == request.user: return True - # Allow admins who hold the required model permission app_label = obj._meta.app_label model_name = obj._meta.model_name - action = getattr(view, "action", None) + action = view.action perm_prefix = self.ACTION_PERM_MAP.get(action) return bool(perm_prefix and request.user.has_perm(f"{app_label}.{perm_prefix}_{model_name}")) - model = view.queryset.model - app_label = model._meta.app_label - model_name = model._meta.model_name - action = str(getattr(view, "action", None)) - - if action == 'retrieve' and request.method in permissions.SAFE_METHODS and hasattr(obj, - 'user') and obj.user is None: - lookup_val = view.kwargs.get(view.lookup_field) - if str(obj.human_readable_id) == lookup_val or str(obj.uuid) == lookup_val: - return True - - perm_prefix = self.ACTION_PERM_MAP.get(action) - return bool(perm_prefix and request.user.has_perm(f"{app_label}.{perm_prefix}_{model_name}")) + perm_prefix = self.ACTION_PERM_MAP.get(view.action) + return bool(perm_prefix and request.user.has_perm( + f"{view.queryset.model._meta.app_label}.{perm_prefix}_{view.queryset.model._meta.model_name}")) def has_queryset_permission(self, request, view, queryset): - """ - Filter the base queryset according to the action and user. - For models with a "user" field, restrict access to records belonging to the request user - unless the admin holds the needed permissions. - """ model = view.queryset.model app_label = model._meta.app_label model_name = model._meta.model_name diff --git a/core/viewsets.py b/core/viewsets.py index fc23a1b7..bd86916d 100644 --- a/core/viewsets.py +++ b/core/viewsets.py @@ -235,6 +235,7 @@ class FeedbackViewSet(EvibesViewSet): @extend_schema_view(**ORDER_SCHEMA) class OrderViewSet(EvibesViewSet): lookup_field = 'lookup_value' + lookup_url_kwarg = 'lookup_value' queryset = Order.objects.prefetch_related("order_products").all() filter_backends = [DjangoFilterBackend] filterset_class = OrderFilter @@ -262,13 +263,12 @@ class OrderViewSet(EvibesViewSet): if user.has_perm("core.view_order"): return qs - return qs.filter(Q(user=user) | Q(user__isnull=True)) + return qs.filter(user=user) def get_object(self): lookup_val = self.kwargs.get(self.lookup_field) - queryset = self.filter_queryset(self.get_queryset()) obj = get_object_or_404( - queryset, + self.get_queryset(), Q(uuid=lookup_val) | Q(human_readable_id=lookup_val) ) self.check_object_permissions(self.request, obj) @@ -285,12 +285,12 @@ class OrderViewSet(EvibesViewSet): ) @action(detail=True, methods=["post"], url_path="buy") - def buy(self, request, *_args, **kwargs): + def buy(self, request, *args, **kwargs): serializer = BuyOrderSerializer(data=request.data) serializer.is_valid(raise_exception=True) - order_uuid = kwargs.get("pk") + lookup_val = kwargs.get(self.lookup_field) try: - order = Order.objects.get(user=request.user, uuid=order_uuid) + order = Order.objects.get(user=request.user, uuid=lookup_val) instance = order.buy( force_balance=serializer.validated_data.get("force_balance"), force_payment=serializer.validated_data.get("force_payment"), @@ -306,7 +306,7 @@ class OrderViewSet(EvibesViewSet): case _: raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}")) except Order.DoesNotExist: - return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": _(f"order {order_uuid} not found")}) + return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": _(f"order {lookup_val} not found")}) @action(detail=False, methods=["post"], url_path="buy_unregistered") @method_decorator(ratelimit(key="ip", rate="5/h" if not DEBUG else "888/h")) @@ -314,7 +314,7 @@ class OrderViewSet(EvibesViewSet): serializer = BuyUnregisteredOrderSerializer(data=request.data) serializer.is_valid(raise_exception=True) order = Order.objects.create(status="MOMENTAL") - products = [product.get("product_uuid") for product in serializer.validated_data.get("products")] + products = [p["product_uuid"] for p in serializer.validated_data["products"]] transaction = order.buy_without_registration( products=products, promocode_uuid=serializer.validated_data.get("promocode_uuid"), @@ -331,8 +331,9 @@ class OrderViewSet(EvibesViewSet): def add_order_product(self, request, **kwargs): serializer = AddOrderProductSerializer(data=request.data) serializer.is_valid(raise_exception=True) + lookup_val = kwargs.get(self.lookup_field) try: - order = Order.objects.get(uuid=kwargs.get("pk")) + order = Order.objects.get(uuid=lookup_val) if not (request.user.has_perm("core.add_orderproduct") or request.user == order.user): raise PermissionDenied(permission_denied_message) @@ -340,7 +341,6 @@ class OrderViewSet(EvibesViewSet): product_uuid=serializer.validated_data.get("product_uuid"), attributes=format_attributes(serializer.validated_data.get("attributes")), ) - return Response(status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data) except Order.DoesNotExist: return Response(status=status.HTTP_404_NOT_FOUND) @@ -349,8 +349,9 @@ class OrderViewSet(EvibesViewSet): def remove_order_product(self, request, **kwargs): serializer = RemoveOrderProductSerializer(data=request.data) serializer.is_valid(raise_exception=True) + lookup_val = kwargs.get(self.lookup_field) try: - order = Order.objects.get(uuid=kwargs.get("pk")) + order = Order.objects.get(uuid=lookup_val) if not (request.user.has_perm("core.delete_orderproduct") or request.user == order.user): raise PermissionDenied(permission_denied_message) @@ -358,41 +359,40 @@ class OrderViewSet(EvibesViewSet): product_uuid=serializer.validated_data.get("product_uuid"), attributes=format_attributes(serializer.validated_data.get("attributes")), ) - return Response(status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data) except Order.DoesNotExist: return Response(status=status.HTTP_404_NOT_FOUND) @action(detail=True, methods=["post"], url_path="bulk_add_order_products") - def bulk_add_order_products(self, request, *_args, **kwargs): + def bulk_add_order_products(self, request, *args, **kwargs): serializer = BulkAddOrderProductsSerializer(data=request.data) serializer.is_valid(raise_exception=True) + lookup_val = kwargs.get(self.lookup_field) try: - order = Order.objects.get(uuid=kwargs.get("pk")) + order = Order.objects.get(uuid=lookup_val) if not (request.user.has_perm("core.add_orderproduct") or request.user == order.user): raise PermissionDenied(permission_denied_message) order = order.bulk_add_products( products=serializer.validated_data.get("products"), ) - return Response(status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data) except Order.DoesNotExist: return Response(status=status.HTTP_404_NOT_FOUND) @action(detail=True, methods=["post"], url_path="bulk_remove_order_products") - def bulk_remove_order_products(self, request, *_args, **kwargs): + def bulk_remove_order_products(self, request, *args, **kwargs): serializer = BulkRemoveOrderProductsSerializer(data=request.data) serializer.is_valid(raise_exception=True) + lookup_val = kwargs.get(self.lookup_field) try: - order = Order.objects.get(uuid=kwargs.get("pk")) + order = Order.objects.get(uuid=lookup_val) if not (request.user.has_perm("core.delete_orderproduct") or request.user == order.user): raise PermissionDenied(permission_denied_message) order = order.bulk_remove_products( products=serializer.validated_data.get("products"), ) - return Response(status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data) except Order.DoesNotExist: return Response(status=status.HTTP_404_NOT_FOUND)