Merge branch 'dev' into release-1.35.0

Author: 4gust

.github/workflows/python-package.yml

@@ -17,12 +17,8 @@ jobs:
       # Fake a TRAVIS env so that the pre-existing test cases would behave like before
       TRAVIS: true
       LAB_APP_CLIENT_ID: ${{ secrets.LAB_APP_CLIENT_ID }}
-      LAB_APP_CLIENT_SECRET: ${{ secrets.LAB_APP_CLIENT_SECRET }}
       LAB_APP_CLIENT_CERT_BASE64: ${{ secrets.LAB_APP_CLIENT_CERT_BASE64 }}
       LAB_APP_CLIENT_CERT_PFX_PATH: lab_cert.pfx
-      LAB_OBO_CLIENT_SECRET: ${{ secrets.LAB_OBO_CLIENT_SECRET }}
-      LAB_OBO_CONFIDENTIAL_CLIENT_ID: ${{ secrets.LAB_OBO_CONFIDENTIAL_CLIENT_ID }}
-      LAB_OBO_PUBLIC_CLIENT_ID: ${{ secrets.LAB_OBO_PUBLIC_CLIENT_ID }}
 
     # Derived from https://docs.github.com/en/actions/guides/building-and-testing-python#starting-with-the-python-workflow-template
     runs-on: ubuntu-22.04

azure-pipelines.yml

@@ -37,5 +37,23 @@ steps:
 
 - script: |
     pip install pytest pytest-azurepipelines
-    pytest
-  displayName: 'pytest'
+    mkdir -p test-results
+    set -o pipefail
+    pytest -vv --junitxml=test-results/junit.xml 2>&1 | tee test-results/pytest.log
+  displayName: 'pytest (verbose + junit + log)'
+
+- task: PublishTestResults@2
+  displayName: 'Publish test results'
+  condition: succeededOrFailed()
+  inputs:
+    testResultsFormat: 'JUnit'
+    testResultsFiles: 'test-results/junit.xml'
+    failTaskOnFailedTests: true
+    testRunTitle: 'Python $(python.version) pytest'
+
+- task: PublishPipelineArtifact@1
+  displayName: 'Publish pytest log artifact'
+  condition: succeededOrFailed()
+  inputs:
+    targetPath: 'test-results'
+    artifact: 'pytest-logs-$(python.version)'

msal/__main__.py

@@ -339,6 +339,8 @@ def _main():
             logging.error("Invalid input: %s", e)
         except KeyboardInterrupt:  # Useful for bailing out a stuck interactive flow
             print("Aborted")
+        except Exception as e:
+            logging.error("Error: %s", e)
 
 if __name__ == "__main__":
     _main()

msal/application.py

@@ -961,7 +961,7 @@ def initiate_auth_code_flow(
 
         :param str response_mode:
             OPTIONAL. Specifies the method with which response parameters should be returned.
-            The default value is equivalent to ``query``, which is still secure enough in MSAL Python
+            The default value is equivalent to ``query``, which was still secure enough in MSAL Python
             (because MSAL Python does not transfer tokens via query parameter in the first place).
             For even better security, we recommend using the value ``form_post``.
             In "form_post" mode, response parameters
@@ -973,6 +973,11 @@ def initiate_auth_code_flow(
             `here <https://openid.net/specs/oauth-v2-multiple-response-types-1_0.html#ResponseModes>`
             and `here <https://openid.net/specs/oauth-v2-form-post-response-mode-1_0.html#FormPostResponseMode>`
 
+            .. note::
+                You should configure your web framework to accept form_post responses instead of query responses.
+                While this parameter still works, it will be removed in a future version.
+                Using query-based response modes is less secure and should be avoided.
+
         :return:
             The auth code flow. It is a dict in this form::
 
@@ -991,6 +996,9 @@ def initiate_auth_code_flow(
             3. and then relay this dict and subsequent auth response to
                :func:`~acquire_token_by_auth_code_flow()`.
         """
+        # Note to maintainers: Do not emit warning for the use of response_mode here,
+        # because response_mode=form_post is still the recommended usage for MSAL Python 1.x.
+        # App developers making the right call shall not be disturbed by unactionable warnings.
         client = _ClientWithCcsRoutingInfo(
             {"authorization_endpoint": self.authority.authorization_endpoint},
             self.client_id,

msal/authority.py

@@ -5,9 +5,7 @@
     from urlparse import urlparse
 import logging
 
-
 logger = logging.getLogger(__name__)
-
 # Endpoints were copied from here
 # https://docs.microsoft.com/en-us/azure/active-directory/develop/authentication-national-cloud#azure-ad-authentication-endpoints
 AZURE_US_GOVERNMENT = "login.microsoftonline.us"
@@ -21,6 +19,29 @@
     'login-us.microsoftonline.com',
     AZURE_US_GOVERNMENT,
     ])
+
+# Trusted issuer hosts for OIDC issuer validation
+# Includes all well-known Microsoft identity provider hosts and national clouds
+TRUSTED_ISSUER_HOSTS = frozenset([
+    # Global/Public cloud
+    "login.microsoftonline.com",
+    "login.microsoft.com",
+    "login.windows.net",
+    "sts.windows.net",
+    # China cloud
+    "login.chinacloudapi.cn",
+    "login.partner.microsoftonline.cn",
+    # Germany cloud (legacy)
+    "login.microsoftonline.de",
+    # US Government clouds
+    "login.microsoftonline.us",
+    "login.usgovcloudapi.net",
+    "login-us.microsoftonline.com",
+    "https://login.sovcloud-identity.fr", # AzureBleu
+    "https://login.sovcloud-identity.de", # AzureDelos
+    "https://login.sovcloud-identity.sg", # AzureGovSG
+])
+
 WELL_KNOWN_B2C_HOSTS = [
     "b2clogin.com",
     "b2clogin.cn",
@@ -67,12 +88,12 @@ def __init__(
             performed.
         """
         self._http_client = http_client
+        self._oidc_authority_url = oidc_authority_url
         if oidc_authority_url:
             logger.debug("Initializing with OIDC authority: %s", oidc_authority_url)
             tenant_discovery_endpoint = self._initialize_oidc_authority(
                 oidc_authority_url)
         else:
-            logger.debug("Initializing with Entra authority: %s", authority_url)
             tenant_discovery_endpoint = self._initialize_entra_authority(
                 authority_url, validate_authority, instance_discovery)
         try:
@@ -93,14 +114,22 @@ def __init__(
                 .format(authority_url)
                 ) + " Also please double check your tenant name or GUID is correct."
             raise ValueError(error_message)
-        openid_config.pop("issuer", None)  # Not used in MSAL.py, so remove it therefore no need to validate it
-        logger.debug(
-            'openid_config("%s") = %s', tenant_discovery_endpoint, openid_config)
+        self._issuer = openid_config.get('issuer')
         self.authorization_endpoint = openid_config['authorization_endpoint']
         self.token_endpoint = openid_config['token_endpoint']
         self.device_authorization_endpoint = openid_config.get('device_authorization_endpoint')
         _, _, self.tenant = canonicalize(self.token_endpoint)  # Usually a GUID
 
+        # Validate the issuer if using OIDC authority
+        if self._oidc_authority_url and not self.has_valid_issuer():
+            raise ValueError((
+                "The issuer '{iss}' does not match the authority '{auth}' or a known pattern. "
+                "When using the 'oidc_authority' parameter in ClientApplication, the authority "
+                "will be validated against the issuer from {auth}/.well-known/openid-configuration ."
+                "If using a known Entra authority (e.g. login.microsoftonline.com) the "
+                "'authority' parameter should be used instead of 'oidc_authority'. "
+                ""
+            ).format(iss=self._issuer, auth=oidc_authority_url))
     def _initialize_oidc_authority(self, oidc_authority_url):
         authority, self.instance, tenant = canonicalize(oidc_authority_url)
         self.is_adfs = tenant.lower() == 'adfs'  # As a convention
@@ -175,6 +204,60 @@ def user_realm_discovery(self, username, correlation_id=None, response=None):
             self.__class__._domains_without_user_realm_discovery.add(self.instance)
         return {}  # This can guide the caller to fall back normal ROPC flow
 
+    def has_valid_issuer(self):
+        """
+        Returns True if the issuer from OIDC discovery is valid for this authority.
+
+        An issuer is valid if one of the following is true:
+        - It exactly matches the authority URL (with/without trailing slash)
+        - It has the same scheme and host as the authority (path can be different)
+        - The issuer host is a well-known Microsoft authority host
+        - The issuer host is a regional variant of a well-known host (e.g., westus2.login.microsoft.com)
+        - For CIAM, hosts that end with well-known B2C hosts (e.g., tenant.b2clogin.com) are accepted as valid issuers
+        """
+        if not self._issuer or not self._oidc_authority_url:
+            return False
+
+        # Case 1: Exact match (most common case, normalized for trailing slashes)
+        if self._issuer.rstrip("/") == self._oidc_authority_url.rstrip("/"):
+            return True
+
+        issuer_parsed = urlparse(self._issuer)
+        authority_parsed = urlparse(self._oidc_authority_url)
+        issuer_host = issuer_parsed.hostname.lower() if issuer_parsed.hostname else None
+
+        if not issuer_host:
+            return False
+        
+        # Case 2: Issuer is from a trusted Microsoft host - O(1) lookup
+        if issuer_host in TRUSTED_ISSUER_HOSTS:
+            return True
+
+        # Case 3: Regional variant check - O(1) lookup
+        # e.g., westus2.login.microsoft.com -> extract "login.microsoft.com"
+        dot_index = issuer_host.find(".")
+        if dot_index > 0:
+            potential_base = issuer_host[dot_index + 1:]
+            if "." not in issuer_host[:dot_index]:
+                # 3a: Base host is a trusted Microsoft host
+                if potential_base in TRUSTED_ISSUER_HOSTS:
+                    return True
+                # 3b: Issuer has a region prefix on the authority host
+                #     e.g. issuer=us.someweb.com, authority=someweb.com
+                authority_host = authority_parsed.hostname.lower() if authority_parsed.hostname else ""
+                if potential_base == authority_host:
+                    return True
+
+        # Case 4: Same scheme and host (path can differ)
+        if (authority_parsed.scheme == issuer_parsed.scheme and 
+            authority_parsed.netloc == issuer_parsed.netloc):
+            return True
+        
+        # Case 5: Check if issuer host ends with any well-known B2C host (e.g., tenant.b2clogin.com)
+        if any(issuer_host.endswith(h) for h in WELL_KNOWN_B2C_HOSTS):
+            return True
+
+        return False
 
 def canonicalize(authority_or_auth_endpoint):
     # Returns (url_parsed_result, hostname_in_lowercase, tenant)
@@ -223,4 +306,3 @@ def tenant_discovery(tenant_discovery_endpoint, http_client, **kwargs):
     resp.raise_for_status()
     raise RuntimeError(  # A fallback here, in case resp.raise_for_status() is no-op
         "Unable to complete OIDC Discovery: %d, %s" % (resp.status_code, resp.text))
-