Migrate github importer to importer improver model

requirements.txt

@@ -8,7 +8,7 @@ django-widget-tweaks>=1.4.8
 packageurl-python>=0.9.4
 binaryornot>=0.4.4
 GitPython>=3.1.17
-univers>=30.1.0
+univers>=30.3.1
 saneyaml>=0.5.2
 beautifulsoup4>=4.9.3
 python-dateutil>=2.8.1

vulnerabilities/helpers.py

@@ -23,7 +23,9 @@
 import bisect
 import dataclasses
 import json
+import logging
 import re
+from functools import total_ordering
 from typing import List
 from typing import Optional
 from typing import Tuple
@@ -34,8 +36,9 @@
 import toml
 import urllib3
 from packageurl import PackageURL
+from univers.version_range import RANGE_CLASS_BY_SCHEMES
 
-# TODO add logging here
+LOGGER = logging.getLogger(__name__)
 
 cve_regex = re.compile(r"CVE-\d{4}-\d{4,7}", re.IGNORECASE)
 is_cve = cve_regex.match
@@ -133,32 +136,36 @@ def requests_with_5xx_retry(max_retries=5, backoff_factor=0.5):
     return session
 
 
-def nearest_patched_package(
-    vulnerable_packages: List[PackageURL], resolved_packages: List[PackageURL]
-) -> List[AffectedPackage]:
-    class PackageURLWithVersionComparator:
-        """
-        This class is used to  get around bisect module's lack of supplying custom
-        compartor. Get rid of this once we use python 3.10 which supports this.
-        See https://github.com/python/cpython/pull/20556
-        """
+@total_ordering
+class VersionedPackage:
+    """
+    A PackageURL with a Version class.
+    This class is used to  get around bisect module's lack of supplying custom
+    comparator. Get rid of this once we use python 3.10 which supports this.
+    See https://github.com/python/cpython/pull/20556
+    """
 
-        def __init__(self, package):
-            self.package = package
-            self.version_object = version_class_by_package_type[package.type](package.version)
+    def __init__(self, purl: PackageURL):
+        self.purl = purl
+        vrc = RANGE_CLASS_BY_SCHEMES.get(purl.type)
+        self.version = vrc.version_class(purl.version)
 
-        def __eq__(self, other):
-            return self.version_object == other.version_object
+    def __eq__(self, other):
+        return self.version == other.version
 
-        def __lt__(self, other):
-            return self.version_object < other.version_object
+    def __lt__(self, other):
+        return self.version < other.version
 
-    vulnerable_packages = sorted(
-        [PackageURLWithVersionComparator(package) for package in vulnerable_packages]
-    )
-    resolved_packages = sorted(
-        [PackageURLWithVersionComparator(package) for package in resolved_packages]
-    )
+
+def nearest_patched_package(
+    vulnerable_packages: List[PackageURL], resolved_packages: List[PackageURL]
+) -> List[AffectedPackage]:
+    """
+    Return a list of Affected Packages for each Patched package.
+    """
+
+    vulnerable_packages = sorted([VersionedPackage(package) for package in vulnerable_packages])
+    resolved_packages = sorted([VersionedPackage(package) for package in resolved_packages])
 
     resolved_package_count = len(resolved_packages)
     affected_package_with_patched_package_objects = []
@@ -167,11 +174,11 @@ def __lt__(self, other):
         patched_package_index = bisect.bisect_right(resolved_packages, vulnerable_package)
         patched_package = None
         if patched_package_index < resolved_package_count:
-            patched_package = resolved_packages[patched_package_index].package
+            patched_package = resolved_packages[patched_package_index]
 
         affected_package_with_patched_package_objects.append(
             AffectedPackage(
-                vulnerable_package=vulnerable_package.package, patched_package=patched_package
+                vulnerable_package=vulnerable_package.purl, patched_package=patched_package.purl
             )
         )
 
@@ -211,3 +218,26 @@ def __init__(self, fget):
 
     def __get__(self, owner_self, owner_cls):
         return self.fget(owner_cls)
+
+
+def get_item(object: dict, *attributes):
+    """
+    Return `item` by going through all the `attributes` present in the `json_object`
+
+    Do a DFS for the `item` in the `json_object` by traversing the `attributes`
+    and return None if can not traverse through the `attributes`
+    For example:
+    >>> get_item({'a': {'b': {'c': 'd'}}}, 'a', 'b', 'c')
+    'd'
+    >>> assert(get_item({'a': {'b': {'c': 'd'}}}, 'a', 'b', 'e')) == None
+    """
+    if not object:
+        LOGGER.error(f"Object is empty: {object}")
+        return
+    item = object
+    for attribute in attributes:
+        if attribute not in item:
+            LOGGER.error(f"Missing attribute {attribute} in {item}")
+            return None
+        item = item[attribute]
+    return item

vulnerabilities/importer.py

@@ -113,6 +113,18 @@ def from_dict(cls, ref: dict):
         )
 
 
+class UnMergeablePackageError(Exception):
+    """
+    Raised when a package cannot be merged with another one.
+    """
+
+
+class NoAffectedPackages(Exception):
+    """
+    Raised when there were no affected packages found.
+    """
+
+
 @dataclasses.dataclass(order=True, frozen=True)
 class AffectedPackage:
     """
@@ -155,18 +167,23 @@ def merge(cls, affected_packages: Iterable):
             affected_version_range: set(VersionRange)
             fixed_versions: set(Version)
         """
-        affected_version_ranges = set()
-        fixed_versions = set()
+        affected_packages = list(affected_packages)
+        if not affected_packages:
+            raise NoAffectedPackages("No affected packages found")
+        affected_version_ranges = list()
+        fixed_versions = list()
         purls = set()
         for pkg in affected_packages:
             if pkg.affected_version_range:
-                affected_version_ranges.add(pkg.affected_version_range)
+                if pkg.affected_version_range not in affected_version_ranges:
+                    affected_version_ranges.append(pkg.affected_version_range)
             if pkg.fixed_version:
-                fixed_versions.add(pkg.fixed_version)
+                if pkg.fixed_version not in fixed_versions:
+                    fixed_versions.append(pkg.fixed_version)
             purls.add(pkg.package)
         if len(purls) > 1:
-            raise TypeError("Cannot merge with different purls", purls)
-        return purls.pop(), affected_version_ranges, fixed_versions
+            raise UnMergeablePackageError("Cannot merge with different purls", purls)
+        return purls.pop(), sorted(affected_version_ranges), sorted(fixed_versions)
 
     def to_dict(self):
         """
@@ -230,6 +247,15 @@ def __post_init__(self):
         if self.date_published and not self.date_published.tzinfo:
             logger.warn(f"AdvisoryData with no tzinfo: {self!r}")
 
+    def to_dict(self):
+        return {
+            "aliases": self.aliases,
+            "summary": self.summary,
+            "affected_packages": [pkg.to_dict() for pkg in self.affected_packages],
+            "references": [ref.to_dict() for ref in self.references],
+            "date_published": self.date_published.isoformat() if self.date_published else None,
+        }
+
 
 class NoLicenseError(Exception):
     pass

vulnerabilities/importers/__init__.py

@@ -20,8 +20,9 @@
 #  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 import alpine_linux
+from vulnerabilities.importers import github
 from vulnerabilities.importers import nginx
 
-IMPORTERS_REGISTRY = [nginx.NginxImporter, alpine_linux.AlpineImporter]
+IMPORTERS_REGISTRY = [nginx.NginxImporter, alpine_linux.AlpineImporter, github.GitHubAPIImporter]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}