fix: risk acceptance permission overrides

backend/core/permissions.py

@@ -37,13 +37,16 @@ def has_object_permission(self, request: Request, view, obj):
             obj, "is_published", False
         ):
             return True
+
+        # Check for view action permission overrides
+        current_action = getattr(view, "action", None)
+
+        if current_action:
+            permission_overrides = getattr(view, "permission_overrides", {})
+            _codename = permission_overrides.get(current_action, _codename)
+
         perm = Permission.objects.get(codename=_codename)
-        # special case of risk acceptance approval
-        if (
-            request.parser_context["request"]._request.resolver_match.url_name
-            == "risk-acceptances-accept"
-        ):
-            perm = Permission.objects.get(codename="approve_riskacceptance")
+
         return RoleAssignment.is_access_allowed(
             user=request.user,
             perm=perm,

backend/core/views.py

@@ -70,6 +70,7 @@
 from rest_framework.response import Response
 from rest_framework.utils.serializer_helpers import ReturnDict
 from rest_framework.views import APIView
+from rest_framework.exceptions import PermissionDenied
 
 
 from weasyprint import HTML
@@ -1675,6 +1676,12 @@ class RiskAcceptanceViewSet(BaseModelViewSet):
     API endpoint that allows risk acceptance to be viewed or edited.
     """
 
+    permission_overrides = {
+        "accept": "approve_riskacceptance",
+        "reject": "approve_riskacceptance",
+        "revoke": "approve_riskacceptance",
+    }
+
     model = RiskAcceptance
     serializer_class = RiskAcceptanceWriteSerializer
     filterset_fields = ["folder", "state", "approver", "risk_scenarios"]
@@ -1710,20 +1717,44 @@ def to_review(self, request):
 
     @action(detail=True, methods=["post"], name="Accept risk acceptance")
     def accept(self, request, pk):
-        if request.user == self.get_object().approver:
-            self.get_object().set_state("accepted")
+        if request.user != self.get_object().approver:
+            logger.error(
+                "Only the approver can accept the risk acceptance",
+                user=request.user,
+                approver=self.get_object().approver,
+            )
+            raise PermissionDenied(
+                {"error": "Only the approver can accept the risk acceptance"}
+            )
+        self.get_object().set_state("accepted")
         return Response({"results": "state updated to accepted"})
 
     @action(detail=True, methods=["post"], name="Reject risk acceptance")
     def reject(self, request, pk):
-        if request.user == self.get_object().approver:
-            self.get_object().set_state("rejected")
+        if request.user != self.get_object().approver:
+            logger.error(
+                "Only the approver can reject the risk acceptance",
+                user=request.user,
+                approver=self.get_object().approver,
+            )
+            raise PermissionDenied(
+                {"error": "Only the approver can reject the risk acceptance"}
+            )
+        self.get_object().set_state("rejected")
         return Response({"results": "state updated to rejected"})
 
     @action(detail=True, methods=["post"], name="Revoke risk acceptance")
     def revoke(self, request, pk):
-        if request.user == self.get_object().approver:
-            self.get_object().set_state("revoked")
+        if request.user != self.get_object().approver:
+            logger.error(
+                "Only the approver can revoke the risk acceptance",
+                user=request.user,
+                approver=self.get_object().approver,
+            )
+            raise PermissionDenied(
+                {"error": "Only the approver can revoke the risk acceptance"}
+            )
+        self.get_object().set_state("revoked")
         return Response({"results": "state updated to revoked"})
 
     @action(detail=False, methods=["get"], name="Get waiting risk acceptances")