Removed fixed version resolver for safetydb

Signed-off-by: Omkar Phansopkar <omkarphansopkar@gmail.com>

Author: OmkarPh

vulntotal/datasources/safetydb.py

@@ -12,10 +12,7 @@
 from typing import List
 
 import requests
-from fetchcode.package_versions import versions
 from packageurl import PackageURL
-from univers.version_range import PypiVersionRange
-from univers.versions import PypiVersion
 
 from vulntotal.validator import DataSource
 from vulntotal.validator import InvalidCVEError
@@ -53,8 +50,7 @@ def datasource_advisory(self, purl) -> Iterable[VendorData]:
             return []
         advisory = self.fetch_advisory()
         self._raw_dump.append(advisory)
-        self._versions = sorted([PypiVersion(ver.value) for ver in versions(str(purl))])
-        return parse_advisory(advisory, purl, self._versions)
+        return parse_advisory(advisory, purl)
 
     def datasource_advisory_from_cve(self, cve: str) -> Iterable[VendorData]:
         if not cve.upper().startswith("CVE-"):
@@ -65,13 +61,10 @@ def datasource_advisory_from_cve(self, cve: str) -> Iterable[VendorData]:
 
     @classmethod
     def supported_ecosystem(cls):
-        # source - @TODO
         return {"pypi": "PyPI"}
 
 
-def parse_advisory(
-    response, purl: PackageURL, all_versions: List[PypiVersion]
-) -> Iterable[VendorData]:
+def parse_advisory(response, purl: PackageURL) -> Iterable[VendorData]:
     """
     Parse response from safetydb API and yield VendorData
 
@@ -83,14 +76,11 @@ def parse_advisory(
     """
 
     for advisory in response.get(purl.name):
-        vulnerable_version_range_string = "vers:pypi/" + advisory.get("v").replace(",", "|")
-        vulnerable_version_range = PypiVersionRange.from_string(vulnerable_version_range_string)
-
         yield VendorData(
             purl=PackageURL(purl.type, purl.namespace, purl.name),
             aliases=[advisory.get("cve"), advisory.get("id")],
             affected_versions=sorted(advisory.get("specs")),
-            fixed_versions=get_patched_versions(all_versions, vulnerable_version_range),
+            fixed_versions=[],
         )
 
 
@@ -109,81 +99,11 @@ def parse_advisory_for_cve(response, cve: str) -> Iterable[VendorData]:
         if package == "$meta":
             continue
 
-        all_versions = sorted(
-            [PypiVersion(ver.value) for ver in versions(str(PackageURL(type="pypi", name=package)))]
-        )
-
         for advisory in advisories:
             if advisory.get("cve") == cve:
-                vulnerable_version_range_string = "vers:pypi/" + advisory.get("v").replace(",", "|")
-                vulnerable_version_range = PypiVersionRange.from_string(
-                    vulnerable_version_range_string
-                )
-
                 yield VendorData(
                     purl=PackageURL(type="pypi", name=package),
                     aliases=[advisory.get("cve"), advisory.get("id")],
                     affected_versions=sorted(advisory.get("specs")),
-                    fixed_versions=get_patched_versions(all_versions, vulnerable_version_range),
+                    fixed_versions=[],
                 )
-
-
-def get_patched_versions(
-    all_versions: List[PypiVersion],
-    vulnerable_version_range: PypiVersionRange,
-):
-    """
-    Get the first patched version from the list of all versions of a package
-
-    Parameters:
-        all_versions: A list containing PackageVersion of a package
-        vulnerable_version_range: A PypiVersionRange object specifying the vulnerable version range
-
-    Returns:
-        A PackageVersion object containing the first patched version of the package
-    """
-
-    # last_patched = None
-    # for version in reversed(all_versions):
-    #     if version in vulnerable_version_range:
-    #         if last_patched is not None:
-    #             return [str(last_patched.value)]
-    #         return []
-    #     last_patched = version
-    # return []
-
-    patched_version_ranges: List[str] = []
-    current_patched_range_start: PypiVersion = None
-    current_patched_range_latest: PypiVersion = None
-
-    def resolve_patched_range():
-        if current_patched_range_start is not None:
-            if current_patched_range_latest == current_patched_range_start:
-                patched_version_ranges.append(str(current_patched_range_start.value))
-            else:
-                patched_version_ranges.append(
-                    f">={current_patched_range_start.value},<={current_patched_range_latest.value}"
-                )
-
-    for version in all_versions:
-        if version in vulnerable_version_range:
-            resolve_patched_range()
-            current_patched_range_start = None
-            current_patched_range_latest = None
-        else:
-            if current_patched_range_start is None:
-                current_patched_range_start = version
-            current_patched_range_latest = version
-    resolve_patched_range()
-
-    # Remove upper bound from the last fixed range
-    if len(patched_version_ranges) > 0:
-        patched_version_ranges[-1] = patched_version_ranges[-1].split(",")[0]
-
-    # Ensure that >= is only present if there are fragmented fixed ranges
-    # eg. For vulnerable spec  "<2.2.5 >=2.3.0 <2.3.2",,fixed range => "2.2.5, >=2.3.2"
-    # eg. For vulnerable spec "<2.2.5", fixed range => "2.2.5
-    if len(patched_version_ranges) == 1:
-        patched_version_ranges[-1] = patched_version_ranges[-1][2:]
-
-    return patched_version_ranges

vulntotal/tests/test_data/safetydb/parse_advisory-expected.json

@@ -2,25 +2,25 @@
   {
     "purl": "pkg:pypi/flask",
     "affected_versions": ["<0.12.3"],
-    "fixed_versions": ["0.12.3"],
+    "fixed_versions": [],
     "aliases": ["CVE-2018-1000656", "pyup.io-36388"]
   },
   {
     "purl": "pkg:pypi/flask",
     "affected_versions": ["<0.12.3"],
-    "fixed_versions": ["0.12.3"],
+    "fixed_versions": [],
     "aliases": ["CVE-2019-1010083", "pyup.io-38654"]
   },
   {
     "purl": "pkg:pypi/flask",
     "affected_versions": ["<0.6.1"],
-    "fixed_versions": ["0.6.1"],
+    "fixed_versions": [],
     "aliases": ["PVE-2021-25820", "pyup.io-25820"]
   },
   {
     "purl": "pkg:pypi/flask",
     "affected_versions": ["<2.2.5", ">=2.3.0,<2.3.2"],
-    "fixed_versions": ["2.2.5", ">=2.3.2"],
+    "fixed_versions": [],
     "aliases": ["CVE-2023-30861", "pyup.io-55261"]
   }
 ]

vulntotal/tests/test_data/safetydb/parse_advisory_cve-expected.json

@@ -2,7 +2,7 @@
   {
     "purl": "pkg:pypi/flask",
     "affected_versions": ["<0.12.3"],
-    "fixed_versions": ["0.12.3"],
+    "fixed_versions": [],
     "aliases": ["CVE-2019-1010083", "pyup.io-38654"]
   }
 ]

vulntotal/tests/test_safetydb.py

@@ -11,10 +11,7 @@
 from pathlib import Path
 
 from commoncode import testcase
-from fetchcode.package_versions import versions
 from packageurl import PackageURL
-from univers.version_range import PypiVersionRange
-from univers.versions import PypiVersion
 
 from vulnerabilities.tests import util_tests
 from vulntotal.datasources import safetydb
@@ -28,9 +25,8 @@ def test_parse_advisory(self):
         advisory_file = self.get_test_loc("advisory.json")
         with open(advisory_file) as f:
             advisory = json.load(f)
-        all_versions = sorted([PypiVersion(ver.value) for ver in versions(str(purl))])
 
-        results = [adv.to_dict() for adv in safetydb.parse_advisory(advisory, purl, all_versions)]
+        results = [adv.to_dict() for adv in safetydb.parse_advisory(advisory, purl)]
         expected_file = self.get_test_loc("parse_advisory-expected.json", must_exist=False)
         util_tests.check_results_against_json(results, expected_file)
 
@@ -43,47 +39,3 @@ def test_parse_advisory_for_cve(self):
         results = [adv.to_dict() for adv in safetydb.parse_advisory_for_cve(advisory, cve)]
         expected_file = self.get_test_loc("parse_advisory_cve-expected.json", must_exist=False)
         util_tests.check_results_against_json(results, expected_file)
-
-    def test_get_patched_versions(self):
-        # Ref - flask package
-        all_versions = [
-            PypiVersion(ver)
-            for ver in [
-                "0.12.2",
-                "0.12.3",
-                "0.12.4",
-                "0.12.5",
-                "1.0",
-                "2.2.4",
-                "2.2.5",
-                "2.3.0",
-                "2.3.1",
-                "2.3.2",
-                "2.3.3",
-                "3.0.0",
-                "3.0.1",
-                "3.0.2",
-                "3.0.3",
-            ]
-        ]
-
-        test_cases = [
-            {
-                "vulnerable_version_range": PypiVersionRange.from_string("vers:pypi/<0.12.3"),
-                "expected_patched_version_ranges": ["0.12.3"],
-            },
-            {
-                "vulnerable_version_range": PypiVersionRange.from_string(
-                    "vers:pypi/<2.2.5|>=2.3.0|<2.3.2"
-                ),
-                "expected_patched_version_ranges": ["2.2.5", ">=2.3.2"],
-            },
-        ]
-
-        for test_case in test_cases:
-            results = safetydb.get_patched_versions(
-                all_versions, test_case["vulnerable_version_range"]
-            )
-            util_tests.check_results_against_expected(
-                results, test_case["expected_patched_version_ranges"]
-            )