Fix permission vulnerabilities: IDOR, data exposure, and decorator misconfigs

Security fixes across the authorization stack:

- serializers.py: Gate accepted_risks behind Risk_Acceptance permission
  so Readers can no longer see risk acceptance details via the API
- views.py (api_v2): Add defense-in-depth permission checks to metadata
  batch operations; add clarifying comments to generate_report actions
- benchmark/views.py: Fix IDOR allowing cross-product benchmark updates
  by validating product ownership after object lookup
- cred/views.py: Fix 14 decorator misconfigurations — standalone views
  now use @user_is_configuration_authorized, product-scoped views use
  correct parent model types (Engagement/Test/Finding instead of Product)
- object/views.py: Harden edit/delete with get_object_or_404 and parent
  product mismatch checks raising PermissionDenied
- tool_product/views.py: Same hardening as object/views.py

New test file: unittests/test_permissions_audit.py with 11 tests covering
risk acceptance exposure, metadata batch auth, note relationship checks,
benchmark IDOR, and object/tool_product parent validation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: devGregA

dojo/api_v2/serializers.py

@@ -1760,9 +1760,7 @@ class FindingSerializer(serializers.ModelSerializer):
     mitigated_by = serializers.PrimaryKeyRelatedField(required=False, allow_null=True, queryset=User.objects.all())
     tags = TagListSerializerField(required=False)
     request_response = serializers.SerializerMethodField()
-    accepted_risks = RiskAcceptanceSerializer(
-        many=True, read_only=True, source="risk_acceptance_set",
-    )
+    accepted_risks = serializers.SerializerMethodField()
     push_to_jira = serializers.BooleanField(default=False)
     found_by = serializers.PrimaryKeyRelatedField(
         queryset=Test_Type.objects.all(), many=True,
@@ -1806,6 +1804,16 @@ def __init__(self, *args, **kwargs):
                 many=True, required=False, queryset=Endpoint.objects.all(),
             )
 
+    def get_accepted_risks(self, obj):
+        request = self.context.get("request")
+        if request is None:
+            return []
+        if not user_has_permission(request.user, obj, Permissions.Risk_Acceptance):
+            return []
+        return RiskAcceptanceSerializer(
+            obj.risk_acceptance_set.all(), many=True,
+        ).data
+
     @extend_schema_field(serializers.DateTimeField())
     def get_jira_creation(self, obj):
         return jira_helper.get_jira_creation(obj)

dojo/api_v2/views.py

@@ -45,6 +45,7 @@
     serializers,
 )
 from dojo.api_v2.prefetch.prefetcher import _Prefetcher
+from dojo.authorization.authorization import user_has_permission_or_403
 from dojo.authorization.roles_permissions import Permissions
 from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.cred.queries import get_authorized_cred_mappings
@@ -351,7 +352,11 @@ def get_queryset(self):
         responses={status.HTTP_200_OK: serializers.ReportGenerateSerializer},
     )
     @action(
-        detail=True, methods=["post"], permission_classes=[IsAuthenticated],
+        detail=True, methods=["post"],
+        # IsAuthenticated only: report generation requires View permission,
+        # enforced by the permission-filtered get_queryset(). The viewset's
+        # permission_classes would check Edit (POST), which is too restrictive.
+        permission_classes=[IsAuthenticated],
     )
     def generate_report(self, request, pk=None):
         endpoint = self.get_object()
@@ -475,7 +480,11 @@ def reopen(self, request, pk=None):
         responses={status.HTTP_200_OK: serializers.ReportGenerateSerializer},
     )
     @action(
-        detail=True, methods=["post"], permission_classes=[IsAuthenticated],
+        detail=True, methods=["post"],
+        # IsAuthenticated only: report generation requires View permission,
+        # enforced by the permission-filtered get_queryset(). The viewset's
+        # permission_classes would check Edit (POST), which is too restrictive.
+        permission_classes=[IsAuthenticated],
     )
     def generate_report(self, request, pk=None):
         engagement = self.get_object()
@@ -1383,7 +1392,11 @@ def set_finding_as_original(self, request, pk, new_fid):
         responses={status.HTTP_200_OK: serializers.ReportGenerateSerializer},
     )
     @action(
-        detail=False, methods=["post"], permission_classes=[IsAuthenticated],
+        detail=False, methods=["post"],
+        # IsAuthenticated only: report generation requires View permission,
+        # enforced by the permission-filtered get_queryset(). The viewset's
+        # permission_classes would check Edit (POST), which is too restrictive.
+        permission_classes=[IsAuthenticated],
     )
     def generate_report(self, request):
         findings = self.get_queryset()
@@ -1728,38 +1741,55 @@ def batch(self, request, pk=None):
         serialized_data = serializers.MetaMainSerializer(data=request.data)
         if serialized_data.is_valid(raise_exception=True):
             if request.method == "POST":
