[cravex2-reachability] Integrate triage rules in CRAVEX API #365

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 = [

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,6 +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": {
@@ -363,6 +370,37 @@ 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 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 AssignTriageRulesetSerializer(serializers.Serializer):
+ ruleset = serializers.UUIDField()
+ assigned = serializers.BooleanField()
+
+
 class ProductViewSet(
  ObjectPermissionsMixin,
  SendAboutFilesMixin,
@@ -441,6 +479,69 @@ 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=["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
+ 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":
+ 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=serializer.validated_data["ruleset"],
+ )
+ assigned = serializer.validated_data["assigned"]
+
+ 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
+ )
+ reevaluate_product_rulesets(product)
+ 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

@@ -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,14 @@
 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 ProductTriageRuleset
+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 +765,163 @@ 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)
+
+ 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_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")
+ 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):

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",
  )
  )