fix(azure): validation and other errors in cosmosdb, defender, storage and vm (#8915)

This commit is contained in:
Daniel Barranquero
2025-11-13 15:17:44 +01:00
committed by GitHub
parent 46666d29d3
commit d1380fc19d
9 changed files with 581 additions and 16 deletions
+1
View File
@@ -62,6 +62,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- False negative in `iam_role_cross_service_confused_deputy_prevention` check [(#9213)](https://github.com/prowler-cloud/prowler/pull/9213)
- Fix M365 Teams `--sp-env-auth` connection error and enhanced timeout logging [(#9191)](https://github.com/prowler-cloud/prowler/pull/9191)
- Rename `get_oci_assessment_summary` to `get_oraclecloud_assessment_summary` in HTML output [(#9200)](https://github.com/prowler-cloud/prowler/pull/9200)
- Fix Validation and other errors in Azure provider [(#8915)](https://github.com/prowler-cloud/prowler/pull/8915)
---
@@ -36,9 +36,14 @@ class CosmosDB(AzureService):
name=private_endpoint_connection.name,
type=private_endpoint_connection.type,
)
for private_endpoint_connection in account.private_endpoint_connections
for private_endpoint_connection in getattr(
account, "private_endpoint_connections", []
)
if private_endpoint_connection
],
disable_local_auth=account.disable_local_auth,
disable_local_auth=getattr(
account, "disable_local_auth", False
),
)
)
except Exception as error:
@@ -112,7 +112,9 @@ class Defender(AzureService):
assessment.display_name: Assesment(
resource_id=assessment.id,
resource_name=assessment.name,
status=assessment.status.code,
status=getattr(
getattr(assessment, "status", None), "code", None
),
)
}
)
@@ -304,7 +306,7 @@ class AutoProvisioningSetting(BaseModel):
class Assesment(BaseModel):
resource_id: str
resource_name: str
status: str
status: Optional[str] = None
class Setting(BaseModel):
@@ -141,10 +141,12 @@ class Storage(AzureService):
container_delete_retention_policy,
"enabled",
False,
),
)
or False,
days=getattr(
container_delete_retention_policy, "days", 0
),
)
or 0,
),
versioning_enabled=versioning_enabled,
)
@@ -220,12 +222,14 @@ class Storage(AzureService):
share_delete_retention_policy,
"enabled",
False,
),
)
or False,
days=getattr(
share_delete_retention_policy,
"days",
0,
),
)
or 0,
),
smb_protocol_settings=SMBProtocolSettings(
channel_encryption=(
@@ -241,6 +245,11 @@ class Storage(AzureService):
),
)
except Exception as error:
if "File is not supported for the account." in str(error).strip():
logger.warning(
f"Subscription name: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
continue
logger.error(
f"Subscription name: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
@@ -1,4 +1,3 @@
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional
@@ -294,16 +293,14 @@ class VirtualMachines(AzureService):
return vm_instance_ids
@dataclass
class UefiSettings:
class UefiSettings(BaseModel):
secure_boot_enabled: bool
v_tpm_enabled: bool
@dataclass
class SecurityProfile:
security_type: str
uefi_settings: Optional[UefiSettings]
class SecurityProfile(BaseModel):
security_type: Optional[str] = None
uefi_settings: Optional[UefiSettings] = None
class OperatingSystemType(Enum):
@@ -53,3 +53,83 @@ class Test_CosmosDB_Service:
is None
)
assert account.accounts[AZURE_SUBSCRIPTION_ID][0].disable_local_auth is None
def mock_cosmosdb_get_accounts_with_none(_):
"""Mock CosmosDB accounts with None private_endpoint_connections"""
from prowler.providers.azure.services.cosmosdb.cosmosdb_service import (
PrivateEndpointConnection,
)
return {
AZURE_SUBSCRIPTION_ID: [
Account(
id="/subscriptions/test/account1",
name="cosmosdb-none-pec",
kind="GlobalDocumentDB",
location="eastus",
type="Microsoft.DocumentDB/databaseAccounts",
tags={},
is_virtual_network_filter_enabled=False,
disable_local_auth=False,
private_endpoint_connections=[], # Empty list from getattr default
),
Account(
id="/subscriptions/test/account2",
name="cosmosdb-with-pec",
kind="MongoDB",
location="westus",
type="Microsoft.DocumentDB/databaseAccounts",
tags={"env": "test"},
is_virtual_network_filter_enabled=True,
disable_local_auth=True,
private_endpoint_connections=[
PrivateEndpointConnection(
id="/subscriptions/test/pec1",
name="pec-1",
type="Microsoft.Network/privateEndpoints",
)
],
),
]
}
@patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_service.CosmosDB._get_accounts",
new=mock_cosmosdb_get_accounts_with_none,
)
class Test_CosmosDB_Service_None_Handling:
"""Test CosmosDB service handling of None values"""
def test_account_with_none_private_endpoint_connections(self):
"""Test that CosmosDB handles None private_endpoint_connections gracefully"""
cosmosdb = CosmosDB(set_mocked_azure_provider())
# Find account with no connections
account = next(
acc
for acc in cosmosdb.accounts[AZURE_SUBSCRIPTION_ID]
if acc.name == "cosmosdb-none-pec"
)
assert account.private_endpoint_connections == []
assert account.disable_local_auth is False
def test_account_with_valid_private_endpoint_connections(self):
"""Test that CosmosDB handles valid private_endpoint_connections"""
cosmosdb = CosmosDB(set_mocked_azure_provider())
# Find account with connections
account = next(
acc
for acc in cosmosdb.accounts[AZURE_SUBSCRIPTION_ID]
if acc.name == "cosmosdb-with-pec"
)
assert len(account.private_endpoint_connections) == 1
assert account.private_endpoint_connections[0].id == "/subscriptions/test/pec1"
assert account.private_endpoint_connections[0].name == "pec-1"
assert (
account.private_endpoint_connections[0].type
== "Microsoft.Network/privateEndpoints"
)
assert account.disable_local_auth is True
@@ -283,3 +283,77 @@ class Test_Defender_Service:
assert policy1.name == "JITPolicy1"
assert policy1.location == "eastus"
assert set(policy1.vm_ids) == {"vm-1", "vm-2"}
def mock_defender_get_assessments_with_none(_):
"""Mock Defender assessments with None and valid statuses"""
return {
AZURE_SUBSCRIPTION_ID: {
"Assessment None": Assesment(
resource_id="/subscriptions/test/assessment1",
resource_name="assessment-none",
status=None, # None status
),
"Assessment Healthy": Assesment(
resource_id="/subscriptions/test/assessment2",
resource_name="assessment-healthy",
status="Healthy",
),
"Assessment Unhealthy": Assesment(
resource_id="/subscriptions/test/assessment3",
resource_name="assessment-unhealthy",
status="Unhealthy",
),
}
}
@patch(
"prowler.providers.azure.services.defender.defender_service.Defender._get_assessments",
new=mock_defender_get_assessments_with_none,
)
class Test_Defender_Service_Assessments_None_Handling:
"""Test Defender service handling of None values in assessments"""
def test_assessment_with_none_status(self):
"""Test that Defender handles assessments with None status gracefully"""
defender = Defender(set_mocked_azure_provider())
# Check assessment with None status
assessment = defender.assessments[AZURE_SUBSCRIPTION_ID]["Assessment None"]
assert assessment.resource_id == "/subscriptions/test/assessment1"
assert assessment.resource_name == "assessment-none"
assert assessment.status is None
def test_assessment_with_valid_status(self):
"""Test that Defender handles assessments with valid status"""
defender = Defender(set_mocked_azure_provider())
# Check assessment with Healthy status
assessment = defender.assessments[AZURE_SUBSCRIPTION_ID]["Assessment Healthy"]
assert assessment.resource_id == "/subscriptions/test/assessment2"
assert assessment.resource_name == "assessment-healthy"
assert assessment.status == "Healthy"
def test_assessment_with_multiple_mixed_statuses(self):
"""Test that Defender handles mix of None and valid statuses"""
defender = Defender(set_mocked_azure_provider())
# Should have all 3 assessments
assert len(defender.assessments[AZURE_SUBSCRIPTION_ID]) == 3
# Check None status
assessment_none = defender.assessments[AZURE_SUBSCRIPTION_ID]["Assessment None"]
assert assessment_none.status is None
# Check Healthy status
assessment_healthy = defender.assessments[AZURE_SUBSCRIPTION_ID][
"Assessment Healthy"
]
assert assessment_healthy.status == "Healthy"
# Check Unhealthy status
assessment_unhealthy = defender.assessments[AZURE_SUBSCRIPTION_ID][
"Assessment Unhealthy"
]
assert assessment_unhealthy.status == "Unhealthy"
@@ -224,3 +224,161 @@ class Test_Storage_Service:
account.file_service_properties.smb_protocol_settings.supported_versions
== []
)
def mock_storage_get_storage_accounts_with_none(_):
"""Mock storage accounts with None values in retention policies"""
blob_properties_none_days = BlobProperties(
id="id-none-days",
name="name-none-days",
type="type",
default_service_version="2019-07-07",
container_delete_retention_policy=DeleteRetentionPolicy(
enabled=True, days=0 # None converted to 0
),
versioning_enabled=False,
)
blob_properties_none_enabled = BlobProperties(
id="id-none-enabled",
name="name-none-enabled",
type="type",
default_service_version=None,
container_delete_retention_policy=DeleteRetentionPolicy(
enabled=False, days=30 # None enabled converted to False
),
versioning_enabled=True,
)
file_service_properties_none_days = FileServiceProperties(
id="id-file-none",
name="name-file-none",
type="type",
share_delete_retention_policy=DeleteRetentionPolicy(
enabled=False, days=0 # None converted to 0
),
smb_protocol_settings=SMBProtocolSettings(
channel_encryption=[], supported_versions=[]
),
)
return {
AZURE_SUBSCRIPTION_ID: [
Account(
id="id-none-days",
name="storage-none-days",
resouce_group_name="rg",
enable_https_traffic_only=True,
infrastructure_encryption=False,
allow_blob_public_access=False,
network_rule_set=NetworkRuleSet(
bypass="AzureServices", default_action="Allow"
),
encryption_type="Microsoft.Storage",
minimum_tls_version="TLS1_2",
key_expiration_period_in_days=None,
private_endpoint_connections=[],
location="eastus",
blob_properties=blob_properties_none_days,
default_to_entra_authorization=False,
replication_settings="Standard_LRS",
allow_cross_tenant_replication=True,
allow_shared_key_access=True,
file_service_properties=None,
),
Account(
id="id-none-enabled",
name="storage-none-enabled",
resouce_group_name="rg2",
enable_https_traffic_only=True,
infrastructure_encryption=False,
allow_blob_public_access=True,
network_rule_set=NetworkRuleSet(bypass="None", default_action="Deny"),
encryption_type="Microsoft.Storage",
minimum_tls_version="TLS1_2",
key_expiration_period_in_days=None,
private_endpoint_connections=[],
location="northeurope",
blob_properties=blob_properties_none_enabled,
default_to_entra_authorization=False,
replication_settings="Premium_LRS",
allow_cross_tenant_replication=False,
allow_shared_key_access=False,
file_service_properties=None,
),
Account(
id="id-file-none",
name="storage-file-none",
resouce_group_name="rg3",
enable_https_traffic_only=True,
infrastructure_encryption=True,
allow_blob_public_access=False,
network_rule_set=NetworkRuleSet(
bypass="AzureServices", default_action="Deny"
),
encryption_type="Microsoft.Keyvault",
minimum_tls_version="TLS1_2",
key_expiration_period_in_days=None,
private_endpoint_connections=[],
location="westus",
blob_properties=None,
default_to_entra_authorization=False,
replication_settings="Standard_GRS",
allow_cross_tenant_replication=True,
allow_shared_key_access=True,
file_service_properties=file_service_properties_none_days,
),
]
}
@patch(
"prowler.providers.azure.services.storage.storage_service.Storage._get_storage_accounts",
new=mock_storage_get_storage_accounts_with_none,
)
class Test_Storage_Service_Retention_Policy_None_Handling:
"""Test Storage service handling of None values in retention policies"""
def test_blob_properties_with_none_retention_days(self):
"""Test that Storage handles None days in container_delete_retention_policy"""
storage = Storage(set_mocked_azure_provider())
# Find account with None days converted to 0
account = next(
acc
for acc in storage.storage_accounts[AZURE_SUBSCRIPTION_ID]
if acc.name == "storage-none-days"
)
assert account.blob_properties is not None
assert account.blob_properties.container_delete_retention_policy.enabled is True
assert account.blob_properties.container_delete_retention_policy.days == 0
def test_blob_properties_with_none_retention_enabled(self):
"""Test that Storage handles None enabled in retention policy"""
storage = Storage(set_mocked_azure_provider())
# Find account with None enabled converted to False
account = next(
acc
for acc in storage.storage_accounts[AZURE_SUBSCRIPTION_ID]
if acc.name == "storage-none-enabled"
)
assert account.blob_properties is not None
assert (
account.blob_properties.container_delete_retention_policy.enabled is False
)
assert account.blob_properties.container_delete_retention_policy.days == 30
def test_file_service_properties_with_none_retention_days(self):
"""Test that Storage handles None days in share_delete_retention_policy"""
storage = Storage(set_mocked_azure_provider())
# Find account with None days in file service
account = next(
acc
for acc in storage.storage_accounts[AZURE_SUBSCRIPTION_ID]
if acc.name == "storage-file-none"
)
assert account.file_service_properties is not None
assert (
account.file_service_properties.share_delete_retention_policy.enabled
is False
)
assert account.file_service_properties.share_delete_retention_policy.days == 0
@@ -1,4 +1,4 @@
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from prowler.providers.azure.services.vm.vm_service import (
Disk,
@@ -226,3 +226,242 @@ def test_virtual_machine_with_linux_configuration():
vm = virtual_machines.virtual_machines[AZURE_SUBSCRIPTION_ID]["vm_id-linux"]
assert vm.linux_configuration is not None
assert vm.linux_configuration.disable_password_authentication is True
class Test_VirtualMachine_SecurityProfile_Validation:
"""Test VirtualMachine SecurityProfile Pydantic validation"""
def test_security_profile_with_all_fields(self):
"""Test that SecurityProfile with all fields validates correctly"""
vm = VirtualMachine(
resource_id="/subscriptions/test/vm1",
resource_name="test-vm",
location="eastus",
security_profile=SecurityProfile(
security_type="TrustedLaunch",
uefi_settings=UefiSettings(
secure_boot_enabled=True,
v_tpm_enabled=True,
),
),
extensions=[],
)
assert vm.security_profile is not None
assert vm.security_profile.security_type == "TrustedLaunch"
assert vm.security_profile.uefi_settings is not None
assert vm.security_profile.uefi_settings.secure_boot_enabled is True
assert vm.security_profile.uefi_settings.v_tpm_enabled is True
def test_security_profile_with_none_uefi_settings(self):
"""Test that SecurityProfile with None uefi_settings validates correctly"""
vm = VirtualMachine(
resource_id="/subscriptions/test/vm2",
resource_name="test-vm-2",
location="westus",
security_profile=SecurityProfile(
security_type="Standard",
uefi_settings=None,
),
extensions=[],
)
assert vm.security_profile is not None
assert vm.security_profile.security_type == "Standard"
assert vm.security_profile.uefi_settings is None
def test_security_profile_with_none_security_type(self):
"""Test that SecurityProfile with None security_type validates correctly"""
vm = VirtualMachine(
resource_id="/subscriptions/test/vm3",
resource_name="test-vm-3",
location="northeurope",
security_profile=SecurityProfile(
security_type=None,
uefi_settings=UefiSettings(
secure_boot_enabled=False,
v_tpm_enabled=False,
),
),
extensions=[],
)
assert vm.security_profile is not None
assert vm.security_profile.security_type is None
assert vm.security_profile.uefi_settings is not None
assert vm.security_profile.uefi_settings.secure_boot_enabled is False
def test_security_profile_with_all_none(self):
"""Test that SecurityProfile with all None values validates correctly"""
vm = VirtualMachine(
resource_id="/subscriptions/test/vm4",
resource_name="test-vm-4",
location="southeastasia",
security_profile=SecurityProfile(
security_type=None,
uefi_settings=None,
),
extensions=[],
)
assert vm.security_profile is not None
assert vm.security_profile.security_type is None
assert vm.security_profile.uefi_settings is None
def test_virtual_machine_with_none_security_profile(self):
"""Test that VirtualMachine with None security_profile validates correctly"""
vm = VirtualMachine(
resource_id="/subscriptions/test/vm5",
resource_name="test-vm-5",
location="japaneast",
security_profile=None,
extensions=[],
)
assert vm.security_profile is None
def test_security_profile_creation_from_azure_sdk_simulation(self):
"""
Test that SecurityProfile can be created from Azure SDK-like objects
This simulates the conversion that happens in _get_virtual_machines
"""
# Simulate Azure SDK SecurityProfile object
mock_azure_security_profile = MagicMock()
mock_azure_security_profile.security_type = "TrustedLaunch"
mock_azure_uefi_settings = MagicMock()
mock_azure_uefi_settings.secure_boot_enabled = True
mock_azure_uefi_settings.v_tpm_enabled = True
# Simulate the conversion that happens in the service
security_type = getattr(mock_azure_security_profile, "security_type", None)
uefi_settings = UefiSettings(
secure_boot_enabled=getattr(
mock_azure_uefi_settings, "secure_boot_enabled", False
),
v_tpm_enabled=getattr(mock_azure_uefi_settings, "v_tpm_enabled", False),
)
security_profile = SecurityProfile(
security_type=security_type,
uefi_settings=uefi_settings,
)
# Create VirtualMachine with converted SecurityProfile
vm = VirtualMachine(
resource_id="/subscriptions/test/vm6",
resource_name="test-vm-6",
location="uksouth",
security_profile=security_profile,
extensions=[],
)
# Verify no ValidationError is raised and data is correct
assert vm.security_profile is not None
assert vm.security_profile.security_type == "TrustedLaunch"
assert vm.security_profile.uefi_settings.secure_boot_enabled is True
assert vm.security_profile.uefi_settings.v_tpm_enabled is True
def test_security_profile_with_dict_input(self):
"""Test that SecurityProfile can be created from dictionary (Pydantic feature)"""
vm = VirtualMachine(
resource_id="/subscriptions/test/vm7",
resource_name="test-vm-7",
location="canadacentral",
security_profile={
"security_type": "ConfidentialVM",
"uefi_settings": {
"secure_boot_enabled": True,
"v_tpm_enabled": True,
},
},
extensions=[],
)
assert vm.security_profile is not None
assert vm.security_profile.security_type == "ConfidentialVM"
assert vm.security_profile.uefi_settings.secure_boot_enabled is True
def test_uefi_settings_boolean_values(self):
"""Test that UefiSettings properly handles boolean values"""
uefi_true = UefiSettings(secure_boot_enabled=True, v_tpm_enabled=True)
assert uefi_true.secure_boot_enabled is True
assert uefi_true.v_tpm_enabled is True
uefi_false = UefiSettings(secure_boot_enabled=False, v_tpm_enabled=False)
assert uefi_false.secure_boot_enabled is False
assert uefi_false.v_tpm_enabled is False
uefi_mixed = UefiSettings(secure_boot_enabled=True, v_tpm_enabled=False)
assert uefi_mixed.secure_boot_enabled is True
assert uefi_mixed.v_tpm_enabled is False
def test_security_profile_full_service_simulation(self):
"""
Full integration test simulating the complete VM service flow
This tests the actual scenario where Azure SDK objects are converted
"""
def mock_list_vms(*args, **kwargs):
# Simulate Azure SDK VM object with security_profile
mock_vm = MagicMock()
mock_vm.id = "/subscriptions/test/resourceGroups/test-rg/providers/Microsoft.Compute/virtualMachines/test-vm"
mock_vm.name = "test-vm-full-sim"
mock_vm.location = "eastus"
# Simulate Azure SDK SecurityProfile (this was causing the ValidationError)
mock_security_profile = MagicMock()
mock_security_profile.security_type = "TrustedLaunch"
mock_uefi_settings = MagicMock()
mock_uefi_settings.secure_boot_enabled = True
mock_uefi_settings.v_tpm_enabled = True
mock_security_profile.uefi_settings = mock_uefi_settings
mock_vm.security_profile = mock_security_profile
mock_vm.resources = []
mock_vm.storage_profile = None
mock_vm.hardware_profile = None
mock_vm.os_profile = None
return [mock_vm]
# Create mock client with properly configured virtual_machines attribute
mock_client = MagicMock()
# Explicitly create virtual_machines as a MagicMock to ensure it has list_all method
# This prevents AttributeError in GitHub Actions where it might be a dict
mock_client.virtual_machines = MagicMock()
mock_client.virtual_machines.list_all.side_effect = mock_list_vms
with (
patch.object(VirtualMachines, "_get_disks", return_value={}),
patch.object(VirtualMachines, "_get_vm_scale_sets", return_value={}),
patch.object(VirtualMachines, "_get_virtual_machines", return_value={}),
):
vm_service = VirtualMachines(set_mocked_azure_provider())
# Replace the client with our mocked one
vm_service.clients[AZURE_SUBSCRIPTION_ID] = mock_client
# Now call _get_virtual_machines with the mocked client (patch is removed)
# This simulates the actual service flow
virtual_machines = vm_service._get_virtual_machines()
# Verify VM was created successfully without ValidationError
assert len(virtual_machines[AZURE_SUBSCRIPTION_ID]) == 1
vm_id = list(virtual_machines[AZURE_SUBSCRIPTION_ID].keys())[0]
vm = virtual_machines[AZURE_SUBSCRIPTION_ID][vm_id]
# Verify the VM object is valid
assert vm.resource_name == "test-vm-full-sim"
assert vm.location == "eastus"
# Verify SecurityProfile was converted correctly (not Azure SDK object)
assert vm.security_profile is not None
assert isinstance(vm.security_profile, SecurityProfile)
assert vm.security_profile.security_type == "TrustedLaunch"
# Verify UefiSettings was converted correctly
assert vm.security_profile.uefi_settings is not None
assert isinstance(vm.security_profile.uefi_settings, UefiSettings)
assert vm.security_profile.uefi_settings.secure_boot_enabled is True
assert vm.security_profile.uefi_settings.v_tpm_enabled is True