Revert "Harden Ubuntu OVAL importer"

This reverts commit 36f43d1.

Author: pombredanne

vulnerabilities/data_source.py

@@ -106,9 +106,9 @@ class DataSource(ContextManager):
     def __init__(
             self,
             batch_size: int,
-            last_run_date: Optional[datetime]=None,
-            cutoff_date: Optional[datetime]=None,
-            config: Optional[Mapping[str, Any]]=None,
+            last_run_date: Optional[datetime] = None,
+            cutoff_date: Optional[datetime] = None,
+            config: Optional[Mapping[str, Any]] = None,
     ):
         """
         Create a DataSource instance.
@@ -240,9 +240,9 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def file_changes(
             self,
-            subdir: str=None,
-            recursive: bool=False,
-            file_ext: Optional[str]=None,
+            subdir: str = None,
+            recursive: bool = False,
+            file_ext: Optional[str] = None,
     ) -> Tuple[Set[str], Set[str]]:
         """
         Returns all added and modified files since last_run_date or cutoff_date (whichever is more
@@ -382,9 +382,9 @@ def _update_from_remote(self, remote, branch) -> None:
 
 def _include_file(
         path: str,
-        subdir: Optional[str]=None,
-        recursive: bool=False,
-        file_ext: Optional[str]=None,
+        subdir: Optional[str] = None,
+        recursive: bool = False,
+        file_ext: Optional[str] = None,
 ) -> bool:
     match = True
 
@@ -408,7 +408,6 @@ class OvalDataSource(DataSource):
     All data sources which collect data from OVAL files must inherit from this
     `OvalDataSource` class. Subclasses must implement the methods `_fetch` and `set_api`.
     """
-
     @staticmethod
     def create_purl(pkg_name: str, pkg_version: str, pkg_data: Mapping) -> PackageURL:
         """
@@ -434,48 +433,29 @@ def _collect_pkgs(parsed_oval_data: Mapping) -> Set:
 
     def _fetch(self) -> Tuple[Mapping, Iterable[ET.ElementTree]]:
         """
-        Return a two-tuple of ({mapping of Package URL data}, it's ET.ElementTree)
+        This method  contains logic to fetch OVAL files and yield them into
+        a tuple of file's metadata and it's ET.ElementTree.
         Subclasses must implement this method.
 
-        Note:  Package URL data MUST INCLUDE a Package URL "type" key so
-        implement _fetch() accordingly.
-        For example::
-
+        Note: Mapping MUST INCLUDE "type" key. Example values of Mapping
               {"type":"deb","qualifiers":{"distro":"buster"} }
+
         """
-        # TODO: enforce that we receive the proper data here
         raise NotImplementedError
 
     def updated_advisories(self) -> List[Advisory]:
-        for purl_data, oval_etree in self._fetch():
-            if 'type' not in purl_data:
-                ets = (oval_etree and ET.tostring(oval_etree)) or 'NO DATA'
-                msg = (
-                    "Failed to get updated_advisories for Ubuntu: purl_data is "
-                    f"missing a package type {purl_data!r}\n"
-                    f"with OVAL XML:\n"
-                    f"{ets}\n"
-                    f"... continuing..."
-                )
-                print(msg)
-                logger.error(msg)
-                continue
-
+        """
+        Note: metadata MUST INCLUDE "type" key, implement _fetch accordingly.
+        """
+        for metadata, oval_file in self._fetch():
             try:
-                oval_data = self.get_data_from_xml_doc(oval_etree, purl_data)
+                oval_data = self.get_data_from_xml_doc(oval_file, metadata)
                 yield oval_data
             except Exception:
-                ets = (oval_etree and ET.tostring(oval_etree)) or 'NO DATA'
-                tb = traceback.format_exc()
-                msg = (
-                    f"Failed to get updated_advisories for Ubuntu:"
-                    f"with {purl_data!r}\n"
-                    f"and with OVAL XML:\n"
-                    f"{ets}\n{tb}\n"
-                    f"... continuing..."
+                logger.error(
+                    f"Failed to get updated_advisories: {oval_file!r} "
+                    "with {metadata!r}:\n" + traceback.format_exc()
                 )
-                print(msg)
-                logger.error(msg)
                 continue
 
     def set_api(self, all_pkgs: Iterable[str]):
@@ -493,89 +473,57 @@ def set_api(self, all_pkgs: Iterable[str]):
 
     def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> List[Advisory]:
         """
-        The orchestration method of the OvalDataSource. This method breaks an
-        OVAL xml ElementTree into a list of `Advisory`.
+        The orchestration method of the OvalDataSource. This method breaks an OVAL xml
+        ElementTree into a list of `Advisory`.
 
-        Note: pkg_metadata is a mapping of Package URL data that MUST INCLUDE
-        "type" key.
-
-        Example value of pkg_metadata:
+        Note: pkg_metadata MUST INCLUDE "type" key. Example value of pkg_metadata,
               {"type":"deb","qualifiers":{"distro":"buster"} }
         """
-        if 'type' not in pkg_metadata:
-            ets = xml_doc and ET.tostring(xml_doc) or 'NO DATA'
-            msg = (
-                "Failed to get_data_from_xml_doc: pkg_metadata is "
-                f"missing a package type {pkg_metadata!r}\n"
-                f"with OVAL XML:\n"
-                f"{ets}"
-            )
-            print(msg)
-            logger.error(msg)
-            raise Exception(msg)
-
         all_adv = []
         oval_doc = OvalParser(self.translations, xml_doc)
-        try:
-            raw_data = oval_doc.get_data()
-        except Exception:
-            ets = xml_doc and ET.tostring(xml_doc) or 'NO DATA'
-            tb = traceback.format_exc()
-            msg = (
-                f"Failed to get_data_from_xml_doc:"
-                f"with {pkg_metadata!r}\n"
-                f"and with OVAL XML:\n"
-                f"{ets}\n{tb}"
-            )
-            print(msg)
-            logger.error(msg)
-            raise Exception(msg)
-
+        raw_data = oval_doc.get_data()
         all_pkgs = self._collect_pkgs(raw_data)
         self.set_api(all_pkgs)
+        for definition_data in raw_data:  # definition_data -> Advisory
 
-        # convert definition_data to Advisory objects
-        for definition_data in raw_data:
-            # These fields are definition level, i.e common for all elements
-            # connected/linked to an OvalDefinition
+            # These fields are definition level, i.e common for all
+            # elements connected/linked to an OvalDefinition
             vuln_id = definition_data['vuln_id']
             description = definition_data['description']
             affected_purls = set()
             safe_purls = set()
-            references = [Reference(url=url) for url in definition_data['reference_urls']]
+            references = [Reference(url=url)
+                          for url in definition_data['reference_urls']]
+
             for test_data in definition_data['test_data']:
                 for package in test_data['package_list']:
                     pkg_name = package
                     if package and len(pkg_name) >= 50:
                         continue
-                    aff_ver_range = test_data['version_ranges'] or set()
+                    aff_ver_range = test_data['version_ranges']
                     all_versions = self.pkg_manager_api.get(package)
-
-                    # 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 = set(filter(lambda x: len(x) < 50, all_versions))
+                    all_versions = set(
+                        filter(
+                            lambda x: len(x) < 50,
+                            all_versions))
                     if not all_versions:
                         continue
-
-                    affected_versions = set(filter(lambda x: x in aff_ver_range, all_versions))
+                    affected_versions = set(
+                        filter(
+                            lambda x: x in aff_ver_range,
+                            all_versions))
                     safe_versions = all_versions - affected_versions
 
                     for version in affected_versions:
                         pkg_url = self.create_purl(
-                            pkg_name=pkg_name,
-                            pkg_version=version,
-                            pkg_data=pkg_metadata,
-                        )
+                            pkg_name=pkg_name, pkg_version=version, pkg_data=pkg_metadata)
                         affected_purls.add(pkg_url)
 
                     for version in safe_versions:
                         pkg_url = self.create_purl(
-                            pkg_name=pkg_name,
-                            pkg_version=version,
-                            pkg_data=pkg_metadata,
-                        )
+                            pkg_name=pkg_name, pkg_version=version, pkg_data=pkg_metadata)
                         safe_purls.add(pkg_url)
 
             all_adv.append(
@@ -584,8 +532,5 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
                     impacted_package_urls=affected_purls,
                     resolved_package_urls=safe_purls,
                     cve_id=vuln_id,
-                    vuln_references=references,
-                ))
-
-        print(f"Processed {len(all_adv)} Advisory from Oval data for {pkg_metadata}")
+                    vuln_references=references))
         return all_adv

