fix(kubernetes): filter apiGroup in permission checks (#5829)

This commit is contained in:
Sergio Garcia
2024-11-20 10:06:06 -04:00
committed by GitHub
parent 2bf835d3d2
commit 4b7883c464
9 changed files with 51 additions and 26 deletions
@@ -1,4 +1,4 @@
def is_rule_allowing_permisions(rules, resources, verbs):
def is_rule_allowing_permissions(rules, resources, verbs):
"""
Check Kubernetes role permissions.
@@ -17,6 +17,9 @@ def is_rule_allowing_permisions(rules, resources, verbs):
if rules:
# Iterate through each rule in the list of rules
for rule in rules:
# Ensure apiGroups are relevant ("" or "v1" for secrets)
if rule.apiGroups and all(api not in ["", "v1"] for api in rule.apiGroups):
continue # Skip rules with unrelated apiGroups
# Check if the rule has resources, verbs, and matches any of the specified resources and verbs
if (
rule.resources
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -22,7 +22,7 @@ class rbac_minimize_csr_approval_access(Check):
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_permisions(
if is_rule_allowing_permissions(
cr.rules,
resources,
verbs,
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -22,7 +22,7 @@ class rbac_minimize_node_proxy_subresource_access(Check):
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_permisions(cr.rules, resources, verbs):
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
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -21,7 +21,7 @@ class rbac_minimize_pod_creation_access(Check):
report.status_extended = (
f"ClusterRole {cr.metadata.name} does not have pod create access."
)
if is_rule_allowing_permisions(cr.rules, resources, verbs):
if is_rule_allowing_permissions(cr.rules, resources, verbs):
report.status = "FAIL"
report.status_extended = (
f"ClusterRole {cr.metadata.name} has pod create access."
@@ -39,7 +39,7 @@ class rbac_minimize_pod_creation_access(Check):
f"Role {role.metadata.name} does not have pod create access."
)
if is_rule_allowing_permisions(role.rules, resources, verbs):
if is_rule_allowing_permissions(role.rules, resources, verbs):
report.status = "FAIL"
report.status_extended = (
f"Role {role.metadata.name} has pod create access."
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -23,7 +23,7 @@ class rbac_minimize_pv_creation_access(Check):
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_permisions(cr.rules, resources, verbs):
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
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -21,7 +21,7 @@ class rbac_minimize_secret_access(Check):
report.status_extended = (
f"ClusterRole {cr.metadata.name} does not have secret access."
)
if is_rule_allowing_permisions(cr.rules, resources, verbs):
if is_rule_allowing_permissions(cr.rules, resources, verbs):
report.status = "FAIL"
report.status_extended = (
f"ClusterRole {cr.metadata.name} has secret access."
@@ -39,7 +39,7 @@ class rbac_minimize_secret_access(Check):
f"Role {role.metadata.name} does not have secret access."
)
if is_rule_allowing_permisions(cr.rules, resources, verbs):
if is_rule_allowing_permissions(cr.rules, resources, verbs):
report.status = "FAIL"
report.status_extended = f"Role {role.metadata.name} has secret access."
findings.append(report)
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -22,7 +22,7 @@ class rbac_minimize_service_account_token_creation(Check):
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_permisions(cr.rules, resources, verbs):
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
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_Kubernetes
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_client import rbac_client
@@ -25,7 +25,7 @@ class rbac_minimize_webhook_config_access(Check):
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_permisions(
if is_rule_allowing_permissions(
cr.rules,
resources,
verbs,
@@ -1,11 +1,11 @@
from prowler.providers.kubernetes.services.rbac.lib.role_permissions import (
is_rule_allowing_permisions,
is_rule_allowing_permissions,
)
from prowler.providers.kubernetes.services.rbac.rbac_service import Rule
class TestCheckRolePermissions:
def test_is_rule_allowing_permisions(self):
def test_is_rule_allowing_permissions(self):
# Define some sample rules, resources, and verbs for testing
rules = [
# Rule 1: Allows 'get' and 'list' on 'pods' and 'services'
@@ -16,7 +16,7 @@ class TestCheckRolePermissions:
resources = ["pods", "deployments"]
verbs = ["get", "create"]
assert is_rule_allowing_permisions(rules, resources, verbs)
assert is_rule_allowing_permissions(rules, resources, verbs)
def test_no_permissions(self):
# Test when there are no rules
@@ -24,7 +24,7 @@ class TestCheckRolePermissions:
resources = ["pods", "deployments"]
verbs = ["get", "create"]
assert not is_rule_allowing_permisions(rules, resources, verbs)
assert not is_rule_allowing_permissions(rules, resources, verbs)
def test_no_matching_rules(self):
# Test when there are rules, but none match the specified resources and verbs
@@ -35,7 +35,7 @@ class TestCheckRolePermissions:
resources = ["deployments", "configmaps"]
verbs = ["get", "create"]
assert not is_rule_allowing_permisions(rules, resources, verbs)
assert not is_rule_allowing_permissions(rules, resources, verbs)
def test_empty_rules(self):
# Test when the rules list is empty
@@ -43,7 +43,7 @@ class TestCheckRolePermissions:
resources = ["pods", "deployments"]
verbs = ["get", "create"]
assert not is_rule_allowing_permisions(rules, resources, verbs)
assert not is_rule_allowing_permissions(rules, resources, verbs)
def test_empty_resources_and_verbs(self):
# Test when resources and verbs are empty lists
@@ -54,7 +54,7 @@ class TestCheckRolePermissions:
resources = []
verbs = []
assert not is_rule_allowing_permisions(rules, resources, verbs)
assert not is_rule_allowing_permissions(rules, resources, verbs)
def test_matching_rule_with_empty_resources_or_verbs(self):
# Test when a rule matches, but either resources or verbs are empty
@@ -65,9 +65,31 @@ class TestCheckRolePermissions:
resources = []
verbs = ["get"]
assert not is_rule_allowing_permisions(rules, resources, verbs)
assert not is_rule_allowing_permissions(rules, resources, verbs)
resources = ["pods"]
verbs = []
assert not is_rule_allowing_permisions(rules, resources, verbs)
assert not is_rule_allowing_permissions(rules, resources, verbs)
def test_rule_with_ignored_api_groups(self):
# Test when a rule has apiGroups that are not relevant
rules = [
Rule(resources=["pods"], verbs=["get"], apiGroups=["test"]),
Rule(resources=["services"], verbs=["list"], apiGroups=["test2"]),
]
resources = ["pods"]
verbs = ["get"]
assert not is_rule_allowing_permissions(rules, resources, verbs)
def test_rule_with_relevant_api_groups(self):
# Test when a rule has apiGroups that are relevant
rules = [
Rule(resources=["pods"], verbs=["get"], apiGroups=["", "v1"]),
Rule(resources=["services"], verbs=["list"], apiGroups=["test2"]),
]
resources = ["pods"]
verbs = ["get"]
assert is_rule_allowing_permissions(rules, resources, verbs)