fix(kubernetes): Validate k8s spec type

checkov/kubernetes/checks/resource/base_root_container_check.py

@@ -44,6 +44,8 @@ def extract_spec(self, conf: Dict[str, Any]) -> Dict:
 
     @staticmethod
     def check_runAsNonRoot(spec):
+        if not isinstance(spec, dict):
+            return "ABSENT"
         security_context = spec.get("securityContext")
         if security_context and isinstance(security_context, dict) and "runAsNonRoot" in security_context:
             if security_context["runAsNonRoot"]:
@@ -54,7 +56,7 @@ def check_runAsNonRoot(spec):
 
     @staticmethod
     def check_runAsUser(spec: Dict[str, Any], uid: int) -> str:
-        if spec.get("securityContext") and isinstance(spec.get("securityContext"), dict) and "runAsUser" in spec["securityContext"]:
+        if isinstance(spec, dict) and spec.get("securityContext") and isinstance(spec.get("securityContext"), dict) and "runAsUser" in spec["securityContext"]:
             if isinstance(spec["securityContext"]["runAsUser"], int) and spec["securityContext"]["runAsUser"] >= uid:
                 return "PASSED"
             else:

checkov/kubernetes/checks/resource/k8s/DockerSocketVolume.py

@@ -33,7 +33,7 @@ def scan_spec_conf(self, conf):
             spec = inner_spec if inner_spec else spec
 
         # Evaluate volumes
-        if spec:
+        if spec and isinstance(spec, dict):
             volumes = spec.get("volumes", [])
             if not isinstance(volumes, list):
                 return CheckResult.UNKNOWN

checkov/kubernetes/checks/resource/k8s/PodSecurityContext.py

@@ -34,7 +34,7 @@ def scan_spec_conf(self, conf):
             inner_spec = self.get_inner_entry(conf, "spec")
             spec = inner_spec if inner_spec else spec
 
-        if spec:
+        if spec and isinstance(spec, dict):
             if spec.get("securityContext"):
                 if spec["securityContext"]:
                     return CheckResult.PASSED

checkov/kubernetes/checks/resource/k8s/RootContainers.py

@@ -18,7 +18,7 @@ def scan_spec_conf(self, conf):
         spec = self.extract_spec(conf)
 
         # Collect results
-        if spec:
+        if spec and isinstance(spec, dict):
             results = {"pod": {}, "container": []}
             results["pod"]["runAsNonRoot"] = self.check_runAsNonRoot(spec)
             results["pod"]["runAsUser"] = self.check_runAsUser(spec, 1)

checkov/kubernetes/checks/resource/k8s/RootContainersHighUID.py

@@ -20,7 +20,7 @@ def scan_spec_conf(self, conf: dict[str, Any]) -> CheckResult:
         spec = self.extract_spec(conf)
 
         # Collect results
-        if spec:
+        if spec and isinstance(spec, dict):
             results = {"pod": {}, "container": []}
             results["pod"]["runAsUser"] = self.check_runAsUser(spec, 10000)
 

tests/kubernetes/checks/example_RootContainers/rootContainersFAILED-malformed-spec.yaml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: pod1
+spec:
+  - containers:
+  - name: main
+    image: alpine
+    command: ["/bin/sleep", "999999"]

tests/kubernetes/checks/example_SecurityContexts/pod-container-FAILED_malformed_spec.yaml

@@ -0,0 +1,26 @@
+# Source: https://kubernetes.io/docs/tasks/configure-pod-container/security-context/
+apiVersion: v1
+kind: Pod
+metadata:
+  name: security-context-demo
+spec:
+  - securityContext:
+    runAsUser: 1000
+    runAsGroup: 3000
+    fsGroup: 2000
+  - volumes:
+  - name: sec-ctx-vol
+    emptyDir: {}
+  - containers:
+  - name: sec-ctx-demo
+    image: busybox
+    command: [ "sh", "-c", "sleep 1h" ]
+    volumeMounts:
+    - name: sec-ctx-vol
+      mountPath: /data/demo
+    securityContext:
+      allowPrivilegeEscalation: false
+# https://kubernetes.io/docs/tasks/configure-pod-container/security-context/
+# cat /proc/1/status
+# allowPrivilegeEscalation: false means NoNewPrivs = 1
+# https://www.kernel.org/doc/html/latest/userspace-api/no_new_privs.html

tests/kubernetes/checks/test_ContainerSecurityContext.py

@@ -16,7 +16,7 @@ def test_summary(self):
         report = runner.run(root_folder=test_files_dir,runner_filter=RunnerFilter(checks=[check.id]))
         summary = report.get_summary()
 
-        self.assertEqual(summary['passed'], 2)
+        self.assertEqual(summary['passed'], 3)
         self.assertEqual(summary['failed'], 2)
         self.assertEqual(summary['skipped'], 0)
         self.assertEqual(summary['parsing_errors'], 0)

tests/kubernetes/checks/test_PodSecurityContext.py

@@ -17,7 +17,7 @@ def test_summary(self):
         summary = report.get_summary()
 
         self.assertEqual(summary['passed'], 2)
-        self.assertEqual(summary['failed'], 2)
+        self.assertEqual(summary['failed'], 3)
         self.assertEqual(summary['skipped'], 0)
         self.assertEqual(summary['parsing_errors'], 0)
 

tests/kubernetes/checks/test_RootContainers.py

@@ -17,7 +17,7 @@ def test_summary(self):
         summary = report.get_summary()
 
         self.assertEqual(summary['passed'], 5)
-        self.assertEqual(summary['failed'], 6)
+        self.assertEqual(summary['failed'], 7)
         self.assertEqual(summary['skipped'], 0)
         self.assertEqual(summary['parsing_errors'], 0)