Migrate debian importer to importer-improver model

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

Author: TG1999

requirements.txt

@@ -107,7 +107,7 @@ toml==0.10.2
 tomli==2.0.1
 traitlets==5.1.1
 typing_extensions==4.1.1
-univers==30.4.0
+univers==30.5.1
 urllib3==1.26.9
 wcwidth==0.2.5
 websocket-client==0.59.0

vulnerabilities/importer.py

@@ -150,7 +150,9 @@ def get_fixed_purl(self):
         return fixed_purl
 
     @classmethod
-    def merge(cls, affected_packages: Iterable):
+    def merge(
+        cls, affected_packages: Iterable
+    ) -> Tuple[PackageURL, List[VersionRange], List[Version]]:
         """
         Return a tuple with all attributes of AffectedPackage as a set
         for all values in the given iterable of AffectedPackage

vulnerabilities/importers/__init__.py

@@ -20,6 +20,7 @@
 #  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 debian
 from vulnerabilities.importers import github
 from vulnerabilities.importers import nginx
 from vulnerabilities.importers import nvd
@@ -31,6 +32,7 @@
     github.GitHubAPIImporter,
     nvd.NVDImporter,
     openssl.OpensslImporter,
+    debian.DebianImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/debian.py

@@ -21,115 +21,203 @@
 #  VulnerableCode is a free software tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
-import dataclasses
+import logging
 from typing import Any
+from typing import Iterable
 from typing import List
 from typing import Mapping
-from typing import Set
 
 import requests
-from dateutil import parser as dateparser
+from django.db.models.query import QuerySet
 from packageurl import PackageURL
+from univers.version_range import DebianVersionRange
+from univers.versions import DebianVersion
 
+from vulnerabilities.helpers import AffectedPackage as LegacyAffectedPackage
+from vulnerabilities.helpers import get_item
 from vulnerabilities.helpers import nearest_patched_package
 from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
+from vulnerabilities.importer import UnMergeablePackageError
+from vulnerabilities.improver import MAX_CONFIDENCE
+from vulnerabilities.improver import Improver
+from vulnerabilities.improver import Inference
+from vulnerabilities.models import Advisory
 
+logger = logging.getLogger(__name__)
 
-class DebianImporter(Importer):
-    def __enter__(self):
-        if self.response_is_new():
-            self._api_response = self._fetch()
+IGNORABLE_VERSIONS = {"3.8.20-4."}
 
-        else:
-            self._api_response = {}
 
-    def updated_advisories(self) -> Set[AdvisoryData]:
-        advisories = []
+class DebianImporter(Importer):
 
-        for pkg_name, records in self._api_response.items():
-            advisories.extend(self._parse(pkg_name, records))
+    spdx_license_expression = "MIT"
+    license_url = "https://www.debian.org/license"
 
-        return self.batch_advisories(advisories)
+    api_url = "https://security-tracker.debian.org/tracker/data/json"
 
-    def _fetch(self) -> Mapping[str, Any]:
-        return requests.get(self.config.debian_tracker_url).json()
+    def get_response(self):
+        return requests.get(self.api_url).json()
 
-    def _parse(self, pkg_name: str, records: Mapping[str, Any]) -> List[AdvisoryData]:
-        advisories = []
-        ignored_versions = {"3.8.20-4."}
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        response = self.get_response()
+        for pkg_name, records in response.items():
+            yield from self.parse(pkg_name, records)
 
+    def parse(
+        self, pkg_name: str, records: Mapping[str, Any], ignored_versions=IGNORABLE_VERSIONS
+    ) -> Iterable[AdvisoryData]:
         for cve_id, record in records.items():
-            impacted_purls, resolved_purls = [], []
+            affected_versions, fixed_versions = list(), list()
             if not cve_id.startswith("CVE"):
+                logger.error(f"Invalid CVE ID: {cve_id} in {record} in package {pkg_name}")
                 continue
 
             # vulnerabilities starting with something else may not be public yet
             # see for instance https://web.archive.org/web/20201215213725/https://security-tracker.debian.org/tracker/TEMP-0000000-A2EB44
             # TODO: this would need to be revisited though to ensure we are not missing out on anything
 
             for release_name, release_record in record["releases"].items():
-                if not release_record.get("repositories", {}).get(release_name):
-                    continue
+                version = get_item(release_record, "repositories", release_name)
 
-                version = release_record["repositories"][release_name]
+                if not version:
+                    logger.error(
+                        f"Version not found for {release_name} in {record} in package {pkg_name}"
+                    )
+                    continue
 
                 if version in ignored_versions:
+                    logger.error(f"Ignoring version {version} in {record} in package {pkg_name}")
                     continue
 
                 purl = PackageURL(
                     name=pkg_name,
                     type="deb",
                     namespace="debian",
-                    version=version,
                     qualifiers={"distro": release_name},
                 )
 
                 if release_record.get("status", "") == "resolved":
-                    resolved_purls.append(purl)
+                    fixed_versions.append(version)
                 else:
-                    impacted_purls.append(purl)
+                    affected_versions.append(version)
 
                 if (
                     "fixed_version" in release_record
                     and release_record["fixed_version"] not in ignored_versions
                 ):
-                    resolved_purls.append(
-                        PackageURL(
-                            name=pkg_name,
-                            type="deb",
-                            namespace="debian",
-                            version=release_record["fixed_version"],
-                            qualifiers={"distro": release_name},
-                        )
-                    )
+                    fixed_versions.append(release_record["fixed_version"])
 
             references = []
             debianbug = record.get("debianbug")
             if debianbug:
                 bug_url = f"https://bugs.debian.org/cgi-bin/bugreport.cgi?bug={debianbug}"
-                references.append(Reference(url=bug_url, reference_id=debianbug))
-            advisories.append(
-                AdvisoryData(
-                    vulnerability_id=cve_id,
-                    affected_packages=nearest_patched_package(impacted_purls, resolved_purls),
-                    summary=record.get("description", ""),
-                    references=references,
+                references.append(Reference(url=bug_url, reference_id=str(debianbug)))
+            affected_versions = list(dict.fromkeys(affected_versions))
+            fixed_versions = list(dict.fromkeys(fixed_versions))
+            affected_version_range = (
+                DebianVersionRange.from_versions(list(affected_versions))
+                if affected_versions
+                else None
+            )
+            affected_packages = []
+            for fixed_version in fixed_versions:
+                affected_packages.append(
+                    AffectedPackage(
+                        package=purl,
+                        affected_version_range=affected_version_range,
+                        fixed_version=DebianVersion(fixed_version),
+                    )
                 )
+            yield AdvisoryData(
+                aliases=[cve_id],
+                summary=record.get("description", ""),
+                affected_packages=affected_packages,
+                references=references,
             )
 
-        return advisories
 
-    def response_is_new(self):
+class DebianBasicImprover(Improver):
+    @property
+    def interesting_advisories(self) -> QuerySet:
+        return Advisory.objects.filter(created_by=DebianImporter.qualified_name)
+
+    def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
         """
-        Return True if a request response is for new data likely changed or
-        updated since we last checked.
+        Yield Inferences for the given advisory data
         """
-        head = requests.head(self.config.debian_tracker_url)
-        date_str = head.headers.get("last-modified")
-        last_modified_date = dateparser.parse(date_str)
-        if self.config.last_run_date:
-            return self.config.last_run_date < last_modified_date
+        if not advisory_data.affected_packages:
+            yield Inference(
+                aliases=advisory_data.aliases,
+                confidence=MAX_CONFIDENCE,
+                summary=advisory_data.summary,
+                affected_purls=[],
+                fixed_purl=None,
+                references=advisory_data.references,
+            )
+        else:
+            try:
+                purl, affected_version_ranges, fixed_versions = AffectedPackage.merge(
+                    advisory_data.affected_packages
+                )
+            except UnMergeablePackageError:
+                logger.error(
+                    f"Cannot merge with different purls {advisory_data.affected_packages!r}"
+                )
+                return iter([])
+
+            pkg_type = purl.type
+            pkg_namespace = purl.namespace
+            pkg_name = purl.name
+            unaffected_purls = [
+                PackageURL(
+                    type=pkg_type, namespace=pkg_namespace, name=pkg_name, version=str(version)
+                )
+                for version in fixed_versions
+            ]
+            if not affected_version_ranges:
+                for unaffected_purl in unaffected_purls:
+                    yield Inference.from_advisory_data(
+                        advisory_data,
+                        confidence=MAX_CONFIDENCE,  # We are getting all valid versions to get this inference
+                        affected_purls=[],
+                        fixed_purl=unaffected_purl,
+                    )
+            else:
+                aff_versions = set()
+                for affected_version_range in affected_version_ranges:
+                    for constraint in affected_version_range.constraints:
+                        aff_versions.add(constraint.version.string)
+                affected_purls = [
+                    PackageURL(
+                        type=pkg_type, namespace=pkg_namespace, name=pkg_name, version=version
+                    )
+                    for version in aff_versions
+                ]
+                affected_packages: List[LegacyAffectedPackage] = nearest_patched_package(
+                    vulnerable_packages=affected_purls, resolved_packages=unaffected_purls
+                )
 
-        return True
+                unique_patched_packages_with_affected_packages = {}
+                for package in affected_packages:
+                    if (
+                        package.patched_package
+                        not in unique_patched_packages_with_affected_packages
+                    ):
+                        unique_patched_packages_with_affected_packages[package.patched_package] = []
+                    unique_patched_packages_with_affected_packages[package.patched_package].append(
+                        package.vulnerable_package
+                    )
+
+                for (
+                    fixed_package,
+                    affected_packages,
+                ) in unique_patched_packages_with_affected_packages.items():
+                    yield Inference.from_advisory_data(
+                        advisory_data,
+                        confidence=MAX_CONFIDENCE,  # We are getting all valid versions to get this inference
+                        affected_purls=affected_packages,
+                        fixed_purl=fixed_package,
+                    )

vulnerabilities/improvers/__init__.py

@@ -28,6 +28,7 @@
     default.DefaultImprover,
     importers.nginx.NginxBasicImprover,
     importers.github.GitHubBasicImprover,
+    importers.debian.DebianBasicImprover,
 ]
 
 IMPROVERS_REGISTRY = {x.qualified_name: x for x in IMPROVERS_REGISTRY}