chore: Add thread safety to overcome race condition during 401 client certificate rotation in transports

packages/google-auth/google/auth/transport/requests.py

@@ -20,6 +20,7 @@
 import http.client as http_client
 import logging
 import numbers
+import threading
 import time
 from typing import Optional
 
@@ -414,6 +415,7 @@ def __init__(
         self._refresh_timeout = refresh_timeout
         self._is_mtls = False
         self._default_host = default_host
+        self._reauth_lock = threading.Lock()
 
         if auth_request is None:
             self._auth_request_session = requests.Session()
@@ -655,33 +657,36 @@ def request(
                     prefix in url for prefix in MTLS_URL_PREFIXES
                 )
                 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:
+                    with self._reauth_lock:
+                        (
+                            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(
-                                "Client certificate has changed, reconfiguring mTLS "
-                                "channel."
+                                "Skipping reconfiguration of mTLS channel because the client"
+                                " certificate has not changed."
                             )
-                            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."
-                        )
             _LOGGER.info(
                 "Refreshing credentials due to a %s response. Attempt %s/%s.",
                 response.status_code,

packages/google-auth/google/auth/transport/urllib3.py

@@ -18,6 +18,7 @@
 
 import http.client as http_client
 import logging
+import threading
 import warnings
 
 # Certifi is Mozilla's certificate bundle. Urllib3 needs a certificate bundle
@@ -309,6 +310,7 @@ def __init__(
         # credentials.refresh).
         self._request = Request(self.http)
         self._is_mtls = False
+        self._reauth_lock = threading.Lock()
 
         # https://google.aip.dev/auth/4111
         # Attempt to use self-signed JWTs when a service account is used.
@@ -436,38 +438,48 @@ def urlopen(self, method, url, body=None, headers=None, **kwargs):
             and _credential_refresh_attempt < self._max_refresh_attempts
         ):
             if response.status == http_client.UNAUTHORIZED:
+                MTLS_URL_PREFIXES = [
+                    "mtls.googleapis.com",
+                    "mtls.sandbox.googleapis.com",
+                ]
+                use_mtls = getattr(self, "_is_mtls", False) and any(
+                    prefix in url for prefix in MTLS_URL_PREFIXES
+                )
+
                 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(
-                                client_cert_callback=lambda: (
-                                    call_cert_bytes,
-                                    call_key_bytes,
+                    with self._reauth_lock:
+                        (
+                            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(
+                                    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."
                             )
-                        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.",

packages/google-auth/tests/transport/test_requests.py

@@ -16,6 +16,7 @@
 import functools
 import http.client as http_client
 import os
+import threading
 from unittest import mock
 
 import freezegun
@@ -1109,3 +1110,66 @@ 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()
+        session = google.auth.transport.requests.AuthorizedSession(credentials)
+        session._is_mtls = True
+        session._cached_cert = b"cert"
+        mock_response = mock.Mock(status_code=http_client.UNAUTHORIZED)
+        mock_success_response = mock.Mock(status_code=http_client.OK)
+        mock_session_request.side_effect = [mock_response, mock_success_response]
+        real_lock = threading.Lock()
+        session._reauth_lock = real_lock
+        mock_check_params.return_value = (
+            b"new_cert_bytes",
+            b"new_key_bytes",
+            "old_fingerprint",
+            "new_fingerprint",
+        )
+        lock_held_during_call = {"held": False}
+
+        def verify_lock_held(*args, **kwargs):
+            lock_held_during_call["held"] = session._reauth_lock.locked()
+
+        session.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held)
+        session.request("GET", "https://example.mtls.googleapis.com/")
+
+        session.configure_mtls_channel.assert_called()
+        assert lock_held_during_call["held"] is True
+
+    @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 = mock.Mock()
+
+        session.request("GET", "https://example.mtls.googleapis.com/")
+
+        session.configure_mtls_channel.assert_not_called()

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
@@ -723,3 +724,59 @@ def test_configure_mtls_channel_subsequent_disabled(
         assert not is_mtls
         assert not authed_http._is_mtls
         assert isinstance(authed_http.http, urllib3.PoolManager)
+
+
+class TestAuthorizedHttpMTLSReauth:
+    @mock.patch(
+        "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+    )
+    def test_reauth_lock_acquired_on_unauthorized(self, mock_check_params):
+        credentials = mock.Mock()
+        http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials)
+        http_obj._is_mtls = True
+        http_obj._cached_cert = b"cert"
+        mock_response = mock.Mock()
+        mock_response.status = http_client.UNAUTHORIZED
+        http_obj.http.urlopen = mock.Mock(return_value=mock_response)
+        real_lock = threading.Lock()
+        http_obj._reauth_lock = real_lock
+        mock_check_params.return_value = (
+            b"new_cert_bytes",
+            b"new_key_bytes",
+            "old_fingerprint",
+            "new_fingerprint",
+        )
+        lock_held_during_call = {"held": False}
+
+        def verify_lock_held(*args, **kwargs):
+            lock_held_during_call["held"] = real_lock.locked()
+
+        http_obj.configure_mtls_channel = mock.Mock(side_effect=verify_lock_held)
+        http_obj.request("GET", "https://example.mtls.googleapis.com/")
+        http_obj.configure_mtls_channel.assert_called()
+        assert lock_held_during_call["held"] is True
+
+    @mock.patch(
+        "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+    )
+    def test_reauth_skipped_when_cert_fingerprint_matches(self, mock_check_params):
+        credentials = mock.Mock()
+        http_obj = google.auth.transport.urllib3.AuthorizedHttp(credentials)
+        http_obj._is_mtls = True
+        http_obj._cached_cert = b"cert"
+        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]
+        )
+        mock_check_params.return_value = (
+            b"same_cert_bytes",
+            b"same_key_bytes",
+            "same_fingerprint",
+            "same_fingerprint",
+        )
+        http_obj.configure_mtls_channel = mock.Mock()
+        http_obj.request("GET", "https://example.mtls.googleapis.com/")
+        http_obj.configure_mtls_channel.assert_not_called()