Tune down CPE. Filter duplicate versions in vers (#227)

* Tune down the CPE alias. Duplicate versions in vers was resulting in comparison failures.

Signed-off-by: Prabhu Subramanian <prabhu@appthreat.com>

* Bug fix

Signed-off-by: Prabhu Subramanian <prabhu@appthreat.com>

---------

Signed-off-by: Prabhu Subramanian <prabhu@appthreat.com>

Author: prabhu

contrib/cpe_research.py

@@ -72,6 +72,7 @@ def propose_pseudo_purls() -> list:
             "citrix",
             "juniper",
             "qnap",
+            "tenable",
         ]
     )
     raw_hits = index_conn.execute(

packages/mcp-server-vdb/src/mcp_server_vdb/server.py

@@ -385,7 +385,7 @@ async def handle_call_tool(
             write_stream,
             InitializationOptions(
                 server_name="appthreat-vulnerability-db",
-                server_version="6.4.2",
+                server_version="6.4.3",
                 capabilities=server.get_capabilities(
                     notification_options=NotificationOptions(),
                     experimental_capabilities={},

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "appthreat-vulnerability-db"
-version = "6.4.2"
+version = "6.4.3"
 description = "AppThreat's vulnerability database and package search library with a built-in sqlite based storage. OSV, CVE, GitHub, npm are the primary sources of vulnerabilities."
 authors = [
     {name = "Team AppThreat", email = "cloud@appthreat.com"},

test/test_source.py

@@ -573,7 +573,7 @@ def test_convert(test_cve_json):
     results_count = len(
         list(search.search_by_any("cpe:2.3:o:google:android:8.1:*:*:*:*:*:*:*"))
     )
-    assert results_count == 25
+    assert results_count == 6
 
     cvesource = CVESource()
     cve = cvesource.convert5(vulnerabilities)
@@ -709,22 +709,22 @@ def test_nvd_api_convert(
             )
         )
     )
-    assert results_count == 2
+    assert results_count == 1
 
     # git_json
     vulnerabilities = nvdlatest.convert(test_nvd_api_git_json)
     assert len(vulnerabilities) == 1
-    assert len(vulnerabilities[0].details) == 2
+    assert len(vulnerabilities[0].details) == 1
 
     db6.clear_all()
     nvdlatest.store(vulnerabilities)
     cve_data_count, cve_index_count = db6.stats()
-    assert cve_data_count == 2
-    assert cve_index_count == 2
+    assert cve_data_count == 1
+    assert cve_index_count == 1
     results_count = len(list(search.search_by_any("CVE-2020-8022")))
     assert results_count == 0
     results_count = len(list(search.search_by_any("CVE-2023-52426")))
-    assert results_count == 2
+    assert results_count == 1
     results_count = len(
         list(
             search.search_by_any("cpe:2.3:a:libexpat_project:libexpat:*:*:*:*:*:*:*:*")
@@ -760,13 +760,13 @@ def test_nvd_api_convert3(test_nvd_api_signal_json):
     # signal ios
     vulnerabilities = nvdlatest.convert(test_nvd_api_signal_json)
     assert len(vulnerabilities) == 1
-    assert len(vulnerabilities[0].details) == 2
+    assert len(vulnerabilities[0].details) == 1
     nvdlatest.store(vulnerabilities)
     cve_data_count, cve_index_count = db6.stats()
-    assert cve_data_count == 2
-    assert cve_index_count == 2
+    assert cve_data_count == 1
+    assert cve_index_count == 1
     results_count = len(list(search.search_by_any("CVE-2018-9840")))
-    assert results_count == 2
+    assert results_count == 1
     results_count = len(
         list(
             search.search_by_any("pkg:generic/ios/signal")

test/test_utils.py

@@ -1097,6 +1097,15 @@ def test_vers_compare():
     assert utils.vers_compare("1.12", "")
 
 
+def test_vers_compare2():
+    assert not utils.vers_compare("2.4.59", "vers:apache/2.4.1|2.4.1")
+    assert not utils.vers_compare("2.4.59", "vers:apache/2.4.1")
+    assert not utils.vers_compare("2.4.59", "vers:apache/2.4.1|2.4.2|2.4.3|2.4.4")
+    assert utils.vers_compare("2.4.59", "vers:apache/2.4.1|2.4.2|2.4.3|2.4.4|2.4.59")
+    assert utils.vers_compare("2.4.59", "vers:apache/2.4.1|2.4.22|2.4.32|2.4.43|2.4.59|2.4.60")
+    assert utils.vers_compare("2.4.59", "vers:apache/2.4.59|2.4.22|2.4.32|2.4.43|2.4.59|2.4.60")
+    assert utils.vers_compare("2.4.59", "vers:apache/2.4.58|2.4.59|2.4.32|2.4.43|2.4.59|2.4.60")
+
 
 def test_clean_cpe_uri():
     test_tuples = (

uv.lock

@@ -31,7 +31,7 @@
 # Size of the stream to read and write to the file
 DOWNLOAD_CHUNK_SIZE = 128
 
-purl_proposal_cache = defaultdict(list)
+cpe_proposal_cache = defaultdict(list)
 
 
 def filterable_cve(cve_id: str) -> bool:
@@ -133,15 +133,23 @@ def get_alt_cpes(cpe_uri, git_urls):
         git_repo_product = f"{purl_obj.get('namespace')}/{purl_obj.get('name')}"
         if not parsed_git_repo_names.get(git_repo_product):
             parsed_git_repo_names[git_repo_product] = True
-            # We only need 2 new aliases
-            if len(purl_proposal_cache.get(cpe_uri, [])) > 2:
-                purl_proposal_cache[cpe_uri].pop(0)
-            purl_proposal_cache[cpe_uri].append(
-                f"cpe:2.3:a:{purl_obj['type']}:{git_repo_product}:*:*:*:*:*:*:*:*"
-            )
+            purl_type = purl_obj["type"]
+            # Tune down the CPE proposal to reduce false positives
+            # See issue #224
+            if cpe_uri.startswith(f"cpe:2.3:a:{purl_type}:") or (
+                purl_type in cpe_uri
+                and "git" not in purl_type
+                and "git" not in git_repo_product
+            ):
+                # We only need 2 new aliases
+                if len(cpe_proposal_cache.get(cpe_uri, [])) > 2:
+                    cpe_proposal_cache[cpe_uri].pop(0)
+                cpe_proposal_cache[cpe_uri].append(
+                    f"cpe:2.3:a:{purl_type}:{git_repo_product}:*:*:*:*:*:*:*:*"
+                )
     # See if there is something useful in the cache
     if not alt_cpes:
-        alt_cpes = purl_proposal_cache.get(cpe_uri, [])
+        alt_cpes = cpe_proposal_cache.get(cpe_uri, [])
     return alt_cpes
 
 

vdb/lib/nvd.py

@@ -458,7 +458,7 @@ def vers_compare(compare_ver: str | int | float | None, vers: str) -> bool:
     if vers == "*" or compare_ver is None or compare_ver == "*":
         return True
     if vers.startswith("vers:"):
-        vers_parts = vers.split("/")[-1].split("|")
+        vers_parts = list(dict.fromkeys(vers.rsplit("/", 1)[-1].split("|")))
         if len(vers_parts) == 1:
             single_version = vers_parts[0].strip().replace(" ", "")
             # Handle wildcard and the special placeholder fix version
@@ -472,6 +472,9 @@ def vers_compare(compare_ver: str | int | float | None, vers: str) -> bool:
                 return False
         for apart in vers_parts:
             apart = apart.strip().replace(" ", "")
+            # Do we have a direct match
+            if apart == compare_ver:
+                return True
             if apart.startswith(">="):
                 min_version = apart.removeprefix(">=")
             elif apart.startswith(">"):
@@ -480,16 +483,15 @@ def vers_compare(compare_ver: str | int | float | None, vers: str) -> bool:
                 max_version = apart.removeprefix("<=")
             elif apart.startswith("<"):
                 max_excluding = apart.removeprefix("<")
-        # There is exactly only one version
+        # Could the vers be simply a list of values
         if (
-            len(vers_parts) == 1
-            and not min_version
+            not min_version
             and not max_version
             and not min_excluding
             and not max_excluding
         ):
             min_version = vers_parts[0].strip().replace(" ", "")
-            max_version = min_version
+            max_version = vers_parts[-1].strip().replace(" ", "")
     return version_compare(
         compare_ver, min_version, max_version, min_excluding, max_excluding
     )
@@ -1143,8 +1145,7 @@ def to_purl_vers(vendor: str, versions: List) -> str:
                     vers_list.append("*")
                 else:
                     vers_list.append(f"<={less_than_or_equal}")
-
-    return f"vers:{scheme}/{'|'.join(vers_list)}" if vers_list else ""
+    return f"vers:{scheme}/{'|'.join(dict.fromkeys(vers_list))}" if vers_list else ""
 
 
 def calculate_hash(content: str, digest_size: int = 16) -> str: