[cravex2-reachability] Process extended reachability

product_portfolio/api.py

@@ -208,6 +208,11 @@ class ProductFilterSet(DataspacedAPIFilterSet):
         field_name="packages__affected_by_vulnerabilities__advisory_id",
         label="Affected by (advisory_id)",
     )
+    has_reachable_vulnerability = django_filters.BooleanFilter(
+        field_name="vulnerability_analyses__is_reachable",
+        label="Has reachable vulnerability",
+        distinct=True,
+    )
 
     class Meta:
         model = Product
@@ -226,6 +231,7 @@ class Meta:
             "last_modified_date",
             "is_vulnerable",
             "affected_by",
+            "has_reachable_vulnerability",
         )
 
 
@@ -885,6 +891,11 @@ class ProductPackageFilterSet(DataspacedAPIFilterSet):
         field_name="package__affected_by_vulnerabilities__advisory_id",
         label="Affected by (advisory_id)",
     )
+    has_reachable_vulnerability = django_filters.BooleanFilter(
+        field_name="vulnerability_analyses__is_reachable",
+        label="Has reachable vulnerability",
+        distinct=True,
+    )
 
     class Meta:
         model = ProductPackage
@@ -898,6 +909,7 @@ class Meta:
             "last_modified_date",
             "is_vulnerable",
             "affected_by",
+            "has_reachable_vulnerability",
         )
 
 

product_portfolio/filters.py

@@ -422,6 +422,7 @@ class ProductPackageFilterSet(BaseProductRelationFilterSet):
             ("unknown", _("Reachability not known")),
         ),
     )
+
     triage_action = django_filters.ChoiceFilter(
         label=_("Triage action"),
         choices=TriageAction.choices,
@@ -462,6 +463,8 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.filters["vulnerability_analyses__state"].extra["null_label"] = "(No values)"
         self.filters["vulnerability_analyses__justification"].extra["null_label"] = "(No values)"
+        is_reachable = self.filters["is_reachable"]
+        is_reachable.extra["widget"].link_content = '<i class="fa-solid fa-circle-radiation"></i>'
 
 
 class ComponentCompletenessListFilter(admin.SimpleListFilter):

product_portfolio/importers.py

@@ -54,6 +54,7 @@
 from product_portfolio.models import ProductPackage
 from product_portfolio.models import ProductRelationStatus
 from product_portfolio.models import ScanCodeProject
+from vulnerabilities.triage.signals import reevaluate_on_analysis_change
 from vulnerabilities.triage.signals import reevaluate_on_product_package_change
 from vulnerabilities.triage.tasks import reevaluate_product_triage_rulesets_task
 
@@ -71,24 +72,35 @@ def log_elapsed(label):
 @contextmanager
 def paused_product_package_reevaluation():
     """
-    Pause the policy and triage re-evaluation signals triggered by ProductPackage changes,
-    for the duration of a bulk import. Call `reevaluate_products()` once the import completes
-    to evaluate each affected product exactly once, instead of once per imported row.
+    Pause re-evaluation signals for the duration of a bulk import.
+
+    Covers ProductPackage add/remove and VulnerabilityAnalysis create/update signals so that
+    each triggers at most once per affected product. Call `reevaluate_products()` after the
+    import to run the evaluation exactly once instead of once per imported row.
     """
-    receivers = [
+    from vulnerabilities.models import VulnerabilityAnalysis
+
+    productpackage_receivers = [
         evaluate_product_rules_on_productpackage_change,
         reevaluate_on_product_package_change,
     ]
-    for receiver in receivers:
+    for receiver in productpackage_receivers:
         post_save.disconnect(receiver, sender=ProductPackage)
         post_delete.disconnect(receiver, sender=ProductPackage)
+
+    post_save.disconnect(reevaluate_on_analysis_change, sender=VulnerabilityAnalysis)
+    post_delete.disconnect(reevaluate_on_analysis_change, sender=VulnerabilityAnalysis)
+
     try:
         yield
     finally:
-        for receiver in receivers:
+        for receiver in productpackage_receivers:
             post_save.connect(receiver, sender=ProductPackage)
             post_delete.connect(receiver, sender=ProductPackage)
 
+        post_save.connect(reevaluate_on_analysis_change, sender=VulnerabilityAnalysis)
+        post_delete.connect(reevaluate_on_analysis_change, sender=VulnerabilityAnalysis)
+
 
 def reevaluate_products(products):
     """Queue the policy and triage re-evaluation once for each of the given products."""

product_portfolio/tests/test_api.py

@@ -1553,6 +1553,57 @@ def test_api_product_endpoint_vulnerabilities_features(self):
         self.assertNotContains(response, self.product1_detail_url)
         self.assertNotContains(response, self.product2_detail_url)
 
+    def test_api_productpackage_has_reachable_vulnerability_filter(self):
+        self.client.login(username="super_user", password="secret")
+        vulnerability = make_vulnerability(self.dataspace, affecting=self.package1)
+        make_vulnerability_analysis(self.pp1, vulnerability, is_reachable=True)
+
+        data = {"has_reachable_vulnerability": "true"}
+        response = self.client.get(self.productpackage_list_url, data)
+        self.assertEqual(1, response.data["count"])
+        self.assertContains(response, self.pp1_detail_url)
+
+        data = {"has_reachable_vulnerability": "false"}
+        response = self.client.get(self.productpackage_list_url, data)
+        self.assertEqual(0, response.data["count"])
+
+    def test_api_product_has_reachable_vulnerability_filter(self):
+        self.client.login(username="super_user", password="secret")
+        vulnerability = make_vulnerability(self.dataspace, affecting=self.package1)
+        make_vulnerability_analysis(self.pp1, vulnerability, is_reachable=True)
+
+        data = {"has_reachable_vulnerability": "true"}
+        response = self.client.get(self.product_list_url, data)
+        self.assertEqual(1, response.data["count"])
+        self.assertContains(response, self.product1_detail_url)
+        self.assertNotContains(response, self.product2_detail_url)
+
+        data = {"has_reachable_vulnerability": "false"}
+        response = self.client.get(self.product_list_url, data)
+        self.assertEqual(0, response.data["count"])
+
+    def test_api_product_has_reachable_vulnerability_filter_no_duplicates(self):
+        self.client.login(username="super_user", password="secret")
+        vulnerability1 = make_vulnerability(self.dataspace, affecting=self.package1)
+        vulnerability2 = make_vulnerability(self.dataspace, affecting=self.package1)
+        make_vulnerability_analysis(self.pp1, vulnerability1, is_reachable=True)
+        make_vulnerability_analysis(self.pp1, vulnerability2, is_reachable=True)
+
+        data = {"has_reachable_vulnerability": "true"}
+        response = self.client.get(self.product_list_url, data)
+        self.assertEqual(1, response.data["count"])
+
+    def test_api_productpackage_has_reachable_vulnerability_filter_no_duplicates(self):
+        self.client.login(username="super_user", password="secret")
+        vulnerability1 = make_vulnerability(self.dataspace, affecting=self.package1)
+        vulnerability2 = make_vulnerability(self.dataspace, affecting=self.package1)
+        make_vulnerability_analysis(self.pp1, vulnerability1, is_reachable=True)
+        make_vulnerability_analysis(self.pp1, vulnerability2, is_reachable=True)
+
+        data = {"has_reachable_vulnerability": "true"}
+        response = self.client.get(self.productpackage_list_url, data)
+        self.assertEqual(1, response.data["count"])
+
     def test_api_codebaseresource_list_endpoint_results(self):
         self.client.login(username="super_user", password="secret")
         response = self.client.get(self.codebase_resource_list_url)

product_portfolio/tests/test_views.py

@@ -322,6 +322,14 @@ def test_product_portfolio_tab_vulnerability_view_filters(self):
             response, "?vulnerabilities-vulnerability_analyses__state=#vulnerabilities"
         )
 
+    def test_product_portfolio_tab_vulnerability_view_is_reachable_filter_in_analysis_header(self):
+        self.client.login(username="nexb_user", password="secret")
+        url = self.product1.get_url("tab_vulnerabilities")
+        response = self.client.get(url)
+        self.assertContains(response, "fa-circle-radiation")
+        self.assertContains(response, "?vulnerabilities-is_reachable=yes#vulnerabilities")
+        self.assertContains(response, "?vulnerabilities-is_reachable=no#vulnerabilities")
+
     def test_product_portfolio_tab_vulnerability_view_packages_row_rendering(self):
         self.client.login(username="nexb_user", password="secret")
         # Each have a unique vulnerability, and p1 p2 are sharing a common one.

product_portfolio/views.py

@@ -1271,6 +1271,17 @@ class ProductTabVulnerabilitiesView(
         ),
     )
 
+    def get_table_headers(self):
+        """Inject the is_reachable filter widget into the Analysis column header."""
+        headers = super().get_table_headers()
+        is_reachable_widget = f'<span class="me-2">{self.filterset.form["is_reachable"]}</span>'
+        return [
+            header._replace(filter=mark_safe(is_reachable_widget + str(header.filter)))
+            if header.field_name == "vulnerability_analyses__state"
+            else header
+            for header in headers
+        ]
+
     def attach_vulnerability_analyses(self, page_obj):
         """Set the matching VulnerabilityAnalysis instance on each prefetched vulnerability."""
         response_labels = dict(VulnerabilityAnalysis.Response.choices)

vulnerabilities/models.py

@@ -19,6 +19,7 @@
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
+from cyclonedx import model as cdx_model
 from cyclonedx.model import vulnerability as cdx_vulnerability
 
 from dje.fields import JSONListField
@@ -266,7 +267,16 @@ def as_cyclonedx(self, affected_instances, analysis=None):
             for instance in affected_instances
         ]
 
-        analysis = analysis.as_cyclonedx() if analysis else None
+        properties = None
+        if analysis is not None and analysis.is_reachable is not None:
+            properties = [
+                cdx_model.Property(
+                    name="aboutcode:is_reachable",
+                    value="true" if analysis.is_reachable else "false",
+                )
+            ]
+
+        cdx_analysis = analysis.as_cyclonedx() if analysis else None
 
         source = cdx_vulnerability.VulnerabilitySource(
             name="VulnerableCode",
@@ -278,7 +288,8 @@ def as_cyclonedx(self, affected_instances, analysis=None):
             source=source,
             description=self.summary,
             affects=affects,
-            analysis=analysis,
+            analysis=cdx_analysis,
+            properties=properties,
         )
 
 

vulnerabilities/tests/test_models.py

@@ -319,6 +319,38 @@ def test_vulnerability_model_as_cyclonedx(self):
         }
         self.assertEqual(expected, as_dict["analysis"])
 
+    def test_vulnerability_model_as_cyclonedx_is_reachable_property(self):
+        vulnerability = make_vulnerability(self.dataspace)
+        package = make_package(self.dataspace)
+        product_package = make_product_package(make_product(self.dataspace), package=package)
+
+        def make_analysis(is_reachable):
+            return VulnerabilityAnalysis(
+                product_package=product_package,
+                vulnerability=vulnerability,
+                dataspace=self.dataspace,
+                state=VulnerabilityAnalysis.State.IN_TRIAGE,
+                is_reachable=is_reachable,
+            )
+
+        cdx = vulnerability.as_cyclonedx(affected_instances=[package], analysis=make_analysis(True))
+        as_dict = json.loads(cdx.as_json())
+        self.assertEqual(
+            [{"name": "aboutcode:is_reachable", "value": "true"}], as_dict["properties"]
+        )
+
+        cdx = vulnerability.as_cyclonedx(
+            affected_instances=[package], analysis=make_analysis(False)
+        )
+        as_dict = json.loads(cdx.as_json())
+        self.assertEqual(
+            [{"name": "aboutcode:is_reachable", "value": "false"}], as_dict["properties"]
+        )
+
+        cdx = vulnerability.as_cyclonedx(affected_instances=[package], analysis=make_analysis(None))
+        as_dict = json.loads(cdx.as_json())
+        self.assertNotIn("properties", as_dict)
+
     def test_vulnerability_model_vulnerability_analysis_save(self):
         vulnerability1 = make_vulnerability(dataspace=self.dataspace)
         product_package1 = make_product_package(make_product(self.dataspace))

vulnerabilities/triage/management/commands/create_triage_rulesets.py

@@ -53,6 +53,19 @@
         "detail": "SSVC decision recommends Attend or Act. Flagged for review by triage.",
         "ruleset_name": "SSVC Attend or Act",
     },
+    {
+        "name": "Flag - Reachable Vulnerability",
+        "description": (
+            "Flag vulnerabilities confirmed as reachable in the product for patch prioritization."
+        ),
+        "state": "in_triage",
+        "is_reachable": True,
+        "detail": (
+            "Vulnerability confirmed reachable in the product context. "
+            "Flagged for patch prioritization."
+        ),
+        "ruleset_name": "Reachable Vulnerability",
+    },
 ]
 
 REFERENCE_RULESETS = [
@@ -63,7 +76,7 @@
             " affecting the product."
         ),
         "recommended_action": TriageAction.UPGRADE,
-        "precedence": 700,
+        "precedence": 800,
         "rules_config": {
             "risk_score": {"is_active": True, "min_risk_score": 8.0},
             "exploited_vulnerability": {"is_active": True},
@@ -76,7 +89,7 @@
             " regardless of severity."
         ),
         "recommended_action": TriageAction.UPGRADE,
-        "precedence": 600,
+        "precedence": 700,
         "rules_config": {
             "exploited_vulnerability": {"is_active": True},
         },
@@ -88,7 +101,7 @@
             " (Attend or Act)."
         ),
         "recommended_action": TriageAction.UPGRADE,
-        "precedence": 550,
+        "precedence": 600,
         "rules_config": {
             "ssvc_decision": {"is_active": True},
         },
@@ -232,6 +245,7 @@ def handle(self, *args, **options):
                 justification=preset_data.get("justification", ""),
                 responses=preset_data.get("responses"),
                 detail=preset_data.get("detail", ""),
+                is_reachable=preset_data.get("is_reachable"),
             )
             self.stdout.write(f"  Created preset: {preset_data['name']}")
 

vulnerabilities/triage/tests/test_commands.py

@@ -38,7 +38,7 @@ def test_creates_the_reference_rulesets_and_presets(self):
         management.call_command("create_triage_rulesets", self.dataspace.name, stdout=StringIO())
 
         self.assertEqual(8, TriageRuleset.objects.filter(dataspace=self.dataspace).count())
-        self.assertEqual(4, AnalysisPreset.objects.filter(dataspace=self.dataspace).count())
+        self.assertEqual(5, AnalysisPreset.objects.filter(dataspace=self.dataspace).count())
 
     def test_raises_when_rulesets_already_exist_without_reset(self):
         management.call_command("create_triage_rulesets", self.dataspace.name, stdout=StringIO())