updated

Author: GladwinJohnson

msal/managed_identity.py

@@ -171,7 +171,6 @@ def __init__(
         token_cache=None,
         http_cache=None,
         client_capabilities: Optional[List[str]] = None,
-        msi_v2_enabled: Optional[bool] = None,
     ):
         """Create a managed identity client.
 
@@ -213,17 +212,6 @@ def __init__(
             Client capability in Managed Identity is relayed as-is
             via ``xms_cc`` parameter on the wire.
 
-        :param bool msi_v2_enabled: (optional)
-            Enable MSI v2 (mTLS PoP) token acquisition.
-            When True (or when the ``MSAL_ENABLE_MSI_V2`` environment variable
-            is set to a truthy value), the client will attempt to acquire tokens
-            using the MSI v2 flow (IMDS /issuecredential + mTLS PoP).
-            If the MSI v2 flow fails, it automatically falls back to MSI v1.
-            MSI v2 only applies to Azure VM (IMDS) environments; it is ignored
-            in other managed identity environments (App Service, Service Fabric,
-            Azure Arc, etc.).
-            Defaults to None (disabled unless the env var is set).
-
         Recipe 1: Hard code a managed identity for your app::
 
             import msal, requests
@@ -270,11 +258,6 @@ def __init__(
         )
         self._token_cache = token_cache or TokenCache()
         self._client_capabilities = client_capabilities
-        # MSI v2 is enabled by the constructor param or the MSAL_ENABLE_MSI_V2 env var
-        if msi_v2_enabled is None:
-            env_val = os.environ.get("MSAL_ENABLE_MSI_V2", "").lower()
-            msi_v2_enabled = env_val in ("1", "true", "yes")
-        self._msi_v2_enabled = msi_v2_enabled
 
     def acquire_token_for_client(
         self,
@@ -309,7 +292,8 @@ def acquire_token_for_client(
             Without this flag the legacy IMDS v1 flow is used.
             Defaults to False.
 
-            This takes precedence over the ``msi_v2_enabled`` constructor parameter.
+            MSI v2 is used only when both ``mtls_proof_of_possession`` and
+            ``with_attestation_support`` are True.
 
         :param bool with_attestation_support: (optional)
             When True (and ``mtls_proof_of_possession`` is also True), attempt
@@ -332,6 +316,27 @@ def acquire_token_for_client(
         client_id_in_cache = self._managed_identity.get(
             ManagedIdentity.ID, "SYSTEM_ASSIGNED_MANAGED_IDENTITY")
         now = time.time()
+        # MSI v2 is opt-in: use it only when BOTH mtls_proof_of_possession and
+        # with_attestation_support are explicitly requested by the caller.
+        # No auto-fallback: if MSI v2 is requested and fails, the error is raised.
+        use_msi_v2 = bool(mtls_proof_of_possession and with_attestation_support)
+
+        if with_attestation_support and not mtls_proof_of_possession:
+            raise ManagedIdentityError(
+                "attestation_requires_pop",
+                "with_attestation_support=True requires mtls_proof_of_possession=True (mTLS PoP)."
+            )
+
+        if use_msi_v2:
+            from .msi_v2 import obtain_token as _obtain_token_v2
+            result = _obtain_token_v2(
+                self._http_client, self._managed_identity, resource,
+                attestation_enabled=True,
+            )
+            if "access_token" in result and "error" not in result:
+                result[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
+            return result
+
         if True:  # Attempt cache search even if receiving claims_challenge,
                   # because we want to locate the existing token (if any) and refresh it
             matches = self._token_cache.search(
@@ -366,41 +371,13 @@ def acquire_token_for_client(
                         break  # With a fallback in hand, we break here to go refresh
                 return access_token_from_cache  # It is still good as new
         try:
-            result = None
-            # Per-call mtls_proof_of_possession takes precedence over the constructor
-            # default (msi_v2_enabled / MSAL_ENABLE_MSI_V2 env var).
-            use_msi_v2 = mtls_proof_of_possession or self._msi_v2_enabled
-            if use_msi_v2:
-                if mtls_proof_of_possession:
-                    # Explicit per-call request: errors are raised, no fallback to v1
-                    from .msi_v2 import obtain_token as _obtain_token_v2
-                    result = _obtain_token_v2(
-                        self._http_client, self._managed_identity, resource,
-                        attestation_enabled=with_attestation_support)
-                    logger.debug("MSI v2 token acquisition succeeded")
-                else:
-                    # Legacy constructor flag: swallow errors and fall back to v1
-                    try:
-                        from .msi_v2 import obtain_token as _obtain_token_v2
-                        result = _obtain_token_v2(
-                            self._http_client, self._managed_identity, resource,
-                            attestation_enabled=with_attestation_support)
-                        logger.debug("MSI v2 token acquisition succeeded")
-                    except MsiV2Error as exc:
-                        logger.warning(
-                            "MSI v2 flow failed, falling back to MSI v1: %s", exc)
-                    except Exception as exc:  # pylint: disable=broad-except
-                        logger.warning(
-                            "MSI v2 encountered unexpected error, "
-                            "falling back to MSI v1: %s", exc)
-            if result is None:
-                result = _obtain_token(
-                    self._http_client, self._managed_identity, resource,
-                    access_token_sha256_to_refresh=hashlib.sha256(
-                        access_token_to_refresh.encode("utf-8")).hexdigest()
-                        if access_token_to_refresh else None,
-                    client_capabilities=self._client_capabilities,
-                )
+            result = _obtain_token(
+                self._http_client, self._managed_identity, resource,
+                access_token_sha256_to_refresh=hashlib.sha256(
+                    access_token_to_refresh.encode("utf-8")).hexdigest()
+                    if access_token_to_refresh else None,
+                client_capabilities=self._client_capabilities,
+            )
             if "access_token" in result:
                 expires_in = result.get("expires_in", 3600)
                 if "refresh_in" not in result and expires_in >= 7200:
@@ -753,4 +730,4 @@ def _obtain_token_on_arc(http_client, endpoint, resource):
     return {
         "error": "invalid_request",
         "error_description": response.text,
-        }
+        }