diff --git a/prowler/changelog.d/azure-aiservices-network-checks.added.md b/prowler/changelog.d/azure-aiservices-network-checks.added.md new file mode 100644 index 0000000000..804dab6feb --- /dev/null +++ b/prowler/changelog.d/azure-aiservices-network-checks.added.md @@ -0,0 +1 @@ +`aiservices_account_uses_private_endpoint`, `aiservices_account_outbound_network_restricted`, `aiservices_account_managed_identity_enabled`, and `aiservices_account_diagnostic_logging_enabled` checks for Azure provider diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/__init__.py b/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled.metadata.json b/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled.metadata.json new file mode 100644 index 0000000000..230ea97743 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled.metadata.json @@ -0,0 +1,38 @@ +{ + "Provider": "azure", + "CheckID": "aiservices_account_diagnostic_logging_enabled", + "CheckTitle": "AI services account sends audit logs through a diagnostic setting", + "CheckType": [], + "ServiceName": "aiservices", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "microsoft.cognitiveservices/accounts", + "ResourceGroup": "ai_ml", + "Description": "**Azure AI services accounts** (including **Azure OpenAI** and **Azure AI Foundry**) emit resource logs only when a **diagnostic setting** routes them to Log Analytics, Storage, or Event Hubs. This check passes when the `Audit` category, or the `audit` or `allLogs` category group, is enabled.", + "Risk": "Without audit logs, data-plane activity on the account leaves no record:\n- **Detection**: key misuse, unusual callers, and prompt abuse go unnoticed\n- **Investigation**: incident responders cannot reconstruct who called which model and when\n- **Compliance**: audit-trail requirements for AI workloads are not met", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://learn.microsoft.com/en-us/azure/ai-services/diagnostic-logging", + "https://learn.microsoft.com/en-us/azure/azure-monitor/platform/diagnostic-settings" + ], + "Remediation": { + "Code": { + "CLI": "az monitor diagnostic-settings create --name --resource --workspace --logs '[{\"categoryGroup\":\"audit\",\"enabled\":true}]'", + "NativeIaC": "```bicep\n// Send audit logs from an Azure AI services account to Log Analytics\nresource account 'Microsoft.CognitiveServices/accounts@2024-10-01' existing = {\n name: ''\n}\n\nresource diagnostics 'Microsoft.Insights/diagnosticSettings@2021-05-01-preview' = {\n name: ''\n scope: account\n properties: {\n workspaceId: ''\n logs: [\n {\n categoryGroup: 'audit' // CRITICAL: sends audit logs\n enabled: true\n }\n ]\n }\n}\n```", + "Other": "1. In the Azure portal, open the AI services account and go to Monitoring > Diagnostic settings\n2. Click + Add diagnostic setting\n3. Under Category groups, select audit (or allLogs)\n4. Choose a destination such as a Log Analytics workspace and click Save", + "Terraform": "```hcl\nresource \"azurerm_monitor_diagnostic_setting\" \"\" {\n name = \"\"\n target_resource_id = azurerm_cognitive_account..id\n log_analytics_workspace_id = azurerm_log_analytics_workspace..id\n\n enabled_log {\n category_group = \"audit\" # CRITICAL: sends audit logs\n }\n}\n```" + }, + "Recommendation": { + "Text": "Add a **diagnostic setting** to each AI services account that sends the **audit** category group to a **Log Analytics workspace**, and alert on unusual activity. Add **RequestResponse** logs where your data-handling policy allows it.", + "Url": "https://hub.prowler.com/check/aiservices_account_diagnostic_logging_enabled" + } + }, + "Categories": [ + "logging", + "gen-ai" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled.py b/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled.py new file mode 100644 index 0000000000..2e89cb0e40 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled.py @@ -0,0 +1,64 @@ +from prowler.lib.check.models import Check, Check_Report_Azure +from prowler.providers.azure.services.aiservices.aiservices_client import ( + aiservices_client, +) + +AUDIT_CATEGORY = "audit" +AUDIT_CATEGORY_GROUPS = {"audit", "alllogs"} + + +def sends_audit_logs(diagnostic_setting) -> bool: + """Tell whether a diagnostic setting sends AI services audit logs. + + Args: + diagnostic_setting: Monitor `DiagnosticSetting` of the account. + + Returns: + True if the `Audit` category, or the `audit` or `allLogs` category + group, is enabled. + """ + return any( + log.enabled + and ( + (log.category or "").lower() == AUDIT_CATEGORY + or (log.category_group or "").lower() in AUDIT_CATEGORY_GROUPS + ) + for log in diagnostic_setting.logs + ) + + +class aiservices_account_diagnostic_logging_enabled(Check): + """AI services account sends audit logs through a diagnostic setting.""" + + def execute(self) -> list[Check_Report_Azure]: + """Evaluate audit logging on every AI services account. + + Returns: + One finding per account. MANUAL when the diagnostic settings + could not be read. + """ + findings = [] + for subscription_id, accounts in aiservices_client.accounts.items(): + subscription_name = aiservices_client.subscriptions.get( + subscription_id, subscription_id + ) + for account in accounts.values(): + report = Check_Report_Azure(metadata=self.metadata(), resource=account) + report.subscription = subscription_id + prefix = f"AI services account {account.name} (kind {account.kind}) from subscription {subscription_name} ({subscription_id})" + settings = account.monitor_diagnostic_settings + if settings is None: + report.status = "MANUAL" + report.status_extended = f"{prefix} diagnostic settings could not be read; grant Microsoft.Insights/diagnosticSettings/read (included in Reader) to evaluate audit logging." + elif any(sends_audit_logs(setting) for setting in settings): + report.status = "PASS" + report.status_extended = ( + f"{prefix} has a diagnostic setting that sends audit logs." + ) + else: + report.status = "FAIL" + report.status_extended = ( + f"{prefix} has no diagnostic setting that sends audit logs." + ) + findings.append(report) + return findings diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/__init__.py b/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled.metadata.json b/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled.metadata.json new file mode 100644 index 0000000000..620a399a67 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled.metadata.json @@ -0,0 +1,38 @@ +{ + "Provider": "azure", + "CheckID": "aiservices_account_managed_identity_enabled", + "CheckTitle": "AI services account has a managed identity", + "CheckType": [], + "ServiceName": "aiservices", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "low", + "ResourceType": "microsoft.cognitiveservices/accounts", + "ResourceGroup": "ai_ml", + "Description": "**Azure AI services accounts** (including **Azure OpenAI** and **Azure AI Foundry**) can use a system-assigned or user-assigned **managed identity** to reach other Azure resources, such as Key Vault for customer-managed keys or Storage and AI Search for data sources, without stored credentials.", + "Risk": "Without a managed identity, the account cannot authenticate to other Azure resources through Microsoft Entra ID:\n- **Credentials**: integrations fall back to connection strings and keys that must be stored and rotated\n- **Encryption**: customer-managed keys cannot be configured, because they need an identity\n- **Auditability**: access by the service is not tied to an Entra ID principal", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/overview", + "https://learn.microsoft.com/en-us/azure/ai-services/openai/how-to/managed-identity" + ], + "Remediation": { + "Code": { + "CLI": "az cognitiveservices account identity assign --name --resource-group ", + "NativeIaC": "```bicep\n// Enable a system-assigned managed identity on an Azure AI services account\nresource account 'Microsoft.CognitiveServices/accounts@2024-10-01' = {\n name: ''\n location: ''\n kind: 'OpenAI'\n sku: { name: 'S0' }\n identity: { type: 'SystemAssigned' } // CRITICAL: gives the account an Entra ID identity\n properties: {\n customSubDomainName: ''\n }\n}\n```", + "Other": "1. In the Azure portal, open the AI services account and go to Resource Management > Identity\n2. On the System assigned tab, set Status to On and click Save, or add a user-assigned identity\n3. Grant the identity least-privilege roles on the resources it must reach", + "Terraform": "```hcl\nresource \"azurerm_cognitive_account\" \"\" {\n name = \"\"\n location = \"\"\n resource_group_name = \"\"\n kind = \"OpenAI\"\n sku_name = \"S0\"\n custom_subdomain_name = \"\"\n\n identity {\n type = \"SystemAssigned\" # CRITICAL: gives the account an Entra ID identity\n }\n}\n```" + }, + "Recommendation": { + "Text": "Enable a **managed identity** on AI services accounts and use it, instead of keys or connection strings, to reach **Key Vault**, **Storage**, and **AI Search**. Grant it only the roles each integration needs.", + "Url": "https://hub.prowler.com/check/aiservices_account_managed_identity_enabled" + } + }, + "Categories": [ + "identity-access", + "gen-ai" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled.py b/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled.py new file mode 100644 index 0000000000..22cc034bb7 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled.py @@ -0,0 +1,40 @@ +from prowler.lib.check.models import Check, Check_Report_Azure +from prowler.providers.azure.services.aiservices.aiservices_client import ( + aiservices_client, +) + +NO_IDENTITY_TYPE = "none" + + +class aiservices_account_managed_identity_enabled(Check): + """AI services account has a managed identity.""" + + def execute(self) -> list[Check_Report_Azure]: + """Evaluate the managed identity of every AI services account. + + Azure reports a removed identity either as no identity block or as + type `None`; both fail. + + Returns: + One finding per account. + """ + findings = [] + for subscription_id, accounts in aiservices_client.accounts.items(): + subscription_name = aiservices_client.subscriptions.get( + subscription_id, subscription_id + ) + for account in accounts.values(): + report = Check_Report_Azure(metadata=self.metadata(), resource=account) + report.subscription = subscription_id + prefix = f"AI services account {account.name} (kind {account.kind}) from subscription {subscription_name} ({subscription_id})" + identity_type = account.identity_type + if identity_type and identity_type.lower() != NO_IDENTITY_TYPE: + report.status = "PASS" + report.status_extended = ( + f"{prefix} has a managed identity ({identity_type})." + ) + else: + report.status = "FAIL" + report.status_extended = f"{prefix} has no managed identity." + findings.append(report) + return findings diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/__init__.py b/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted.metadata.json b/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted.metadata.json new file mode 100644 index 0000000000..08117f1b7a --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted.metadata.json @@ -0,0 +1,37 @@ +{ + "Provider": "azure", + "CheckID": "aiservices_account_outbound_network_restricted", + "CheckTitle": "AI services account restricts outbound network access to an allowed FQDN list", + "CheckType": [], + "ServiceName": "aiservices", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "microsoft.cognitiveservices/accounts", + "ResourceGroup": "ai_ml", + "Description": "**Azure AI services accounts** (including **Azure OpenAI** and **Azure AI Foundry**) can call external endpoints, for example to fetch data for features that reach out to URLs. Setting `restrictOutboundNetworkAccess` to `true` limits those calls to the domains in `allowedFqdnList`.", + "Risk": "Unrestricted outbound access lets the service send data to any endpoint:\n- **Exfiltration**: a prompt-injected or misconfigured feature can post data to an attacker-controlled host\n- **Trust boundary**: the account can reach destinations no one reviewed\n- **Compliance**: data-loss-prevention controls for AI workloads are not enforced", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://learn.microsoft.com/en-us/azure/ai-services/cognitive-services-data-loss-prevention" + ], + "Remediation": { + "Code": { + "CLI": "az resource update --ids --set properties.restrictOutboundNetworkAccess=true properties.allowedFqdnList='[\"\"]'", + "NativeIaC": "```bicep\n// Restrict outbound calls from an Azure AI services account\nresource account 'Microsoft.CognitiveServices/accounts@2024-10-01' = {\n name: ''\n location: ''\n kind: 'OpenAI'\n sku: { name: 'S0' }\n properties: {\n customSubDomainName: ''\n restrictOutboundNetworkAccess: true // CRITICAL: blocks outbound calls outside the allow list\n allowedFqdnList: [\n ''\n ]\n }\n}\n```", + "Other": "1. List the external domains the account must reach, for example storage endpoints used by your data sources\n2. Deploy the Bicep or Terraform example with those domains in the allowed FQDN list\n3. Or run the Azure CLI command above with your domains in allowedFqdnList\n4. Test the features that call external endpoints; an empty allow list blocks every outbound call", + "Terraform": "```hcl\nresource \"azurerm_cognitive_account\" \"\" {\n name = \"\"\n location = \"\"\n resource_group_name = \"\"\n kind = \"OpenAI\"\n sku_name = \"S0\"\n custom_subdomain_name = \"\"\n\n outbound_network_access_restricted = true # CRITICAL: blocks outbound calls outside the allow list\n fqdns = [\"\"]\n}\n```" + }, + "Recommendation": { + "Text": "Enable **outbound network restriction** on AI services accounts and keep the **allowed FQDN list** to the domains each workload needs. Review the list when data sources change.", + "Url": "https://hub.prowler.com/check/aiservices_account_outbound_network_restricted" + } + }, + "Categories": [ + "trust-boundaries", + "gen-ai" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted.py b/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted.py new file mode 100644 index 0000000000..481ffa9843 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted.py @@ -0,0 +1,34 @@ +from prowler.lib.check.models import Check, Check_Report_Azure +from prowler.providers.azure.services.aiservices.aiservices_client import ( + aiservices_client, +) + + +class aiservices_account_outbound_network_restricted(Check): + """AI services account restricts outbound network access.""" + + def execute(self) -> list[Check_Report_Azure]: + """Evaluate outbound network restriction on every AI services account. + + Returns: + One finding per account. + """ + findings = [] + for subscription_id, accounts in aiservices_client.accounts.items(): + subscription_name = aiservices_client.subscriptions.get( + subscription_id, subscription_id + ) + for account in accounts.values(): + report = Check_Report_Azure(metadata=self.metadata(), resource=account) + report.subscription = subscription_id + prefix = f"AI services account {account.name} (kind {account.kind}) from subscription {subscription_name} ({subscription_id})" + if account.restrict_outbound_network_access: + report.status = "PASS" + report.status_extended = f"{prefix} restricts outbound network access to an allowed FQDN list." + else: + report.status = "FAIL" + report.status_extended = ( + f"{prefix} allows unrestricted outbound network access." + ) + findings.append(report) + return findings diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/__init__.py b/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint.metadata.json b/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint.metadata.json new file mode 100644 index 0000000000..70f7967150 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint.metadata.json @@ -0,0 +1,38 @@ +{ + "Provider": "azure", + "CheckID": "aiservices_account_uses_private_endpoint", + "CheckTitle": "AI services account has an approved private endpoint connection", + "CheckType": [], + "ServiceName": "aiservices", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "microsoft.cognitiveservices/accounts", + "ResourceGroup": "ai_ml", + "Description": "**Azure AI services accounts** (including **Azure OpenAI** and **Azure AI Foundry**) can be reached through a **private endpoint**, a private IP address in your virtual network. Only connections in the `Approved` state carry traffic; `Pending` and `Rejected` connections do not.", + "Risk": "Without an approved private endpoint, clients reach the account over its public endpoint:\n- **Confidentiality**: prompts and completions cross the public internet\n- **Exposure**: the endpoint stays reachable for credential stuffing and key replay\n- **Control**: network-layer controls such as NSGs and private DNS do not apply", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://learn.microsoft.com/en-us/azure/ai-services/cognitive-services-virtual-networks", + "https://learn.microsoft.com/en-us/azure/private-link/private-endpoint-overview" + ], + "Remediation": { + "Code": { + "CLI": "az network private-endpoint create --name --resource-group --vnet-name --subnet --private-connection-resource-id --group-id account --connection-name ", + "NativeIaC": "```bicep\n// Private endpoint for an Azure AI services account\nresource privateEndpoint 'Microsoft.Network/privateEndpoints@2023-11-01' = {\n name: ''\n location: ''\n properties: {\n subnet: { id: '' }\n privateLinkServiceConnections: [\n {\n name: ''\n properties: {\n privateLinkServiceId: ''\n groupIds: [ 'account' ] // CRITICAL: connects the account through the private endpoint\n }\n }\n ]\n }\n}\n```", + "Other": "1. In the Azure portal, open the AI services account and go to Resource Management > Networking\n2. Select the Private endpoint connections tab and click + Private endpoint\n3. Choose the virtual network and subnet, keep the target sub-resource account, and enable private DNS integration\n4. Approve the connection if it shows Pending, then set Public network access to Disabled", + "Terraform": "```hcl\nresource \"azurerm_private_endpoint\" \"\" {\n name = \"\"\n location = \"\"\n resource_group_name = \"\"\n subnet_id = azurerm_subnet..id\n\n private_service_connection {\n name = \"\"\n private_connection_resource_id = azurerm_cognitive_account..id\n subresource_names = [\"account\"] # CRITICAL: connects the account through the private endpoint\n is_manual_connection = false\n }\n}\n```" + }, + "Recommendation": { + "Text": "Connect AI services accounts to your virtual network through an **approved private endpoint** with **private DNS** integration. Then disable **public network access** so the public endpoint stops accepting traffic.", + "Url": "https://hub.prowler.com/check/aiservices_account_uses_private_endpoint" + } + }, + "Categories": [ + "internet-exposed", + "gen-ai" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint.py b/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint.py new file mode 100644 index 0000000000..cde9cac618 --- /dev/null +++ b/prowler/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint.py @@ -0,0 +1,44 @@ +from prowler.lib.check.models import Check, Check_Report_Azure +from prowler.providers.azure.services.aiservices.aiservices_client import ( + aiservices_client, +) + +APPROVED_STATUS = "approved" + + +class aiservices_account_uses_private_endpoint(Check): + """AI services account has an approved private endpoint connection.""" + + def execute(self) -> list[Check_Report_Azure]: + """Evaluate private endpoint connections on every AI services account. + + Pending and rejected connections carry no traffic, so only an + approved connection passes. + + Returns: + One finding per account. + """ + findings = [] + for subscription_id, accounts in aiservices_client.accounts.items(): + subscription_name = aiservices_client.subscriptions.get( + subscription_id, subscription_id + ) + for account in accounts.values(): + report = Check_Report_Azure(metadata=self.metadata(), resource=account) + report.subscription = subscription_id + prefix = f"AI services account {account.name} (kind {account.kind}) from subscription {subscription_name} ({subscription_id})" + if any( + status.lower() == APPROVED_STATUS + for status in account.private_endpoint_connection_statuses + ): + report.status = "PASS" + report.status_extended = ( + f"{prefix} has an approved private endpoint connection." + ) + else: + report.status = "FAIL" + report.status_extended = ( + f"{prefix} has no approved private endpoint connection." + ) + findings.append(report) + return findings diff --git a/prowler/providers/azure/services/aiservices/aiservices_service.py b/prowler/providers/azure/services/aiservices/aiservices_service.py index d4e8e71d01..ed39fc9341 100644 --- a/prowler/providers/azure/services/aiservices/aiservices_service.py +++ b/prowler/providers/azure/services/aiservices/aiservices_service.py @@ -45,6 +45,9 @@ class AIServices(AzureService): for sdk_account in sdk_accounts: try: account = self._to_account(sdk_account) + account.monitor_diagnostic_settings = self._get_diagnostic_settings( + subscription, account.id + ) accounts[subscription][account.id] = account except Exception as error: logger.error( @@ -57,7 +60,8 @@ class AIServices(AzureService): """Map an SDK account to the Prowler model. Unset properties follow Azure defaults: public network access on, - local (key) authentication on, and Microsoft-managed encryption. + local (key) authentication on, Microsoft-managed encryption, no + managed identity, and unrestricted outbound access. Args: sdk_account: `azure.mgmt.cognitiveservices.models.Account`. @@ -81,8 +85,72 @@ class AIServices(AzureService): getattr(encryption, "key_source", None) or MICROSOFT_MANAGED_KEY_SOURCE ), encryption_key_name=getattr(key_vault_properties, "key_name", None), + private_endpoint_connection_statuses=AIServices._private_endpoint_statuses( + properties + ), + identity_type=getattr(getattr(sdk_account, "identity", None), "type", None), + restrict_outbound_network_access=bool( + getattr(properties, "restrict_outbound_network_access", False) + ), ) + @staticmethod + def _private_endpoint_statuses(properties) -> list[str]: + """Get the connection status of each private endpoint on an account. + + Args: + properties: `azure.mgmt.cognitiveservices.models.AccountProperties`, + or `None`. + + Returns: + One status (`Approved`, `Pending`, `Rejected`) per connection that + reports one. + """ + statuses = [] + for connection in ( + getattr(properties, "private_endpoint_connections", None) or [] + ): + state = getattr( + getattr(connection, "properties", None), + "private_link_service_connection_state", + None, + ) + status = getattr(state, "status", None) + if status: + statuses.append(status) + return statuses + + @staticmethod + def _get_diagnostic_settings(subscription: str, account_id: str) -> Optional[list]: + """Get the Azure Monitor diagnostic settings of one account. + + Args: + subscription: Subscription ID that holds the account. + account_id: Account resource ID. + + Returns: + The account's `DiagnosticSetting` items, or `None` when they cannot + be read. + """ + try: + # Imported here because the Monitor client is built from the global + # provider at import time; importing this module must not need it. + from prowler.providers.azure.services.monitor.monitor_client import ( + monitor_client, + ) + + return monitor_client.diagnostic_settings_with_uri( + subscription, + account_id, + monitor_client.clients[subscription], + raise_errors=True, + ) + except Exception as error: + logger.error( + f"Subscription ID: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return None + class Account(BaseModel): """Azure AI services account.""" @@ -95,3 +163,9 @@ class Account(BaseModel): disable_local_auth: bool encryption_key_source: str encryption_key_name: Optional[str] = None + private_endpoint_connection_statuses: list[str] = [] + identity_type: Optional[str] = None + restrict_outbound_network_access: bool = False + # Monitor DiagnosticSetting dataclasses. Left untyped: Pydantic would + # re-validate them and reject log entries whose category is None. + monitor_diagnostic_settings: Optional[list] = None diff --git a/prowler/providers/azure/services/monitor/monitor_service.py b/prowler/providers/azure/services/monitor/monitor_service.py index 07d41d58f2..1bfa63c76f 100644 --- a/prowler/providers/azure/services/monitor/monitor_service.py +++ b/prowler/providers/azure/services/monitor/monitor_service.py @@ -33,7 +33,23 @@ class Monitor(AzureService): ) return diagnostics_settings - def diagnostic_settings_with_uri(self, subscription, uri, client): + def diagnostic_settings_with_uri( + self, subscription, uri, client, raise_errors: bool = False + ): + """Get the diagnostic settings of one resource. + + Args: + subscription: Subscription ID, used for logging. + uri: Resource URI to read the diagnostic settings from. + client: `MonitorManagementClient` of the subscription. + raise_errors: When `True`, API errors are re-raised unlogged so the + caller can handle and log them and tell a failure from "no + settings configured". + + Returns: + The resource's `DiagnosticSetting` items. On error with + `raise_errors=False`, the items collected so far (usually `[]`). + """ diagnostics_settings = [] try: settings = client.diagnostic_settings.list(resource_uri=uri) @@ -60,6 +76,8 @@ class Monitor(AzureService): ) ) except Exception as error: + if raise_errors: + raise logger.error( f"Subscription ID: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) diff --git a/tests/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled_test.py b/tests/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled_test.py new file mode 100644 index 0000000000..8723d093d2 --- /dev/null +++ b/tests/providers/azure/services/aiservices/aiservices_account_diagnostic_logging_enabled/aiservices_account_diagnostic_logging_enabled_test.py @@ -0,0 +1,142 @@ +from unittest import mock + +from prowler.providers.azure.services.aiservices.aiservices_service import Account +from prowler.providers.azure.services.monitor.monitor_service import ( + DiagnosticSetting, + LogSettings, +) +from tests.providers.azure.azure_fixtures import ( + AZURE_SUBSCRIPTION_DISPLAY, + AZURE_SUBSCRIPTION_ID, + AZURE_SUBSCRIPTION_NAME, + RESOURCE_GROUP, + set_mocked_azure_provider, +) + +CHECK = "aiservices_account_diagnostic_logging_enabled" +CHECK_PATH = f"prowler.providers.azure.services.aiservices.{CHECK}.{CHECK}" +ACCOUNT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.CognitiveServices/accounts/openai1" +PREFIX = f"AI services account openai1 (kind OpenAI) from subscription {AZURE_SUBSCRIPTION_DISPLAY}" + + +def build_setting(*logs: LogSettings) -> DiagnosticSetting: + return DiagnosticSetting( + id=f"{ACCOUNT_ID}/providers/microsoft.insights/diagnosticSettings/diag1", + storage_account_id=None, + storage_account_name=None, + logs=list(logs), + name="diag1", + workspace_id="/subscriptions/x/resourceGroups/rg/providers/Microsoft.OperationalInsights/workspaces/law1", + ) + + +def build_account(settings) -> Account: + return Account( + id=ACCOUNT_ID, + name="openai1", + location="eastus", + kind="OpenAI", + public_network_access=False, + disable_local_auth=True, + encryption_key_source="Microsoft.CognitiveServices", + monitor_diagnostic_settings=settings, + ) + + +def run_check(accounts): + aiservices_client = mock.MagicMock() + aiservices_client.subscriptions = {AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME} + aiservices_client.accounts = accounts + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_azure_provider(), + ), + mock.patch(f"{CHECK_PATH}.aiservices_client", new=aiservices_client), + ): + from prowler.providers.azure.services.aiservices.aiservices_account_diagnostic_logging_enabled.aiservices_account_diagnostic_logging_enabled import ( + aiservices_account_diagnostic_logging_enabled, + ) + + return aiservices_account_diagnostic_logging_enabled().execute() + + +def single(settings): + return run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(settings)}}) + + +class Test_aiservices_account_diagnostic_logging_enabled: + def test_no_resources(self): + result = run_check({}) + assert len(result) == 0 + + def test_settings_unreadable_is_manual(self): + result = single(None) + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert ( + result[0].status_extended + == f"{PREFIX} diagnostic settings could not be read; grant Microsoft.Insights/diagnosticSettings/read (included in Reader) to evaluate audit logging." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_no_diagnostic_settings(self): + result = single([]) + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"{PREFIX} has no diagnostic setting that sends audit logs." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_audit_category_disabled(self): + result = single([build_setting(LogSettings("Audit", None, False))]) + assert result[0].status == "FAIL" + + def test_only_request_response_category(self): + result = single([build_setting(LogSettings("RequestResponse", None, True))]) + assert result[0].status == "FAIL" + + def test_audit_category_enabled(self): + result = single( + [ + build_setting( + LogSettings("RequestResponse", None, True), + LogSettings("Audit", None, True), + ) + ] + ) + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"{PREFIX} has a diagnostic setting that sends audit logs." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_audit_category_group_enabled(self): + result = single([build_setting(LogSettings(None, "audit", True))]) + assert result[0].status == "PASS" + + def test_all_logs_category_group_enabled(self): + result = single([build_setting(LogSettings(None, "allLogs", True))]) + assert result[0].status == "PASS" + + def test_category_group_case_insensitive(self): + result = single([build_setting(LogSettings(None, "alllogs", True))]) + assert result[0].status == "PASS" + + def test_second_setting_sends_audit_logs(self): + result = single( + [ + build_setting(), + build_setting(LogSettings(None, "audit", True)), + ] + ) + assert result[0].status == "PASS" diff --git a/tests/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled_test.py b/tests/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled_test.py new file mode 100644 index 0000000000..d39bc69730 --- /dev/null +++ b/tests/providers/azure/services/aiservices/aiservices_account_managed_identity_enabled/aiservices_account_managed_identity_enabled_test.py @@ -0,0 +1,99 @@ +from typing import Optional +from unittest import mock + +from prowler.providers.azure.services.aiservices.aiservices_service import Account +from tests.providers.azure.azure_fixtures import ( + AZURE_SUBSCRIPTION_DISPLAY, + AZURE_SUBSCRIPTION_ID, + AZURE_SUBSCRIPTION_NAME, + RESOURCE_GROUP, + set_mocked_azure_provider, +) + +CHECK = "aiservices_account_managed_identity_enabled" +CHECK_PATH = f"prowler.providers.azure.services.aiservices.{CHECK}.{CHECK}" +ACCOUNT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.CognitiveServices/accounts/openai1" +PREFIX = f"AI services account openai1 (kind OpenAI) from subscription {AZURE_SUBSCRIPTION_DISPLAY}" + + +def build_account(identity_type: Optional[str]) -> Account: + return Account( + id=ACCOUNT_ID, + name="openai1", + location="eastus", + kind="OpenAI", + public_network_access=False, + disable_local_auth=True, + encryption_key_source="Microsoft.CognitiveServices", + identity_type=identity_type, + ) + + +def run_check(accounts): + aiservices_client = mock.MagicMock() + aiservices_client.subscriptions = {AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME} + aiservices_client.accounts = accounts + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_azure_provider(), + ), + mock.patch(f"{CHECK_PATH}.aiservices_client", new=aiservices_client), + ): + from prowler.providers.azure.services.aiservices.aiservices_account_managed_identity_enabled.aiservices_account_managed_identity_enabled import ( + aiservices_account_managed_identity_enabled, + ) + + return aiservices_account_managed_identity_enabled().execute() + + +class Test_aiservices_account_managed_identity_enabled: + def test_no_resources(self): + result = run_check({}) + assert len(result) == 0 + + def test_no_identity(self): + result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(None)}}) + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == f"{PREFIX} has no managed identity." + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_identity_type_none_string(self): + result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account("None")}}) + assert result[0].status == "FAIL" + assert result[0].status_extended == f"{PREFIX} has no managed identity." + + def test_identity_type_none_string_lowercase(self): + result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account("none")}}) + assert result[0].status == "FAIL" + + def test_system_assigned_identity(self): + result = run_check( + {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account("SystemAssigned")}} + ) + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"{PREFIX} has a managed identity (SystemAssigned)." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_system_and_user_assigned_identity(self): + result = run_check( + { + AZURE_SUBSCRIPTION_ID: { + ACCOUNT_ID: build_account("SystemAssigned, UserAssigned") + } + } + ) + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"{PREFIX} has a managed identity (SystemAssigned, UserAssigned)." + ) diff --git a/tests/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted_test.py b/tests/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted_test.py new file mode 100644 index 0000000000..e9be286cb2 --- /dev/null +++ b/tests/providers/azure/services/aiservices/aiservices_account_outbound_network_restricted/aiservices_account_outbound_network_restricted_test.py @@ -0,0 +1,76 @@ +from unittest import mock + +from prowler.providers.azure.services.aiservices.aiservices_service import Account +from tests.providers.azure.azure_fixtures import ( + AZURE_SUBSCRIPTION_DISPLAY, + AZURE_SUBSCRIPTION_ID, + AZURE_SUBSCRIPTION_NAME, + RESOURCE_GROUP, + set_mocked_azure_provider, +) + +CHECK = "aiservices_account_outbound_network_restricted" +CHECK_PATH = f"prowler.providers.azure.services.aiservices.{CHECK}.{CHECK}" +ACCOUNT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.CognitiveServices/accounts/openai1" +PREFIX = f"AI services account openai1 (kind OpenAI) from subscription {AZURE_SUBSCRIPTION_DISPLAY}" + + +def build_account(restrict_outbound_network_access: bool) -> Account: + return Account( + id=ACCOUNT_ID, + name="openai1", + location="eastus", + kind="OpenAI", + public_network_access=False, + disable_local_auth=True, + encryption_key_source="Microsoft.CognitiveServices", + restrict_outbound_network_access=restrict_outbound_network_access, + ) + + +def run_check(accounts): + aiservices_client = mock.MagicMock() + aiservices_client.subscriptions = {AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME} + aiservices_client.accounts = accounts + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_azure_provider(), + ), + mock.patch(f"{CHECK_PATH}.aiservices_client", new=aiservices_client), + ): + from prowler.providers.azure.services.aiservices.aiservices_account_outbound_network_restricted.aiservices_account_outbound_network_restricted import ( + aiservices_account_outbound_network_restricted, + ) + + return aiservices_account_outbound_network_restricted().execute() + + +class Test_aiservices_account_outbound_network_restricted: + def test_no_resources(self): + result = run_check({}) + assert len(result) == 0 + + def test_outbound_unrestricted(self): + result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(False)}}) + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"{PREFIX} allows unrestricted outbound network access." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_outbound_restricted(self): + result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(True)}}) + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"{PREFIX} restricts outbound network access to an allowed FQDN list." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID diff --git a/tests/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint_test.py b/tests/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint_test.py new file mode 100644 index 0000000000..ebaec879d9 --- /dev/null +++ b/tests/providers/azure/services/aiservices/aiservices_account_uses_private_endpoint/aiservices_account_uses_private_endpoint_test.py @@ -0,0 +1,102 @@ +from unittest import mock + +from prowler.providers.azure.services.aiservices.aiservices_service import Account +from tests.providers.azure.azure_fixtures import ( + AZURE_SUBSCRIPTION_DISPLAY, + AZURE_SUBSCRIPTION_ID, + AZURE_SUBSCRIPTION_NAME, + RESOURCE_GROUP, + set_mocked_azure_provider, +) + +CHECK = "aiservices_account_uses_private_endpoint" +CHECK_PATH = f"prowler.providers.azure.services.aiservices.{CHECK}.{CHECK}" +ACCOUNT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.CognitiveServices/accounts/openai1" +PREFIX = f"AI services account openai1 (kind OpenAI) from subscription {AZURE_SUBSCRIPTION_DISPLAY}" + + +def build_account(statuses: list[str]) -> Account: + return Account( + id=ACCOUNT_ID, + name="openai1", + location="eastus", + kind="OpenAI", + public_network_access=False, + disable_local_auth=True, + encryption_key_source="Microsoft.CognitiveServices", + private_endpoint_connection_statuses=statuses, + ) + + +def run_check(accounts): + aiservices_client = mock.MagicMock() + aiservices_client.subscriptions = {AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME} + aiservices_client.accounts = accounts + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_azure_provider(), + ), + mock.patch(f"{CHECK_PATH}.aiservices_client", new=aiservices_client), + ): + from prowler.providers.azure.services.aiservices.aiservices_account_uses_private_endpoint.aiservices_account_uses_private_endpoint import ( + aiservices_account_uses_private_endpoint, + ) + + return aiservices_account_uses_private_endpoint().execute() + + +class Test_aiservices_account_uses_private_endpoint: + def test_no_resources(self): + result = run_check({}) + assert len(result) == 0 + + def test_no_private_endpoints(self): + result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account([])}}) + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"{PREFIX} has no approved private endpoint connection." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_only_pending_and_rejected_private_endpoints(self): + result = run_check( + { + AZURE_SUBSCRIPTION_ID: { + ACCOUNT_ID: build_account(["Pending", "Rejected"]) + } + } + ) + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"{PREFIX} has no approved private endpoint connection." + ) + + def test_approved_private_endpoint(self): + result = run_check( + { + AZURE_SUBSCRIPTION_ID: { + ACCOUNT_ID: build_account(["Pending", "Approved"]) + } + } + ) + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"{PREFIX} has an approved private endpoint connection." + ) + assert result[0].resource_id == ACCOUNT_ID + assert result[0].resource_name == "openai1" + assert result[0].subscription == AZURE_SUBSCRIPTION_ID + + def test_approved_status_is_case_insensitive(self): + result = run_check( + {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(["approved"])}} + ) + assert result[0].status == "PASS" diff --git a/tests/providers/azure/services/aiservices/aiservices_service_test.py b/tests/providers/azure/services/aiservices/aiservices_service_test.py index 2e872ec297..78e60b171e 100644 --- a/tests/providers/azure/services/aiservices/aiservices_service_test.py +++ b/tests/providers/azure/services/aiservices/aiservices_service_test.py @@ -1,5 +1,8 @@ +import sys from unittest.mock import MagicMock, patch +import pytest + from prowler.providers.azure.services.aiservices.aiservices_service import ( Account, AIServices, @@ -13,6 +16,19 @@ from tests.providers.azure.azure_fixtures import ( ACCOUNT_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.CognitiveServices/accounts/openai1" SERVICE_PATH = "prowler.providers.azure.services.aiservices.aiservices_service" +MONITOR_CLIENT_MODULE = "prowler.providers.azure.services.monitor.monitor_client" + + +@pytest.fixture(autouse=True) +def monitor_client(): + """Replace the Monitor client module so no real client is built.""" + client = MagicMock() + client.clients = {AZURE_SUBSCRIPTION_ID: MagicMock()} + client.diagnostic_settings_with_uri.return_value = [] + with patch.dict( + sys.modules, {MONITOR_CLIENT_MODULE: MagicMock(monitor_client=client)} + ): + yield client def mock_get_accounts(_): @@ -37,21 +53,35 @@ def build_sdk_account( disable_local_auth=True, encryption=None, properties_none=False, + identity=None, + private_endpoint_connections=None, + restrict_outbound_network_access=None, ): account = MagicMock() account.id = ACCOUNT_ID account.name = "openai1" account.location = "eastus" account.kind = "OpenAI" + account.identity = identity if properties_none: account.properties = None return account account.properties.public_network_access = public_network_access account.properties.disable_local_auth = disable_local_auth account.properties.encryption = encryption + account.properties.private_endpoint_connections = private_endpoint_connections + account.properties.restrict_outbound_network_access = ( + restrict_outbound_network_access + ) return account +def build_sdk_private_endpoint_connection(status): + connection = MagicMock() + connection.properties.private_link_service_connection_state.status = status + return connection + + def build_service(mock_client, resource_groups=None): with patch(f"{SERVICE_PATH}.AIServices._get_accounts", return_value={}): aiservices = AIServices(set_mocked_azure_provider()) @@ -177,6 +207,81 @@ class Test_AIServices_get_accounts: assert account.public_network_access is True assert account.disable_local_auth is False assert account.encryption_key_source == "Microsoft.CognitiveServices" + assert account.private_endpoint_connection_statuses == [] + assert account.restrict_outbound_network_access is False + + def test_new_fields_default_when_unset(self): + mock_client = MagicMock() + mock_client.accounts.list.return_value = [build_sdk_account()] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.private_endpoint_connection_statuses == [] + assert account.identity_type is None + assert account.restrict_outbound_network_access is False + + def test_private_endpoint_statuses_mapped(self): + broken = MagicMock() + broken.properties = None + mock_client = MagicMock() + mock_client.accounts.list.return_value = [ + build_sdk_account( + private_endpoint_connections=[ + build_sdk_private_endpoint_connection("Approved"), + build_sdk_private_endpoint_connection("Pending"), + broken, + ] + ) + ] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.private_endpoint_connection_statuses == ["Approved", "Pending"] + + def test_private_endpoint_status_enum_mapped_to_value(self): + from azure.mgmt.cognitiveservices.models import ( + PrivateEndpointServiceConnectionStatus, + ) + + mock_client = MagicMock() + mock_client.accounts.list.return_value = [ + build_sdk_account( + private_endpoint_connections=[ + build_sdk_private_endpoint_connection( + PrivateEndpointServiceConnectionStatus.APPROVED + ) + ] + ) + ] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.private_endpoint_connection_statuses == ["Approved"] + + def test_identity_type_mapped(self): + identity = MagicMock() + identity.type = "SystemAssigned" + mock_client = MagicMock() + mock_client.accounts.list.return_value = [build_sdk_account(identity=identity)] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.identity_type == "SystemAssigned" + + def test_restrict_outbound_network_access_mapped(self): + mock_client = MagicMock() + mock_client.accounts.list.return_value = [ + build_sdk_account(restrict_outbound_network_access=True) + ] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.restrict_outbound_network_access is True def test_list_failure_leaves_subscription_empty(self): mock_client = MagicMock() @@ -198,3 +303,54 @@ class Test_AIServices_get_accounts: result = aiservices._get_accounts() assert list(result[AZURE_SUBSCRIPTION_ID]) == [ACCOUNT_ID] + + def test_diagnostic_settings_fetched_per_account(self, monitor_client): + setting = MagicMock() + monitor_client.diagnostic_settings_with_uri.return_value = [setting] + mock_client = MagicMock() + mock_client.accounts.list.return_value = [build_sdk_account()] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + monitor_client.diagnostic_settings_with_uri.assert_called_once_with( + AZURE_SUBSCRIPTION_ID, + ACCOUNT_ID, + monitor_client.clients[AZURE_SUBSCRIPTION_ID], + raise_errors=True, + ) + assert account.monitor_diagnostic_settings == [setting] + + def test_diagnostic_settings_none_configured_is_empty_list(self): + mock_client = MagicMock() + mock_client.accounts.list.return_value = [build_sdk_account()] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.monitor_diagnostic_settings == [] + + def test_diagnostic_settings_failure_is_none(self, monitor_client): + monitor_client.diagnostic_settings_with_uri.side_effect = Exception("boom") + mock_client = MagicMock() + mock_client.accounts.list.return_value = [build_sdk_account()] + aiservices = build_service(mock_client) + + result = aiservices._get_accounts() + + assert ( + result[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID].monitor_diagnostic_settings + is None + ) + + def test_diagnostic_settings_no_monitor_client_for_subscription_is_none( + self, monitor_client + ): + monitor_client.clients = {} + mock_client = MagicMock() + mock_client.accounts.list.return_value = [build_sdk_account()] + aiservices = build_service(mock_client) + + account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID] + + assert account.monitor_diagnostic_settings is None diff --git a/tests/providers/azure/services/monitor/monitor_service_test.py b/tests/providers/azure/services/monitor/monitor_service_test.py index 2bc7904c51..40246cc86f 100644 --- a/tests/providers/azure/services/monitor/monitor_service_test.py +++ b/tests/providers/azure/services/monitor/monitor_service_test.py @@ -1,6 +1,7 @@ from unittest import mock -from unittest.mock import patch +from unittest.mock import MagicMock, patch +import pytest from azure.mgmt.monitor.models import AlertRuleAnyOfOrLeafCondition from prowler.providers.azure.services.monitor.lib.monitor_alerts import check_alert_rule @@ -112,6 +113,27 @@ class Test_Monitor_Service: == "/subscriptions/1234a5-123a-123a-123a-1234567890ab/resourceGroups/rg/providers/Microsoft.Storage/storageAccounts/storageaccountname" ) + def test_diagnostic_settings_with_uri_error_returns_empty_by_default(self): + monitor = Monitor(set_mocked_azure_provider()) + client = MagicMock() + client.diagnostic_settings.list.side_effect = Exception("boom") + + result = monitor.diagnostic_settings_with_uri( + AZURE_SUBSCRIPTION_ID, "uri", client + ) + + assert result == [] + + def test_diagnostic_settings_with_uri_error_raises_when_requested(self): + monitor = Monitor(set_mocked_azure_provider()) + client = MagicMock() + client.diagnostic_settings.list.side_effect = Exception("boom") + + with pytest.raises(Exception, match="boom"): + monitor.diagnostic_settings_with_uri( + AZURE_SUBSCRIPTION_ID, "uri", client, raise_errors=True + ) + def test__monitor_alerts_false__(self): alert_rule = AlertRule( id="id",