Show severity and score from providers in addition to NVD in the summary report (#72)

Co-authored-by: Kenji Sugimura <kensugim@amazon.co.jp>

Author: s-kenji

entrypoint/entrypoint/pkg_vuln.py

@@ -10,6 +10,7 @@
 import os
 import urllib.parse
 
+from dataclasses import dataclass
 from io import StringIO
 from typing import List
 
@@ -38,6 +39,12 @@ def __init__(self):
         self.cwes = "null"
 
 
+@dataclass
+class CvssRating:
+    severity: str = "null"
+    cvss_score: str = "null"
+
+
 def vulns_to_obj(inspector_scan_json) -> List[Vulnerability]:
     """
     this function parses JSON from Inspector's ScanSbom API
@@ -70,18 +77,7 @@ def vulns_to_obj(inspector_scan_json) -> List[Vulnerability]:
 
         # get vuln severity and EPSS score
         ratings = v.get("ratings")
-        if ratings:
-            nvd_severity = get_nvd_severity(ratings)
-            if nvd_severity:
-                vuln_obj.severity = nvd_severity
-
-            nvd_score = get_nvd_score(ratings)
-            if nvd_score:
-                vuln_obj.cvss_score = nvd_score
-
-            epss_score = get_epss_score(ratings)
-            if epss_score:
-                vuln_obj.epss_score = epss_score
+        add_ratings(ratings, vuln_obj)
 
         # get vulnerability published date
         published = v.get("created")
@@ -207,34 +203,37 @@ def getPropertyValueFromKey(vuln_json, key):
     return None
 
 
-def get_nvd_severity(ratings):
-    for rating in ratings:
-        source = rating.get("source")
-        if not source:
-            continue
+def add_ratings(ratings, vulnerability):
+    if ratings is None:
+        return
 
-        if source["name"] == "NVD":
-            severity = rating["severity"]
-            if severity:
-                return severity
-    return None
+    rating = get_cvss_rating(ratings, vulnerability)
+    vulnerability.severity = rating.severity
+    vulnerability.cvss_score = rating.cvss_score
 
+    epss_score = get_epss_score(ratings)
+    if epss_score:
+        vulnerability.epss_score = epss_score
 
-def get_nvd_score(ratings):
-    for rating in ratings:
-        source = rating.get("source")
-        if not source:
-            continue
 
-        method = rating.get("method")
-        if not method:
-            continue
+def get_cvss_rating(ratings, vulnerability) -> CvssRating:
+    rating_provider_priority = [
+        'NVD',
+        'MITRE',
+        'AMAZON_INSPECTOR'
+    ]
+    for provider in rating_provider_priority:
+        for rating in ratings:
+            if rating['source']['name'] != provider:
+                continue
 
-        if source["name"] == "NVD" and method == "CVSSv31":
-            score = rating["score"]
-            if score:
-                return score
-    return None
+            severity = rating["severity"]
+            cvss_score = rating["score"] if rating['method'] == 'CVSSv31' else "null"
+            if severity and cvss_score:
+                return CvssRating(severity=severity, cvss_score=cvss_score)
+
+    logging.info(f"No CVSS rating is provided for {vulnerability.vuln_id}")
+    return CvssRating()
 
 
 def get_epss_score(ratings):

entrypoint/tests/__init__.py

@@ -1,3 +1,8 @@
-from entrypoint import log_conf
+import logging
 
-log_conf.init(enable_verbose=True)
+
+logger = logging.getLogger()
+logger.setLevel(logging.ERROR)
+
+handler = logging.StreamHandler()
+handler.setLevel(logging.ERROR)

entrypoint/tests/test_pkg_vuln.py

@@ -64,6 +64,67 @@ def test_post_github_step_summary_no_vulns(self):
             self.assertIn(expected, zero_vuln_summary_md)
         cleanup_stale_markdown_report(markdown_dst_path)
 
+    def test_get_rating(self):
+        tests = [
+            {
+                "name": "NVD should take a priority over other providers",
+                "ratings": [
+                    {
+                        "method": "CVSSv31",
+                        "score": 7.2,
+                        "severity": "high",
+                        "source": {"name": "MITRE"},
+                    },
+                    {
+                        "method": "CVSSv31",
+                        "score": 5.3,
+                        "severity": "medium",
+                        "source": {"name": "NVD"},
+                    },
+                    {
+                        "method": "other",
+                        "score": 0.00051,
+                        "severity": "none",
+                        "source": {"name": "EPSS"},
+                    },
+                ],
+                "expected": pkg_vuln.CvssRating("medium", 5.3),
+            },
+            {
+                "name": 'score should be "null" when method is "other"',
+                "ratings": [
+                    {
+                        "method": "other",
+                        "severity": "medium",
+                        "source": {"name": "AMAZON_INSPECTOR"},
+                    },
+                    {
+                        "method": "other",
+                        "score": 0.00051,
+                        "severity": "none",
+                        "source": {"name": "EPSS"},
+                    },
+                ],
+                "expected": pkg_vuln.CvssRating("medium"),
+            },
+            {
+                "name": 'rating should be "null" when neither NVD, MITRE, nor AMAZON_INSPECTOR provides a rating',
+                "ratings": [
+                    {
+                        "method": "other",
+                        "score": 0.00051,
+                        "severity": "none",
+                        "source": {"name": "EPSS"},
+                    },
+                ],
+                "expected": pkg_vuln.CvssRating(),
+            },
+        ]
+        for test in tests:
+            with self.subTest(msg=test["name"]):
+                rating = pkg_vuln.get_cvss_rating(test["ratings"], pkg_vuln.Vulnerability())
+                self.assertEqual(test["expected"], rating)
+
 
 def get_scan_body(test_file):
     # test_file = "tests/test_data/artifacts/containers/dockerfile_checks/inspector-scan-cdx.json"