diff --git a/safety/safety.py b/safety/safety.py index a86c1679..df7cf2f2 100644 --- a/safety/safety.py +++ b/safety/safety.py @@ -15,7 +15,7 @@ from authlib.integrations.base_client.errors import OAuthError import httpx -from packaging.specifiers import SpecifierSet +from packaging.specifiers import InvalidSpecifier, SpecifierSet from packaging.utils import canonicalize_name from packaging.version import parse as parse_version, Version from pydantic_core import to_jsonable_python @@ -753,7 +753,15 @@ def check( if name in vulnerable_packages: # we have a candidate here, build the spec set for specifier in db["vulnerable_packages"][name]: - spec_set = SpecifierSet(specifiers=specifier) + try: + spec_set = SpecifierSet(specifiers=specifier) + except InvalidSpecifier: + LOG.warning( + "Skipping invalid specifier '%s' for package '%s'", + specifier, + name, + ) + continue if is_vulnerable(spec_set, req, pkg): if not db_full: diff --git a/tests/test_safety.py b/tests/test_safety.py index 0ea8c354..17062f6d 100644 --- a/tests/test_safety.py +++ b/tests/test_safety.py @@ -224,6 +224,57 @@ def mock_fetch_side_effect(*args, **kwargs): self.assertEqual(len(vulns), 1) + @patch("safety.safety.fetch_database_url") + def test_check_invalid_specifier_in_db(self, mock_fetch_db): + # A vulnerability DB entry with a malformed specifier such as + # "=1.7.0" (single equals) must be skipped instead of crashing. + insecure_path = os.path.join(self.dirname, "test_db", "insecure.json") + insecure_full_path = os.path.join(self.dirname, "test_db", "insecure_full.json") + + with open(insecure_path) as f: + insecure_db = json.load(f) + with open(insecure_full_path) as f: + insecure_full_db = json.load(f) + + insecure_db["vulnerable_packages"]["insecure-package"] = ["=1.7.0", "<1.0"] + insecure_full_db["vulnerable_packages"]["insecure-package"] = [ + { + "specs": ["<1.0"], + "advisory": "Test advisory for insecure-package", + "transitive": False, + "more_info_path": "/v/test/path", + "ids": [{"type": "pyup", "id": "test-id"}], + } + ] + + def mock_fetch_side_effect(*args, **kwargs): + db_name = kwargs.get( + "db_name", args[2] if len(args) > 2 else "insecure.json" + ) + if db_name == "insecure_full.json": + return insecure_full_db + else: + return insecure_db + + mock_fetch_db.side_effect = mock_fetch_side_effect + + reqs = StringIO("insecure-package==0.1") + packages = read_requirements(reqs) + + vulns, _ = check( + auth=self.auth, + packages=packages, + db_mirror=False, + cached=0, + ignore_vulns={}, + ignore_severity_rules=None, + proxy={}, + telemetry=False, + ) + + # The valid "<1.0" specifier is still matched. + self.assertEqual(len(vulns), 1) + @patch("safety.safety.fetch_database_url") def test_check_cached(self, mock_fetch_db): from safety.constants import DB_CACHE_FILE