Migrate apache httpd importer#971

CHANGELOG.rst

@@ -2,10 +2,16 @@ Release notes
 =============
 
 
+Version v31.1.1
+---------------
+
+- We re-enabled support for the Apache HTTPD security advisories importer.
+
+
 Version v31.1.0
 ----------------
 
-- We re-enabled support for the NPM vulnerabilities advisories importer. 
+- We re-enabled support for the NPM vulnerabilities advisories importer.
 - We re-enabled support for the Retiredotnet vulnerabilities advisories importer.
 - We are now handling purl fragments in package search. For example:
   you can now serch using queries in the UI like this : ``cherrypy@2.1.1``,
@@ -30,7 +36,7 @@ Version v31.0.0
 - We made bulk search faster by pre-computing `package_url` and
   `plain_package_url` in Package model.  And provided two options in package
   bulk search  ``purl_only`` option to get only vulnerable purls without any
-  extra details, ``plain_purl`` option to filter purls without qualifiers and 
+  extra details, ``plain_purl`` option to filter purls without qualifiers and
   subpath and also return them without qualifiers and subpath. The names used
   are provisional and may be updated in a future release.
 

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.9.0
+univers==30.9.1
 urllib3==1.26.9
 wcwidth==0.2.5
 websocket-client==0.59.0
@@ -120,4 +120,4 @@ drf-spectacular-sidecar==2022.10.1
 drf-spectacular==0.24.2
 coreapi==2.3.3
 coreschema==0.0.4
-itypes==1.2.0
+itypes==1.2.0

setup.cfg

@@ -70,7 +70,7 @@ install_requires =
 
     #essentials
     packageurl-python>=0.10.5rc1
-    univers>=30.9.0
+    univers>=30.9.1
     license-expression>=21.6.14
 
     # file and data formats

vulnerabilities/importers/__init__.py

@@ -8,6 +8,7 @@
 #
 
 from vulnerabilities.importers import alpine_linux
+from vulnerabilities.importers import apache_httpd
 from vulnerabilities.importers import archlinux
 from vulnerabilities.importers import debian
 from vulnerabilities.importers import debian_oval
@@ -41,6 +42,7 @@
     debian_oval.DebianOvalImporter,
     npm.NpmImporter,
     retiredotnet.RetireDotnetImporter,
+    apache_httpd.ApacheHTTPDImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/apache_httpd.py

@@ -13,43 +13,32 @@
 import requests
 from bs4 import BeautifulSoup
 from packageurl import PackageURL
-from univers.version_range import VersionRange
+from univers.version_constraint import VersionConstraint
+from univers.version_range import ApacheVersionRange
 from univers.versions import SemverVersion
 
 from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
-from vulnerabilities.package_managers import GitHubTagsAPI
 from vulnerabilities.severity_systems import APACHE_HTTPD
-from vulnerabilities.utils import nearest_patched_package
 
 
 class ApacheHTTPDImporter(Importer):
 
     base_url = "https://httpd.apache.org/security/json/"
+    spdx_license_expression = "Apache-2.0"
+    license_url = "https://www.apache.org/licenses/LICENSE-2.0"
 
-    def set_api(self):
-        self.version_api = GitHubTagsAPI()
-        asyncio.run(self.version_api.load_api(["apache/httpd"]))
-        self.version_api.cache["apache/httpd"] = set(
-            filter(
-                lambda version: version.value not in ignore_tags,
-                self.version_api.cache["apache/httpd"],
-            )
-        )
-
-    def updated_advisories(self):
+    def advisory_data(self):
         links = fetch_links(self.base_url)
-        self.set_api()
-        advisories = []
         for link in links:
             data = requests.get(link).json()
-            advisories.append(self.to_advisory(data))
-        return self.batch_advisories(advisories)
+            yield self.to_advisory(data)
 
     def to_advisory(self, data):
