Add PyPL OSV importer

Reference: #607
Signed-off-by: Ziad <[email protected]>

Add PyPL OSV

Signed-off-by: Ziad <[email protected]>

rename pypl_osv to pysec.py , add a test

Signed-off-by: Ziad <[email protected]>

squash! Add PyPI OSV importer

Signed-off-by: Ziad <[email protected]>

check items before accessing them , add logs

Signed-off-by: Ziad <[email protected]>

fix_version ranges and add logs

Signed-off-by: Ziad <[email protected]>

add more test , add multiple fixed_version for affected_pkg

Signed-off-by: Ziad <[email protected]>

add fixed_filter with a test . fix aliases , purl

Signed-off-by: Ziad <[email protected]>

add NotImplemented GIT-Version logs

Signed-off-by: Ziad <[email protected]>

add unit tests

Signed-off-by: Ziad <[email protected]>

add affected_pkg ranges , fix_version ecosystem support,

Signed-off-by: Ziad <[email protected]>

requirements.txt

@@ -20,3 +20,6 @@ defusedxml==0.7.1
 license-expression>=21.6.14
 
 Markdown==3.3.4
+
+
+dateparser~=1.1.1

vulnerabilities/importers/__init__.py

@@ -21,7 +21,8 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 from vulnerabilities.importers import alpine_linux
 from vulnerabilities.importers import nginx
+from vulnerabilities.importers import pysec
 
-IMPORTERS_REGISTRY = [nginx.NginxImporter, alpine_linux.AlpineImporter]
+IMPORTERS_REGISTRY = [nginx.NginxImporter, alpine_linux.AlpineImporter, pysec.PyPIImporter]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/pysec.py

@@ -0,0 +1,248 @@
+# 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 MavenVersionRange
+from univers.version_range import NugetVersionRange
+from univers.version_range import PypiVersionRange
+from univers.versions import InvalidVersion
+from univers.versions import MavenVersion
+from univers.versions import NugetVersion
+from univers.versions import PypiVersion
+from univers.versions import SemverVersion
+
+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):
+    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"
+        try:
+            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)
+        except requests.exceptions.RequestException:
+            logger.error("Failed to fetch osv-vulnerabilities PyPI")
+
+
+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_package = []
+    if "affected" in raw_data:
+        for affected_pkg in raw_data["affected"]:
+            purl = get_aff_purl(affected_pkg, raw_data["id"])
+            affected_version_range = get_aff_version_range(affected_pkg, raw_data["id"], purl)
+
+            for fixed_range in affected_pkg.get("ranges", []):
+                fixed_version = get_fixed_version(fixed_range, raw_data["id"], purl)
+
+                for version in fixed_version:
+                    affected_package.append(
+                        AffectedPackage(
+                            package=purl,
+                            affected_version_range=affected_version_range,
+                            fixed_version=version,
+                        )
+                    )
+    if not "affected" in raw_data:
+        logger.error(f"affected_package not found - {raw_data['id'] !r}")
+
+    return AdvisoryData(
+        aliases=aliases,
+        summary=summary,
+        affected_packages=affected_package,
+        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
+    """
+    vulnerability_id = raw_data.get("id")
+    vulnerability_aliases = raw_data.get("aliases") or []
+    if vulnerability_id:
+        vulnerability_aliases.append(vulnerability_id)
+    return list(set(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, purl):
+    if "versions" in affected_pkg and len(affected_pkg["versions"]) < 200:
+        if purl.type == "pypi":
+            try:
+                return PypiVersionRange(affected_pkg["versions"])
+            except Exception as e:
+                logger.error(f"affected_pkg_version_range Error - {raw_id !r} - purl: {e !r}")
+
+        elif purl.type == "maven":
+            try:
+                return MavenVersionRange(affected_pkg["versions"])
+            except Exception as e:
+                logger.error(f"affected_pkg_version_range Error - {raw_id !r} - purl: {e !r}")
+
+        elif purl.type == "nuget":
+            try:
+                NugetVersionRange(affected_pkg["versions"])
+            except Exception as e:
+                logger.error(f"Invalid Version - NugetVersion - {raw_id !r} - {e !r}")
+
+        else:
+            logger.error(f"NotImplementedError Ecosystem fixed range - {raw_id !r} ")
+
+
+def get_fixed_version(fixed_range, raw_id, purl) -> {}:
+    fixed_version = set()
+    purl_type = purl.type
+    if "type" in fixed_range:
+        list_fixed = fixed_filter(fixed_range)
+        for i in list_fixed:
+            if fixed_range["type"] == "ECOSYSTEM":
+                if purl_type == "pypi":
+                    try:
+                        fixed_version.add(PypiVersion(i))
+                    except InvalidVersion:
+                        logger.error(f"Invalid Version - PypiVersion - {raw_id !r} - {i !r}")
+                elif purl_type == "maven":
+                    try:
+                        fixed_version.add(MavenVersion(i))
+                    except InvalidVersion:
+                        logger.error(f"Invalid Version - MavenVersion - {raw_id !r} - {i !r}")
+                elif purl_type == "nuget":
+                    try:
+                        fixed_version.add(NugetVersion(i))
+                    except InvalidVersion:
+                        logger.error(f"Invalid Version - NugetVersion - {raw_id !r} - {i !r}")
+
+                else:
+                    logger.error(f"NotImplementedError Ecosystem aff range - {raw_id !r} - {i !r}")
+            if fixed_range["type"] == "SEMVER":
+                try:
+                    fixed_version.add(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 fixed_version