add alpine improver

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

Author: TG1999

vulnerabilities/importer.py

@@ -159,7 +159,8 @@ def merge(cls, affected_packages: Iterable):
         fixed_versions = set()
         purls = set()
         for pkg in affected_packages:
-            affected_version_ranges.add(pkg.affected_version_range)
+            if pkg.affected_version_range:
+                affected_version_ranges.add(pkg.affected_version_range)
             if pkg.fixed_version:
                 fixed_versions.add(pkg.fixed_version)
             purls.add(pkg.package)
@@ -171,9 +172,12 @@ def to_dict(self):
         """
         Return a serializable dict that can be converted back using self.from_dict
         """
+        affected_version_range = None
+        if self.affected_version_range:
+            affected_version_range = str(self.affected_version_range)
         return {
             "package": self.package.to_dict(),
-            "affected_version_range": str(self.affected_version_range),
+            "affected_version_range": affected_version_range,
             "fixed_version": str(self.fixed_version) if self.fixed_version else None,
         }
 
@@ -183,9 +187,11 @@ def from_dict(cls, affected_pkg: dict):
         Return an AffectedPackage object from dict generated by self.to_dict
         """
         package = PackageURL(**affected_pkg["package"])
-        affected_version_range = VersionRange.from_string(affected_pkg["affected_version_range"])
+        affected_version_range = None
+        if affected_pkg["affected_version_range"]:
+            affected_version_range = VersionRange.from_string(affected_pkg["affected_version_range"])
         fixed_version = affected_pkg["fixed_version"]
-        if fixed_version:
+        if fixed_version and affected_version_range:
             # TODO: revisit after https://github.com/nexB/univers/issues/10
             fixed_version = affected_version_range.version_class(fixed_version)
 
@@ -224,6 +230,10 @@ class NoLicenseError(Exception):
     pass
 
 
+class InvalidSPDXLicense(Exception):
+    pass
+
+
 class Importer:
     """
     An Importer collects data from various upstreams and returns corresponding AdvisoryData objects
@@ -239,8 +249,8 @@ def __init__(self):
         licensing = Licensing()
         try:
             licensing.parse(self.spdx_license_expression)
-        except Exception as e:
-            raise Exception(
+        except InvalidSPDXLicense as e:
+            raise ValueError(
                 f"{self.spdx_license_expression!r} is not a valid SPDX license expression"
             ) from e
 

vulnerabilities/importers/alpine_linux.py

@@ -25,16 +25,22 @@
 from typing import Iterable
 from typing import List
 from typing import Mapping
+from urllib.parse import urljoin
 
 import requests
 from bs4 import BeautifulSoup
+from django.db.models.query import QuerySet
 from packageurl import PackageURL
 from univers.versions import AlpineLinuxVersion
 
 from vulnerabilities.helpers import is_cve
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
+from vulnerabilities.improver import MAX_CONFIDENCE
+from vulnerabilities.improver import Improver
+from vulnerabilities.improver import Inference
+from vulnerabilities.models import Advisory
 from vulnerabilities.references import WireSharkReference
 from vulnerabilities.references import XsaReference
 from vulnerabilities.references import ZbxReference
@@ -60,7 +66,8 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
         for link in advisory_links:
             record = fetch_response(link).json()
             if not record["packages"]:
-                raise Exception(f'"packages" not found in {link}')
+                LOGGER.error(f'"packages" not found in {link!r}')
+                continue
             advisories.extend(process_record(record))
         return advisories
 
@@ -72,50 +79,67 @@ def fetch_response(url):
     response = requests.get(url)
     if response.status_code == 200:
         return response
-    raise Exception("Failed to fetch data from the URL")
+    raise Exception(f"Failed to fetch data from {url!r}")
 
 
-def fetch_advisory_directory_links(page_response_content):
+def fetch_advisory_directory_links(page_response_content: str) -> List[str]:
     """
-    Return `advisory_directory_links` present in `index_page`
+    Return a list of advisory directory links present in `page_response_content` html string
     """
     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}"
+    if not alpine_versions:
+        LOGGER.error(f"No versions found in {BASE_URL!r}")
+        return []
 
-    advisory_directory_links = [f"{BASE_URL}{version}" for version in alpine_versions]
+    advisory_directory_links = [urljoin(BASE_URL, version) for version in alpine_versions]
 
     return advisory_directory_links
 
 
-def fetch_advisory_links(advisory_directory_page, advisory_directory_link):
+def fetch_advisory_links(
+    advisory_directory_page: str, advisory_directory_link: str
+) -> Iterable[str]:
     """
     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}"
+    if not anchor_tags:
+        LOGGER.error(f"No anchor tags found in {advisory_directory_link!r}")
+        return iter([])
     for anchor_tag in anchor_tags:
         if anchor_tag.text.endswith("json"):
             yield f"{advisory_directory_link}{anchor_tag.text}"
 
 
+def check_for_attributes(record) -> bool:
+    attributes = ["distroversion", "reponame", "archs"]
+    for attribute in attributes:
+        if attribute not in record:
+            LOGGER.error(f'"{attribute!r}" not found in {record!r}')
+            return False
+    return True
+
+
 def process_record(record: dict) -> List[AdvisoryData]:
     """
     Return a list of AdvisoryData objects by processing data
     present in that `record`
     """
     if not record["packages"]:
-        raise Exception(f'"packages" not found in this {record!r}')
+        LOGGER.error(f'"packages" not found in this record {record!r}')
+        return []
 
     advisories: List[AdvisoryData] = []
 
     for package in record["packages"]:
-        assert package["pkg"]
-        attributes = ["distroversion", "reponame", "archs"]
-        for attribute in attributes:
-            assert attribute in record
+        if not package["pkg"]:
+            LOGGER.error(f'"pkg" not found in this package {package!r}')
+            continue
+        if not check_for_attributes(record):
+            continue
         loaded_advisories = load_advisories(
             package["pkg"],
             record["distroversion"],
@@ -133,18 +157,23 @@ def load_advisories(
     archs: List[str],
 ) -> Iterable[AdvisoryData]:
     """
-    Yields AdvisoryData by mapping data from `pkg_infos`
+    Yield AdvisoryData by mapping data from `pkg_infos`
     and form PURL for AffectedPackages by using
     `distroversion`, `reponame`, `archs`
     """
-    assert pkg_infos.get("name"), '"name" is not available in package'
+    if not pkg_infos.get("name"):
+        LOGGER.error(f'"name" is not available in package {pkg_infos!r}')
+        return []
 
     for version, fixed_vulns in pkg_infos["secfixes"].items():
-
-        assert fixed_vulns, f"No fixed vulnerabilities in {version}"
+        if not fixed_vulns:
+            LOGGER.error(f"No fixed vulnerabilities in version {version!r}")
+            continue
 
         for vuln_ids in fixed_vulns:
-            assert isinstance(vuln_ids, str)
+            if not isinstance(vuln_ids, str):
+                LOGGER.error(f"{vuln_ids!r} is not of `str` instance")
+                continue
             vuln_ids = vuln_ids.split()
             aliases = []
             vuln_id = vuln_ids[0]
@@ -176,8 +205,11 @@ def load_advisories(
             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)
+                LOGGER.error(f"{version!r} is not a valid AlpineVersion {e!r}")
+                continue
+            if not isinstance(archs, List):
+                LOGGER.error(f"{archs!r} is not of `List` instance")
+                continue
             if archs:
                 for arch in archs:
                     qualifiers["arch"] = arch
@@ -209,3 +241,21 @@ def load_advisories(
                 affected_packages=affected_packages,
                 aliases=aliases,
             )
+
+
+class AlpineBasicImprover(Improver):
+    @property
+    def interesting_advisories(self) -> QuerySet:
+        return Advisory.objects.filter(created_by=AlpineImporter.qualified_name)
+
+    def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
+        """
+        Generate and return Inferences for the given advisory data
+        """
+        for affected_package in advisory_data.affected_packages:
+            fixed_purl = affected_package.get_fixed_purl()
+            yield Inference.from_advisory_data(
+                advisory_data,
+                confidence=MAX_CONFIDENCE,
+                fixed_purl=fixed_purl,
+            )

vulnerabilities/improve_runner.py

@@ -75,16 +75,16 @@ def process_inferences(inferences: List[Inference], advisory: Advisory, improver
                 )
                 if updated:
                     logger.info("Severity updated for reference {ref!r} to {severity.value!r}")
-
-        for pkg in inference.affected_purls:
-            vulnerable_package, _ = _get_or_create_package(pkg)
-            models.PackageRelatedVulnerability(
-                vulnerability=vuln,
-                package=vulnerable_package,
-                created_by=improver_name,
-                confidence=inference.confidence,
-                fix=False,
-            ).update_or_create()
+        if inference.affected_purls:
+            for pkg in inference.affected_purls:
+                vulnerable_package, _ = _get_or_create_package(pkg)
+                models.PackageRelatedVulnerability(
+                    vulnerability=vuln,
+                    package=vulnerable_package,
+                    created_by=improver_name,
+                    confidence=inference.confidence,
+                    fix=False,
+                ).update_or_create()
 
         if inference.fixed_purl:
             fixed_package, _ = _get_or_create_package(inference.fixed_purl)

vulnerabilities/improver.py

@@ -27,10 +27,10 @@ class Inference:
     """
 
     vulnerability_id: str = None
-    aliases: List[str] = dataclasses.field(default_factory=list)
+    aliases: Optional[List[str]] = None
     confidence: int = MAX_CONFIDENCE
     summary: Optional[str] = None
-    affected_purls: List[PackageURL] = dataclasses.field(default_factory=list)
+    affected_purls: Optional[List[PackageURL]] = None
     fixed_purl: PackageURL = None
     references: List[Reference] = dataclasses.field(default_factory=list)
 
@@ -48,7 +48,12 @@ def __post_init__(self):
         )
 
         versionless_purls = []
-        for purl in self.affected_purls + [self.fixed_purl]:
+        purls = []
+        if self.fixed_purl:
+            purls = [self.fixed_purl]
+        if self.affected_purls:
+            purls = purls + self.affected_purls
+        for purl in purls:
             if purl and not purl.version:
                 versionless_purls.append(purl)
 
@@ -57,7 +62,7 @@ def __post_init__(self):
         ), f"Version-less purls are not supported in an Inference: {versionless_purls}"
 
     @classmethod
-    def from_advisory_data(cls, advisory_data, confidence, affected_purls, fixed_purl):
+    def from_advisory_data(cls, advisory_data, confidence, fixed_purl, affected_purls=None):
         """
         Return an Inference object while keeping the same values as of advisory_data
         for vulnerability_id, summary and references

vulnerabilities/improvers/__init__.py

@@ -1,6 +1,10 @@
 from vulnerabilities import importers
 from vulnerabilities.improvers import default
 
-IMPROVERS_REGISTRY = [default.DefaultImprover, importers.nginx.NginxBasicImprover]
+IMPROVERS_REGISTRY = [
+    default.DefaultImprover,
+    importers.nginx.NginxBasicImprover,
+    importers.alpine_linux.AlpineBasicImprover,
+]
 
 IMPROVERS_REGISTRY = {x.qualified_name: x for x in IMPROVERS_REGISTRY}

vulnerabilities/references.py

@@ -12,11 +12,11 @@ def from_id(cls, xsa_id):
         Return a new XsaReference from an XSA-XXXX id.
         """
         if not xsa_id or not xsa_id.lower().startswith("xsa"):
-            return ValueError(f"Not a Xen advisory. Does not start with XSA: {xsa_id!r}")
-        numid = xsa_id.split("-")[-1]
+            return ValueError(f"Not a Xen reference. Does not start with XSA: {xsa_id!r}")
+        _, numid = xsa_id.rsplit("-")
         return cls(
             reference_id=xsa_id,
-            url="https://xenbits.xen.org/xsa/advisory-{}.html".format(numid),
+            url=f"https://xenbits.xen.org/xsa/advisory-{numid}.html",
         )
 
 
@@ -31,10 +31,10 @@ def from_id(cls, zbx_id):
         Return a new ZbxReference from an ZBX-XXXX id.
         """
         if not zbx_id or not zbx_id.lower().startswith("zbx"):
-            return ValueError(f"Not a Zabbix advisory. Does not start with ZBX: {zbx_id!r}")
+            return ValueError(f"Not a Zabbix reference. Does not start with ZBX: {zbx_id!r}")
         return cls(
             reference_id=zbx_id,
-            url="https://support.zabbix.com/browse/{}".format(zbx_id),
+            url=f"https://support.zabbix.com/browse/{zbx_id}",
         )
 
 
@@ -50,9 +50,9 @@ def from_id(cls, wnpa_sec_id):
         """
         if not wnpa_sec_id or not wnpa_sec_id.lower().startswith("wnpa-sec"):
             return ValueError(
-                f"Not a WireShark advisory. Does not start with wnpa-sec: {wnpa_sec_id!r}"
+                f"Not a WireShark reference. Does not start with wnpa-sec: {wnpa_sec_id!r}"
             )
         return cls(
             reference_id=wnpa_sec_id,
-            url="https://www.wireshark.org/security/{}.html".format(wnpa_sec_id),
+            url=f"https://www.wireshark.org/security/{wnpa_sec_id}.html",
         )