Fail closed on mTLS PoP token_type downgrade; honor verify; lock session build

Harden the mTLS Proof-of-Possession path with three fixes surfaced by review.

Fail closed on a token_type downgrade. A cert-bound request is driven by our
request flag, not the response, so if ESTS answers an mtls_pop request with a
non-cert-bound token we previously still attached binding_certificate and cached
it as bound. Now acquire_token_for_client returns token_type_mismatch when the
response token_type is not "mtls_pop", and _MtlsClient only re-injects key_id
(the cache binding) when the response is actually mtls_pop. A downgraded token
therefore caches as an ordinary token that the cert-bound silent query can never
match, so the flow re-fetches and fails closed every time. Mirrors MSAL .NET/Go.

Honor verify in the mTLS transport. A custom ssl_context bypasses requests' own
verify handling, so verify was silently ignored: verify=False raised a cryptic
ValueError and the default used the system store instead of certifi. The context
builder now matches requests' semantics (certifi default, CA file/dir path, or
disabled) so verify behaves as documented.

Serialize first session build. Wrap the lazy session construction in a
double-checked lock so a cold-start burst on a multi-threaded confidential
client no longer builds N transports (each writing the private key to a temp
file) and orphans all but one.

Also drop stray blank lines that had crept into the acquire_token_for_client
mTLS setup block.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: Robbie-Microsoft

msal/application.py

@@ -300,9 +300,15 @@ def obtain_token_for_client(self, scope=None, **kwargs):
             outer = kwargs.pop("on_obtaining_tokens", None)
 
             def _reinject_key_id(event):
-                # Re-attach key_id to the event's data so token_cache.add()
-                # binds the stored mtls_pop AT to the certificate.
-                event["data"] = dict(event.get("data", {}), key_id=key_id)
+                # Bind the stored AT to the certificate (via key_id) only when
+                # ESTS actually returned a cert-bound token. On a token_type
+                # downgrade (e.g. a Bearer response) leave key_id off, so the
+                # token is never cached as a bound mtls_pop token - it caches as
+                # an ordinary Bearer, and the fail-closed check in
+                # acquire_token_for_client returns token_type_mismatch.
+                response = event.get("response") or {}
+                if (response.get("token_type") or "").lower() == "mtls_pop":
+                    event["data"] = dict(event.get("data", {}), key_id=key_id)
                 (outer or self.on_obtaining_tokens)(event)
 
             kwargs["on_obtaining_tokens"] = _reinject_key_id
@@ -2799,56 +2805,47 @@ def acquire_token_for_client(
             kwargs["data"] = kwargs.get("data", {})
             kwargs["data"]["fmi_path"] = fmi_path
         if mtls_proof_of_possession:
-
             # An mTLS transport is required to present the certificate for a
-
             # cert-bound PoP request.
-
             if self._http_client_is_custom:
-
                 raise ValueError(
-
                     "mtls_proof_of_possession=True is not supported with a "
-
                     "custom http_client, because MSAL must own the TLS transport "
-
                     "to present the client certificate in the mutual-TLS "
-
                     "handshake. Omit the http_client argument to use MSAL's "
-
                     "built-in mTLS transport.")
-
             if self.authority.tenant.lower() in ("common", "organizations"):
-
                 raise ValueError(
-
                     "mtls_proof_of_possession=True requires a tenanted authority. "
-
                     "Use a specific tenant id or domain instead of /common or "
-
                     "/organizations.")
-
             # Parse/validate the certificate now (fail fast).
-
             mtls_cert = self._get_mtls_pop_cert()
-
             # Cert-bound PoP: request an mtls_pop token and bind its cache
-
             # entry to the cert via key_id (base64url x5t#S256). token_type
-
             # also routes _acquire_token_for_client() to the mTLS client.
-
             data = dict(kwargs.get("data") or {})
-
             data["token_type"] = "mtls_pop"
-
             data["key_id"] = mtls_cert["key_id"]
-
             kwargs["data"] = data
 
         result = _clean_up(self._acquire_token_silent_with_error(
             scopes, None, claims_challenge=claims_challenge, **kwargs))
         if mtls_proof_of_possession and result and "access_token" in result:
+            # Fail closed on a token_type downgrade: if ESTS returned a
+            # non-cert-bound token (e.g. a Bearer downgrade) for our mtls_pop
+            # request, do not surface it as bound. The cache side is handled in
+            # _MtlsClient (key_id binds the entry only when the response is
+            # mtls_pop), so a downgraded token is never stored as bound either.
+            # Mirrors the fail-closed behavior of MSAL .NET and MSAL Go.
+            if (result.get("token_type") or "").lower() != "mtls_pop":
+                return {
+                    "error": "token_type_mismatch",
+                    "error_description": (
+                        "The identity provider returned token_type {!r} instead "
+                        "of 'mtls_pop'; the access token is not "
+                        "certificate-bound.".format(result.get("token_type"))),
+                    }
             # Surface the PUBLIC binding certificate (never the private key), so
             # callers can correlate the token to its cert. Survives _clean_up
             # (the key has no "_" prefix).

msal/mtls.py

@@ -12,6 +12,7 @@
 ``RegionAndMtlsDiscoveryProvider`` so MSAL Python stays cross-SDK consistent.
 """
 import logging
+import threading
 try:
     from urllib.parse import urlparse, urlunparse
 except ImportError:  # Python 2
@@ -116,18 +117,25 @@ def __init__(self, cert_pem, key_pem, *,
         self._proxies = proxies
         self._timeout = timeout
         self._session = None
+        self._session_lock = threading.Lock()
 
     def _ensure_session(self):
+        # Double-checked locking: confidential clients are typically
+        # multi-threaded servers, and a cold-start burst must not build N
+        # sessions (each writing the private key to a temp file N times) and
+        # orphan all but one. Mirrors the client-level lock in application.py.
         if self._session is None:
-            import requests  # Lazy import, same as the rest of MSAL
-            session = requests.Session()
-            session.verify = self._verify
-            if self._proxies:
-                session.proxies = self._proxies
-            adapter = _make_mtls_adapter(
-                _create_ssl_context(self._cert_pem, self._key_pem))
-            session.mount("https://", adapter)
-            self._session = session
+            with self._session_lock:
+                if self._session is None:
+                    import requests  # Lazy import, same as the rest of MSAL
+                    session = requests.Session()
+                    session.verify = self._verify
+                    if self._proxies:
+                        session.proxies = self._proxies
+                    adapter = _make_mtls_adapter(_create_ssl_context(
+                        self._cert_pem, self._key_pem, self._verify))
+                    session.mount("https://", adapter)
+                    self._session = session
         return self._session
 
     def post(self, url, **kwargs):
@@ -166,18 +174,43 @@ def proxy_manager_for(self, *args, **kwargs):
     return _MtlsHTTPAdapter(ssl_context)
 
 
-def _create_ssl_context(cert_pem, key_pem):
+def _new_verifying_context(verify):
+    """Build an ssl context whose server verification matches ``verify`` (the
+    same semantics as ``requests``). A custom ssl_context otherwise bypasses
+    ``requests``' own ``verify`` handling, so we honor it here: ``True`` uses
+    certifi (as the rest of MSAL does), a filesystem path uses that CA file/dir,
+    and ``False`` disables verification (instead of the cryptic ValueError that
+    ``requests`` would raise when it tries to set CERT_NONE on this context).
+    """
+    import ssl
+    import os
+    if verify is False:
+        context = ssl.create_default_context()
+        context.check_hostname = False  # Must be cleared before CERT_NONE
+        context.verify_mode = ssl.CERT_NONE
+        return context
+    if isinstance(verify, str):
+        return (ssl.create_default_context(capath=verify)
+                if os.path.isdir(verify)
+                else ssl.create_default_context(cafile=verify))
+    try:  # Default: match requests' trust store (certifi), not the system one
+        import certifi
+        return ssl.create_default_context(cafile=certifi.where())
+    except ImportError:  # pragma: no cover
+        return ssl.create_default_context()
+
+
+def _create_ssl_context(cert_pem, key_pem, verify=True):
     """Build a client ``ssl.SSLContext`` that presents ``cert_pem``/``key_pem``.
 
     ``ssl.SSLContext.load_cert_chain`` requires a file path, but our key is
     in memory. We write a ``0600`` temp PEM (mkstemp defaults to owner-only),
     load it, then unlink it immediately - the context keeps the material in
     memory, so nothing sensitive lingers on disk.
     """
-    import ssl
     import os
     import tempfile
-    context = ssl.create_default_context()  # Verifies the server (ESTS) as usual
+    context = _new_verifying_context(verify)  # Verifies the server (ESTS)
     fd, path = tempfile.mkstemp(suffix=".pem")  # Owner-only (0600) by default
     try:
         # os.fdopen() takes ownership of fd and guarantees it is closed even if

tests/test_application.py

@@ -1670,6 +1670,26 @@ def test_backward_compatible_bearer_is_unchanged(self):
         self.assertEqual("Bearer", result.get("token_type"))
         self.assertNotIn("binding_certificate", result)
 
+    def test_token_type_downgrade_fails_closed(self):
+        # ESTS returns a non-cert-bound token (a Bearer downgrade) for an
+        # mtls_pop request. MSAL must fail closed: an error result, no
+        # binding_certificate, and nothing cached as a bound (key_id) token.
+        app = ConfidentialClientApplication(
+            "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
+        result = app.acquire_token_for_client(
+            ["s1"], mtls_proof_of_possession=True,
+            post=self._capturing_post([], token_type="Bearer"))
+        self.assertEqual("token_type_mismatch", result.get("error"))
+        self.assertNotIn("access_token", result)
+        self.assertNotIn("binding_certificate", result)
+        bound = [
+            at for at in app.token_cache.search(
+                msal.TokenCache.CredentialType.ACCESS_TOKEN)
+            if at.get("key_id")
+            or (at.get("token_type") or "").lower() == "mtls_pop"]
+        self.assertEqual(
+            [], bound, "A downgraded token must never be cached as bound")
+
     def test_regional_mtls_endpoint(self):
         app = ConfidentialClientApplication(
             "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY,