Merge pull request #188 from sbs2001/ubuntu_oval_rewrite

Ubuntu oval rewrite

Author: haikoschol

requirements.txt

@@ -1,3 +1,4 @@
+aiohttp==3.6.2
 asgiref==3.2.7
 attrs==19.3.0
 backcall==0.1.0

vulnerabilities/data_source.py

@@ -35,10 +35,13 @@
 from typing import Sequence
 from typing import Set
 from typing import Tuple
+import xml.etree.ElementTree as ET
 
 import pygit2
 from packageurl import PackageURL
 
+from vulnerabilities.oval_parser import OvalParser
+
 
 @dataclasses.dataclass
 class Advisory:
@@ -257,7 +260,8 @@ def file_changes(
 
             return {str(p) for p in path.glob(glob) if p.is_file()}, set()
 
-        return self._collect_file_changes(subdir=subdir, recursive=recursive, file_ext=file_ext)
+        return self._collect_file_changes(
+            subdir=subdir, recursive=recursive, file_ext=file_ext)
 
     def _collect_file_changes(
             self,
@@ -269,7 +273,8 @@ def _collect_file_changes(
         previous_commit = None
         added_files, updated_files = set(), set()
 
-        for commit in self._repo.walk(self._repo.head.target, pygit2.GIT_SORT_TIME):
+        for commit in self._repo.walk(
+                self._repo.head.target, pygit2.GIT_SORT_TIME):
             commit_time = commit.commit_time + commit.commit_time_offset  # convert to UTC
 
             if commit_time < self.cutoff_timestamp:
@@ -280,13 +285,16 @@ def _collect_file_changes(
                 continue
 
             for d in commit.tree.diff_to_tree(previous_commit.tree).deltas:
-                if not _include_file(d.new_file.path, subdir, recursive, file_ext) or d.is_binary:
+                if not _include_file(
+                        d.new_file.path, subdir, recursive, file_ext) or d.is_binary:
                     continue
 
-                abspath = os.path.join(self.config.working_directory, d.new_file.path)
+                abspath = os.path.join(
+                    self.config.working_directory, d.new_file.path)
                 # TODO
                 # Just filtering on the two status values for "added" and "modified" is too
-                # simplistic. This does not cover file renames, copies & deletions.
+                # simplistic. This does not cover file renames, copies &
+                # deletions.
                 if d.status == pygit2.GIT_DELTA_ADDED:
                     added_files.add(abspath)
                 elif d.status == pygit2.GIT_DELTA_MODIFIED:
@@ -381,3 +389,127 @@ def _include_file(
         match = match and path.endswith(f'.{file_ext}')
 
     return match
+
+
+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:
+        """
+        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)
+
+    @staticmethod
+    def _collect_pkgs(parsed_oval_data: Mapping) -> Set:
+        """
+        Helper method, used for loading the API. It expects data from
+        OvalParser.get_data().
+        """
+        all_pkgs = set()
+        for definition_data in parsed_oval_data:
+            for test_data in definition_data['test_data']:
+                for package in test_data['package_list']:
+                    all_pkgs.add(package)
+
+        return all_pkgs
+
+    def _fetch() -> Tuple[Mapping, Iterable[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: Mapping MUST INCLUDE "type" key. Example values of Mapping
+              {"type":"deb","qualifiers":{"distro":"buster"} }
+
+        """
+        raise NotImplementedError
+
+    def updated_advisories(self) -> List[Advisory]:
+        """
+        Note: metadata MUST INCLUDE "type" key, implement _fetch accordingly.
+        """
+        advisories = []
+        for metadata, oval_file in self._fetch():
+            advisories.extend(self.get_data_from_xml_doc(oval_file, metadata))
+        return self.batch_advisories(advisories)
+
+    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[Advisory]:
+        """
+        The orchestration method of the OvalDataSource. This method breaks an OVAL xml
+        ElementTree into a list of `Advisory`.
+
+        Note: pkg_metadata MUST INCLUDE "type" key. 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)
+        for definition_data in raw_data:  # definition_data -> Advisory
+
+            # 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()
+            urls = definition_data['reference_urls']
+
+            for test_data in definition_data['test_data']:
+                for package in test_data['package_list']:
+                    pkg_name = package
+                    aff_ver_range = test_data['version_ranges']
+                    all_versions = self.pkg_manager_api.get(package)
+                    # 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))
+                    if not all_versions:
+                        continue
+                    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)
+                        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)
+                        safe_purls.add(pkg_url)
+
+            all_adv.append(
+                Advisory(
+                    summary=description,
+                    impacted_package_urls=affected_purls,
+                    resolved_package_urls=safe_purls,
+                    cve_id=vuln_id,
+                    reference_urls=urls))
+        return all_adv

vulnerabilities/importers/__init__.py

@@ -20,10 +20,12 @@
 #  VulnerableCode is a free software code scanning tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
+
 from vulnerabilities.importers.alpine_linux import AlpineDataSource
 from vulnerabilities.importers.archlinux import ArchlinuxDataSource
 from vulnerabilities.importers.debian import DebianDataSource
 from vulnerabilities.importers.npm import NpmDataSource
 from vulnerabilities.importers.rust import RustDataSource
 from vulnerabilities.importers.safety_db import SafetyDbDataSource
 from vulnerabilities.importers.ruby import RubyDataSource
+from vulnerabilities.importers.ubuntu import UbuntuDataSource

vulnerabilities/importers/ubuntu.py

@@ -1,4 +1,3 @@
-#
 # Copyright (c) 2017 nexB Inc. and others. All rights reserved.
 # http://nexb.com and https://github.com/nexB/vulnerablecode/
 # The VulnerableCode software is licensed under the Apache License version 2.0.
@@ -21,36 +20,112 @@
 #  VulnerableCode is a free software code scanning tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
-from urllib.request import urlopen
 
-import bs4
+import asyncio
+import bz2
+import dataclasses
+from typing import Iterable
+from typing import List
+from typing import Mapping
+from typing import Set
+import xml.etree.ElementTree as ET
+
+
+from aiohttp import ClientSession
+from aiohttp.client_exceptions import ClientResponseError
+import requests
+
+
+from vulnerabilities.data_source import OvalDataSource, DataSourceConfiguration
+
+
+@dataclasses.dataclass
+class UbuntuConfiguration(DataSourceConfiguration):
+    releases: list
+    etags: dict
+
+
+class UbuntuDataSource(OvalDataSource):
+
+    CONFIG_CLASS = UbuntuConfiguration
+
+    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 = VersionAPI()
 
+    def _fetch(self):
+        base_url = 'https://people.canonical.com/~ubuntu-security/oval/'
+        file_name = 'com.ubuntu.{}.cve.oval.xml.bz2'
+        releases = self.config.releases
+        for release in releases:
+            file_url = base_url + file_name.format(release)
+            if not self.create_etag(file_url):
+                continue
+            resp = requests.get(file_url)
+            extracted = bz2.decompress(resp.content)
+            yield (
+                {'type': 'deb', 'namespace': 'ubuntu'},
+                ET.ElementTree(ET.fromstring(extracted.decode('utf-8')))
+            )
+        # 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 []
 
-UBUNTU_ROOT_URL = 'https://people.canonical.com/~ubuntu-security/cve/main.html'
+    def set_api(self, packages):
+        asyncio.run(self.pkg_manager_api.load_api(packages))
 
+    def create_etag(self, url):
 
-def extract_cves(html):
-    soup = bs4.BeautifulSoup(html, 'lxml')
+        etag = requests.head(url).headers.get('ETag')
+        if not etag:
+            # Kind of inaccurate to return True since etag is
+            # not created
+            return True
+        elif url in self.config.etags:
+            if self.config.etags[url] == etag:
+                return False
+        self.config.etags[url] = etag
+        return True
 
-    # Exclude the header row which has no class attribute
-    rows = soup.find_all('tr', attrs={'class': True})
 
-    cves = []
-    for row in rows:
-        columns = row.text.split()
-        cves.append({
-            'cve_id': columns[0],
-            'package_name': columns[1],
-            'vulnerability_status': row.get('class')[0],
-        })
+class VersionAPI:
+    def __init__(self, cache: Mapping[str, Set[str]] = None):
+        self.cache = cache or {}
 
-    return cves
+    def get(self, package_name: str) -> Set[str]:
+        return self.cache[package_name]
 
+    async def load_api(self, pkg_set):
+        async with ClientSession(raise_for_status=True) as session:
+            await asyncio.gather(*[self.set_api(pkg, session)
+                                   for pkg in pkg_set if pkg not in self.cache])
 
-def scrape_cves():
-    """
-    Runs the full scraping process of Ubuntu CVEs.
-    """
-    html = urlopen(UBUNTU_ROOT_URL).read()
-    cves = extract_cves(html)
-    return cves
+    async def set_api(self, pkg, session):
+        if pkg in self.cache:
+            return
+        url = ('https://api.launchpad.net/1.0/ubuntu/+archive/'
+               'primary?ws.op=getPublishedSources&'
+               'source_name={}&exact_match=true'.format(pkg))
+        try:
+            all_versions = set()
+            while True:
+                response = await session.request(method='GET', url=url)
+                resp_json = await response.json()
+                if resp_json['entries'] == []:
+                    self.cache[pkg] = {}
+                    break
+                for release in resp_json['entries']:
+                    all_versions.add(release['source_package_version'])
+                if resp_json.get('next_collection_link'):
+                    url = resp_json['next_collection_link']
+                else:
+                    break
+            self.cache[pkg] = all_versions
+        except (ClientResponseError, asyncio.exceptions.TimeoutError):
+            self.cache[pkg] = {}