Fix cpe 2.3 name parsing (#269)

CveXplore/VERSION

@@ -1 +1 @@
-0.3.23.dev1
+0.3.24.dev6

CveXplore/common/cpe_converters.py

@@ -8,6 +8,15 @@
 from CveXplore.database.helpers.cpe_conversion import cpe_uri_to_fs, cpe_fs_to_uri
 
 
+def split_cpe_name(cpename: str) -> list[str]:
+    """
+    Split CPE 2.3 into its components, accounting for escaped colons.
+    """
+    non_escaped_colon = r"(?<!\\):"
+    split_name = re.split(non_escaped_colon, cpename)
+    return split_name
+
+
 def from2to3CPE(cpe: str, autofill: bool = False) -> str:
     """
     Method to transform cpe2.2 to cpe2.3 format
@@ -18,7 +27,7 @@ def from2to3CPE(cpe: str, autofill: bool = False) -> str:
             return False
         cpe = cpe_uri_to_fs(cpe)
     if autofill:
-        e = cpe.split(":")
+        e = split_cpe_name(cpe)
         for x in range(0, 13 - len(e)):
             cpe += ":-"
     return cpe

CveXplore/core/database_maintenance/api_handlers.py

@@ -1,5 +1,7 @@
+import re
 from abc import abstractmethod
 
+from CveXplore.common.cpe_converters import split_cpe_name
 from CveXplore.core.database_actions.db_action import DatabaseAction
 from CveXplore.core.database_maintenance.download_handler import DownloadHandler
 from CveXplore.core.nvd_nist.nvd_nist_api import NvdNistApi
@@ -35,8 +37,11 @@ def process_item(self, item: dict):
                 return item
 
     @staticmethod
-    def stem(cpe_uri: str):
-        cpe_stem = cpe_uri.split(":")
+    def split_cpe_name(cpename: str) -> list[str]:
+        return split_cpe_name(cpename)
+
+    def stem(self, cpe_uri: str):
+        cpe_stem = self.split_cpe_name(cpe_uri)
         return ":".join(cpe_stem[:5])
 
     @staticmethod

CveXplore/core/database_maintenance/sources_process.py

@@ -48,9 +48,8 @@ def __init__(self):
     def file_to_queue(self, *args):
         pass
 
-    @staticmethod
-    def parse_cpe_version(cpename: str):
-        cpe_list = cpename.split(":")
+    def parse_cpe_version(self, cpename: str):
+        cpe_list = self.split_cpe_name(cpename)
         version_stem = cpe_list[5]
 
         if cpe_list[6] != "*" and cpe_list[6] != "-":
@@ -81,11 +80,12 @@ def process_the_item(self, item: dict = None):
 
         version = self.parse_cpe_version(cpename=item["cpeName"])
 
+        split_cpe_name = self.split_cpe_name(item["cpeName"])
         cpe = {
             "title": title,
             "cpeName": item["cpeName"],
-            "vendor": item["cpeName"].split(":")[3],
-            "product": item["cpeName"].split(":")[4],
+            "vendor": split_cpe_name[3],
+            "product": split_cpe_name[4],
             "version": version,
             "padded_version": self.padded_version(version),
             "stem": self.stem(item["cpeName"]),
@@ -97,15 +97,14 @@ def process_the_item(self, item: dict = None):
         }
 
         sha1_hash = hashlib.sha1(
-            cpe["cpeName"].encode("utf-8")
-            + item["cpeName"].split(":")[5].encode("utf-8")
+            cpe["cpeName"].encode("utf-8") + split_cpe_name[5].encode("utf-8")
         ).hexdigest()
 
         cpe["id"] = sha1_hash
 
         return cpe
 
-    def process_downloads(self, sites: list = None):
+    def process_downloads(self, sites: list | None = None):
         """
         Method to download and process files
         """
@@ -366,10 +365,10 @@ def add_if_missing(cve: dict, key: str, value: Any):
             cve[key].append(value)
         return cve
 
-    @staticmethod
-    def get_vendor_product(cpeUri: str):
-        vendor = cpeUri.split(":")[3]
-        product = cpeUri.split(":")[4]
+    def get_vendor_product(self, cpeUri: str):
+        split_cpe_uri = self.split_cpe_name(cpeUri)
+        vendor = split_cpe_uri[3]
+        product = split_cpe_uri[4]
         return vendor, product
 
     def file_to_queue(self, *args):