Use package type information if available. Fixes #29

depscan/cli.py

@@ -113,18 +113,20 @@ def build_args():
     return parser.parse_args()
 
 
-def scan(db, pkg_list, suggest_mode):
+def scan(db, project_type, pkg_list, suggest_mode):
     """
     Method to search packages in our vulnerability database
 
     :param db: Reference to db
+    :param project_type: Project Type
     :param pkg_list: List of packages
+    :param suggest_mode: True if package fix version should be normalized across findings
     """
     if not pkg_list:
         LOG.debug("Empty package search attempted!")
     else:
         LOG.info("Scanning {} oss dependencies for issues".format(len(pkg_list)))
-    results, pkg_aliases = utils.search_pkgs(db, pkg_list)
+    results, pkg_aliases = utils.search_pkgs(db, project_type, pkg_list)
     # pkg_aliases is a dict that can be used to find the original vendor and package name
     # This way we consistently use the same names used by the caller irrespective of how
     # the result was obtained
@@ -192,6 +194,7 @@ def summarise(
     :return: Summary of the results
     """
     if not results:
+        LOG.info(f"No oss vulnerabilities detected for type {project_type} ✅")
         return None
     if report_file:
         jsonl_report(
@@ -305,7 +308,9 @@ def main():
                     src_dir, project_type
                 )
             )
-            results, pkg_aliases, sug_version_dict = scan(db, pkg_list, args.suggest)
+            results, pkg_aliases, sug_version_dict = scan(
+                db, project_type, pkg_list, args.suggest
+            )
         # Summarise and print results
         summary = summarise(
             project_type,

depscan/lib/analysis.py

@@ -189,6 +189,7 @@ def jsonl_report(
             data_obj = {
                 "id": id,
                 "package": full_pkg,
+                "package_type": vuln_occ_dict.get("type"),
                 "package_usage": package_usage,
                 "version": package_issue.affected_location.version,
                 "fix_version": fixed_location,

depscan/lib/config.py

@@ -119,3 +119,23 @@
     ".terraform",
     ".serverless",
 ]
+
+# Package types allowed for each language
+LANG_PKG_TYPES = {
+    "python": "pypi",
+    "java": "maven",
+    "jvm": "maven",
+    "groovy": "maven",
+    "kotlin": "maven",
+    "scala": "maven",
+    "js": "npm",
+    "javascript": "npm",
+    "nodejs": "npm",
+    "go": "golang",
+    "golang": "golang",
+    "ruby": "rubygems",
+    "php": "composer",
+    "dotnet": "nuget",
+    "csharp": "nuget",
+    "rust": "crates",
+}

depscan/lib/normalize.py

@@ -1,3 +1,5 @@
+from vdb.lib import KNOWN_PKG_TYPES
+
 from depscan.lib import config as config
 
 # Common package suffixes
@@ -110,10 +112,11 @@ def create_pkg_variations(pkg_dict):
     return pkg_list
 
 
-def dealias_packages(pkg_list, pkg_aliases):
+def dealias_packages(project_type, pkg_list, pkg_aliases):
     """Method to dealias package names by looking up vendor and name information
     in the aliases list
 
+    :param project_type: Project type
     :param pkg_list: List of packages to dealias
     :param pkg_aliases: Package aliases
     """
@@ -137,16 +140,23 @@ def dealias_packages(pkg_list, pkg_aliases):
     return dealias_dict
 
 
-def dedup(pkg_list, pkg_aliases):
+def dedup(project_type, pkg_list, pkg_aliases):
     """Method to trim duplicates in the results based on the id. The logic should ideally be based on package alias but is kept simple for now.
 
+    :param project_type: Project type
     :param pkg_list: List of packages to dedup
     :param pkg_aliases: Package aliases
     """
     dedup_dict = {}
     ret_list = []
     for res in pkg_list:
         vid = res.id
+        vuln_occ_dict = res.to_dict()
+        package_type = vuln_occ_dict.get("type")
+        allowed_type = config.LANG_PKG_TYPES.get(project_type)
+        if package_type and package_type in KNOWN_PKG_TYPES and allowed_type:
+            if allowed_type != package_type:
+                dedup_dict[vid] = True
         if vid not in dedup_dict:
             ret_list.append(res)
             dedup_dict[vid] = True

depscan/lib/utils.py

@@ -123,11 +123,12 @@ def get_pkg_vendor_name(pkg):
     return vendor, name
 
 
-def search_pkgs(db, pkg_list):
+def search_pkgs(db, project_type, pkg_list):
     """
     Method to search packages in our vulnerability database
 
     :param db: DB instance
+    :param project_type: Project type
     :param pkg_list: List of packages to search
     """
     expanded_list = []
@@ -142,8 +143,10 @@ def search_pkgs(db, pkg_list):
         ]
     quick_res = dbLib.bulk_index_search(expanded_list)
     raw_results = dbLib.pkg_bulk_search(db, quick_res)
-    raw_results = normalize.dedup(raw_results, pkg_aliases=pkg_aliases)
-    pkg_aliases = normalize.dealias_packages(raw_results, pkg_aliases=pkg_aliases)
+    raw_results = normalize.dedup(project_type, raw_results, pkg_aliases=pkg_aliases)
+    pkg_aliases = normalize.dealias_packages(
+        project_type, raw_results, pkg_aliases=pkg_aliases
+    )
     return raw_results, pkg_aliases
 
 

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="appthreat-depscan",
-    version="1.10.0",
+    version="1.10.1",
     author="Team AppThreat",
     author_email="[email protected]",
     description="Fully open-source security audit for project dependencies based on known vulnerabilities and advisories.",

test/test_bom.py

@@ -133,7 +133,7 @@ def test_search(test_db):
         os.path.dirname(os.path.realpath(__file__)), "data", "bom.xml"
     )
     pkg_list = get_pkg_list(test_bom)
-    search_res, pkg_aliases = search_pkgs(test_db, pkg_list)
+    search_res, pkg_aliases = search_pkgs(test_db, "java", pkg_list)
     assert not len(search_res)
 
 
@@ -142,7 +142,7 @@ def test_go_search(test_db):
         os.path.dirname(os.path.realpath(__file__)), "data", "bom-go.xml"
     )
     pkg_list = get_pkg_list(test_bom)
-    search_res, pkg_aliases = search_pkgs(test_db, pkg_list)
+    search_res, pkg_aliases = search_pkgs(test_db, "golang", pkg_list)
     assert not len(search_res)
 
 
@@ -151,7 +151,7 @@ def test_search_webgoat(test_db):
         os.path.dirname(os.path.realpath(__file__)), "data", "bom-webgoat.xml"
     )
     pkg_list = get_pkg_list(test_bom)
-    search_res, pkg_aliases = search_pkgs(test_db, pkg_list)
+    search_res, pkg_aliases = search_pkgs(test_db, "java", pkg_list)
     assert not len(search_res)
 
 
@@ -161,5 +161,5 @@ def test_search_webgoat_json(test_db):
     )
     pkg_list = get_pkg_list(test_bom)
     assert len(pkg_list) == 157
-    search_res, pkg_aliases = search_pkgs(test_db, pkg_list)
+    search_res, pkg_aliases = search_pkgs(test_db, "java", pkg_list)
     assert not len(search_res)