Migrate mozilla importer

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

Author: TG1999

CHANGELOG.rst

@@ -6,7 +6,7 @@ Version v31.1.0
 ----------------
 
 - We re-enabled support for the NPM vulnerabilities advisories importer. 
-
+- We re-enabled support for the mozilla vulnerabilities advisories importer.
 
 Version v31.0.0
 ----------------

vulnerabilities/importers/mozilla.py

@@ -8,72 +8,72 @@
 #
 
 import re
+import logging
+from pathlib import Path
+from typing import Iterable
 from typing import List
-from typing import Set
 
 import yaml
 from bs4 import BeautifulSoup
 from markdown import markdown
 from packageurl import PackageURL
+from univers.versions import SemverVersion
 
 from vulnerabilities import severity_systems
 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.importer import VulnerabilitySeverity
 from vulnerabilities.utils import is_cve
 from vulnerabilities.utils import split_markdown_front_matter
 
-REPOSITORY = "mozilla/foundation-security-advisories"
 MFSA_FILENAME_RE = re.compile(r"mfsa(\d{4}-\d{2,3})\.(md|yml)$")
+logger = logging.getLogger(__name__)
 
+class MozillaImporter(Importer):
+    spdx_license_expression = "MPL-2.0"
+    license_url = "https://github.com/mozilla/foundation-security-advisories/blob/master/LICENSE"
+    repo_url = "git+https://github.com/mozilla/foundation-security-advisories/"
 
-class MozillaImporter(GitImporter):
-    def __enter__(self):
-        super(MozillaImporter, self).__enter__()
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        try:
+            self.clone(self.repo_url)
+            path = Path(self.vcs_response.dest_dir)
 
-        if not getattr(self, "_added_files", None):
-            self._added_files, self._updated_files = self.file_changes(
-                recursive=True, subdir="announce"
-            )
-
-    def updated_advisories(self) -> Set[AdvisoryData]:
-        files = self._updated_files.union(self._added_files)
-        files = [
-            f for f in files if f.endswith(".md") or f.endswith(".yml")
-        ]  # skip irrelevant files
-
-        advisories = []
-        for path in files:
-            advisories.extend(to_advisories(path))
-
-        return self.batch_advisories(advisories)
+            vuln = path / "announce"
+            paths = list(vuln.glob("**/*.yml")) + list(vuln.glob("**/*.md"))
+            for file_path in paths:
+                yield from to_advisories(file_path)
+        finally:
+            if self.vcs_response:
+                self.vcs_response.delete()
 
 
 def to_advisories(path: str) -> List[AdvisoryData]:
     """
     Convert a file to corresponding advisories.
     This calls proper method to handle yml/md files.
     """
+    path = str(path)
     mfsa_id = mfsa_id_from_filename(path)
     if not mfsa_id:
         return []
 
     with open(path) as lines:
         if path.endswith(".md"):
-            return get_advisories_from_md(mfsa_id, lines)
+            yield from get_advisories_from_md(mfsa_id, lines)
         if path.endswith(".yml"):
-            return get_advisories_from_yml(mfsa_id, lines)
+            yield from get_advisories_from_yml(mfsa_id, lines)
 
     return []
 
 
 def get_advisories_from_yml(mfsa_id, lines) -> List[AdvisoryData]:
-    advisories = []
     data = yaml.safe_load(lines)
     data["mfsa_id"] = mfsa_id
 
-    fixed_package_urls = get_package_urls(data.get("fixed_in"))
+    affected_packages = get_affected_packages(data.get("fixed_in") or [])
     references = get_yml_references(data)
 
     if not data.get("advisories"):
@@ -82,47 +82,36 @@ def get_advisories_from_yml(mfsa_id, lines) -> List[AdvisoryData]:
     for cve, advisory in data["advisories"].items():
         # These may contain HTML tags
         summary = BeautifulSoup(advisory.get("description", ""), features="lxml").get_text()
-
-        advisories.append(
-            AdvisoryData(
+        alias = is_cve(cve)
+        if alias:
+            yield AdvisoryData(
                 summary=summary,
-                vulnerability_id=cve if is_cve(cve) else "",
-                impacted_package_urls=[],
-                resolved_package_urls=fixed_package_urls,
+                aliases=[cve],
                 references=references,
+                affected_packages=list(affected_packages),
             )
-        )
-
-    return advisories
 
 
 def get_advisories_from_md(mfsa_id, lines) -> List[AdvisoryData]:
     yamltext, mdtext = split_markdown_front_matter(lines.read())
     data = yaml.safe_load(yamltext)
     data["mfsa_id"] = mfsa_id
 
-    fixed_package_urls = get_package_urls(data.get("fixed_in"))
+    affected_packages = get_affected_packages(data.get("fixed_in") or [])
     references = get_yml_references(data)
     cves = re.findall(r"CVE-\d+-\d+", yamltext + mdtext, re.IGNORECASE)
+    description = html_get_p_under_h3(markdown(mdtext), "description")
     for cve in cves:
-        references.append(
-            Reference(
-                reference_id=cve,
-                url=f"https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve}",
-            )
+        cve_ref = Reference(
+            reference_id=cve,
+            url=f"https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve}",
         )
-
-    description = html_get_p_under_h3(markdown(mdtext), "description")
-
-    return [
-        AdvisoryData(
+        yield AdvisoryData(
             summary=description,
-            vulnerability_id="",
-            impacted_package_urls=[],
-            resolved_package_urls=fixed_package_urls,
-            references=references,
+            aliases=[cve],
+            affected_packages=list(affected_packages),
+            references=references + [cve_ref],
         )
-    ]
 
 
 def html_get_p_under_h3(html, h3: str):
@@ -146,18 +135,25 @@ def mfsa_id_from_filename(filename):
     return None
 
 
-def get_package_urls(pkgs: List[str]) -> List[PackageURL]:
-    package_urls = [
-        PackageURL(
-            type="mozilla",
+def get_affected_packages(pkgs: List[str]) -> List[PackageURL]:
+    for pkg in pkgs:
+        if not pkg:
+            continue
             # pkg is of the form "Firefox ESR 1.21" or "Thunderbird 2.21"
-            name=pkg.rsplit(None, 1)[0],
-            version=pkg.rsplit(None, 1)[1],
-        )
-        for pkg in pkgs
-        if pkg
-    ]
-    return package_urls
+        version = pkg.rsplit(None, 1)[1]
+        name = pkg.rsplit(None, 1)[0]
+        if version and name:
+            try:
+                fixed_version=SemverVersion(version)
+                yield AffectedPackage(
+                    package=PackageURL(
+                        type="mozilla",
+                        name=name,
+                    ),
+                    fixed_version=fixed_version,
+                )
+            except Exception:
+                logger.exception(f"Error parsing version {version} for {name}")
 
 
 def get_yml_references(data: any) -> List[Reference]: