Merge pull request #1056 from TG1999/migrate/gentoo

Migrate gentoo importer #1055

CHANGELOG.rst

@@ -6,6 +6,7 @@ Next release
 ----------------
 
 - We re-enabled support for the mozilla vulnerabilities advisories importer.
+- We re-enabled support for the gentoo vulnerabilities advisories importer.
 
 
 Version v31.1.1

vulnerabilities/importers/__init__.py

@@ -12,6 +12,7 @@
 from vulnerabilities.importers import archlinux
 from vulnerabilities.importers import debian
 from vulnerabilities.importers import debian_oval
+from vulnerabilities.importers import gentoo
 from vulnerabilities.importers import github
 from vulnerabilities.importers import gitlab
 from vulnerabilities.importers import mozilla
@@ -45,6 +46,7 @@
     retiredotnet.RetireDotnetImporter,
     apache_httpd.ApacheHTTPDImporter,
     mozilla.MozillaImporter,
+    gentoo.GentooImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/gentoo.py

@@ -10,73 +10,72 @@
 
 import re
 import xml.etree.ElementTree as ET
-from typing import Set
+from pathlib import Path
+from typing import Iterable
 
 from packageurl import PackageURL
+from univers.version_constraint import VersionConstraint
+from univers.version_range import EbuildVersionRange
+from univers.versions import GentooVersion
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import GitImporter
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
-from vulnerabilities.utils import nearest_patched_package
 
 
-class GentooImporter(GitImporter):
-    def __enter__(self):
-        super(GentooImporter, self).__enter__()
-
-        if not getattr(self, "_added_files", None):
-            self._added_files, self._updated_files = self.file_changes(
-                recursive=True, file_ext="xml"
-            )
-
-    def updated_advisories(self) -> Set[AdvisoryData]:
-        files = self._updated_files.union(self._added_files)
-        advisories = []
-        for f in files:
-            processed_data = self.process_file(f)
-            advisories.extend(processed_data)
-        return self.batch_advisories(advisories)
+class GentooImporter(Importer):
+    repo_url = "git+https://anongit.gentoo.org/git/data/glsa.git"
+    spdx_license_expression = "CC-BY-SA-4.0"
+    # the license notice is at this url https://anongit.gentoo.org/ says:
+    # The contents of this document, unless otherwise expressly stated, are licensed
+    # under the [CC-BY-SA-4.0](https://creativecommons.org/licenses/by-sa/4.0/) license.
+    license_url = "https://creativecommons.org/licenses/by-sa/4.0/"
+
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        try:
+            self.clone(repo_url=self.repo_url)
+            base_path = Path(self.vcs_response.dest_dir)
+            for file_path in base_path.glob("**/*.xml"):
+                yield from self.process_file(file_path)
+        finally:
+            if self.vcs_response:
+                self.vcs_response.delete()
 
     def process_file(self, file):
-        xml_data = {}
+        cves = []
+        summary = ""
+        vuln_references = []
         xml_root = ET.parse(file).getroot()
-        glsa = "GLSA-" + xml_root.attrib["id"]
-        vuln_reference = [
-            Reference(
-                reference_id=glsa,
-                url="https://security.gentoo.org/glsa/{}".format(xml_root.attrib["id"]),
-            )
-        ]
+        id = xml_root.attrib.get("id")
+        if id:
+            glsa = "GLSA-" + id
+            vuln_references = [
+                Reference(
+                    reference_id=glsa,
+                    url=f"https://security.gentoo.org/glsa/{id}",
+                )
+            ]
 
         for child in xml_root:
             if child.tag == "references":
-                xml_data["cves"] = self.cves_from_reference(child)
+                cves = self.cves_from_reference(child)
 
             if child.tag == "synopsis":
-                xml_data["description"] = child.text
+                summary = child.text
 
             if child.tag == "affected":
-                (
-                    xml_data["affected_purls"],
-                    xml_data["unaffected_purls"],
-                ) = self.affected_and_safe_purls(child)
-                xml_data["unaffected_purls"] = list(xml_data["unaffected_purls"])
-                xml_data["affected_purls"] = list(xml_data["affected_purls"])
-
-        advisory_list = []
+                affected_packages = list(self.affected_and_safe_purls(child))
+
         # It is very inefficient, to create new Advisory for each CVE
         # this way, but there seems no alternative.
-        for cve in xml_data["cves"]:
-            advisory = AdvisoryData(
-                vulnerability_id=cve,
-                summary=xml_data["description"],
-                affected_packages=nearest_patched_package(
-                    xml_data["affected_purls"], xml_data["unaffected_purls"]
-                ),
-                references=vuln_reference,
+        for cve in cves:
+            yield AdvisoryData(
+                aliases=[cve],
+                summary=summary,
+                references=vuln_references,
+                affected_packages=affected_packages,
             )
-            advisory_list.append(advisory)
-        return advisory_list
 
     @staticmethod
     def cves_from_reference(reference):
@@ -91,40 +90,63 @@ def cves_from_reference(reference):
 
     @staticmethod
     def affected_and_safe_purls(affected_elem):
-        safe_purls = set()
-        affected_purls = set()
-        skip_versions = {"1.3*", "7.3*", "7.4*"}
+        constraints = []
         for pkg in affected_elem:
-            for info in pkg:
-                if info.text in skip_versions:
+            name = pkg.attrib.get("name")
+            if not name:
+                continue
+            pkg_ns, _, pkg_name = name.rpartition("/")
+            purl = PackageURL(type="ebuild", name=pkg_name, namespace=pkg_ns)
+            safe_versions, affected_versions = GentooImporter.get_safe_and_affected_versions(pkg)
+
+            for version in safe_versions:
+                constraints.append(
+                    VersionConstraint(version=GentooVersion(version), comparator="=").invert()
+                )
+
+            for version in affected_versions:
+                constraints.append(
+                    VersionConstraint(version=GentooVersion(version), comparator="=")
+                )
+
+            if not constraints:
+                continue
+
+            yield AffectedPackage(
+                package=purl, affected_version_range=EbuildVersionRange(constraints=constraints)
+            )
+
+    @staticmethod
+    def get_safe_and_affected_versions(pkg):
+        # TODO : Revisit why we are skipping some versions in gentoo importer
+        skip_versions = {"1.3*", "7.3*", "7.4*"}
+        safe_versions = set()
+        affected_versions = set()
+        for info in pkg:
+            if info.text in skip_versions:
+                continue
+
+            if info.attrib.get("range"):
+                if len(info.attrib.get("range")) > 2:
                     continue
-                pkg_ns, pkg_name, = pkg.attrib[
-                    "name"
-                ].split("/")
-                purl = PackageURL(type="ebuild", name=pkg_name, version=info.text, namespace=pkg_ns)
-
-                if info.attrib.get("range"):
-                    if len(info.attrib.get("range")) > 2:
-                        continue
-
-                if info.tag == "unaffected":
-                    # quick hack, to know whether this
-                    # version lies in this range, 'e' stands for
-                    # equal, which is paired with 'greater' or 'less'.
-                    # All possible values of info.attrib['range'] =
-                    # {'gt', 'lt', 'rle', 'rge', 'rgt', 'le', 'ge', 'eq'}, out of
-                    # which ('rle', 'rge', 'rgt') are ignored, because they compare
-                    # 'release' not the 'version'.
-
-                    if "e" in info.attrib["range"]:
-                        safe_purls.add(purl)
-                    else:
-                        affected_purls.add(purl)
-
-                elif info.tag == "vulnerable":
-                    if "e" in info.attrib["range"]:
-                        affected_purls.add(purl)
-                    else:
-                        safe_purls.add(purl)
-
-        return (affected_purls, safe_purls)
+
+            if info.tag == "unaffected":
+                # quick hack, to know whether this
+                # version lies in this range, 'e' stands for
+                # equal, which is paired with 'greater' or 'less'.
+                # All possible values of info.attrib['range'] =
+                # {'gt', 'lt', 'rle', 'rge', 'rgt', 'le', 'ge', 'eq'}, out of
+                # which ('rle', 'rge', 'rgt') are ignored, because they compare
+                # 'release' not the 'version'.
+                if "e" in info.attrib["range"]:
+                    safe_versions.add(info.text)
+                else:
+                    affected_versions.add(info.text)
+
+            elif info.tag == "vulnerable":
+                if "e" in info.attrib["range"]:
+                    affected_versions.add(info.text)
+                else:
+                    safe_versions.add(info.text)
+
+        return safe_versions, affected_versions

vulnerabilities/tests/conftest.py

@@ -29,7 +29,6 @@ def no_rmtree(monkeypatch):
     "test_apache_tomcat.py",
     "test_api.py",
     "test_elixir_security.py",
-    "test_gentoo.py",
     "test_istio.py",
     "test_models.py",
     "test_msr2019.py",

vulnerabilities/tests/test_data/gentoo/gentoo-expected.json

@@ -0,0 +1,30 @@
+[
+  {
+    "aliases": [
+      "CVE-2017-9800"
+    ],
+    "summary": "A command injection vulnerability in Subversion may allow remote\n    attackers to execute arbitrary code.\n  ",
+    "affected_packages": [
+      {
+        "package": {
+          "type": "ebuild",
+          "namespace": "dev-vcs",
+          "name": "subversion",
+          "version": null,
+          "qualifiers": null,
+          "subpath": null
+        },
+        "affected_version_range": "vers:ebuild/0.1.1|!=1.9.7",
+        "fixed_version": null
+      }
+    ],
+    "references": [
+      {
+        "reference_id": "GLSA-201709-09",
+        "url": "https://security.gentoo.org/glsa/201709-09",
+        "severities": []
+      }
+    ],
+    "date_published": null
+  }
+]

vulnerabilities/tests/test_gentoo.py

@@ -19,101 +19,16 @@
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Reference
 from vulnerabilities.importers.gentoo import GentooImporter
+from vulnerabilities.tests import util_tests
 from vulnerabilities.utils import AffectedPackage
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-TEST_DATA = os.path.join(BASE_DIR, "test_data/gentoo/glsa-201709-09.xml")
+TEST_DIR = os.path.join(BASE_DIR, "test_data/gentoo")
 
 
-class TestGentooImporter(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        data_source_cfg = {
-            "repository_url": "https://example.git",
-        }
-        cls.data_src = GentooImporter(1, config=data_source_cfg)
-        cls.xml_doc = ET.parse(TEST_DATA)
-        cls.references = []
-        for child in cls.xml_doc.getroot():
-
-            if child.tag == "references":
-                cls.references.append(child)
-
-            if child.tag == "affected":
-                cls.affected = child
-
-    def test_affected_and_safe_purls(self):
-        exp_affected = {
-            PackageURL(
-                type="ebuild",
-                namespace="dev-vcs",
-                name="subversion",
-                version="0.1.1",
-                qualifiers=OrderedDict(),
-                subpath=None,
-            )
-        }
-        exp_safe = {
-            PackageURL(
-                type="ebuild",
-                namespace="dev-vcs",
-                name="subversion",
-                version="1.9.7",
-                qualifiers=OrderedDict(),
-                subpath=None,
-            )
-        }
-
-        aff, safe = GentooImporter.affected_and_safe_purls(self.affected)
-
-        assert aff == exp_affected
-        assert safe == exp_safe
-
-    def test_cves_from_reference(self):
-
-        exp_cves = {"CVE-2017-9800"}
-        found_cves = set()
-        for ref in self.references:
-            found_cves.update(GentooImporter.cves_from_reference(ref))
-
-        assert exp_cves == found_cves
-
-    def test_process_file(self):
-
-        expected_advisories = [
-            Advisory(
-                summary=(
-                    "A command injection vulnerability in "
-                    "Subversion may allow remote\n    "
-                    "attackers to execute arbitrary code.\n  "
-                ),
-                affected_packages=[
-                    AffectedPackage(
-                        vulnerable_package=PackageURL(
-                            type="ebuild",
-                            namespace="dev-vcs",
-                            name="subversion",
-                            version="0.1.1",
-                        ),
-                        patched_package=PackageURL(
-                            type="ebuild",
-                            namespace="dev-vcs",
-                            name="subversion",
-                            version="1.9.7",
-                        ),
-                    )
-                ],
-                references=[
-                    Reference(
-                        url="https://security.gentoo.org/glsa/201709-09",
-                        reference_id="GLSA-201709-09",
-                    )
-                ],
-                vulnerability_id="CVE-2017-9800",
-            )
-        ]
-
-        found_advisories = self.data_src.process_file(TEST_DATA)
-        found_advisories = list(map(Advisory.normalized, found_advisories))
-        expected_advisories = list(map(Advisory.normalized, expected_advisories))
-        assert sorted(found_advisories) == sorted(expected_advisories)
+def test_gentoo_import():
+    file = os.path.join(TEST_DIR, "glsa-201709-09.xml")
+    advisories = GentooImporter().process_file(file)
+    result = [adv.to_dict() for adv in advisories]
+    expected_file = os.path.join(TEST_DIR, "gentoo-expected.json")
+    util_tests.check_results_against_json(result, expected_file)