migrate current alpine importer to alpine importer-improver model

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

Author: TG1999

vulnerabilities/importers/alpine_linux.py

@@ -20,16 +20,15 @@
 #  for any legal advice.
 #  VulnerableCode is a free software code scanning tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
-from typing import Any
+from typing import Any, Iterable
 from typing import List
 from typing import Mapping
-from typing import Set
 
 import requests
 import saneyaml
 from bs4 import BeautifulSoup
 
-from vulnerabilities.importer import Advisory
+from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.helpers import is_cve
@@ -38,106 +37,96 @@
 
 
 class AlpineImporter(Importer):
-    @staticmethod
-    def fetch_advisory_links():
-        index_page = BeautifulSoup(requests.get(BASE_URL).content, features="lxml")
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        advisory_links = fetch_advisory_links()
+        for link in advisory_links:
+            yield (process_link(link))
 
-        alpine_versions = [
-            link.text for link in index_page.find_all("a") if link.text.startswith("v")
-        ]
 
-        advisory_directory_links = [f"{BASE_URL}{version}" for version in alpine_versions]
-
-        advisory_links = []
-        for advisory_directory_link in advisory_directory_links:
-            advisory_directory_page = requests.get(advisory_directory_link).content
-            advisory_directory_page = BeautifulSoup(advisory_directory_page, features="lxml")
-            advisory_links.extend(
-                [
-                    f"{advisory_directory_link}{anchore_tag.text}"
-                    for anchore_tag in advisory_directory_page.find_all("a")
-                    if anchore_tag.text.endswith("yaml")
-                ]
-            )
+def fetch_advisory_links():
+    index_page = BeautifulSoup(requests.get(BASE_URL).content, features="lxml")
 
-        return advisory_links
+    alpine_versions = [
+        link.text for link in index_page.find_all("a") if link.text.startswith("v")
+    ]
 
-    def updated_advisories(self) -> Set[Advisory]:
-        advisories = []
-        advisory_links = self.fetch_advisory_links()
-        for link in advisory_links:
-            advisories.extend(self._process_link(link))
+    advisory_directory_links = [f"{BASE_URL}{version}" for version in alpine_versions]
 
-        return self.batch_advisories(advisories)
+    for advisory_directory_link in advisory_directory_links:
+        advisory_directory_page = requests.get(advisory_directory_link).content
+        advisory_directory_page = BeautifulSoup(
+            advisory_directory_page, features="lxml"
+        )
+        yield [
+            f"{advisory_directory_link}{anchore_tag.text}"
+            for anchore_tag in advisory_directory_page.find_all("a")
+            if anchore_tag.text.endswith("yaml")
+        ]
 
-    def _process_link(self, link) -> List[Advisory]:
-        advisories = []
-        yaml_response = requests.get(link).content
-        record = saneyaml.load(yaml_response)
 
-        if record["packages"] is None:
-            return advisories
+def process_link(link) -> List[AdvisoryData]:
+    yaml_response = requests.get(link).content
+    record = saneyaml.load(yaml_response)
 
-        for p in record["packages"]:
-            advisories.extend(
-                self._load_advisories(
-                    p["pkg"],
-                )
-            )
+    if record["packages"] is None:
+        return []
 
-        return advisories
+    for package in record["packages"]:
+        yield load_advisories(
+            package["pkg"],
+        )
 
-    @staticmethod
-    def _load_advisories(
-        pkg_infos: Mapping[str, Any],
-    ) -> List[Advisory]:
 
-        advisories = []
+def load_advisories(
+    pkg_infos: Mapping[str, Any],
+) -> List[AdvisoryData]:
 
-        for fixed_vulns in pkg_infos["secfixes"].values():
+    # assert type(pkg_infos["secfixes"]) == Mapping[str, List]
+    for fixed_vulns in pkg_infos["secfixes"].values():
 
-            if fixed_vulns is None:
-                continue
+        if fixed_vulns is None:
+            continue
 
-            for vuln_ids in fixed_vulns:
-                vuln_ids = vuln_ids.split()
-                references = []
-                for reference_id in vuln_ids[1:]:
+        for vuln_ids in fixed_vulns:
+            vuln_ids = vuln_ids.split()
+            references = []
+            for reference_id in vuln_ids[1:]:
 
-                    if reference_id.startswith("XSA"):
-                        xsa_id = reference_id.split("-")[-1]
-                        references.append(
-                            Reference(
-                                reference_id=reference_id,
-                                url="https://xenbits.xen.org/xsa/advisory-{}.html".format(xsa_id),
-                            )
+                if reference_id.startswith("XSA"):
+                    xsa_id = reference_id.split("-")[-1]
+                    references.append(
+                        Reference(
+                            reference_id=reference_id,
+                            url="https://xenbits.xen.org/xsa/advisory-{}.html".format(
+                                xsa_id
+                            ),
                         )
+                    )
 
-                    elif reference_id.startswith("ZBX"):
-                        references.append(
-                            Reference(
-                                reference_id=reference_id,
-                                url="https://support.zabbix.com/browse/{}".format(reference_id),
-                            )
+                elif reference_id.startswith("ZBX"):
+                    references.append(
+                        Reference(
+                            reference_id=reference_id,
+                            url="https://support.zabbix.com/browse/{}".format(
+                                reference_id
+                            ),
                         )
+                    )
 
-                    elif reference_id.startswith("wnpa-sec"):
-                        references.append(
-                            Reference(
-                                reference_id=reference_id,
-                                url="https://www.wireshark.org/security/{}.html".format(
-                                    reference_id
-                                ),
-                            )
+                elif reference_id.startswith("wnpa-sec"):
+                    references.append(
+                        Reference(
+                            reference_id=reference_id,
+                            url="https://www.wireshark.org/security/{}.html".format(
+                                reference_id
+                            ),
                         )
-
-                # TODO: Handle the CVE-????-????? case
-                advisories.append(
-                    Advisory(
-                        summary="",
-                        references=references,
-                        vulnerability_id=vuln_ids[0] if is_cve(vuln_ids[0]) else "",
                     )
-                )
 
-        return advisories
+            # TODO: Handle the CVE-????-????? case
+            yield AdvisoryData(
+                summary="",
+                references=references,
+                affected_packages=[],
+                aliases=[vuln_ids[0] if is_cve(vuln_ids[0]) else ""],
+            )

vulnerabilities/importers/nginx.py

@@ -20,8 +20,6 @@
 #  VulnerableCode is a free software tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
-import dataclasses
-import datetime
 from typing import Iterable
 import logging
 
@@ -40,7 +38,6 @@
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.improver import Inference
 from vulnerabilities.improver import Improver
-from vulnerabilities.helpers import nearest_patched_package
 from vulnerabilities.models import Advisory
 from vulnerabilities.package_managers import GitHubTagsAPI
 from vulnerabilities.package_managers import Version

vulnerabilities/improvers/default.py

@@ -1,4 +1,4 @@
-from typing import Iterable
+from typing import Iterable, Tuple
 from typing import List
 from itertools import chain
 
@@ -38,7 +38,7 @@ def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
             )
 
 
-def get_exact_purls(affected_package: AffectedPackage) -> (List[PackageURL], PackageURL):
+def get_exact_purls(affected_package: AffectedPackage) -> Tuple[List[PackageURL], PackageURL]:
     """
     Return a list of affected purls and the fixed package found in the ``affected_package``
     AffectedPackage disregarding any ranges.
@@ -67,7 +67,7 @@ def get_exact_purls(affected_package: AffectedPackage) -> (List[PackageURL], Pac
     range_versions = [c.version for c in vr.constraints if c]
     resolved_versions = [v for v in range_versions if v and v in vr]
 
-    affected_purls = []
+    affected_purls: List[PackageURL] = []
     for version in resolved_versions:
         affected_purl = affected_package.package._replace(version=str(version))
         affected_purls.append(affected_purl)