diff --git a/docs/index.rst b/docs/index.rst index bc844db4..6c0b7dbf 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -243,6 +243,24 @@ 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 containing the leg-1 result's + ``binding_certificate``. This reuses the actual certificate snapshot rather + than reopening a PFX path that may have changed between the two legs. The + leg-1 assertion is then sent as a ``jwt-pop`` client assertion over mTLS + using that same certificate. A certificate credential dictionary remains + supported when the caller already guarantees certificate continuity. + A fixed assertion string expires and emits a deprecation warning; the callable + lets the leg-1 application reuse cached tokens and acquire fresh ones as needed. + The target application must have a federated identity credential trusting the + 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 d4b96c7c..3554f3e2 100644 --- a/msal/application.py +++ b/msal/application.py @@ -412,15 +412,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 @@ -440,6 +439,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) @@ -455,9 +455,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(): @@ -633,6 +642,52 @@ 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. Pass the + :class:`~msal.MtlsBindingContext` returned with the leg-1 token + to present that exact certificate during the leg-2 token request. + A non-empty certificate credential dictionary (same shape as the + top-level cert credential) is also supported. Invalid or + incomplete values raise ``ValueError``. + Use a callable to acquire the leg-1 token on demand; the leg-1 + application's cache reuses valid tokens and renews expired ones:: + + leg1 = leg1_app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], + mtls_proof_of_possession=True, + ) + if "access_token" not in leg1: + raise RuntimeError("FIC leg 1 failed") + binding = leg1["binding_certificate"] + + def get_leg1_assertion(): + result = leg1_app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], + mtls_proof_of_possession=True, + ) + if "access_token" not in result: + raise RuntimeError("FIC leg 1 failed") + if result["binding_certificate"].x5c != binding.x5c: + raise RuntimeError("FIC certificate binding changed") + return result["access_token"] + + { + "client_assertion": get_leg1_assertion, + "mtls_binding_certificate": binding, + } + + 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. @@ -999,13 +1054,40 @@ 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, mtls.MtlsBindingContext) and ( + not isinstance(binding_certificate, dict) or not ( + binding_certificate.get("private_key_pfx_path") + or (binding_certificate.get("private_key") + and isinstance( + binding_certificate.get("public_certificate"), str) + and binding_certificate["public_certificate"]))): + raise ValueError( + "mtls_binding_certificate must be an MtlsBindingContext " + "or a non-empty certificate credential dictionary with " + "'private_key_pfx_path', or " + "'private_key' plus a PEM 'public_certificate'.") + if not client_credential.get("client_assertion"): + raise ValueError( + "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 = ( + dict(binding_certificate) + if isinstance(binding_certificate, dict) else binding_certificate) + self._mtls_is_fic_leg2 = True + elif isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( 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) @@ -1201,10 +1283,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'.") material_options = {} if (self._mtls_cert_credential is self.client_credential and isinstance(self.client_credential, dict) @@ -1221,13 +1304,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: @@ -1252,16 +1339,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( @@ -1284,6 +1379,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 @@ -1649,6 +1760,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) @@ -1727,6 +1839,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 @@ -1951,6 +2064,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: @@ -2020,6 +2134,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: @@ -2368,6 +2483,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, @@ -2422,6 +2538,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: @@ -3021,10 +3138,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. @@ -3081,6 +3203,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( @@ -3094,7 +3223,7 @@ 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 (self.client_claims and isinstance(self.client_credential, dict) and not self.client_credential.get("client_assertion") @@ -3105,48 +3234,59 @@ def acquire_token_for_client( "assertion and cannot use constructor client_claims.") 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"], @@ -3173,9 +3313,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 @@ -3249,6 +3390,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", {}) @@ -3321,6 +3463,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..9338dd4a 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,19 @@ 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. Leg 2 reuses the actual certificate binding returned +with the leg-1 assertion, not a second load of the PFX path. Each application +keeps its certificate snapshot; replacing the file does not rotate those apps. + Then run:: python confidential_client_mtls_pop_sample.py @@ -110,6 +126,93 @@ 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 acquire_leg1(): + # Reuse the leg-1 cache, acquiring a fresh token when the old one expires. + leg1 = leg1_app.acquire_token_for_client( + [exchange_scope], mtls_proof_of_possession=True) + if not leg1.get("access_token"): + _print_result_summary(leg1, "FIC leg 1 result:") + raise RuntimeError("FIC leg 1 failed; inspect the result to diagnose.") + if (leg1.get("token_type") != "mtls_pop" or not isinstance( + leg1.get("binding_certificate"), msal.MtlsBindingContext)): + raise RuntimeError("FIC leg 1 must return an mTLS PoP certificate binding.") + return leg1 + + first_leg1 = acquire_leg1() + binding = first_leg1["binding_certificate"] + + def get_leg1_assertion(): + leg1 = acquire_leg1() + current_binding = leg1["binding_certificate"] + if (current_binding.thumbprint_sha256 != binding.thumbprint_sha256 + or current_binding.x5c != binding.x5c): + raise RuntimeError( + "FIC leg 1 certificate binding changed; recreate both applications.") + 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": binding, + }, + 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 23fc9e62..82e17bf8 100644 --- a/tests/README.md +++ b/tests/README.md @@ -37,6 +37,84 @@ 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. + +Both default-transport cells require real token POST evidence for each leg, +the expected global or `westus3` mTLS endpoint, and the same lab certificate. +Leg 2 must invoke the assertion provider and send its exact, unchanged leg-1 +value as a `jwt-pop` client assertion. The exchange assertion may be opaque; +it is not assumed to be a resource JWT. Leg 2 receives the actual +`MtlsBindingContext` returned with the leg-1 assertion rather than reloading +the PFX path. The sample also checks binding continuity when the callable +renews the assertion. Final Storage Bearer tokens must have the Storage +audience. Graph owns its token format: the final Graph token is treated as +opaque and requires matching returned certificate binding plus HTTP 200 from +the mTLS Graph resource using that context, not client-side JWT inspection. +The same token must also fail the shared no-certificate, wrong-certificate, +and Bearer-scheme controls with their exact rejection markers. + +The required FIC inventory is exactly these two methods in +`tests.test_e2e.FicTwoLegTestCase`: + +- `test_credential_fic_output_bearer` +- `test_credential_fic_output_pop` + +They run without skips, including on tagged releases. Custom HTTP-factory +acquisitions are not part of this default-FIC branch. +This branch's required live inventory totals eight methods: the six inherited +`CertificateScenariosTestCase` cells documented below and these two FIC cells. +`tests\test_lab_config.py` checks the exact method sets and proves that all +eight produce credential errors, not skips, when the certificate is missing. + +The FIC mapping reuses the already pinned .NET +`c8b90e9de5bc4974de9928835532faac5c88e2d2` [client-credentials cases][dotnet-cc] +and Go `30f4770ffcbd9c4c51d278cdab6dbda71a0f7337` +[two-leg integration cases][go-fic]: + +| Active reference case(s) | Python FIC method | Required observation | +| --- | --- | --- | +| Go `TestTwoLegFICMtlsPoP_SNI`; .NET `Credential_Fic_Output_Pop_TestAsync` / `RunTwoLegS2sFicBothLegsPopAsync` | `test_credential_fic_output_pop` | Global exchange to Graph; invoked provider, unchanged `jwt-pop` assertion, actual returned-binding handoff, final PoP type/binding, authoritative resource HTTP 200. | +| Go `TestTwoLegFICMtlsBearer_SNI`; .NET `Credential_Fic_Output_Bearer_TestAsync` | `test_credential_fic_output_bearer` | `westus3` exchange to Storage; provider and real `jwt-pop` POST, certificate continuity, Storage Bearer, no output binding or downstream resource mTLS. | + +Run these two live cells with: + +```powershell +python -m pytest -q tests\test_e2e.py::FicTwoLegTestCase +``` + +The offline `tests\test_fic_e2e_helpers.py` checks the two-leg provider/wire +contract, opaque exchange and Graph tokens, strict final binding/type and +Storage audience, Graph rejection failures, and exact assertion handoff. +`tests.test_application.TestFicCertificateHandoff` replaces a PFX with a second +certificate after caching the leg-1 assertion, checks both token and actual +TLS certificate/key continuity, and checks renewal and Bearer/PoP output +contracts without reopening that file. The recorder privacy regression uses +pytest's actual traceback renderer with arguments shown and locals disabled, +and rejects full tokens and truncated token fragments. These offline guards +are not evidence of live acquisition or a custom HTTP factory. + +```powershell +python -m pytest -q tests\test_fic_e2e_helpers.py tests\test_lab_config.py tests\test_e2e.py::FicScenarioTestCase +``` + +[go-fic]: https://github.com/AzureAD/microsoft-authentication-library-for-go/blob/30f4770ffcbd9c4c51d278cdab6dbda71a0f7337/apps/tests/integration/mtls_pop_integration_test.go + ### Exportable-certificate scope and deferred protected keys This feature supports exportable PEM/PFX certificate credentials. A custom HTTP diff --git a/tests/test_application.py b/tests/test_application.py index bd9b977a..cbfbb0b8 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -5,7 +5,10 @@ import logging import os import pickle +import runpy +import shutil import sys +import tempfile import time import warnings from unittest.mock import patch, Mock @@ -2287,6 +2290,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) @@ -2737,6 +2741,256 @@ 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"} + leg1_app = Mock() + leg1_app.acquire_token_for_client.return_value = { + "access_token": "assertion", "token_type": "mtls_pop", + "binding_certificate": msal.MtlsBindingContext( + b"cert", b"key", ["certificate"], "thumbprint"), + } + leg2_app = Mock() + leg2_app.acquire_token_for_client.return_value = result + with patch.dict(os.environ, { + "CLIENT_ID": "leg1", "AUTHORITY": self._AUTHORITY, + "LEG2_CLIENT_ID": "leg2", "LEG2_AUTHORITY": self._AUTHORITY, + "PFX_PATH": "unused.pfx", "SCOPE": "s1", + "RESOURCE_URL": "https://resource.invalid"}), patch( + "msal.ConfidentialClientApplication", + side_effect=[leg1_app, leg2_app]), patch("builtins.print"), patch( + "msal.MtlsBindingContext.create_requests_session") as session: + self.assertIs(result, sample["fic_two_leg_over_mtls_pop"]()) + session.assert_not_called() + + 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. + for binding in ( + _MTLS_CERT_CRED, + msal.MtlsBindingContext(b"cert", b"key", ["certificate"], "thumbprint")): + with self.subTest(binding_type=type(binding).__name__): + with self.assertRaisesRegex( + ValueError, "also requires a client_assertion"): + ConfidentialClientApplication( + "cid", authority=self._AUTHORITY, + client_credential={"mtls_binding_certificate": binding}) + + def test_invalid_binding_certificate_is_rejected(self): + for certificate in ( + {}, 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_government_alias_requests_normalized_global_and_regional_endpoints(self): authority = "https://login.usgovcloudapi.net/my_tenant" endpoint = authority + "/oauth2/v2.0/token" @@ -2770,6 +3024,13 @@ def test_secret_credential_with_flag_raises(self): 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, @@ -2778,6 +3039,226 @@ def test_custom_http_client_with_flag_fails_fast(self): app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True) +@patch(_OIDC_DISCOVERY, new=_MTLS_OIDC_MOCK) +class TestFicCertificateHandoff(unittest.TestCase): + _AUTHORITY = TestMtlsProofOfPossession._AUTHORITY + + @classmethod + def setUpClass(cls): + from datetime import datetime, timedelta, timezone + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.hazmat.primitives.serialization import pkcs12 + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute( + x509.oid.NameOID.COMMON_NAME, "FIC handoff replacement")]) + now = datetime.now(timezone.utc) + cert = (x509.CertificateBuilder() + .subject_name(name).issuer_name(name).public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=1)) + .sign(key, hashes.SHA256())) + cls._replacement_pfx = pkcs12.serialize_key_and_certificates( + b"fic-replacement", key, cert, None, + serialization.BestAvailableEncryption(b"password")) + + def setUp(self): + from msal.application import _load_mtls_cert_material + + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.pfx_path = os.path.join(directory.name, "credential.pfx") + shutil.copyfile(_MTLS_PFX, self.pfx_path) + self.credential = dict(_MTLS_CERT_CRED, private_key_pfx_path=self.pfx_path) + self.material = _load_mtls_cert_material(self.credential) + self.binding = self._binding(self.material) + self.requests = [] + self.sample = runpy.run_path(os.path.join( + os.path.dirname(__file__), os.pardir, "sample", + "confidential_client_mtls_pop_sample.py")) + environment = patch.dict(os.environ, { + "CLIENT_ID": "leg1", "AUTHORITY": self._AUTHORITY, + "LEG2_CLIENT_ID": "leg2", "LEG2_AUTHORITY": self._AUTHORITY, + "PFX_PATH": self.pfx_path, "PFX_PASSPHRASE": "password", + "SCOPE": "final-scope", "RESOURCE_URL": "", + "FIC_EXCHANGE_SCOPE": "api://AzureADTokenExchange/.default", + }) + environment.start() + os.environ.pop("AZURE_REGION", None) + self.addCleanup(environment.stop) + + @staticmethod + def _binding(material): + return msal.MtlsBindingContext( + material["cert_pem"], material["private_key_pem"], + material["x5c"], material["key_id"]) + + def _replace_pfx(self): + with open(self.pfx_path, "wb") as stream: + stream.write(self._replacement_pfx) + + def _post(self, transport, url, data=None, **kwargs): + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + + cert = x509.load_pem_x509_certificate(transport._cert_pem) + key = serialization.load_pem_private_key(transport._key_pem, password=None) + self.assertEqual( + self.material["sha256_thumbprint"], + cert.fingerprint(hashes.SHA256()).hex()) + self.assertEqual( + cert.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo), + key.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo)) + transport._ensure_session(url) # Build the real certificate-bearing TLS context. + self.addCleanup(transport.close) + self.requests.append(dict(data or {})) + return MinimalResponse(status_code=200, text=json.dumps({ + "access_token": "unit-assertion-{}".format(len(self.requests)), + "token_type": (data or {}).get("token_type", "Bearer"), + "expires_in": 3600, + })) + + def test_fic_sample_reuses_cached_assertion_certificate_after_pfx_replacement(self): + factory = msal.ConfidentialClientApplication + apps = [] + pfx_reads = [] + original_open = open + + def capture_open(path, *args, **kwargs): + mode = kwargs.get("mode", args[0] if args else "r") + if path == self.pfx_path and "r" in mode: + pfx_reads.append(True) + return original_open(path, *args, **kwargs) + + def create_app(*args, **kwargs): + app = factory(*args, **kwargs) + apps.append(app) + if len(apps) == 1: + app.acquire_token_for_client( + ["api://AzureADTokenExchange/.default"], + mtls_proof_of_possession=True) + self._replace_pfx() + return app + + with patch("builtins.open", side_effect=capture_open), patch( + "msal.ConfidentialClientApplication", side_effect=create_app), patch( + "msal.mtls._MtlsHttpClient.post", + new=lambda transport, url, **kwargs: + self._post(transport, url, **kwargs)), patch("builtins.print"): + result = self.sample["fic_two_leg_over_mtls_pop"]() + self.assertEqual(["leg1", "leg2"], [ + request["client_id"] for request in self.requests]) + self.assertEqual( + "unit-assertion-1", self.requests[1]["client_assertion"]) + self.assertEqual(_JWT_POP, self.requests[1]["client_assertion_type"]) + binding = apps[1].client_credential["mtls_binding_certificate"] + self.assertIsInstance(binding, msal.MtlsBindingContext) + self.assertEqual( + self.binding.thumbprint_sha256, + result["binding_certificate"].thumbprint_sha256) + for entry in list(apps[0].token_cache.search( + msal.TokenCache.CredentialType.ACCESS_TOKEN, for_removal=True)): + apps[0].token_cache.modify( + msal.TokenCache.CredentialType.ACCESS_TOKEN, entry, + {"expires_on": "0"}) + renewed = apps[1].client_credential["client_assertion"]() + self.assertEqual("unit-assertion-3", renewed) + self.assertEqual(["leg1", "leg2", "leg1"], [ + request["client_id"] for request in self.requests]) + self.assertEqual(1, len(pfx_reads), "PFX must be read once per app.") + + def test_fic_binding_context_preserves_bearer_and_pop_contracts(self): + self._replace_pfx() + with patch( + "msal.application._parse_pfx", + side_effect=AssertionError("FIC must not reload the PFX path.")), patch( + "msal.mtls._MtlsHttpClient.post", + new=lambda transport, url, **kwargs: + self._post(transport, url, **kwargs)): + for pop in (False, True): + with self.subTest(pop=pop): + app = ConfidentialClientApplication( + "leg2", authority=self._AUTHORITY, azure_region=False, + client_credential={ + "client_assertion": lambda: "cached-assertion-A", + "mtls_binding_certificate": self.binding}) + result = app.acquire_token_for_client( + ["final-scope"], mtls_proof_of_possession=pop) + self.assertEqual( + "mtls_pop" if pop else "Bearer", result["token_type"]) + self.assertEqual(pop, "binding_certificate" in result) + self.assertEqual( + "cached-assertion-A", self.requests[-1]["client_assertion"]) + self.assertEqual(_JWT_POP, self.requests[-1]["client_assertion_type"]) + + def test_fic_certificate_dictionary_is_copied_before_caller_mutation(self): + credential = dict(self.credential) + app = ConfidentialClientApplication( + "leg2", authority=self._AUTHORITY, azure_region=False, + client_credential={ + "client_assertion": lambda: "cached-assertion-A", + "mtls_binding_certificate": credential}) + credential.clear() + credential["private_key_pfx_path"] = "missing-replacement.pfx" + with patch( + "msal.mtls._MtlsHttpClient.post", + new=lambda transport, url, **kwargs: + self._post(transport, url, **kwargs)): + result = app.acquire_token_for_client( + ["final-scope"], mtls_proof_of_possession=True) + self.assertEqual("mtls_pop", result["token_type"]) + self.assertEqual( + self.binding.thumbprint_sha256, + result["binding_certificate"].thumbprint_sha256) + self.assertEqual("cached-assertion-A", self.requests[0]["client_assertion"]) + self.assertEqual(_JWT_POP, self.requests[0]["client_assertion_type"]) + + def test_fic_sample_rejects_changed_assertion_binding(self): + from msal.application import _load_mtls_cert_material + + self._replace_pfx() + replacement = self._binding(_load_mtls_cert_material(self.credential)) + leg1 = Mock() + leg1.acquire_token_for_client.side_effect = [{ + "access_token": "unit-assertion", "token_type": "mtls_pop", + "binding_certificate": binding, + } for binding in (self.binding, replacement)] + factory = msal.ConfidentialClientApplication + + def create_app(client_id, **kwargs): + return leg1 if client_id == "leg1" else factory(client_id, **kwargs) + + with patch("msal.ConfidentialClientApplication", side_effect=create_app), patch( + "msal.mtls._MtlsHttpClient.post") as post: + with self.assertRaisesRegex(RuntimeError, "certificate binding changed"): + self.sample["fic_two_leg_over_mtls_pop"]() + post.assert_not_called() + + def test_fic_sample_requires_leg_one_binding_before_constructing_leg_two(self): + for result in ( + {"error": "invalid_client"}, + {"access_token": "unit-assertion", "token_type": "Bearer"}, + {"access_token": "unit-assertion", "token_type": "mtls_pop"}, + {"access_token": "unit-assertion", "token_type": "mtls_pop", + "binding_certificate": {}}): + with self.subTest(fields=tuple(result)): + leg1 = Mock() + leg1.acquire_token_for_client.return_value = result + with patch( + "msal.ConfidentialClientApplication", + return_value=leg1) as factory, patch("builtins.print"): + with self.assertRaises(RuntimeError): + self.sample["fic_two_leg_over_mtls_pop"]() + self.assertEqual(1, factory.call_count) + + @patch(_OIDC_DISCOVERY, new=Mock(return_value={ "authorization_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/authorize", "token_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/token", diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 6c4d6e6c..12d0f332 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. """ @@ -34,7 +36,7 @@ UserSecrets, AppSecrets, LAB_APP_CLIENT_ID, clean_env, ) from tests.mtls_e2e_helpers import ( - GRAPH_REJECTION_MARKERS, JWT_BEARER, + GRAPH_REJECTION_MARKERS, JWT_BEARER, JWT_POP, binding_from_credential, jwt_claims, record_token_requests, unrelated_binding, ) @@ -682,6 +684,19 @@ def test_subject_name_issuer_authentication(self): self._assert_token_result(result, self._MTLS_POP_SCOPE, "Bearer") self._assert_cached_result(app, self._MTLS_POP_SCOPE, result) + # ── SN/I certificate over mTLS Proof-of-Possession ───────────────────────── + # These reuse the same lab SN/I cert as test_subject_name_issuer_authentication + # (it is not CNG-backed, so it works as a TLS client cert on the CI agents). + # NOTE: MSAL must own the TLS transport for mTLS, so unlike the tests above we + # 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. + # 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" + def _acquire_x509(self, scope, pop=False, send=False, region=False, custom=False): credential, binding = self._certificate() if send: @@ -714,6 +729,10 @@ def _acquire_x509(self, scope, pop=False, send=False, region=False, custom=False self.assertTrue(http_client._closed and child._closed) return result + # ── 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 + # unittest, but the matrix names line up 1:1 with the sibling SDKs. def test_credential_x509_output_pop(self): credential, expected_binding = self._certificate() app = self._sni_app(credential) @@ -783,6 +802,102 @@ def test_bearer_then_mtls_pop_cache_isolated(self): def test_mtls_pop_then_bearer_cache_isolated(self): self._assert_cache_isolated(pop_first=True) +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): + __tracebackhide__ = True + credential, binding = self._certificate() + leg1_app = self._sni_app(credential, azure_region=azure_region) + leg1, leg1_requests = self._acquire_recorded( + leg1_app, self._FIC_EXCHANGE_SCOPE, binding, pop=True) + self._assert_token_result( + leg1, self._FIC_EXCHANGE_SCOPE, "mtls_pop", binding) + first_request = self._assert_token_request( + leg1_requests, self._FIC_EXCHANGE_SCOPE, "client_credentials", + region=azure_region) + self.assertEqual("mtls_pop", first_request["token_type"]) + self.assertEqual(200, first_request["status_code"]) + + provider_calls = [] + + def assertion_provider(): + provider_calls.append(True) + return leg1["access_token"] + + leg2_app = self._sni_app( + azure_region=azure_region, + credential={ + "client_assertion": assertion_provider, + "mtls_binding_certificate": leg1["binding_certificate"]}) + result, leg2_requests = self._acquire_recorded( + leg2_app, scope, binding, pop=mtls_proof_of_possession, + expected_assertion=leg1["access_token"]) + self.assertTrue(provider_calls, "The assertion provider must actually be invoked.") + self._assert_token_result( + result, scope, "mtls_pop" if mtls_proof_of_possession else "Bearer", + binding) + self._assert_audience(result, scope) + second_request = self._assert_token_request( + leg2_requests, scope, "client_credentials", region=azure_region, + assertion_type=JWT_POP) + self.assertTrue( + second_request["assertion_matches"], + "The actual jwt-pop POST must carry the leg-1 assertion unchanged.") + self.assertEqual( + "mtls_pop" if mtls_proof_of_possession else None, + second_request["token_type"]) + self.assertEqual(200, second_request["status_code"]) + return leg1["binding_certificate"].thumbprint_sha256, result + + def test_credential_fic_output_bearer(self): + self._acquire_fic_token(self._FIC_BEARER_SCOPE, azure_region="westus3") + + def test_credential_fic_output_pop(self): + _, result = self._acquire_fic_token( + self._MTLS_POP_SCOPE, azure_region=False, + mtls_proof_of_possession=True) + self._assert_graph_token_binding( + result["access_token"], result["binding_certificate"]) + + +class FicScenarioTestCase(unittest.TestCase): + """Keep the live FIC cells fail-closed when lab configuration is missing.""" + + def test_missing_certificate_fails_before_acquisition(self): + scenario = FicTwoLegTestCase("test_credential_fic_output_pop") + for method in ( + scenario.test_credential_fic_output_pop, + scenario.test_credential_fic_output_bearer): + for value in ("", "$(LAB_APP_CLIENT_CERT_PFX_PATH)"): + with self.subTest(method=method.__name__, value=value): + with patch.dict(os.environ, { + "LAB_APP_CLIENT_CERT_PFX_PATH": value}, clear=True), patch( + "msal.ConfidentialClientApplication") as application: + with self.assertRaisesRegex( + EnvironmentError, "LAB_APP_CLIENT_CERT_PFX_PATH"): + method() + application.assert_not_called() + + def test_missing_certificate_is_a_runner_error_even_on_tagged_releases(self): + with patch.object(E2eTestCase, "__unittest_skip__", True, create=True): + 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]) + + class DeviceFlowTestCase(E2eTestCase): # A leaf class so it will be run only once @classmethod def setUpClass(cls): diff --git a/tests/test_fic_e2e_helpers.py b/tests/test_fic_e2e_helpers.py new file mode 100644 index 00000000..f3ddfb88 --- /dev/null +++ b/tests/test_fic_e2e_helpers.py @@ -0,0 +1,276 @@ +"""Offline default-FIC guards, not evidence of live token acquisitions.""" +from unittest.mock import MagicMock, Mock, patch + +import pytest +import requests + +from tests import test_e2e +from tests.mtls_e2e_helpers import ( + GRAPH_REJECTION_MARKERS, JWT_BEARER, JWT_POP, + token_request_summary, unrelated_binding, +) +from tests.test_mtls_e2e_helpers import ( + CertificateFixture, _CREDENTIAL, _ENDPOINT, _jwt, +) + + +class TestDefaultFicCapture(CertificateFixture): + def test_jwt_pop_capture_requires_the_exact_leg_one_assertion(self): + form = dict( + self._signed_form(), client_assertion="unit-fic-assertion", + client_assertion_type=JWT_POP) + for expected, matches in (("unit-fic-assertion", True), ("different", False)): + summary = token_request_summary( + self._response(form=form), self.binding, + requests.adapters.HTTPAdapter(), expected_assertion=expected) + self.assertEqual(matches, summary["assertion_matches"]) + self.assertFalse("unit-fic-assertion" in repr(summary)) + + def test_default_fic_capture_failure_traceback_hides_the_expected_assertion(self): + assertion = _jwt({"aud": "api://AzureADTokenExchange"}) + app = Mock() + app.acquire_token_for_client.side_effect = AssertionError( + "Unit acquisition failed before receiving a response.") + scenario = test_e2e.FicTwoLegTestCase("test_credential_fic_output_pop") + with pytest.raises(AssertionError) as failure: + scenario._acquire_recorded( + app, scenario._MTLS_POP_SCOPE, self.binding, + pop=True, expected_assertion=assertion) + self._assert_private_traceback(failure, assertion) + + +class TestFicScenarioGuards(CertificateFixture): + def setUp(self): + self.leg1 = Mock() + self.leg2 = Mock() + self.leg1.acquire_token_for_client.return_value = { + "access_token": _jwt({ + "aud": "api://AzureADTokenExchange", + "cnf": {"x5t#S256": self.binding.thumbprint_sha256}}), + "token_type": "mtls_pop", "binding_certificate": self.binding, + } + self.leg2.acquire_token_for_client.return_value = { + "access_token": _jwt({ + "aud": "https://graph.microsoft.com", + "cnf": {"x5t#S256": self.binding.thumbprint_sha256}}), + "token_type": "mtls_pop", "binding_certificate": self.binding, + } + self.factory = Mock(side_effect=[self.leg1, self.leg2]) + self.scenario = test_e2e.FicTwoLegTestCase("test_credential_fic_output_pop") + self.invoke_provider = True + self.omit_final_record = False + self.final_record_changes = {} + self.graph_response = Mock(status_code=200, text="") + self.graph_requests = [] + self.invalid_graph_marker = None + patchers = ( + patch("tests.lab_config.get_client_certificate", return_value=_CREDENTIAL), + patch("msal.ConfidentialClientApplication", self.factory), + patch.object(self.scenario, "_acquire_recorded", + side_effect=self._recorded), + patch.object( + self.scenario, "_call_graph_with_mtls_pop_token", + wraps=self.scenario._call_graph_with_mtls_pop_token), + patch("msal.MtlsBindingContext.create_requests_session", + autospec=True, side_effect=self._graph_session_for), + patch("requests.Session", side_effect=self._graph_session_for), + ) + for patcher in patchers: + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(self.scenario.doCleanups) + + def _graph_session_for(self, binding=None): + def get(url, headers, **kwargs): + self.assertEqual(self.scenario._MTLS_GRAPH_RESOURCE, url) + self.assertEqual({"timeout": 30, "allow_redirects": False}, kwargs) + header = headers["Authorization"] + if binding is None: + marker = GRAPH_REJECTION_MARKERS[0] + elif header.startswith("Bearer "): + marker = GRAPH_REJECTION_MARKERS[1] + elif binding.thumbprint_sha256 != self.binding.thumbprint_sha256: + marker = GRAPH_REJECTION_MARKERS[2] + else: + marker = None + self.graph_requests.append({ + "binding": binding, "header": header, "marker": marker}) + if marker is None: + return self.graph_response + return Mock( + status_code=200 if marker == self.invalid_graph_marker else 401, + text=marker) + + session = MagicMock() + session.__enter__.return_value = session + session.get.side_effect = get + return session + + def _recorded(self, app, scope, binding, pop=False, + http_client=None, expected_assertion=None): + config = self.factory.call_args.kwargs + final = app is self.leg2 + assertion = None + if final and self.invoke_provider: + assertion = config["client_credential"]["client_assertion"]() + result = app.acquire_token_for_client(scope, mtls_proof_of_possession=pop) + region = config["azure_region"] + record = { + "url": _ENDPOINT.replace( + "://mtlsauth.", "://{}.mtlsauth.".format(region)) if region else _ENDPOINT, + "fields": frozenset( + ("scope", "client_id", "grant_type") + + (("client_assertion", "client_assertion_type") if final else ()) + + (("token_type",) if pop else ())), + "grant_type": "client_credentials", "scope": frozenset(scope), + "client_assertion_type": JWT_POP if final else None, + "client_assertion_present": final, "signed_client_assertion": False, + "assertion_matches": assertion == expected_assertion, + "token_type": "mtls_pop" if pop else None, + "status_code": 200, "tls_adapter": "_MtlsHTTPAdapter", + } + if final: + record.update(self.final_record_changes) + return result, [] if final and self.omit_final_record else [record] + + def test_global_pairing_provider_wire_type_and_resource_validation(self): + self.scenario.test_credential_fic_output_pop() + calls = self.factory.call_args_list + self.assertEqual(2, len(calls)) + for app_call in calls: + self.assertEqual(self.scenario._SNI_ALLOWLISTED_CLIENT_ID, app_call.args[0]) + self.assertEqual( + self.scenario._SNI_ALLOWLISTED_AUTHORITY, app_call.kwargs["authority"]) + self.assertIs(False, app_call.kwargs["azure_region"]) + self.assertIs( + self.leg1.acquire_token_for_client.return_value["binding_certificate"], + calls[1].kwargs["client_credential"]["mtls_binding_certificate"]) + self.leg1.acquire_token_for_client.assert_called_once_with( + ["api://AzureADTokenExchange/.default"], mtls_proof_of_possession=True) + self.leg2.acquire_token_for_client.assert_called_once_with( + ["https://graph.microsoft.com/.default"], mtls_proof_of_possession=True) + self.scenario._call_graph_with_mtls_pop_token.assert_called_once() + self.assertEqual(4, len(self.graph_requests)) + token = self.leg2.acquire_token_for_client.return_value["access_token"] + self.assertCountEqual( + ["mtls_pop " + token] * 3 + ["Bearer " + token], + [request["header"] for request in self.graph_requests]) + self.assertEqual(2, sum( + request["binding"] is self.binding for request in self.graph_requests)) + self.assertCountEqual( + [None] + list(GRAPH_REJECTION_MARKERS), + [request["marker"] for request in self.graph_requests]) + + def test_regional_bearer_has_storage_audience_and_no_resource_mtls(self): + self.leg2.acquire_token_for_client.return_value = { + "access_token": _jwt({"aud": "https://storage.azure.com"}), + "token_type": "Bearer"} + self.scenario.test_credential_fic_output_bearer() + self.assertTrue(all( + call.kwargs["azure_region"] == "westus3" + for call in self.factory.call_args_list)) + self.leg2.acquire_token_for_client.assert_called_once_with( + ["https://storage.azure.com/.default"], mtls_proof_of_possession=False) + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_leg_one_errors_fail_instead_of_skipping(self): + for method in ( + self.scenario.test_credential_fic_output_pop, + self.scenario.test_credential_fic_output_bearer): + for error in ("token_type_mismatch", "invalid_client", "invalid_grant"): + with self.subTest(error=error, method=method.__name__): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg1.acquire_token_for_client.return_value = {"error": error} + with self.assertRaisesRegex(AssertionError, error): + method() + self.leg2.acquire_token_for_client.assert_not_called() + + def test_leg_two_errors_fail_instead_of_skipping(self): + for method in ( + self.scenario.test_credential_fic_output_pop, + self.scenario.test_credential_fic_output_bearer): + for error in ("invalid_client", "invalid_grant", "temporarily_unavailable"): + with self.subTest(error=error, method=method.__name__): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = {"error": error} + with self.assertRaisesRegex(AssertionError, error): + method() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_federated_assertion_is_opaque_but_still_requires_certificate_binding(self): + self.leg1.acquire_token_for_client.return_value["access_token"] = "opaque" + self.scenario.test_credential_fic_output_pop() + self.leg2.acquire_token_for_client.reset_mock() + for binding in (None, {}, unrelated_binding()): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg1.acquire_token_for_client.return_value["binding_certificate"] = binding + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.leg2.acquire_token_for_client.assert_not_called() + + def test_final_pop_requires_type_and_matching_certificate_binding(self): + original = self.leg2.acquire_token_for_client.return_value + for changes in ( + {"token_type": "Bearer"}, {"binding_certificate": None}, + {"binding_certificate": {}}, + {"binding_certificate": unrelated_binding()}): + with self.subTest(fields=tuple(changes)): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = dict(original, **changes) + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_graph_format_is_opaque_and_resource_acceptance_is_authoritative(self): + for token in ( + "opaque-graph-token", + _jwt({"aud": "https://wrong.example"}), + _jwt({"cnf": {"x5t#S256": "different-certificate"}})): + for status in (200, 401, 403, 302, 500): + with self.subTest(format="jwt" if "." in token else "opaque", + status=status): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value["access_token"] = token + self.graph_response.status_code = status + self.graph_requests.clear() + if status == 200: + self.scenario.test_credential_fic_output_pop() + else: + with self.assertRaisesRegex(AssertionError, "Graph resource"): + self.scenario.test_credential_fic_output_pop() + self.assertEqual("mtls_pop " + token, self.graph_requests[0]["header"]) + + def test_fic_graph_requires_all_same_token_rejection_controls(self): + for marker in GRAPH_REJECTION_MARKERS: + with self.subTest(marker=marker): + self.factory.side_effect = [self.leg1, self.leg2] + self.invalid_graph_marker = marker + self.graph_requests.clear() + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.assertEqual(marker, self.graph_requests[-1]["marker"]) + + def test_final_bearer_rejects_bound_or_wrong_audience_results(self): + for result in ( + {"access_token": "unit-token", "token_type": "mtls_pop"}, + {"access_token": _jwt({"aud": "https://storage.azure.com"}), + "token_type": "Bearer", "binding_certificate": {}}, + {"access_token": _jwt({"aud": "https://wrong.example"}), + "token_type": "Bearer"}): + self.factory.side_effect = [self.leg1, self.leg2] + self.leg2.acquire_token_for_client.return_value = result + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_bearer() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() + + def test_fic_requires_provider_invocation_and_actual_wire_evidence(self): + for invoke, omit, changes in ( + (False, False, {}), (True, True, {}), + (True, False, {"client_assertion_type": JWT_BEARER}), + (True, False, {"assertion_matches": False})): + self.factory.side_effect = [self.leg1, self.leg2] + self.invoke_provider, self.omit_final_record = invoke, omit + self.final_record_changes = changes + with self.assertRaises(AssertionError): + self.scenario.test_credential_fic_output_pop() + self.scenario._call_graph_with_mtls_pop_token.assert_not_called() diff --git a/tests/test_lab_config.py b/tests/test_lab_config.py index dc7301e8..3ee41d2a 100644 --- a/tests/test_lab_config.py +++ b/tests/test_lab_config.py @@ -58,6 +58,10 @@ class TestRequiredCertificateConfiguration(unittest.TestCase): "test_bearer_then_mtls_pop_cache_isolated", "test_mtls_pop_then_bearer_cache_isolated", }, + test_e2e.FicTwoLegTestCase: { + "test_credential_fic_output_pop", + "test_credential_fic_output_bearer", + }, } def test_required_scenario_inventory_is_complete(self): @@ -117,8 +121,8 @@ def test_required_scenarios_fail_without_credentials_instead_of_skipping(self): unittest.defaultTestLoader.loadTestsFromTestCase(case) for case in self._REQUIRED_SCENARIOS) result = unittest.TextTestRunner(stream=io.StringIO()).run(suite) - self.assertEqual(6, result.testsRun) - self.assertEqual(6, len(result.errors)) + self.assertEqual(8, result.testsRun) + self.assertEqual(8, len(result.errors)) self.assertEqual([], result.skipped) self.assertEqual([], result.failures) for _, traceback in result.errors: