Merge branch 'v5.13' into backport/v5.13/pr-9208

This commit is contained in:
Pedro Martín
2025-11-19 14:29:11 +01:00
committed by GitHub
26 changed files with 620 additions and 51 deletions
+2 -2
View File
@@ -24,7 +24,7 @@ Standard results will be shown and additionally the framework information as the
**If Prowler can't find a resource related with a check from a compliance requirement, this requirement won't appear on the output**
</Note>
## List Available Compliance Frameworks
## List Available Compliance Frameworks
To see which compliance frameworks are covered by Prowler, use the `--list-compliance` option:
@@ -34,7 +34,7 @@ prowler <provider> --list-compliance
Or you can visit [Prowler Hub](https://hub.prowler.com/compliance).
## List Requirements of Compliance Frameworks
## List Requirements of Compliance Frameworks
To list requirements for a compliance framework, use the `--list-compliance-requirements` option:
```sh
@@ -94,7 +94,7 @@ The following list includes all the Azure checks with configurable variables tha
### Configurable Checks
## Kubernetes
## Kubernetes
### Configurable Checks
The following list includes all the Kubernetes checks with configurable variables that can be changed in the configuration yaml file:
@@ -2,7 +2,7 @@
title: 'Integrations'
---
## Integration with Slack
## Integration with Slack
Prowler can be integrated with [Slack](https://slack.com/) to send a summary of the execution having configured a Slack APP in your channel with the following command:
+3 -3
View File
@@ -14,7 +14,7 @@ prowler <provider> -V/-v/--version
Prowler provides various execution settings.
### Verbose Execution
### Verbose Execution
To enable verbose mode in Prowler, similar to Version 2, use:
@@ -54,7 +54,7 @@ To run Prowler without color formatting:
prowler <provider> --no-color
```
### Checks in Prowler
### Checks in Prowler
Prowler provides various security checks per cloud provider. Use the following options to list, execute, or exclude specific checks:
@@ -96,7 +96,7 @@ prowler <provider> -e/--excluded-checks ec2 rds
prowler <provider> -C/--checks-file <checks_list>.json
```
## Custom Checks in Prowler
## Custom Checks in Prowler
Prowler supports custom security checks, allowing users to define their own logic.
+5 -4
View File
@@ -27,7 +27,7 @@ If any of the criteria do not match, the check is not muted.
Remember that mutelist can be used with regular expressions.
</Note>
## Mutelist Specification
## Mutelist Specification
<Note>
- For Azure provider, the Account ID is the Subscription Name and the Region is the Location.
@@ -40,9 +40,10 @@ The Mutelist file uses the [YAML](https://en.wikipedia.org/wiki/YAML) format wit
```yaml
### Account, Check and/or Region can be * to apply for all the cases.
### Resources and tags are lists that can have either Regex or Keywords.
### Tags is an optional list that matches on tuples of 'key=value' and are "ANDed" together.
### Use an alternation Regex to match one of multiple tags with "ORed" logic.
### For each check you can except Accounts, Regions, Resources and/or Tags.
### Multiple tags in the list are "ANDed" together (ALL must match).
### Use regex alternation (|) within a single tag for "OR" logic (e.g., "env=dev|env=stg").
### For each check you can use Exceptions to unmute specific Accounts, Regions, Resources and/or Tags.
### All conditions (Account, Check, Region, Resource, Tags) are ANDed together.
########################### MUTELIST EXAMPLE ###########################
Mutelist:
Accounts:
@@ -10,7 +10,7 @@ This can help for really large accounts, but please be aware of AWS API rate lim
2. **API Rate Limits**: Most of the rate limits in AWS are applied at the API level. Each API call to an AWS service counts towards the rate limit for that service.
3. **Throttling Responses**: When you exceed the rate limit for a service, AWS responds with a throttling error. In AWS SDKs, these are typically represented as `ThrottlingException` or `RateLimitExceeded` errors.
For information on Prowler's retrier configuration please refer to this [page](https://docs.prowler.cloud/en/latest/tutorials/aws/boto3-configuration/).
For information on Prowler's retrier configuration please refer to this [page](https://docs.prowler.com/user-guide/providers/aws/boto3-configuration/).
<Note>
You might need to increase the `--aws-retries-max-attempts` parameter from the default value of 3. The retrier follows an exponential backoff strategy.
@@ -24,6 +24,6 @@ By default, it extracts resources from all the regions, you could use `-f`/`--fi
![Quick Inventory Example](/images/quick-inventory.jpg)
## Objections
## Objections
The inventorying process is carried out with `resourcegroupstaggingapi` calls, which means that only resources they have or have had tags will appear (except for the IAM and S3 resources which are done with Boto3 API calls).
+2 -2
View File
@@ -22,7 +22,7 @@ prowler <provider> --output-formats json-asff
All compliance-related reports are automatically generated when Prowler is executed. These outputs are stored in the `/output/compliance` directory.
## Custom Output Flags
## Custom Output Flags
By default, Prowler creates a file inside the `output` directory named: `prowler-output-ACCOUNT_NUM-OUTPUT_DATE.format`.
@@ -53,7 +53,7 @@ Both flags can be used simultaneously to provide a custom directory and filename
By default, the timestamp format of the output files is ISO 8601. This can be changed with the flag `--unix-timestamp` generating the timestamp fields in pure unix timestamp format.
## Supported Output Formats
## Supported Output Formats
Prowler natively supports the following reporting output formats:
@@ -14,7 +14,7 @@ prowler <provider> --scan-unused-services
## Services Ignored
### AWS
### AWS
#### ACM (AWS Certificate Manager)
@@ -22,21 +22,21 @@ Certificates stored in ACM without active usage in AWS resources are excluded. B
- `acm_certificates_expiration_check`
#### Athena
#### Athena
Upon AWS account creation, Athena provisions a default primary workgroup for the user. Prowler verifies if this workgroup is enabled and used by checking for queries within the last 45 days. If Athena is unused, findings related to its checks will not appear.
- `athena_workgroup_encryption`
- `athena_workgroup_enforce_configuration`
#### AWS CloudTrail
#### AWS CloudTrail
AWS CloudTrail should have at least one trail with a data event to record all S3 object-level API operations. Before flagging this issue, Prowler verifies if S3 buckets exist in the account.
- `cloudtrail_s3_dataevents_read_enabled`
- `cloudtrail_s3_dataevents_write_enabled`
#### AWS Elastic Compute Cloud (EC2)
#### AWS Elastic Compute Cloud (EC2)
If Amazon Elastic Block Store (EBS) default encyption is not enabled, sensitive data at rest will remain unprotected in EC2. However, Prowler will only generate a finding if EBS volumes exist where default encryption could be enforced.
@@ -56,7 +56,7 @@ Prowler scans only attached security groups to report vulnerabilities in activel
- `ec2_networkacl_allow_ingress_X_port`
#### AWS Glue
#### AWS Glue
AWS Glue best practices recommend encrypting metadata and connection passwords in Data Catalogs.
@@ -71,7 +71,7 @@ Amazon Inspector is a vulnerability discovery service that automates continuous
- `inspector2_is_enabled`
#### Amazon Macie
#### Amazon Macie
Amazon Macie leverages machine learning to automatically discover, classify, and protect sensitive data in S3 buckets. Prowler only generates findings if Macie is disabled and there are S3 buckets in the AWS account.
@@ -83,7 +83,7 @@ A network firewall is essential for monitoring and controlling traffic within a
- `networkfirewall_in_all_vpc`
#### Amazon S3
#### Amazon S3
To prevent unintended data exposure:
@@ -91,7 +91,7 @@ Public Access Block should be enabled at the account level. Prowler only checks
- `s3_account_level_public_access_blocks`
#### Virtual Private Cloud (VPC)
#### Virtual Private Cloud (VPC)
VPC settings directly impact network security and availability.
+3
View File
@@ -10,6 +10,9 @@ 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)
- Update documentation URLs from docs.prowler.cloud to docs.prowler.com [(#9240)](https://github.com/prowler-cloud/prowler/pull/9240)
- Fix file name parsing for checks on Windows [(#9268)](https://github.com/prowler-cloud/prowler/pull/9268)
## [v5.13.1] (Prowler v5.13.1)
+1 -1
View File
@@ -422,7 +422,7 @@ def prowler():
else:
# Refactor(CLI)
logger.critical(
"Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_NAME environment variables (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack)."
"Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_NAME environment variables (see more in https://docs.prowler.com/user-guide/cli/tutorials/integrations#configuration-of-the-integration-with-slack)."
)
sys.exit(1)
+2 -1
View File
@@ -457,7 +457,8 @@ class Check(ABC, CheckMetadata):
# Verify names consistency
check_id = self.CheckID
class_name = self.__class__.__name__
file_name = file_path.split(sep="/")[-1]
# os.path.basename handles Windows and POSIX paths reliably
file_name = os.path.basename(file_path)
errors = []
if check_id != class_name:
+4 -4
View File
@@ -301,7 +301,7 @@ Detailed documentation at https://docs.prowler.com
"--checks-folder",
"-x",
nargs="?",
help="Specify external directory with custom checks (each check must have a folder with the required files, see more in https://docs.prowler.cloud/en/latest/tutorials/misc/#custom-checks).",
help="Specify external directory with custom checks (each check must have a folder with the required files, see more in https://docs.prowler.com/user-guide/cli/tutorials/misc#custom-checks-in-prowler).",
)
def __init_list_checks_parser__(self):
@@ -354,7 +354,7 @@ Detailed documentation at https://docs.prowler.com
"--mutelist-file",
"-w",
nargs="?",
help="Path for mutelist YAML file. See example prowler/config/<provider>_mutelist.yaml for reference and format. For AWS provider, it also accepts AWS DynamoDB Table, Lambda ARNs or S3 URIs, see more in https://docs.prowler.cloud/en/latest/tutorials/mutelist/",
help="Path for mutelist YAML file. See example prowler/config/<provider>_mutelist.yaml for reference and format. For AWS provider, it also accepts AWS DynamoDB Table, Lambda ARNs or S3 URIs, see more in https://docs.prowler.com/user-guide/cli/tutorials/mutelist",
)
def __init_config_parser__(self):
@@ -381,7 +381,7 @@ Detailed documentation at https://docs.prowler.com
"--custom-checks-metadata-file",
nargs="?",
default=None,
help="Path for the custom checks metadata YAML file. See example prowler/config/custom_checks_metadata_example.yaml for reference and format. See more in https://docs.prowler.cloud/en/latest/tutorials/custom-checks-metadata/",
help="Path for the custom checks metadata YAML file. See example prowler/config/custom_checks_metadata_example.yaml for reference and format. See more in https://docs.prowler.com/user-guide/cli/tutorials/custom-checks-metadata/",
)
def __init_third_party_integrations_parser__(self):
@@ -399,5 +399,5 @@ Detailed documentation at https://docs.prowler.com
third_party_subparser.add_argument(
"--slack",
action="store_true",
help="Send a summary of the execution with a Slack APP in your channel. Environment variables SLACK_API_TOKEN and SLACK_CHANNEL_NAME are required (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack).",
help="Send a summary of the execution with a Slack APP in your channel. Environment variables SLACK_API_TOKEN and SLACK_CHANNEL_NAME are required (see more in https://docs.prowler.com/user-guide/cli/tutorials/integrations#configuration-of-the-integration-with-slack/).",
)
+1 -1
View File
@@ -64,7 +64,7 @@ def open_file(input_file: str, mode: str = "r") -> TextIOWrapper:
except OSError as os_error:
if os_error.strerror == "Too many open files":
logger.critical(
"Ooops! You reached your user session maximum open files. To solve this issue, increase the shell session limit by running this command `ulimit -n 4096`. For more info visit https://docs.prowler.cloud/en/latest/troubleshooting/"
"Ooops! You reached your user session maximum open files. To solve this issue, increase the shell session limit by running this command `ulimit -n 4096`. For more info visit https://docs.prowler.com/troubleshooting/"
)
else:
logger.critical(
@@ -297,7 +297,7 @@ def create_output(resources: list, provider: AwsProvider, args):
csv_file.close()
print(
f"\n{Fore.YELLOW}WARNING: Only resources that have or have had tags will appear (except for IAM and S3).\nSee more in https://docs.prowler.cloud/en/latest/tutorials/quick-inventory/#objections{Style.RESET_ALL}"
f"\n{Fore.YELLOW}WARNING: Only resources that have or have had tags will appear (except for IAM and S3).\nSee more in https://docs.prowler.com/user-guide/cli/tutorials/quick-inventory/#objections{Style.RESET_ALL}"
)
print("\nMore details in files:")
print(f" - CSV: {args.output_directory}/{output_file + csv_file_suffix}")
@@ -256,7 +256,7 @@ class SecurityHub:
security_hub_client.list_enabled_products_for_import()
):
logger.warning(
f"Security Hub is enabled in {region} but Prowler integration does not accept findings. More info: https://docs.prowler.cloud/en/latest/tutorials/aws/securityhub/"
f"Security Hub is enabled in {region} but Prowler integration does not accept findings. More info: https://docs.prowler.com/user-guide/providers/aws/securityhub#aws-security-hub-integration-with-prowler"
)
return region, None
else:
@@ -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):
+1 -1
View File
@@ -83,7 +83,7 @@ prowler = "prowler.__main__:prowler"
[project.urls]
"Changelog" = "https://github.com/prowler-cloud/prowler/releases"
"Documentation" = "https://docs.prowler.cloud"
"Documentation" = "https://docs.prowler.com"
"Homepage" = "https://github.com/prowler-cloud/prowler"
"Issue tracker" = "https://github.com/prowler-cloud/prowler/issues"
@@ -133,7 +133,7 @@ class TestSecurityHub:
(
"root",
WARNING,
f"Security Hub is enabled in {AWS_REGION_EU_WEST_1} but Prowler integration does not accept findings. More info: https://docs.prowler.cloud/en/latest/tutorials/aws/securityhub/",
f"Security Hub is enabled in {AWS_REGION_EU_WEST_1} but Prowler integration does not accept findings. More info: https://docs.prowler.com/user-guide/providers/aws/securityhub#aws-security-hub-integration-with-prowler",
)
]
@@ -1376,7 +1376,7 @@ class TestSecurityHub:
(
"root",
WARNING,
f"Security Hub is enabled in {AWS_REGION_EU_WEST_1} but Prowler integration does not accept findings. More info: https://docs.prowler.cloud/en/latest/tutorials/aws/securityhub/",
f"Security Hub is enabled in {AWS_REGION_EU_WEST_1} but Prowler integration does not accept findings. More info: https://docs.prowler.com/user-guide/providers/aws/securityhub#aws-security-hub-integration-with-prowler",
),
]
@@ -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