Migrate debian-oval and ubuntu importer

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

Author: TG1999

vulnerabilities/importer.py

@@ -36,16 +36,17 @@
 from typing import Tuple
 
 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
 
 from vulnerabilities.helpers import classproperty
 from vulnerabilities.helpers import evolve_purl
-from vulnerabilities.helpers import nearest_patched_package
 from vulnerabilities.oval_parser import OvalParser
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 from vulnerabilities.severity_systems import ScoringSystem
@@ -510,13 +511,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:
@@ -550,28 +551,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`.
@@ -584,10 +574,11 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
         """
 
         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()
+        # all_pkgs = self._collect_pkgs(raw_data)
 
         # convert definition_data to Advisory objects
         for definition_data in raw_data:
@@ -599,49 +590,28 @@ 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,
-                )
+            yield AdvisoryData(
+                aliases=[vuln_id],
+                summary=description,
+                affected_packages=affected_packages,
+                references=references,
+                date_published=dateparser.parse(timestamp),
             )
-        return all_adv

vulnerabilities/importers/__init__.py

@@ -21,12 +21,14 @@
 #  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 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,
@@ -37,6 +39,8 @@
     redhat.RedhatImporter,
     pysec.PyPIImporter,
     debian.DebianImporter,
+    debian_oval.DebianOvalImporter,
+    ubuntu.UbuntuImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/debian_oval.py

@@ -21,7 +21,6 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 
-import asyncio
 import xml.etree.ElementTree as ET
 
 import requests
@@ -32,16 +31,17 @@
 
 
 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"]
         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"):
@@ -53,6 +53,3 @@ def _fetch(self):
                 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/github.py

@@ -50,7 +50,9 @@
 from vulnerabilities.improver import Inference
 from vulnerabilities.models import Advisory
 from vulnerabilities.package_managers import ComposerVersionAPI
+from vulnerabilities.package_managers import DebianVersionAPI
 from vulnerabilities.package_managers import GoproxyVersionAPI
+from vulnerabilities.package_managers import LaunchpadVersionAPI
 from vulnerabilities.package_managers import MavenVersionAPI
 from vulnerabilities.package_managers import NugetVersionAPI
 from vulnerabilities.package_managers import PypiVersionAPI
@@ -286,7 +288,7 @@ def get_api_package_name(purl: PackageURL) -> str:
     if purl.type == "composer":
         return f"{purl.namespace}/{purl.name}"
 
-    if purl.type in ("nuget", "pypi", "gem", "golang"):
+    if purl.type in ("nuget", "pypi", "gem", "golang", "deb"):
         return purl.name
 
     logger.error(f"get_api_package_name: Unknown PURL {purl!r}")

vulnerabilities/importers/ubuntu.py

@@ -21,7 +21,6 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 
-import asyncio
 import bz2
 import logging
 import xml.etree.ElementTree as ET
@@ -35,17 +34,19 @@
 
 
 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
+        releases = ["bionic", "trusty", "focal", "eoan", "xenial"]
         for i, release in enumerate(releases, 1):
             file_url = f"{base_url}/com.ubuntu.{release}.cve.oval.xml.bz2"  # nopep8
             logger.info(f"Fetching Ubuntu Oval: {file_url}")
@@ -63,6 +64,3 @@ def _fetch(self):
             )
 
         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,12 +23,14 @@
 
 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,
+    oval.OvalBasicImprover,
 ]
 
 IMPROVERS_REGISTRY = {x.qualified_name: x for x in IMPROVERS_REGISTRY}