fix(azure): migrate resource models to avoid using SDK defaults (#7043)

Co-authored-by: Rubén De la Torre Vico <[email protected]>

prowler/providers/azure/services/aks/aks_service.py

@@ -2,7 +2,6 @@
 from typing import List
 
 from azure.mgmt.containerservice import ContainerServiceClient
-from azure.mgmt.containerservice.models import ManagedClusterAgentPoolProfile
 
 from prowler.lib.logger import logger
 from prowler.providers.azure.azure_provider import AzureProvider
@@ -42,9 +41,19 @@ def _get_clusters(self):
                                         if getattr(cluster, "network_profile", None)
                                         else None
                                     ),
-                                    agent_pool_profiles=getattr(
-                                        cluster, "agent_pool_profiles", []
-                                    ),
+                                    agent_pool_profiles=[
+                                        ManagedClusterAgentPoolProfile(
+                                            name=agent_pool_profile.name,
+                                            enable_node_public_ip=getattr(
+                                                agent_pool_profile,
+                                                "enable_node_public_ip",
+                                                False,
+                                            ),
+                                        )
+                                        for agent_pool_profile in getattr(
+                                            cluster, "agent_pool_profiles", []
+                                        )
+                                    ],
                                     rbac_enabled=getattr(cluster, "enable_rbac", False),
                                 )
                             }
@@ -57,6 +66,12 @@ def _get_clusters(self):
         return clusters
 
 
+@dataclass
+class ManagedClusterAgentPoolProfile:
+    name: str
+    enable_node_public_ip: bool
+
+
 @dataclass
 class Cluster:
     id: str

prowler/providers/azure/services/app/app_service.py

@@ -1,8 +1,7 @@
-from dataclasses import dataclass
-from typing import Dict
+from dataclasses import dataclass, field
+from typing import Dict, List
 
 from azure.mgmt.web import WebSiteManagementClient
-from azure.mgmt.web.models import ManagedServiceIdentity, SiteConfigResource
 
 from prowler.lib.logger import logger
 from prowler.providers.azure.azure_provider import AzureProvider
@@ -37,6 +36,11 @@ def _get_apps(self):
                             None,
                         )
 
+                        # Get app configurations
+                        app_configurations = client.web_apps.get_configuration(
+                            resource_group_name=app.resource_group, name=app.name
+                        )
+
                         apps[subscription_name].update(
                             {
                                 app.id: WebApp(
@@ -47,9 +51,30 @@ def _get_apps(self):
                                         if platform_auth
                                         else False
                                     ),
-                                    configurations=client.web_apps.get_configuration(
-                                        resource_group_name=app.resource_group,
-                                        name=app.name,
+                                    configurations=SiteConfigResource(
+                                        id=app_configurations.id,
+                                        name=app_configurations.name,
+                                        linux_fx_version=getattr(
+                                            app_configurations, "linux_fx_version", ""
+                                        ),
+                                        java_version=getattr(
+                                            app_configurations, "java_version", ""
+                                        ),
+                                        php_version=getattr(
+                                            app_configurations, "php_version", ""
+                                        ),
+                                        python_version=getattr(
+                                            app_configurations, "python_version", ""
+                                        ),
+                                        http20_enabled=getattr(
+                                            app_configurations, "http20_enabled", False
+                                        ),
+                                        ftps_state=getattr(
+                                            app_configurations, "ftps_state", ""
+                                        ),
+                                        min_tls_version=getattr(
+                                            app_configurations, "min_tls_version", ""
+                                        ),
                                     ),
                                     client_cert_mode=self._get_client_cert_mode(
                                         getattr(app, "client_cert_enabled", False),
@@ -59,7 +84,21 @@ def _get_apps(self):
                                         app.name, app.resource_group, subscription_name
                                     ),
                                     https_only=getattr(app, "https_only", False),
-                                    identity=getattr(app, "identity", None),
+                                    identity=ManagedServiceIdentity(
+                                        principal_id=getattr(
+                                            getattr(app, "identity", {}),
+                                            "principal_id",
+                                            "",
+                                        ),
+                                        tenant_id=getattr(
+                                            getattr(app, "identity", {}),
+                                            "tenant_id",
+                                            "",
+                                        ),
+                                        type=getattr(
+                                            getattr(app, "identity", {}), "type", ""
+                                        ),
+                                    ),
                                     location=app.location,
                                     kind=app.kind,
                                 )
@@ -171,6 +210,26 @@ def _get_app_monitor_settings(self, app_name, resource_group, subscription):
         return monitor_diagnostics_settings
 
 
+@dataclass
+class ManagedServiceIdentity:
+    principal_id: str
+    tenant_id: str
+    type: str
+
+
+@dataclass
+class SiteConfigResource:
+    id: str
+    name: str
+    linux_fx_version: str
+    java_version: str
+    php_version: str
+    python_version: str
+    http20_enabled: bool
+    ftps_state: str
+    min_tls_version: str
+
+
 @dataclass
 class WebApp:
     resource_id: str
@@ -181,7 +240,7 @@ class WebApp:
     client_cert_mode: str = "Ignore"
     auth_enabled: bool = False
     https_only: bool = False
-    monitor_diagnostic_settings: list[DiagnosticSetting] = None
+    monitor_diagnostic_settings: List[DiagnosticSetting] = field(default_factory=list)
     kind: str = "app"
 
 

prowler/providers/azure/services/containerregistry/containerregistry_service.py

@@ -1,10 +1,6 @@
 from dataclasses import dataclass
 
 from azure.mgmt.containerregistry import ContainerRegistryManagementClient
-from azure.mgmt.containerregistry.models import (
-    NetworkRuleSet,
-    PrivateEndpointConnection,
-)
 
 from prowler.lib.logger import logger
 from prowler.providers.azure.azure_provider import AzureProvider
@@ -40,7 +36,9 @@ def _get_container_registries(self):
                                 public_network_access=(
                                     False
                                     if getattr(
-                                        registry, "public_network_access" "Enabled"
+                                        registry,
+                                        "public_network_access_enabled",
+                                        "Enabled",
                                     )
                                     == "Disabled"
                                     else True
@@ -54,9 +52,16 @@ def _get_container_registries(self):
                                 monitor_diagnostic_settings=self._get_registry_monitor_settings(
                                     registry.name, resource_group, subscription
                                 ),
-                                private_endpoint_connections=getattr(
-                                    registry, "private_endpoint_connections", []
-                                ),
+                                private_endpoint_connections=[
+                                    PrivateEndpointConnection(
+                                        id=pec.id,
+                                        name=pec.name,
+                                        type=pec.type,
+                                    )
+                                    for pec in getattr(
+                                        registry, "private_endpoint_connections", []
+                                    )
+                                ],
                             )
                         },
                     )
@@ -90,6 +95,13 @@ def _get_registry_monitor_settings(
         return monitor_diagnostics_settings
 
 
+@dataclass
+class PrivateEndpointConnection:
+    id: str
+    name: str
+    type: str
+
+
 @dataclass
 class ContainerRegistryInfo:
     id: str
@@ -100,6 +112,5 @@ class ContainerRegistryInfo:
     login_server: str
     public_network_access: bool
     admin_user_enabled: bool
-    network_rule_set: NetworkRuleSet
     monitor_diagnostic_settings: list[DiagnosticSetting]
     private_endpoint_connections: list[PrivateEndpointConnection]

prowler/providers/azure/services/cosmosdb/cosmosdb_service.py

@@ -1,7 +1,7 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
+from typing import List
 
 from azure.mgmt.cosmosdb import CosmosDBManagementClient
-from azure.mgmt.cosmosdb.models import PrivateEndpointConnection
 
 from prowler.lib.logger import logger
 from prowler.providers.azure.azure_provider import AzureProvider
@@ -30,7 +30,14 @@ def _get_accounts(self):
                             type=account.type,
                             tags=account.tags,
                             is_virtual_network_filter_enabled=account.is_virtual_network_filter_enabled,
-                            private_endpoint_connections=account.private_endpoint_connections,
+                            private_endpoint_connections=[
+                                PrivateEndpointConnection(
+                                    id=private_endpoint_connection.id,
+                                    name=private_endpoint_connection.name,
+                                    type=private_endpoint_connection.type,
+                                )
+                                for private_endpoint_connection in account.private_endpoint_connections
+                            ],
                             disable_local_auth=account.disable_local_auth,
                         )
                     )
@@ -41,6 +48,13 @@ def _get_accounts(self):
         return accounts
 
 
+@dataclass
+class PrivateEndpointConnection:
+    id: str
+    name: str
+    type: str
+
+
 @dataclass
 class Account:
     id: str
@@ -50,7 +64,5 @@ class Account:
     tags: dict
     is_virtual_network_filter_enabled: bool
     location: str
-    private_endpoint_connections: list[PrivateEndpointConnection] = field(
-        default_factory=list
-    )
+    private_endpoint_connections: List[PrivateEndpointConnection]
     disable_local_auth: bool = False

prowler/providers/azure/services/entra/entra_service.py

