diff --git a/packages/google-auth/google/auth/aio/transport/mtls.py b/packages/google-auth/google/auth/aio/transport/mtls.py index 267a1e401ff0..47adb3df7d5b 100644 --- a/packages/google-auth/google/auth/aio/transport/mtls.py +++ b/packages/google-auth/google/auth/aio/transport/mtls.py @@ -211,7 +211,7 @@ def _fetch_fingerprints(): cached_cert ) else: - cached_fingerprint = None + cached_fingerprint = current_fingerprint return cached_fingerprint, current_fingerprint cached_fingerprint, current_cert_fingerprint = await _run_in_executor( diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py index f2ced1280e50..0763ea4cbb34 100644 --- a/packages/google-auth/google/auth/aio/transport/sessions.py +++ b/packages/google-auth/google/auth/aio/transport/sessions.py @@ -20,7 +20,7 @@ import inspect import logging import time -from typing import Mapping, Optional, TYPE_CHECKING, Union +from typing import Mapping, Optional, Tuple, TYPE_CHECKING, Union import urllib.parse import warnings @@ -58,6 +58,19 @@ AIOHTTP_INSTALLED = False +def _retrieve_task_exception(task: "asyncio.Task") -> None: + """Mark a finished task's exception as retrieved. + + Used as a done-callback so that a background mTLS initialization failure + that nobody ends up awaiting (for example, because the only caller timed + out) does not later surface as an unraisable + "Task exception was never retrieved" warning during garbage collection. + Callers that *do* await the task still observe the exception normally. + """ + if not task.cancelled(): + task.exception() + + @asynccontextmanager async def timeout_guard(timeout): """ @@ -154,7 +167,7 @@ def __init__( if not _auth_request and AIOHTTP_INSTALLED: _auth_request = AiohttpRequest() self._is_mtls = False - self._mtls_init_task = None + self._mtls_init_task: Optional[asyncio.Task] = None self._cached_cert = None self._client_cert_callback = None self._old_auth_requests: list[transport.Request] = [] @@ -165,10 +178,70 @@ def __init__( self._auth_request = _auth_request self._mtls_rotation_lock: Optional[asyncio.Lock] = None self._mtls_check_counter = 0 + # Incremented every time the mTLS channel is successfully reconfigured. + # Unlike a coroutine-local flag, this lets a request that skipped the + # rotation check (because a concurrent request already performed it) + # still observe that the channel changed since its own 401. + self._mtls_reconfig_counter = 0 + # Value of `_mtls_reconfig_counter` observed when the most recently + # completed credential refresh *started*. Counting refresh completions + # is not sufficient to decide whether a token is usable on the current + # channel: a refresh that began before a rotation and finished after it + # was still minted over the old transport, and a certificate-bound + # token from the old channel is rejected by the new one. Recording the + # generation a refresh started in lets us tell the two apart. + self._last_refresh_reconfig_gen = -1 + # Serializes the decision to create a new mTLS initialization task so + # that two concurrent callers cannot both spawn `_do_configure()`. + self._mtls_init_lock: Optional[asyncio.Lock] = None self._refresh_lock: Optional[asyncio.Lock] = None self._refresh_counter = 0 + # Set by `close()`. Guarded by `_mtls_init_lock` so that an in-flight + # certificate rotation cannot install a new transport on a session that + # has already been torn down. + self._closed = False + + async def _trim_old_auth_requests(self) -> None: + """Close retired transports, keeping at most the two most recent.""" + while len(self._old_auth_requests) > 2: + oldest_auth_request = self._old_auth_requests.pop(0) + try: + if hasattr(oldest_auth_request, "close"): + res = oldest_auth_request.close() + if inspect.isawaitable(res): + await res + except Exception as caught_exc: + _LOGGER.debug( + "Failed to close a retired auth transport: %s", caught_exc + ) - async def configure_mtls_channel(self, client_cert_callback=None): + async def _reset_non_mtls_state(self) -> None: + """Clear mTLS state, retiring a library-owned mTLS transport. + + If the session previously installed its own mTLS-enabled transport, + that transport still presents the old client certificate. Simply + clearing the flags would leave subsequent requests sending a stale + cert, so the transport is retired and replaced with a fresh non-mTLS + one. A caller-supplied custom transport is never replaced. + """ + was_mtls = self._is_mtls + self._is_mtls = False + self._cached_cert = None + if ( + was_mtls + and AIOHTTP_INSTALLED + and isinstance(self._auth_request, AiohttpRequest) + ): + self._old_auth_requests.append(self._auth_request) + self._auth_request = AiohttpRequest() + await self._trim_old_auth_requests() + + async def configure_mtls_channel( + self, + client_cert_callback=None, + force: bool = False, + _cert_key_override: Optional[Tuple[bytes, bytes]] = None, + ): """Configure the client certificate and key for SSL connection. This method configures mTLS if client certificates are explicitly enabled @@ -188,83 +261,176 @@ async def configure_mtls_channel(self, client_cert_callback=None): key bytes both in PEM format. If the callback is None, application default SSL credentials will be used. + force (bool): + Whether to force reconfiguration even if the channel is already configured + with the same callback. + _cert_key_override (Optional[Tuple[bytes, bytes]]): + Internal use only. An explicit (cert, key) pair to install, + bypassing ``client_cert_callback`` resolution. Used by the + certificate-rotation path so it does not have to temporarily + mutate ``self._client_cert_callback``, which would otherwise + be visible to (and clobber) concurrent callers. When set, the + channel is always reconfigured and the user-supplied callback + is left untouched. Raises: google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel creation failed for any reason. + google.auth.exceptions.InvalidOperation: If the session has already + been closed. """ - if self._mtls_init_task is None: - self._client_cert_callback = client_cert_callback - - async def _do_configure(): - # Run the blocking check in an executor - use_client_cert = await mtls._run_in_executor( - google.auth.transport._mtls_helper.check_use_client_cert + if self._mtls_init_lock is None: + self._mtls_init_lock = asyncio.Lock() + init_lock = self._mtls_init_lock + + # Serialize the decide-and-create step so two concurrent callers cannot + # both spawn `_do_configure()`. The lock is released before awaiting the + # task itself, so a slow configuration does not block unrelated callers. + async with init_lock: + if self._closed: + # Without this, a rotation triggered by an in-flight request + # could build and install a brand new transport after `close()` + # has already drained everything, leaking it permanently. + # + # A task that was *already* running when `close()` landed needs + # no separate check: `close()` sets `_closed` and cancels the + # task without yielding in between, so a running + # `_do_configure` is always interrupted at one of its awaits + # before it reaches the point where it installs a transport. + raise exceptions.InvalidOperation( + "Cannot configure the mTLS channel on a closed session." + ) + if _cert_key_override is not None: + needs_reconfig = True + else: + is_explicit_reconfig = ( + client_cert_callback != self._client_cert_callback + ) + task_failed = ( + self._mtls_init_task is not None + and self._mtls_init_task.done() + and ( + self._mtls_init_task.cancelled() + or self._mtls_init_task.exception() is not None + ) + ) + needs_reconfig = ( + self._mtls_init_task is None + or is_explicit_reconfig + or task_failed + or force ) - if not use_client_cert: - return - - try: - ( - is_mtls, - cert, - key, - ) = await mtls.get_client_cert_and_key(client_cert_callback) - - if is_mtls: - # Re-create the auth request with the new SSL context - if AIOHTTP_INSTALLED and isinstance( - self._auth_request, AiohttpRequest - ): - ssl_context = await mtls._run_in_executor( - mtls.make_client_cert_ssl_context, cert, key - ) - connector = aiohttp.TCPConnector(ssl=ssl_context) - new_session = aiohttp.ClientSession(connector=connector) - - old_auth_request = self._auth_request - self._auth_request = AiohttpRequest(session=new_session) - self._old_auth_requests.append(old_auth_request) - while len(self._old_auth_requests) > 2: - oldest_auth_request = self._old_auth_requests[0] - try: - if hasattr(oldest_auth_request, "close"): - res = oldest_auth_request.close() - if inspect.isawaitable(res): - await res - except Exception: - pass - self._old_auth_requests.pop(0) + if not needs_reconfig: + task = self._mtls_init_task + else: + old_task = self._mtls_init_task + if old_task is not None and not old_task.done(): + old_task.cancel() + # `asyncio.wait` does not re-raise the awaited task's + # exception, and does not convert that task's cancellation + # into ours -- while still letting a cancellation targeted + # at *this* coroutine propagate normally. + await asyncio.wait({old_task}) + if _cert_key_override is None: + # Only a user-driven call may change the stored callback. + # The internal rotation path leaves it untouched. + self._client_cert_callback = client_cert_callback + + async def _do_configure(): + # Run the blocking check in an executor. It reads and parses + # a config file, so it can fail in ways the caller is + # promised to see as `MutualTLSChannelError`. + try: + use_client_cert = await mtls._run_in_executor( + google.auth.transport._mtls_helper.check_use_client_cert + ) + except Exception as caught_exc: + raise exceptions.MutualTLSChannelError( + caught_exc + ) from caught_exc + if not use_client_cert: + return + try: + if _cert_key_override is not None: + is_mtls = True + cert, key = _cert_key_override else: - is_mtls = False - warnings.warn( - "Attempted to establish mTLS, but a custom async transport was provided. " - "google-auth cannot automatically configure custom transports for mTLS. " - "Falling back to standard TLS. If your custom transport is not manually " - "configured for mTLS, you may encounter 401 Unauthorized errors when " - "using Certificate-Bound Tokens.", - UserWarning, - ) + ( + is_mtls, + cert, + key, + ) = await mtls.get_client_cert_and_key(client_cert_callback) + + if is_mtls: + # Re-create the auth request with the new SSL context + if AIOHTTP_INSTALLED and isinstance( + self._auth_request, AiohttpRequest + ): + ssl_context = await mtls._run_in_executor( + mtls.make_client_cert_ssl_context, cert, key + ) + connector = aiohttp.TCPConnector(ssl=ssl_context) + new_session = aiohttp.ClientSession(connector=connector) - self._is_mtls = is_mtls - if is_mtls: - self._cached_cert = cert - else: - self._cached_cert = None + old_auth_request = self._auth_request + self._auth_request = AiohttpRequest(session=new_session) + self._is_mtls = True + self._cached_cert = cert + self._old_auth_requests.append(old_auth_request) + await self._trim_old_auth_requests() - except Exception as caught_exc: - new_exc = exceptions.MutualTLSChannelError(caught_exc) - raise new_exc from caught_exc - - self._mtls_init_task = asyncio.create_task(_do_configure()) + else: + await self._reset_non_mtls_state() + warnings.warn( + "Attempted to establish mTLS, but a custom async transport was provided. " + "google-auth cannot automatically configure custom transports for mTLS. " + "Falling back to standard TLS. If your custom transport is not manually " + "configured for mTLS, you may encounter 401 Unauthorized errors when " + "using Certificate-Bound Tokens.", + UserWarning, + ) + else: + await self._reset_non_mtls_state() + + except Exception as caught_exc: + new_exc = exceptions.MutualTLSChannelError(caught_exc) + raise new_exc from caught_exc + + task = asyncio.create_task(_do_configure()) + # If every awaiter goes away (e.g. the only caller timed out) + # a failure would otherwise surface as an unraisable + # "Task exception was never retrieved" warning at GC time. + task.add_done_callback(_retrieve_task_exception) + self._mtls_init_task = task + + if task is None: # pragma: no cover - defensive + raise exceptions.MutualTLSChannelError( + "mTLS initialization task was not created." + ) - try: - return await self._mtls_init_task - except BaseException: - self._mtls_init_task = None - raise + while True: + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + if not task.cancelled(): + # The cancellation targeted this caller rather than the + # initialization task, so it must propagate. + raise + # The task we were waiting on was cancelled by a concurrent + # reconfiguration. That coroutine holds `init_lock` across + # both the cancel and the creation of the replacement, so + # acquiring it here waits until the replacement is installed. + async with init_lock: + current = self._mtls_init_task + if current is None or current is task: + # Nobody installed a replacement (e.g. the session is + # closing); surface the cancellation. + raise + # Follow the replacement rather than reporting a cancellation + # this caller never requested. + task = current async def request( self, @@ -312,13 +478,14 @@ async def request( channel reconfiguration fails for any reason during certificate rotation. """ _auth_retry_count = kwargs.pop("_auth_retry_count", 0) - if self._mtls_init_task and not self._mtls_init_task.done(): - try: - await asyncio.shield(self._mtls_init_task) - except Exception: - # Suppress all exceptions from the background mTLS initialization task, - # allowing the request to fail naturally elsewhere. - pass + # Wait for any in-flight mTLS initialization to settle. `asyncio.wait` + # neither re-raises the task's exception (the request should fail + # naturally elsewhere instead) nor turns that task's cancellation into + # ours, while a cancellation aimed at *this* coroutine still + # propagates. Looping re-reads the attribute in case a concurrent + # `configure_mtls_channel()` swapped in a replacement task. + while self._mtls_init_task is not None and not self._mtls_init_task.done(): + await asyncio.wait({self._mtls_init_task}) retries = _exponential_backoff.AsyncExponentialBackoff( total_attempts=total_attempts, ) @@ -326,6 +493,7 @@ async def request( start_time = time.monotonic() refresh_counter_at_error = self._refresh_counter check_counter_at_error = self._mtls_check_counter + reconfig_counter_at_error = self._mtls_reconfig_counter async with timeout_guard(max_allowed_time) as with_timeout: await with_timeout( # Note: before_request will attempt to refresh credentials if expired. @@ -409,10 +577,12 @@ async def _recover_auth_state(): exceptions.MutualTLSChannelError, OSError, ValueError, + TypeError, ImportError, ) as e: _LOGGER.warning( - "Failed to check client certificate parameters: %s. Proceeding with original response.", + "Failed to check client certificate parameters: %s. " + "Falling back to credential refresh and retry.", e, ) else: @@ -421,30 +591,34 @@ async def _recover_auth_state(): and cached_fingerprint != current_cert_fingerprint ): - saved_callback = ( - self._client_cert_callback - ) try: _LOGGER.info( "Client certificate has changed, reconfiguring mTLS " "channel." ) - if self._mtls_init_task is not None: - if ( - not self._mtls_init_task.done() - ): - try: - await self._mtls_init_task - except Exception: - pass - self._mtls_init_task = None + # Pass the rotated cert/key + # directly rather than + # temporarily swapping + # `self._client_cert_callback`, + # which is shared state and + # could be observed (or + # clobbered) by concurrent + # callers. await self.configure_mtls_channel( - lambda: ( + _cert_key_override=( call_cert_bytes, call_key_bytes, ) ) + self._mtls_reconfig_counter += 1 except Exception as e: + # NOTE: `_mtls_check_counter` + # is deliberately left + # un-incremented below, so a + # queued coroutine retries + # the reconfiguration rather + # than inheriting this + # failure. _LOGGER.error( "Failed to reconfigure mTLS channel: %s", e, @@ -452,10 +626,6 @@ async def _recover_auth_state(): raise exceptions.MutualTLSChannelError( "Failed to reconfigure mTLS channel" ) from e - finally: - self._client_cert_callback = ( - saved_callback - ) else: if current_cert_fingerprint is None: _LOGGER.info( @@ -469,27 +639,64 @@ async def _recover_auth_state(): ) # Always increment so waiting tasks skip the check block self._mtls_check_counter += 1 + + # Derived from session state rather than a local flag: + # a concurrent request may have performed the rotation + # on our behalf (we then skipped the check block), and + # that request still needs to retry on the new channel. + # + # Evaluated at each use rather than snapshotted here: a + # concurrent request can rotate the channel while this + # coroutine is queued on `_refresh_lock` or waiting for + # its own refresh to fail. A value captured at this + # point would miss that rotation and drop a retry that + # would have succeeded on the new channel. + def channel_reconfigured() -> bool: + return ( + self._mtls_reconfig_counter > reconfig_counter_at_error + ) + if self._refresh_lock is None: self._refresh_lock = asyncio.Lock() async with self._refresh_lock: - # Check if another task already refreshed credentials while we were waiting - if self._refresh_counter > refresh_counter_at_error: + # A concurrent refresh only makes this one redundant + # if it completed after this request's 401 *and* it + # was started on the channel we are about to retry + # on. Completion order alone is not enough: a + # refresh that began before a rotation and finished + # after it minted its token over the old transport, + # and the rotated channel will reject it. + already_refreshed = ( + self._refresh_counter > refresh_counter_at_error + and self._last_refresh_reconfig_gen + >= self._mtls_reconfig_counter + ) + if already_refreshed: _LOGGER.debug( "Credentials were already refreshed by a concurrent task. Skipping duplicate refresh." ) else: + # Snapshot the generation *before* awaiting so a + # rotation that lands mid-refresh is not + # credited to the token we are about to mint. + reconfig_gen = self._mtls_reconfig_counter try: await self._credentials.refresh(self._auth_request) except NotImplementedError: _LOGGER.debug( "Credentials do not implement refresh()." ) - return response - except ( - exceptions.RefreshError, - getattr(exceptions, "InvalidOperation", Exception), - ) as e: + if not channel_reconfigured(): + return response + except exceptions.InvalidOperation as e: + _LOGGER.debug( + "Credentials cannot be refreshed: %s", + e, + ) + if not channel_reconfigured(): + return response + except exceptions.RefreshError as e: _LOGGER.debug( "Credential refresh failed, returning 401 response. Error: %s", e, @@ -497,6 +704,7 @@ async def _recover_auth_state(): return response else: self._refresh_counter += 1 + self._last_refresh_reconfig_gen = reconfig_gen if is_streaming: return response @@ -513,8 +721,10 @@ async def _recover_auth_state(): res = response.close() if inspect.isawaitable(res): await res - except Exception: - pass + except Exception as close_exc: + _LOGGER.debug( + "Failed to close the 401 response: %s", close_exc + ) raise # If it returned a response (meaning streaming or error), bail out if early_return_response is not None: @@ -524,8 +734,8 @@ async def _recover_auth_state(): res = response.close() if inspect.isawaitable(res): await res - except Exception: - pass + except Exception as close_exc: + _LOGGER.debug("Failed to close the 401 response: %s", close_exc) if max_allowed_time is not None: remaining_time = max( 0.0, max_allowed_time - (time.monotonic() - start_time) @@ -820,25 +1030,43 @@ def is_mtls(self): async def close(self) -> None: """ Close the underlying auth request session. + + Once closed, the session refuses further mTLS (re)configuration, so an + in-flight certificate rotation cannot resurrect it with a freshly built + transport that nothing would ever close. """ - if self._mtls_init_task and not self._mtls_init_task.done(): - self._mtls_init_task.cancel() - try: - await self._mtls_init_task - except asyncio.CancelledError: - pass + if self._mtls_init_lock is None: + self._mtls_init_lock = asyncio.Lock() + # Flip the flag under the same lock `configure_mtls_channel` uses to + # decide whether to spawn a task, so the two cannot interleave. + async with self._mtls_init_lock: + self._closed = True + init_task = self._mtls_init_task try: - if hasattr(self._auth_request, "close"): - res = self._auth_request.close() - if inspect.isawaitable(res): - await res + if init_task and not init_task.done(): + init_task.cancel() + # Same rationale as `configure_mtls_channel`: `asyncio.wait` + # lets the cancelled initialization task unwind without + # absorbing a cancellation aimed at this `close()` call. + await asyncio.wait({init_task}) finally: - for old_request in self._old_auth_requests: + while self._old_auth_requests: + old_request = self._old_auth_requests.pop(0) try: if hasattr(old_request, "close"): res = old_request.close() if inspect.isawaitable(res): await res - except Exception: - pass - self._old_auth_requests.clear() + except Exception as caught_exc: + _LOGGER.debug( + "Failed to close a retired auth transport: %s", caught_exc + ) + try: + if hasattr(self._auth_request, "close"): + res = self._auth_request.close() + if inspect.isawaitable(res): + await res + except Exception as caught_exc: + _LOGGER.debug( + "Failed to close the active auth transport: %s", caught_exc + ) diff --git a/packages/google-auth/tests/transport/aio/test_mtls.py b/packages/google-auth/tests/transport/aio/test_mtls.py index d0ff6b749719..585ed1f3ef9c 100644 --- a/packages/google-auth/tests/transport/aio/test_mtls.py +++ b/packages/google-auth/tests/transport/aio/test_mtls.py @@ -159,9 +159,9 @@ def callback(): assert cert == CERT_BYTES assert key == KEY_BYTES - assert cached_fp is None + assert cached_fp == "FINGERPRINT_CURRENT" assert current_fp == "FINGERPRINT_CURRENT" - assert cached_fp != current_fp + assert cached_fp == current_fp mock_get_cached.assert_not_called() diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py index f6f1185a660e..a0f313fe5899 100644 --- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py +++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py @@ -131,6 +131,125 @@ async def test_configure_mtls_channel_invalid_fields(self): await session.configure_mtls_channel() await session.close() + @pytest.mark.asyncio + async def test_configure_mtls_channel_mock_callback(self): + callback = mock.AsyncMock(return_value=(b"cert", b"key")) + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ) as mock_make_context, + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + await session.configure_mtls_channel(callback) + assert session.is_mtls is True + assert session._cached_cert == b"cert" + mock_make_context.assert_called_once_with(b"cert", b"key") + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_custom_request(self): + custom_req = mock.AsyncMock(spec=transport.Request) + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds, auth_request=custom_req) + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ), + pytest.warns( + UserWarning, + match="Attempted to establish mTLS, but a custom async transport was provided", + ), + ): + await session.configure_mtls_channel() + assert session.is_mtls is False + assert session._cached_cert is None + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_exception_preserves_flag(self): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + session._is_mtls = True + session._cached_cert = b"old_cert" + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + side_effect=Exception("Disk failure"), + ), + pytest.raises(exceptions.MutualTLSChannelError), + ): + await session.configure_mtls_channel(lambda: (b"new_cert", b"new_key")) + assert session.is_mtls is True + assert session._cached_cert == b"old_cert" + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_transport_error_resets_flag(self): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.ClientSession", side_effect=Exception("Session error")), + pytest.raises(exceptions.MutualTLSChannelError), + ): + await session.configure_mtls_channel() + assert session.is_mtls is False + assert session._cached_cert is None + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_atomic_on_exception(self): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + orig_req = session._auth_request + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + side_effect=RuntimeError("Fatal error"), + ), + pytest.raises(exceptions.MutualTLSChannelError), + ): + await session.configure_mtls_channel() + assert session._auth_request is orig_req + assert session.is_mtls is False + await session.close() + @pytest.mark.asyncio async def test_configure_mtls_channel_close_exception_does_not_abort(self): """Tests that an exception in old_auth_request.close() during eviction does not abort configuration.""" @@ -343,12 +462,14 @@ async def test_cert_rotation_success_and_retry(self): assert resp == mock_resp_200 mock_conf.assert_called_once() - cb = ( - mock_conf.call_args.args[0] - if mock_conf.call_args.args - else mock_conf.call_args.kwargs["client_cert_callback"] + # The rotation path passes the rotated cert/key explicitly rather + # than temporarily swapping the shared `_client_cert_callback`, + # which must therefore be left untouched. + assert mock_conf.call_args.kwargs["_cert_key_override"] == ( + new_cert, + new_key, ) - assert cb() == (new_cert, new_key) + assert session._client_cert_callback is None mock_creds.refresh.assert_called_once() assert mock_auth_req.call_count == 2 mock_resp_401.close.assert_called_once() @@ -523,12 +644,14 @@ async def test_psc_endpoint_triggers_cert_rotation(self): assert resp == mock_resp_200 mock_check.assert_called_once() mock_conf.assert_called_once() - cb = ( - mock_conf.call_args.args[0] - if mock_conf.call_args.args - else mock_conf.call_args.kwargs["client_cert_callback"] + # The rotation path passes the rotated cert/key explicitly rather + # than temporarily swapping the shared `_client_cert_callback`, + # which must therefore be left untouched. + assert mock_conf.call_args.kwargs["_cert_key_override"] == ( + new_cert, + new_key, ) - assert cb() == (new_cert, new_key) + assert session._client_cert_callback is None await session.close() @@ -888,13 +1011,17 @@ async def dummy_completed(): session._mtls_init_task = initial_task # Pre-populate completed task new_cert = b"new_cert" new_key = b"new_key" + + async def fake_configure(cb=None, **kwargs): + session._mtls_init_task = asyncio.create_task(dummy_completed()) + with ( mock.patch( "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response", new_callable=mock.AsyncMock, ) as mock_check, mock.patch.object( - session, "configure_mtls_channel", new_callable=mock.AsyncMock + session, "configure_mtls_channel", side_effect=fake_configure ) as mock_conf, ): mock_check.return_value = (new_cert, new_key, b"old_fp", b"new_fp") @@ -918,21 +1045,41 @@ async def test_401_retry_raises_timeout_before_refresh(self): """ mock_creds = mock.AsyncMock(spec=credentials.Credentials) mock_auth_request = mock.AsyncMock(spec=transport.Request) - mock_resp_401 = mock.Mock(spec=transport.Response, status_code=401) - current_time = 0.0 - # When the 401 request completes, advance time past max_allowed_time + status_access_count = 0 + + class _CustomResponse: + def __init__(self): + self.close = mock.AsyncMock() + + @property + def status_code(self): + nonlocal status_access_count + status_access_count += 1 + return 401 + + mock_resp_401 = _CustomResponse() + async def fake_auth_request(*args, **kwargs): - nonlocal current_time - current_time = 100.0 # Expire timeout before refresh starts return mock_resp_401 mock_auth_request.side_effect = fake_auth_request session = sessions.AsyncAuthorizedSession( mock_creds, auth_request=mock_auth_request ) - with mock.patch("time.monotonic", side_effect=lambda: current_time): - with pytest.raises(exceptions.TimeoutError): + + def mock_time(): + if status_access_count >= 2: + return 100.0 + return 0.1 + + with mock.patch( + "google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time + ): + with pytest.raises( + exceptions.TimeoutError, + match="Timeout exceeded before credential refresh could begin", + ): await session.request( "GET", "https://example.com", max_allowed_time=1.0 ) @@ -949,21 +1096,34 @@ async def test_401_retry_raises_timeout_before_subsequent_retry(self): mock_creds = mock.AsyncMock(spec=credentials.Credentials) mock_auth_request = mock.AsyncMock(spec=transport.Request) mock_resp_401 = mock.Mock(spec=transport.Response, status_code=401) + mock_resp_401.close = mock.AsyncMock() mock_auth_request.return_value = mock_resp_401 - current_time = 0.0 - # Allow initial request to proceed at t=0.0, but expire timeout during refresh async def fake_refresh(auth_request): - nonlocal current_time - current_time = 100.0 # Expire timeout during refresh before retry return None mock_creds.refresh = mock.AsyncMock(side_effect=fake_refresh) session = sessions.AsyncAuthorizedSession( mock_creds, auth_request=mock_auth_request ) - with mock.patch("time.monotonic", side_effect=lambda: current_time): - with pytest.raises(exceptions.TimeoutError): + + after_refresh_count = 0 + + def mock_time(): + nonlocal after_refresh_count + if mock_creds.refresh.called: + after_refresh_count += 1 + if after_refresh_count >= 2: + return 100.0 + return 0.1 + + with mock.patch( + "google.auth.aio.transport.sessions.time.monotonic", side_effect=mock_time + ): + with pytest.raises( + exceptions.TimeoutError, + match=r"(Timeout exceeded before retrying the request|Context manager exceeded the configured timeout)", + ): await session.request( "GET", "https://example.com", max_allowed_time=1.0 ) @@ -1093,7 +1253,10 @@ async def test_request_cancellation_propagates_and_leaves_mtls_init_running(self in the background. """ mock_creds = mock.AsyncMock(spec=credentials.Credentials) - session = sessions.AsyncAuthorizedSession(mock_creds) + mock_auth_request = mock.AsyncMock(spec=transport.Request) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_request + ) init_started = asyncio.Event() init_can_finish = asyncio.Event() @@ -1126,3 +1289,1150 @@ async def slow_mtls_init(): assert not session._mtls_init_task.cancelled() assert session._is_mtls is True await session.close() + + @pytest.mark.asyncio + async def test_401_mtls_rotation_e2e_unmocked_configure(self): + """End-to-end 401 recovery test with unmocked configure_mtls_channel.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(return_value=None) + + cert_v1 = b"cert_v1" + key_v1 = b"key_v1" + cert_v2 = b"cert_v2" + key_v2 = b"key_v2" + + certs_queue = [(True, cert_v1, key_v1), (True, cert_v2, key_v2)] + + async def mock_get_cert(cb=None): + if certs_queue: + return certs_queue.pop(0) + return (True, cert_v2, key_v2) + + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + side_effect=mock_get_cert, + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + session = sessions.AsyncAuthorizedSession(mock_creds) + await session.configure_mtls_channel() + assert session.is_mtls is True + assert session._cached_cert == cert_v1 + first_auth_req = session._auth_request + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_resp_200 = mock.Mock() + mock_resp_200.status_code = http_client.OK + mock_resp_200.close = mock.AsyncMock() + + with ( + mock.patch( + "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response", + new_callable=mock.AsyncMock, + return_value=(cert_v2, key_v2, b"fp1", b"fp2"), + ), + mock.patch.object( + sessions.AiohttpRequest, + "__call__", + side_effect=[mock_resp_401, mock_resp_200], + ), + ): + resp = await session.request( + "GET", "https://pubsub.mtls.googleapis.com/test" + ) + assert resp.status_code == 200 + assert session.is_mtls is True + assert session._cached_cert == cert_v2 + assert session._auth_request is not first_auth_req + assert mock_creds.refresh.call_count == 1 + await session.close() + + @pytest.mark.asyncio + async def test_401_cert_check_type_error_falls_back_to_refresh(self): + """Verifies that TypeError in cert check logs warning and falls back to refresh and retry.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(return_value=None) + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_resp_200 = mock.Mock() + mock_resp_200.status_code = http_client.OK + mock_resp_200.close = mock.AsyncMock() + + mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200]) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + session._is_mtls = True + session._cached_cert = b"some_cert" + + with ( + mock.patch( + "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response", + new_callable=mock.AsyncMock, + side_effect=TypeError("Callback returned invalid type"), + ), + mock.patch.object( + session, "configure_mtls_channel", new_callable=mock.AsyncMock + ) as mock_conf, + mock.patch.object(sessions._LOGGER, "warning") as mock_warn, + ): + resp = await session.request( + "GET", "https://pubsub.mtls.googleapis.com/test" + ) + assert resp == mock_resp_200 + mock_conf.assert_not_called() + mock_creds.refresh.assert_called_once() + assert any( + "Falling back to credential refresh and retry." in str(call) + for call in mock_warn.call_args_list + ) + await session.close() + + @pytest.mark.asyncio + async def test_401_cert_check_without_cached_cert_skips_reconfiguration(self): + """Verifies that when cached_cert is None, cert check produces equal fingerprints and skips reconfiguration.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(return_value=None) + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_resp_200 = mock.Mock() + mock_resp_200.status_code = http_client.OK + mock_resp_200.close = mock.AsyncMock() + + mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200]) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + session._is_mtls = True + session._cached_cert = None + + with ( + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + new_callable=mock.AsyncMock, + return_value=(True, b"cert_bytes", b"key_bytes"), + ), + mock.patch("google.auth._agent_identity_utils.parse_certificate"), + mock.patch( + "google.auth._agent_identity_utils.calculate_certificate_fingerprint", + return_value="FINGERPRINT_1", + ), + mock.patch.object( + session, "configure_mtls_channel", new_callable=mock.AsyncMock + ) as mock_conf, + mock.patch.object(sessions._LOGGER, "info") as mock_info, + ): + resp = await session.request( + "GET", "https://pubsub.mtls.googleapis.com/test" + ) + assert resp == mock_resp_200 + mock_conf.assert_not_called() + mock_creds.refresh.assert_called_once() + assert any( + "certificate has not changed" in str(call) + for call in mock_info.call_args_list + ) + await session.close() + + @pytest.mark.asyncio + async def test_401_refresh_raises_invalid_operation_returns_401(self): + """Verifies that exceptions.InvalidOperation during refresh is caught and returns the 401 response.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock( + side_effect=exceptions.InvalidOperation("Invalid operation") + ) + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_auth_req = mock.AsyncMock(return_value=mock_resp_401) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + + resp = await session.request("GET", "https://example.com") + assert resp == mock_resp_401 + mock_creds.refresh.assert_called_once() + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_idempotent_when_called_repeatedly(self): + """Tests that calling configure_mtls_channel repeatedly without a new callback reuses the existing task.""" + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert_bytes", b"key_bytes"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + await session.configure_mtls_channel() + assert session.is_mtls is True + first_auth_req = session._auth_request + first_task = session._mtls_init_task + + # Call configure_mtls_channel again without new callback + await session.configure_mtls_channel() + assert session._auth_request is first_auth_req + assert session._mtls_init_task is first_task + + # Call configure_mtls_channel with a new callback - should reconfigure + def new_callback(): + return b"new_cert", b"new_key" + + await session.configure_mtls_channel(new_callback) + assert session._auth_request is not first_auth_req + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_exception_preserves_existing_mtls_state(self): + """Tests that an exception during re-configuration does not clear existing is_mtls and cached_cert.""" + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert_v1", b"key_v1"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + await session.configure_mtls_channel() + assert session.is_mtls is True + assert session._cached_cert == b"cert_v1" + first_auth_req = session._auth_request + + # Now attempt reconfiguring with a failing callback/context + with ( + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + side_effect=RuntimeError("Reconfig failure"), + ), + pytest.raises(exceptions.MutualTLSChannelError), + ): + await session.configure_mtls_channel(lambda: (b"cert_v2", b"key_v2")) + + # Session should still retain its previous mTLS state and active auth_request + assert session.is_mtls is True + assert session._cached_cert == b"cert_v1" + assert session._auth_request is first_auth_req + await session.close() + + @pytest.mark.asyncio + async def test_cert_rotation_credential_refresh_not_implemented_retries(self): + """Validate credentials that raise NotImplementedError on refresh() + still trigger a retry after mTLS reconfiguration, not return the 401.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(side_effect=NotImplementedError) + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_resp_200 = mock.Mock() + mock_resp_200.status_code = http_client.OK + mock_resp_200.close = mock.AsyncMock() + + mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200]) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + session._is_mtls = True + session._cached_cert = b"old_cert" + + with mock.patch( + "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response", + new_callable=mock.AsyncMock, + ) as mock_check: + with mock.patch.object( + session, "configure_mtls_channel", new_callable=mock.AsyncMock + ) as mock_conf: + mock_check.return_value = ( + b"new_cert", + b"new_key", + b"old_fp", + b"new_fp", + ) + + resp = await session.request( + "GET", "https://pubsub.mtls.googleapis.com/test" + ) + + # Validate that the handler falls through to `return None` + # on NotImplementedError in order to signal retry. + assert resp == mock_resp_200 + mock_conf.assert_called_once() + mock_creds.refresh.assert_called_once() + assert mock_auth_req.call_count == 2 + mock_resp_401.close.assert_called_once() + + await session.close() + + @pytest.mark.asyncio + async def test_401_mtls_consecutive_multi_rotation(self): + """Verifies that consecutive rotations (v1 -> v2 -> v3) succeed. + + `configure_mtls_channel` is deliberately NOT mocked here. The + low-level helpers are patched instead so the real reconfiguration path + executes; otherwise this test would still pass even if rotation + stopped swapping the transport or started clobbering the shared + `_client_cert_callback`. + """ + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(return_value=None) + + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert_v1", b"key_v1"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + + def user_cb(): + return (b"cert_v1", b"key_v1") + + session = sessions.AsyncAuthorizedSession(mock_creds) + # Configure with an explicit, non-None user callback. A rotation + # that clobbers this shared attribute is then observable; with a + # default of None the overwrite would be a silent no-op. + await session.configure_mtls_channel(user_cb) + assert session._cached_cert == b"cert_v1" + assert session._client_cert_callback is user_cb + + rotations = ((b"cert_v1", b"cert_v2"), (b"cert_v2", b"cert_v3")) + for old_cert, new_cert in rotations: + prev_auth_request = session._auth_request + with ( + mock.patch( + "google.auth.aio.transport.mtls." + "check_parameters_for_unauthorized_response", + new_callable=mock.AsyncMock, + return_value=(new_cert, b"key", b"fp_old", b"fp_new"), + ) as mock_check, + mock.patch.object( + sessions.AiohttpRequest, + "__call__", + side_effect=[ + mock.Mock(status_code=401, close=mock.AsyncMock()), + mock.Mock(status_code=200, close=mock.AsyncMock()), + ], + ), + ): + resp = await session.request( + "GET", "https://pubsub.mtls.googleapis.com/test" + ) + + assert resp.status_code == 200 + # The check is driven by the previously cached cert and the + # user's callback (None), so the on-disk cert can be read. + mock_check.assert_called_once_with(old_cert, user_cb) + # Real reconfiguration ran: cert cached and transport swapped. + assert session._cached_cert == new_cert + assert session._auth_request is not prev_auth_request + assert session.is_mtls is True + # Shared callback state must survive rotation untouched. + assert session._client_cert_callback is user_cb + + await session.close() + + @pytest.mark.asyncio + async def test_non_mtls_not_implemented_refresh_returns_401_without_retry(self): + """Verifies that non-mTLS 401s on credentials that raise NotImplementedError + return the 401 immediately without retrying.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(side_effect=NotImplementedError) + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_auth_req = mock.AsyncMock(return_value=mock_resp_401) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + + resp = await session.request("GET", "https://example.com") + assert resp == mock_resp_401 + assert mock_auth_req.call_count == 1 + mock_creds.refresh.assert_called_once() + mock_resp_401.close.assert_not_called() + await session.close() + + @pytest.mark.asyncio + async def test_cert_rotation_credential_refresh_invalid_operation_retries(self): + """Verifies that when mTLS is reconfigured, credentials raising InvalidOperation + (e.g., StaticCredentials) still retry on the reconfigured mTLS channel.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock( + side_effect=exceptions.InvalidOperation("Static credentials") + ) + + mock_resp_401 = mock.Mock() + mock_resp_401.status_code = http_client.UNAUTHORIZED + mock_resp_401.close = mock.AsyncMock() + + mock_resp_200 = mock.Mock() + mock_resp_200.status_code = http_client.OK + mock_resp_200.close = mock.AsyncMock() + + mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200]) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + session._is_mtls = True + session._cached_cert = b"old_cert" + + with ( + mock.patch( + "google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response", + new_callable=mock.AsyncMock, + return_value=(b"new_cert", b"new_key", b"old_fp", b"new_fp"), + ), + mock.patch.object( + session, "configure_mtls_channel", new_callable=mock.AsyncMock + ) as mock_conf, + ): + resp = await session.request( + "GET", "https://pubsub.mtls.googleapis.com/test" + ) + assert resp == mock_resp_200 + mock_conf.assert_called_once() + mock_creds.refresh.assert_called_once() + assert mock_auth_req.call_count == 2 + mock_resp_401.close.assert_called_once() + + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_reverts_to_default_when_callback_none(self): + """Tests that passing callback=None when a callback was previously set reconfigures back to ADC.""" + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + + def custom_cb(): + return b"c", b"k" + + await session.configure_mtls_channel(custom_cb) + assert session._client_cert_callback is custom_cb + task1 = session._mtls_init_task + + # Revert to default + await session.configure_mtls_channel(None) + assert session._client_cert_callback is None + assert session._mtls_init_task is not task1 + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_force_reconfigures_same_callback(self): + """Tests that force=True reconfigures even if callback is identical.""" + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession(mock_creds) + + await session.configure_mtls_channel() + task1 = session._mtls_init_task + + # Same callback with force=True + await session.configure_mtls_channel(force=True) + assert session._mtls_init_task is not task1 + await session.close() + + @pytest.mark.asyncio + async def test_concurrent_rotation_retries_for_non_refreshable_credentials(self): + """Concurrent 401s with non-refreshable credentials must both retry. + + Only one coroutine performs the rotation; the other skips the check + block via the dedupe counter. The skipping coroutine must still observe + that the channel was reconfigured since its own 401 and retry, rather + than returning the stale 401 to the caller. + """ + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(side_effect=NotImplementedError) + + def _resp(status): + return mock.Mock(status_code=status, close=mock.AsyncMock()) + + mock_resp_401_1 = _resp(http_client.UNAUTHORIZED) + mock_resp_401_2 = _resp(http_client.UNAUTHORIZED) + mock_resp_200_1 = _resp(http_client.OK) + mock_resp_200_2 = _resp(http_client.OK) + + mock_auth_req = mock.AsyncMock( + side_effect=[ + mock_resp_401_1, + mock_resp_401_2, + mock_resp_200_1, + mock_resp_200_2, + ] + ) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + session._is_mtls = True + session._cached_cert = b"old_cert" + + async def slow_check(*args, **kwargs): + # Hold the rotation lock long enough that the second coroutine + # queues behind it and then takes the dedupe path. + await asyncio.sleep(0.05) + return (b"new_cert", b"new_key", b"old_fp", b"new_fp") + + with ( + mock.patch( + "google.auth.aio.transport.mtls." + "check_parameters_for_unauthorized_response", + side_effect=slow_check, + ) as mock_check, + mock.patch.object( + session, "configure_mtls_channel", new_callable=mock.AsyncMock + ) as mock_conf, + ): + results = await asyncio.gather( + session.request("GET", "https://pubsub.mtls.googleapis.com/t1"), + session.request("GET", "https://pubsub.mtls.googleapis.com/t2"), + ) + + # Exactly one coroutine ran the check and the rotation. + assert mock_check.call_count == 1 + assert mock_conf.call_count == 1 + # Both requests must have been retried on the rotated channel. + assert results == [mock_resp_200_1, mock_resp_200_2] + assert mock_auth_req.call_count == 4 + + await session.close() + + @pytest.mark.asyncio + async def test_reconfigure_to_non_mtls_replaces_stale_mtls_transport(self): + """A session leaving mTLS must not keep serving the old client cert.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + with mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ): + session = sessions.AsyncAuthorizedSession(mock_creds) + await session.configure_mtls_channel() + + assert session.is_mtls is True + mtls_transport = session._auth_request + + # The workload stops providing a client certificate. + with mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(False, None, None), + ): + await session.configure_mtls_channel(force=True) + + assert session.is_mtls is False + assert session._cached_cert is None + # The stale mTLS transport must be retired, not silently reused. + assert session._auth_request is not mtls_transport + assert mtls_transport in session._old_auth_requests + + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_propagates_caller_cancellation(self): + """Cancelling the caller must raise, not be swallowed by cleanup.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + + async def slow_get_cert(cb=None): + await asyncio.sleep(10) + return (True, b"cert", b"key") + + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + side_effect=slow_get_cert, + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + session = sessions.AsyncAuthorizedSession(mock_creds) + caller = asyncio.create_task(session.configure_mtls_channel()) + await asyncio.sleep(0.02) + + caller.cancel() + with pytest.raises(asyncio.CancelledError): + await caller + + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_follows_replacement_task(self): + """A caller awaiting a task that gets replaced follows the new one.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + + release = asyncio.Event() + + async def gated_get_cert(cb=None): + await release.wait() + return (True, b"cert", b"key") + + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + side_effect=gated_get_cert, + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession"), + ): + session = sessions.AsyncAuthorizedSession(mock_creds) + + # First caller starts and blocks on the gated configuration. + waiter = asyncio.create_task(session.configure_mtls_channel()) + await asyncio.sleep(0.02) + first_task = session._mtls_init_task + assert first_task is not None + + # A forced reconfiguration cancels and replaces that task. + release.set() + await session.configure_mtls_channel(force=True) + assert session._mtls_init_task is not first_task + + # The original caller must not surface a cancellation it never + # requested; it follows the replacement task instead. + await waiter + assert session.is_mtls is True + + await session.close() + + @pytest.mark.asyncio + async def test_retrieve_task_exception_helper(self): + """The done-callback marks failures retrieved and tolerates cancels.""" + + async def boom(): + raise RuntimeError("boom") + + task = asyncio.create_task(boom()) + task.add_done_callback(sessions._retrieve_task_exception) + with pytest.raises(RuntimeError): + await task + + async def sleeper(): + await asyncio.sleep(10) + + cancelled = asyncio.create_task(sleeper()) + cancelled.cancel() + with pytest.raises(asyncio.CancelledError): + await cancelled + # Must not raise CancelledError/InvalidStateError when inspected. + sessions._retrieve_task_exception(cancelled) + + @pytest.mark.asyncio + async def test_reconfigure_cancellation_while_retiring_old_task_propagates(self): + """A cancellation aimed at the reconfiguring coroutine must not be eaten. + + `configure_mtls_channel` cancels the task it is replacing and waits for + it to unwind. That wait has to absorb only the *retired task's* + cancellation; a cancellation targeting the caller still has to + propagate. + """ + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock() + ) + + calls = {"n": 0} + + async def run_in_executor(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + # Take a measurable amount of time to unwind so the replacing + # coroutine is still parked in the wait when we cancel it. + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + await asyncio.sleep(0.05) + raise + # Any replacement task finishes immediately. + return False + + with mock.patch.object(sessions.mtls, "_run_in_executor", new=run_in_executor): + first = asyncio.create_task(session.configure_mtls_channel()) + await asyncio.sleep(0.01) + assert session._mtls_init_task is not None + + replacer = asyncio.create_task(session.configure_mtls_channel(force=True)) + await asyncio.sleep(0.01) # parked waiting on the retired task + + replacer.cancel() + with pytest.raises(asyncio.CancelledError): + await replacer + + first.cancel() + try: + await first + except (asyncio.CancelledError, exceptions.MutualTLSChannelError): + pass + + await session.close() + + @pytest.mark.asyncio + async def test_close_cancels_in_flight_mtls_init(self): + """`close()` must actually cancel a still-running initialization task.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock() + ) + + async def never(*args, **kwargs): + await asyncio.sleep(10) + + with mock.patch.object(sessions.mtls, "_run_in_executor", new=never): + waiter = asyncio.create_task(session.configure_mtls_channel()) + await asyncio.sleep(0.01) + init_task = session._mtls_init_task + assert init_task is not None + assert not init_task.done() + + await session.close() + + assert init_task.done() + assert init_task.cancelled() + + with pytest.raises(asyncio.CancelledError): + await waiter + + @pytest.mark.asyncio + async def test_close_propagates_its_own_cancellation(self): + """A cancellation aimed at `close()` must not be absorbed.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock() + ) + + async def stubborn(*args, **kwargs): + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + # Unwind slowly so `close()` is parked in the wait. + await asyncio.sleep(0.05) + raise + + with mock.patch.object(sessions.mtls, "_run_in_executor", new=stubborn): + waiter = asyncio.create_task(session.configure_mtls_channel()) + await asyncio.sleep(0.01) + + closing = asyncio.create_task(session.close()) + await asyncio.sleep(0.01) + closing.cancel() + with pytest.raises(asyncio.CancelledError): + await closing + + waiter.cancel() + try: + await waiter + except (asyncio.CancelledError, exceptions.MutualTLSChannelError): + pass + + @pytest.mark.asyncio + async def test_failed_mtls_init_without_awaiter_is_marked_retrieved(self): + """A background init failure nobody awaits must not warn at GC time.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock() + ) + + async def slow_boom(*args, **kwargs): + await asyncio.sleep(0.05) + raise RuntimeError("cert check exploded") + + with mock.patch.object(sessions.mtls, "_run_in_executor", new=slow_boom): + # The only caller gives up before the task fails. + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(session.configure_mtls_channel(), 0.01) + task = session._mtls_init_task + assert task is not None + await asyncio.wait({task}) + + assert task.done() + assert not task.cancelled() + # The done-callback must have consumed the exception already. Without + # it, CPython would still have the task flagged for the + # "Task exception was never retrieved" report at collection time. + # + # This has to be checked BEFORE calling `task.exception()` below, since + # retrieving the exception here would clear the flag by itself and make + # the assertion vacuous. + assert task._log_traceback is False + assert isinstance(task.exception(), exceptions.MutualTLSChannelError) + + await session.close() + + @pytest.mark.asyncio + async def test_rotation_forces_refresh_when_earlier_refresh_lands_late(self): + """A refresh straddling a rotation must not satisfy the post-rotation one. + + A refresh that starts before the channel is rotated mints its token + over the old transport. Even though it completes after the rotation, + the rotating coroutine still has to perform its own refresh, otherwise + it retries carrying a token the new channel will reject. + """ + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + + async def slow_refresh(_transport): + await asyncio.sleep(0.10) + + mock_creds.refresh = mock.AsyncMock(side_effect=slow_refresh) + + seen = {} + + async def auth_req(url, *args, **kwargs): + seen[url] = seen.get(url, 0) + 1 + status = http_client.UNAUTHORIZED if seen[url] == 1 else http_client.OK + return mock.Mock(status_code=status, close=mock.AsyncMock()) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock(side_effect=auth_req) + ) + session._is_mtls = True + session._cached_cert = b"old_cert" + + async def fast_check(*args, **kwargs): + await asyncio.sleep(0.01) + return (b"new_cert", b"new_key", b"old_fp", b"new_fp") + + with ( + mock.patch( + "google.auth.aio.transport.mtls." + "check_parameters_for_unauthorized_response", + side_effect=fast_check, + ), + mock.patch.object( + session, "configure_mtls_channel", new_callable=mock.AsyncMock + ) as mock_conf, + ): + # The non-mTLS request takes `_refresh_lock` before the rotation + # begins and releases it only after the rotation has finished. + plain = asyncio.create_task( + session.request("GET", "https://example.com/plain") + ) + await asyncio.sleep(0) + rotating = asyncio.create_task( + session.request("GET", "https://pubsub.mtls.googleapis.com/x") + ) + await asyncio.gather(plain, rotating) + + assert mock_conf.call_count == 1 + # One refresh from the plain request (old channel) plus one forced by + # the rotation. Counting completions alone would wrongly treat the + # first as satisfying the second. + assert mock_creds.refresh.call_count == 2 + assert session._last_refresh_reconfig_gen == session._mtls_reconfig_counter + + await session.close() + + @pytest.mark.asyncio + async def test_configure_mtls_channel_rejects_a_closed_session(self): + """A closed session must refuse to build new mTLS state.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock() + ) + await session.close() + + with pytest.raises(exceptions.InvalidOperation): + await session.configure_mtls_channel() + assert session._mtls_init_task is None + + @pytest.mark.asyncio + async def test_rotation_racing_close_does_not_leak_a_transport(self): + """An in-flight rotation must not install a transport after `close()`. + + Otherwise `close()` returns, the rotation then builds a fresh + `aiohttp.ClientSession` and installs it on the dead session, and + nothing ever closes it. + """ + created = [] + + class _FakeClientSession: + def __init__(self, *args, **kwargs): + self.closed = False + created.append(self) + + async def close(self): + self.closed = True + + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + mock_creds.refresh = mock.AsyncMock(return_value=None) + + seen = {} + + async def auth_req(url, *args, **kwargs): + seen[url] = seen.get(url, 0) + 1 + status = http_client.UNAUTHORIZED if seen[url] == 1 else http_client.OK + return mock.Mock(status_code=status, close=mock.AsyncMock()) + + gate = asyncio.Event() + + async def blocking_check(*args, **kwargs): + await gate.wait() + return (b"new_cert", b"new_key", b"old_fp", b"new_fp") + + with ( + mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + return_value=True, + ), + mock.patch( + "google.auth.aio.transport.mtls.make_client_cert_ssl_context", + return_value=mock.Mock(spec=ssl.SSLContext), + ), + mock.patch("aiohttp.TCPConnector"), + mock.patch("aiohttp.ClientSession", _FakeClientSession), + mock.patch( + "google.auth.aio.transport.mtls.get_client_cert_and_key", + return_value=(True, b"cert", b"key"), + ), + mock.patch( + "google.auth.aio.transport.mtls." + "check_parameters_for_unauthorized_response", + side_effect=blocking_check, + ), + mock.patch.object( + sessions.AiohttpRequest, "__call__", side_effect=auth_req + ), + ): + session = sessions.AsyncAuthorizedSession(mock_creds) + await session.configure_mtls_channel() + assert len(created) == 1 + + pending = asyncio.create_task( + session.request("GET", "https://pubsub.mtls.googleapis.com/x") + ) + await asyncio.sleep(0.02) # park inside the rotation + + await session.close() + + gate.set() + with pytest.raises(exceptions.MutualTLSChannelError): + await pending + + # No second transport was built, and the original one was closed. + assert len(created) == 1 + assert created[0].closed is True + + @pytest.mark.asyncio + async def test_configure_mtls_channel_wraps_use_client_cert_failure(self): + """A failure reading the cert config must surface as MutualTLSChannelError.""" + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock.AsyncMock() + ) + + with mock.patch( + "google.auth.transport._mtls_helper.check_use_client_cert", + side_effect=UnicodeDecodeError("utf-8", b"\xff", 0, 1, "bad byte"), + ): + with pytest.raises(exceptions.MutualTLSChannelError): + await session.configure_mtls_channel() + + await session.close() + + @pytest.mark.asyncio + async def test_rotation_landing_during_refresh_still_triggers_retry(self): + """A rotation that lands while we await refresh must not be missed. + + Whether the channel moved since this request's 401 has to be read at + the point the decision is made, not snapshotted before acquiring + `_refresh_lock`. A coroutine whose credentials cannot be refreshed + would otherwise return its stale 401 even though a concurrent + coroutine rotated the channel in the meantime. + """ + a_in_refresh = asyncio.Event() + b_rotated = asyncio.Event() + + mock_creds = mock.AsyncMock(spec=credentials.Credentials) + mock_creds.before_request = mock.AsyncMock(return_value=None) + + first_refresh = {"done": False} + + async def refresh_side_effect(*args, **kwargs): + if not first_refresh["done"]: + first_refresh["done"] = True + a_in_refresh.set() + await b_rotated.wait() + raise exceptions.InvalidOperation("static cert-bound token") + + mock_creds.refresh = mock.AsyncMock(side_effect=refresh_side_effect) + + seen = set() + + async def auth_request(url, method, data, headers, timeout, **kwargs): + if url not in seen: + seen.add(url) + return mock.Mock( + status_code=http_client.UNAUTHORIZED, close=mock.AsyncMock() + ) + return mock.Mock(status_code=http_client.OK, close=mock.AsyncMock()) + + mock_auth_req = mock.AsyncMock(side_effect=auth_request) + + session = sessions.AsyncAuthorizedSession( + mock_creds, auth_request=mock_auth_req + ) + session._is_mtls = True + session._cached_cert = b"old_cert" + + checks = {"n": 0} + + async def check_side_effect(cached_cert, callback): + checks["n"] += 1 + if checks["n"] == 1: + # The first coroutine sees an unchanged certificate. + return (b"old_cert", b"old_key", b"same_fp", b"same_fp") + return (b"new_cert", b"new_key", b"old_fp", b"new_fp") + + async def fake_configure(*args, **kwargs): + b_rotated.set() + + with ( + mock.patch( + "google.auth.aio.transport.mtls." + "check_parameters_for_unauthorized_response", + side_effect=check_side_effect, + ), + mock.patch.object( + session, "configure_mtls_channel", side_effect=fake_configure + ), + ): + task_a = asyncio.create_task( + session.request("GET", "https://pubsub.mtls.googleapis.com/a") + ) + await asyncio.wait_for(a_in_refresh.wait(), 5) + task_b = asyncio.create_task( + session.request("GET", "https://pubsub.mtls.googleapis.com/b") + ) + resp_a, resp_b = await asyncio.wait_for(asyncio.gather(task_a, task_b), 5) + + assert session._mtls_reconfig_counter == 1 + # The rotating coroutine retries, and so must the one that only + # learned about the rotation after its own refresh failed. + assert resp_b.status_code == http_client.OK + assert resp_a.status_code == http_client.OK + + await session.close()