Bypass B2C

Author: rayluo

msal/application.py

@@ -466,9 +466,11 @@ def __init__(
                     self.http_client, validate_authority=False)
             else:
                 raise
-        self._enable_broker = (
-            isinstance(self, PublicClientApplication)  # Exclude Confidential ROPC
-            and sys.platform == "win32" and not self.authority.is_adfs)
+        is_public_app = (isinstance(self, PublicClientApplication) or
+            (isinstance(self, ClientApplication) and not self.client_credential))
+        self._enable_broker = (is_public_app
+            and sys.platform == "win32"
+            and not self.authority.is_adfs and not self.authority._is_b2c)
 
         self.token_cache = token_cache or TokenCache()
         self._region_configured = azure_region
@@ -1225,7 +1227,7 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
                 try:
                     from .wam import _acquire_token_silently
                     response = _acquire_token_silently(
-                        "https://{}/{}".format(self.authority.instance, self.authority.tenant),  # TODO: What about B2C & ADFS?
+                        "https://{}/{}".format(self.authority.instance, self.authority.tenant),
                         self.client_id,
                         account["local_account_id"],
                         scopes,
@@ -1438,14 +1440,14 @@ def acquire_token_by_username_password(
             try:
                 from .wam import _signin_silently, RedirectUriError
                 response = _signin_silently(
-                    "https://{}/{}".format(self.authority.instance, self.authority.tenant),  # TODO: What about B2C?
+                    "https://{}/{}".format(self.authority.instance, self.authority.tenant),
                     self.client_id,
                     scopes,  # Decorated scopes won't work due to offline_access
                     MSALRuntime_Username=username,
                     MSALRuntime_Password=password,
                     validateAuthority="no"
                         if self.authority._validate_authority is False
-                        or self.authority.is_adfs
+                        or self.authority.is_adfs or self.authority._is_b2c
                         else None,
                     claims=claims,
                     )
@@ -1625,12 +1627,12 @@ def acquire_token_interactive(
                 if "welcome_template" in kwargs:
                     logger.debug(kwargs["welcome_template"])  # Experimental
                 response = _signin_interactively(
-                    "https://{}/{}".format(self.authority.instance, self.authority.tenant),  # TODO: What about B2C?
+                    "https://{}/{}".format(self.authority.instance, self.authority.tenant),
                     self.client_id,
                     scopes,
                     validateAuthority="no"
                         if self.authority._validate_authority is False
-                        or self.authority.is_adfs
+                        or self.authority.is_adfs or self.authority._is_b2c
                         else None,
                     login_hint=login_hint,
                     prompt=prompt,

msal/authority.py

@@ -77,10 +77,10 @@ def __init__(self, authority_url, http_client, validate_authority=True):
             authority_url = str(authority_url)
         authority, self.instance, tenant = canonicalize(authority_url)
         parts = authority.path.split('/')
-        is_b2c = any(self.instance.endswith("." + d) for d in WELL_KNOWN_B2C_HOSTS) or (
+        self._is_b2c = any(self.instance.endswith("." + d) for d in WELL_KNOWN_B2C_HOSTS) or (
             len(parts) == 3 and parts[2].lower().startswith("b2c_"))
         self._validate_authority = True if validate_authority is None else bool(validate_authority)
-        if (tenant != "adfs" and (not is_b2c) and self._validate_authority
+        if (tenant != "adfs" and (not self._is_b2c) and self._validate_authority
                 and self.instance not in WELL_KNOWN_AUTHORITY_HOSTS):
             payload = instance_discovery(
                 "https://{}{}/oauth2/v2.0/authorize".format(

tests/test_e2e.py

@@ -683,18 +683,9 @@ def test_adfs2019_onprem_acquire_token_interactive(self):
         config["authority"] = "https://fs.%s.com/adfs" % config["lab_name"]
         config["scope"] = self.adfs2019_scopes
         config["port"] = 8080
-        username_uri = "https://msidlab.com/api/user?usertype=onprem&federationprovider=ADFSv2019"
-        try:
-            import pymsalruntime
-            logger.warning("Absorbing an AssertionError because PyMsalRuntime does not yet support onprem ADFS")
-            with self.assertRaises(AssertionError):  # Expecting a failure because
-                # PyMsalRuntime does not yet support on-prem ADFS.
-                # But if this expectation is not met,
-                # it would mean the latest PyMsalRuntime supports onprem ADFS.
-                # At that time we would revert this patch.
-                self._test_acquire_token_interactive(username_uri=username_uri, **config)
-        except ImportError:  # Then use browser-based interactive flow, which will work
-            self._test_acquire_token_interactive(username_uri=username_uri, **config)
+        self._test_acquire_token_interactive(
+            username_uri="https://msidlab.com/api/user?usertype=onprem&federationprovider=ADFSv2019",
+            **config)
 
     @unittest.skipUnless(
         os.getenv("LAB_OBO_CLIENT_SECRET"),