feat(kubernetes): add etcd, controllermanager and rbac services (#3261)

Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
2026-07-24 04:51:51 +00:00 · 2024-02-19 13:19:07 +01:00
parent 36fc575e40
commit 3cc367e0a3
21 changed files with 321 additions and 10 deletions
@@ -9,9 +9,9 @@ class APIServer(KubernetesService):
        super().__init__(audit_info)
        self.client = core_client

-        self.apiserver_pods = self.__get_apiserver_pod__()
+        self.apiserver_pods = self.__get_apiserver_pods__()

-    def __get_apiserver_pod__(self):
+    def __get_apiserver_pods__(self):
        try:
            apiserver_pods = []
            for pod in self.client.pods.values():
@@ -0,0 +1,6 @@
+from prowler.providers.common.common import global_provider
+from prowler.providers.kubernetes.services.controllermanager.controllermanager_service import (
+    ControllerManager,
+)
+
+controllermanager_client = ControllerManager(global_provider)
@@ -0,0 +1,36 @@
+{
+  "Provider": "kubernetes",
+  "CheckID": "controllermanager_garbage_collection",
+  "CheckTitle": "Ensure that the --terminated-pod-gc-threshold argument is set as appropriate",
+  "CheckType": [
+    "Resource Management",
+    "Performance Optimization"
+  ],
+  "ServiceName": "kube-controller-manager",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "KubernetesControllerManager",
+  "Description": "Activate garbage collector on pod termination, as appropriate. Garbage collection is crucial for maintaining resource availability and performance. The default threshold for garbage collection is 12,500 terminated pods, which may be too high for some systems. Adjusting this threshold based on system resources and performance tests is recommended.",
+  "Risk": "A high threshold for garbage collection can lead to degraded performance and resource exhaustion. In extreme cases, it might cause system crashes or prolonged unavailability.",
+  "RelatedUrl": "https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-garbage-collection",
+  "Remediation": {
+    "Code": {
+      "CLI": "Edit the Controller Manager pod specification file /etc/kubernetes/manifests/kube-controller-manager.yaml and set --terminated-pod-gc-threshold to an appropriate value, e.g., --terminated-pod-gc-threshold=10",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Review and adjust the --terminated-pod-gc-threshold argument in the kube-controller-manager to ensure efficient garbage collection and optimal resource utilization.",
+      "Url": "https://kubernetes.io/docs/reference/command-line-tools-reference/kube-controller-manager/"
+    }
+  },
+  "Categories": [
+    "Cluster Stability",
+    "Operational Best Practices"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "The default value of --terminated-pod-gc-threshold is 12500. Adjust according to your specific cluster workload and performance requirements."
+}
@@ -0,0 +1,22 @@
+from prowler.lib.check.models import Check, Check_Report_Kubernetes
+from prowler.providers.kubernetes.services.controllermanager.controllermanager_client import (
+    controllermanager_client,
+)
+
+
+class controllermanager_garbage_collection(Check):
+    def execute(self) -> Check_Report_Kubernetes:
+        findings = []
+        for pod in controllermanager_client.controllermanager_pods:
+            report = Check_Report_Kubernetes(self.metadata())
+            report.namespace = pod.namespace
+            report.resource_name = pod.name
+            report.resource_id = pod.uid
+            report.status = "PASS"
+            report.status_extended = f"Controller Manager has an appropriate garbage collection threshold in pod {pod.name}."
+            for container in pod.containers.values():
+                if "--terminated-pod-gc-threshold=12500" in str(container.command):
+                    report.status = "FAIL"
+                    report.status_extended = f"Controller Manager has the default garbage collection threshold in pod {pod.name}."
+            findings.append(report)
+        return findings
@@ -0,0 +1,26 @@
+from prowler.lib.logger import logger
+from prowler.providers.kubernetes.lib.service.service import KubernetesService
+from prowler.providers.kubernetes.services.core.core_client import core_client
+
+
+################## ControllerManager ##################
+class ControllerManager(KubernetesService):
+    def __init__(self, audit_info):
+        super().__init__(audit_info)
+        self.client = core_client
+
+        self.controllermanager_pods = self.__get_controllermanager_pods__()
+
+    def __get_controllermanager_pods__(self):
+        try:
+            controllermanager_pods = []
+            for pod in self.client.pods.values():
+                if pod.namespace == "kube-system" and pod.name.startswith(
+                    "kube-controller-manager"
+                ):
+                    controllermanager_pods.append(pod)
+            return controllermanager_pods
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
@@ -0,0 +1,4 @@
+from prowler.providers.common.common import global_provider
+from prowler.providers.kubernetes.services.etcd.etcd_service import Etcd
+
+etcd_client = Etcd(global_provider)
@@ -0,0 +1,24 @@
+from prowler.lib.logger import logger
+from prowler.providers.kubernetes.lib.service.service import KubernetesService
+from prowler.providers.kubernetes.services.core.core_client import core_client
+
+
+################## Etcd ##################
+class Etcd(KubernetesService):
+    def __init__(self, audit_info):
+        super().__init__(audit_info)
+        self.client = core_client
+
+        self.etcd_pods = self.__get_etcd_pods__()
+
+    def __get_etcd_pods__(self):
+        try:
+            etcd_pods = []
+            for pod in self.client.pods.values():
+                if pod.namespace == "kube-system" and pod.name.startswith("etcd"):
+                    etcd_pods.append(pod)
+            return etcd_pods
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
@@ -0,0 +1,36 @@
+{
+  "Provider": "kubernetes",
+  "CheckID": "etcd_tls_encryption",
+  "CheckTitle": "Ensure that the --cert-file and --key-file arguments are set as appropriate for etcd",
+  "CheckType": [
+    "Security",
+    "Configuration"
+  ],
+  "ServiceName": "etcd",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "Etcd",
+  "Description": "This check verifies that the etcd service in a Kubernetes cluster is configured with appropriate TLS encryption settings. etcd, being a key value store for all Kubernetes REST API objects, should have its communication encrypted to protect these sensitive objects in transit.",
+  "Risk": "Without proper TLS configuration, data stored in etcd can be susceptible to interception and unauthorized access, posing a significant security risk to the entire Kubernetes cluster.",
+  "RelatedUrl": "https://kubernetes.io/docs/tasks/administer-cluster/configure-upgrade-etcd/#limiting-access-of-etcd-clusters",
+  "Remediation": {
+    "Code": {
+      "CLI": "Edit the etcd pod specification file /etc/kubernetes/manifests/etcd.yaml on the master node and set the --cert-file and --key-file arguments appropriately.",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that the etcd service is configured with TLS encryption for secure communication. The --cert-file and --key-file arguments should point to a valid TLS certificate and key.",
+      "Url": "https://kubernetes.io/docs/tasks/administer-cluster/configure-upgrade-etcd/"
+    }
+  },
+  "Categories": [
+    "Cluster Security",
+    "Data Protection"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "By default, etcd may not be configured with TLS encryption. It is crucial to enable TLS to protect the sensitive data handled by etcd."
+}
@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_Kubernetes
+from prowler.providers.kubernetes.services.etcd.etcd_client import etcd_client
+
+
+class etcd_tls_encryption(Check):
+    def execute(self) -> Check_Report_Kubernetes:
+        findings = []
+        for pod in etcd_client.etcd_pods:
+            report = Check_Report_Kubernetes(self.metadata())
+            report.namespace = pod.namespace
+            report.resource_name = pod.name
+            report.resource_id = pod.uid
+            report.status = "FAIL"
+            report.status_extended = (
+                f"Etcd does not have TLS encryption configured in pod {pod.name}."
+            )
+            for container in pod.containers.values():
+                if "--cert-file" in str(container.command) and "--key-file" in str(
+                    container.command
+                ):
+
+                    report.status = "PASS"
+                    report.status_extended = (
+                        f"Etcd has configured TLS encryption in pod {pod.name}."
+                    )
+            findings.append(report)
+        return findings
@@ -0,0 +1,4 @@
+from prowler.providers.common.common import global_provider
+from prowler.providers.kubernetes.services.rbac.rbac_service import Rbac
+
+rbac_client = Rbac(global_provider)
@@ -0,0 +1,36 @@
+{
+  "Provider": "kubernetes",
+  "CheckID": "rbac_cluster_admin_usage",
+  "CheckTitle": "Ensure that the cluster-admin role is only used where required",
+  "CheckType": [
+    "Security",
+    "Compliance"
+  ],
+  "ServiceName": "RBAC",
+  "SubServiceName": "Authorization",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "ClusterRoleBinding",
+  "Description": "This check ensures that the 'cluster-admin' role, which provides wide-ranging powers, is used only where necessary. The 'cluster-admin' role grants super-user access to perform any action on any resource, including all namespaces. It should be applied cautiously to avoid excessive privileges.",
+  "Risk": "Inappropriate use of the 'cluster-admin' role can lead to excessive privileges, increasing the risk of malicious actions and potentially impacting the cluster's security posture.",
+  "RelatedUrl": "https://kubernetes.io/docs/reference/access-authn-authz/rbac/#user-facing-roles",
+  "Remediation": {
+    "Code": {
+      "CLI": "Review and, if necessary, modify the ClusterRoleBindings to limit the use of 'cluster-admin'. Use 'kubectl delete clusterrolebinding [name]' to remove unnecessary bindings.",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Audit and assess the use of 'cluster-admin' role in all ClusterRoleBindings. Ensure it is assigned only to subjects that require such extensive privileges. Consider using more restrictive roles wherever possible.",
+      "Url": "https://kubernetes.io/docs/reference/access-authn-authz/rbac/#clusterrolebinding-example"
+    }
+  },
+  "Categories": [
+    "Access Control",
+    "Least Privilege Principle"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Modifying ClusterRoleBindings should be done with caution to avoid unintended access issues. Always ensure that critical system components have the necessary permissions to operate effectively."
+}
@@ -0,0 +1,23 @@
+from prowler.lib.check.models import Check, Check_Report_Kubernetes
+from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
+
+
+class rbac_cluster_admin_usage(Check):
+    def execute(self) -> Check_Report_Kubernetes:
+        findings = []
+        # Iterate through the bindings
+        for binding in rbac_client.cluster_role_bindings:
+            # Check if the binding refers to the cluster-admin role
+            if binding.roleRef.name == "cluster-admin":
+                report = Check_Report_Kubernetes(self.metadata())
+                report.namespace = (
+                    "kube-system"
+                    if not binding.metadata.namespace
+                    else binding.metadata.namespace
+                )
+                report.resource_name = binding.metadata.name
+                report.resource_id = binding.metadata.uid
+                report.status = "MANUAL"
+                report.status_extended = f"Cluster Role Binding {binding.metadata.name} uses cluster-admin role."
+                findings.append(report)
+        return findings
@@ -0,0 +1,64 @@
+from typing import Any, List, Optional
+
+from kubernetes import client
+from pydantic import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.providers.kubernetes.lib.service.service import KubernetesService
+
+
+################## Rbac ##################
+class Rbac(KubernetesService):
+    def __init__(self, audit_info):
+        super().__init__(audit_info)
+        self.client = client.RbacAuthorizationV1Api()
+
+        self.cluster_role_bindings = self.__list_cluster_role_binding__()
+
+    def __list_cluster_role_binding__(self):
+        try:
+            bindings_list = []
+            for binding in self.client.list_cluster_role_binding().items:
+                # For each binding, create a ClusterRoleBinding object and append it to the list
+                formatted_binding = {
+                    "metadata": binding.metadata,
+                    "subjects": []
+                    if not binding.subjects
+                    else [
+                        {
+                            "kind": subject.kind,
+                            "name": subject.name,
+                            "namespace": getattr(subject, "namespace", ""),
+                        }
+                        for subject in binding.subjects
+                    ],
+                    "roleRef": {
+                        "kind": binding.role_ref.kind,
+                        "name": binding.role_ref.name,
+                        "apiGroup": binding.role_ref.api_group,
+                    },
+                }
+                bindings_list.append(ClusterRoleBinding(**formatted_binding))
+            return bindings_list
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class Subject(BaseModel):
+    kind: str
+    name: str
+    namespace: Optional[str] = None
+
+
+class RoleRef(BaseModel):
+    kind: str
+    name: str
+    apiGroup: str
+
+
+class ClusterRoleBinding(BaseModel):
+    metadata: Any
+    subjects: List[Subject]
+    roleRef: RoleRef
@@ -12,12 +12,15 @@ class scheduler_profiling(Check):
            report.namespace = pod.namespace
            report.resource_name = pod.name
            report.resource_id = pod.uid
-            report.status = "PASS"
-            report.status_extended = "Scheduler does not have profiling enabled."
+            report.status = "FAIL"
+            report.status_extended = (
+                f"Scheduler has profiling enabled in pod {pod.name}."
+            )
            for container in pod.containers.values():
-                if "--profiling=true" in container.command:
-                    report.resource_id = container.name
-                    report.status = "FAIL"
-                    report.status_extended = f"Scheduler has profiling enabled in container {container.name}."
+                if "--profiling=false" in str(container.command):
+                    report.status = "PASS"
+                    report.status_extended = (
+                        f"Scheduler does not have profiling enabled in pod {pod.name}."
+                    )
            findings.append(report)
        return findings
@@ -9,9 +9,9 @@ class Scheduler(KubernetesService):
        super().__init__(audit_info)
        self.client = core_client

-        self.scheduler_pods = self.__get_scheduler_pod__()
+        self.scheduler_pods = self.__get_scheduler_pods__()

-    def __get_scheduler_pod__(self):
+    def __get_scheduler_pods__(self):
        try:
            scheduler_pods = []
            for pod in self.client.pods.values():