-        cve = data["CVE_data_meta"]["ID"]
+        alias = data["CVE_data_meta"]["ID"]
         descriptions = data["description"]["description_data"]
         description = None
         for desc in descriptions:
@@ -66,12 +55,13 @@ def to_advisory(self, data):
                     VulnerabilitySeverity(
                         system=APACHE_HTTPD,
                         value=value,
+                        scoring_elements="",
                     )
                 )
                 break
         reference = Reference(
-            reference_id=cve,
-            url=urllib.parse.urljoin(self.base_url, f"{cve}.json"),
+            reference_id=alias,
+            url=urllib.parse.urljoin(self.base_url, f"{alias}.json"),
             severities=severities,
         )
 
@@ -81,56 +71,68 @@ def to_advisory(self, data):
                 for version_data in products["version"]["version_data"]:
                     versions_data.append(version_data)
 
-        fixed_version_ranges, affected_version_ranges = self.to_version_ranges(versions_data)
+        fixed_versions = []
+        for timeline_object in data.get("timeline") or []:
+            timeline_value = timeline_object["value"]
+            if "release" in timeline_value:
+                split_timeline_value = timeline_value.split(" ")
+                if "never" in timeline_value:
+                    continue
+                if "release" in split_timeline_value[-1]:
+                    fixed_versions.append(split_timeline_value[0])
+                if "release" in split_timeline_value[0]:
+                    fixed_versions.append(split_timeline_value[-1])
 
         affected_packages = []
