Add improver to get valid versions for stored ranges

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

Author: TG1999

vulnerabilities/importer.py

@@ -247,6 +247,7 @@ class AdvisoryData:
     references: List[Reference] = dataclasses.field(default_factory=list)
     date_published: Optional[datetime.datetime] = None
     weaknesses: List[int] = dataclasses.field(default_factory=list)
+    created_by: Optional[str] = None
 
     def __post_init__(self):
         if self.date_published and not self.date_published.tzinfo:
@@ -260,22 +261,27 @@ def to_dict(self):
             "references": [ref.to_dict() for ref in self.references],
             "date_published": self.date_published.isoformat() if self.date_published else None,
             "weaknesses": self.weaknesses,
+            "created_by": self.created_by or "",
         }
 
     @classmethod
     def from_dict(cls, advisory_data):
         date_published = advisory_data["date_published"]
         transformed = {
-            "aliases": advisory_data["aliases"],
-            "summary": advisory_data["summary"],
+            "aliases": advisory_data.get("aliases") or [],
+            "summary": advisory_data.get("summary") or "",
             "affected_packages": [
-                AffectedPackage.from_dict(pkg) for pkg in advisory_data["affected_packages"]
+                AffectedPackage.from_dict(pkg)
+                for pkg in advisory_data.get("affected_packages") or []
+            ],
+            "references": [
+                Reference.from_dict(ref) for ref in advisory_data.get("references") or []
             ],
-            "references": [Reference.from_dict(ref) for ref in advisory_data["references"]],
             "date_published": datetime.datetime.fromisoformat(date_published)
             if date_published
             else None,
-            "weaknesses": advisory_data["weaknesses"],
+            "weaknesses": advisory_data.get("weaknesses") or [],
+            "created_by": advisory_data.get("created_by") or "",
         }
         return cls(**transformed)
 

vulnerabilities/importers/apache_httpd.py

@@ -9,15 +9,9 @@
 
 import logging
 import urllib
-from datetime import datetime
-from typing import Iterable
-from typing import List
-from typing import Mapping
-from typing import Optional
 
 import requests
 from bs4 import BeautifulSoup
-from django.db.models.query import QuerySet
 from packageurl import PackageURL
 from univers.version_constraint import VersionConstraint
 from univers.version_range import ApacheVersionRange
@@ -27,18 +21,8 @@
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
-from vulnerabilities.importer import UnMergeablePackageError
 from vulnerabilities.importer import VulnerabilitySeverity
-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.severity_systems import APACHE_HTTPD
-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
 
 logger = logging.getLogger(__name__)
 
@@ -227,80 +211,3 @@ def fetch_links(url):
     "post_ajp_proxy",
     "pre_ajp_proxy",
 }
