Search by purl. Also fixes #185 (#186)

Signed-off-by: Prabhu Subramanian <prabhu@appthreat.com>

README.md

@@ -108,22 +108,26 @@ Use the `/scan` endpoint to perform scans.
 > [!NOTE]
 > The `type` parameter is mandatory in server mode.
 
-* Scanning a local directory.
+- Scanning a local directory.
+
 ```bash
 curl --json '{"path": "/tmp/vulnerable-aws-koa-app", "type": "js"}' http://0.0.0.0:7070/scan
 ```
 
-* Scanning a SBOM file (present locally).
+- Scanning a SBOM file (present locally).
+
 ```bash
 curl --json '{"path": "/tmp/vulnerable-aws-koa-app/sbom_file.json", "type": "js"}' http://0.0.0.0:7070/scan
 ```
 
-* Scanning a GitHub repo.
+- Scanning a GitHub repo.
+
 ```bash
 curl --json '{"url": "https://github.com/HooliCorp/vulnerable-aws-koa-app", "type": "js"}' http://0.0.0.0:7070/scan -o app.vdr.json
 ```
 
-* Uploading a SBOM file and generating results based on it.
+- Uploading a SBOM file and generating results based on it.
+
 ```bash
 curl -X POST -H 'Content-Type: multipart/form-data' -F 'file=@/tmp/app/sbom_file.json' http://0.0.0.0:7070/scan?type=js
 ```
@@ -203,6 +207,7 @@ options:
   --explain             Makes depscan to explain the various analysis. Useful for creating detailed reports.
   --reachables-slices-file REACHABLES_SLICES_FILE
                         Path for the reachables slices file created by atom.
