Feature/oras (#131)

* Support for downloading precompiled image using oras

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

---------

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

Author: prabhu

README.md

@@ -39,22 +39,39 @@ Multiple upstream sources are used by vdb to improve accuracy and reduce false n
 ## Installation
 
 ```shell
-pip install appthreat-vulnerability-db>=6.0.0
+pip install appthreat-vulnerability-db>=6.0.1
 ```
 
-VDB v6 is a major rewrite to use sqlite database. Current users of depscan v5 must continue using version 5.6.x
+To install vdb with optional dependencies such as `oras` use the `[oras]` or `[all]` dependency group.
 
 ```shell
-pip install appthreat-vulnerability-db==5.6.6
+pip install appthreat-vulnerability-db[all]
+```
+
+**NOTE:** VDB v6 is a major rewrite to use sqlite database. Current users of depscan v5 must continue using version 5.6.x
+
+```shell
+pip install appthreat-vulnerability-db==5.6.7
 ```
 
 ## Usage
 
 This package is ideal as a library for managing vulnerabilities. This is used by [owasp-dep-scan](http://github.com/owasp-dep-scan/dep-scan), a free open-source dependency audit tool. However, there is a limited cli capability available with few features to test this tool directly.
 
-### Download pre-built database (Recommended)
+### Option 1: Download pre-built database (Recommended)
+
+To download a pre-built sqlite database ([refreshed](https://github.com/AppThreat/vdb/actions) every 6 hours) containing all application and OS vulnerabilities. This step is recommended for all users.
+
+```shell
+# pip install appthreat-vulnerability-db[all]
+vdb --download-image
+```
+
+You can execute this command daily or when a fresh database is required.
 
-Use the [ORAS cli](https://oras.land/cli/) to download a pre-built sqlite database ([refreshed](https://github.com/AppThreat/vdb/actions) every 6 hours) containing all application and OS vulnerabilities. This is recommended for all users.
+### Option 2: Download pre-built database (ORAS)
+
+Using [ORAS cli](https://oras.land/) might be slightly faster.
 
 ```
 export VDB_HOME=$HOME/vdb
@@ -73,7 +90,7 @@ Use any sqlite browser or cli tools to load and query the two databases.
 
 <img src="./docs/vdb6.png" alt="database" width="400">
 
-### Manually create the vulnerability database
+### Option 3: Manually create the vulnerability database (ADVANCED users)
 
 Cache application vulnerabilities
 
@@ -102,7 +119,7 @@ It is possible to customize the cache behavior by increasing the historic data p
 - NVD_START_YEAR - Default: 2018. Supports up to 2002
 - GITHUB_PAGE_COUNT - Default: 2. Supports up to 20
 
-## Usage
+## CLI Usage
 
 ```shell
 usage: vdb [-h] [--clean] [--cache] [--cache-os] [--only-osv] [--only-aqua] [--only-ghsa] [--search SEARCH] [--list-malware] [--bom BOM_FILE]
@@ -120,6 +137,7 @@ options:
   --search SEARCH  Search for the package or CVE ID in the database. Use purl, cpe, or git http url.
   --list-malware   List latest malwares with CVE ID beginning with MAL-.
   --bom BOM_FILE   Search for packages in the CycloneDX BOM file.
+  --download-image  Downloaded pre-created vdb image to platform specific user_data_dir.
 ```
 
 ### CLI search
@@ -158,6 +176,8 @@ vdb --bom bom.json
 
 ### List recent malware
 
+To list malware entries with the `MAL-` prefix, use the following command.
+
 ```shell
 vdb --list-malware
 ```

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "appthreat-vulnerability-db"
-version = "6.0.0"
+version = "6.0.1"
 description = "AppThreat's vulnerability database and package search library with a built-in sqlite based storage. OSV, CVE, GitHub, npm are the primary sources of vulnerabilities."
 authors = [
     {name = "Team AppThreat", email = "[email protected]"},
@@ -51,6 +51,8 @@ dev = [
     "pytest",
     "pytest-cov"
 ]
+oras = ["oras"]
+all = ["oras"]
 
 [tool.setuptools]
 packages = ["vdb", "vdb.lib", "vdb.lib.cve_model"]
@@ -61,3 +63,4 @@ addopts="--showlocals -v --cov-report=term-missing --no-cov-on-fail --cov vdb"
 [tool.pylint]
 disable = ["broad-exception-caught", "too-many-branches", "too-many-statements", "too-many-nested-blocks", "too-many-locals", "missing-function-docstring", "too-many-lines", "missing-module-docstring"]
 ignore-paths = ["vdb/lib/cve_model/*"]
+generated-member = ["orjson.loads", "orjson.dumps", "orjson.OPT_NAIVE_UTC"]

vdb/cli.py

@@ -18,6 +18,14 @@
 from vdb.lib.gha import GitHubSource
 from vdb.lib.osv import OSVSource
 
+ORAS_AVAILABLE = False
+# oras is an optional dependency
+try:
+    from vdb.lib.orasclient import download_image
+    ORAS_AVAILABLE = True
+except ImportError:
+    pass
+
 console = Console()
 
 logging.basicConfig(
@@ -101,6 +109,13 @@ def build_args():
         dest="bom_file",
         help="Search for packages in the CycloneDX BOM file.",
     )
+    parser.add_argument(
+        "--download-image",
+        action="store_true",
+        default=False,
+        dest="download_image",
+        help="Downloaded pre-created vdb image to platform specific user_data_dir.",
+    )
     return parser.parse_args()
 
 
@@ -159,7 +174,13 @@ def main():
     if args.clean:
         if os.path.exists(config.DATA_DIR):
             shutil.rmtree(config.DATA_DIR, ignore_errors=True)
-    if args.cache or args.cache_os:
+    if args.download_image:
+        if ORAS_AVAILABLE:
+            LOG.info("Downloading vdb image from %s to %s", config.VDB_DATABASE_URL, config.DATA_DIR)
+            download_image(config.VDB_DATABASE_URL, config.DATA_DIR)
+        else:
+            console.print("Oras library is not available. Install using pip install appthreat-vulnerability-db[oras] and then re-run this command.")
+    elif args.cache or args.cache_os:
         db_lib.get()
         db_lib.clear_all()
         if args.only_osv:

vdb/lib/aqua.py

@@ -143,13 +143,10 @@ def alsa_to_vuln(cve_data):
             references = references.decode("utf-8", "ignore")
         description = cve_data.get("description", "")
         if not description and cve_data.get("title"):
-            description = """# {}
-{}
-{}
-            """.format(
-                cve_data.get("summary"), cve_data.get("title"), cve_data.get("solution")
-            )
-
+            description = f"""# {cve_data.get("summary")}
+{cve_data.get("title")}
+{cve_data.get("solution")}
+            """
         assigner = cve_data.get("fromstr", "")
         severity = config.THREAT_TO_SEVERITY[cve_data.get("severity").lower()]
         score, severity, vector_string, attack_complexity = get_default_cve_data(
@@ -689,11 +686,9 @@ def photon_to_vuln(cve_data):
         pkg_name = cve_data.get("pkg")
         cwe_id = ""
         references = []
-        description = """Summary
-{}
-        """.format(
-            cve_data.get("aff_ver")
-        )
+        description = f"""Summary
+{cve_data.get("aff_ver")}
+        """
         assigner = "vmware"
         score = cve_data.get("cve_score")
         severity = convert_score_severity(score)
@@ -779,12 +774,10 @@ def debian_to_vuln(cve_data):
                 and ann.get("Type") == "xref"
                 and ann.get("Bugs")
             ):
-                aliases_block = """
+                aliases_block = f"""
 ## Related CVE(s)
-{}
-            """.format(
-                    ", ".join(ann.get("Bugs"))
-                )
+{", ".join(ann.get("Bugs"))}
+            """
                 description += aliases_block
                 for bug in ann.get("Bugs"):
                     if bug.startswith("CVE"):

vdb/lib/config.py

@@ -144,3 +144,6 @@
     "coreos",
     "ebuild",
 )
+
+# URL for the pre-compiled database
+VDB_DATABASE_URL = os.getenv("VDB_DATABASE_URL", "ghcr.io/appthreat/vdbxz:v6")

vdb/lib/nvd.py

@@ -2,7 +2,7 @@
 import gzip
 import logging
 from collections import defaultdict
-from urllib.parse import parse_qs, urlparse
+from urllib.parse import urlparse
 
 import httpx
 import orjson
@@ -148,7 +148,7 @@ class NvdSource(CVESource):
 
     def download_all(self):
         """Download all historic cve data"""
-        super.download_all()
+        super().download_all()
         for y in range(now.year, int(start_year) - 1, -1):
             data = self.fetch(y)
             if not data:
@@ -216,7 +216,6 @@ def bulk_search(self, app_info, pkg_list):
         Bulk search the resource instead of downloading the information
         :return: Vulnerability result
         """
-        pass
 
     @staticmethod
     def convert_vuln(vuln: dict) -> Vulnerability | None:

vdb/lib/orasclient.py

@@ -0,0 +1,58 @@
+import os
+import tarfile
+
+import oras.client
+import oras.provider
+from oras.logger import setup_logger
+
+
+setup_logger(quiet=True, debug=False)
+
+
+class VdbDistributionRegistry(oras.provider.Registry):
+    """
+    We override the default registry to make things compatible with ghcr. Without this, the below error is thrown.
+
+    jsonschema.exceptions.ValidationError: Additional properties are not allowed ('artifactType' was unexpected)
+    """
+
+    def get_manifest(self, container, allowed_media_type=None, refresh_headers=True):
+        """
+        Retrieve a manifest for a package.
+
+        :param container:  parsed container URI
+        :type container: oras.container.Container or str
+        :param allowed_media_type: one or more allowed media types
+        :type allowed_media_type: str
+        """
+        if not allowed_media_type:
+            allowed_media_type = [oras.defaults.default_manifest_media_type]
+        headers = {"Accept": ";".join(allowed_media_type)}
+
+        get_manifest = f"{self.prefix}://{container.manifest_url()}"  # type: ignore
+        response = self.do_request(get_manifest, "GET", headers=headers)
+        self._check_200_response(response)
+        manifest = response.json()
+        return manifest
+
+
+def download_image(target, outdir):
+    """
+    Method to download vdb files from a oci registry
+    """
+    oras_client = oras.client.OrasClient(registry=VdbDistributionRegistry())
+    paths_list = oras_client.pull(
+        target=target,
+        outdir=outdir,
+        allowed_media_type=[],
+        overwrite=True,
+    )
+    for apath in paths_list:
+        if apath.endswith(".tar.gz") or apath.endswith(".tar.xz"):
+            with tarfile.open(apath, "r") as tarf:
+                tarf.extractall(path=outdir)
+            try:
+                os.remove(apath)
+            except OSError:
+                pass
+    return paths_list