Migrate gentoo importer

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

Author: TG1999

CHANGELOG.rst

@@ -2,6 +2,12 @@ Release notes
 =============
 
 
+Version v31.2.0
+----------------
+
+- We re-enabled support for the Gentoo vulnerabilities advisories importer.
+
+
 Version v31.1.0
 ----------------
 

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

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/gentoo.py

@@ -10,73 +10,69 @@
 
 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__()
+class GentooImporter(Importer):
+    repo_url = "https://anongit.gentoo.org/git/data/glsa.git"
+    spdx_license_expression = "CC-BY-SA-4.0"
+    license_url = "https://anongit.gentoo.org/"
 
-        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)
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        try:
+            self.clone()
+            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_reference = []
         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_reference = [
+                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"]
-                ),
+        for cve in cves:
+            yield AdvisoryData(
+                aliases=[cve],
+                summary=summary,
                 references=vuln_reference,
+                affected_packages=affected_packages,
             )
-            advisory_list.append(advisory)
-        return advisory_list
 
     @staticmethod
     def cves_from_reference(reference):
@@ -91,8 +87,8 @@ def cves_from_reference(reference):
 
     @staticmethod
     def affected_and_safe_purls(affected_elem):
-        safe_purls = set()
-        affected_purls = set()
+        safe_versions = set()
+        affected_versions = set()
         skip_versions = {"1.3*", "7.3*", "7.4*"}
         for pkg in affected_elem:
             for info in pkg:
@@ -101,7 +97,7 @@ def affected_and_safe_purls(affected_elem):
                 pkg_ns, pkg_name, = pkg.attrib[
                     "name"
                 ].split("/")
-                purl = PackageURL(type="ebuild", name=pkg_name, version=info.text, namespace=pkg_ns)
+                purl = PackageURL(type="ebuild", name=pkg_name, namespace=pkg_ns)
 
                 if info.attrib.get("range"):
                     if len(info.attrib.get("range")) > 2:
@@ -117,14 +113,28 @@ def affected_and_safe_purls(affected_elem):
                     # 'release' not the 'version'.
 
                     if "e" in info.attrib["range"]:
-                        safe_purls.add(purl)
+                        safe_versions.add(info.text)
                     else:
-                        affected_purls.add(purl)
+                        affected_versions.add(info.text)
 
                 elif info.tag == "vulnerable":
                     if "e" in info.attrib["range"]:
-                        affected_purls.add(purl)
+                        affected_versions.add(info.text)
                     else:
-                        safe_purls.add(purl)
+                        safe_versions.add(info.text)
+
+                constraints = []
+
+                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="=")
+                    )
 
-        return (affected_purls, safe_purls)
+                yield AffectedPackage(
+                    package=purl, affected_version_range=EbuildVersionRange(constraints=constraints)
+                )

vulnerabilities/tests/conftest.py

@@ -30,7 +30,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_mozilla.py",

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

@@ -0,0 +1,54 @@
+[
+  {
+    "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/!=1.9.7",
+        "fixed_version": null
+      },
+      {
+        "package": {
+          "type": "ebuild",
+          "namespace": "dev-vcs",
+          "name": "subversion",
+          "version": null,
+          "qualifiers": null,
+          "subpath": null
+        },
+        "affected_version_range": "vers:ebuild/!=1.9.7",
+        "fixed_version": null
+      },
+      {
+        "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)