feat(azure): add aiservices network ACL and prompt shield checks (#12997)

Co-authored-by: Rubén De la Torre Vico <98956809+puchy22@users.noreply.github.com>

prowler/changelog.d/azure-aiservices-content-filter-checks.added.md

@@ -0,0 +1 @@
+`aiservices_account_network_acls_default_action_deny` and `aiservices_deployment_content_filter_prompt_shield_enabled` checks for Azure provider

prowler/providers/azure/services/aiservices/aiservices_account_network_acls_default_action_deny/aiservices_account_network_acls_default_action_deny.metadata.json

@@ -0,0 +1,40 @@
+{
+  "Provider": "azure",
+  "CheckID": "aiservices_account_network_acls_default_action_deny",
+  "CheckTitle": "AI services account network access rules deny public traffic by default",
+  "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**) that keep **public network access** enabled can set the network ACL default action (`networkAcls.defaultAction`) to `Deny`, so the public endpoint only accepts traffic from allowed virtual networks and IP ranges. Accounts with public network access `Disabled` also pass.",
+  "Risk": "With the default action set to `Allow`, the public endpoint accepts traffic from any network:\n- **Confidentiality**: prompts, completions, and grounding data are reachable by anyone holding a leaked key or token\n- **Integrity**: unauthorized model calls and deployment changes from any source\n- **Availability**: quota exhaustion and cost abuse from scanning and automated traffic",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://learn.microsoft.com/en-us/azure/ai-services/cognitive-services-virtual-networks",
+    "https://learn.microsoft.com/en-us/security/benchmark/azure/baselines/cognitive-services-security-baseline"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "az resource update --ids <account_resource_id> --set properties.networkAcls.defaultAction=Deny",
+      "NativeIaC": "```bicep\n// Deny public traffic by default on an Azure AI services account\nresource account 'Microsoft.CognitiveServices/accounts@2024-10-01' = {\n  name: '<example_resource_name>'\n  location: '<location>'\n  kind: 'OpenAI'\n  sku: { name: 'S0' }\n  properties: {\n    customSubDomainName: '<example_resource_name>'\n    networkAcls: {\n      defaultAction: 'Deny' // CRITICAL: only allowed networks reach the public endpoint\n      ipRules: [\n        { value: '<allowed_ip_range>' }\n      ]\n    }\n  }\n}\n```",
+      "Other": "1. In the Azure portal, open the Azure AI services or Azure OpenAI resource\n2. Go to Resource Management > Networking\n3. Under Firewalls and virtual networks, select Selected Networks and Private Endpoints\n4. Add the required virtual networks and IP ranges\n5. Click Save",
+      "Terraform": "```hcl\nresource \"azurerm_cognitive_account\" \"<example_resource_name>\" {\n  name                  = \"<example_resource_name>\"\n  location              = \"<location>\"\n  resource_group_name   = \"<resource_group>\"\n  kind                  = \"OpenAI\"\n  sku_name              = \"S0\"\n  custom_subdomain_name = \"<example_resource_name>\"\n\n  network_acls {\n    default_action = \"Deny\" # CRITICAL: only allowed networks reach the public endpoint\n    ip_rules       = [\"<allowed_ip_range>\"]\n  }\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Prefer disabling **public network access** and using **Private Link**. Where the public endpoint must stay on, set the network ACL default action to `Deny` and allow only the virtual networks and IP ranges each workload needs.",
+      "Url": "https://hub.prowler.com/check/aiservices_account_network_acls_default_action_deny"
+    }
+  },
+  "Categories": [
+    "internet-exposed",
+    "gen-ai"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "aiservices_account_public_network_access_disabled"
+  ],
+  "Notes": "Mirrors the Azure Policy built-in \"Azure AI Services resources should restrict network access\", which flags accounts where public network access is not Disabled and the network ACL default action is not Deny. Add the needed IP ranges and virtual networks before switching to Deny to avoid losing connectivity."
+}

prowler/providers/azure/services/aiservices/aiservices_account_network_acls_default_action_deny/aiservices_account_network_acls_default_action_deny.py

