Merge remote-tracking branch 'origin/main' into faster-vuln-search

Author: pombredanne

requirements.txt

@@ -106,11 +106,11 @@ toml==0.10.2
 tomli==2.0.1
 traitlets==5.1.1
 typing_extensions==4.1.1
-univers==30.7.0
+univers==30.9.0
 urllib3==1.26.9
 wcwidth==0.2.5
 websocket-client==0.59.0
 yarl==1.7.2
 zipp==3.8.0
 dateparser==1.1.1
-fetchcode==0.1.0
+fetchcode==0.2.0

setup.cfg

@@ -65,7 +65,7 @@ install_requires =
 
     #essentials
     packageurl-python>=0.9.4
-    univers>=30.3.1
+    univers>=30.9.0
     license-expression>=21.6.14
 
     # file and data formats
@@ -82,7 +82,7 @@ install_requires =
     # networking
     GitPython>=3.1.17
     requests>=2.25.1
-    fetchcode>=0.1.0
+    fetchcode>=0.2.0
 
 [options.extras_require]
 dev =

vulnerabilities/importer.py

@@ -22,12 +22,16 @@
 from typing import Set
 from typing import Tuple
 
+import pytz
+from dateutil import parser as dateparser
 from fetchcode.vcs import fetch_via_vcs
 from license_expression import Licensing
 from packageurl import PackageURL
+from univers.version_range import RANGE_CLASS_BY_SCHEMES
 from univers.version_range import VersionRange
 from univers.versions import Version
 
+from vulnerabilities import severity_systems
 from vulnerabilities.oval_parser import OvalParser
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 from vulnerabilities.severity_systems import ScoringSystem
@@ -350,13 +354,13 @@ class OvalImporter(Importer):
     """
 
     @staticmethod
-    def create_purl(pkg_name: str, pkg_version: str, pkg_data: Mapping) -> PackageURL:
+    def create_purl(pkg_name: str, pkg_data: Mapping) -> PackageURL:
         """
         Helper method for creating different purls for subclasses without them reimplementing
         get_data_from_xml_doc  method
         Note: pkg_data must include 'type' of package
         """
-        return PackageURL(name=pkg_name, version=pkg_version, **pkg_data)
+        return PackageURL(name=pkg_name, **pkg_data)
 
     @staticmethod
     def _collect_pkgs(parsed_oval_data: Mapping) -> Set:
@@ -390,28 +394,17 @@ def advisory_data(self) -> List[AdvisoryData]:
         for metadata, oval_file in self._fetch():
             try:
                 oval_data = self.get_data_from_xml_doc(oval_file, metadata)
-                yield oval_data
+                yield from oval_data
             except Exception:
                 logger.error(
                     f"Failed to get updated_advisories: {oval_file!r} "
                     f"with {metadata!r}:\n" + traceback.format_exc()
                 )
                 continue
 
-    def set_api(self, all_pkgs: Iterable[str]):
-        """
-        This method loads the self.pkg_manager_api with the specified packages.
-        It fetches and caches all the versions of these packages and exposes
-        them through self.pkg_manager_api.get(<package_name>). Example
-
-        >> self.set_api(['electron'])
-        Assume 'electron' has only versions 1.0.0 and 1.2.0
-        >> assert  self.pkg_manager_api.get('electron') == {'1.0.0','1.2.0'}
-
-        """
-        raise NotImplementedError
-
-    def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> List[AdvisoryData]:
+    def get_data_from_xml_doc(
+        self, xml_doc: ET.ElementTree, pkg_metadata={}
+    ) -> Iterable[AdvisoryData]:
         """
         The orchestration method of the OvalDataSource. This method breaks an
         OVAL xml ElementTree into a list of `Advisory`.