vulnerabilities/importers/ubuntu.py

@@ -59,25 +59,21 @@ def __init__(self, *args, **kwargs):
 
     def _fetch(self):
         releases = self.config.releases
-        for i, release in enumerate(releases, 1):
+        for release in releases:
             file_url = f"https://people.canonical.com/~ubuntu-security/oval/com.ubuntu.{release}.cve.oval.xml.bz2"  # nopep8
-#             if not create_etag(data_src=self, url=file_url, etag_key="ETag"):
-#                 print(f"Ubuntu Oval Etag not changed, not re-fetching: {file_url}")
-#                 continue
-
-            print(f"Fetching Ubuntu Oval: {file_url}")
-            response = requests.get(file_url)
-            if response.status_code != requests.codes.ok:
-                print(f"Failed to fetch Ubuntu Oval: HTTP {response.status_code} : {file_url}")
+            if not create_etag(data_src=self, url=file_url, etag_key="ETag"):
                 continue
-
-            extracted = bz2.decompress(response.content)
+            resp = requests.get(file_url)
+            extracted = bz2.decompress(resp.content)
             yield (
                 {"type": "deb", "namespace": "ubuntu"},
                 ET.ElementTree(ET.fromstring(extracted.decode("utf-8"))),
             )
-
-        print(f"Fetched {i} Ubuntu Oval releases https://people.canonical.com/~ubuntu-security/oval/")
+        # In case every file is latest, _fetch won't yield anything(due to checking for new etags),
+        # this would return None to added_advisories
+        # which will cause error, hence this
+        # function return an empty list
+        return []
 
     def set_api(self, packages):
         asyncio.run(self.pkg_manager_api.load_api(packages))