@@ -0,0 +1,42 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.aiservices.aiservices_client import (
+    aiservices_client,
+)
+
+
+class aiservices_account_network_acls_default_action_deny(Check):
+    """AI services account denies public network access by default."""
+
+    def execute(self) -> list[Check_Report_Azure]:
+        """Evaluate the network ACL default action on every AI services account.
+
+        PASS when public network access is disabled, or when the network ACL
+        default action is `Deny` so only selected networks reach the endpoint.
+
+        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 not account.public_network_access:
+                    report.status = "PASS"
+                    report.status_extended = (
+                        f"{prefix} has public network access disabled."
+                    )
+                elif account.network_acls_default_action.lower() == "deny":
+                    report.status = "PASS"
+                    report.status_extended = f"{prefix} restricts public network access to selected networks."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"{prefix} allows public network access from all networks."
+                    )
+                findings.append(report)
+        return findings

prowler/providers/azure/services/aiservices/aiservices_deployment_content_filter_prompt_shield_enabled/aiservices_deployment_content_filter_prompt_shield_enabled.metadata.json

@@ -0,0 +1,38 @@
+{
+  "Provider": "azure",
+  "CheckID": "aiservices_deployment_content_filter_prompt_shield_enabled",
+  "CheckTitle": "AI services model deployment blocks jailbreak attempts with Prompt Shields",
+  "CheckType": [],
+  "ServiceName": "aiservices",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "microsoft.cognitiveservices/accounts/deployments",
+  "ResourceGroup": "ai_ml",
+  "Description": "**Azure AI services model deployments** (including **Azure OpenAI** and **Azure AI Foundry**) apply a **content filter policy** (`raiPolicyName`). **Prompt Shields** for direct attacks is the policy's `Jailbreak` filter; it should be enabled and blocking on prompts so user prompt injection attempts are rejected before the model runs.",
+  "Risk": "Without a blocking jailbreak filter, crafted prompts can override the system prompt and safety training:\n- **Integrity**: the model ignores its instructions and produces harmful or policy-violating output\n- **Confidentiality**: attackers extract system prompts, grounding data, or tool results\n- **Excessive agency**: agents with tools act on attacker instructions",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/content-filter-prompt-shields",
+    "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/how-to/content-filters",
+    "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/default-safety-policies"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "az resource update --ids <deployment_resource_id> --set properties.raiPolicyName=Microsoft.DefaultV2",
+      "NativeIaC": "```bicep\n// Deploy a model with a content filter policy that blocks jailbreak attempts\nresource deployment 'Microsoft.CognitiveServices/accounts/deployments@2024-10-01' = {\n  parent: account\n  name: '<deployment_name>'\n  sku: { name: 'Standard', capacity: <capacity> }\n  properties: {\n    model: { format: 'OpenAI', name: '<model_name>', version: '<model_version>' }\n    raiPolicyName: 'Microsoft.DefaultV2' // CRITICAL: policy includes a blocking Jailbreak filter on prompts\n  }\n}\n```",
+      "Other": "1. In the Azure AI Foundry portal, open the resource and go to Guardrails + controls > Content filters\n2. Create or edit a content filter and, under Input filter, turn on Prompt Shields for jailbreak attacks with the Annotate and block action\n3. Go to Deployments, open the deployment, and assign that content filter\n4. Save the deployment",
+      "Terraform": "```hcl\nresource \"azurerm_cognitive_deployment\" \"<example_resource_name>\" {\n  name                 = \"<deployment_name>\"\n  cognitive_account_id = \"<account_resource_id>\"\n  rai_policy_name      = \"Microsoft.DefaultV2\" # CRITICAL: policy includes a blocking Jailbreak filter on prompts\n\n  model {\n    format  = \"OpenAI\"\n    name    = \"<model_name>\"\n    version = \"<model_version>\"\n  }\n\n  sku {\n    name     = \"Standard\"\n    capacity = <capacity>\n  }\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Assign every model deployment an explicit content filter policy whose **Prompt Shields** jailbreak filter is enabled and set to **block** on prompts, such as `Microsoft.DefaultV2` or a custom policy. Avoid relying on the account default, which cannot be verified from configuration.",
+      "Url": "https://hub.prowler.com/check/aiservices_deployment_content_filter_prompt_shield_enabled"
+    }
+  },
+  "Categories": [
+    "gen-ai"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Deployments with an empty raiPolicyName use the account default content filter policy. Azure Resource Manager does not report which policy that is, and the legacy Microsoft.Default policy has no Jailbreak filter, so these deployments are reported as MANUAL. Deployments whose capabilities show no text generation (embeddings, transcription, image generation) are skipped. A custom policy must list its own Jailbreak filter; filters inherited from basePolicyName are not evaluated."
+}

prowler/providers/azure/services/aiservices/aiservices_deployment_content_filter_prompt_shield_enabled/aiservices_deployment_content_filter_prompt_shield_enabled.py

@@ -0,0 +1,117 @@
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.aiservices.aiservices_client import (
+    aiservices_client,
+)
+from prowler.providers.azure.services.aiservices.aiservices_service import (
+    Deployment,
+    RaiPolicy,
+)
+
+# Deployment capabilities that take text prompts, where Prompt Shields applies.
+TEXT_GENERATION_CAPABILITIES = {
+    "chatcompletion",
+    "completion",
+    "responses",
+    "assistants",
+}
+# Capabilities of models that take no text prompts.
+NON_TEXT_CAPABILITIES = {"embeddings", "imagegenerations"}
+
+
+class aiservices_deployment_content_filter_prompt_shield_enabled(Check):
+    """AI services deployment blocks jailbreak attempts with Prompt Shields."""
+
+    def execute(self) -> list[Check_Report_Azure]:
+        """Evaluate the content filter policy of every model deployment.
+
+        PASS when the deployment's policy has the `Jailbreak` filter enabled
+        and blocking on prompts. MANUAL when the deployment uses the account
+        default policy (empty `raiPolicyName`), because Azure does not say
+        which policy that is, or when the policy cannot be read.
+
+        Deployments whose capabilities show only non-text models (embeddings,
+        image generation) are skipped. All other deployments are evaluated,
+        including those that report no or unrecognized capabilities.
+
+        Returns:
+            One finding per evaluated deployment, plus one MANUAL finding per account
+            whose deployments cannot be read.
+        """
+        findings = []
+        for subscription_id, accounts in aiservices_client.accounts.items():
+            subscription_name = aiservices_client.subscriptions.get(
+                subscription_id, subscription_id
+            )
+            subscription = f"subscription {subscription_name} ({subscription_id})"
+            for account in accounts.values():
+                if account.deployments is None:
+                    report = Check_Report_Azure(
+                        metadata=self.metadata(), resource=account
+                    )
+                    report.subscription = subscription_id
+                    report.status = "MANUAL"
+                    report.status_extended = f"Deployments of AI services account {account.name} (kind {account.kind}) from {subscription} could not be read."
+                    findings.append(report)
+                    continue
+                # ARM resource names are case-insensitive.
+                rai_policies = (
+                    None
+                    if account.rai_policies is None
+                    else {
+                        name.lower(): policy
+                        for name, policy in account.rai_policies.items()
+                    }
+                )
+                for deployment in account.deployments:
+                    if not _takes_text_prompts(deployment):
+                        continue
+                    report = Check_Report_Azure(
+                        metadata=self.metadata(), resource=deployment
+                    )
+                    report.subscription = subscription_id
+                    prefix = f"AI services deployment {deployment.name} (model {deployment.model_name}) in account {account.name} from {subscription}"
+                    policy_name = deployment.rai_policy_name
+                    if not policy_name:
+                        report.status = "MANUAL"
+                        report.status_extended = f"{prefix} uses the account default content filter policy, which Azure does not identify. Confirm that it blocks jailbreak attempts with Prompt Shields."
+                    elif rai_policies is None:
+                        report.status = "MANUAL"
+                        report.status_extended = f"{prefix} uses content filter policy {policy_name}, which could not be read."
+                    elif policy_name.lower() not in rai_policies:
+                        report.status = "MANUAL"
+                        report.status_extended = f"{prefix} uses content filter policy {policy_name}, which was not found on the account."
+                    elif _blocks_jailbreak(rai_policies[policy_name.lower()]):
+                        report.status = "PASS"
+                        report.status_extended = f"{prefix} uses content filter policy {policy_name}, which blocks jailbreak attempts with Prompt Shields."
+                    else:
+                        report.status = "FAIL"
+                        report.status_extended = f"{prefix} uses content filter policy {policy_name}, which does not block jailbreak attempts with Prompt Shields."
+                    findings.append(report)
+        return findings
+
+
+def _takes_text_prompts(deployment: Deployment) -> bool:
+    """Return whether a deployment may serve a model that takes text prompts.
+
+    Capabilities are an open map, so only a known non-text capability with no
+    text-generation capability rules a deployment out.
+    """
+    enabled = {
+        name.lower()
+        for name, value in deployment.capabilities.items()
+        if str(value).lower() == "true"
+    }
+    return bool(enabled & TEXT_GENERATION_CAPABILITIES) or not (
+        enabled & NON_TEXT_CAPABILITIES
+    )
+
+
+def _blocks_jailbreak(policy: RaiPolicy) -> bool:
+    """Return whether a policy blocks jailbreak attempts on prompts."""
+    return any(
+        content_filter.name.lower() == "jailbreak"
+        and content_filter.source.lower() == "prompt"
+        and content_filter.enabled
+        and content_filter.blocking
+        for content_filter in policy.content_filters
+    )

prowler/providers/azure/services/aiservices/aiservices_service.py

@@ -48,6 +48,12 @@ class AIServices(AzureService):
                     account.monitor_diagnostic_settings = self._get_diagnostic_settings(
                         subscription, account.id
                     )
+                    account.deployments = self._get_deployments(
+                        subscription, client, account
+                    )
+                    account.rai_policies = self._get_rai_policies(
+                        subscription, client, account
+                    )
                     accounts[subscription][account.id] = account
                 except Exception as error:
                     logger.error(
@@ -61,7 +67,8 @@ class AIServices(AzureService):
 
         Unset properties follow Azure defaults: public network access on,
         local (key) authentication on, Microsoft-managed encryption, no
-        managed identity, and unrestricted outbound access.
+        managed identity, unrestricted outbound access, and a network ACL
+        default action of `Allow`.
 
         Args:
             sdk_account: `azure.mgmt.cognitiveservices.models.Account`.
@@ -72,6 +79,9 @@ class AIServices(AzureService):
         properties = sdk_account.properties
         encryption = getattr(properties, "encryption", None)
         key_vault_properties = getattr(encryption, "key_vault_properties", None)
+        default_action = getattr(
+            getattr(properties, "network_acls", None), "default_action", None
+        )
         return Account(
             id=sdk_account.id,
             name=sdk_account.name,
@@ -92,8 +102,111 @@ class AIServices(AzureService):
             restrict_outbound_network_access=bool(
                 getattr(properties, "restrict_outbound_network_access", False)
             ),
+            # NetworkRuleAction is a str enum; keep the plain value.
+            network_acls_default_action=(
+                getattr(default_action, "value", default_action) or "Allow"
+            ),
         )
 
+    @staticmethod
+    def _get_deployments(
+        subscription: str, client, account: "Account"
+    ) -> Optional[list["Deployment"]]:
+        """Get the model deployments of one account.
+
+        Args:
+            subscription: Subscription ID that holds the account.
+            client: `CognitiveServicesManagementClient` for the subscription.
+            account: The account whose deployments are listed.
+
+        Returns:
+            The account's deployments, or `None` when they cannot be read.
+        """
+        try:
+            deployments = []
+            for sdk_deployment in client.deployments.list(
+                resource_group_name=_resource_group(account.id),
+                account_name=account.name,
+            ):
+                properties = getattr(sdk_deployment, "properties", None)
+                capabilities = getattr(properties, "capabilities", None)
+                deployments.append(
+                    Deployment(
+                        id=sdk_deployment.id,
+                        name=sdk_deployment.name,
+                        location=account.location,
+                        model_name=getattr(
+                            getattr(properties, "model", None), "name", None
+                        )
+                        or "",
+                        # An empty name means the deployment uses the account
+                        # default policy, which ARM does not identify.
+                        rai_policy_name=getattr(properties, "rai_policy_name", None)
+                        or "",
+                        capabilities=(
+                            capabilities if isinstance(capabilities, dict) else {}
+                        ),
+                    )
+                )
+            return deployments
+        except Exception as error:
+            logger.error(
+                f"Subscription ID: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return None
+
+    @staticmethod
+    def _get_rai_policies(
+        subscription: str, client, account: "Account"
+    ) -> Optional[dict[str, "RaiPolicy"]]:
+        """Get the content filter (RAI) policies of one account.
+
+        Uses `list` because `get` returns 404 for system-managed policies such
+        as `Microsoft.Default`.
+
+        Args:
+            subscription: Subscription ID that holds the account.
+            client: `CognitiveServicesManagementClient` for the subscription.
+            account: The account whose policies are listed.
+
+        Returns:
+            Policies keyed by name, or `None` when they cannot be read.
+        """
+        try:
+            policies = {}
+            for sdk_policy in client.rai_policies.list(
+                resource_group_name=_resource_group(account.id),
+                account_name=account.name,
+            ):
+                content_filters = []
+                for sdk_filter in (
+                    getattr(
+                        getattr(sdk_policy, "properties", None),
+                        "content_filters",
+                        None,
+                    )
+                    or []
+                ):
+                    source = getattr(sdk_filter, "source", None)
+                    content_filters.append(
+                        ContentFilter(
+                            name=getattr(sdk_filter, "name", None) or "",
+                            # RaiPolicyContentSource is a str enum; keep the plain value.
+                            source=getattr(source, "value", source) or "",
+                            enabled=bool(getattr(sdk_filter, "enabled", False)),
+                            blocking=bool(getattr(sdk_filter, "blocking", False)),
+                        )
+                    )
+                policies[sdk_policy.name] = RaiPolicy(
+                    name=sdk_policy.name, content_filters=content_filters
+                )
+            return policies
+        except Exception as error:
+            logger.error(
+                f"Subscription ID: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return None
+
     @staticmethod
     def _private_endpoint_statuses(properties) -> list[str]:
         """Get the connection status of each private endpoint on an account.
@@ -152,6 +265,39 @@ class AIServices(AzureService):
             return None
 
 
+def _resource_group(resource_id: str) -> str:
+    """Get the resource group name from an ARM resource ID."""
+    return resource_id.split("/")[4]
+
+
+class ContentFilter(BaseModel):
+    """One content filter of a content filter (RAI) policy."""
+
+    name: str
+    source: str
+    enabled: bool
+    blocking: bool
+
+
+class RaiPolicy(BaseModel):
+    """Content filter (RAI) policy of an AI services account."""
+
+    name: str
+    content_filters: list[ContentFilter] = []
+
+
+class Deployment(BaseModel):
+    """Model deployment of an AI services account."""
+
+    id: str
+    name: str
+    location: str
+    model_name: str = ""
+    rai_policy_name: str = ""
+    # Model capabilities, for example {"chatCompletion": "true"}.
+    capabilities: dict[str, str] = {}
+
+
 class Account(BaseModel):
     """Azure AI services account."""
 
@@ -166,6 +312,10 @@ class Account(BaseModel):
     private_endpoint_connection_statuses: list[str] = []
     identity_type: Optional[str] = None
     restrict_outbound_network_access: bool = False
+    network_acls_default_action: str = "Allow"
+    # None when the list call fails, so checks can report MANUAL.
+    deployments: Optional[list[Deployment]] = None
+    rai_policies: Optional[dict[str, RaiPolicy]] = None
     # Monitor DiagnosticSetting dataclasses. Left untyped: Pydantic would
     # re-validate them and reject log entries whose category is None.
     monitor_diagnostic_settings: Optional[list] = None

tests/providers/azure/services/aiservices/aiservices_account_network_acls_default_action_deny/aiservices_account_network_acls_default_action_deny_test.py

@@ -0,0 +1,100 @@
+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_network_acls_default_action_deny"
+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(public_network_access: bool, default_action: str) -> Account:
+    return Account(
+        id=ACCOUNT_ID,
+        name="openai1",
+        location="eastus",
+        kind="OpenAI",
+        public_network_access=public_network_access,
+        disable_local_auth=True,
+        encryption_key_source="Microsoft.CognitiveServices",
+        network_acls_default_action=default_action,
+    )
+
+
+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_network_acls_default_action_deny.aiservices_account_network_acls_default_action_deny import (
+            aiservices_account_network_acls_default_action_deny,
+        )
+
+        return aiservices_account_network_acls_default_action_deny().execute()
+
+
+class Test_aiservices_account_network_acls_default_action_deny:
+    def test_no_resources(self):
+        result = run_check({})
+        assert len(result) == 0
+
+    def test_public_access_from_all_networks(self):
+        result = run_check(
+            {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(True, "Allow")}}
+        )
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert (
+            result[0].status_extended
+            == f"{PREFIX} allows public network access from all networks."
+        )
+        assert result[0].resource_id == ACCOUNT_ID
+        assert result[0].resource_name == "openai1"
+        assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+
+    def test_default_action_deny(self):
+        result = run_check(
+            {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(True, "Deny")}}
+        )
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == f"{PREFIX} restricts public network access to selected networks."
+        )
+        assert result[0].resource_id == ACCOUNT_ID
+        assert result[0].resource_name == "openai1"
+        assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+
+    def test_public_network_access_disabled(self):
+        result = run_check(
+            {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(False, "Allow")}}
+        )
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended == f"{PREFIX} has public network access disabled."
+        )
+        assert result[0].resource_id == ACCOUNT_ID
+        assert result[0].resource_name == "openai1"
+        assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+
+    def test_default_action_deny_case_insensitive(self):
+        result = run_check(
+            {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(True, "deny")}}
+        )
+        assert len(result) == 1
+        assert result[0].status == "PASS"

tests/providers/azure/services/aiservices/aiservices_deployment_content_filter_prompt_shield_enabled/aiservices_deployment_content_filter_prompt_shield_enabled_test.py

@@ -0,0 +1,240 @@
+from typing import Optional
+from unittest import mock
+
+from prowler.providers.azure.services.aiservices.aiservices_service import (
+    Account,
+    ContentFilter,
+    Deployment,
+    RaiPolicy,
+)
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_DISPLAY,
+    AZURE_SUBSCRIPTION_ID,
+    AZURE_SUBSCRIPTION_NAME,
+    RESOURCE_GROUP,
+    set_mocked_azure_provider,
+)
+
+CHECK = "aiservices_deployment_content_filter_prompt_shield_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"
+DEPLOYMENT_ID = f"{ACCOUNT_ID}/deployments/gpt-4o"
+PREFIX = f"AI services deployment gpt-4o (model gpt-4o) in account openai1 from subscription {AZURE_SUBSCRIPTION_DISPLAY}"
+
+
+def jailbreak_filter(enabled=True, blocking=True, source="Prompt") -> ContentFilter:
+    return ContentFilter(
+        name="Jailbreak", source=source, enabled=enabled, blocking=blocking
+    )
+
+
+def build_account(
+    rai_policy_name: str = "Microsoft.DefaultV2",
+    rai_policies: Optional[dict] = None,
+    deployments: Optional[list] = "default",
+    capabilities: Optional[dict] = None,
+) -> Account:
+    if deployments == "default":
+        deployments = [
+            Deployment(
+                id=DEPLOYMENT_ID,
+                name="gpt-4o",
+                location="eastus",
+                model_name="gpt-4o",
+                rai_policy_name=rai_policy_name,
+                capabilities=(
+                    {"chatCompletion": "true"} if capabilities is None else capabilities
+                ),
+            )
+        ]
+    return Account(
+        id=ACCOUNT_ID,
+        name="openai1",
+        location="eastus",
+        kind="OpenAI",
+        public_network_access=False,
+        disable_local_auth=True,
+        encryption_key_source="Microsoft.CognitiveServices",
+        deployments=deployments,
+        rai_policies=rai_policies,
+    )
+
+
+def policies(*content_filters, name="Microsoft.DefaultV2") -> dict:
+    return {name: RaiPolicy(name=name, content_filters=list(content_filters))}
+
+
+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_deployment_content_filter_prompt_shield_enabled.aiservices_deployment_content_filter_prompt_shield_enabled import (
+            aiservices_deployment_content_filter_prompt_shield_enabled,
+        )
+
+        return aiservices_deployment_content_filter_prompt_shield_enabled().execute()
+
+
+def single(account):
+    result = run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: account}})
+    assert len(result) == 1
+    assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+    return result[0]
+
+
+class Test_aiservices_deployment_content_filter_prompt_shield_enabled:
+    def test_no_resources(self):
+        assert run_check({}) == []
+
+    def test_account_without_deployments(self):
+        assert (
+            run_check(
+                {AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: build_account(deployments=[])}}
+            )
+            == []
+        )
+
+    def test_jailbreak_filter_blocking(self):
+        finding = single(build_account(rai_policies=policies(jailbreak_filter())))
+        assert finding.status == "PASS"
+        assert (
+            finding.status_extended
+            == f"{PREFIX} uses content filter policy Microsoft.DefaultV2, which blocks jailbreak attempts with Prompt Shields."
+        )
+        assert finding.resource_id == DEPLOYMENT_ID
+        assert finding.resource_name == "gpt-4o"
+        assert finding.location == "eastus"
+
+    def test_jailbreak_filter_match_is_case_insensitive(self):
+        content_filter = ContentFilter(
+            name="jailbreak", source="prompt", enabled=True, blocking=True
+        )
+        finding = single(build_account(rai_policies=policies(content_filter)))
+        assert finding.status == "PASS"
+
+    def test_jailbreak_filter_missing(self):
+        finding = single(
+            build_account(
+                rai_policy_name="Microsoft.Default",
+                rai_policies=policies(
+                    ContentFilter(
+                        name="Hate", source="Prompt", enabled=True, blocking=True
+                    ),
+                    name="Microsoft.Default",
+                ),
+            )
+        )
+        assert finding.status == "FAIL"
+        assert (
+            finding.status_extended
+            == f"{PREFIX} uses content filter policy Microsoft.Default, which does not block jailbreak attempts with Prompt Shields."
+        )
+        assert finding.resource_id == DEPLOYMENT_ID
+
+    def test_jailbreak_filter_disabled(self):
+        finding = single(
+            build_account(rai_policies=policies(jailbreak_filter(enabled=False)))
+        )
+        assert finding.status == "FAIL"
+
+    def test_jailbreak_filter_annotate_only(self):
+        finding = single(
+            build_account(rai_policies=policies(jailbreak_filter(blocking=False)))
+        )
+        assert finding.status == "FAIL"
+
+    def test_jailbreak_filter_on_completion_only(self):
+        finding = single(
+            build_account(rai_policies=policies(jailbreak_filter(source="Completion")))
+        )
+        assert finding.status == "FAIL"
+
+    def test_policy_name_match_is_case_insensitive(self):
+        finding = single(
+            build_account(
+                rai_policy_name="microsoft.defaultv2",
+                rai_policies=policies(jailbreak_filter()),
+            )
+        )
+        assert finding.status == "PASS"
+
+    def test_embeddings_deployment_is_skipped(self):
+        account = build_account(
+            rai_policy_name="",
+            rai_policies=policies(jailbreak_filter()),
+            capabilities={"embeddings": "true", "area": "EUR"},
+        )
+        assert run_check({AZURE_SUBSCRIPTION_ID: {ACCOUNT_ID: account}}) == []
+
+    def test_unrecognized_capabilities_is_evaluated(self):
+        finding = single(
+            build_account(
+                rai_policies=policies(jailbreak_filter()),
+                capabilities={"realtime": "true", "area": "EUR"},
+            )
+        )
+        assert finding.status == "PASS"
+
+    def test_embeddings_with_text_capability_is_evaluated(self):
+        finding = single(
+            build_account(
+                rai_policies=policies(jailbreak_filter()),
+                capabilities={"embeddings": "true", "chatCompletion": "true"},
+            )
+        )
+        assert finding.status == "PASS"
+
+    def test_missing_capabilities_is_evaluated(self):
+        finding = single(
+            build_account(rai_policies=policies(jailbreak_filter()), capabilities={})
+        )
+        assert finding.status == "PASS"
+
+    def test_default_policy_is_manual(self):
+        finding = single(
+            build_account(rai_policy_name="", rai_policies=policies(jailbreak_filter()))
+        )
+        assert finding.status == "MANUAL"
+        assert (
+            finding.status_extended
+            == f"{PREFIX} uses the account default content filter policy, which Azure does not identify. Confirm that it blocks jailbreak attempts with Prompt Shields."
+        )
+        assert finding.resource_id == DEPLOYMENT_ID
+
+    def test_policies_unreadable_is_manual(self):
+        finding = single(build_account(rai_policies=None))
+        assert finding.status == "MANUAL"
+        assert (
+            finding.status_extended
+            == f"{PREFIX} uses content filter policy Microsoft.DefaultV2, which could not be read."
+        )
+
+    def test_policy_not_found_is_manual(self):
+        finding = single(
+            build_account(
+                rai_policy_name="custom-policy",
+                rai_policies=policies(jailbreak_filter()),
+            )
+        )
+        assert finding.status == "MANUAL"
+        assert (
+            finding.status_extended
+            == f"{PREFIX} uses content filter policy custom-policy, which was not found on the account."
+        )
+
+    def test_deployments_unreadable_is_manual(self):
+        finding = single(build_account(deployments=None, rai_policies={}))
+        assert finding.status == "MANUAL"
+        assert (
+            finding.status_extended
+            == f"Deployments of AI services account openai1 (kind OpenAI) from subscription {AZURE_SUBSCRIPTION_DISPLAY} could not be read."
+        )
+        assert finding.resource_id == ACCOUNT_ID
+        assert finding.resource_name == "openai1"

