🎉 Make Trivy Operator K8s vulnids consistent

dojo/settings/.settings.dist.py.sha256sum

@@ -1 +1 @@
-58e2f6cb0ed2c041fe2741d955b72cb7540bfb0923f489d6324717fcf00039da
+16d7a27d3146421a9aa6a8b1283f3d71b5c41b8bdb7c88ca70b0160e251034d1

dojo/settings/settings.dist.py

@@ -1744,6 +1744,8 @@ def saml2_attrib_map_format(dict):
     "ELSA": "https://linux.oracle.com/errata/&&.html",  # e.g. https://linux.oracle.com/errata/ELSA-2024-12714.html
     "ELBA": "https://linux.oracle.com/errata/&&.html",  # e.g. https://linux.oracle.com/errata/ELBA-2024-7457.html
     "RXSA": "https://errata.rockylinux.org/",  # e.g. https://errata.rockylinux.org/RXSA-2024:4928
+    "AVD": "https://avd.aquasec.com/misconfig/",  # e.g. https://avd.aquasec.com/misconfig/avd-ksv-01010
+    "KHV": "https://avd.aquasec.com/misconfig/kubernetes/",  # e.g. https://avd.aquasec.com/misconfig/kubernetes/khv045
     "CAPEC": "https://capec.mitre.org/data/definitions/&&.html",  # e.g. https://capec.mitre.org/data/definitions/157.html
     "CWE": "https://cwe.mitre.org/data/definitions/&&.html",  # e.g. https://cwe.mitre.org/data/definitions/79.html
     "TEMP": "https://security-tracker.debian.org/tracker/",  # e.g. https://security-tracker.debian.org/tracker/TEMP-0841856-B18BAF

dojo/templatetags/display_tags.py

@@ -780,6 +780,8 @@ def vulnerability_url(vulnerability_id):
 
     for key in settings.VULNERABILITY_URLS:
         if vulnerability_id.upper().startswith(key):
+            if key in ["AVD", "KHV"]:
+                return settings.VULNERABILITY_URLS[key] + str(vulnerability_id.lower())
             if "&&" in settings.VULNERABILITY_URLS[key]:
                 # Process specific keys specially if need
                 if key in ["CAPEC", "CWE"]:

dojo/tools/trivy_operator/checks_handler.py

@@ -1,4 +1,5 @@
 from dojo.models import Finding
+from dojo.tools.trivy_operator.uniform_vulnid import UniformTrivyVulnID
 
 TRIVY_SEVERITIES = {
     "CRITICAL": "Critical",
@@ -47,6 +48,6 @@ def handle_checks(self, labels, checks, test):
                 tags=[resource_namespace],
             )
             if check_id:
-                finding.unsaved_vulnerability_ids = [check_id]
+                finding.unsaved_vulnerability_ids = [UniformTrivyVulnID().return_uniformed_vulnid(check_id)]
             findings.append(finding)
         return findings

dojo/tools/trivy_operator/compliance_handler.py

@@ -1,4 +1,5 @@
 from dojo.models import Finding
+from dojo.tools.trivy_operator.uniform_vulnid import UniformTrivyVulnID
 
 TRIVY_SEVERITIES = {
     "CRITICAL": "Critical",
@@ -54,6 +55,6 @@ def handle_compliance(self, benchmarkreport, test):
                         dynamic_finding=True,
                     )
                     if check_checkID:
-                        finding.unsaved_vulnerability_ids = [check_checkID]
+                        finding.unsaved_vulnerability_ids = [UniformTrivyVulnID().return_uniformed_vulnid(check_checkID)]
                     findings.append(finding)
         return findings

dojo/tools/trivy_operator/secrets_handler.py

@@ -42,6 +42,7 @@ def handle_secrets(self, labels, secrets, test):
             secret_description += "\n**resource.kind:** " + resource_kind
             secret_description += "\n**resource.name:** " + resource_name
             secret_description += "\n**resource.namespace:** " + resource_namespace
+            secret_description += "\n**ruleID:** " + secret_rule_id
             finding = Finding(
                 test=test,
                 title=title,
@@ -54,7 +55,5 @@ def handle_secrets(self, labels, secrets, test):
                 service=service,
                 tags=[resource_namespace],
             )
-            if secret_rule_id:
-                finding.unsaved_vulnerability_ids = [secret_rule_id]
             findings.append(finding)
         return findings

dojo/tools/trivy_operator/uniform_vulnid.py

@@ -0,0 +1,20 @@
+import re
+
+
+class UniformTrivyVulnID:
+    def return_uniformed_vulnid(self, vulnid):
+        if vulnid is None:
+            return vulnid
+        if "cve" in vulnid.lower():
+            return vulnid
+        if "khv" in vulnid.lower():
+            temp = re.compile("([a-zA-Z-_]+)([0-9]+)")
+            number = str(temp.match(vulnid).groups()[1]).zfill(3)
+            avd_category = str(temp.match(vulnid.lower()).groups()[0])
+            return avd_category.upper() + number
+        if "ksv" in vulnid.lower() or "kcv" in vulnid.lower():
+            temp = re.compile("([a-zA-Z-_]+)([0-9]+)")
+            number = str(temp.match(vulnid).groups()[1]).zfill(4)
+            avd_category = str(temp.match(vulnid.lower().replace("_", "").replace("-", "")).groups()[0].replace("avd", ""))
+            return "AVD-" + avd_category.upper() + "-" + number
+        return vulnid

dojo/tools/trivy_operator/vulnerability_handler.py

@@ -1,4 +1,5 @@
 from dojo.models import Finding
+from dojo.tools.trivy_operator.uniform_vulnid import UniformTrivyVulnID
 
 DESCRIPTION_TEMPLATE = """{title}
 **Fixed version:** {fixed_version}
@@ -85,6 +86,6 @@ def handle_vulns(self, labels, vulnerabilities, test):
                 tags=finding_tags,
             )
             if vuln_id:
-                finding.unsaved_vulnerability_ids = [vuln_id]
+                finding.unsaved_vulnerability_ids = [UniformTrivyVulnID().return_uniformed_vulnid(vuln_id)]
             findings.append(finding)
         return findings

unittests/tools/test_trivy_operator_parser.py

@@ -25,7 +25,7 @@ def test_configauditreport_single_vulns(self):
             finding = findings[0]
             self.assertEqual("Low", finding.severity)
             self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
-            self.assertEqual("KSV014", finding.unsaved_vulnerability_ids[0])
+            self.assertEqual("AVD-KSV-0014", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("KSV014 - Root file system is not read-only", finding.title)
 
     def test_configauditreport_many_vulns(self):
@@ -36,12 +36,12 @@ def test_configauditreport_many_vulns(self):
             finding = findings[0]
             self.assertEqual("Low", finding.severity)
             self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
-            self.assertEqual("KSV014", finding.unsaved_vulnerability_ids[0])
+            self.assertEqual("AVD-KSV-0014", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("KSV014 - Root file system is not read-only", finding.title)
             finding = findings[1]
             self.assertEqual("Low", finding.severity)
             self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
-            self.assertEqual("KSV016", finding.unsaved_vulnerability_ids[0])
+            self.assertEqual("AVD-KSV-0016", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("KSV016 - Memory requests not specified", finding.title)
 
     def test_vulnerabilityreport_no_vuln(self):
@@ -96,8 +96,6 @@ def test_exposedsecretreport_single_vulns(self):
             self.assertEqual(len(findings), 1)
             finding = findings[0]
             self.assertEqual("Critical", finding.severity)
-            self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
-            self.assertEqual("aws-secret-access-key", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("aws-secret-access-key", finding.references)
             self.assertEqual("root/aws_secret.txt", finding.file_path)
             self.assertEqual("Secret detected in root/aws_secret.txt - AWS Secret Access Key", finding.title)
@@ -109,15 +107,11 @@ def test_exposedsecretreport_many(self):
             self.assertEqual(len(findings), 2)
             finding = findings[0]
             self.assertEqual("Critical", finding.severity)
-            self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
-            self.assertEqual("aws-secret-access-key", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("aws-secret-access-key", finding.references)
             self.assertEqual("root/aws_secret.txt", finding.file_path)
             self.assertEqual("Secret detected in root/aws_secret.txt - AWS Secret Access Key", finding.title)
             finding = findings[1]
             self.assertEqual("Critical", finding.severity)
-            self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
-            self.assertEqual("github-pat", finding.unsaved_vulnerability_ids[0])
             self.assertEqual("github-pat", finding.references)
             self.assertEqual("root/github_secret.txt", finding.file_path)
             self.assertEqual("Secret detected in root/github_secret.txt - GitHub Personal Access Token", finding.title)