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

Add tests to test scraping of webpages and parsing of data

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

Author: TG1999

AUTHORS.rst

@@ -13,3 +13,4 @@ The following organizations or individuals have contributed to this repo:
 - Navonil Das @NavonilDas
 - Tushar Upadhyay @tushar912
 - Hritik Vijay @hritik14
+- Tushar Goel @TG1999

requirements.txt

@@ -8,7 +8,7 @@ django-widget-tweaks>=1.4.8
 packageurl-python>=0.9.4
 binaryornot>=0.4.4
 GitPython>=3.1.17
-univers>=30.0.0
+univers>=30.1.0
 saneyaml>=0.5.2
 beautifulsoup4>=4.9.3
 python-dateutil>=2.8.1
@@ -17,5 +17,6 @@ lxml>=4.6.4
 gunicorn>=20.1.0
 django-environ==0.4.5
 defusedxml==0.7.1
+license-expression>=21.6.14
 
 Markdown==3.3.4

vulnerabilities/importer.py

@@ -38,6 +38,7 @@
 from binaryornot.helpers import is_binary_string
 from git import DiffIndex
 from git import Repo
+from license_expression import Licensing
 from packageurl import PackageURL
 from univers.version_range import VersionRange
 from univers.versions import Version
@@ -117,22 +118,28 @@ 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] = None
     fixed_version: Optional[Version] = None
 
     def __post_init__(self):
         if self.package.version:
-            raise ValueError
+            raise ValueError("The PackageURL supplied must not have a version")
+        if not (self.affected_version_range or self.fixed_version):
+            raise ValueError(
+                "Affected Package should at least have either a fixed version or affected version range"
+            )
 
     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("Affected package should have a fixed version")
+        fixed_purl = self.package._replace(version=str(self.fixed_version))
         return fixed_purl
 
     @classmethod
@@ -203,7 +210,7 @@ class AdvisoryData:
     """
 
     aliases: List[str] = dataclasses.field(default_factory=list)
-    summary: str = None
+    summary: Optional[str] = None
     affected_packages: List[AffectedPackage] = dataclasses.field(default_factory=list)
     references: List[Reference] = dataclasses.field(default_factory=list)
     date_published: Optional[datetime.datetime] = None
@@ -224,10 +231,18 @@ class Importer:
     """
 
     spdx_license_expression = ""
+    license_url = ""
 
     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 Exception as e:
+            raise Exception(
+                f"{self.spdx_license_expression!r} is not a valid SPDX license expression"
+            ) from e
 
     @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

@@ -21,123 +21,190 @@
 #  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 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
+from vulnerabilities.references import WireSharkReference
+from vulnerabilities.references import XsaReference
+from vulnerabilities.references import ZbxReference
 
 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]
+    spdx_license_expression = "CC-BY-SA-4.0"
+    license_url = "https://secdb.alpinelinux.org/license.txt"
 
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        advisories = []
+        page_response_content = fetch_response(BASE_URL).content
+        advisory_directory_links = fetch_advisory_directory_links(page_response_content)
         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_directory_page = fetch_response(advisory_directory_link).content
             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")
-                ]
+                fetch_advisory_links(advisory_directory_page, advisory_directory_link)
             )
-
-        return advisory_links
-
-    def updated_advisories(self) -> Set[Advisory]:
-        advisories = []
-        advisory_links = self.fetch_advisory_links()
         for link in advisory_links:
-            advisories.extend(self._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)
-
-        if record["packages"] is None:
-            return advisories
-
-        for p in record["packages"]:
-            advisories.extend(
-                self._load_advisories(
-                    p["pkg"],
-                )
-            )
-
+            record = fetch_response(link).json()
+            if not record["packages"]:
+                raise Exception(f'"packages" not found in {link}')
+            advisories.extend(process_record(record))
         return advisories
 
-    @staticmethod
-    def _load_advisories(
-        pkg_infos: Mapping[str, Any],
-    ) -> List[Advisory]:
 
-        advisories = []
-
-        for fixed_vulns in pkg_infos["secfixes"].values():
-
-            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:]:
-
-                    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),
-                            )
+def fetch_response(url):
+    """
+    Fetch and return `response` from the `url`
+    """
+    response = requests.get(url)
+    if response.status_code == 200:
+        return response
+    raise Exception("Failed to fetch data from the URL")
+
+
+def fetch_advisory_directory_links(page_response_content):
+    """
+    Return `advisory_directory_links` present in `index_page`
+    """
+    index_page = BeautifulSoup(page_response_content, features="lxml")
+    alpine_versions = [link.text for link in index_page.find_all("a") if link.text.startswith("v")]
+
+    assert alpine_versions, f"No versions found in {BASE_URL}"
+
+    advisory_directory_links = [f"{BASE_URL}{version}" for version in alpine_versions]
+
+    return advisory_directory_links
+
+
+def fetch_advisory_links(advisory_directory_page, advisory_directory_link):
+    """
+    Yield json file urls present in `advisory_directory_page`
+    """
+    advisory_directory_page = BeautifulSoup(advisory_directory_page, features="lxml")
+    anchor_tags = advisory_directory_page.find_all("a")
+    assert anchor_tags, f"No anchor tags found in {advisory_directory_link}"
+    for anchor_tag in anchor_tags:
+        if anchor_tag.text.endswith("json"):
+            yield f"{advisory_directory_link}{anchor_tag.text}"
+
+
+def process_record(record: dict) -> List[AdvisoryData]:
+    """
+    Return a list of AdvisoryData objects by processing data
+    present in that `link`
+    """
+
+    advisories: List[AdvisoryData] = []
+
+    for package in record["packages"]:
+        assert package["pkg"]
+        attributes = ["distroversion", "reponame", "archs"]
+        for attribute in attributes:
+            assert attribute in record
+        loaded_advisories = load_advisories(
+            package["pkg"],
+            record["distroversion"],
+            record["reponame"],
+            record["archs"],
+        )
+        advisories.extend(loaded_advisories)
+    return advisories
+
+
+def load_advisories(
+    pkg_infos: Mapping[str, Any],
+    distroversion: str,
+    reponame: str,
+    archs: Iterable[str],
+) -> Iterable[AdvisoryData]:
+    """
+    Yields AdvisoryData by mapping data from `pkg_infos`
+    and form PURL for AffectedPackages by using
+    `distroversion`, `reponame`, `archs`
+    """
+
+    if not pkg_infos.get("name"):
+        raise Exception('"name" is not available in package')
+
+    for version, fixed_vulns in pkg_infos["secfixes"].items():
+
+        if fixed_vulns is None:
+            raise Exception(f"No fixed vulnerabilities in {version}")
+
+        for vuln_ids in fixed_vulns:
+            assert isinstance(vuln_ids, str)
+            vuln_ids = vuln_ids.split()
+            aliases = []
+            vuln_id = vuln_ids[0]
+            # check for valid vuln ID, if there is valid vuln ID then iterate over
+            # the remaining elements of the list else iterate over the whole list
+            # and also check if the initial element is a reference or not
+            if is_cve(vuln_id):
+                aliases = [vuln_id]
+                vuln_ids = vuln_ids[1:]
+            references = []
+            for reference_id in vuln_ids:
+
+                if reference_id.startswith("XSA"):
+                    references.append(XsaReference.from_id(xsa_id=reference_id))
+
+                elif reference_id.startswith("ZBX"):
+                    references.append(ZbxReference.from_id(zbx_id=reference_id))
+
+                elif reference_id.startswith("wnpa-sec"):
+                    references.append(WireSharkReference.from_id(wnpa_sec_id=reference_id))
+
+            qualifiers = {
+                "distroversion": distroversion,
+                "reponame": reponame,
+            }
+
+            affected_packages = []
+
+            try:
+                fixed_version = AlpineLinuxVersion(version)
+            except Exception as e:
+                raise Exception(f"{version!r} Not a valid Alpine Version") from e
+            assert isinstance(archs, List)
+            if archs:
+                for arch in archs:
+                    qualifiers["arch"] = arch
+                    affected_packages.append(
+                        AffectedPackage(
+                            package=PackageURL(
+                                type="alpine",
+                                name=pkg_infos["name"],
+                                qualifiers=qualifiers,
+                            ),
+                            fixed_version=fixed_version,
                         )
-
-                    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 "",
+                    )
+            else:
+                # there is no arch, this is not an arch-specific package
+                affected_packages.append(
+                    AffectedPackage(
+                        package=PackageURL(
+                            type="alpine",
+                            name=pkg_infos["name"],
+                            qualifiers=qualifiers,
+                        ),
+                        fixed_version=fixed_version,
                     )
                 )
 
-        return advisories
+            yield AdvisoryData(
+                references=references,
+                affected_packages=affected_packages,
+                aliases=aliases,
+            )

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