diff --git a/packages/google-auth/google/auth/transport/requests.py b/packages/google-auth/google/auth/transport/requests.py index 3f1ec345765c..6c263051b8df 100644 --- a/packages/google-auth/google/auth/transport/requests.py +++ b/packages/google-auth/google/auth/transport/requests.py @@ -17,6 +17,7 @@ from __future__ import absolute_import import collections +import contextlib import copy import functools import http.client as http_client @@ -116,6 +117,36 @@ def __exit__(self, exc_type, exc_value, traceback): raise self._timeout_error_type() +def _replace_adapter(session, prefix, adapter): + """Mounts an adapter on a session without mutating its adapters in place. + + :meth:`requests.Session.mount` mutates the ``session.adapters`` + OrderedDict in place. If another thread is iterating over it in + :meth:`requests.Session.get_adapter` at the same time, that thread fails + with ``RuntimeError: OrderedDict mutated during iteration``. This builds + an updated copy and assigns it in one step instead, so threads that are + already iterating over the old dictionary are not affected. + + This only protects threads that read ``session.adapters``. The copy and + the assignment are not atomic together, so if two threads call this (or + :meth:`requests.Session.mount`) on the same session at the same time, + one of the new adapters can be lost. Callers must serialize writes, for + example by holding a lock. + + Args: + session (requests.Session): The session to mount the adapter on. + prefix (str): The URL prefix to mount the adapter for. + adapter (requests.adapters.BaseAdapter): The adapter to mount. + """ + new_adapters = session.adapters.copy() + new_adapters[prefix] = adapter + # Keep the longest prefixes first, as requests.Session.mount() does, so + # that get_adapter() matches the most specific prefix. + session.adapters = collections.OrderedDict( + sorted(new_adapters.items(), key=lambda item: len(item[0]), reverse=True) + ) + + class Request(transport.Request): """Requests request adapter. @@ -213,18 +244,7 @@ def _configure_mtls_if_needed(self, url): pool_block=session_adapter._pool_block, ) - # Replace session.adapters atomically instead of calling session.mount(), - # which mutates the OrderedDict in place and can raise RuntimeError if - # another thread is iterating over session.adapters in get_adapter(). - new_adapters = self.session.adapters.copy() - new_adapters[prefix] = self._mtls_adapter - self.session.adapters = collections.OrderedDict( - sorted( - new_adapters.items(), - key=lambda item: len(item[0]), - reverse=True, - ) - ) + _replace_adapter(self.session, prefix, self._mtls_adapter) def __call__( self, @@ -514,6 +534,13 @@ def __init__( self._refresh_timeout = refresh_timeout self._is_mtls = False self._default_host = default_host + self._mtls_reauth_lock = threading.Lock() + self._mtls_reauth_lock_owner = None + # Incremented after every client certificate check that runs on a + # 401 response, so that threads that were waiting for + # _mtls_reauth_lock can skip a check that ran after they sent their + # request. + self._mtls_cert_check_generation = 0 if auth_request is None: self._auth_request_session = requests.Session() @@ -541,6 +568,26 @@ def __init__( "https://{}/".format(self._default_host) if self._default_host else None ) + @contextlib.contextmanager + def _acquire_mtls_reauth_lock(self): + """Acquires ``_mtls_reauth_lock`` reentrantly for the current thread. + + A reentrant wrapper around ``threading.Lock`` is used so that a + ``client_cert_callback`` that sends a request through this session on + the same thread does not deadlock, while keeping ``.locked()`` + available on Python < 3.14. + """ + current_thread = threading.get_ident() + if self._mtls_reauth_lock_owner == current_thread: + yield + return + with self._mtls_reauth_lock: + self._mtls_reauth_lock_owner = current_thread + try: + yield + finally: + self._mtls_reauth_lock_owner = None + def configure_mtls_channel(self, client_cert_callback=None): """Configure the client certificate and key for SSL connection. @@ -557,16 +604,26 @@ def configure_mtls_channel(self, client_cert_callback=None): If the callback is None, application default SSL credentials will be used. - .. warning:: - Calling this method mutates the underlying `requests.Session` adapter - dictionary. It is not thread-safe to call this explicitly while other - threads are making requests. + .. note:: + Replaces and closes adapters. Calls are serialized with each other + and with the automatic reconfiguration that runs on 401 responses. + ``client_cert_callback`` runs while the lock is held. It may send + requests through this session on the same thread, but it must not + wait for other threads that use this session. Raises: google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel creation failed for any reason. The existing session state (such as adapter mounts) remains unmodified if this error is raised. """ + with self._acquire_mtls_reauth_lock(): + self._configure_mtls_channel_locked(client_cert_callback) + + def _configure_mtls_channel_locked(self, client_cert_callback=None): + """Implements :meth:`configure_mtls_channel`. + + The caller must hold ``_mtls_reauth_lock``. + """ use_client_cert = google.auth.transport._mtls_helper.check_use_client_cert() if not use_client_cert: return @@ -580,6 +637,8 @@ def configure_mtls_channel(self, client_cert_callback=None): client_cert_callback ) + # Read current adapters after the callback returns so any adapter + # mounted by a reentrant call inside the callback is closed below. old_adapter = self.adapters.get("https://") kwargs = {} @@ -641,13 +700,18 @@ def configure_mtls_channel(self, client_cert_callback=None): new_exc = exceptions.MutualTLSChannelError(caught_exc) raise new_exc from caught_exc - self.mount("https://", new_adapter) + _replace_adapter(self, "https://", new_adapter) if old_adapter is not None and old_adapter is not new_adapter: old_adapter.close() if self._auth_request_session is not None and new_auth_adapter is not None: - self._auth_request_session.mount("https://", new_auth_adapter) + # Request._configure_mtls_if_needed() also writes the adapters of + # this session during token refresh, under the Request's own lock. + with self._auth_request._mtls_lock: + _replace_adapter( + self._auth_request_session, "https://", new_auth_adapter + ) if ( old_auth_adapter is not None @@ -724,6 +788,13 @@ def request( self.credentials.before_request(auth_request, method, url, request_headers) remaining_time = guard.remaining_timeout + # Remember the client certificate that this request is sent with, and + # how many certificate checks have run so far. If the request receives + # a 401 response, this tells whether another thread has already + # checked or rotated the certificate in the meantime. + cert_at_request = getattr(self, "_cached_cert", None) + check_generation_at_request = self._mtls_cert_check_generation + with TimeoutGuard(remaining_time) as guard: _helpers.request_log(_LOGGER, method, url, data, headers) response = super(AuthorizedSession, self).request( @@ -749,33 +820,11 @@ def request( if response.status_code == http_client.UNAUTHORIZED: use_mtls = self.is_mtls and _mtls_helper.is_mtls_endpoint(url) if use_mtls: - ( - call_cert_bytes, - call_key_bytes, - cached_fingerprint, - current_cert_fingerprint, - ) = _mtls_helper.check_parameters_for_unauthorized_response( - self._cached_cert - ) - if cached_fingerprint != current_cert_fingerprint: - try: - _LOGGER.info( - "Client certificate has changed, reconfiguring mTLS " - "channel." - ) - self.configure_mtls_channel( - lambda: (call_cert_bytes, call_key_bytes) - ) - except Exception as e: - _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) - raise exceptions.MutualTLSChannelError( - "Failed to reconfigure mTLS channel" - ) from e - else: - _LOGGER.info( - "Skipping reconfiguration of mTLS channel because the client" - " certificate has not changed." + with TimeoutGuard(remaining_time) as guard: + self._reconfigure_mtls_channel_if_cert_changed( + cert_at_request, check_generation_at_request ) + remaining_time = guard.remaining_timeout _LOGGER.info( "Refreshing credentials due to a %s response. Attempt %s/%s.", response.status_code, @@ -810,6 +859,98 @@ def request( return response + def _reconfigure_mtls_channel_if_cert_changed( + self, cert_at_request, check_generation_at_request + ): + """Reconfigures the mTLS channel if the client certificate has changed. + + This is called when a request to an mTLS endpoint receives a 401 + response. The certificate check and the channel reconfiguration run + while holding ``_mtls_reauth_lock``. When several threads receive a 401 + response at the same time, only the first one reads the current + certificate, and remounts the adapters if the certificate has + changed. Threads that were waiting for the lock see that the + certificate was already checked after they sent their request, and + skip the check. + + Args: + cert_at_request (Optional[bytes]): The cached client certificate at + the time the request that received the 401 response was sent. + check_generation_at_request (int): The value of + ``_mtls_cert_check_generation`` at the time the request that + received the 401 response was sent. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If the client + certificate has changed but the mTLS channel could not be + reconfigured. + """ + with self._acquire_mtls_reauth_lock(): + if not self.is_mtls: + # mTLS was disabled by another thread while this one waited. + return + + if getattr(self, "_cached_cert", None) != cert_at_request: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because it has already" + " been reconfigured with a new client certificate." + ) + return + + if self._mtls_cert_check_generation != check_generation_at_request: + # Another thread checked the certificate after this request was + # sent, whether or not the certificate had changed. Checking it + # again would only load the same certificate again. + _LOGGER.info( + "Skipping client certificate check because another request" + " already checked it." + ) + return + + try: + ( + call_cert_bytes, + call_key_bytes, + cached_fingerprint, + current_cert_fingerprint, + ) = _mtls_helper.check_parameters_for_unauthorized_response( + cert_at_request + ) + except Exception as e: + # Failing to load the current certificate (for example, because + # the certificate file is missing or the certificate provider + # failed) must not prevent the token refresh that follows. + _LOGGER.warning( + "Failed to check whether the client certificate has changed: %s", + e, + exc_info=True, + ) + return + finally: + self._mtls_cert_check_generation += 1 + + if cached_fingerprint == current_cert_fingerprint: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because the client" + " certificate has not changed." + ) + return + + try: + _LOGGER.info( + "Client certificate has changed, reconfiguring mTLS channel." + ) + self._configure_mtls_channel_locked( + lambda: (call_cert_bytes, call_key_bytes) + ) + except Exception as e: + _LOGGER.error( + "Failed to reconfigure mTLS channel: %s", e, exc_info=True + ) + raise exceptions.MutualTLSChannelError( + "Failed to reconfigure mTLS channel" + ) from e + @property def is_mtls(self): """Indicates if the created SSL channel is mutual TLS.""" diff --git a/packages/google-auth/google/auth/transport/urllib3.py b/packages/google-auth/google/auth/transport/urllib3.py index b4b4ebe1daac..59e69cdadef0 100644 --- a/packages/google-auth/google/auth/transport/urllib3.py +++ b/packages/google-auth/google/auth/transport/urllib3.py @@ -16,8 +16,10 @@ from __future__ import absolute_import +import contextlib import http.client as http_client import logging +import threading import warnings # Certifi is Mozilla's certificate bundle. Urllib3 needs a certificate bundle @@ -308,6 +310,13 @@ def __init__( # credentials.refresh). self._request = Request(self.http) self._is_mtls = False + self._mtls_reauth_lock = threading.Lock() + self._mtls_reauth_lock_owner = None + # Incremented after every client certificate check that runs on a + # 401 response, so that threads that were waiting for + # _mtls_reauth_lock can skip a check that ran after they sent their + # request. + self._mtls_cert_check_generation = 0 # https://google.aip.dev/auth/4111 # Attempt to use self-signed JWTs when a service account is used. @@ -318,6 +327,31 @@ def __init__( super(AuthorizedHttp, self).__init__() + @property + def is_mtls(self): + """Indicates if the created SSL channel is mutual TLS.""" + return self._is_mtls + + @contextlib.contextmanager + def _acquire_mtls_reauth_lock(self): + """Acquires ``_mtls_reauth_lock`` reentrantly for the current thread. + + A reentrant wrapper around ``threading.Lock`` is used so that a + ``client_cert_callback`` that sends a request through this instance on + the same thread does not deadlock, while keeping ``.locked()`` + available on Python < 3.14. + """ + current_thread = threading.get_ident() + if self._mtls_reauth_lock_owner == current_thread: + yield + return + with self._mtls_reauth_lock: + self._mtls_reauth_lock_owner = current_thread + try: + yield + finally: + self._mtls_reauth_lock_owner = None + def configure_mtls_channel(self, client_cert_callback=None): """Configures mutual TLS channel using the given client_cert_callback or application default SSL credentials. @@ -334,10 +368,13 @@ def configure_mtls_channel(self, client_cert_callback=None): If the callback is None, application default SSL credentials will be used. - .. warning:: - Calling this method mutates the underlying `urllib3.PoolManager`. - It is not thread-safe to call this explicitly while other - threads are making requests. + .. note:: + Replaces the pool manager and clears the old one. Calls are + serialized with each other and with the automatic reconfiguration + that runs on 401 responses. ``client_cert_callback`` runs while the + lock is held. It may send requests through this instance on the + same thread, but it must not wait for other threads that use this + instance. Returns: True if the channel is mutual TLS and False otherwise. @@ -347,6 +384,14 @@ def configure_mtls_channel(self, client_cert_callback=None): creation failed for any reason. The existing channel state (the HTTP client) remains unmodified if this error is raised. """ + with self._acquire_mtls_reauth_lock(): + return self._configure_mtls_channel_locked(client_cert_callback) + + def _configure_mtls_channel_locked(self, client_cert_callback=None): + """Implements :meth:`configure_mtls_channel`. + + The caller must hold ``_mtls_reauth_lock``. + """ use_client_cert = transport._mtls_helper.check_use_client_cert() if not use_client_cert: return False @@ -371,6 +416,8 @@ def configure_mtls_channel(self, client_cert_callback=None): new_exc = exceptions.MutualTLSChannelError(caught_exc) raise new_exc from caught_exc + # Read the current pool manager after the callback returns so any pool + # manager created by a reentrant call inside the callback is cleared below. old_http = self.http self.http = new_http @@ -414,6 +461,13 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): self.credentials.before_request(self._request, method, url, request_headers) + # Remember the client certificate that this request is sent with, and + # how many certificate checks have run so far. If the request receives + # a 401 response, this tells whether another thread has already + # checked or rotated the certificate in the meantime. + cert_at_request = getattr(self, "_cached_cert", None) + check_generation_at_request = self._mtls_cert_check_generation + response = self.http.urlopen( method, url, body=body, headers=request_headers, **kwargs ) @@ -430,39 +484,11 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): and _credential_refresh_attempt < self._max_refresh_attempts ): if response.status == http_client.UNAUTHORIZED: - use_mtls = self._is_mtls and _mtls_helper.is_mtls_endpoint(url) + use_mtls = self.is_mtls and _mtls_helper.is_mtls_endpoint(url) if use_mtls: - ( - call_cert_bytes, - call_key_bytes, - cached_fingerprint, - current_cert_fingerprint, - ) = _mtls_helper.check_parameters_for_unauthorized_response( - self._cached_cert + self._reconfigure_mtls_channel_if_cert_changed( + cert_at_request, check_generation_at_request ) - if cached_fingerprint != current_cert_fingerprint: - try: - _LOGGER.info( - "Client certificate has changed, reconfiguring mTLS " - "channel." - ) - self.configure_mtls_channel( - client_cert_callback=lambda: ( - call_cert_bytes, - call_key_bytes, - ) - ) - except Exception as e: - _LOGGER.error("Failed to reconfigure mTLS channel: %s", e) - raise exceptions.MutualTLSChannelError( - "Failed to reconfigure mTLS channel" - ) from e - - else: - _LOGGER.info( - "Skipping reconfiguration of mTLS channel because the " - "client certificate has not changed." - ) _LOGGER.info( "Refreshing credentials due to a %s response. Attempt %s/%s.", @@ -485,6 +511,98 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs): return response + def _reconfigure_mtls_channel_if_cert_changed( + self, cert_at_request, check_generation_at_request + ): + """Reconfigures the mTLS channel if the client certificate has changed. + + This is called when a request to an mTLS endpoint receives a 401 + response. The certificate check and the channel reconfiguration run + while holding ``_mtls_reauth_lock``. When several threads receive a 401 + response at the same time, only the first one reads the current + certificate, and replaces the pool manager if the certificate has + changed. Threads that were waiting for the lock see that the + certificate was already checked after they sent their request, and + skip the check. + + Args: + cert_at_request (Optional[bytes]): The cached client certificate at + the time the request that received the 401 response was sent. + check_generation_at_request (int): The value of + ``_mtls_cert_check_generation`` at the time the request that + received the 401 response was sent. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If the client + certificate has changed but the mTLS channel could not be + reconfigured. + """ + with self._acquire_mtls_reauth_lock(): + if not self.is_mtls: + # mTLS was disabled by another thread while this one waited. + return + + if getattr(self, "_cached_cert", None) != cert_at_request: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because it has already" + " been reconfigured with a new client certificate." + ) + return + + if self._mtls_cert_check_generation != check_generation_at_request: + # Another thread checked the certificate after this request was + # sent, whether or not the certificate had changed. Checking it + # again would only load the same certificate again. + _LOGGER.info( + "Skipping client certificate check because another request" + " already checked it." + ) + return + + try: + ( + call_cert_bytes, + call_key_bytes, + cached_fingerprint, + current_cert_fingerprint, + ) = _mtls_helper.check_parameters_for_unauthorized_response( + cert_at_request + ) + except Exception as e: + # Failing to load the current certificate (for example, because + # the certificate file is missing or the certificate provider + # failed) must not prevent the token refresh that follows. + _LOGGER.warning( + "Failed to check whether the client certificate has changed: %s", + e, + exc_info=True, + ) + return + finally: + self._mtls_cert_check_generation += 1 + + if cached_fingerprint == current_cert_fingerprint: + _LOGGER.info( + "Skipping reconfiguration of mTLS channel because the " + "client certificate has not changed." + ) + return + + try: + _LOGGER.info( + "Client certificate has changed, reconfiguring mTLS channel." + ) + self._configure_mtls_channel_locked( + client_cert_callback=lambda: (call_cert_bytes, call_key_bytes) + ) + except Exception as e: + _LOGGER.error( + "Failed to reconfigure mTLS channel: %s", e, exc_info=True + ) + raise exceptions.MutualTLSChannelError( + "Failed to reconfigure mTLS channel" + ) from e + # Proxy methods for compliance with the urllib3.PoolManager interface def __enter__(self): diff --git a/packages/google-auth/tests/transport/test_requests.py b/packages/google-auth/tests/transport/test_requests.py index 5ca8788ca42b..ab484ca7131c 100644 --- a/packages/google-auth/tests/transport/test_requests.py +++ b/packages/google-auth/tests/transport/test_requests.py @@ -20,6 +20,7 @@ import os import pickle import ssl +import threading from unittest import mock import freezegun @@ -730,6 +731,47 @@ def test_request_timeout_w_refresh_timeout_timeout_error(self, frozen_time): "GET", self.TEST_URL, timeout=60, max_allowed_time=2.9 ) + def test_request_timeout_w_mtls_reconfigure_timeout_error(self, frozen_time): + tick_one_second = functools.partial( + frozen_time.tick, delta=datetime.timedelta(seconds=1.0) + ) + + credentials = mock.Mock( + wraps=TimeTickCredentialsStub(time_tick=tick_one_second) + ) + adapter = TimeTickAdapterStub( + time_tick=tick_one_second, + responses=[ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ], + ) + + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=100 + ) + authed_session.mount(self.MTLS_TEST_URL, adapter) + authed_session._is_mtls = True + authed_session._cached_cert = b"old_cert" + + def check_params(cached_cert): + tick_one_second() + return (b"old_cert", b"old_key", "same_fp", "same_fp") + + # First request (1s) + mTLS cert check (1s) = 2s, which exceeds + # max_allowed_time=1.9 inside the mTLS TimeoutGuard before refresh. + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=check_params, + ): + with pytest.raises(requests.exceptions.Timeout): + authed_session.request( + "GET", self.MTLS_TEST_URL, timeout=60, max_allowed_time=1.9 + ) + + credentials.refresh.assert_not_called() + def test_authorized_session_without_default_host(self): credentials = mock.create_autospec(service_account.Credentials) @@ -1213,7 +1255,7 @@ def test_cert_rotation_failure_raises_error(self): ): with mock.patch.object( authed_session, - "configure_mtls_channel", + "_configure_mtls_channel_locked", side_effect=Exception("Failed to reconfigure"), ): with pytest.raises(exceptions.MutualTLSChannelError): @@ -1224,7 +1266,12 @@ def test_cert_rotation_failure_raises_error(self): def test_cert_rotation_check_params_fails(self): credentials = mock.Mock(wraps=CredentialsStub()) - adapter = AdapterStub([make_response(status=http_client.UNAUTHORIZED)]) + adapter = AdapterStub( + [ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ] + ) authed_session = google.auth.transport.requests.AuthorizedSession( credentials, refresh_timeout=60 @@ -1237,11 +1284,11 @@ def test_cert_rotation_check_params_fails(self): "google.auth.transport.requests._mtls_helper.check_parameters_for_unauthorized_response", side_effect=Exception("check_params failed"), ) as mock_check_params: - with pytest.raises(Exception, match="check_params failed"): - authed_session.request("GET", self.MTLS_TEST_URL) + response = authed_session.request("GET", self.MTLS_TEST_URL) mock_check_params.assert_called_once() - credentials.refresh.assert_not_called() + assert credentials.refresh.called + assert response.status_code == http_client.OK def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): """ @@ -1433,6 +1480,512 @@ def test_configure_mtls_channel_subsequent_disabled(self): requests.adapters.HTTPAdapter, ) + def test_unauthorized_cert_discovery_exception_proceeds_to_token_refresh(self): + credentials = mock.Mock(wraps=CredentialsStub()) + adapter = AdapterStub( + [ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ] + ) + authed_session = google.auth.transport.requests.AuthorizedSession(credentials) + authed_session.mount(self.MTLS_TEST_URL, adapter) + authed_session._is_mtls = True + # Simulate discovery failure (e.g. Enterprise cert provider error or file missing) + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=Exception("Certificate discovery failed"), + ): + result = authed_session.request("GET", self.MTLS_TEST_URL) + # Token refresh should still be called despite discovery failure + credentials.refresh.assert_called_once() + assert result.status_code == http_client.OK + assert authed_session._mtls_cert_check_generation == 1 + assert not authed_session._mtls_reauth_lock.locked() + + def test_cert_check_skipped_when_cert_rotated_during_request(self): + """ + Tests that a request that receives a 401 response does not read the + client certificate again if another thread has already rotated it + while the request was in flight. + """ + credentials = mock.Mock(wraps=CredentialsStub()) + adapter = AdapterStub( + [ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ] + ) + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials, refresh_timeout=60 + ) + authed_session.mount(self.MTLS_TEST_URL, adapter) + authed_session._is_mtls = True + authed_session._cached_cert = b"old_cert" + send_request = adapter.send + + def send_request_and_rotate_cert(request, **kwargs): + # Simulate another thread rotating the client certificate while + # this request is in flight. + authed_session._cached_cert = b"new_cert" + return send_request(request, **kwargs) + + adapter.send = send_request_and_rotate_cert + + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + ) as mock_check_params: + with mock.patch.object( + authed_session, "_configure_mtls_channel_locked" + ) as mock_configure_mtls_channel: + result = authed_session.request("GET", self.MTLS_TEST_URL) + + assert result.status_code == http_client.OK + mock_check_params.assert_not_called() + mock_configure_mtls_channel.assert_not_called() + credentials.refresh.assert_called_once() + + def test_reconfigure_mtls_channel_skipped_if_mtls_disabled(self): + """ + Tests that the client certificate is not checked if another thread + disabled mTLS while this thread was waiting for the lock. + """ + authed_session = google.auth.transport.requests.AuthorizedSession( + mock.sentinel.credentials + ) + authed_session._is_mtls = False + + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + ) as mock_check_params: + authed_session._reconfigure_mtls_channel_if_cert_changed(b"old_cert", 0) + + mock_check_params.assert_not_called() + assert not authed_session._mtls_reauth_lock.locked() + + @mock.patch.object(requests.adapters.HTTPAdapter, "send", autospec=True) + def test_concurrent_unauthorized_responses_reconfigure_mtls_channel_once( + self, mock_send + ): + """ + Tests that when several threads that share an AuthorizedSession receive + a 401 response at the same time, the client certificate is read and the + real mTLS channel reconfiguration runs only once, and every request is + retried successfully through the newly mounted mTLS adapter. + """ + num_threads = 8 + old_cert = b"old_cert" + all_requests_sent = threading.Barrier(num_threads) + + authed_session = google.auth.transport.requests.AuthorizedSession( + CredentialsStub(), refresh_timeout=60 + ) + authed_session._is_mtls = True + authed_session._cached_cert = old_cert + old_adapter = authed_session.adapters["https://"] + old_auth_adapter = authed_session._auth_request_session.adapters["https://"] + + def send(adapter, request, **kwargs): + if isinstance(adapter, google.auth.transport.requests._MutualTlsAdapter): + return make_response(status=http_client.OK) + # Wait until every thread has sent its request through the old + # adapter, so that all of them receive a 401 response. + all_requests_sent.wait(timeout=30) + return make_response(status=http_client.UNAUTHORIZED) + + mock_send.side_effect = send + + def check_parameters_for_unauthorized_response(cached_cert): + cached_fingerprint = "old_fp" if cached_cert == old_cert else "new_fp" + return ( + pytest.public_cert_bytes, + pytest.private_key_bytes, + cached_fingerprint, + "new_fp", + ) + + results = [] + errors = [] + + def make_request(): + try: + results.append(authed_session.request("GET", self.MTLS_TEST_URL)) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + with ( + mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ), + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=check_parameters_for_unauthorized_response, + ) as mock_check_params, + mock.patch.object( + authed_session, + "_configure_mtls_channel_locked", + wraps=authed_session._configure_mtls_channel_locked, + ) as spy_configure_mtls_channel, + ): + threads = [ + threading.Thread(target=make_request, daemon=True) + for _ in range(num_threads) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=60) + assert not any(thread.is_alive() for thread in threads) + + assert errors == [] + assert [r.status_code for r in results] == [http_client.OK] * num_threads + mock_check_params.assert_called_once() + spy_configure_mtls_channel.assert_called_once() + assert authed_session.is_mtls + assert authed_session._cached_cert == pytest.public_cert_bytes + new_adapter = authed_session.adapters["https://"] + new_auth_adapter = authed_session._auth_request_session.adapters["https://"] + assert isinstance(new_adapter, google.auth.transport.requests._MutualTlsAdapter) + assert isinstance( + new_auth_adapter, google.auth.transport.requests._MutualTlsAdapter + ) + assert new_adapter is not old_adapter + assert new_auth_adapter is not old_auth_adapter + assert not authed_session._mtls_reauth_lock.locked() + + @mock.patch.object(requests.adapters.HTTPAdapter, "send", autospec=True) + def test_concurrent_unauthorized_responses_check_unchanged_cert_once( + self, mock_send + ): + """ + Tests that when several threads that share an AuthorizedSession receive + a 401 response at the same time and the client certificate has not + changed (for example, because the token expired), the certificate is + loaded only once, the mTLS channel is not reconfigured, and every + request is retried successfully. + """ + num_threads = 8 + all_requests_sent = threading.Barrier(num_threads) + sent_lock = threading.Lock() + sent = {"count": 0} + + authed_session = google.auth.transport.requests.AuthorizedSession( + CredentialsStub(), refresh_timeout=60 + ) + authed_session._is_mtls = True + authed_session._cached_cert = pytest.public_cert_bytes + adapter = authed_session.adapters["https://"] + + def send(adapter, request, **kwargs): + with sent_lock: + sent["count"] += 1 + first_attempt = sent["count"] <= num_threads + if first_attempt: + # Wait until every thread has sent its first request, so that + # all of them receive a 401 response before any cert check. + all_requests_sent.wait(timeout=30) + return make_response(status=http_client.UNAUTHORIZED) + return make_response(status=http_client.OK) + + mock_send.side_effect = send + + results = [] + errors = [] + + def make_request(): + try: + results.append(authed_session.request("GET", self.MTLS_TEST_URL)) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + with ( + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=( + pytest.public_cert_bytes, + pytest.private_key_bytes, + "same_fp", + "same_fp", + ), + ) as mock_check_params, + mock.patch.object( + authed_session, "_configure_mtls_channel_locked" + ) as mock_configure_mtls_channel, + ): + threads = [ + threading.Thread(target=make_request, daemon=True) + for _ in range(num_threads) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=60) + assert not any(thread.is_alive() for thread in threads) + + assert errors == [] + assert [r.status_code for r in results] == [http_client.OK] * num_threads + mock_check_params.assert_called_once() + mock_configure_mtls_channel.assert_not_called() + assert authed_session.adapters["https://"] is adapter + assert authed_session._mtls_cert_check_generation == 1 + assert not authed_session._mtls_reauth_lock.locked() + + @mock.patch.object(requests.adapters.HTTPAdapter, "send", autospec=True) + def test_configure_mtls_channel_serialized_with_unauthorized_reconfiguration( + self, mock_send + ): + """ + Tests that an explicit configure_mtls_channel() call made while another + thread is reconfiguring the channel after a 401 response waits for it, + so that each replaced adapter is closed exactly once and the cached + certificate matches the adapter that ends up mounted. + """ + with open(os.path.join(pytest.data_dir, "es256_public_cert.pem"), "rb") as fh: + explicit_cert = fh.read() + with open(os.path.join(pytest.data_dir, "es256_privatekey.pem"), "rb") as fh: + explicit_key = fh.read() + + authed_session = google.auth.transport.requests.AuthorizedSession( + CredentialsStub(), refresh_timeout=60 + ) + authed_session._is_mtls = True + authed_session._cached_cert = b"old_cert" + initial_adapter = authed_session.adapters["https://"] + + def send(adapter, request, **kwargs): + if isinstance(adapter, google.auth.transport.requests._MutualTlsAdapter): + return make_response(status=http_client.OK) + return make_response(status=http_client.UNAUTHORIZED) + + mock_send.side_effect = send + + check_started = threading.Event() + release_check = threading.Event() + + def check_parameters_for_unauthorized_response(cached_cert): + check_started.set() + release_check.wait(timeout=30) + return ( + pytest.public_cert_bytes, + pytest.private_key_bytes, + "old_fp", + "new_fp", + ) + + mounted = [] + replace_adapter = google.auth.transport.requests._replace_adapter + + def record_replace_adapter(session, prefix, adapter): + if session is authed_session: + mounted.append(adapter) + replace_adapter(session, prefix, adapter) + + results = [] + errors = [] + + def make_request(): + try: + results.append(authed_session.request("GET", self.MTLS_TEST_URL)) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + def configure(): + try: + authed_session.configure_mtls_channel( + lambda: (explicit_cert, explicit_key) + ) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + with ( + mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ), + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=check_parameters_for_unauthorized_response, + ), + mock.patch.object( + google.auth.transport.requests, + "_replace_adapter", + side_effect=record_replace_adapter, + ), + mock.patch.object( + requests.adapters.HTTPAdapter, "close", autospec=True + ) as mock_close, + ): + request_thread = threading.Thread(target=make_request, daemon=True) + request_thread.start() + assert check_started.wait(timeout=30) + + configure_thread = threading.Thread(target=configure, daemon=True) + configure_thread.start() + # The explicit call waits for the 401 reconfiguration to finish. + configure_thread.join(timeout=0.5) + assert configure_thread.is_alive() + assert mounted == [] + + release_check.set() + request_thread.join(timeout=60) + configure_thread.join(timeout=60) + assert not request_thread.is_alive() + assert not configure_thread.is_alive() + + assert errors == [] + assert [r.status_code for r in results] == [http_client.OK] + # The 401 reconfiguration ran first, then the explicit call. + assert len(mounted) == 2 + assert authed_session.adapters["https://"] is mounted[-1] + assert authed_session._cached_cert == explicit_cert + closed = [call.args[0] for call in mock_close.call_args_list] + # Each replaced adapter of the session is closed exactly once, and the + # mounted adapter is not closed. + assert closed.count(initial_adapter) == 1 + assert closed.count(mounted[0]) == 1 + assert mounted[-1] not in closed + assert not authed_session._mtls_reauth_lock.locked() + + @mock.patch.object(requests.adapters.HTTPAdapter, "send", autospec=True) + def test_configure_mtls_channel_callback_can_reenter_session(self, mock_send): + """ + Tests that _acquire_mtls_reauth_lock() is reentrant on the same thread, + so a custom client_cert_callback that sends a request through the same + session (even one that receives a 401 from an mTLS endpoint) does not + deadlock, while _mtls_reauth_lock remains held against other threads. + """ + with open(os.path.join(pytest.data_dir, "es256_public_cert.pem"), "rb") as fh: + callback_cert = fh.read() + with open(os.path.join(pytest.data_dir, "es256_privatekey.pem"), "rb") as fh: + callback_key = fh.read() + + authed_session = google.auth.transport.requests.AuthorizedSession( + CredentialsStub(), refresh_timeout=60 + ) + authed_session._is_mtls = True + authed_session._cached_cert = b"old_cert" + initial_adapter = authed_session.adapters["https://"] + + responses = iter( + [ + make_response(status=http_client.UNAUTHORIZED), + make_response(status=http_client.OK), + ] + ) + mock_send.side_effect = lambda adapter, request, **kwargs: next(responses) + + mounted = [] + replace_adapter = google.auth.transport.requests._replace_adapter + + def record_replace_adapter(session, prefix, adapter): + if session is authed_session: + mounted.append(adapter) + replace_adapter(session, prefix, adapter) + + lock_held_in_callback = {"held": None} + + def client_cert_callback(): + lock_held_in_callback["held"] = authed_session._mtls_reauth_lock.locked() + response = authed_session.request("GET", self.MTLS_TEST_URL) + assert response.status_code == http_client.OK + return callback_cert, callback_key + + with ( + mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ), + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=( + pytest.public_cert_bytes, + pytest.private_key_bytes, + "old_fp", + "new_fp", + ), + ), + mock.patch.object( + google.auth.transport.requests, + "_replace_adapter", + side_effect=record_replace_adapter, + ), + mock.patch.object( + requests.adapters.HTTPAdapter, "close", autospec=True + ) as mock_close, + ): + authed_session.configure_mtls_channel(client_cert_callback) + + assert lock_held_in_callback["held"] is True + assert len(mounted) == 2 + assert authed_session.adapters["https://"] is mounted[-1] + assert authed_session._cached_cert == callback_cert + closed = [call.args[0] for call in mock_close.call_args_list] + assert closed.count(initial_adapter) == 1 + assert closed.count(mounted[0]) == 1 + assert mounted[-1] not in closed + assert not authed_session._mtls_reauth_lock.locked() + + @mock.patch( + "google.auth.transport._mtls_helper.get_client_cert_and_key", autospec=True + ) + def test_configure_mtls_channel_does_not_mutate_adapters_in_place( + self, mock_get_client_cert_and_key + ): + """ + Tests that configure_mtls_channel replaces the adapters dictionaries of + the session and of the auth request session instead of mutating them, + so that a thread iterating over them in get_adapter() does not fail + with "OrderedDict mutated during iteration". + """ + mock_get_client_cert_and_key.return_value = ( + True, + pytest.public_cert_bytes, + pytest.private_key_bytes, + ) + authed_session = google.auth.transport.requests.AuthorizedSession( + credentials=mock.Mock() + ) + custom_adapter = AdapterStub([]) + authed_session.mount(self.MTLS_TEST_URL, custom_adapter) + auth_session = authed_session._auth_request_session + + adapters_before = authed_session.adapters + adapters_snapshot = adapters_before.copy() + adapters_iter = iter(adapters_before.items()) + next(adapters_iter) + auth_adapters_before = auth_session.adapters + auth_adapters_snapshot = auth_adapters_before.copy() + auth_adapters_iter = iter(auth_adapters_before.items()) + next(auth_adapters_iter) + + with mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ): + authed_session.configure_mtls_channel() + + # Iterators over the previous dictionaries are still valid. + list(adapters_iter) + list(auth_adapters_iter) + assert adapters_before == adapters_snapshot + assert auth_adapters_before == auth_adapters_snapshot + assert authed_session.adapters is not adapters_before + assert auth_session.adapters is not auth_adapters_before + assert isinstance( + authed_session.get_adapter("https://example.com/"), + google.auth.transport.requests._MutualTlsAdapter, + ) + assert isinstance( + auth_session.get_adapter("https://example.com/"), + google.auth.transport.requests._MutualTlsAdapter, + ) + # The more specific prefix mounted earlier still takes precedence. + assert authed_session.get_adapter(self.MTLS_TEST_URL) is custom_adapter + class TestMutualTlsOffloadAdapter(object): @mock.patch.object(requests.adapters.HTTPAdapter, "init_poolmanager") @@ -1499,3 +2052,63 @@ def test_success_should_use_provider( adapter.proxy_manager_for() mock_proxy_manager_for.assert_called_with(ssl_context=adapter._ctx_proxymanager) + + +class TestAuthorizedSessionMTLSReauth: + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + @mock.patch("google.auth.transport.requests.requests.Session.request") + def test_reauth_lock_acquired_on_unauthorized( + self, mock_session_request, mock_check_params + ): + credentials = mock.Mock(spec=google.auth.credentials.Credentials) + session = google.auth.transport.requests.AuthorizedSession(credentials) + session._is_mtls = True + mock_session_request.side_effect = [ + mock.Mock(status_code=http_client.UNAUTHORIZED), + mock.Mock(status_code=http_client.OK), + ] + mock_check_params.return_value = (b"cert", b"key", "old_fp", "new_fp") + lock_held_during_call = {"held": False} + + def mock_configure_mtls_channel(callback): + lock_held_during_call["held"] = session._mtls_reauth_lock.locked() + + session._configure_mtls_channel_locked = mock.Mock( + side_effect=mock_configure_mtls_channel + ) + + session.request("GET", "https://example.mtls.googleapis.com/") + + session._configure_mtls_channel_locked.assert_called_once() + assert lock_held_during_call["held"] is True + assert not session._mtls_reauth_lock.locked() + + @mock.patch( + "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response" + ) + @mock.patch("google.auth.transport.requests.requests.Session.request") + def test_reauth_skipped_when_cert_fingerprint_matches( + self, mock_session_request, mock_check_params + ): + credentials = mock.Mock() + session = google.auth.transport.requests.AuthorizedSession(credentials) + session._is_mtls = True + session._cached_cert = b"cert" + + mock_session_request.side_effect = [ + mock.Mock(status_code=http_client.UNAUTHORIZED), + mock.Mock(status_code=http_client.OK), + ] + mock_check_params.return_value = ( + b"same_cert_bytes", + b"same_key_bytes", + "same_fingerprint", + "same_fingerprint", + ) + session._configure_mtls_channel_locked = mock.Mock() + + session.request("GET", "https://example.mtls.googleapis.com/") + + session._configure_mtls_channel_locked.assert_not_called() diff --git a/packages/google-auth/tests/transport/test_urllib3.py b/packages/google-auth/tests/transport/test_urllib3.py index 7ffd50aeee32..60260d01defd 100644 --- a/packages/google-auth/tests/transport/test_urllib3.py +++ b/packages/google-auth/tests/transport/test_urllib3.py @@ -14,6 +14,7 @@ import http.client as http_client import os +import threading from unittest import mock import pytest # type: ignore @@ -124,6 +125,14 @@ def test_authed_http_defaults(self): assert authed_http.credentials == mock.sentinel.credentials assert isinstance(authed_http.http, urllib3.PoolManager) + assert not authed_http.is_mtls + + def test_is_mtls_property(self): + credentials = mock.Mock(spec=google.auth.credentials.Credentials) + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) + assert http_obj.is_mtls is False + http_obj._is_mtls = True + assert http_obj.is_mtls is True def test_urlopen_no_refresh(self): credentials = mock.Mock(wraps=CredentialsStub()) @@ -590,7 +599,7 @@ def test_cert_rotation_failure_raises_error(self): ) as mock_check_params: with mock.patch.object( authed_http, - "configure_mtls_channel", + "_configure_mtls_channel_locked", side_effect=Exception("Failed to reconfigure"), ) as mock_reconfigure: with pytest.raises(exceptions.MutualTLSChannelError): @@ -602,7 +611,12 @@ def test_cert_rotation_failure_raises_error(self): def test_cert_rotation_check_params_fails(self): credentials = mock.Mock(wraps=CredentialsStub()) - http = HttpStub([ResponseStub(status=http_client.UNAUTHORIZED)]) + http = HttpStub( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.OK), + ] + ) authed_http = google.auth.transport.urllib3.AuthorizedHttp( credentials, http=http @@ -614,11 +628,13 @@ def test_cert_rotation_check_params_fails(self): "google.auth.transport.urllib3._mtls_helper.check_parameters_for_unauthorized_response", side_effect=Exception("check_params failed"), ) as mock_check_params: - with pytest.raises(Exception, match="check_params failed"): - authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + result = authed_http.urlopen("GET", "http://example.mtls.googleapis.com") + # A failure to check the certificate must not prevent the token + # refresh and retry. mock_check_params.assert_called_once() - credentials.refresh.assert_not_called() + credentials.refresh.assert_called_once() + assert result.status == http_client.OK def test_cert_rotation_logic_skipped_on_other_refresh_status_codes(self): """ @@ -799,3 +815,491 @@ def test_configure_mtls_channel_subsequent_disabled( assert not is_mtls assert not authed_http._is_mtls assert isinstance(authed_http.http, urllib3.PoolManager) + + def test_reauth_lock_acquired_on_unauthorized(self): + credentials = mock.Mock(spec=google.auth.credentials.Credentials) + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) + http_obj._is_mtls = True + + mock_response_unauth = mock.Mock() + mock_response_unauth.status = http_client.UNAUTHORIZED + mock_response_ok = mock.Mock() + mock_response_ok.status = http_client.OK + + lock_held_during_call = {"held": False} + + def mock_configure_mtls_channel(client_cert_callback=None): + lock_held_during_call["held"] = http_obj._mtls_reauth_lock.locked() + + http_obj._configure_mtls_channel_locked = mock.Mock( + side_effect=mock_configure_mtls_channel + ) + http_obj.http.urlopen = mock.Mock( + side_effect=[mock_response_unauth, mock_response_ok] + ) + + with mock.patch.object( + google.auth.transport.urllib3._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=(b"cert", b"key", "old_fp", "new_fp"), + ): + http_obj.request("GET", "https://example.mtls.googleapis.com/") + + http_obj._configure_mtls_channel_locked.assert_called_once() + assert lock_held_during_call["held"] is True + assert not http_obj._mtls_reauth_lock.locked() + + def test_unauthorized_cert_discovery_exception_proceeds_to_token_refresh(self): + credentials = mock.Mock(spec=google.auth.credentials.Credentials) + http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials) + http_obj._is_mtls = True + + mock_response_unauth = mock.Mock() + mock_response_unauth.status = http_client.UNAUTHORIZED + mock_response_ok = mock.Mock() + mock_response_ok.status = http_client.OK + + http_obj.http.urlopen = mock.Mock( + side_effect=[mock_response_unauth, mock_response_ok] + ) + + # Simulate discovery failure (e.g. Enterprise cert provider error or file missing) + with mock.patch.object( + google.auth.transport.urllib3._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=Exception("Certificate discovery failed"), + ): + http_obj.request("GET", "https://example.mtls.googleapis.com/") + + # Token refresh should still be called despite discovery failure + credentials.refresh.assert_called_once() + assert http_obj._mtls_cert_check_generation == 1 + assert not http_obj._mtls_reauth_lock.locked() + + def test_cert_check_skipped_when_cert_rotated_during_request(self): + """ + Tests that a request that receives a 401 response does not read the + client certificate again if another thread has already rotated it + while the request was in flight. + """ + credentials = mock.Mock(wraps=CredentialsStub()) + http = HttpStub( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.OK), + ] + ) + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + credentials, http=http + ) + authed_http._is_mtls = True + authed_http._cached_cert = b"old_cert" + send_request = http.urlopen + + def send_request_and_rotate_cert(*args, **kwargs): + # Simulate another thread rotating the client certificate while + # this request is in flight. + authed_http._cached_cert = b"new_cert" + return send_request(*args, **kwargs) + + http.urlopen = send_request_and_rotate_cert + + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + ) as mock_check_params: + with mock.patch.object( + authed_http, "_configure_mtls_channel_locked" + ) as mock_configure_mtls_channel: + result = authed_http.urlopen( + "GET", "https://example.mtls.googleapis.com" + ) + + assert result.status == http_client.OK + mock_check_params.assert_not_called() + mock_configure_mtls_channel.assert_not_called() + credentials.refresh.assert_called_once() + + def test_reconfigure_mtls_channel_skipped_if_mtls_disabled(self): + """ + Tests that the client certificate is not checked if another thread + disabled mTLS while this thread was waiting for the lock. + """ + authed_http = google.auth.transport.urllib3.AuthorizedHttp( + mock.sentinel.credentials + ) + authed_http._is_mtls = False + + with mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + ) as mock_check_params: + authed_http._reconfigure_mtls_channel_if_cert_changed(b"old_cert", 0) + + mock_check_params.assert_not_called() + assert not authed_http._mtls_reauth_lock.locked() + + @mock.patch.object(urllib3.PoolManager, "urlopen", autospec=True) + def test_concurrent_unauthorized_responses_reconfigure_mtls_channel_once( + self, mock_urlopen + ): + """ + Tests that when several threads that share an AuthorizedHttp receive a + 401 response at the same time, the client certificate is read and the + real mTLS channel reconfiguration runs only once, and every request is + retried successfully through the new mTLS pool manager. + """ + num_threads = 8 + old_cert = b"old_cert" + all_requests_sent = threading.Barrier(num_threads) + + authed_http = google.auth.transport.urllib3.AuthorizedHttp(CredentialsStub()) + authed_http._is_mtls = True + authed_http._cached_cert = old_cert + old_http = authed_http.http + + def urlopen(pool_manager, method, url, body=None, headers=None, **kwargs): + if "ssl_context" in pool_manager.connection_pool_kw: + # Pool manager created by _make_mutual_tls_http(). + return ResponseStub(status=http_client.OK) + # Wait until every thread has sent its request through the old + # pool manager, so that all of them receive a 401 response. + all_requests_sent.wait(timeout=30) + return ResponseStub(status=http_client.UNAUTHORIZED) + + mock_urlopen.side_effect = urlopen + + def check_parameters_for_unauthorized_response(cached_cert): + cached_fingerprint = "old_fp" if cached_cert == old_cert else "new_fp" + return ( + pytest.public_cert_bytes, + pytest.private_key_bytes, + cached_fingerprint, + "new_fp", + ) + + results = [] + errors = [] + + def make_request(): + try: + results.append( + authed_http.urlopen("GET", "https://example.mtls.googleapis.com") + ) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + with ( + mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ), + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=check_parameters_for_unauthorized_response, + ) as mock_check_params, + mock.patch.object( + authed_http, + "_configure_mtls_channel_locked", + wraps=authed_http._configure_mtls_channel_locked, + ) as spy_configure_mtls_channel, + ): + threads = [ + threading.Thread(target=make_request, daemon=True) + for _ in range(num_threads) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=60) + assert not any(thread.is_alive() for thread in threads) + + assert errors == [] + assert [r.status for r in results] == [http_client.OK] * num_threads + mock_check_params.assert_called_once() + spy_configure_mtls_channel.assert_called_once() + assert authed_http.is_mtls + assert authed_http._cached_cert == pytest.public_cert_bytes + assert authed_http.http is not old_http + assert "ssl_context" in authed_http.http.connection_pool_kw + # Token refreshes use the new pool manager too. + assert authed_http._request.http is authed_http.http + assert not authed_http._mtls_reauth_lock.locked() + + @mock.patch.object(urllib3.PoolManager, "urlopen", autospec=True) + def test_concurrent_unauthorized_responses_check_unchanged_cert_once( + self, mock_urlopen + ): + """ + Tests that when several threads that share an AuthorizedHttp receive a + 401 response at the same time and the client certificate has not + changed (for example, because the token expired), the certificate is + loaded only once, the mTLS channel is not reconfigured, and every + request is retried successfully. + """ + num_threads = 8 + all_requests_sent = threading.Barrier(num_threads) + sent_lock = threading.Lock() + sent = {"count": 0} + + authed_http = google.auth.transport.urllib3.AuthorizedHttp(CredentialsStub()) + authed_http._is_mtls = True + authed_http._cached_cert = pytest.public_cert_bytes + http = authed_http.http + + def urlopen(pool_manager, method, url, body=None, headers=None, **kwargs): + with sent_lock: + sent["count"] += 1 + first_attempt = sent["count"] <= num_threads + if first_attempt: + # Wait until every thread has sent its first request, so that + # all of them receive a 401 response before any cert check. + all_requests_sent.wait(timeout=30) + return ResponseStub(status=http_client.UNAUTHORIZED) + return ResponseStub(status=http_client.OK) + + mock_urlopen.side_effect = urlopen + + results = [] + errors = [] + + def make_request(): + try: + results.append( + authed_http.urlopen("GET", "https://example.mtls.googleapis.com") + ) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + with ( + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=( + pytest.public_cert_bytes, + pytest.private_key_bytes, + "same_fp", + "same_fp", + ), + ) as mock_check_params, + mock.patch.object( + authed_http, "_configure_mtls_channel_locked" + ) as mock_configure_mtls_channel, + ): + threads = [ + threading.Thread(target=make_request, daemon=True) + for _ in range(num_threads) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=60) + assert not any(thread.is_alive() for thread in threads) + + assert errors == [] + assert [r.status for r in results] == [http_client.OK] * num_threads + mock_check_params.assert_called_once() + mock_configure_mtls_channel.assert_not_called() + assert authed_http.http is http + assert authed_http._mtls_cert_check_generation == 1 + assert not authed_http._mtls_reauth_lock.locked() + + @mock.patch.object(urllib3.PoolManager, "urlopen", autospec=True) + def test_configure_mtls_channel_serialized_with_unauthorized_reconfiguration( + self, mock_urlopen + ): + """ + Tests that an explicit configure_mtls_channel() call made while another + thread is reconfiguring the channel after a 401 response waits for it, + so that each replaced pool manager is cleared exactly once and the + cached certificate matches the pool manager that ends up in use. + """ + with open(os.path.join(pytest.data_dir, "es256_public_cert.pem"), "rb") as fh: + explicit_cert = fh.read() + with open(os.path.join(pytest.data_dir, "es256_privatekey.pem"), "rb") as fh: + explicit_key = fh.read() + + authed_http = google.auth.transport.urllib3.AuthorizedHttp(CredentialsStub()) + authed_http._is_mtls = True + authed_http._cached_cert = b"old_cert" + initial_http = authed_http.http + + def urlopen(pool_manager, method, url, body=None, headers=None, **kwargs): + if "ssl_context" in pool_manager.connection_pool_kw: + return ResponseStub(status=http_client.OK) + return ResponseStub(status=http_client.UNAUTHORIZED) + + mock_urlopen.side_effect = urlopen + + check_started = threading.Event() + release_check = threading.Event() + + def check_parameters_for_unauthorized_response(cached_cert): + check_started.set() + release_check.wait(timeout=30) + return ( + pytest.public_cert_bytes, + pytest.private_key_bytes, + "old_fp", + "new_fp", + ) + + created = [] + make_mutual_tls_http = google.auth.transport.urllib3._make_mutual_tls_http + + def record_make_mutual_tls_http(cert, key): + http = make_mutual_tls_http(cert, key) + created.append(http) + return http + + results = [] + errors = [] + + def make_request(): + try: + results.append( + authed_http.urlopen("GET", "https://example.mtls.googleapis.com") + ) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + def configure(): + try: + authed_http.configure_mtls_channel( + lambda: (explicit_cert, explicit_key) + ) + except Exception as exc: # pragma: NO COVER + errors.append(exc) + + with ( + mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ), + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + side_effect=check_parameters_for_unauthorized_response, + ), + mock.patch.object( + google.auth.transport.urllib3, + "_make_mutual_tls_http", + side_effect=record_make_mutual_tls_http, + ), + mock.patch.object( + urllib3.PoolManager, "clear", autospec=True + ) as mock_clear, + ): + request_thread = threading.Thread(target=make_request, daemon=True) + request_thread.start() + assert check_started.wait(timeout=30) + + configure_thread = threading.Thread(target=configure, daemon=True) + configure_thread.start() + # The explicit call waits for the 401 reconfiguration to finish. + configure_thread.join(timeout=0.5) + assert configure_thread.is_alive() + assert created == [] + + release_check.set() + request_thread.join(timeout=60) + configure_thread.join(timeout=60) + assert not request_thread.is_alive() + assert not configure_thread.is_alive() + + assert errors == [] + assert [r.status for r in results] == [http_client.OK] + # The 401 reconfiguration ran first, then the explicit call. + assert len(created) == 2 + assert authed_http.http is created[-1] + assert authed_http._request.http is created[-1] + assert authed_http._cached_cert == explicit_cert + cleared = [call.args[0] for call in mock_clear.call_args_list] + # Each replaced pool manager is cleared exactly once, and the one in + # use is not cleared. + assert cleared.count(initial_http) == 1 + assert cleared.count(created[0]) == 1 + assert created[-1] not in cleared + assert not authed_http._mtls_reauth_lock.locked() + + @mock.patch.object(urllib3.PoolManager, "urlopen", autospec=True) + def test_configure_mtls_channel_callback_can_reenter_session(self, mock_urlopen): + """ + Tests that _acquire_mtls_reauth_lock() is reentrant on the same thread, + so a custom client_cert_callback that sends a request through the same + AuthorizedHttp (even one that receives a 401 from an mTLS endpoint) + does not deadlock, while _mtls_reauth_lock remains held against other + threads. + """ + with open(os.path.join(pytest.data_dir, "es256_public_cert.pem"), "rb") as fh: + callback_cert = fh.read() + with open(os.path.join(pytest.data_dir, "es256_privatekey.pem"), "rb") as fh: + callback_key = fh.read() + + authed_http = google.auth.transport.urllib3.AuthorizedHttp(CredentialsStub()) + authed_http._is_mtls = True + authed_http._cached_cert = b"old_cert" + initial_http = authed_http.http + + responses = iter( + [ + ResponseStub(status=http_client.UNAUTHORIZED), + ResponseStub(status=http_client.OK), + ] + ) + mock_urlopen.side_effect = ( + lambda pool_manager, method, url, body=None, headers=None, **kwargs: next( + responses + ) + ) + + created = [] + make_mutual_tls_http = google.auth.transport.urllib3._make_mutual_tls_http + + def record_make_mutual_tls_http(cert, key): + http = make_mutual_tls_http(cert, key) + created.append(http) + return http + + lock_held_in_callback = {"held": None} + + def client_cert_callback(): + lock_held_in_callback["held"] = authed_http._mtls_reauth_lock.locked() + response = authed_http.urlopen("GET", "https://example.mtls.googleapis.com") + assert response.status == http_client.OK + return callback_cert, callback_key + + with ( + mock.patch.dict( + os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"} + ), + mock.patch.object( + google.auth.transport._mtls_helper, + "check_parameters_for_unauthorized_response", + return_value=( + pytest.public_cert_bytes, + pytest.private_key_bytes, + "old_fp", + "new_fp", + ), + ), + mock.patch.object( + google.auth.transport.urllib3, + "_make_mutual_tls_http", + side_effect=record_make_mutual_tls_http, + ), + mock.patch.object( + urllib3.PoolManager, "clear", autospec=True + ) as mock_clear, + ): + assert authed_http.configure_mtls_channel(client_cert_callback) is True + + assert lock_held_in_callback["held"] is True + assert len(created) == 2 + assert authed_http.http is created[-1] + assert authed_http._request.http is created[-1] + assert authed_http._cached_cert == callback_cert + cleared = [call.args[0] for call in mock_clear.call_args_list] + assert cleared.count(initial_http) == 1 + assert cleared.count(created[0]) == 1 + assert created[-1] not in cleared + assert not authed_http._mtls_reauth_lock.locked()