@@ -422,66 +415,58 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
         Example value of pkg_metadata:
                 {"type":"deb","qualifiers":{"distro":"buster"} }
         """
-
-        all_adv = []
-        oval_doc = OvalParser(self.translations, xml_doc)
-        raw_data = oval_doc.get_data()
-        all_pkgs = self._collect_pkgs(raw_data)
-        self.set_api(all_pkgs)
+        oval_parsed_data = OvalParser(self.translations, xml_doc)
+        raw_data = oval_parsed_data.get_data()
+        oval_doc = oval_parsed_data.oval_document
+        timestamp = oval_doc.getGenerator().getTimestamp()
 
         # convert definition_data to Advisory objects
         for definition_data in raw_data:
             # These fields are definition level, i.e common for all elements
             # connected/linked to an OvalDefinition
             vuln_id = definition_data["vuln_id"]
             description = definition_data["description"]
-            references = [Reference(url=url) for url in definition_data["reference_urls"]]
+            severities = (
+                [
+                    VulnerabilitySeverity(
+                        system=severity_systems.GENERIC, value=definition_data.get("severity")
+                    )
+                ]
+                if definition_data.get("severity")
+                else []
+            )
+            references = [
+                Reference(url=url, severities=severities)
+                for url in definition_data["reference_urls"]
+            ]
             affected_packages = []
             for test_data in definition_data["test_data"]:
                 for package_name in test_data["package_list"]:
-                    if package_name and len(package_name) >= 50:
-                        continue
-
-                    affected_version_range = test_data["version_ranges"] or set()
-                    version_class = version_class_by_package_type[pkg_metadata["type"]]
-                    version_scheme = version_class.scheme
-
-                    affected_version_range = VersionRange.from_scheme_version_spec_string(
-                        version_scheme, affected_version_range
-                    )
-                    all_versions = self.pkg_manager_api.get(package_name).valid_versions
-
-                    # FIXME: what is this 50 DB limit? that's too small for versions
-                    # FIXME: we should not drop data this way
-                    # This filter is for filtering out long versions.
-                    # 50 is limit because that's what db permits atm.
-                    all_versions = [version for version in all_versions if len(version) < 50]
-                    if not all_versions:
-                        continue
-
-                    affected_purls = []
-                    safe_purls = []
-                    for version in all_versions:
-                        purl = self.create_purl(
-                            pkg_name=package_name,
-                            pkg_version=version,
-                            pkg_data=pkg_metadata,
+                    affected_version_range = test_data["version_ranges"]
+                    vrc = RANGE_CLASS_BY_SCHEMES[pkg_metadata["type"]]
+                    if affected_version_range:
+                        try:
+                            affected_version_range = vrc.from_native(affected_version_range)
+                        except Exception as e:
+                            logger.error(
+                                f"Failed to parse version range {affected_version_range!r} "
+                                f"for package {package_name!r}:\n{e}"
+                            )
+                            continue
+                    if package_name:
+                        affected_packages.append(
+                            AffectedPackage(
+                                package=self.create_purl(package_name, pkg_metadata),
+                                affected_version_range=affected_version_range,
+                            )
                         )
-                        if version_class(version) in affected_version_range:
-                            affected_purls.append(purl)
-                        else:
-                            safe_purls.append(purl)
-
-                    affected_packages.extend(
-                        nearest_patched_package(affected_purls, safe_purls),
-                    )
-
-            all_adv.append(
-                AdvisoryData(
-                    summary=description,
-                    affected_packages=affected_packages,
-                    vulnerability_id=vuln_id,
-                    references=references,
-                )
+            date_published = dateparser.parse(timestamp)
+            if not date_published.tzinfo:
+                date_published = date_published.replace(tzinfo=pytz.UTC)
+            yield AdvisoryData(
+                aliases=[vuln_id],
+                summary=description,
+                affected_packages=affected_packages,
+                references=sorted(references),
+                date_published=date_published,
             )
-        return all_adv

vulnerabilities/importers/__init__.py

@@ -8,7 +8,9 @@
 #
 
 from vulnerabilities.importers import alpine_linux
+from vulnerabilities.importers import archlinux
 from vulnerabilities.importers import debian
+from vulnerabilities.importers import debian_oval
 from vulnerabilities.importers import github
 from vulnerabilities.importers import gitlab
 from vulnerabilities.importers import nginx
@@ -17,6 +19,7 @@
 from vulnerabilities.importers import pypa
 from vulnerabilities.importers import pysec
 from vulnerabilities.importers import redhat
+from vulnerabilities.importers import ubuntu
 
 IMPORTERS_REGISTRY = [
     nginx.NginxImporter,
@@ -29,6 +32,9 @@
     debian.DebianImporter,
     gitlab.GitLabAPIImporter,
     pypa.PyPaImporter,
+    archlinux.ArchlinuxImporter,
+    ubuntu.UbuntuImporter,
+    debian_oval.DebianOvalImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/alpine_linux.py

@@ -31,6 +31,7 @@
 from vulnerabilities.references import WireSharkReference
 from vulnerabilities.references import XsaReference
 from vulnerabilities.references import ZbxReference
+from vulnerabilities.utils import fetch_response
 from vulnerabilities.utils import is_cve
 
 LOGGER = logging.getLogger(__name__)
@@ -58,16 +59,6 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
             yield from process_record(record)
 
 
-def fetch_response(url):
-    """
-    Fetch and return `response` from the `url`
-    """
-    response = requests.get(url)
-    if response.status_code == 200:
-        return response
-    raise Exception(f"Failed to fetch data from {url!r} with status code: {response.status_code!r}")
-
-
 def fetch_advisory_directory_links(page_response_content: str) -> List[str]:
     """
     Return a list of advisory directory links present in `page_response_content` html string

