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())