updated API for fmi

Author: 4gust

msal/application.py

@@ -2427,7 +2427,7 @@ class ConfidentialClientApplication(ClientApplication):  # server-side web app
     except that ``allow_broker`` parameter shall remain ``None``.
     """
 
-    def acquire_token_for_client(self, scopes, claims_challenge=None, **kwargs):
+    def acquire_token_for_client(self, scopes, claims_challenge=None, fmi_path=None, **kwargs):
         """Acquires token for the current confidential client, not for an end user.
 
         Since MSAL Python 1.23, it will automatically look for token from cache,
@@ -2440,7 +2440,17 @@ def acquire_token_for_client(self, scopes, claims_challenge=None, **kwargs):
             in the form of a claims_challenge directive in the www-authenticate header to be
             returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
             It is a string of a JSON object which contains lists of claims being requested from these locations.
-
+        :param str fmi_path:
+            Optional. The Federated Managed Identity (FMI) credential path.
+            When provided, it is sent as the ``fmi_path`` parameter in the
+            token request body, and the resulting token is cached separately
+            so that different FMI paths do not share cached tokens.
+            Example usage::
+
+                result = cca.acquire_token_for_client(
+                    scopes=["api://resource/.default"],
+                    fmi_path="SomeFmiPath/FmiCredentialPath",
+                )
         :return: A dict representing the json response from Microsoft Entra:
 
             - A successful response would contain "access_token" key,
@@ -2450,6 +2460,12 @@ def acquire_token_for_client(self, scopes, claims_challenge=None, **kwargs):
             raise ValueError(  # We choose to disallow force_refresh
                 "Historically, this method does not support force_refresh behavior. "
             )
+        if fmi_path is not None:
+            if not isinstance(fmi_path, str):
+                raise ValueError(
+                    "fmi_path must be a string, got {}".format(type(fmi_path).__name__))
+            kwargs["data"] = kwargs.get("data", {})
+            kwargs["data"]["fmi_path"] = fmi_path
         return _clean_up(self._acquire_token_silent_with_error(
             scopes, None, claims_challenge=claims_challenge, **kwargs))
 
@@ -2494,32 +2510,6 @@ def remove_tokens_for_client(self):
                 self.token_cache.remove_at(at)
         # acquire_token_for_client() obtains no RTs, so we have no RT to remove
 
-    def acquire_token_for_client_with_fmi_path(self, scopes, fmi_path, claims_challenge=None, **kwargs):
-        """Acquires token for the current confidential client with a Federated Managed Identity (FMI) path.
-
-        This is a convenience wrapper around :func:`~acquire_token_for_client`
-        that attaches the ``fmi_path`` parameter to the token request body.
-
-        :param list[str] scopes: (Required)
-            Scopes requested to access a protected API (a resource).
-        :param str fmi_path: (Required)
-            The Federated Managed Identity path to attach to the request.
-        :param claims_challenge:
-            The claims_challenge parameter requests specific claims requested by the resource provider
-            in the form of a claims_challenge directive in the www-authenticate header to be
-            returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
-            It is a string of a JSON object which contains lists of claims being requested from these locations.
-
-        :return: A dict representing the json response from Microsoft Entra:
-
-            - A successful response would contain "access_token" key,
-            - an error response would contain "error" and usually "error_description".
-        """
-        data = kwargs.pop("data", {})
-        data["fmi_path"] = fmi_path
-        return self.acquire_token_for_client(
-            scopes, claims_challenge=claims_challenge, data=data, **kwargs)
-
     def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=None, **kwargs):
         """Acquires token using on-behalf-of (OBO) flow.
 

msal/token_cache.py

@@ -20,11 +20,11 @@
 #
 # Excluded fields and reasons:
 #   - "client_id"              : Standard OAuth2 client identifier, same for every request
-#   - "grant_type"             : Standard OAuth2 grant type (e.g. jwt-bearer, refresh_token)
+#   - "grant_type"             : It is possible to combine grants to get tokens, e.g. obo + refresh_token, auth_code + refresh_token etc.
 #   - "scope"                  : Already represented as "target" in the AT cache key
 #   - "claims"                 : Handled separately; its presence forces a token refresh
-#   - "username"               : Standard ROPC grant parameter
-#   - "password"               : Standard ROPC grant parameter
+#   - "username"               : Standard ROPC grant parameter. Tokens are cached by user ID (subject or oid+tid) instead
+#   - "password"               : Standard ROPC grant parameter. Tokens are tied to credentials.
 #   - "refresh_token"          : Standard refresh grant parameter
 #   - "code"                   : Standard authorization code grant parameter
 #   - "redirect_uri"           : Standard authorization code grant parameter

tests/test_application.py

@@ -709,7 +709,15 @@ def test_organizations_authority_should_emit_warning(self):
 
 @patch(_OIDC_DISCOVERY, new=_OIDC_DISCOVERY_MOCK)
 class TestAcquireTokenForClientWithFmiPath(unittest.TestCase):
-    """Test that acquire_token_for_client_with_fmi_path attaches fmi_path to HTTP body."""
+    """Test that acquire_token_for_client(fmi_path=...) attaches fmi_path to HTTP body."""
+
+    def test_fmi_path_rejects_non_string_types(self):
+        app = ConfidentialClientApplication(
+            "client_id", client_credential="secret",
+            authority="https://login.microsoftonline.com/my_tenant")
+        for bad_value in [123, True, ["path"], {"path": "value"}, b"bytes"]:
+            with self.assertRaises(ValueError, msg="fmi_path={!r} should raise".format(bad_value)):
+                app.acquire_token_for_client(["scope"], fmi_path=bad_value)
 
     def test_fmi_path_is_included_in_request_body(self):
         app = ConfidentialClientApplication(
@@ -726,8 +734,8 @@ def mock_post(url, headers=None, data=None, *args, **kwargs):
                     "expires_in": 3600,
                 }))
 
-        result = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], fmi_path, post=mock_post)
+        result = app.acquire_token_for_client(
+            ["scope"], fmi_path=fmi_path, post=mock_post)
         self.assertIn("access_token", result)
         self.assertIn("fmi_path", captured_data,
             "fmi_path should be present in the HTTP request body")
@@ -749,8 +757,8 @@ def mock_post(url, headers=None, data=None, *args, **kwargs):
                     "expires_in": 3600,
                 }))
 
-        result = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], fmi_path, post=mock_post)
+        result = app.acquire_token_for_client(
+            ["scope"], fmi_path=fmi_path, post=mock_post)
         self.assertIn("access_token", result)
         self.assertEqual(fmi_path, captured_data["fmi_path"])
         self.assertEqual("client_credentials", captured_data.get("grant_type"))
@@ -770,8 +778,8 @@ def mock_post(url, headers=None, data=None, *args, **kwargs):
                     "expires_in": 3600,
                 }))
 
-        result = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], fmi_path,
+        result = app.acquire_token_for_client(
+            ["scope"], fmi_path=fmi_path,
             data={"extra_key": "extra_value"},
             post=mock_post)
         self.assertIn("access_token", result)
@@ -794,13 +802,13 @@ def mock_post(url, headers=None, data=None, *args, **kwargs):
                     "expires_in": 3600,
                 }))
 
-        result1 = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], fmi_path, post=mock_post)
+        result1 = app.acquire_token_for_client(
+            ["scope"], fmi_path=fmi_path, post=mock_post)
         self.assertIn("access_token", result1)
         self.assertEqual(result1[app._TOKEN_SOURCE], app._TOKEN_SOURCE_IDP)
 
-        result2 = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], fmi_path, post=mock_post)
+        result2 = app.acquire_token_for_client(
+            ["scope"], fmi_path=fmi_path, post=mock_post)
         self.assertIn("access_token", result2)
         self.assertEqual(result2[app._TOKEN_SOURCE], app._TOKEN_SOURCE_CACHE,
             "Second call should return token from cache")
@@ -821,20 +829,20 @@ def mock_post(url, headers=None, data=None, *args, **kwargs):
             return mock_post
 
         # Acquire token with path A
-        result_a = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], "PathA/credential", post=mock_post_factory("AT_for_path_A"))
+        result_a = app.acquire_token_for_client(
+            ["scope"], fmi_path="PathA/credential", post=mock_post_factory("AT_for_path_A"))
         self.assertEqual("AT_for_path_A", result_a["access_token"])
 
         # Acquire token with path B (should NOT get path A's cached token)
-        result_b = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], "PathB/credential", post=mock_post_factory("AT_for_path_B"))
+        result_b = app.acquire_token_for_client(
+            ["scope"], fmi_path="PathB/credential", post=mock_post_factory("AT_for_path_B"))
         self.assertEqual("AT_for_path_B", result_b["access_token"])
         self.assertEqual(result_b[app._TOKEN_SOURCE], app._TOKEN_SOURCE_IDP,
             "Different FMI path should NOT return a cached token from another path")
 
         # Verify path A still returns its own cached token
-        result_a2 = app.acquire_token_for_client_with_fmi_path(
-            ["scope"], "PathA/credential", post=mock_post_factory("should_not_be_used"))
+        result_a2 = app.acquire_token_for_client(
+            ["scope"], fmi_path="PathA/credential", post=mock_post_factory("should_not_be_used"))
         self.assertEqual("AT_for_path_A", result_a2["access_token"])
         self.assertEqual(result_a2[app._TOKEN_SOURCE], app._TOKEN_SOURCE_CACHE,
             "Same FMI path should return cached token")
@@ -846,8 +854,8 @@ def test_fmi_token_does_not_interfere_with_non_fmi_token(self):
             authority="https://login.microsoftonline.com/my_tenant")
 
         # First, cache a token via FMI path
-        app.acquire_token_for_client_with_fmi_path(
-            ["scope"], "some/fmi/path",
+        app.acquire_token_for_client(
+            ["scope"], fmi_path="some/fmi/path",
             post=lambda url, **kwargs: MinimalResponse(
                 status_code=200, text=json.dumps({
                     "access_token": "FMI_AT", "expires_in": 3600})))

tests/test_fmi_e2e.py

@@ -45,8 +45,8 @@ def _get_fmi_credential_from_rma():
         authority=_AUTHORITY_URL,
         http_client=MinimalHttpClient(),
     )
-    result = app.acquire_token_for_client_with_fmi_path(
-        [_FMI_SCOPE_FOR_RMA], _FMI_PATH)
+    result = app.acquire_token_for_client(
+        [_FMI_SCOPE_FOR_RMA], fmi_path=_FMI_PATH)
     if "access_token" not in result:
         raise RuntimeError(
             "Failed to acquire FMI token from RMA: {}: {}".format(
@@ -73,17 +73,17 @@ def test_acquire_and_cache_with_fmi_path(self):
         scopes = [_FMI_SCOPE]
 
         # 1. Acquire token by credential with FMI path
-        result = app.acquire_token_for_client_with_fmi_path(scopes, _FMI_PATH)
+        result = app.acquire_token_for_client(scopes, fmi_path=_FMI_PATH)
         self.assertIn("access_token", result,
-            "acquire_token_for_client_with_fmi_path() failed: {}: {}".format(
+            "acquire_token_for_client(fmi_path=...) failed: {}: {}".format(
                 result.get("error"), result.get("error_description")))
         self.assertNotEqual("", result["access_token"],
-            "acquire_token_for_client_with_fmi_path() returned empty access token")
+            "acquire_token_for_client(fmi_path=...) returned empty access token")
 
         first_token = result["access_token"]
 
         # 2. Verify silent token acquisition works (should retrieve from cache)
-        cache_result = app.acquire_token_for_client_with_fmi_path(scopes, _FMI_PATH)
+        cache_result = app.acquire_token_for_client(scopes, fmi_path=_FMI_PATH)
         self.assertIn("access_token", cache_result,
             "Second call failed: {}: {}".format(
                 cache_result.get("error"), cache_result.get("error_description")))
@@ -124,16 +124,16 @@ def test_acquire_with_assertion_callback_and_fmi_path(self):
         fmi_path = "SomeFmiPath/Path"
 
         # 1. Acquire token by credential with FMI path
-        result = app.acquire_token_for_client_with_fmi_path(scopes, fmi_path)
+        result = app.acquire_token_for_client(scopes, fmi_path=fmi_path)
         self.assertIn("access_token", result,
-            "acquire_token_for_client_with_fmi_path() failed: {}: {}".format(
+            "acquire_token_for_client(fmi_path=...) failed: {}: {}".format(
                 result.get("error"), result.get("error_description")))
         self.assertNotEqual("", result["access_token"],
-            "acquire_token_for_client_with_fmi_path() returned empty access token")
+            "acquire_token_for_client(fmi_path=...) returned empty access token")
         first_token = result["access_token"]
 
         # 2. Verify cached token acquisition works
-        cache_result = app.acquire_token_for_client_with_fmi_path(scopes, fmi_path)
+        cache_result = app.acquire_token_for_client(scopes, fmi_path=fmi_path)
         self.assertIn("access_token", cache_result,
             "Second call failed: {}: {}".format(
                 cache_result.get("error"), cache_result.get("error_description")))
@@ -166,15 +166,15 @@ def test_different_fmi_paths_are_cached_separately(self):
         scopes = [_FMI_SCOPE]
 
         # Acquire token with path A
-        result_a = app.acquire_token_for_client_with_fmi_path(
-            scopes, "PathA/credential")
+        result_a = app.acquire_token_for_client(
+            scopes, fmi_path="PathA/credential")
         self.assertIn("access_token", result_a,
             "Path A acquisition failed: {}: {}".format(
                 result_a.get("error"), result_a.get("error_description")))
 
         # Acquire token with path B — should NOT get path A's cached token
-        result_b = app.acquire_token_for_client_with_fmi_path(
-            scopes, "PathB/credential")
+        result_b = app.acquire_token_for_client(
+            scopes, fmi_path="PathB/credential")
         self.assertIn("access_token", result_b,
             "Path B acquisition failed: {}: {}".format(
                 result_b.get("error"), result_b.get("error_description")))
@@ -183,8 +183,8 @@ def test_different_fmi_paths_are_cached_separately(self):
             "Different FMI path should NOT return cached token from another path")
 
         # Verify path A still returns its own cached token
-        result_a2 = app.acquire_token_for_client_with_fmi_path(
-            scopes, "PathA/credential")
+        result_a2 = app.acquire_token_for_client(
+            scopes, fmi_path="PathA/credential")
         self.assertIn("access_token", result_a2)
         self.assertEqual(
             result_a2.get("token_source"), "cache",
@@ -201,7 +201,7 @@ def test_fmi_token_does_not_interfere_with_non_fmi_token(self):
         scopes = [_FMI_SCOPE]
 
         # Cache a token via FMI path
-        fmi_result = app.acquire_token_for_client_with_fmi_path(scopes, _FMI_PATH)
+        fmi_result = app.acquire_token_for_client(scopes, fmi_path=_FMI_PATH)
         self.assertIn("access_token", fmi_result)
 
         # Regular acquire_token_for_client should NOT get the FMI token
@@ -230,14 +230,14 @@ def test_two_fmi_paths_produce_separate_cache_entries(self):
         path_b = "PathBeta/Credential"
 
         # 1. Acquire token with path A
-        result_a = app.acquire_token_for_client_with_fmi_path(scopes, path_a)
+        result_a = app.acquire_token_for_client(scopes, fmi_path=path_a)
         self.assertIn("access_token", result_a,
             "Path A acquisition failed: {}: {}".format(
                 result_a.get("error"), result_a.get("error_description")))
         token_a = result_a["access_token"]
 
         # 2. Acquire token with path B
-        result_b = app.acquire_token_for_client_with_fmi_path(scopes, path_b)
+        result_b = app.acquire_token_for_client(scopes, fmi_path=path_b)
         self.assertIn("access_token", result_b,
             "Path B acquisition failed: {}: {}".format(
                 result_b.get("error"), result_b.get("error_description")))
@@ -268,12 +268,12 @@ def test_two_fmi_paths_produce_separate_cache_entries(self):
             "ext_cache_key values for different FMI paths must differ")
 
         # 5. Verify each path still returns its own cached token
-        cached_a = app.acquire_token_for_client_with_fmi_path(scopes, path_a)
+        cached_a = app.acquire_token_for_client(scopes, fmi_path=path_a)
         self.assertEqual("cache", cached_a.get("token_source"))
         self.assertEqual(token_a, cached_a["access_token"],
             "Path A should return its own cached token")
 
-        cached_b = app.acquire_token_for_client_with_fmi_path(scopes, path_b)
+        cached_b = app.acquire_token_for_client(scopes, fmi_path=path_b)
         self.assertEqual("cache", cached_b.get("token_source"))
         self.assertEqual(token_b, cached_b["access_token"],
             "Path B should return its own cached token")