Rework the previous work stopped at 3-years ago

Author: rayluo

msal/application.py

@@ -624,7 +624,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).
@@ -639,17 +639,26 @@ 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, scope, authority=None, policy=''):
-        the_authority = Authority(authority) if authority else self.authority
-        return oauth2.Client(
-            self.client_id, token_endpoint=the_authority.token_endpoint,
-            default_body=self._build_auth_parameters(
-                self.client_credential, the_authority.token_endpoint,
-                self.client_id)
-            )._get_token(  # TODO: Avoid using internal methods
-                "urn:ietf:params:oauth:grant-type:jwt-bearer",
-                assertion=user_assertion, requested_token_use='on_behalf_of',
-                scope=scope,  # This grant flow requires no scope decoration???
-                query={'p': policy} if policy else None)
+    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 already receives 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 service, on behalf of that user.
+
+        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
+        """
+        # 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=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.
+            data=dict(kwargs.pop("data", {}), requested_token_use="on_behalf_of"),
+            **kwargs)
 

tests/test_application.py

@@ -194,12 +194,3 @@ def test_acquire_token_silent(self):
         self.assertNotEqual(None, at)
         self.assertEqual(self.access_token, at.get('access_token'))
 
-    def test_acquire_token_obo(self):
-        token = self.app.acquire_token_on_behalf_of(
-            self.token['access_token'], self.scope2)
-        error_description = token.get('error_description', "")
-        if 'grant is not supported by this API version' in error_description:
-            raise unittest.SkipTest(
-                "OBO is not yet supported by service: %s" % error_description)
-        self.assertEqual(error_description, "")
-

tests/test_e2e.py

@@ -327,3 +327,33 @@ 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):  # It hardcodes many pre-defined resources
+        obo_client_id = "23c64cd8-21e4-41dd-9756-ab9e2c23f58c"
+        obo_scopes = ["https://graph.microsoft.com/User.Read"]
+        config = get_lab_user(isFederated=False)
+        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
+            )
+        self.assertNotEqual(None, pca_result.get("access_token"), "PCA should work")
+
+        cca = msal.ConfidentialClientApplication(
+            obo_client_id,
+            client_credential=os.getenv("OBO_CLIENT_SECRET"),
+            authority=config.get("authority"))
+        cca_result = cca.acquire_token_on_behalf_of(
+            pca_result['access_token'], obo_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)
+            self.assertEqual(cca_result["access_token"], result["access_token"])
+