From f98961f60cb82687cee40532e08f57f447e19d5f Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 31 Jul 2026 16:11:36 -0400 Subject: [PATCH 01/13] 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 Date: Fri, 31 Jul 2026 16:20:51 -0400 Subject: [PATCH 02/13] 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 Date: Fri, 31 Jul 2026 16:26:45 -0400 Subject: [PATCH 03/13] 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 Date: Fri, 31 Jul 2026 16:43:16 -0400 Subject: [PATCH 04/13] 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 Date: Fri, 31 Jul 2026 16:45:05 -0400 Subject: [PATCH 05/13] 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 Date: Fri, 31 Jul 2026 16:49:59 -0400 Subject: [PATCH 06/13] 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 Date: Wed, 21 Oct 2026 16:18:22 -0400 Subject: [PATCH 07/13] 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 Date: Thu, 22 Oct 2026 09:56:55 -0400 Subject: [PATCH 08/13] 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") From 30281ce10572ad4f24ccc5f96c8b438f2052861a Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 14:06:01 -0400 Subject: [PATCH 09/13] Fix Bearer-over-mTLS routing, certificate overrides and PFX compatibility Route password, federated password and user FIC grants through the mTLS selector. Reject certificate overrides before transport access, preserve ordinary PFX loading and restore lazy EdDSA imports. Add focused regression coverage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 54 ++++++++------ msal/mtls.py | 4 + tests/test_application.py | 138 +++++++++++++++++++++++++---------- tests/test_mtls_transport.py | 75 ++++++++++++++++++- 4 files changed, 207 insertions(+), 64 deletions(-) diff --git a/msal/application.py b/msal/application.py index c4f52fcf..ee361318 100644 --- a/msal/application.py +++ b/msal/application.py @@ -128,8 +128,7 @@ def _extract_cert_and_thumbprints(cert): 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.hazmat.primitives.asymmetric import rsa, ec, dsa, padding from cryptography.x509.oid import SignatureAlgorithmOID if cert.issuer != issuer.subject: @@ -156,10 +155,12 @@ def _is_certificate_issuer(cert, issuer): 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") + from cryptography.hazmat.primitives.asymmetric import ed25519, ed448 + if 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 @@ -386,10 +387,12 @@ def obtain_token_by_username_password(self, username, password, **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.") + for option in ("post", "cert"): + if option in kwargs: + raise ValueError( + "send_certificate_over_mtls=True does not support a {} " + "override. Omit {} to use MSAL's built-in mTLS transport " + "with the configured certificate.".format(option, option)) return super(_MtlsBearerClient, self)._obtain_token(*args, **kwargs) @@ -657,14 +660,17 @@ def get_client_assertion(): (: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`), including refresh-token - redemption during :func:`~acquire_token_silent`. It defaults to + :func:`~acquire_token_by_authorization_code`, + :func:`~acquire_token_by_refresh_token`, + :func:`~acquire_token_by_username_password`, and + :func:`~acquire_token_by_user_federated_identity_credential`), + 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``. 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. + ``post`` or ``cert`` 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. @@ -1379,10 +1385,9 @@ def _get_mtls_bearer_client(self, central_authority): 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, + """Pick the Bearer-over-mTLS client for a confidential-client user flow + 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: @@ -1417,7 +1422,7 @@ def _build_client(self, client_credential, authority, skip_regional_client=False if client_credential.get("private_key_pfx_path"): private_key, sha256_thumbprint, sha1_thumbprint, x5c, _ = _parse_pfx( client_credential["private_key_pfx_path"], - passphrase_bytes, include_chain=True) + passphrase_bytes) if client_credential.get("public_certificate") is True and x5c: headers["x5c"] = x5c elif client_credential.get("private_key"): # PEM blob @@ -2570,7 +2575,7 @@ def acquire_token_by_username_password( data=data, headers=headers, **kwargs)) if response is None: # Either ADFS or not federated - response = _clean_up(self.client.obtain_token_by_username_password( + response = _clean_up(self._client_for_confidential_request().obtain_token_by_username_password( username, password, scope=scopes, headers=headers, data=data, @@ -2612,9 +2617,10 @@ def _acquire_token_by_username_password_federated( if not grant_type: raise RuntimeError( "RSTR returned unknown token type: %s", wstrust_result.get("type")) - self.client.grant_assertion_encoders.setdefault( # Register a non-standard type - grant_type, self.client.encode_saml_assertion) - return self.client.obtain_token_by_assertion( + client = self._client_for_confidential_request() + client.grant_assertion_encoders.setdefault( # Register a non-standard type + grant_type, client.encode_saml_assertion) + return client.obtain_token_by_assertion( wstrust_result["token"], grant_type, scope=scopes, on_obtaining_tokens=lambda event: self.token_cache.add(dict( event, @@ -3441,7 +3447,7 @@ def acquire_token_by_user_federated_identity_credential( user_object_id, self.authority.tenant) _data = kwargs.pop("data", {}) _stash_client_claims(forwarded_client_claims, _data) - response = _clean_up(self.client.obtain_token_by_user_fic( + response = _clean_up(self._client_for_confidential_request().obtain_token_by_user_fic( scope=self._decorate_scope(scopes), assertion=assertion, username=username, diff --git a/msal/mtls.py b/msal/mtls.py index 67b2e31b..d6aab526 100644 --- a/msal/mtls.py +++ b/msal/mtls.py @@ -236,6 +236,10 @@ def get(self, url, **kwargs): def _request(self, method, url, **kwargs): from requests.exceptions import HTTPError + if "cert" in kwargs: + raise ValueError( + "MSAL's mTLS transport does not support a cert override. " + "Use the configured client certificate.") if self._timeout is not None: kwargs.setdefault("timeout", self._timeout) # The certificate and token request must stay at the selected authority. diff --git a/tests/test_application.py b/tests/test_application.py index eb7804d1..f2e57261 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2847,6 +2847,53 @@ def test_refresh_flag_uses_global_mtls_and_refresh_grant(self): self.assertTrue(req["data"].get("client_assertion")) self.assertNotEqual("mtls_pop", req["data"].get("token_type")) + def test_password_and_user_fic_flows_honor_mtls_flag(self): + from msal.wstrust_response import SAML_TOKEN_TYPE_V1 + + saml_grant = "urn:ietf:params:oauth:grant-type:saml1_1-bearer" + for send_over_mtls in (False, True): + for grant in ("password", saml_grant, "user_fic"): + with self.subTest(send_over_mtls=send_over_mtls, grant=grant): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, azure_region=False, + client_credential=dict( + _MTLS_CERT_CRED, + send_certificate_over_mtls=send_over_mtls)) + regular, mtls = [], [] + realm = {"account_type": ( + "Federated" if grant == saml_grant else "Managed"), + "federation_active_auth_url": "https://federation.example"} + with patch.object( + app.authority, "user_realm_discovery", return_value=realm), \ + patch("msal.application.wst_send_request", return_value={ + "token": b"saml_assertion", "type": SAML_TOKEN_TYPE_V1}), \ + patch.object(app.http_client, "post", + side_effect=self._capturing_post(regular)), \ + patch("msal.mtls._MtlsHttpClient.post", + side_effect=self._capturing_post(mtls)): + if grant == "user_fic": + result = app.acquire_token_by_user_federated_identity_credential( + ["s1"], "fic_assertion", user_object_id="uid") + else: + result = app.acquire_token_by_username_password( + "user@example.com", "password", ["s1"]) + requests = mtls if send_over_mtls else regular + self.assertEqual([], regular if send_over_mtls else mtls) + self.assertEqual(1, len(requests)) + self.assertEqual( + "https://{}/my_tenant/oauth2/v2.0/token".format( + "mtlsauth.microsoft.com" if send_over_mtls + else "login.microsoftonline.com"), + requests[0]["url"]) + self.assertEqual(grant, requests[0]["data"]["grant_type"]) + self.assertTrue(requests[0]["data"]["client_assertion"]) + self.assertEqual("Bearer", result["token_type"]) + self.assertNotIn("binding_certificate", result) + if grant == saml_grant: + self.assertEqual( + base64.urlsafe_b64encode(b"saml_assertion").rstrip(b"="), + requests[0]["data"]["assertion"]) + def test_auth_code_flag_uses_global_mtls_and_auth_code_grant(self): app = ConfidentialClientApplication( "cid", authority=self._AUTHORITY, client_credential=dict( @@ -3036,41 +3083,58 @@ def test_cc_second_call_is_bearer_cache_hit(self): self.assertEqual("Bearer", result.get("token_type")) self.assertNotIn("binding_certificate", result) - def test_post_override_rejected_for_all_mtls_bearer_grants(self): + def test_transport_overrides_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() + "refresh", "silent_refresh", "password", "federated", "user_fic"): + for option in ("post", "cert"): + with self.subTest(grant=grant, option=option): + self._assert_transport_override_rejected(grant, option) + + def _assert_transport_override_rejected(self, grant, option): + 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=[]), \ + patch.object(app.authority, "user_realm_discovery", return_value={ + "account_type": "Federated" if grant == "federated" else "Managed", + "federation_active_auth_url": "https://federation.example"}), \ + patch("msal.application.wst_send_request", return_value={ + "token": b"assertion", + "type": msal.application.SAML_TOKEN_TYPE_V1}): + override = Mock() + kwargs = {option: override} + with self.assertRaisesRegex( + ValueError, "does not support a {} override".format(option)): + if grant == "client_credentials": + app.acquire_token_for_client(["s2"], **kwargs) + elif grant == "obo": + app.acquire_token_on_behalf_of("assertion", ["s2"], **kwargs) + elif grant == "auth_code": + app.acquire_token_by_authorization_code("code", ["s2"], **kwargs) + 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"]}, **kwargs) + elif grant == "refresh": + app.acquire_token_by_refresh_token("a_rt", ["s2"], **kwargs) + elif grant in ("password", "federated"): + app.acquire_token_by_username_password( + "user@example.com", "password", ["s2"], **kwargs) + elif grant == "user_fic": + app.acquire_token_by_user_federated_identity_credential( + ["s2"], "assertion", user_object_id="uid", **kwargs) + else: + app.acquire_token_silent_with_error( + ["s1"], account, force_refresh=True, **kwargs) + override.assert_not_called() + transport.assert_not_called() + regular_transport.assert_not_called() diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py index 070e5c1e..2560091f 100644 --- a/tests/test_mtls_transport.py +++ b/tests/test_mtls_transport.py @@ -214,6 +214,17 @@ def test_owned_transport_disables_redirects_even_with_override(self): method.upper(), "https://mtlsauth.microsoft.com/token", allow_redirects=False, timeout=30) + def test_owned_transport_rejects_certificate_overrides_before_session_access(self): + client = mtls._MtlsHttpClient(b"cert", b"key") + with patch.object(client, "_ensure_session") as ensure_session: + for method in ("get", "post"): + for cert in (None, "replacement.pem", ("cert.pem", "key.pem")): + with self.subTest(method=method, cert=cert): + with self.assertRaisesRegex(ValueError, "cert override"): + getattr(client, method)( + "https://mtlsauth.microsoft.com/token", cert=cert) + ensure_session.assert_not_called() + def test_unsupported_mtls_dependencies_fail_with_upgrade_instructions(self): binding = MtlsBindingContext(b"cert", b"key", [], "thumbprint") unsupported = ( @@ -457,7 +468,9 @@ 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"]) + self.assertEqual( + expected_x5c if send_over_mtls else expected_x5c[:1], + decoded["x5c"]) return MinimalResponse( status_code=200, text='{"access_token": "at", "token_type": "Bearer"}') @@ -468,7 +481,54 @@ def post(url, data=None, **kwargs): else: app.acquire_token_for_client(["s1"], post=post) + def test_flagless_pfx_does_not_require_issuer_signature_verification(self): + from cryptography.exceptions import UnsupportedAlgorithm + 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") + with patch("cryptography.hazmat.primitives.serialization.pkcs12." + "load_key_and_certificates", + return_value=(self.leaf_key, self.leaf, [self.intermediate])), \ + patch("msal.application._is_certificate_issuer", side_effect=UnsupportedAlgorithm( + "Upgrade cryptography to verify RSA-PSS certificate chains")), \ + patch("msal.authority.tenant_discovery", return_value={ + "authorization_endpoint": endpoint.replace("/token", "/authorize"), + "token_endpoint": endpoint, + "issuer": "https://login.microsoftonline.com/tenant/v2.0"}): + for public_certificate in (False, True): + with self.subTest(public_certificate=public_certificate): + app = ConfidentialClientApplication( + "cid", authority="https://login.microsoftonline.com/tenant", + azure_region=False, client_credential={ + "private_key_pfx_path": path, + "public_certificate": public_certificate}) + captured = [] + + def post(url, data=None, **kwargs): + captured.append(data["client_assertion"]) + return MinimalResponse( + status_code=200, + text='{"access_token": "at", "token_type": "Bearer"}') + + result = app.acquire_token_for_client(["s1"], post=post) + self.assertEqual("at", result["access_token"]) + header = captured[0].split(b".")[0] + decoded = json.loads(base64.urlsafe_b64decode( + header + b"=" * (-len(header) % 4))) + if public_certificate: + self.assertEqual( + extract_certs(self.chain_pem.decode("ascii"))[:1], + decoded["x5c"]) + else: + self.assertNotIn("x5c", decoded) + with self.assertRaises(UnsupportedAlgorithm): + _load_mtls_cert_material({"private_key_pfx_path": path}) + def test_issuer_signature_verification_on_older_cryptography(self): + import builtins + class LegacyCertificate: issuer = self.leaf.issuer signature = self.leaf.signature @@ -476,8 +536,17 @@ class LegacyCertificate: 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)) + import_module = builtins.__import__ + + def legacy_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "cryptography.hazmat.primitives.asymmetric" and any( + module in fromlist for module in ("ed25519", "ed448")): + raise ImportError("EdDSA modules require cryptography 2.6") + return import_module(name, globals, locals, fromlist, level) + + with patch("builtins.__import__", side_effect=legacy_import): + 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: From 820c96f928a5c0d3d4a517c09dea233978b6e015 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 15:09:37 -0400 Subject: [PATCH 10/13] Align Bearer-over-mTLS E2E failure and coverage semantics Fail missing lab credentials instead of skipping supported scenarios and remove the unsupported live user-flow placeholder. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_e2e.py | 41 ++++++++++++++++++----------------------- 1 file changed, 18 insertions(+), 23 deletions(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 706dc5b9..aedc16b7 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -573,8 +573,6 @@ def test_credential_x509_send_certificate_over_mtls_output_bearer(self): # .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, @@ -597,17 +595,11 @@ def test_credential_x509_send_certificate_over_mtls_output_bearer(self): 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. + # Like .NET/Go, cover these flows through request capture while their + # apps lack live mTLS enablement (AADSTS700027 / AADSTS392189). + # Replacing post below proves request construction, not a TLS handshake + # or successful live acquisition. 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): @@ -646,17 +638,20 @@ def _assert_mtls(req, grant_key, grant_value): _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): - # 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 BearerOverMtlsScenarioTestCase(unittest.TestCase): + def test_missing_certificate_fails_before_acquisition(self): + for test_name in ( + "test_credential_x509_send_certificate_over_mtls_output_bearer", + "test_send_certificate_over_mtls_user_flows_capture_request"): + with self.subTest(test_name=test_name): + scenario = PublicCloudScenariosTestCase(test_name) + with patch.dict(os.environ, {}, clear=True), patch( + "msal.ConfidentialClientApplication") as factory: + with self.assertRaisesRegex( + EnvironmentError, "LAB_APP_CLIENT_CERT_PFX_PATH"): + getattr(scenario, test_name)() + factory.assert_not_called() class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once From f4813801648db1acaa596bc885ea4ac66af71bcc Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 10:45:56 -0400 Subject: [PATCH 11/13] Relocate default Bearer-over-mTLS E2E coverage Replace mocked user-grant capture with real transport observations and exact service rejections; retain safe sample summaries. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- sample/confidential_client_mtls_pop_sample.py | 2 +- tests/README.md | 8 +- tests/test_bearer_mtls_e2e_helpers.py | 44 +++++++ tests/test_e2e.py | 122 ++++++++---------- tests/test_lab_config.py | 24 +++- 5 files changed, 127 insertions(+), 73 deletions(-) create mode 100644 tests/test_bearer_mtls_e2e_helpers.py diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index 381bc56c..017eb2d4 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -69,7 +69,7 @@ def _print_result_summary(result, header): print(header) if "access_token" in result: print(" acquired: yes") - print(" token_type: ", result.get("token_type")) + print(" token_type is mtls_pop: ", result.get("token_type") == "mtls_pop") print(" carries binding_certificate: ", "binding_certificate" in result) print(" served from cache: ", result.get("token_source") == "cache") else: diff --git a/tests/README.md b/tests/README.md index e7f77bf0..f2d0132e 100644 --- a/tests/README.md +++ b/tests/README.md @@ -65,8 +65,12 @@ uses client `163ffef9-a313-45b4-ab2f-c7e2f5e0e23e` in tenant Bearer-over-mTLS also has a live client-credentials case for `westus3` and `https://vault.azure.net/.default`. OBO, refresh-token, and authorization-code -coverage captures request construction only, matching the reference SDKs' -live-app enablement limitation; it does not prove live token issuance. +captures delegate to the shipped TLS transport and require real service +responses, verified signed client assertions, grant-specific fields, and exact +OAuth rejection/status checks. The OBO flag-off control requires the ordinary +endpoint. Transport, credential, and configuration failures cannot pass as +successful captures. These deliberately invalid grants do not prove successful +user-token issuance; the reference SDKs also leave those acquisitions disabled. They do not require fetching unrelated user configuration from Key Vault. Legacy interactive and other optional E2E skips are unchanged. diff --git a/tests/test_bearer_mtls_e2e_helpers.py b/tests/test_bearer_mtls_e2e_helpers.py new file mode 100644 index 00000000..3a90d065 --- /dev/null +++ b/tests/test_bearer_mtls_e2e_helpers.py @@ -0,0 +1,44 @@ +"""Offline rejection guards for the real Bearer-over-mTLS E2E captures.""" +from contextlib import nullcontext +from unittest.mock import Mock, patch + +import requests + +from tests import test_e2e +from tests.mtls_e2e_helpers import token_request_summary +from tests.test_mtls_e2e_helpers import CertificateFixture, _CREDENTIAL + + +class TestBearerMtlsE2eHelpers(CertificateFixture): + def test_user_capture_requires_exact_rejection_not_success_or_credential_failure(self): + scenario = test_e2e.CertificateScenariosTestCase( + "test_send_certificate_over_mtls_on_behalf_of_request_capture") + self.addCleanup(scenario.doCleanups) + form = self._signed_form() + form.update( + grant_type="urn:ietf:params:oauth:grant-type:jwt-bearer", + assertion="unit-user-assertion", requested_token_use="on_behalf_of") + del form["refresh_token"] + app = Mock() + for status, payload, accepted in ( + (200, {"access_token": "must-not-be-accepted"}, False), + (400, {"error": "invalid_client", "error_codes": [700027]}, False), + (503, {"error": "invalid_grant"}, False), + (400, {"error": "invalid_grant", "error_codes": [50027]}, False), + (400, {"error": "invalid_request", "error_codes": [700027]}, False), + (400, {"error": "invalid_request", "error_codes": [50027]}, True)): + with self.subTest(status=status, accepted=accepted): + response = self._response(form=form, status=status, payload=payload) + summary = token_request_summary( + response, self.binding, requests.adapters.HTTPAdapter()) + summary["tls_adapter"] = "_MtlsHTTPAdapter" + app.acquire_token_on_behalf_of.return_value = payload + with patch( + "tests.lab_config.get_client_certificate", + return_value=_CREDENTIAL), patch( + "msal.ConfidentialClientApplication", return_value=app), patch( + "tests.test_e2e.record_token_requests", + return_value=nullcontext([summary])): + with (nullcontext() if accepted + else self.assertRaises(AssertionError)): + scenario.test_send_certificate_over_mtls_on_behalf_of_request_capture() diff --git a/tests/test_e2e.py b/tests/test_e2e.py index bd7216a6..1656884d 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -570,74 +570,60 @@ def test_credential_x509_output_bearer(self): # 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 - 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) - # 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): - # Like .NET/Go, cover these flows through request capture while their - # apps lack live mTLS enablement (AADSTS700027 / AADSTS392189). - # Replacing post below proves request construction, not a TLS handshake - # or successful live acquisition. - from tests.lab_config import get_client_certificate - 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 = [] - 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 = [] - 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 = [] - 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") + self._acquire_x509( + self._KEY_VAULT_SCOPE, send=True, region="westus3") + + def _capture_invalid_grant(self, grant, send=True): + credential, binding = self._certificate() + credential = dict(credential, send_certificate_over_mtls=send) + app = self._sni_app(credential) + with record_token_requests(binding) as records: + if grant == "urn:ietf:params:oauth:grant-type:jwt-bearer": + result = app.acquire_token_on_behalf_of( + "fake-user-assertion-for-request-capture", self._KEY_VAULT_SCOPE) + elif grant == "refresh_token": + result = app.acquire_token_by_refresh_token( + "fake-refresh-token-for-request-capture", self._KEY_VAULT_SCOPE) + else: + result = app.acquire_token_by_authorization_code( + "fake-auth-code-for-request-capture", self._KEY_VAULT_SCOPE, + redirect_uri="https://localhost/redirect") + request = self._assert_token_request( + records, self._KEY_VAULT_SCOPE, grant, + assertion_type=JWT_BEARER, regular_endpoint=not send) + self.assertNotIn("token_type", request["fields"]) + field = { + "urn:ietf:params:oauth:grant-type:jwt-bearer": "assertion", + "refresh_token": "refresh_token", "authorization_code": "code", + }[grant] + self.assertIn(field, request["fields"]) + if field == "assertion": + self.assertEqual("on_behalf_of", request["requested_token_use"]) + if field == "code": + self.assertIn("redirect_uri", request["fields"]) + expected_error = "invalid_request" if field == "assertion" else "invalid_grant" + diagnostic = "Expected the malformed-grant service rejection: {!r}".format(request) + self.assertEqual(400, request["status_code"], diagnostic) + self.assertEqual(expected_error, request["error"], diagnostic) + self.assertEqual(expected_error, result.get("error"), diagnostic) + if field == "assertion": + self.assertEqual((50027,), request["error_codes"], diagnostic) + self.assertEqual([50027], result.get("error_codes"), diagnostic) + self.assertFalse( + "access_token" in result, "Intentionally invalid input must not acquire a token.") + + def test_send_certificate_over_mtls_on_behalf_of_request_capture(self): + self._capture_invalid_grant("urn:ietf:params:oauth:grant-type:jwt-bearer") + + def test_send_certificate_over_mtls_refresh_token_request_capture(self): + self._capture_invalid_grant("refresh_token") + + def test_send_certificate_over_mtls_auth_code_request_capture(self): + self._capture_invalid_grant("authorization_code") + + def test_without_send_certificate_over_mtls_on_behalf_of_regular_endpoint(self): + self._capture_invalid_grant( + "urn:ietf:params:oauth:grant-type:jwt-bearer", send=False) class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once diff --git a/tests/test_lab_config.py b/tests/test_lab_config.py index 5ebdb394..1bcc2989 100644 --- a/tests/test_lab_config.py +++ b/tests/test_lab_config.py @@ -11,6 +11,26 @@ class TestRequiredCertificateConfiguration(unittest.TestCase): + _REQUIRED_SCENARIOS = { + "test_subject_name_issuer_authentication", + "test_credential_x509_output_pop", + "test_credential_x509_output_pop_regional", + "test_credential_x509_output_bearer", + "test_bearer_then_mtls_pop_cache_isolated", + "test_mtls_pop_then_bearer_cache_isolated", + "test_credential_x509_send_certificate_over_mtls_output_bearer", + "test_send_certificate_over_mtls_on_behalf_of_request_capture", + "test_send_certificate_over_mtls_refresh_token_request_capture", + "test_send_certificate_over_mtls_auth_code_request_capture", + "test_without_send_certificate_over_mtls_on_behalf_of_regular_endpoint", + } + + def test_required_scenario_inventory_is_complete(self): + self.assertEqual( + self._REQUIRED_SCENARIOS, + set(unittest.defaultTestLoader.getTestCaseNames( + test_e2e.CertificateScenariosTestCase))) + def test_missing_certificate_configuration_fails(self): for value in (None, "", "$(LAB_APP_CLIENT_CERT_PFX_PATH)"): with self.subTest(value=value), patch.dict(os.environ, {}, clear=True): @@ -60,8 +80,8 @@ def test_required_scenarios_fail_without_credentials_instead_of_skipping(self): suite = unittest.defaultTestLoader.loadTestsFromTestCase( test_e2e.CertificateScenariosTestCase) result = unittest.TextTestRunner(stream=io.StringIO()).run(suite) - self.assertEqual(5, result.testsRun) - self.assertEqual(5, len(result.errors)) + self.assertEqual(11, result.testsRun) + self.assertEqual(11, len(result.errors)) self.assertEqual([], result.skipped) self.assertEqual([], result.failures) for _, traceback in result.errors: From 8e6acee5781470107d1cd61689da08ff79453356 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 11:58:33 -0400 Subject: [PATCH 12/13] Treat an unspecified certificate consistently across mTLS requests Accept cert=None as omission on default transport and Bearer grant paths; retain real certificate override rejection and cover cold and cached token acquisitions. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 5 +++- msal/mtls.py | 2 +- tests/test_application.py | 48 ++++++++++++++++++++++++++++++++++++ tests/test_mtls_transport.py | 15 ++++++++++- 4 files changed, 67 insertions(+), 3 deletions(-) diff --git a/msal/application.py b/msal/application.py index ee361318..89003ceb 100644 --- a/msal/application.py +++ b/msal/application.py @@ -388,7 +388,8 @@ def obtain_token_by_username_password(self, username, password, **kwargs): class _MtlsBearerClient(_ClientWithCcsRoutingInfo): def _obtain_token(self, *args, **kwargs): for option in ("post", "cert"): - if option in kwargs: + if option in kwargs and ( + option != "cert" or kwargs[option] is not None): raise ValueError( "send_certificate_over_mtls=True does not support a {} " "override. Omit {} to use MSAL's built-in mTLS transport " @@ -671,6 +672,8 @@ def get_client_assertion(): ``ValueError``. Custom ``http_client`` instances and per-request ``post`` or ``cert`` overrides are rejected when making an mTLS token request, because MSAL must own the certificate-bearing transport. + Passing ``cert=None`` is equivalent to omitting ``cert`` and + does not replace the configured client certificate. 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. diff --git a/msal/mtls.py b/msal/mtls.py index d6aab526..d87e16de 100644 --- a/msal/mtls.py +++ b/msal/mtls.py @@ -236,7 +236,7 @@ def get(self, url, **kwargs): def _request(self, method, url, **kwargs): from requests.exceptions import HTTPError - if "cert" in kwargs: + if kwargs.get("cert") is not None: raise ValueError( "MSAL's mTLS transport does not support a cert override. " "Use the configured client certificate.") diff --git a/tests/test_application.py b/tests/test_application.py index f2e57261..447aa5e0 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2321,6 +2321,30 @@ def test_certificate_override_is_rejected_before_cache_lookup(self): cert=("other-cert.pem", "other-key.pem")) lookup.assert_not_called() + def test_none_certificate_matches_omission_on_network_and_cache(self): + for first_kwargs in ({}, {"cert": None}): + with self.subTest(first_kwargs=first_kwargs): + app = ConfidentialClientApplication( + "cid", client_credential=_MTLS_CERT_CRED, + authority=self._AUTHORITY, azure_region=False) + session = Mock() + session.request.return_value = MinimalResponse( + status_code=200, text=json.dumps({ + "access_token": "an_at", "expires_in": 3600, + "token_type": "mtls_pop"})) + with patch("msal.mtls._MtlsHttpClient._ensure_session", + return_value=session): + first = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, **first_kwargs) + for kwargs in ({}, {"cert": None}): + cached = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, **kwargs) + self.assertEqual( + first["access_token"], cached["access_token"]) + self.assertEqual( + app._TOKEN_SOURCE_CACHE, cached[app._TOKEN_SOURCE]) + self.assertEqual(1, session.request.call_count) + def test_failed_region_discovery_is_not_repeated(self): for region in ("westus3", True): with self.subTest(region=region), patch( @@ -3090,6 +3114,30 @@ def test_transport_overrides_rejected_for_all_mtls_bearer_grants(self): with self.subTest(grant=grant, option=option): self._assert_transport_override_rejected(grant, option) + def test_none_certificate_matches_omission_on_network_and_cache(self): + for first_kwargs in ({}, {"cert": None}): + with self.subTest(first_kwargs=first_kwargs): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, azure_region=False, + client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + session = Mock() + session.request.return_value = MinimalResponse( + status_code=200, text=json.dumps({ + "access_token": "an_at", "expires_in": 3600, + "token_type": "Bearer"})) + with patch("msal.mtls._MtlsHttpClient._ensure_session", + return_value=session): + first = app.acquire_token_for_client( + ["s1"], **first_kwargs) + for kwargs in ({}, {"cert": None}): + cached = app.acquire_token_for_client(["s1"], **kwargs) + self.assertEqual( + first["access_token"], cached["access_token"]) + self.assertEqual( + app._TOKEN_SOURCE_CACHE, cached[app._TOKEN_SOURCE]) + self.assertEqual(1, session.request.call_count) + def _assert_transport_override_rejected(self, grant, option): app = ConfidentialClientApplication( "cid", authority=self._AUTHORITY, client_credential=dict( diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py index 2560091f..8e1603cd 100644 --- a/tests/test_mtls_transport.py +++ b/tests/test_mtls_transport.py @@ -218,13 +218,26 @@ def test_owned_transport_rejects_certificate_overrides_before_session_access(sel client = mtls._MtlsHttpClient(b"cert", b"key") with patch.object(client, "_ensure_session") as ensure_session: for method in ("get", "post"): - for cert in (None, "replacement.pem", ("cert.pem", "key.pem")): + for cert in ("replacement.pem", ("cert.pem", "key.pem")): with self.subTest(method=method, cert=cert): with self.assertRaisesRegex(ValueError, "cert override"): getattr(client, method)( "https://mtlsauth.microsoft.com/token", cert=cert) ensure_session.assert_not_called() + def test_owned_transport_treats_none_certificate_as_unspecified(self): + client = mtls._MtlsHttpClient(b"cert", b"key") + session = Mock() + session.request.return_value = Mock(status_code=200) + with patch.object(client, "_ensure_session", return_value=session): + for method in ("get", "post"): + with self.subTest(method=method): + getattr(client, method)( + "https://mtlsauth.microsoft.com/token", cert=None) + session.request.assert_called_with( + method.upper(), "https://mtlsauth.microsoft.com/token", + cert=None, allow_redirects=False) + def test_unsupported_mtls_dependencies_fail_with_upgrade_instructions(self): binding = MtlsBindingContext(b"cert", b"key", [], "thumbprint") unsupported = ( From 4ec453670ed010d8f69262a713bbdf441ac177d6 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 12:03:05 -0400 Subject: [PATCH 13/13] Reject real Bearer certificate overrides before cache lookup Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 4 ++++ tests/test_application.py | 6 ++++++ 2 files changed, 10 insertions(+) diff --git a/msal/application.py b/msal/application.py index 89003ceb..ff1a684d 100644 --- a/msal/application.py +++ b/msal/application.py @@ -3207,6 +3207,10 @@ def acquire_token_for_client( # and stripped from the wire body by the oauth2 layer. kwargs["data"] = kwargs.get("data", {}) _stash_client_claims(forwarded_client_claims, kwargs["data"]) + if self._send_certificate_over_mtls and kwargs.get("cert") is not None: + raise ValueError( + "send_certificate_over_mtls=True does not support a cert override. " + "Use the configured client certificate.") if mtls_proof_of_possession: if kwargs.get("cert") is not None: raise ValueError( diff --git a/tests/test_application.py b/tests/test_application.py index 447aa5e0..87d2ac71 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -3136,6 +3136,12 @@ def test_none_certificate_matches_omission_on_network_and_cache(self): first["access_token"], cached["access_token"]) self.assertEqual( app._TOKEN_SOURCE_CACHE, cached[app._TOKEN_SOURCE]) + with patch.object( + app, "_acquire_token_silent_with_error") as lookup: + with self.assertRaisesRegex(ValueError, "cert override"): + app.acquire_token_for_client( + ["s1"], cert=("other.pem", "other-key.pem")) + lookup.assert_not_called() self.assertEqual(1, session.request.call_count) def _assert_transport_override_rejected(self, grant, option):