Skip to content

Add FIC two-leg exchange over mTLS Proof-of-Possession (follow-up to #938) #939

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Robbie-Microsoft wants to merge 1 commit into
base: rginsburg/sni-mtls-pop
Choose a base branch
from
Open

Add FIC two-leg exchange over mTLS Proof-of-Possession (follow-up to #938) #939

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/index.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
141 changes: 87 additions & 54 deletions msal/application.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Comment on lines +862 to +865
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(
Expand Down Expand Up @@ -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'.")
Comment on lines 1034 to +1038
with self._mtls_lock:
if self._mtls_pop_cert_material is None:
self._mtls_pop_cert_material = _load_mtls_cert_material(
Expand All @@ -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:
Expand All @@ -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(
Expand Down Expand Up @@ -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``.
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down
1 change: 1 addition & 0 deletions msal/oauth2cli/oauth2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
52 changes: 52 additions & 0 deletions sample/confidential_client_mtls_pop_sample.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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()
43 changes: 43 additions & 0 deletions tests/test_application.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading