fix(awssecurityhub): extract CVSS v3/v4 scores from Inspector findings

Author: samiat4911

dojo/tools/awssecurityhub/inspector.py

@@ -31,6 +31,10 @@ def get_item(self, finding: dict, test):
         references = []
         unsaved_vulnerability_ids = []
         epss_score = finding.get("EpssScore")
+        cvssv3 = None
+        cvssv3_score = None
+        cvssv4 = None
+        cvssv4_score = None
         description = f"This is an Inspector Finding\n{finding.get('Description', '')}" + "\n"
         description += f"**AWS Finding ARN:** {finding_id}\n"
         description += f"**AwsAccountId:** {finding.get('AwsAccountId', '')}\n"
@@ -52,6 +56,15 @@ def get_item(self, finding: dict, test):
                     references.append(vendor_url)
             if vulnerability.get("EpssScore") is not None:
                 epss_score = vulnerability.get("EpssScore")
+            # Extract CVSS v3/v4 scores from the Cvss array
+            for cvss_entry in vulnerability.get("Cvss", []):
+                version = cvss_entry.get("Version", "")
+                if version.startswith("3") and cvssv3 is None:
+                    cvssv3 = cvss_entry.get("BaseVector")
+                    cvssv3_score = cvss_entry.get("BaseScore")
+                elif version.startswith("4") and cvssv4 is None:
+                    cvssv4 = cvss_entry.get("BaseVector")
+                    cvssv4_score = cvss_entry.get("BaseScore")
         if finding.get("ProductFields", {}).get("aws/inspector/FindingStatus", "ACTIVE") == "ACTIVE":
             mitigated = None
             is_Mitigated = False
@@ -120,6 +133,25 @@ def get_item(self, finding: dict, test):
             result.unsaved_endpoints = locations
         if epss_score is not None:
             result.epss_score = epss_score
+        if cvssv3 is not None:
+            result.cvssv3 = cvssv3
+        if cvssv3_score is not None:
+            result.cvssv3_score = cvssv3_score
+        if cvssv4 is not None:
+            result.cvssv4 = cvssv4
+        if cvssv4_score is not None:
+            result.cvssv4_score = cvssv4_score
+        # Build severity justification from available CVSS data
+        severity_parts = []
+        if cvssv3 is not None:
+            severity_parts.append(f"CVSS v3 vector: {cvssv3} (base score: {cvssv3_score})")
+        if cvssv4 is not None:
+            severity_parts.append(f"CVSS v4 vector: {cvssv4} (base score: {cvssv4_score})")
+        severity_label = finding.get("Severity", {}).get("Label", "")
+        if severity_label:
+            severity_parts.append(f"AWS severity: {severity_label}")
+        if severity_parts:
+            result.severity_justification = "\n".join(severity_parts)
         # Add the unsaved vulnerability ids
         result.unsaved_vulnerability_ids = unsaved_vulnerability_ids
         return result

…ests/tools/test_awssecurityhub_parser.py …ts/tools/test_aws_security_hub_parser.pyunittests/tools/test_awssecurityhub_parser.py renamed to unittests/tools/test_aws_security_hub_parser.py unittests/tools/test_awssecurityhub_parser.py renamed to unittests/tools/test_aws_security_hub_parser.py

@@ -72,6 +72,11 @@ def test_inspector_ec2(self):
             self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
             self.assertEqual("CVE-2022-3643", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("- Update kernel-4.14.301\n\t- yum update kernel\n", finding.mitigation)
+            # Verify CVSS v3 extraction from the Cvss array
+            self.assertEqual("CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H", finding.cvssv3)
+            self.assertEqual(10.0, finding.cvssv3_score)
+            self.assertIn("CVSS v3 vector:", finding.severity_justification)
+            self.assertIn("AWS severity: CRITICAL", finding.severity_justification)
             location = self.get_unsaved_locations(finding)[0]
             self.assertEqual("AwsEc2Instance_arn_aws_ec2_us-east-1_XXXXXXXXXXXX_i-11111111111111111".lower(), location.host.lower())
 
@@ -97,6 +102,9 @@ def test_inspector_ec2_ghsa(self):
             self.assertIn("GHSA-p98r-538v-jgw5", finding.title)
             self.assertSetEqual({"CVE-2023-34256", "GHSA-p98r-538v-jgw5"}, set(finding.unsaved_vulnerability_ids))
             self.assertEqual("https://github.com/bottlerocket-os/bottlerocket/security/advisories/GHSA-p98r-538v-jgw5", finding.references)
+            # Verify backward compatibility: no CVSS data in this fixture
+            self.assertIsNone(finding.cvssv3)
+            self.assertIsNone(finding.cvssv3_score)
             location = self.get_unsaved_locations(finding)[0]
             self.assertEqual("AwsEc2Instance_arn_aws_ec2_eu-central-1_012345678912_instance_i-07c11cc535d830123".lower(), location.host.lower())
 
@@ -115,6 +123,9 @@ def test_inspector_ecr(self):
             self.assertIn("repo-os/sha256:af965ef68c78374a5f987fce98c0ddfa45801df2395bf012c50b863e65978d74", finding.impact)
             self.assertIn("Repository: repo-os", finding.impact)
             self.assertEqual(0.0014, finding.epss_score)
+            # Verify CVSS v3 extraction from the ECR fixture
+            self.assertEqual("CVSS:3.1/AV:N/AC:L/PR:N/UI:R/S:U/C:N/I:N/A:H", finding.cvssv3)
+            self.assertEqual(6.5, finding.cvssv3_score)
             location = self.get_unsaved_locations(finding)[0]
             self.assertEqual("AwsEcrContainerImage_arn_aws_ecr_eu-central-1_123456789012_repository_repo-os_sha256_af965ef68c78374a5f987fce98c0ddfa45801df2395bf012c50b863e65978d74".lower(), location.host.lower())