Add istio improver

vulnerabilities/importers/istio.py

@@ -6,13 +6,20 @@
 # See https://github.com/nexB/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
+import logging
 import re
+from datetime import datetime
 from pathlib import Path
+from typing import Iterable
+from typing import List
+from typing import Mapping
+from typing import Optional
 from typing import Set
 
 import pytz
 import saneyaml
 from dateutil import parser
+from django.db.models.query import QuerySet
 from packageurl import PackageURL
 from univers.version_constraint import VersionConstraint
 from univers.version_range import GitHubVersionRange
@@ -23,10 +30,22 @@
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
+from vulnerabilities.importer import UnMergeablePackageError
+from vulnerabilities.improver import Improver
+from vulnerabilities.improver import Inference
+from vulnerabilities.models import Advisory
+from vulnerabilities.package_managers import GitHubTagsAPI
+from vulnerabilities.package_managers import VersionAPI
+from vulnerabilities.utils import AffectedPackage as LegacyAffectedPackage
+from vulnerabilities.utils import get_affected_packages_by_patched_package
+from vulnerabilities.utils import nearest_patched_package
+from vulnerabilities.utils import resolve_version_range
 from vulnerabilities.utils import split_markdown_front_matter
 
 is_release = re.compile(r"^[\d.]+$", re.IGNORECASE).match
 
+logger = logging.getLogger(__name__)
+
 
 class IstioImporter(Importer):
     spdx_license_expression = "Apache-2.0"
@@ -136,3 +155,70 @@ def get_data_from_md(self, path):
         with open(path) as f:
             front_matter, _ = split_markdown_front_matter(f.read())
             return saneyaml.load(front_matter)
+
+
+class IstioImprover(Improver):
+    def __init__(self) -> None:
+        self.versions_fetcher_by_purl: Mapping[str, VersionAPI] = {}
+        self.vesions_by_purl = {}
+
+    @property
+    def interesting_advisories(self) -> QuerySet:
+        return Advisory.objects.filter(created_by=IstioImporter.qualified_name)
+
+    def get_package_versions(
+        self, package_url: PackageURL, until: Optional[datetime] = None
+    ) -> List[str]:
+        """
+        Return a list of `valid_versions` for the `package_url`
+        """
+        api_name = "istio/istio"
+        versions_fetcher = GitHubTagsAPI()
+        return versions_fetcher.get_until(package_name=api_name, until=until).valid_versions
+
+    def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
+        """
+        Yield Inferences for the given advisory data
+        """
+        if not advisory_data.affected_packages:
+            return
+        for affected_package in advisory_data.affected_packages:
+            purl = affected_package.package
+            affected_version_range = affected_package.affected_version_range
+            pkg_type = purl.type
+            pkg_namespace = purl.namespace
+            pkg_name = purl.name
+            if not self.vesions_by_purl.get("istio/istio"):
+                valid_versions = self.get_package_versions(
+                    package_url=purl, until=advisory_data.date_published
+                )
+                self.vesions_by_purl["istio/istio"] = valid_versions
+            valid_versions = self.vesions_by_purl["istio/istio"]
+            aff_vers, unaff_vers = resolve_version_range(
+                affected_version_range=affected_version_range,
+                package_versions=valid_versions,
+            )
+            affected_purls = [
+                PackageURL(type=pkg_type, namespace=pkg_namespace, name=pkg_name, version=version)
+                for version in aff_vers
+            ]
+
+            unaffected_purls = [
+                PackageURL(type=pkg_type, namespace=pkg_namespace, name=pkg_name, version=version)
+                for version in unaff_vers
+            ]
+
+            affected_packages: List[LegacyAffectedPackage] = nearest_patched_package(
+                vulnerable_packages=affected_purls, resolved_packages=unaffected_purls
+            )
+
+            for (
+                fixed_package,
+                affected_packages,
+            ) in get_affected_packages_by_patched_package(affected_packages).items():
+                yield Inference.from_advisory_data(
+                    advisory_data,
+                    confidence=100,  # We are getting all valid versions to get this inference
+                    affected_purls=affected_packages,
+                    fixed_purl=fixed_package,
+                )

vulnerabilities/improvers/__init__.py

@@ -17,6 +17,7 @@
     importers.github.GitHubBasicImprover,
     importers.debian.DebianBasicImprover,
     importers.gitlab.GitLabBasicImprover,
+    importers.istio.IstioImprover,
     oval.DebianOvalBasicImprover,
     oval.UbuntuOvalBasicImprover,
 ]