diff --git a/docs/index.rst b/docs/index.rst index bc844db4..e36455ae 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -243,6 +243,10 @@ Notes and requirements: ``os.PathLike`` objects (including ``pathlib.Path``); these replace the default trust roots. ``True`` or ``None`` selects the Requests CA bundle, while ``False`` disables server verification. +* To request Bearer-over-mTLS instead, set + ``send_certificate_over_mtls=True`` in the certificate credential and omit + ``mtls_proof_of_possession``. The resulting token uses the ``Bearer`` scheme, + is not certificate-bound, and does not include ``binding_certificate``. mTLS-bound access tokens are persisted in a separate ``AccessToken_mTLS`` cache bucket. Older SDK versions ignore that bucket and retain their normal Bearer diff --git a/msal/application.py b/msal/application.py index 4b6d9ff5..ee361318 100644 --- a/msal/application.py +++ b/msal/application.py @@ -385,6 +385,17 @@ def obtain_token_by_username_password(self, username, password, **kwargs): username, password, headers=headers, **kwargs) +class _MtlsBearerClient(_ClientWithCcsRoutingInfo): + def _obtain_token(self, *args, **kwargs): + 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) + + def _validate_mtls_pop_response(response): if (response and "access_token" in response and (response.get("token_type") or "").lower() != "mtls_pop"): @@ -629,6 +640,41 @@ 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_auth_code_flow`, + :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`` 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. + :type client_credential: Union[dict, str, None] :param dict client_claims: @@ -974,6 +1020,10 @@ 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_clients = {} # Keyed by authority and resolved region self._mtls_lock = Lock() if isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( @@ -983,6 +1033,24 @@ 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 + 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 " + "'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)): @@ -1241,6 +1309,91 @@ 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() + cache_key = ( + central_authority.instance, central_authority.token_endpoint, region) + with self._mtls_lock: + 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) + 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 = _MtlsBearerClient( + 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. + client = self._mtls_bearer_clients.setdefault(cache_key, client) + return client + + def _client_for_confidential_request(self): + """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: + 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 @@ -1609,7 +1762,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, @@ -1699,7 +1852,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(), @@ -2239,9 +2392,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) @@ -2329,7 +2485,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(), @@ -2419,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, @@ -2461,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, @@ -3122,6 +3279,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", {}) @@ -3199,7 +3360,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: @@ -3286,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/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index 17c1afc7..381bc56c 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 is mtls_pop: ", result.get("token_type") == "mtls_pop") + print(" token_type: ", result.get("token_type")) print(" carries binding_certificate: ", "binding_certificate" in result) print(" served from cache: ", result.get("token_source") == "cache") else: @@ -77,6 +77,27 @@ def _print_result_summary(result, header): print(" (inspect result['error'] / result['error_description'] to diagnose)") +def _call_resource(result): + resource_url = os.getenv("RESOURCE_URL") + if not resource_url or "access_token" not in result: + return + binding = result["binding_certificate"] + with binding.create_requests_session() as session: + response = session.get( + resource_url, + headers={ + "Authorization": + "{} {}".format( + result["token_type"], + result["access_token"], + ), + }, + timeout=30, + ) + response.raise_for_status() + print(" resource call status: ", response.status_code) + + def vanilla_sni_mtls_pop(): """Acquire an mTLS-bound PoP token using the SN/I cert as the TLS client cert.""" app = msal.ConfidentialClientApplication( @@ -94,22 +115,27 @@ def vanilla_sni_mtls_pop(): ) _print_result_summary(result, "Vanilla SN/I -> mTLS PoP result:") - resource_url = os.getenv("RESOURCE_URL") - if resource_url and "access_token" in result: - binding = result["binding_certificate"] - with binding.create_requests_session() as session: - response = session.get( - resource_url, - headers={ - "Authorization": "mtls_pop " + result["access_token"], - }, - timeout=30, - ) - response.raise_for_status() - print(" resource call status: ", response.status_code) + _call_resource(result) + return result + + +def bearer_over_mtls(): + """Acquire an ordinary Bearer token while authenticating over mTLS.""" + credential = _build_cert_credential() + credential["send_certificate_over_mtls"] = True + app = msal.ConfidentialClientApplication( + os.environ["CLIENT_ID"], + authority=os.environ["AUTHORITY"], + client_credential=credential, + azure_region=os.getenv("AZURE_REGION"), + ) + result = app.acquire_token_for_client(os.environ["SCOPE"].split()) + _print_result_summary(result, "SN/I -> Bearer-over-mTLS result:") return result if __name__ == "__main__": print("=== Vanilla SN/I -> mTLS PoP ===") vanilla_sni_mtls_pop() + print("\n=== SN/I -> Bearer-over-mTLS ===") + bearer_over_mtls() diff --git a/tests/README.md b/tests/README.md index 293073dc..e7f77bf0 100644 --- a/tests/README.md +++ b/tests/README.md @@ -62,6 +62,11 @@ PoP acquisition with a certificate-bound Graph resource call and cache reuse, and ordinary certificate Bearer authentication in `westus3`. The X509 matrix uses client `163ffef9-a313-45b4-ab2f-c7e2f5e0e23e` in tenant `bea21ebe-8b64-4d06-9f6d-6a889b120a7c`, matching the .NET and Go lab setup. + +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. 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_application.py b/tests/test_application.py index 03c1607a..f2e57261 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2636,3 +2636,505 @@ 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)) + + 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( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + 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( + 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 = [] + 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") + + 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 = [] + 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") + + 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 = [] + 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 + 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 = [] + 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"], + "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 = [] + 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") + + 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 = [] + 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"], + "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_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( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + 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"], + "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_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)) + with patch("msal.mtls._MtlsHttpClient.post", + return_value=MinimalResponse( + status_code=200, text=_build_user_token_response( + 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 = [] + + 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). + 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") + + 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)) + 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( + ["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) + + 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", "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_e2e.py b/tests/test_e2e.py index d4df481d..bd7216a6 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -561,6 +561,84 @@ 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 + 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") + class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once @classmethod diff --git a/tests/test_lab_config.py b/tests/test_lab_config.py index 43108d4b..5ebdb394 100644 --- a/tests/test_lab_config.py +++ b/tests/test_lab_config.py @@ -60,8 +60,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(3, result.testsRun) - self.assertEqual(3, len(result.errors)) + self.assertEqual(5, result.testsRun) + self.assertEqual(5, len(result.errors)) self.assertEqual([], result.skipped) self.assertEqual([], result.failures) for _, traceback in result.errors: diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py index e77aa4dc..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 = ( @@ -372,34 +383,42 @@ def test_pem_intermediates_reach_tls_handshake(self): "public_certificate": self.chain_pem.decode("ascii"), })) - def test_non_mtls_assertion_preserves_pem_chain(self): + 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")) - with 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"), - }) - - 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) + 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"}') + + 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 @@ -414,14 +433,14 @@ def test_pfx_intermediates_reach_tls_handshake(self): material = _load_mtls_cert_material({"private_key_pfx_path": path}) self._assert_chain_and_handshake(material) - def test_unordered_pfx_chain_preserves_non_mtls_leaf_only_assertion(self): - from cryptography.hazmat.primitives import serialization + 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).decode("ascii") + 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] @@ -429,33 +448,87 @@ def test_unordered_pfx_chain_preserves_non_mtls_leaf_only_assertion(self): "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.encode("ascii"), material["cert_pem"]) - self.assertEqual(extract_certs(expected_pem), material["x5c"]) + self.assertEqual(expected_pem, material["cert_pem"]) + self.assertEqual(expected_x5c, material["x5c"]) self.assertEqual(self.key_id, material["key_id"]) - with patch("msal.authority.tenant_discovery", return_value={ + 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 if send_over_mtls else expected_x5c[:1], + 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_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"}): - app = ConfidentialClientApplication( - "cid", authority="https://login.microsoftonline.com/tenant", - azure_region=False, client_credential={ - "private_key_pfx_path": path, "public_certificate": True}) - - def post(url, data=None, **kwargs): - self.assertEqual(endpoint, url) - header = data["client_assertion"].split(b".")[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))) - leaf_pem = self.leaf.public_bytes( - serialization.Encoding.PEM).decode("ascii") - self.assertEqual(extract_certs(leaf_pem), decoded["x5c"]) - return MinimalResponse( - status_code=200, - text='{"access_token": "at", "token_type": "Bearer"}') - - app.acquire_token_for_client(["s1"], post=post) + 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 @@ -463,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: