🐛 Fix safetydb and disable it by default (license issue)

Signed-off-by: Shivam Sandbhor <shivam.sandbhor@gmail.com>

Author: sbs2001

vulnerabilities/importer_yielder.py

@@ -56,16 +56,16 @@
             "debian_tracker_url": "https://security-tracker.debian.org/tracker/data/json"
         },
     },
-    {
-        "name": "safetydb",
-        "license": "cc-by-nc-4.0",
-        "last_run": None,
-        "data_source": "SafetyDbDataSource",
-        "data_source_cfg": {
-            "url": "https://raw.githubusercontent.com/pyupio/safety-db/master/data/insecure_full.json",  # nopep8
-            "etags": {},
-        },
-    },
+    # {
+    #     "name": "safetydb",
+    #     "license": "cc-by-nc-4.0",
+    #     "last_run": None,
+    #     "data_source": "SafetyDbDataSource",
+    #     "data_source_cfg": {
+    #         "url": "https://raw.githubusercontent.com/pyupio/safety-db/master/data/insecure_full.json",
+    #         "etags": {},
+    #     },
+    # },
     {
         "name": "npm",
         "license": "mit",

vulnerabilities/importers/safety_db.py

@@ -36,6 +36,7 @@
 import requests
 from univers.version_specifier import VersionSpecifier
 from univers.versions import PYPIVersion
+from univers.versions import InvalidVersion
 from packageurl import PackageURL
 from schema import Or
 from schema import Regex
@@ -97,8 +98,6 @@ def collect_packages(self):
         return {pkg for pkg in self._api_response}
 
     def updated_advisories(self) -> Set[Advisory]:
-        advisories = []
-
         for package_name in self._api_response:
             if package_name == "$meta" or package_name == "cumin":
                 # This is the first entry in the data feed. It contains metadata of the feed.
@@ -129,7 +128,7 @@ def updated_advisories(self) -> Set[Advisory]:
                 )
 
                 reference = [Reference(reference_id=advisory["id"])]
-
+                advisories = []
                 for cve_id in cve_ids:
                     advisories.append(
                         Advisory(
@@ -142,7 +141,7 @@ def updated_advisories(self) -> Set[Advisory]:
                         )
                     )
 
-        return advisories
+                yield advisories
 
     # FIXME: This is duplicate code. Use the the helper instead.
     def create_etag(self, url):
@@ -172,10 +171,12 @@ def categorize_versions(
     for version_spec in version_specs:
         vurl_specs.append(VersionSpecifier.from_scheme_version_spec_string("pypi", version_spec))
 
+    invalid_versions = set()
     for version in all_versions:
         try:
             version_object = PYPIVersion(version)
-        except:
+        except InvalidVersion:
+            invalid_versions.add(version)
             continue
 
         if any([version_object in vurl_spec for vurl_spec in vurl_specs]):
@@ -189,6 +190,7 @@ def categorize_versions(
             )
 
     resolved_purls = []
+    all_versions -= invalid_versions
     for version in all_versions - impacted_versions:
         resolved_purls.append(PackageURL(name=package_name, type="pypi", version=version))
     return impacted_purls, resolved_purls

vulnerabilities/tests/test_safety_db.py

@@ -154,8 +154,14 @@ def test_import(self):
             ),
         ]
 
-        found_data = [adv for adv in data_src.updated_advisories()]
+        found_data = []
+        # FIXME: This is messed up
+        for adv_batch in data_src.updated_advisories():
+            found_data.extend(adv_batch)
+            # found_data = [list(adv) for adv in data_src.updated_advisories()]
 
+        print(expected_data)
+        print("\n", found_data)
         assert expected_data == found_data