ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(aks): use Check_Report_Azure constructor properly in AKS checks (#6509)

Browse Source
This commit is contained in:
Rubén De la Torre Vico
2025-01-14 20:14:02 +01:00
committed by GitHub
parent 90fd9b0eb8
commit ce60c286dc
11 changed files with 57 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/check/models.py
+23 -4
View File
@@ -5,7 +5,7 @@ import sys
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Any, Set
from typing import Any, Dict, Set
from pydantic import BaseModel, ValidationError, validator
@@ -410,10 +410,29 @@ class Check_Report:
resource_tags: list
muted: bool
def __init__(self, metadata, resource=None):
def __init__(self, metadata: Dict, resource: Any = None) -> None:
"""Initialize the Check's finding information.
Args:
metadata: The metadata of the check.
resource: Basic information about the resource. Defaults to None.
Only accepted BaseModels (dict attribute), custom models (to_dict attribute) or objects with __dict__.
"""
self.status = ""
self.check_metadata = CheckMetadata.parse_raw(metadata)
self.resource_metadata = resource.dict() if resource else {}
if hasattr(resource, "dict"):
self.resource_metadata = resource.dict()
elif hasattr(resource, "to_dict"):
self.resource_metadata = resource.to_dict()
elif hasattr(resource, "__dict__"):
self.resource_metadata = resource.__dict__
else:
logger.error(
f"Resource metadata {type(resource)} could not be converted to dict"
)
self.resource_metadata = {}
self.status_extended = ""
self.resource_details = ""
self.resource_tags = getattr(resource, "tags", []) if resource else []
@@ -453,7 +472,7 @@ class Check_Report_Azure(Check_Report):
subscription: str
location: str
def __init__(self, metadata: dict, resource_metadata: Any = None) -> None:
def __init__(self, metadata: Dict, resource_metadata: Any = None) -> None:
"""Initialize the Azure Check's finding information.
Args:
prowler/providers/azure/services/aks/aks_cluster_rbac_enabled/aks_cluster_rbac_enabled.py
+5 -6
View File
@@ -7,13 +7,12 @@ class aks_cluster_rbac_enabled(Check):
findings = []
for subscription_name, clusters in aks_client.clusters.items():
for cluster_id, cluster in clusters.items():
report = Check_Report_Azure(self.metadata())
report.status = "PASS"
for cluster in clusters.values():
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=cluster
)
report.subscription = subscription_name
report.resource_name = cluster.name
report.resource_id = cluster_id
report.location = cluster.location
report.status = "PASS"
report.status_extended = f"RBAC is enabled for cluster '{cluster.name}' in subscription '{subscription_name}'."
if not cluster.rbac_enabled:
prowler/providers/azure/services/aks/aks_clusters_created_with_private_nodes/aks_clusters_created_with_private_nodes.py
+5 -6
View File
@@ -7,13 +7,12 @@ class aks_clusters_created_with_private_nodes(Check):
findings = []
for subscription_name, clusters in aks_client.clusters.items():
for cluster_id, cluster in clusters.items():
report = Check_Report_Azure(self.metadata())
report.status = "PASS"
for cluster in clusters.values():
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=cluster
)
report.subscription = subscription_name
report.resource_name = cluster.name
report.resource_id = cluster_id
report.location = cluster.location
report.status = "PASS"
report.status_extended = f"Cluster '{cluster.name}' was created with private nodes in subscription '{subscription_name}'"
for agent_pool in cluster.agent_pool_profiles:
prowler/providers/azure/services/aks/aks_clusters_public_access_disabled/aks_clusters_public_access_disabled.py
+5 -6
View File
@@ -7,13 +7,12 @@ class aks_clusters_public_access_disabled(Check):
findings = []
for subscription_name, clusters in aks_client.clusters.items():
for cluster_id, cluster in clusters.items():
report = Check_Report_Azure(self.metadata())
report.status = "FAIL"
for cluster in clusters.values():
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=cluster
)
report.subscription = subscription_name
report.resource_name = cluster.name
report.resource_id = cluster_id
report.location = cluster.location
report.status = "FAIL"
report.status_extended = f"Public access to nodes is enabled for cluster '{cluster.name}' in subscription '{subscription_name}'"
if cluster.private_fqdn:
prowler/providers/azure/services/aks/aks_network_policy_enabled/aks_network_policy_enabled.py
+4 -5
View File
@@ -8,12 +8,11 @@ class aks_network_policy_enabled(Check):
for subscription_name, clusters in aks_client.clusters.items():
for cluster_id, cluster in clusters.items():
report = Check_Report_Azure(self.metadata())
report.status = "PASS"
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=cluster
)
report.subscription = subscription_name
report.resource_name = cluster.name
report.resource_id = cluster_id
report.location = cluster.location
report.status = "PASS"
report.status_extended = f"Network policy is enabled for cluster '{cluster.name}' in subscription '{subscription_name}'."
if not getattr(cluster, "network_policy", False):
prowler/providers/azure/services/aks/aks_service.py
+4 -2
View File
@@ -1,4 +1,5 @@
from dataclasses import dataclass
from typing import List
from azure.mgmt.containerservice import ContainerServiceClient
from azure.mgmt.containerservice.models import ManagedClusterAgentPoolProfile
@@ -8,7 +9,6 @@ from prowler.providers.azure.azure_provider import AzureProvider
from prowler.providers.azure.lib.service.service import AzureService
########################## AKS (Azure Kubernetes Service)
class AKS(AzureService):
def __init__(self, provider: AzureProvider):
super().__init__(ContainerServiceClient, provider)
@@ -28,6 +28,7 @@ class AKS(AzureService):
clusters[subscription_name].update(
{
cluster.id: Cluster(
id=cluster.id,
name=cluster.name,
public_fqdn=cluster.fqdn,
private_fqdn=cluster.private_fqdn,
@@ -58,10 +59,11 @@ class AKS(AzureService):
@dataclass
class Cluster:
id: str
name: str
public_fqdn: str
private_fqdn: str
network_policy: str
agent_pool_profiles: list[ManagedClusterAgentPoolProfile]
agent_pool_profiles: List[ManagedClusterAgentPoolProfile]
rbac_enabled: bool
location: str
tests/providers/azure/services/aks/aks_cluster_rbac_enabled/aks_cluster_rbac_enabled_test.py
+2
View File
@@ -53,6 +53,7 @@ class Test_aks_cluster_rbac_enabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn=None,
@@ -94,6 +95,7 @@ class Test_aks_cluster_rbac_enabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn=None,
tests/providers/azure/services/aks/aks_clusters_created_with_private_nodes/aks_clusters_created_with_private_nodes_test.py
+3
View File
@@ -53,6 +53,7 @@ class Test_aks_clusters_created_with_private_nodes:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn="",
@@ -94,6 +95,7 @@ class Test_aks_clusters_created_with_private_nodes:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn="private_fqdn",
@@ -135,6 +137,7 @@ class Test_aks_clusters_created_with_private_nodes:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn="private_fqdn",
tests/providers/azure/services/aks/aks_clusters_public_access_disabled/aks_clusters_public_access_disabled_test.py
+3
View File
@@ -53,6 +53,7 @@ class Test_aks_clusters_public_access_disabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn=None,
@@ -94,6 +95,7 @@ class Test_aks_clusters_public_access_disabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn="private_fqdn",
@@ -135,6 +137,7 @@ class Test_aks_clusters_public_access_disabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn="private_fqdn",
tests/providers/azure/services/aks/aks_network_policy_enabled/aks_network_policy_enabled_test.py
+2
View File
@@ -53,6 +53,7 @@ class Test_aks_network_policy_enabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn=None,
@@ -94,6 +95,7 @@ class Test_aks_network_policy_enabled:
aks_client.clusters = {
AZURE_SUBSCRIPTION_ID: {
cluster_id: Cluster(
id=cluster_id,
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn=None,
tests/providers/azure/services/aks/aks_service_test.py
+1
View File
@@ -11,6 +11,7 @@ def mock_aks_get_clusters(_):
return {
AZURE_SUBSCRIPTION_ID: {
"cluster_id-1": Cluster(
id="cluster_id-1",
name="cluster_name",
public_fqdn="public_fqdn",
private_fqdn="private_fqdn",