diff --git a/msal/application.py b/msal/application.py index 74395c33..2d2649f4 100644 --- a/msal/application.py +++ b/msal/application.py @@ -545,6 +545,33 @@ def get_client_assertion(): If your .pfx file contains both the private key and public cert, you can opt in for Subject Name/Issuer Auth by setting "public_certificate" to ``True``. + .. admonition:: Sending the certificate over mTLS for a Bearer token + + A certificate credential can additionally opt in to + **Bearer-over-mTLS**: MSAL presents the certificate as the client + certificate on the mutual-TLS (mTLS) handshake to Microsoft + Entra's mTLS endpoint, but requests an ordinary, *non-cert-bound* + Bearer access token (contrast ``mtls_proof_of_possession``, which + binds the token). Set ``send_certificate_over_mtls`` to ``True`` + on the certificate credential dictionary:: + + { + "private_key_pfx_path": "/path/to/your.pfx", + "public_certificate": True, + "send_certificate_over_mtls": True, + } + + It applies to every confidential-client flow + (:func:`~acquire_token_for_client`, + :func:`~acquire_token_on_behalf_of`, + :func:`~acquire_token_by_authorization_code`, and + :func:`~acquire_token_by_refresh_token`), defaults to ``False`` + (so leaving it unset changes no behavior), and requires a + certificate credential -- otherwise construction raises + ``ValueError``. A per-request ``mtls_proof_of_possession=True`` on + :func:`~acquire_token_for_client` always takes precedence, yielding + a cert-bound ``mtls_pop`` token instead of a Bearer one. + :type client_credential: Union[dict, str, None] :param dict client_claims: @@ -880,6 +907,11 @@ def get_client_assertion(): self._mtls_pop_cert_material = None # Cached parse of the cert below self._mtls_client = None # Lazily built global mTLS client self._mtls_regional_client = None # Lazily built regional mTLS client + # Bearer-over-mTLS reuses the same mTLS transport but signs a + # private_key_jwt client_assertion (x5c forced) instead of requesting a + # cert-bound token; its client(s) are likewise built lazily. + self._mtls_bearer_client = None # Lazily built global Bearer-over-mTLS client + self._mtls_bearer_regional_client = None # Lazily built regional one self._mtls_lock = Lock() if isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( @@ -889,6 +921,20 @@ def get_client_assertion(): else: # No certificate available to present over mTLS self._mtls_cert_credential = None + # Bearer-over-mTLS opt-in (app-level): present the certificate on the + # TLS handshake to the mTLS endpoint but receive a plain Bearer token + # (NOT cert-bound). Requires a certificate credential; a per-request + # mtls_proof_of_possession=True always takes precedence over this flag. + self._send_certificate_over_mtls = bool( + isinstance(client_credential, dict) + and client_credential.get("send_certificate_over_mtls")) + if self._send_certificate_over_mtls and self._mtls_cert_credential is None: + raise ValueError( + "send_certificate_over_mtls=True requires this confidential " + "client to be configured with a certificate credential (a " + "'private_key_pfx_path', or a 'private_key' plus " + "'public_certificate').") + # Warn if using a static string/bytes client_assertion (discouraged for long-running apps) if isinstance(client_credential, dict) and isinstance( client_credential.get("client_assertion"), (str, bytes)): @@ -1124,6 +1170,99 @@ def _get_mtls_client(self, central_authority): client = self._mtls_client return client + def _get_mtls_bearer_client(self, central_authority): + """Lazily build (and cache) the Bearer-over-mTLS client for + ``central_authority``. + + Like :func:`_get_mtls_client`, the token endpoint host is transformed + ``login.* -> [region.]mtlsauth.*`` and the certificate is presented over + mutual-TLS. UNLIKE mtls_pop, the request also carries a signed + private_key_jwt ``client_assertion`` (with the x5c chain FORCED on), and + does NOT request a cert-bound token (no ``token_type=mtls_pop`` / no + ``key_id`` / no ``req_cnf``). The result is therefore a plain Bearer + access token whose cache entry is not fenced by the certificate. + """ + if self._http_client_is_custom: + raise ValueError( + "send_certificate_over_mtls=True is not supported with a custom " + "http_client, because MSAL must own the TLS transport to present " + "the client certificate in the mutual-TLS handshake. Omit the " + "http_client argument to use MSAL's built-in mTLS transport.") + region = self._compute_region_to_use() + with self._mtls_lock: + cached = ( + self._mtls_bearer_regional_client if region + else self._mtls_bearer_client) + if cached is not None: + return cached + cert = self._get_mtls_pop_cert() # May raise ValueError (no cert configured) + token_endpoint = mtls.transform_token_endpoint( # login.* -> mtlsauth.* + central_authority.token_endpoint, central_authority.instance, region) + logger.debug("Bearer-over-mTLS token endpoint: %s", token_endpoint) + configuration = { + "authorization_endpoint": central_authority.authorization_endpoint, + "token_endpoint": token_endpoint, + "device_authorization_endpoint": + central_authority.device_authorization_endpoint, + } + http_client = ThrottledHttpClient( + mtls._MtlsHttpClient( + cert["cert_pem"], cert["private_key_pem"], + **self._http_client_config), + http_cache=self._http_cache, + default_throttle_time=5, + ) + # Sign a private_key_jwt assertion, FORCING the x5c chain onto the + # header regardless of the app's public_certificate opt-in (Bearer-over- + # mTLS always attaches the chain), so ESTS accepts the assertion on the + # mTLS endpoint. This is the seam that distinguishes it from mtls_pop, + # whose vanilla SN/I request sends no client_assertion at all. + assertion = JwtAssertionCreator( + cert["private_key_pem"], algorithm="PS256", + sha256_thumbprint=cert["sha256_thumbprint"], + headers={"x5c": cert["x5c"]}) + client_assertion = assertion.create_regenerative_assertion( + audience=token_endpoint, issuer=self.client_id, + additional_claims=self.client_claims or {}) + client = _ClientWithCcsRoutingInfo( + configuration, + self.client_id, + http_client=http_client, + default_headers=self._default_client_headers(), + default_body={"client_info": 1}, + client_assertion=client_assertion, + client_assertion_type=Client.CLIENT_ASSERTION_TYPE_JWT, + # Cache under the ORIGINAL login.* host (like the rest of the app), + # as a PLAIN Bearer AT (no key_id fence), so ordinary Bearer cache + # lookups in this app return it. + on_obtaining_tokens=lambda event: self.token_cache.add(dict( + event, environment=central_authority.instance)), + on_removing_rt=self.token_cache.remove_rt, + on_updating_rt=self.token_cache.update_rt) + with self._mtls_lock: + # Double-check under the lock (mirrors _get_mtls_client): another + # thread may have built and cached a client while we built ours. + if region: + if self._mtls_bearer_regional_client is None: + self._mtls_bearer_regional_client = client + client = self._mtls_bearer_regional_client + else: + if self._mtls_bearer_client is None: + self._mtls_bearer_client = client + client = self._mtls_bearer_client + return client + + def _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 @@ -1582,7 +1721,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(), @@ -2211,7 +2350,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(), @@ -2984,6 +3123,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", {}) @@ -3061,7 +3204,7 @@ def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=No _data = kwargs.pop("data", {}) _stash_client_claims(forwarded_client_claims, _data) # The implementation is NOT based on Token Exchange (RFC 8693) - response = _clean_up(self.client.obtain_token_by_assertion( # bases on assertion RFC 7521 + response = _clean_up(self._client_for_confidential_request().obtain_token_by_assertion( # bases on assertion RFC 7521 user_assertion, self.client.GRANT_TYPE_JWT, # IDTs and AAD ATs are all JWTs scope=self._decorate_scope(scopes), # Decoration is used for: diff --git a/tests/test_application.py b/tests/test_application.py index e1bf9a02..f9374a62 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2175,3 +2175,252 @@ def test_common_authority_with_flag_raises(self): authority="https://login.microsoftonline.com/common") with self.assertRaises(ValueError): app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True) + + +_MTLS_CERT_CRED_NO_X5C = { # A cert cred that does NOT opt into public_certificate + "private_key_pfx_path": _MTLS_PFX, + "passphrase": "password", +} + + +def _jwt_header_of(assertion): + """Decode the (unverified) JOSE header of a compact-serialized JWT.""" + if isinstance(assertion, bytes): # create_normal_assertion returns bytes + assertion = assertion.decode("ascii") + header_b64 = assertion.split(".")[0] + header_b64 += "=" * (-len(header_b64) % 4) # Restore base64url padding + return json.loads(base64.urlsafe_b64decode(header_b64)) + + +@patch(_OIDC_DISCOVERY, new=_MTLS_OIDC_MOCK) +class TestSendCertificateOverMtls(unittest.TestCase): + """Bearer-over-mTLS: present the SN/I cert as the client TLS cert on the + handshake to the mTLS endpoint, but receive a *plain Bearer* token (the + token is NOT cert-bound). Distinct from mtls_pop (which binds the token and + fences the cache entry by key_id) and from the classic private_key_jwt + Bearer path (which never puts the cert on the TLS handshake).""" + + _AUTHORITY = "https://login.microsoftonline.com/my_tenant" + + def _capturing_post(self, captured, token_type="Bearer"): + def mock_post(url, headers=None, data=None, *a, **k): + captured.append({ + "url": url, "data": dict(data or {}), "headers": dict(headers or {})}) + return MinimalResponse(status_code=200, text=json.dumps({ + "access_token": "an_at", "expires_in": 3600, "token_type": token_type})) + return mock_post + + # --- A. Config / builder ------------------------------------------------- + def test_flag_defaults_false(self): + app = ConfidentialClientApplication( + "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY) + self.assertFalse(app._send_certificate_over_mtls) + + def test_flag_stored_true_on_cert_credential(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + self.assertTrue(app._send_certificate_over_mtls) + + def test_flag_with_non_cert_credential_raises_at_construction(self): + with self.assertRaises(ValueError) as ctx: + ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": "a.static.jwt", + "send_certificate_over_mtls": True}) + self.assertIn("send_certificate_over_mtls", str(ctx.exception)) + + # --- B. Client-credentials behavior ------------------------------------- + def test_cc_flag_no_region_uses_global_mtls_and_returns_bearer(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + result = app.acquire_token_for_client( + ["s1"], post=self._capturing_post(captured)) + req = captured[0] + # Routed to the GLOBAL mtls host (login.* -> mtlsauth.*) + self.assertEqual( + req["url"], + "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + # Body: signed private_key_jwt client_assertion is present ... + self.assertTrue(req["data"].get("client_assertion")) + self.assertEqual(_JWT_BEARER, req["data"].get("client_assertion_type")) + # ... but NONE of the PoP wire markers + self.assertNotEqual("mtls_pop", req["data"].get("token_type")) + self.assertNotIn("req_cnf", req["data"]) + self.assertNotIn("key_id", req["data"]) + # Result is a plain Bearer token, NOT cert-bound + self.assertEqual("Bearer", result.get("token_type")) + self.assertNotIn("binding_certificate", result) + + def test_cc_x5c_is_forced_on_assertion(self): + # Bearer-over-mTLS forces the x5c chain onto the assertion header even + # when public_certificate was NOT opted into; the regular (no-flag) + # path with the same cred omits it. That contrast proves "forced". + forced = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED_NO_X5C, send_certificate_over_mtls=True)) + captured = [] + forced.acquire_token_for_client(["s1"], post=self._capturing_post(captured)) + self.assertIn( + "x5c", _jwt_header_of(captured[0]["data"]["client_assertion"]), + "Bearer-over-mTLS must force x5c onto the assertion header") + + plain = ConfidentialClientApplication( + "cid", client_credential=_MTLS_CERT_CRED_NO_X5C, authority=self._AUTHORITY) + plain_captured = [] + plain.acquire_token_for_client(["s1"], post=self._capturing_post(plain_captured)) + self.assertNotIn( + "x5c", _jwt_header_of(plain_captured[0]["data"]["client_assertion"]), + "The no-flag path must NOT force x5c (contrast)") + + def test_cc_flag_with_region_uses_regional_mtls(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, azure_region="westus3", + client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + app.acquire_token_for_client(["s1"], post=self._capturing_post(captured)) + self.assertEqual( + captured[0]["url"], + "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + + def test_cc_per_request_mtls_pop_wins_over_flag(self): + # Precedence: a per-request mtls_proof_of_possession=True ALWAYS wins + # over the app-level Bearer-over-mTLS flag. + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, + post=self._capturing_post(captured, token_type="mtls_pop")) + req = captured[0] + self.assertEqual("mtls_pop", req["data"].get("token_type")) + # The mtls_pop path is vanilla SN/I: no signed assertion on the wire + self.assertNotIn("client_assertion", req["data"]) + self.assertEqual("mtls_pop", result.get("token_type")) + self.assertTrue(result.get("binding_certificate")) + + # --- C. User flows (OBO / refresh / auth-code) -------------------------- + def _capturing_post_full(self, captured): + # A capturing post that returns a full user-token response (AT + + # id_token + client_info), needed by flows that create an account. + def mock_post(url, headers=None, data=None, *a, **k): + captured.append({ + "url": url, "data": dict(data or {}), "headers": dict(headers or {})}) + return MinimalResponse(status_code=200, text=_build_user_token_response()) + return mock_post + + def test_obo_flag_uses_global_mtls_and_obo_grant(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + app.acquire_token_on_behalf_of( + "a_user_assertion", ["s1"], post=self._capturing_post(captured)) + req = captured[0] + self.assertEqual( + req["url"], + "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + self.assertEqual("on_behalf_of", req["data"].get("requested_token_use")) + self.assertTrue(req["data"].get("client_assertion")) + self.assertEqual(_JWT_BEARER, req["data"].get("client_assertion_type")) + self.assertNotEqual("mtls_pop", req["data"].get("token_type")) + self.assertNotIn("key_id", req["data"]) + + def test_obo_flag_with_region_uses_regional_mtls(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, azure_region="westus3", + client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + app.acquire_token_on_behalf_of( + "a_user_assertion", ["s1"], post=self._capturing_post(captured)) + self.assertEqual( + captured[0]["url"], + "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + + def test_refresh_flag_uses_global_mtls_and_refresh_grant(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + app.acquire_token_by_refresh_token( + "a_rt", ["s1"], post=self._capturing_post(captured)) + req = captured[0] + self.assertEqual( + req["url"], + "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + self.assertEqual("refresh_token", req["data"].get("grant_type")) + self.assertTrue(req["data"].get("client_assertion")) + self.assertNotEqual("mtls_pop", req["data"].get("token_type")) + + def test_auth_code_flag_uses_global_mtls_and_auth_code_grant(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + captured = [] + app.acquire_token_by_authorization_code( + "a_code", ["s1"], post=self._capturing_post_full(captured)) + req = captured[0] + self.assertEqual( + req["url"], + "https://mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + self.assertEqual("authorization_code", req["data"].get("grant_type")) + self.assertTrue(req["data"].get("client_assertion")) + self.assertNotEqual("mtls_pop", req["data"].get("token_type")) + + def test_obo_without_flag_uses_regular_login_endpoint(self): + # Negative control: without the app-level flag, OBO stays on the + # regular login.* endpoint (no mTLS routing). + app = ConfidentialClientApplication( + "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY) + captured = [] + app.acquire_token_on_behalf_of( + "a_user_assertion", ["s1"], post=self._capturing_post(captured)) + self.assertEqual( + captured[0]["url"], + "https://login.microsoftonline.com/my_tenant/oauth2/v2.0/token") + + def test_cc_second_call_is_bearer_cache_hit(self): + # Regression (mirrors .NET MtlsBearerUserFlowTests' 2nd-call test, + # adapted to msal-python's cache seam): a 2nd Bearer-over-mTLS call is + # served from cache with no 2nd network round-trip and no crash on + # region/instance-metadata resolution. In .NET the crash arises because + # the AT is cached under the mtlsauth.* host and instance discovery + # rejects that host. msal-python decouples the cache/metadata authority + # (login.*, where this plain Bearer AT lives) from the transport + # endpoint (mtlsauth.*), so the 2nd lookup resolves login.* -- the + # normal, always-safe path -- and returns the cached plain Bearer AT + # (NOT cert-bound). Only client-credentials auto-serves from cache in + # msal-python (OBO/refresh/auth-code have no silent-first lookup), so CC + # is the flow that exercises this regression. + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, client_credential=dict( + _MTLS_CERT_CRED, send_certificate_over_mtls=True)) + first = app.acquire_token_for_client( + ["s1"], post=self._capturing_post([])) + self.assertEqual("Bearer", first.get("token_type")) + second = [] + result = app.acquire_token_for_client( + ["s1"], post=self._capturing_post(second)) + self.assertEqual([], second, "Second call must be served from cache") + # 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) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 200c5cdb..ef3a76d4 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -567,6 +567,105 @@ def test_credential_x509_output_bearer(self): self.assertIsNone(result.get("binding_certificate")) self.assertCacheWorksForApp(result, self._MTLS_POP_SCOPE) + # ── Bearer-over-mTLS (send_certificate_over_mtls) ────────────────────────── + # Present the SN/I cert as the client TLS certificate ON the handshake to the + # mTLS endpoint, but receive a *plain Bearer* token (the cert authenticates + # the transport; the token is NOT cert-bound). Opt-in via the app-level + # client_credential flag send_certificate_over_mtls=True. Distinct from the + # two cells above: mtls_pop BINDS the token; the classic X509 Bearer cell + # signs a private_key_jwt to the REGULAR login.* endpoint (cert never on the + # TLS handshake). Ports MSAL .NET CertificateOptions.SendCertificateOverMtls. + def test_credential_x509_send_certificate_over_mtls_output_bearer(self): + # LIVE client-credentials acquisition, allow-listed (mirrors the MSAL + # .NET live CC cell: westus3 + the Key Vault resource). Proves the cert + # goes on the TLS handshake to mtlsauth.* yet ESTS returns a plain Bearer. + from tests.lab_config import get_client_certificate + if not clean_env("LAB_APP_CLIENT_CERT_PFX_PATH"): + self.skipTest("LAB_APP_CLIENT_CERT_PFX_PATH not set") + scope = ["https://vault.azure.net/.default"] + self.app = msal.ConfidentialClientApplication( + self._SNI_ALLOWLISTED_CLIENT_ID, + authority=self._SNI_ALLOWLISTED_AUTHORITY, + azure_region="westus3", + client_credential=dict( + get_client_certificate(), send_certificate_over_mtls=True)) + result = self.app.acquire_token_for_client(scope) + self.assertIn( + "access_token", result, + "Bearer-over-mTLS request failed: %s" % result) + # A PLAIN Bearer token, NOT cert-bound (contrast the mtls_pop cell). + self.assertEqual("Bearer", result.get("token_type")) + self.assertNotIn("binding_certificate", result) + self.assertIsNone(result.get("binding_certificate")) + # The AT is an ordinary Bearer entry (no key_id/thumbprint fence), so a + # normal flagless acquire_token_for_client lookup returns it from cache + # -- the caller-visible proof of the feature's goal. (mtls_pop needed the + # special _assert_pop_cache_hit because its AT is key-id-fenced; Bearer- + # over-mTLS deliberately is not.) + self.assertCacheWorksForApp(result, scope) + + def test_send_certificate_over_mtls_user_flows_capture_request(self): + # The allow-listed app is entitled to Bearer-over-mTLS for + # client-credentials only; the OBO/refresh/auth-code apps are not + # mTLS-enabled server-side (AADSTS700027 / AADSTS392189), exactly as + # MSAL .NET [Ignore]s those live. So prove MSAL builds the correct wire + # request for every confidential user flow -- the mtlsauth endpoint + a + # private_key_jwt client_assertion + the flow's own grant, with none of + # the mtls_pop markers -- by capturing the outgoing POST over the REAL + # lab SN/I certificate, without needing those apps allow-listed. + from tests.lab_config import get_client_certificate + if not clean_env("LAB_APP_CLIENT_CERT_PFX_PATH"): + self.skipTest("LAB_APP_CLIENT_CERT_PFX_PATH not set") + cred = dict( + get_client_certificate(), send_certificate_over_mtls=True) + def _capture(sink): + def post(url, headers=None, data=None, *a, **k): + sink.append({"url": url, "data": dict(data or {})}) + return MinimalResponse(status_code=400, text=json.dumps({ + "error": "invalid_grant", + "error_description": "request captured; live acquisition " + "not required for this app"})) + return post + def _new_app(): + return msal.ConfidentialClientApplication( + self._SNI_ALLOWLISTED_CLIENT_ID, + authority=self._SNI_ALLOWLISTED_AUTHORITY, + client_credential=cred) + scope = ["https://graph.microsoft.com/.default"] + def _assert_mtls(req, grant_key, grant_value): + self.assertTrue( + req["url"].startswith("https://mtlsauth.microsoft.com/"), + "Expected the mTLS endpoint, got %s" % req["url"]) + self.assertEqual(grant_value, req["data"].get(grant_key)) + self.assertTrue(req["data"].get("client_assertion")) + self.assertNotEqual("mtls_pop", req["data"].get("token_type")) + self.assertNotIn("key_id", req["data"]) + self.assertNotIn("req_cnf", req["data"]) + cap = [] + _new_app().acquire_token_on_behalf_of( + "a_user_assertion", scope, post=_capture(cap)) + _assert_mtls(cap[0], "requested_token_use", "on_behalf_of") + cap = [] + _new_app().acquire_token_by_refresh_token( + "a_refresh_token", scope, post=_capture(cap)) + _assert_mtls(cap[0], "grant_type", "refresh_token") + cap = [] + _new_app().acquire_token_by_authorization_code( + "an_auth_code", scope, post=_capture(cap)) + _assert_mtls(cap[0], "grant_type", "authorization_code") + + def test_send_certificate_over_mtls_user_flows_live_acquire(self): + # Live OBO/refresh acquisition with Bearer-over-mTLS is pending the + # downstream apps being mTLS-enabled server-side; today ESTS rejects it + # (AADSTS700027 / AADSTS392189) -- the same class of block as the FIC + # leg-1 gate (AADSTS51000) and MSAL .NET's [Ignore]'d live user-flow + # cells. Kept as an explicit, self-documenting gate that flips to a real + # acquisition once the apps are enabled; the request shape is already + # proven by test_send_certificate_over_mtls_user_flows_capture_request. + self.skipTest( + "pending app mTLS-enablement for OBO/refresh Bearer-over-mTLS " + "(AADSTS700027 / AADSTS392189)") + class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once @classmethod