feat: Use ManagedIdentityCredential instead of DefaultAzureCredentials

infra/main.bicep

@@ -807,6 +807,10 @@ module avmContainerApp 'br/public:avm/res/app/container-app:0.17.0' = {
             name: 'APP_CONFIG_ENDPOINT'
             value: ''
           }
+          {
+            name: 'APP_ENV'
+            value: 'prod'
+          }
         ]
       }
     ]
@@ -851,6 +855,10 @@ module avmContainerApp_API 'br/public:avm/res/app/container-app:0.17.0' = {
             name: 'APP_CONFIG_ENDPOINT'
             value: ''
           }
+          {
+            name: 'APP_ENV'
+            value: 'prod'
+          }
         ]
         probes: [
           // Liveness Probe - Checks if the app is still running
@@ -1266,6 +1274,10 @@ module avmContainerApp_update 'br/public:avm/res/app/container-app:0.17.0' = {
             name: 'APP_CONFIG_ENDPOINT'
             value: avmAppConfig.outputs.endpoint
           }
+          {
+            name: 'APP_ENV'
+            value: 'prod'
+          }
         ]
       }
     ]
@@ -1321,6 +1333,10 @@ module avmContainerApp_API_update 'br/public:avm/res/app/container-app:0.17.0' =
             name: 'APP_CONFIG_ENDPOINT'
             value: avmAppConfig.outputs.endpoint
           }