+  --purl SEARCH_PURL    Scan a single package url.
   -v, --version         Display the version
 ```
 

depscan/cli.py

@@ -298,6 +298,11 @@ def build_args():
         dest="reachables_slices_file",
         help="Path for the reachables slices file created by atom.",
     )
+    parser.add_argument(
+        "--purl",
+        dest="search_purl",
+        help="Scan a single package url.",
+    )
     parser.add_argument(
         "-v",
         "--version",
@@ -448,7 +453,7 @@ def summarise(
         reached_purls=reached_purls,
     )
     pkg_vulnerabilities, pkg_group_rows = prepare_vdr(options)
-    vdr_file = bom_file.replace(".json", ".vdr.json")
+    vdr_file = bom_file.replace(".json", ".vdr.json") if bom_file else None
     if pkg_vulnerabilities and bom_file:
         try:
             with open(bom_file, encoding="utf-8") as fp:
@@ -508,9 +513,15 @@ def download_rafs_based_image():
                 target=vdb_rafs_database_url, outdir=rafs_data_dir.name
             )
 
-            if paths_list and os.path.exists(
-                os.path.join(rafs_data_dir.name, "data.rafs")
-            ) and os.path.exists(os.path.join(rafs_data_dir.name, "meta.rafs")):
+            if (
+                paths_list
+                and os.path.exists(
+                    os.path.join(rafs_data_dir.name, "data.rafs")
+                )
+                and os.path.exists(
+                    os.path.join(rafs_data_dir.name, "meta.rafs")
+                )
+            ):
                 nydus_download_command = [
                     f"{nydus_image_command}",
                     "unpack",
@@ -540,7 +551,8 @@ def download_rafs_based_image():
 
         except Exception:
             LOG.info(
-                "Unable to pull the vulnerability database (rafs image) from %s. Trying to pull the non-rafs-based VDB image.", vdb_rafs_database_url
+                "Unable to pull the vulnerability database (rafs image) from %s. Trying to pull the non-rafs-based VDB image.",
+                vdb_rafs_database_url,
             )
             rafs_image_downloaded = False
 
@@ -819,6 +831,12 @@ def main():
     # Detect the project types and perform the right type of scan
     if args.project_type:
         project_types_list = args.project_type.split(",")
+    elif args.search_purl:
+        purl_obj = parse_purl(args.search_purl)
+        purl_obj["purl"] = args.search_purl
+        purl_obj["vendor"] = purl_obj.get("namespace")
+        project_types_list = [purl_obj.get("type")]
+        pkg_list = [purl_obj]
     elif args.bom and not args.project_type:
         project_types_list = ["bom"]
     elif not args.non_universal_scan:
@@ -857,7 +875,12 @@ def main():
         risk_report_file = areport_file.replace(
             ".json", f"-risk.{project_type}.json"
         )
-        if args.bom and os.path.exists(args.bom):
+        # Are we scanning a single purl
+        if args.search_purl:
+            bom_file = None
+            creation_status = True
+        # Are we scanning a bom file
+        elif args.bom and os.path.exists(args.bom):
             bom_file = args.bom
             creation_status = True
         else:
@@ -876,14 +899,15 @@ def main():
         if not creation_status:
             LOG.debug("Bom file %s was not created successfully", bom_file)
             continue
-        LOG.debug("Scanning using the bom file %s", bom_file)
-        if not args.bom:
-            LOG.info(
-                "To improve performance, cache the bom file and invoke "
-                "depscan with --bom %s instead of -i",
-                bom_file,
-            )
-        pkg_list = get_pkg_list(bom_file)
+        if bom_file:
+            LOG.debug("Scanning using the bom file %s", bom_file)
+            if not args.bom:
+                LOG.info(
+                    "To improve performance, cache the bom file and invoke "
+                    "depscan with --bom %s instead of -i",
+                    bom_file,
+                )
+            pkg_list = get_pkg_list(bom_file)
         if not pkg_list:
             LOG.debug("No packages found in the project!")
             continue
@@ -1097,7 +1121,10 @@ def main():
             result_file=os.path.join(reports_dir, args.report_name),
         )
     elif args.report_template:
-        LOG.warning("Template file %s doesn't exist, custom report not created.", args.report_template)
+        LOG.warning(
+            "Template file %s doesn't exist, custom report not created.",
+            args.report_template,
+        )
     # Submit vdr/vex files to threatdb server
     if args.threatdb_server and (args.threatdb_username or args.threatdb_token):
         submit_bom(

depscan/lib/analysis.py

@@ -11,7 +11,7 @@
 from rich.table import Table
 from rich.tree import Tree
 from vdb.lib import CPE_FULL_REGEX
-from vdb.lib.config import placeholder_fix_version
+from vdb.lib.config import placeholder_exclude_version, placeholder_fix_version
 from vdb.lib.utils import parse_cpe, parse_purl
 
 from depscan.lib import config
@@ -84,7 +84,7 @@ def retrieve_bom_dependency_tree(bom_file):
     :return: Dependency tree as a list
     """
     if not bom_file:
-        return []
+        return [], None
     try:
         with open(bom_file, encoding="utf-8") as bfp:
             bom_data = json.load(bfp)
@@ -97,6 +97,8 @@ def retrieve_bom_dependency_tree(bom_file):
 
 def retrieve_oci_properties(bom_data):
     props = {}
+    if not bom_data:
+        return props
     for p in bom_data.get("metadata", {}).get("properties", []):
         if p.get("name", "").startswith("oci:image:"):
             props[p.get("name")] = p.get("value")
@@ -1174,7 +1176,11 @@ def suggest_version(results, pkg_aliases=None, purl_aliases=None):
         else:
             full_pkg = pkg_aliases.get(full_pkg, full_pkg)
         version_upgrades = pkg_fix_map.get(full_pkg, set())
-        version_upgrades.add(fixed_location)
+        if (
+            fixed_location != placeholder_fix_version
+            and fixed_location != placeholder_exclude_version
+        ):
+            version_upgrades.add(fixed_location)
         pkg_fix_map[full_pkg] = version_upgrades
     for k, v in pkg_fix_map.items():
         # Don't go near certain packages

depscan/lib/audit.py

@@ -4,14 +4,16 @@
 from depscan.lib.pkg_query import npm_metadata, pypi_metadata
 
 # Dict mapping project type to the audit source
-type_audit_map = {"nodejs": NpmSource(), "js": NpmSource()}
+type_audit_map = {"nodejs": NpmSource(), "js": NpmSource(), "npm": NpmSource()}
 
 # Dict mapping project type to risk audit
 risk_audit_map = {
+    "npm": npm_metadata,
     "nodejs": npm_metadata,
     "js": npm_metadata,
     "python": pypi_metadata,
     "py": pypi_metadata,
+    "pypi": pypi_metadata,
 }
 
 

depscan/lib/csaf.py

@@ -1577,6 +1577,7 @@ def parse_toml(metadata):
         'current_release_date'.
     """
     tracking = parse_revision_history(metadata.get("tracking"))
+    # FIXME: Could this be simplified as list comprehension without append
     refs = []
     [refs.append(v) for v in metadata.get("reference")]
     notes = []

depscan/lib/normalize.py

@@ -54,10 +54,10 @@ def create_pkg_variations(pkg_dict):
             if purl_obj:
                 pkg_type = purl_obj.get("type")
                 qualifiers = purl_obj.get("qualifiers", {})
-                if qualifiers.get("distro_name"):
+                if qualifiers and qualifiers.get("distro_name"):
                     os_distro_name = qualifiers.get("distro_name")
                     name_aliases.add(f"""{os_distro_name}/{name}""")
-                if qualifiers.get("distro"):
+                if qualifiers and qualifiers.get("distro"):
                     os_distro = qualifiers.get("distro")
                     name_aliases.add(f"""{os_distro}/{name}""")
                     # almalinux-9.2 becomes almalinux-9
@@ -70,6 +70,7 @@ def create_pkg_variations(pkg_dict):
     if vendor:
         vendor_aliases.add(vendor)
         vendor_aliases.add(vendor.lower())
+        vendor_aliases.add(vendor.lstrip("@"))
         if (
             vendor.startswith("org.")
             or vendor.startswith("io.")
@@ -94,7 +95,6 @@ def create_pkg_variations(pkg_dict):
             vendor_aliases.add("golang")
     if pkg_type not in config.OS_PKG_TYPES:
         name_aliases.add("package_" + name)
-        vendor_aliases.add(pkg_type)
         if purl.startswith("pkg:composer"):
             vendor_aliases.add("get" + name)
             vendor_aliases.add(name + "_project")
@@ -168,7 +168,7 @@ def create_pkg_variations(pkg_dict):
             name_aliases.add(name + "-bin")
     else:
         # Filter vendor aliases that are also name aliases
-        vendor_aliases = [x for x in vendor_aliases if x not in name_aliases]
+        vendor_aliases = [x for x in vendor_aliases if x not in name_aliases or x == vendor]
     if len(vendor_aliases) > 0:
         for vvar in list(vendor_aliases):
             for nvar in list(name_aliases):

pyproject.toml

@@ -1,12 +1,12 @@
 [project]
 name = "owasp-depscan"
-version = "5.0.2"
+version = "5.0.3"
 description = "Fully open-source security audit for project dependencies based on known vulnerabilities and advisories."
 authors = [
     {name = "Team AppThreat", email = "cloud@appthreat.com"},
 ]
 dependencies = [
-    "appthreat-vulnerability-db>=5.5.4",
+    "appthreat-vulnerability-db>=5.5.5",
     "defusedxml",
     "oras",
     "PyYAML",