From 01b676b0815fea5dbe980965db5f246ba6bf1f20 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Mon, 6 Jul 2026 19:06:20 -0400 Subject: [PATCH 01/11] 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 02/11] 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 03/11] 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 04/11] 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 05/11] 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 06/11] 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 07/11] 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 From 3a9c636370b2f2ce5b1100f5fc334573c3768069 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 10:43:43 -0400 Subject: [PATCH 08/11] Restore default FIC E2E wire and binding guarantees Extract the default FIC live and offline checks from validated combined commit 0cd290d without custom HTTP factory behavior. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/README.md | 25 ++++ tests/test_e2e.py | 242 ++++++++-------------------------- tests/test_fic_e2e_helpers.py | 207 +++++++++++++++++++++++++++++ 3 files changed, 289 insertions(+), 185 deletions(-) create mode 100644 tests/test_fic_e2e_helpers.py diff --git a/tests/README.md b/tests/README.md index 636401fa..7ea8d20c 100644 --- a/tests/README.md +++ b/tests/README.md @@ -55,12 +55,37 @@ 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. +Both default-transport cells require real token POST evidence for each leg, +the expected global or `westus3` mTLS endpoint, and the same lab certificate. +Leg 2 must invoke the assertion provider and send its exact, unchanged leg-1 +value as a `jwt-pop` client assertion. The exchange assertion may be opaque; +it is not assumed to be a resource JWT. Final resource tokens must be JWTs +with the Graph or Storage audience. Final Graph PoP additionally requires a +matching SHA256 `cnf` and returned `MtlsBindingContext`, then HTTP 200 from the +mTLS Graph resource using that context. + +The required FIC inventory is exactly these two methods in +`tests.test_e2e.FicTwoLegTestCase`: + +- `test_credential_fic_output_bearer` +- `test_credential_fic_output_pop` + +They run without skips, including on tagged releases. Custom HTTP-factory +acquisitions are not part of this default-FIC branch. + Run these two live cells with: ```powershell python -m pytest -q tests\test_e2e.py::FicTwoLegTestCase ``` +The offline `tests\test_fic_e2e_helpers.py` checks the two-leg provider/wire +contract, opaque exchange assertions, strict final binding/audience, and exact +assertion handoff. Its privacy regression uses pytest's actual traceback +renderer with arguments shown and locals disabled, and rejects full tokens +and truncated token fragments. These mock-driven guards are not evidence of +live acquisition or a custom HTTP factory. + ## 4) Run unit/integration tests Run all non-E2E tests quickly: diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 15bbbcf0..610a2ccf 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -19,9 +19,9 @@ from urllib.parse import urlparse, parse_qs import sys try: - from unittest.mock import patch, ANY, Mock, call, mock_open + from unittest.mock import patch, ANY except: - from mock import patch, ANY, Mock, call, mock_open + from mock import patch, ANY import requests @@ -34,6 +34,7 @@ get_user_config, get_app_config, get_user_password, get_secret, UserSecrets, AppSecrets, LAB_APP_CLIENT_ID, clean_env, ) +from tests.mtls_e2e_helpers import JWT_POP logger = logging.getLogger(__name__) @@ -568,107 +569,78 @@ class FicTwoLegTestCase(MtlsPopTestCase): def _acquire_fic_token( self, scope, azure_region, mtls_proof_of_possession=False): - from tests.lab_config import get_client_certificate - cert = get_client_certificate() - leg1_app = msal.ConfidentialClientApplication( - CertificateScenariosTestCase._SNI_ALLOWLISTED_CLIENT_ID, - authority=CertificateScenariosTestCase._SNI_ALLOWLISTED_AUTHORITY, - client_credential=cert, - azure_region=azure_region) - leg1 = leg1_app.acquire_token_for_client( - 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( - CertificateScenariosTestCase._SNI_ALLOWLISTED_CLIENT_ID, - authority=CertificateScenariosTestCase._SNI_ALLOWLISTED_AUTHORITY, - client_credential={ - "client_assertion": lambda: leg1["access_token"], - "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 + __tracebackhide__ = True + credential, binding = self._certificate() + leg1_app = self._sni_app(credential, azure_region=azure_region) + leg1, leg1_requests = self._acquire_recorded( + leg1_app, self._FIC_EXCHANGE_SCOPE, binding, pop=True) + self._assert_token_result( + leg1, self._FIC_EXCHANGE_SCOPE, "mtls_pop", binding) + first_request = self._assert_token_request( + leg1_requests, self._FIC_EXCHANGE_SCOPE, "client_credentials", + region=azure_region) + self.assertEqual("mtls_pop", first_request["token_type"]) + self.assertEqual(200, first_request["status_code"]) + + provider_calls = [] + + def assertion_provider(): + provider_calls.append(True) + return leg1["access_token"] + + leg2_app = self._sni_app( + azure_region=azure_region, + credential={ + "client_assertion": assertion_provider, + "mtls_binding_certificate": credential}) + result, leg2_requests = self._acquire_recorded( + leg2_app, scope, binding, pop=mtls_proof_of_possession, + expected_assertion=leg1["access_token"]) + self.assertTrue(provider_calls, "The assertion provider must actually be invoked.") + self._assert_token_result( + result, scope, "mtls_pop" if mtls_proof_of_possession else "Bearer", + binding) + self._assert_audience(result, scope) + second_request = self._assert_token_request( + leg2_requests, scope, "client_credentials", region=azure_region, + assertion_type=JWT_POP) + self.assertTrue( + second_request["assertion_matches"], + "The actual jwt-pop POST must carry the leg-1 assertion unchanged.") + self.assertEqual( + "mtls_pop" if mtls_proof_of_possession else None, + second_request["token_type"]) + self.assertEqual(200, second_request["status_code"]) + return leg1["binding_certificate"].thumbprint_sha256, 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) + self._acquire_fic_token(self._FIC_BEARER_SCOPE, azure_region="westus3") def test_credential_fic_output_pop(self): - leg1_thumbprint, leg2_pop = self._acquire_fic_token( + _, result = 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 - 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") - 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) + result["access_token"], result["binding_certificate"]) 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", - } - 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.pop_binding = Mock(spec=msal.MtlsBindingContext) - self.pop_binding.get.return_value = "cert-thumbprint" - self.leg2 = Mock() - 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.open", mock_open(read_data=b"unit-test-pfx")), - patch( - "tests.lab_config.pkcs12.load_key_and_certificates", - return_value=(Mock(), Mock(), [])), - 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) + """Keep the live FIC cells fail-closed when lab configuration is missing.""" def test_missing_certificate_fails_before_acquisition(self): + scenario = FicTwoLegTestCase("test_credential_fic_output_pop") for method in ( - self.scenario.test_credential_fic_output_pop, - self.scenario.test_credential_fic_output_bearer): + scenario.test_credential_fic_output_pop, + 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 patch.dict(os.environ, { + "LAB_APP_CLIENT_CERT_PFX_PATH": value}, clear=True), patch( + "msal.ConfidentialClientApplication") as application: with self.assertRaisesRegex( EnvironmentError, "LAB_APP_CLIENT_CERT_PFX_PATH"): method() - self.factory.assert_not_called() + application.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): @@ -684,106 +656,6 @@ def test_missing_certificate_is_a_runner_error_even_on_tagged_releases(self): 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): - 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_shared_lab_defaults_and_opaque_token_resource_validation(self): - self.scenario.test_credential_fic_output_pop() - 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.assertIs(False, app_call.kwargs["azure_region"]) - self.assertEqual(self.cert, calls[0].kwargs["client_credential"]) - 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://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"}, - {"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.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_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 diff --git a/tests/test_fic_e2e_helpers.py b/tests/test_fic_e2e_helpers.py new file mode 100644 index 00000000..78316a69 --- /dev/null +++ b/tests/test_fic_e2e_helpers.py @@ -0,0 +1,207 @@ +"""Offline default-FIC guards, not evidence of live token acquisitions.""" +from unittest.mock import Mock, patch + +import pytest +import requests + +from tests import test_e2e +from tests.mtls_e2e_helpers import ( + JWT_BEARER, JWT_POP, token_request_summary, unrelated_binding, +) +from tests.test_mtls_e2e_helpers import ( + CertificateFixture, _CREDENTIAL, _ENDPOINT, _jwt, +) + + +class TestDefaultFicCapture(CertificateFixture): + def test_jwt_pop_capture_requires_the_exact_leg_one_assertion(self): + form = dict( + self._signed_form(), client_assertion="unit-fic-assertion", + client_assertion_type=JWT_POP) + for expected, matches in (("unit-fic-assertion", True), ("different", False)): + summary = token_request_summary( + self._response(form=form), self.binding, + requests.adapters.HTTPAdapter(), expected_assertion=expected) + self.assertEqual(matches, summary["assertion_matches"]) + self.assertFalse("unit-fic-assertion" in repr(summary)) + + def test_default_fic_capture_failure_traceback_hides_the_expected_assertion(self): + assertion = _jwt({"aud": "api://AzureADTokenExchange"}) + app = Mock() + app.acquire_token_for_client.side_effect = AssertionError( + "Unit acquisition failed before receiving a response.") + scenario = test_e2e.FicTwoLegTestCase("test_credential_fic_output_pop") + with pytest.raises(AssertionError) as failure: + scenario._acquire_recorded( + app, scenario._MTLS_POP_SCOPE, self.binding, + pop=True, expected_assertion=assertion) + self._assert_private_traceback(failure, assertion) + + +class TestFicScenarioGuards(CertificateFixture): + def setUp(self): + self.leg1 = Mock() + self.leg2 = Mock() + self.leg1.acquire_token_for_client.return_value = { + "access_token": _jwt({ + "aud": "api://AzureADTokenExchange", + "cnf": {"x5t#S256": self.binding.thumbprint_sha256}}), + "token_type": "mtls_pop", "binding_certificate": self.binding, + } + self.leg2.acquire_token_for_client.return_value = { + "access_token": _jwt({ + "aud": "https://graph.microsoft.com", + "cnf": {"x5t#S256": self.binding.thumbprint_sha256}}), + "token_type": "mtls_pop", "binding_certificate": self.binding, + } + self.factory = Mock(side_effect=[self.leg1, self.leg2]) + self.scenario = test_e2e.FicTwoLegTestCase("test_credential_fic_output_pop") + self.invoke_provider = True + self.omit_final_record = False + self.final_record_changes = {} + patchers = ( + patch("tests.lab_config.get_client_certificate", return_value=_CREDENTIAL), + patch("msal.ConfidentialClientApplication", self.factory), + patch.object(self.scenario, "_acquire_recorded", + side_effect=self._recorded), + patch.object(self.scenario, "_call_graph_with_mtls_pop_token"), + ) + for patcher in patchers: + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(self.scenario.doCleanups) + + def _recorded(self, app, scope, binding, pop=False, + http_client=None, expected_assertion=None): + config = self.factory.call_args.kwargs + final = app is self.leg2 + assertion = None + if final and self.invoke_provider: + assertion = config["client_credential"]["client_assertion"]() + result = app.acquire_token_for_client(scope, mtls_proof_of_possession=pop) + region = config["azure_region"] + record = { + "url": _ENDPOINT.replace( + "://mtlsauth.", "://{}.mtlsauth.".format(region)) if region else _ENDPOINT, + "fields": frozenset( + ("scope", "client_id", "grant_type") + + (("client_assertion", "client_assertion_type") if final else ()) + + (("token_type",) if pop else ())), + "grant_type": "client_credentials", "scope": frozenset(scope), + "client_assertion_type": JWT_POP if final else None, + "client_assertion_present": final, "signed_client_assertion": False, + "assertion_matches": assertion == expected_assertion, + "token_type": "mtls_pop" if pop else None, + "status_code": 200, "tls_adapter": "_MtlsHTTPAdapter", + } + if final: + record.update(self.final_record_changes) + return result, [] if final and self.omit_final_record else [record] + + def test_global_pairing_provider_wire_type_and_resource_validation(self): + self.scenario.test_credential_fic_output_pop() + calls = self.factory.call_args_list + self.assertEqual(2, len(calls)) + for app_call in calls: + self.assertEqual(self.scenario._SNI_ALLOWLISTED_CLIENT_ID, app_call.args[0]) + self.assertEqual( + self.scenario._SNI_ALLOWLISTED_AUTHORITY, app_call.kwargs["authority"]) + self.assertIs(False, app_call.kwargs["azure_region"]) + self.assertIs( + calls[0].kwargs["client_credential"], + calls[1].kwargs["client_credential"]["mtls_binding_certificate"]) + 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://graph.microsoft.com/.default"], mtls_proof_of_possession=True) + self.scenario._call_graph_with_mtls_pop_token.assert_called_once() + + def test_regional_bearer_has_storage_audience_and_no_resource_mtls(self): + self.leg2.acquire_token_for_client.return_value = { + "access_token": _jwt({"aud": "https://storage.azure.com"}), + "token_type": "Bearer"} + self.scenario.test_credential_fic_output_bearer() + self.assertTrue(all( + call.kwargs["azure_region"] == "westus3" + for call in self.factory.call_args_list)) + 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_leg_one_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 ("token_type_mismatch", "invalid_client", "invalid_grant"): + with self.subTest(error=error, method=method.__name__): + 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_leg_two_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(error=error, method=method.__name__): + 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_federated_assertion_is_opaque_but_still_requires_certificate_binding(self): + self.leg1.acquire_token_for_client.return_value["access_token"] = "opaque" + self.scenario.test_credential_fic_output_pop() + self.leg2.acquire_token_for_client.reset_mock() + for binding in (None, {}, unrelated_binding()): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg1.acquire_token_for_client.return_value["binding_certificate"] = binding + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.leg2.acquire_token_for_client.assert_not_called() + + def test_final_pop_requires_type_binding_cnf_and_audience(self): + original = self.leg2.acquire_token_for_client.return_value + for changes in ( + {"token_type": "Bearer"}, {"binding_certificate": None}, + {"binding_certificate": {}}, {"access_token": "opaque"}, + {"access_token": _jwt({ + "aud": "https://graph.microsoft.com", + "cnf": {"x5t#S256": "different-certificate"}})}, + {"access_token": _jwt({ + "aud": "https://wrong.example", + "cnf": {"x5t#S256": self.binding.thumbprint_sha256}})}): + with self.subTest(fields=tuple(changes)): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = dict(original, **changes) + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_final_bearer_rejects_bound_or_wrong_audience_results(self): + for result in ( + {"access_token": "unit-token", "token_type": "mtls_pop"}, + {"access_token": _jwt({"aud": "https://storage.azure.com"}), + "token_type": "Bearer", "binding_certificate": {}}, + {"access_token": _jwt({"aud": "https://wrong.example"}), + "token_type": "Bearer"}): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = result + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_bearer() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_fic_requires_provider_invocation_and_actual_wire_evidence(self): + for invoke, omit, changes in ( + (False, False, {}), (True, True, {}), + (True, False, {"client_assertion_type": JWT_BEARER}), + (True, False, {"assertion_matches": False})): + self.factory.side_effect = [self.leg1, self.leg2] + self.invoke_provider, self.omit_final_record = invoke, omit + self.final_record_changes = changes + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() From cd295d4a98b76b2e895f5677c41257f4285c390d Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 10:44:46 -0400 Subject: [PATCH 09/11] Document targeted offline FIC guard invocation Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/README.md b/tests/README.md index 7ea8d20c..aeea5538 100644 --- a/tests/README.md +++ b/tests/README.md @@ -86,6 +86,10 @@ renderer with arguments shown and locals disabled, and rejects full tokens and truncated token fragments. These mock-driven guards are not evidence of live acquisition or a custom HTTP factory. +```powershell +python -m pytest -q tests\test_fic_e2e_helpers.py tests\test_lab_config.py tests\test_e2e.py::FicScenarioTestCase +``` + ## 4) Run unit/integration tests Run all non-E2E tests quickly: From 0611fbdcd21cf6e8af017dbd94804a7a81f09a48 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Mon, 2 Nov 2026 08:19:21 -0500 Subject: [PATCH 10/11] Preserve actual certificate binding across FIC legs Carry the returned leg-one binding context into leg two, guard assertion renewal, and cover cached assertions after PFX replacement. Adapt default FIC guards to the shared opaque-Graph resource contract. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/index.rst | 8 +- msal/application.py | 40 ++-- sample/confidential_client_mtls_pop_sample.py | 26 ++- tests/README.md | 27 ++- tests/test_application.py | 219 +++++++++++++++++- tests/test_e2e.py | 4 +- tests/test_fic_e2e_helpers.py | 93 +++++++- 7 files changed, 363 insertions(+), 54 deletions(-) diff --git a/docs/index.rst b/docs/index.rst index a57b29ff..6c0b7dbf 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -245,8 +245,12 @@ Notes and requirements: ``False`` disables server verification. * For a Federated Identity Credential (FIC) exchange, configure the leg-2 client 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 mTLS using the same certificate as leg 1. + ``mtls_binding_certificate`` sub-key containing the leg-1 result's + ``binding_certificate``. This reuses the actual certificate snapshot rather + than reopening a PFX path that may have changed between the two legs. The + leg-1 assertion is then sent as a ``jwt-pop`` client assertion over mTLS + using that same certificate. A certificate credential dictionary remains + supported when the caller already guarantees certificate continuity. 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 diff --git a/msal/application.py b/msal/application.py index b739417c..4ca97fa4 100644 --- a/msal/application.py +++ b/msal/application.py @@ -620,31 +620,37 @@ def get_client_assertion(): 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. - An explicitly supplied binding certificate must be a non-empty - certificate credential dictionary; invalid or incomplete values - raise ``ValueError``. + carry an ``mtls_binding_certificate`` sub-key. Pass the + :class:`~msal.MtlsBindingContext` returned with the leg-1 token + to present that exact certificate during the leg-2 token request. + A non-empty certificate credential dictionary (same shape as the + top-level cert credential) is also supported. 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:: + leg1 = leg1_app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], + mtls_proof_of_possession=True, + ) + if "access_token" not in leg1: + raise RuntimeError("FIC leg 1 failed") + binding = leg1["binding_certificate"] + 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"))) + raise RuntimeError("FIC leg 1 failed") + if result["binding_certificate"].x5c != binding.x5c: + raise RuntimeError("FIC certificate binding changed") return result["access_token"] { "client_assertion": get_leg1_assertion, - "mtls_binding_certificate": { - "private_key_pfx_path": "/path/to/your.pfx", - "public_certificate": True, - }, + "mtls_binding_certificate": binding, } When ``mtls_binding_certificate`` is present, MSAL sends the @@ -1025,15 +1031,17 @@ def get_leg1_assertion(): 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 ( + if not isinstance(binding_certificate, mtls.MtlsBindingContext) and ( + 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"])): + and binding_certificate["public_certificate"]))): raise ValueError( - "mtls_binding_certificate must be a non-empty certificate " - "credential dictionary with 'private_key_pfx_path', or " + "mtls_binding_certificate must be an MtlsBindingContext " + "or 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( diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index 88f0c370..9338dd4a 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -37,7 +37,9 @@ 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. +subject, and audience. Leg 2 reuses the actual certificate binding returned +with the leg-1 assertion, not a second load of the PFX path. Each application +keeps its certificate snapshot; replacing the file does not rotate those apps. Then run:: @@ -147,14 +149,28 @@ def fic_two_leg_over_mtls_pop(): exchange_scope = os.getenv( "FIC_EXCHANGE_SCOPE", "api://AzureADTokenExchange/.default") - def get_leg1_assertion(): + def acquire_leg1(): # 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: + if not leg1.get("access_token"): _print_result_summary(leg1, "FIC leg 1 result:") + raise RuntimeError("FIC leg 1 failed; inspect the result to diagnose.") + if (leg1.get("token_type") != "mtls_pop" or not isinstance( + leg1.get("binding_certificate"), msal.MtlsBindingContext)): + raise RuntimeError("FIC leg 1 must return an mTLS PoP certificate binding.") + return leg1 + + first_leg1 = acquire_leg1() + binding = first_leg1["binding_certificate"] + + def get_leg1_assertion(): + leg1 = acquire_leg1() + current_binding = leg1["binding_certificate"] + if (current_binding.thumbprint_sha256 != binding.thumbprint_sha256 + or current_binding.x5c != binding.x5c): raise RuntimeError( - "FIC leg 1 failed: {}".format(leg1.get("error"))) + "FIC leg 1 certificate binding changed; recreate both applications.") return leg1["access_token"] # --- Leg 2: exchange the leg-1 token for the final token --------------- @@ -163,7 +179,7 @@ def get_leg1_assertion(): authority=os.environ["LEG2_AUTHORITY"], client_credential={ "client_assertion": get_leg1_assertion, - "mtls_binding_certificate": cert, + "mtls_binding_certificate": binding, }, azure_region=os.getenv("AZURE_REGION"), ) diff --git a/tests/README.md b/tests/README.md index 74bbb5ab..417680e6 100644 --- a/tests/README.md +++ b/tests/README.md @@ -59,10 +59,15 @@ Both default-transport cells require real token POST evidence for each leg, the expected global or `westus3` mTLS endpoint, and the same lab certificate. Leg 2 must invoke the assertion provider and send its exact, unchanged leg-1 value as a `jwt-pop` client assertion. The exchange assertion may be opaque; -it is not assumed to be a resource JWT. Final resource tokens must be JWTs -with the Graph or Storage audience. Final Graph PoP additionally requires a -matching SHA256 `cnf` and returned `MtlsBindingContext`, then HTTP 200 from the -mTLS Graph resource using that context. +it is not assumed to be a resource JWT. Leg 2 receives the actual +`MtlsBindingContext` returned with the leg-1 assertion rather than reloading +the PFX path. The sample also checks binding continuity when the callable +renews the assertion. Final Storage Bearer tokens must have the Storage +audience. Graph owns its token format: the final Graph token is treated as +opaque and requires matching returned certificate binding plus HTTP 200 from +the mTLS Graph resource using that context, not client-side JWT inspection. +The same token must also fail the shared no-certificate, wrong-certificate, +and Bearer-scheme controls with their exact rejection markers. The required FIC inventory is exactly these two methods in `tests.test_e2e.FicTwoLegTestCase`: @@ -84,7 +89,7 @@ and Go `30f4770ffcbd9c4c51d278cdab6dbda71a0f7337` | Active reference case(s) | Python FIC method | Required observation | | --- | --- | --- | -| Go `TestTwoLegFICMtlsPoP_SNI`; .NET `Credential_Fic_Output_Pop_TestAsync` / `RunTwoLegS2sFicBothLegsPopAsync` | `test_credential_fic_output_pop` | Global exchange to Graph; invoked provider, unchanged `jwt-pop` assertion, certificate continuity, strict final audience/type/`cnf`, resource HTTP 200. | +| Go `TestTwoLegFICMtlsPoP_SNI`; .NET `Credential_Fic_Output_Pop_TestAsync` / `RunTwoLegS2sFicBothLegsPopAsync` | `test_credential_fic_output_pop` | Global exchange to Graph; invoked provider, unchanged `jwt-pop` assertion, actual returned-binding handoff, final PoP type/binding, authoritative resource HTTP 200. | | Go `TestTwoLegFICMtlsBearer_SNI`; .NET `Credential_Fic_Output_Bearer_TestAsync` | `test_credential_fic_output_bearer` | `westus3` exchange to Storage; provider and real `jwt-pop` POST, certificate continuity, Storage Bearer, no output binding or downstream resource mTLS. | Run these two live cells with: @@ -94,10 +99,14 @@ python -m pytest -q tests\test_e2e.py::FicTwoLegTestCase ``` The offline `tests\test_fic_e2e_helpers.py` checks the two-leg provider/wire -contract, opaque exchange assertions, strict final binding/audience, and exact -assertion handoff. Its privacy regression uses pytest's actual traceback -renderer with arguments shown and locals disabled, and rejects full tokens -and truncated token fragments. These mock-driven guards are not evidence of +contract, opaque exchange and Graph tokens, strict final binding/type and +Storage audience, Graph rejection failures, and exact assertion handoff. +`tests.test_application.TestFicCertificateHandoff` replaces a PFX with a second +certificate after caching the leg-1 assertion, checks both token and actual +TLS certificate/key continuity, and checks renewal and Bearer/PoP output +contracts without reopening that file. The recorder privacy regression uses +pytest's actual traceback renderer with arguments shown and locals disabled, +and rejects full tokens and truncated token fragments. These offline guards are not evidence of live acquisition or a custom HTTP factory. ```powershell diff --git a/tests/test_application.py b/tests/test_application.py index 92190053..26b654ee 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -6,7 +6,9 @@ import os import pickle import runpy +import shutil import sys +import tempfile import time import warnings from unittest.mock import patch, Mock @@ -2741,15 +2743,21 @@ def test_fic_sample_skips_resource_call_for_bearer_result(self): 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 + leg1_app = Mock() + leg1_app.acquire_token_for_client.return_value = { + "access_token": "assertion", "token_type": "mtls_pop", + "binding_certificate": msal.MtlsBindingContext( + b"cert", b"key", ["certificate"], "thumbprint"), + } + leg2_app = Mock() + leg2_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( + side_effect=[leg1_app, leg2_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() @@ -2758,11 +2766,15 @@ 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.assertRaisesRegex( - ValueError, "also requires a client_assertion"): - ConfidentialClientApplication( - "cid", authority=self._AUTHORITY, - client_credential={"mtls_binding_certificate": _MTLS_CERT_CRED}) + for binding in ( + _MTLS_CERT_CRED, + msal.MtlsBindingContext(b"cert", b"key", ["certificate"], "thumbprint")): + with self.subTest(binding_type=type(binding).__name__): + with self.assertRaisesRegex( + ValueError, "also requires a client_assertion"): + ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={"mtls_binding_certificate": binding}) def test_invalid_binding_certificate_is_rejected(self): for certificate in ( @@ -2873,6 +2885,197 @@ def test_custom_http_client_with_flag_fails_fast(self): app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True) +@patch(_OIDC_DISCOVERY, new=_MTLS_OIDC_MOCK) +class TestFicCertificateHandoff(unittest.TestCase): + _AUTHORITY = TestMtlsProofOfPossession._AUTHORITY + + @classmethod + def setUpClass(cls): + from datetime import datetime, timedelta, timezone + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.hazmat.primitives.serialization import pkcs12 + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute( + x509.oid.NameOID.COMMON_NAME, "FIC handoff replacement")]) + now = datetime.now(timezone.utc) + cert = (x509.CertificateBuilder() + .subject_name(name).issuer_name(name).public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=1)) + .sign(key, hashes.SHA256())) + cls._replacement_pfx = pkcs12.serialize_key_and_certificates( + b"fic-replacement", key, cert, None, + serialization.BestAvailableEncryption(b"password")) + + def setUp(self): + from msal.application import _load_mtls_cert_material + + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.pfx_path = os.path.join(directory.name, "credential.pfx") + shutil.copyfile(_MTLS_PFX, self.pfx_path) + self.credential = dict(_MTLS_CERT_CRED, private_key_pfx_path=self.pfx_path) + self.material = _load_mtls_cert_material(self.credential) + self.binding = self._binding(self.material) + self.requests = [] + self.sample = runpy.run_path(os.path.join( + os.path.dirname(__file__), os.pardir, "sample", + "confidential_client_mtls_pop_sample.py")) + environment = patch.dict(os.environ, { + "CLIENT_ID": "leg1", "AUTHORITY": self._AUTHORITY, + "LEG2_CLIENT_ID": "leg2", "LEG2_AUTHORITY": self._AUTHORITY, + "PFX_PATH": self.pfx_path, "PFX_PASSPHRASE": "password", + "SCOPE": "final-scope", "RESOURCE_URL": "", + "AZURE_REGION": "", + "FIC_EXCHANGE_SCOPE": "api://AzureADTokenExchange/.default", + }) + environment.start() + self.addCleanup(environment.stop) + + @staticmethod + def _binding(material): + return msal.MtlsBindingContext( + material["cert_pem"], material["private_key_pem"], + material["x5c"], material["key_id"]) + + def _replace_pfx(self): + with open(self.pfx_path, "wb") as stream: + stream.write(self._replacement_pfx) + + def _post(self, transport, url, data=None, **kwargs): + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + + cert = x509.load_pem_x509_certificate(transport._cert_pem) + key = serialization.load_pem_private_key(transport._key_pem, password=None) + self.assertEqual( + self.material["sha256_thumbprint"], + cert.fingerprint(hashes.SHA256()).hex()) + self.assertEqual( + cert.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo), + key.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo)) + transport._ensure_session(url) # Build the real certificate-bearing TLS context. + self.addCleanup(transport.close) + self.requests.append(dict(data or {})) + return MinimalResponse(status_code=200, text=json.dumps({ + "access_token": "unit-assertion-{}".format(len(self.requests)), + "token_type": (data or {}).get("token_type", "Bearer"), + "expires_in": 3600, + })) + + def test_fic_sample_reuses_cached_assertion_certificate_after_pfx_replacement(self): + factory = msal.ConfidentialClientApplication + apps = [] + + def create_app(*args, **kwargs): + app = factory(*args, **kwargs) + apps.append(app) + if len(apps) == 1: + app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], + mtls_proof_of_possession=True) + self._replace_pfx() + return app + + from msal.application import _parse_pfx + with patch("msal.application._parse_pfx", wraps=_parse_pfx) as parse, patch( + "msal.ConfidentialClientApplication", side_effect=create_app), patch( + "msal.mtls._MtlsHttpClient.post", + new=lambda transport, url, **kwargs: + self._post(transport, url, **kwargs)), patch("builtins.print"): + result = self.sample["fic_two_leg_over_mtls_pop"]() + self.assertEqual(["leg1", "leg2"], [ + request["client_id"] for request in self.requests]) + self.assertEqual( + "unit-assertion-1", self.requests[1]["client_assertion"]) + self.assertEqual(_JWT_POP, self.requests[1]["client_assertion_type"]) + binding = apps[1].client_credential["mtls_binding_certificate"] + self.assertIsInstance(binding, msal.MtlsBindingContext) + self.assertEqual( + self.binding.thumbprint_sha256, + result["binding_certificate"].thumbprint_sha256) + for entry in list(apps[0].token_cache.search( + msal.TokenCache.CredentialType.ACCESS_TOKEN, for_removal=True)): + apps[0].token_cache.modify( + msal.TokenCache.CredentialType.ACCESS_TOKEN, entry, + {"expires_on": "0"}) + renewed = apps[1].client_credential["client_assertion"]() + self.assertEqual("unit-assertion-3", renewed) + self.assertEqual(["leg1", "leg2", "leg1"], [ + request["client_id"] for request in self.requests]) + self.assertEqual(1, parse.call_count, "PFX must be loaded once per app.") + + def test_fic_binding_context_preserves_bearer_and_pop_contracts(self): + self._replace_pfx() + with patch( + "msal.application._parse_pfx", + side_effect=AssertionError("FIC must not reload the PFX path.")), patch( + "msal.mtls._MtlsHttpClient.post", + new=lambda transport, url, **kwargs: + self._post(transport, url, **kwargs)): + for pop in (False, True): + with self.subTest(pop=pop): + app = ConfidentialClientApplication( + "leg2", authority=self._AUTHORITY, azure_region=False, + client_credential={ + "client_assertion": lambda: "cached-assertion-A", + "mtls_binding_certificate": self.binding}) + result = app.acquire_token_for_client( + ["final-scope"], mtls_proof_of_possession=pop) + self.assertEqual( + "mtls_pop" if pop else "Bearer", result["token_type"]) + self.assertEqual(pop, "binding_certificate" in result) + self.assertEqual( + "cached-assertion-A", self.requests[-1]["client_assertion"]) + self.assertEqual(_JWT_POP, self.requests[-1]["client_assertion_type"]) + + def test_fic_sample_rejects_changed_assertion_binding(self): + from msal.application import _load_mtls_cert_material + + self._replace_pfx() + replacement = self._binding(_load_mtls_cert_material(self.credential)) + leg1 = Mock() + leg1.acquire_token_for_client.side_effect = [{ + "access_token": "unit-assertion", "token_type": "mtls_pop", + "binding_certificate": binding, + } for binding in (self.binding, replacement)] + factory = msal.ConfidentialClientApplication + + def create_app(client_id, **kwargs): + return leg1 if client_id == "leg1" else factory(client_id, **kwargs) + + with patch("msal.ConfidentialClientApplication", side_effect=create_app), patch( + "msal.mtls._MtlsHttpClient.post") as post: + with self.assertRaisesRegex(RuntimeError, "certificate binding changed"): + self.sample["fic_two_leg_over_mtls_pop"]() + post.assert_not_called() + + def test_fic_sample_requires_leg_one_binding_before_constructing_leg_two(self): + for result in ( + {"error": "invalid_client"}, + {"access_token": "unit-assertion", "token_type": "Bearer"}, + {"access_token": "unit-assertion", "token_type": "mtls_pop"}, + {"access_token": "unit-assertion", "token_type": "mtls_pop", + "binding_certificate": {}}): + with self.subTest(fields=tuple(result)): + leg1 = Mock() + leg1.acquire_token_for_client.return_value = result + with patch( + "msal.ConfidentialClientApplication", + return_value=leg1) as factory, patch("builtins.print"): + with self.assertRaises(RuntimeError): + self.sample["fic_two_leg_over_mtls_pop"]() + self.assertEqual(1, factory.call_count) + + @patch(_OIDC_DISCOVERY, new=Mock(return_value={ "authorization_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/authorize", "token_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/token", diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 99878388..c6ae5b4d 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -826,7 +826,7 @@ def assertion_provider(): azure_region=azure_region, credential={ "client_assertion": assertion_provider, - "mtls_binding_certificate": credential}) + "mtls_binding_certificate": leg1["binding_certificate"]}) result, leg2_requests = self._acquire_recorded( leg2_app, scope, binding, pop=mtls_proof_of_possession, expected_assertion=leg1["access_token"]) @@ -854,7 +854,7 @@ def test_credential_fic_output_pop(self): _, result = self._acquire_fic_token( self._MTLS_POP_SCOPE, azure_region=False, mtls_proof_of_possession=True) - self._call_graph_with_mtls_pop_token( + self._assert_graph_token_binding( result["access_token"], result["binding_certificate"]) diff --git a/tests/test_fic_e2e_helpers.py b/tests/test_fic_e2e_helpers.py index 78316a69..f3ddfb88 100644 --- a/tests/test_fic_e2e_helpers.py +++ b/tests/test_fic_e2e_helpers.py @@ -1,12 +1,13 @@ """Offline default-FIC guards, not evidence of live token acquisitions.""" -from unittest.mock import Mock, patch +from unittest.mock import MagicMock, Mock, patch import pytest import requests from tests import test_e2e from tests.mtls_e2e_helpers import ( - JWT_BEARER, JWT_POP, token_request_summary, unrelated_binding, + GRAPH_REJECTION_MARKERS, JWT_BEARER, JWT_POP, + token_request_summary, unrelated_binding, ) from tests.test_mtls_e2e_helpers import ( CertificateFixture, _CREDENTIAL, _ENDPOINT, _jwt, @@ -59,18 +60,52 @@ def setUp(self): self.invoke_provider = True self.omit_final_record = False self.final_record_changes = {} + self.graph_response = Mock(status_code=200, text="") + self.graph_requests = [] + self.invalid_graph_marker = None patchers = ( patch("tests.lab_config.get_client_certificate", return_value=_CREDENTIAL), patch("msal.ConfidentialClientApplication", self.factory), patch.object(self.scenario, "_acquire_recorded", side_effect=self._recorded), - patch.object(self.scenario, "_call_graph_with_mtls_pop_token"), + patch.object( + self.scenario, "_call_graph_with_mtls_pop_token", + wraps=self.scenario._call_graph_with_mtls_pop_token), + patch("msal.MtlsBindingContext.create_requests_session", + autospec=True, side_effect=self._graph_session_for), + patch("requests.Session", side_effect=self._graph_session_for), ) for patcher in patchers: patcher.start() self.addCleanup(patcher.stop) self.addCleanup(self.scenario.doCleanups) + def _graph_session_for(self, binding=None): + def get(url, headers, **kwargs): + self.assertEqual(self.scenario._MTLS_GRAPH_RESOURCE, url) + self.assertEqual({"timeout": 30, "allow_redirects": False}, kwargs) + header = headers["Authorization"] + if binding is None: + marker = GRAPH_REJECTION_MARKERS[0] + elif header.startswith("Bearer "): + marker = GRAPH_REJECTION_MARKERS[1] + elif binding.thumbprint_sha256 != self.binding.thumbprint_sha256: + marker = GRAPH_REJECTION_MARKERS[2] + else: + marker = None + self.graph_requests.append({ + "binding": binding, "header": header, "marker": marker}) + if marker is None: + return self.graph_response + return Mock( + status_code=200 if marker == self.invalid_graph_marker else 401, + text=marker) + + session = MagicMock() + session.__enter__.return_value = session + session.get.side_effect = get + return session + def _recorded(self, app, scope, binding, pop=False, http_client=None, expected_assertion=None): config = self.factory.call_args.kwargs @@ -108,13 +143,23 @@ def test_global_pairing_provider_wire_type_and_resource_validation(self): self.scenario._SNI_ALLOWLISTED_AUTHORITY, app_call.kwargs["authority"]) self.assertIs(False, app_call.kwargs["azure_region"]) self.assertIs( - calls[0].kwargs["client_credential"], + self.leg1.acquire_token_for_client.return_value["binding_certificate"], calls[1].kwargs["client_credential"]["mtls_binding_certificate"]) 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://graph.microsoft.com/.default"], mtls_proof_of_possession=True) self.scenario._call_graph_with_mtls_pop_token.assert_called_once() + self.assertEqual(4, len(self.graph_requests)) + token = self.leg2.acquire_token_for_client.return_value["access_token"] + self.assertCountEqual( + ["mtls_pop " + token] * 3 + ["Bearer " + token], + [request["header"] for request in self.graph_requests]) + self.assertEqual(2, sum( + request["binding"] is self.binding for request in self.graph_requests)) + self.assertCountEqual( + [None] + list(GRAPH_REJECTION_MARKERS), + [request["marker"] for request in self.graph_requests]) def test_regional_bearer_has_storage_audience_and_no_resource_mtls(self): self.leg2.acquire_token_for_client.return_value = { @@ -163,17 +208,12 @@ def test_federated_assertion_is_opaque_but_still_requires_certificate_binding(se self.scenario.test_credential_fic_output_pop() self.leg2.acquire_token_for_client.assert_not_called() - def test_final_pop_requires_type_binding_cnf_and_audience(self): + def test_final_pop_requires_type_and_matching_certificate_binding(self): original = self.leg2.acquire_token_for_client.return_value for changes in ( {"token_type": "Bearer"}, {"binding_certificate": None}, - {"binding_certificate": {}}, {"access_token": "opaque"}, - {"access_token": _jwt({ - "aud": "https://graph.microsoft.com", - "cnf": {"x5t#S256": "different-certificate"}})}, - {"access_token": _jwt({ - "aud": "https://wrong.example", - "cnf": {"x5t#S256": self.binding.thumbprint_sha256}})}): + {"binding_certificate": {}}, + {"binding_certificate": unrelated_binding()}): with self.subTest(fields=tuple(changes)): self.factory.side_effect = [self.leg1, self.leg2] self.leg2.acquire_token_for_client.return_value = dict(original, **changes) @@ -181,6 +221,35 @@ def test_final_pop_requires_type_binding_cnf_and_audience(self): self.scenario.test_credential_fic_output_pop() self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + def test_graph_format_is_opaque_and_resource_acceptance_is_authoritative(self): + for token in ( + "opaque-graph-token", + _jwt({"aud": "https://wrong.example"}), + _jwt({"cnf": {"x5t#S256": "different-certificate"}})): + for status in (200, 401, 403, 302, 500): + with self.subTest(format="jwt" if "." in token else "opaque", + status=status): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value["access_token"] = token + self.graph_response.status_code = status + self.graph_requests.clear() + if status == 200: + self.scenario.test_credential_fic_output_pop() + else: + with self.assertRaisesRegex(AssertionError, "Graph resource"): + self.scenario.test_credential_fic_output_pop() + self.assertEqual("mtls_pop " + token, self.graph_requests[0]["header"]) + + def test_fic_graph_requires_all_same_token_rejection_controls(self): + for marker in GRAPH_REJECTION_MARKERS: + with self.subTest(marker=marker): + self.factory.side_effect = [self.leg1, self.leg2] + self.invalid_graph_marker = marker + self.graph_requests.clear() + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.assertEqual(marker, self.graph_requests[-1]["marker"]) + def test_final_bearer_rejects_bound_or_wrong_audience_results(self): for result in ( {"access_token": "unit-token", "token_type": "mtls_pop"}, From bf763f9f1071a978ff38e53cbd611ca8390cd710 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Mon, 2 Nov 2026 08:24:44 -0500 Subject: [PATCH 11/11] Snapshot nested FIC certificate credential dictionaries Keep caller mutations of a binding credential dictionary from changing the certificate selected by an existing FIC application. Preserve context-object handoff for the pending API decision. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- msal/application.py | 4 +++- tests/test_application.py | 22 ++++++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/msal/application.py b/msal/application.py index 4ca97fa4..09bed76c 100644 --- a/msal/application.py +++ b/msal/application.py @@ -1049,7 +1049,9 @@ def get_leg1_assertion(): "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 = binding_certificate + self._mtls_cert_credential = ( + dict(binding_certificate) + if isinstance(binding_certificate, dict) else binding_certificate) self._mtls_is_fic_leg2 = True elif isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( diff --git a/tests/test_application.py b/tests/test_application.py index 26b654ee..c12ef548 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -3037,6 +3037,28 @@ def test_fic_binding_context_preserves_bearer_and_pop_contracts(self): "cached-assertion-A", self.requests[-1]["client_assertion"]) self.assertEqual(_JWT_POP, self.requests[-1]["client_assertion_type"]) + def test_fic_certificate_dictionary_is_copied_before_caller_mutation(self): + credential = dict(self.credential) + app = ConfidentialClientApplication( + "leg2", authority=self._AUTHORITY, azure_region=False, + client_credential={ + "client_assertion": lambda: "cached-assertion-A", + "mtls_binding_certificate": credential}) + credential.clear() + credential["private_key_pfx_path"] = "missing-replacement.pfx" + with patch( + "msal.mtls._MtlsHttpClient.post", + new=lambda transport, url, **kwargs: + self._post(transport, url, **kwargs)): + result = app.acquire_token_for_client( + ["final-scope"], mtls_proof_of_possession=True) + self.assertEqual("mtls_pop", result["token_type"]) + self.assertEqual( + self.binding.thumbprint_sha256, + result["binding_certificate"].thumbprint_sha256) + self.assertEqual("cached-assertion-A", self.requests[0]["client_assertion"]) + self.assertEqual(_JWT_POP, self.requests[0]["client_assertion_type"]) + def test_fic_sample_rejects_changed_assertion_binding(self): from msal.application import _load_mtls_cert_material