Merge pull request #55 from AppThreat/feature/tune-perf

Reduce db size. Lint fixes

.github/workflows/pythonapp.yml

@@ -29,7 +29,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install .[dev]
+        pip install ".[dev]"
     - name: Lint with flake8
       run: |
         # stop the build if there are Python syntax errors or undefined names
@@ -45,16 +45,3 @@ jobs:
       run: |
         pip uninstall -y orjson
         pytest --cov=vdb test
-    - name: Self sast-scan
-      uses: AppThreat/sast-scan-action@master
-      with:
-        output: reports
-        type: python,credscan
-      env:
-        SCAN_ID: ${{ github.sha }}
-        WORKSPACE: https://github.com/${{ github.repository }}/blob/${{ github.sha }}
-    - name: Upload scan reports
-      uses: actions/upload-artifact@v1.0.0
-      with:
-        name: sast-scan-reports
-        path: reports

.github/workflows/pythonpublish.yml

@@ -16,7 +16,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.10'
+          python-version: '3.11'
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "appthreat-vulnerability-db"
-version = "5.1.4"
+version = "5.2.0"
 description = "AppThreat's vulnerability database and package search library with a built-in file based storage. OSV, CVE, GitHub, npm are the primary sources of vulnerabilities."
 authors = [
     {name = "Team AppThreat", email = "cloud@appthreat.com"},
@@ -24,10 +24,10 @@ classifiers = [
     "Intended Audience :: System Administrators",
     "License :: OSI Approved :: MIT License",
     "Operating System :: OS Independent",
-    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
-    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.8",
     "Topic :: Security",
     "Topic :: Utilities",
 ]
@@ -48,7 +48,8 @@ dev = [
     "bandit",
     "flake8",
     "pytest",
-    "pytest-cov",]
+    "pytest-cov"
+]
 
 [tool.setuptools]
 packages = ["test", "vdb", "vdb.lib"]

test/test_db.py

@@ -57,7 +57,7 @@ def test_search_slow(test_db, test_vuln_data):
         res = db.pkg_search(
             table,
             d["details"]["package"],
-            d["details"]["max_affected_version_including"],
+            d["details"]["mai"],
         )
         assert len(res)
         assert res[0].to_dict()["package_issue"]
@@ -74,7 +74,7 @@ def test_search_fast(test_db, test_vuln_data):
     search_list = [
         {
             "name": d["details"]["package"],
-            "version": d["details"]["max_affected_version_including"],
+            "version": d["details"]["mai"],
         }
         for d in all_data
     ]
@@ -96,7 +96,7 @@ def test_gha_search_slow(test_db, test_gha_data):
     all_data = db.list_all(table)
     assert all_data
     for d in all_data:
-        version = d["details"]["max_affected_version_including"]
+        version = d["details"]["mai"]
         if version and version != "*":
             res = db.pkg_search(
                 table,
@@ -117,7 +117,7 @@ def test_gha_vendor_search(test_db, test_gha_data):
     assert all_data
     for d in all_data:
         vendor, _, _, cve_type = parse_cpe(d["details"]["cpe_uri"])
-        version = d["details"]["max_affected_version_including"]
+        version = d["details"]["mai"]
         if version and version != "*":
             res = db.vendor_pkg_search(
                 table,
@@ -140,10 +140,10 @@ def test_gha_search_bulk(test_db, test_gha_data):
     tmp_list = [
         {
             "name": d["details"]["package"],
-            "version": d["details"]["max_affected_version_including"],
+            "version": d["details"]["mai"],
         }
         for d in all_data
-        if d["details"]["max_affected_version_including"] != "*"
+        if d["details"]["mai"] != "*"
     ]
     res = db.bulk_index_search(tmp_list)
     assert len(res)
@@ -160,7 +160,7 @@ def test_index_search(test_db, test_vuln_data):
     assert all_data
     tmp_list = []
     for d in all_data[:40]:
-        version = d["details"]["max_affected_version_including"]
+        version = d["details"]["mai"]
         if version and version != "*":
             tmp_list.append({"name": d["details"]["package"], "version": version})
     res = db.bulk_index_search(tmp_list)
@@ -187,7 +187,7 @@ def test_vendor_index_search(test_db, test_vuln_data):
             {
                 "vendor": vendor,
                 "name": d["details"]["package"],
-                "version": d["details"]["max_affected_version_including"],
+                "version": d["details"]["mai"],
             }
         )
     res = db.bulk_index_search(tmp_list)

vdb/cli.py

@@ -5,10 +5,11 @@
 import logging
 import os
 import re
+import shutil
 
 from tabulate import tabulate
 
-from vdb.lib import config as config
+from vdb.lib import config
 from vdb.lib import db as dbLib
 from vdb.lib.aqua import AquaSource
 from vdb.lib.gha import GitHubSource
@@ -134,21 +135,18 @@ def print_results(results):
     ]
     for res in results:
         vuln_occ_dict = res.to_dict()
-        id = vuln_occ_dict.get("id")
+        vid = vuln_occ_dict.get("id")
         package_type = vuln_occ_dict.get("type")
-        if id not in added_list:
+        if vid not in added_list:
             package_issue = res.package_issue
             full_pkg = package_issue.affected_location.package
             if package_issue.affected_location.vendor:
-                full_pkg = "{}:{}".format(
-                    package_issue.affected_location.vendor,
-                    package_issue.affected_location.package,
-                )
+                full_pkg = f"{package_issue.affected_location.vendor}:{package_issue.affected_location.package}"
             if package_type and package_type not in ("*", "o", "h"):
                 full_pkg = package_type + ":" + full_pkg
             table.append(
                 [
-                    id,
+                    vid,
                     full_pkg,
                     package_issue.affected_location.version,
                     package_issue.fixed_location,
@@ -158,19 +156,17 @@ def print_results(results):
                     vuln_occ_dict.get("short_description"),
                 ]
             )
-            added_list.append(id)
+            added_list.append(vid)
     print(tabulate(table, headers, tablefmt="grid"))
 
 
 def main():
+    """Main function"""
     args = build_args()
     print(at_logo)
     if args.clean:
         if os.path.exists(config.data_dir):
-            try:
-                os.rmdir(config.data_dir)
-            except Exception:
-                pass
+            shutil.rmtree(config.data_dir, ignore_errors=True)
     if args.cache or args.cache_os:
         if args.only_osv:
             sources = [OSVSource()]
@@ -181,19 +177,19 @@ def main():
         if args.cache_os:
             sources.insert(0, AquaSource())
         for s in sources:
-            LOG.info("Refreshing {}".format(s.__class__.__name__))
+            LOG.info("Refreshing %s", s.__class__.__name__)
             s.refresh()
     elif args.sync:
         for s in [GitHubSource(), NvdSource()]:
-            LOG.info("Syncing {}".format(s.__class__.__name__))
+            LOG.info("Syncing %s", s.__class__.__name__)
             s.download_recent()
     if args.sync_npm:
         for s in [NpmSource()]:
-            LOG.info("Syncing {}".format(s.__class__.__name__))
+            LOG.info("Syncing %s", s.__class__.__name__)
             s.download_recent()
     if args.sync_github:
         for s in [GitHubSource()]:
-            LOG.info("Syncing {}".format(s.__class__.__name__))
+            LOG.info("Syncing %s", s.__class__.__name__)
             s.download_recent()
     if args.search_npm:
         source = NpmSource()
@@ -204,7 +200,7 @@ def main():
         results = dbLib.list_all_occurrence(db)
         print_results(results)
     elif args.search:
-        LOG.info("Vulnerability database loaded from {}".format(config.vdb_bin_file))
+        LOG.info("Vulnerability database loaded from %s", config.vdb_bin_file)
         db = dbLib.get()
         search_list = re.split(r"[,|;]", args.search)
         for pkg_info in search_list:

vdb/lib/__init__.py

