Migrate oval importer and improver

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importer.py

@@ -35,11 +35,14 @@
 from typing import Set
 from typing import Tuple
 
+import pytz
 from binaryornot.helpers import is_binary_string
+from dateutil import parser as dateparser
 from git import DiffIndex
 from git import Repo
 from license_expression import Licensing
 from packageurl import PackageURL
+from univers.version_range import RANGE_CLASS_BY_SCHEMES
 from univers.version_range import VersionRange
 from univers.versions import Version
 
@@ -521,13 +524,13 @@ class OvalImporter(Importer):
     """
 
     @staticmethod
-    def create_purl(pkg_name: str, pkg_version: str, pkg_data: Mapping) -> PackageURL:
+    def create_purl(pkg_name: str, pkg_data: Mapping) -> PackageURL:
         """
         Helper method for creating different purls for subclasses without them reimplementing
         get_data_from_xml_doc  method
         Note: pkg_data must include 'type' of package
         """
-        return PackageURL(name=pkg_name, version=pkg_version, **pkg_data)
+        return PackageURL(name=pkg_name, **pkg_data)
 
     @staticmethod
     def _collect_pkgs(parsed_oval_data: Mapping) -> Set:
@@ -561,28 +564,17 @@ def advisory_data(self) -> List[AdvisoryData]:
         for metadata, oval_file in self._fetch():
             try:
                 oval_data = self.get_data_from_xml_doc(oval_file, metadata)
-                yield oval_data
+                yield from oval_data
             except Exception:
                 logger.error(
                     f"Failed to get updated_advisories: {oval_file!r} "
                     f"with {metadata!r}:\n" + traceback.format_exc()
                 )
                 continue
 
-    def set_api(self, all_pkgs: Iterable[str]):
-        """
-        This method loads the self.pkg_manager_api with the specified packages.
-        It fetches and caches all the versions of these packages and exposes
-        them through self.pkg_manager_api.get(<package_name>). Example
-
-        >> self.set_api(['electron'])
-        Assume 'electron' has only versions 1.0.0 and 1.2.0
-        >> assert  self.pkg_manager_api.get('electron') == {'1.0.0','1.2.0'}
-
-        """
-        raise NotImplementedError
-
-    def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> List[AdvisoryData]:
+    def get_data_from_xml_doc(
+        self, xml_doc: ET.ElementTree, pkg_metadata={}
+    ) -> Iterable[AdvisoryData]:
         """
         The orchestration method of the OvalDataSource. This method breaks an
         OVAL xml ElementTree into a list of `Advisory`.
@@ -593,12 +585,10 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
         Example value of pkg_metadata:
                 {"type":"deb","qualifiers":{"distro":"buster"} }
         """
-
-        all_adv = []
-        oval_doc = OvalParser(self.translations, xml_doc)
-        raw_data = oval_doc.get_data()
-        all_pkgs = self._collect_pkgs(raw_data)
-        self.set_api(all_pkgs)
+        oval_parsed_data = OvalParser(self.translations, xml_doc)
+        raw_data = oval_parsed_data.get_data()
+        oval_doc = oval_parsed_data.oval_document
+        timestamp = oval_doc.getGenerator().getTimestamp()
 
         # convert definition_data to Advisory objects
         for definition_data in raw_data:
@@ -610,49 +600,31 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
             affected_packages = []
             for test_data in definition_data["test_data"]:
                 for package_name in test_data["package_list"]:
-                    if package_name and len(package_name) >= 50:
-                        continue
-
-                    affected_version_range = test_data["version_ranges"] or set()
-                    version_class = version_class_by_package_type[pkg_metadata["type"]]
-                    version_scheme = version_class.scheme
-
-                    affected_version_range = VersionRange.from_scheme_version_spec_string(
-                        version_scheme, affected_version_range
-                    )
-                    all_versions = self.pkg_manager_api.get(package_name).valid_versions
-
-                    # FIXME: what is this 50 DB limit? that's too small for versions
-                    # FIXME: we should not drop data this way
-                    # This filter is for filtering out long versions.
-                    # 50 is limit because that's what db permits atm.
-                    all_versions = [version for version in all_versions if len(version) < 50]
-                    if not all_versions:
-                        continue
-
-                    affected_purls = []
-                    safe_purls = []
-                    for version in all_versions:
-                        purl = self.create_purl(
-                            pkg_name=package_name,
-                            pkg_version=version,
-                            pkg_data=pkg_metadata,
+                    affected_version_range = test_data["version_ranges"]
+                    vrc = RANGE_CLASS_BY_SCHEMES[pkg_metadata["type"]]
+                    if affected_version_range:
+                        try:
+                            affected_version_range = vrc.from_native(affected_version_range)
+                        except Exception as e:
+                            logger.error(
+                                f"Failed to parse version range {affected_version_range!r} "
+                                f"for package {package_name!r}:\n{e}"
+                            )
+                            continue
+                    if package_name:
+                        affected_packages.append(
+                            AffectedPackage(
+                                package=self.create_purl(package_name, pkg_metadata),
+                                affected_version_range=affected_version_range,
+                            )
                         )
-                        if version_class(version) in affected_version_range:
-                            affected_purls.append(purl)
-                        else:
-                            safe_purls.append(purl)
-
-                    affected_packages.extend(
-                        nearest_patched_package(affected_purls, safe_purls),
-                    )
-
-            all_adv.append(
-                AdvisoryData(
-                    summary=description,
-                    affected_packages=affected_packages,
-                    vulnerability_id=vuln_id,
-                    references=references,
-                )
+            date_published = dateparser.parse(timestamp)
+            if not date_published.tzinfo:
+                date_published = date_published.replace(tzinfo=pytz.UTC)
+            yield AdvisoryData(
+                aliases=[vuln_id],
+                summary=description,
+                affected_packages=affected_packages,
+                references=sorted(references),
+                date_published=date_published,
             )
-        return all_adv

vulnerabilities/importers/__init__.py

@@ -21,13 +21,15 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 from vulnerabilities.importers import alpine_linux
 from vulnerabilities.importers import debian
+from vulnerabilities.importers import debian_oval
 from vulnerabilities.importers import github
 from vulnerabilities.importers import gitlab
 from vulnerabilities.importers import nginx
 from vulnerabilities.importers import nvd
 from vulnerabilities.importers import openssl
 from vulnerabilities.importers import pysec
 from vulnerabilities.importers import redhat
+from vulnerabilities.importers import ubuntu
 
 IMPORTERS_REGISTRY = [
     nginx.NginxImporter,
@@ -39,6 +41,8 @@
     pysec.PyPIImporter,
     debian.DebianImporter,
     gitlab.GitLabAPIImporter,
+    debian_oval.DebianOvalImporter,
+    ubuntu.UbuntuImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/debian_oval.py

@@ -21,38 +21,29 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 
-import asyncio
 import xml.etree.ElementTree as ET
 
 import requests
 
 from vulnerabilities.importer import OvalImporter
-from vulnerabilities.package_managers import DebianVersionAPI
-from vulnerabilities.utils import create_etag
 
 
 class DebianOvalImporter(OvalImporter):
+    spdx_license_expression = "LicenseRef-scancode-unknown"
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         # we could avoid setting translations, and have it
         # set by default in the OvalParser, but we don't yet know
         # whether all OVAL providers use the same format
         self.translations = {"less than": "<"}
-        self.pkg_manager_api = DebianVersionAPI()
 
     def _fetch(self):
-        releases = self.config.releases
+        releases = ["wheezy", "stretch", "jessie", "buster", "bullseye"]
         for release in releases:
             file_url = f"https://www.debian.org/security/oval/oval-definitions-{release}.xml"
-            if not create_etag(data_src=self, url=file_url, etag_key="ETag"):
-                continue
-
             resp = requests.get(file_url).content
             yield (
                 {"type": "deb", "namespace": "debian", "qualifiers": {"distro": release}},
                 ET.ElementTree(ET.fromstring(resp.decode("utf-8"))),
             )
-        return []
-
-    def set_api(self, packages):
-        asyncio.run(self.pkg_manager_api.load_api(packages))

vulnerabilities/importers/ubuntu.py

@@ -21,32 +21,32 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 
-import asyncio
 import bz2
 import logging
 import xml.etree.ElementTree as ET
 
 import requests
 
 from vulnerabilities.importer import OvalImporter
-from vulnerabilities.package_managers import LaunchpadVersionAPI
 
 logger = logging.getLogger(__name__)
 
 
 class UbuntuImporter(OvalImporter):
+    spdx_license_expression = "GPL"
+    license_url = "https://ubuntu.com/legal/terms"
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         # we could avoid setting translations, and have it
         # set by default in the OvalParser, but we don't yet know
         # whether all OVAL providers use the same format
         self.translations = {"less than": "<"}
-        self.pkg_manager_api = LaunchpadVersionAPI()
 
     def _fetch(self):
         base_url = "https://people.canonical.com/~ubuntu-security/oval"
-        releases = self.config.releases
-        for i, release in enumerate(releases, 1):
+        releases = ["bionic", "trusty", "focal", "eoan", "xenial"]
+        for release in releases:
             file_url = f"{base_url}/com.ubuntu.{release}.cve.oval.xml.bz2"  # nopep8
             logger.info(f"Fetching Ubuntu Oval: {file_url}")
             response = requests.get(file_url)
@@ -61,8 +61,3 @@ def _fetch(self):
                 {"type": "deb", "namespace": "ubuntu"},
                 ET.ElementTree(ET.fromstring(extracted.decode("utf-8"))),
             )
-
-        logger.info(f"Fetched {i} Ubuntu Oval releases from {base_url}")
-
-    def set_api(self, packages):
-        asyncio.run(self.pkg_manager_api.load_api(packages))

vulnerabilities/improvers/__init__.py

@@ -23,13 +23,15 @@
 
 from vulnerabilities import importers
 from vulnerabilities.improvers import default
+from vulnerabilities.improvers import oval
 
 IMPROVERS_REGISTRY = [
     default.DefaultImprover,
     importers.nginx.NginxBasicImprover,
     importers.github.GitHubBasicImprover,
     importers.debian.DebianBasicImprover,
     importers.gitlab.GitLabBasicImprover,
+    oval.OvalBasicImprover,
 ]
 
 IMPROVERS_REGISTRY = {x.qualified_name: x for x in IMPROVERS_REGISTRY}