-
-
-class ApacheHTTPDImprover(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=ApacheHTTPDImporter.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 = "apache/httpd"
-        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
-        try:
-            purl, affected_version_ranges, _ = 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
-
-        if not self.vesions_by_purl.get(str(purl)):
-            valid_versions = self.get_package_versions(
-                package_url=purl, until=advisory_data.date_published
-            )
-            self.vesions_by_purl[str(purl)] = valid_versions
-
-        valid_versions = self.vesions_by_purl[str(purl)]
-
-        for affected_version_range in affected_version_ranges:
-            aff_vers, unaff_vers = resolve_version_range(
-                affected_version_range=affected_version_range,
-                package_versions=valid_versions,
-                ignorable_versions=IGNORE_TAGS,
-            )
-            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/importers/debian.py

@@ -164,74 +164,3 @@ def parse(self, pkg_name: str, records: Mapping[str, Any]) -> Iterable[AdvisoryD
                 affected_packages=affected_packages,
                 references=references,
             )
-
-
-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]:
-        """
-        Yield Inferences for the given advisory data
-        """
-        if not advisory_data.affected_packages:
-            return
-        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
-
-        pkg_type = purl.type
-        pkg_namespace = purl.namespace
-        pkg_name = purl.name
-        pkg_qualifiers = purl.qualifiers
-        fixed_purls = [
-            PackageURL(
-                type=pkg_type,
-                namespace=pkg_namespace,
-                name=pkg_name,
-                version=str(version),
-                qualifiers=pkg_qualifiers,
-            )
-            for version in fixed_versions
-        ]
-        if not affected_version_ranges:
-            for fixed_purl in fixed_purls:
-                yield Inference.from_advisory_data(
-                    advisory_data,  # We are getting all valid versions to get this inference
-                    confidence=MAX_CONFIDENCE,
-                    affected_purls=[],
-                    fixed_purl=fixed_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,
-                    qualifiers=pkg_qualifiers,
-                )
-                for version in aff_versions
-            ]
-            affected_packages: List[LegacyAffectedPackage] = nearest_patched_package(
-                vulnerable_packages=affected_purls, resolved_packages=fixed_purls
-            )
-
-            for (fixed_package, affected_packages,) in get_affected_packages_by_patched_package(
-                affected_packages=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/importers/github.py

@@ -8,14 +8,10 @@
 #
 
 import logging
-from datetime import datetime
 from typing import Iterable
-from typing import List
-from typing import Mapping
 from typing import Optional
 
 from dateutil import parser as dateparser
-from django.db.models.query import QuerySet
 from packageurl import PackageURL
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
 from univers.version_range import build_range_from_github_advisory_constraint
@@ -26,21 +22,9 @@
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
-from vulnerabilities.importer import UnMergeablePackageError
 from vulnerabilities.importer import VulnerabilitySeverity
-from vulnerabilities.improver import Improver
-from vulnerabilities.improver import Inference
-from vulnerabilities.models import Advisory
-from vulnerabilities.package_managers import VERSION_API_CLASSES_BY_PACKAGE_TYPE
-from vulnerabilities.package_managers import GoproxyVersionAPI
-from vulnerabilities.package_managers import VersionAPI
-from vulnerabilities.package_managers import get_api_package_name
-from vulnerabilities.utils import AffectedPackage as LegacyAffectedPackage
 from vulnerabilities.utils import dedupe
-from vulnerabilities.utils import get_affected_packages_by_patched_package
 from vulnerabilities.utils import get_item
-from vulnerabilities.utils import nearest_patched_package
-from vulnerabilities.utils import resolve_version_range
 
 logger = logging.getLogger(__name__)
 
@@ -319,92 +303,3 @@ def process_response(resp: dict, package_type: str) -> Iterable[AdvisoryData]:
             affected_packages=affected_packages,
             date_published=date_published,
         )
-
-
-class GitHubBasicImprover(Improver):
-    def __init__(self) -> None:
-        self.versions_fetcher_by_purl: Mapping[str, VersionAPI] = {}
-
-    @property
-    def interesting_advisories(self) -> QuerySet:
-        return Advisory.objects.filter(created_by=GitHubAPIImporter.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 = get_api_package_name(package_url)
-        if not api_name:
-            logger.error(f"Could not get versions for {package_url!r}")
-            return []
-        versions_fetcher = self.versions_fetcher_by_purl.get(package_url)
-        if not versions_fetcher:
-            versions_fetcher: VersionAPI = VERSION_API_CLASSES_BY_PACKAGE_TYPE[package_url.type]
-            self.versions_fetcher_by_purl[package_url] = versions_fetcher()
-
-        versions_fetcher = self.versions_fetcher_by_purl[package_url]
-
-        self.versions_fetcher_by_purl[package_url] = versions_fetcher
-        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
-        try:
-            purl, affected_version_ranges, _ = 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
-        if purl.type == "golang":
-            # Problem with the Golang and Go that they provide full path
-            # FIXME: We need to get the PURL subpath for Go module
-            versions_fetcher = self.versions_fetcher_by_purl.get(purl)
-            if not versions_fetcher:
-                versions_fetcher = GoproxyVersionAPI()
-                self.versions_fetcher_by_purl[purl] = versions_fetcher
-            pkg_name = versions_fetcher.module_name_by_package_name.get(pkg_name, pkg_name)
-
-        valid_versions = self.get_package_versions(
-            package_url=purl, until=advisory_data.date_published
-        )
-        for affected_version_range in affected_version_ranges:
-            aff_vers, unaff_vers = resolve_version_range(
-                affected_version_range=affected_version_range,
-                package_versions=valid_versions,
-                ignorable_versions=WEIRD_IGNORABLE_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,
-                )