Added safetydb datasource

pyproject.toml

@@ -59,7 +59,6 @@ addopts = [
     "--ignore=vulnerabilities/importers/retiredotnet.py",
     "--ignore=vulnerabilities/importers/ruby.py",
     "--ignore=vulnerabilities/importers/rust.py",
-    "--ignore=vulnerabilities/importers/safety_db.py",
     "--ignore=vulnerabilities/importers/suse_backports.py",
     "--ignore=vulnerabilities/importers/suse_scores.py",
     "--ignore=vulnerabilities/importers/ubuntu_usn.py",

vulntotal/datasources/__init__.py

@@ -12,13 +12,15 @@
 from vulntotal.datasources import gitlab
 from vulntotal.datasources import oss_index
 from vulntotal.datasources import osv
+from vulntotal.datasources import safetydb
 from vulntotal.datasources import snyk
 from vulntotal.datasources import vulnerablecode
 from vulntotal.validator import DataSource
 
 DATASOURCE_REGISTRY = {
     "deps": deps.DepsDataSource,
     "github": github.GithubDataSource,
+    "safetydb": safetydb.SafetydbDataSource,
     "gitlab": gitlab.GitlabDataSource,
     "oss_index": oss_index.OSSDataSource,
     "osv": osv.OSVDataSource,

vulntotal/datasources/safetydb.py

@@ -0,0 +1,109 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import logging
+from typing import Iterable
+from typing import List
+
+import requests
+from packageurl import PackageURL
+
+from vulntotal.validator import DataSource
+from vulntotal.validator import InvalidCVEError
+from vulntotal.validator import VendorData
+
+logger = logging.getLogger(__name__)
+
+
+class SafetydbDataSource(DataSource):
+    spdx_license_expression = "CC-BY-NC-4.0"
+    license_url = "https://github.com/pyupio/safety-db/blob/master/LICENSE.txt"
+    url = "https://raw.githubusercontent.com/pyupio/safety-db/master/data/insecure_full.json"
+
+    def fetch_advisory(self):
+        """
+        Fetch entire JSON advisory from pyupio repository
+
+        Parameters:
+
+        Returns:
+            A JSON object containing the advisory information for insecure packages, or None if an error occurs while fetching data from safetydb repo's URL.
+        """
+
+        response = requests.get(self.url)
+        try:
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            logger.error(f"Error while fetching safetydb advisories: {e}")
+            return
+
+        return response.json()
+
+    def datasource_advisory(self, purl) -> Iterable[VendorData]:
+        if purl.type not in self.supported_ecosystem():
+            return []
+        advisory = self.fetch_advisory()
+        self._raw_dump.append(advisory)
+        return parse_advisory(advisory, purl)
+
+    def datasource_advisory_from_cve(self, cve: str) -> Iterable[VendorData]:
+        if not cve.upper().startswith("CVE-"):
+            raise InvalidCVEError
+        advisory = self.fetch_advisory()
+        self._raw_dump.append(advisory)
+        return parse_advisory_for_cve(advisory, cve)
+
+    @classmethod
+    def supported_ecosystem(cls):
+        return {"pypi": "PyPI"}
+
+
+def parse_advisory(response, purl: PackageURL) -> Iterable[VendorData]:
+    """
+    Parse response from safetydb API and yield VendorData
+
+    Parameters:
+        response: A JSON object containing the response data from the safetydb datasource.
+
+    Yields:
+        VendorData instance containing the advisory information for the package.
+    """
+
+    for advisory in response.get(purl.name, []):
+        yield VendorData(
+            purl=PackageURL(purl.type, purl.namespace, purl.name),
+            aliases=[advisory.get("cve"), advisory.get("id")],
+            affected_versions=sorted(advisory.get("specs")),
+            fixed_versions=[],
+        )
+
+
+def parse_advisory_for_cve(response, cve: str) -> Iterable[VendorData]:
+    """
+    Parse response from safetydb API and yield VendorData with specified CVE
+
+    Parameters:
+        response: A JSON object containing the response data from the safetydb datasource.
+
+    Yields:
+        VendorData instance containing the advisory information for the package.
+    """
+
+    for package, advisories in response.items():
+        if package == "$meta":
+            continue
+
+        for advisory in advisories:
+            if advisory.get("cve") == cve:
+                yield VendorData(
+                    purl=PackageURL(type="pypi", name=package),
+                    aliases=[advisory.get("cve"), advisory.get("id")],
+                    affected_versions=sorted(advisory.get("specs")),
+                    fixed_versions=[],
+                )

vulntotal/tests/test_data/safetydb/advisory.json

@@ -0,0 +1,50 @@
+{
+  "$meta": {
+    "advisory": "PyUp.io metadata",
+    "base_domain": "https://pyup.io",
+    "timestamp": 1714543250
+  },
+  "flask": [
+      {
+          "advisory": "flask version Before 0.12.3 contains a CWE-20: Improper Input Validation vulnerability in flask that can result in Large amount of memory usage possibly leading to denial of service. This attack appear to be exploitable via Attacker provides JSON data in incorrect encoding. This vulnerability appears to have been fixed in 0.12.3.",
+          "cve": "CVE-2018-1000656",
+          "id": "pyup.io-36388",
+          "more_info_path": "/vulnerabilities/CVE-2018-1000656/36388",
+          "specs": [
+              "<0.12.3"
+          ],
+          "v": "<0.12.3"
+      },
+      {
+          "advisory": "Flask 0.12.3 includes a fix for CVE-2019-1010083: Unexpected memory usage. The impact is denial of service. The attack vector is crafted encoded JSON data. NOTE: this may overlap CVE-2018-1000656.\r\nhttps://github.com/pallets/flask/pull/2695/commits/0e1e9a04aaf29ab78f721cfc79ac2a691f6e3929",
+          "cve": "CVE-2019-1010083",
+          "id": "pyup.io-38654",
+          "more_info_path": "/vulnerabilities/CVE-2019-1010083/38654",
+          "specs": [
+              "<0.12.3"
+          ],
+          "v": "<0.12.3"
+      },
+      {
+          "advisory": "flask 0.6.1 fixes a security problem that allowed clients to download arbitrary files  if the host server was a windows based operating system and the client  uses backslashes to escape the directory the files where exposed from.\r\nhttps://data.safetycli.com/vulnerabilities/PVE-2021-25820/25820/",
+          "cve": "PVE-2021-25820",
+          "id": "pyup.io-25820",
+          "more_info_path": "/vulnerabilities/PVE-2021-25820/25820",
+          "specs": [
+              "<0.6.1"
+          ],
+          "v": "<0.6.1"
+      },
+      {
+          "advisory": "Flask 2.2.5 and 2.3.2 include a fix for CVE-2023-30861: When all of the following conditions are met, a response containing data intended for one client may be cached and subsequently sent by the proxy to other clients. If the proxy also caches 'Set-Cookie' headers, it may send one client's 'session' cookie to other clients. The severity depends on the application's use of the session and the proxy's behavior regarding cookies. The risk depends on all these conditions being met:\r\n1. The application must be hosted behind a caching proxy that does not strip cookies or ignore responses with cookies.\r\n2. The application sets 'session.permanent = True'\r\n3. The application does not access or modify the session at any point during a request.\r\n4. 'SESSION_REFRESH_EACH_REQUEST' enabled (the default).\r\n5. The application does not set a 'Cache-Control' header to indicate that a page is private or should not be cached.\r\nThis happens because vulnerable versions of Flask only set the 'Vary: Cookie' header when the session is accessed or modified, not when it is refreshed (re-sent to update the expiration) without being accessed or modified.\r\nhttps://github.com/pallets/flask/security/advisories/GHSA-m2qf-hxjv-5gpq",
+          "cve": "CVE-2023-30861",
+          "id": "pyup.io-55261",
+          "more_info_path": "/vulnerabilities/CVE-2023-30861/55261",
+          "specs": [
+              "<2.2.5",
+              ">=2.3.0,<2.3.2"
+          ],
+          "v": "<2.2.5,>=2.3.0,<2.3.2"
+      }
+  ]
+}

vulntotal/tests/test_data/safetydb/parse_advisory-expected.json

@@ -0,0 +1,26 @@
+[
+  {
+    "purl": "pkg:pypi/flask",
+    "affected_versions": ["<0.12.3"],
+    "fixed_versions": [],
+    "aliases": ["CVE-2018-1000656", "pyup.io-36388"]
+  },
+  {
+    "purl": "pkg:pypi/flask",
+    "affected_versions": ["<0.12.3"],
+    "fixed_versions": [],
+    "aliases": ["CVE-2019-1010083", "pyup.io-38654"]
+  },
+  {
+    "purl": "pkg:pypi/flask",
+    "affected_versions": ["<0.6.1"],
+    "fixed_versions": [],
+    "aliases": ["PVE-2021-25820", "pyup.io-25820"]
+  },
+  {
+    "purl": "pkg:pypi/flask",
+    "affected_versions": ["<2.2.5", ">=2.3.0,<2.3.2"],
+    "fixed_versions": [],
+    "aliases": ["CVE-2023-30861", "pyup.io-55261"]
+  }
+]

vulntotal/tests/test_data/safetydb/parse_advisory_cve-expected.json

@@ -0,0 +1,8 @@
+[
+  {
+    "purl": "pkg:pypi/flask",
+    "affected_versions": ["<0.12.3"],
+    "fixed_versions": [],
+    "aliases": ["CVE-2019-1010083", "pyup.io-38654"]
+  }
+]

vulntotal/tests/test_safetydb.py

@@ -0,0 +1,41 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import json
+from pathlib import Path
+
+from commoncode import testcase
+from packageurl import PackageURL
+
+from vulnerabilities.tests import util_tests
+from vulntotal.datasources import safetydb
+
+
+class TestSafetydb(testcase.FileBasedTesting):
+    test_data_dir = str(Path(__file__).resolve().parent / "test_data" / "safetydb")
+
+    def test_parse_advisory(self):
+        purl = PackageURL.from_string("pkg:pypi/flask")
+        advisory_file = self.get_test_loc("advisory.json")
+        with open(advisory_file) as f:
+            advisory = json.load(f)
+
+        results = [adv.to_dict() for adv in safetydb.parse_advisory(advisory, purl)]
+        expected_file = self.get_test_loc("parse_advisory-expected.json", must_exist=False)
+        util_tests.check_results_against_json(results, expected_file)
+
+    def test_parse_advisory_for_cve(self):
+        cve = "CVE-2019-1010083"
+        advisory_file = self.get_test_loc("advisory.json")
+        with open(advisory_file) as f:
+            advisory = json.load(f)
+
+        results = [adv.to_dict() for adv in safetydb.parse_advisory_for_cve(advisory, cve)]
+        expected_file = self.get_test_loc("parse_advisory_cve-expected.json", must_exist=False)
+        util_tests.check_results_against_json(results, expected_file)