From 01b676b0815fea5dbe980965db5f246ba6bf1f20 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Mon, 6 Jul 2026 19:06:20 -0400 Subject: [PATCH 1/7] Add FIC two-leg exchange over mTLS Proof-of-Possession Stacked on top of the SN/I mTLS PoP base support. Enables a Federated Identity Credential (FIC) two-leg exchange over mutual-TLS: a leg-1 cert-bound mtls_pop token is presented as a jwt-pop client assertion over an mTLS connection bound by the same certificate to obtain the final token. - Detect the client_credential mtls_binding_certificate sub-key (FIC leg 2) and carry the leg-1 assertion as a jwt-pop client assertion over mTLS - Route every FIC leg-2 request over the mTLS endpoint, so even a Bearer token travels the cert-bound connection (implicit Bearer-over-mTLS) - Add unit + E2E coverage, docs, and a two-leg sample Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/index.rst | 4 + msal/application.py | 126 +++++++++++++----- msal/oauth2cli/oauth2.py | 1 + sample/confidential_client_mtls_pop_sample.py | 52 ++++++++ tests/test_application.py | 52 ++++++++ tests/test_e2e.py | 37 +++++ 6 files changed, 240 insertions(+), 32 deletions(-) diff --git a/docs/index.rst b/docs/index.rst index 98659dcd..aad61ffc 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -202,6 +202,10 @@ Notes and requirements: certificate can be used either as an assertion signer (Bearer) or as the TLS client certificate (mtls_pop). * mTLS PoP currently targets the public and Azure Government (Arlington) clouds. +* For a Federated Identity Credential (FIC) exchange, configure the leg-2 client + with both a ``client_assertion`` (the leg-1 token) and an + ``mtls_binding_certificate`` sub-key; the leg-1 assertion is then sent as a + ``jwt-pop`` client assertion over the same mTLS connection. Exceptions diff --git a/msal/application.py b/msal/application.py index 08b9c810..c10bb6b0 100644 --- a/msal/application.py +++ b/msal/application.py @@ -143,8 +143,9 @@ def _private_key_to_unencrypted_pem(private_key, passphrase_bytes=None): def _load_mtls_cert_material(cert_credential): """Load client-cert material for an mTLS PoP handshake from a cert credential. - ``cert_credential`` is the app's main ``client_credential`` (vanilla SN/I), - a certificate credential dict. Returns a dict with the unencrypted-PEM private key, + ``cert_credential`` is a certificate credential dict - either the app's main + ``client_credential`` (vanilla SN/I) or the ``mtls_binding_certificate`` + sub-dict (FIC leg 2). Returns a dict with the unencrypted-PEM private key, the leaf cert PEM, ``x5c``, the SHA-256 thumbprint (hex), and ``key_id`` (base64url ``x5t#S256``, used for cache binding). Raises ``ValueError`` when the credential cannot yield mTLS-capable certificate material. @@ -497,6 +498,30 @@ def get_client_assertion(): still supported for backward compatibility but is discouraged because the assertion will eventually expire. + .. admonition:: Binding an assertion to an mTLS certificate (FIC leg 2) + + *Added in version 1.38.0*: + For a Federated Identity Credential (FIC) exchange over mutual-TLS + Proof-of-Possession, the ``client_assertion`` container may also + carry an ``mtls_binding_certificate`` sub-key -- a certificate + credential (same shape as the top-level cert credential) that is + presented as the client TLS certificate during the token request:: + + { + "client_assertion": leg1_result["access_token"], # the leg-1 mtls_pop token + "mtls_binding_certificate": { + "private_key_pfx_path": "/path/to/your.pfx", + "public_certificate": True, + }, + } + + When ``mtls_binding_certificate`` is present, MSAL sends the + assertion with ``client_assertion_type`` set to the ``jwt-pop`` + type and routes the request over mTLS using that binding + certificate. See + :func:`~msal.ConfidentialClientApplication.acquire_token_for_client`'s + ``mtls_proof_of_possession`` parameter for the full two-leg flow. + .. admonition:: Supporting reading client certificates from PFX files This usage will automatically use SHA-256 thumbprint of the certificate. @@ -848,13 +873,25 @@ def get_client_assertion(): self._mtls_client = None # Lazily built global mTLS client self._mtls_regional_client = None # Lazily built regional mTLS client self._mtls_lock = Lock() - if isinstance(client_credential, dict) and not client_credential.get( + if isinstance(client_credential, dict) and client_credential.get( + "mtls_binding_certificate"): # FIC leg 2 (assertion + binding cert) + if not client_credential.get("client_assertion"): + raise ValueError( + "mtls_binding_certificate (FIC leg 2) also requires a " + "client_assertion (the leg-1 mtls_pop token) in the same " + "client_credential; the binding certificate is presented " + "to carry that assertion.") + self._mtls_cert_credential = client_credential["mtls_binding_certificate"] + self._mtls_is_fic_leg2 = True + elif isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( client_credential.get("private_key_pfx_path") or client_credential.get("private_key")): # Vanilla SN/I cert self._mtls_cert_credential = client_credential + self._mtls_is_fic_leg2 = False else: # No certificate available to present over mTLS self._mtls_cert_credential = None + self._mtls_is_fic_leg2 = False # Warn if using a static string/bytes client_assertion (discouraged for long-running apps) if isinstance(client_credential, dict) and isinstance( @@ -1014,10 +1051,11 @@ def _get_mtls_pop_cert(self): suitable certificate credential is configured.""" if self._mtls_cert_credential is None: raise ValueError( - "mtls_proof_of_possession=True requires this confidential client " + "mTLS Proof-of-Possession requires this confidential client " "to be configured with a certificate credential (a " "'private_key_pfx_path', or a 'private_key' plus " - "'public_certificate').") + "'public_certificate'); or, for a Federated Identity Credential " + "exchange, a 'client_assertion' plus an 'mtls_binding_certificate'.") with self._mtls_lock: if self._mtls_pop_cert_material is None: self._mtls_pop_cert_material = _load_mtls_cert_material( @@ -1029,13 +1067,17 @@ def _get_mtls_client(self, central_authority): The token endpoint host is transformed ``login.* -> [region.]mtlsauth.*`` (region honored when configured, global otherwise), and the client - presents the configured certificate over mutual-TLS. + presents the configured certificate over mutual-TLS. For a FIC leg-2 + credential (``client_assertion`` + ``mtls_binding_certificate``) the + assertion is sent with ``client_assertion_type = ...:jwt-pop``. """ if self._http_client_is_custom: raise ValueError( - "mtls_proof_of_possession=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 " + "mTLS 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. mTLS is engaged when you pass " + "mtls_proof_of_possession=True, or when the client_credential " + "carries mtls_binding_certificate (FIC leg 2). Omit the " "http_client argument to use MSAL's built-in mTLS transport.") region = self._compute_region_to_use() with self._mtls_lock: @@ -1059,13 +1101,20 @@ def _get_mtls_client(self, central_authority): http_cache=self._http_cache, default_throttle_time=5, ) - # Vanilla SN/I: the TLS certificate alone authenticates the client. + if self._mtls_is_fic_leg2: # FIC leg 2: assertion carried as jwt-pop + client_assertion = self.client_credential["client_assertion"] + client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT_POP + else: # Vanilla SN/I: the TLS certificate alone authenticates the client + client_assertion = None + client_assertion_type = None client = _MtlsClient( 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_assertion_type, # Cache under the ORIGINAL login.* host, never the mtlsauth.* host, # so mtls_pop ATs share the environment with the rest of the app. on_obtaining_tokens=lambda event: self.token_cache.add(dict( @@ -2779,7 +2828,9 @@ def acquire_token_for_client( Requirements: the app must be configured with a certificate credential (``private_key_pfx_path``, or ``private_key`` + - ``public_certificate``); the ``authority`` must be tenanted (not + ``public_certificate``) - or, for a Federated Identity Credential + (FIC) exchange, a ``client_assertion`` plus an + ``mtls_binding_certificate``; the ``authority`` must be tenanted (not ``/common`` or ``/organizations``); and MSAL's built-in HTTP transport must be in use (a custom ``http_client`` cannot perform the mTLS handshake). Any of these unmet raises ``ValueError``. @@ -2813,31 +2864,41 @@ def acquire_token_for_client( "fmi_path must be a string, got {}".format(type(fmi_path).__name__)) kwargs["data"] = kwargs.get("data", {}) kwargs["data"]["fmi_path"] = fmi_path - if mtls_proof_of_possession: - # An mTLS transport is required to present the certificate for a - # cert-bound PoP request. + if mtls_proof_of_possession or self._mtls_is_fic_leg2: + # An mTLS transport is required whenever we present the certificate: + # for a cert-bound PoP request (mtls_proof_of_possession=True) and + # for every FIC leg-2 request - there the leg-1 assertion is itself + # bound to the certificate, so even a Bearer final token must travel + # over the same mTLS connection ("implicit Bearer-over-mTLS"). if self._http_client_is_custom: raise ValueError( - "mtls_proof_of_possession=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 " + "mTLS 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. mTLS is engaged " + "when you pass mtls_proof_of_possession=True, or when the " + "client_credential carries mtls_binding_certificate (FIC " + "leg 2). Omit the http_client argument to use MSAL's " "built-in mTLS transport.") if self.authority.tenant.lower() in ("common", "organizations"): raise ValueError( - "mtls_proof_of_possession=True requires a tenanted authority. " - "Use a specific tenant id or domain instead of /common or " - "/organizations.") + "mTLS Proof-of-Possession requires a tenanted authority. It " + "is engaged when you pass mtls_proof_of_possession=True, or " + "when the client_credential carries mtls_binding_certificate " + "(FIC leg 2). Use a specific tenant id or domain instead of " + "/common or /organizations.") # Parse/validate the certificate now (fail fast). mtls_cert = self._get_mtls_pop_cert() - # Cert-bound PoP: request an mtls_pop token and bind its cache - # entry to the cert via key_id (base64url x5t#S256). token_type - # also routes _acquire_token_for_client() to the mTLS client. data = dict(kwargs.get("data") or {}) - data["token_type"] = "mtls_pop" - data["key_id"] = mtls_cert["key_id"] + if mtls_proof_of_possession: + # Cert-bound PoP: request an mtls_pop token and bind its cache + # entry to the cert via key_id (base64url x5t#S256). token_type + # also routes _acquire_token_for_client() to the mTLS client. + data["token_type"] = "mtls_pop" + data["key_id"] = mtls_cert["key_id"] + # else: FIC leg-2 without the flag -> Bearer-over-mTLS. No token_type + # / key_id, so the final Bearer token caches normally; the mTLS + # client is still selected because self._mtls_is_fic_leg2 is True. kwargs["data"] = data - result = _clean_up(self._acquire_token_silent_with_error( scopes, None, claims_challenge=claims_challenge, **kwargs)) if mtls_proof_of_possession and result and "access_token" in result: @@ -2856,8 +2917,8 @@ def acquire_token_for_client( "certificate-bound.".format(result.get("token_type"))), } # Surface the PUBLIC binding certificate (never the private key), so - # callers can correlate the token to its cert. Survives _clean_up - # (the key has no "_" prefix). + # callers - including FIC leg-2 / cross-app hand-off - can correlate + # the token to its cert. Survives _clean_up (key has no "_" prefix). result["binding_certificate"] = { "x5c": mtls_cert["x5c"], "thumbprint_sha256": mtls_cert["key_id"], # base64url x5t#S256 @@ -2882,9 +2943,10 @@ def _acquire_token_for_client( telemetry_context = self._build_telemetry_context( self.ACQUIRE_TOKEN_FOR_CLIENT_ID, refresh_reason=refresh_reason, token_type=data.get("token_type")) - if is_mtls_pop: - # Present the certificate over mTLS to obtain a cert-bound - # (mtls_pop) token. + if is_mtls_pop or self._mtls_is_fic_leg2: + # Present the certificate over mTLS to obtain a cert-bound token + # (mtls_pop), or to carry a cert-bound FIC leg-1 assertion as jwt-pop + # (Bearer-over-mTLS when the flag is absent). client = self._get_mtls_client(self.authority) else: client = self._regional_client or self.client diff --git a/msal/oauth2cli/oauth2.py b/msal/oauth2cli/oauth2.py index 4590d52d..fa2419f9 100644 --- a/msal/oauth2cli/oauth2.py +++ b/msal/oauth2cli/oauth2.py @@ -43,6 +43,7 @@ def encode_saml_assertion(assertion): CLIENT_ASSERTION_TYPE_JWT = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" CLIENT_ASSERTION_TYPE_SAML2 = "urn:ietf:params:oauth:client-assertion-type:saml2-bearer" + CLIENT_ASSERTION_TYPE_JWT_POP = "urn:ietf:params:oauth:client-assertion-type:jwt-pop" client_assertion_encoders = {CLIENT_ASSERTION_TYPE_SAML2: encode_saml_assertion} @property diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index cf399e37..44e58cd1 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -4,6 +4,9 @@ mutual-TLS (mTLS) handshake to Microsoft Entra, and obtains an mTLS-bound Proof-of-Possession (PoP) access token (``token_type == "mtls_pop"``). +It also shows the optional two-leg Federated Identity Credential (FIC) exchange +over mTLS PoP. + Prerequisites ------------- * A confidential-client app registration configured with a certificate. @@ -95,6 +98,55 @@ def vanilla_sni_mtls_pop(): return result +def fic_two_leg_over_mtls_pop(): + """Optional: two-leg Federated Identity Credential (FIC) exchange over mTLS PoP. + + Leg 1: the SN/I confidential client acquires a cert-bound mtls_pop token for + the token-exchange audience. + Leg 2: a second confidential client presents that token as a ``jwt-pop`` + client assertion, over an mTLS connection bound by the SAME cert, to + obtain the final token (Bearer or mtls_pop). + """ + cert = _build_cert_credential() + + # --- Leg 1: acquire the federated (cert-bound) assertion --------------- + leg1_app = msal.ConfidentialClientApplication( + os.environ["CLIENT_ID"], + authority=os.environ["AUTHORITY"], + client_credential=cert, + azure_region=os.getenv("AZURE_REGION"), + ) + # The exchange audience is caller-supplied, not hard-coded by MSAL. + exchange_scope = os.getenv( + "FIC_EXCHANGE_SCOPE", "api://AzureADTokenExchange/.default") + leg1 = leg1_app.acquire_token_for_client( + [exchange_scope], mtls_proof_of_possession=True) + if "access_token" not in leg1: + _print_result_summary(leg1, "FIC leg 1 result:") + return leg1 + + # --- Leg 2: exchange the leg-1 token for the final token --------------- + leg2_app = msal.ConfidentialClientApplication( + os.getenv("LEG2_CLIENT_ID", os.environ["CLIENT_ID"]), + authority=os.environ["AUTHORITY"], + client_credential={ + "client_assertion": leg1["access_token"], # the leg-1 mtls_pop token + "mtls_binding_certificate": cert, # binds the leg-2 TLS handshake + }, + azure_region=os.getenv("AZURE_REGION"), + ) + + # Final token as mTLS PoP (drop mtls_proof_of_possession for Bearer-over-mTLS): + leg2 = leg2_app.acquire_token_for_client( + os.environ["SCOPE"].split(), mtls_proof_of_possession=True) + _print_result_summary(leg2, "FIC leg 2 result:") + return leg2 + + if __name__ == "__main__": print("=== Vanilla SN/I -> mTLS PoP ===") vanilla_sni_mtls_pop() + + if os.getenv("RUN_FIC_SAMPLE"): + print("\n=== FIC two-leg over mTLS PoP ===") + fic_two_leg_over_mtls_pop() diff --git a/tests/test_application.py b/tests/test_application.py index cc54e1ff..48f182fc 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -1602,6 +1602,7 @@ def mock_post(url, headers=None, data=None, *args, **kwargs): "issuer": "https://login.microsoftonline.com/my_tenant/v2.0", }) _JWT_BEARER = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" +_JWT_POP = "urn:ietf:params:oauth:client-assertion-type:jwt-pop" @patch(_OIDC_DISCOVERY, new=_MTLS_OIDC_MOCK) @@ -1701,12 +1702,63 @@ def test_regional_mtls_endpoint(self): captured[0]["url"], "https://westus3.mtlsauth.microsoft.com/my_tenant/oauth2/v2.0/token") + def test_fic_leg2_pop_sends_jwt_pop_over_mtls(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": "LEG1_TOKEN", + "mtls_binding_certificate": _MTLS_CERT_CRED}) + captured = [] + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, post=self._capturing_post(captured)) + req = captured[0] + self.assertTrue(req["url"].startswith("https://mtlsauth.microsoft.com/")) + self.assertEqual("LEG1_TOKEN", req["data"].get("client_assertion")) + self.assertEqual(_JWT_POP, req["data"].get("client_assertion_type")) + self.assertEqual("mtls_pop", req["data"].get("token_type")) + self.assertEqual("mtls_pop", result.get("token_type")) + + def test_fic_leg2_bearer_still_travels_over_mtls(self): + # No flag -> Bearer final token, but the cert-bound leg-1 assertion still + # requires the mTLS endpoint and jwt-pop ("implicit Bearer-over-mTLS"). + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": "LEG1_TOKEN", + "mtls_binding_certificate": _MTLS_CERT_CRED}) + captured = [] + result = app.acquire_token_for_client( + ["s1"], post=self._capturing_post(captured, token_type="Bearer")) + req = captured[0] + self.assertTrue(req["url"].startswith("https://mtlsauth.microsoft.com/")) + self.assertEqual(_JWT_POP, req["data"].get("client_assertion_type")) + self.assertNotIn("token_type", req["data"]) + self.assertEqual("4|730,2|", req["headers"].get(CLIENT_CURRENT_TELEMETRY)) + self.assertEqual("Bearer", result.get("token_type")) + self.assertNotIn("binding_certificate", result) + + def test_binding_cert_without_client_assertion_raises(self): + # FIC leg 2 needs the binding cert AND the leg-1 assertion. A binding + # cert with no client_assertion must fail fast at construction with a + # clear ValueError, not a later KeyError when the request is built. + with self.assertRaises(ValueError): + ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={"mtls_binding_certificate": _MTLS_CERT_CRED}) + def test_secret_credential_with_flag_raises(self): app = ConfidentialClientApplication( "cid", client_credential="a_secret", authority=self._AUTHORITY) with self.assertRaises(ValueError): app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True) + def test_string_assertion_without_binding_cert_with_flag_raises(self): + app = ConfidentialClientApplication( + "cid", client_credential={"client_assertion": "just_a_string"}, + authority=self._AUTHORITY) + with self.assertRaises(ValueError): + app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True) + def test_custom_http_client_with_flag_fails_fast(self): app = ConfidentialClientApplication( "cid", client_credential=_MTLS_CERT_CRED, authority=self._AUTHORITY, diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 9d80a9b6..9f58deb2 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -508,6 +508,43 @@ def test_sni_mtls_pop_for_client_regional(self): self.assertIn("access_token", result, "Regional mTLS PoP failed: %s" % result) self.assertEqual("mtls_pop", result.get("token_type")) + def test_fic_two_leg_over_mtls_pop(self): + 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") + authority = "https://login.microsoftonline.com/microsoft.onmicrosoft.com" + # Leg 1: SN/I cert -> cert-bound (mtls_pop) federated assertion for the + # caller-supplied token-exchange audience. + leg1_app = msal.ConfidentialClientApplication( + LAB_APP_CLIENT_ID, authority=authority, + client_credential=get_client_certificate()) + leg1 = leg1_app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], mtls_proof_of_possession=True) + if "access_token" not in leg1: + self.skipTest("FIC leg 1 not pre-authorized for token exchange: %s" % leg1) + self.assertEqual("mtls_pop", leg1.get("token_type"), "Leg 1 must be mtls_pop") + leg1_thumbprint = leg1["binding_certificate"]["thumbprint_sha256"] + + # Leg 2: same cert binds the TLS handshake, leg-1 token carried as jwt-pop. + leg2_app = msal.ConfidentialClientApplication( + LAB_APP_CLIENT_ID, authority=authority, + client_credential={ + "client_assertion": leg1["access_token"], + "mtls_binding_certificate": get_client_certificate()}) + # Leg 2 -> Bearer over mTLS (implicit Bearer-over-mTLS, no flag) + leg2_bearer = leg2_app.acquire_token_for_client(self._MTLS_POP_SCOPE) + self.assertIn("access_token", leg2_bearer, "Leg 2 (Bearer) failed: %s" % leg2_bearer) + # Leg 2 -> mTLS PoP final token + leg2_pop = leg2_app.acquire_token_for_client( + self._MTLS_POP_SCOPE, mtls_proof_of_possession=True) + self.assertIn("access_token", leg2_pop, "Leg 2 (PoP) failed: %s" % leg2_pop) + self.assertEqual("mtls_pop", leg2_pop.get("token_type")) + # The final PoP token must be bound to the leg-1 certificate thumbprint + cnf = self._cnf_x5t_s256(leg2_pop["access_token"]) + if cnf: + self.assertEqual(leg1_thumbprint, cnf, + "Final token cnf must be bound to the leg-1 cert thumbprint") + class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once @classmethod From 90d0e67a4e44718bf7ac54007e3806971c64fc32 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Wed, 8 Jul 2026 17:00:24 -0400 Subject: [PATCH 2/7] Fail fast when a FIC leg-2 app is used with a non-mTLS flow A confidential client configured for a Federated Identity Credential (FIC) leg-2 exchange (a client_assertion plus an mtls_binding_certificate) is only meaningful through acquire_token_for_client, which presents the binding certificate over mutual-TLS and carries the leg-1 assertion as jwt-pop. If such an app is instead used with any other flow (authorization code, on-behalf-of, refresh token, username/password, or user-FIC), MSAL would send that cert-bound leg-1 assertion as an ordinary jwt-bearer over a non-mTLS connection. ESTS rejects it, but with a confusing server error. Add a small _reject_if_mtls_binding_cert guard and call it at the top of those flows so the misuse fails fast with a clear ValueError that points the caller to acquire_token_for_client. The guard is a no-op for every non-FIC-leg-2 app, so public clients and ordinary confidential clients are unaffected. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 22 ++++++++++++++++++++++ tests/test_application.py | 24 ++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/msal/application.py b/msal/application.py index c10bb6b0..241ddc95 100644 --- a/msal/application.py +++ b/msal/application.py @@ -1137,6 +1137,22 @@ def _get_mtls_client(self, central_authority): client = self._mtls_client return client + def _reject_if_mtls_binding_cert(self, api_name): + # A FIC leg-2 credential (client_assertion + mtls_binding_certificate) + # is bound to the certificate: the leg-1 assertion is only valid when + # presented over the mTLS connection that acquire_token_for_client() + # establishes. Any other flow would send that cert-bound assertion as + # an ordinary jwt-bearer over a non-mTLS connection, which ESTS rejects. + # Fail fast with a clear message instead of a confusing server error. + if getattr(self, "_mtls_is_fic_leg2", False): + raise ValueError( + "This application is configured for a Federated Identity " + "Credential (FIC) leg-2 exchange (a 'client_assertion' plus an " + "'mtls_binding_certificate'). That credential can only be " + "presented over mutual-TLS, so {}() is not supported on this " + "application - use acquire_token_for_client() instead.".format( + api_name)) + def _build_client(self, client_credential, authority, skip_regional_client=False): client_assertion = None client_assertion_type = None @@ -1502,6 +1518,7 @@ def authorize(): # A controller in a web app pass # Simply ignore them return redirect(url_for("index")) """ + self._reject_if_mtls_binding_cert("acquire_token_by_auth_code_flow") self._validate_ssh_cert_input_data(kwargs.get("data", {})) telemetry_context = self._build_telemetry_context( self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID) @@ -1567,6 +1584,7 @@ def acquire_token_by_authorization_code( - A successful response would contain "access_token" key, - an error response would contain "error" and usually "error_description". """ + self._reject_if_mtls_binding_cert("acquire_token_by_authorization_code") # If scope is absent on the wire, STS will give you a token associated # to the FIRST scope sent during the authorization request. # So in theory, you can omit scope here when you were working with only @@ -2179,6 +2197,7 @@ def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs): * A dict contains "error" and some other keys, when error happened. * A dict contains no "error" key means migration was successful. """ + self._reject_if_mtls_binding_cert("acquire_token_by_refresh_token") self._validate_ssh_cert_input_data(kwargs.get("data", {})) telemetry_context = self._build_telemetry_context( self.ACQUIRE_TOKEN_BY_REFRESH_TOKEN, @@ -2233,6 +2252,7 @@ def acquire_token_by_username_password( Migration guide: https://aka.ms/msal-ropc-migration """ + self._reject_if_mtls_binding_cert("acquire_token_by_username_password") is_confidential_app = self.client_credential or isinstance( self, ConfidentialClientApplication) if not is_confidential_app: @@ -3001,6 +3021,7 @@ def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=No - A successful response would contain "access_token" key, - an error response would contain "error" and usually "error_description". """ + self._reject_if_mtls_binding_cert("acquire_token_on_behalf_of") telemetry_context = self._build_telemetry_context( self.ACQUIRE_TOKEN_ON_BEHALF_OF_ID) # The implementation is NOT based on Token Exchange (RFC 8693) @@ -3057,6 +3078,7 @@ def acquire_token_by_user_federated_identity_credential( - A successful response would contain "access_token" key, - an error response would contain "error" and usually "error_description". """ + self._reject_if_mtls_binding_cert("acquire_token_by_user_federated_identity_credential") # Input validation if not assertion: raise ValueError("assertion is required and must be non-empty") diff --git a/tests/test_application.py b/tests/test_application.py index 48f182fc..3d6db5a6 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -1746,6 +1746,30 @@ def test_binding_cert_without_client_assertion_raises(self): "cid", authority=self._AUTHORITY, client_credential={"mtls_binding_certificate": _MTLS_CERT_CRED}) + def test_fic_leg2_app_rejects_non_client_credential_flows(self): + # A FIC leg-2 app (client_assertion + mtls_binding_certificate) is only + # valid via acquire_token_for_client over mTLS. Any other flow would put + # the cert-bound leg-1 assertion on a non-mTLS jwt-bearer request, so it + # must fail fast with a clear ValueError rather than a confusing 4xx. + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": "LEG1_TOKEN", + "mtls_binding_certificate": _MTLS_CERT_CRED}) + with self.assertRaises(ValueError): + app.acquire_token_by_authorization_code("code", ["s1"]) + with self.assertRaises(ValueError): + app.acquire_token_on_behalf_of("user_token", ["s1"]) + with self.assertRaises(ValueError): + app.acquire_token_by_refresh_token("refresh_token", ["s1"]) + with self.assertRaises(ValueError): + app.acquire_token_by_username_password("user", "password", ["s1"]) + with self.assertRaises(ValueError): + app.acquire_token_by_auth_code_flow({}, {}) + with self.assertRaises(ValueError): + app.acquire_token_by_user_federated_identity_credential( + ["s1"], "assertion_token", username="user@contoso.com") + def test_secret_credential_with_flag_raises(self): app = ConfidentialClientApplication( "cid", client_credential="a_secret", authority=self._AUTHORITY) From acd78ed25346cdb6dd09ee6adfc5a9a191840f31 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Wed, 8 Jul 2026 17:58:26 -0400 Subject: [PATCH 3/7] Reject FIC leg-2 apps on silent flows too The _reject_if_mtls_binding_cert guard already covered the six non-client-credential acquisition flows, but not acquire_token_silent / acquire_token_silent_with_error. With a shared token cache holding user refresh tokens, a FIC leg-2 app (client_assertion + mtls_binding_certificate) could silently redeem one of those RTs, putting the cert-bound leg-1 assertion on a non-mTLS jwt-bearer request -- which ESTS rejects with a confusing invalid_client rather than a clear message. Add the guard to both public silent entry points. Both call the private _acquire_token_silent_with_error directly (not each other), and acquire_token_for_client also uses that private helper, so the guard lives on the two public wrappers -- never on the private method -- to keep the legitimate mTLS client-credential flow working. No-op for normal apps. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 2 ++ tests/test_application.py | 10 ++++++++++ 2 files changed, 12 insertions(+) diff --git a/msal/application.py b/msal/application.py index 241ddc95..8dfdfc7c 100644 --- a/msal/application.py +++ b/msal/application.py @@ -1804,6 +1804,7 @@ def acquire_token_silent( if cache lookup succeeded. - None when cache lookup does not yield a token. """ + self._reject_if_mtls_binding_cert("acquire_token_silent") if not account: return None # A backward-compatible NO-OP to drop the account=None usage result = _clean_up(self._acquire_token_silent_with_error( @@ -1858,6 +1859,7 @@ def acquire_token_silent_with_error( - None when there is simply no token in the cache. - A dict containing an "error" key, when token refresh failed. """ + self._reject_if_mtls_binding_cert("acquire_token_silent_with_error") if not account: return None # A backward-compatible NO-OP to drop the account=None usage return _clean_up(self._acquire_token_silent_with_error( diff --git a/tests/test_application.py b/tests/test_application.py index 3d6db5a6..b4e5966a 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -1769,6 +1769,16 @@ def test_fic_leg2_app_rejects_non_client_credential_flows(self): with self.assertRaises(ValueError): app.acquire_token_by_user_federated_identity_credential( ["s1"], "assertion_token", username="user@contoso.com") + # Silent flows are also unsupported: a shared cache could hold user RTs + # whose redemption would put the cert-bound assertion on a non-mTLS + # jwt-bearer request. Guard both silent entry points (they call the + # private _acquire_token_silent_with_error directly, not each other). + account = {"home_account_id": "uid.utid", + "environment": "login.microsoftonline.com", "username": "u"} + with self.assertRaises(ValueError): + app.acquire_token_silent(["s1"], account) + with self.assertRaises(ValueError): + app.acquire_token_silent_with_error(["s1"], account) def test_secret_credential_with_flag_raises(self): app = ConfidentialClientApplication( From 1e41f24eea25e6d0f39e53a58eb713ddbb160e93 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Thu, 22 Oct 2026 08:53:20 -0400 Subject: [PATCH 4/7] Address FIC mTLS review feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/index.rst | 10 +- msal/application.py | 43 +++- sample/confidential_client_mtls_pop_sample.py | 34 ++- tests/test_application.py | 36 +++ tests/test_e2e.py | 222 +++++++++++++----- 5 files changed, 264 insertions(+), 81 deletions(-) diff --git a/docs/index.rst b/docs/index.rst index f060eae0..02d04a52 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -209,9 +209,15 @@ Notes and requirements: (non-mTLS) certificate path is unchanged and still sends a signed ``client_assertion`` (``private_key_jwt``) to obtain an ordinary token. * For a Federated Identity Credential (FIC) exchange, configure the leg-2 client - with both a ``client_assertion`` (the leg-1 token) and an + with both a ``client_assertion`` callable (acquiring the leg-1 token) and an ``mtls_binding_certificate`` sub-key; the leg-1 assertion is then sent as a - ``jwt-pop`` client assertion over the same mTLS connection. + ``jwt-pop`` client assertion over mTLS using the same certificate as leg 1. + A fixed assertion string expires and emits a deprecation warning; the callable + lets the leg-1 application reuse cached tokens and acquire fresh ones as needed. + The target application must have a federated identity credential trusting the + leg-1 token's issuer, subject, and audience. See + ``fic_two_leg_over_mtls_pop()`` in + ``sample/confidential_client_mtls_pop_sample.py`` for the two-leg setup. Exceptions diff --git a/msal/application.py b/msal/application.py index 7cf63e20..30ddd966 100644 --- a/msal/application.py +++ b/msal/application.py @@ -537,15 +537,29 @@ def get_client_assertion(): .. admonition:: Binding an assertion to an mTLS certificate (FIC leg 2) - *Added in version 1.38.0*: For a Federated Identity Credential (FIC) exchange over mutual-TLS Proof-of-Possession, the ``client_assertion`` container may also carry an ``mtls_binding_certificate`` sub-key -- a certificate credential (same shape as the top-level cert credential) that is - presented as the client TLS certificate during the token request:: + presented as the client TLS certificate during the token request. + An explicitly supplied binding certificate must be a non-empty + certificate credential dictionary; invalid or incomplete values + raise ``ValueError``. + Use a callable to acquire the leg-1 token on demand; the leg-1 + application's cache reuses valid tokens and renews expired ones:: + + def get_leg1_assertion(): + result = leg1_app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], + mtls_proof_of_possession=True, + ) + if "access_token" not in result: + raise RuntimeError( + "FIC leg 1 failed: {}".format(result.get("error"))) + return result["access_token"] { - "client_assertion": leg1_result["access_token"], # the leg-1 mtls_pop token + "client_assertion": get_leg1_assertion, "mtls_binding_certificate": { "private_key_pfx_path": "/path/to/your.pfx", "public_certificate": True, @@ -554,8 +568,10 @@ def get_client_assertion(): When ``mtls_binding_certificate`` is present, MSAL sends the assertion with ``client_assertion_type`` set to the ``jwt-pop`` - type and routes the request over mTLS using that binding - certificate. See + type and routes the request over mTLS using the same certificate + as the leg-1 application. A fixed token string is supported for + short-lived use, but emits the static-assertion deprecation + warning and cannot be renewed automatically. See :func:`~msal.ConfidentialClientApplication.acquire_token_for_client`'s ``mtls_proof_of_possession`` parameter for the full two-leg flow. @@ -926,15 +942,26 @@ def get_client_assertion(): self._mtls_client = None # Lazily built global mTLS client self._mtls_regional_client = None # Lazily built regional mTLS client self._mtls_lock = Lock() - if isinstance(client_credential, dict) and client_credential.get( - "mtls_binding_certificate"): # FIC leg 2 (assertion + binding cert) + if (isinstance(client_credential, dict) + and "mtls_binding_certificate" in client_credential): + binding_certificate = client_credential["mtls_binding_certificate"] + if not isinstance(binding_certificate, dict) or not ( + binding_certificate.get("private_key_pfx_path") + or (binding_certificate.get("private_key") + and isinstance( + binding_certificate.get("public_certificate"), str) + and binding_certificate["public_certificate"])): + raise ValueError( + "mtls_binding_certificate must be a non-empty certificate " + "credential dictionary with 'private_key_pfx_path', or " + "'private_key' plus a PEM 'public_certificate'.") if not client_credential.get("client_assertion"): raise ValueError( "mtls_binding_certificate (FIC leg 2) also requires a " "client_assertion (the leg-1 mtls_pop token) in the same " "client_credential; the binding certificate is presented " "to carry that assertion.") - self._mtls_cert_credential = client_credential["mtls_binding_certificate"] + self._mtls_cert_credential = binding_certificate self._mtls_is_fic_leg2 = True elif isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index e2765620..7750bbc0 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -27,6 +27,17 @@ SCOPE e.g. https://graph.microsoft.com/.default AZURE_REGION (optional) e.g. westus3; omit for the global mtls endpoint +For the optional FIC sample, also configure:: + + RUN_FIC_SAMPLE set to 1 to run the two-leg exchange + LEG2_CLIENT_ID target application with a provisioned federated identity credential + LEG2_AUTHORITY tenanted authority for that target application + FIC_EXCHANGE_SCOPE (optional) defaults to api://AzureADTokenExchange/.default + +The leg-1 app must be authorized for mTLS PoP at the exchange audience. The +leg-2 app's federated identity credential must trust the leg-1 token's issuer, +subject, and audience. Both legs use the SAME local certificate. + Then run:: python confidential_client_mtls_pop_sample.py @@ -120,19 +131,24 @@ def fic_two_leg_over_mtls_pop(): # The exchange audience is caller-supplied, not hard-coded by MSAL. exchange_scope = os.getenv( "FIC_EXCHANGE_SCOPE", "api://AzureADTokenExchange/.default") - leg1 = leg1_app.acquire_token_for_client( - [exchange_scope], mtls_proof_of_possession=True) - if "access_token" not in leg1: - _print_result_summary(leg1, "FIC leg 1 result:") - return leg1 + + def get_leg1_assertion(): + # Reuse the leg-1 cache, acquiring a fresh token when the old one expires. + leg1 = leg1_app.acquire_token_for_client( + [exchange_scope], mtls_proof_of_possession=True) + if "access_token" not in leg1: + _print_result_summary(leg1, "FIC leg 1 result:") + raise RuntimeError( + "FIC leg 1 failed: {}".format(leg1.get("error"))) + return leg1["access_token"] # --- Leg 2: exchange the leg-1 token for the final token --------------- leg2_app = msal.ConfidentialClientApplication( - os.getenv("LEG2_CLIENT_ID", os.environ["CLIENT_ID"]), - authority=os.environ["AUTHORITY"], + os.environ["LEG2_CLIENT_ID"], + authority=os.environ["LEG2_AUTHORITY"], client_credential={ - "client_assertion": leg1["access_token"], # the leg-1 mtls_pop token - "mtls_binding_certificate": cert, # binds the leg-2 TLS handshake + "client_assertion": get_leg1_assertion, + "mtls_binding_certificate": cert, }, azure_region=os.getenv("AZURE_REGION"), ) diff --git a/tests/test_application.py b/tests/test_application.py index 6550d615..4c47d27b 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2402,6 +2402,42 @@ def test_binding_cert_without_client_assertion_raises(self): "cid", authority=self._AUTHORITY, client_credential={"mtls_binding_certificate": _MTLS_CERT_CRED}) + def test_invalid_binding_certificate_is_rejected(self): + for certificate in ( + {}, None, "", [], ["certificate"], True, 1, "certificate", + {"passphrase": "password"}, {"private_key": "key"}, + {"private_key": "key", "public_certificate": True}): + with self.subTest(certificate=certificate): + with self.assertRaisesRegex(ValueError, "mtls_binding_certificate"): + ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": lambda: "LEG1_TOKEN", + "mtls_binding_certificate": certificate}) + + def test_fic_callable_renews_assertion_without_deprecation_warning(self): + assertion = Mock(side_effect=["LEG1_TOKEN", "RENEWED_LEG1_TOKEN"]) + captured = [] + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": assertion, + "mtls_binding_certificate": _MTLS_CERT_CRED}) + for scope in ("s1", "s2"): + result = app.acquire_token_for_client( + [scope], mtls_proof_of_possession=True, + post=self._capturing_post(captured)) + self.assertIn("access_token", result) + self.assertEqual( + ["LEG1_TOKEN", "RENEWED_LEG1_TOKEN"], + [request["data"]["client_assertion"] for request in captured]) + self.assertTrue(all( + request["data"]["client_assertion_type"] == _JWT_POP + and request["url"].startswith("https://mtlsauth.microsoft.com/") + for request in captured)) + def test_fic_leg2_app_rejects_non_client_credential_flows(self): # A FIC leg-2 app (client_assertion + mtls_binding_certificate) is only # valid via acquire_token_for_client over mTLS. Any other flow would put diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 1f68dd6b..73e6b119 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -1,5 +1,11 @@ """If the following ENV VAR were available, many end-to-end test cases would run. LAB_APP_CLIENT_CERT_PFX_PATH=... + +The FIC two-leg test additionally requires LAB_FIC_LEG1_CLIENT_ID, +LAB_FIC_LEG1_AUTHORITY, LAB_FIC_LEG2_CLIENT_ID, LAB_FIC_LEG2_AUTHORITY, and +LAB_FIC_EXCHANGE_SCOPE. Provision the leg-1 app for mTLS PoP at the exchange +audience, and a federated identity credential on the leg-2 app trusting the +leg-1 token's issuer, subject, and audience. Both legs use the lab certificate. """ try: from dotenv import load_dotenv # Use this only in local dev machine @@ -15,9 +21,9 @@ from urllib.parse import urlparse, parse_qs import sys try: - from unittest.mock import patch, ANY + from unittest.mock import patch, ANY, Mock except: - from mock import patch, ANY + from mock import patch, ANY, Mock import requests @@ -338,7 +344,55 @@ def test_access_token_should_be_obtained_for_a_supported_scope(self): self.assertIsNotNone(result.get("access_token")) -class PublicCloudScenariosTestCase(E2eTestCase): +class MtlsPopTestCase(E2eTestCase): + _MTLS_POP_SCOPE = ["https://graph.microsoft.com/.default"] + # Only the mTLS Graph host validates the certificate binding on the channel. + _MTLS_GRAPH_RESOURCE = "https://mtlstb.graph.microsoft.com/v1.0/applications?$top=1" + + @staticmethod + def _cnf_x5t_s256(access_token): + # Best-effort decode of an mtls_pop JWT's cnf.x5t#S256 binding claim. + try: + payload = json.loads(decode_part(access_token.split(".")[1])) + return payload.get("cnf", {}).get("x5t#S256") + except Exception: # Opaque/non-JWT token - skip the offline binding check + return None + + def _call_graph_with_mtls_pop_token(self, token, cert_credential): + """Prove an ``mtls_pop`` token is genuinely usable: present it to the + mTLS Graph endpoint over a TLS channel bound by the SAME certificate and + require HTTP 200. Acquiring the token is not enough - a resource only + honors it when the binding certificate is on the TLS handshake and the + auth scheme is ``mtls_pop``; a 401/403 means the binding was lost, i.e. a + real regression rather than a transient failure. + + Shared test infrastructure: the FIC two-leg e2e in PR #939 reuses this + as-is (please do not inline it), which is why it takes an explicit + ``cert_credential`` rather than reading ``self.app``'s certificate. + """ + from msal.application import _load_mtls_cert_material + from msal.mtls import _create_ssl_context, _make_mtls_adapter + material = _load_mtls_cert_material(cert_credential) + session = requests.Session() + session.mount("https://", _make_mtls_adapter(_create_ssl_context( + material["cert_pem"], material["private_key_pem"]))) + try: + resp = session.get( + self._MTLS_GRAPH_RESOURCE, + headers={"Authorization": "mtls_pop " + token}, + timeout=30) + finally: + session.close() + self.assertEqual( + 200, resp.status_code, + "mtls_pop token rejected by the resource (HTTP %s). A 401/403 means " + "the binding certificate was not presented on the TLS handshake, or " + "the 'mtls_pop' auth scheme was wrong. Body[:500]=%s" % ( + resp.status_code, (resp.text or "")[:500])) + return resp + + +class PublicCloudScenariosTestCase(MtlsPopTestCase): # Historically this class was driven by tests/config.json for semi-automated runs. # It now uses lab config + env vars so it can run automatically on any CI # (including Azure DevOps) as long as LAB_APP_CLIENT_CERT_PFX_PATH is set. @@ -448,17 +502,11 @@ def test_subject_name_issuer_authentication(self): # do NOT pass http_client=MinimalHttpClient() here. # ESTS gates mTLS PoP on the final resource audience (must be allow-listed, # e.g. MS Graph), not on the client app, so we target Microsoft Graph. - _MTLS_POP_SCOPE = ["https://graph.microsoft.com/.default"] # SN/I app + tenant that ESTS allow-lists for mTLS PoP. The generic # LAB_APP_CLIENT_ID + microsoft.onmicrosoft.com works for Bearer SN/I but is # rejected for mTLS PoP with AADSTS700025. Mirrors the MSAL .NET/Java e2e config. _SNI_ALLOWLISTED_CLIENT_ID = "163ffef9-a313-45b4-ab2f-c7e2f5e0e23e" _SNI_ALLOWLISTED_AUTHORITY = "https://login.microsoftonline.com/bea21ebe-8b64-4d06-9f6d-6a889b120a7c" - # The mTLS Graph host (NOT plain graph.microsoft.com): only this host performs - # the client-certificate TLS handshake that an mtls_pop token's binding is - # validated against. Presenting the token here, with the binding cert on the - # handshake, must return 200; a 401/403 means the binding was lost. - _MTLS_GRAPH_RESOURCE = "https://mtlstb.graph.microsoft.com/v1.0/applications?$top=1" def _assert_pop_cache_hit(self, app, scope): # A second mTLS-PoP call for the same scope must be served from cache. @@ -468,48 +516,6 @@ def _assert_pop_cache_hit(self, app, scope): self.assertIsNotNone(again) self.assertEqual(app._TOKEN_SOURCE_CACHE, again[app._TOKEN_SOURCE]) - @staticmethod - def _cnf_x5t_s256(access_token): - # Best-effort decode of an mtls_pop JWT's cnf.x5t#S256 binding claim. - try: - payload = json.loads(decode_part(access_token.split(".")[1])) - return payload.get("cnf", {}).get("x5t#S256") - except Exception: # Opaque/non-JWT token - skip the offline binding check - return None - - def _call_graph_with_mtls_pop_token(self, token, cert_credential): - """Prove an ``mtls_pop`` token is genuinely usable: present it to the - mTLS Graph endpoint over a TLS channel bound by the SAME certificate and - require HTTP 200. Acquiring the token is not enough - a resource only - honors it when the binding certificate is on the TLS handshake and the - auth scheme is ``mtls_pop``; a 401/403 means the binding was lost, i.e. a - real regression rather than a transient failure. - - Shared test infrastructure: the FIC two-leg e2e in PR #939 reuses this - as-is (please do not inline it), which is why it takes an explicit - ``cert_credential`` rather than reading ``self.app``'s certificate. - """ - from msal.application import _load_mtls_cert_material - from msal.mtls import _create_ssl_context, _make_mtls_adapter - material = _load_mtls_cert_material(cert_credential) - session = requests.Session() - session.mount("https://", _make_mtls_adapter(_create_ssl_context( - material["cert_pem"], material["private_key_pem"]))) - try: - resp = session.get( - self._MTLS_GRAPH_RESOURCE, - headers={"Authorization": "mtls_pop " + token}, - timeout=30) - finally: - session.close() - self.assertEqual( - 200, resp.status_code, - "mtls_pop token rejected by the resource (HTTP %s). A 401/403 means " - "the binding certificate was not presented on the TLS handshake, or " - "the 'mtls_pop' auth scheme was wrong. Body[:500]=%s" % ( - resp.status_code, (resp.text or "")[:500])) - return resp - # ── Credential x Output matrix (mirrors the MSAL .NET e2e naming) ────────── # Credential__Output_. Task 1 covers the two X509 cells # (this file); the Fic cells live in the FIC follow-up PR. snake_case per @@ -567,32 +573,45 @@ def test_credential_x509_output_bearer(self): self.assertIsNone(result.get("binding_certificate")) self.assertCacheWorksForApp(result, self._MTLS_POP_SCOPE) + +class FicTwoLegTestCase(MtlsPopTestCase): def test_fic_two_leg_over_mtls_pop(self): 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") - authority = "https://login.microsoftonline.com/microsoft.onmicrosoft.com" + required = ( + "LAB_APP_CLIENT_CERT_PFX_PATH", + "LAB_FIC_LEG1_CLIENT_ID", "LAB_FIC_LEG1_AUTHORITY", + "LAB_FIC_LEG2_CLIENT_ID", "LAB_FIC_LEG2_AUTHORITY", + "LAB_FIC_EXCHANGE_SCOPE", + ) + config = {name: clean_env(name) for name in required} + missing = [name for name, value in config.items() if not value] + if missing: + self.skipTest("FIC configuration missing: " + ", ".join(missing)) + cert = get_client_certificate() # Leg 1: SN/I cert -> cert-bound (mtls_pop) federated assertion for the # caller-supplied token-exchange audience. leg1_app = msal.ConfidentialClientApplication( - LAB_APP_CLIENT_ID, authority=authority, - client_credential=get_client_certificate()) + config["LAB_FIC_LEG1_CLIENT_ID"], + authority=config["LAB_FIC_LEG1_AUTHORITY"], + client_credential=cert) leg1 = leg1_app.acquire_token_for_client( - ["api://AzureADTokenExchange/.default"], mtls_proof_of_possession=True) - if "access_token" not in leg1: - self.skipTest("FIC leg 1 not pre-authorized for token exchange: %s" % leg1) + [config["LAB_FIC_EXCHANGE_SCOPE"]], mtls_proof_of_possession=True) + self.assertIn("access_token", leg1, "FIC leg 1 failed: %s" % leg1) self.assertEqual("mtls_pop", leg1.get("token_type"), "Leg 1 must be mtls_pop") leg1_thumbprint = leg1["binding_certificate"]["thumbprint_sha256"] # Leg 2: same cert binds the TLS handshake, leg-1 token carried as jwt-pop. leg2_app = msal.ConfidentialClientApplication( - LAB_APP_CLIENT_ID, authority=authority, + config["LAB_FIC_LEG2_CLIENT_ID"], + authority=config["LAB_FIC_LEG2_AUTHORITY"], client_credential={ - "client_assertion": leg1["access_token"], - "mtls_binding_certificate": get_client_certificate()}) + "client_assertion": lambda: leg1["access_token"], + "mtls_binding_certificate": cert}) # Leg 2 -> Bearer over mTLS (implicit Bearer-over-mTLS, no flag) leg2_bearer = leg2_app.acquire_token_for_client(self._MTLS_POP_SCOPE) self.assertIn("access_token", leg2_bearer, "Leg 2 (Bearer) failed: %s" % leg2_bearer) + self.assertEqual("Bearer", leg2_bearer.get("token_type")) + self.assertNotIn("binding_certificate", leg2_bearer) # Leg 2 -> mTLS PoP final token leg2_pop = leg2_app.acquire_token_for_client( self._MTLS_POP_SCOPE, mtls_proof_of_possession=True) @@ -603,6 +622,85 @@ def test_fic_two_leg_over_mtls_pop(self): if cnf: self.assertEqual(leg1_thumbprint, cnf, "Final token cnf must be bound to the leg-1 cert thumbprint") + self._call_graph_with_mtls_pop_token(leg2_pop["access_token"], cert) + + +class FicScenarioTestCase(unittest.TestCase): + """Exercise the FIC integration test's failure gates without lab access.""" + + def setUp(self): + self.config = { + "LAB_APP_CLIENT_CERT_PFX_PATH": "configured.pfx", + "LAB_FIC_LEG1_CLIENT_ID": "leg1-client", + "LAB_FIC_LEG1_AUTHORITY": "https://login.microsoftonline.com/leg1-tenant", + "LAB_FIC_LEG2_CLIENT_ID": "leg2-client", + "LAB_FIC_LEG2_AUTHORITY": "https://login.microsoftonline.com/leg2-tenant", + "LAB_FIC_EXCHANGE_SCOPE": "api://exchange/.default", + } + self.leg1 = Mock() + self.leg1.acquire_token_for_client.return_value = { + "access_token": "leg1-token", "token_type": "mtls_pop", + "binding_certificate": {"thumbprint_sha256": "cert-thumbprint"}, + } + self.leg2 = Mock() + self.leg2.acquire_token_for_client.side_effect = [ + {"access_token": "bearer-token", "token_type": "Bearer"}, + {"access_token": "opaque-pop-token", "token_type": "mtls_pop"}, + ] + self.scenario = FicTwoLegTestCase("test_fic_two_leg_over_mtls_pop") + self.cert = {"private_key_pfx_path": "configured.pfx"} + self.factory = Mock(side_effect=[self.leg1, self.leg2]) + patchers = ( + patch.dict(os.environ, self.config, clear=True), + patch("tests.lab_config.get_client_certificate", return_value=self.cert), + patch("msal.ConfidentialClientApplication", self.factory), + patch.object(self.scenario, "_call_graph_with_mtls_pop_token"), + ) + for patcher in patchers: + patcher.start() + self.addCleanup(patcher.stop) + + def test_missing_configuration_skips_before_acquisition(self): + for name in self.config: + with self.subTest(name=name), patch.dict(os.environ, {name: ""}): + with self.assertRaisesRegex(unittest.SkipTest, name): + self.scenario.test_fic_two_leg_over_mtls_pop() + self.factory.assert_not_called() + + def test_configured_leg1_errors_fail_instead_of_skipping(self): + self.leg1.acquire_token_for_client.return_value = { + "error": "token_type_mismatch"} + with self.assertRaisesRegex(AssertionError, "token_type_mismatch"): + self.scenario.test_fic_two_leg_over_mtls_pop() + self.leg2.acquire_token_for_client.assert_not_called() + + def test_explicit_configuration_and_opaque_token_resource_validation(self): + self.scenario.test_fic_two_leg_over_mtls_pop() + calls = self.factory.call_args_list + self.assertEqual("leg1-client", calls[0].args[0]) + self.assertEqual(self.config["LAB_FIC_LEG1_AUTHORITY"], + calls[0].kwargs["authority"]) + self.assertEqual("leg2-client", calls[1].args[0]) + self.assertEqual(self.config["LAB_FIC_LEG2_AUTHORITY"], + calls[1].kwargs["authority"]) + credential = calls[1].kwargs["client_credential"] + self.assertIs(self.cert, credential["mtls_binding_certificate"]) + self.assertEqual("leg1-token", credential["client_assertion"]()) + self.leg1.acquire_token_for_client.assert_called_once_with( + [self.config["LAB_FIC_EXCHANGE_SCOPE"]], mtls_proof_of_possession=True) + self.scenario._call_graph_with_mtls_pop_token.assert_called_once_with( + "opaque-pop-token", self.cert) + + def test_bearer_contract_is_checked(self): + for response in ( + {"access_token": "token", "token_type": "mtls_pop"}, + {"access_token": "token", "token_type": "Bearer", + "binding_certificate": {}}): + with self.subTest(response=response): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.side_effect = [response] + with self.assertRaises(AssertionError): + self.scenario.test_fic_two_leg_over_mtls_pop() class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once From d2b9ab22967afa16a31c3b8d9fa5fb228c64a611 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 10:26:45 -0400 Subject: [PATCH 5/7] Align FIC Bearer and PoP result binding Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/index.rst | 6 ++-- msal/application.py | 8 ++--- sample/confidential_client_mtls_pop_sample.py | 7 +++-- tests/test_application.py | 3 +- tests/test_e2e.py | 31 +++++++++---------- 5 files changed, 29 insertions(+), 26 deletions(-) diff --git a/docs/index.rst b/docs/index.rst index a94dd2a2..8a2d1ad5 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -241,8 +241,10 @@ Notes and requirements: leg-1 token's issuer, subject, and audience. See ``fic_two_leg_over_mtls_pop()`` in ``sample/confidential_client_mtls_pop_sample.py`` for the two-leg setup. - Both final ``mtls_pop`` and Bearer-over-mTLS results include an - ``MtlsBindingContext`` for the downstream resource connection. + A final ``mtls_pop`` result includes an ``MtlsBindingContext`` for the + downstream resource connection. If leg 2 does not request PoP, its bound + assertion still travels as ``jwt-pop`` over mTLS, but the final Bearer token + is not certificate-bound and does not include ``binding_certificate``. .. autoclass:: msal.MtlsBindingContext :members: diff --git a/msal/application.py b/msal/application.py index cec24559..a79307b4 100644 --- a/msal/application.py +++ b/msal/application.py @@ -3167,8 +3167,7 @@ def acquire_token_for_client( kwargs["data"] = data result = _clean_up(self._acquire_token_silent_with_error( scopes, None, claims_challenge=claims_challenge, **kwargs)) - if ((mtls_proof_of_possession or self._mtls_is_fic_leg2) - and result and "access_token" in result): + if (mtls_proof_of_possession and result and "access_token" in result): # Fail closed on a token_type downgrade: if ESTS returned a # non-cert-bound token (e.g. a Bearer downgrade) for our mtls_pop # request, do not surface it as bound. The cache side is handled in @@ -3183,8 +3182,9 @@ def acquire_token_for_client( "of 'mtls_pop'; the access token is not " "certificate-bound.".format(result.get("token_type"))), } - # Attach process-local TLS capability after cache cleanup. The - # private key is never added to the token result or token cache. + # Attach process-local TLS capability only when the final token is + # certificate-bound. A FIC assertion can travel as jwt-pop over + # mTLS while the final token remains an ordinary Bearer token. result["binding_certificate"] = mtls.MtlsBindingContext( mtls_cert["cert_pem"], mtls_cert["private_key_pem"], diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index 9409f841..00050a0f 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -131,7 +131,8 @@ def fic_two_leg_over_mtls_pop(): the token-exchange audience. Leg 2: a second confidential client presents that token as a ``jwt-pop`` client assertion, over an mTLS connection bound by the SAME cert, to - obtain the final token (Bearer or mtls_pop). + obtain the final token. Requesting PoP produces ``mtls_pop`` and a + binding context; omitting PoP produces Bearer without one. """ cert = _build_cert_credential() @@ -167,7 +168,9 @@ def get_leg1_assertion(): azure_region=os.getenv("AZURE_REGION"), ) - # Final token as mTLS PoP (drop mtls_proof_of_possession for Bearer-over-mTLS): + # Final token as mTLS PoP. Omit mtls_proof_of_possession for a final Bearer + # token; its assertion still travels as jwt-pop over mTLS, but the result + # does not contain binding_certificate. leg2 = leg2_app.acquire_token_for_client( os.environ["SCOPE"].split(), mtls_proof_of_possession=True) _print_result_summary(leg2, "FIC leg 2 result:") diff --git a/tests/test_application.py b/tests/test_application.py index 372ff0a4..431712e8 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -2546,8 +2546,7 @@ def test_fic_leg2_bearer_still_travels_over_mtls(self): self.assertNotIn("token_type", req["data"]) self.assertEqual("4|730,2|", req["headers"].get(CLIENT_CURRENT_TELEMETRY)) self.assertEqual("Bearer", result.get("token_type")) - self.assertIsInstance( - result.get("binding_certificate"), msal.MtlsBindingContext) + self.assertNotIn("binding_certificate", result) def test_fic_leg2_rejects_post_override(self): app = ConfidentialClientApplication( diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 9e6219c7..b17571fa 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -358,26 +358,25 @@ def _cnf_x5t_s256(access_token): except Exception: # Opaque/non-JWT token - skip the offline binding check return None - def _call_graph_with_mtls_pop_token( - self, token, binding_certificate, token_type="mtls_pop"): - """Prove an mTLS token is genuinely usable: present it to the + def _call_graph_with_mtls_pop_token(self, token, binding_certificate): + """Prove an ``mtls_pop`` token is genuinely usable: present it to the mTLS Graph endpoint over a TLS channel bound by the SAME certificate and require HTTP 200. Acquiring the token is not enough - a resource only honors it when the binding certificate is on the TLS handshake and the - auth scheme matches the token type; a 401/403 means the binding was - lost, i.e. a real regression rather than a transient failure. + auth scheme is ``mtls_pop``; a 401/403 means the binding was lost, i.e. + a real regression rather than a transient failure. """ with binding_certificate.create_requests_session() as session: resp = session.get( self._MTLS_GRAPH_RESOURCE, - headers={"Authorization": token_type + " " + token}, + headers={"Authorization": "mtls_pop " + token}, timeout=30) self.assertEqual( 200, resp.status_code, - "%s token rejected by the resource (HTTP %s). A 401/403 means " + "mtls_pop token rejected by the resource (HTTP %s). A 401/403 means " "the binding certificate was not presented on the TLS handshake, or " "the auth scheme was wrong. Body[:500]=%s" % ( - token_type, resp.status_code, (resp.text or "")[:500])) + resp.status_code, (resp.text or "")[:500])) return resp @@ -565,6 +564,8 @@ def test_credential_x509_output_bearer(self): class FicTwoLegTestCase(MtlsPopTestCase): + _FIC_BEARER_SCOPE = ["https://storage.azure.com/.default"] + def test_fic_two_leg_over_mtls_pop(self): from tests.lab_config import get_client_certificate required = ( @@ -598,13 +599,10 @@ def test_fic_two_leg_over_mtls_pop(self): "client_assertion": lambda: leg1["access_token"], "mtls_binding_certificate": cert}) # Leg 2 -> Bearer over mTLS (implicit Bearer-over-mTLS, no flag) - leg2_bearer = leg2_app.acquire_token_for_client(self._MTLS_POP_SCOPE) + leg2_bearer = leg2_app.acquire_token_for_client(self._FIC_BEARER_SCOPE) self.assertIn("access_token", leg2_bearer, "Leg 2 (Bearer) failed: %s" % leg2_bearer) self.assertEqual("Bearer", leg2_bearer.get("token_type")) - bearer_binding = leg2_bearer.get("binding_certificate") - self.assertIsInstance(bearer_binding, msal.MtlsBindingContext) - self._call_graph_with_mtls_pop_token( - leg2_bearer["access_token"], bearer_binding, token_type="Bearer") + self.assertNotIn("binding_certificate", leg2_bearer) # Leg 2 -> mTLS PoP final token leg2_pop = leg2_app.acquire_token_for_client( self._MTLS_POP_SCOPE, mtls_proof_of_possession=True) @@ -617,6 +615,9 @@ def test_fic_two_leg_over_mtls_pop(self): "Final token cnf must be bound to the leg-1 cert thumbprint") pop_binding = leg2_pop.get("binding_certificate") self.assertIsInstance(pop_binding, msal.MtlsBindingContext) + self.assertEqual( + leg1_thumbprint, pop_binding.get("thumbprint_sha256"), + "Final binding context must carry the leg-1 certificate") self._call_graph_with_mtls_pop_token( leg2_pop["access_token"], pop_binding) @@ -638,14 +639,13 @@ def setUp(self): "access_token": "leg1-token", "token_type": "mtls_pop", "binding_certificate": {"thumbprint_sha256": "cert-thumbprint"}, } - self.bearer_binding = Mock(spec=msal.MtlsBindingContext) self.pop_binding = Mock(spec=msal.MtlsBindingContext) + self.pop_binding.get.return_value = "cert-thumbprint" self.leg2 = Mock() self.leg2.acquire_token_for_client.side_effect = [ { "access_token": "bearer-token", "token_type": "Bearer", - "binding_certificate": self.bearer_binding, }, { "access_token": "opaque-pop-token", @@ -696,7 +696,6 @@ def test_explicit_configuration_and_opaque_token_resource_validation(self): [self.config["LAB_FIC_EXCHANGE_SCOPE"]], mtls_proof_of_possession=True) self.assertEqual( [ - call("bearer-token", self.bearer_binding, token_type="Bearer"), call("opaque-pop-token", self.pop_binding), ], self.scenario._call_graph_with_mtls_pop_token.call_args_list) From b70842b4c9469d9e40e95ee8f51efbba9b9d6906 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 14:07:04 -0400 Subject: [PATCH 6/7] Address FIC mTLS review feedback Validate final token types before returning or caching responses, reject unsupported FIC outputs, validate credentials before client construction, and guard the sample resource call. Cover cache isolation and same-scope assertion renewal. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 39 ++++-- sample/confidential_client_mtls_pop_sample.py | 2 +- tests/test_application.py | 119 +++++++++++++++++- 3 files changed, 143 insertions(+), 17 deletions(-) diff --git a/msal/application.py b/msal/application.py index 1c8e042b..b739417c 100644 --- a/msal/application.py +++ b/msal/application.py @@ -386,15 +386,14 @@ def obtain_token_by_username_password(self, username, password, **kwargs): username, password, headers=headers, **kwargs) -def _validate_mtls_pop_response(response): +def _validate_mtls_response(response, expected_token_type="mtls_pop"): if (response and "access_token" in response - and (response.get("token_type") or "").lower() != "mtls_pop"): + and (response.get("token_type") or "").lower() != expected_token_type): return { "error": "token_type_mismatch", "error_description": ( "The identity provider returned token_type {!r} instead " - "of 'mtls_pop'; the access token is not " - "certificate-bound.".format(response.get("token_type"))), + "of {!r}.".format(response.get("token_type"), expected_token_type)), } return response @@ -414,6 +413,7 @@ class _MtlsClient(_ClientWithCcsRoutingInfo): def obtain_token_for_client(self, scope=None, **kwargs): data = kwargs.pop("data", {}) key_id = data.pop("key_id", None) # Keep key_id OFF the wire + expected_token_type = (data.get("token_type") or "Bearer").lower() if key_id: outer = kwargs.pop("on_obtaining_tokens", None) @@ -429,11 +429,18 @@ def _reinject_key_id(event): (outer or self.on_obtaining_tokens)(event) kwargs["on_obtaining_tokens"] = _reinject_key_id + else: + outer = kwargs.pop("on_obtaining_tokens", None) + + def _cache_bearer_response(event): + response = event.get("response") or {} + if (response.get("token_type") or "").lower() == "bearer": + (outer or self.on_obtaining_tokens)(event) + + kwargs["on_obtaining_tokens"] = _cache_bearer_response response = super(_MtlsClient, self).obtain_token_for_client( scope=scope, data=data, **kwargs) - return (_validate_mtls_pop_response(response) - if (data.get("token_type") or "").lower() == "mtls_pop" - else response) + return _validate_mtls_response(response, expected_token_type) def _msal_extension_check(): @@ -1006,9 +1013,6 @@ def get_leg1_assertion(): self._region_detected = None self._region_resolved = False self._region_to_use = None - self.client, self._regional_client = self._build_client( - client_credential, self.authority) - # mTLS Proof-of-Possession state. The certificate material and the # dedicated mTLS client(s) are parsed/built lazily on the first # mtls_proof_of_possession request (see _get_mtls_pop_cert / @@ -1049,6 +1053,9 @@ def get_leg1_assertion(): self._mtls_cert_credential = None self._mtls_is_fic_leg2 = False + self.client, self._regional_client = self._build_client( + client_credential, self.authority) + # 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)): @@ -3085,6 +3092,9 @@ def acquire_token_for_client( ``/common`` or ``/organizations``); and MSAL's built-in HTTP transport must be in use (a custom ``http_client`` cannot perform the mTLS handshake). Any of these unmet raises ``ValueError``. + FIC exchanges only support final ``Bearer`` or ``mtls_pop`` tokens; + requesting other token types, including SSH certificates, raises + ``ValueError``. A ``post`` transport override is also rejected. If the provider returns another token type, the result is ``token_type_mismatch`` and the rejected response is not cached. @@ -3137,6 +3147,13 @@ def acquire_token_for_client( raise ValueError( "Request mTLS PoP using mtls_proof_of_possession=True, " "not data['token_type'].") + if (self._mtls_is_fic_leg2 + and (requested_token_type or "").lower() not in ( + "", "bearer", "mtls_pop")): + raise ValueError( + "FIC leg-2 exchanges only support Bearer or mtls_pop final " + "tokens; token_type {!r} is not supported.".format( + requested_token_type)) if fmi_path is not None: if not isinstance(fmi_path, str): raise ValueError( @@ -3202,7 +3219,7 @@ def acquire_token_for_client( # _MtlsClient, which does not cache downgraded responses. # Mirrors the fail-closed behavior of MSAL .NET and MSAL Go. if (result.get("token_type") or "").lower() != "mtls_pop": - return _validate_mtls_pop_response(result) + return _validate_mtls_response(result) # Attach process-local TLS capability only when the final token is # certificate-bound. A FIC assertion can travel as jwt-pop over # mTLS while the final token remains an ordinary Bearer token. diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index a1c048dd..88f0c370 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -175,7 +175,7 @@ def get_leg1_assertion(): os.environ["SCOPE"].split(), mtls_proof_of_possession=True) _print_result_summary(leg2, "FIC leg 2 result:") resource_url = os.getenv("RESOURCE_URL") - if resource_url and "access_token" in leg2: + if resource_url and "access_token" in leg2 and "binding_certificate" in leg2: with leg2["binding_certificate"].create_requests_session() as session: response = session.get( resource_url, diff --git a/tests/test_application.py b/tests/test_application.py index 91f76270..92190053 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -5,6 +5,7 @@ import logging import os import pickle +import runpy import sys import time import warnings @@ -2662,11 +2663,103 @@ def test_fic_leg2_rejects_post_override(self): app.acquire_token_for_client( ["s1"], post=self._capturing_post([])) + def test_fic_leg2_bearer_rejects_unexpected_response_type(self): + for token_type in ("mtls_pop", "ssh-cert", None): + with self.subTest(token_type=token_type): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": lambda: "LEG1_TOKEN", + "mtls_binding_certificate": _MTLS_CERT_CRED}) + with patch( + "msal.mtls._MtlsHttpClient.post", + side_effect=self._capturing_post( + [], token_type=token_type)): + result = app.acquire_token_for_client(["s1"]) + self.assertEqual("token_type_mismatch", result.get("error")) + self.assertNotIn("access_token", result) + self.assertEqual([], list(app.token_cache.search( + msal.TokenCache.CredentialType.ACCESS_TOKEN, + for_removal=True))) + + def test_fic_leg2_rejects_unsupported_final_token_types_before_cache_lookup(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": lambda: "LEG1_TOKEN", + "mtls_binding_certificate": _MTLS_CERT_CRED}) + for token_type in ("ssh-cert", "pop", "unknown"): + for request_pop in (False, True): + with self.subTest(token_type=token_type, request_pop=request_pop): + with patch.object( + app, "_acquire_token_silent_with_error") as lookup: + with self.assertRaisesRegex( + ValueError, "only support Bearer or mtls_pop"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=request_pop, + data={ + "token_type": token_type, + "key_id": "ssh-key", "req_cnf": "{}"}) + lookup.assert_not_called() + + def test_fic_leg2_bearer_and_pop_cache_entries_are_isolated(self): + for order in ((False, True), (True, False)): + with self.subTest(order=order): + assertion = Mock(return_value="LEG1_TOKEN") + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": assertion, + "mtls_binding_certificate": _MTLS_CERT_CRED}) + for request_pop in order: + captured = [] + token_type = "mtls_pop" if request_pop else "Bearer" + with patch( + "msal.mtls._MtlsHttpClient.post", + side_effect=self._capturing_post( + captured, token_type=token_type)): + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=request_pop) + self.assertEqual(1, len(captured)) + self.assertEqual(token_type, result["token_type"]) + with patch("msal.mtls._MtlsHttpClient.post") as post: + for request_pop in order: + cached = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=request_pop) + self.assertEqual( + app._TOKEN_SOURCE_CACHE, cached[app._TOKEN_SOURCE]) + self.assertEqual( + "mtls_pop" if request_pop else "Bearer", + cached["token_type"]) + self.assertEqual( + request_pop, "binding_certificate" in cached) + post.assert_not_called() + self.assertEqual(2, assertion.call_count) + + def test_fic_sample_skips_resource_call_for_bearer_result(self): + sample = runpy.run_path(os.path.join( + os.path.dirname(__file__), os.pardir, "sample", + "confidential_client_mtls_pop_sample.py")) + result = {"access_token": "token", "token_type": "Bearer"} + app = Mock() + app.acquire_token_for_client.return_value = result + with patch.dict(os.environ, { + "CLIENT_ID": "leg1", "AUTHORITY": self._AUTHORITY, + "LEG2_CLIENT_ID": "leg2", "LEG2_AUTHORITY": self._AUTHORITY, + "PFX_PATH": "unused.pfx", "SCOPE": "s1", + "RESOURCE_URL": "https://resource.invalid"}), patch( + "msal.ConfidentialClientApplication", + return_value=app), patch("builtins.print"), patch( + "msal.MtlsBindingContext.create_requests_session") as session: + self.assertIs(result, sample["fic_two_leg_over_mtls_pop"]()) + session.assert_not_called() + def test_binding_cert_without_client_assertion_raises(self): # FIC leg 2 needs the binding cert AND the leg-1 assertion. A binding # cert with no client_assertion must fail fast at construction with a # clear ValueError, not a later KeyError when the request is built. - with self.assertRaises(ValueError): + with self.assertRaisesRegex( + ValueError, "also requires a client_assertion"): ConfidentialClientApplication( "cid", authority=self._AUTHORITY, client_credential={"mtls_binding_certificate": _MTLS_CERT_CRED}) @@ -2697,10 +2790,26 @@ def test_fic_callable_renews_assertion_without_deprecation_warning(self): with patch( "msal.mtls._MtlsHttpClient.post", side_effect=self._capturing_post(captured)): - for scope in ("s1", "s2"): - result = app.acquire_token_for_client( - [scope], mtls_proof_of_possession=True) - self.assertIn("access_token", result) + first = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertIn("access_token", first) + cached = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertEqual(app._TOKEN_SOURCE_CACHE, cached[app._TOKEN_SOURCE]) + self.assertIsInstance( + cached.get("binding_certificate"), msal.MtlsBindingContext) + self.assertEqual(1, assertion.call_count) + self.assertEqual(1, len(captured)) + for token in list(app.token_cache.search( + msal.TokenCache.CredentialType.ACCESS_TOKEN, + for_removal=True)): + app.token_cache.modify( + msal.TokenCache.CredentialType.ACCESS_TOKEN, token, + {"expires_on": "0"}) + renewed = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertIn("access_token", renewed) + self.assertEqual(app._TOKEN_SOURCE_IDP, renewed[app._TOKEN_SOURCE]) self.assertEqual( ["LEG1_TOKEN", "RENEWED_LEG1_TOKEN"], [request["data"]["client_assertion"] for request in captured]) From 84ff3e860335cce8b185d8a5c7b9dbc8ccda7c47 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 15:12:00 -0400 Subject: [PATCH 7/7] Align FIC E2E cells with shared SDK lab setup Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/README.md | 24 ++++++ tests/test_e2e.py | 210 ++++++++++++++++++++++++++++++---------------- 2 files changed, 164 insertions(+), 70 deletions(-) diff --git a/tests/README.md b/tests/README.md index 0b044a60..3d1c9eb0 100644 --- a/tests/README.md +++ b/tests/README.md @@ -34,6 +34,30 @@ Notes: - `tests/test_e2e.py` loads `.env` automatically when `python-dotenv` is installed. - For certificate auth, `LAB_APP_CLIENT_CERT_PFX_PATH` should be an absolute path. +### FIC over mTLS lab parity + +The FIC tests reuse the existing MSAL.NET/Go SN/I lab configuration for both +legs: app `163ffef9-a313-45b4-ab2f-c7e2f5e0e23e`, authority +`https://login.microsoftonline.com/bea21ebe-8b64-4d06-9f6d-6a889b120a7c`, +and the shared LabAuth certificate supplied through +`LAB_APP_CLIENT_CERT_PFX_PATH`. No Python-only `LAB_FIC_*` configuration or +new app provisioning is needed. + +Leg 1 requests `api://AzureADTokenExchange/.default` as an mTLS PoP assertion. +The PoP-output cell uses global endpoints for both legs, requests +`https://graph.microsoft.com/.default`, and calls the mTLS Graph resource with +the final token's `binding_certificate`. The Bearer-output cell uses `westus3` +for both legs and requests `https://storage.azure.com/.default`; its final +token must not contain `binding_certificate` or trigger a downstream mTLS call. +Missing or invalid certificates and real token-acquisition errors fail the +FIC tests rather than skipping or accepting an error as success. + +Run these two live cells with: + +```powershell +python -m pytest -q tests\test_e2e.py::FicTwoLegTestCase +``` + ## 4) Run unit/integration tests Run all non-E2E tests quickly: diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 7f49a65f..c980eb90 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -1,11 +1,9 @@ """If the following ENV VAR were available, many end-to-end test cases would run. LAB_APP_CLIENT_CERT_PFX_PATH=... -The FIC two-leg test additionally requires LAB_FIC_LEG1_CLIENT_ID, -LAB_FIC_LEG1_AUTHORITY, LAB_FIC_LEG2_CLIENT_ID, LAB_FIC_LEG2_AUTHORITY, and -LAB_FIC_EXCHANGE_SCOPE. Provision the leg-1 app for mTLS PoP at the exchange -audience, and a federated identity credential on the leg-2 app trusting the -leg-1 token's issuer, subject, and audience. Both legs use the lab certificate. +The FIC two-leg tests use the same preconfigured SN/I lab app, tenant, and +LabAuth certificate as MSAL.NET and Go. No LAB_FIC_* variables are required. +Missing certificates and token-acquisition errors fail these tests. """ try: from dotenv import load_dotenv # Use this only in local dev machine @@ -565,48 +563,48 @@ def test_credential_x509_output_bearer(self): class FicTwoLegTestCase(MtlsPopTestCase): + __unittest_skip__ = False # Required cells must also run on tagged releases. + _FIC_EXCHANGE_SCOPE = ["api://AzureADTokenExchange/.default"] _FIC_BEARER_SCOPE = ["https://storage.azure.com/.default"] - def test_fic_two_leg_over_mtls_pop(self): + def _acquire_fic_token( + self, scope, azure_region, mtls_proof_of_possession=False): from tests.lab_config import get_client_certificate - required = ( - "LAB_APP_CLIENT_CERT_PFX_PATH", - "LAB_FIC_LEG1_CLIENT_ID", "LAB_FIC_LEG1_AUTHORITY", - "LAB_FIC_LEG2_CLIENT_ID", "LAB_FIC_LEG2_AUTHORITY", - "LAB_FIC_EXCHANGE_SCOPE", - ) - config = {name: clean_env(name) for name in required} - missing = [name for name, value in config.items() if not value] - if missing: - self.skipTest("FIC configuration missing: " + ", ".join(missing)) cert = get_client_certificate() - # Leg 1: SN/I cert -> cert-bound (mtls_pop) federated assertion for the - # caller-supplied token-exchange audience. leg1_app = msal.ConfidentialClientApplication( - config["LAB_FIC_LEG1_CLIENT_ID"], - authority=config["LAB_FIC_LEG1_AUTHORITY"], - client_credential=cert) + PublicCloudScenariosTestCase._SNI_ALLOWLISTED_CLIENT_ID, + authority=PublicCloudScenariosTestCase._SNI_ALLOWLISTED_AUTHORITY, + client_credential=cert, + azure_region=azure_region) leg1 = leg1_app.acquire_token_for_client( - [config["LAB_FIC_EXCHANGE_SCOPE"]], mtls_proof_of_possession=True) + self._FIC_EXCHANGE_SCOPE, mtls_proof_of_possession=True) self.assertIn("access_token", leg1, "FIC leg 1 failed: %s" % leg1) self.assertEqual("mtls_pop", leg1.get("token_type"), "Leg 1 must be mtls_pop") leg1_thumbprint = leg1["binding_certificate"]["thumbprint_sha256"] # Leg 2: same cert binds the TLS handshake, leg-1 token carried as jwt-pop. leg2_app = msal.ConfidentialClientApplication( - config["LAB_FIC_LEG2_CLIENT_ID"], - authority=config["LAB_FIC_LEG2_AUTHORITY"], + PublicCloudScenariosTestCase._SNI_ALLOWLISTED_CLIENT_ID, + authority=PublicCloudScenariosTestCase._SNI_ALLOWLISTED_AUTHORITY, client_credential={ "client_assertion": lambda: leg1["access_token"], - "mtls_binding_certificate": cert}) - # Leg 2 -> Bearer over mTLS (implicit Bearer-over-mTLS, no flag) - leg2_bearer = leg2_app.acquire_token_for_client(self._FIC_BEARER_SCOPE) + "mtls_binding_certificate": cert}, + azure_region=azure_region) + result = leg2_app.acquire_token_for_client( + scope, mtls_proof_of_possession=mtls_proof_of_possession) + return leg1_thumbprint, result + + def test_credential_fic_output_bearer(self): + _, leg2_bearer = self._acquire_fic_token( + self._FIC_BEARER_SCOPE, azure_region="westus3") self.assertIn("access_token", leg2_bearer, "Leg 2 (Bearer) failed: %s" % leg2_bearer) self.assertEqual("Bearer", leg2_bearer.get("token_type")) self.assertNotIn("binding_certificate", leg2_bearer) - # Leg 2 -> mTLS PoP final token - leg2_pop = leg2_app.acquire_token_for_client( - self._MTLS_POP_SCOPE, mtls_proof_of_possession=True) + + def test_credential_fic_output_pop(self): + leg1_thumbprint, leg2_pop = self._acquire_fic_token( + self._MTLS_POP_SCOPE, azure_region=False, + mtls_proof_of_possession=True) self.assertIn("access_token", leg2_pop, "Leg 2 (PoP) failed: %s" % leg2_pop) self.assertEqual("mtls_pop", leg2_pop.get("token_type")) # The final PoP token must be bound to the leg-1 certificate thumbprint @@ -629,11 +627,6 @@ class FicScenarioTestCase(unittest.TestCase): def setUp(self): self.config = { "LAB_APP_CLIENT_CERT_PFX_PATH": "configured.pfx", - "LAB_FIC_LEG1_CLIENT_ID": "leg1-client", - "LAB_FIC_LEG1_AUTHORITY": "https://login.microsoftonline.com/leg1-tenant", - "LAB_FIC_LEG2_CLIENT_ID": "leg2-client", - "LAB_FIC_LEG2_AUTHORITY": "https://login.microsoftonline.com/leg2-tenant", - "LAB_FIC_EXCHANGE_SCOPE": "api://exchange/.default", } self.leg1 = Mock() self.leg1.acquire_token_for_client.return_value = { @@ -643,23 +636,17 @@ def setUp(self): self.pop_binding = Mock(spec=msal.MtlsBindingContext) self.pop_binding.get.return_value = "cert-thumbprint" self.leg2 = Mock() - self.leg2.acquire_token_for_client.side_effect = [ - { - "access_token": "bearer-token", - "token_type": "Bearer", - }, - { - "access_token": "opaque-pop-token", - "token_type": "mtls_pop", - "binding_certificate": self.pop_binding, - }, - ] - self.scenario = FicTwoLegTestCase("test_fic_two_leg_over_mtls_pop") - self.cert = {"private_key_pfx_path": "configured.pfx"} + self.leg2.acquire_token_for_client.return_value = { + "access_token": "opaque-pop-token", + "token_type": "mtls_pop", + "binding_certificate": self.pop_binding, + } + self.scenario = FicTwoLegTestCase("test_credential_fic_output_pop") + self.cert = { + "private_key_pfx_path": "configured.pfx", "public_certificate": True} self.factory = Mock(side_effect=[self.leg1, self.leg2]) patchers = ( patch.dict(os.environ, self.config, clear=True), - patch("tests.lab_config.get_client_certificate", return_value=self.cert), patch("msal.ConfidentialClientApplication", self.factory), patch.object(self.scenario, "_call_graph_with_mtls_pop_token"), ) @@ -667,40 +654,112 @@ def setUp(self): patcher.start() self.addCleanup(patcher.stop) - def test_missing_configuration_skips_before_acquisition(self): - for name in self.config: - with self.subTest(name=name), patch.dict(os.environ, {name: ""}): - with self.assertRaisesRegex(unittest.SkipTest, name): - self.scenario.test_fic_two_leg_over_mtls_pop() + def test_missing_certificate_fails_before_acquisition(self): + for method in ( + self.scenario.test_credential_fic_output_pop, + self.scenario.test_credential_fic_output_bearer): + for value in ("", "$(LAB_APP_CLIENT_CERT_PFX_PATH)"): + with self.subTest(method=method.__name__, value=value): + with patch.dict( + os.environ, {"LAB_APP_CLIENT_CERT_PFX_PATH": value}): + with self.assertRaisesRegex( + EnvironmentError, "LAB_APP_CLIENT_CERT_PFX_PATH"): + method() + self.factory.assert_not_called() + + def test_missing_certificate_is_a_runner_error_even_on_tagged_releases(self): + with patch.object(E2eTestCase, "__unittest_skip__", True, create=True): + with patch.dict(os.environ, {}, clear=True): + for name in ( + "test_credential_fic_output_pop", + "test_credential_fic_output_bearer"): + with self.subTest(name=name): + result = unittest.TestResult() + FicTwoLegTestCase(name).run(result) + self.assertEqual(1, result.testsRun) + self.assertEqual([], result.skipped) + self.assertEqual(1, len(result.errors)) + self.assertIn( + "LAB_APP_CLIENT_CERT_PFX_PATH", result.errors[0][1]) self.factory.assert_not_called() def test_configured_leg1_errors_fail_instead_of_skipping(self): - self.leg1.acquire_token_for_client.return_value = { - "error": "token_type_mismatch"} - with self.assertRaisesRegex(AssertionError, "token_type_mismatch"): - self.scenario.test_fic_two_leg_over_mtls_pop() + for method in ( + self.scenario.test_credential_fic_output_pop, + self.scenario.test_credential_fic_output_bearer): + for error in ("token_type_mismatch", "invalid_client", "invalid_grant"): + with self.subTest(method=method.__name__, error=error): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg1.acquire_token_for_client.return_value = { + "error": error} + with self.assertRaisesRegex(AssertionError, error): + method() self.leg2.acquire_token_for_client.assert_not_called() - def test_explicit_configuration_and_opaque_token_resource_validation(self): - self.scenario.test_fic_two_leg_over_mtls_pop() + def test_shared_lab_defaults_and_opaque_token_resource_validation(self): + self.scenario.test_credential_fic_output_pop() calls = self.factory.call_args_list - self.assertEqual("leg1-client", calls[0].args[0]) - self.assertEqual(self.config["LAB_FIC_LEG1_AUTHORITY"], - calls[0].kwargs["authority"]) - self.assertEqual("leg2-client", calls[1].args[0]) - self.assertEqual(self.config["LAB_FIC_LEG2_AUTHORITY"], - calls[1].kwargs["authority"]) + self.assertEqual(2, len(calls)) + for app_call in calls: + self.assertEqual( + "163ffef9-a313-45b4-ab2f-c7e2f5e0e23e", app_call.args[0]) + self.assertEqual( + "https://login.microsoftonline.com/bea21ebe-8b64-4d06-9f6d-6a889b120a7c", + app_call.kwargs["authority"]) + self.assertIs(False, app_call.kwargs["azure_region"]) + self.assertEqual(self.cert, calls[0].kwargs["client_credential"]) credential = calls[1].kwargs["client_credential"] - self.assertIs(self.cert, credential["mtls_binding_certificate"]) + self.assertIs( + calls[0].kwargs["client_credential"], + credential["mtls_binding_certificate"]) self.assertEqual("leg1-token", credential["client_assertion"]()) self.leg1.acquire_token_for_client.assert_called_once_with( - [self.config["LAB_FIC_EXCHANGE_SCOPE"]], mtls_proof_of_possession=True) + ["api://AzureADTokenExchange/.default"], mtls_proof_of_possession=True) + self.leg2.acquire_token_for_client.assert_called_once_with( + ["https://graph.microsoft.com/.default"], mtls_proof_of_possession=True) self.assertEqual( [ call("opaque-pop-token", self.pop_binding), ], self.scenario._call_graph_with_mtls_pop_token.call_args_list) + def test_bearer_uses_regional_shared_lab_without_downstream_mtls(self): + self.leg2.acquire_token_for_client.return_value = { + "access_token": "bearer-token", "token_type": "Bearer"} + self.scenario.test_credential_fic_output_bearer() + calls = self.factory.call_args_list + self.assertEqual(2, len(calls)) + for app_call in calls: + self.assertEqual( + "163ffef9-a313-45b4-ab2f-c7e2f5e0e23e", app_call.args[0]) + self.assertEqual( + "https://login.microsoftonline.com/bea21ebe-8b64-4d06-9f6d-6a889b120a7c", + app_call.kwargs["authority"]) + self.assertEqual("westus3", app_call.kwargs["azure_region"]) + credential = calls[1].kwargs["client_credential"] + self.assertIs( + calls[0].kwargs["client_credential"], + credential["mtls_binding_certificate"]) + self.assertEqual("leg1-token", credential["client_assertion"]()) + self.leg1.acquire_token_for_client.assert_called_once_with( + ["api://AzureADTokenExchange/.default"], mtls_proof_of_possession=True) + self.leg2.acquire_token_for_client.assert_called_once_with( + ["https://storage.azure.com/.default"], mtls_proof_of_possession=False) + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_leg2_errors_fail_instead_of_skipping(self): + for method in ( + self.scenario.test_credential_fic_output_pop, + self.scenario.test_credential_fic_output_bearer): + for error in ("invalid_client", "invalid_grant", "temporarily_unavailable"): + with self.subTest(method=method.__name__, error=error): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = { + "error": error} + with self.assertRaisesRegex(AssertionError, error): + method() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + def test_bearer_contract_is_checked(self): for response in ( {"access_token": "token", "token_type": "mtls_pop"}, @@ -708,9 +767,20 @@ def test_bearer_contract_is_checked(self): "binding_certificate": {}}): with self.subTest(response=response): self.factory.side_effect = [self.leg1, self.leg2] - self.leg2.acquire_token_for_client.side_effect = [response] + self.leg2.acquire_token_for_client.return_value = response + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_bearer() + + def test_pop_requires_matching_final_binding_certificate(self): + for binding in (None, {}, Mock(spec=msal.MtlsBindingContext)): + with self.subTest(binding=binding): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = { + "access_token": "opaque-pop-token", "token_type": "mtls_pop", + "binding_certificate": binding} with self.assertRaises(AssertionError): - self.scenario.test_fic_two_leg_over_mtls_pop() + self.scenario.test_credential_fic_output_pop() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once