From 1c10771840e413cce25c407d46f80d013d8a1146 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 19 Aug 2026 09:27:24 +0200 Subject: [PATCH 1/5] add triage_records action on product REST API Signed-off-by: tdruez --- product_portfolio/api.py | 29 +++++++++++++++++++ product_portfolio/tests/test_api.py | 43 +++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/product_portfolio/api.py b/product_portfolio/api.py index 5c74d9c0..3b1dcb43 100644 --- a/product_portfolio/api.py +++ b/product_portfolio/api.py @@ -51,6 +51,7 @@ from product_portfolio.models import ProductPolicyViolation from product_portfolio.models import ScanCodeProject from vulnerabilities.api import VulnerabilityAnalysisSerializer +from vulnerabilities.triage.models import TriageRecord base_extra_kwargs = { "licenses": { @@ -363,6 +364,24 @@ class Meta: ) +class TriageRecordSerializer(serializers.ModelSerializer): + advisory_id = serializers.ReadOnlyField(source="vulnerability.advisory_id") + ruleset = serializers.ReadOnlyField(source="ruleset.name") + request = serializers.StringRelatedField() + + class Meta: + model = TriageRecord + fields = ( + "advisory_id", + "ruleset", + "recommended_action", + "matched_rules", + "request", + "detected_date", + "last_checked", + ) + + class ProductViewSet( ObjectPermissionsMixin, SendAboutFilesMixin, @@ -441,6 +460,16 @@ def policy_violations(self, request, uuid): serializer = ProductPolicyViolationSerializer(violations, many=True) return Response(serializer.data) + @action(detail=True, url_path="triage_records") + def triage_records(self, request, uuid): + """List active triage recommendations for this product, one per vulnerability.""" + product = self.get_object() + records = product.triage_records.highest_precedence().select_related( + "vulnerability", "ruleset", "request" + ) + serializer = TriageRecordSerializer(records, many=True) + return Response(serializer.data) + @action(detail=True, methods=["post"], serializer_class=LoadSBOMsFormSerializer) def load_sboms(self, request, *args, **kwargs): """ diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py index 23803db8..bcf3a824 100644 --- a/product_portfolio/tests/test_api.py +++ b/product_portfolio/tests/test_api.py @@ -24,6 +24,7 @@ from component_catalog.models import Component from component_catalog.models import ComponentKeyword from component_catalog.models import Package +from component_catalog.tests import make_package from dje.models import Dataspace from dje.models import History from dje.tests import MaxQueryMixin @@ -47,8 +48,13 @@ from product_portfolio.models import ProductRelationStatus from product_portfolio.models import ProductStatus from product_portfolio.models import ScanCodeProject +from product_portfolio.tests import make_product_package 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 TriageAction +from vulnerabilities.triage.tests import make_product_triage_ruleset +from vulnerabilities.triage.tests import make_triage_ruleset class ProductAPITestCase(MaxQueryMixin, TestCase): @@ -758,6 +764,43 @@ def test_api_product_endpoint_policy_violations_action(self): self.assertEqual(status.HTTP_200_OK, response.status_code) self.assertEqual([], response.data) + def test_api_product_endpoint_triage_records_action(self): + url = reverse("api_v2:product-triage-records", 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) + + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual([], response.data) + + 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}}, + ) + make_product_triage_ruleset(self.product1, ruleset=ruleset) + evaluate_ruleset(ruleset, self.product1) + + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual(1, len(response.data)) + entry = response.data[0] + self.assertEqual(vulnerability.advisory_id, entry["advisory_id"]) + self.assertEqual(ruleset.name, entry["ruleset"]) + self.assertEqual(TriageAction.UPGRADE, entry["recommended_action"]) + self.assertEqual(["risk_score"], entry["matched_rules"]) + self.assertIsNone(entry["request"]) + self.assertIn("detected_date", entry) + self.assertIn("last_checked", entry) + class ProductRelatedAPITestCase(TestCase): def setUp(self): From 49ff7cf8453792888bd3ff30c76773540560b5c3 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 19 Aug 2026 11:24:55 +0200 Subject: [PATCH 2/5] enhance VulnerabilityAnalysisSerializer with new fields Signed-off-by: tdruez --- vulnerabilities/api.py | 13 +++++++++ vulnerabilities/tests/test_api.py | 46 +++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/vulnerabilities/api.py b/vulnerabilities/api.py index b8be931b..e85335e7 100644 --- a/vulnerabilities/api.py +++ b/vulnerabilities/api.py @@ -9,6 +9,7 @@ from django.db.models import Prefetch +import django_filters from rest_framework import serializers from rest_framework import viewsets @@ -134,6 +135,9 @@ def get_queryset(self): class VulnerabilityAnalysisSerializer(DataspacedSerializer, serializers.ModelSerializer): advisory_uid = serializers.ReadOnlyField(source="vulnerability.advisory_uid") + applied_by_preset = serializers.ReadOnlyField(source="applied_by_preset.name", allow_null=True) + created_by = serializers.StringRelatedField() + last_modified_by = serializers.StringRelatedField() class Meta: model = VulnerabilityAnalysis @@ -148,6 +152,9 @@ class Meta: "responses", "detail", "is_reachable", + "applied_by_preset", + "created_by", + "last_modified_by", "first_issued", "last_updated", ) @@ -170,6 +177,9 @@ class Meta: class VulnerabilityAnalysisFilterSet(DataspacedAPIFilterSet): uuid = MultipleUUIDFilter() last_updated = LastModifiedDateFilter() + applied_by_preset__isnull = django_filters.BooleanFilter( + field_name="applied_by_preset", lookup_expr="isnull" + ) class Meta: model = VulnerabilityAnalysis @@ -203,5 +213,8 @@ def get_queryset(self): .select_related( "vulnerability", "product_package", + "applied_by_preset", + "created_by", + "last_modified_by", ) ) diff --git a/vulnerabilities/tests/test_api.py b/vulnerabilities/tests/test_api.py index ccae963b..5f0f6574 100644 --- a/vulnerabilities/tests/test_api.py +++ b/vulnerabilities/tests/test_api.py @@ -20,6 +20,9 @@ from vulnerabilities.models import VulnerabilityAnalysis from vulnerabilities.tests import make_vulnerability from vulnerabilities.tests import make_vulnerability_analysis +from vulnerabilities.triage.engine import apply_preset_for_vulnerabilities +from vulnerabilities.triage.models import AnalysisPreset +from vulnerabilities.triage.tests import make_analysis_preset class VulnerabilitiesAPITestCase(MaxQueryMixin, TestCase): @@ -154,6 +157,49 @@ def test_api_vulnerability_analysis_detail_endpoint(self): self.assertEqual(str(analysis1.uuid), response.data["uuid"]) self.assertTrue(response.data["is_reachable"]) + def test_api_vulnerability_analysis_detail_endpoint_applied_by_preset_and_authors(self): + self.client.login(username="super_user", password="secret") + + human_analysis = make_vulnerability_analysis( + self.product_package1, + self.vulnerability1, + created_by=self.super_user, + last_modified_by=self.super_user, + ) + detail_url = reverse("api_v2:vulnerabilityanalysis-detail", args=[human_analysis.uuid]) + response = self.client.get(detail_url) + self.assertIsNone(response.data["applied_by_preset"]) + self.assertEqual(self.super_user.username, response.data["created_by"]) + self.assertEqual(self.super_user.username, response.data["last_modified_by"]) + + preset = make_analysis_preset(self.dataspace, state=AnalysisPreset.State.NOT_AFFECTED) + apply_preset_for_vulnerabilities(preset, self.product1, [self.vulnerability2.pk]) + auto_analysis = VulnerabilityAnalysis.objects.get(vulnerability=self.vulnerability2) + detail_url = reverse("api_v2:vulnerabilityanalysis-detail", args=[auto_analysis.uuid]) + + response = self.client.get(detail_url) + self.assertEqual(preset.name, response.data["applied_by_preset"]) + self.assertIsNone(response.data["created_by"]) + self.assertIsNone(response.data["last_modified_by"]) + + def test_api_vulnerability_analysis_list_endpoint_filters_applied_by_preset(self): + self.client.login(username="super_user", password="secret") + make_vulnerability_analysis(self.product_package1, self.vulnerability1) + preset = make_analysis_preset(self.dataspace, state=AnalysisPreset.State.NOT_AFFECTED) + apply_preset_for_vulnerabilities(preset, self.product1, [self.vulnerability2.pk]) + + data = {"applied_by_preset__isnull": "true"} + response = self.client.get(self.analysis_list_url, data) + self.assertEqual(1, response.data["count"]) + self.assertContains(response, self.vulnerability1.advisory_id) + self.assertNotContains(response, self.vulnerability2.advisory_id) + + data = {"applied_by_preset__isnull": "false"} + response = self.client.get(self.analysis_list_url, data) + self.assertEqual(1, response.data["count"]) + self.assertNotContains(response, self.vulnerability1.advisory_id) + self.assertContains(response, self.vulnerability2.advisory_id) + def test_api_vulnerability_analysis_endpoint_create(self): self.client.login(username="super_user", password="secret") response = self.client.post(self.analysis_list_url) From fa83063ac5791fff4038a26635f4cc5d23cc9f9d Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 19 Aug 2026 11:51:22 +0200 Subject: [PATCH 3/5] add API endpoints for AnalysisPreset and TriageRuleset Signed-off-by: tdruez --- dejacode/urls.py | 4 + vulnerabilities/triage/api.py | 137 +++++++++++++++++++ vulnerabilities/triage/tests/test_api.py | 159 +++++++++++++++++++++++ 3 files changed, 300 insertions(+) create mode 100644 vulnerabilities/triage/api.py create mode 100644 vulnerabilities/triage/tests/test_api.py diff --git a/dejacode/urls.py b/dejacode/urls.py index 5b381868..eb656555 100644 --- a/dejacode/urls.py +++ b/dejacode/urls.py @@ -55,6 +55,8 @@ from reporting.api import ReportViewSet from vulnerabilities.api import VulnerabilityAnalysisViewSet from vulnerabilities.api import VulnerabilityViewSet +from vulnerabilities.triage.api import AnalysisPresetViewSet +from vulnerabilities.triage.api import TriageRulesetViewSet from workflow.api import RequestTemplateViewSet from workflow.api import RequestViewSet @@ -83,6 +85,8 @@ api_router.register("usage_policies", UsagePolicyViewSet) api_router.register("vulnerabilities", VulnerabilityViewSet) api_router.register("vulnerability_analyses", VulnerabilityAnalysisViewSet) +api_router.register("triage_rulesets", TriageRulesetViewSet) +api_router.register("analysis_presets", AnalysisPresetViewSet) urlpatterns = [ diff --git a/vulnerabilities/triage/api.py b/vulnerabilities/triage/api.py new file mode 100644 index 00000000..f3a833ac --- /dev/null +++ b/vulnerabilities/triage/api.py @@ -0,0 +1,137 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# DejaCode is a trademark of nexB Inc. +# SPDX-License-Identifier: AGPL-3.0-only +# See https://github.com/aboutcode-org/dejacode for support or download. +# See https://aboutcode.org for more information about AboutCode FOSS projects. +# + +from rest_framework import mixins +from rest_framework import serializers + +from dje.api import CreateRetrieveUpdateListViewSet +from dje.api import DataspacedSerializer +from dje.api import ExtraPermissionsViewSetMixin +from dje.api_custom import TabPermission +from vulnerabilities.triage.models import AnalysisPreset +from vulnerabilities.triage.models import TriageRuleset + + +class AnalysisPresetSerializer(DataspacedSerializer): + class Meta: + model = AnalysisPreset + fields = ( + "api_url", + "uuid", + "name", + "description", + "state", + "justification", + "responses", + "detail", + "is_reachable", + ) + extra_kwargs = { + "api_url": { + "view_name": "api_v2:analysispreset-detail", + "lookup_field": "uuid", + }, + } + + def validate(self, data): + content_fields = ("state", "justification", "responses", "detail") + + def get_value(field_name): + if field_name in data: + return data[field_name] + if self.instance is not None: + return getattr(self.instance, field_name) + return None + + if not any(get_value(field_name) for field_name in content_fields): + raise serializers.ValidationError( + "At least one of state, justification, responses or detail must be provided." + ) + return data + + +class AnalysisPresetViewSet( + ExtraPermissionsViewSetMixin, + mixins.DestroyModelMixin, + CreateRetrieveUpdateListViewSet, +): + queryset = AnalysisPreset.objects.all() + serializer_class = AnalysisPresetSerializer + lookup_field = "uuid" + extra_permissions = (TabPermission,) + search_fields = ( + "name", + "description", + ) + ordering_fields = ("name",) + allow_reference_access = True + + +class TriageRulesetSerializer(DataspacedSerializer): + class Meta: + model = TriageRuleset + fields = ( + "api_url", + "uuid", + "name", + "description", + "recommended_action", + "precedence", + "enabled", + "rules_config", + "analysis_preset", + "request_template", + ) + extra_kwargs = { + "api_url": { + "view_name": "api_v2:triageruleset-detail", + "lookup_field": "uuid", + }, + "analysis_preset": { + "view_name": "api_v2:analysispreset-detail", + "lookup_field": "uuid", + "required": False, + "allow_null": True, + }, + "request_template": { + "view_name": "api_v2:requesttemplate-detail", + "lookup_field": "uuid", + "required": False, + "allow_null": True, + }, + } + + def validate_request_template(self, value): + if value and not value.created_by_id: + raise serializers.ValidationError( + "This request template has no creator and cannot be used to open requests." + ) + return value + + +class TriageRulesetViewSet( + ExtraPermissionsViewSetMixin, + mixins.DestroyModelMixin, + CreateRetrieveUpdateListViewSet, +): + queryset = TriageRuleset.objects.all() + serializer_class = TriageRulesetSerializer + lookup_field = "uuid" + extra_permissions = (TabPermission,) + search_fields = ( + "name", + "description", + ) + ordering_fields = ( + "name", + "precedence", + ) + allow_reference_access = True + + def get_queryset(self): + return super().get_queryset().select_related("analysis_preset", "request_template") diff --git a/vulnerabilities/triage/tests/test_api.py b/vulnerabilities/triage/tests/test_api.py new file mode 100644 index 00000000..2a0e0766 --- /dev/null +++ b/vulnerabilities/triage/tests/test_api.py @@ -0,0 +1,159 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# DejaCode is a trademark of nexB Inc. +# SPDX-License-Identifier: AGPL-3.0-only +# See https://github.com/aboutcode-org/dejacode for support or download. +# See https://aboutcode.org for more information about AboutCode FOSS projects. +# + +from django.contrib.contenttypes.models import ContentType +from django.test import TestCase +from django.urls import reverse + +from rest_framework import status + +from dje.models import Dataspace +from dje.tests import create_superuser +from product_portfolio.models import Product +from vulnerabilities.triage.models import AnalysisPreset +from vulnerabilities.triage.models import TriageAction +from vulnerabilities.triage.models import TriageRuleset +from vulnerabilities.triage.tests import make_analysis_preset +from vulnerabilities.triage.tests import make_triage_ruleset +from workflow.models import RequestTemplate + + +class AnalysisPresetAPITestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.alternate = Dataspace.objects.create(name="Alternate") + self.super_user = create_superuser("super_user", self.dataspace) + + self.list_url = reverse("api_v2:analysispreset-list") + self.preset = make_analysis_preset( + self.dataspace, name="Preset1", state=AnalysisPreset.State.NOT_AFFECTED + ) + self.detail_url = reverse("api_v2:analysispreset-detail", args=[self.preset.uuid]) + make_analysis_preset(self.alternate, name="OtherPreset") + + def test_api_analysispreset_list_endpoint_user_available_scope(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.list_url) + self.assertEqual(1, response.data["count"]) + self.assertContains(response, self.preset.name) + + def test_api_analysispreset_detail_endpoint(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.detail_url) + self.assertEqual(self.preset.name, response.data["name"]) + self.assertEqual(AnalysisPreset.State.NOT_AFFECTED, response.data["state"]) + + def test_api_analysispreset_endpoint_create(self): + self.client.login(username="super_user", password="secret") + data = {"name": "New Preset", "detail": "Some detail"} + response = self.client.post(self.list_url, data) + self.assertEqual(status.HTTP_201_CREATED, response.status_code) + preset = AnalysisPreset.objects.get(name="New Preset") + self.assertEqual("Some detail", preset.detail) + + def test_api_analysispreset_endpoint_create_rejects_no_content_field(self): + self.client.login(username="super_user", password="secret") + data = {"name": "No content", "is_reachable": True} + response = self.client.post(self.list_url, data) + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + msg = "At least one of state, justification, responses or detail must be provided." + self.assertIn(msg, response.data["non_field_errors"]) + + def test_api_analysispreset_endpoint_update(self): + self.client.login(username="super_user", password="secret") + data = {"detail": "Updated detail"} + response = self.client.patch(self.detail_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.preset.refresh_from_db() + self.assertEqual("Updated detail", self.preset.detail) + + def test_api_analysispreset_endpoint_delete(self): + self.client.login(username="super_user", password="secret") + response = self.client.delete(self.detail_url) + self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code) + self.assertFalse(AnalysisPreset.objects.filter(pk=self.preset.pk).exists()) + + +class TriageRulesetAPITestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.alternate = Dataspace.objects.create(name="Alternate") + self.super_user = create_superuser("super_user", self.dataspace) + + self.list_url = reverse("api_v2:triageruleset-list") + self.ruleset = make_triage_ruleset( + self.dataspace, + name="Ruleset1", + recommended_action=TriageAction.UPGRADE, + precedence=100, + ) + self.detail_url = reverse("api_v2:triageruleset-detail", args=[self.ruleset.uuid]) + make_triage_ruleset(self.alternate, name="OtherRuleset") + + def test_api_triageruleset_list_endpoint_user_available_scope(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.list_url) + self.assertEqual(1, response.data["count"]) + self.assertContains(response, self.ruleset.name) + + def test_api_triageruleset_detail_endpoint(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.detail_url) + self.assertEqual(self.ruleset.name, response.data["name"]) + self.assertEqual(TriageAction.UPGRADE, response.data["recommended_action"]) + self.assertEqual(100, response.data["precedence"]) + + def test_api_triageruleset_endpoint_create(self): + self.client.login(username="super_user", password="secret") + data = { + "name": "New Ruleset", + "precedence": 200, + "recommended_action": TriageAction.NOTIFY, + "rules_config": {"risk_score": {"is_active": True, "min_risk_score": 8.0}}, + } + response = self.client.post(self.list_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_201_CREATED, response.status_code) + ruleset = TriageRuleset.objects.get(name="New Ruleset") + self.assertEqual( + {"is_active": True, "min_risk_score": 8.0}, ruleset.rules_config["risk_score"] + ) + + def test_api_triageruleset_endpoint_create_rejects_request_template_with_no_creator(self): + self.client.login(username="super_user", password="secret") + request_template = RequestTemplate.objects.create( + name="Broken Template", + description="Header", + dataspace=self.dataspace, + content_type=ContentType.objects.get_for_model(Product), + ) + request_template_url = reverse( + "api_v2:requesttemplate-detail", args=[request_template.uuid] + ) + data = { + "name": "New Ruleset", + "precedence": 200, + "request_template": request_template_url, + } + response = self.client.post(self.list_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + msg = "This request template has no creator and cannot be used to open requests." + self.assertIn(msg, response.data["request_template"]) + + def test_api_triageruleset_endpoint_update(self): + self.client.login(username="super_user", password="secret") + data = {"enabled": False} + response = self.client.patch(self.detail_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.ruleset.refresh_from_db() + self.assertFalse(self.ruleset.enabled) + + def test_api_triageruleset_endpoint_delete(self): + self.client.login(username="super_user", password="secret") + response = self.client.delete(self.detail_url) + self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code) + self.assertFalse(TriageRuleset.objects.filter(pk=self.ruleset.pk).exists()) From c5087aa342b411fadc4dad34d1f6c09b30794699 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 19 Aug 2026 12:21:57 +0200 Subject: [PATCH 4/5] add a manage_triage_rulesets action on Product API endpoint Signed-off-by: tdruez --- product_portfolio/api.py | 67 ++++++++++++++++++ product_portfolio/tests/test_api.py | 102 ++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+) diff --git a/product_portfolio/api.py b/product_portfolio/api.py index 3b1dcb43..c8cc7607 100644 --- a/product_portfolio/api.py +++ b/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": "", "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): """ diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py index bcf3a824..609bb8ed 100644 --- a/product_portfolio/tests/test_api.py +++ b/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): From d86f4715599484be5bbe38faa13d448cef57faa6 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 19 Aug 2026 14:28:30 +0200 Subject: [PATCH 5/5] refine implementation Signed-off-by: tdruez --- product_portfolio/api.py | 35 ++++++++++++++---------- product_portfolio/tests/test_api.py | 19 +++++++++++++ vulnerabilities/triage/api.py | 26 ++++++++++-------- vulnerabilities/triage/tests/test_api.py | 13 +++++++++ 4 files changed, 66 insertions(+), 27 deletions(-) diff --git a/product_portfolio/api.py b/product_portfolio/api.py index c8cc7607..571012dc 100644 --- a/product_portfolio/api.py +++ b/product_portfolio/api.py @@ -396,6 +396,11 @@ class TriageRulesetAssignmentSerializer(serializers.Serializer): assigned = serializers.BooleanField(read_only=True) +class AssignTriageRulesetSerializer(serializers.Serializer): + ruleset = serializers.UUIDField() + assigned = serializers.BooleanField() + + class ProductViewSet( ObjectPermissionsMixin, SendAboutFilesMixin, @@ -484,7 +489,12 @@ 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") + @action( + detail=True, + methods=["get", "post"], + url_path="manage_triage_rulesets", + serializer_class=AssignTriageRulesetSerializer, + ) def manage_triage_rulesets(self, request, uuid): """ GET: list every enabled triage ruleset in this product's dataspace, each flagged @@ -496,22 +506,16 @@ def manage_triage_rulesets(self, request, uuid): 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, - ) + serializer = AssignTriageRulesetSerializer(data=request.data) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + ruleset = get_object_or_404( TriageRuleset.objects.scope(product.dataspace).filter(enabled=True), - uuid=ruleset_uuid, + uuid=serializer.validated_data["ruleset"], ) + assigned = serializer.validated_data["assigned"] + with transaction.atomic(): assignment = ProductTriageRuleset.objects.filter( product=product, ruleset=ruleset @@ -520,10 +524,11 @@ def manage_triage_rulesets(self, request, uuid): ProductTriageRuleset.objects.create( product=product, ruleset=ruleset, dataspace=product.dataspace ) + reevaluate_product_rulesets(product) elif not assigned and assignment: assignment.delete() delete_triage_records_for_assignment(ruleset=ruleset, product=product) - reevaluate_product_rulesets(product) + reevaluate_product_rulesets(product) return Response(status=status.HTTP_200_OK) assigned_ruleset_ids = set( diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py index 609bb8ed..e68bb5fd 100644 --- a/product_portfolio/tests/test_api.py +++ b/product_portfolio/tests/test_api.py @@ -868,6 +868,25 @@ def test_api_product_endpoint_manage_triage_rulesets_post_unassigns(self): ProductTriageRuleset.objects.filter(product=self.product1, ruleset=ruleset).exists() ) + def test_api_product_endpoint_manage_triage_rulesets_post_unassigns_form_encoded(self): + # Regression: a form-encoded "false" string must not be treated as truthy. + 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) + + 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") diff --git a/vulnerabilities/triage/api.py b/vulnerabilities/triage/api.py index f3a833ac..2e90bbf4 100644 --- a/vulnerabilities/triage/api.py +++ b/vulnerabilities/triage/api.py @@ -10,6 +10,7 @@ from rest_framework import serializers from dje.api import CreateRetrieveUpdateListViewSet +from dje.api import DataspacedHyperlinkedRelatedField from dje.api import DataspacedSerializer from dje.api import ExtraPermissionsViewSetMixin from dje.api_custom import TabPermission @@ -73,6 +74,19 @@ class AnalysisPresetViewSet( class TriageRulesetSerializer(DataspacedSerializer): + analysis_preset = DataspacedHyperlinkedRelatedField( + view_name="api_v2:analysispreset-detail", + lookup_field="uuid", + required=False, + allow_null=True, + ) + request_template = DataspacedHyperlinkedRelatedField( + view_name="api_v2:requesttemplate-detail", + lookup_field="uuid", + required=False, + allow_null=True, + ) + class Meta: model = TriageRuleset fields = ( @@ -92,18 +106,6 @@ class Meta: "view_name": "api_v2:triageruleset-detail", "lookup_field": "uuid", }, - "analysis_preset": { - "view_name": "api_v2:analysispreset-detail", - "lookup_field": "uuid", - "required": False, - "allow_null": True, - }, - "request_template": { - "view_name": "api_v2:requesttemplate-detail", - "lookup_field": "uuid", - "required": False, - "allow_null": True, - }, } def validate_request_template(self, value): diff --git a/vulnerabilities/triage/tests/test_api.py b/vulnerabilities/triage/tests/test_api.py index 2a0e0766..8c89c309 100644 --- a/vulnerabilities/triage/tests/test_api.py +++ b/vulnerabilities/triage/tests/test_api.py @@ -144,6 +144,19 @@ def test_api_triageruleset_endpoint_create_rejects_request_template_with_no_crea msg = "This request template has no creator and cannot be used to open requests." self.assertIn(msg, response.data["request_template"]) + def test_api_triageruleset_endpoint_create_rejects_cross_dataspace_analysis_preset(self): + self.client.login(username="super_user", password="secret") + other_preset = make_analysis_preset(self.alternate, name="OtherPreset") + other_preset_url = reverse("api_v2:analysispreset-detail", args=[other_preset.uuid]) + data = { + "name": "New Ruleset", + "precedence": 200, + "analysis_preset": other_preset_url, + } + response = self.client.post(self.list_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("analysis_preset", response.data) + def test_api_triageruleset_endpoint_update(self): self.client.login(username="super_user", password="secret") data = {"enabled": False}