-                self.process_post(request.data)
+                self.process_post(request)
                 status_code = status.HTTP_201_CREATED
             if request.method == "PATCH":
-                self.process_patch(request.data)
+                self.process_patch(request)
                 status_code = status.HTTP_200_OK
 
         return Response(status=status_code, data=serialized_data.data)
 
-    def process_post(self: object, data: dict):
-        product = Product.objects.filter(id=data.get("product")).first()
-        finding = Finding.objects.filter(id=data.get("finding")).first()
-        endpoint = Endpoint.objects.filter(id=data.get("endpoint")).first()
+    def _fetch_and_authorize_parents(self, request, permission_map):
+        """Fetch parent objects and verify the user has the required permissions."""
+        data = request.data
+        parents = {}
+        for field, (model, permission) in permission_map.items():
+            obj = model.objects.filter(id=data.get(field)).first()
+            if obj:
+                user_has_permission_or_403(request.user, obj, permission)
+            parents[field] = obj
+        return parents
+
+    def process_post(self, request):
+        data = request.data
+        parents = self._fetch_and_authorize_parents(request, {
+            "product": (Product, Permissions.Product_Edit),
+            "finding": (Finding, Permissions.Finding_Edit),
+            "endpoint": (Endpoint, Permissions.Location_Edit),
+        })
         metalist = data.get("metadata")
         for metadata in metalist:
             try:
                 DojoMeta.objects.create(
-                    product=product,
-                    finding=finding,
-                    endpoint=endpoint,
+                    product=parents["product"],
+                    finding=parents["finding"],
+                    endpoint=parents["endpoint"],
                     name=metadata.get("name"),
                     value=metadata.get("value"),
                     )
             except (IntegrityError) as ex:  # this should not happen as the data was validated in the batch call
                 raise ValidationError(str(ex))
 
-    def process_patch(self: object, data: dict):
-        product = Product.objects.filter(id=data.get("product")).first()
-        finding = Finding.objects.filter(id=data.get("finding")).first()
-        endpoint = Endpoint.objects.filter(id=data.get("endpoint")).first()
+    def process_patch(self, request):
+        data = request.data
+        parents = self._fetch_and_authorize_parents(request, {
+            "product": (Product, Permissions.Product_Edit),
+            "finding": (Finding, Permissions.Finding_Edit),
+            "endpoint": (Endpoint, Permissions.Location_Edit),
+        })
         metalist = data.get("metadata")
         for metadata in metalist:
-            dojometa = DojoMeta.objects.filter(product=product, finding=finding, endpoint=endpoint, name=metadata.get("name"))
+            dojometa = DojoMeta.objects.filter(product=parents["product"], finding=parents["finding"], endpoint=parents["endpoint"], name=metadata.get("name"))
             if dojometa:
                 try:
                     dojometa.update(
@@ -1815,7 +1845,11 @@ def destroy(self, request, *args, **kwargs):
         responses={status.HTTP_200_OK: serializers.ReportGenerateSerializer},
     )
     @action(
-        detail=True, methods=["post"], permission_classes=[IsAuthenticated],
+        detail=True, methods=["post"],
+        # IsAuthenticated only: report generation requires View permission,
+        # enforced by the permission-filtered get_queryset(). The viewset's
+        # permission_classes would check Edit (POST), which is too restrictive.
+        permission_classes=[IsAuthenticated],
     )
     def generate_report(self, request, pk=None):
         product = self.get_object()
@@ -1956,7 +1990,11 @@ def destroy(self, request, *args, **kwargs):
         responses={status.HTTP_200_OK: serializers.ReportGenerateSerializer},
     )
     @action(
-        detail=True, methods=["post"], permission_classes=[IsAuthenticated],
+        detail=True, methods=["post"],
+        # IsAuthenticated only: report generation requires View permission,
+        # enforced by the permission-filtered get_queryset(). The viewset's
+        # permission_classes would check Edit (POST), which is too restrictive.
+        permission_classes=[IsAuthenticated],
     )
     def generate_report(self, request, pk=None):
         product_type = self.get_object()
@@ -2143,7 +2181,11 @@ def get_serializer_class(self):
         responses={status.HTTP_200_OK: serializers.ReportGenerateSerializer},
     )
     @action(
-        detail=True, methods=["post"], permission_classes=[IsAuthenticated],
+        detail=True, methods=["post"],
+        # IsAuthenticated only: report generation requires View permission,
+        # enforced by the permission-filtered get_queryset(). The viewset's
+        # permission_classes would check Edit (POST), which is too restrictive.
+        permission_classes=[IsAuthenticated],
     )
     def generate_report(self, request, pk=None):
         test = self.get_object()

dojo/benchmark/views.py

@@ -46,6 +46,8 @@ def update_benchmark(request, pid, _type):
         field = request.POST.get("field")
         value = request.POST.get("value")
         value = {"true": True, "false": False}.get(value, value)
+        product = get_object_or_404(Product, id=pid)
+        bench = get_object_or_404(Benchmark_Product.objects.filter(product=product), id=bench_id)
 
         if field in {
             "enabled",
@@ -54,7 +56,6 @@ def update_benchmark(request, pid, _type):
             "get_notes",
             "delete_notes",
         }:
-            bench = Benchmark_Product.objects.get(id=bench_id)
             if field == "enabled":
                 bench.enabled = value
             elif field == "pass_fail":
@@ -90,21 +91,22 @@ def update_benchmark(request, pid, _type):
 @user_is_authorized(Product, Permissions.Benchmark_Edit, "pid")
 def update_benchmark_summary(request, pid, _type, summary):
     if request.method == "POST":
+        product = get_object_or_404(Product, id=pid)
+        benchmark_summary = get_object_or_404(Benchmark_Product_Summary.objects.filter(product=product), id=summary)
         field = request.POST.get("field")
         value = request.POST.get("value")
         value = {"true": True, "false": False}.get(value, value)
 
         if field in {"publish", "desired_level"}:
-            summary = Benchmark_Product_Summary.objects.get(id=summary)
             data = {}
             if field == "publish":
-                summary.publish = value
+                benchmark_summary.publish = value
                 data = {"publish": value}
             elif field == "desired_level":
-                summary.desired_level = value
-                data = {"desired_level": value, "text": asvs_level(summary)}
+                benchmark_summary.desired_level = value
+                data = {"desired_level": value, "text": asvs_level(benchmark_summary)}
 
-            summary.save()
+            benchmark_summary.save()
             return JsonResponse(data)
 
     return redirect_to_return_url_or_else(

dojo/cred/views.py

@@ -47,7 +47,7 @@ def all_cred_product(request, pid):
     return render(request, "dojo/view_cred_prod.html", {"product_tab": product_tab, "creds": creds, "prod": prod})
 
 
-@user_is_authorized(Cred_User, Permissions.Credential_Edit, "ttid")
+@user_is_configuration_authorized(Permissions.Credential_Edit)
 def edit_cred(request, ttid):
     tool_config = Cred_User.objects.get(pk=ttid)
     if request.method == "POST":
@@ -79,7 +79,7 @@ def edit_cred(request, ttid):
     })
 
 
-@user_is_authorized(Cred_User, Permissions.Credential_View, "ttid")
+@user_is_configuration_authorized(Permissions.Credential_View)
 def view_cred_details(request, ttid):
     cred = Cred_User.objects.get(pk=ttid)
     notes = cred.notes.all()
@@ -127,7 +127,7 @@ def cred(request):
 
 
 @user_is_authorized(Product, Permissions.Product_View, "pid")
-@user_is_authorized(Cred_User, Permissions.Credential_View, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_View, "ttid")
 def view_cred_product(request, pid, ttid):
     cred = get_object_or_404(
         Cred_Mapping.objects.select_related("cred_id"), id=ttid)
@@ -182,8 +182,8 @@ def view_cred_product(request, pid, ttid):
         })
 
 
-@user_is_authorized(Product, Permissions.Engagement_View, "eid")
-@user_is_authorized(Cred_User, Permissions.Credential_View, "ttid")
+@user_is_authorized(Engagement, Permissions.Engagement_View, "eid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_View, "ttid")
 def view_cred_product_engagement(request, eid, ttid):
     cred = get_object_or_404(
         Cred_Mapping.objects.select_related("cred_id"), id=ttid)
@@ -231,8 +231,8 @@ def view_cred_product_engagement(request, eid, ttid):
         })
 
 
-@user_is_authorized(Product, Permissions.Test_View, "tid")
-@user_is_authorized(Cred_User, Permissions.Credential_View, "ttid")
+@user_is_authorized(Test, Permissions.Test_View, "tid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_View, "ttid")
 def view_cred_engagement_test(request, tid, ttid):
     cred = get_object_or_404(
         Cred_Mapping.objects.select_related("cred_id"), id=ttid)
@@ -282,8 +282,8 @@ def view_cred_engagement_test(request, tid, ttid):
         })
 
 