@@ -207,12 +207,12 @@ def __init__(
         # Occasionally, NVD CPE value could be invalid. We need to guard against this
         if parts:
             self.package = package if package else parts.group("package")
-            self.min_affected_version_including = (
+            self.mii = (
                 min_affected_version_including
                 if min_affected_version_including
                 else parts.group("version")
             )
-            self.max_affected_version_including = (
+            self.mai = (
                 max_affected_version_including
                 if max_affected_version_including
                 else parts.group("version")
@@ -225,20 +225,20 @@ def __init__(
                 if len(cpe_parts) > 4:
                     package_workaround = cpe_parts[4]
             self.package = package if package else package_workaround
-            self.min_affected_version_including = (
+            self.mii = (
                 min_affected_version_including
                 if min_affected_version_including
                 else "*"
             )
-            self.max_affected_version_including = (
+            self.mai = (
                 max_affected_version_including
                 if max_affected_version_including
                 else "*"
             )
-        self.min_affected_version_excluding = (
+        self.mie = (
             min_affected_version_excluding if min_affected_version_excluding else None
         )
-        self.max_affected_version_excluding = (
+        self.mae = (
             max_affected_version_excluding if max_affected_version_excluding else None
         )
         self.severity = Severity.from_str(severity)
@@ -288,10 +288,10 @@ def from_dict(detail):
         return VulnerabilityDetail(
             detail.get("cpe_uri"),
             detail.get("package"),
-            detail.get("min_affected_version_including"),
-            detail.get("max_affected_version_including"),
-            detail.get("min_affected_version_excluding"),
-            detail.get("max_affected_version_excluding"),
+            detail.get("mii"),
+            detail.get("mai"),
+            detail.get("mie"),
+            detail.get("mae"),
             detail.get("severity"),
             detail.get("description"),
             detail.get("fixed_location"),
@@ -308,21 +308,21 @@ def __init__(
         self,
         affected_location,
         fixed_location,
-        min_affected_version_including=None,
-        max_affected_version_including=None,
-        min_affected_version_excluding=None,
-        max_affected_version_excluding=None,
+        mii=None,
+        mai=None,
+        mie=None,
+        mae=None,
     ):
         self.affected_location = VulnerabilityLocation.from_values(
             affected_location,
-            min_affected_version_including,
-            max_affected_version_including,
-            min_affected_version_excluding,
-            max_affected_version_excluding,
+            mii,
+            mai,
+            mie,
+            mae,
         )
         # If there is no fixed_location but there is max excluded version then consider that as the fix
-        if not fixed_location and max_affected_version_excluding:
-            self.fixed_location = max_affected_version_excluding
+        if not fixed_location and mae:
+            self.fixed_location = mae
         else:
             if fixed_location and fixed_location.startswith("cpe"):
                 # Extract the fixed version from fixed_location cpe uri
@@ -408,36 +408,26 @@ def __init__(self, cpe_uri, vendor, package, version):
     @staticmethod
     def from_values(
         cpe_uri,
-        min_affected_version_including=None,
-        max_affected_version_including=None,
-        min_affected_version_excluding=None,
-        max_affected_version_excluding=None,
+        mii=None,
+        mai=None,
+        mie=None,
+        mae=None,
     ):
-        if (
-            not cpe_uri
-            and not min_affected_version_including
-            and not max_affected_version_including
-            and not min_affected_version_excluding
-            and not max_affected_version_excluding
-        ):
+        if not cpe_uri and not mii and not mai and not mie and not mae:
             return None
         if cpe_uri:
             parts = CPE_REGEX.match(cpe_uri)
-            version = (
-                max_affected_version_including
-                if max_affected_version_including
-                else parts.group(3)
-            )
+            version = mai if mai else parts.group(3)
         version_left = ""
         version_right = ""
-        if min_affected_version_excluding:
-            version_left = ">" + min_affected_version_excluding
-        if min_affected_version_including and min_affected_version_including != "*":
-            version_left = ">=" + min_affected_version_including
-        if max_affected_version_excluding:
-            version_right = "<" + max_affected_version_excluding
-        if max_affected_version_including and max_affected_version_including != "*":
-            version_right = "<=" + max_affected_version_including
+        if mie:
+            version_left = ">" + mie
+        if mii and mii != "*":
+            version_left = ">=" + mii
+        if mae:
+            version_right = "<" + mae
+        if mai and mai != "*":
+            version_right = "<=" + mai
         if version_left and not version_right:
             version = version_left
             # Convert >0.0.0 to *
@@ -446,8 +436,8 @@ def from_values(
         elif not version_left and version_right:
             version = version_right
         elif version_left and version_right:
-            if min_affected_version_including == max_affected_version_including:
-                version = max_affected_version_including
+            if mii == mai:
+                version = mai
             else:
                 version = "{}-{}".format(version_left, version_right)
         if parts: