feat(azure): Add APIM threat detection for LLM jacking attacks (#8571)

Co-authored-by: Rubén De la Torre Vico <ruben@prowler.com>
2026-07-23 20:42:02 +00:00 · 2025-08-28 11:42:07 +02:00
parent c69187f484
commit f484b83f15
22 changed files with 1399 additions and 4 deletions
@@ -83,6 +83,9 @@ The following list includes all the Azure checks with configurable variables tha
 | `vm_sufficient_daily_backup_retention_period`                 | `vm_backup_min_daily_retention_days`             | Integer         |
 | `vm_desired_sku_size`                                         | `desired_vm_sku_sizes`                           | List of Strings |
 | `defender_attack_path_notifications_properly_configured`      | `defender_attack_path_minimal_risk_level`        | String          |
+| `apim_threat_detection_llm_jacking`                           | `apim_threat_detection_llm_jacking_threshold`    | Float           |
+| `apim_threat_detection_llm_jacking`                           | `apim_threat_detection_llm_jacking_minutes`      | Integer         |
+| `apim_threat_detection_llm_jacking`                           | `apim_threat_detection_llm_jacking_actions`      | List of Strings |


 ## GCP
@@ -494,6 +497,48 @@ azure:
      "Standard_DS3_v2",
      "Standard_D4s_v3",
    ]
+  # Azure VM Backup Configuration
+  # azure.vm_sufficient_daily_backup_retention_period
+  vm_backup_min_daily_retention_days: 7
+
+  # Azure API Management Threat Detection Configuration
+  # azure.apim_threat_detection_llm_jacking
+  apim_threat_detection_llm_jacking_threshold: 0.1
+  apim_threat_detection_llm_jacking_minutes: 1440
+  apim_threat_detection_llm_jacking_actions:
+    [
+      # OpenAI API endpoints
+      "ImageGenerations_Create",
+      "ChatCompletions_Create",
+      "Completions_Create",
+      "Embeddings_Create",
+      "FineTuning_Jobs_Create",
+      "Models_List",
+
+      # Azure OpenAI endpoints
+      "Deployments_List",
+      "Deployments_Get",
+      "Deployments_Create",
+      "Deployments_Delete",
+
+      # Anthropic endpoints
+      "Messages_Create",
+      "Claude_Create",
+
+      # Google AI endpoints
+      "GenerateContent",
+      "GenerateText",
+      "GenerateImage",
+
+      # Meta AI endpoints
+      "Llama_Create",
+      "CodeLlama_Create",
+
+      # Other LLM endpoints
+      "Gemini_Generate",
+      "Claude_Generate",
+      "Llama_Generate"
+    ]

 # GCP Configuration
 gcp:
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.1.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.

 [[package]]
 name = "about-time"
@@ -394,6 +394,24 @@ cryptography = ">=2.1.4"
 isodate = ">=0.6.1"
 typing-extensions = ">=4.0.1"

+[[package]]
+name = "azure-mgmt-apimanagement"
+version = "5.0.0"
+description = "Microsoft Azure API Management Client Library for Python"
+optional = false
+python-versions = ">=3.8"
+groups = ["main"]
+files = [
+    {file = "azure_mgmt_apimanagement-5.0.0-py3-none-any.whl", hash = "sha256:b88c42a392333b60722fb86f15d092dfc19a8d67510dccd15c217381dff4e6ec"},
+    {file = "azure_mgmt_apimanagement-5.0.0.tar.gz", hash = "sha256:0ab7fe17e70fe3154cd840ff47d19d7a4610217003eaa7c21acf3511a6e57999"},
+]
+
+[package.dependencies]
+azure-common = ">=1.1"
+azure-mgmt-core = ">=1.3.2"
+isodate = ">=0.6.1"
+typing-extensions = ">=4.6.0"
+
 [[package]]
 name = "azure-mgmt-applicationinsights"
 version = "4.1.0"
@@ -551,6 +569,23 @@ azure-mgmt-core = ">=1.3.2"
 isodate = ">=0.6.1"
 typing-extensions = ">=4.6.0"

+[[package]]
+name = "azure-mgmt-loganalytics"
+version = "12.0.0"
+description = "Microsoft Azure Log Analytics Management Client Library for Python"
+optional = false
+python-versions = "*"
+groups = ["main"]
+files = [
+    {file = "azure-mgmt-loganalytics-12.0.0.zip", hash = "sha256:da128a7e0291be7fa2063848df92a9180cf5c16d42adc09d2bc2efd711536bfb"},
+    {file = "azure_mgmt_loganalytics-12.0.0-py2.py3-none-any.whl", hash = "sha256:75ac1d47dd81179905c40765be8834643d8994acff31056ddc1863017f3faa02"},
+]
+
+[package.dependencies]
+azure-common = ">=1.1,<2.0"
+azure-mgmt-core = ">=1.2.0,<2.0.0"
+msrest = ">=0.6.21"
+
 [[package]]
 name = "azure-mgmt-monitor"
 version = "6.0.2"
@@ -761,6 +796,23 @@ azure-mgmt-core = ">=1.3.2"
 isodate = ">=0.6.1"
 typing-extensions = ">=4.6.0"

+[[package]]
+name = "azure-monitor-query"
+version = "2.0.0"
+description = "Microsoft Corporation Azure Monitor Query Client Library for Python"
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "azure_monitor_query-2.0.0-py3-none-any.whl", hash = "sha256:8f52d581271d785e12f49cd5aaa144b8910fb843db2373855a7ef94c7fc462ea"},
+    {file = "azure_monitor_query-2.0.0.tar.gz", hash = "sha256:7b05f2fcac4fb67fc9f77a7d4c5d98a0f3099fb73b57c69ec1b080773994671b"},
+]
+
+[package.dependencies]
+azure-core = ">=1.30.0"
+isodate = ">=0.6.1"
+typing-extensions = ">=4.6.0"
+
 [[package]]
 name = "azure-storage-blob"
 version = "12.24.1"
@@ -2352,8 +2404,6 @@ python-versions = "*"
 groups = ["dev"]
 files = [
    {file = "jsonpath-ng-1.7.0.tar.gz", hash = "sha256:f6f5f7fd4e5ff79c785f1573b394043b39849fb2bb47bcead935d12b00beab3c"},
-    {file = "jsonpath_ng-1.7.0-py2-none-any.whl", hash = "sha256:898c93fc173f0c336784a3fa63d7434297544b7198124a68f9a3ef9597b0ae6e"},
-    {file = "jsonpath_ng-1.7.0-py3-none-any.whl", hash = "sha256:f3d7f9e848cba1b6da28c55b1c26ff915dc9e0b1ba7e752a53d6da8d5cbd00b6"},
 ]

 [package.dependencies]
@@ -5841,4 +5891,4 @@ type = ["pytest-mypy"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">3.9.1,<3.13"
-content-hash = "fdd2cbdb6913d0dd8d05030ee41c0d36d3954e473783d43b730ec163e697ec15"
+content-hash = "aea38b0311bfabac00d4bf9ee5d2fa0a7f3e32dd2ee5c5d27eb54c69a80b35e9"
@@ -15,6 +15,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `eks_cluster_deletion_protection_enabled` check for AWS provider [(#8536)](https://github.com/prowler-cloud/prowler/pull/8536)
 - ECS privilege escalation patterns (StartTask and RunTask) for AWS provider [(#8541)](https://github.com/prowler-cloud/prowler/pull/8541)
 - Resource Explorer enumeration v2 API actions in `cloudtrail_threat_detection_enumeration` check [(#8557)](https://github.com/prowler-cloud/prowler/pull/8557)
+- `apim_threat_detection_llm_jacking` check for Azure provider [(#8571)](https://github.com/prowler-cloud/prowler/pull/8571)
 - GCP `--skip-api-check` command line flag [(#8575)](https://github.com/prowler-cloud/prowler/pull/8575)

 ### Changed
@@ -463,6 +463,45 @@ azure:
  # azure.vm_sufficient_daily_backup_retention_period
  vm_backup_min_daily_retention_days: 7

+  # Azure API Management Configuration
+  # azure.apim_threat_detection_llm_jacking
+  apim_threat_detection_llm_jacking_threshold: 0.1
+  apim_threat_detection_llm_jacking_minutes: 1440
+  apim_threat_detection_llm_jacking_actions:
+    [
+      # OpenAI API endpoints
+      "ImageGenerations_Create",
+      "ChatCompletions_Create",
+      "Completions_Create",
+      "Embeddings_Create",
+      "FineTuning_Jobs_Create",
+      "Models_List",
+
+      # Azure OpenAI endpoints
+      "Deployments_List",
+      "Deployments_Get",
+      "Deployments_Create",
+      "Deployments_Delete",
+
+      # Anthropic endpoints
+      "Messages_Create",
+      "Claude_Create",
+
+      # Google AI endpoints
+      "GenerateContent",
+      "GenerateText",
+      "GenerateImage",
+
+      # Meta AI endpoints
+      "Llama_Create",
+      "CodeLlama_Create",
+
+      # Other LLM endpoints
+      "Gemini_Generate",
+      "Claude_Generate",
+      "Llama_Generate"
+    ]
+
 # GCP Configuration
 gcp:
  # GCP Compute Configuration
@@ -25,6 +25,9 @@ class AzureService:
        try:
            if "GraphServiceClient" in str(service):
                clients.update({identity.tenant_domain: service(credentials=session)})
+            elif "LogsQueryClient" in str(service):
+                for display_name, id in identity.subscriptions.items():
+                    clients.update({display_name: service(credential=session)})
            else:
                for display_name, id in identity.subscriptions.items():
                    clients.update(
@@ -0,0 +1,4 @@
+from prowler.providers.azure.services.apim.apim_service import APIM
+from prowler.providers.common.provider import Provider
+
+apim_client = APIM(Provider.get_global_provider())
@@ -0,0 +1,251 @@
+from datetime import datetime, timedelta
+from typing import List, Optional
+
+from azure.mgmt.apimanagement import ApiManagementClient
+from pydantic.v1 import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.providers.azure.azure_provider import AzureProvider
+from prowler.providers.azure.lib.service.service import AzureService
+from prowler.providers.azure.services.logs.loganalytics_client import (
+    loganalytics_client,
+)
+from prowler.providers.azure.services.logs.logsquery_client import logsquery_client
+from prowler.providers.azure.services.monitor.monitor_client import monitor_client
+
+
+class APIMInstance(BaseModel):
+    """APIM Instance model"""
+
+    id: str
+    name: str
+    location: str
+    log_analytics_workspace_id: Optional[str] = None
+
+
+class LogsQueryLogEntry(BaseModel):
+    """Represents a log entry from Azure Log Analytics query results."""
+
+    time_generated: datetime
+    operation_id: str
+    caller_ip_address: str
+    correlation_id: str
+
+
+class APIM(AzureService):
+    def __init__(self, provider: AzureProvider):
+        """Initialize the APIM service client.
+
+        Args:
+            provider: The Azure provider instance containing authentication and client configuration
+        """
+        super().__init__(ApiManagementClient, provider)
+        self.instances = self._get_instances()
+
+    def _get_workspace_customer_id(
+        self, subscription: str, workspace_arm_id: str
+    ) -> Optional[str]:
+        """Get the Customer ID (GUID) for a workspace from its full ARM ID.
+
+        This method extracts the resource group and workspace name from the ARM ID
+        and queries the Log Analytics client to retrieve the customer ID (GUID)
+        needed for workspace-specific queries.
+
+        Args:
+            subscription: The Azure subscription ID
+            workspace_arm_id: The full ARM ID of the Log Analytics workspace
+
+        Returns:
+            The customer ID (GUID) of the workspace if successful, None otherwise
+        """
+        try:
+            resource_group = workspace_arm_id.split("/")[4]
+            workspace_name = workspace_arm_id.split("/")[-1]
+
+            workspace = loganalytics_client.clients[subscription].workspaces.get(
+                resource_group_name=resource_group, workspace_name=workspace_name
+            )
+            return workspace.customer_id
+        except Exception as error:
+            logger.error(
+                f"Failed to get customer ID for workspace {workspace_arm_id}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return None
+
+    def _get_log_analytics_workspace_id(
+        self, instance_id: str, subscription: str
+    ) -> Optional[str]:
+        """Retrieve the Log Analytics workspace ARM ID from an APIM instance's diagnostic settings.
+
+        This method queries the Azure Monitor diagnostic settings for a specific APIM
+        instance to find the configured Log Analytics workspace. It specifically looks
+        for diagnostic settings that have GatewayLogs enabled, which are essential for
+        monitoring APIM API calls and operations.
+
+        Args:
+            instance_id: The ARM ID of the APIM instance
+            subscription: The Azure subscription ID
+
+        Returns:
+            The ARM ID of the Log Analytics workspace if diagnostic settings are found
+            and GatewayLogs are enabled, None otherwise
+        """
+        try:
+            diagnostic_settings = monitor_client.diagnostic_settings_with_uri(
+                subscription, instance_id, monitor_client.clients[subscription]
+            )
+            for setting in diagnostic_settings:
+                if setting.workspace_id and setting.logs:
+                    for log_setting in setting.logs:
+                        if (
+                            log_setting.enabled
+                            and log_setting.category == "GatewayLogs"
+                        ):
+                            logger.info(
+                                f"Found enabled Log Analytics workspace for APIM instance {instance_id} with category {log_setting.category}"
+                            )
+                            return setting.workspace_id
+        except Exception as error:
+            logger.error(
+                f"Failed to get diagnostic settings for {instance_id}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        return None
+
+    def _get_instances(self):
+        """Get all APIM instances and their configured Log Analytics workspace.
+
+        This method iterates through all accessible Azure subscriptions and retrieves
+        all APIM instances within each subscription. For each instance, it also
+        determines the associated Log Analytics workspace by checking diagnostic
+        settings. The method populates the instances dictionary with APIMInstance
+        objects containing all relevant metadata and configuration.
+
+        Returns:
+            A dictionary mapping subscription IDs to lists of APIMInstance objects.
+            Each APIMInstance contains the instance details and its associated
+            Log Analytics workspace ID if configured.
+        """
+        logger.info("APIM - Getting instances...")
+        instances = {}
+
+        for subscription, client in self.clients.items():
+            try:
+                instances.update({subscription: []})
+                apim_instances = client.api_management_service.list()
+
+                for instance in apim_instances:
+                    workspace_id = self._get_log_analytics_workspace_id(
+                        instance.id, subscription
+                    )
+                    instances[subscription].append(
+                        APIMInstance(
+                            id=instance.id,
+                            name=instance.name,
+                            location=instance.location,
+                            log_analytics_workspace_id=workspace_id,
+                        )
+                    )
+            except Exception as error:
+                logger.error(
+                    f"Subscription name: {subscription} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
+        return instances
+
+    def query_logs(
+        self,
+        subscription: str,
+        query: str,
+        timespan: timedelta,
+        workspace_customer_id: str,
+    ) -> List[LogsQueryLogEntry]:
+        """Query a specific Log Analytics workspace using its Customer ID (GUID).
+
+        This method executes Kusto Query Language (KQL) queries against a specific
+        Log Analytics workspace. It's used to retrieve log data for analysis and
+        monitoring purposes. The method handles the response parsing and converts
+        the tabular results into a list of dictionaries for easy consumption.
+
+        Args:
+            subscription: The Azure subscription ID
+            query: The KQL query string to execute
+            timespan: The time range for the query as a timedelta
+            workspace_customer_id: The customer ID (GUID) of the Log Analytics workspace
+
+        Returns:
+            A list of dictionaries where each dictionary represents a row from the
+            query results. The keys are the column names from the query response.
+            Returns an empty list if the query fails or returns no results.
+        """
+        try:
+            response = logsquery_client.clients[subscription].query_workspace(
+                workspace_id=workspace_customer_id,
+                query=query,
+                timespan=timespan,
+            )
+
+            if response.tables:
+                columns = response.tables[0].columns
+                rows = response.tables[0].rows
+                result = []
+
+                for row in rows:
+                    # Create a mapping from Azure column names to our snake_case field names
+                    row_dict = dict(zip(columns, row))
+                    mapped_dict = {
+                        "time_generated": row_dict.get("TimeGenerated"),
+                        "operation_id": row_dict.get("OperationId"),
+                        "caller_ip_address": row_dict.get("CallerIpAddress"),
+                        "correlation_id": row_dict.get("CorrelationId"),
+                    }
+                    result.append(LogsQueryLogEntry(**mapped_dict))
+
+                return result
+
+        except Exception as error:
+            logger.error(
+                f"Failed to query Log Analytics workspace with customer ID {workspace_customer_id}: {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return []
+
+    def get_llm_operations_logs(
+        self, subscription: str, instance: APIMInstance, minutes: int = 1440
+    ) -> List[LogsQueryLogEntry]:
+        """Get LLM-related operations from the APIM instance's specific Log Analytics workspace.
+
+        This method retrieves logs related to Large Language Model (LLM) operations
+        from a specific APIM instance. It queries the GatewayLogs table in the
+        associated Log Analytics workspace to find API calls and operations that
+        may be related to LLM services. The method automatically handles the
+        translation from workspace ARM ID to customer ID for querying.
+
+        Args:
+            subscription: The Azure subscription ID
+            instance: The APIMInstance object containing the instance details
+            minutes: The time range in minutes to look back (default: 1440 = 24 hours)
+
+        Returns:
+            A list of dictionaries containing log entries with fields like
+            time_generated, operation_id, caller_ip_address, and correlation_id.
+            Returns an empty list if no workspace is configured or if the query fails.
+        """
+        if not instance.log_analytics_workspace_id:
+            logger.warning(
+                f"APIM instance {instance.name} has no configured Log Analytics workspace."
+            )
+            return []
+
+        # Translate the workspace ARM ID to the Customer ID (GUID) before querying
+        workspace_customer_id = self._get_workspace_customer_id(
+            subscription, instance.log_analytics_workspace_id
+        )
+        if not workspace_customer_id:
+            return []
+
+        query = f"""
+        ApiManagementGatewayLogs
+        | where _ResourceId has '{instance.id}'
+        | project TimeGenerated, OperationId, CallerIpAddress, CorrelationId
+        """
+        timespan = timedelta(minutes=minutes)
+        return self.query_logs(subscription, query, timespan, workspace_customer_id)
@@ -0,0 +1,34 @@
+{
+  "Provider": "azure",
+  "CheckID": "apim_threat_detection_llm_jacking",
+  "CheckTitle": "Ensure Azure API Management is protected against LLM Jacking attacks",
+  "CheckType": [],
+  "ServiceName": "apim",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "Azure API Management Instance",
+  "Description": "This check analyzes Azure API Management diagnostic logs in Log Analytics to detect potential LLM Jacking attacks by monitoring the frequency of LLM-related operations (ImageGenerations_Create, ChatCompletions_Create, Completions_Create) from individual IP addresses within a configurable time window.",
+  "Risk": "LLM Jacking attacks can lead to unauthorized access to AI models, potential data exfiltration, increased costs, and abuse of AI services. Attackers may use these endpoints to generate content, bypass rate limits, or access premium AI capabilities without proper authorization.",
+  "RelatedUrl": "https://learn.microsoft.com/en-us/azure/api-management/monitor-api-management",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To protect against LLM Jacking attacks: 1. Enable diagnostic logging for APIM instances and send logs to Log Analytics workspace 2. Configure appropriate thresholds for LLM operation frequency monitoring 3. Set up alerts for suspicious activity patterns 4. Implement rate limiting and IP allowlisting for sensitive AI endpoints 5. Regularly review and analyze APIM access logs for anomalies",
+      "Url": "https://learn.microsoft.com/en-us/azure/api-management/monitor-api-management"
+    }
+  },
+  "Categories": [
+    "threat-detection",
+    "monitoring",
+    "logging"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "This check requires: 1. APIM diagnostic logging to be enabled and configured to send logs to Log Analytics workspace 2. Log Analytics workspace ID and key to be configured in the audit configuration 3. Appropriate permissions to query Log Analytics workspace"
+}
@@ -0,0 +1,107 @@
+from typing import List
+
+from prowler.lib.check.models import Check, Check_Report_Azure
+from prowler.providers.azure.services.apim.apim_client import apim_client
+from prowler.providers.azure.services.apim.apim_service import LogsQueryLogEntry
+
+
+class apim_threat_detection_llm_jacking(Check):
+    def execute(self):
+        findings = []
+
+        # Get configuration from audit config with defaults
+        threshold = float(
+            getattr(apim_client, "audit_config", {}).get(
+                "apim_threat_detection_llm_jacking_threshold", 0.1
+            )
+        )
+        threat_detection_minutes = getattr(apim_client, "audit_config", {}).get(
+            "apim_threat_detection_llm_jacking_minutes", 1440
+        )
+        monitored_actions = getattr(apim_client, "audit_config", {}).get(
+            "apim_threat_detection_llm_jacking_actions",
+            [
+                # OpenAI API endpoints
+                "ImageGenerations_Create",
+                "ChatCompletions_Create",
+                "Completions_Create",
+                "Embeddings_Create",
+                "FineTuning_Jobs_Create",
+                "Models_List",
+                # Azure OpenAI endpoints
+                "Deployments_List",
+                "Deployments_Get",
+                "Deployments_Create",
+                "Deployments_Delete",
+                # Anthropic endpoints
+                "Messages_Create",
+                "Claude_Create",
+                # Google AI endpoints
+                "GenerateContent",
+                "GenerateText",
+                "GenerateImage",
+                # Meta AI endpoints
+                "Llama_Create",
+                "CodeLlama_Create",
+                # Other LLM endpoints
+                "Gemini_Generate",
+                "Claude_Generate",
+                "Llama_Generate",
+            ],
+        )
+
+        # 1. Aggregate logs from all APIM instances first
+        all_llm_logs: List[LogsQueryLogEntry] = []
+        for subscription, instances in apim_client.instances.items():
+            for instance in instances:
+                if instance.log_analytics_workspace_id:
+                    logs = apim_client.get_llm_operations_logs(
+                        subscription, instance, threat_detection_minutes
+                    )
+                    all_llm_logs.extend(logs)
+
+            # 2. Perform a single, global analysis on all collected logs
+            potential_llm_jacking_attackers = {}
+            for log in all_llm_logs:
+                operation_name = log.operation_id
+                caller_ip = log.caller_ip_address
+
+                if operation_name in monitored_actions and caller_ip:
+                    # Use IP address as the principal identifier
+                    if caller_ip not in potential_llm_jacking_attackers:
+                        potential_llm_jacking_attackers[caller_ip] = set()
+                    potential_llm_jacking_attackers[caller_ip].add(operation_name)
+
+            # 3. Check each principal against the threshold and report failures
+            found_potential_llm_jacking_attackers = False
+            for (
+                principal_ip,
+                distinct_actions,
+            ) in potential_llm_jacking_attackers.items():
+                action_ratio = round(len(distinct_actions) / len(monitored_actions), 2)
+
+                if action_ratio > threshold:
+                    found_potential_llm_jacking_attackers = True
+                    # Build Identity resource for the report
+                    resource = {
+                        "name": principal_ip,
+                        "id": principal_ip,
+                    }
+                    # Report against the subscription, identifying the offending principal (IP)
+                    report = Check_Report_Azure(self.metadata(), resource=resource)
+                    report.subscription = subscription
+                    report.status = "FAIL"
+                    report.status_extended = f"Potential LLM Jacking attack detected from IP address {principal_ip} with a threshold of {action_ratio}."
+                    findings.append(report)
+
+            # 4. If no threats were found after checking all principals, create a single PASS report
+            if not found_potential_llm_jacking_attackers:
+                report = Check_Report_Azure(self.metadata(), resource={})
+                report.resource_name = "Azure API Management"
+                report.resource_id = "Azure API Management"
+                report.subscription = subscription
+                report.status = "PASS"
+                report.status_extended = f"No potential LLM Jacking attacks detected across all monitored APIM instances in the last {threat_detection_minutes} minutes."
+                findings.append(report)
+
+        return findings
@@ -0,0 +1,4 @@
+from prowler.providers.azure.services.logs.logs_service import LogAnalytics
+from prowler.providers.common.provider import Provider
+
+loganalytics_client = LogAnalytics(Provider.get_global_provider())
@@ -0,0 +1,15 @@
+from azure.mgmt.loganalytics import LogAnalyticsManagementClient
+from azure.monitor.query import LogsQueryClient
+
+from prowler.providers.azure.azure_provider import AzureProvider
+from prowler.providers.azure.lib.service.service import AzureService
+
+
+class LogsQuery(AzureService):
+    def __init__(self, provider: AzureProvider):
+        super().__init__(LogsQueryClient, provider)
+
+
+class LogAnalytics(AzureService):
+    def __init__(self, provider: AzureProvider):
+        super().__init__(LogAnalyticsManagementClient, provider)
@@ -0,0 +1,4 @@
+from prowler.providers.azure.services.logs.logs_service import LogsQuery
+from prowler.providers.common.provider import Provider
+
+logsquery_client = LogsQuery(Provider.get_global_provider())
@@ -56,6 +56,7 @@ class Monitor(AzureService):
                            for log_settings in (getattr(setting, "logs", []) or [])
                        ],
                        storage_account_id=setting.storage_account_id,
+                        workspace_id=getattr(setting, "workspace_id", None),
                    )
                )
        except Exception as error:
@@ -112,6 +113,7 @@ class DiagnosticSetting:
    storage_account_name: str
    logs: List[LogSettings]
    name: str
+    workspace_id: Optional[str] = None


@dataclass
@@ -35,6 +35,9 @@ dependencies = [
  "azure-mgmt-storage==22.1.1",
  "azure-mgmt-subscription==3.1.1",
  "azure-mgmt-web==8.0.0",
+  "azure-mgmt-apimanagement==5.0.0",
+  "azure-mgmt-loganalytics==12.0.0",
+  "azure-monitor-query==2.0.0",
  "azure-storage-blob==12.24.1",
  "boto3==1.39.15",
  "botocore==1.39.15",
@@ -92,6 +92,30 @@ class TestAzureProvider:
                    "Standard_D4s_v3",
                ],
                "defender_attack_path_minimal_risk_level": "High",
+                "apim_threat_detection_llm_jacking_threshold": 0.1,
+                "apim_threat_detection_llm_jacking_minutes": 1440,
+                "apim_threat_detection_llm_jacking_actions": [
+                    "ImageGenerations_Create",
+                    "ChatCompletions_Create",
+                    "Completions_Create",
+                    "Embeddings_Create",
+                    "FineTuning_Jobs_Create",
+                    "Models_List",
+                    "Deployments_List",
+                    "Deployments_Get",
+                    "Deployments_Create",
+                    "Deployments_Delete",
+                    "Messages_Create",
+                    "Claude_Create",
+                    "GenerateContent",
+                    "GenerateText",
+                    "GenerateImage",
+                    "Llama_Create",
+                    "CodeLlama_Create",
+                    "Gemini_Generate",
+                    "Claude_Generate",
+                    "Llama_Generate",
+                ],
            }

    def test_azure_provider_not_auth_methods(self):
@@ -0,0 +1,318 @@
+from datetime import timedelta
+from unittest import TestCase, mock
+from unittest.mock import patch
+
+from azure.mgmt.loganalytics.models import Workspace
+from azure.mgmt.monitor.models import DiagnosticSettingsResource
+from azure.monitor.query import LogsQueryResult
+
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+# Define constants for reusable mock data
+APIM_INSTANCE_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/rg/providers/Microsoft.ApiManagement/service/apim1"
+APIM_INSTANCE_NAME = "apim1"
+LOCATION = "West US"
+RESOURCE_GROUP = "rg"
+WORKSPACE_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourcegroups/rg/providers/microsoft.operationalinsights/workspaces/loganalytics"
+WORKSPACE_CUSTOMER_ID = "12345678-1234-1234-1234-1234567890ab"
+
+
+def mock_apim_get_instances(_):
+    """Mock function to replace APIM._get_instances."""
+    from prowler.providers.azure.services.apim.apim_service import APIMInstance
+
+    return {
+        AZURE_SUBSCRIPTION_ID: [
+            APIMInstance(
+                id=APIM_INSTANCE_ID,
+                name=APIM_INSTANCE_NAME,
+                location=LOCATION,
+                log_analytics_workspace_id=WORKSPACE_ID,
+            )
+        ]
+    }
+
+
+class Test_APIM_Service(TestCase):
+    def test_get_client(self):
+        """Test that the APIM service client is created correctly."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with (
+            patch(
+                "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                return_value={},
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            apim = APIM(set_mocked_azure_provider())
+            self.assertEqual(
+                apim.clients[AZURE_SUBSCRIPTION_ID].__class__.__name__,
+                "ApiManagementClient",
+            )
+
+    def test_get_subscriptions(self):
+        """Test that subscriptions are retrieved correctly."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with (
+            patch(
+                "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                return_value={},
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            apim = APIM(set_mocked_azure_provider())
+            self.assertEqual(apim.subscriptions.__class__.__name__, "dict")
+
+    def test_get_instances(self):
+        """Test that APIM instances are retrieved and parsed correctly."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with (
+            patch(
+                "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                new=mock_apim_get_instances,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            apim = APIM(set_mocked_azure_provider())
+            self.assertEqual(len(apim.instances), 1)
+            self.assertEqual(len(apim.instances[AZURE_SUBSCRIPTION_ID]), 1)
+            instance = apim.instances[AZURE_SUBSCRIPTION_ID][0]
+            self.assertEqual(instance.id, APIM_INSTANCE_ID)
+            self.assertEqual(instance.name, APIM_INSTANCE_NAME)
+            self.assertEqual(instance.location, LOCATION)
+            self.assertEqual(instance.log_analytics_workspace_id, WORKSPACE_ID)
+
+    def test_get_log_analytics_workspace_id_success(self):
+        """Test retrieving a Log Analytics workspace ID successfully."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with patch(
+            "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+            return_value=mock_provider,
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            with (
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                    return_value={},
+                ),
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.monitor_client"
+                ) as mock_monitor_client,
+            ):
+                apim = APIM(set_mocked_azure_provider())
+                mock_log_setting = mock.MagicMock(enabled=True, category="GatewayLogs")
+                mock_setting = DiagnosticSettingsResource(
+                    workspace_id=WORKSPACE_ID, logs=[mock_log_setting]
+                )
+                mock_monitor_client.diagnostic_settings_with_uri.return_value = [
+                    mock_setting
+                ]
+                workspace_id = apim._get_log_analytics_workspace_id(
+                    APIM_INSTANCE_ID, AZURE_SUBSCRIPTION_ID
+                )
+                self.assertEqual(workspace_id, WORKSPACE_ID)
+
+    def test_get_log_analytics_workspace_id_not_enabled(self):
+        """Test that no workspace ID is returned if GatewayLogs are not enabled."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with patch(
+            "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+            return_value=mock_provider,
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            with (
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                    return_value={},
+                ),
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.monitor_client"
+                ) as mock_monitor_client,
+            ):
+                apim = APIM(set_mocked_azure_provider())
+                mock_log_setting = mock.MagicMock(enabled=False, category="GatewayLogs")
+                mock_setting = DiagnosticSettingsResource(
+                    workspace_id=WORKSPACE_ID, logs=[mock_log_setting]
+                )
+                mock_monitor_client.diagnostic_settings_with_uri.return_value = [
+                    mock_setting
+                ]
+                workspace_id = apim._get_log_analytics_workspace_id(
+                    APIM_INSTANCE_ID, AZURE_SUBSCRIPTION_ID
+                )
+                self.assertIsNone(workspace_id)
+
+    def test_get_workspace_customer_id_success(self):
+        """Test retrieving a workspace customer ID successfully."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with patch(
+            "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+            return_value=mock_provider,
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            with (
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                    return_value={},
+                ),
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.loganalytics_client"
+                ) as mock_loganalytics_client,
+            ):
+                apim = APIM(set_mocked_azure_provider())
+                mock_workspace = Workspace(location=LOCATION)
+                # Set customer_id after creation since it's readonly
+                mock_workspace.customer_id = WORKSPACE_CUSTOMER_ID
+
+                # Properly mock the nested client structure
+                mock_client = mock.MagicMock()
+                mock_workspaces = mock.MagicMock()
+                mock_workspaces.get.return_value = mock_workspace
+                mock_client.workspaces = mock_workspaces
+                mock_loganalytics_client.clients = {AZURE_SUBSCRIPTION_ID: mock_client}
+
+                customer_id = apim._get_workspace_customer_id(
+                    AZURE_SUBSCRIPTION_ID, WORKSPACE_ID
+                )
+                self.assertEqual(customer_id, WORKSPACE_CUSTOMER_ID)
+
+    def test_query_logs_success(self):
+        """Test querying logs successfully."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with patch(
+            "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+            return_value=mock_provider,
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            with (
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                    return_value={},
+                ),
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.logsquery_client"
+                ) as mock_logsquery_client,
+            ):
+                apim = APIM(set_mocked_azure_provider())
+                # Create a mock table with the expected structure for LogsQueryLogEntry
+                mock_table = mock.MagicMock()
+                mock_table.columns = [
+                    "TimeGenerated",
+                    "OperationId",
+                    "CallerIpAddress",
+                    "CorrelationId",
+                ]
+                from datetime import datetime
+
+                mock_table.rows = [
+                    [
+                        datetime.fromisoformat("2024-01-01T10:00:00+00:00"),
+                        "test-operation",
+                        "192.168.1.100",
+                        "test-correlation",
+                    ]
+                ]
+
+                mock_response = LogsQueryResult(tables=[mock_table], status="Success")
+
+                # Properly mock the nested client structure
+                mock_client = mock.MagicMock()
+                mock_client.query_workspace.return_value = mock_response
+                mock_logsquery_client.clients = {AZURE_SUBSCRIPTION_ID: mock_client}
+
+                result = apim.query_logs(
+                    AZURE_SUBSCRIPTION_ID,
+                    "query",
+                    timedelta(minutes=60),
+                    WORKSPACE_CUSTOMER_ID,
+                )
+                self.assertEqual(len(result), 1)
+                # The result should be LogsQueryLogEntry objects
+                from datetime import datetime
+
+                self.assertEqual(
+                    result[0].time_generated,
+                    datetime.fromisoformat("2024-01-01T10:00:00+00:00"),
+                )
+                self.assertEqual(result[0].operation_id, "test-operation")
+                self.assertEqual(result[0].caller_ip_address, "192.168.1.100")
+                self.assertEqual(result[0].correlation_id, "test-correlation")
+
+    def test_get_llm_operations_logs_no_workspace_id(self):
+        """Test getting logs when the APIM instance has no workspace configured."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with patch(
+            "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+            return_value=mock_provider,
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            with patch(
+                "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                return_value={},
+            ):
+                apim = APIM(set_mocked_azure_provider())
+                instance = mock.MagicMock(
+                    log_analytics_workspace_id=None, name="test-apim"
+                )
+                result = apim.get_llm_operations_logs(AZURE_SUBSCRIPTION_ID, instance)
+                self.assertEqual(result, [])
+
+    def test_get_llm_operations_logs_success(self):
+        """Test the successful retrieval of LLM operation logs."""
+        mock_provider = mock.MagicMock()
+        mock_provider.identity = mock.MagicMock()
+        with patch(
+            "prowler.providers.azure.azure_provider.Provider.get_global_provider",
+            return_value=mock_provider,
+        ):
+            from prowler.providers.azure.services.apim.apim_service import APIM
+
+            with (
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM._get_instances",
+                    new=mock_apim_get_instances,
+                ),
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM.query_logs",
+                    return_value=[{"log": "data"}],
+                ),
+                patch(
+                    "prowler.providers.azure.services.apim.apim_service.APIM._get_workspace_customer_id",
+                    return_value=WORKSPACE_CUSTOMER_ID,
+                ),
+            ):
+                apim = APIM(set_mocked_azure_provider())
+                instance = apim.instances[AZURE_SUBSCRIPTION_ID][0]
+                result = apim.get_llm_operations_logs(AZURE_SUBSCRIPTION_ID, instance)
+                self.assertEqual(result, [{"log": "data"}])
@@ -0,0 +1,491 @@
+from datetime import datetime
+from unittest import mock
+
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+
+# Create a mock LogsQueryLogEntry class for testing
+class MockLogsQueryLogEntry:
+    def __init__(self, **kwargs):
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+
+def mock_get_llm_operations_logs(subscription, instance, minutes):
+    """Mock LLM operations logs for testing - returns 2 operations"""
+    return [
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:00:00+00:00"),
+            operation_id="ChatCompletions_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-1",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:01:00+00:00"),
+            operation_id="ImageGenerations_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-2",
+        ),
+    ]
+
+
+def mock_get_llm_operations_logs_6_operations(subscription, instance, minutes):
+    """Mock LLM operations logs for testing - returns 6 operations"""
+    return [
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:00:00+00:00"),
+            operation_id="ChatCompletions_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-1",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:01:00+00:00"),
+            operation_id="ImageGenerations_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-2",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:02:00+00:00"),
+            operation_id="Completions_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-3",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:03:00+00:00"),
+            operation_id="Embeddings_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-4",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:04:00+00:00"),
+            operation_id="FineTuning_Jobs_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-5",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:05:00+00:00"),
+            operation_id="Models_List",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-6",
+        ),
+    ]
+
+
+def mock_get_llm_operations_logs_2_operations(subscription, instance, minutes):
+    """Mock LLM operations logs for testing - returns 2 operations"""
+    return [
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:00:00+00:00"),
+            operation_id="ChatCompletions_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-1",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:01:00+00:00"),
+            operation_id="ImageGenerations_Create",
+            caller_ip_address="192.168.1.100",
+            correlation_id="test-correlation-id-2",
+        ),
+    ]
+
+
+def mock_get_llm_operations_logs_attacker(subscription, instance, minutes):
+    """Mock LLM operations logs showing potential attack"""
+    return [
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:00:00+00:00"),
+            operation_id="ChatCompletions_Create",
+            caller_ip_address="10.0.0.50",
+            correlation_id="test-correlation-id-1",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:01:00+00:00"),
+            operation_id="ImageGenerations_Create",
+            caller_ip_address="10.0.0.50",
+            correlation_id="test-correlation-id-2",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:02:00+00:00"),
+            operation_id="Completions_Create",
+            caller_ip_address="10.0.0.50",
+            correlation_id="test-correlation-id-3",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:03:00+00:00"),
+            operation_id="Embeddings_Create",
+            caller_ip_address="10.0.0.50",
+            correlation_id="test-correlation-id-4",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:04:00+00:00"),
+            operation_id="FineTuning_Jobs_Create",
+            caller_ip_address="10.0.0.50",
+            correlation_id="test-correlation-id-5",
+        ),
+        MockLogsQueryLogEntry(
+            time_generated=datetime.fromisoformat("2024-01-01T10:05:00+00:00"),
+            operation_id="Models_List",
+            caller_ip_address="10.0.0.50",
+            correlation_id="test-correlation-id-6",
+        ),
+    ]
+
+
+def mock_get_llm_operations_logs_no_workspace(subscription, instance, minutes):
+    """Mock LLM operations logs for instance without workspace"""
+    return []
+
+
+class Test_apim_threat_detection_llm_jacking:
+    def test_no_apim_instances(self):
+        """Test when there are no APIM instances"""
+        apim_client = mock.MagicMock()
+        apim_client.instances = {}
+        apim_client.audit_config = {
+            "apim_threat_detection_llm_jacking_threshold": 0.1,
+            "apim_threat_detection_llm_jacking_minutes": 1440,
+            "apim_threat_detection_llm_jacking_actions": [
+                "ChatCompletions_Create",
+                "ImageGenerations_Create",
+            ],
+        }
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking.apim_client",
+                new=apim_client,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking import (
+                apim_threat_detection_llm_jacking,
+            )
+
+            check = apim_threat_detection_llm_jacking()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    def test_no_potential_llm_jacking(self):
+        """Test when no potential LLM jacking is detected"""
+        apim_client = mock.MagicMock()
+        apim_client.instances = {
+            AZURE_SUBSCRIPTION_ID: [
+                mock.MagicMock(
+                    id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.ApiManagement/service/test-apim",
+                    name="test-apim",
+                    log_analytics_workspace_id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.OperationalInsights/workspaces/test-workspace",
+                )
+            ]
+        }
+        apim_client.audit_config = {
+            "apim_threat_detection_llm_jacking_threshold": 0.9,
+            "apim_threat_detection_llm_jacking_minutes": 1440,
+            "apim_threat_detection_llm_jacking_actions": [
+                "ChatCompletions_Create",
+                "ImageGenerations_Create",
+                "Completions_Create",
+                "Embeddings_Create",
+                "FineTuning_Jobs_Create",
+                "Models_List",
+                "Deployments_List",
+                "Deployments_Get",
+                "Deployments_Create",
+                "Deployments_Delete",
+                "Messages_Create",
+                "Claude_Create",
+                "GenerateContent",
+                "GenerateText",
+                "GenerateImage",
+                "Llama_Create",
+                "CodeLlama_Create",
+                "Gemini_Generate",
+                "Claude_Generate",
+                "Llama_Generate",
+            ],
+        }
+        apim_client.get_llm_operations_logs = mock_get_llm_operations_logs_6_operations
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking.apim_client",
+                new=apim_client,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking import (
+                apim_threat_detection_llm_jacking,
+            )
+
+            check = apim_threat_detection_llm_jacking()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                "No potential LLM Jacking attacks detected" in result[0].status_extended
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+
+    def test_potential_llm_jacking_detected(self):
+        """Test when potential LLM jacking is detected"""
+        apim_client = mock.MagicMock()
+        apim_client.instances = {
+            AZURE_SUBSCRIPTION_ID: [
+                mock.MagicMock(
+                    id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.ApiManagement/service/test-apim",
+                    name="test-apim",
+                    log_analytics_workspace_id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.OperationalInsights/workspaces/test-workspace",
+                )
+            ]
+        }
+        apim_client.audit_config = {
+            "apim_threat_detection_llm_jacking_threshold": 0.1,
+            "apim_threat_detection_llm_jacking_minutes": 1440,
+            "apim_threat_detection_llm_jacking_actions": [
+                "ChatCompletions_Create",
+                "ImageGenerations_Create",
+                "Completions_Create",
+                "Embeddings_Create",
+                "FineTuning_Jobs_Create",
+                "Models_List",
+            ],
+        }
+        apim_client.get_llm_operations_logs = mock_get_llm_operations_logs_attacker
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking.apim_client",
+                new=apim_client,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking import (
+                apim_threat_detection_llm_jacking,
+            )
+
+            check = apim_threat_detection_llm_jacking()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                "Potential LLM Jacking attack detected from IP address 10.0.0.50"
+                in result[0].status_extended
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource["name"] == "10.0.0.50"
+            assert result[0].resource["id"] == "10.0.0.50"
+
+    def test_higher_threshold_no_detection(self):
+        """Test when threshold is higher and no attack is detected"""
+        apim_client = mock.MagicMock()
+        apim_client.instances = {
+            AZURE_SUBSCRIPTION_ID: [
+                mock.MagicMock(
+                    id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.ApiManagement/service/test-apim",
+                    name="test-apim",
+                    log_analytics_workspace_id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.OperationalInsights/workspaces/test-workspace",
+                )
+            ]
+        }
+        apim_client.audit_config = {
+            "apim_threat_detection_llm_jacking_threshold": 0.9,
+            "apim_threat_detection_llm_jacking_minutes": 1440,
+            "apim_threat_detection_llm_jacking_actions": [
+                "ChatCompletions_Create",
+                "ImageGenerations_Create",
+                "Completions_Create",
+                "Embeddings_Create",
+                "FineTuning_Jobs_Create",
+                "Models_List",
+                "Deployments_List",
+                "Deployments_Get",
+                "Deployments_Create",
+                "Deployments_Delete",
+                "Messages_Create",
+                "Claude_Create",
+                "GenerateContent",
+                "GenerateText",
+                "GenerateImage",
+                "Llama_Create",
+                "CodeLlama_Create",
+                "Gemini_Generate",
+                "Claude_Generate",
+                "Llama_Generate",
+            ],
+        }
+        apim_client.get_llm_operations_logs = mock_get_llm_operations_logs_6_operations
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking.apim_client",
+                new=apim_client,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking import (
+                apim_threat_detection_llm_jacking,
+            )
+
+            check = apim_threat_detection_llm_jacking()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                "No potential LLM Jacking attacks detected" in result[0].status_extended
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+
+    def test_instance_without_workspace(self):
+        """Test when APIM instance has no Log Analytics workspace configured"""
+        apim_client = mock.MagicMock()
+        apim_client.instances = {
+            AZURE_SUBSCRIPTION_ID: [
+                mock.MagicMock(
+                    id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.ApiManagement/service/test-apim",
+                    name="test-apim",
+                    log_analytics_workspace_id=None,
+                )
+            ]
+        }
+        apim_client.audit_config = {
+            "apim_threat_detection_llm_jacking_threshold": 0.9,
+            "apim_threat_detection_llm_jacking_minutes": 1440,
+            "apim_threat_detection_llm_jacking_actions": [
+                "ChatCompletions_Create",
+                "ImageGenerations_Create",
+                "Completions_Create",
+                "Embeddings_Create",
+                "FineTuning_Jobs_Create",
+                "Models_List",
+                "Deployments_List",
+                "Deployments_Get",
+                "Deployments_Create",
+                "Deployments_Delete",
+                "Messages_Create",
+                "Claude_Create",
+                "GenerateContent",
+                "GenerateText",
+                "GenerateImage",
+                "Llama_Create",
+                "CodeLlama_Create",
+                "Gemini_Generate",
+                "Claude_Generate",
+                "Llama_Generate",
+            ],
+        }
+        apim_client.get_llm_operations_logs = mock_get_llm_operations_logs_2_operations
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking.apim_client",
+                new=apim_client,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking import (
+                apim_threat_detection_llm_jacking,
+            )
+
+            check = apim_threat_detection_llm_jacking()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                "No potential LLM Jacking attacks detected" in result[0].status_extended
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+
+    def test_multiple_subscriptions(self):
+        """Test with multiple subscriptions"""
+        apim_client = mock.MagicMock()
+        apim_client.instances = {
+            AZURE_SUBSCRIPTION_ID: [
+                mock.MagicMock(
+                    id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.ApiManagement/service/test-apim",
+                    name="test-apim",
+                    log_analytics_workspace_id="/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.OperationalInsights/workspaces/test-workspace",
+                )
+            ],
+            "another-subscription": [
+                mock.MagicMock(
+                    id="/subscriptions/another-sub/resourceGroups/test-rg/providers/Microsoft.ApiManagement/service/another-apim",
+                    name="another-apim",
+                    log_analytics_workspace_id="/subscriptions/another-sub/resourceGroups/test-rg/providers/Microsoft.OperationalInsights/workspaces/another-workspace",
+                )
+            ],
+        }
+        apim_client.audit_config = {
+            "apim_threat_detection_llm_jacking_threshold": 0.9,
+            "apim_threat_detection_llm_jacking_minutes": 1440,
+            "apim_threat_detection_llm_jacking_actions": [
+                "ChatCompletions_Create",
+                "ImageGenerations_Create",
+                "Completions_Create",
+                "Embeddings_Create",
+                "FineTuning_Jobs_Create",
+                "Models_List",
+                "Deployments_List",
+                "Deployments_Get",
+                "Deployments_Create",
+                "Deployments_Delete",
+                "Messages_Create",
+                "Claude_Create",
+                "GenerateContent",
+                "GenerateText",
+                "GenerateImage",
+                "Llama_Create",
+                "CodeLlama_Create",
+                "Gemini_Generate",
+                "Claude_Generate",
+                "Llama_Generate",
+            ],
+        }
+        apim_client.get_llm_operations_logs = mock_get_llm_operations_logs_2_operations
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_azure_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking.apim_client",
+                new=apim_client,
+            ),
+        ):
+            from prowler.providers.azure.services.apim.apim_threat_detection_llm_jacking.apim_threat_detection_llm_jacking import (
+                apim_threat_detection_llm_jacking,
+            )
+
+            check = apim_threat_detection_llm_jacking()
+            result = check.execute()
+
+            assert len(result) == 2
+            # Both subscriptions should have PASS results
+            for report in result:
+                assert report.status == "PASS"
+                assert (
+                    "No potential LLM Jacking attacks detected"
+                    in report.status_extended
+                )