-@user_is_authorized(Product, Permissions.Finding_View, "fid")
-@user_is_authorized(Cred_User, Permissions.Credential_View, "ttid")
+@user_is_authorized(Finding, Permissions.Finding_View, "fid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_View, "ttid")
 def view_cred_finding(request, fid, ttid):
     cred = get_object_or_404(
         Cred_Mapping.objects.select_related("cred_id"), id=ttid)
@@ -334,7 +334,7 @@ def view_cred_finding(request, fid, ttid):
 
 
 @user_is_authorized(Product, Permissions.Product_Edit, "pid")
-@user_is_authorized(Cred_User, Permissions.Credential_Edit, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_Edit, "ttid")
 def edit_cred_product(request, pid, ttid):
     cred = get_object_or_404(
         Cred_Mapping.objects.select_related("cred_id"), id=ttid)
@@ -362,7 +362,7 @@ def edit_cred_product(request, pid, ttid):
 
 
 @user_is_authorized(Engagement, Permissions.Engagement_Edit, "eid")
-@user_is_authorized(Cred_User, Permissions.Credential_Edit, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_Edit, "ttid")
 def edit_cred_product_engagement(request, eid, ttid):
     cred = get_object_or_404(
         Cred_Mapping.objects.select_related("cred_id"), id=ttid)
@@ -582,7 +582,6 @@ def new_cred_finding(request, fid):
         })
 
 
-@user_is_authorized(Cred_User, Permissions.Credential_Delete, "ttid")
 def delete_cred_controller(request, destination_url, elem_id, ttid):
     cred = Cred_Mapping.objects.filter(pk=ttid).first()
     if request.method == "POST":
@@ -662,30 +661,30 @@ def delete_cred_controller(request, destination_url, elem_id, ttid):
     })
 
 
-@user_is_authorized(Cred_User, Permissions.Credential_Delete, "ttid")
+@user_is_configuration_authorized(Permissions.Credential_Delete)
 def delete_cred(request, ttid):
     return delete_cred_controller(request, "cred", 0, ttid=ttid)
 
 
 @user_is_authorized(Product, Permissions.Product_Edit, "pid")
-@user_is_authorized(Cred_User, Permissions.Credential_Delete, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_Delete, "ttid")
 def delete_cred_product(request, pid, ttid):
     return delete_cred_controller(request, "all_cred_product", pid, ttid)
 
 
 @user_is_authorized(Engagement, Permissions.Engagement_Edit, "eid")
-@user_is_authorized(Cred_User, Permissions.Credential_Delete, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_Delete, "ttid")
 def delete_cred_engagement(request, eid, ttid):
     return delete_cred_controller(request, "view_engagement", eid, ttid)
 
 
 @user_is_authorized(Test, Permissions.Test_Edit, "tid")
-@user_is_authorized(Cred_User, Permissions.Credential_Delete, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_Delete, "ttid")
 def delete_cred_test(request, tid, ttid):
     return delete_cred_controller(request, "view_test", tid, ttid)
 
 
 @user_is_authorized(Finding, Permissions.Finding_Edit, "fid")
-@user_is_authorized(Cred_User, Permissions.Credential_Delete, "ttid")
+@user_is_authorized(Cred_Mapping, Permissions.Credential_Delete, "ttid")
 def delete_cred_finding(request, fid, ttid):
     return delete_cred_controller(request, "view_finding", fid, ttid)

dojo/object/views.py

@@ -1,7 +1,7 @@
 import logging
 
 from django.contrib import messages
-from django.core.exceptions import BadRequest
+from django.core.exceptions import PermissionDenied
 from django.http import HttpResponseRedirect
 from django.shortcuts import get_object_or_404, render
 from django.urls import reverse
@@ -63,12 +63,10 @@ def view_objects(request, pid):
 
 @user_is_authorized(Product, Permissions.Product_Tracking_Files_Edit, "pid")
 def edit_object(request, pid, ttid):
-    object_prod = Objects_Product.objects.get(pk=ttid)
+    object_prod = get_object_or_404(Objects_Product, pk=ttid)
     product = get_object_or_404(Product, id=pid)
     if object_prod.product != product:
-        msg = labels.ASSET_TRACKED_FILES_ID_MISMATCH_ERROR_MESSAGE % {"asset_id": pid,
-                                                                      "object_asset_id": object_prod.product.id}
-        raise BadRequest(msg)
+        raise PermissionDenied
 
     if request.method == "POST":
         tform = ObjectSettingsForm(request.POST, instance=object_prod)
@@ -94,12 +92,10 @@ def edit_object(request, pid, ttid):
 
 @user_is_authorized(Product, Permissions.Product_Tracking_Files_Delete, "pid")
 def delete_object(request, pid, ttid):
-    object_prod = Objects_Product.objects.get(pk=ttid)
+    object_prod = get_object_or_404(Objects_Product, pk=ttid)
     product = get_object_or_404(Product, id=pid)
     if object_prod.product != product:
-        msg = labels.ASSET_TRACKED_FILES_ID_MISMATCH_ERROR_MESSAGE % {"asset_id": pid,
-                                                                      "object_asset_id": object_prod.product.id}
-        raise BadRequest(msg)
+        raise PermissionDenied
 
     if request.method == "POST":
         tform = ObjectSettingsForm(request.POST, instance=object_prod)