Include pods of kubernetes_deployment in kubernetes_pod checks (4/4)

bridgecrewio · Oct 25, 2022 · a526b6c · a526b6c
1 parent d0d2ffa
commit a526b6c
Show file tree

Hide file tree

Showing 15 changed files with 4,221 additions and 217 deletions.
diff --git a/checkov/terraform/checks/resource/kubernetes/Secrets.py b/checkov/terraform/checks/resource/kubernetes/Secrets.py
@@ -9,7 +9,8 @@ def __init__(self):
         name = "Prefer using secrets as files over secrets as environment variables"
         id = "CKV_K8S_35"
 
-        supported_resources = ['kubernetes_pod', 'kubernetes_pod_v1']
+        supported_resources = ['kubernetes_pod', "kubernetes_pod_v1",
+                               'kubernetes_deployment', 'kubernetes_deployment_v1']
         categories = [CheckCategories.GENERAL_SECURITY]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
@@ -18,6 +19,14 @@ def scan_resource_conf(self, conf) -> CheckResult:
             self.evaluated_keys = [""]
             return CheckResult.FAILED
         spec = conf['spec'][0]
+        evaluated_keys_path = "spec"
+
+        if spec.get("template") and isinstance(spec.get("template"), list):
+            template = spec.get("template")[0]
+            if template.get("spec") and isinstance(template.get("spec"), list):
+                spec = template.get("spec")[0]
+                evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'
+
         containers = spec.get("container")
         if containers:
 
@@ -33,14 +42,14 @@ def scan_resource_conf(self, conf) -> CheckResult:
                                 value_from = env.get("value_from")[0]
                                 if value_from.get("secret_key_ref"):
                                     self.evaluated_keys = \
-                                        [f"spec/[0]/container/[{idx}]/env/[{idy}]/value_from/secret_key_ref"]
+                                        [f"{evaluated_keys_path}/[0]/container/[{idx}]/env/[{idy}]/value_from/secret_key_ref"]
                                     return CheckResult.FAILED
                 if container.get("env_from") and isinstance(container.get("env_from"), list):
                     env_from = container.get("env_from")[0]
                     for idy, ef in enumerate(env_from):
                         if "secret_ref" in ef:
                             self.evaluated_keys = \
-                                [f"spec/[0]/container/[{idx}]/env_from/[{idy}]/secret_ref"]
+                                [f"{evaluated_keys_path}/[0]/container/[{idx}]/env_from/[{idy}]/secret_ref"]
                             return CheckResult.FAILED
             return CheckResult.PASSED
 

diff --git a/checkov/terraform/checks/resource/kubernetes/ShareHostIPC.py b/checkov/terraform/checks/resource/kubernetes/ShareHostIPC.py
@@ -11,11 +11,14 @@ def __init__(self):
         # CIS-1.5 5.2.3
         name = "Do not admit containers wishing to share the host IPC namespace"
         id = "CKV_K8S_18"
-        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1"]
+        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1",
+                               "kubernetes_deployment", "kubernetes_deployment_v1"]
         categories = [CheckCategories.GENERAL_SECURITY]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
     def get_inspected_key(self):
+        if "kubernetes_deployment" == self.entity_type or "kubernetes_deployment_v1" == self.entity_type:
+            return "spec/[0]/template/[0]/spec/[0]/host_ipc"
         return "spec/[0]/host_ipc"
 
     def get_forbidden_values(self) -> List[Any]:

diff --git a/checkov/terraform/checks/resource/kubernetes/ShareHostPID.py b/checkov/terraform/checks/resource/kubernetes/ShareHostPID.py
@@ -11,12 +11,15 @@ def __init__(self):
         # CIS-1.5 5.2.2
         name = "Do not admit containers wishing to share the host process ID namespace"
         id = "CKV_K8S_17"
-        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1"]
+        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1",
+                               "kubernetes_deployment", "kubernetes_deployment_v1"]
         categories = [CheckCategories.GENERAL_SECURITY]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources,
                          missing_block_result=CheckResult.PASSED)
 
     def get_inspected_key(self) -> str:
+        if "kubernetes_deployment" == self.entity_type or "kubernetes_deployment_v1" == self.entity_type:
+            return "spec/[0]/template/[0]/spec/[0]/host_pid"
         return "spec/[0]/host_pid"
 
     def get_expected_value(self) -> Any:

diff --git a/checkov/terraform/checks/resource/kubernetes/SharedHostNetworkNamespace.py b/checkov/terraform/checks/resource/kubernetes/SharedHostNetworkNamespace.py
@@ -11,12 +11,15 @@ def __init__(self):
         # CIS-1.5 5.2.4
         name = "Do not admit containers wishing to share the host network namespace"
         id = "CKV_K8S_19"
-        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1"]
+        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1",
+                               "kubernetes_deployment", "kubernetes_deployment_v1"]
         categories = [CheckCategories.NETWORKING]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources,
                          missing_block_result=CheckResult.PASSED)
 
     def get_inspected_key(self) -> str:
+        if "kubernetes_deployment" == self.entity_type or "kubernetes_deployment_v1" == self.entity_type:
+            return "spec/[0]/template/[0]/spec/[0]/host_network"
         return "spec/[0]/host_network"
 
     def get_expected_value(self) -> Any:

diff --git a/checkov/terraform/checks/resource/kubernetes/Tiller.py b/checkov/terraform/checks/resource/kubernetes/Tiller.py
@@ -10,7 +10,8 @@ class Tiller(BaseResourceCheck):
     def __init__(self) -> None:
         name = "Ensure that Tiller (Helm v2) is not deployed"
         id = "CKV_K8S_34"
-        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1"]
+        supported_resources = ["kubernetes_pod", "kubernetes_pod_v1",
+                               "kubernetes_deployment", "kubernetes_deployment_v1"]
         categories = [CheckCategories.GENERAL_SECURITY]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
@@ -32,6 +33,26 @@ def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
             self.evaluated_keys = [""]
             return CheckResult.FAILED
         spec = conf['spec'][0]
+        evaluated_keys_path = "spec"
+
+        if spec.get("template") and isinstance(spec.get("template"), list):
+            template = spec.get("template")[0]
+            if template.get("metadata") and isinstance(template.get("metadata"), list):
+                metadata = template.get("metadata")[0]
+
+                if metadata.get("labels") and isinstance(metadata.get("labels"), list) and isinstance(metadata.get("labels")[0], dict):
+                    labels = metadata.get("labels")[0]
+                    self.evaluated_keys = [f"{evaluated_keys_path}/[0]/template/[0]/metadata/[0]/labels"]
+                    if labels.get("app") == "helm":
+                        self.evaluated_keys = [f"{evaluated_keys_path}/[0]/template/[0]/metadata/[0]/labels/[0]/app"]
+                        return CheckResult.FAILED
+                    elif labels.get("name") == "tiller":
+                        self.evaluated_keys = [f"{evaluated_keys_path}/[0]/template/[0]/metadata/[0]/labels/[0]/name"]
+                        return CheckResult.FAILED
+
+            if template.get("spec") and isinstance(template.get("spec"), list):
+                spec = template.get("spec")[0]
+                evaluated_keys_path = f'{evaluated_keys_path}/[0]/template/[0]/spec'
 
         containers = spec.get("container")
         if not containers:
@@ -42,7 +63,7 @@ def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
             if container.get("image") and isinstance(container.get("image"), list):
                 image = container.get("image")[0]
                 if "tiller" in image:
-                    self.evaluated_keys = [f'spec/[0]/container/[{idx}]/image']
+                    self.evaluated_keys = [f'{evaluated_keys_path}/[0]/container/[{idx}]/image']
                     return CheckResult.FAILED
 
         return CheckResult.PASSED