Create impacted pkg during advisory insertion in v2 base importer

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulnerabilities/pipelines/__init__.py

@@ -15,29 +15,18 @@
 from traceback import format_exc as traceback_format_exc
 from typing import Iterable
 from typing import List
-from typing import Optional
 
 from aboutcode.pipeline import LoopProgress
 from aboutcode.pipeline import PipelineDefinition
 from aboutcode.pipeline import humanize_time
-from fetchcode import package_versions
-from packageurl import PackageURL
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
-from vulnerabilities.importer import UnMergeablePackageError
 from vulnerabilities.improver import MAX_CONFIDENCE
 from vulnerabilities.models import Advisory
-from vulnerabilities.models import PackageV2
 from vulnerabilities.models import PipelineRun
 from vulnerabilities.pipes.advisory import import_advisory
 from vulnerabilities.pipes.advisory import insert_advisory
 from vulnerabilities.pipes.advisory import insert_advisory_v2
-from vulnerabilities.utils import AffectedPackage as LegacyAffectedPackage
-from vulnerabilities.utils import classproperty
-from vulnerabilities.utils import get_affected_packages_by_patched_package
-from vulnerabilities.utils import nearest_patched_package
-from vulnerabilities.utils import resolve_version_range
 
 module_logger = logging.getLogger(__name__)
 
@@ -271,13 +260,15 @@ class VulnerableCodeBaseImporterPipelineV2(VulnerableCodePipeline):
     license_url = None
     spdx_license_expression = None
     repo_url = None
-    advisory_confidence = MAX_CONFIDENCE
     ignorable_versions = []
-    unfurl_version_ranges = False
 
     @classmethod
     def steps(cls):
-        return (cls.collect_and_store_advisories,)
+        return (
+            # Add step for downloading/cloning resource as required.
+            cls.collect_and_store_advisories,
+            # Add step for removing downloaded/cloned resource as required.
+        )
 
     def collect_advisories(self) -> Iterable[AdvisoryData]:
         """
@@ -311,7 +302,6 @@ def collect_and_store_advisories(self):
                 if _obj := insert_advisory_v2(
                     advisory=advisory,
                     pipeline_id=self.pipeline_id,
-                    get_advisory_packages=self.get_advisory_packages,
                     logger=self.log,
                 ):
                     collected_advisory_count += 1
@@ -323,203 +313,3 @@ def collect_and_store_advisories(self):
                 continue
 
         self.log(f"Successfully collected {collected_advisory_count:,d} advisories")
-
-    def get_advisory_packages(self, advisory_data: AdvisoryData) -> list:
-        """
-        Return the list of packages for the given advisory.
-
-        Used by ``import_advisory`` to get the list of packages for the advisory.
-        """
-        from vulnerabilities.improvers import default
-
-        affected_purls = []
-        fixed_purls = []
-        for affected_package in advisory_data.affected_packages:
-            package_affected_purls, package_fixed_purls = default.get_exact_purls(
-                affected_package=affected_package
-            )
-            affected_purls.extend(package_affected_purls)
-            fixed_purls.extend(package_fixed_purls)
-
-        if self.unfurl_version_ranges:
-            vulnerable_pvs, fixed_pvs = self.get_impacted_packages(
-                affected_packages=advisory_data.affected_packages,
-                advisory_date_published=advisory_data.date_published,
-            )
-            affected_purls.extend(vulnerable_pvs)
-            fixed_purls.extend(fixed_pvs)
-
-        vulnerable_packages = []
-        fixed_packages = []
-
-        for affected_purl in affected_purls:
-            vulnerable_package, _ = PackageV2.objects.get_or_create_from_purl(purl=affected_purl)
-            vulnerable_packages.append(vulnerable_package)
-
-        for fixed_purl in fixed_purls:
-            fixed_package, _ = PackageV2.objects.get_or_create_from_purl(purl=fixed_purl)
-            fixed_packages.append(fixed_package)
-
-        return vulnerable_packages, fixed_packages
-
-    def get_published_package_versions(
-        self, package_url: PackageURL, until: Optional[datetime] = None
-    ) -> List[str]:
-        """
-        Return a list of versions published before `until` for the `package_url`
-        """
-        versions_before_until = []
-        try:
-            versions = package_versions.versions(str(package_url))
-            for version in versions or []:
-                if (
-                    version.release_date
-                    and version.release_date.tzinfo
-                    and until
-                    and until.tzinfo is None
-                ):
-                    until = until.replace(tzinfo=timezone.utc)
-                if until and version.release_date and version.release_date > until:
-                    continue
-                versions_before_until.append(version.value)
-
-            return versions_before_until
-        except Exception as e:
-            self.log(
-                f"Failed to fetch versions for package {str(package_url)} {e!r}",
-                level=logging.ERROR,
-            )
-            return []
-
-    def get_impacted_packages(self, affected_packages, advisory_date_published):
-        """
-        Return a tuple of lists of affected and fixed PackageURLs
-        """
-        if not affected_packages:
-            return [], []
-
-        mergable = True
-
-        # TODO: We should never had the exception in first place
-        try:
-            purl, affected_version_ranges, fixed_versions = AffectedPackage.merge(affected_packages)
-        except UnMergeablePackageError:
-            self.log(f"Cannot merge with different purls {affected_packages!r}", logging.ERROR)
-            mergable = False
-
-        if not mergable:
-            vulnerable_packages = []
-            fixed_packages = []
-            for affected_package in affected_packages:
-                purl = affected_package.package
-                affected_version_range = affected_package.affected_version_range
-                fixed_version = affected_package.fixed_version
-                pkg_type = purl.type
-                pkg_namespace = purl.namespace
-                pkg_name = purl.name
-                if not affected_version_range and fixed_version:
-                    fixed_packages.append(
-                        PackageURL(
-                            type=pkg_type,
-                            namespace=pkg_namespace,
-                            name=pkg_name,
-                            version=str(fixed_version),
-                        )
-                    )
-                else:
-                    valid_versions = self.get_published_package_versions(
-                        package_url=purl, until=advisory_date_published
-                    )
-                    affected_pvs, fixed_pvs = self.resolve_package_versions(
-                        affected_version_range=affected_version_range,
-                        pkg_type=pkg_type,
-                        pkg_namespace=pkg_namespace,
-                        pkg_name=pkg_name,
-                        valid_versions=valid_versions,
-                    )
-                    vulnerable_packages.extend(affected_pvs)
-                    fixed_packages.extend(fixed_pvs)
-            return vulnerable_packages, fixed_packages
-        else:
-            pkg_type = purl.type
-            pkg_namespace = purl.namespace
-            pkg_name = purl.name
-            pkg_qualifiers = purl.qualifiers
-            fixed_purls = [
-                PackageURL(
-                    type=pkg_type,
-                    namespace=pkg_namespace,
-                    name=pkg_name,
-                    version=str(version),
-                    qualifiers=pkg_qualifiers,
-                )
-                for version in fixed_versions
-            ]
-            if not affected_version_ranges:
-                return [], fixed_purls
-            else:
-                valid_versions = self.get_published_package_versions(
-                    package_url=purl, until=advisory_date_published
-                )
-                vulnerable_packages = []
-                fixed_packages = []
-                for affected_version_range in affected_version_ranges:
-                    vulnerable_pvs, fixed_pvs = self.resolve_package_versions(
-                        affected_version_range=affected_version_range,
-                        pkg_type=pkg_type,
-                        pkg_namespace=pkg_namespace,
-                        pkg_name=pkg_name,
-                        valid_versions=valid_versions,
-                    )
-                    vulnerable_packages.extend(vulnerable_pvs)
-                    fixed_packages.extend(fixed_pvs)
-                return vulnerable_packages, fixed_packages
-
-    def resolve_package_versions(
-        self,
-        affected_version_range,
-        pkg_type,
-        pkg_namespace,
-        pkg_name,
-        valid_versions,
-    ):
-        """
-        Return a tuple of lists of ``affected_packages`` and ``fixed_packages`` PackageURL for the given `affected_version_range` and `valid_versions`.
-
-        ``valid_versions`` are the valid version listed on the package registry for that package
-
-        """
-        aff_vers, unaff_vers = resolve_version_range(
-            affected_version_range=affected_version_range,
-            ignorable_versions=self.ignorable_versions,
-            package_versions=valid_versions,
-        )
-
-        affected_purls = list(
-            self.expand_verion_range_to_purls(pkg_type, pkg_namespace, pkg_name, aff_vers)
-        )
-
-        unaffected_purls = list(
-            self.expand_verion_range_to_purls(pkg_type, pkg_namespace, pkg_name, unaff_vers)
-        )
-
-        fixed_packages = []
-        affected_packages = []
-
-        patched_packages = nearest_patched_package(
-            vulnerable_packages=affected_purls, resolved_packages=unaffected_purls
-        )
-
-        for (
-            fixed_package,
-            affected_purls,
-        ) in get_affected_packages_by_patched_package(patched_packages).items():
-            if fixed_package:
-                fixed_packages.append(fixed_package)
-            affected_packages.extend(affected_purls)
-
-        return affected_packages, fixed_packages
-
-    def expand_verion_range_to_purls(self, pkg_type, pkg_namespace, pkg_name, versions):
-        for version in versions:
-            yield PackageURL(type=pkg_type, namespace=pkg_namespace, name=pkg_name, version=version)

vulnerabilities/pipes/advisory.py

@@ -35,6 +35,7 @@
 from vulnerabilities.models import VulnerabilityRelatedReference
 from vulnerabilities.models import VulnerabilitySeverity
 from vulnerabilities.models import Weakness
+from vulnerabilities.pipes.univers_utils import get_exact_purls_v2
 
 
 def get_or_create_aliases(aliases: List) -> QuerySet:
@@ -43,9 +44,6 @@ def get_or_create_aliases(aliases: List) -> QuerySet:
     return Alias.objects.filter(alias__in=aliases)
 
 
-from django.db.models import Q
-
-
 def get_or_create_advisory_aliases(aliases: List[str]) -> List[AdvisoryAlias]:
     existing = AdvisoryAlias.objects.filter(alias__in=aliases)
     existing_aliases = {a.alias for a in existing}
@@ -139,9 +137,10 @@ def insert_advisory(advisory: AdvisoryData, pipeline_id: str, logger: Callable =
 def insert_advisory_v2(
     advisory: AdvisoryData,
     pipeline_id: str,
-    get_advisory_packages: Callable,
     logger: Callable = None,
 ):
+    from vulnerabilities.models import ImpactedPackage
+    from vulnerabilities.models import PackageV2
     from vulnerabilities.utils import compute_content_id
 
     advisory_obj = None
@@ -150,7 +149,6 @@ def insert_advisory_v2(
     severities = get_or_create_advisory_severities(severities=advisory.severities)
     weaknesses = get_or_create_advisory_weaknesses(weaknesses=advisory.weaknesses)
     content_id = compute_content_id(advisory_data=advisory)
-    affecting_packages, fixed_by_packages = get_advisory_packages(advisory_data=advisory)
     try:
         default_data = {
             "datasource_id": pipeline_id,
@@ -162,7 +160,7 @@ def insert_advisory_v2(
             "original_advisory_text": advisory.original_advisory_text,
         }
 
-        advisory_obj, _ = AdvisoryV2.objects.get_or_create(
+        advisory_obj, created = AdvisoryV2.objects.get_or_create(
             unique_content_id=content_id,
             url=advisory.url,
             defaults=default_data,
@@ -172,8 +170,6 @@ def insert_advisory_v2(
             "references": references,
             "severities": severities,
             "weaknesses": weaknesses,
-            "fixed_by_packages": fixed_by_packages,
-            "affecting_packages": affecting_packages,
         }
 
         for field_name, values in related_fields.items():
@@ -192,6 +188,29 @@ def insert_advisory_v2(
                 level=logging.ERROR,
             )
 
+    if created:
+        for affected_pkg in advisory.affected_packages:
+            impact = ImpactedPackage.objects.create(
+                advisory=advisory_obj,
+                base_purl=str(affected_pkg.package),
+                affecting_vers=str(affected_pkg.affected_version_range),
+                fixed_vers=str(affected_pkg.fixed_version_range),
+            )
+            package_affected_purls, package_fixed_purls = get_exact_purls_v2(
+                affected_package=affected_pkg,
+                logger=logger,
+            )
+            affected_packages_v2 = [
+                PackageV2.objects.get_or_create_from_purl(purl=purl)[0]
+                for purl in package_affected_purls
+            ]
+            fixed_packages_v2 = [
+                PackageV2.objects.get_or_create_from_purl(purl=purl)[0]
+                for purl in package_fixed_purls
+            ]
+            impact.affecting_packages.add(*affected_packages_v2)
+            impact.fixed_by_packages.add(*fixed_packages_v2)
+
     return advisory_obj