add a manage_triage_rulesets action on Product API endpoint

Signed-off-by: tdruez <tdruez@aboutcode.org>

Author: tdruez

product_portfolio/api.py

@@ -7,6 +7,8 @@
 #
 
 from django.core.exceptions import ValidationError
+from django.db import transaction
+from django.shortcuts import get_object_or_404
 
 import django_filters
 from rest_framework import permissions
@@ -51,7 +53,11 @@
 from product_portfolio.models import ProductPolicyViolation
 from product_portfolio.models import ScanCodeProject
 from vulnerabilities.api import VulnerabilityAnalysisSerializer
+from vulnerabilities.triage.engine import delete_triage_records_for_assignment
+from vulnerabilities.triage.engine import reevaluate_product_rulesets
+from vulnerabilities.triage.models import ProductTriageRuleset
 from vulnerabilities.triage.models import TriageRecord
+from vulnerabilities.triage.models import TriageRuleset
 
 base_extra_kwargs = {
     "licenses": {
@@ -382,6 +388,14 @@ class Meta:
         )
 
 
+class TriageRulesetAssignmentSerializer(serializers.Serializer):
+    uuid = serializers.UUIDField(read_only=True)
+    name = serializers.CharField(read_only=True)
+    recommended_action = serializers.CharField(read_only=True)
+    precedence = serializers.IntegerField(read_only=True)
+    assigned = serializers.BooleanField(read_only=True)
+
+
 class ProductViewSet(
     ObjectPermissionsMixin,
     SendAboutFilesMixin,
@@ -470,6 +484,59 @@ def triage_records(self, request, uuid):
         serializer = TriageRecordSerializer(records, many=True)
         return Response(serializer.data)
 
+    @action(detail=True, methods=["get", "post"], url_path="manage_triage_rulesets")
+    def manage_triage_rulesets(self, request, uuid):
+        """
+        GET: list every enabled triage ruleset in this product's dataspace, each flagged
+        with whether it is currently assigned to this product.
+
+        POST: assign or unassign a single ruleset for this product.
+        Body: {"ruleset": "<uuid>", "assigned": true}
+        """
+        product = self.get_object()
+
+        if request.method == "POST":
+            if not isinstance(request.data, dict):
+                return Response(
+                    {"error": "Expected a JSON object with 'ruleset' and 'assigned'."},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+            ruleset_uuid = request.data.get("ruleset")
+            assigned = request.data.get("assigned")
+            if ruleset_uuid is None or assigned is None:
+                return Response(
+                    {"error": "Both 'ruleset' and 'assigned' are required."},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+            ruleset = get_object_or_404(
+                TriageRuleset.objects.scope(product.dataspace).filter(enabled=True),
+                uuid=ruleset_uuid,
+            )
+            with transaction.atomic():
+                assignment = ProductTriageRuleset.objects.filter(
+                    product=product, ruleset=ruleset
+                ).first()
+                if assigned and not assignment:
+                    ProductTriageRuleset.objects.create(
+                        product=product, ruleset=ruleset, dataspace=product.dataspace
+                    )
+                elif not assigned and assignment:
+                    assignment.delete()
+                    delete_triage_records_for_assignment(ruleset=ruleset, product=product)
+                reevaluate_product_rulesets(product)
+            return Response(status=status.HTTP_200_OK)
+
+        assigned_ruleset_ids = set(
+            product.product_triage_rulesets.values_list("ruleset_id", flat=True)
+        )
+        rulesets = TriageRuleset.objects.filter(dataspace=product.dataspace, enabled=True).order_by(
+            "-precedence", "name"
+        )
+        for ruleset in rulesets:
+            ruleset.assigned = ruleset.id in assigned_ruleset_ids
+        serializer = TriageRulesetAssignmentSerializer(rulesets, many=True)
+        return Response(serializer.data)
+
     @action(detail=True, methods=["post"], serializer_class=LoadSBOMsFormSerializer)
     def load_sboms(self, request, *args, **kwargs):
         """

product_portfolio/tests/test_api.py

@@ -52,6 +52,7 @@
 from vulnerabilities.tests import make_vulnerability
 from vulnerabilities.tests import make_vulnerability_analysis
 from vulnerabilities.triage.engine import evaluate_ruleset
+from vulnerabilities.triage.models import ProductTriageRuleset
 from vulnerabilities.triage.models import TriageAction
 from vulnerabilities.triage.tests import make_product_triage_ruleset
 from vulnerabilities.triage.tests import make_triage_ruleset
@@ -801,6 +802,107 @@ def test_api_product_endpoint_triage_records_action(self):
         self.assertIn("detected_date", entry)
         self.assertIn("last_checked", entry)
 
+    def test_api_product_endpoint_manage_triage_rulesets_get(self):
+        url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid])
+        self.client.login(username=self.base_user.username, password="secret")
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+
+        assigned_ruleset = make_triage_ruleset(self.dataspace, name="Assigned Ruleset")
+        make_product_triage_ruleset(self.product1, ruleset=assigned_ruleset)
+        unassigned_ruleset = make_triage_ruleset(self.dataspace, name="Unassigned Ruleset")
+        make_triage_ruleset(self.dataspace, name="Disabled Ruleset", enabled=False)
+
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_200_OK, response.status_code)
+        self.assertEqual(2, len(response.data))
+        entries_by_name = {entry["name"]: entry for entry in response.data}
+        self.assertTrue(entries_by_name[assigned_ruleset.name]["assigned"])
+        self.assertFalse(entries_by_name[unassigned_ruleset.name]["assigned"])
+        self.assertNotIn("Disabled Ruleset", entries_by_name)
+
+    def test_api_product_endpoint_manage_triage_rulesets_post_assigns(self):
+        url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid])
+        self.client.login(username=self.base_user.username, password="secret")
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+        assign_perm("change_product", self.base_user, self.product1)
+
+        package = make_package(self.dataspace)
+        make_product_package(self.product1, package=package)
+        vulnerability = make_vulnerability(self.dataspace, affecting=package, risk_score=9.0)
+        ruleset = make_triage_ruleset(
+            self.dataspace,
+            recommended_action=TriageAction.UPGRADE,
+            rules_config={"risk_score": {"is_active": True, "min_risk_score": 8.0}},
+        )
+
+        data = {"ruleset": str(ruleset.uuid), "assigned": True}
+        response = self.client.post(url, data=data, content_type="application/json")
+
+        self.assertEqual(status.HTTP_200_OK, response.status_code)
+        self.assertTrue(
+            ProductTriageRuleset.objects.filter(product=self.product1, ruleset=ruleset).exists()
+        )
+        triage_record = self.product1.triage_records.get()
+        self.assertEqual(vulnerability, triage_record.vulnerability)
+
+    def test_api_product_endpoint_manage_triage_rulesets_post_unassigns(self):
+        url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid])
+        self.client.login(username=self.base_user.username, password="secret")
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+        assign_perm("change_product", self.base_user, self.product1)
+
+        ruleset = make_triage_ruleset(self.dataspace)
+        make_product_triage_ruleset(self.product1, ruleset=ruleset)
+
+        data = {"ruleset": str(ruleset.uuid), "assigned": False}
+        response = self.client.post(url, data=data, content_type="application/json")
+
+        self.assertEqual(status.HTTP_200_OK, response.status_code)
+        self.assertFalse(
+            ProductTriageRuleset.objects.filter(product=self.product1, ruleset=ruleset).exists()
+        )
+
+    def test_api_product_endpoint_manage_triage_rulesets_post_requires_both_fields(self):
+        url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid])
+        self.client.login(username=self.base_user.username, password="secret")
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+        assign_perm("change_product", self.base_user, self.product1)
+
+        response = self.client.post(url, data={}, content_type="application/json")
+
+        self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+
+    def test_api_product_endpoint_manage_triage_rulesets_post_rejects_non_dict_body(self):
+        url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid])
+        self.client.login(username=self.base_user.username, password="secret")
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+        assign_perm("change_product", self.base_user, self.product1)
+
+        response = self.client.post(url, data=[], content_type="application/json")
+
+        self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+
+    def test_api_product_endpoint_manage_triage_rulesets_post_rejects_disabled_ruleset(self):
+        url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid])
+        self.client.login(username=self.base_user.username, password="secret")
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+        assign_perm("change_product", self.base_user, self.product1)
+
+        ruleset = make_triage_ruleset(self.dataspace, enabled=False)
+        data = {"ruleset": str(ruleset.uuid), "assigned": True}
+        response = self.client.post(url, data=data, content_type="application/json")
+
+        self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+
 
 class ProductRelatedAPITestCase(TestCase):
     def setUp(self):