Merge pull request #92 from AzureAD/obo

On-behalf-of (OBO) Implementation

Author: rayluo

msal/application.py

@@ -627,7 +627,7 @@ def _acquire_token_by_username_password_federated(
 class ConfidentialClientApplication(ClientApplication):  # server-side web app
 
     def acquire_token_for_client(self, scopes, **kwargs):
-        """Acquires token from the service for the confidential client.
+        """Acquires token for the current confidential client, not for an end user.
 
         :param list[str] scopes: (Required)
             Scopes requested to access a protected API (a resource).
@@ -642,6 +642,38 @@ def acquire_token_for_client(self, scopes, **kwargs):
                 scope=scopes,  # This grant flow requires no scope decoration
                 **kwargs)
 
-    def acquire_token_on_behalf_of(self, user_assertion, scopes, authority=None):
-        raise NotImplementedError()
+    def acquire_token_on_behalf_of(self, user_assertion, scopes, **kwargs):
+        """Acquires token using on-behalf-of (OBO) flow.
+
+        The current app is a middle-tier service which was called with a token
+        representing an end user.
+        The current app can use such token (a.k.a. a user assertion) to request
+        another token to access downstream web API, on behalf of that user.
+        See `detail docs here <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow>`_ .
+
+        The current middle-tier app has no user interaction to obtain consent.
+        See how to gain consent upfront for your middle-tier app from this article.
+        https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow#gaining-consent-for-the-middle-tier-application
+
+        :param str user_assertion: The incoming token already received by this app
+        :param list[str] scopes: Scopes required by downstream API (a resource).
+
+        :return: A dict representing the json response from AAD:
+
+            - A successful response would contain "access_token" key,
+            - an error response would contain "error" and usually "error_description".
+        """
+        # The implementation is NOT based on Token Exchange
+        # https://tools.ietf.org/html/draft-ietf-oauth-token-exchange-16
+        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=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

@@ -327,3 +327,49 @@ def test_adfs2019_fed_user(self):
         self._test_username_password(
             password=self.get_lab_user_secret(config["lab"]["labname"]), **config)
 
+    @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):
+        # Some hardcoded, pre-defined settings
+        obo_client_id = "23c64cd8-21e4-41dd-9756-ab9e2c23f58c"
+        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=[  # The OBO app's scope. Yours might be different.
+                "%s/access_as_user" % obo_client_id],
+            )
+        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"),
+            # 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'], downstream_scopes)
+        self.assertNotEqual(None, cca_result.get("access_token"), str(cca_result))
+
+        # 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"])
+