cve_bin_tool: cvedb.py: Add Common Weakness Enumeration (CWE) table

The CWE is a category system for hardware and software weakness and
vulnerabilities with the goal of understanding flaws.

Signed-off-by: Maik Otto <[email protected]>

Author: motto-phytec

cve_bin_tool/cvedb.py

@@ -136,6 +136,14 @@ class CVEDB:
             UNIQUE(purl,cpe)
         )
         """,
+        "cve_cwe": """
+        CREATE TABLE IF NOT EXISTS cve_cwe (
+            cve_number TEXT,
+            cwe TEXT,
+            data_source TEXT,
+            FOREIGN KEY(cve_number, data_source) REFERENCES cve_severity(cve_number, data_source)
+        )
+        """,
     }
 
     # This is mostly to make bandit happier because we won't be
@@ -148,6 +156,7 @@ class CVEDB:
         "metrics": "DROP TABLE metrics",
         "mismatch": "DROP TABLE mismatch",
         "purl2cpe": "DROP TABLE purl2cpe",
+        "cve_cwe": "DROP TABLE cve_cwe",
     }
 
     INDEXES = {
@@ -163,6 +172,7 @@ class CVEDB:
         "metrics": "SELECT * FROM metrics WHERE 1=0",
         "mismatch": "SELECT * FROM mismatch WHERE 1=0",
         "purl2cpe": "SELECT * FROM purl2cpe WHERE 1=0",
+        "cve_cwe": "SELECT * FROM cve_cwe WHERE 1=0"
     }
 
     INSERT_QUERIES = {
@@ -217,6 +227,14 @@ class CVEDB:
             )
             VALUES (?, ?)
         """,
+        "insert_cve_cwe": """
+        INSERT or REPLACE INTO cve_cwe (
+            cve_number,
+            cwe,
+            data_source
+            )
+            VALUES (?,?,?)
+        """,
     }
 
     def __init__(
@@ -333,6 +351,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("cve_cwe", self.TABLE_SCHEMAS["cve_cwe"])
             ):
                 self.refresh_cache_and_update_db()
                 self.time_of_last_update = datetime.datetime.today()
@@ -513,6 +532,7 @@ def populate_db(self) -> None:
                         severity_data, cursor, data_source=source_name
                     )
                     self.populate_cve_metrics(severity_data, cursor)
+                    self.populate_cve_cwe(severity_data, cursor,data_source=source_name)
                 if affected_data is not None:
                     self.populate_affected(
                         affected_data,
@@ -599,6 +619,27 @@ def populate_cve_metrics(self, severity_data, cursor):
             except Exception as e:
                 LOGGER.info(f"Unable to insert data for {e}\n{cve}")
 
+    def populate_cve_cwe(self, severity_data, cursor, data_source):
+        """Adds data into CVE CWE table."""
+        insert_cve_cwe = self.INSERT_QUERIES["insert_cve_cwe"]
+
+        for cve in severity_data:
+            try:
+                if "cwes" in cve and len(cve["cwes"]) >0:
+                    cursor.executemany(
+                        insert_cve_cwe,
+                        [
+                            (
+                                cve["ID"],
+                                cwe,
+                                data_source,
+                            )
+                            for cwe in cve["cwes"]
+                        ],
+                    )
+            except Exception as e:
+                LOGGER.info(f"Unable to insert data for {e}\n{cve}")
+
     def populate_affected(self, affected_data, cursor, data_source):
         """Populate database with affected versions."""
         insert_cve_range = self.INSERT_QUERIES["insert_cve_range"]
@@ -899,6 +940,7 @@ def delete_old_files_if_exists(self, path):
             "cve_severity",
             "cve_metrics",
             "metrics",
+            "cve_cwe",
         ]
         for directory in DIRECTORIES:
             if (path / directory).exists():
@@ -1058,6 +1100,8 @@ def json_to_db(self, cursor, db_column, json_data):
             cursor.executemany(self.INSERT_QUERIES["insert_cve_metrics"], values)
         elif db_column == "metrics":
             cursor.executemany(self.INSERT_QUERIES["insert_metrics"], values)
+        elif db_column == "cve_cwe":
+            cursor.executemany(self.INSERT_QUERIES["insert_cve_cwe"], values)
 
     def json_to_db_wrapper(self, path, pubkey, ignore_signature, log_signature_error):
         """Initialize the process wrapper to insert records into database from JSON."""
@@ -1083,6 +1127,7 @@ def json_to_db_wrapper(self, path, pubkey, ignore_signature, log_signature_error
             cursor.execute(self.TABLE_SCHEMAS["cve_exploited"])
             cursor.execute(self.TABLE_SCHEMAS["cve_metrics"])
             cursor.execute(self.TABLE_SCHEMAS["metrics"])
+            cursor.execute(self.TABLE_SCHEMAS["cve_cwe"])
             index_range = "CREATE INDEX IF NOT EXISTS product_index ON cve_range (cve_number, vendor, product)"
             cursor.execute(index_range)
             metadata_fd = open(path / "metadata.json")