diff --git a/docs/index.rst b/docs/index.rst index bc844db4..a57b29ff 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -243,6 +243,20 @@ Notes and requirements: ``os.PathLike`` objects (including ``pathlib.Path``); these replace the default trust roots. ``True`` or ``None`` selects the Requests CA bundle, while ``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. + 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. + 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``. mTLS-bound access tokens are persisted in a separate ``AccessToken_mTLS`` cache bucket. Older SDK versions ignore that bucket and retain their normal Bearer diff --git a/msal/application.py b/msal/application.py index 4b6d9ff5..b739417c 100644 --- a/msal/application.py +++ b/msal/application.py @@ -234,8 +234,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 certificate chain 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. @@ -385,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 @@ -413,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) @@ -428,9 +429,18 @@ def _reinject_key_id(event): (outer or self.on_obtaining_tokens)(event) kwargs["on_obtaining_tokens"] = _reinject_key_id - return _validate_mtls_pop_response( - super(_MtlsClient, self).obtain_token_for_client( - scope=scope, data=data, **kwargs)) + 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_response(response, expected_token_type) def _msal_extension_check(): @@ -606,6 +616,46 @@ 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) + + 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``. + 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": get_leg1_assertion, + "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 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. + .. admonition:: Supporting reading client certificates from PFX files This usage will automatically use SHA-256 thumbprint of the certificate. @@ -963,9 +1013,6 @@ def get_client_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 / @@ -975,13 +1022,39 @@ 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 "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 = 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 + + 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( @@ -1163,10 +1236,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( @@ -1178,13 +1252,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: @@ -1209,16 +1287,24 @@ def _get_mtls_client(self, central_authority): http_cache={}, default_throttle_time=5, ) - # Vanilla SN/I: the TLS certificate alone authenticates the client, so - # this mTLS path sends NO signed client_assertion. (The classic, non-mTLS - # certificate path still signs a private_key_jwt client_assertion; only - # mTLS PoP replaces that assertion with the mutual-TLS handshake.) + 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, so + # this mTLS path sends NO signed client_assertion. (The classic, non-mTLS + # certificate path still signs a private_key_jwt client_assertion; only + # mTLS PoP replaces that assertion with the mutual-TLS handshake.) + 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( @@ -1241,6 +1327,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 @@ -1606,6 +1708,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) @@ -1684,6 +1787,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 @@ -1908,6 +2012,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 if forwarded_client_claims is not None: @@ -1977,6 +2082,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 if forwarded_client_claims is not None: @@ -2325,6 +2431,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, @@ -2379,6 +2486,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: @@ -2978,10 +3086,15 @@ 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``. + 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. @@ -3034,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( @@ -3047,51 +3167,62 @@ def acquire_token_for_client( # and stripped from the wire body by the oauth2 layer. kwargs["data"] = kwargs.get("data", {}) _stash_client_claims(forwarded_client_claims, kwargs["data"]) - if mtls_proof_of_possession: + if mtls_proof_of_possession or self._mtls_is_fic_leg2: if kwargs.get("cert") is not None: raise ValueError( - "mtls_proof_of_possession=True does not support a cert " + "mTLS token requests do not support a cert " "override. MSAL must present the configured binding certificate.") if "post" in kwargs: raise ValueError( - "mtls_proof_of_possession=True does not support a post " - "override. Omit post to use MSAL's built-in mTLS transport.") - # An mTLS transport is required to present the certificate for a - # cert-bound PoP request. + "mTLS token requests do not support a post override. " + "Omit post to use MSAL's built-in mTLS transport.") + # 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: + 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 # _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) - # Attach process-local TLS capability after cache cleanup. The - # private key is never added to the token result or token cache. + 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. result["binding_certificate"] = mtls.MtlsBindingContext( mtls_cert["cert_pem"], mtls_cert["private_key_pem"], @@ -3118,9 +3249,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 @@ -3194,6 +3326,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) _data = kwargs.pop("data", {}) @@ -3266,6 +3399,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/msal/oauth2cli/oauth2.py b/msal/oauth2cli/oauth2.py index 5b38519c..2a49c8ba 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 17c1afc7..88f0c370 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. @@ -25,6 +28,17 @@ RESOURCE_URL optional mTLS resource URL to call with the acquired token AZURE_REGION (optional) e.g. westus3; omission honors MSAL_FORCE_REGION +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 @@ -110,6 +124,79 @@ 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. Requesting PoP produces ``mtls_pop`` and a + binding context; omitting PoP produces Bearer without one. + """ + 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") + + 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.environ["LEG2_CLIENT_ID"], + authority=os.environ["LEG2_AUTHORITY"], + client_credential={ + "client_assertion": get_leg1_assertion, + "mtls_binding_certificate": cert, + }, + azure_region=os.getenv("AZURE_REGION"), + ) + + # 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:") + resource_url = os.getenv("RESOURCE_URL") + 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, + headers={ + "Authorization": + "{} {}".format( + leg2["token_type"], + leg2["access_token"], + ), + }, + timeout=30, + ) + response.raise_for_status() + print(" resource call status: ", response.status_code) + 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/README.md b/tests/README.md index 293073dc..636401fa 100644 --- a/tests/README.md +++ b/tests/README.md @@ -37,6 +37,30 @@ Notes: from the MSID Lab `LabAuth` secret, without a PFX password. Missing, unreadable, or invalid PFX files fail these scenarios; they never count as skipped coverage. +### 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_application.py b/tests/test_application.py index 03c1607a..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 @@ -2287,6 +2288,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) @@ -2610,12 +2612,259 @@ 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 = [] + with patch("msal.mtls._MtlsHttpClient.post", + side_effect=self._capturing_post(captured)): + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + 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")) + self.assertIsInstance( + result.get("binding_certificate"), msal.MtlsBindingContext) + + 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 = [] + with patch("msal.mtls._MtlsHttpClient.post", + side_effect=self._capturing_post( + captured, token_type="Bearer")): + result = app.acquire_token_for_client(["s1"]) + 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_fic_leg2_rejects_post_override(self): + app = ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={ + "client_assertion": "LEG1_TOKEN", + "mtls_binding_certificate": _MTLS_CERT_CRED}) + with self.assertRaisesRegex(ValueError, "post override"): + 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.assertRaisesRegex( + ValueError, "also requires a client_assertion"): + ConfidentialClientApplication( + "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}) + with patch( + "msal.mtls._MtlsHttpClient.post", + side_effect=self._capturing_post(captured)): + 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]) + 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 + # 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") + # 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( "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 d4df481d..15bbbcf0 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -1,5 +1,7 @@ """Unattended certificate E2E scenarios require LAB_APP_CLIENT_CERT_PFX_PATH. +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 or invalid lab credentials fail the required certificate scenarios; legacy interactive and other optional scenarios retain their existing gates. """ @@ -17,9 +19,9 @@ from urllib.parse import urlparse, parse_qs import sys try: - from unittest.mock import patch, ANY + from unittest.mock import patch, ANY, Mock, call, mock_open except: - from mock import patch, ANY + from mock import patch, ANY, Mock, call, mock_open import requests @@ -340,7 +342,43 @@ 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, 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 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": "mtls_pop " + token}, + timeout=30) + 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 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. @@ -429,7 +467,7 @@ def test_client_secret(self): self.assertCacheWorksForApp(result, scope) -class CertificateScenariosTestCase(E2eTestCase): +class CertificateScenariosTestCase(MtlsPopTestCase): # Required unattended scenarios must also run during tagged releases. __unittest_skip__ = False @@ -453,17 +491,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. @@ -473,39 +505,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, 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 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). - """ - with binding_certificate.create_requests_session() as session: - resp = session.get( - self._MTLS_GRAPH_RESOURCE, - headers={"Authorization": "mtls_pop " + token}, - timeout=30) - 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 @@ -562,6 +561,231 @@ def test_credential_x509_output_bearer(self): self.assertCacheWorksForApp(result, self._MTLS_POP_SCOPE) +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 _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 + + 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) + + 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 + 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) + + +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) + + 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): + 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 @classmethod def setUpClass(cls):