Update V2 pipelines to create AffectedPackageV2

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

Author: keshav-space

vulnerabilities/importer.py

@@ -360,7 +360,7 @@ def __post_init__(self):
             )
 
     def __lt__(self, other):
-        if not isinstance(other, AffectedPackage):
+        if not isinstance(other, AffectedPackageV2):
             return NotImplemented
         return self._cmp_key() < other._cmp_key()
 
@@ -446,8 +446,6 @@ class AdvisoryData:
     original_advisory_text: Optional[str] = None
 
     def __post_init__(self):
-        if self.date_published and not self.date_published.tzinfo:
-            logger.warning(f"AdvisoryData with no tzinfo: {self!r}")
         if self.summary:
             self.summary = self.clean_summary(self.summary)
 

vulnerabilities/importers/osv.py

@@ -23,6 +23,7 @@
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.importer import VulnerabilitySeverity
@@ -144,19 +145,27 @@ def parse_advisory_data_v2(
             supported_ecosystem=purl.type,
         )
 
+        fixed_versions = []
+        fixed_version_range = None
         for fixed_range in affected_pkg.get("ranges") or []:
             fixed_version = get_fixed_versions(
                 fixed_range=fixed_range, raw_id=advisory_id, supported_ecosystem=purl.type
             )
+            fixed_versions.extend([v.string for v in fixed_version])
 
-            for version in fixed_version:
-                affected_packages.append(
-                    AffectedPackage(
-                        package=purl,
-                        affected_version_range=affected_version_range,
-                        fixed_version=version,
-                    )
+        fixed_version_range = (
+            get_fixed_version_range(fixed_versions, purl.type) if fixed_versions else None
+        )
+
+        if fixed_version_range or affected_version_range:
+            affected_packages.append(
+                AffectedPackageV2(
+                    package=purl,
+                    affected_version_range=affected_version_range,
+                    fixed_version_range=fixed_version_range,
                 )
+            )
+
     database_specific = raw_data.get("database_specific") or {}
     cwe_ids = database_specific.get("cwe_ids") or []
     weaknesses = list(map(get_cwe_id, cwe_ids))
@@ -334,6 +343,13 @@ def get_affected_version_range(affected_pkg, raw_id, supported_ecosystem):
             )
 
 
+def get_fixed_version_range(versions, ecosystem):
+    try:
+        return RANGE_CLASS_BY_SCHEMES[ecosystem].from_versions(versions)
+    except Exception as e:
+        logger.error(f"Failed to create VersionRange from: {versions}: error:{e!r}")
+
+
 def get_fixed_versions(fixed_range, raw_id, supported_ecosystem) -> List[Version]:
     """
     Return a list of unique fixed univers Versions given a ``fixed_range``

vulnerabilities/pipelines/v2_importers/apache_httpd_importer.py

@@ -22,7 +22,7 @@
 from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
@@ -151,7 +151,6 @@ class ApacheHTTPDImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     spdx_license_expression = "Apache-2.0"
     license_url = "https://www.apache.org/licenses/LICENSE-2.0"
     base_url = "https://httpd.apache.org/security/json/"
-    unfurl_version_ranges = True
 
     links = []
 
@@ -219,7 +218,6 @@ class ApacheHTTPDImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
             "pre_ajp_proxy",
         ]
     )
-    unfurl_version_ranges = True
 
     @classmethod
     def steps(cls):
@@ -292,7 +290,7 @@ def to_advisory(self, data):
         affected_version_range = self.to_version_ranges(versions_data, fixed_versions)
         if affected_version_range:
             affected_packages.append(
-                AffectedPackage(
+                AffectedPackageV2(
                     package=PackageURL(
                         type="apache",
                         name="httpd",

vulnerabilities/pipelines/v2_importers/curl_importer.py

@@ -12,10 +12,9 @@
 from cwe2.database import Database
 from packageurl import PackageURL
 from univers.version_range import GenericVersionRange
-from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
@@ -37,7 +36,6 @@ class CurlImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     license_url = "https://curl.se/docs/copyright.html"
     repo_url = "https://github.com/curl/curl-www/"
     url = "https://curl.se/docs/vuln.json"
-    unfurl_version_ranges = True
 
     @classmethod
     def steps(cls):
@@ -77,16 +75,16 @@ def parse_curl_advisory(raw_data) -> AdvisoryData:
     version_type = get_item(ranges, "type") if get_item(ranges, "type") else ""
     fixed_version = events.get("fixed")
     if version_type == "SEMVER" and fixed_version:
-        fixed_version = SemverVersion(fixed_version)
+        fixed_version_range = GenericVersionRange.from_versions([fixed_version])
 
     purl = PackageURL(type="generic", namespace="curl.se", name="curl")
     versions = affected.get("versions") or []
     affected_version_range = GenericVersionRange.from_versions(versions)
 
-    affected_package = AffectedPackage(
+    affected_package = AffectedPackageV2(
         package=purl,
         affected_version_range=affected_version_range,
-        fixed_version=fixed_version,
+        fixed_version_range=fixed_version_range,
     )
 
     database_specific = raw_data.get("database_specific") or {}

vulnerabilities/pipelines/v2_importers/elixir_security_importer.py

@@ -17,7 +17,7 @@
 from univers.version_range import HexVersionRange
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 from vulnerabilities.utils import is_cve
@@ -26,7 +26,7 @@
 
 class ElixirSecurityImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     """
-    Elixir Security Advisiories Importer Pipeline
+    Elixir Security Advisories Importer Pipeline
 
     This pipeline imports security advisories for elixir.
     """
@@ -35,7 +35,6 @@ class ElixirSecurityImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     spdx_license_expression = "CC0-1.0"
     license_url = "https://github.com/dependabot/elixir-security-advisories/blob/master/LICENSE.txt"
     repo_url = "git+https://github.com/dependabot/elixir-security-advisories"
-    unfurl_version_ranges = True
 
     @classmethod
     def steps(cls):
@@ -118,7 +117,7 @@ def process_file(self, file, base_path) -> Iterable[AdvisoryData]:
         affected_packages = []
         if pkg_name:
             affected_packages.append(
-                AffectedPackage(
+                AffectedPackageV2(
                     package=PackageURL(type="hex", name=pkg_name),
                     affected_version_range=HexVersionRange(constraints=constraints),
                 )

vulnerabilities/pipelines/v2_importers/github_osv_importer.py

@@ -29,7 +29,6 @@ class GithubOSVImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     spdx_license_expression = "CC-BY-4.0"
     license_url = "https://github.com/github/advisory-database/blob/main/LICENSE.md"
     repo_url = "git+https://github.com/github/advisory-database/"
-    unfurl_version_ranges = True
 
     @classmethod
     def steps(cls):

vulnerabilities/pipelines/v2_importers/gitlab_importer.py

@@ -12,7 +12,6 @@
 import traceback
 from pathlib import Path
 from typing import Iterable
-from typing import List
 from typing import Tuple
 
 import pytz
@@ -21,12 +20,10 @@
 from fetchcode.vcs import fetch_via_vcs
 from packageurl import PackageURL
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
-from univers.version_range import VersionRange
 from univers.version_range import from_gitlab_native
-from univers.versions import Version
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 from vulnerabilities.utils import build_description
@@ -45,7 +42,6 @@ class GitLabImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     spdx_license_expression = "MIT"
     license_url = "https://gitlab.com/gitlab-org/advisories-community/-/blob/main/LICENSE"
     repo_url = "git+https://gitlab.com/gitlab-org/advisories-community/"
-    unfurl_version_ranges = True
 
     @classmethod
     def steps(cls):
@@ -177,27 +173,6 @@ def get_purl(package_slug, purl_type_by_gitlab_scheme, logger):
     return
 
 
-def extract_affected_packages(
-    affected_version_range: VersionRange,
-    fixed_versions: List[Version],
-    purl: PackageURL,
-) -> Iterable[AffectedPackage]:
-    """
-    Yield AffectedPackage objects, one for each fixed_version
-
-    In case of gitlab advisory data we get a list of fixed_versions and a affected_version_range.
-    Since we can not determine which package fixes which range.
-    We store the all the fixed_versions with the same affected_version_range in the advisory.
-    Later the advisory data is used to be inferred in the GitLabBasicImprover.
-    """
-    for fixed_version in fixed_versions:
-        yield AffectedPackage(
-            package=purl,
-            fixed_version=fixed_version,
-            affected_version_range=affected_version_range,
-        )
-
-
 def parse_gitlab_advisory(
     file, base_path, gitlab_scheme_by_purl_type, purl_type_by_gitlab_scheme, logger
 ):
@@ -276,7 +251,7 @@ def parse_gitlab_advisory(
     fixed_versions = gitlab_advisory.get("fixed_versions") or []
     affected_range = gitlab_advisory.get("affected_range")
     gitlab_native_schemes = set(["pypi", "gem", "npm", "go", "packagist", "conan"])
-    vrc: VersionRange = RANGE_CLASS_BY_SCHEMES[purl.type]
+    vrc = RANGE_CLASS_BY_SCHEMES[purl.type]
     gitlab_scheme = gitlab_scheme_by_purl_type[purl.type]
     try:
         if affected_range:
@@ -296,38 +271,33 @@ def parse_gitlab_advisory(
     for fixed_version in fixed_versions:
         try:
             fixed_version = vrc.version_class(fixed_version)
-            parsed_fixed_versions.append(fixed_version)
+            parsed_fixed_versions.append(fixed_version.string)
         except Exception as e:
             logger(
                 f"parse_yaml_file: fixed_version is not parsable`: {fixed_version!r} error: {e!r}\n {traceback.format_exc()}",
                 level=logging.ERROR,
             )
 
-    if parsed_fixed_versions:
-        affected_packages = list(
-            extract_affected_packages(
-                affected_version_range=affected_version_range,
-                fixed_versions=parsed_fixed_versions,
-                purl=purl,
-            )
-        )
-    else:
-        if not affected_version_range:
-            affected_packages = []
-        else:
-            affected_packages = [
-                AffectedPackage(
-                    package=purl,
-                    affected_version_range=affected_version_range,
-                )
-            ]
+    if affected_version_range:
+        vrc = affected_version_range.__class__
+
+    fixed_version_range = vrc.from_versions(parsed_fixed_versions)
+    if not fixed_version_range and not affected_version_range:
+        return
+
+    affected_package = AffectedPackageV2(
+        package=purl,
+        affected_version_range=affected_version_range,
+        fixed_version_range=fixed_version_range,
+    )
+
     return AdvisoryData(
         advisory_id=advisory_id,
         aliases=aliases,
         summary=summary,
         references_v2=references,
         date_published=date_published,
-        affected_packages=affected_packages,
+        affected_packages=[affected_package],
         weaknesses=cwe_list,
         url=advisory_url,
         original_advisory_text=json.dumps(gitlab_advisory, indent=2, ensure_ascii=False),