+          {
+            name: 'APP_ENV'
+            value: 'prod'
+          }
         ]
         probes: [
           // Liveness Probe - Checks if the app is still running

src/ContentProcessor/conftest.py

@@ -0,0 +1,34 @@
+"""
+Global test configuration and fixtures for ContentProcessor tests.
+"""
+import sys
+import os
+import pytest
+from unittest.mock import patch, MagicMock
+
+# Add src directory to Python path for imports
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
+
+pytest_plugins = ["pytest_mock"]
+
+
+@pytest.fixture(autouse=True, scope="function")
+def mock_azure_credentials_for_helpers(request):
+    """
+    Mock Azure credentials for azure_helper classes only.
+    Skip this for credential utility tests that need to test the actual logic.
+    """
+    # Skip mocking for credential utility tests
+    if "test_azure_credential_utils" in str(request.fspath):
+        yield
+        return
+
+    with patch("helpers.azure_credential_utils.get_azure_credential") as mock_get_cred, \
+         patch("helpers.azure_credential_utils.get_azure_credential_async") as mock_get_cred_async:
+
+        # Create mock credential objects
+        mock_credential = MagicMock()
+        mock_get_cred.return_value = mock_credential
+        mock_get_cred_async.return_value = mock_credential
+
+        yield mock_credential

src/ContentProcessor/pytest.ini

@@ -0,0 +1,6 @@
+[tool:pytest]
+addopts = -v --strict-markers --disable-warnings
+python_files = tests.py test_*.py *_tests.py
+testpaths = src/tests
+markers =
+    asyncio: marks tests as async

src/ContentProcessor/src/helpers/azure_credential_utils.py

@@ -0,0 +1,44 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import os
+from azure.identity import ManagedIdentityCredential, DefaultAzureCredential
+from azure.identity.aio import ManagedIdentityCredential as AioManagedIdentityCredential, DefaultAzureCredential as AioDefaultAzureCredential
+
+
+async def get_azure_credential_async(client_id=None):
+    """
+    Returns an Azure credential asynchronously based on the application environment.
+
+    If the environment is 'dev', it uses AioDefaultAzureCredential.
+    Otherwise, it uses AioManagedIdentityCredential.
+
+    Args:
+        client_id (str, optional): The client ID for the Managed Identity Credential.
+
+    Returns:
+        Credential object: Either AioDefaultAzureCredential or AioManagedIdentityCredential.
+    """
+    if os.getenv("APP_ENV", "prod").lower() == 'dev':
+        return AioDefaultAzureCredential()  # CodeQL [SM05139] Okay use of DefaultAzureCredential as it is only used in development
+    else:
+        return AioManagedIdentityCredential(client_id=client_id)
+
+
+def get_azure_credential(client_id=None):
+    """
+    Returns an Azure credential based on the application environment.
+
+    If the environment is 'dev', it uses DefaultAzureCredential.
+    Otherwise, it uses ManagedIdentityCredential.
+
+    Args:
+        client_id (str, optional): The client ID for the Managed Identity Credential.
+
+    Returns:
+        Credential object: Either DefaultAzureCredential or ManagedIdentityCredential.
+    """
+    if os.getenv("APP_ENV", "prod").lower() == 'dev':
+        return DefaultAzureCredential()  # CodeQL [SM05139] Okay use of DefaultAzureCredential as it is only used in development
+    else:
+        return ManagedIdentityCredential(client_id=client_id)

src/ContentProcessor/src/libs/application/application_context.py

@@ -1,4 +1,6 @@
-from azure.identity import DefaultAzureCredential
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+from typing import Any
 
 from libs.application.application_configuration import AppConfiguration
 from libs.base.application_models import AppModelBase
@@ -11,10 +13,10 @@ class AppContext(AppModelBase):
     """
 
     configuration: AppConfiguration = None
-    credential: DefaultAzureCredential = None
+    credential: Any = None  # Azure credential object
 
     def set_configuration(self, configuration: AppConfiguration):
         self.configuration = configuration
 
-    def set_credential(self, credential: DefaultAzureCredential):
+    def set_credential(self, credential: Any):
         self.credential = credential

src/ContentProcessor/src/libs/azure_helper/app_configuration.py

@@ -1,16 +1,18 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
 import os
 
 from azure.appconfiguration import AzureAppConfigurationClient
-from azure.identity import DefaultAzureCredential
+from helpers.azure_credential_utils import get_azure_credential
 
 
 class AppConfigurationHelper:
-    credential: DefaultAzureCredential = None
     app_config_endpoint: str = None
     app_config_client: AzureAppConfigurationClient = None
 
     def __init__(self, app_config_endpoint: str):
-        self.credential = DefaultAzureCredential()
+        self.credential = get_azure_credential()
         self.app_config_endpoint = app_config_endpoint
         self._initialize_client()
 

src/ContentProcessor/src/libs/azure_helper/azure_openai.py

@@ -1,9 +1,13 @@
-from azure.identity import DefaultAzureCredential, get_bearer_token_provider
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from azure.identity import get_bearer_token_provider
+from helpers.azure_credential_utils import get_azure_credential
 from openai import AzureOpenAI
 
 
 def get_openai_client(azure_openai_endpoint: str) -> AzureOpenAI:
-    credential = DefaultAzureCredential()
+    credential = get_azure_credential()
     token_provider = get_bearer_token_provider(
         credential, "https://cognitiveservices.azure.com/.default"
     )

src/ContentProcessor/src/libs/azure_helper/content_understanding.py

@@ -7,22 +7,21 @@
 from pathlib import Path
 
 import requests
-from azure.identity import DefaultAzureCredential
+from helpers.azure_credential_utils import get_azure_credential
 from requests.models import Response
 
 COGNITIVE_SERVICES_SCOPE = "https://cognitiveservices.azure.com/.default"
 
 
 class AzureContentUnderstandingHelper:
-    credential: DefaultAzureCredential = None
 
     def __init__(
         self,
         endpoint: str,
         api_version: str = "2024-12-01-preview",
         x_ms_useragent: str = "cps-contentunderstanding/client",
     ):
-        self.credential = DefaultAzureCredential()
+        self.credential = get_azure_credential()
 
         if not api_version:
             raise ValueError("API version must be provided.")

src/ContentProcessor/src/libs/azure_helper/storage_blob.py

@@ -3,20 +3,19 @@
 
 from typing import IO, Union
 
-from azure.identity import DefaultAzureCredential
+from helpers.azure_credential_utils import get_azure_credential
 from azure.storage.blob import BlobServiceClient
 
 
 class StorageBlobHelper:
-    credential: DefaultAzureCredential = None
     blob_service_client: BlobServiceClient = None
 
     @staticmethod
     def get(account_url: str, container_name: str = None):
         return StorageBlobHelper(account_url=account_url, container_name=container_name)
 
     def __init__(self, account_url: str, container_name=None):
-        self.credential = DefaultAzureCredential()
+        self.credential = get_azure_credential()
         self.blob_service_client = BlobServiceClient(
             account_url=account_url, credential=self.credential
         )

src/ContentProcessor/src/libs/pipeline/pipeline_queue_helper.py

@@ -4,7 +4,7 @@
 import logging
 
 from azure.core.exceptions import ResourceNotFoundError
-from azure.identity import DefaultAzureCredential
+from helpers.azure_credential_utils import get_azure_credential
 from azure.storage.queue import QueueClient, QueueMessage
 
 from libs.pipeline import pipeline_step_helper
@@ -28,7 +28,7 @@ def invalidate_queue(queue_client: QueueClient):
 
 
 def create_or_get_queue_client(
-    queue_name: str, accouont_url: str, credential: DefaultAzureCredential
+    queue_name: str, accouont_url: str, credential: get_azure_credential
 ) -> QueueClient:
     queue_client = QueueClient(
         account_url=accouont_url, queue_name=queue_name, credential=credential
@@ -55,7 +55,7 @@ def has_messages(queue_client: QueueClient) -> bool:
 
 
 def pass_data_pipeline_to_next_step(
-    data_pipeline: DataPipeline, account_url: str, credential: DefaultAzureCredential
+    data_pipeline: DataPipeline, account_url: str, credential: get_azure_credential
 ):
     next_step_name = pipeline_step_helper.get_next_step_name(
         data_pipeline.pipeline_status, data_pipeline.pipeline_status.active_step
@@ -70,7 +70,7 @@ def pass_data_pipeline_to_next_step(
 
 
 def _create_queue_client(
-    account_url: str, queue_name: str, credential: DefaultAzureCredential
+    account_url: str, queue_name: str, credential: get_azure_credential
 ) -> QueueClient:
     queue_client = QueueClient(
         account_url=account_url, queue_name=queue_name, credential=credential

src/ContentProcessor/src/libs/utils/remote_module_loader.py

@@ -4,7 +4,7 @@
 import importlib.util
 import sys
 
-from azure.identity import DefaultAzureCredential
+from helpers.azure_credential_utils import get_azure_credential
 from azure.storage.blob import BlobServiceClient
 
 
@@ -27,7 +27,7 @@ def load_schema_from_blob(
 
 def _download_blob_content(container_name, blob_name, account_url):
     # Create the BlobServiceClient object which will be used to create a container client
-    credential = DefaultAzureCredential()
+    credential = get_azure_credential()
     blob_service_client = BlobServiceClient(
         account_url=account_url, credential=credential
     )

src/ContentProcessor/src/main.py

@@ -5,7 +5,7 @@
 import os
 import sys
 
-from azure.identity import DefaultAzureCredential
+from helpers.azure_credential_utils import get_azure_credential
 
 from libs.base.application_main import AppMainBase
 from libs.process_host import handler_type_loader
@@ -29,7 +29,7 @@ def __init__(self, **data):
 
     def _initialize_application(self):
         # Add Azure Credential
-        self.application_context.set_credential(DefaultAzureCredential())
+        self.application_context.set_credential(get_azure_credential())
 
     async def run(self, test_mode: bool = False):
         # Get Process lists from the configuration - ex. ["extract", "transform", "evaluate", "save", "custom1", "custom2"....]

src/ContentProcessor/src/tests/azure_helper/test_storage_blob.py

@@ -1,20 +1,22 @@
 import pytest
 from io import BytesIO
-from libs.azure_helper.storage_blob import StorageBlobHelper
+from unittest.mock import MagicMock, patch
+
+# Ensure Azure credentials are mocked before any imports
+with patch("helpers.azure_credential_utils.get_azure_credential") as mock_cred:
+    mock_cred.return_value = MagicMock()
+    from libs.azure_helper.storage_blob import StorageBlobHelper
 
 
 @pytest.fixture
 def mock_blob_service_client(mocker):
+    """Mock BlobServiceClient class for tests."""
     return mocker.patch("libs.azure_helper.storage_blob.BlobServiceClient")
 
 
 @pytest.fixture
-def mock_default_azure_credential(mocker):
-    return mocker.patch("libs.azure_helper.storage_blob.DefaultAzureCredential")
-
-
-@pytest.fixture
-def storage_blob_helper(mock_blob_service_client, mock_default_azure_credential):
+def storage_blob_helper(mock_blob_service_client):
+    """Create StorageBlobHelper with mocked BlobServiceClient."""
     return StorageBlobHelper(
         account_url="https://testaccount.blob.core.windows.net",
         container_name="testcontainer",

src/ContentProcessor/src/tests/conftest.py

@@ -0,0 +1,4 @@
+"""
+Test configuration and fixtures for ContentProcessor tests.
+"""
+# Note: pytest_plugins is now defined in the top-level conftest.py