diff --git a/prowler/providers/kubernetes/services/rbac/rbac_minimize_csr_approval_access/rbac_minimize_csr_approval_access.py b/prowler/providers/kubernetes/services/rbac/rbac_minimize_csr_approval_access/rbac_minimize_csr_approval_access.py index f2527b4606..9a554d65cc 100644 --- a/prowler/providers/kubernetes/services/rbac/rbac_minimize_csr_approval_access/rbac_minimize_csr_approval_access.py +++ b/prowler/providers/kubernetes/services/rbac/rbac_minimize_csr_approval_access/rbac_minimize_csr_approval_access.py @@ -11,24 +11,26 @@ resources = ["certificatesigningrequests/approval"] class rbac_minimize_csr_approval_access(Check): def execute(self) -> Check_Report_Kubernetes: findings = [] + # Collect unique subjects and the ClusterRole names bound to them + subjects_bound_roles = {} for crb in rbac_client.cluster_role_bindings.values(): for subject in crb.subjects: if subject.kind in ["User", "Group"]: - report = Check_Report_Kubernetes( - metadata=self.metadata(), resource=subject - ) - report.status = "PASS" - report.status_extended = f"User or group '{subject.name}' does not have access to update the CSR approval sub-resource." - for cr in rbac_client.cluster_roles.values(): - if cr.metadata.name == crb.roleRef.name: - if is_rule_allowing_permissions( - cr.rules, - resources, - verbs, - ): - report.status = "FAIL" - report.status_extended = f"User or group '{subject.name}' has access to update the CSR approval sub-resource." - break - findings.append(report) + key = (subject.kind, subject.name, subject.namespace) + if key not in subjects_bound_roles: + subjects_bound_roles[key] = (subject, set()) + subjects_bound_roles[key][1].add(crb.roleRef.name) + + for (_kind, _name, _ns), (subject, role_names) in subjects_bound_roles.items(): + report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) + report.status = "PASS" + report.status_extended = f"User or group '{subject.name}' does not have access to update the CSR approval sub-resource." + for cr in rbac_client.cluster_roles.values(): + if cr.metadata.name in role_names: + if is_rule_allowing_permissions(cr.rules, resources, verbs): + report.status = "FAIL" + report.status_extended = f"User or group '{subject.name}' has access to update the CSR approval sub-resource." + break + findings.append(report) return findings diff --git a/prowler/providers/kubernetes/services/rbac/rbac_minimize_node_proxy_subresource_access/rbac_minimize_node_proxy_subresource_access.py b/prowler/providers/kubernetes/services/rbac/rbac_minimize_node_proxy_subresource_access/rbac_minimize_node_proxy_subresource_access.py index 913d968f31..3ee59070e3 100644 --- a/prowler/providers/kubernetes/services/rbac/rbac_minimize_node_proxy_subresource_access/rbac_minimize_node_proxy_subresource_access.py +++ b/prowler/providers/kubernetes/services/rbac/rbac_minimize_node_proxy_subresource_access/rbac_minimize_node_proxy_subresource_access.py @@ -11,20 +11,26 @@ resources = ["nodes/proxy"] class rbac_minimize_node_proxy_subresource_access(Check): def execute(self) -> Check_Report_Kubernetes: findings = [] + # Collect unique subjects and the ClusterRole names bound to them + subjects_bound_roles = {} for crb in rbac_client.cluster_role_bindings.values(): for subject in crb.subjects: if subject.kind in ["User", "Group"]: - report = Check_Report_Kubernetes( - metadata=self.metadata(), resource=subject - ) - report.status = "PASS" - report.status_extended = f"User or group '{subject.name}' does not have access to the node proxy sub-resource." - for cr in rbac_client.cluster_roles.values(): - if cr.metadata.name == crb.roleRef.name: - if is_rule_allowing_permissions(cr.rules, resources, verbs): - report.status = "FAIL" - report.status_extended = f"User or group '{subject.name}' has access to the node proxy sub-resource." - break - findings.append(report) + key = (subject.kind, subject.name, subject.namespace) + if key not in subjects_bound_roles: + subjects_bound_roles[key] = (subject, set()) + subjects_bound_roles[key][1].add(crb.roleRef.name) + + for (_kind, _name, _ns), (subject, role_names) in subjects_bound_roles.items(): + report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) + report.status = "PASS" + report.status_extended = f"User or group '{subject.name}' does not have access to the node proxy sub-resource." + for cr in rbac_client.cluster_roles.values(): + if cr.metadata.name in role_names: + if is_rule_allowing_permissions(cr.rules, resources, verbs): + report.status = "FAIL" + report.status_extended = f"User or group '{subject.name}' has access to the node proxy sub-resource." + break + findings.append(report) return findings diff --git a/prowler/providers/kubernetes/services/rbac/rbac_minimize_pv_creation_access/rbac_minimize_pv_creation_access.py b/prowler/providers/kubernetes/services/rbac/rbac_minimize_pv_creation_access/rbac_minimize_pv_creation_access.py index 204942c57e..4b675a5c42 100644 --- a/prowler/providers/kubernetes/services/rbac/rbac_minimize_pv_creation_access/rbac_minimize_pv_creation_access.py +++ b/prowler/providers/kubernetes/services/rbac/rbac_minimize_pv_creation_access/rbac_minimize_pv_creation_access.py @@ -11,21 +11,26 @@ resources = ["persistentvolumes"] class rbac_minimize_pv_creation_access(Check): def execute(self) -> Check_Report_Kubernetes: findings = [] - # Check each ClusterRoleBinding for access to create PersistentVolumes + # Collect unique subjects and the ClusterRole names bound to them + subjects_bound_roles = {} for crb in rbac_client.cluster_role_bindings.values(): for subject in crb.subjects: if subject.kind in ["User", "Group"]: - report = Check_Report_Kubernetes( - metadata=self.metadata(), resource=subject - ) - report.status = "PASS" - report.status_extended = f"User or group '{subject.name}' does not have access to create PersistentVolumes." - for cr in rbac_client.cluster_roles.values(): - if cr.metadata.name == crb.roleRef.name: - if is_rule_allowing_permissions(cr.rules, resources, verbs): - report.status = "FAIL" - report.status_extended = f"User or group '{subject.name}' has access to create PersistentVolumes." - break - findings.append(report) + key = (subject.kind, subject.name, subject.namespace) + if key not in subjects_bound_roles: + subjects_bound_roles[key] = (subject, set()) + subjects_bound_roles[key][1].add(crb.roleRef.name) + + for (_kind, _name, _ns), (subject, role_names) in subjects_bound_roles.items(): + report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) + report.status = "PASS" + report.status_extended = f"User or group '{subject.name}' does not have access to create PersistentVolumes." + for cr in rbac_client.cluster_roles.values(): + if cr.metadata.name in role_names: + if is_rule_allowing_permissions(cr.rules, resources, verbs): + report.status = "FAIL" + report.status_extended = f"User or group '{subject.name}' has access to create PersistentVolumes." + break + findings.append(report) return findings diff --git a/prowler/providers/kubernetes/services/rbac/rbac_minimize_service_account_token_creation/rbac_minimize_service_account_token_creation.py b/prowler/providers/kubernetes/services/rbac/rbac_minimize_service_account_token_creation/rbac_minimize_service_account_token_creation.py index 9b1318c92f..59d88cdc14 100644 --- a/prowler/providers/kubernetes/services/rbac/rbac_minimize_service_account_token_creation/rbac_minimize_service_account_token_creation.py +++ b/prowler/providers/kubernetes/services/rbac/rbac_minimize_service_account_token_creation/rbac_minimize_service_account_token_creation.py @@ -11,20 +11,26 @@ resources = ["serviceaccounts/token"] class rbac_minimize_service_account_token_creation(Check): def execute(self) -> Check_Report_Kubernetes: findings = [] + # Collect unique subjects and the ClusterRole names bound to them + subjects_bound_roles = {} for crb in rbac_client.cluster_role_bindings.values(): for subject in crb.subjects: if subject.kind in ["User", "Group"]: - report = Check_Report_Kubernetes( - metadata=self.metadata(), resource=subject - ) - report.status = "PASS" - report.status_extended = f"User or group '{subject.name}' does not have access to create service account tokens." - for cr in rbac_client.cluster_roles.values(): - if cr.metadata.name == crb.roleRef.name: - if is_rule_allowing_permissions(cr.rules, resources, verbs): - report.status = "FAIL" - report.status_extended = f"User or group '{subject.name}' has access to create service account tokens." - break - findings.append(report) + key = (subject.kind, subject.name, subject.namespace) + if key not in subjects_bound_roles: + subjects_bound_roles[key] = (subject, set()) + subjects_bound_roles[key][1].add(crb.roleRef.name) + + for (_kind, _name, _ns), (subject, role_names) in subjects_bound_roles.items(): + report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) + report.status = "PASS" + report.status_extended = f"User or group '{subject.name}' does not have access to create service account tokens." + for cr in rbac_client.cluster_roles.values(): + if cr.metadata.name in role_names: + if is_rule_allowing_permissions(cr.rules, resources, verbs): + report.status = "FAIL" + report.status_extended = f"User or group '{subject.name}' has access to create service account tokens." + break + findings.append(report) return findings diff --git a/prowler/providers/kubernetes/services/rbac/rbac_minimize_webhook_config_access/rbac_minimize_webhook_config_access.py b/prowler/providers/kubernetes/services/rbac/rbac_minimize_webhook_config_access/rbac_minimize_webhook_config_access.py index 2da9893dab..1b680307a8 100644 --- a/prowler/providers/kubernetes/services/rbac/rbac_minimize_webhook_config_access/rbac_minimize_webhook_config_access.py +++ b/prowler/providers/kubernetes/services/rbac/rbac_minimize_webhook_config_access/rbac_minimize_webhook_config_access.py @@ -14,24 +14,26 @@ verbs = ["create", "update", "delete"] class rbac_minimize_webhook_config_access(Check): def execute(self) -> Check_Report_Kubernetes: findings = [] + # Collect unique subjects and the ClusterRole names bound to them + subjects_bound_roles = {} for crb in rbac_client.cluster_role_bindings.values(): for subject in crb.subjects: if subject.kind in ["User", "Group"]: - report = Check_Report_Kubernetes( - metadata=self.metadata(), resource=subject - ) - report.status = "PASS" - report.status_extended = f"User or group '{subject.name}' does not have access to create, update, or delete webhook configurations." - for cr in rbac_client.cluster_roles.values(): - if cr.metadata.name == crb.roleRef.name: - if is_rule_allowing_permissions( - cr.rules, - resources, - verbs, - ): - report.status = "FAIL" - report.status_extended = f"User or group '{subject.name}' has access to create, update, or delete webhook configurations." - break - findings.append(report) + key = (subject.kind, subject.name, subject.namespace) + if key not in subjects_bound_roles: + subjects_bound_roles[key] = (subject, set()) + subjects_bound_roles[key][1].add(crb.roleRef.name) + + for (_kind, _name, _ns), (subject, role_names) in subjects_bound_roles.items(): + report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) + report.status = "PASS" + report.status_extended = f"User or group '{subject.name}' does not have access to create, update, or delete webhook configurations." + for cr in rbac_client.cluster_roles.values(): + if cr.metadata.name in role_names: + if is_rule_allowing_permissions(cr.rules, resources, verbs): + report.status = "FAIL" + report.status_extended = f"User or group '{subject.name}' has access to create, update, or delete webhook configurations." + break + findings.append(report) return findings