From 28b0126a5193fbdd86fcb1a651490a0e985425c0 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 11:28:10 -0400 Subject: [PATCH 01/11] Add certificate-aware mTLS HTTP client extensibility Allow custom HTTP pipelines to create isolated certificate-bearing token clients using MtlsBindingContext. Preserve networking policy, pooling and Requests TLS compatibility, enforce observable transport contracts, and document ownership and trust boundaries. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/index.rst | 112 ++- msal/application.py | 154 ++-- msal/mtls.py | 100 ++- ...idential_client_mtls_http_client_sample.py | 114 +++ sample/confidential_client_mtls_pop_sample.py | 4 +- tests/test_mtls_http_client.py | 674 ++++++++++++++++++ 6 files changed, 1072 insertions(+), 86 deletions(-) create mode 100644 sample/confidential_client_mtls_http_client_sample.py create mode 100644 tests/test_mtls_http_client.py diff --git a/docs/index.rst b/docs/index.rst index e450c78f..1b49f116 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -211,9 +211,11 @@ certificate credential and pass ``mtls_proof_of_possession=True`` to Notes and requirements: * The ``authority`` must be **tenanted** (not ``/common`` or ``/organizations``). -* MSAL must own the TLS transport, so a custom ``http_client`` is not supported - together with ``mtls_proof_of_possession=True``. - A per-request ``post`` override is also rejected. +* The default HTTP transport handles mTLS automatically. A custom + ``http_client`` must explicitly implement the certificate-aware factory + described below; legacy custom clients fail with ``ValueError`` for mTLS. + Per-request ``post`` and ``cert`` overrides and redirects are rejected for + mTLS token requests, including when a token could otherwise come from cache. * If the provider returns a different token type, MSAL returns ``token_type_mismatch`` without caching the rejected response. * ``azure_region`` is **optional**: omit it to use the global mTLS endpoint @@ -235,6 +237,110 @@ Notes and requirements: .. autoclass:: msal.MtlsBindingContext :members: +Custom HTTP pipelines for mTLS token requests +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +An application that does not supply ``http_client`` needs no new configuration. +MSAL owns its certificate-bearing Requests transport, separate from discovery +and ordinary token traffic. + +A custom HTTP client opts in by implementing this additional method:: + + def create_mtls_client(self, binding_context): + # binding_context is an MSAL-created msal.MtlsBindingContext. + # Return a NEW pipeline with the same middleware and network policy, + # but isolated certificate-bearing connections. + ... + +There is no marker class to inherit and no additional constructor parameter. +The ordinary client's ``get`` and ``post`` still handle discovery and non-mTLS +requests. On the first mTLS token network request, MSAL calls the factory with +the application's selected certificate. The returned object must have a +callable ``post(url, **kwargs)`` using the normal MSAL HTTP response contract +(``status_code``, ``text``, optional ``headers``, and ``raise_for_status()``). +It must accept and honor ``allow_redirects=False``. Factory and request +exceptions propagate; MSAL never falls back to the ordinary client. + +For Requests-based pipelines, mount +``binding_context.create_requests_adapter(...)`` on a **new** session. This +helper accepts ``HTTPAdapter`` options for retry and pooling policy, retains +MSAL's Requests-version compatibility handling, and selects separate +certificate-bearing TLS contexts and pools using the effective Requests +``verify`` value. Session/request CA paths, ``REQUESTS_CA_BUNDLE`` and +``CURL_CA_BUNDLE`` (when ``session.trust_env`` allows them), and HTTPS proxies +retain their Requests semantics. For other HTTP libraries, +``binding_context.create_ssl_context(verify=...)`` creates the certificate-bearing +``ssl.SSLContext``; the implementer must resolve its own library's effective +server verification and proxy settings before using it. + +For example, inside a Requests-based factory:: + + session = MyPolicySession() # A fresh session retaining your middleware + session.verify = self.ca_bundle + session.trust_env = self.trust_env + session.proxies.update(self.proxies) + session.mount("https://", binding_context.create_requests_adapter( + max_retries=self.retries, + pool_connections=self.pool_connections, + pool_maxsize=self.pool_maxsize, + pool_block=True, + )) + # Set the pipeline's timeout, hooks and other policy as for ordinary traffic. + # Keep session alive until all applications using this factory have finished. + return session + +The runnable ``sample/confidential_client_mtls_http_client_sample.py`` shows a +factory that retains response middleware, a header policy, CA/proxy/environment +settings, timeouts and pooling, and closes its derived pipelines explicitly. +It uses the same certificate and environment configuration as the default +mTLS-PoP sample. + +Trust, isolation and lifetime contract +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +* Return a fresh pipeline for every factory call. Do not mutate or return the + ordinary client, its shared transport, or a previously returned pipeline. + Never share certificate-bearing connection pools, adapters, mutable TLS + contexts or TLS session caches with another certificate or ordinary traffic. + Shared immutable policy is fine. +* Preserve all required middleware and networking policy explicitly: MSAL cannot + clone an opaque client. Constructor ``verify``, ``proxies`` and ``timeout`` + options still do not override a custom HTTP client's settings. +* MSAL serializes factory creation within each application and caches one + returned pipeline for that application's certificate. Global and regional + token endpoints reuse it. Separate applications invoke the factory separately, + even when they share an HTTP factory or caches. The factory must be safe to + call concurrently from different applications, and its returned pipelines + must support the application's concurrency requirements. +* Certificate credentials are fixed for the application's lifetime. To rotate, + create a new application and pipeline, stop in-flight work using the old + application, then release its pipelines. Do not mutate a supplied binding + context or a TLS context already used by a connection pool. +* The custom factory owns and must retain/close every client it creates, + including partial constructions when its own code raises. MSAL does not + close caller-owned clients or the ordinary pipeline. Close them only after + all applications and in-flight requests using them have finished. +* MSAL rejects a missing/non-callable factory, a return value without callable + ``post``, returning the ordinary client itself, and returning a client already + assigned to another live application. This identity check does not keep + applications alive or require custom clients to support weak references. + MSAL restricts its token calls to the configured HTTPS endpoint, passes + ``allow_redirects=False``, + and rejects redirect responses or observable redirect history/final-endpoint + changes (excluding query parameters). These checks cannot undo a redirect + already followed by a faulty custom implementation. +* This is a **trusted implementation contract**, not proof that arbitrary code + uses the supplied certificate. MSAL cannot inspect nested transports, detect + all shared pools (or all reuse after an application is collected), enforce an + opaque client's verification/proxy policy, or prevent custom middleware from + ignoring TLS or redirect instructions. The implementer is responsible for + those behaviors; do not opt in unless the complete pipeline honors them. + +The factory receives a process-local certificate capability, not a successful +token result. ``binding_certificate`` remains present only on a final +``mtls_pop`` result, never on an ordinary Bearer result. This extension does not +change downstream resource TLS or the requirements for mTLS PoP. + Exceptions ---------- diff --git a/msal/application.py b/msal/application.py index 13400341..5347f1bd 100644 --- a/msal/application.py +++ b/msal/application.py @@ -673,6 +673,23 @@ def get_client_assertion(): If you are providing your own http_client, it will be your http_client's duty to decide whether to perform retry. + To support mTLS token requests, it must also implement + ``create_mtls_client(binding_context)``. MSAL passes an + :py:class:`msal.MtlsBindingContext`; the factory returns a new, + dedicated client with a callable ``post(url, **kwargs)`` that uses + that certificate's TLS configuration and honors + ``allow_redirects=False``. The factory must preserve its networking + policy and middleware, isolate connection pools and TLS sessions + from ordinary traffic and other certificates, and own cleanup of + its returned clients. MSAL reuses one returned client per + application, including regional token endpoints. Creating a new + application is required to rotate the certificate. + + MSAL validates the returned interface and rejects reuse of the + ordinary client or a client assigned to another live application, + but cannot prove that arbitrary custom code uses the supplied TLS + configuration or disables redirects. + :param verify: (optional) It will be passed to the `verify parameter in the underlying requests library @@ -872,11 +889,8 @@ def get_client_assertion(): 'Invalid exclude_scopes={}. You can not opt out "openid" scope'.format( repr(exclude_scopes))) - # Retained for the mTLS PoP transport, which needs its own dedicated - # requests.Session carrying the client certificate (see msal/mtls.py). - # A user-supplied http_client cannot perform the mTLS handshake, so - # requesting mtls_proof_of_possession=True with one fails fast. - self._http_client_is_custom = bool(http_client) + # A custom pipeline must explicitly opt in to certificate-aware TLS. + self._custom_http_client = http_client if http_client else None self._http_client_config = { "verify": verify, "proxies": proxies, "timeout": timeout} self._http_cache = http_cache # Reused by the lazily-built mTLS client @@ -958,6 +972,7 @@ def get_client_assertion(): self._mtls_pop_cert_material = None # Cached parse of the cert below self._mtls_client = None # Lazily built global mTLS client self._mtls_regional_client = None # Lazily built regional mTLS client + self._mtls_http_client = None # Shared by this certificate's endpoints self._mtls_lock = Lock() if isinstance(client_credential, dict) and not client_credential.get( "client_assertion") and ( @@ -1153,6 +1168,17 @@ def _get_mtls_pop_cert(self): self._mtls_cert_credential) return self._mtls_pop_cert_material + def _get_mtls_http_client_factory(self): + if self._custom_http_client is None: + return None + factory = getattr(self._custom_http_client, "create_mtls_client", None) + if not callable(factory): + raise ValueError( + "mTLS requires a custom http_client to implement " + "create_mtls_client(binding_context). Omit http_client to use " + "MSAL's built-in mTLS transport.") + return factory + def _get_mtls_client(self, central_authority): """Lazily build (and cache) the mTLS PoP client for ``central_authority``. @@ -1160,17 +1186,8 @@ def _get_mtls_client(self, central_authority): (region honored when configured, global otherwise), and the client presents the configured certificate over mutual-TLS. """ - 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 built-in mTLS transport.") + factory = self._get_mtls_http_client_factory() region = self._compute_region_to_use() - with self._mtls_lock: - cached = self._mtls_regional_client if region else self._mtls_client - if cached is not None: - return cached cert = self._get_mtls_pop_cert() # May raise ValueError (no cert configured) token_endpoint = mtls.transform_token_endpoint( # login.* -> mtlsauth.* central_authority.token_endpoint, central_authority.instance, region) @@ -1181,44 +1198,53 @@ def _get_mtls_client(self, central_authority): "device_authorization_endpoint": central_authority.device_authorization_endpoint, } - http_client = ThrottledHttpClient( - mtls._MtlsHttpClient( - cert["cert_pem"], cert["private_key_pem"], - **self._http_client_config), - # The TLS credential is absent from HTTP cache keys. - 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.) - client = _MtlsClient( - configuration, - self.client_id, - http_client=http_client, - default_headers=self._default_client_headers(), - default_body={"client_info": 1}, - # 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( - event, environment=central_authority.instance)), - on_removing_rt=self.token_cache.remove_rt, - on_updating_rt=self.token_cache.update_rt) with self._mtls_lock: - # Double-check under the lock: another thread may have built and - # cached a client while we were building ours (we released the lock - # to build). If so, keep the already-cached instance and let ours be - # discarded, so every caller shares one client instead of orphaning - # a redundant build. + cached = self._mtls_regional_client if region else self._mtls_client + if cached is not None: + return cached + if self._mtls_http_client is None: + if factory is None: + http_client = mtls._MtlsHttpClient( + cert["cert_pem"], cert["private_key_pem"], + **self._http_client_config) + else: + binding = mtls.MtlsBindingContext( + cert["cert_pem"], cert["private_key_pem"], + cert["x5c"], cert["key_id"]) + http_client = factory(binding) + if (http_client is self._custom_http_client + or http_client is self.http_client + or http_client is self.http_client.http_client): + raise ValueError( + "create_mtls_client must return a new, dedicated " + "client, not the ordinary http_client.") + if not callable(getattr(http_client, "post", None)): + raise TypeError( + "create_mtls_client must return a client with " + "a callable post(url, **kwargs).") + mtls._claim_mtls_http_client(http_client, self) + self._mtls_http_client = http_client + http_client = ThrottledHttpClient( + mtls._MtlsTokenHttpClient(self._mtls_http_client, token_endpoint), + # The TLS credential is absent from HTTP cache keys. + http_cache={}, + default_throttle_time=5, + ) + # The TLS certificate authenticates this path; no client assertion. + client = _MtlsClient( + configuration, + self.client_id, + http_client=http_client, + default_headers=self._default_client_headers(), + default_body={"client_info": 1}, + on_obtaining_tokens=lambda event: self.token_cache.add(dict( + event, environment=central_authority.instance)), + on_removing_rt=self.token_cache.remove_rt, + on_updating_rt=self.token_cache.update_rt) if region: - if self._mtls_regional_client is None: - self._mtls_regional_client = client - client = self._mtls_regional_client + self._mtls_regional_client = client else: - if self._mtls_client is None: - self._mtls_client = client - client = self._mtls_client + self._mtls_client = client return client def _build_client(self, client_credential, authority, skip_regional_client=False): @@ -2958,12 +2984,13 @@ 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 - ``/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``. - 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. + ``/common`` or ``/organizations``); and a custom ``http_client`` + must implement ``create_mtls_client(binding_context)`` as described + in :py:class:`msal.ClientApplication`. Any of these unmet raises + ``ValueError``. The default transport needs no extra configuration. + ``post`` and ``cert`` overrides and redirects are rejected. If the + provider returns another token type, the result is + ``token_type_mismatch`` and the rejected response is not cached. On success the result also contains an :py:class:`msal.MtlsBindingContext` as ``binding_certificate``. It @@ -3026,19 +3053,8 @@ def acquire_token_for_client( kwargs["data"] = kwargs.get("data", {}) _stash_client_claims(forwarded_client_claims, kwargs["data"]) if mtls_proof_of_possession: - 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. - 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 " - "built-in mTLS transport.") + mtls._validate_mtls_request(kwargs) + self._get_mtls_http_client_factory() if self.authority.tenant.lower() in ("common", "organizations"): raise ValueError( "mtls_proof_of_possession=True requires a tenanted authority. " diff --git a/msal/mtls.py b/msal/mtls.py index 6710ae0e..11aaafcc 100644 --- a/msal/mtls.py +++ b/msal/mtls.py @@ -13,6 +13,7 @@ """ import logging import threading +import weakref try: from collections.abc import Mapping except ImportError: # Python 2 @@ -26,6 +27,8 @@ logger = logging.getLogger(__name__) _MTLS_POP_DOC_LINK = "https://aka.ms/msal-net-pop" +_MTLS_HTTP_CLIENT_OWNERS = weakref.WeakValueDictionary() +_MTLS_HTTP_CLIENT_OWNERS_LOCK = threading.Lock() # The global public mTLS host. The three public "login.*" hosts all normalize to # this single global endpoint (ESTSR provides regional failover), matching @@ -97,9 +100,10 @@ class MtlsBindingContext(Mapping): """The certificate binding and TLS helpers for an mTLS-bound access token. Instances are created by MSAL and returned as ``binding_certificate`` in a - successful token result. The certificate metadata remains available through - mapping access for compatibility, while the private key is exposed only - through TLS contexts and sessions created by this object. + successful mTLS-PoP token result. MSAL also supplies an instance to a custom + HTTP client's ``create_mtls_client(binding_context)`` factory before sending + mTLS token requests. The certificate metadata remains available through + mapping access, while the private key is exposed only through TLS helpers. This object is process-local and must not be serialized or persisted. """ @@ -133,6 +137,23 @@ def create_ssl_context(self, verify=True): """ return _create_ssl_context(self.__cert_pem, self.__key_pem, verify) + def create_requests_adapter(self, **kwargs): + """Create a certificate-bearing ``requests.adapters.HTTPAdapter``. + + Mount this on a new, dedicated session, never on a session used for + ordinary requests or another certificate. ``kwargs`` are HTTPAdapter + options, such as ``max_retries``, ``pool_connections``, ``pool_maxsize`` + and ``pool_block``. By default, one connection retry is enabled. + + Server verification follows the effective Requests ``verify`` setting, + including environment, session and request CA configuration. Separate + TLS contexts and pools are used for different verification settings. + A ``cert`` override is rejected to preserve the selected certificate. + The caller owns the adapter and its connection pools. + """ + return _make_mtls_adapter( + ssl_context_factory=self.create_ssl_context, **kwargs) + def create_requests_session(self, verify=True, proxies=None): """Create a ``requests.Session`` that presents the binding certificate. @@ -145,9 +166,7 @@ def create_requests_session(self, verify=True, proxies=None): session.verify = effective_verify if proxies: session.proxies.update(proxies) - session.mount( - "https://", - _make_mtls_adapter(ssl_context_factory=self.create_ssl_context)) + session.mount("https://", self.create_requests_adapter()) return session def __getitem__(self, key): @@ -178,9 +197,8 @@ class _MtlsHttpClient(object): """A minimal http client (``post``/``get``/``close``) whose HTTPS connections present a client certificate for mutual-TLS. - MSAL owns this transport. A caller's plain custom ``http_client`` cannot - perform the mTLS handshake, which is why requesting mTLS PoP with a - non-mTLS-capable custom transport fails fast (see application.py). + MSAL owns this default transport. Custom pipelines instead opt in through + ``http_client.create_mtls_client(binding_context)`` (see application.py). The ``ssl.SSLContext`` (and its temp key file) is built lazily on first use, so Bearer-only certificate apps never pay the cost nor touch the disk. @@ -240,7 +258,60 @@ def close(self): session.close() -def _make_mtls_adapter(ssl_context=None, ssl_context_factory=None): +def _claim_mtls_http_client(http_client, application): + # Values are weak: tracking direct reuse must not keep applications alive. + with _MTLS_HTTP_CLIENT_OWNERS_LOCK: + owner = _MTLS_HTTP_CLIENT_OWNERS.get(id(http_client)) + if owner is not None and owner is not application: + raise ValueError( + "create_mtls_client must return a new, dedicated client for " + "each application, not a client already used by another application.") + _MTLS_HTTP_CLIENT_OWNERS[id(http_client)] = application + + +def _validate_mtls_request(kwargs): + if "post" in kwargs: + raise ValueError( + "mtls_proof_of_possession=True does not support a post override. " + "Use the configured certificate-aware HTTP transport.") + if kwargs.get("cert") is not None: + raise ValueError( + "mTLS token requests do not support a cert override. " + "Use the application's certificate credential.") + if kwargs.get("allow_redirects", False) is not False: + raise ValueError("mTLS token requests do not support redirects.") + + +class _MtlsTokenHttpClient(object): + """Restrict a certificate-scoped pipeline to its token endpoint.""" + + def __init__(self, http_client, token_endpoint): + if urlparse(token_endpoint).scheme != "https": + raise ValueError("The mTLS token endpoint must use HTTPS.") + self._http_client = http_client + self._token_endpoint = token_endpoint + + def post(self, url, **kwargs): + if url != self._token_endpoint: + raise ValueError( + "The mTLS HTTP transport can only call its token endpoint.") + _validate_mtls_request(kwargs) + kwargs["allow_redirects"] = False + response = self._http_client.post(url, **kwargs) + response_url = getattr(response, "url", None) + # Request params may change the query without a redirect. + endpoint_changed = response_url is not None and ( + urlparse(response_url)._replace(query="", fragment="") + != urlparse(url)._replace(query="", fragment="")) + if (300 <= response.status_code < 400 + or getattr(response, "history", None) + or endpoint_changed): + raise ValueError("The mTLS token endpoint must not redirect.") + return response + + +def _make_mtls_adapter( + ssl_context=None, ssl_context_factory=None, **adapter_kwargs): """Return a ``requests`` HTTPAdapter that injects ``ssl_context`` (with the client certificate loaded) into every connection pool it creates.""" from requests.adapters import HTTPAdapter # Lazy import @@ -250,16 +321,21 @@ def __init__(self, ssl_context): self._ssl_context = ssl_context self._adapters = {} self._adapter_lock = threading.Lock() - super(_MtlsHTTPAdapter, self).__init__(max_retries=1) + options = dict(adapter_kwargs) + options.setdefault("max_retries", 1) + super(_MtlsHTTPAdapter, self).__init__(**options) def send(self, request, **kwargs): + if kwargs.get("cert") is not None: + raise ValueError( + "The mTLS adapter does not support a cert override.") if ssl_context_factory is None: return super(_MtlsHTTPAdapter, self).send(request, **kwargs) verify = kwargs.get("verify", True) with self._adapter_lock: if verify not in self._adapters: self._adapters[verify] = _make_mtls_adapter( - ssl_context_factory(verify)) + ssl_context_factory(verify), **adapter_kwargs) adapter = self._adapters[verify] return adapter.send(request, **kwargs) diff --git a/sample/confidential_client_mtls_http_client_sample.py b/sample/confidential_client_mtls_http_client_sample.py new file mode 100644 index 00000000..233fef29 --- /dev/null +++ b/sample/confidential_client_mtls_http_client_sample.py @@ -0,0 +1,114 @@ +"""A custom HTTP pipeline for MSAL's mTLS token transport. + +Use the CLIENT_ID, AUTHORITY, PFX_PATH, optional PFX_PASSPHRASE, SCOPE and +optional AZURE_REGION environment variables from the default mTLS-PoP sample. +REQUESTS_CA_BUNDLE, CURL_CA_BUNDLE and proxy environment variables retain their +Requests behavior. Run: + + python sample\\confidential_client_mtls_http_client_sample.py + +The factory owns both ordinary and derived sessions. Quiesce all applications +and in-flight work before leaving its context manager. No token or private-key +material is logged. +""" +import os +import threading + +import requests + +import msal + + +class PolicyHttpClient: + """A small example; apply your production policy in both pipeline paths.""" + + def __init__( + self, verify=True, proxies=None, timeout=30, trust_env=True, + binding_context=None): + self._config = { + "verify": verify, "proxies": dict(proxies or {}), + "timeout": timeout, "trust_env": trust_env} + self._lock = threading.Lock() + self._closed = False + self._children = [] + self.response_count = 0 + self._session = requests.Session() + self._session.verify = verify + self._session.trust_env = trust_env + self._session.proxies.update(self._config["proxies"]) + self._session.headers["X-Msal-Pipeline"] = "custom-policy" + self._session.hooks["response"].append(self._record_response) + options = { + "max_retries": 1, "pool_connections": 4, + "pool_maxsize": 8, "pool_block": True} + self._session.mount("https://", + binding_context.create_requests_adapter(**options) + if binding_context is not None + else requests.adapters.HTTPAdapter(**options)) + + def _record_response(self, response, **kwargs): + with self._lock: + self.response_count += 1 + return response + + def _check_open(self): + if self._closed: + raise RuntimeError("The HTTP pipeline has been closed.") + + def create_mtls_client(self, binding_context): + with self._lock: + self._check_open() + child = PolicyHttpClient( + binding_context=binding_context, **self._config) + self._children.append(child) + return child + + def get(self, url, **kwargs): + self._check_open() + kwargs.setdefault("timeout", self._config["timeout"]) + return self._session.get(url, **kwargs) + + def post(self, url, **kwargs): + self._check_open() + kwargs.setdefault("timeout", self._config["timeout"]) + return self._session.post(url, **kwargs) + + def close(self): + with self._lock: + if not self._closed: + self._closed = True + for child in self._children: + child.close() + self._session.close() + + def __enter__(self): + self._check_open() + return self + + def __exit__(self, *args): + self.close() + + +def main(): + credential = { + "private_key_pfx_path": os.environ["PFX_PATH"], + "passphrase": os.getenv("PFX_PASSPHRASE"), + "public_certificate": True, + } + with PolicyHttpClient() as http_client: + app = msal.ConfidentialClientApplication( + os.environ["CLIENT_ID"], + authority=os.environ["AUTHORITY"], + client_credential=credential, + http_client=http_client, + azure_region=os.getenv("AZURE_REGION"), + ) + result = app.acquire_token_for_client( + os.environ["SCOPE"].split(), mtls_proof_of_possession=True) + print("Acquired mTLS-PoP token:", result.get("token_type") == "mtls_pop") + if "error" in result: + print("Error code:", result["error"]) + + +if __name__ == "__main__": + main() diff --git a/sample/confidential_client_mtls_pop_sample.py b/sample/confidential_client_mtls_pop_sample.py index 4918559c..f0d95a41 100644 --- a/sample/confidential_client_mtls_pop_sample.py +++ b/sample/confidential_client_mtls_pop_sample.py @@ -83,8 +83,8 @@ def vanilla_sni_mtls_pop(): os.environ["CLIENT_ID"], authority=os.environ["AUTHORITY"], # MUST be tenanted client_credential=_build_cert_credential(), - # Note: do NOT pass a custom http_client here. MSAL must own the - # transport to present the client certificate during the TLS handshake. + # The default transport handles mTLS automatically. For a custom + # pipeline, see confidential_client_mtls_http_client_sample.py. azure_region=os.getenv("AZURE_REGION"), # optional; None -> global endpoint ) diff --git a/tests/test_mtls_http_client.py b/tests/test_mtls_http_client.py new file mode 100644 index 00000000..bea0bb3c --- /dev/null +++ b/tests/test_mtls_http_client.py @@ -0,0 +1,674 @@ +import ipaddress +import json +import os +import select +import socket +import ssl +import tempfile +import threading +import time +import unittest +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timedelta, timezone +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import SimpleNamespace +from unittest.mock import Mock, patch +from urllib.parse import parse_qs + +import requests + +import msal +from msal import mtls +from sample.confidential_client_mtls_http_client_sample import PolicyHttpClient +from tests.http_client import MinimalResponse + + +_AUTHORITY = "https://login.microsoftonline.com/tenant" +_METADATA = { + "authorization_endpoint": _AUTHORITY + "/oauth2/v2.0/authorize", + "token_endpoint": _AUTHORITY + "/oauth2/v2.0/token", + "issuer": _AUTHORITY + "/v2.0", +} +_CREDENTIAL = { + "private_key_pfx_path": os.path.join( + os.path.dirname(__file__), "certificate-with-password.pfx"), + "passphrase": "password", + "public_certificate": True, +} +_TOKEN = '{"access_token": "at", "token_type": "mtls_pop", "expires_in": 3600}' +_TOKEN_PATH = "/tenant/oauth2/v2.0/token" +_ORDINARY_PATH = "/tenant/oauth2/v2.0/ordinary" + + +@patch("msal.authority.tenant_discovery", return_value=_METADATA) +class TestMtlsHttpClientFactory(unittest.TestCase): + def setUp(self): + self.pipeline = Mock(spec=["post"]) + self.pipeline.post.return_value = MinimalResponse( + status_code=200, text=_TOKEN) + self.ordinary = Mock(spec=["get", "post", "close", "create_mtls_client"]) + self.ordinary.create_mtls_client.return_value = self.pipeline + self.ordinary.post.return_value = MinimalResponse( + status_code=200, + text='{"access_token": "ordinary", "token_type": "Bearer"}') + + def _app(self, http_client=None, **kwargs): + return msal.ConfidentialClientApplication( + "cid", authority=_AUTHORITY, client_credential=_CREDENTIAL, + http_client=self.ordinary if http_client is None else http_client, + azure_region=False, instance_discovery=False, **kwargs) + + def test_factory_receives_selected_binding_and_request_options(self, discovery): + app = self._app(verify=False, timeout=999, + proxies={"https": "http://unused.example"}) + self.ordinary.create_mtls_client.assert_not_called() + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, timeout=7, + verify="request-ca.pem", proxies={"https": "http://request.example"}) + binding = self.ordinary.create_mtls_client.call_args.args[0] + self.assertIsInstance(binding, msal.MtlsBindingContext) + self.assertEqual(binding.thumbprint_sha256, + result["binding_certificate"].thumbprint_sha256) + self.assertNotIn("PRIVATE KEY", repr(binding)) + request = self.pipeline.post.call_args + self.assertEqual( + "https://mtlsauth.microsoft.com/tenant/oauth2/v2.0/token", + request.args[0]) + self.assertEqual(7, request.kwargs["timeout"]) + self.assertEqual("request-ca.pem", request.kwargs["verify"]) + self.assertEqual({"https": "http://request.example"}, + request.kwargs["proxies"]) + self.assertIs(False, request.kwargs["allow_redirects"]) + self.assertNotIn("client_assertion", request.kwargs["data"]) + self.assertNotIn("key_id", request.kwargs["data"]) + self.ordinary.post.assert_not_called() + + def test_ordinary_http_and_post_override_do_not_use_factory(self, discovery): + app = self._app() + result = app.acquire_token_for_client(["ordinary"]) + self.assertEqual("Bearer", result["token_type"]) + self.assertNotIn("binding_certificate", result) + self.assertNotIn("allow_redirects", self.ordinary.post.call_args.kwargs) + post = Mock(return_value=self.ordinary.post.return_value) + result = app.acquire_token_for_client(["override"], post=post) + post.assert_called_once() + self.assertNotIn("binding_certificate", result) + self.ordinary.create_mtls_client.assert_not_called() + self.pipeline.post.assert_not_called() + + def test_factory_is_reused_for_network_calls_and_not_called_on_cache_hit( + self, discovery): + app = self._app() + for scope in ("s1", "s1", "s2"): + result = app.acquire_token_for_client( + [scope], mtls_proof_of_possession=True) + self.assertIsInstance( + result["binding_certificate"], msal.MtlsBindingContext) + self.ordinary.create_mtls_client.assert_called_once() + self.assertEqual(2, self.pipeline.post.call_count) + other = self._app(token_cache=app.token_cache) + result = other.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertEqual("cache", result["token_source"]) + self.assertIsNone(other._mtls_http_client) + self.ordinary.create_mtls_client.assert_called_once() + + def test_global_and_regional_clients_share_certificate_pipeline(self, discovery): + app = self._app() + with patch.object(app, "_compute_region_to_use", + side_effect=[None, "westus3", "westus3"]): + central = app._get_mtls_client(app.authority) + regional = app._get_mtls_client(app.authority) + self.assertIs(regional, app._get_mtls_client(app.authority)) + self.assertIsNot(central, regional) + self.assertIn("westus3.mtlsauth.microsoft.com", + regional.configuration["token_endpoint"]) + self.ordinary.create_mtls_client.assert_called_once() + + def test_concurrent_creation_does_not_orphan_pipelines(self, discovery): + app = self._app() + + def factory(binding): + time.sleep(0.02) + return self.pipeline + + self.ordinary.create_mtls_client.side_effect = factory + with ThreadPoolExecutor(max_workers=12) as executor: + clients = list(executor.map( + lambda _: app._get_mtls_client(app.authority), range(24))) + self.assertTrue(all(client is clients[0] for client in clients)) + self.ordinary.create_mtls_client.assert_called_once() + + def test_legacy_or_noncallable_capability_fails_before_cache_lookup( + self, discovery): + legacy = Mock(spec=["get", "post"]) + malformed = Mock(spec=["get", "post", "create_mtls_client"]) + malformed.create_mtls_client = 42 + for client in (legacy, malformed): + with self.subTest(client=client): + app = self._app(http_client=client) + with patch.object(app, "_acquire_token_silent_with_error") as cache: + with self.assertRaisesRegex(ValueError, "create_mtls_client"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + cache.assert_not_called() + client.post.assert_not_called() + + def test_malformed_factory_results_fail_without_fallback(self, discovery): + for result in (None, object(), SimpleNamespace(post=None), + SimpleNamespace(post=42)): + with self.subTest(result=result): + self.ordinary.create_mtls_client.return_value = result + app = self._app() + with self.assertRaisesRegex(TypeError, "callable post"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertIsNone(app._mtls_http_client) + self.ordinary.post.assert_not_called() + + def test_factory_cannot_return_ordinary_client_or_msal_wrapper(self, discovery): + app = self._app() + for client in (self.ordinary, app.http_client): + with self.subTest(client=client): + self.ordinary.create_mtls_client.return_value = client + with self.assertRaisesRegex(ValueError, "new, dedicated"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertIsNone(app._mtls_http_client) + self.ordinary.post.assert_not_called() + + def test_factory_cannot_reuse_client_across_live_applications(self, discovery): + first, second = self._app(), self._app() + first._get_mtls_client(first.authority) + with self.assertRaisesRegex(ValueError, "another application"): + second._get_mtls_client(second.authority) + self.assertIsNone(second._mtls_http_client) + self.pipeline.post.assert_not_called() + + def test_client_identity_tracking_does_not_retain_application(self, discovery): + import gc + import weakref + + app = self._app() + # A post-only, non-weak-referenceable client is a valid pipeline. + self.ordinary.create_mtls_client.return_value = SimpleNamespace( + post=self.pipeline.post) + app._get_mtls_client(app.authority) + reference = weakref.ref(app) + del app + gc.collect() + self.assertIsNone(reference()) + + def test_factory_errors_propagate_and_creation_can_be_retried(self, discovery): + app = self._app() + for error in (TypeError("factory type error"), OSError("factory failed"), + ValueError("factory configuration error")): + with self.subTest(error=error): + self.ordinary.create_mtls_client.side_effect = error + with self.assertRaises(type(error)) as raised: + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertIs(error, raised.exception) + self.assertIsNone(app._mtls_http_client) + self.ordinary.create_mtls_client.side_effect = None + self.assertIn("access_token", app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True)) + self.ordinary.post.assert_not_called() + + def test_request_errors_propagate_without_replacing_pipeline(self, discovery): + app = self._app() + error = requests.exceptions.ConnectionError("network policy failure") + self.pipeline.post.side_effect = [ + error, MinimalResponse(status_code=200, text=_TOKEN)] + with self.assertRaises(requests.exceptions.ConnectionError) as raised: + app.acquire_token_for_client(["s1"], mtls_proof_of_possession=True) + self.assertIs(error, raised.exception) + self.assertIn("access_token", app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True)) + self.ordinary.create_mtls_client.assert_called_once() + self.ordinary.post.assert_not_called() + + def test_request_overrides_are_rejected_before_cache_lookup(self, discovery): + for custom in (False, True): + app = self._app() if custom else msal.ConfidentialClientApplication( + "cid", authority=_AUTHORITY, client_credential=_CREDENTIAL, + azure_region=False, instance_discovery=False) + self.addCleanup(app.http_client.close) + for kwargs in ({"post": None}, {"post": Mock()}, + {"cert": ("other.pem", "other.key")}, + {"allow_redirects": True}, {"allow_redirects": None}): + with self.subTest(custom=custom, kwargs=kwargs): + with patch.object( + app, "_acquire_token_silent_with_error") as cache: + with self.assertRaises(ValueError): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, **kwargs) + cache.assert_not_called() + self.ordinary.create_mtls_client.assert_not_called() + + def test_observable_redirects_are_not_cached(self, discovery): + endpoint = "https://mtlsauth.microsoft.com/tenant/oauth2/v2.0/token" + for changes in ({"status_code": 307}, {"history": [object()]}, + {"url": "https://other.example/token"}): + with self.subTest(changes=changes): + response = SimpleNamespace( + status_code=200, text=_TOKEN, url=endpoint, headers={}) + for name, value in changes.items(): + setattr(response, name, value) + pipeline = Mock(spec=["post"]) + pipeline.post.return_value = response + self.ordinary.create_mtls_client.return_value = pipeline + app = self._app() + with self.assertRaisesRegex(ValueError, "must not redirect"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertEqual({}, app.token_cache._cache.get("AccessToken", {})) + self.ordinary.post.assert_not_called() + + def test_downgraded_custom_response_has_no_binding_or_cache_entry(self, discovery): + app = self._app() + self.pipeline.post.return_value = self.ordinary.post.return_value + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertEqual("token_type_mismatch", result["error"]) + self.assertNotIn("binding_certificate", result) + self.assertEqual({}, app.token_cache._cache.get("AccessToken", {})) + + def test_request_query_parameters_are_not_treated_as_redirects(self, discovery): + self.pipeline.post.return_value = SimpleNamespace( + status_code=200, text=_TOKEN, headers={}, + url="https://mtlsauth.microsoft.com/tenant/oauth2/v2.0/token?extra=value") + result = self._app().acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, params={"extra": "value"}) + self.assertEqual("mtls_pop", result["token_type"]) + self.assertEqual({"extra": "value"}, self.pipeline.post.call_args.kwargs["params"]) + + +class TestMtlsPipelineHelpers(unittest.TestCase): + def test_token_transport_is_restricted_to_https_token_endpoint(self): + pipeline = Mock(spec=["post"]) + with self.assertRaisesRegex(ValueError, "HTTPS"): + mtls._MtlsTokenHttpClient(pipeline, "http://token.example") + client = mtls._MtlsTokenHttpClient(pipeline, "https://token.example/token") + with self.assertRaisesRegex(ValueError, "only call its token endpoint"): + client.post("https://token.example/discovery") + pipeline.post.assert_not_called() + + def test_public_adapter_retains_pool_and_retry_options(self): + from msal.application import _load_mtls_cert_material + + material = _load_mtls_cert_material(_CREDENTIAL) + binding = msal.MtlsBindingContext( + material["cert_pem"], material["private_key_pem"], + material["x5c"], material["key_id"]) + adapter = binding.create_requests_adapter( + pool_connections=3, pool_maxsize=7, pool_block=True, max_retries=0) + self.addCleanup(adapter.close) + request = requests.Request("POST", "https://token.example/token").prepare() + with patch("requests.adapters.HTTPAdapter.send"): + adapter.send(request, verify=False) + adapter.send(request, verify=False) + child = adapter._adapters[False] + self.assertEqual(3, child._pool_connections) + self.assertEqual(7, child._pool_maxsize) + self.assertTrue(child._pool_block) + self.assertEqual(0, child.max_retries.total) + self.assertEqual(ssl.CERT_NONE, child._ssl_context.verify_mode) + self.assertEqual(1, len(adapter._adapters)) + for cert in ("other.pem", ("other.pem", "other.key")): + with self.subTest(cert=cert), self.assertRaisesRegex( + ValueError, "cert override"): + adapter.send(request, cert=cert) + + +class TestCustomPipelineLocalTls(unittest.TestCase): + """Real loopback TLS, with no lab credentials or external network access.""" + + @classmethod + def setUpClass(cls): + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID + + def key(): + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + def certificate(name, private_key, issuer, issuer_key, ca=False, server=False): + now = datetime.now(timezone.utc) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)]) + builder = (x509.CertificateBuilder() + .subject_name(subject).issuer_name(issuer.subject if issuer else subject) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=1)) + .add_extension(x509.BasicConstraints(ca=ca, path_length=None), True) + .add_extension(x509.SubjectKeyIdentifier.from_public_key( + private_key.public_key()), False) + .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key( + issuer_key.public_key()), False) + .add_extension(x509.KeyUsage( + digital_signature=True, content_commitment=False, + key_encipherment=not ca, data_encipherment=False, + key_agreement=False, key_cert_sign=ca, crl_sign=ca, + encipher_only=False, decipher_only=False), True)) + if not ca: + builder = builder.add_extension(x509.ExtendedKeyUsage([ + ExtendedKeyUsageOID.SERVER_AUTH if server + else ExtendedKeyUsageOID.CLIENT_AUTH]), False) + if server: + builder = builder.add_extension(x509.SubjectAlternativeName([ + x509.DNSName("localhost"), + x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), False) + return builder.sign(issuer_key, hashes.SHA256()) + + def pem(private_key): + return private_key.private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption()) + + root_key = key() + root = certificate("test root", root_key, None, root_key, ca=True) + cls.root_pem = root.public_bytes(serialization.Encoding.PEM) + server_key = key() + server = certificate( + "localhost", server_key, root, root_key, server=True) + cls.server_pem = server.public_bytes(serialization.Encoding.PEM) + pem(server_key) + cls.credentials = [] + cls.client_der = [] + for name in ("first client", "rotated client"): + private_key = key() + cert = certificate(name, private_key, root, root_key) + cls.credentials.append({ + "private_key": pem(private_key).decode("ascii"), + "public_certificate": cert.public_bytes( + serialization.Encoding.PEM).decode("ascii"), + "thumbprint": cert.fingerprint(hashes.SHA1()).hex(), + }) + cls.client_der.append(cert.public_bytes(serialization.Encoding.DER)) + + def _start_server(self, handler, context=None): + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + if context is not None: + server.socket = context.wrap_socket(server.socket, server_side=True) + thread = threading.Thread( + target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True) + thread.start() + self.addCleanup(server.server_close) + self.addCleanup(thread.join, 5) + self.addCleanup(server.shutdown) + return server + + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.ca_path = os.path.join(directory.name, "ca.pem") + server_path = os.path.join(directory.name, "server.pem") + for path, content in ((self.ca_path, self.root_pem), + (server_path, self.server_pem)): + with open(path, "wb") as stream: + stream.write(content) + environment = patch.dict(os.environ, { + "REQUESTS_CA_BUNDLE": "", "CURL_CA_BUNDLE": "", + "HTTP_PROXY": "", "HTTPS_PROXY": "", "ALL_PROXY": ""}) + environment.start() + self.addCleanup(environment.stop) + owner = self + self.records = [] + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_GET(self): + self.do_POST() + + def do_POST(self): + body = self.rfile.read(int(self.headers.get("Content-Length", 0))) + data = parse_qs(body.decode("ascii")) + certificate = self.connection.getpeercert(binary_form=True) + owner.records.append({ + "path": self.path, "certificate": certificate, "data": data, + "port": self.client_address[1], + "middleware": self.headers.get("X-Msal-Pipeline"), + }) + status = 200 + response = { + "access_token": "local-at", "token_type": "Bearer", + "expires_in": 3600} + if self.path == "/metadata": + response = dict(_METADATA, token_endpoint=owner.url + _ORDINARY_PATH) + elif data.get("scope") == ["redirect"]: + status = 307 + response = {} + elif self.path == _TOKEN_PATH: + if not certificate or (data.get("scope") == ["reject-first"] + and certificate == owner.client_der[0]): + status, response = 400, {"error": "invalid_client"} + else: + response["token_type"] = "mtls_pop" + payload = json.dumps(response).encode("utf-8") + self.send_response(status) + if status == 307: + self.send_header("Location", owner.url + "/forbidden") + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(server_path) + context.load_verify_locations(cafile=self.ca_path) + context.verify_mode = ssl.CERT_OPTIONAL + self.server = self._start_server(Handler, context) + self.url = "https://127.0.0.1:{}".format(self.server.server_port) + self.http_client = PolicyHttpClient( + verify=self.ca_path, timeout=(2, 3), trust_env=False) + self.addCleanup(self.http_client.close) + + def _app(self, http_client=None, certificate=0, discover=True, **kwargs): + def discovery(endpoint, client): + response = client.get(self.url + "/metadata") + response.raise_for_status() + return json.loads(response.text) + + with patch("msal.authority.tenant_discovery", + side_effect=discovery if discover else None, + return_value=dict(_METADATA, token_endpoint=self.url + _ORDINARY_PATH)): + return msal.ConfidentialClientApplication( + "cid", authority=_AUTHORITY, + client_credential=self.credentials[certificate], + http_client=self.http_client if http_client is None else http_client, + azure_region=False, instance_discovery=False, **kwargs) + + def _acquire(self, app, scope="s1", **kwargs): + with patch("msal.mtls.transform_token_endpoint", + return_value=self.url + _TOKEN_PATH): + return app.acquire_token_for_client( + [scope], mtls_proof_of_possession=True, **kwargs) + + def test_custom_pipeline_preserves_certificate_middleware_policy_and_pooling(self): + adapter = self.http_client._session.adapters["https://"] + app = self._app(verify=False, timeout=0.001, + proxies={"https": "http://unused.example"}) + ordinary = app.acquire_token_for_client(["ordinary"]) + first = self._acquire(app) + child = self.http_client._children[0] + with patch.object(child._session, "send", wraps=child._session.send) as send: + self._acquire(app, "s2") + self.assertEqual((2, 3), send.call_args.kwargs["timeout"]) + self.assertEqual(self.ca_path, send.call_args.kwargs["verify"]) + self.assertEqual(2, child.response_count) + self.assertIs(adapter, self.http_client._session.adapters["https://"]) + self.assertEqual(1, len(self.http_client._children)) + self.assertNotIn("binding_certificate", ordinary) + self.assertIsInstance(first["binding_certificate"], msal.MtlsBindingContext) + self.assertEqual( + ["/metadata", _ORDINARY_PATH, _TOKEN_PATH, _TOKEN_PATH], + [record["path"] for record in self.records]) + self.assertTrue(all(record["middleware"] == "custom-policy" + for record in self.records)) + self.assertTrue(all(record["certificate"] is None for record in self.records[:2])) + self.assertTrue(all(record["certificate"] == self.client_der[0] + for record in self.records[2:])) + self.assertEqual(self.records[2]["port"], self.records[3]["port"]) + self.assertNotEqual(self.records[1]["port"], self.records[2]["port"]) + self._acquire(app) + self.assertEqual(4, len(self.records), "A cache hit must not use HTTP") + + def test_certificate_rotation_isolates_connections_tokens_and_http_errors(self): + token_cache, http_cache = msal.TokenCache(), {} + first = self._app(token_cache=token_cache, http_cache=http_cache) + rotated = self._app(certificate=1, token_cache=token_cache, http_cache=http_cache) + results = [self._acquire(app) for app in (first, rotated)] + self.assertNotEqual(results[0]["binding_certificate"].thumbprint_sha256, + results[1]["binding_certificate"].thumbprint_sha256) + self._acquire(first, "s2") + tokens = [record for record in self.records if record["path"] == _TOKEN_PATH] + self.assertEqual( + [self.client_der[0], self.client_der[1], self.client_der[0]], + [record["certificate"] for record in tokens]) + self.assertEqual(tokens[0]["port"], tokens[2]["port"]) + self.assertNotEqual(tokens[0]["port"], tokens[1]["port"]) + self.assertEqual(2, len(self.http_client._children)) + for _ in range(2): + self.assertEqual("invalid_client", self._acquire(first, "reject-first")["error"]) + self.assertEqual("mtls_pop", self._acquire(rotated, "reject-first")["token_type"]) + failures = [record for record in self.records + if record["data"].get("scope") == ["reject-first"]] + self.assertEqual([self.client_der[0], self.client_der[1]], + [record["certificate"] for record in failures]) + cached = self._acquire(first) + self.assertEqual("cache", cached["token_source"]) + self.assertEqual(results[0]["binding_certificate"].thumbprint_sha256, + cached["binding_certificate"].thumbprint_sha256) + + def test_effective_ca_settings_and_verification_pools(self): + from cryptography import x509 + from cryptography.hazmat.primitives import serialization + + expected_root = x509.load_pem_x509_certificate(self.root_pem).public_bytes( + serialization.Encoding.DER) + for setting in ("session", "request", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE"): + environment = {"REQUESTS_CA_BUNDLE": "", "CURL_CA_BUNDLE": ""} + if setting in environment: + environment[setting] = self.ca_path + with self.subTest(setting=setting), patch.dict(os.environ, environment): + with PolicyHttpClient( + verify=self.ca_path if setting == "session" else True, + timeout=(2, 3)) as client: + app = self._app(http_client=client, discover=False) + options = {"verify": self.ca_path} if setting == "request" else {} + self._acquire(app, **options) + adapter = client._children[0]._session.adapters["https://"] + verified = adapter._adapters[self.ca_path] + self.assertEqual([expected_root], + verified._ssl_context.get_ca_certs(binary_form=True)) + with self.assertWarns(requests.packages.urllib3.exceptions.InsecureRequestWarning): + self._acquire(app, "unverified", verify=False) + self._acquire(app, "verified-again", **options) + self.assertIs(verified, adapter._adapters[self.ca_path]) + self.assertEqual(ssl.CERT_NONE, + adapter._adapters[False]._ssl_context.verify_mode) + records = self.records[-3:] + self.assertEqual(records[0]["port"], records[2]["port"]) + self.assertNotEqual(records[0]["port"], records[1]["port"]) + self.assertTrue(all(record["certificate"] == self.client_der[0] + for record in records)) + + def test_untrusted_server_failure_does_not_fall_back(self): + app = self._app() + with self.assertRaises(requests.exceptions.SSLError): + self._acquire(app, verify=True) + self.assertEqual(["/metadata"], [record["path"] for record in self.records]) + self.assertEqual(1, len(self.http_client._children)) + self.assertEqual("mtls_pop", self._acquire(app)["token_type"]) + self.assertEqual(self.client_der[0], self.records[-1]["certificate"]) + + def test_default_and_custom_pipelines_do_not_follow_token_redirects(self): + custom = self._app() + with patch("msal.authority.tenant_discovery", + return_value=dict(_METADATA, token_endpoint=self.url + _ORDINARY_PATH)): + default = msal.ConfidentialClientApplication( + "cid", authority=_AUTHORITY, client_credential=self.credentials[0], + verify=self.ca_path, timeout=(2, 3), azure_region=False, + instance_discovery=False) + self.addCleanup(default.http_client.close) + for app in (default, custom): + with self.subTest(custom=app is custom): + self.assertEqual("mtls_pop", self._acquire(app)["token_type"]) + with self.assertRaisesRegex(ValueError, "must not redirect"): + self._acquire(app, "redirect") + default._mtls_http_client.close() + self.assertNotIn("/forbidden", [record["path"] for record in self.records]) + tokens = [record for record in self.records if record["path"] == _TOKEN_PATH] + self.assertEqual(4, len(tokens)) + self.assertTrue(all(record["certificate"] == self.client_der[0] for record in tokens)) + + def test_factory_owns_lifecycle_and_closing_child_does_not_close_ordinary(self): + app = self._app() + self._acquire(app) + child = self.http_client._children[0] + child.close() + with self.assertRaisesRegex(RuntimeError, "closed"): + self._acquire(app, "after-close") + self.assertEqual(1, len(self.http_client._children)) + response = self.http_client.get(self.url + _ORDINARY_PATH) + self.assertEqual(200, response.status_code) + self.assertIsNone(self.records[-1]["certificate"]) + self.http_client.close() + self.http_client.close() + with self.assertRaisesRegex(RuntimeError, "closed"): + self.http_client.create_mtls_client(None) + self.assertTrue(child._closed) + + def test_https_proxy_preserves_custom_pipeline_and_certificate(self): + target = self.server.server_address + tunnels = [] + + class ProxyHandler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_CONNECT(self): + if self.path != "{}:{}".format(*target): + self.send_error(403) + return + tunnels.append(self.path) + with socket.create_connection(target, timeout=3) as upstream: + self.send_response(200) + self.end_headers() + self.close_connection = True + sockets = (self.connection, upstream) + while True: + readable, _, _ = select.select(sockets, [], [], 3) + if not readable: + return + for source in readable: + data = source.recv(65536) + if not data: + return + destination = upstream if source is self.connection else self.connection + destination.sendall(data) + + proxy = self._start_server(ProxyHandler) + proxies = {"https": "http://127.0.0.1:{}".format(proxy.server_port)} + with PolicyHttpClient( + verify=self.ca_path, proxies=proxies, timeout=(2, 3), + trust_env=False) as client: + app = self._app(http_client=client) + result = self._acquire(app) + self.assertEqual("mtls_pop", result["token_type"]) + self.assertEqual(proxies, client._children[0]._session.proxies) + self.assertEqual(2, len(tunnels)) + self.assertEqual([None, self.client_der[0]], + [record["certificate"] for record in self.records]) + self.assertTrue(all(record["middleware"] == "custom-policy" + for record in self.records)) + + +if __name__ == "__main__": + unittest.main() From 18bfd0d7a9e11d411cae6b228306d846c7450d7d Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 13:53:49 -0400 Subject: [PATCH 02/11] Fix custom mTLS transport response handling and dependency validation Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- msal/mtls.py | 6 ++- tests/test_mtls_http_client.py | 92 +++++++++++++++++++++++++++++++++- tests/test_mtls_transport.py | 3 +- 3 files changed, 98 insertions(+), 3 deletions(-) diff --git a/msal/mtls.py b/msal/mtls.py index 572ade03..f2550421 100644 --- a/msal/mtls.py +++ b/msal/mtls.py @@ -152,6 +152,7 @@ def create_requests_adapter(self, **kwargs): A ``cert`` override is rejected to preserve the selected certificate. The caller owns the adapter and its connection pools. """ + _require_mtls_dependencies() return _make_mtls_adapter( ssl_context_factory=self.create_ssl_context, **kwargs) @@ -317,11 +318,14 @@ def post(self, url, **kwargs): response_url = getattr(response, "url", None) # Request params may change the query without a redirect. endpoint_changed = response_url is not None and ( - urlparse(response_url)._replace(query="", fragment="") + urlparse(str(response_url))._replace(query="", fragment="") != urlparse(url)._replace(query="", fragment="")) if (300 <= response.status_code < 400 or getattr(response, "history", None) or endpoint_changed): + close = getattr(response, "close", None) + if callable(close): + close() raise ValueError("The mTLS token endpoint must not redirect.") return response diff --git a/tests/test_mtls_http_client.py b/tests/test_mtls_http_client.py index db69c5d2..f149f293 100644 --- a/tests/test_mtls_http_client.py +++ b/tests/test_mtls_http_client.py @@ -265,6 +265,53 @@ def test_observable_redirects_are_not_cached(self, discovery): self.assertEqual({}, app.token_cache._cache.get("AccessToken", {})) self.ordinary.post.assert_not_called() + def test_rejected_responses_are_closed_without_closing_pipeline(self, discovery): + endpoint = "https://mtlsauth.microsoft.com/tenant/oauth2/v2.0/token" + for changes in ({"status_code": 307}, {"history": [object()]}, + {"url": "https://other.example/token"}): + with self.subTest(changes=changes): + response = SimpleNamespace( + status_code=200, text=_TOKEN, url=endpoint, headers={}, + close=Mock()) + for name, value in changes.items(): + setattr(response, name, value) + pipeline = Mock(spec=["post", "close"]) + pipeline.post.return_value = response + self.ordinary.create_mtls_client.return_value = pipeline + app = self._app() + with self.assertRaisesRegex(ValueError, "must not redirect"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + response.close.assert_called_once_with() + pipeline.close.assert_not_called() + self.ordinary.close.assert_not_called() + self.assertEqual({}, app.token_cache._cache.get("AccessToken", {})) + + def test_url_objects_preserve_endpoint_and_redirect_validation(self, discovery): + from urllib3.util import parse_url + + endpoint = "https://mtlsauth.microsoft.com/tenant/oauth2/v2.0/token" + for url in (endpoint + "?extra=value", "https://other.example/token"): + with self.subTest(url=url): + response = SimpleNamespace( + status_code=200, text=_TOKEN, headers={}, + url=parse_url(url), close=Mock()) + pipeline = Mock(spec=["post"]) + pipeline.post.return_value = response + self.ordinary.create_mtls_client.return_value = pipeline + app = self._app() + if url.startswith(endpoint): + result = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertEqual("mtls_pop", result["token_type"]) + response.close.assert_not_called() + else: + with self.assertRaisesRegex(ValueError, "must not redirect"): + app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True) + self.assertEqual( + {}, app.token_cache._cache.get("AccessToken", {})) + def test_downgraded_custom_response_has_no_binding_or_cache_entry(self, discovery): app = self._app() self.pipeline.post.return_value = self.ordinary.post.return_value @@ -285,6 +332,17 @@ def test_request_query_parameters_are_not_treated_as_redirects(self, discovery): class TestMtlsPipelineHelpers(unittest.TestCase): + def test_noncallable_response_close_does_not_mask_redirect_error(self): + endpoint = "https://token.example/token" + for close in (None, 42): + with self.subTest(close=close): + pipeline = Mock(spec=["post"]) + pipeline.post.return_value = SimpleNamespace( + status_code=307, url=endpoint, close=close) + client = mtls._MtlsTokenHttpClient(pipeline, endpoint) + with self.assertRaisesRegex(ValueError, "must not redirect"): + client.post(endpoint) + def test_token_transport_is_restricted_to_https_token_endpoint(self): pipeline = Mock(spec=["post"]) with self.assertRaisesRegex(ValueError, "HTTPS"): @@ -444,6 +502,9 @@ def do_POST(self): elif data.get("scope") == ["redirect"]: status = 307 response = {} + elif data.get("scope") == ["redirect-without-location"]: + status = 307 + response = {} elif self.path == _TOKEN_PATH: if not certificate or (data.get("scope") == ["reject-first"] and certificate == owner.client_der[0]): @@ -452,7 +513,7 @@ def do_POST(self): response["token_type"] = "mtls_pop" payload = json.dumps(response).encode("utf-8") self.send_response(status) - if status == 307: + if status == 307 and data.get("scope") != ["redirect-without-location"]: self.send_header("Location", owner.url + "/forbidden") self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) @@ -629,6 +690,35 @@ def test_factory_owns_lifecycle_and_closing_child_does_not_close_ordinary(self): self.http_client.create_mtls_client(None) self.assertTrue(child._closed) + def test_rejected_streamed_response_releases_blocking_pool_slot(self): + app = self._app() + binding = self._acquire(app)["binding_certificate"] + child = self.http_client._children[0] + adapter = binding.create_requests_adapter( + pool_connections=1, pool_maxsize=1, pool_block=True) + self.addCleanup(adapter.close) + child._session.adapters["https://"].close() + child._session.mount("https://", adapter) + responses = [] + + def capture(response, **kwargs): + responses.append(response) + self.addCleanup(response.close) + return response + + child._session.hooks["response"].append(capture) + with self.assertRaisesRegex(ValueError, "must not redirect"): + self._acquire(app, "redirect-without-location", stream=True) + response = responses[0] + self.assertFalse(response._content_consumed) + self.assertIsNone(response.raw._connection) + verified = adapter._adapters[self.ca_path] + pool = verified.poolmanager.connection_from_url( + self.url, pool_kwargs={"ssl_context": verified._ssl_context}) + self.assertEqual(1, pool.pool.qsize()) + self.assertEqual("mtls_pop", self._acquire(app, "after-redirect")["token_type"]) + self.assertFalse(child._closed) + def test_https_proxy_preserves_custom_pipeline_and_certificate(self): target = self.server.server_address tunnels = [] diff --git a/tests/test_mtls_transport.py b/tests/test_mtls_transport.py index e77aa4dc..af2802e3 100644 --- a/tests/test_mtls_transport.py +++ b/tests/test_mtls_transport.py @@ -226,7 +226,8 @@ def test_unsupported_mtls_dependencies_fail_with_upgrade_instructions(self): with dependency: for factory in ( lambda: mtls._MtlsHttpClient(b"cert", b"key"), - binding.create_requests_session): + binding.create_requests_session, + binding.create_requests_adapter): with self.subTest(factory=factory), self.assertRaisesRegex( ValueError, r"mTLS requires Requests >= 2\.25\.0.*" r"python -m pip install --upgrade"): From 3bc3505ac51b44c36105525a0d63293ab4da9221 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 15:09:40 -0400 Subject: [PATCH 03/11] Clarify default-only mTLS live E2E coverage Distinguish the inherited live matrix from custom HTTP factory loopback TLS coverage without claiming cross-SDK live factory parity. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_e2e.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 65093477..b498053d 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -444,8 +444,9 @@ def test_subject_name_issuer_authentication(self): # ── 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. + # These live matrix cells exercise MSAL's default mTLS transport, not the + # custom HTTP factory. Custom-pipeline coverage remains local TLS coverage + # in test_mtls_http_client.py; these tests do not establish live factory parity. # 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"] From 751aa0b0576c7d39777dacb0c7e0b314968e3be9 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Fri, 23 Oct 2026 15:15:18 -0400 Subject: [PATCH 04/11] Make required E2E fixture checks independent of release environment Allow the mocked legacy release-skip attribute to be created when the E2E module was imported without TRAVIS_TAG. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_lab_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_lab_config.py b/tests/test_lab_config.py index 09501273..43108d4b 100644 --- a/tests/test_lab_config.py +++ b/tests/test_lab_config.py @@ -55,7 +55,7 @@ def test_valid_certificate_configuration_is_preserved(self): def test_required_scenarios_fail_without_credentials_instead_of_skipping(self): with patch.dict(os.environ, {"TRAVIS_TAG": "test-release"}, clear=True), patch.object( - test_e2e.E2eTestCase, "__unittest_skip__", True), patch( + test_e2e.E2eTestCase, "__unittest_skip__", True, create=True), patch( "tests.test_e2e.msal.ConfidentialClientApplication") as application: suite = unittest.defaultTestLoader.loadTestsFromTestCase( test_e2e.CertificateScenariosTestCase) From 9814be5162867ad0302af307127cb5c321e69678 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 10:43:18 -0400 Subject: [PATCH 05/11] Move custom mTLS recorder and sample safety coverage to its owning PR Redistribute the approved factory recorder/privacy/lifecycle slice from combined source 0cd290d without importing enabled Bearer-over-mTLS or real FIC integration. Preserve sample middleware, factory ownership, safe diagnostics, and the explicit loopback TLS floor. Shared generic E2E fixtures are supplied by the dependent foundation. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...idential_client_mtls_http_client_sample.py | 4 +- tests/mtls_http_e2e_helpers.py | 38 ++++++++ tests/test_mtls_http_client.py | 25 +++++ tests/test_mtls_http_e2e_helpers.py | 95 +++++++++++++++++++ 4 files changed, 161 insertions(+), 1 deletion(-) create mode 100644 tests/mtls_http_e2e_helpers.py create mode 100644 tests/test_mtls_http_e2e_helpers.py diff --git a/sample/confidential_client_mtls_http_client_sample.py b/sample/confidential_client_mtls_http_client_sample.py index 233fef29..d1a3dc62 100644 --- a/sample/confidential_client_mtls_http_client_sample.py +++ b/sample/confidential_client_mtls_http_client_sample.py @@ -107,7 +107,9 @@ def main(): os.environ["SCOPE"].split(), mtls_proof_of_possession=True) print("Acquired mTLS-PoP token:", result.get("token_type") == "mtls_pop") if "error" in result: - print("Error code:", result["error"]) + print( + "Token acquisition failed; inspect result['error'] / " + "result['error_description'] locally to diagnose.") if __name__ == "__main__": diff --git a/tests/mtls_http_e2e_helpers.py b/tests/mtls_http_e2e_helpers.py new file mode 100644 index 00000000..83ac4e35 --- /dev/null +++ b/tests/mtls_http_e2e_helpers.py @@ -0,0 +1,38 @@ +"""Custom-pipeline recording for the mTLS HTTP factory E2E tests.""" + +from sample.confidential_client_mtls_http_client_sample import PolicyHttpClient +from tests.mtls_e2e_helpers import token_request_summary + + +class RecordingPolicyHttpClient(PolicyHttpClient): + """Add safe recording to the existing sample's real certificate pipeline.""" + + def __init__(self, binding_context=None, expected_binding=None, + expected_assertion=None, **kwargs): + __tracebackhide__ = True + self.binding_context = binding_context + self.expected_binding = expected_binding or binding_context + self.expected_assertion = expected_assertion + self.token_requests = [] + super().__init__(binding_context=binding_context, **kwargs) + + def create_mtls_client(self, binding_context): + with self._lock: + self._check_open() + child = RecordingPolicyHttpClient( + binding_context=binding_context, + expected_binding=self.expected_binding, + expected_assertion=self.expected_assertion, **self._config) + self._children.append(child) + return child + + def post(self, url, **kwargs): + __tracebackhide__ = True + response = super().post(url, **kwargs) + summary = token_request_summary( + response, self.expected_binding, + self._session.get_adapter(response.request.url), + expected_assertion=self.expected_assertion) + if summary is not None: + self.token_requests.append(summary) + return response diff --git a/tests/test_mtls_http_client.py b/tests/test_mtls_http_client.py index f149f293..c166d036 100644 --- a/tests/test_mtls_http_client.py +++ b/tests/test_mtls_http_client.py @@ -1,4 +1,5 @@ import ipaddress +import io import json import os import select @@ -9,6 +10,7 @@ import time import unittest from concurrent.futures import ThreadPoolExecutor +from contextlib import redirect_stdout from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import SimpleNamespace @@ -40,6 +42,28 @@ _ORDINARY_PATH = "/tenant/oauth2/v2.0/ordinary" +class TestMtlsSampleLogging(unittest.TestCase): + def test_custom_http_sample_reports_failure_without_dynamic_error_values(self): + from sample.confidential_client_mtls_http_client_sample import main + + sensitive = "sensitive-unit-result-value" + app = Mock() + app.acquire_token_for_client.return_value = { + "token_type": sensitive, "error": sensitive, "error_description": sensitive} + environment = { + "PFX_PATH": _CREDENTIAL["private_key_pfx_path"], + "PFX_PASSPHRASE": _CREDENTIAL["passphrase"], + "CLIENT_ID": "sample-client", "AUTHORITY": _AUTHORITY, + "SCOPE": "https://vault.azure.net/.default", + } + with patch.dict(os.environ, environment), patch( + "msal.ConfidentialClientApplication", return_value=app), redirect_stdout( + io.StringIO()) as output: + main() + self.assertNotIn(sensitive, output.getvalue()) + self.assertIn("Token acquisition failed", output.getvalue()) + + @patch("msal.authority.tenant_discovery", return_value=_METADATA) class TestMtlsHttpClientFactory(unittest.TestCase): def setUp(self): @@ -521,6 +545,7 @@ def do_POST(self): self.wfile.write(payload) context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.minimum_version = ssl.TLSVersion.TLSv1_2 context.load_cert_chain(server_path) context.load_verify_locations(cafile=self.ca_path) context.verify_mode = ssl.CERT_OPTIONAL diff --git a/tests/test_mtls_http_e2e_helpers.py b/tests/test_mtls_http_e2e_helpers.py new file mode 100644 index 00000000..a1b9eb68 --- /dev/null +++ b/tests/test_mtls_http_e2e_helpers.py @@ -0,0 +1,95 @@ +"""Offline custom-recorder guards, not evidence of live E2E acquisitions.""" + +import unittest +from unittest.mock import Mock, patch + +import pytest +import requests + +from tests import test_e2e +from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient +from tests.test_mtls_e2e_helpers import CertificateFixture, _ENDPOINT, _jwt + + +class TestCustomMtlsE2eHelpers(CertificateFixture): + def setUp(self): + self.scenario = test_e2e.CertificateScenariosTestCase( + "test_credential_x509_custom_http_output_pop") + + def _assert_private_traceback(self, failure, *secrets): + rendered = str(failure.getrepr( + style="long", showlocals=False, funcargs=True)) + self.assertIn(type(failure.value).__name__, rendered) + self.assertFalse( + "PRIVATE KEY" in rendered, + "Rendered tracebacks must not expose private key material.") + for secret in secrets: + fragments = (secret,) + ( + (secret[:32], secret[-32:]) if len(secret) > 64 else ()) + for fragment in fragments: + self.assertFalse( + fragment in rendered, + "Sensitive arguments must not appear in rendered tracebacks.") + + def test_custom_capture_failure_traceback_hides_the_expected_assertion(self): + assertion = _jwt({"aud": "api://AzureADTokenExchange"}) + app = Mock() + with RecordingPolicyHttpClient( + expected_binding=self.binding, expected_assertion=assertion) as client: + app.acquire_token_for_client.return_value = {"access_token": assertion} + with pytest.raises(AssertionError) as failure: + self.scenario._acquire_recorded( + app, self.scenario._MTLS_POP_SCOPE, self.binding, + pop=True, http_client=client, expected_assertion=assertion) + self._assert_private_traceback(failure, assertion) + + def test_custom_recorder_failure_traceback_hides_posted_credentials(self): + form = self._signed_form() + + def send(adapter, request, **kwargs): + response = self._response(payload=[]) + response.request = request + return response + + with RecordingPolicyHttpClient(expected_binding=self.binding) as owner: + child = owner.create_mtls_client(self.binding) + with patch.object(requests.adapters.HTTPAdapter, "send", + autospec=True, side_effect=send): + with pytest.raises(AssertionError) as failure: + child.post(_ENDPOINT, data=form) + self._assert_private_traceback( + failure, form["client_assertion"], form["refresh_token"]) + + def test_recorder_setup_failure_traceback_hides_the_expected_assertion(self): + assertion = _jwt({"aud": "api://AzureADTokenExchange"}) + with pytest.raises(TypeError) as failure: + RecordingPolicyHttpClient( + expected_assertion=assertion, unexpected_option=True) + self._assert_private_traceback(failure, assertion) + + def test_custom_recorder_retains_sample_pipeline_and_owner_cleanup(self): + def send(adapter, request, **kwargs): + response = self._response() + response.request = request + return response + + with RecordingPolicyHttpClient(expected_binding=self.binding) as owner: + child = owner.create_mtls_client(self.binding) + with patch.object(requests.adapters.HTTPAdapter, "send", + autospec=True, side_effect=send): + child.post(_ENDPOINT, data=self._signed_form()) + self.assertEqual(1, child.response_count) + self.assertEqual(1, len(child.token_requests)) + self.assertEqual( + "custom-policy", child.token_requests[0]["middleware"]) + self.assertEqual( + "_MtlsHTTPAdapter", child.token_requests[0]["tls_adapter"]) + self.assertEqual([], owner.token_requests) + self.assertFalse(owner._closed or child._closed) + self.assertTrue(owner._closed and child._closed) + with self.assertRaisesRegex(RuntimeError, "closed"): + child.post(_ENDPOINT) + + +if __name__ == "__main__": + unittest.main() From 960195b7191009b994b4bc9f482616264a85d283 Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 10:45:05 -0400 Subject: [PATCH 06/11] Document the custom source inventory and reuse shared privacy guards Keep the extensibility branch inventory separate from combined Bearer/FIC scenarios and reuse the foundation certificate fixture for rendered traceback redaction. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/README.md | 35 +++++++++++++++++++++++++++++ tests/test_mtls_http_e2e_helpers.py | 15 ------------- 2 files changed, 35 insertions(+), 15 deletions(-) diff --git a/tests/README.md b/tests/README.md index 293073dc..08446fdd 100644 --- a/tests/README.md +++ b/tests/README.md @@ -65,6 +65,41 @@ uses client `163ffef9-a313-45b4-ab2f-c7e2f5e0e23e` in tenant They do not require fetching unrelated user configuration from Key Vault. Legacy interactive and other optional E2E skips are unchanged. +### Custom HTTP coverage in the extensibility follow-up + +This branch adds two required cells to the six inherited X.509/default-transport +cells, for **eight required certificate scenarios**, not the combined suite's +21-cell inventory. They run in `CertificateScenariosTestCase` and fail, rather +than skip, when the required lab certificate is missing or invalid. + +| Required method | Observation | +| --- | --- | +| `test_credential_x509_custom_http_output_pop` | Real Graph PoP acquisition through the derived certificate-aware factory pipeline, returned/JWT certificate binding, the same cached token, resource HTTP 200, preserved sample middleware, and caller-owned cleanup. | +| `test_custom_http_without_send_certificate_over_mtls_on_behalf_of_regular_endpoint` | Ordinary custom-client OBO POST with the mTLS flag off, signed certificate assertion and exact HTTP 400 / `invalid_request` / AADSTS50027 rejection; no derived certificate-bearing child is created. This is a malformed-user-assertion request capture, not a successful OBO acquisition. | + +These are source-specific tests of the existing factory API. They do not add +Bearer-over-mTLS or FIC APIs, and do not claim live custom-Bearer or real +custom-FIC integration coverage; those combined cases remain in the combined +follow-up. Collecting these tests or passing the offline guards is not evidence +that a live service run passed. + +The custom recorder is isolated in `tests\mtls_http_e2e_helpers.py`; generic +recording/binding helpers remain in `tests\mtls_e2e_helpers.py`, without a +dependency on the custom sample in the foundation branch. Run its offline +privacy/lifecycle guards and existing loopback suite with: + +```powershell +python -m pytest -q tests\test_lab_config.py tests\test_mtls_e2e_helpers.py tests\test_mtls_http_e2e_helpers.py tests\test_mtls_http_client.py +``` + +The four custom-recorder guards render actual pytest failures with argument +display enabled and locals disabled, check constructor/POST/capture redaction, +and retain the sample's response middleware and owner/child cleanup. +`test_custom_capture_failure_traceback_hides_the_expected_assertion` mocks +acquisition: it checks generic recorder privacy, not real FIC support. +The custom sample's output guard prevents dynamic error values from being +printed, and the real loopback TLS fixture explicitly requires TLS 1.2 or newer. + ## 5) Manual-intervention E2E tests Manual tests (interactive browser/device-flow/POP manual scenarios) are separated into: diff --git a/tests/test_mtls_http_e2e_helpers.py b/tests/test_mtls_http_e2e_helpers.py index a1b9eb68..31a2c2c8 100644 --- a/tests/test_mtls_http_e2e_helpers.py +++ b/tests/test_mtls_http_e2e_helpers.py @@ -16,21 +16,6 @@ def setUp(self): self.scenario = test_e2e.CertificateScenariosTestCase( "test_credential_x509_custom_http_output_pop") - def _assert_private_traceback(self, failure, *secrets): - rendered = str(failure.getrepr( - style="long", showlocals=False, funcargs=True)) - self.assertIn(type(failure.value).__name__, rendered) - self.assertFalse( - "PRIVATE KEY" in rendered, - "Rendered tracebacks must not expose private key material.") - for secret in secrets: - fragments = (secret,) + ( - (secret[:32], secret[-32:]) if len(secret) > 64 else ()) - for fragment in fragments: - self.assertFalse( - fragment in rendered, - "Sensitive arguments must not appear in rendered tracebacks.") - def test_custom_capture_failure_traceback_hides_the_expected_assertion(self): assertion = _jwt({"aud": "api://AzureADTokenExchange"}) app = Mock() From c203bfe7b404a0a9a7cfc55ede7b79fc617d329e Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 10:59:35 -0400 Subject: [PATCH 07/11] Keep cleanup regression portable to shared mTLS transports Assert the owned shared transport closes once without closing endpoint wrappers, while retaining the legacy-layout assertions. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_mtls_e2e_helpers.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/test_mtls_e2e_helpers.py b/tests/test_mtls_e2e_helpers.py index e010042f..2cfc7f01 100644 --- a/tests/test_mtls_e2e_helpers.py +++ b/tests/test_mtls_e2e_helpers.py @@ -338,13 +338,21 @@ def test_base_application_closes_ordinary_global_and_regional_clients(self): global_client = app._get_mtls_client(app.authority) app._region_configured = "westus3" regional_client = app._get_mtls_client(app.authority) + shared = getattr(app, "_mtls_http_client", None) with patch.object(app.http_client, "close") as ordinary_close, patch.object( global_client._http_client, "close") as global_close, patch.object( - regional_client._http_client, "close") as regional_close: + regional_client._http_client, "close") as regional_close, ( + patch.object(shared, "close") if shared is not None + else nullcontext()) as shared_close: test_e2e.MtlsPopTestCase._close_default_http_clients(app) ordinary_close.assert_called_once_with() - global_close.assert_called_once_with() - regional_close.assert_called_once_with() + if shared is None: + global_close.assert_called_once_with() + regional_close.assert_called_once_with() + else: + shared_close.assert_called_once_with() + global_close.assert_not_called() + regional_close.assert_not_called() def test_default_bearer_clients_are_closed_once_without_shared_transport(self): ordinary, pop, bearer = Mock(), Mock(), Mock() From 5908a19caec619a83c0abd680755a92a869048ac Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 11:58:21 -0400 Subject: [PATCH 08/11] Sanitize custom recorder transport failure tracebacks Catch Requests errors at the hidden outer POST boundary so real downstream DNS frames cannot render posted assertions or refresh tokens. Cover ordinary and derived clients through the actual Requests/urllib3 path. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/mtls_http_e2e_helpers.py | 9 ++++++++- tests/test_mtls_http_e2e_helpers.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/tests/mtls_http_e2e_helpers.py b/tests/mtls_http_e2e_helpers.py index 83ac4e35..9c58d865 100644 --- a/tests/mtls_http_e2e_helpers.py +++ b/tests/mtls_http_e2e_helpers.py @@ -1,5 +1,7 @@ """Custom-pipeline recording for the mTLS HTTP factory E2E tests.""" +import requests + from sample.confidential_client_mtls_http_client_sample import PolicyHttpClient from tests.mtls_e2e_helpers import token_request_summary @@ -28,7 +30,12 @@ def create_mtls_client(self, binding_context): def post(self, url, **kwargs): __tracebackhide__ = True - response = super().post(url, **kwargs) + try: + response = super().post(url, **kwargs) + except requests.exceptions.RequestException: + raise AssertionError( + "The custom token request transport failed before receiving a response." + ) from None summary = token_request_summary( response, self.expected_binding, self._session.get_adapter(response.request.url), diff --git a/tests/test_mtls_http_e2e_helpers.py b/tests/test_mtls_http_e2e_helpers.py index 31a2c2c8..a85d6a80 100644 --- a/tests/test_mtls_http_e2e_helpers.py +++ b/tests/test_mtls_http_e2e_helpers.py @@ -1,10 +1,12 @@ """Offline custom-recorder guards, not evidence of live E2E acquisitions.""" +import socket import unittest from unittest.mock import Mock, patch import pytest import requests +import urllib3 from tests import test_e2e from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient @@ -45,6 +47,33 @@ def send(adapter, request, **kwargs): self._assert_private_traceback( failure, form["client_assertion"], form["refresh_token"]) + def test_custom_dns_failure_traceback_hides_posted_credentials(self): + form = self._signed_form() + form["assertion"] = _jwt({"sub": "unit-user-assertion-secret"}) + form["refresh_token"] = "unit-refresh-secret-" * 8 + secrets = tuple(form[key] for key in ( + "client_assertion", "assertion", "refresh_token")) + for mtls in (False, True): + with self.subTest(mtls=mtls), RecordingPolicyHttpClient( + expected_binding=self.binding, trust_env=False) as owner: + client = owner.create_mtls_client(self.binding) if mtls else owner + with patch("socket.getaddrinfo", side_effect=socket.gaierror( + socket.EAI_NONAME, "offline DNS resolution failure")) as resolve: + with pytest.raises(( + AssertionError, requests.exceptions.RequestException)) as failure: + client.post(_ENDPOINT, data=form) + self._assert_private_traceback(failure, *secrets) + self.assertIsInstance(failure.value, AssertionError) + self.assertEqual( + "The custom token request transport failed before receiving a response.", + str(failure.value)) + original = failure.value.__context__ + self.assertIsInstance(original, requests.exceptions.ConnectionError) + self.assertIsInstance(original.args[0], urllib3.exceptions.MaxRetryError) + self.assertTrue(failure.value.__suppress_context__) + self.assertTrue(resolve.called) + self.assertEqual([], client.token_requests) + def test_recorder_setup_failure_traceback_hides_the_expected_assertion(self): assertion = _jwt({"aud": "api://AzureADTokenExchange"}) with pytest.raises(TypeError) as failure: From 55434684f8817f68a81e272e652b55d22f80ca7a Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 12:02:19 -0400 Subject: [PATCH 09/11] Sanitize outer custom acquisition tracebacks and guard cert None Rethrow only the private recorder transport failure outside OAuth call frames. Exercise real application client-credential and OBO DNS failures with expanded pytest tracebacks. Keep cert=None cold/cache behavior covered without changing production code. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/mtls_http_e2e_helpers.py | 20 ++++++++++- tests/test_e2e.py | 21 ++++++++---- tests/test_mtls_http_client.py | 19 +++++++++++ tests/test_mtls_http_e2e_helpers.py | 52 ++++++++++++++++++++++++++++- 4 files changed, 103 insertions(+), 9 deletions(-) diff --git a/tests/mtls_http_e2e_helpers.py b/tests/mtls_http_e2e_helpers.py index 9c58d865..88ec582d 100644 --- a/tests/mtls_http_e2e_helpers.py +++ b/tests/mtls_http_e2e_helpers.py @@ -1,11 +1,29 @@ """Custom-pipeline recording for the mTLS HTTP factory E2E tests.""" +from contextlib import contextmanager + import requests from sample.confidential_client_mtls_http_client_sample import PolicyHttpClient from tests.mtls_e2e_helpers import token_request_summary +class _CustomTokenTransportError(AssertionError): + pass + + +@contextmanager +def record_custom_transport_failures(): + """Drop upper OAuth frames only for this recorder's transport failures.""" + __tracebackhide__ = True + try: + yield + except _CustomTokenTransportError: + raise AssertionError( + "The custom token request transport failed before receiving a response." + ) from None + + class RecordingPolicyHttpClient(PolicyHttpClient): """Add safe recording to the existing sample's real certificate pipeline.""" @@ -33,7 +51,7 @@ def post(self, url, **kwargs): try: response = super().post(url, **kwargs) except requests.exceptions.RequestException: - raise AssertionError( + raise _CustomTokenTransportError( "The custom token request transport failed before receiving a response." ) from None summary = token_request_summary( diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 7322ccf2..6c265400 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -515,8 +515,11 @@ def _acquire_recorded( expected_assertion=None): __tracebackhide__ = True if http_client is not None: - result = app.acquire_token_for_client( - scope, mtls_proof_of_possession=pop) + from tests.mtls_http_e2e_helpers import record_custom_transport_failures + + with record_custom_transport_failures(): + result = app.acquire_token_for_client( + scope, mtls_proof_of_possession=pop) records = self._assert_factory_used( http_client, binding).token_requests else: @@ -676,12 +679,14 @@ def _acquire_x509(self, scope, pop=False, send=False, region=False, custom=False credential = dict(credential, send_certificate_over_mtls=True) http_client = None if custom: - from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient + from tests.mtls_http_e2e_helpers import ( + RecordingPolicyHttpClient, record_custom_transport_failures) http_client = RecordingPolicyHttpClient(expected_binding=binding) self.addCleanup(http_client.close) app = self._sni_app(credential, azure_region=region, http_client=http_client) - with (nullcontext() if custom else record_token_requests(binding)) as records: + with (record_custom_transport_failures() if custom + else record_token_requests(binding)) as records: result = app.acquire_token_for_client( scope, mtls_proof_of_possession=pop) self._assert_token_result( @@ -755,15 +760,17 @@ def test_credential_x509_custom_http_output_pop(self): result["access_token"], result["binding_certificate"]) def test_custom_http_without_send_certificate_over_mtls_on_behalf_of_regular_endpoint(self): - from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient + from tests.mtls_http_e2e_helpers import ( + RecordingPolicyHttpClient, record_custom_transport_failures) credential, binding = self._certificate() credential = dict(credential, send_certificate_over_mtls=False) http_client = RecordingPolicyHttpClient(expected_binding=binding) self.addCleanup(http_client.close) app = self._sni_app(credential, http_client=http_client) - result = app.acquire_token_on_behalf_of( - "fake-user-assertion-for-request-capture", self._KEY_VAULT_SCOPE) + with record_custom_transport_failures(): + result = app.acquire_token_on_behalf_of( + "fake-user-assertion-for-request-capture", self._KEY_VAULT_SCOPE) self.assertEqual([], http_client._children) request = self._assert_token_request( http_client.token_requests, self._KEY_VAULT_SCOPE, diff --git a/tests/test_mtls_http_client.py b/tests/test_mtls_http_client.py index c166d036..865669c0 100644 --- a/tests/test_mtls_http_client.py +++ b/tests/test_mtls_http_client.py @@ -149,6 +149,25 @@ def test_global_and_regional_clients_share_certificate_pipeline(self, discovery) regional.configuration["token_endpoint"]) self.ordinary.create_mtls_client.assert_called_once() + def test_custom_factory_cert_none_works_on_cold_and_cached_pop(self, discovery): + app = self._app() + original = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, cert=None) + cached = app.acquire_token_for_client( + ["s1"], mtls_proof_of_possession=True, cert=None) + binding = self.ordinary.create_mtls_client.call_args.args[0] + for result in (original, cached): + self.assertEqual("mtls_pop", result["token_type"]) + self.assertEqual( + binding.thumbprint_sha256, + result["binding_certificate"].thumbprint_sha256) + self.assertEqual("cache", cached["token_source"]) + self.assertEqual(original["access_token"], cached["access_token"]) + self.ordinary.create_mtls_client.assert_called_once() + self.pipeline.post.assert_called_once() + self.assertIsNone(self.pipeline.post.call_args.kwargs["cert"]) + self.ordinary.post.assert_not_called() + def test_concurrent_creation_does_not_orphan_pipelines(self, discovery): app = self._app() diff --git a/tests/test_mtls_http_e2e_helpers.py b/tests/test_mtls_http_e2e_helpers.py index a85d6a80..8eb29c47 100644 --- a/tests/test_mtls_http_e2e_helpers.py +++ b/tests/test_mtls_http_e2e_helpers.py @@ -3,14 +3,16 @@ import socket import unittest from unittest.mock import Mock, patch +from urllib.parse import parse_qs +import msal import pytest import requests import urllib3 from tests import test_e2e from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient -from tests.test_mtls_e2e_helpers import CertificateFixture, _ENDPOINT, _jwt +from tests.test_mtls_e2e_helpers import CertificateFixture, _CREDENTIAL, _ENDPOINT, _jwt class TestCustomMtlsE2eHelpers(CertificateFixture): @@ -74,6 +76,54 @@ def test_custom_dns_failure_traceback_hides_posted_credentials(self): self.assertTrue(resolve.called) self.assertEqual([], client.token_requests) + def test_custom_acquisition_dns_traceback_hides_oauth_arguments(self): + authority = self.scenario._SNI_ALLOWLISTED_AUTHORITY + metadata = { + "authorization_endpoint": authority + "/oauth2/v2.0/authorize", + "token_endpoint": authority + "/oauth2/v2.0/token", + "issuer": authority + "/v2.0", + } + bodies = [] + prepare = requests.Session.prepare_request + + def capture(session, request): + prepared = prepare(session, request) + bodies.append(prepared.body) + return prepared + + with RecordingPolicyHttpClient( + expected_binding=self.binding, trust_env=False) as owner, patch( + "msal.authority.tenant_discovery", return_value=metadata): + app = msal.ConfidentialClientApplication( + "client", authority=authority, client_credential=_CREDENTIAL, + http_client=owner, azure_region=False, instance_discovery=False) + for operation in ("recorded", "obo"): + with self.subTest(operation=operation), patch.object( + requests.Session, "prepare_request", + autospec=True, side_effect=capture), patch( + "socket.getaddrinfo", side_effect=socket.gaierror( + socket.EAI_NONAME, "offline DNS resolution failure")): + with pytest.raises(AssertionError) as failure: + if operation == "recorded": + self.scenario._acquire_recorded( + app, self.scenario._KEY_VAULT_SCOPE, + self.binding, http_client=owner) + else: + with patch.object(self.scenario, "_certificate", + return_value=(_CREDENTIAL, self.binding)), patch.object( + self.scenario, "_sni_app", return_value=app): + try: + self.scenario.test_custom_http_without_send_certificate_over_mtls_on_behalf_of_regular_endpoint() + finally: + self.scenario.doCleanups() + form = parse_qs(bodies[-1]) + self._assert_private_traceback( + failure, form["client_assertion"][0], + *form.get("assertion", [])) + self.assertTrue(failure.value.__suppress_context__) + self.assertEqual([], owner.token_requests) + self.assertEqual([], owner._children) + def test_recorder_setup_failure_traceback_hides_the_expected_assertion(self): assertion = _jwt({"aud": "api://AzureADTokenExchange"}) with pytest.raises(TypeError) as failure: From 6d408b3f0087b45be91d6e1b48856df255596b4a Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 12:17:49 -0400 Subject: [PATCH 10/11] Isolate synthetic OBO acquisition from traceback source rendering Python 3.9 renders the inline dummy assertion in the visible leaf source even after transport exceptions are sanitized. Keep only the acquisition call in a hidden helper; leave every validation assertion visible and preserve all full/prefix/suffix privacy checks. Reproduced before and verified after on Python 3.9.25 with pytest 8.4.2. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_e2e.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 6c265400..4981b291 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -759,18 +759,23 @@ def test_credential_x509_custom_http_output_pop(self): self._call_graph_with_mtls_pop_token( result["access_token"], result["binding_certificate"]) + def _acquire_custom_http_obo_invalid_grant(self, app): + __tracebackhide__ = True + from tests.mtls_http_e2e_helpers import record_custom_transport_failures + + with record_custom_transport_failures(): + return app.acquire_token_on_behalf_of( + "fake-user-assertion-for-request-capture", self._KEY_VAULT_SCOPE) + def test_custom_http_without_send_certificate_over_mtls_on_behalf_of_regular_endpoint(self): - from tests.mtls_http_e2e_helpers import ( - RecordingPolicyHttpClient, record_custom_transport_failures) + from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient credential, binding = self._certificate() credential = dict(credential, send_certificate_over_mtls=False) http_client = RecordingPolicyHttpClient(expected_binding=binding) self.addCleanup(http_client.close) app = self._sni_app(credential, http_client=http_client) - with record_custom_transport_failures(): - result = app.acquire_token_on_behalf_of( - "fake-user-assertion-for-request-capture", self._KEY_VAULT_SCOPE) + result = self._acquire_custom_http_obo_invalid_grant(app) self.assertEqual([], http_client._children) request = self._assert_token_request( http_client.token_requests, self._KEY_VAULT_SCOPE, From 7a5756bea00e2e56908247b416950a6296af072e Mon Sep 17 00:00:00 2001 From: Robbie Ginsburg Date: Tue, 27 Oct 2026 12:44:53 -0400 Subject: [PATCH 11/11] Sanitize custom recorder summary validation at the acquisition boundary Use the shared private validation sentinel only around token_request_summary and rethrow it outside OAuth frames with a static validation diagnostic, distinct from transport errors. Cover real-app bad JSON and signed-assertion validation in client credentials and OBO without masking unrelated assertions. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/mtls_http_e2e_helpers.py | 21 ++++++++--- tests/test_mtls_http_e2e_helpers.py | 58 +++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 6 deletions(-) diff --git a/tests/mtls_http_e2e_helpers.py b/tests/mtls_http_e2e_helpers.py index 88ec582d..807c44cf 100644 --- a/tests/mtls_http_e2e_helpers.py +++ b/tests/mtls_http_e2e_helpers.py @@ -5,7 +5,7 @@ import requests from sample.confidential_client_mtls_http_client_sample import PolicyHttpClient -from tests.mtls_e2e_helpers import token_request_summary +from tests.mtls_e2e_helpers import _TokenValidationError, token_request_summary class _CustomTokenTransportError(AssertionError): @@ -14,7 +14,7 @@ class _CustomTokenTransportError(AssertionError): @contextmanager def record_custom_transport_failures(): - """Drop upper OAuth frames only for this recorder's transport failures.""" + """Drop upper OAuth frames only for private recorder failures.""" __tracebackhide__ = True try: yield @@ -22,6 +22,10 @@ def record_custom_transport_failures(): raise AssertionError( "The custom token request transport failed before receiving a response." ) from None + except _TokenValidationError: + raise AssertionError( + "The real token exchange failed recording validation." + ) from None class RecordingPolicyHttpClient(PolicyHttpClient): @@ -54,10 +58,15 @@ def post(self, url, **kwargs): raise _CustomTokenTransportError( "The custom token request transport failed before receiving a response." ) from None - summary = token_request_summary( - response, self.expected_binding, - self._session.get_adapter(response.request.url), - expected_assertion=self.expected_assertion) + adapter = self._session.get_adapter(response.request.url) + try: + summary = token_request_summary( + response, self.expected_binding, adapter, + expected_assertion=self.expected_assertion) + except AssertionError: + raise _TokenValidationError( + "The real token exchange failed recording validation." + ) from None if summary is not None: self.token_requests.append(summary) return response diff --git a/tests/test_mtls_http_e2e_helpers.py b/tests/test_mtls_http_e2e_helpers.py index 8eb29c47..5d6498b0 100644 --- a/tests/test_mtls_http_e2e_helpers.py +++ b/tests/test_mtls_http_e2e_helpers.py @@ -11,6 +11,7 @@ import urllib3 from tests import test_e2e +from tests.mtls_e2e_helpers import unrelated_binding from tests.mtls_http_e2e_helpers import RecordingPolicyHttpClient from tests.test_mtls_e2e_helpers import CertificateFixture, _CREDENTIAL, _ENDPOINT, _jwt @@ -131,6 +132,63 @@ def test_recorder_setup_failure_traceback_hides_the_expected_assertion(self): expected_assertion=assertion, unexpected_option=True) self._assert_private_traceback(failure, assertion) + def _assert_custom_summary_validation_privacy(self, invalid_json): + authority = self.scenario._SNI_ALLOWLISTED_AUTHORITY + metadata = { + "authorization_endpoint": authority + "/oauth2/v2.0/authorize", + "token_endpoint": authority + "/oauth2/v2.0/token", + "issuer": authority + "/v2.0", + } + posted = [] + + def send(adapter, request, **kwargs): + posted.append(parse_qs(request.body)) + response = requests.Response() + response.status_code = 400 + response._content = ( + b"not-json" if invalid_json else b'{"error":"invalid_grant"}') + response.request = request + return response + + with RecordingPolicyHttpClient( + expected_binding=self.binding if invalid_json else unrelated_binding(), + trust_env=False) as owner, patch( + "msal.authority.tenant_discovery", return_value=metadata): + app = msal.ConfidentialClientApplication( + "client", authority=authority, client_credential=_CREDENTIAL, + http_client=owner, azure_region=False, instance_discovery=False) + for operation in ("recorded", "obo"): + with self.subTest(operation=operation), patch.object( + requests.adapters.HTTPAdapter, "send", + autospec=True, side_effect=send): + with pytest.raises(AssertionError) as failure: + if operation == "recorded": + self.scenario._acquire_recorded( + app, self.scenario._KEY_VAULT_SCOPE, + self.binding, http_client=owner) + else: + self.scenario._acquire_custom_http_obo_invalid_grant(app) + form = posted[-1] + self._assert_private_traceback( + failure, form["client_assertion"][0], + *form.get("assertion", [])) + self.assertEqual( + "The real token exchange failed recording validation.", + str(failure.value)) + self.assertTrue(failure.value.__suppress_context__) + self.assertEqual( + "_TokenValidationError", + type(failure.value.__context__).__name__) + self.assertEqual([], owner.token_requests) + self.assertEqual({}, app.token_cache._cache.get("AccessToken", {})) + self.assertEqual(len(posted), owner.response_count) + + def test_custom_acquisition_bad_json_traceback_hides_oauth_arguments(self): + self._assert_custom_summary_validation_privacy(invalid_json=True) + + def test_custom_acquisition_invalid_signed_assertion_traceback_hides_oauth_arguments(self): + self._assert_custom_summary_validation_privacy(invalid_json=False) + def test_custom_recorder_retains_sample_pipeline_and_owner_cleanup(self): def send(adapter, request, **kwargs): response = self._response()