test: add e2e tests for iwa flow

Author: shajia-deshaw

msal/application.py

@@ -239,6 +239,7 @@ class ClientApplication(object):
         "You can enable broker by following these instructions. "
         "https://msal-python.readthedocs.io/en/latest/#publicclientapplication")
 
+
     def __init__(
             self, client_id,
             client_credential=None, authority=None, validate_authority=True,
@@ -1889,11 +1890,10 @@ def _acquire_token_by_username_password_federated(
             wstrust_endpoint.get("action"), self.http_client)
         if not ("token" in wstrust_result and "type" in wstrust_result):
             raise RuntimeError("Unsuccessful RSTR. %s" % wstrust_result)
-        GRANT_TYPE_SAML1_1 = 'urn:ietf:params:oauth:grant-type:saml1_1-bearer'
         grant_type = {
-            SAML_TOKEN_TYPE_V1: GRANT_TYPE_SAML1_1,
+            SAML_TOKEN_TYPE_V1: self.client.GRANT_TYPE_SAML1_1,
             SAML_TOKEN_TYPE_V2: self.client.GRANT_TYPE_SAML2,
-            WSS_SAML_TOKEN_PROFILE_V1_1: GRANT_TYPE_SAML1_1,
+            WSS_SAML_TOKEN_PROFILE_V1_1: self.client.GRANT_TYPE_SAML1_1,
             WSS_SAML_TOKEN_PROFILE_V2: self.client.GRANT_TYPE_SAML2
             }.get(wstrust_result.get("type"))
         if not grant_type:
@@ -2387,11 +2387,10 @@ def _acquire_token_by_iwa_federated(
             wstrust_endpoint.get("action"), self.http_client)
         if not ("token" in wstrust_result and "type" in wstrust_result):
             raise RuntimeError("Unsuccessful RSTR. %s" % wstrust_result)
-        GRANT_TYPE_SAML1_1 = 'urn:ietf:params:oauth:grant-type:saml1_1-bearer'
         grant_type = {
-            SAML_TOKEN_TYPE_V1: GRANT_TYPE_SAML1_1,
+            SAML_TOKEN_TYPE_V1: self.client.GRANT_TYPE_SAML1_1,
             SAML_TOKEN_TYPE_V2: self.client.GRANT_TYPE_SAML2,
-            WSS_SAML_TOKEN_PROFILE_V1_1: GRANT_TYPE_SAML1_1,
+            WSS_SAML_TOKEN_PROFILE_V1_1: self.client.GRANT_TYPE_SAML1_1,
             WSS_SAML_TOKEN_PROFILE_V2: self.client.GRANT_TYPE_SAML2
         }.get(wstrust_result.get("type"))
         if not grant_type:
@@ -2405,6 +2404,7 @@ def _acquire_token_by_iwa_federated(
                 event,
                 environment=self.authority.instance,
                 username=username,  # Useful in case IDT contains no such info
+                iwa=True
             )),
             **kwargs)
 

msal/oauth2cli/oauth2.py

@@ -300,6 +300,7 @@ class Client(BaseClient):  # We choose to implement all 4 grants in 1 class
         "DEVICE_CODE": "device_code",
         }
     DEVICE_FLOW_RETRIABLE_ERRORS = ("authorization_pending", "slow_down")
+    GRANT_TYPE_SAML1_1 = 'urn:ietf:params:oauth:grant-type:saml1_1-bearer'
     GRANT_TYPE_SAML2 = "urn:ietf:params:oauth:grant-type:saml2-bearer"  # RFC7522
     GRANT_TYPE_JWT = "urn:ietf:params:oauth:grant-type:jwt-bearer"  # RFC7523
     grant_assertion_encoders = {GRANT_TYPE_SAML2: BaseClient.encode_saml_assertion}

msal/token_cache.py

@@ -240,8 +240,8 @@ def __add(self, event, now=None):
             # Only use decode_id_token() when necessary, it contains time-sensitive validation
             decode_id_token(id_token, client_id=event["client_id"]) if id_token else {})
         client_info, home_account_id = self.__parse_account(response, id_token_claims)
-
         target = ' '.join(sorted(event.get("scope") or []))  # Schema should have required sorting
+        iwa = event.get("iwa", False)  # Integrated Windows Authentication
 
         with self._lock:
             now = int(time.time() if now is None else now)
@@ -277,7 +277,7 @@ def __add(self, event, now=None):
                     at["refresh_on"] = str(now + refresh_in)  # Schema wants a string
                 self.modify(self.CredentialType.ACCESS_TOKEN, at, at)
 
-            if client_info and not event.get("skip_account_creation"):
+            if (client_info or iwa) and not event.get("skip_account_creation"):
                 account = {
                     "home_account_id": home_account_id,
                     "environment": environment,

sample/integrated_windows_authentication_sample.py

@@ -25,6 +25,7 @@
 
 import requests
 import msal
+from msal.token_cache import  TokenCache
 
 
 # Optional logging
@@ -54,11 +55,11 @@ def acquire_and_use_token():
     # Firstly, check the cache to see if this end user has signed in before
     accounts = global_app.get_accounts(username=config["username"])
     if accounts:
-        logging.info("Account(s) exists in cache, probably with token too. Let's try.")
+        print("Account(s) exists in cache, probably with token too. Let's try.")
         result = global_app.acquire_token_silent(config["scope"], account=accounts[0])
 
     if not result:
-        logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
+        print("No suitable token exists in cache. Let's get a new one from AAD.")
         # See this page for constraints of Username Password Flow.
         # https://github.com/AzureAD/microsoft-authentication-library-for-python/wiki/Username-Password-Authentication
         result = global_app.acquire_token_integrated_windows_auth(

tests/test_e2e.py

@@ -230,6 +230,29 @@ def _test_username_password(self,
             )
         return result
 
+    def _test_iwa(self,
+                  authority=None, client_id=None, username=None, scope=None,
+                  client_secret=None,
+                  azure_region=None,
+                  http_client=None,
+                  **ignored):
+        assert authority and client_id and username and scope
+        self.app = self._build_app(
+            client_id, authority=authority,
+            http_client=requests,
+            azure_region=azure_region, client_credential=client_secret)
+        self.assertEqual(
+            self.app.get_accounts(username=username), [], "Cache starts empty")
+        result = self.app.acquire_token_integrated_windows_auth(
+            username, scopes=scope)
+        self.assertLoosely(result)
+        self.assertCacheWorksForUser(
+            result, scope,
+            username=username
+        )
+        return result
+
+
     @unittest.skipIf(
         os.getenv("TRAVIS"),  # It is set when running on TravisCI or Github Actions
         "Although it is doable, we still choose to skip device flow to save time")
@@ -348,6 +371,10 @@ def test_username_password(self):
         self.skipUnlessWithConfig(["client_id", "username", "password", "scope"])
         self._test_username_password(**self.config)
 
+    def test_iwa(self):
+        self.skipUnlessWithConfig(["client_id","username","scope"])
+        return self._test_iwa(**self.config)
+
     def _get_app_and_auth_code(self, scopes=None, **kwargs):
         return _get_app_and_auth_code(
             self.config["client_id"],