fix: unknown values in metrics cvedb

Author: 22f1001635

cve_bin_tool/cvedb.py

@@ -218,6 +218,12 @@ class CVEDB:
             VALUES (?, ?)
         """,
     }
+    METRICS = [
+        (UNKNOWN_METRIC_ID, "UNKNOWN"),
+        (EPSS_METRIC_ID, "EPSS"),
+        (CVSS_2_METRIC_ID, "CVSS-2"),
+        (CVSS_3_METRIC_ID, "CVSS-3"),
+    ]
 
     def __init__(
         self,
@@ -315,7 +321,8 @@ def refresh_cache_and_update_db(self) -> None:
         if not self.latest_schema("metrics", self.TABLE_SCHEMAS["metrics"], cursor):
             self.LOGGER.info("Updating metrics data.")
             self.populate_metrics()
-            self.connection.commit()
+            if self.connection is not None:
+                self.connection.commit()
         self.db_close()
 
     def get_cvelist_if_stale(self) -> None:
@@ -341,6 +348,7 @@ def get_cvelist_if_stale(self) -> None:
                 or not self.latest_schema(
                     "cve_exploited", self.TABLE_SCHEMAS["cve_exploited"]
                 )
+                or not self.latest_schema("metrics", self.TABLE_SCHEMAS["metrics"])
             ):
                 self.refresh_cache_and_update_db()
                 self.time_of_last_update = datetime.datetime.today()
@@ -368,7 +376,6 @@ def latest_schema(
 
         # getting schema from command
         lines = table_schema.split("(")[1].split(",")
-
         table_schema = [x.split("\n")[1].strip().split(" ")[0] for x in lines]
         table_schema.pop()
 
@@ -378,13 +385,16 @@ def latest_schema(
         if table_schema == current_schema:
             schema_latest = True
 
-        # Check for metrics table schema
+        # Check for metrics table data integrity
         if table_name == "metrics":
-            result = cursor.execute(
-                "SELECT * FROM metrics WHERE metrics_id=?", (UNKNOWN_METRIC_ID,)
-            )
-            if not result.fetchone():
-                schema_latest = False
+            for metric_id, metric_name in self.METRICS:
+                result = cursor.execute(
+                    "SELECT * FROM metrics WHERE metrics_id=? AND metrics_name=?",
+                    (metric_id, metric_name),
+                )
+                if not result.fetchone():
+                    schema_latest = False
+                    break  # Early exit if any metric is missing
 
         return schema_latest
 
@@ -640,18 +650,12 @@ def populate_affected(self, affected_data, cursor, data_source):
     def populate_metrics(self):
         """Adding data to metric table."""
         cursor = self.db_open_and_get_cursor()
-        # Insert a row without specifying cve_metrics_id
         insert_metrics = self.INSERT_QUERIES["insert_metrics"]
-        data = [
-            (UNKNOWN_METRIC_ID, "UNKNOWN"),
-            (EPSS_METRIC_ID, "EPSS"),
-            (CVSS_2_METRIC_ID, "CVSS-2"),
-            (CVSS_3_METRIC_ID, "CVSS-3"),
-        ]
-        # Execute the insert query for each row
-        for row in data:
+        # Use the METRICS constant to populate the table
+        for row in self.METRICS:
             cursor.execute(insert_metrics, row)
-        self.connection.commit()
+        if self.connection is not None:
+            self.connection.commit()
         self.db_close()
 
     def metric_finder(self, cursor, cve):
@@ -869,23 +873,26 @@ def create_exploit_db(self):
         create_exploit_table = self.TABLE_SCHEMAS["cve_exploited"]
         cursor = self.db_open_and_get_cursor()
         cursor.execute(create_exploit_table)
-        self.connection.commit()
+        if self.connection is not None:
+            self.connection.commit()
         self.db_close()
 
     def populate_exploit_db(self, exploits):
         """Add exploits to the exploits database table."""
         insert_exploit = self.INSERT_QUERIES["insert_exploit"]
         cursor = self.db_open_and_get_cursor()
         cursor.executemany(insert_exploit, exploits)
-        self.connection.commit()
+        if self.connection is not None:
+            self.connection.commit()
         self.db_close()
 
     def store_epss_data(self, epss_data):
         """Insert Exploit Prediction Scoring System (EPSS) data into database."""
         insert_cve_metrics = self.INSERT_QUERIES["insert_cve_metrics"]
         cursor = self.db_open_and_get_cursor()
         cursor.executemany(insert_cve_metrics, epss_data)
-        self.connection.commit()
+        if self.connection is not None:
+            self.connection.commit()
         self.db_close()
 
     def dict_factory(self, cursor, row):
@@ -1157,7 +1164,8 @@ def json_to_db_wrapper(self, path, pubkey, ignore_signature, log_signature_error
                                     shutil.rmtree(temp_gnupg_home)
                                 return ERROR_CODES[SigningError]
                     self.json_to_db(cursor, dir, json.loads(data))
-                    self.connection.commit()
+                    if self.connection is not None:
+                        self.connection.commit()
 
             if is_signed and not ignore_signature and temp_gnupg_home.exists():
                 shutil.rmtree(temp_gnupg_home)