From 776c7c55ac43e29a940860eeb4cf4842962a39ba Mon Sep 17 00:00:00 2001 From: Hameed Kunkanoor Date: Tue, 7 Jul 2026 15:57:05 +0530 Subject: [PATCH] msal: native agentic token for UserManagedIdentity via fmi_path exchange MsalAuth.get_agentic_application_token raised RuntimeError for AuthTypes.user_managed_identity because MSAL's ManagedIdentityClient does not support the federated fmi_path exchange required for agentic application tokens. Callers had to monkey-patch get_agentic_application_token to work around this. This adds a native branch that, for user_managed_identity, acquires the token via azure-identity's DefaultAzureCredential with identity_config={"fmi_path": ...} (and managed_identity_client_id from CLIENT_ID when set) against the managed- identity endpoints. azure-identity is imported lazily and a clear ImportError is raised if it is not installed, so it stays an optional dependency. - New MsalAuth._acquire_agentic_token_via_managed_identity helper. - Updated the unsupported-client test to use system_managed_identity (still raises); added a user_managed_identity test that injects a fake DefaultAzureCredential (no azure-identity needed to run the test). - black clean (libraries + tests); 30/30 msal_auth tests pass. Note: complements #413 (identity_proxy_manager), which added a distinct IDPM resource path but does not perform the fmi_path federated exchange. --- .../authentication/msal/msal_auth.py | 65 ++++++++++++++++++- tests/authentication_msal/test_msal_auth.py | 52 ++++++++++++++- 2 files changed, 115 insertions(+), 2 deletions(-) diff --git a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py index e3400d86..339ca876 100644 --- a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py +++ b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py @@ -388,10 +388,73 @@ async def get_agentic_application_token( return None + if ( + self._msal_configuration.AUTH_TYPE == AuthTypes.user_managed_identity + and isinstance(msal_auth_client, ManagedIdentityClient) + ): + return await self._acquire_agentic_token_via_managed_identity( + agent_app_instance_id + ) + raise RuntimeError( - "Agentic token acquisition supports ConfidentialClientApplication, or ManagedIdentityClient when AUTH_TYPE is AuthTypes.identity_proxy_manager." + "Agentic token acquisition supports ConfidentialClientApplication, or ManagedIdentityClient " + "when AUTH_TYPE is AuthTypes.identity_proxy_manager or AuthTypes.user_managed_identity." + ) + + async def _acquire_agentic_token_via_managed_identity( + self, agent_app_instance_id: str + ) -> Optional[str]: + """Acquire an agentic application token for a user-assigned managed identity. + + MSAL's ``ManagedIdentityClient`` does not support the federated ``fmi_path`` + token exchange required for agentic application tokens. This uses + azure-identity's ``DefaultAzureCredential``, which accepts + ``identity_config={"fmi_path": ...}`` and handles the managed-identity + (IMDS / App Service) endpoints, to perform the federated exchange. + + :param agent_app_instance_id: The agent application instance ID, used as the + federated identity ``fmi_path``. + :type agent_app_instance_id: str + :return: The agentic application token, or None if acquisition failed. + :rtype: Optional[str] + """ + try: + from azure.identity.aio import DefaultAzureCredential + except ImportError as exc: + raise ImportError( + "Acquiring an agentic application token for AuthTypes.user_managed_identity " + "requires the 'azure-identity' package. Install it with: pip install azure-identity." + ) from exc + + credential_kwargs: dict = { + "identity_config": {"fmi_path": agent_app_instance_id}, + } + client_id = getattr(self._msal_configuration, "CLIENT_ID", None) + if client_id: + credential_kwargs["managed_identity_client_id"] = client_id + + logger.info( + "Acquiring agentic application token via DefaultAzureCredential for agent_app_instance_id=%s", + agent_app_instance_id, ) + credential = DefaultAzureCredential(**credential_kwargs) + try: + token = await credential.get_token("api://AzureAdTokenExchange/.default") + return token.token + except Exception: # noqa: BLE001 - failures are logged and surfaced as None + logger.exception( + "Failed to acquire agentic application token via DefaultAzureCredential " + "for agent_app_instance_id=%s", + agent_app_instance_id, + ) + return None + finally: + try: + await credential.close() + except Exception: # noqa: BLE001 - best-effort cleanup + logger.debug("Error closing DefaultAzureCredential", exc_info=True) + async def get_agentic_instance_token( self, tenant_id: str, agent_app_instance_id: str ) -> tuple[str, str]: diff --git a/tests/authentication_msal/test_msal_auth.py b/tests/authentication_msal/test_msal_auth.py index adb5c140..48491c84 100644 --- a/tests/authentication_msal/test_msal_auth.py +++ b/tests/authentication_msal/test_msal_auth.py @@ -335,12 +335,62 @@ async def test_get_agentic_application_token_identity_proxy_manager(self, mocker resource="https://custom-resource/.default" ) + @pytest.mark.asyncio + async def test_get_agentic_application_token_user_managed_identity(self, mocker): + """UserManagedIdentity acquires the agentic token via DefaultAzureCredential + using the federated ``fmi_path`` exchange (no monkey-patch required).""" + import sys + import types + + config = AgentAuthConfiguration( + auth_type=AuthTypes.user_managed_identity, + client_id="test-client-id", + ) + auth = MsalAuth(config) + + mock_client = mocker.Mock(spec=ManagedIdentityClient) + mocker.patch.object(auth, "_get_client", return_value=mock_client) + + # Inject a fake azure.identity.aio.DefaultAzureCredential so the test does + # not require azure-identity to be installed. + captured = {} + + class _FakeToken: + token = "umi-fmi-token" + + class _FakeCredential: + def __init__(self, **kwargs): + captured["kwargs"] = kwargs + + async def get_token(self, scope): + captured["scope"] = scope + return _FakeToken() + + async def close(self): + captured["closed"] = True + + fake_module = types.ModuleType("azure.identity.aio") + fake_module.DefaultAzureCredential = _FakeCredential + mocker.patch.dict(sys.modules, {"azure.identity.aio": fake_module}) + + token = await auth.get_agentic_application_token( + "test-tenant-id", "test-agent-app-instance-id" + ) + + assert token == "umi-fmi-token" + assert captured["kwargs"]["identity_config"] == { + "fmi_path": "test-agent-app-instance-id" + } + assert captured["kwargs"]["managed_identity_client_id"] == "test-client-id" + assert captured["scope"] == "api://AzureAdTokenExchange/.default" + assert captured.get("closed") is True + @pytest.mark.asyncio async def test_get_agentic_application_token_unsupported_client_raises( self, mocker ): config = AgentAuthConfiguration( - auth_type=AuthTypes.user_managed_identity, + auth_type=AuthTypes.system_managed_identity, client_id="test-client-id", ) auth = MsalAuth(config)