fureunoir/schon
fureunoir/schon
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Fixes: OrderViewSet fixes

Browse source
This commit is contained in:
Egor Pavlovich Gorbunov 2025-06-07 18:57:45 +03:00
parent f6ade70b57
commit 71dcf8e922
2 changed files with 27 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
core/permissions.py
View file

@ -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

36
core/viewsets.py
View file

@ -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)