diff --git a/docs/index.rst b/docs/index.rst index e450c78f..76b9903d 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -231,6 +231,11 @@ Notes and requirements: and SHA-256 thumbprint and can create an ``ssl.SSLContext`` or ``requests.Session`` that presents the same certificate to the resource. It is not serializable and must not be persisted. +* 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, + and its result still includes ``binding_certificate`` for constructing the + downstream certificate-bearing TLS connection. .. autoclass:: msal.MtlsBindingContext :members: diff --git a/msal/application.py b/msal/application.py index ae1c5b97..64a9ef59 100644 --- a/msal/application.py +++ b/msal/application.py @@ -128,7 +128,8 @@ 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, padding + from cryptography.hazmat.primitives.asymmetric import ( + rsa, ec, dsa, ed25519, ed448, padding) from cryptography.x509.oid import SignatureAlgorithmOID if cert.issuer != issuer.subject: @@ -155,12 +156,10 @@ 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: - 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") + raise UnsupportedAlgorithm("Unsupported certificate issuer key") except (InvalidSignature, ValueError): return False return True @@ -383,6 +382,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). @@ -613,6 +621,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`, and + :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``. 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. + Successful results include ``binding_certificate`` as an + :class:`msal.MtlsBindingContext`, which can create the + certificate-bearing TLS session for the downstream resource. + 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: @@ -956,6 +999,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 ( @@ -965,6 +1012,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)): @@ -1151,6 +1216,19 @@ def _get_mtls_pop_cert(self): self._mtls_cert_credential) return self._mtls_pop_cert_material + def _decorate_mtls_binding_context(self, result): + if (not self._send_certificate_over_mtls + or not result or "access_token" not in result): + return result + cert = self._get_mtls_pop_cert() + result["binding_certificate"] = mtls.MtlsBindingContext( + cert["cert_pem"], + cert["private_key_pem"], + cert["x5c"], + cert["key_id"], + ) + return result + def _get_mtls_client(self, central_authority): """Lazily build (and cache) the mTLS PoP client for ``central_authority``. @@ -1218,6 +1296,92 @@ 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 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 @@ -1246,7 +1410,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) + passphrase_bytes, include_chain=True) if client_credential.get("public_certificate") is True and x5c: headers["x5c"] = x5c elif client_credential.get("private_key"): # PEM blob @@ -1586,7 +1750,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, @@ -1600,7 +1764,7 @@ def authorize(): # A controller in a web app if "access_token" in response: response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP telemetry_context.update_telemetry(response) - return response + return self._decorate_mtls_binding_context(response) def acquire_token_by_authorization_code( self, @@ -1676,7 +1840,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(), @@ -1691,7 +1855,7 @@ def acquire_token_by_authorization_code( if "access_token" in response: response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP telemetry_context.update_telemetry(response) - return response + return self._decorate_mtls_binding_context(response) def get_accounts(self, username=None): """Get a list of accounts which previously signed in, i.e. exists in cache. @@ -1893,6 +2057,7 @@ def acquire_token_silent( result = _clean_up(self._acquire_token_silent_with_error( scopes, account, authority=authority, force_refresh=force_refresh, claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs)) + result = self._decorate_mtls_binding_context(result) return result if result and "error" not in result else None def acquire_token_silent_with_error( @@ -1959,9 +2124,10 @@ def acquire_token_silent_with_error( if forwarded_client_claims is not None: kwargs["data"] = kwargs.get("data", {}) _stash_client_claims(forwarded_client_claims, kwargs["data"]) - return _clean_up(self._acquire_token_silent_with_error( - scopes, account, authority=authority, force_refresh=force_refresh, - claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs)) + return self._decorate_mtls_binding_context(_clean_up( + self._acquire_token_silent_with_error( + scopes, account, authority=authority, force_refresh=force_refresh, + claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs))) def _acquire_token_silent_with_error( self, @@ -2215,9 +2381,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) @@ -2305,7 +2474,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(), @@ -2316,7 +2485,7 @@ def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs): if "access_token" in response: response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP telemetry_context.update_telemetry(response) - return response + return self._decorate_mtls_binding_context(response) def acquire_token_by_username_password( self, username, password, scopes, claims_challenge=None, @@ -3070,6 +3239,8 @@ def acquire_token_for_client( mtls_cert["x5c"], mtls_cert["key_id"], # base64url x5t#S256 ) + elif self._send_certificate_over_mtls: + result = self._decorate_mtls_binding_context(result) return result def _acquire_token_for_client( @@ -3094,6 +3265,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", {}) @@ -3171,7 +3346,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: @@ -3193,7 +3368,7 @@ def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=No if "access_token" in response: response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP telemetry_context.update_telemetry(response) - return response + return self._decorate_mtls_binding_context(response) def acquire_token_by_user_federated_identity_credential( self, scopes, assertion, username=None, user_object_id=None, diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index 4918559c..b61425fc 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,28 @@ 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 a Bearer token while presenting the certificate 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:") + _call_resource(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/test_application.py b/tests/test_application.py index 9ad5a575..399ce917 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2464,3 +2464,443 @@ 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.assertIsInstance( + result.get("binding_certificate"), msal.MtlsBindingContext) + + 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_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.assertIsInstance( + result.get("binding_certificate"), msal.MtlsBindingContext) + + 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 6c4f38fb..50dad86e 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -559,6 +559,107 @@ 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")) + binding = result.get("binding_certificate") + self.assertIsInstance(binding, msal.MtlsBindingContext) + self.assertTrue(binding.x5c) + self.assertTrue(binding.thumbprint_sha256) + # 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 = [] + 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") + + 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 diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py index d9eeb7dd..89079f40 100644 --- a/tests/test_mtls_transport.py +++ b/tests/test_mtls_transport.py @@ -343,7 +343,7 @@ 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): + def test_unordered_pfx_chain_reaches_non_mtls_assertion(self): from cryptography.hazmat.primitives import serialization from msal.application import ConfidentialClientApplication from tests.http_client import MinimalResponse @@ -375,9 +375,7 @@ def post(url, data=None, **kwargs): header = data["client_assertion"].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"]) + self.assertEqual(extract_certs(expected_pem), decoded["x5c"]) return MinimalResponse( status_code=200, text='{"access_token": "at", "token_type": "Bearer"}') @@ -418,6 +416,231 @@ def test_environment_ca_bundles_replace_default_roots(self): unverified.get_adapter("https://")._ssl_context.verify_mode) +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) + 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, 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, 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) + 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"}') + + 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 + 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, 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") + 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."""