vulnerabilities/importers/archlinux.py

@@ -6,66 +6,64 @@
 # See https://github.com/nexB/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
-import dataclasses
-import json
+
 from typing import Iterable
 from typing import List
 from typing import Mapping
-from typing import Set
 from urllib.request import urlopen
 
 from packageurl import PackageURL
+from univers.version_range import ArchLinuxVersionRange
+from univers.versions import ArchLinuxVersion
 
 from vulnerabilities import severity_systems
 from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
-from vulnerabilities.utils import nearest_patched_package
+from vulnerabilities.utils import fetch_response
 
 
 class ArchlinuxImporter(Importer):
-    def __enter__(self):
-        self._api_response = self._fetch()
-
-    def updated_advisories(self) -> Set[AdvisoryData]:
-        advisories = []
+    url = "https://security.archlinux.org/json"
+    spdx_license_expression = "MIT"
+    license_url = "https://github.com/archlinux/arch-security-tracker/blob/master/LICENSE"
 
-        for record in self._api_response:
-            advisories.extend(self._parse(record))
+    def fetch(self) -> Iterable[Mapping]:
+        response = fetch_response(self.url)
+        return response.json()
 
-        return self.batch_advisories(advisories)
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        for record in self.fetch():
+            yield from self.parse_advisory(record)
 
-    def _fetch(self) -> Iterable[Mapping]:
-        with urlopen(self.config.archlinux_tracker_url) as response:
-            return json.load(response)
-
-    def _parse(self, record) -> List[AdvisoryData]:
+    def parse_advisory(self, record) -> List[AdvisoryData]:
         advisories = []
-
-        for cve_id in record["issues"]:
+        aliases = record.get("issues") or []
+        for alias in aliases:
             affected_packages = []
             for name in record["packages"]:
-                impacted_purls, resolved_purls = [], []
-                impacted_purls.append(
-                    PackageURL(
+                summary = record.get("type") or ""
+                if summary == "unknown":
+                    summary = ""
+                affected = record.get("affected") or ""
+                affected_version_range = (
+                    ArchLinuxVersionRange.from_versions([affected]) if affected else None
+                )
+                fixed = record.get("fixed") or ""
+                fixed_version = ArchLinuxVersion(fixed) if fixed else None
+                affected_packages = []
+                affected_package = AffectedPackage(
+                    package=PackageURL(
                         name=name,
-                        type="pacman",
+                        type="alpm",
                         namespace="archlinux",
-                        version=record["affected"],
-                    )
+                    ),
+                    affected_version_range=affected_version_range,
+                    fixed_version=fixed_version,
                 )
-
-                if record["fixed"]:
-                    resolved_purls.append(
-                        PackageURL(
-                            name=name,
-                            type="pacman",
-                            namespace="archlinux",
-                            version=record["fixed"],
-                        )
-                    )
-                affected_packages.extend(nearest_patched_package(impacted_purls, resolved_purls))
+                affected_packages.append(affected_package)
 
             references = []
             references.append(
@@ -89,9 +87,9 @@ def _parse(self, record) -> List[AdvisoryData]:
                 )
 
             advisories.append(
-                Advisory(
-                    vulnerability_id=cve_id,
-                    summary="",
+                AdvisoryData(
+                    aliases=[alias],
+                    summary=summary,
                     affected_packages=affected_packages,
                     references=references,
                 )

vulnerabilities/importers/debian.py

@@ -39,7 +39,7 @@
 
 class DebianImporter(Importer):
 
-    spdx_license_expression = "MIT"
+    spdx_license_expression = "LicenseRef-scancode-other-permissive"
     license_url = "https://www.debian.org/license"
     notice = """
     From: Tushar Goel <[email protected]>