vulnerabilities/oval_parser.py

@@ -30,12 +30,8 @@
 
 from dephell_specifier import RangeSpecifier
 
-from vulnerabilities.lib_oval import OvalDefinition
-from vulnerabilities.lib_oval import OvalDocument
-from vulnerabilities.lib_oval import OvalObject
-from vulnerabilities.lib_oval import OvalState
-from vulnerabilities.lib_oval import OvalTest
-import traceback
+from vulnerabilities.lib_oval import (
+    OvalDefinition, OvalDocument, OvalTest, OvalObject, OvalState)
 
 
 class OvalParser:
@@ -49,7 +45,8 @@ def __init__(self, translations: Dict, oval_document: ET.ElementTree):
 
     def get_data(self) -> List[Dict]:
         """
-        Return a list of OvalDefinition mappings.
+        This is the orchestration method, it returns a list of dictionaries,
+        where each dictionary represents data from an OvalDefinition
         """
         oval_data = []
         for definition in self.all_definitions:
@@ -58,22 +55,27 @@ def get_data(self) -> List[Dict]:
             if not matching_tests:
                 continue
             definition_data = {'test_data': []}
-            # TODO:this could use some data cleaning
-            definition_data['description'] = definition.getMetadata().getDescription() or ''
+            definition_data['description'] = definition.getMetadata(
+            ).getDescription()  # this could use some data cleaning
 
-            definition_data['vuln_id'] = self.get_vuln_id_from_definition(definition)
-            definition_data['reference_urls'] = self.get_urls_from_definition(definition)
+            if not definition_data['description']:
+                definition_data['description'] = ''
 
+            definition_data['vuln_id'] = self.get_vuln_id_from_definition(
+                definition)
+            definition_data['reference_urls'] = self.get_urls_from_definition(
+                definition
+            )
             for test in matching_tests:
                 test_obj, test_state = self.get_object_state_of_test(test)
                 if not test_obj or not test_state:
                     continue
                 test_data = {'package_list': []}
-                test_data['package_list'].extend(self.get_pkgs_from_obj(test_obj))
-                version_ranges = self.get_versionsrngs_from_state(test_state)
-                test_data['version_ranges'] = version_ranges
+                test_data['package_list'].extend(
+                    self.get_pkgs_from_obj(test_obj))
+                test_data['version_ranges'] = self.get_versionsrngs_from_state(
+                    test_state)
                 definition_data['test_data'].append(test_data)
-
             oval_data.append(definition_data)
 
         return oval_data
@@ -136,27 +138,18 @@ def get_pkgs_from_obj(self, obj: OvalObject) -> List[str]:
 
         return pkg_list
 
-    # TODO: this method needs a better name
     def get_versionsrngs_from_state(self, state: OvalState) -> Optional[RangeSpecifier]:
         """
-        Return a version range(s)? from a state
+        returns  all related version ranges within a state
         """
         for var in state.element:
-            operation = var.get('operation')
-            if not operation:
-                continue
-            operand = self.translations.get(operation) or ''
-            if not operand:
-                continue
-            version = var.text or ''
-            if not version:
-                continue
-            version_range = operand + version
-            try:
+            if var.get('operation'):
+                if var.get('operation') not in self.translations:
+                    continue
+                operand = self.translations[var.get('operation')]
+                version = var.text
+                version_range = operand + version
                 return RangeSpecifier(version_range)
-            except Exception:
-                # FIXME: we should not continue
-                print(f"Failed to process invalid version_range in OvalState: {version_range}...continuing")
 
     @staticmethod
     def get_urls_from_definition(definition: OvalDefinition) -> Set[str]: