migrate alpine importer to importer improver model

Current alpine importer models need to be refactored to give AdvisoryData instead of Advisory, also add some validation to parse license expression and make affected_version_range optional

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

Author: TG1999

vulnerabilities/importer.py

@@ -41,6 +41,7 @@
 from packageurl import PackageURL
 from univers.version_range import VersionRange
 from univers.versions import Version
+from license_expression import Licensing
 
 from vulnerabilities.helpers import classproperty
 from vulnerabilities.helpers import nearest_patched_package
@@ -117,22 +118,24 @@ class AffectedPackage:
     """
     Contains a range of affected versions and a fixed version of a given package
     The PackageURL supplied must *not* have a version
+    It must contain either `affected_version_range` or `fixed_version`
     """
 
     package: PackageURL
-    affected_version_range: VersionRange
+    affected_version_range: Optional[VersionRange]
     fixed_version: Optional[Version] = None
 
     def __post_init__(self):
-        if self.package.version:
+        if self.package.version or not (self.affected_version_range or self.fixed_version):
             raise ValueError
 
     def get_fixed_purl(self):
         """
         Return PackageURL corresponding to object's fixed_version
         """
-        fixed_version = self.fixed_version
-        fixed_purl = self.package._replace(version=str(fixed_version))
+        if not self.fixed_version:
+            raise ValueError
+        fixed_purl = self.package._replace(version=str(self.fixed_version))
         return fixed_purl
 
     @classmethod
@@ -228,6 +231,13 @@ class Importer:
     def __init__(self):
         if not self.spdx_license_expression:
             raise Exception(f"Cannot run importer {self!r} without a license")
+        licensing = Licensing()
+        try:
+            licensing.parse(self.spdx_license_expression)
+        except:
+            raise Exception(
+                f"{self.spdx_license_expression} is not a valid spdx license expression"
+            )
 
     @classproperty
     def qualified_name(cls):

vulnerabilities/importers/__init__.py

@@ -19,8 +19,9 @@
 #  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 vulnerabilities.importers import alpine_linux
 from vulnerabilities.importers import nginx
 
-IMPORTERS_REGISTRY = [nginx.NginxImporter]
+IMPORTERS_REGISTRY = [nginx.NginxImporter, alpine_linux.AlpineImporter]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/alpine_linux.py

@@ -20,124 +20,146 @@
 #  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.
+import json
 from typing import Any
+from typing import Iterable
 from typing import List
 from typing import Mapping
-from typing import Set
 
 import requests
-import saneyaml
 from bs4 import BeautifulSoup
+from packageurl import PackageURL
+from univers.versions import AlpineLinuxVersion
 
 from vulnerabilities.helpers import is_cve
-from vulnerabilities.importer import Advisory
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 
-BASE_URL = "https://secdb.alpinelinux.org/"
-
 
 class AlpineImporter(Importer):
-    @staticmethod
-    def fetch_advisory_links():
-        index_page = BeautifulSoup(requests.get(BASE_URL).content, features="lxml")
-
-        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")
-                ]
-            )
-
-        return advisory_links
+    spdx_license_expression = "CC-BY-SA-4.0"
 
-    def updated_advisories(self) -> Set[Advisory]:
+    def advisory_data(self) -> Iterable[AdvisoryData]:
         advisories = []
-        advisory_links = self.fetch_advisory_links()
+        advisory_links = fetch_advisory_links()
         for link in advisory_links:
-            advisories.extend(self._process_link(link))
+            advisories.extend(process_link(link))
 
-        return self.batch_advisories(advisories)
 
-    def _process_link(self, link) -> List[Advisory]:
-        advisories = []
-        yaml_response = requests.get(link).content
-        record = saneyaml.load(yaml_response)
+def fetch_advisory_links():
+    BASE_URL = "https://secdb.alpinelinux.org/"
+    index_page = BeautifulSoup(requests.get(BASE_URL).content, features="lxml")
 
-        if record["packages"] is None:
-            return advisories
+    alpine_versions = [link.text for link in index_page.find_all("a") if link.text.startswith("v")]
 
-        for p in record["packages"]:
-            advisories.extend(
-                self._load_advisories(
-                    p["pkg"],
-                )
-            )
+    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("json")
+            ]
+        )
+
+    return advisory_links
+
+
+def process_link(link) -> List[AdvisoryData]:
+    json_response = requests.get(link).content
+    record = json.loads(json_response)
+    advisories: List[AdvisoryData] = []
+
+    if record["packages"] is None:
         return advisories
 
-    @staticmethod
-    def _load_advisories(
-        pkg_infos: Mapping[str, Any],
-    ) -> List[Advisory]:
+    for package in record["packages"]:
+        assert isinstance(record["archs"], Iterable[str])
+        assert isinstance(record["reponame"], str)
+        assert isinstance(record["distroversion"], str)
+        advisories.extend(
+            load_advisories(
+                package["pkg"],
+                record["distroversion"],
+                record["reponame"],
+                record["archs"],
+            )
+        )
+    return advisories
+
 
-        advisories = []
+def load_advisories(
+    pkg_infos: Mapping[str, Any],
+    distroversion: str,
+    reponame: str,
+    archs: Iterable[str],
+) -> Iterable[AdvisoryData]:
 
-        for fixed_vulns in pkg_infos["secfixes"].values():
+    if not (pkg_infos.get("name") and isinstance(pkg_infos["name"], str)):
+        return []
 
-            if fixed_vulns is None:
-                continue
+    assert isinstance(pkg_infos["secfixes"], Mapping[str, List])
 
-            for vuln_ids in fixed_vulns:
-                vuln_ids = vuln_ids.split()
-                references = []
-                for reference_id in vuln_ids[1:]:
+    for version, fixed_vulns in pkg_infos["secfixes"].items():
 
-                    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 fixed_vulns is None:
+            continue
 
-                    elif reference_id.startswith("ZBX"):
-                        references.append(
-                            Reference(
-                                reference_id=reference_id,
-                                url="https://support.zabbix.com/browse/{}".format(reference_id),
-                            )
+        for vuln_ids in fixed_vulns:
+            assert isinstance(vuln_ids, str)
+            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),
                         )
+                    )
 
-                    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("ZBX"):
+                    references.append(
+                        Reference(
+                            reference_id=reference_id,
+                            url="https://support.zabbix.com/browse/{}".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 "",
+                elif reference_id.startswith("wnpa-sec"):
+                    references.append(
+                        Reference(
+                            reference_id=reference_id,
+                            url="https://www.wireshark.org/security/{}.html".format(reference_id),
+                        )
                     )
-                )
 
-        return advisories
+            # TODO: Handle the CVE-????-????? case
+            yield AdvisoryData(
+                summary="",
+                references=references,
+                affected_packages=[
+                    AffectedPackage(
+                        package=PackageURL(
+                            type="alpine",
+                            name=pkg_infos["name"],
+                            qualifiers={
+                                "arch": arch,
+                                "distroversion": distroversion,
+                                "reponame": reponame,
+                            },
+                        ),
+                        fixed_version=AlpineLinuxVersion(version),
+                    )
+                    for arch in archs
+                ],
+                aliases=[vuln_ids[0] if is_cve(vuln_ids[0]) else ""],
+            )

vulnerabilities/importers/nginx.py

@@ -19,7 +19,6 @@
 #  for any legal advice.
 #  VulnerableCode is a free software tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
-
 import asyncio
 import dataclasses
 import datetime

vulnerabilities/improvers/default.py

@@ -1,6 +1,7 @@
 from itertools import chain
 from typing import Iterable
 from typing import List
+from typing import Tuple
 
 from django.db.models.query import QuerySet
 from packageurl import PackageURL
@@ -38,7 +39,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 +68,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)