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 5b461f30..5f6856a6 100644 --- a/msal/application.py +++ b/msal/application.py @@ -135,8 +135,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. @@ -483,6 +484,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. @@ -834,13 +859,19 @@ 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) + 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( @@ -1003,7 +1034,8 @@ def _get_mtls_pop_cert(self): "mtls_proof_of_possession=True 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( @@ -1015,13 +1047,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: @@ -1045,13 +1081,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( @@ -2755,7 +2798,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``. @@ -2789,60 +2834,47 @@ 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: # 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 @@ -2867,9 +2899,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 da429195..7ed21413 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) @@ -1681,12 +1682,54 @@ 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_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 92af3642..1bc6285f 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -503,6 +503,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