@@ -48,7 +48,7 @@ async def _get_users(self):
                 for user in users_list.value:
                     users[tenant].update(
                         {
-                            user.user_principal_name: User(
+                            user.id: User(
                                 id=user.id,
                                 name=user.display_name,
                                 authentication_methods=[
@@ -261,11 +261,9 @@ async def _get_directory_roles(self):
                             directory_role.display_name: DirectoryRole(
                                 id=directory_role.id,
                                 members=[
-                                    self.users[tenant][member.user_principal_name]
+                                    self.users[tenant][member.id]
                                     for member in directory_role_members.value
-                                    if self.users[tenant].get(
-                                        member.user_principal_name, None
-                                    )
+                                    if self.users[tenant].get(member.id, None)
                                 ],
                             )
                         }

prowler/providers/azure/services/iam/iam_service.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
+from typing import List
 
 from azure.mgmt.authorization import AuthorizationManagementClient
-from azure.mgmt.authorization.v2022_04_01.models import Permission
 
 from prowler.lib.logger import logger
 from prowler.providers.azure.azure_provider import AzureProvider
@@ -33,7 +33,16 @@ def _get_roles(self):
                                 name=role.role_name,
                                 type=role.role_type,
                                 assignable_scopes=role.assignable_scopes,
-                                permissions=role.permissions,
+                                permissions=[
+                                    Permission(
+                                        condition=getattr(permission, "condition", ""),
+                                        condition_version=getattr(
+                                            permission, "condition_version", ""
+                                        ),
+                                        actions=getattr(permission, "actions", []),
+                                    )
+                                    for permission in getattr(role, "permissions", [])
+                                ],
                             )
                         )
                     else:
@@ -82,13 +91,20 @@ def _get_role_assignments(self):
         return role_assignments
 
 
+@dataclass
+class Permission:
+    actions: List[str]
+    condition: str
+    condition_version: str
+
+
 @dataclass
 class Role:
     id: str
     name: str
     type: str
-    assignable_scopes: list[str]
-    permissions: list[Permission]
+    assignable_scopes: List[str]
+    permissions: List[Permission]
 
 
 @dataclass

prowler/providers/azure/services/keyvault/keyvault_key_expiration_set_in_non_rbac/keyvault_key_expiration_set_in_non_rbac.py

@@ -16,11 +16,7 @@ def execute(self) -> Check_Report_Azure:
                     report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has all the keys with expiration date set."
                     has_key_without_expiration = False
                     for key in keyvault.keys:
-                        if (
-                            key.attributes
-                            and not key.attributes.expires
-                            and key.enabled
-                        ):
+                        if not key.attributes.expires and key.enabled:
                             report.status = "FAIL"
                             report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has the key {key.name} without expiration date set."
                             has_key_without_expiration = True

prowler/providers/azure/services/keyvault/keyvault_non_rbac_secret_expiration_set/keyvault_non_rbac_secret_expiration_set.py

@@ -19,11 +19,7 @@ def execute(self) -> Check_Report_Azure:
                     report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has all the secrets with expiration date set."
                     has_secret_without_expiration = False
                     for secret in keyvault.secrets:
-                        if (
-                            secret.attributes
-                            and not secret.attributes.expires
-                            and secret.enabled
-                        ):
+                        if not secret.attributes.expires and secret.enabled:
                             report.status = "FAIL"
                             report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has the secret {secret.name} without expiration date set."
                             has_secret_without_expiration = True

prowler/providers/azure/services/keyvault/keyvault_rbac_key_expiration_set/keyvault_rbac_key_expiration_set.py

@@ -16,11 +16,7 @@ def execute(self) -> Check_Report_Azure:
                     report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has all the keys with expiration date set."
                     has_key_without_expiration = False
                     for key in keyvault.keys:
-                        if (
-                            key.attributes
-                            and not key.attributes.expires
-                            and key.enabled
-                        ):
+                        if not key.attributes.expires and key.enabled:
                             report.status = "FAIL"
                             report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has the key {key.name} without expiration date set."
                             has_key_without_expiration = True

prowler/providers/azure/services/keyvault/keyvault_rbac_secret_expiration_set/keyvault_rbac_secret_expiration_set.py

@@ -16,11 +16,7 @@ def execute(self) -> Check_Report_Azure:
                     report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has all the secrets with expiration date set."
                     has_secret_without_expiration = False
                     for secret in keyvault.secrets:
-                        if (
-                            secret.attributes
-                            and not secret.attributes.expires
-                            and secret.enabled
-                        ):
+                        if not secret.attributes.expires and secret.enabled:
                             report.status = "FAIL"
                             report.status_extended = f"Keyvault {keyvault.name} from subscription {subscription} has the secret {secret.name} without expiration date set."
                             has_secret_without_expiration = True