🎉 Add Trivy Operator clustercompliance report

dojo/tools/trivy_operator/clustercompliance_handler.py

@@ -0,0 +1,45 @@
+from dojo.models import Finding
+
+TRIVY_SEVERITIES = {
+    "CRITICAL": "Critical",
+    "HIGH": "High",
+    "MEDIUM": "Medium",
+    "LOW": "Low",
+    "UNKNOWN": "Info",
+}
+
+
+class TrivyClusterComplianceHandler:
+    def handle_clustercompliance(self, controls, clustercompliance, test):
+        findings = []
+        for result in clustercompliance.get("controlCheck"):
+            if int(result.get("totalFail", 0)) > 0:
+                description = ""
+                result_id = result.get("id", "")
+                vulnerabilityids = []
+                for control in controls:
+                    if control.get("id") == result_id:
+                        vulnids = control.get("checks", [])
+                        for vulnid in vulnids:
+                            vulnerabilityids.append(vulnid.get("id"))
+                        description += "**description:** " + control.get("description") + "\n"
+                result_name = result.get("name", "")
+                result_severity = result.get("severity", "")
+                result_totalfail = str(result.get("totalFail", ""))
+                severity = TRIVY_SEVERITIES[result_severity]
+                description += "**id:** " + result_id + "\n"
+                description += "**name:** " + result_name + "\n"
+                description += "**totalfail:** " + result_totalfail + "\n"
+                title = "TrivyClusterCompliance " + result_id + " totalFail: " + result_totalfail
+                finding = Finding(
+                    test=test,
+                    title=title,
+                    description=description,
+                    severity=severity,
+                    static_finding=False,
+                    dynamic_finding=True,
+                )
+                if vulnerabilityids != []:
+                    finding.unsaved_vulnerability_ids = vulnerabilityids
+                findings.append(finding)
+        return findings

dojo/tools/trivy_operator/parser.py

@@ -3,6 +3,7 @@
 import json
 
 from dojo.tools.trivy_operator.checks_handler import TrivyChecksHandler
+from dojo.tools.trivy_operator.clustercompliance_handler import TrivyClusterComplianceHandler
 from dojo.tools.trivy_operator.compliance_handler import TrivyComplianceHandler
 from dojo.tools.trivy_operator.secrets_handler import TrivySecretsHandler
 from dojo.tools.trivy_operator.vulnerability_handler import TrivyVulnerabilityHandler
@@ -38,6 +39,7 @@ def get_findings(self, scan_file, test):
         return findings
 
     def output_findings(self, data, test):
+        findings = []
         if data is None:
             return []
         metadata = data.get("metadata", None)
@@ -47,10 +49,6 @@ def output_findings(self, data, test):
         if labels is None:
             return []
         report = data.get("report", None)
-        benchmark = data.get("status", None)
-        if benchmark is not None:
-            benchmarkreport = benchmark.get("detailReport", None)
-        findings = []
         if report is not None:
             vulnerabilities = report.get("vulnerabilities", None)
             if vulnerabilities is not None:
@@ -61,6 +59,13 @@ def output_findings(self, data, test):
             secrets = report.get("secrets", None)
             if secrets is not None:
                 findings += TrivySecretsHandler().handle_secrets(labels, secrets, test)
-        elif benchmarkreport is not None:
-            findings += TrivyComplianceHandler().handle_compliance(benchmarkreport, test)
+        status = data.get("status", None)
+        if status is not None:
+            benchmarkreport = status.get("detailReport", None)
+            if benchmarkreport is not None:
+                findings += TrivyComplianceHandler().handle_compliance(benchmarkreport, test)
+            clustercompliance = status.get("summaryReport", None)
+            if clustercompliance is not None:
+                if int(status.get("summary").get("failCount", 0)) > 0:
+                    findings += TrivyClusterComplianceHandler().handle_clustercompliance(controls=data.get("spec").get("compliance").get("controls"), clustercompliance=clustercompliance, test=test)
         return findings

unittests/scans/trivy_operator/clustercompliancereport.json

@@ -0,0 +1,105 @@
+[
+    {
+      "apiVersion": "aquasecurity.github.io/v1alpha1",
+      "kind": "ClusterComplianceReport",
+      "metadata": {
+        "annotations": {
+          "meta.helm.sh/release-name": "trivy-operator-asdf",
+          "meta.helm.sh/release-namespace": "asdf"
+        },
+        "creationTimestamp": "2021-1-1T10:10:10Z",
+        "generation": 1,
+        "labels": {
+          "app.kubernetes.io/instance": "trivy-operator",
+          "app.kubernetes.io/managed-by": "Helm",
+          "app.kubernetes.io/name": "trivy-operator",
+          "app.kubernetes.io/version": "0.17.1"
+        },
+        "name": "cis",
+        "resourceVersion": "asdf",
+        "uid": "afsewfwefwfw"
+      },
+      "spec": {
+        "compliance": {
+          "controls": [
+            {
+              "checks": [
+                {
+                  "id": "AVD-KCV-0087"
+                }
+              ],
+              "description": "Security relevant information should be captured. The --event-qps flag on the Kubelet can be used to limit the rate at which events are gathered",
+              "id": "4.2.9",
+              "name": "Ensure that the --event-qps argument is set to 0 or a level which ensures appropriate event capture",
+              "severity": "HIGH"
+            },
+            {
+              "checks": [
+                {
+                  "id": "AVD-KCV-0088"
+                },
+                {
+                  "id": "AVD-KCV-0089"
+                }
+              ],
+              "description": "Setup TLS connection on the Kubelets",
+              "id": "4.2.10",
+              "name": "Ensure that the --tls-cert-file and --tls-private-key-file arguments are set as appropriate",
+              "severity": "CRITICAL"
+            },
+            {
+              "checks": [
+                {
+                  "id": "AVD-KCV-0090"
+                }
+              ],
+              "description": "Enable kubelet client certificate rotation",
+              "id": "4.2.11",
+              "name": "Ensure that the --rotate-certificates argument is not set to false",
+              "severity": "CRITICAL"
+            }
+          ],
+          "description": "CIS Kubernetes Benchmarks",
+          "id": "cis",
+          "relatedResources": [
+            "https://www.cisecurity.org/benchmark/kubernetes"
+          ],
+          "title": "CIS Kubernetes Benchmarks v1.23",
+          "version": "1.0"
+        },
+        "cron": "0 */6 * * *",
+        "reportType": "summary"
+      },
+      "status": {
+        "summary": {
+          "failCount": 2,
+          "passCount": 1
+        },
+        "summaryReport": {
+          "controlCheck": [
+            {
+              "id": "4.2.9",
+              "name": "Ensure that the --event-qps argument is set to 0 or a level which ensures appropriate event capture",
+              "severity": "HIGH",
+              "totalFail": 0
+            },
+            {
+              "id": "4.2.10",
+              "name": "Ensure that the --tls-cert-file and --tls-private-key-file arguments are set as appropriate",
+              "severity": "CRITICAL",
+              "totalFail": 10
+            },
+            {
+              "id": "4.2.11",
+              "name": "Ensure that the --rotate-certificates argument is not set to false",
+              "severity": "CRITICAL",
+              "totalFail": 5
+            }
+          ],
+          "id": "cis",
+          "title": "CIS Kubernetes Benchmarks v1.23"
+        },
+        "updateTimestamp": "2021-1-1T10:10:10Z"
+      }
+    }
+  ]

unittests/tools/test_trivy_operator_parser.py

@@ -163,3 +163,9 @@ def test_findings_all_reports_in_dict(self):
             parser = TrivyOperatorParser()
             findings = parser.get_findings(test_file, Test())
             self.assertEqual(len(findings), 43)
+
+    def test_findings_clustercompliancereport(self):
+        with open(sample_path("clustercompliancereport.json"), encoding="utf-8") as test_file:
+            parser = TrivyOperatorParser()
+            findings = parser.get_findings(test_file, Test())
+            self.assertEqual(len(findings), 2)