Refactor to provide IDT, and demo how to do cache

Author: rayluo

msal/application.py

@@ -656,9 +656,12 @@ def acquire_token_on_behalf_of(self, user_assertion, scopes, **kwargs):
         return self.client.obtain_token_by_assertion(  # bases on assertion RFC 7521
             user_assertion,
             self.client.GRANT_TYPE_JWT,  # IDTs and AAD ATs are all JWTs
-            scope=scopes,  # Without decorate_scope(...), it still gets an AT.
-                # As of 2019, AAD would even issue RT, and ClientInfo i.e. account.
-                # No IDT will be issued. OBO app probably does not need one anyway.
+            scope=decorate_scope(scopes, self.client_id),  # Decoration is used for:
+                # 1. Explicitly requesting an RT, without relying on AAD default
+                #    behavior, even though it currently still issues an RT.
+                # 2. Requesting an IDT (which would otherwise be unavailable)
+                #    so that the calling app could use id_token_claims to implement
+                #    their own cache mapping, which is likely needed in web apps.
             data=dict(kwargs.pop("data", {}), requested_token_use="on_behalf_of"),
             **kwargs)
 

tests/test_e2e.py

@@ -330,30 +330,46 @@ def test_adfs2019_fed_user(self):
     @unittest.skipUnless(
         os.getenv("OBO_CLIENT_SECRET"),
         "Need OBO_CLIENT_SECRET from https://buildautomation.vault.azure.net/secrets/IdentityDivisionDotNetOBOServiceSecret")
-    def test_acquire_token_obo(self):  # It hardcodes many pre-defined resources
+    def test_acquire_token_obo(self):
+        # Some hardcoded, pre-defined settings
         obo_client_id = "23c64cd8-21e4-41dd-9756-ab9e2c23f58c"
-        obo_scopes = ["https://graph.microsoft.com/User.Read"]
+        downstream_scopes = ["https://graph.microsoft.com/User.Read"]
         config = get_lab_user(isFederated=False)
+
+        # 1. An app obtains a token representing a user, for our mid-tier service
         pca = msal.PublicClientApplication(
             "be9b0186-7dfd-448a-a944-f771029105bf", authority=config.get("authority"))
         pca_result = pca.acquire_token_by_username_password(
             config["username"],
             self.get_lab_user_secret(config["lab"]["labname"]),
-            scopes=["%s/access_as_user" % obo_client_id],  # Need setup beforehand
+            scopes=[  # The OBO app's scope. Yours might be different.
+                "%s/access_as_user" % obo_client_id],
             )
-        self.assertNotEqual(None, pca_result.get("access_token"), "PCA should work")
+        self.assertIsNotNone(pca_result.get("access_token"), "PCA should work")
 
+        # 2. Our mid-tier service uses OBO to obtain a token for downstream service
         cca = msal.ConfidentialClientApplication(
             obo_client_id,
             client_credential=os.getenv("OBO_CLIENT_SECRET"),
-            authority=config.get("authority"))
+            authority=config.get("authority"),
+            # token_cache= ...,  # Default token cache is all-tokens-store-in-memory.
+                # That's fine if OBO app uses short-lived msal instance per session.
+                # Otherwise, the OBO app need to implement a one-cache-per-user setup.
+            )
         cca_result = cca.acquire_token_on_behalf_of(
-            pca_result['access_token'], obo_scopes)
+            pca_result['access_token'], downstream_scopes)
         self.assertNotEqual(None, cca_result.get("access_token"), str(cca_result))
 
-        # Cache would also work, with the one-cache-per-user caveat.
-        if len(cca.get_accounts()) == 1:
-            account = cca.get_accounts()[0]  # This test involves only 1 account
-            result = cca.acquire_token_silent(obo_scopes, account)
+        # 3. Now the OBO app can simply store downstream token(s) in same session.
+        #    Alternatively, if you want to persist the downstream AT, and possibly
+        #    the RT (if any) for prolonged access even after your own AT expires,
+        #    now it is the time to persist current cache state for current user.
+        #    Assuming you already did that (which is not shown in this test case),
+        #    the following part shows one of the ways to obtain an AT from cache.
+        username = cca_result.get("id_token_claims", {}).get("preferred_username")
+        self.assertEqual(config["username"], username)
+        if username:  # A precaution so that we won't use other user's token
+            account = cca.get_accounts(username=username)[0]
+            result = cca.acquire_token_silent(downstream_scopes, account)
             self.assertEqual(cca_result["access_token"], result["access_token"])