-        fixed_packages = []
-
-        for version_range in fixed_version_ranges:
-            fixed_packages.extend(
-                [
-                    PackageURL(type="apache", name="httpd", version=version)
-                    for version in self.version_api.get("apache/httpd").valid_versions
-                    if SemverVersion(version) in version_range
-                ]
-            )
-
-        for version_range in affected_version_ranges:
-            affected_packages.extend(
-                [
-                    PackageURL(type="apache", name="httpd", version=version)
-                    for version in self.version_api.get("apache/httpd").valid_versions
-                    if SemverVersion(version) in version_range
-                ]
+        affected_version_range = self.to_version_ranges(versions_data, fixed_versions)
+        if affected_version_range:
+            affected_packages.append(
+                AffectedPackage(
+                    package=PackageURL(
+                        type="apache",
+                        name="httpd",
+                    ),
+                    affected_version_range=affected_version_range,
+                )
             )
 
         return AdvisoryData(
-            vulnerability_id=cve,
+            aliases=[alias],
             summary=description,
-            affected_packages=nearest_patched_package(affected_packages, fixed_packages),
+            affected_packages=affected_packages,
             references=[reference],
         )
 
-    def to_version_ranges(self, versions_data):
-        fixed_version_ranges = []
-        affected_version_ranges = []
+    def to_version_ranges(self, versions_data, fixed_versions):
+        constraints = []
         for version_data in versions_data:
             version_value = version_data["version_value"]
             range_expression = version_data["version_affected"]
-            if range_expression == "<":
-                fixed_version_ranges.append(
-                    VersionRange.from_scheme_version_spec_string(
-                        "semver", ">={}".format(version_value)
-                    )
-                )
-            elif range_expression == "=" or range_expression == "?=":
-                affected_version_ranges.append(
-                    VersionRange.from_scheme_version_spec_string(
-                        "semver", "{}".format(version_value)
-                    )
+            if range_expression not in {"<=", ">=", "?=", "!<", "="}:
+                raise ValueError(f"unknown comparator found! {range_expression}")
+            comparator_by_range_expression = {
+                ">=": ">=",
+                "!<": ">=",
+                "<=": "<=",
+                "=": "=",
+            }
+            comparator = comparator_by_range_expression.get(range_expression)
+            if comparator:
+                constraints.append(
+                    VersionConstraint(comparator=comparator, version=SemverVersion(version_value))
                 )
 
-        return (fixed_version_ranges, affected_version_ranges)
+        for fixed_version in fixed_versions:
+            # The VersionConstraint method `invert()` inverts the fixed_version's comparator,
+            # enabling inclusion of multiple fixed versions with the `affected_version_range` values.
+            constraints.append(
+                VersionConstraint(
+                    comparator="=",
+                    version=SemverVersion(fixed_version),
+                ).invert()
+            )
+
+        return ApacheVersionRange(constraints=constraints)
 
 
 def fetch_links(url):

vulnerabilities/improvers/default.py

@@ -39,21 +39,22 @@ def interesting_advisories(self) -> QuerySet:
         return Advisory.objects.all()
 
     def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
-
         if not advisory_data:
             return []
 
         if advisory_data.affected_packages:
             for affected_package in advisory_data.affected_packages:
-                affected_purls, fixed_purl = get_exact_purls(affected_package)
-                yield Inference(
-                    aliases=advisory_data.aliases,
-                    confidence=MAX_CONFIDENCE,
-                    summary=advisory_data.summary,
-                    affected_purls=affected_purls,
-                    fixed_purl=fixed_purl,
-                    references=advisory_data.references,
-                )
+                # To deal with multiple fixed versions in a single affected package
+                affected_purls, fixed_purls = get_exact_purls(affected_package)
+                for fixed_purl in fixed_purls:
+                    yield Inference(
+                        aliases=advisory_data.aliases,
+                        confidence=MAX_CONFIDENCE,
+                        summary=advisory_data.summary,
+                        affected_purls=affected_purls,
+                        fixed_purl=fixed_purl,
+                        references=advisory_data.references,
+                    )
 
         else:
             yield Inference.from_advisory_data(
@@ -78,7 +79,7 @@ def get_exact_purls(affected_package: AffectedPackage) -> Tuple[List[PackageURL]
     >>> got = get_exact_purls(affected_package)
     >>> expected = (
     ...     [PackageURL(type='turtle', namespace=None, name='green', version='2.0.0', qualifiers={}, subpath=None)],
-    ...      PackageURL(type='turtle', namespace=None, name='green', version='5.0.0', qualifiers={}, subpath=None)
+    ...     [PackageURL(type='turtle', namespace=None, name='green', version='5.0.0', qualifiers={}, subpath=None)]
     ... )
     >>> assert expected == got
     """
@@ -89,16 +90,25 @@ def get_exact_purls(affected_package: AffectedPackage) -> Tuple[List[PackageURL]
     # TODO: Revisit after https://github.com/nexB/univers/issues/33
     try:
         affected_purls = []
+        fixed_versions = []
         if vr:
             range_versions = [c.version for c in vr.constraints if c]
+            # Any version that's not affected by a vulnerability is considered
+            # fixed.
+            fixed_versions = [c.version for c in vr.constraints if c and c.comparator == "!="]
             resolved_versions = [v for v in range_versions if v and v in vr]
             for version in resolved_versions:
                 affected_purl = evolve_purl(purl=affected_package.package, version=str(version))
                 affected_purls.append(affected_purl)
 
-        fixed_purl = affected_package.get_fixed_purl() if affected_package.fixed_version else None
+        if affected_package.fixed_version:
+            fixed_versions.append(affected_package.fixed_version)
 
-        return affected_purls, fixed_purl
+        fixed_purls = [
+            evolve_purl(purl=affected_package.package, version=str(version))
+            for version in fixed_versions
+        ]
+        return affected_purls, fixed_purls
     except Exception as e:
         logger.error(f"Failed to get exact purls for {affected_package} {e}")
-        return [], None
+        return [], []

vulnerabilities/tests/conftest.py

@@ -25,7 +25,6 @@ def no_rmtree(monkeypatch):
 # Step 2: Run test for importer only if it is activated (pytestmark = pytest.mark.skipif(...))
 # Step 3: Migrate all the tests
 collect_ignore = [
-    "test_apache_httpd.py",
     "test_apache_kafka.py",
     "test_apache_tomcat.py",
     "test_api.py",