Add PyPI OSV importer

Reference: #607
Signed-off-by: Ziad <ziadhany2016@gmail.com>

add the necessary changes

Signed-off-by: Ziad <ziadhany2016@gmail.com>

remove aliases de-duplicate

Signed-off-by: Ziad <ziadhany2016@gmail.com>

reslove conflicts

Signed-off-by: Ziad <ziadhany2016@gmail.com>

add dateparser to setup.cfg

Signed-off-by: Ziad <ziadhany2016@gmail.com>

Resolving conflicts

Signed-off-by: Ziad <ziadhany2016@gmail.com>

Author: ziadhany

requirements.txt

@@ -113,3 +113,4 @@ wcwidth==0.2.5
 websocket-client==0.59.0
 yarl==1.7.2
 zipp==3.8.0
+dateparser==1.1.1

setup.cfg

@@ -75,6 +75,7 @@ install_requires =
     lxml>=4.6.4
     defusedxml>=0.7.1
     Markdown>=3.3.0
+    dateparser>=1.1.1
 
     # networking
     GitPython>=3.1.17

vulnerabilities/helpers.py

@@ -242,3 +242,16 @@ def get_item(object: dict, *attributes):
             return None
         item = item[attribute]
     return item
+
+
+def dedupe(original: List) -> List:
+    """
+    Remove all duplicate items and return a new list
+    >>> dedupe(["z","i","a","a","d","d"])
+    ['z', 'i', 'a', 'd']
+    """
+    new_list = []
+    for i in original:
+        if i not in new_list:
+            new_list.append(i)
+    return new_list

vulnerabilities/importers/__init__.py

@@ -24,13 +24,15 @@
 from vulnerabilities.importers import nginx
 from vulnerabilities.importers import nvd
 from vulnerabilities.importers import openssl
+from vulnerabilities.importers import pysec
 
 IMPORTERS_REGISTRY = [
     nginx.NginxImporter,
     alpine_linux.AlpineImporter,
     github.GitHubAPIImporter,
     nvd.NVDImporter,
     openssl.OpensslImporter,
+    pysec.PyPIImporter,
 ]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/pysec.py

@@ -0,0 +1,220 @@
+# Copyright (c) 2017 nexB Inc. and others. All rights reserved.
+# http://nexb.com and https://github.com/nexB/vulnerablecode/
+# The VulnerableCode software is licensed under the Apache License version 2.0.
+# Data generated with VulnerableCode require an acknowledgment.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# When you publish or redistribute any data created with VulnerableCode or any VulnerableCode
+# derivative work, you must accompany this data with the following acknowledgment:
+#
+#  Generated with VulnerableCode and provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+#  OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+#  VulnerableCode should be considered or used as legal advice. Consult an Attorney
+#  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
+import logging
+from builtins import set
+from io import BytesIO
+from typing import Iterable
+from zipfile import ZipFile
+
+import requests
+from dateparser import parse
+from packageurl import PackageURL
+from univers.version_range import PypiVersionRange
+from univers.versions import InvalidVersion
+from univers.versions import PypiVersion
+from univers.versions import SemverVersion
+
+from vulnerabilities.helpers import dedupe
+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.severity_systems import SCORING_SYSTEMS
+
+logger = logging.getLogger(__name__)
+
+
+class PyPIImporter(Importer):
+    license_url = "https://github.com/pypa/advisory-database/blob/main/LICENSE"
+    spdx_license_expression = "CC-BY-4.0"
+
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        """
+        1. Fetch the data from osv api
+        2. unzip the file
+        3. open the file one by one
+        4. yield the json file to parse_advisory_data
+        """
+        url = "https://osv-vulnerabilities.storage.googleapis.com/PyPI/all.zip"
+        response = requests.get(url).content
+        with ZipFile(BytesIO(response)) as zip_file:
+            for file_name in zip_file.namelist():
+                with zip_file.open(file_name) as f:
+                    vul_info = json.loads(f.read())
+                    yield parse_advisory_data(vul_info)
+
+
+def parse_advisory_data(raw_data: dict) -> AdvisoryData:
+    summary = raw_data.get("summary") or ""
+    aliases = get_aliases(raw_data)
+    date_published = get_published_date(raw_data)
+    severity = get_severity(raw_data)
+    references = get_references(raw_data, severity)
+
+    affected_packages = []
+    if not "affected" in raw_data:
+        logger.error(f"affected_packages not found - {raw_data['id'] !r}")
+
+    if "affected" in raw_data:
+        for affected_pkg in raw_data["affected"]:
+            purl = get_aff_purl(affected_pkg, raw_data["id"])
+            if purl.type == "pypi":
+                affected_version_range = get_aff_version_range(affected_pkg, raw_data["id"])
+                for fixed_range in affected_pkg.get("ranges", []):
+                    fixed_version = get_fixed_version(fixed_range, raw_data["id"])
+
+                    for version in fixed_version:
+                        affected_packages.append(
+                            AffectedPackage(
+                                package=purl,
+                                affected_version_range=affected_version_range,
+                                fixed_version=version,
+                            )
+                        )
+
+    return AdvisoryData(
+        aliases=aliases,
+        summary=summary,
+        affected_packages=affected_packages,
+        references=references,
+        date_published=date_published,
+    )
+
+
+def fixed_filter(fixed_range) -> []:
+    """
+    >>> fixed_filter({"type": "SEMVER", "events": [{"introduced": "0"}, {"fixed": "1.6.0"}]})
+    ['1.6.0']
+    >>> fixed_filter({"type": "ECOSYSTEM","events":[{"introduced": "0"},{"fixed": "1.0.0"},{"fixed": "9.0.0"}]})
+    ['1.0.0', '9.0.0']
+    """
+    filter_fixed = list(filter(lambda x: x.keys() == {"fixed"}, fixed_range["events"]))
+    list_fixed = [i["fixed"] for i in filter_fixed]
+    return list_fixed
+
+
+def get_aliases(raw_data) -> []:
+    """
+    aliases field is optional , id is required and these are all aliases from our perspective
+    converting list of two fields to a dict then , convert it to a list to make sure a list is unique
+    >>> get_aliases({"id": "GHSA-j3f7-7rmc-6wqj"})
+    ['GHSA-j3f7-7rmc-6wqj']
+    >>> get_aliases({"aliases": ["CVE-2021-40831"]})
+    ['CVE-2021-40831']
+    >>> get_aliases({"aliases": ["CVE-2022-22817", "GHSA-8vj2-vxx3-667w"], "id": "GHSA-j3f7-7rmc-6wqj"})
+    ['CVE-2022-22817', 'GHSA-8vj2-vxx3-667w', 'GHSA-j3f7-7rmc-6wqj']
+    """
+    vulnerability_id = raw_data.get("id")
+    vulnerability_aliases = raw_data.get("aliases") or []
+    if vulnerability_id:
+        vulnerability_aliases.append(vulnerability_id)
+    return vulnerability_aliases
+
+
+def get_published_date(raw_data):
+    if "published" in raw_data:
+        return parse(raw_data["published"])
+    else:
+        logger.warning(f"date_published not found {raw_data['id'] !r}")
+
+
+def get_severity(raw_data) -> []:
+    severity = []
+    if "severity" in raw_data:
+        for sever_list in raw_data["severity"]:
+            if "type" in sever_list and sever_list["type"] == "CVSS_V3":
+                severity.append(
+                    VulnerabilitySeverity(
+                        system=SCORING_SYSTEMS["cvssv3_vector"],
+                        value=sever_list["score"],
+                    )
+                )
+    if "ecosystem_specific" in raw_data and "severity" in raw_data["ecosystem_specific"]:
+        severity.append(
+            VulnerabilitySeverity(
+                system=SCORING_SYSTEMS["generic_textual"],
+                value=raw_data["ecosystem_specific"]["severity"],
+            )
+        )
+    else:
+        logger.warning(f"severity not found- {raw_data['id']!r}")
+
+    return severity
+
+
+def get_references(raw_data, severity) -> []:
+    if "references" in raw_data:
+        return [
+            Reference(url=ref["url"], severities=severity) for ref in raw_data["references"] if ref
+        ]
+    else:
+        return []
+
+
+def get_aff_purl(affected_pkg, raw_id):
+    package = affected_pkg["package"]
+    if "purl" in package:
+        try:
+            return PackageURL.from_string(package["purl"])
+        except ValueError:
+            logger.error(f"PackageURL ValueError - {raw_id !r} - purl: {package['purl'] !r}")
+
+    if "ecosystem" in package and "name" in package:
+        return PackageURL(type=package["ecosystem"], name=package["name"])
+    else:
+        logger.error(f"purl affected_pkg not found - {raw_id !r}")
+
+
+def get_aff_version_range(affected_pkg, raw_id):
+    if "versions" in affected_pkg:
+        try:
+            return PypiVersionRange(affected_pkg["versions"])
+        except Exception as e:
+            logger.error(f"affected_pkg_version_range Error - {raw_id !r} - purl: {e !r}")
+    else:
+        logger.error(f"affected_pkg_version_range not found - {raw_id !r} ")
+
+
+def get_fixed_version(fixed_range, raw_id) -> []:
+    fixed_version = []
+    if "type" in fixed_range:
+        list_fixed = fixed_filter(fixed_range)
+        for i in list_fixed:
+            if fixed_range["type"] == "ECOSYSTEM":
+                try:
+                    fixed_version.append(PypiVersion(i))
+                except InvalidVersion:
+                    logger.error(f"Invalid Version - PypiVersion - {raw_id !r} - {i !r}")
+            if fixed_range["type"] == "SEMVER":
+                try:
+                    fixed_version.append(SemverVersion(i))
+                except InvalidVersion:
+                    logger.error(f"Invalid Version - SemverVersion - {raw_id !r} - {i !r}")
+            if fixed_range["type"] == "GIT":
+                # TODO add GitHubVersion univers fix_version
+                logger.error(f"NotImplementedError GIT Version - {raw_id !r} - {i !r}")
+    else:
+        logger.error(f"Invalid type - {raw_id!r}")
+
+    return dedupe(fixed_version)