fix: Resolve CodeQL issues to avoid unsafe use of DefaultAzureCredential in python application.

Author: Avijit-Microsoft

src/ContentProcessor/src/libs/utils/azure_credential_utils.py

@@ -130,7 +130,11 @@ def get_azure_credential():
     logging.info(
         "[AUTH] All CLI credentials failed - falling back to DefaultAzureCredential"
     )
-    return DefaultAzureCredential()
+    raise RuntimeError(
+        "No Azure authentication available. "
+        "Use Managed Identity in Azure or run "
+        "'az login' / 'azd auth login' locally."
+    )
 
 
 def get_async_azure_credential():

src/ContentProcessor/src/libs/utils/credential_util.py

@@ -130,7 +130,11 @@ def get_azure_credential():
     logging.info(
         "[AUTH] All CLI credentials failed - falling back to DefaultAzureCredential"
     )
-    return DefaultAzureCredential()
+    raise RuntimeError(
+        "No Azure authentication available. "
+        "Use Managed Identity in Azure or run "
+        "'az login' / 'azd auth login' locally."
+    )
 
 
 def get_async_azure_credential():

src/ContentProcessorAPI/app/libs/base/application_base.py

@@ -15,7 +15,7 @@
 import os
 from abc import ABC, abstractmethod
 
-from azure.identity import DefaultAzureCredential
+from app.utils.azure_credential_utils import get_azure_credential
 from dotenv import load_dotenv
 
 from app.libs.application.application_configuration import (
@@ -72,7 +72,7 @@ def __init__(self, env_file_path: str | None = None, **data):
         self._load_env(env_file_path=env_file_path)
 
         self.application_context = AppContext()
-        self.application_context.set_credential(DefaultAzureCredential())
+        self.application_context.set_credential(get_azure_credential())
 
         app_config_endpoint: str | None = EnvConfiguration().app_config_endpoint
         if app_config_endpoint != "" and app_config_endpoint is not None:

src/ContentProcessorWorkflow/src/libs/azure/app_configuration.py

@@ -91,7 +91,13 @@ def __init__(
             ValueError: If *app_configuration_url* is ``None`` or the
                 credential is missing after defaulting.
         """
-        self.credential = credential or DefaultAzureCredential()
+        if credential is None:
+            raise ValueError(
+                "Azure credential is required. "
+                "Use Managed Identity, AzureCliCredential, or AzureDeveloperCliCredential."
+            )
+
+        self.credential = credential
         self.app_config_endpoint = app_configuration_url
         self._initialize_client()
 

src/ContentProcessorWorkflow/src/libs/base/application_base.py

@@ -35,7 +35,7 @@ def run(self):
 import os
 from abc import ABC, abstractmethod
 
-from azure.identity import DefaultAzureCredential
+from utils.credential_util import get_azure_credential
 from dotenv import load_dotenv
 
 from libs.agent_framework.agent_framework_settings import AgentFrameworkSettings
@@ -117,7 +117,7 @@ def __init__(self, env_file_path: str | None = None, **data):
         self._load_env(env_file_path=env_file_path)
 
         self.application_context = AppContext()
-        self.application_context.set_credential(DefaultAzureCredential())
+        self.application_context.set_credential(get_azure_credential())
 
         app_config_url: str | None = _envConfiguration().app_config_endpoint
         if app_config_url != "" and app_config_url is not None:

src/ContentProcessorWorkflow/src/utils/credential_util.py

@@ -126,7 +126,12 @@ def get_azure_credential():
     logging.info(
         "[AUTH] All CLI credentials failed - falling back to DefaultAzureCredential"
     )
-    return DefaultAzureCredential()
+    
+    raise RuntimeError(
+    "No Azure authentication available. "
+    "Use Managed Identity in Azure or run "
+    "'az login' / 'azd auth login' locally."
+    )
 
 
 def get_async_azure_credential():

src/tests/ContentProcessor/utils/test_azure_credential_utils.py

@@ -7,6 +7,8 @@
 
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 import libs.utils.azure_credential_utils as azure_credential_utils
 
 MODULE = "libs.utils.azure_credential_utils"
@@ -45,16 +47,16 @@ def test_returns_user_assigned_with_client_id(self, mock_managed):
         mock_managed.assert_called_once_with(client_id="test-client-id")
         assert credential == mock_instance
 
-    @patch(f"{MODULE}.DefaultAzureCredential")
     @patch(f"{MODULE}.AzureDeveloperCliCredential", side_effect=Exception("no azd"))
     @patch(f"{MODULE}.AzureCliCredential", side_effect=Exception("no az"))
     @patch.dict("os.environ", {}, clear=True)
-    def test_falls_back_to_default(self, mock_cli, mock_dev_cli, mock_default):
-        mock_instance = MagicMock()
-        mock_default.return_value = mock_instance
-        credential = azure_credential_utils.get_azure_credential()
-        mock_default.assert_called_once()
-        assert credential == mock_instance
+    def test_raises_when_no_credentials_available(
+        self, mock_cli, mock_dev_cli
+    ):
+        with pytest.raises(RuntimeError) as exc:
+            azure_credential_utils.get_azure_credential()
+
+        assert "No Azure authentication available" in str(exc.value)
 
 
 # ── TestGetAsyncAzureCredential ─────────────────────────────────────────

src/tests/ContentProcessor/utils/test_azure_credential_utils_extended.py

@@ -42,26 +42,22 @@ def test_get_azure_credential_with_website_site_name(self, monkeypatch):
             assert credential == mock_instance
 
     def test_get_azure_credential_cli_failure_fallback(self, monkeypatch):
-        """Test fallback to DefaultAzureCredential when CLI credentials fail"""
+        """Test RuntimeError when all credential options fail"""
         # Clear all Azure environment indicators
         for key in ["WEBSITE_SITE_NAME", "AZURE_CLIENT_ID", "MSI_ENDPOINT",
                     "IDENTITY_ENDPOINT", "KUBERNETES_SERVICE_HOST", "CONTAINER_REGISTRY_LOGIN"]:
             monkeypatch.delenv(key, raising=False)
 
         with patch('libs.utils.azure_credential_utils.AzureCliCredential') as mock_cli_cred, \
-             patch('libs.utils.azure_credential_utils.AzureDeveloperCliCredential') as mock_azd_cred, \
-             patch('libs.utils.azure_credential_utils.DefaultAzureCredential') as mock_default:
+             patch('libs.utils.azure_credential_utils.AzureDeveloperCliCredential') as mock_azd_cred:
 
-            # Make both CLI credentials raise exceptions
             mock_cli_cred.side_effect = Exception("CLI credential failed")
             mock_azd_cred.side_effect = Exception("AZD credential failed")
-            mock_default_instance = Mock()
-            mock_default.return_value = mock_default_instance
 
-            credential = get_azure_credential()
+            with pytest.raises(RuntimeError) as exc:
+                get_azure_credential()
 
-            assert credential == mock_default_instance
-            mock_default.assert_called_once()
+            assert "No Azure authentication available" in str(exc.value)
 
     def test_get_azure_credential_azd_success(self, monkeypatch):
         """Test successful Azure Developer CLI credential"""

src/tests/ContentProcessorWorkflow/utils/test_credential_util_extended.py

@@ -1,5 +1,6 @@
 """Extended tests for credential_util.py to improve coverage"""
 from unittest.mock import Mock, patch
+import pytest
 from utils.credential_util import (
     get_azure_credential,
     get_async_azure_credential,
@@ -40,24 +41,27 @@ def test_get_azure_credential_app_service_environment(self, monkeypatch):
             assert credential == mock_instance
 
     def test_get_azure_credential_all_cli_fail(self, monkeypatch):
-        """Test fallback when all CLI credentials fail"""
-        for key in ["WEBSITE_SITE_NAME", "AZURE_CLIENT_ID", "MSI_ENDPOINT",
-                    "IDENTITY_ENDPOINT", "KUBERNETES_SERVICE_HOST", "CONTAINER_REGISTRY_LOGIN"]:
+        """Test RuntimeError when all credential options fail"""
+        for key in [
+            "WEBSITE_SITE_NAME",
+            "AZURE_CLIENT_ID",
+            "MSI_ENDPOINT",
+            "IDENTITY_ENDPOINT",
+            "KUBERNETES_SERVICE_HOST",
+            "CONTAINER_REGISTRY_LOGIN",
+        ]:
             monkeypatch.delenv(key, raising=False)
 
         with patch('utils.credential_util.AzureCliCredential') as mock_cli, \
-             patch('utils.credential_util.AzureDeveloperCliCredential') as mock_azd, \
-             patch('utils.credential_util.DefaultAzureCredential') as mock_default:
+             patch('utils.credential_util.AzureDeveloperCliCredential') as mock_azd:
 
             mock_cli.side_effect = Exception("AzureCLI not available")
             mock_azd.side_effect = Exception("AzureDeveloperCLI not available")
-            mock_default_instance = Mock()
-            mock_default.return_value = mock_default_instance
 
-            credential = get_azure_credential()
+            with pytest.raises(RuntimeError) as exc:
+                get_azure_credential()
 
-            assert credential == mock_default_instance
-            mock_default.assert_called_once()
+            assert "No Azure authentication available" in str(exc.value)
 
     def test_get_azure_credential_cli_success(self, monkeypatch):
         """Test successful Azure CLI credential"""