Migrate elixir security importer

Signed-off-by: Tushar Goel <[email protected]>

Author: TG1999

vulnerabilities/importers/__init__.py

@@ -13,6 +13,7 @@
 from vulnerabilities.importers import debian
 from vulnerabilities.importers import debian_oval
 from vulnerabilities.importers import gentoo
+from vulnerabilities.importers import elixir_security
 from vulnerabilities.importers import github
 from vulnerabilities.importers import gitlab
 from vulnerabilities.importers import istio
@@ -53,6 +54,7 @@
     istio.IstioImporter,
     project_kb_msr2019.ProjectKBMSRImporter,
     suse_scores.SUSESeverityScoreImporter,
+    elixir_security.ElixirSecurityImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/elixir_security.py

@@ -6,119 +6,98 @@
 # See https://github.com/nexB/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
-import asyncio
+import logging
+from pathlib import Path
 from typing import Set
 
 from packageurl import PackageURL
-from univers.version_range import VersionRange
-from univers.versions import SemverVersion
+from univers.version_constraint import VersionConstraint
+from univers.version_range import HexVersionRange
 
 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.package_managers import HexVersionAPI
 from vulnerabilities.utils import load_yaml
-from vulnerabilities.utils import nearest_patched_package
 
+logger = logging.getLogger(__name__)
 
-class ElixirSecurityImporter(GitImporter):
-    def __enter__(self):
-        super(ElixirSecurityImporter, self).__enter__()
 
-        if not getattr(self, "_added_files", None):
-            self._added_files, self._updated_files = self.file_changes(
-                recursive=True, file_ext="yml", subdir="./packages"
-            )
-        self.pkg_manager_api = HexVersionAPI()
-        self.set_api(self.collect_packages())
-
-    def set_api(self, packages):
-        asyncio.run(self.pkg_manager_api.load_api(packages))
-
-    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)
-            if processed_data:
-                advisories.append(processed_data)
-        return self.batch_advisories(advisories)
-
-    def collect_packages(self):
-        packages = set()
-        files = self._updated_files.union(self._added_files)
-        for f in files:
-            data = load_yaml(f)
-            if data.get("package"):
-                packages.add(data["package"])
-
-        return packages
-
-    def get_versions_for_pkg_from_range_list(self, version_range_list, pkg_name):
-        # Takes a list of version ranges(pathced and unaffected) of a package
-        # as parameter and returns a tuple of safe package versions and
-        # vulnerable package versions
-
-        safe_pkg_versions = []
-        vuln_pkg_versions = []
-        all_version_list = self.pkg_manager_api.get(pkg_name).valid_versions
-        if not version_range_list:
-            return [], all_version_list
-        version_ranges = [
-            VersionRange.from_scheme_version_spec_string("semver", r) for r in version_range_list
-        ]
-        for version in all_version_list:
-            version_obj = SemverVersion(version)
-            if any([version_obj in v for v in version_ranges]):
-                safe_pkg_versions.append(version)
-
-        vuln_pkg_versions = set(all_version_list) - set(safe_pkg_versions)
-        return safe_pkg_versions, vuln_pkg_versions
+class ElixirSecurityImporter(Importer):
+
+    repo_url = "git+https://github.com/dependabot/elixir-security-advisories"
+    license_url = "https://github.com/dependabot/elixir-security-advisories/blob/master/LICENSE.txt"
+    spdx_license_expression = "CC0-1.0"
+
+    def advisory_data(self) -> Set[AdvisoryData]:
+        try:
+            self.clone(self.repo_url)
+            path = Path(self.vcs_response.dest_dir)
+            vuln = path / "packages"
+            for file in vuln.glob("**/*.yml"):
+                yield from self.process_file(file)
+        finally:
+            if self.vcs_response:
+                self.vcs_response.delete()
 
     def process_file(self, path):
+        path = str(path)
         yaml_file = load_yaml(path)
-        pkg_name = yaml_file["package"]
-        safe_pkg_versions = []
-        vuln_pkg_versions = []
-        if not yaml_file.get("patched_versions"):
-            yaml_file["patched_versions"] = []
-
-        if not yaml_file.get("unaffected_versions"):
-            yaml_file["unaffected_versions"] = []
-
-        safe_pkg_versions, vuln_pkg_versions = self.get_versions_for_pkg_from_range_list(
-            yaml_file["patched_versions"] + yaml_file["unaffected_versions"],
-            pkg_name,
+        cve_id = ""
+        summary = yaml_file.get("description") or ""
+        pkg_name = yaml_file.get("package") or ""
+        if not pkg_name:
+            return []
+
+        cve = yaml_file.get("cve") or ""
+
+        if cve and not cve.startswith("CVE-"):
+            cve = yaml_file["cve"]
+            cve_id = f"CVE-{cve}"
+
+        references = []
+        link = yaml_file.get("link") or ""
+        if link:
+            references.append(
+                Reference(
+                    url=link,
+                )
+            )
+
+        affected_packages = []
+
+        unaffected_versions = yaml_file.get("unaffected_versions") or []
+        patched_versions = yaml_file.get("patched_versions") or []
+
+        constraints = []
+        vrc = HexVersionRange.version_class
+
+        for version in unaffected_versions:
+            constraints.append(VersionConstraint.from_string(version_class=vrc, string=version))
+
+        for version in patched_versions:
+            if version.startswith("~>"):
+                version = version[2:]
+            constraints.append(
+                VersionConstraint.from_string(version_class=vrc, string=version).invert()
+            )
+
+        affected_packages.append(
+            AffectedPackage(
+                package=PackageURL(
+                    type="hex",
+                    name=pkg_name,
+                ),
+                affected_version_range=HexVersionRange(constraints=constraints),
+            )
         )
 
-        if yaml_file.get("cve"):
-            cve_id = "CVE-" + yaml_file["cve"]
-        else:
-            cve_id = ""
-
-        safe_purls = []
-        vuln_purls = []
-
-        safe_purls = [
-            PackageURL(name=pkg_name, type="hex", version=version) for version in safe_pkg_versions
-        ]
-
-        vuln_purls = [
-            PackageURL(name=pkg_name, type="hex", version=version) for version in vuln_pkg_versions
-        ]
-
-        references = [
-            Reference(
-                reference_id=yaml_file["id"],
-            ),
-            Reference(
-                url=yaml_file["link"],
-            ),
-        ]
-
-        return AdvisoryData(
-            summary=yaml_file["description"],
-            affected_packages=nearest_patched_package(vuln_purls, safe_purls),
-            vulnerability_id=cve_id,
+        if not cve_id:
+            return []
+
+        yield AdvisoryData(
+            aliases=[cve_id],
+            summary=summary,
             references=references,
+            affected_packages=affected_packages,
         )