Add pipeline to unfurl affected VERS range in V2 impacts

Signed-off-by: Keshav Priyadarshi <[email protected]>

Author: keshav-space

vulnerabilities/improvers/__init__.py

@@ -30,6 +30,7 @@
     enhance_with_metasploit as enhance_with_metasploit_v2,
 )
 from vulnerabilities.pipelines.v2_improvers import flag_ghost_packages as flag_ghost_packages_v2
+from vulnerabilities.pipelines.v2_improvers import unfurl_version_range as unfurl_version_range_v2
 from vulnerabilities.utils import create_registry
 
 IMPROVERS_REGISTRY = create_registry(
@@ -67,6 +68,7 @@
         compute_package_risk_v2.ComputePackageRiskPipeline,
         compute_version_rank_v2.ComputeVersionRankPipeline,
         compute_advisory_todo_v2.ComputeToDo,
+        unfurl_version_range_v2.UnfurlVersionRangePipeline,
         compute_advisory_todo.ComputeToDo,
     ]
 )

vulnerabilities/pipelines/v2_improvers/unfurl_version_range.py

@@ -0,0 +1,130 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import logging
+from traceback import format_exc as traceback_format_exc
+
+from aboutcode.pipeline import LoopProgress
+from fetchcode.package_versions import SUPPORTED_ECOSYSTEMS as FETCHCODE_SUPPORTED_ECOSYSTEMS
+from packageurl import PackageURL
+from univers.version_range import RANGE_CLASS_BY_SCHEMES
+from univers.version_range import VersionRange
+
+from vulnerabilities.models import ImpactedPackage
+from vulnerabilities.models import PackageV2
+from vulnerabilities.pipelines import VulnerableCodePipeline
+from vulnerabilities.pipes.fetchcode_utils import get_versions
+from vulnerabilities.utils import update_purl_version
+
+
+class UnfurlVersionRangePipeline(VulnerableCodePipeline):
+
+    pipeline_id = "unfurl_version_range_v2"
+
+    @classmethod
+    def steps(cls):
+        return (cls.unfurl_version_range,)
+
+    def unfurl_version_range(self):
+        impacted_packages = ImpactedPackage.objects.all().order_by("-created_at")
+        impacted_packages_count = impacted_packages.count()
+
+        processed_impacted_packages_count = 0
+        processed_affected_packages_count = 0
+        cached_versions = {}
+        self.log(f"Unfurl affected vers range for {impacted_packages_count:,d} ImpactedPackage.")
+        progress = LoopProgress(total_iterations=impacted_packages_count, logger=self.log)
+        for impact in progress.iter(impacted_packages):
+            purl = PackageURL.from_string(impact.base_purl)
+            if not impact.affecting_vers or not any(
+                c in impact.affecting_vers for c in ("<", ">", "!")
+            ):
+                continue
+            if purl.type not in FETCHCODE_SUPPORTED_ECOSYSTEMS:
+                continue
+            if purl.type not in RANGE_CLASS_BY_SCHEMES:
+                continue
+
+            versions = get_purl_versions(purl, cached_versions)
+            affected_purls = get_affected_purls(
+                versions=versions,
+                affecting_vers=impact.affecting_vers,
+                base_purl=purl,
+                logger=self.log,
+            )
+            if not affected_purls:
+                continue
+
+            processed_affected_packages_count += bulk_create_with_m2m(
+                purls=affected_purls,
+                impact=impact,
+                relation=ImpactedPackage.affecting_packages.through,
+                logger=self.log,
+            )
+            processed_impacted_packages_count += 1
+
+        self.log(f"Successfully processed {processed_impacted_packages_count:,d} ImpactedPackage.")
+        self.log(f"{processed_affected_packages_count:,d} new Impact-Package relation created.")
+
+
+def get_affected_purls(versions, affecting_vers, base_purl, logger):
+    affecting_version_range = VersionRange.from_string(affecting_vers)
+    version_class = affecting_version_range.version_class
+
+    try:
+        versions = [version_class(v) for v in versions]
+    except Exception as e:
+        logger(
+            f"Error while parsing versions for {base_purl!s}: {e!r} \n {traceback_format_exc()}",
+            level=logging.ERROR,
+        )
+        return
+
+    affected_purls = []
+    for version in versions:
+        try:
+            if version in affecting_version_range:
+                affected_purls.append(
+                    update_purl_version(
+                        purl=base_purl,
+                        version=str(version),
+                    )
+                )
+        except Exception as e:
+            logger(
+                f"Error while checking {version!s} in {affecting_version_range!s}: {e!r} \n {traceback_format_exc()}",
+                level=logging.ERROR,
+            )
+    return affected_purls
+
+
+def get_purl_versions(purl, cached_versions={}):
+    if not purl in cached_versions:
+        cached_versions[purl] = get_versions(purl)
+    return cached_versions[purl]
+
+
+def bulk_create_with_m2m(purls, impact, relation, logger):
+    """Bulk create PackageV2 and also bulk populate M2M Impact and Package relationships."""
+    if not purls:
+        return 0
+
+    affected_packages_v2 = PackageV2.objects.bulk_get_or_create_from_purls(purls=purls)
+
+    relations = [
+        relation(impactedpackage=impact, packagev2=package) for package in affected_packages_v2
+    ]
+
+    try:
+        relation.objects.bulk_create(relations, ignore_conflicts=True)
+    except Exception as e:
+        logger(f"Error creating ImpactedPackage {relation}: {e!r} \n {traceback_format_exc()}")
+        return 0
+
+    return len(relations)

vulnerabilities/pipes/fetchcode_utils.py

@@ -10,14 +10,18 @@
 import logging
 from traceback import format_exc as traceback_format_exc
 from typing import Callable
+from typing import Union
 
 from fetchcode.package_versions import SUPPORTED_ECOSYSTEMS as FETCHCODE_SUPPORTED_ECOSYSTEMS
 from fetchcode.package_versions import versions
 from packageurl import PackageURL
 
 
-def get_versions(purl: PackageURL, logger: Callable = None):
+def get_versions(purl: Union[PackageURL, str], logger: Callable = None):
     """Return set of known versions for the given purl."""
+    if isinstance(purl, str):
+        purl = PackageURL.from_string(purl)
+
     if purl.type not in FETCHCODE_SUPPORTED_ECOSYSTEMS:
         return
 

vulnerabilities/tests/pipelines/v2_importers/test_redhat_importer_v2.py

@@ -7,10 +7,8 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import json
-import os
+
 from pathlib import Path
-from unittest.mock import Mock
 from unittest.mock import patch
 
 from django.test import TestCase

vulnerabilities/tests/pipelines/v2_improvers/test_unfurl_version_range.py

@@ -0,0 +1,47 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import ImpactedPackage
+from vulnerabilities.models import PackageV2
+from vulnerabilities.pipelines.v2_improvers.unfurl_version_range import UnfurlVersionRangePipeline
+
+
+class TestUnfurlVersionRangePipeline(TestCase):
+    def setUp(self):
+        self.advisory1 = AdvisoryV2.objects.create(
+            datasource_id="ghsa",
+            advisory_id="GHSA-1234",
+            avid="ghsa/GHSA-1234",
+            unique_content_id="f" * 64,
+            url="https://example.com/advisory",
+            date_collected="2025-07-01T00:00:00Z",
+        )
+
+        self.impact1 = ImpactedPackage.objects.create(
+            advisory=self.advisory1,
+            base_purl="pkg:npm/foobar",
+            affecting_vers="vers:npm/>3.2.1|<4.0.0",
+            fixed_vers=None,
+        )
+
+    @patch("vulnerabilities.pipelines.v2_improvers.unfurl_version_range.get_purl_versions")
+    def test_affecting_version_range_unfurl(self, mock_fetch):
+        self.assertEqual(0, PackageV2.objects.count())
+        mock_fetch.return_value = {"3.4.1", "3.9.0", "2.1.0", "4.0.0", "4.1.0"}
+        pipeline = UnfurlVersionRangePipeline()
+        pipeline.execute()
+
+        self.assertEqual(2, PackageV2.objects.count())
+        self.assertEqual(2, self.impact1.affecting_packages.count())