From f98961f60cb82687cee40532e08f57f447e19d5f Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Fri, 31 Jul 2026 16:11:36 -0400
Subject: [PATCH 1/8] Add Bearer-over-mTLS: app-level flag + client-credentials
 path

Introduce an app-level opt-in `send_certificate_over_mtls` on the
confidential-client `client_credential` dict. When set, MSAL presents the
SN/I certificate on the TLS handshake to the [region.]mtlsauth.* endpoint and
signs a private_key_jwt client_assertion (x5c forced), but requests a plain
Bearer token (no token_type=mtls_pop / no key_id / no req_cnf). The result is
an ordinary, non-cert-bound Bearer access token.

- Validate the flag at construction: it requires a certificate credential,
  else ValueError naming the flag.
- Add _get_mtls_bearer_client(): reuses the Task-1 mTLS transport
  (msal/mtls.py) + the login.*->[region.]mtlsauth.* endpoint transform, adds a
  signed x5c-forced assertion, and caches a plain (non-key_id-fenced) token.
- Route acquire_token_for_client() to it when the flag is set and the request
  is not mtls_pop. A per-request mtls_proof_of_possession=True always wins.
- Tests (RED-first) in TestSendCertificateOverMtls: config/builder (default
  false, stored true, flag + non-cert credential raises) and CC behavior
  (global/regional mtlsauth endpoint, client_assertion + jwt-bearer, x5c
  forced, no PoP wire markers, plain-Bearer result, mtls_pop precedence).

Ports MSAL.NET CertificateOptions.SendCertificateOverMtls (client-credentials
leg). OBO/refresh/auth-code coverage follows.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 msal/application.py       | 105 +++++++++++++++++++++++++++++++
 tests/test_application.py | 128 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 233 insertions(+)

diff --git a/msal/application.py b/msal/application.py
index 74395c33..ff0fa384 100644
--- a/msal/application.py
+++ b/msal/application.py
@@ -880,6 +880,11 @@ def get_client_assertion():
         self._mtls_pop_cert_material = None  # Cached parse of the cert below
         self._mtls_client = None  # Lazily built global mTLS client
         self._mtls_regional_client = None  # Lazily built regional mTLS client
+        # Bearer-over-mTLS reuses the same mTLS transport but signs a
+        # private_key_jwt client_assertion (x5c forced) instead of requesting a
+        # cert-bound token; its client(s) are likewise built lazily.
+        self._mtls_bearer_client = None  # Lazily built global Bearer-over-mTLS client
+        self._mtls_bearer_regional_client = None  # Lazily built regional one
         self._mtls_lock = Lock()
         if isinstance(client_credential, dict) and not client_credential.get(
                 "client_assertion") and (
@@ -889,6 +894,20 @@ def get_client_assertion():
         else:  # No certificate available to present over mTLS
             self._mtls_cert_credential = None
 
+        # Bearer-over-mTLS opt-in (app-level): present the certificate on the
+        # TLS handshake to the mTLS endpoint but receive a plain Bearer token
+        # (NOT cert-bound). Requires a certificate credential; a per-request
+        # mtls_proof_of_possession=True always takes precedence over this flag.
+        self._send_certificate_over_mtls = bool(
+            isinstance(client_credential, dict)
+            and client_credential.get("send_certificate_over_mtls"))
+        if self._send_certificate_over_mtls and self._mtls_cert_credential is None:
+            raise ValueError(
+                "send_certificate_over_mtls=True requires this confidential "
+                "client to be configured with a certificate credential (a "
+                "'private_key_pfx_path', or a 'private_key' plus "
+                "'public_certificate').")
+
         # Warn if using a static string/bytes client_assertion (discouraged for long-running apps)
         if isinstance(client_credential, dict) and isinstance(
                 client_credential.get("client_assertion"), (str, bytes)):
@@ -1124,6 +1143,88 @@ def _get_mtls_client(self, central_authority):
                 client = self._mtls_client
         return client
 
+    def _get_mtls_bearer_client(self, central_authority):
+        """Lazily build (and cache) the Bearer-over-mTLS client for
+        ``central_authority``.
+
+        Like :func:`_get_mtls_client`, the token endpoint host is transformed
+        ``login.* -> [region.]mtlsauth.*`` and the certificate is presented over
+        mutual-TLS. UNLIKE mtls_pop, the request also carries a signed
+        private_key_jwt ``client_assertion`` (with the x5c chain FORCED on), and
+        does NOT request a cert-bound token (no ``token_type=mtls_pop`` / no
+        ``key_id`` / no ``req_cnf``). The result is therefore a plain Bearer
+        access token whose cache entry is not fenced by the certificate.
+        """
+        if self._http_client_is_custom:
+            raise ValueError(
+                "send_certificate_over_mtls=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.")
+        region = self._compute_region_to_use()
+        with self._mtls_lock:
+            cached = (
+                self._mtls_bearer_regional_client if region
+                else self._mtls_bearer_client)
+        if cached is not None:
+            return cached
+        cert = self._get_mtls_pop_cert()  # May raise ValueError (no cert configured)
+        token_endpoint = mtls.transform_token_endpoint(  # login.* -> mtlsauth.*
+            central_authority.token_endpoint, central_authority.instance, region)
+        logger.debug("Bearer-over-mTLS token endpoint: %s", token_endpoint)
+        configuration = {
+            "authorization_endpoint": central_authority.authorization_endpoint,
+            "token_endpoint": token_endpoint,
+            "device_authorization_endpoint":
+                central_authority.device_authorization_endpoint,
+            }
+        http_client = ThrottledHttpClient(
+            mtls._MtlsHttpClient(
+                cert["cert_pem"], cert["private_key_pem"],
+                **self._http_client_config),
+            http_cache=self._http_cache,
+            default_throttle_time=5,
+            )
+        # Sign a private_key_jwt assertion, FORCING the x5c chain onto the
+        # header regardless of the app's public_certificate opt-in (Bearer-over-
+        # mTLS always attaches the chain), so ESTS accepts the assertion on the
+        # mTLS endpoint. This is the seam that distinguishes it from mtls_pop,
+        # whose vanilla SN/I request sends no client_assertion at all.
+        assertion = JwtAssertionCreator(
+            cert["private_key_pem"], algorithm="PS256",
+            sha256_thumbprint=cert["sha256_thumbprint"],
+            headers={"x5c": cert["x5c"]})
+        client_assertion = assertion.create_regenerative_assertion(
+            audience=token_endpoint, issuer=self.client_id,
+            additional_claims=self.client_claims or {})
+        client = _ClientWithCcsRoutingInfo(
+            configuration,
+            self.client_id,
+            http_client=http_client,
+            default_headers=self._default_client_headers(),
+            default_body={"client_info": 1},
+            client_assertion=client_assertion,
+            client_assertion_type=Client.CLIENT_ASSERTION_TYPE_JWT,
+            # Cache under the ORIGINAL login.* host (like the rest of the app),
+            # as a PLAIN Bearer AT (no key_id fence), so ordinary Bearer cache
+            # lookups in this app return it.
+            on_obtaining_tokens=lambda event: self.token_cache.add(dict(
+                event, environment=central_authority.instance)),
+            on_removing_rt=self.token_cache.remove_rt,
+            on_updating_rt=self.token_cache.update_rt)
+        with self._mtls_lock:
+            # Double-check under the lock (mirrors _get_mtls_client): another
+            # thread may have built and cached a client while we built ours.
+            if region:
+                if self._mtls_bearer_regional_client is None:
+                    self._mtls_bearer_regional_client = client
+                client = self._mtls_bearer_regional_client
+            else:
+                if self._mtls_bearer_client is None:
+                    self._mtls_bearer_client = client
+                client = self._mtls_bearer_client
+        return client
+
     def _build_client(self, client_credential, authority, skip_regional_client=False):
         client_assertion = None
         client_assertion_type = None
@@ -2984,6 +3085,10 @@ def _acquire_token_for_client(
             # Present the certificate over mTLS to obtain a cert-bound
             # (mtls_pop) token.
             client = self._get_mtls_client(self.authority)
+        elif self._send_certificate_over_mtls:
+            # Bearer-over-mTLS: present the certificate over mTLS but receive a
+            # plain Bearer token. (mtls_pop above always wins over this flag.)
+            client = self._get_mtls_bearer_client(self.authority)
         else:
             client = self._regional_client or self.client
         request_data = kwargs.pop("data", {})
diff --git a/tests/test_application.py b/tests/test_application.py
index e1bf9a02..9ccc0217 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2175,3 +2175,131 @@ def test_common_authority_with_flag_raises(self):
             authority="https://login.microsoftonline.com/common")
         with self.assertRaises(ValueError):
             app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True)
+
+
+_MTLS_CERT_CRED_NO_X5C = {  # A cert cred that does NOT opt into public_certificate
+    "private_key_pfx_path": _MTLS_PFX,
+    "passphrase": "password",
+}
+
+
+def _jwt_header_of(assertion):
+    """Decode the (unverified) JOSE header of a compact-serialized JWT."""
+    if isinstance(assertion, bytes):  # create_normal_assertion returns bytes
+        assertion = assertion.decode("ascii")
+    header_b64 = assertion.split(".")[0]
+    header_b64 += "=" * (-len(header_b64) % 4)  # Restore base64url padding
+    return json.loads(base64.urlsafe_b64decode(header_b64))
+
+
+@patch(_OIDC_DISCOVERY, new=_MTLS_OIDC_MOCK)
+class TestSendCertificateOverMtls(unittest.TestCase):
+    """Bearer-over-mTLS: present the SN/I cert as the client TLS cert on the
+    handshake to the mTLS endpoint, but receive a *plain Bearer* token (the
+    token is NOT cert-bound). Distinct from mtls_pop (which binds the token and
+    fences the cache entry by key_id) and from the classic private_key_jwt
+    Bearer path (which never puts the cert on the TLS handshake)."""
+
+    _AUTHORITY = "https://login.microsoftonline.com/my_tenant"
+
+    def _capturing_post(self, captured, token_type="Bearer"):
+        def mock_post(url, headers=None, data=None, *a, **k):
+            captured.append({
+                "url": url, "data": dict(data or {}), "headers": dict(headers or {})})
+            return MinimalResponse(status_code=200, text=json.dumps({
+                "access_token": "an_at", "expires_in": 3600, "token_type": token_type}))
+        return mock_post
+
+    # --- A. Config / builder -------------------------------------------------
+    def test_flag_defaults_false(self):
+        app = ConfidentialClientApplication(
+            "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
+        self.assertFalse(app._send_certificate_over_mtls)
+
+    def test_flag_stored_true_on_cert_credential(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        self.assertTrue(app._send_certificate_over_mtls)
+
+    def test_flag_with_non_cert_credential_raises_at_construction(self):
+        with self.assertRaises(ValueError) as ctx:
+            ConfidentialClientApplication(
+                "cid", authority=self._AUTHORITY,
+                client_credential={
+                    "client_assertion": "a.static.jwt",
+                    "send_certificate_over_mtls": True})
+        self.assertIn("send_certificate_over_mtls", str(ctx.exception))
+
+    # --- B. Client-credentials behavior -------------------------------------
+    def test_cc_flag_no_region_uses_global_mtls_and_returns_bearer(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        result = app.acquire_token_for_client(
+            ["s1"], post=self._capturing_post(captured))
+        req = captured[0]
+        # Routed to the GLOBAL mtls host (login.* -> mtlsauth.*)
+        self.assertEqual(
+            req["url"],
+            "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
+        # Body: signed private_key_jwt client_assertion is present ...
+        self.assertTrue(req["data"].get("client_assertion"))
+        self.assertEqual(_JWT_BEARER, req["data"].get("client_assertion_type"))
+        # ... but NONE of the PoP wire markers
+        self.assertNotEqual("mtls_pop", req["data"].get("token_type"))
+        self.assertNotIn("req_cnf", req["data"])
+        self.assertNotIn("key_id", req["data"])
+        # Result is a plain Bearer token, NOT cert-bound
+        self.assertEqual("Bearer", result.get("token_type"))
+        self.assertNotIn("binding_certificate", result)
+
+    def test_cc_x5c_is_forced_on_assertion(self):
+        # Bearer-over-mTLS forces the x5c chain onto the assertion header even
+        # when public_certificate was NOT opted into; the regular (no-flag)
+        # path with the same cred omits it. That contrast proves "forced".
+        forced = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED_NO_X5C, send_certificate_over_mtls=True))
+        captured = []
+        forced.acquire_token_for_client(["s1"], post=self._capturing_post(captured))
+        self.assertIn(
+            "x5c", _jwt_header_of(captured[0]["data"]["client_assertion"]),
+            "Bearer-over-mTLS must force x5c onto the assertion header")
+
+        plain = ConfidentialClientApplication(
+            "cid", client_credential=_MTLS_CERT_CRED_NO_X5C, authority=self._AUTHORITY)
+        plain_captured = []
+        plain.acquire_token_for_client(["s1"], post=self._capturing_post(plain_captured))
+        self.assertNotIn(
+            "x5c", _jwt_header_of(plain_captured[0]["data"]["client_assertion"]),
+            "The no-flag path must NOT force x5c (contrast)")
+
+    def test_cc_flag_with_region_uses_regional_mtls(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, azure_region="westus3",
+            client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        app.acquire_token_for_client(["s1"], post=self._capturing_post(captured))
+        self.assertEqual(
+            captured[0]["url"],
+            "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
+
+    def test_cc_per_request_mtls_pop_wins_over_flag(self):
+        # Precedence: a per-request mtls_proof_of_possession=True ALWAYS wins
+        # over the app-level Bearer-over-mTLS flag.
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        result = app.acquire_token_for_client(
+            ["s1"], mtls_proof_of_possession=True,
+            post=self._capturing_post(captured, token_type="mtls_pop"))
+        req = captured[0]
+        self.assertEqual("mtls_pop", req["data"].get("token_type"))
+        # The mtls_pop path is vanilla SN/I: no signed assertion on the wire
+        self.assertNotIn("client_assertion", req["data"])
+        self.assertEqual("mtls_pop", result.get("token_type"))
+        self.assertTrue(result.get("binding_certificate"))

From 93faa2fa30fe9df7953ecd81cdb69c88607f0129 Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Fri, 31 Jul 2026 16:20:51 -0400
Subject: [PATCH 2/8] Route OBO/refresh/auth-code through Bearer-over-mTLS when
 flag is set

Extend the app-level send_certificate_over_mtls opt-in beyond
client-credentials to every confidential-client user flow: on-behalf-of,
refresh-token, and authorization-code.

Add _client_for_confidential_request(): when the flag is set it returns the
Bearer-over-mTLS client (SN/I cert on the TLS handshake, routed to the
[region.]mtlsauth.microsoft.com endpoint, private_key_jwt client_assertion
with the x5c chain forced, plain Bearer token back), otherwise the regular
client. Swap it in at the three user-flow call sites. Client-credentials
routing stays in _acquire_token_for_client, where a per-request
mtls_proof_of_possession still wins over the app-level flag.

Tests (tests/test_application.py, TestSendCertificateOverMtls): OBO / refresh
/ auth-code with the flag hit the global mtlsauth endpoint, carry
client_assertion + client_assertion_type=jwt-bearer with no key_id/req_cnf,
and use the correct grant (on_behalf_of / refresh_token / authorization_code);
OBO with the flag + a configured region hits the regional mtlsauth host; OBO
without the flag stays on the regular login endpoint (negative control).

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 msal/application.py       | 17 ++++++--
 tests/test_application.py | 81 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 95 insertions(+), 3 deletions(-)

diff --git a/msal/application.py b/msal/application.py
index ff0fa384..f6b835aa 100644
--- a/msal/application.py
+++ b/msal/application.py
@@ -1225,6 +1225,17 @@ def _get_mtls_bearer_client(self, central_authority):
                 client = self._mtls_bearer_client
         return client
 
+    def _client_for_confidential_request(self):
+        """Pick the OAuth2 client for a confidential-client user flow (OBO,
+        refresh-token, auth-code): the Bearer-over-mTLS client when the
+        app-level send_certificate_over_mtls flag is set, else the regular
+        client. (Client-credentials routing lives in _acquire_token_for_client,
+        which additionally honors a per-request mtls_pop that wins over the flag.)
+        """
+        if self._send_certificate_over_mtls:
+            return self._get_mtls_bearer_client(self.authority)
+        return self.client
+
     def _build_client(self, client_credential, authority, skip_regional_client=False):
         client_assertion = None
         client_assertion_type = None
@@ -1683,7 +1694,7 @@ def acquire_token_by_authorization_code(
                 self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID)
             _data = kwargs.pop("data", {})
             _stash_client_claims(forwarded_client_claims, _data)
-            response = _clean_up(self.client.obtain_token_by_authorization_code(
+            response = _clean_up(self._client_for_confidential_request().obtain_token_by_authorization_code(
                 code, redirect_uri=redirect_uri,
                 scope=self._decorate_scope(scopes),
                 headers=telemetry_context.generate_headers(),
@@ -2312,7 +2323,7 @@ def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs):
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_BY_REFRESH_TOKEN,
             refresh_reason=msal.telemetry.FORCE_REFRESH)
-        response = _clean_up(self.client.obtain_token_by_refresh_token(
+        response = _clean_up(self._client_for_confidential_request().obtain_token_by_refresh_token(
             refresh_token,
             scope=self._decorate_scope(scopes),
             headers=telemetry_context.generate_headers(),
@@ -3166,7 +3177,7 @@ def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=No
         _data = kwargs.pop("data", {})
         _stash_client_claims(forwarded_client_claims, _data)
         # The implementation is NOT based on Token Exchange (RFC 8693)
-        response = _clean_up(self.client.obtain_token_by_assertion(  # bases on assertion RFC 7521
+        response = _clean_up(self._client_for_confidential_request().obtain_token_by_assertion(  # bases on assertion RFC 7521
             user_assertion,
             self.client.GRANT_TYPE_JWT,  # IDTs and AAD ATs are all JWTs
             scope=self._decorate_scope(scopes),  # Decoration is used for:
diff --git a/tests/test_application.py b/tests/test_application.py
index 9ccc0217..95f43175 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2303,3 +2303,84 @@ def test_cc_per_request_mtls_pop_wins_over_flag(self):
         self.assertNotIn("client_assertion", req["data"])
         self.assertEqual("mtls_pop", result.get("token_type"))
         self.assertTrue(result.get("binding_certificate"))
+
+    # --- C. User flows (OBO / refresh / auth-code) --------------------------
+    def _capturing_post_full(self, captured):
+        # A capturing post that returns a full user-token response (AT +
+        # id_token + client_info), needed by flows that create an account.
+        def mock_post(url, headers=None, data=None, *a, **k):
+            captured.append({
+                "url": url, "data": dict(data or {}), "headers": dict(headers or {})})
+            return MinimalResponse(status_code=200, text=_build_user_token_response())
+        return mock_post
+
+    def test_obo_flag_uses_global_mtls_and_obo_grant(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        app.acquire_token_on_behalf_of(
+            "a_user_assertion", ["s1"], post=self._capturing_post(captured))
+        req = captured[0]
+        self.assertEqual(
+            req["url"],
+            "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
+        self.assertEqual("on_behalf_of", req["data"].get("requested_token_use"))
+        self.assertTrue(req["data"].get("client_assertion"))
+        self.assertEqual(_JWT_BEARER, req["data"].get("client_assertion_type"))
+        self.assertNotEqual("mtls_pop", req["data"].get("token_type"))
+        self.assertNotIn("key_id", req["data"])
+
+    def test_obo_flag_with_region_uses_regional_mtls(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, azure_region="westus3",
+            client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        app.acquire_token_on_behalf_of(
+            "a_user_assertion", ["s1"], post=self._capturing_post(captured))
+        self.assertEqual(
+            captured[0]["url"],
+            "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
+
+    def test_refresh_flag_uses_global_mtls_and_refresh_grant(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        app.acquire_token_by_refresh_token(
+            "a_rt", ["s1"], post=self._capturing_post(captured))
+        req = captured[0]
+        self.assertEqual(
+            req["url"],
+            "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
+        self.assertEqual("refresh_token", req["data"].get("grant_type"))
+        self.assertTrue(req["data"].get("client_assertion"))
+        self.assertNotEqual("mtls_pop", req["data"].get("token_type"))
+
+    def test_auth_code_flag_uses_global_mtls_and_auth_code_grant(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        captured = []
+        app.acquire_token_by_authorization_code(
+            "a_code", ["s1"], post=self._capturing_post_full(captured))
+        req = captured[0]
+        self.assertEqual(
+            req["url"],
+            "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
+        self.assertEqual("authorization_code", req["data"].get("grant_type"))
+        self.assertTrue(req["data"].get("client_assertion"))
+        self.assertNotEqual("mtls_pop", req["data"].get("token_type"))
+
+    def test_obo_without_flag_uses_regular_login_endpoint(self):
+        # Negative control: without the app-level flag, OBO stays on the
+        # regular login.* endpoint (no mTLS routing).
+        app = ConfidentialClientApplication(
+            "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
+        captured = []
+        app.acquire_token_on_behalf_of(
+            "a_user_assertion", ["s1"], post=self._capturing_post(captured))
+        self.assertEqual(
+            captured[0]["url"],
+            "https://login.microsoftonline.com/my_tenant/oauth2/v2.0/token")

From bf79927f719a5408c7a01b19c3f62f9d0a2c8b8a Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Fri, 31 Jul 2026 16:26:45 -0400
Subject: [PATCH 3/8] Add Bearer-over-mTLS second-call cache-hit regression

Guard the .NET MtlsBearerUserFlowTests second-call regression, adapted to
msal-python's cache seam: a 2nd Bearer-over-mTLS client-credentials call is
served from cache with no 2nd network round-trip and no crash on
region/instance-metadata resolution, returning the cached plain Bearer AT
(token_type=Bearer, NOT cert-bound).

In .NET the 2nd call crashes because the AT is cached under the mtlsauth.*
host and instance-metadata discovery rejects that host. msal-python
structurally avoids this by decoupling the cache/metadata authority (login.*,
where the plain Bearer AT lives) from the transport endpoint (mtlsauth.*): the
2nd lookup resolves login.* -- the normal, always-safe path -- so the cached
Bearer AT is returned without feeding the mtlsauth.* host into discovery.

Only client-credentials auto-serves from cache in msal-python (OBO, refresh,
and auth-code have no silent-first lookup), so CC is the flow that exercises
this regression.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 tests/test_application.py | 27 +++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/tests/test_application.py b/tests/test_application.py
index 95f43175..88fd6d76 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2384,3 +2384,30 @@ def test_obo_without_flag_uses_regular_login_endpoint(self):
         self.assertEqual(
             captured[0]["url"],
             "https://login.microsoftonline.com/my_tenant/oauth2/v2.0/token")
+
+    def test_cc_second_call_is_bearer_cache_hit(self):
+        # Regression (mirrors .NET MtlsBearerUserFlowTests' 2nd-call test,
+        # adapted to msal-python's cache seam): a 2nd Bearer-over-mTLS call is
+        # served from cache with no 2nd network round-trip and no crash on
+        # region/instance-metadata resolution. In .NET the crash arises because
+        # the AT is cached under the mtlsauth.* host and instance discovery
+        # rejects that host. msal-python decouples the cache/metadata authority
+        # (login.*, where this plain Bearer AT lives) from the transport
+        # endpoint (mtlsauth.*), so the 2nd lookup resolves login.* -- the
+        # normal, always-safe path -- and returns the cached plain Bearer AT
+        # (NOT cert-bound). Only client-credentials auto-serves from cache in
+        # msal-python (OBO/refresh/auth-code have no silent-first lookup), so CC
+        # is the flow that exercises this regression.
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        first = app.acquire_token_for_client(
+            ["s1"], post=self._capturing_post([]))
+        self.assertEqual("Bearer", first.get("token_type"))
+        second = []
+        result = app.acquire_token_for_client(
+            ["s1"], post=self._capturing_post(second))
+        self.assertEqual([], second, "Second call must be served from cache")
+        self.assertEqual(app._TOKEN_SOURCE_CACHE, result[app._TOKEN_SOURCE])
+        self.assertEqual("Bearer", result.get("token_type"))
+        self.assertNotIn("binding_certificate", result)

From ab024e7db5ef4e9f27cc3036cbd843dfc2952042 Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Fri, 31 Jul 2026 16:43:16 -0400
Subject: [PATCH 4/8] Add Bearer-over-mTLS end-to-end cells

Cover the send_certificate_over_mtls feature in tests/test_e2e.py, mirroring
MSAL .NET's live + capture split and reusing the SNI allow-listed app/authority
constants already in PublicCloudScenariosTestCase.

- test_credential_x509_send_certificate_over_mtls_output_bearer: LIVE
  client-credentials acquisition (allow-listed app, westus3, Key Vault resource)
  that presents the SN/I cert on the TLS handshake to the mTLS endpoint and
  asserts a PLAIN Bearer token back (token_type=Bearer, binding_certificate
  absent) plus a normal flagless cache hit -- the caller-visible proof that the
  AT is an ordinary Bearer entry, not a key-id-fenced mtls_pop entry.
- test_send_certificate_over_mtls_user_flows_capture_request: request-capture
  for OBO, refresh-token, and auth-code over the real lab SN/I certificate --
  asserts each builds the mtlsauth endpoint + a private_key_jwt client_assertion
  + the flow's own grant with none of the mtls_pop markers, without needing
  those (non-mTLS-enabled) apps allow-listed.
- test_send_certificate_over_mtls_user_flows_live_acquire: skip-gated live
  OBO/refresh placeholder documenting the pending server-side app
  mTLS-enablement (AADSTS700027 / AADSTS392189), mirroring MSAL .NET's [Ignore].

All three are gated (lab cert / explicit skip), collect cleanly, and skip
locally; the live and capture cells run in the lab pipeline.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 tests/test_e2e.py | 99 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 99 insertions(+)

diff --git a/tests/test_e2e.py b/tests/test_e2e.py
index 200c5cdb..ef3a76d4 100644
--- a/tests/test_e2e.py
+++ b/tests/test_e2e.py
@@ -567,6 +567,105 @@ def test_credential_x509_output_bearer(self):
         self.assertIsNone(result.get("binding_certificate"))
         self.assertCacheWorksForApp(result, self._MTLS_POP_SCOPE)
 
+    # ── Bearer-over-mTLS (send_certificate_over_mtls) ──────────────────────────
+    # Present the SN/I cert as the client TLS certificate ON the handshake to the
+    # mTLS endpoint, but receive a *plain Bearer* token (the cert authenticates
+    # the transport; the token is NOT cert-bound). Opt-in via the app-level
+    # client_credential flag send_certificate_over_mtls=True. Distinct from the
+    # two cells above: mtls_pop BINDS the token; the classic X509 Bearer cell
+    # signs a private_key_jwt to the REGULAR login.* endpoint (cert never on the
+    # TLS handshake). Ports MSAL .NET CertificateOptions.SendCertificateOverMtls.
+    def test_credential_x509_send_certificate_over_mtls_output_bearer(self):
+        # LIVE client-credentials acquisition, allow-listed (mirrors the MSAL
+        # .NET live CC cell: westus3 + the Key Vault resource). Proves the cert
+        # goes on the TLS handshake to mtlsauth.* yet ESTS returns a plain Bearer.
+        from tests.lab_config import get_client_certificate
+        if not clean_env("LAB_APP_CLIENT_CERT_PFX_PATH"):
+            self.skipTest("LAB_APP_CLIENT_CERT_PFX_PATH not set")
+        scope = ["https://vault.azure.net/.default"]
+        self.app = msal.ConfidentialClientApplication(
+            self._SNI_ALLOWLISTED_CLIENT_ID,
+            authority=self._SNI_ALLOWLISTED_AUTHORITY,
+            azure_region="westus3",
+            client_credential=dict(
+                get_client_certificate(), send_certificate_over_mtls=True))
+        result = self.app.acquire_token_for_client(scope)
+        self.assertIn(
+            "access_token", result,
+            "Bearer-over-mTLS request failed: %s" % result)
+        # A PLAIN Bearer token, NOT cert-bound (contrast the mtls_pop cell).
+        self.assertEqual("Bearer", result.get("token_type"))
+        self.assertNotIn("binding_certificate", result)
+        self.assertIsNone(result.get("binding_certificate"))
+        # The AT is an ordinary Bearer entry (no key_id/thumbprint fence), so a
+        # normal flagless acquire_token_for_client lookup returns it from cache
+        # -- the caller-visible proof of the feature's goal. (mtls_pop needed the
+        # special _assert_pop_cache_hit because its AT is key-id-fenced; Bearer-
+        # over-mTLS deliberately is not.)
+        self.assertCacheWorksForApp(result, scope)
+
+    def test_send_certificate_over_mtls_user_flows_capture_request(self):
+        # The allow-listed app is entitled to Bearer-over-mTLS for
+        # client-credentials only; the OBO/refresh/auth-code apps are not
+        # mTLS-enabled server-side (AADSTS700027 / AADSTS392189), exactly as
+        # MSAL .NET [Ignore]s those live. So prove MSAL builds the correct wire
+        # request for every confidential user flow -- the mtlsauth endpoint + a
+        # private_key_jwt client_assertion + the flow's own grant, with none of
+        # the mtls_pop markers -- by capturing the outgoing POST over the REAL
+        # lab SN/I certificate, without needing those apps allow-listed.
+        from tests.lab_config import get_client_certificate
+        if not clean_env("LAB_APP_CLIENT_CERT_PFX_PATH"):
+            self.skipTest("LAB_APP_CLIENT_CERT_PFX_PATH not set")
+        cred = dict(
+            get_client_certificate(), send_certificate_over_mtls=True)
+        def _capture(sink):
+            def post(url, headers=None, data=None, *a, **k):
+                sink.append({"url": url, "data": dict(data or {})})
+                return MinimalResponse(status_code=400, text=json.dumps({
+                    "error": "invalid_grant",
+                    "error_description": "request captured; live acquisition "
+                        "not required for this app"}))
+            return post
+        def _new_app():
+            return msal.ConfidentialClientApplication(
+                self._SNI_ALLOWLISTED_CLIENT_ID,
+                authority=self._SNI_ALLOWLISTED_AUTHORITY,
+                client_credential=cred)
+        scope = ["https://graph.microsoft.com/.default"]
+        def _assert_mtls(req, grant_key, grant_value):
+            self.assertTrue(
+                req["url"].startswith("https://mtlsauth.microsoft.com/"),
+                "Expected the mTLS endpoint, got %s" % req["url"])
+            self.assertEqual(grant_value, req["data"].get(grant_key))
+            self.assertTrue(req["data"].get("client_assertion"))
+            self.assertNotEqual("mtls_pop", req["data"].get("token_type"))
+            self.assertNotIn("key_id", req["data"])
+            self.assertNotIn("req_cnf", req["data"])
+        cap = []
+        _new_app().acquire_token_on_behalf_of(
+            "a_user_assertion", scope, post=_capture(cap))
+        _assert_mtls(cap[0], "requested_token_use", "on_behalf_of")
+        cap = []
+        _new_app().acquire_token_by_refresh_token(
+            "a_refresh_token", scope, post=_capture(cap))
+        _assert_mtls(cap[0], "grant_type", "refresh_token")
+        cap = []
+        _new_app().acquire_token_by_authorization_code(
+            "an_auth_code", scope, post=_capture(cap))
+        _assert_mtls(cap[0], "grant_type", "authorization_code")
+
+    def test_send_certificate_over_mtls_user_flows_live_acquire(self):
+        # Live OBO/refresh acquisition with Bearer-over-mTLS is pending the
+        # downstream apps being mTLS-enabled server-side; today ESTS rejects it
+        # (AADSTS700027 / AADSTS392189) -- the same class of block as the FIC
+        # leg-1 gate (AADSTS51000) and MSAL .NET's [Ignore]'d live user-flow
+        # cells. Kept as an explicit, self-documenting gate that flips to a real
+        # acquisition once the apps are enabled; the request shape is already
+        # proven by test_send_certificate_over_mtls_user_flows_capture_request.
+        self.skipTest(
+            "pending app mTLS-enablement for OBO/refresh Bearer-over-mTLS "
+            "(AADSTS700027 / AADSTS392189)")
+
 
 class DeviceFlowTestCase(E2eTestCase):  # A leaf class so it will be run only once
     @classmethod

From 92421b0bb96f5584ca0976eed72be7a44feff031 Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Fri, 31 Jul 2026 16:45:05 -0400
Subject: [PATCH 5/8] Document the send_certificate_over_mtls client_credential
 flag

Add a docstring admonition to the client_credential parameter describing the
Bearer-over-mTLS opt-in: set send_certificate_over_mtls=True on the certificate
credential dictionary to present the cert on the mTLS handshake and receive a
plain (non-cert-bound) Bearer token. Notes that it covers all four confidential
flows, defaults to False, requires a certificate credential, and that a
per-request mtls_proof_of_possession=True always takes precedence.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 msal/application.py | 27 +++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/msal/application.py b/msal/application.py
index f6b835aa..2d2649f4 100644
--- a/msal/application.py
+++ b/msal/application.py
@@ -545,6 +545,33 @@ def get_client_assertion():
                 If your .pfx file contains both the private key and public cert,
                 you can opt in for Subject Name/Issuer Auth by setting "public_certificate" to ``True``.
 
+            .. admonition:: Sending the certificate over mTLS for a Bearer token
+
+                A certificate credential can additionally opt in to
+                **Bearer-over-mTLS**: MSAL presents the certificate as the client
+                certificate on the mutual-TLS (mTLS) handshake to Microsoft
+                Entra's mTLS endpoint, but requests an ordinary, *non-cert-bound*
+                Bearer access token (contrast ``mtls_proof_of_possession``, which
+                binds the token). Set ``send_certificate_over_mtls`` to ``True``
+                on the certificate credential dictionary::
+
+                    {
+                        "private_key_pfx_path": "/path/to/your.pfx",
+                        "public_certificate": True,
+                        "send_certificate_over_mtls": True,
+                    }
+
+                It applies to every confidential-client flow
+                (:func:`~acquire_token_for_client`,
+                :func:`~acquire_token_on_behalf_of`,
+                :func:`~acquire_token_by_authorization_code`, and
+                :func:`~acquire_token_by_refresh_token`), defaults to ``False``
+                (so leaving it unset changes no behavior), and requires a
+                certificate credential -- otherwise construction raises
+                ``ValueError``. A per-request ``mtls_proof_of_possession=True`` on
+                :func:`~acquire_token_for_client` always takes precedence, yielding
+                a cert-bound ``mtls_pop`` token instead of a Bearer one.
+
         :type client_credential: Union[dict, str, None]
 
         :param dict client_claims:

From 60a9f248622f1a219cdc0ad01cc90f277aefa9a6 Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Fri, 31 Jul 2026 16:49:59 -0400
Subject: [PATCH 6/8] Harden Bearer-over-mTLS 2nd-call cache test with an
 env-lock assertion

Add a negative assertion to test_cc_second_call_is_bearer_cache_hit that the
cached plain Bearer AT is stored under the ORIGINAL login.* host and its
environment never contains "mtlsauth". This is the assertion that would fail if
the cache seam were ever switched to the .NET-style mtlsauth.* environment
(Option A), which would silently break msal-python's login.*-keyed cache lookup
and defeat the 2nd-call cache hit. Also assert result1 == result2 (the same
cached AT is returned) alongside the existing exactly-one-network-POST guard.

Mirrors the cross-SDK finding that a 2nd-call cache test can pass without
actually proving the cache environment; this makes the env dependence explicit
and deterministic.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 tests/test_application.py | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/tests/test_application.py b/tests/test_application.py
index 88fd6d76..f9374a62 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2408,6 +2408,19 @@ def test_cc_second_call_is_bearer_cache_hit(self):
         result = app.acquire_token_for_client(
             ["s1"], post=self._capturing_post(second))
         self.assertEqual([], second, "Second call must be served from cache")
+        # result1 == result2: the SAME cached AT is returned on the 2nd call.
+        self.assertEqual(first.get("access_token"), result.get("access_token"))
+        # Env-lock (hardens Option B, per the cross-SDK 2nd-call finding): the
+        # plain Bearer AT is cached under the ORIGINAL login.* host, NEVER the
+        # rewritten mtlsauth.* endpoint. This is the assertion that would fail if
+        # anyone re-cached under mtlsauth.* (Option A) -- which would silently
+        # break the login.*-keyed lookup that served the 2nd call above.
+        at_entries = list(app.token_cache.search(
+            msal.TokenCache.CredentialType.ACCESS_TOKEN, query={}))
+        self.assertEqual(1, len(at_entries), "exactly one plain Bearer AT cached")
+        cached_env = at_entries[0].get("environment")
+        self.assertEqual("login.microsoftonline.com", cached_env)
+        self.assertNotIn("mtlsauth", cached_env or "")
         self.assertEqual(app._TOKEN_SOURCE_CACHE, result[app._TOKEN_SOURCE])
         self.assertEqual("Bearer", result.get("token_type"))
         self.assertNotIn("binding_certificate", result)

From 6e2c371473a65ff9576cff88084eabd58f65099d Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Wed, 21 Oct 2026 16:18:22 -0400
Subject: [PATCH 7/8] Address mTLS review feedback

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 docs/index.rst               |   3 +
 msal/application.py          |  71 +++++++-----
 msal/mtls.py                 |  64 +++++------
 msal/token_cache.py          |  87 ++++++++------
 tests/test_application.py    | 213 ++++++++++++++++++++++++++++++++---
 tests/test_mtls_transport.py | 173 +++++++++++++++++++++++++++-
 tests/test_token_cache.py    | 107 +++++++++++++++++-
 7 files changed, 602 insertions(+), 116 deletions(-)

diff --git a/docs/index.rst b/docs/index.rst
index bfb71ada..ec0efffe 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -198,6 +198,9 @@ Notes and requirements:
 * The ``authority`` must be **tenanted** (not ``/common`` or ``/organizations``).
 * MSAL must own the TLS transport, so a custom ``http_client`` is not supported
   together with ``mtls_proof_of_possession=True``.
+  A per-request ``post`` override is also rejected.
+* If the provider returns a different token type, MSAL returns
+  ``token_type_mismatch`` without caching the rejected response.
 * ``azure_region`` is **optional**: omit it to use the global mTLS endpoint
   (``mtlsauth.microsoft.com``); set it (e.g. ``westus3``) only to pin a region.
 * The existing SN/I + Bearer (client-assertion) flow is unchanged; the same
diff --git a/msal/application.py b/msal/application.py
index 477c9675..0e2f7576 100644
--- a/msal/application.py
+++ b/msal/application.py
@@ -130,12 +130,17 @@ def _parse_pfx(pfx_path, passphrase_bytes):
     # Cert concepts https://security.stackexchange.com/a/226758/125264
     from cryptography.hazmat.primitives.serialization import pkcs12
     with open(pfx_path, 'rb') as f:
-        private_key, cert, _ = pkcs12.load_key_and_certificates(  # cryptography 2.5+
+        private_key, cert, additional_certs = pkcs12.load_key_and_certificates(  # cryptography 2.5+
             # https://cryptography.io/en/latest/hazmat/primitives/asymmetric/serialization/#cryptography.hazmat.primitives.serialization.pkcs12.load_key_and_certificates
             f.read(), passphrase_bytes)
     if not (private_key and cert):
         raise ValueError("Your PFX file shall contain both private key and cert")
     sha256_thumbprint, sha1_thumbprint, x5c, cert_pem = _extract_cert_and_thumbprints(cert)
+    for additional_cert in additional_certs or []:
+        _, _, additional_x5c, additional_pem = _extract_cert_and_thumbprints(
+            additional_cert)
+        x5c.extend(additional_x5c)
+        cert_pem += additional_pem
     return private_key, sha256_thumbprint, sha1_thumbprint, x5c, cert_pem
 
 
@@ -182,7 +187,7 @@ def _load_mtls_cert_material(cert_credential):
 
     ``cert_credential`` is the app's main ``client_credential`` (vanilla SN/I),
     a certificate credential dict. Returns a dict with the unencrypted-PEM private key,
-    the leaf cert PEM, ``x5c``, the SHA-256 thumbprint (hex), and ``key_id``
+    the certificate chain PEM, ``x5c``, the SHA-256 thumbprint (hex), and ``key_id``
     (base64url ``x5t#S256``, used for cache binding). Raises ``ValueError`` when
     the credential cannot yield mTLS-capable certificate material.
     """
@@ -202,6 +207,8 @@ def _load_mtls_cert_material(cert_credential):
         from cryptography import x509
         cert = x509.load_pem_x509_certificate(_str2bytes(public_cert))
         sha256_thumbprint, _, x5c, cert_pem = _extract_cert_and_thumbprints(cert)
+        x5c = extract_certs(public_cert)
+        cert_pem = public_cert
         private_key = (
             _load_private_key_from_pem_str(
                 cert_credential['private_key'], passphrase_bytes)
@@ -348,13 +355,12 @@ def _reinject_key_id(event):
                 # 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
+                # response is never cached, 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)
+                    (outer or self.on_obtaining_tokens)(event)
 
             kwargs["on_obtaining_tokens"] = _reinject_key_id
         return super(_MtlsClient, self).obtain_token_for_client(
@@ -576,10 +582,12 @@ def get_client_assertion():
                 It applies to every confidential-client flow
                 (:func:`~acquire_token_for_client`,
                 :func:`~acquire_token_on_behalf_of`,
+                :func:`~acquire_token_by_auth_code_flow`,
                 :func:`~acquire_token_by_authorization_code`, and
-                :func:`~acquire_token_by_refresh_token`), defaults to ``False``
-                (so leaving it unset changes no behavior), and requires a
-                certificate credential -- otherwise construction raises
+                :func:`~acquire_token_by_refresh_token`), including refresh-token
+                redemption during :func:`~acquire_token_silent`. It defaults to
+                ``False`` (so leaving it unset changes no behavior) and requires
+                a certificate credential -- otherwise construction raises
                 ``ValueError``. A per-request ``mtls_proof_of_possession=True`` on
                 :func:`~acquire_token_for_client` always takes precedence, yielding
                 a cert-bound ``mtls_pop`` token instead of a Bearer one.
@@ -930,8 +938,7 @@ def get_client_assertion():
         # Bearer-over-mTLS reuses the same mTLS transport but signs a
         # private_key_jwt client_assertion (x5c forced) instead of requesting a
         # cert-bound token; its client(s) are likewise built lazily.
-        self._mtls_bearer_client = None  # Lazily built global Bearer-over-mTLS client
-        self._mtls_bearer_regional_client = None  # Lazily built regional one
+        self._mtls_bearer_clients = {}  # Keyed by authority and resolved region
         self._mtls_lock = Lock()
         if isinstance(client_credential, dict) and not client_credential.get(
                 "client_assertion") and (
@@ -948,7 +955,11 @@ def get_client_assertion():
         self._send_certificate_over_mtls = bool(
             isinstance(client_credential, dict)
             and client_credential.get("send_certificate_over_mtls"))
-        if self._send_certificate_over_mtls and self._mtls_cert_credential is None:
+        if self._send_certificate_over_mtls and (
+                self._mtls_cert_credential is None
+                or (not client_credential.get("private_key_pfx_path") and not (
+                    isinstance(client_credential.get("public_certificate"), str)
+                    and client_credential["public_certificate"].strip()))):
             raise ValueError(
                 "send_certificate_over_mtls=True requires this confidential "
                 "client to be configured with a certificate credential (a "
@@ -1227,10 +1238,10 @@ def _get_mtls_bearer_client(self, central_authority):
                 "the client certificate in the mutual-TLS handshake. Omit the "
                 "http_client argument to use MSAL's built-in mTLS transport.")
         region = self._compute_region_to_use()
+        cache_key = (
+            central_authority.instance, central_authority.token_endpoint, region)
         with self._mtls_lock:
-            cached = (
-                self._mtls_bearer_regional_client if region
-                else self._mtls_bearer_client)
+            cached = self._mtls_bearer_clients.get(cache_key)
         if cached is not None:
             return cached
         cert = self._get_mtls_pop_cert()  # May raise ValueError (no cert configured)
@@ -1280,14 +1291,7 @@ def _get_mtls_bearer_client(self, central_authority):
         with self._mtls_lock:
             # Double-check under the lock (mirrors _get_mtls_client): another
             # thread may have built and cached a client while we built ours.
-            if region:
-                if self._mtls_bearer_regional_client is None:
-                    self._mtls_bearer_regional_client = client
-                client = self._mtls_bearer_regional_client
-            else:
-                if self._mtls_bearer_client is None:
-                    self._mtls_bearer_client = client
-                client = self._mtls_bearer_client
+            client = self._mtls_bearer_clients.setdefault(cache_key, client)
         return client
 
     def _client_for_confidential_request(self):
@@ -1669,7 +1673,7 @@ def authorize():  # A controller in a web app
         self._validate_ssh_cert_input_data(kwargs.get("data", {}))
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID)
-        response = _clean_up(self.client.obtain_token_by_auth_code_flow(
+        response = _clean_up(self._client_for_confidential_request().obtain_token_by_auth_code_flow(
             auth_code_flow,
             auth_response,
             scope=self._decorate_scope(scopes) if scopes else None,
@@ -2298,9 +2302,12 @@ def _acquire_token_silent_by_finding_specific_refresh_token(
         response = None  # A distinguishable value to mean cache is empty
         if not matches:  # Then exit early to avoid expensive operations
             return response
-        client, _ = self._build_client(
-            # Potentially expensive if building regional client
-            self.client_credential, authority, skip_regional_client=True)
+        if self._send_certificate_over_mtls:
+            client = self._get_mtls_bearer_client(authority)
+        else:
+            client, _ = self._build_client(
+                # Potentially expensive if building regional client
+                self.client_credential, authority, skip_regional_client=True)
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_SILENT_ID,
             correlation_id=correlation_id, refresh_reason=refresh_reason)
@@ -2999,7 +3006,7 @@ class ConfidentialClientApplication(ClientApplication):  # server-side web app
 
     def acquire_token_for_client(
             self, scopes, claims_challenge=None, fmi_path=None,
-            mtls_proof_of_possession=False, forwarded_client_claims=None,
+            forwarded_client_claims=None, *, mtls_proof_of_possession=False,
             **kwargs):
         """Acquires token for the current confidential client, not for an end user.
 
@@ -3041,6 +3048,9 @@ def acquire_token_for_client(
             ``/common`` or ``/organizations``); and MSAL's built-in HTTP
             transport must be in use (a custom ``http_client`` cannot perform
             the mTLS handshake). Any of these unmet raises ``ValueError``.
+            A ``post`` transport override is also rejected. If the provider
+            returns another token type, the result is ``token_type_mismatch``
+            and the rejected response is not cached.
 
             On success the result also contains a ``binding_certificate`` dict
             with the **public** binding material only (never the private key)::
@@ -3092,6 +3102,10 @@ def acquire_token_for_client(
             kwargs["data"] = kwargs.get("data", {})
             _stash_client_claims(forwarded_client_claims, kwargs["data"])
         if mtls_proof_of_possession:
+            if "post" in kwargs:
+                raise ValueError(
+                    "mtls_proof_of_possession=True does not support a post "
+                    "override. Omit post to use MSAL's built-in mTLS transport.")
             # An mTLS transport is required to present the certificate for a
             # cert-bound PoP request.
             if self._http_client_is_custom:
@@ -3122,8 +3136,7 @@ def acquire_token_for_client(
             # 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.
+            # _MtlsClient, which does not cache downgraded responses.
             # Mirrors the fail-closed behavior of MSAL .NET and MSAL Go.
             if (result.get("token_type") or "").lower() != "mtls_pop":
                 return {
diff --git a/msal/mtls.py b/msal/mtls.py
index f55684aa..1e3bab22 100644
--- a/msal/mtls.py
+++ b/msal/mtls.py
@@ -39,20 +39,15 @@
 # ─────────────────────────────────────────────────────────────────────────────
 # SOVEREIGN GUARDRAIL - single override point for mTLS cloud availability.
 #
-# mTLS PoP is currently rejected for the deprecated sovereign login hosts below
-# and for any non-"login." host. ``mtlsauth.*`` is rolling out across clouds
-# (Azure Government / AGC: available; Bleu / Delos: TBD). To enable a cloud,
-# remove its entry here (and, if needed, relax the non-"login." host check in
-# ``mtls_pop_host``). This is the ONLY place cloud eligibility is enforced, so
-# do not scatter equivalent checks elsewhere in the codebase.
+# Only public-cloud aliases and Azure Government are supported. Add an explicit
+# mapping here when another cloud becomes available.
 # ─────────────────────────────────────────────────────────────────────────────
 _MTLS_POP_UNSUPPORTED_HOSTS = {
     "login.usgovcloudapi.net":
         "login.usgovcloudapi.net is not supported for mTLS PoP, "
         "please use login.microsoftonline.us",
     "login.chinacloudapi.cn":
-        "login.chinacloudapi.cn is not supported for mTLS PoP, "
-        "please use login.partner.microsoftonline.cn",
+        "login.chinacloudapi.cn is not supported for mTLS PoP.",
     }
 
 
@@ -62,8 +57,7 @@ def mtls_pop_host(instance, region=None):
     :param str instance: The authority host, e.g. ``login.microsoftonline.com``.
     :param region: Optional region, e.g. ``westus3``. When provided, a regional
         mTLS host ``{region}.mtlsauth...`` is returned; otherwise the global one.
-    :raises ValueError: When mTLS PoP is not supported for the given host
-        (the sovereign guardrail above, or a non-``login.`` host).
+    :raises ValueError: When mTLS PoP is not supported for the given host.
     """
     instance = instance.lower()
     if instance in _MTLS_POP_UNSUPPORTED_HOSTS:  # Sovereign guardrail
@@ -72,12 +66,12 @@ def mtls_pop_host(instance, region=None):
         # Known public aliases (incl. legacy login.windows.net / sts.windows.net)
         # all normalize to the single global mTLS host.
         base = _PUBLIC_MTLS_HOST
-    elif instance.startswith("login."):  # e.g. login.microsoftonline.us
-        base = "mtlsauth" + instance[len("login"):]  # -> mtlsauth.microsoftonline.us
+    elif instance == "login.microsoftonline.us":
+        base = "mtlsauth.microsoftonline.us"
     else:
         raise ValueError(
-            "mTLS PoP is only supported for hosts that start with 'login.'. "
-            "The provided authority host ({}) does not meet this requirement. "
+            "mTLS PoP is only supported for public-cloud and Azure Government "
+            "authority hosts. The provided authority host ({}) is not supported. "
             "See {} for details.".format(instance, _MTLS_POP_DOC_LINK))
     return "{}.{}".format(region, base) if region else base
 
@@ -121,40 +115,45 @@ def __init__(self, cert_pem, key_pem, *,
         self._proxies = proxies
         self._timeout = timeout
         self._session = None
+        self._sessions = {}
         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:
-            with self._session_lock:
-                if self._session is None:
-                    import requests  # Lazy import, same as the rest of MSAL
+    def _ensure_session(self, url="", verify=None):
+        import requests
+        with self._session_lock:
+            session = self._session or requests.Session()
+            session.verify = self._verify
+            if self._proxies:
+                session.proxies = self._proxies
+            effective_verify = session.merge_environment_settings(
+                url, {}, None, verify, None)["verify"]
+            if effective_verify not in self._sessions:
+                if self._sessions:
                     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
+                adapter = _make_mtls_adapter(_create_ssl_context(
+                    self._cert_pem, self._key_pem, effective_verify))
+                session.mount("https://", adapter)
+                self._sessions[effective_verify] = session
+                self._session = self._session or session
+            return self._sessions[effective_verify]
 
     def post(self, url, **kwargs):
         if self._timeout is not None:
             kwargs.setdefault("timeout", self._timeout)
-        return self._ensure_session().post(url, **kwargs)
+        return self._ensure_session(url, kwargs.get("verify")).post(url, **kwargs)
 
     def get(self, url, **kwargs):
         if self._timeout is not None:
             kwargs.setdefault("timeout", self._timeout)
-        return self._ensure_session().get(url, **kwargs)
+        return self._ensure_session(url, kwargs.get("verify")).get(url, **kwargs)
 
     def close(self):
-        if self._session is not None:
-            self._session.close()
+        with self._session_lock:
+            for session in self._sessions.values():
+                session.close()
 
 
 def _make_mtls_adapter(ssl_context):
@@ -229,4 +228,3 @@ def _create_ssl_context(cert_pem, key_pem, verify=True):
         except OSError:  # pragma: no cover
             logger.warning("Unable to remove temporary mTLS key file")
     return context
-
diff --git a/msal/token_cache.py b/msal/token_cache.py
index 1b1a8d65..034649ee 100644
--- a/msal/token_cache.py
+++ b/msal/token_cache.py
@@ -16,15 +16,32 @@
 
 
 def _key_id_to_str(key_id):
-    """Coerce a key_id (base64url x5t#S256) into a str for the AT cache key.
-
-    It is normally an ASCII str, but tolerate bytes (decoded as ASCII) or any
-    other type so that building the cache key never raises ``TypeError``.
-    """
+    """Normalize a key_id to text for cache storage and lookup."""
     if isinstance(key_id, bytes):
-        return key_id.decode("ascii")
+        try:
+            return key_id.decode("ascii")
+        except UnicodeDecodeError as exc:
+            raise ValueError("key_id bytes must contain ASCII text") from exc
     return key_id if isinstance(key_id, str) else str(key_id)
 
+
+def _make_access_token_key(
+        home_account_id=None, environment=None, client_id=None, realm=None,
+        target=None, ext_cache_key=None, key_id=None):
+    components = [
+        home_account_id or "", environment or "",
+        "atext" if ext_cache_key else "AccessToken",
+        client_id or "", realm or "", target or "",
+        ] + ([ext_cache_key] if ext_cache_key else [])
+    components = [value.lower() for value in components]
+    key_id = _key_id_to_str(key_id) if key_id is not None else None
+    if key_id:
+        # A hyphen-free namespace cannot collide with the legacy joined keys.
+        payload = json.dumps(
+            components + [key_id], separators=(",", ":"))
+        return "atkey:" + hashlib.sha256(payload.encode("utf-8")).hexdigest()
+    return "-".join(components)
+
 # Fields in the request data dict that should NOT be included in the extended
 # cache key hash. Everything else in data IS included, because those are extra
 # body parameters going on the wire and must differentiate cached tokens.
@@ -246,22 +263,9 @@ def __init__(self):
                             # same app+scope). key_id is absent for Bearer, which
                             # keeps the Bearer cache key byte-for-byte unchanged.
                         **ignored_payload_from_a_real_token:
-                    "-".join([  # Note: Could use a hash here to shorten key length
-                        home_account_id or "",
-                        environment or "",
-                        # Use "atext" credential type when ext_cache_key is
-                        # present, matching MSAL Go and MSAL .NET behaviour.
-                        "atext" if ext_cache_key else "AccessToken",
-                        client_id or "",
-                        realm or "",
-                        target or "",
-                        ] + ([ext_cache_key] if ext_cache_key else [])
-                        ).lower()
-                        # key_id is a base64url x5t#S256 and is case-sensitive,
-                        # so it is appended AFTER lower-casing the rest, to keep
-                        # ATs bound to different keys/certs isolated. Coerce it to
-                        # a str first so a non-str key_id never breaks caching.
-                        + ("-" + _key_id_to_str(key_id) if key_id else ""),
+                    _make_access_token_key(
+                        home_account_id, environment, client_id, realm, target,
+                        ext_cache_key, key_id),
             self.CredentialType.ID_TOKEN:
                 lambda home_account_id=None, environment=None, client_id=None,
                         realm=None, **ignored_payload_from_a_real_token:
@@ -345,6 +349,9 @@ def search(self, credential_type, target=None, query=None, *, now=None, for_remo
         """
         target = sorted(target or [])  # Match the order sorted by add()
         assert isinstance(target, list), "Invalid parameter type"
+        target_set = set(target)
+        if query and query.get("key_id") is not None:
+            query = dict(query, key_id=_key_id_to_str(query["key_id"]))
 
         preferred_result = None
         if (credential_type == self.CredentialType.ACCESS_TOKEN
@@ -357,13 +364,14 @@ def search(self, credential_type, target=None, query=None, *, now=None, for_remo
                 query["client_id"], query["realm"], target,
                 ext_cache_key=query.get("ext_cache_key"),
                 key_id=query.get("key_id"))
-            if preferred_result and self._is_matching(
-                preferred_result, query,
-                # Needs no target_set here because it is satisfied by dict key
+            if (preferred_result
+                and self._is_matching(preferred_result, query, target_set)
+                and (for_removal or all(
+                    not preferred_result.get(field) or query.get(field)
+                    for field in ("key_id", "ext_cache_key")))
             ):
                 yield preferred_result
 
-        target_set = set(target)
         with self._lock:
             # O(n) search. The key is NOT used in search.
             now = int(time.time() if now is None else now)
@@ -389,8 +397,8 @@ def search(self, credential_type, target=None, query=None, *, now=None, for_remo
                     # to enumerate these ATs and delete them.
                     if (credential_type == self.CredentialType.ACCESS_TOKEN
                         and not for_removal
-                        and "ext_cache_key" in entry
-                        and "ext_cache_key" not in (query or {})
+                        and entry.get("ext_cache_key")
+                        and not (query or {}).get("ext_cache_key")
                     ):
                         continue
                     # Cache isolation for key-bound tokens (e.g. mtls_pop, SSH-cert).
@@ -401,8 +409,8 @@ def search(self, credential_type, target=None, query=None, *, now=None, for_remo
                     # to enumerate these key-bound ATs and delete them.
                     if (credential_type == self.CredentialType.ACCESS_TOKEN
                         and not for_removal
-                        and "key_id" in entry
-                        and "key_id" not in (query or {})
+                        and entry.get("key_id")
+                        and not (query or {}).get("key_id")
                     ):
                         continue
                     yield entry
@@ -592,13 +600,25 @@ def modify(self, credential_type, old_entry, new_key_value_pairs=None):
         # You can monkeypatch self.key_makers to support more types on-the-fly.
         key = self.key_makers[credential_type](**old_entry)
         with self._lock:
+            entries = self._cache.setdefault(credential_type, {})
             if new_key_value_pairs:  # Update with them
-                entries = self._cache.setdefault(credential_type, {})
-                entries[key] = dict(
+                entry = dict(
                     old_entry,  # Do not use entries[key] b/c it might not exist
                     **new_key_value_pairs)
+                if (credential_type == self.CredentialType.ACCESS_TOKEN
+                        and entry.get("key_id") is not None):
+                    entry["key_id"] = _key_id_to_str(entry["key_id"])
+                entries[key] = entry
+            elif credential_type == self.CredentialType.ACCESS_TOKEN:
+                # Persisted caches may still use an older AT key format.
+                if entries.get(key) == old_entry:
+                    entries.pop(key)
+                else:
+                    for stored_key in [
+                            k for k, entry in entries.items() if entry == old_entry]:
+                        entries.pop(stored_key)
             else:  # Remove old_entry
-                self._cache.setdefault(credential_type, {}).pop(key, None)
+                entries.pop(key, None)
 
     def remove_rt(self, rt_item):
         assert rt_item.get("credential_type") == self.CredentialType.REFRESH_TOKEN
@@ -681,4 +701,3 @@ def serialize(self):
         with self._lock:
             self.has_state_changed = False
             return json.dumps(self._cache, indent=4)
-
diff --git a/tests/test_application.py b/tests/test_application.py
index 0e2e4abb..5f525b8c 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2278,8 +2278,10 @@ def test_vanilla_sni_mtls_pop_request_and_result(self):
         app = ConfidentialClientApplication(
             "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
         captured = []
-        result = app.acquire_token_for_client(
-            ["s1"], mtls_proof_of_possession=True, post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            result = app.acquire_token_for_client(
+                ["s1"], mtls_proof_of_possession=True)
         req = captured[0]
         # Endpoint transformed to the global mtls host
         self.assertEqual(
@@ -2301,15 +2303,40 @@ def test_vanilla_sni_mtls_pop_request_and_result(self):
     def test_repeat_call_is_cache_hit(self):
         app = ConfidentialClientApplication(
             "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
-        app.acquire_token_for_client(
-            ["s1"], mtls_proof_of_possession=True, post=self._capturing_post([]))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post([])):
+            app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True)
         second = []
-        result = app.acquire_token_for_client(
-            ["s1"], mtls_proof_of_possession=True, post=self._capturing_post(second))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(second)):
+            result = app.acquire_token_for_client(
+                ["s1"], mtls_proof_of_possession=True)
         self.assertEqual([], second, "Second call must be served from cache")
         self.assertEqual(app._TOKEN_SOURCE_CACHE, result[app._TOKEN_SOURCE])
+        self.assertEqual("mtls_pop", result["token_type"])
         self.assertTrue(result.get("binding_certificate"))
 
+    def test_released_positional_claims_argument_is_preserved(self):
+        app = ConfidentialClientApplication(
+            "cid", client_credential="a_secret", authority=self._AUTHORITY)
+        captured = []
+        claims = '{"access_token": {"example": {"value": "value"}}}'
+        result = app.acquire_token_for_client(
+            ["s1"], None, None, claims,
+            post=self._capturing_post(captured, token_type="Bearer"))
+        self.assertEqual(json.loads(claims), json.loads(captured[0]["data"]["claims"]))
+        self.assertEqual("Bearer", result["token_type"])
+        self.assertNotIn("binding_certificate", result)
+
+    def test_post_override_is_rejected_for_pop(self):
+        app = ConfidentialClientApplication(
+            "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
+        post = Mock()
+        with self.assertRaisesRegex(ValueError, "post"):
+            app.acquire_token_for_client(
+                ["s1"], mtls_proof_of_possession=True, post=post)
+        post.assert_not_called()
+
     def test_backward_compatible_bearer_is_unchanged(self):
         app = ConfidentialClientApplication(
             "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
@@ -2332,9 +2359,10 @@ def test_token_type_downgrade_fails_closed(self):
         # 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"))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post([], token_type="Bearer")):
+            result = app.acquire_token_for_client(
+                ["s1"], mtls_proof_of_possession=True)
         self.assertEqual("token_type_mismatch", result.get("error"))
         self.assertNotIn("access_token", result)
         self.assertNotIn("binding_certificate", result)
@@ -2345,14 +2373,31 @@ def test_token_type_downgrade_fails_closed(self):
             or (at.get("token_type") or "").lower() == "mtls_pop"]
         self.assertEqual(
             [], bound, "A downgraded token must never be cached as bound")
+        self.assertEqual({}, app.token_cache._cache.get("AccessToken", {}))
+        captured = []
+        ordinary = app.acquire_token_for_client(
+            ["s1"], post=self._capturing_post(captured, token_type="Bearer"))
+        self.assertEqual(1, len(captured))
+        self.assertEqual(app._TOKEN_SOURCE_IDP, ordinary[app._TOKEN_SOURCE])
+
+    def test_missing_token_type_is_not_cached(self):
+        app = ConfidentialClientApplication(
+            "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY)
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post([], token_type=None)):
+            result = app.acquire_token_for_client(
+                ["s1"], mtls_proof_of_possession=True)
+        self.assertEqual("token_type_mismatch", result["error"])
+        self.assertEqual({}, app.token_cache._cache.get("AccessToken", {}))
 
     def test_regional_mtls_endpoint(self):
         app = ConfidentialClientApplication(
             "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY,
             azure_region="westus3")
         captured = []
-        app.acquire_token_for_client(
-            ["s1"], mtls_proof_of_possession=True, post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True)
         self.assertEqual(
             captured[0]["url"],
             "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
@@ -2439,6 +2484,25 @@ def test_flag_with_non_cert_credential_raises_at_construction(self):
                     "send_certificate_over_mtls": True})
         self.assertIn("send_certificate_over_mtls", str(ctx.exception))
 
+    def test_flag_with_incomplete_pem_credential_raises_at_construction(self):
+        from msal.application import _load_mtls_cert_material
+        material = _load_mtls_cert_material(_MTLS_CERT_CRED)
+        credential = {
+            "private_key": material["private_key_pem"].decode("ascii"),
+            "thumbprint": "0" * 40,
+        }
+        ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=credential)
+        for public_certificate in (None, "", " ", True):
+            with self.subTest(public_certificate=public_certificate):
+                with self.assertRaisesRegex(
+                        ValueError, "send_certificate_over_mtls.*public_certificate"):
+                    ConfidentialClientApplication(
+                        "cid", authority=self._AUTHORITY,
+                        client_credential=dict(
+                            credential, send_certificate_over_mtls=True,
+                            public_certificate=public_certificate))
+
     # --- B. Client-credentials behavior -------------------------------------
     def test_cc_flag_no_region_uses_global_mtls_and_returns_bearer(self):
         app = ConfidentialClientApplication(
@@ -2502,9 +2566,10 @@ def test_cc_per_request_mtls_pop_wins_over_flag(self):
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        result = app.acquire_token_for_client(
-            ["s1"], mtls_proof_of_possession=True,
-            post=self._capturing_post(captured, token_type="mtls_pop"))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured, token_type="mtls_pop")):
+            result = app.acquire_token_for_client(
+                ["s1"], mtls_proof_of_possession=True)
         req = captured[0]
         self.assertEqual("mtls_pop", req["data"].get("token_type"))
         # The mtls_pop path is vanilla SN/I: no signed assertion on the wire
@@ -2581,6 +2646,126 @@ def test_auth_code_flag_uses_global_mtls_and_auth_code_grant(self):
         self.assertTrue(req["data"].get("client_assertion"))
         self.assertNotEqual("mtls_pop", req["data"].get("token_type"))
 
+    def test_auth_code_flow_uses_mtls_and_preserves_state_validation(self):
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        flow = app.initiate_auth_code_flow(
+            ["s1"], redirect_uri="https://localhost/callback",
+            response_mode="form_post")
+        captured = []
+        with patch.object(app.http_client, "post",
+                          side_effect=AssertionError("Expected mTLS transport")), \
+                patch("msal.mtls._MtlsHttpClient.post",
+                      side_effect=self._capturing_post(captured)) as post:
+            with self.assertRaises(ValueError):
+                app.acquire_token_by_auth_code_flow(
+                    flow, {"code": "a_code", "state": "wrong_state"})
+            post.assert_not_called()
+            result = app.acquire_token_by_auth_code_flow(
+                flow, {"code": "a_code", "state": flow["state"]})
+        self.assertIn("access_token", result)
+        self.assertEqual(
+            "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token",
+            captured[0]["url"])
+        self.assertEqual("authorization_code", captured[0]["data"]["grant_type"])
+        self.assertEqual(flow["code_verifier"], captured[0]["data"]["code_verifier"])
+
+    def test_silent_refresh_uses_mtls_and_preserves_cache_callbacks(self):
+        for succeeds in (True, False):
+            with self.subTest(succeeds=succeeds):
+                app = ConfidentialClientApplication(
+                    "cid", authority=self._AUTHORITY, client_credential=dict(
+                        _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+                app.acquire_token_on_behalf_of(
+                    "user_assertion", ["s1"],
+                    post=lambda *a, **k: MinimalResponse(
+                        status_code=200, text=_build_user_token_response(
+                            client_id="cid", refresh_token="original_rt")))
+                account = next(app.token_cache.search(
+                    msal.TokenCache.CredentialType.ACCOUNT))
+                captured = []
+
+                def refresh(url, data=None, **kwargs):
+                    captured.append({"url": url, "data": data})
+                    if succeeds:
+                        app.token_cache.remove_account(account)
+                    return MinimalResponse(
+                        status_code=200 if succeeds else 400,
+                        text=_build_user_token_response(
+                            client_id="cid", refresh_token="rotated_rt")
+                        if succeeds else '{"error": "invalid_grant"}')
+
+                with patch.object(app.http_client, "post",
+                                  side_effect=AssertionError("Expected mTLS transport")), \
+                        patch("msal.mtls._MtlsHttpClient.post", side_effect=refresh), \
+                        patch.object(app, "_get_authority_aliases", return_value=[]):
+                    result = app.acquire_token_silent_with_error(
+                        ["s1"], account, force_refresh=True)
+                self.assertEqual(
+                    "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token",
+                    captured[0]["url"])
+                self.assertEqual("refresh_token", captured[0]["data"]["grant_type"])
+                refresh_tokens = list(app.token_cache.search(
+                    msal.TokenCache.CredentialType.REFRESH_TOKEN))
+                self.assertEqual(1, len(refresh_tokens))
+                self.assertEqual(
+                    "rotated_rt" if succeeds else "original_rt",
+                    refresh_tokens[0]["secret"])
+                if succeeds:
+                    self.assertIn("access_token", result)
+                    self.assertEqual([], list(app.token_cache.search(
+                        msal.TokenCache.CredentialType.ACCOUNT)))
+                else:
+                    self.assertEqual("invalid_grant", result["error"])
+
+    def test_silent_refresh_preserves_effective_authority(self):
+        from msal.authority import Authority
+        app = ConfidentialClientApplication(
+            "cid", authority=self._AUTHORITY, client_credential=dict(
+                _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+        alias = "login.windows.net"
+        endpoint = "https://{}/other_tenant/oauth2/v2.0/token".format(alias)
+        with patch(_OIDC_DISCOVERY, return_value={
+                "authorization_endpoint": endpoint.replace("/token", "/authorize"),
+                "token_endpoint": endpoint,
+                "issuer": "https://{}/other_tenant/v2.0".format(alias)}):
+            authority = Authority(
+                "https://{}/other_tenant".format(alias), app.http_client,
+                instance_discovery=False)
+        rt = {
+            "credential_type": "RefreshToken",
+            "environment": alias, "home_account_id": "uid.utid",
+            "client_id": "cid", "secret": "a_rt",
+        }
+        app.token_cache.modify(msal.TokenCache.CredentialType.REFRESH_TOKEN, rt, rt)
+        for region in (None, "westus3"):
+            with self.subTest(region=region), patch.object(
+                    app, "_compute_region_to_use", return_value=region):
+                original_client = app._get_mtls_bearer_client(app.authority)
+                captured = []
+                with patch.object(app.http_client, "post",
+                                  side_effect=AssertionError("Expected mTLS transport")), \
+                        patch("msal.mtls._MtlsHttpClient.post",
+                              side_effect=self._capturing_post(captured)), \
+                        patch.object(app.token_cache, "add",
+                                     wraps=app.token_cache.add) as add:
+                    result = app._acquire_token_silent_by_finding_specific_refresh_token(
+                        authority, ["s1"], {
+                            "environment": alias, "home_account_id": "uid.utid",
+                            "client_id": "cid"})
+                expected = "https://{}mtlsauth.microsoft.com/other_tenant/oauth2/v2.0/token".format(
+                    region + "." if region else "")
+                self.assertIn("access_token", result)
+                self.assertEqual(expected, captured[0]["url"])
+                event = add.call_args[0][0]
+                self.assertEqual(alias, event["environment"])
+                self.assertTrue(event["skip_account_creation"])
+                alias_client = app._get_mtls_bearer_client(authority)
+                self.assertIsNot(original_client, alias_client)
+                self.assertIs(alias_client, app._get_mtls_bearer_client(authority))
+                self.assertIs(original_client, app._get_mtls_bearer_client(app.authority))
+
     def test_obo_without_flag_uses_regular_login_endpoint(self):
         # Negative control: without the app-level flag, OBO stays on the
         # regular login.* endpoint (no mTLS routing).
diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py
index 6baa69cd..ae7b6e99 100644
--- a/tests/test_mtls_transport.py
+++ b/tests/test_mtls_transport.py
@@ -1,10 +1,15 @@
+import base64
+import json
 import os
 import ssl
+import tempfile
 import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
 
 from msal import mtls
 from msal.application import (
-    _load_mtls_cert_material, _private_key_to_unencrypted_pem)
+    _load_mtls_cert_material, _private_key_to_unencrypted_pem, extract_certs)
 
 
 class TestMtlsEndpointTransform(unittest.TestCase):
@@ -66,7 +71,15 @@ def test_sovereign_us_gov_fails_fast(self):
     def test_sovereign_china_fails_fast(self):
         with self.assertRaises(ValueError) as cm:
             mtls.mtls_pop_host("login.chinacloudapi.cn")
-        self.assertIn("login.partner.microsoftonline.cn", str(cm.exception))
+        self.assertIn("not supported", str(cm.exception))
+
+    def test_unsupported_login_hosts_fail_fast(self):
+        for host in ("login.example.com", "login.partner.microsoftonline.cn",
+                     "login.microsoftonline.de"):
+            with self.subTest(host=host), self.assertRaisesRegex(
+                    ValueError, "not supported"):
+                mtls.transform_token_endpoint(
+                    "https://{}/tenant/token".format(host), host)
 
     def test_non_login_host_fails_fast(self):
         with self.assertRaises(ValueError):
@@ -143,6 +156,162 @@ def test_http_client_builds_session_lazily(self):
         client.close()
 
 
+class TestMtlsCertificateChain(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        from cryptography import x509
+        from cryptography.hazmat.primitives import hashes, serialization
+        from cryptography.hazmat.primitives.asymmetric import rsa
+        from cryptography.x509.oid import NameOID
+
+        def make_cert(name, key, issuer, issuer_key, ca):
+            subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)])
+            now = datetime.now(timezone.utc)
+            return (x509.CertificateBuilder()
+                .subject_name(subject).issuer_name(issuer or subject)
+                .public_key(key.public_key()).serial_number(x509.random_serial_number())
+                .not_valid_before(now - timedelta(days=1))
+                .not_valid_after(now + timedelta(days=1))
+                .add_extension(x509.BasicConstraints(ca=ca, path_length=None), True)
+                .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), False)
+                .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(
+                    issuer_key.public_key()), False)
+                .add_extension(x509.KeyUsage(
+                    digital_signature=True, content_commitment=False,
+                    key_encipherment=False, data_encipherment=False,
+                    key_agreement=False, key_cert_sign=ca, crl_sign=ca,
+                    encipher_only=False, decipher_only=False), True)
+                .sign(issuer_key, hashes.SHA256()))
+
+        root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        root = make_cert("root", root_key, None, root_key, True)
+        intermediate_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        cls.intermediate = make_cert(
+            "intermediate", intermediate_key, root.subject, root_key, True)
+        cls.leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        cls.leaf = make_cert(
+            "client", cls.leaf_key, cls.intermediate.subject, intermediate_key, False)
+        cls.chain_pem = b"".join(cert.public_bytes(serialization.Encoding.PEM)
+            for cert in (cls.leaf, cls.intermediate))
+        cls.key_pem = _private_key_to_unencrypted_pem(cls.leaf_key)
+        cls.root_pem = root.public_bytes(serialization.Encoding.PEM)
+        cls.server_pem = cls.root_pem + _private_key_to_unencrypted_pem(root_key)
+        cls.key_id = base64.urlsafe_b64encode(
+            cls.leaf.fingerprint(hashes.SHA256())).rstrip(b"=").decode("ascii")
+
+    def _assert_chain_and_handshake(self, material):
+        self.assertEqual(self.chain_pem, material["cert_pem"])
+        self.assertEqual(extract_certs(self.chain_pem.decode("ascii")), material["x5c"])
+        self.assertEqual(self.key_id, material["key_id"])
+        client = mtls._create_ssl_context(
+            material["cert_pem"], material["private_key_pem"], verify=False)
+        server = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        server.verify_mode = ssl.CERT_REQUIRED
+        server.load_verify_locations(cadata=self.root_pem.decode("ascii"))
+        with tempfile.TemporaryDirectory() as directory:
+            path = os.path.join(directory, "server.pem")
+            with open(path, "wb") as stream:
+                stream.write(self.server_pem)
+            server.load_cert_chain(path)
+        client_in, client_out, server_in, server_out = [
+            ssl.MemoryBIO() for _ in range(4)]
+        client_tls = client.wrap_bio(client_in, client_out, server_side=False)
+        server_tls = server.wrap_bio(server_in, server_out, server_side=True)
+        client_done = server_done = False
+        for _ in range(10):
+            if not client_done:
+                try:
+                    client_tls.do_handshake()
+                    client_done = True
+                except ssl.SSLWantReadError:
+                    pass
+            if client_out.pending:
+                server_in.write(client_out.read())
+            if not server_done:
+                try:
+                    server_tls.do_handshake()
+                    server_done = True
+                except ssl.SSLWantReadError:
+                    pass
+            if server_out.pending:
+                client_in.write(server_out.read())
+            if client_done and server_done:
+                break
+        self.assertTrue(client_done and server_done, "mTLS handshake did not complete")
+
+    def test_pem_intermediates_reach_tls_handshake(self):
+        self._assert_chain_and_handshake(_load_mtls_cert_material({
+            "private_key": self.key_pem.decode("ascii"),
+            "public_certificate": self.chain_pem.decode("ascii"),
+            }))
+
+    def test_bearer_mtls_assertion_preserves_pem_chain(self):
+        from cryptography.hazmat.primitives import hashes
+        from msal.application import ConfidentialClientApplication
+        from tests.http_client import MinimalResponse
+
+        endpoint = "https://login.microsoftonline.com/tenant/oauth2/v2.0/token"
+        expected_chain = extract_certs(self.chain_pem.decode("ascii"))
+        for send_over_mtls in (False, True):
+            with self.subTest(send_over_mtls=send_over_mtls), patch(
+                    "msal.authority.tenant_discovery", return_value={
+                        "authorization_endpoint": endpoint.replace("/token", "/authorize"),
+                        "token_endpoint": endpoint,
+                        "issuer": "https://login.microsoftonline.com/tenant/v2.0"}):
+                app = ConfidentialClientApplication(
+                    "cid", authority="https://login.microsoftonline.com/tenant",
+                    azure_region=False, client_credential={
+                        "private_key": self.key_pem.decode("ascii"),
+                        "thumbprint": self.leaf.fingerprint(hashes.SHA1()).hex(),
+                        "public_certificate": self.chain_pem.decode("ascii"),
+                        "send_certificate_over_mtls": send_over_mtls,
+                    })
+
+                def post(url, data=None, **kwargs):
+                    header = data["client_assertion"].split(b".")[0]
+                    decoded = json.loads(base64.urlsafe_b64decode(
+                        header + b"=" * (-len(header) % 4)))
+                    self.assertEqual(expected_chain, decoded["x5c"])
+                    return MinimalResponse(
+                        status_code=200, text='{"access_token": "at", "token_type": "Bearer"}')
+
+                app.acquire_token_for_client(["s1"], post=post)
+
+    def test_pfx_intermediates_reach_tls_handshake(self):
+        from cryptography.hazmat.primitives import serialization
+        from cryptography.hazmat.primitives.serialization import pkcs12
+        with tempfile.TemporaryDirectory() as directory:
+            path = os.path.join(directory, "chain.pfx")
+            with open(path, "wb") as stream:
+                stream.write(pkcs12.serialize_key_and_certificates(
+                    b"client", self.leaf_key, self.leaf, [self.intermediate],
+                    serialization.NoEncryption()))
+            material = _load_mtls_cert_material({"private_key_pfx_path": path})
+        self._assert_chain_and_handshake(material)
+
+    def test_environment_ca_bundles_replace_default_roots(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = os.path.join(directory, "private-ca.pem")
+            with open(path, "wb") as stream:
+                stream.write(self.root_pem)
+            for variable in ("REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE"):
+                with self.subTest(variable=variable), patch.dict(
+                        os.environ, {"REQUESTS_CA_BUNDLE": "", "CURL_CA_BUNDLE": "",
+                                     variable: path}):
+                    client = mtls._MtlsHttpClient(self.chain_pem, self.key_pem)
+                    self.addCleanup(client.close)
+                    session = client._ensure_session("https://mtlsauth.microsoft.com")
+                    context = session.get_adapter("https://")._ssl_context
+                    self.assertEqual(1, len(context.get_ca_certs()))
+                    self.assertEqual(
+                        session, client._ensure_session("https://mtlsauth.microsoft.com"))
+                    unverified = client._ensure_session(
+                        "https://mtlsauth.microsoft.com", verify=False)
+                    self.assertIsNot(session, unverified)
+                    self.assertEqual(ssl.CERT_NONE,
+                        unverified.get_adapter("https://")._ssl_context.verify_mode)
+
+
 class TestUnencryptedPemLoading(unittest.TestCase):
     """_private_key_to_unencrypted_pem pins the cryptography backend and turns
     low-level load failures into a clear, actionable ValueError."""
diff --git a/tests/test_token_cache.py b/tests/test_token_cache.py
index 758fae0a..923ab7fc 100644
--- a/tests/test_token_cache.py
+++ b/tests/test_token_cache.py
@@ -292,12 +292,111 @@ def test_key_id_is_coerced_to_str_for_cache_key(self):
         self.assertEqual("THUMB", _key_id_to_str(b"THUMB"))
         self.assertEqual("THUMB", _key_id_to_str("THUMB"))
         self.assertEqual("123", _key_id_to_str(123))  # Any other type -> str()
-        # Adding + searching an AT with a bytes key_id no longer raises; the
-        # helper below stores and then finds it by that same key_id.
-        self._test_data_should_be_saved_and_searchable_in_access_token(
-            {"key_id": b"THUMB"})
+        # Bytes IDs are normalized before persistence and during lookup.
+        self.cache = SerializableTokenCache()
+        self.cache.add({
+            "client_id": "my_client_id", "scope": ["s1"],
+            "token_endpoint": "https://login.example.com/contoso/v2/token",
+            "data": {"key_id": b"THUMB"},
+            "response": build_response(access_token="bound_at"),
+            }, now=1000)
+        self.cache.deserialize(self.cache.serialize())
+        for key_id in (b"THUMB", "THUMB"):
+            entry = self.assertFoundAccessToken(
+                scopes=["s1"], now=1000, query=dict(
+                    client_id="my_client_id", environment="login.example.com",
+                    realm="contoso", home_account_id=None, key_id=key_id))
+            self.assertEqual("THUMB", entry["key_id"])
         self.assertEqual(1, len(self.cache._cache["AccessToken"]))
 
+    def test_non_ascii_key_id_bytes_fail_explicitly(self):
+        with self.assertRaisesRegex(ValueError, "ASCII"):
+            self.cache.add({
+                "data": {"key_id": b"\xff"},
+                "response": build_response(access_token="bound_at"),
+                })
+
+    def test_bound_scope_suffix_cannot_collide_with_unbound_scope(self):
+        common = dict(
+            client_id="my_client_id",
+            token_endpoint="https://login.example.com/contoso/v2/token")
+        events = [
+            dict(common, scope=["scope"], data={"key_id": "abc"},
+                 response=build_response(access_token="bound_at", token_type="mtls_pop")),
+            dict(common, scope=["scope-abc"],
+                 response=build_response(access_token="bearer_at", token_type="Bearer")),
+            ]
+        query = dict(
+            client_id="my_client_id", environment="login.example.com",
+            realm="contoso", home_account_id=None)
+        for insertion_order in (events, list(reversed(events))):
+            with self.subTest(order=[e["scope"] for e in insertion_order]):
+                self.cache = TokenCache()
+                for event in insertion_order:
+                    self.cache.add(event, now=1000)
+                self.assertEqual(2, len(self.cache._cache["AccessToken"]))
+                self.assertEqual("bearer_at", self.assertFoundAccessToken(
+                    scopes=["scope-abc"], query=query, now=1000)["secret"])
+                self.assertEqual("bound_at", self.assertFoundAccessToken(
+                    scopes=["scope"], query=dict(query, key_id="abc"),
+                    now=1000)["secret"])
+
+    def test_legacy_bound_entries_are_isolated_and_removable(self):
+        self.cache.add({
+            "client_id": "my_client_id", "scope": ["s1"],
+            "token_endpoint": "https://login.example.com/contoso/v2/token",
+            "data": {"key_id": "THUMB"},
+            "response": build_response(access_token="legacy_at", token_type="ssh-cert"),
+            }, now=1000)
+        entry = next(iter(self.cache._cache["AccessToken"].values()))
+        legacy_key = self.at_key_maker(**dict(entry, key_id=None))
+        for stored_key in (legacy_key, legacy_key + "-THUMB"):
+            for expired in (False, True):
+                with self.subTest(key=stored_key, expired=expired):
+                    self.cache = SerializableTokenCache()
+                    self.cache.deserialize(json.dumps({
+                        "AccessToken": {stored_key: entry}}))
+                    query = dict(
+                        client_id="my_client_id", environment="login.example.com",
+                        realm="contoso", home_account_id=None)
+                    self.assertEqual([], list(self.cache.search(
+                        "AccessToken", target=["s1"], query=query, now=1000)))
+                    if expired:
+                        list(self.cache.search("AccessToken", now=5000))
+                    else:
+                        self.assertFoundAccessToken(
+                            scopes=["s1"], query=dict(query, key_id="THUMB"),
+                            now=1000)
+                        self.cache.remove_at(entry)
+                    self.assertEqual({}, self.cache._cache["AccessToken"])
+                    self.assertTrue(self.cache.has_state_changed)
+
+    def test_fast_lookup_checks_stored_scope(self):
+        entry = {
+            "credential_type": "AccessToken", "home_account_id": None,
+            "environment": "env", "client_id": "cid", "realm": "tenant",
+            "target": "different", "secret": "wrong_at", "expires_on": "4600",
+            }
+        query = dict(
+            home_account_id=None, environment="env", client_id="cid", realm="tenant")
+        key = self.at_key_maker(**dict(query, target="requested"))
+        self.cache._cache["AccessToken"] = {key: entry}
+        self.assertEqual([], list(self.cache.search(
+            "AccessToken", target=["requested"], query=query, now=1000)))
+
+    def test_case_distinct_key_ids_remain_distinct_after_serialization(self):
+        self.cache = SerializableTokenCache()
+        for key_id in ("Thumb", "thumb"):
+            self.cache.add({
+                "scope": ["s1"], "data": {"key_id": key_id},
+                "response": build_response(access_token=key_id),
+                }, now=1000)
+        self.cache.deserialize(self.cache.serialize())
+        for key_id in ("Thumb", "thumb"):
+            entries = list(self.cache.search(
+                "AccessToken", target=["s1"], query={"key_id": key_id}, now=1000))
+            self.assertEqual([key_id], [entry["secret"] for entry in entries])
+
     def test_bearer_and_mtls_pop_tokens_coexist_and_isolate(self):
         # The crux of mTLS PoP cache isolation (plan C6): a Bearer token and an
         # mtls_pop token for the SAME app/scope/tenant must coexist, and an

From a3ee8075651bf9ee486417190a1d2901f4100846 Mon Sep 17 00:00:00 2001
From: Robbie Ginsburg <rginsburg@microsoft.com>
Date: Thu, 22 Oct 2026 09:56:55 -0400
Subject: [PATCH 8/8] Fix Bearer-over-mTLS transport enforcement and PFX chains

Reject post overrides centrally for Bearer-over-mTLS grants and capture requests through the mTLS transport in tests. Build ordered PFX issuer chains using signature verification, excluding unrelated and duplicate certificates.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 msal/application.py          | 66 ++++++++++++++++++++++++++--
 tests/test_application.py    | 85 ++++++++++++++++++++++++++++--------
 tests/test_e2e.py            | 12 ++---
 tests/test_mtls_transport.py | 80 ++++++++++++++++++++++++++++++---
 4 files changed, 211 insertions(+), 32 deletions(-)

diff --git a/msal/application.py b/msal/application.py
index 0e2f7576..2507f0b7 100644
--- a/msal/application.py
+++ b/msal/application.py
@@ -126,6 +126,45 @@ def _extract_cert_and_thumbprints(cert):
     sha1_thumbprint = cert.fingerprint(hashes.SHA1()).hex()  # CodeQL [SM02167] for legacy support such as ADFS
     return sha256_thumbprint, sha1_thumbprint, x5c, cert_pem
 
+def _is_certificate_issuer(cert, issuer):
+    from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
+    from cryptography.hazmat.primitives.asymmetric import (
+        rsa, ec, dsa, ed25519, ed448, padding)
+    from cryptography.x509.oid import SignatureAlgorithmOID
+
+    if cert.issuer != issuer.subject:
+        return False
+    try:
+        if hasattr(cert, "verify_directly_issued_by"):
+            cert.verify_directly_issued_by(issuer)
+        else:  # Support cryptography versions before 40.
+            key = issuer.public_key()
+            if isinstance(key, rsa.RSAPublicKey):
+                parameters = getattr(cert, "signature_algorithm_parameters", None)
+                if (cert.signature_algorithm_oid == SignatureAlgorithmOID.RSASSA_PSS
+                        and parameters is None):
+                    raise UnsupportedAlgorithm(
+                        "Upgrade cryptography to verify RSA-PSS certificate chains")
+                key.verify(
+                    cert.signature, cert.tbs_certificate_bytes,
+                    parameters or padding.PKCS1v15(), cert.signature_hash_algorithm)
+            elif isinstance(key, ec.EllipticCurvePublicKey):
+                key.verify(
+                    cert.signature, cert.tbs_certificate_bytes,
+                    ec.ECDSA(cert.signature_hash_algorithm))
+            elif isinstance(key, dsa.DSAPublicKey):
+                key.verify(
+                    cert.signature, cert.tbs_certificate_bytes,
+                    cert.signature_hash_algorithm)
+            elif isinstance(key, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)):
+                key.verify(cert.signature, cert.tbs_certificate_bytes)
+            else:
+                raise UnsupportedAlgorithm("Unsupported certificate issuer key")
+    except (InvalidSignature, ValueError):
+        return False
+    return True
+
+
 def _parse_pfx(pfx_path, passphrase_bytes):
     # Cert concepts https://security.stackexchange.com/a/226758/125264
     from cryptography.hazmat.primitives.serialization import pkcs12
@@ -136,11 +175,20 @@ def _parse_pfx(pfx_path, passphrase_bytes):
     if not (private_key and cert):
         raise ValueError("Your PFX file shall contain both private key and cert")
     sha256_thumbprint, sha1_thumbprint, x5c, cert_pem = _extract_cert_and_thumbprints(cert)
-    for additional_cert in additional_certs or []:
+    remaining = [candidate for candidate in additional_certs or [] if candidate != cert]
+    current = cert
+    while remaining:
+        additional_cert = next(
+            (candidate for candidate in remaining
+             if _is_certificate_issuer(current, candidate)), None)
+        if additional_cert is None:
+            break  # A PFX may contain only a partial chain.
+        remaining = [candidate for candidate in remaining if candidate != additional_cert]
         _, _, additional_x5c, additional_pem = _extract_cert_and_thumbprints(
             additional_cert)
         x5c.extend(additional_x5c)
         cert_pem += additional_pem
+        current = additional_cert
     return private_key, sha256_thumbprint, sha1_thumbprint, x5c, cert_pem
 
 
@@ -333,6 +381,15 @@ def obtain_token_by_username_password(self, username, password, **kwargs):
             username, password, headers=headers, **kwargs)
 
 
+class _MtlsBearerClient(_ClientWithCcsRoutingInfo):
+    def _obtain_token(self, *args, **kwargs):
+        if "post" in kwargs:
+            raise ValueError(
+                "send_certificate_over_mtls=True does not support a post "
+                "override. Omit post to use MSAL's built-in mTLS transport.")
+        return super(_MtlsBearerClient, self)._obtain_token(*args, **kwargs)
+
+
 class _MtlsClient(_ClientWithCcsRoutingInfo):
     """A confidential-client OAuth2 client whose token request presents a
     client certificate over mutual-TLS (mTLS Proof-of-Possession).
@@ -588,7 +645,10 @@ def get_client_assertion():
                 redemption during :func:`~acquire_token_silent`. It defaults to
                 ``False`` (so leaving it unset changes no behavior) and requires
                 a certificate credential -- otherwise construction raises
-                ``ValueError``. A per-request ``mtls_proof_of_possession=True`` on
+                ``ValueError``. Custom ``http_client`` instances and per-request
+                ``post`` overrides are rejected when making an mTLS token request,
+                because MSAL must own the certificate-bearing transport.
+                A per-request ``mtls_proof_of_possession=True`` on
                 :func:`~acquire_token_for_client` always takes precedence, yielding
                 a cert-bound ``mtls_pop`` token instead of a Bearer one.
 
@@ -1273,7 +1333,7 @@ def _get_mtls_bearer_client(self, central_authority):
         client_assertion = assertion.create_regenerative_assertion(
             audience=token_endpoint, issuer=self.client_id,
             additional_claims=self.client_claims or {})
-        client = _ClientWithCcsRoutingInfo(
+        client = _MtlsBearerClient(
             configuration,
             self.client_id,
             http_client=http_client,
diff --git a/tests/test_application.py b/tests/test_application.py
index 5f525b8c..980bffc5 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -2509,8 +2509,9 @@ def test_cc_flag_no_region_uses_global_mtls_and_returns_bearer(self):
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        result = app.acquire_token_for_client(
-            ["s1"], post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            result = app.acquire_token_for_client(["s1"])
         req = captured[0]
         # Routed to the GLOBAL mtls host (login.* -> mtlsauth.*)
         self.assertEqual(
@@ -2535,7 +2536,9 @@ def test_cc_x5c_is_forced_on_assertion(self):
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED_NO_X5C, send_certificate_over_mtls=True))
         captured = []
-        forced.acquire_token_for_client(["s1"], post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            forced.acquire_token_for_client(["s1"])
         self.assertIn(
             "x5c", _jwt_header_of(captured[0]["data"]["client_assertion"]),
             "Bearer-over-mTLS must force x5c onto the assertion header")
@@ -2554,7 +2557,9 @@ def test_cc_flag_with_region_uses_regional_mtls(self):
             client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        app.acquire_token_for_client(["s1"], post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            app.acquire_token_for_client(["s1"])
         self.assertEqual(
             captured[0]["url"],
             "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
@@ -2592,8 +2597,9 @@ def test_obo_flag_uses_global_mtls_and_obo_grant(self):
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        app.acquire_token_on_behalf_of(
-            "a_user_assertion", ["s1"], post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            app.acquire_token_on_behalf_of("a_user_assertion", ["s1"])
         req = captured[0]
         self.assertEqual(
             req["url"],
@@ -2610,8 +2616,9 @@ def test_obo_flag_with_region_uses_regional_mtls(self):
             client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        app.acquire_token_on_behalf_of(
-            "a_user_assertion", ["s1"], post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            app.acquire_token_on_behalf_of("a_user_assertion", ["s1"])
         self.assertEqual(
             captured[0]["url"],
             "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token")
@@ -2621,8 +2628,9 @@ def test_refresh_flag_uses_global_mtls_and_refresh_grant(self):
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        app.acquire_token_by_refresh_token(
-            "a_rt", ["s1"], post=self._capturing_post(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post(captured)):
+            app.acquire_token_by_refresh_token("a_rt", ["s1"])
         req = captured[0]
         self.assertEqual(
             req["url"],
@@ -2636,8 +2644,9 @@ def test_auth_code_flag_uses_global_mtls_and_auth_code_grant(self):
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
         captured = []
-        app.acquire_token_by_authorization_code(
-            "a_code", ["s1"], post=self._capturing_post_full(captured))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post_full(captured)):
+            app.acquire_token_by_authorization_code("a_code", ["s1"])
         req = captured[0]
         self.assertEqual(
             req["url"],
@@ -2677,11 +2686,11 @@ def test_silent_refresh_uses_mtls_and_preserves_cache_callbacks(self):
                 app = ConfidentialClientApplication(
                     "cid", authority=self._AUTHORITY, client_credential=dict(
                         _MTLS_CERT_CRED, send_certificate_over_mtls=True))
-                app.acquire_token_on_behalf_of(
-                    "user_assertion", ["s1"],
-                    post=lambda *a, **k: MinimalResponse(
+                with patch("msal.mtls._MtlsHttpClient.post",
+                           return_value=MinimalResponse(
                         status_code=200, text=_build_user_token_response(
-                            client_id="cid", refresh_token="original_rt")))
+                            client_id="cid", refresh_token="original_rt"))):
+                    app.acquire_token_on_behalf_of("user_assertion", ["s1"])
                 account = next(app.token_cache.search(
                     msal.TokenCache.CredentialType.ACCOUNT))
                 captured = []
@@ -2794,8 +2803,9 @@ def test_cc_second_call_is_bearer_cache_hit(self):
         app = ConfidentialClientApplication(
             "cid", authority=self._AUTHORITY, client_credential=dict(
                 _MTLS_CERT_CRED, send_certificate_over_mtls=True))
-        first = app.acquire_token_for_client(
-            ["s1"], post=self._capturing_post([]))
+        with patch("msal.mtls._MtlsHttpClient.post",
+                   side_effect=self._capturing_post([])):
+            first = app.acquire_token_for_client(["s1"])
         self.assertEqual("Bearer", first.get("token_type"))
         second = []
         result = app.acquire_token_for_client(
@@ -2817,3 +2827,42 @@ def test_cc_second_call_is_bearer_cache_hit(self):
         self.assertEqual(app._TOKEN_SOURCE_CACHE, result[app._TOKEN_SOURCE])
         self.assertEqual("Bearer", result.get("token_type"))
         self.assertNotIn("binding_certificate", result)
+
+    def test_post_override_rejected_for_all_mtls_bearer_grants(self):
+        for grant in ("client_credentials", "obo", "auth_code", "auth_code_flow",
+                      "refresh", "silent_refresh"):
+            with self.subTest(grant=grant):
+                app = ConfidentialClientApplication(
+                    "cid", authority=self._AUTHORITY, client_credential=dict(
+                        _MTLS_CERT_CRED, send_certificate_over_mtls=True))
+                with patch("msal.mtls._MtlsHttpClient.post",
+                           return_value=MinimalResponse(
+                               status_code=200, text=_build_user_token_response(
+                                   client_id="cid", refresh_token="a_rt"))):
+                    app.acquire_token_on_behalf_of("assertion", ["s1"])
+                account = next(app.token_cache.search(
+                    msal.TokenCache.CredentialType.ACCOUNT))
+                with patch("msal.mtls._MtlsHttpClient.post") as transport, \
+                        patch.object(app.http_client, "post") as regular_transport, \
+                        patch.object(app, "_get_authority_aliases", return_value=[]):
+                    override = Mock()
+                    with self.assertRaisesRegex(ValueError, "does not support a post"):
+                        if grant == "client_credentials":
+                            app.acquire_token_for_client(["s2"], post=override)
+                        elif grant == "obo":
+                            app.acquire_token_on_behalf_of("assertion", ["s2"], post=override)
+                        elif grant == "auth_code":
+                            app.acquire_token_by_authorization_code("code", ["s2"], post=override)
+                        elif grant == "auth_code_flow":
+                            flow = app.initiate_auth_code_flow(["s2"])
+                            app.acquire_token_by_auth_code_flow(
+                                flow, {"code": "code", "state": flow["state"]},
+                                post=override)
+                        elif grant == "refresh":
+                            app.acquire_token_by_refresh_token("a_rt", ["s2"], post=override)
+                        else:
+                            app.acquire_token_silent_with_error(
+                                ["s1"], account, force_refresh=True, post=override)
+                    override.assert_not_called()
+                    transport.assert_not_called()
+                    regular_transport.assert_not_called()
diff --git a/tests/test_e2e.py b/tests/test_e2e.py
index 02fa1e51..8fb9bb1a 100644
--- a/tests/test_e2e.py
+++ b/tests/test_e2e.py
@@ -642,16 +642,16 @@ def _assert_mtls(req, grant_key, grant_value):
             self.assertNotIn("key_id", req["data"])
             self.assertNotIn("req_cnf", req["data"])
         cap = []
-        _new_app().acquire_token_on_behalf_of(
-            "a_user_assertion", scope, post=_capture(cap))
+        with patch("msal.mtls._MtlsHttpClient.post", side_effect=_capture(cap)):
+            _new_app().acquire_token_on_behalf_of("a_user_assertion", scope)
         _assert_mtls(cap[0], "requested_token_use", "on_behalf_of")
         cap = []
-        _new_app().acquire_token_by_refresh_token(
-            "a_refresh_token", scope, post=_capture(cap))
+        with patch("msal.mtls._MtlsHttpClient.post", side_effect=_capture(cap)):
+            _new_app().acquire_token_by_refresh_token("a_refresh_token", scope)
         _assert_mtls(cap[0], "grant_type", "refresh_token")
         cap = []
-        _new_app().acquire_token_by_authorization_code(
-            "an_auth_code", scope, post=_capture(cap))
+        with patch("msal.mtls._MtlsHttpClient.post", side_effect=_capture(cap)):
+            _new_app().acquire_token_by_authorization_code("an_auth_code", scope)
         _assert_mtls(cap[0], "grant_type", "authorization_code")
 
     def test_send_certificate_over_mtls_user_flows_live_acquire(self):
diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py
index ae7b6e99..50e82fb7 100644
--- a/tests/test_mtls_transport.py
+++ b/tests/test_mtls_transport.py
@@ -9,7 +9,8 @@
 
 from msal import mtls
 from msal.application import (
-    _load_mtls_cert_material, _private_key_to_unencrypted_pem, extract_certs)
+    _is_certificate_issuer, _load_mtls_cert_material,
+    _private_key_to_unencrypted_pem, extract_certs)
 
 
 class TestMtlsEndpointTransform(unittest.TestCase):
@@ -185,14 +186,21 @@ def make_cert(name, key, issuer, issuer_key, ca):
 
         root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         root = make_cert("root", root_key, None, root_key, True)
+        cls.root = root
+        upper_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        cls.upper_intermediate = make_cert(
+            "upper", upper_key, root.subject, root_key, True)
         intermediate_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         cls.intermediate = make_cert(
-            "intermediate", intermediate_key, root.subject, root_key, True)
+            "intermediate", intermediate_key, cls.upper_intermediate.subject, upper_key, True)
+        unrelated_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        cls.unrelated = make_cert("unrelated", unrelated_key, None, unrelated_key, True)
+        cls.impostor = make_cert("intermediate", unrelated_key, None, unrelated_key, True)
         cls.leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         cls.leaf = make_cert(
             "client", cls.leaf_key, cls.intermediate.subject, intermediate_key, False)
         cls.chain_pem = b"".join(cert.public_bytes(serialization.Encoding.PEM)
-            for cert in (cls.leaf, cls.intermediate))
+            for cert in (cls.leaf, cls.intermediate, cls.upper_intermediate))
         cls.key_pem = _private_key_to_unencrypted_pem(cls.leaf_key)
         cls.root_pem = root.public_bytes(serialization.Encoding.PEM)
         cls.server_pem = cls.root_pem + _private_key_to_unencrypted_pem(root_key)
@@ -275,7 +283,11 @@ def post(url, data=None, **kwargs):
                     return MinimalResponse(
                         status_code=200, text='{"access_token": "at", "token_type": "Bearer"}')
 
-                app.acquire_token_for_client(["s1"], post=post)
+                if send_over_mtls:
+                    with patch("msal.mtls._MtlsHttpClient.post", side_effect=post):
+                        app.acquire_token_for_client(["s1"])
+                else:
+                    app.acquire_token_for_client(["s1"], post=post)
 
     def test_pfx_intermediates_reach_tls_handshake(self):
         from cryptography.hazmat.primitives import serialization
@@ -284,11 +296,69 @@ def test_pfx_intermediates_reach_tls_handshake(self):
             path = os.path.join(directory, "chain.pfx")
             with open(path, "wb") as stream:
                 stream.write(pkcs12.serialize_key_and_certificates(
-                    b"client", self.leaf_key, self.leaf, [self.intermediate],
+                    b"client", self.leaf_key, self.leaf,
+                    [self.intermediate, self.upper_intermediate],
                     serialization.NoEncryption()))
             material = _load_mtls_cert_material({"private_key_pfx_path": path})
         self._assert_chain_and_handshake(material)
 
+    def test_unordered_pfx_chain_excludes_unrelated_and_wrong_signer(self):
+        from msal.application import ConfidentialClientApplication
+        from tests.http_client import MinimalResponse
+
+        endpoint = "https://login.microsoftonline.com/tenant/oauth2/v2.0/token"
+        path = os.path.join(os.path.dirname(__file__), "certificate-with-password.pfx")
+        expected_pem = self.chain_pem + self.root_pem
+        expected_x5c = extract_certs(expected_pem.decode("ascii"))
+        additional = [
+            self.root, self.impostor, self.upper_intermediate,
+            self.unrelated, self.intermediate, self.leaf, self.intermediate]
+        with patch("cryptography.hazmat.primitives.serialization.pkcs12."
+                   "load_key_and_certificates",
+                   return_value=(self.leaf_key, self.leaf, additional)):
+            material = _load_mtls_cert_material({"private_key_pfx_path": path})
+            self.assertEqual(expected_pem, material["cert_pem"])
+            self.assertEqual(expected_x5c, material["x5c"])
+            self.assertEqual(self.key_id, material["key_id"])
+            for send_over_mtls in (False, True):
+                with self.subTest(send_over_mtls=send_over_mtls), patch(
+                        "msal.authority.tenant_discovery", return_value={
+                            "authorization_endpoint": endpoint.replace("/token", "/authorize"),
+                            "token_endpoint": endpoint,
+                            "issuer": "https://login.microsoftonline.com/tenant/v2.0"}):
+                    app = ConfidentialClientApplication(
+                        "cid", authority="https://login.microsoftonline.com/tenant",
+                        azure_region=False, client_credential={
+                            "private_key_pfx_path": path,
+                            "public_certificate": True,
+                            "send_certificate_over_mtls": send_over_mtls})
+
+                    def post(url, data=None, **kwargs):
+                        header = data["client_assertion"].split(b".")[0]
+                        decoded = json.loads(base64.urlsafe_b64decode(
+                            header + b"=" * (-len(header) % 4)))
+                        self.assertEqual(expected_x5c, decoded["x5c"])
+                        return MinimalResponse(
+                            status_code=200,
+                            text='{"access_token": "at", "token_type": "Bearer"}')
+
+                    if send_over_mtls:
+                        with patch("msal.mtls._MtlsHttpClient.post", side_effect=post):
+                            app.acquire_token_for_client(["s1"])
+                    else:
+                        app.acquire_token_for_client(["s1"], post=post)
+
+    def test_issuer_signature_verification_on_older_cryptography(self):
+        class LegacyCertificate:
+            issuer = self.leaf.issuer
+            signature = self.leaf.signature
+            tbs_certificate_bytes = self.leaf.tbs_certificate_bytes
+            signature_algorithm_oid = self.leaf.signature_algorithm_oid
+            signature_hash_algorithm = self.leaf.signature_hash_algorithm
+
+        self.assertFalse(_is_certificate_issuer(LegacyCertificate(), self.impostor))
+        self.assertTrue(_is_certificate_issuer(LegacyCertificate(), self.intermediate))
+
     def test_environment_ca_bundles_replace_default_roots(self):
         with tempfile.TemporaryDirectory() as directory:
             path = os.path.join(directory, "private-ca.pem")