Kube-hunter Parser (#8593)

* Create a parser for kube-hunter

* fix syntax

* adding doc

* using None instead empty string

* removing unnecessary extra sentence added to step_to_reproduce

docs/content/en/integrations/parsers/file/kubehunter.md

@@ -0,0 +1,5 @@
+---
+title: "kubeHunter Scanner"
+toc_hide: true
+---
+Import JSON reports of kube-hunter scans. Use "kube-hunter --report json" to produce the report in json format.

dojo/tools/kubehunter/parser.py

@@ -0,0 +1,102 @@
+import hashlib
+import json
+from dojo.models import Finding
+
+
+class KubeHunterParser(object):
+    """
+    kube-hunter hunts for security weaknesses in Kubernetes clusters. The tool was developed to increase awareness and visibility for security issues in Kubernetes environments.
+    """
+
+    def get_scan_types(self):
+        return ["KubeHunter Scan"]
+
+    def get_label_for_scan_types(self, scan_type):
+        return "KubeHunter Scan"
+
+    def get_description_for_scan_types(self, scan_type):
+        return "KubeHunter JSON vulnerability report format.."
+
+    def get_findings(self, file, test):
+        data = json.load(file)
+
+        dupes = dict()
+
+        # Find any missing attribute
+        vulnerabilities = data['vulnerabilities']
+        check_required_attributes(vulnerabilities)
+
+        for item in vulnerabilities:
+            vulnerability_id = item.get('vid')
+            title = item['vulnerability']
+
+            # Finding details information
+            findingdetail = '**Hunter**: ' + item.get('hunter') + '\n\n'
+            findingdetail += '**Category**: ' + item.get('category') + '\n\n'
+            findingdetail += '**Location**: ' + item.get('location') + '\n\n'
+            findingdetail += '**Description**:\n' + item.get('description') + '\n\n'
+
+            # Finding severity
+            severity = item.get('severity', 'info')
+            allowed_severity = ['info', 'low', 'medium', 'high', "critical"]
+            if severity.lower() in allowed_severity:
+                severity = severity.capitalize()
+            else:
+                severity = 'Info'
+
+            # Finding mitigation and reference
+            avd_reference = item.get('avd_reference')
+
+            if avd_reference and avd_reference != '' and vulnerability_id != 'None':
+                mitigation = f"Further details can be found in kube-hunter documentation available at : {avd_reference}"
+                references = "**Kube-hunter AVD reference**: " + avd_reference
+            else:
+                mitigation = None
+                references = None
+
+            # Finding evidence
+            evidence = item.get('evidence')
+            if evidence and evidence != '' and evidence != 'none':
+                steps_to_reproduce = '**Evidence**: ' + item.get('evidence')
+            else:
+                steps_to_reproduce = None
+
+            finding = Finding(
+                title=title,
+                test=test,
+                description=findingdetail,
+                severity=severity,
+                mitigation=mitigation,
+                references=references,
+                static_finding=False,
+                dynamic_finding=True,
+                duplicate=False,
+                out_of_scope=False,
+                vuln_id_from_tool=vulnerability_id,
+                steps_to_reproduce=steps_to_reproduce
+            )
+
+            # internal de-duplication
+            if finding.steps_to_reproduce is None:
+                finding.steps_to_reproduce = ''
+            dupe_key = hashlib.sha256(str(finding.description + finding.title + finding.steps_to_reproduce + finding.vuln_id_from_tool).encode('utf-8')).hexdigest()
+
+            if dupe_key not in dupes:
+                dupes[dupe_key] = finding
+
+        return list(dupes.values())
+
+
+def check_required_attributes(vulnerabilities):
+    required_attributes = ["hunter", "category", "location", "description", "evidence", "avd_reference", "severity"]
+
+    missing_vulnerabilities = []
+
+    for idx, vulnerability in enumerate(vulnerabilities, start=1):
+        missing_attributes = [attr for attr in required_attributes if attr not in vulnerability]
+
+        if missing_attributes:
+            missing_vulnerabilities.append(f"Vulnerability {idx}: Missing attributes: {', '.join(missing_attributes)}")
+
+    if missing_vulnerabilities:
+        raise ValueError("\n`".join(missing_vulnerabilities))

unittests/scans/kubehunter/dupe.json

@@ -0,0 +1 @@
+{"nodes": [{"type": "Node/Master", "location": "10.1.1.1"}, {"type": "Node/Master", "location": "10.2.2.0"}], "services": [{"service": "Kubelet API (readonly)", "location": "10.0.1.1:10255"}, {"service": "Kubelet API", "location": "10.0.1.1:10250"}, {"service": "API Server", "location": "10.0.0.1:443"}], "vulnerabilities": [{"location": "10.0.1.1:10255", "vid": "KHV044", "category": "Privilege Escalation // Privileged container", "severity": "high", "vulnerability": "Privileged Container", "description": "A Privileged container exist on a node\n    could expose the node/cluster to unwanted root operations", "evidence": "pod: kube-proxy, container: kube-proxy, count: 1", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv044/", "hunter": "Kubelet Readonly Ports Hunter"},{"location": "10.0.1.1:10255", "vid": "KHV044", "category": "Privilege Escalation // Privileged container", "severity": "high", "vulnerability": "Privileged Container", "description": "A Privileged container exist on a node\n    could expose the node/cluster to unwanted root operations", "evidence": "pod: kube-proxy, container: kube-proxy, count: 1", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv044/", "hunter": "Kubelet Readonly Ports Hunter"}]}

unittests/scans/kubehunter/kubehunter_many_vul.json

@@ -0,0 +1 @@
+{"nodes": [{"type": "Node/Master", "location": "10.1.1.1"}, {"type": "Node/Master", "location": "10.2.2.0"}], "services": [{"service": "Kubelet API (readonly)", "location": "10.0.1.1:10255"}, {"service": "Kubelet API", "location": "10.0.1.1:10250"}, {"service": "API Server", "location": "10.0.0.1:443"}], "vulnerabilities": [{"location": "Local to Pod (kube-hunter-5pmjs)", "vid": "KHV050", "category": "Credential Access // Access container service account", "severity": "low", "vulnerability": "Read access to pod's service account token", "description": "Accessing the pod service account token gives an attacker the option to use the server API", "evidence": "TOKEN", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv050/", "hunter": "Access Secrets"}, {"location": "Local to Pod (kube-hunter-5pmjs)", "vid": "None", "category": "Lateral Movement // ARP poisoning and IP spoofing", "severity": "medium", "vulnerability": "CAP_NET_RAW Enabled", "description": "CAP_NET_RAW is enabled by default for pods.\n    If an attacker manages to compromise a pod,\n    they could potentially take advantage of this capability to perform network\n    attacks on other pods running on the same node", "evidence": "", "avd_reference": "https://avd.aquasec.com/kube-hunter/none/", "hunter": "Pod Capabilities Hunter"}, {"location": "Local to Pod (kube-hunter-5pmjs)", "vid": "None", "category": "Credential Access // Access container service account", "severity": "low", "vulnerability": "Access to pod's secrets", "description": "Accessing the pod's secrets within a compromised pod might disclose valuable data to a potential attacker", "evidence": "['/var/run/secrets/kubernetes.io/serviceaccount/namespace', '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt']", "avd_reference": "https://avd.aquasec.com/kube-hunter/none/", "hunter": "Access Secrets"}, {"location": "10.0.1.1:10255", "vid": "KHV044", "category": "Privilege Escalation // Privileged container", "severity": "high", "vulnerability": "Privileged Container", "description": "A Privileged container exist on a node\n    could expose the node/cluster to unwanted root operations", "evidence": "pod: kube-proxy, container: kube-proxy, count: 1", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv044/", "hunter": "Kubelet Readonly Ports Hunter"}, {"location": "10.0.1.1:10255", "vid": "KHV043", "category": "Initial Access // General Sensitive Information", "severity": "low", "vulnerability": "Cluster Health Disclosure", "description": "By accessing the open /healthz handler,\n    an attacker could get the cluster health state without authenticating", "evidence": "status: ok", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv043/", "hunter": "Kubelet Readonly Ports Hunter"}, {"location": "10.0.1.1:10255", "vid": "KHV052", "category": "Discovery // Access Kubelet API", "severity": "medium", "vulnerability": "Exposed Pods", "description": "An attacker could view sensitive information about pods that are\n    bound to a Node using the /pods endpoint", "evidence": "count: 7", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv052/", "hunter": "Kubelet Readonly Ports Hunter"}, {"location": "10.0.0.1:443", "vid": "KHV002", "category": "Initial Access // Exposed sensitive interfaces", "severity": "high", "vulnerability": "K8s Version Disclosure", "description": "The kubernetes version could be obtained from the /version endpoint", "evidence": "v1", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv002/", "hunter": "Api Version Hunter"}, {"location": "10.16.0.1:443", "vid": "KHV005", "category": "Discovery // Access the K8S API Server", "severity": "medium", "vulnerability": "Access to API using service account token", "description": "The API Server port is accessible.\n    Depending on your RBAC settings this could expose access to or control of your cluster.", "evidence": "b'{\"kind\":\"APIVersions\",\"versions\":[\"v1\"],\"serverAddressByClientCIDRs\":[{\"clientCIDR\":\"0.0.0.0/0\",\"serverAddress\":\"10.1.1.1:443\"}]}\\n'", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv005/", "hunter": "API Server Hunter"}]}

unittests/scans/kubehunter/kubehunter_one_vul.json

@@ -0,0 +1 @@
+{"nodes": [{"type": "Node/Master", "location": "10.1.1.1"}, {"type": "Node/Master", "location": "10.2.2.0"}], "services": [{"service": "Kubelet API (readonly)", "location": "10.0.1.1:10255"}, {"service": "Kubelet API", "location": "10.0.1.1:10250"}, {"service": "API Server", "location": "10.0.0.1:443"}], "vulnerabilities": [{"location": "10.0.1.1:10255", "vid": "KHV044", "category": "Privilege Escalation // Privileged container", "severity": "high", "vulnerability": "Privileged Container", "description": "A Privileged container exist on a node\n    could expose the node/cluster to unwanted root operations", "evidence": "pod: kube-proxy, container: kube-proxy, count: 1", "avd_reference": "https://avd.aquasec.com/kube-hunter/khv044/", "hunter": "Kubelet Readonly Ports Hunter"}]}

unittests/scans/kubehunter/kubehunter_zero_vul.json

@@ -0,0 +1 @@
+{"nodes": [{"type": "Node/Master", "location": "10.1.1.1"}, {"type": "Node/Master", "location": "10.2.2.0"}], "services": [{"service": "Kubelet API (readonly)", "location": "10.0.1.1:10255"}, {"service": "Kubelet API", "location": "10.0.1.1:10250"}, {"service": "API Server", "location": "10.0.0.1:443"}], "vulnerabilities": []}

unittests/tools/test_kubehunter_parser.py

@@ -0,0 +1,56 @@
+from django.test import TestCase
+from dojo.tools.kubehunter.parser import KubeHunterParser
+from dojo.models import Test
+
+
+class TestKubeHunterParser(TestCase):
+
+    def test_kubehunter_parser_with_no_vuln_has_no_findings(self):
+        testfile = open("unittests/scans/kubehunter/kubehunter_zero_vul.json")
+        parser = KubeHunterParser()
+        findings = parser.get_findings(testfile, Test())
+        testfile.close()
+        self.assertEqual(0, len(findings))
+
+    def test_kubehunter_parser_with_one_criticle_vuln_has_one_findings(self):
+        testfile = open("unittests/scans/kubehunter/kubehunter_one_vul.json")
+        parser = KubeHunterParser()
+        findings = parser.get_findings(testfile, Test())
+        testfile.close()
+        for finding in findings:
+            for endpoint in finding.unsaved_endpoints:
+                endpoint.clean()
+        self.assertEqual(1, len(findings))
+        self.assertEqual("KHV044", findings[0].vuln_id_from_tool)
+        self.assertEqual("Privileged Container", findings[0].title)
+        self.assertEqual(True, finding.active)
+
+        self.assertEqual(False, finding.duplicate)
+        self.assertEqual(finding.severity, 'High')
+
+    def test_kubehunter_parser_with_many_vuln_has_many_findings(self):
+        testfile = open("unittests/scans/kubehunter/kubehunter_many_vul.json")
+        parser = KubeHunterParser()
+        findings = parser.get_findings(testfile, Test())
+        testfile.close()
+
+        self.assertEqual(8, len(findings))
+
+    def test_kubehunter_parser_empty_with_error(self):
+        with self.assertRaises(ValueError) as context:
+            testfile = open("unittests/scans/kubehunter/empty.json")
+            parser = KubeHunterParser()
+            findings = parser.get_findings(testfile, Test())
+            testfile.close()
+
+            self.assertTrue(
+                "KubeHunter report contains errors:" in str(context.exception)
+            )
+            self.assertTrue("ECONNREFUSED" in str(context.exception))
+
+    def test_kubehunter_parser_dupe(self):
+        testfile = open("unittests/scans/kubehunter/dupe.json")
+        parser = KubeHunterParser()
+        findings = parser.get_findings(testfile, Test())
+        testfile.close()
+        self.assertEqual(1, len(findings))