tests/providers/azure/services/aiservices/aiservices_service_test.py

@@ -56,6 +56,7 @@ def build_sdk_account(
     identity=None,
     private_endpoint_connections=None,
     restrict_outbound_network_access=None,
+    network_acls=None,
 ):
     account = MagicMock()
     account.id = ACCOUNT_ID
@@ -73,6 +74,7 @@ def build_sdk_account(
     account.properties.restrict_outbound_network_access = (
         restrict_outbound_network_access
     )
+    account.properties.network_acls = network_acls
     return account
 
 
@@ -82,6 +84,34 @@ def build_sdk_private_endpoint_connection(status):
     return connection
 
 
+def build_sdk_deployment(
+    name="gpt-4o", rai_policy_name=None, model_name="gpt-4o", capabilities=None
+):
+    deployment = MagicMock()
+    deployment.id = f"{ACCOUNT_ID}/deployments/{name}"
+    deployment.name = name
+    deployment.properties.rai_policy_name = rai_policy_name
+    deployment.properties.model.name = model_name
+    deployment.properties.capabilities = capabilities
+    return deployment
+
+
+def build_sdk_rai_policy(name, content_filters):
+    policy = MagicMock()
+    policy.name = name
+    policy.properties.content_filters = content_filters
+    return policy
+
+
+def build_sdk_content_filter(name, source, enabled=True, blocking=True):
+    content_filter = MagicMock()
+    content_filter.name = name
+    content_filter.source = source
+    content_filter.enabled = enabled
+    content_filter.blocking = blocking
+    return content_filter
+
+
 def build_service(mock_client, resource_groups=None):
     with patch(f"{SERVICE_PATH}.AIServices._get_accounts", return_value={}):
         aiservices = AIServices(set_mocked_azure_provider())
@@ -209,6 +239,7 @@ class Test_AIServices_get_accounts:
         assert account.encryption_key_source == "Microsoft.CognitiveServices"
         assert account.private_endpoint_connection_statuses == []
         assert account.restrict_outbound_network_access is False
+        assert account.network_acls_default_action == "Allow"
 
     def test_new_fields_default_when_unset(self):
         mock_client = MagicMock()
@@ -220,6 +251,7 @@ class Test_AIServices_get_accounts:
         assert account.private_endpoint_connection_statuses == []
         assert account.identity_type is None
         assert account.restrict_outbound_network_access is False
+        assert account.network_acls_default_action == "Allow"
 
     def test_private_endpoint_statuses_mapped(self):
         broken = MagicMock()
@@ -283,6 +315,125 @@ class Test_AIServices_get_accounts:
 
         assert account.restrict_outbound_network_access is True
 
+    def test_network_acls_default_action_enum_mapped_to_value(self):
+        from azure.mgmt.cognitiveservices.models import NetworkRuleAction
+
+        network_acls = MagicMock()
+        network_acls.default_action = NetworkRuleAction.DENY
+        mock_client = MagicMock()
+        mock_client.accounts.list.return_value = [
+            build_sdk_account(network_acls=network_acls)
+        ]
+        aiservices = build_service(mock_client)
+
+        account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID]
+
+        assert account.network_acls_default_action == "Deny"
+
+    def test_network_acls_default_action_none_treated_as_allow(self):
+        network_acls = MagicMock()
+        network_acls.default_action = None
+        mock_client = MagicMock()
+        mock_client.accounts.list.return_value = [
+            build_sdk_account(network_acls=network_acls)
+        ]
+        aiservices = build_service(mock_client)
+
+        account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID]
+
+        assert account.network_acls_default_action == "Allow"
+
+    def test_deployments_and_rai_policies_mapped(self):
+        from azure.mgmt.cognitiveservices.models import RaiPolicyContentSource
+
+        mock_client = MagicMock()
+        mock_client.accounts.list.return_value = [build_sdk_account()]
+        mock_client.deployments.list.return_value = [
+            build_sdk_deployment(
+                rai_policy_name="Microsoft.DefaultV2",
+                capabilities={"chatCompletion": "true", "area": "EUR"},
+            ),
+            build_sdk_deployment(name="gpt-4o-mini", rai_policy_name=""),
+        ]
+        mock_client.rai_policies.list.return_value = [
+            build_sdk_rai_policy(
+                "Microsoft.DefaultV2",
+                [
+                    build_sdk_content_filter(
+                        "Jailbreak", RaiPolicyContentSource.PROMPT
+                    ),
+                    build_sdk_content_filter("Hate", "Completion", blocking=False),
+                ],
+            )
+        ]
+        aiservices = build_service(mock_client)
+
+        account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID]
+
+        mock_client.deployments.list.assert_called_once_with(
+            resource_group_name=RESOURCE_GROUP, account_name="openai1"
+        )
+        mock_client.rai_policies.list.assert_called_once_with(
+            resource_group_name=RESOURCE_GROUP, account_name="openai1"
+        )
+        assert [d.name for d in account.deployments] == ["gpt-4o", "gpt-4o-mini"]
+        assert account.deployments[0].id == f"{ACCOUNT_ID}/deployments/gpt-4o"
+        assert account.deployments[0].location == "eastus"
+        assert account.deployments[0].model_name == "gpt-4o"
+        assert account.deployments[0].rai_policy_name == "Microsoft.DefaultV2"
+        assert account.deployments[1].rai_policy_name == ""
+        assert account.deployments[0].capabilities == {
+            "chatCompletion": "true",
+            "area": "EUR",
+        }
+        assert account.deployments[1].capabilities == {}
+        policy = account.rai_policies["Microsoft.DefaultV2"]
+        assert policy.content_filters[0].name == "Jailbreak"
+        assert policy.content_filters[0].source == "Prompt"
+        assert policy.content_filters[0].enabled is True
+        assert policy.content_filters[0].blocking is True
+        assert policy.content_filters[1].blocking is False
+
+    def test_deployment_without_properties_uses_empty_policy_name(self):
+        deployment = build_sdk_deployment()
+        deployment.properties = None
+        mock_client = MagicMock()
+        mock_client.accounts.list.return_value = [build_sdk_account()]
+        mock_client.deployments.list.return_value = [deployment]
+        mock_client.rai_policies.list.return_value = []
+        aiservices = build_service(mock_client)
+
+        account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID]
+
+        assert account.deployments[0].rai_policy_name == ""
+        assert account.deployments[0].model_name == ""
+        assert account.deployments[0].capabilities == {}
+        assert account.rai_policies == {}
+
+    def test_deployments_failure_is_none(self):
+        mock_client = MagicMock()
+        mock_client.accounts.list.return_value = [build_sdk_account()]
+        mock_client.deployments.list.side_effect = Exception("boom")
+        mock_client.rai_policies.list.return_value = []
+        aiservices = build_service(mock_client)
+
+        account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID]
+
+        assert account.deployments is None
+        assert account.rai_policies == {}
+
+    def test_rai_policies_failure_is_none(self):
+        mock_client = MagicMock()
+        mock_client.accounts.list.return_value = [build_sdk_account()]
+        mock_client.deployments.list.return_value = [build_sdk_deployment()]
+        mock_client.rai_policies.list.side_effect = Exception("boom")
+        aiservices = build_service(mock_client)
+
+        account = aiservices._get_accounts()[AZURE_SUBSCRIPTION_ID][ACCOUNT_ID]
+
+        assert len(account.deployments) == 1
+        assert account.rai_policies is None
+
     def test_list_failure_leaves_subscription_empty(self):
         mock_client = MagicMock()
         mock_client.accounts.list.side_effect = Exception("boom")