Added search by cve and stats method

Signed-off-by: Prabhu Subramanian <[email protected]>

vdb/cli.py

@@ -121,10 +121,14 @@ def main():
         for s in sources:
             LOG.info("Refreshing %s", s.__class__.__name__)
             s.refresh()
+        cve_data_count, cve_index_count = db_lib.stats()
+        print("cve_data_count", cve_data_count, "cve_index_count", cve_index_count)
         db_lib.optimize_and_close_all()
     if args.search:
         if args.search.startswith("pkg:"):
             results = search.search_by_purl(args.search, with_data=True)
+        elif args.search.startswith("CVE-") or args.search.startswith("GHSA-") or args.search.startswith("MAL-"):
+            results = search.search_by_cve(args.search, with_data=True)
         else:
             results = search.search_by_cpe_like(args.search, with_data=True)
         if results:

vdb/lib/db6.py

@@ -1,8 +1,5 @@
-try:
-    from pysqlite3 import dbapi2 as sqlite3
-except ImportError:
-    import sqlite3
 import os
+import sqlite3
 import sys
 
 from vdb.lib import config
@@ -41,6 +38,19 @@ def get(db_file: str = config.VDB_BIN_FILE, index_file: str = config.VDB_BIN_IND
     return db_conn, index_conn
 
 
+def stats():
+    global db_conn, index_conn
+    cve_data_count = 0
+    res = db_conn.execute("SELECT count(*) FROM cve_data").fetchone()
+    if res:
+        cve_data_count = res[0]
+    cve_index_count = 0
+    res = index_conn.execute("SELECT count(*) FROM cve_index").fetchone()
+    if res:
+        cve_index_count = res[0]
+    return cve_data_count, cve_index_count
+
+
 def clear_all():
     if db_conn:
         db_conn.execute("DELETE FROM cve_data;")

vdb/lib/search.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 import orjson
 
 from vdb.lib import db6, utils
@@ -20,9 +22,21 @@ def _filter_hits(raw_hits: list, compare_ver: str) -> list:
     return filtered_list
 
 
-def _data_list(db_conn, filtered_list, search_str):
+def get_cve_data(db_conn, index_hits: list[dict, Any], search_str: str) -> list[dict[str, str | CVE | None]]:
+    """Get CVE data for the index results
+
+    Args:
+        db_conn: DB Connection or None to create a new one
+        index_hits: Hits from one of the search methods
+        search_str: Original search string used
+
+    Returns:
+        list: List of CVE data with original source data as a pydantic model
+    """
+    if not db_conn:
+        db_conn, _ = db6.get(read_only=True)
     data_list = []
-    for ahit in filtered_list:
+    for ahit in index_hits:
         results = exec_query(db_conn,
                              "SELECT cve_id, type, namespace, name, source_data, override_data FROM cve_data WHERE cve_id = ? AND type = ? ORDER BY cve_id DESC;",
                              (ahit["cve_id"], ahit["type"]))
@@ -41,6 +55,7 @@ def _data_list(db_conn, filtered_list, search_str):
 
 
 def search_by_cpe_like(cpe: str, with_data=False) -> list | None:
+    """Search by CPE or colon-separate strings"""
     db_conn, index_conn = db6.get(read_only=True)
     if cpe.startswith("cpe:"):
         vendor, package, version, _ = utils.parse_cpe(cpe)
@@ -57,11 +72,12 @@ def search_by_cpe_like(cpe: str, with_data=False) -> list | None:
                               (vendor, package))
     filtered_list = _filter_hits(raw_hits, version)
     if with_data:
-        return _data_list(db_conn, filtered_list, cpe)
+        return get_cve_data(db_conn, filtered_list, cpe)
     return filtered_list
 
 
 def search_by_purl(purl: str, with_data=False) -> list | None:
+    """Search by purl"""
     db_conn, index_conn = db6.get(read_only=True)
     purl_obj = utils.parse_purl(purl)
     if purl_obj:
@@ -83,10 +99,20 @@ def search_by_purl(purl: str, with_data=False) -> list | None:
                               args)
         filtered_list = _filter_hits(raw_hits, version)
         if with_data:
-            return _data_list(db_conn, filtered_list, purl)
+            return get_cve_data(db_conn, filtered_list, purl)
         return filtered_list
 
 
+def search_by_cve(cve_id: str, with_data=False) -> list | None:
+    """Search by CVE"""
+    db_conn, index_conn = db6.get(read_only=True)
+    raw_hits = exec_query(index_conn, "SELECT cve_id, type, namespace, name, vers FROM cve_index where cve_id = ?", (cve_id, ))
+    filtered_list = _filter_hits(raw_hits, "*")
+    if with_data:
+        return get_cve_data(db_conn, filtered_list, cve_id)
+    return filtered_list
+
+
 def exec_query(conn, query: str, args: tuple[str, ...]) -> list:
     res = conn.execute(query, args)
     return res.fetchall()

vdb/lib/utils.py

@@ -447,7 +447,9 @@ def trim_epoch(
 def vers_compare(compare_ver: str | int | float, vers: str) -> bool:
     """Purl vers based version comparison"""
     min_version, max_version, min_excluding, max_excluding = None, None, None, None
-    if vers.startswith("vers:"):
+    if vers == "*":
+        return True
+    elif vers.startswith("vers:"):
         vers_parts = vers.split("/")[-1].split("|")
         if len(vers_parts) == 1:
             single_version = vers_parts[0].strip().replace(" ", "")