diff --git a/product_portfolio/api.py b/product_portfolio/api.py
index 571012dc..d6d8d75e 100644
--- a/product_portfolio/api.py
+++ b/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",
)
diff --git a/product_portfolio/filters.py b/product_portfolio/filters.py
index 5c40ff65..9dff4bf6 100644
--- a/product_portfolio/filters.py
+++ b/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 = ''
class ComponentCompletenessListFilter(admin.SimpleListFilter):
diff --git a/product_portfolio/importers.py b/product_portfolio/importers.py
index d8966f56..8a06e75c 100644
--- a/product_portfolio/importers.py
+++ b/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."""
diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py
index e68bb5fd..ffdd2cf5 100644
--- a/product_portfolio/tests/test_api.py
+++ b/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)
diff --git a/product_portfolio/tests/test_views.py b/product_portfolio/tests/test_views.py
index 18ed8b56..750e7000 100644
--- a/product_portfolio/tests/test_views.py
+++ b/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.
diff --git a/product_portfolio/views.py b/product_portfolio/views.py
index 46eb0901..d837504d 100644
--- a/product_portfolio/views.py
+++ b/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'{self.filterset.form["is_reachable"]}'
+ 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)
diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py
index a0496a63..eb286bb2 100644
--- a/vulnerabilities/models.py
+++ b/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,
)
diff --git a/vulnerabilities/tests/test_models.py b/vulnerabilities/tests/test_models.py
index 2f96d012..9f5f58e8 100644
--- a/vulnerabilities/tests/test_models.py
+++ b/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))
diff --git a/vulnerabilities/triage/management/commands/create_triage_rulesets.py b/vulnerabilities/triage/management/commands/create_triage_rulesets.py
index 9356d85e..8aaa19a2 100644
--- a/vulnerabilities/triage/management/commands/create_triage_rulesets.py
+++ b/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']}")
diff --git a/vulnerabilities/triage/tests/test_commands.py b/vulnerabilities/triage/tests/test_commands.py
index a37936e6..17303d48 100644
--- a/vulnerabilities/triage/tests/test_commands.py
+++ b/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())