Add Common Weakness Enumeration (CWE) table to cve db

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")

cve_bin_tool/data_sources/curl_source.py

@@ -39,6 +39,7 @@ def __init__(self, error_mode=ErrorMode.TruncTrace):
         self.error_mode = error_mode
         self.session = None
         self.affected_data = None
+        self.severity_data = None
         self.source_name = self.SOURCE
         self.vulnerability_data = []
 
@@ -47,7 +48,7 @@ async def get_cve_data(self):
         await self.fetch_cves()
         self.get_cve_list()
 
-        return (None, self.affected_data), self.source_name
+        return (self.severity_data, self.affected_data), self.source_name
 
     async def fetch_cves(self):
         """Fetch cURL vulnerabilities data."""
@@ -74,20 +75,44 @@ async def download_curl_vulnerabilities(self, session: RateLimiter) -> None:
     def get_cve_list(self):
         """Get a list of affected cURL vulnerabilities."""
         self.affected_data = []
+        self.severity_data = []
+
+        for cve_item in self.vulnerability_data:
+            score=0
+            severity=cve_item["database_specific"]["severity"]
+            if "Low" in severity:
+                score=2
+            elif "Medium" in severity:
+                score=5
+            elif "High" in severity:
+                score=7
+            else:
+                score=9
+
+            cve = {
+                "ID": cve_item["aliases"][0],
+                "severity": severity,
+                "description": cve_item["details"],
+                "score": score,
+                "CVSS_version": "unknown",
+                "CVSS_vector": "unknown",
+                "last_modified": cve_item["modified"],
+                "cwes": [cve_item["database_specific"]["CWE"]["id"]] if cve_item["database_specific"]["CWE"] is not None else [],
+            }
+            self.severity_data.append(cve)
 
-        for cve in self.vulnerability_data:
             affected = {
-                "cve_id": cve["aliases"][0],
+                "cve_id": cve_item["aliases"][0],
                 "vendor": "haxx",
                 "product": "curl",
                 "version": "*",
-                "versionStartIncluding": cve["affected"][0]["ranges"][0]["events"][0][
+                "versionStartIncluding": cve_item["affected"][0]["ranges"][0]["events"][0][
                     "introduced"
                 ],
                 "versionStartExcluding": "",
-                "versionEndIncluding": cve["affected"][0]["versions"][0],
+                "versionEndIncluding": cve_item["affected"][0]["versions"][0],
                 "versionEndExcluding": "",
             }
             self.affected_data.append(affected)
 
-        return self.affected_data
+        return (self.severity_data, self.affected_data)

cve_bin_tool/data_sources/nvd_source.py

@@ -140,6 +140,7 @@ def format_data(self, all_cve_entries):
                     if cve_item.get("lastModifiedDate", None)
                     else cve_item["publishedDate"]
                 ),
+                "cwes": [],
             }
             if cve["description"].startswith("** REJECT **"):
                 # Skip this CVE if it's marked as 'REJECT'
@@ -168,6 +169,14 @@ def format_data(self, all_cve_entries):
             cve["score"] = cve["score"] if cve["score"] is not None else "unknown"
             cve_data.append(cve)
 
+            # Add all CWE
+            if "problemtype" in cve_item["cve"]:
+                for problemtype_data in cve_item["cve"]["problemtype"]["problemtype_data"]:
+                    if "description" in problemtype_data:
+                        for description in problemtype_data["description"]:
+                            if "value" in description and description["value"].startswith("CWE-"):
+                                cve["cwes"].append(description["value"])
+
             # walk the nodes with version data
             # return list of versions
             affects_list = []
@@ -250,6 +259,7 @@ def format_data_api2(self, all_cve_entries):
                     if cve_item.get("lastModified", None)
                     else cve_item["published"]
                 ),
+                "cwes": [],
             }
             if cve["description"].startswith("** REJECT **"):
                 # Skip this CVE if it's marked as 'REJECT'
@@ -336,6 +346,14 @@ def format_data_api2(self, all_cve_entries):
 
             cve_data.append(cve)
 
+            # Add all CWE
+            if "problemtype" in cve_item:
+                for problemtype_data in cve_item["problemtype"]["problemtype_data"]:
+                    if "description" in problemtype_data:
+                        for description in problemtype_data["description"]:
+                            if "value" in description and description["value"].startswith("CWE-"):
+                                cve["cwes"].append(description["value"])
+
             # walk the nodes with version data
             # return list of versions
             affects_list = []

cve_bin_tool/data_sources/redhat_source.py

@@ -3,6 +3,7 @@
 
 import datetime
 import json
+import re
 from pathlib import Path
 
 import aiohttp
@@ -215,6 +216,7 @@ def format_data(self, all_cve_entries):
                 "CVSS_version": str(version) if vector[0] is not None else "unknown",
                 "CVSS_vector": vector[0] if vector[0] is not None else "unknown",
                 "last_modified": cve_item["public_date"],
+                "cwes": re.findall(r'(CWE-\d+)',cve_item["CWE"]) if cve_item["CWE"] is not None else [],
             }
             # Only add CVE if at least one related product found
             cve_to_write = True