From 4910fb358227db9e557819cb576f9accb62d3913 Mon Sep 17 00:00:00 2001 From: Pepe Fagoaga Date: Fri, 1 May 2026 10:37:20 +0200 Subject: [PATCH] perf(k8s): index ClusterRoles by name in RBAC checks --- .../rbac_minimize_csr_approval_access.py | 15 +++++++++------ ...rbac_minimize_node_proxy_subresource_access.py | 15 +++++++++------ .../rbac_minimize_pv_creation_access.py | 15 +++++++++------ ...bac_minimize_service_account_token_creation.py | 15 +++++++++------ .../rbac_minimize_webhook_config_access.py | 15 +++++++++------ 5 files changed, 45 insertions(+), 30 deletions(-) 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 2740b65ab0..17e56add55 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 @@ -22,18 +22,21 @@ class rbac_minimize_csr_approval_access(Check): subjects_bound_roles[key] = (subject, set()) subjects_bound_roles[key][1].add(crb.roleRef.name) + cluster_roles_by_name = { + cr.metadata.name: cr for cr in rbac_client.cluster_roles.values() + } for _, (subject, role_names) in subjects_bound_roles.items(): report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) report.resource_name = f"{subject.kind}:{subject.name}" report.resource_id = f"{subject.kind}/{subject.name}" 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 + for role_name in role_names: + cr = cluster_roles_by_name.get(role_name) + if cr and 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 4a2305c445..377e5345da 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 @@ -22,18 +22,21 @@ class rbac_minimize_node_proxy_subresource_access(Check): subjects_bound_roles[key] = (subject, set()) subjects_bound_roles[key][1].add(crb.roleRef.name) + cluster_roles_by_name = { + cr.metadata.name: cr for cr in rbac_client.cluster_roles.values() + } for _, (subject, role_names) in subjects_bound_roles.items(): report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) report.resource_name = f"{subject.kind}:{subject.name}" report.resource_id = f"{subject.kind}/{subject.name}" 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 + for role_name in role_names: + cr = cluster_roles_by_name.get(role_name) + if cr and 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 80389f3fba..2fb76bbed7 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 @@ -22,18 +22,21 @@ class rbac_minimize_pv_creation_access(Check): subjects_bound_roles[key] = (subject, set()) subjects_bound_roles[key][1].add(crb.roleRef.name) + cluster_roles_by_name = { + cr.metadata.name: cr for cr in rbac_client.cluster_roles.values() + } for _, (subject, role_names) in subjects_bound_roles.items(): report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) report.resource_name = f"{subject.kind}:{subject.name}" report.resource_id = f"{subject.kind}/{subject.name}" 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 + for role_name in role_names: + cr = cluster_roles_by_name.get(role_name) + if cr and 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 20c113797c..8e492309db 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 @@ -22,18 +22,21 @@ class rbac_minimize_service_account_token_creation(Check): subjects_bound_roles[key] = (subject, set()) subjects_bound_roles[key][1].add(crb.roleRef.name) + cluster_roles_by_name = { + cr.metadata.name: cr for cr in rbac_client.cluster_roles.values() + } for _, (subject, role_names) in subjects_bound_roles.items(): report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) report.resource_name = f"{subject.kind}:{subject.name}" report.resource_id = f"{subject.kind}/{subject.name}" 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 + for role_name in role_names: + cr = cluster_roles_by_name.get(role_name) + if cr and 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 55669012cf..ffdad52a19 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 @@ -25,18 +25,21 @@ class rbac_minimize_webhook_config_access(Check): subjects_bound_roles[key] = (subject, set()) subjects_bound_roles[key][1].add(crb.roleRef.name) + cluster_roles_by_name = { + cr.metadata.name: cr for cr in rbac_client.cluster_roles.values() + } for _, (subject, role_names) in subjects_bound_roles.items(): report = Check_Report_Kubernetes(metadata=self.metadata(), resource=subject) report.resource_name = f"{subject.kind}:{subject.name}" report.resource_id = f"{subject.kind}/{subject.name}" 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 + for role_name in role_names: + cr = cluster_roles_by_name.get(role_name) + if cr and 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