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]>

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.PyPlImporter]
 
 IMPORTERS_REGISTRY = {x.qualified_name: x for x in IMPORTERS_REGISTRY}

vulnerabilities/importers/pysec.py

@@ -0,0 +1,174 @@
+# 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 datetime import datetime
+from datetime import timezone
+from io import BytesIO
+from typing import Iterable
+from zipfile import ZipFile
+
+import requests
+from packageurl import PackageURL
+from univers.version_range import GitHubVersionRange
+from univers.version_range import PypiVersionRange
+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 PyPlImporter(Importer):
+    spdx_license_expression = "Apache-2.0"
+
+    def advisory_data(self) -> Iterable[AdvisoryData]:
+        url = "https://osv-vulnerabilities.storage.googleapis.com/PyPI/all.zip"
+        response = requests.get(url).content
+        try:
+            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(f"Failed to fetch osv-vulnerabilities PyPI: HTTP ")
+
+
+def parse_advisory_data(raw_data: dict) -> AdvisoryData:
+
+    if "summary" in raw_data:
+        summary = raw_data["summary"]
+    elif "details" in raw_data:
+        summary = raw_data["details"][slice(50)]
+    else:
+        summary = ""
+        logger.error(f"summary not found- {raw_data['id'] !r}")
+
+    aliases = raw_data["aliases"] if "aliases" in raw_data else raw_data["id"]
+
+    if raw_data["published"] is not None:
+        date_published = datetime.strptime(
+            raw_data["published"][0:19], "%Y-%m-%dT%H:%M:%S"
+        ).replace(tzinfo=timezone.utc)
+    else:
+        date_published = None
+        logger.error("date_published not found " + raw_data["id"])
+
+    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"],
+                    )
+                )
+    elif "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:
+        severity = []
+        logger.error(f"severity not found- {raw_data['id'] !r}")
+
+    references = []
+    if "references" in raw_data:
+        for ref in raw_data["references"]:
+            if ref["url"] is None:
+                continue
+            else:
+                references.append(Reference(url=ref["url"], severities=severity))
+    else:
+        logger.error(f"references not found - {raw_data['id'] !r}")
+
+    affected_package = []
+    if "affected" in raw_data:
+        for affected_pkg in raw_data["affected"]:
+            if (affected_pkg["package"]["ecosystem"] is not None) and (
+                affected_pkg["package"]["name"] is not None
+            ):
+                purl = PackageURL(
+                    type=affected_pkg["package"]["ecosystem"], name=affected_pkg["package"]["name"]
+                )
+            else:
+                purl = ""
+                logger.error(f"purl affected_pkg not found - {raw_data['id'] !r}")
+
+            affected_version_range = (
+                PypiVersionRange(affected_pkg["versions"][slice(50)])
+                if "versions" in affected_pkg
+                else None
+            )
+
+            if "ranges" in affected_pkg:
+                for fix_range in affected_pkg["ranges"]:
+                    fixed_version = set()
+                    if "type" in fix_range:
+                        if fix_range["type"] == "ECOSYSTEM":
+                            filter_fixed = list(
+                                filter(lambda x: x.keys() == {"fixed"}, fix_range["events"])
+                            )
+                            list_fixed = [i["fixed"] for i in filter_fixed]
+                            fixed_version.add(PypiVersionRange(list_fixed))
+
+                        if fix_range["type"] == "SEMVER":
+                            filter_fixed = list(
+                                filter(lambda x: x.keys() == {"fixed"}, fix_range["events"])
+                            )
+                            list_fixed = [i["fixed"] for i in filter_fixed]
+                            for i in list_fixed:
+                                fixed_version.add(SemverVersion(i))
+
+                        if fix_range["type"] == "GIT":
+                            filter_fixed = list(
+                                filter(lambda x: x.keys() == {"fixed"}, fix_range["events"])
+                            )
+                            list_fixed = [i["fixed"] for i in filter_fixed]
+                            fixed_version.add(GitHubVersionRange(list_fixed))
+
+                    affected_package.append(
+                        AffectedPackage(
+                            package=purl,
+                            affected_version_range=affected_version_range,
+                            fixed_version=fixed_version,
+                        )
+                    )
+    else:
+        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,
+    )