From ad645fe0ad8ef29ff01dfc00ad0137b9a1b88d97 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Tue, 9 Jun 2026 04:45:55 +0000
Subject: [PATCH 01/14] feat: add streaming support for transparent cert
 rotation retries

---
 google/auth/transport/grpc.py                 | 293 +++++++++++++++++-
 .../transport/test_aiohttp_requests.py        |  14 +-
 2 files changed, 295 insertions(+), 12 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index e541d20ca..e76af4bde 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -16,8 +16,10 @@
 
 from __future__ import absolute_import
 
+import logging
 import logging
 
+_LOGGER = logging.getLogger(__name__)
 from google.auth import exceptions
 from google.auth.transport import _mtls_helper
 from google.oauth2 import service_account
@@ -208,7 +210,7 @@ def my_client_cert_callback():
 
         channel = google.auth.transport.grpc.secure_authorized_channel(
             credentials, request, mtls_endpoint)
-
+      
     Args:
         credentials (google.auth.credentials.Credentials): The credentials to
             add to requests.
@@ -253,6 +255,7 @@ def my_client_cert_callback():
         )
 
     # If SSL credentials are not explicitly set, try client_cert_callback and ADC.
+    cached_cert = None
     if not ssl_credentials:
         use_client_cert = _mtls_helper.check_use_client_cert()
         if use_client_cert and client_cert_callback:
@@ -261,10 +264,12 @@ def my_client_cert_callback():
             ssl_credentials = grpc.ssl_channel_credentials(
                 certificate_chain=cert, private_key=key
             )
+            cached_cert = cert   
         elif use_client_cert:
             # Use application default SSL credentials.
-            adc_ssl_credentils = SslCredentials()
-            ssl_credentials = adc_ssl_credentils.ssl_credentials
+            adc_ssl_credentials = SslCredentials()
+            ssl_credentials = adc_ssl_credentials.ssl_credentials
+            cached_cert = adc_ssl_credentials._cached_cert
         else:
             ssl_credentials = grpc.ssl_channel_credentials()
 
@@ -272,9 +277,27 @@ def my_client_cert_callback():
     composite_credentials = grpc.composite_channel_credentials(
         ssl_credentials, google_auth_credentials
     )
-
-    return grpc.secure_channel(target, composite_credentials, **kwargs)
-
+    is_retry = kwargs.pop("_is_retry", False)
+    channel = grpc.secure_channel(target, composite_credentials, **kwargs)
+    # Check if we are already inside a retry to avoid infinite recursion
+    if cached_cert and not is_retry:
+        # Package arguments to recreate the channel if rotation occurs
+        factory_args = {
+            "credentials": credentials,
+            "request": request, 
+            "target": target,
+            "ssl_credentials": None,
+            "client_cert_callback": client_cert_callback,
+            "_is_retry": True, # Hidden flag to stop recursion
+            **kwargs
+        }
+        interceptor = _MTLSCallInterceptor()
+
+        wrapper = _MTLSRefreshingChannel(target, factory_args, channel, cached_cert)
+     
+        interceptor._wrapper = wrapper   
+        return grpc.intercept_channel(wrapper, interceptor)
+    return channel
 
 class SslCredentials:
     """Class for application default SSL credentials.
@@ -292,6 +315,7 @@ class SslCredentials:
 
     def __init__(self):
         use_client_cert = _mtls_helper.check_use_client_cert()
+        self._cached_cert = None
         if not use_client_cert:
             self._is_mtls = False
         else:
@@ -323,6 +347,7 @@ def ssl_credentials(self):
                 self._ssl_credentials = grpc.ssl_channel_credentials(
                     certificate_chain=cert, private_key=key
                 )
+                self._cached_cert = cert
             except exceptions.ClientCertError as caught_exc:
                 new_exc = exceptions.MutualTLSChannelError(caught_exc)
                 raise new_exc from caught_exc
@@ -335,3 +360,259 @@ def ssl_credentials(self):
     def is_mtls(self):
         """Indicates if the created SSL channel credentials is mutual TLS."""
         return self._is_mtls
+
+class _MTLSCallInterceptor(
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor,
+):
+    def __init__(self):
+        self._wrapper = None
+        self._max_retries = 2 # Set your desired limit here
+
+    def _should_retry(self, code, retry_count):
+        if code != grpc.StatusCode.UNAUTHENTICATED or not self._wrapper:
+            return False
+
+        if retry_count >= self._max_retries:
+            _LOGGER.debug("Max retries reached (%d/%d).", retry_count, self._max_retries)
+            return False
+
+        # Fingerprint check logic
+        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._wrapper._cached_cert)
+        return cached_fp != current_fp
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        retry_count = 0
+
+        while True:
+            try:
+                # Every time we call continuation(), our Wrapper (which is the channel
+                # being intercepted) will point to its CURRENT active raw channel.
+                response = continuation(client_call_details, request)
+                status_code = response.code()
+            except grpc.RpcError as e:
+                status_code = e.code()
+                if not self._should_retry(status_code, retry_count):
+                    raise e
+                # If we should retry, we fall through to the refresh logic below
+
+            if self._should_retry(status_code, retry_count):
+                retry_count += 1
+                # Tell the wrapper to swap the channel.
+                # We don't need the wrapper to execute the retry; the loop does it!
+                self._wrapper.refresh_logic(retry_count)
+                continue # Jump back to the start of the while loop
+
+            return response
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        return _RetryableUnaryStreamCall(continuation, client_call_details, request, self)
+
+    def intercept_stream_unary(self, continuation, client_call_details, request_iterator):
+        response = continuation(client_call_details, request_iterator)
+        return _RefreshTriggeringFuture(response, self)
+
+    def intercept_stream_stream(self, continuation, client_call_details, request_iterator):
+        return _RetryableStreamStreamCall(continuation, client_call_details, request_iterator, self)
+
+class _MTLSRefreshingChannel(grpc.Channel):
+    def __init__(self, target, factory_args, initial_channel, initial_cert):
+        self._target = target
+        self._factory_args = factory_args
+        self._channel = initial_channel
+        self._cached_cert = initial_cert
+        self._lock = threading.Lock()
+
+    def refresh_logic(self, count):
+        with self._lock:
+            # Re-check inside lock to prevent race conditions
+            _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
+            if cached_fp != current_fp:
+                _LOGGER.debug("Wrapper: Refreshing mTLS channel. Retry count: %d", count)
+                old_channel = self._channel
+                self._channel = secure_authorized_channel(**self._factory_args)
+
+                creds = _mtls_helper.get_client_ssl_credentials()
+                self._cached_cert = creds[1]
+                old_channel.close()
+
+    def unary_unary(self, method, *args, **kwargs):
+        # Always return a callable from the CURRENT channel
+        return self._channel.unary_unary(method, *args, **kwargs)
+
+    # Mandatory passthroughs
+    def unary_stream(self, method, *args, **kwargs): return self._channel.unary_stream(method, *args, **kwargs)
+    def stream_unary(self, method, *args, **kwargs): return self._channel.stream_unary(method, *args, **kwargs)
+    def stream_stream(self, method, *args, **kwargs): return self._channel.stream_stream(method, *args, **kwargs)
+    def subscribe(self, *args, **kwargs): return self._channel.subscribe(*args, **kwargs)
+    def unsubscribe(self, *args, **kwargs): return self._channel.unsubscribe(*args, **kwargs)
+    def close(self): self._channel.close()
+
+
+class _RetryableUnaryStreamCall(grpc.Call, collections.abc.Iterator):
+    def __init__(self, continuation, client_call_details, request, interceptor):
+        self._continuation = continuation
+        self._client_call_details = client_call_details
+        self._request = request
+        self._interceptor = interceptor
+        self._retry_count = 0
+        self._call = None
+        self._iterator = None
+        self._yielded_any = False
+        self._start_call()
+
+    def _start_call(self):
+        self._call = self._continuation(self._client_call_details, self._request)
+        self._iterator = iter(self._call)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        while True:
+            try:
+                val = next(self._iterator)
+                self._yielded_any = True
+                return val
+            except grpc.RpcError as e:
+                status_code = e.code()
+                if not self._yielded_any and self._interceptor._should_retry(status_code, self._retry_count):
+                    self._retry_count += 1
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    continue
+                
+                if self._interceptor._should_retry(status_code, 0):
+                    self._interceptor._wrapper.refresh_logic(1)
+                raise e
+
+    def cancel(self): self._call.cancel()
+    def is_active(self): return self._call.is_active()
+    def time_remaining(self): return self._call.time_remaining()
+    def add_callback(self, callback): self._call.add_callback(callback)
+    def initial_metadata(self): return self._call.initial_metadata()
+    def trailing_metadata(self): return self._call.trailing_metadata()
+    def code(self): return self._call.code()
+    def details(self): return self._call.details()
+
+
+class _RefreshTriggeringFuture(grpc.Call, grpc.Future):
+    def __init__(self, target_future, interceptor):
+        self._target_future = target_future
+        self._interceptor = interceptor
+
+    def result(self, timeout=None):
+        try:
+            return self._target_future.result(timeout)
+        except grpc.RpcError as e:
+            if self._interceptor._should_retry(e.code(), 0):
+                self._interceptor._wrapper.refresh_logic(1)
+            raise e
+
+    def exception(self, timeout=None):
+        exc = self._target_future.exception(timeout)
+        if isinstance(exc, grpc.RpcError):
+            if self._interceptor._should_retry(exc.code(), 0):
+                self._interceptor._wrapper.refresh_logic(1)
+        return exc
+
+    def cancel(self): return self._target_future.cancel()
+    def cancelled(self): return self._target_future.cancelled()
+    def running(self): return self._target_future.running()
+    def done(self): return self._target_future.done()
+    def add_done_callback(self, fn): self._target_future.add_done_callback(fn)
+    def is_active(self): return self._target_future.is_active()
+    def time_remaining(self): return self._target_future.time_remaining()
+    def add_callback(self, callback): self._target_future.add_callback(callback)
+    def initial_metadata(self): return self._target_future.initial_metadata()
+    def trailing_metadata(self): return self._target_future.trailing_metadata()
+    def code(self): return self._target_future.code()
+    def details(self): return self._target_future.details()
+
+
+class _ReplayableIterator(object):
+    def __init__(self, target_iterator):
+        self._target_iterator = target_iterator
+        self._buffer = []
+        self._lock = threading.Lock()
+        self._exhausted = False
+
+    def __iter__(self):
+        return _ReplayableIteratorReader(self)
+
+
+class _ReplayableIteratorReader(object):
+    def __init__(self, parent):
+        self._parent = parent
+        self._read_index = 0
+
+    def __next__(self):
+        with self._parent._lock:
+            if self._read_index < len(self._parent._buffer):
+                val = self._parent._buffer[self._read_index]
+                self._read_index += 1
+                return val
+            
+            if self._parent._exhausted:
+                raise StopIteration()
+
+            try:
+                val = next(self._parent._target_iterator)
+                self._parent._buffer.append(val)
+                self._read_index += 1
+                return val
+            except StopIteration:
+                self._parent._exhausted = True
+                raise
+
+
+class _RetryableStreamStreamCall(grpc.Call, collections.abc.Iterator):
+    def __init__(self, continuation, client_call_details, request_iterator, interceptor):
+        self._continuation = continuation
+        self._client_call_details = client_call_details
+        self._replayable_request_iterator = _ReplayableIterator(request_iterator)
+        self._interceptor = interceptor
+        self._retry_count = 0
+        self._call = None
+        self._response_iterator = None
+        self._yielded_any_response = False
+        self._start_call()
+
+    def _start_call(self):
+        req_iter = iter(self._replayable_request_iterator)
+        self._call = self._continuation(self._client_call_details, req_iter)
+        self._response_iterator = iter(self._call)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        while True:
+            try:
+                val = next(self._response_iterator)
+                self._yielded_any_response = True
+                return val
+            except grpc.RpcError as e:
+                status_code = e.code()
+                if not self._yielded_any_response and self._interceptor._should_retry(status_code, self._retry_count):
+                    self._retry_count += 1
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    continue
+                
+                if self._interceptor._should_retry(status_code, 0):
+                    self._interceptor._wrapper.refresh_logic(1)
+                raise e
+
+    def cancel(self): self._call.cancel()
+    def is_active(self): return self._call.is_active()
+    def time_remaining(self): return self._call.time_remaining()
+    def add_callback(self, callback): self._call.add_callback(callback)
+    def initial_metadata(self): return self._call.initial_metadata()
+    def trailing_metadata(self): return self._call.trailing_metadata()
+    def code(self): return self._call.code()
+    def details(self): return self._call.details()
diff --git a/tests_async/transport/test_aiohttp_requests.py b/tests_async/transport/test_aiohttp_requests.py
index d6a24da2e..4f4a41265 100644
--- a/tests_async/transport/test_aiohttp_requests.py
+++ b/tests_async/transport/test_aiohttp_requests.py
@@ -121,12 +121,14 @@ async def test_unsupported_session(self):
         with pytest.raises(ValueError):
             await aiohttp_requests.Request(http)
 
-    def test_timeout(self):
-        http = mock.create_autospec(
-            aiohttp.ClientSession, instance=True, _auto_decompress=False
-        )
-        request = aiohttp_requests.Request(http)
-        request(url="http://example.com", method="GET", timeout=5)
+    @pytest.mark.asyncio
+    async def test_timeout(self):
+    http = mock.create_autospec(
+        aiohttp.ClientSession, instance=True, _auto_decompress=False
+    )
+    request = aiohttp_requests.Request(http)
+    await request(url="http://example.com", method="GET", timeout=5)
+
 
 
 class CredentialsStub(google.auth._credentials_async.Credentials):

From 5e52bcc142ec50430ffb7430e9b782954566e0d2 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 11 Jun 2026 20:47:31 +0000
Subject: [PATCH 02/14] feat: enable transparent mid-stream retries by removing
 yielded checks

---
 google/auth/transport/grpc.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index e76af4bde..455a267d1 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -478,7 +478,7 @@ def __next__(self):
                 return val
             except grpc.RpcError as e:
                 status_code = e.code()
-                if not self._yielded_any and self._interceptor._should_retry(status_code, self._retry_count):
+                if self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     time.sleep(random.uniform(0.1, 1.0))
@@ -597,7 +597,7 @@ def __next__(self):
                 return val
             except grpc.RpcError as e:
                 status_code = e.code()
-                if not self._yielded_any_response and self._interceptor._should_retry(status_code, self._retry_count):
+                if self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     time.sleep(random.uniform(0.1, 1.0))

From 23a831c35d9059130dd9c4b846c55566866a3490 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 11 Jun 2026 20:54:06 +0000
Subject: [PATCH 03/14] feat: add threshold clear logic to prevent OOM during
 mid-stream retries

---
 google/auth/transport/grpc.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 455a267d1..b808b7b1a 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -478,7 +478,8 @@ def __next__(self):
                 return val
             except grpc.RpcError as e:
                 status_code = e.code()
-                if self._interceptor._should_retry(status_code, self._retry_count):
+                can_replay = self._replayable_request_iterator.can_replay()
+                if can_replay and self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     time.sleep(random.uniform(0.1, 1.0))

From 603d283625409afbef18df37898e1c76bda0c049 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 11 Jun 2026 21:26:12 +0000
Subject: [PATCH 04/14] feat: add transparent auto-retry for stream-unary calls

---
 google/auth/transport/grpc.py | 44 ++++++++++++++++++++++++++---------
 1 file changed, 33 insertions(+), 11 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index b808b7b1a..9eb167d80 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -411,8 +411,7 @@ def intercept_unary_stream(self, continuation, client_call_details, request):
         return _RetryableUnaryStreamCall(continuation, client_call_details, request, self)
 
     def intercept_stream_unary(self, continuation, client_call_details, request_iterator):
-        response = continuation(client_call_details, request_iterator)
-        return _RefreshTriggeringFuture(response, self)
+        return _RetryableStreamUnaryFuture(continuation, client_call_details, request_iterator, self)
 
     def intercept_stream_stream(self, continuation, client_call_details, request_iterator):
         return _RetryableStreamStreamCall(continuation, client_call_details, request_iterator, self)
@@ -500,18 +499,41 @@ def code(self): return self._call.code()
     def details(self): return self._call.details()
 
 
-class _RefreshTriggeringFuture(grpc.Call, grpc.Future):
-    def __init__(self, target_future, interceptor):
-        self._target_future = target_future
+class _RetryableStreamUnaryFuture(grpc.Call, grpc.Future):
+    def __init__(self, continuation, client_call_details, request_iterator, interceptor):
+        self._continuation = continuation
+        self._client_call_details = client_call_details
+        self._replayable_request_iterator = _ReplayableIterator(request_iterator)
         self._interceptor = interceptor
+        self._retry_count = 0
+        self._target_future = None
+        self._start_call()
+
+    def _start_call(self):
+        req_iter = iter(self._replayable_request_iterator)
+        self._target_future = self._continuation(self._client_call_details, req_iter)
 
     def result(self, timeout=None):
-        try:
-            return self._target_future.result(timeout)
-        except grpc.RpcError as e:
-            if self._interceptor._should_retry(e.code(), 0):
-                self._interceptor._wrapper.refresh_logic(1)
-            raise e
+        while True:
+            try:
+                return self._target_future.result(timeout)
+            except grpc.RpcError as e:
+                status_code = e.code()
+                can_replay = self._replayable_request_iterator.can_replay()
+                
+                if can_replay and self._interceptor._should_retry(status_code, self._retry_count):
+                    self._retry_count += 1
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    
+                    import time, random
+                    time.sleep(random.uniform(0.1, 1.0))
+                    
+                    self._start_call()
+                    continue
+                
+                if self._interceptor._should_retry(status_code, 0):
+                    self._interceptor._wrapper.refresh_logic(1)
+                raise e
 
     def exception(self, timeout=None):
         exc = self._target_future.exception(timeout)

From 25cade0ab1751bda20430d18d25f985ae2576f59 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Thu, 11 Jun 2026 21:36:22 +0000
Subject: [PATCH 05/14] feat: add logging during transparent stream retries

---
 google/auth/transport/grpc.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 9eb167d80..143a434ca 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -481,6 +481,7 @@ def __next__(self):
                 if can_replay and self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
                     time.sleep(random.uniform(0.1, 1.0))
                     self._start_call()
                     continue
@@ -524,6 +525,7 @@ def result(self, timeout=None):
                 if can_replay and self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
                     
                     import time, random
                     time.sleep(random.uniform(0.1, 1.0))
@@ -623,6 +625,7 @@ def __next__(self):
                 if self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
                     time.sleep(random.uniform(0.1, 1.0))
                     self._start_call()
                     continue

From d8eeb25b6f5acb7a00f28e924d93d6ee145bb251 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Fri, 10 Jul 2026 22:47:48 +0000
Subject: [PATCH 06/14] feat: Implement streaming resilience buffers and First
 Byte safeguard

---
 google/auth/transport/grpc.py | 24 ++++++++++++++++++------
 1 file changed, 18 insertions(+), 6 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 143a434ca..596efb0d8 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -17,7 +17,10 @@
 from __future__ import absolute_import
 
 import logging
-import logging
+import threading
+import collections.abc
+import time
+import random
 
 _LOGGER = logging.getLogger(__name__)
 from google.auth import exceptions
@@ -477,8 +480,7 @@ def __next__(self):
                 return val
             except grpc.RpcError as e:
                 status_code = e.code()
-                can_replay = self._replayable_request_iterator.can_replay()
-                if can_replay and self._interceptor._should_retry(status_code, self._retry_count):
+                if not self._yielded_any and self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
@@ -559,15 +561,20 @@ def details(self): return self._target_future.details()
 
 
 class _ReplayableIterator(object):
-    def __init__(self, target_iterator):
+    def __init__(self, target_iterator, max_items=1000):
         self._target_iterator = target_iterator
+        self._max_items = max_items
         self._buffer = []
         self._lock = threading.Lock()
         self._exhausted = False
+        self._can_replay = True
 
     def __iter__(self):
         return _ReplayableIteratorReader(self)
 
+    def can_replay(self):
+        return self._can_replay
+
 
 class _ReplayableIteratorReader(object):
     def __init__(self, parent):
@@ -586,7 +593,11 @@ def __next__(self):
 
             try:
                 val = next(self._parent._target_iterator)
-                self._parent._buffer.append(val)
+                if self._parent._can_replay:
+                    self._parent._buffer.append(val)
+                    if len(self._parent._buffer) > self._parent._max_items:
+                        self._parent._buffer.clear()
+                        self._parent._can_replay = False
                 self._read_index += 1
                 return val
             except StopIteration:
@@ -622,7 +633,8 @@ def __next__(self):
                 return val
             except grpc.RpcError as e:
                 status_code = e.code()
-                if self._interceptor._should_retry(status_code, self._retry_count):
+                can_replay = self._replayable_request_iterator.can_replay()
+                if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")

From 9ac670e92ac0c0ac32e2cb1dbbdf71c39f5f0ce5 Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Tue, 14 Jul 2026 14:03:07 +0000
Subject: [PATCH 07/14] Fix BidiRpc done_callback integration for streaming
 cert rotation interceptors

---
 google/auth/transport/grpc.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 596efb0d8..2e060e959 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -650,6 +650,7 @@ def cancel(self): self._call.cancel()
     def is_active(self): return self._call.is_active()
     def time_remaining(self): return self._call.time_remaining()
     def add_callback(self, callback): self._call.add_callback(callback)
+    def add_done_callback(self, callback): self._call.add_done_callback(callback)
     def initial_metadata(self): return self._call.initial_metadata()
     def trailing_metadata(self): return self._call.trailing_metadata()
     def code(self): return self._call.code()

From ff9b2913d3ced12b966073920b02a0f69cde38ac Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Tue, 14 Jul 2026 20:45:04 +0000
Subject: [PATCH 08/14] fix(grpc): Handle bidi.py add_done_callback missing
 method and fix refresh_logic fallback behavior

---
 google/auth/transport/grpc.py | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 2e060e959..923d6ed64 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -434,12 +434,20 @@ def refresh_logic(self, count):
             if cached_fp != current_fp:
                 _LOGGER.debug("Wrapper: Refreshing mTLS channel. Retry count: %d", count)
                 old_channel = self._channel
+                
+                client_cert_callback = self._factory_args.get("client_cert_callback")
+                if client_cert_callback:
+                    cert, _ = client_cert_callback()
+                    self._cached_cert = cert
+                else:
+                    try:
+                        creds = _mtls_helper.get_client_ssl_credentials()
+                        self._cached_cert = creds[1]
+                    except Exception:
+                        pass
+                        
                 self._channel = secure_authorized_channel(**self._factory_args)
-
-                creds = _mtls_helper.get_client_ssl_credentials()
-                self._cached_cert = creds[1]
                 old_channel.close()
-
     def unary_unary(self, method, *args, **kwargs):
         # Always return a callable from the CURRENT channel
         return self._channel.unary_unary(method, *args, **kwargs)

From 6a4a4b57d908ff98052d329daac2a1257a66f7fd Mon Sep 17 00:00:00 2001
From: Radhika Agrawal <agrawalradhika@google.com>
Date: Wed, 15 Jul 2026 19:03:57 +0000
Subject: [PATCH 09/14] feat: Fix streaming bidi flow on grpc

Signed-off-by: Radhika Agrawal <agrawalradhika@google.com>
---
 google/auth/transport/grpc.py | 61 ++++++++++++++++++++---------------
 1 file changed, 35 insertions(+), 26 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 923d6ed64..3f8ecfcc9 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -434,7 +434,6 @@ def refresh_logic(self, count):
             if cached_fp != current_fp:
                 _LOGGER.debug("Wrapper: Refreshing mTLS channel. Retry count: %d", count)
                 old_channel = self._channel
-                
                 client_cert_callback = self._factory_args.get("client_cert_callback")
                 if client_cert_callback:
                     cert, _ = client_cert_callback()
@@ -620,46 +619,56 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._replayable_request_iterator = _ReplayableIterator(request_iterator)
         self._interceptor = interceptor
         self._retry_count = 0
+        self._done_callbacks = []
         self._call = None
         self._response_iterator = None
         self._yielded_any_response = False
         self._start_call()
-
+    def _on_inner_call_done(self, inner_call):
+        status_code = inner_call.code()
+        if status_code == grpc.StatusCode.UNAUTHENTICATED:
+            can_replay = self._replayable_request_iterator.can_replay()
+            if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count):
+                # IMPORTANT: Swallow the callback so bidi.py does not tear down 
+                # the router tracking threads while we attempt to reconstruct the stream!
+                return
+
+        for cb in self._done_callbacks:
+            cb(self)
     def _start_call(self):
         req_iter = iter(self._replayable_request_iterator)
         self._call = self._continuation(self._client_call_details, req_iter)
         self._response_iterator = iter(self._call)
+        self._call.add_done_callback(self._on_inner_call_done)
+
+    def add_done_callback(self, callback):
+        # Store requested callbacks natively instead of forwarding blindly
+        self._done_callbacks.append(callback)
 
     def __iter__(self):
         return self
 
     def __next__(self):
-        while True:
-            try:
-                val = next(self._response_iterator)
-                self._yielded_any_response = True
-                return val
-            except grpc.RpcError as e:
-                status_code = e.code()
-                can_replay = self._replayable_request_iterator.can_replay()
-                if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count):
-                    self._retry_count += 1
-                    self._interceptor._wrapper.refresh_logic(self._retry_count)
-                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
-                    time.sleep(random.uniform(0.1, 1.0))
-                    self._start_call()
-                    continue
-                
-                if self._interceptor._should_retry(status_code, 0):
-                    self._interceptor._wrapper.refresh_logic(1)
-                raise e
-
-    def cancel(self): self._call.cancel()
+        try:
+            response = next(self._response_iterator)
+            self._yielded_any_response = True
+            return response
+        except grpc.RpcError as e:
+            if not self._yielded_any_response and self._interceptor._should_retry(
+                e.code(), self._retry_count
+            ):
+                self._interceptor._wrapper.refresh_logic(self._retry_count)
+                self._retry_count += 1
+                self._start_call()
+                return next(self)
+            raise e
+
+    # Simple pass-throughs for the remaining gRPC methods
+    def cancel(self): return self._call.cancel()
+    def code(self): return self._call.code()
+    def details(self): return self._call.details()
     def is_active(self): return self._call.is_active()
     def time_remaining(self): return self._call.time_remaining()
     def add_callback(self, callback): self._call.add_callback(callback)
-    def add_done_callback(self, callback): self._call.add_done_callback(callback)
     def initial_metadata(self): return self._call.initial_metadata()
     def trailing_metadata(self): return self._call.trailing_metadata()
-    def code(self): return self._call.code()
-    def details(self): return self._call.details()

From d32869984782a68d6fba9082df6362e596eb9795 Mon Sep 17 00:00:00 2001
From: Jetski <jetski@google.com>
Date: Thu, 16 Jul 2026 19:23:46 +0000
Subject: [PATCH 10/14] fix(grpc): Robust mTLS streaming disconnect handling
 and test fixes

- Handled race conditions when old streams fail during cert rotation by tracking `attempt_cert`.
- Dropped the flawed `_channel.close()` that abruptly killed in-flight streams. Subscriptions are now safely migrated and the old channel is allowed to drain.
- Implemented `_ReplayableIterator` with dual-lock synchronization and `_can_replay` checks. This guarantees zombie threads gracefully abort context handoffs instead of dropping data or deadlocking.
- Fixed the `IndentationError` in `test_aiohttp_requests.py`.
- Added new test suite to cover the tricky streaming handoff concurrency bugs.
---
 google/auth/transport/grpc.py                 | 152 ++++++++++++------
 tests/transport/test_grpc_mtls_streaming.py   | 123 ++++++++++++++
 .../transport/test_aiohttp_requests.py        |  11 +-
 3 files changed, 235 insertions(+), 51 deletions(-)
 create mode 100644 tests/transport/test_grpc_mtls_streaming.py

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 3f8ecfcc9..3588ed798 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -374,7 +374,7 @@ def __init__(self):
         self._wrapper = None
         self._max_retries = 2 # Set your desired limit here
 
-    def _should_retry(self, code, retry_count):
+    def _should_retry(self, code, retry_count, attempt_cert):
         if code != grpc.StatusCode.UNAUTHENTICATED or not self._wrapper:
             return False
 
@@ -382,14 +382,19 @@ def _should_retry(self, code, retry_count):
             _LOGGER.debug("Max retries reached (%d/%d).", retry_count, self._max_retries)
             return False
 
+        # If the wrapper has already rotated to a new cert, we can retry immediately
+        if attempt_cert != self._wrapper._cached_cert:
+            return True
+
         # Fingerprint check logic
-        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._wrapper._cached_cert)
+        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
         return cached_fp != current_fp
 
     def intercept_unary_unary(self, continuation, client_call_details, request):
         retry_count = 0
 
         while True:
+            attempt_cert = self._wrapper._cached_cert if self._wrapper else None
             try:
                 # Every time we call continuation(), our Wrapper (which is the channel
                 # being intercepted) will point to its CURRENT active raw channel.
@@ -397,11 +402,11 @@ def intercept_unary_unary(self, continuation, client_call_details, request):
                 status_code = response.code()
             except grpc.RpcError as e:
                 status_code = e.code()
-                if not self._should_retry(status_code, retry_count):
+                if not self._should_retry(status_code, retry_count, attempt_cert):
                     raise e
                 # If we should retry, we fall through to the refresh logic below
 
-            if self._should_retry(status_code, retry_count):
+            if self._should_retry(status_code, retry_count, attempt_cert):
                 retry_count += 1
                 # Tell the wrapper to swap the channel.
                 # We don't need the wrapper to execute the retry; the loop does it!
@@ -426,6 +431,7 @@ def __init__(self, target, factory_args, initial_channel, initial_cert):
         self._channel = initial_channel
         self._cached_cert = initial_cert
         self._lock = threading.Lock()
+        self._subscribers = set()
 
     def refresh_logic(self, count):
         with self._lock:
@@ -446,7 +452,14 @@ def refresh_logic(self, count):
                         pass
                         
                 self._channel = secure_authorized_channel(**self._factory_args)
-                old_channel.close()
+                
+                for callback in self._subscribers:
+                    try:
+                        old_channel.unsubscribe(callback)
+                    except Exception:
+                        pass
+                    self._channel.subscribe(callback)
+
     def unary_unary(self, method, *args, **kwargs):
         # Always return a callable from the CURRENT channel
         return self._channel.unary_unary(method, *args, **kwargs)
@@ -455,8 +468,17 @@ def unary_unary(self, method, *args, **kwargs):
     def unary_stream(self, method, *args, **kwargs): return self._channel.unary_stream(method, *args, **kwargs)
     def stream_unary(self, method, *args, **kwargs): return self._channel.stream_unary(method, *args, **kwargs)
     def stream_stream(self, method, *args, **kwargs): return self._channel.stream_stream(method, *args, **kwargs)
-    def subscribe(self, *args, **kwargs): return self._channel.subscribe(*args, **kwargs)
-    def unsubscribe(self, *args, **kwargs): return self._channel.unsubscribe(*args, **kwargs)
+    
+    def subscribe(self, callback, try_to_connect=False):
+        with self._lock:
+            self._subscribers.add(callback)
+            return self._channel.subscribe(callback, try_to_connect=try_to_connect)
+
+    def unsubscribe(self, callback):
+        with self._lock:
+            self._subscribers.discard(callback)
+            return self._channel.unsubscribe(callback)
+            
     def close(self): self._channel.close()
 
 
@@ -473,6 +495,7 @@ def __init__(self, continuation, client_call_details, request, interceptor):
         self._start_call()
 
     def _start_call(self):
+        self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         self._call = self._continuation(self._client_call_details, self._request)
         self._iterator = iter(self._call)
 
@@ -487,7 +510,7 @@ def __next__(self):
                 return val
             except grpc.RpcError as e:
                 status_code = e.code()
-                if not self._yielded_any and self._interceptor._should_retry(status_code, self._retry_count):
+                if not self._yielded_any and self._interceptor._should_retry(status_code, self._retry_count, self._attempt_cert):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
@@ -495,8 +518,9 @@ def __next__(self):
                     self._start_call()
                     continue
                 
-                if self._interceptor._should_retry(status_code, 0):
-                    self._interceptor._wrapper.refresh_logic(1)
+                if getattr(self._interceptor, "_wrapper", None):
+                    if self._interceptor._should_retry(status_code, 0, self._attempt_cert):
+                        self._interceptor._wrapper.refresh_logic(1)
                 raise e
 
     def cancel(self): self._call.cancel()
@@ -520,6 +544,7 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._start_call()
 
     def _start_call(self):
+        self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
         self._target_future = self._continuation(self._client_call_details, req_iter)
 
@@ -531,26 +556,27 @@ def result(self, timeout=None):
                 status_code = e.code()
                 can_replay = self._replayable_request_iterator.can_replay()
                 
-                if can_replay and self._interceptor._should_retry(status_code, self._retry_count):
+                if can_replay and self._interceptor._should_retry(status_code, self._retry_count, self._attempt_cert):
                     self._retry_count += 1
                     self._interceptor._wrapper.refresh_logic(self._retry_count)
                     _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
                     
-                    import time, random
                     time.sleep(random.uniform(0.1, 1.0))
                     
                     self._start_call()
                     continue
                 
-                if self._interceptor._should_retry(status_code, 0):
-                    self._interceptor._wrapper.refresh_logic(1)
+                if getattr(self._interceptor, "_wrapper", None):
+                    if self._interceptor._should_retry(status_code, 0, self._attempt_cert):
+                        self._interceptor._wrapper.refresh_logic(1)
                 raise e
 
     def exception(self, timeout=None):
         exc = self._target_future.exception(timeout)
         if isinstance(exc, grpc.RpcError):
-            if self._interceptor._should_retry(exc.code(), 0):
-                self._interceptor._wrapper.refresh_logic(1)
+            if getattr(self._interceptor, "_wrapper", None):
+                if self._interceptor._should_retry(exc.code(), 0, getattr(self, "_attempt_cert", None)):
+                    self._interceptor._wrapper.refresh_logic(1)
         return exc
 
     def cancel(self): return self._target_future.cancel()
@@ -572,15 +598,22 @@ def __init__(self, target_iterator, max_items=1000):
         self._target_iterator = target_iterator
         self._max_items = max_items
         self._buffer = []
-        self._lock = threading.Lock()
         self._exhausted = False
         self._can_replay = True
+        
+        self._lock = threading.Lock()
+        self._consumer_lock = threading.Lock()
+        self._active_reader = None
 
     def __iter__(self):
-        return _ReplayableIteratorReader(self)
+        reader = _ReplayableIteratorReader(self)
+        with self._lock:
+            self._active_reader = reader
+        return reader
 
     def can_replay(self):
-        return self._can_replay
+        with self._lock:
+            return self._can_replay
 
 
 class _ReplayableIteratorReader(object):
@@ -589,27 +622,48 @@ def __init__(self, parent):
         self._read_index = 0
 
     def __next__(self):
-        with self._parent._lock:
-            if self._read_index < len(self._parent._buffer):
-                val = self._parent._buffer[self._read_index]
-                self._read_index += 1
-                return val
-            
-            if self._parent._exhausted:
-                raise StopIteration()
+        while True:
+            with self._parent._lock:
+                if self._read_index < len(self._parent._buffer):
+                    val = self._parent._buffer[self._read_index]
+                    self._read_index += 1
+                    return val
+
+                if self._parent._exhausted:
+                    raise StopIteration()
+
+                if self._parent._active_reader is not self:
+                    raise StopIteration()
+
+            with self._parent._consumer_lock:
+                with self._parent._lock:
+                    if self._read_index < len(self._parent._buffer):
+                        continue
+                    if self._parent._active_reader is not self:
+                        raise StopIteration()
+
+                try:
+                    val = next(self._parent._target_iterator)
+                except StopIteration:
+                    with self._parent._lock:
+                        if self._parent._active_reader is self:
+                            self._parent._exhausted = True
+                    raise
+
+            with self._parent._lock:
+                if self._parent._active_reader is not self:
+                    if self._parent._can_replay:
+                        self._parent._buffer.append(val)
+                    raise StopIteration()
 
-            try:
-                val = next(self._parent._target_iterator)
                 if self._parent._can_replay:
                     self._parent._buffer.append(val)
                     if len(self._parent._buffer) > self._parent._max_items:
                         self._parent._buffer.clear()
                         self._parent._can_replay = False
+
                 self._read_index += 1
                 return val
-            except StopIteration:
-                self._parent._exhausted = True
-                raise
 
 
 class _RetryableStreamStreamCall(grpc.Call, collections.abc.Iterator):
@@ -636,6 +690,7 @@ def _on_inner_call_done(self, inner_call):
         for cb in self._done_callbacks:
             cb(self)
     def _start_call(self):
+        self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
         self._call = self._continuation(self._client_call_details, req_iter)
         self._response_iterator = iter(self._call)
@@ -649,19 +704,26 @@ def __iter__(self):
         return self
 
     def __next__(self):
-        try:
-            response = next(self._response_iterator)
-            self._yielded_any_response = True
-            return response
-        except grpc.RpcError as e:
-            if not self._yielded_any_response and self._interceptor._should_retry(
-                e.code(), self._retry_count
-            ):
-                self._interceptor._wrapper.refresh_logic(self._retry_count)
-                self._retry_count += 1
-                self._start_call()
-                return next(self)
-            raise e
+        while True:
+            try:
+                val = next(self._response_iterator)
+                self._yielded_any_response = True
+                return val
+            except grpc.RpcError as e:
+                status_code = e.code()
+                can_replay = self._replayable_request_iterator.can_replay()
+                if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count, getattr(self, "_attempt_cert", None)):
+                    self._retry_count += 1
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    continue
+                
+                if getattr(self._interceptor, "_wrapper", None):
+                    if self._interceptor._should_retry(status_code, 0, getattr(self, "_attempt_cert", None)):
+                        self._interceptor._wrapper.refresh_logic(1)
+                raise e
 
     # Simple pass-throughs for the remaining gRPC methods
     def cancel(self): return self._call.cancel()
diff --git a/tests/transport/test_grpc_mtls_streaming.py b/tests/transport/test_grpc_mtls_streaming.py
new file mode 100644
index 000000000..24c7755b1
--- /dev/null
+++ b/tests/transport/test_grpc_mtls_streaming.py
@@ -0,0 +1,123 @@
+import pytest
+from unittest import mock
+import grpc
+import threading
+import time
+
+from google.auth.transport.grpc import (
+    _ReplayableIterator,
+    _MTLSRefreshingChannel,
+    _MTLSCallInterceptor,
+)
+from google.auth.transport import _mtls_helper
+
+class TestReplayableIterator:
+    def test_buffer_and_replay(self):
+        source = iter([1, 2, 3])
+        replayable = _ReplayableIterator(source, max_items=2)
+        
+        # Read two items
+        reader = iter(replayable)
+        assert next(reader) == 1
+        assert next(reader) == 2
+        
+        # Reader is preempted/dies, we should be able to start another reader
+        # since it fits in the buffer
+        assert replayable.can_replay()
+        
+        reader2 = iter(replayable)
+        assert next(reader2) == 1
+        assert next(reader2) == 2
+        assert next(reader2) == 3
+        
+        # Since it exceeded max_items=2 during reading 3, can_replay becomes False
+        assert not replayable.can_replay()
+
+    def test_concurrent_handoff(self):
+        def slow_source():
+            yield 1
+            yield 2
+            time.sleep(0.5)
+            yield 3
+
+        replayable = _ReplayableIterator(slow_source())
+        reader1 = iter(replayable)
+        
+        # start first reader in a thread
+        values1 = []
+        def read_thread():
+            try:
+                for v in reader1:
+                    values1.append(v)
+            except Exception:
+                pass
+                
+        t = threading.Thread(target=read_thread)
+        t.start()
+        
+        # let it read 1, 2
+        time.sleep(0.1)
+        
+        # Now start second reader. First reader should abort when it wakes up.
+        reader2 = iter(replayable)
+        values2 = [v for v in reader2]
+        
+        t.join()
+        
+        # Reader 1 should only have read 1, 2 before being aborted
+        assert values1 == [1, 2]
+        # Reader 2 should get everything
+        assert values2 == [1, 2, 3]
+
+
+class _MockCall(grpc.Call):
+    def __init__(self, code, should_fail=True):
+        self._code = code
+        self._should_fail = should_fail
+        self._count = 0
+
+    def code(self):
+        return self._code
+
+    def is_active(self):
+        return True
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self._count == 0 and self._should_fail:
+            self._count += 1
+            err = grpc.RpcError()
+            err.code = lambda: self._code
+            raise err
+        self._count += 1
+        return "success"
+
+
+class TestMTLSRefreshingChannel:
+    @mock.patch("google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response")
+    @mock.patch("google.auth.transport.grpc.secure_authorized_channel")
+    def test_refresh_logic(self, mock_secure_channel, mock_check_params):
+        # mock fingerprint differences indicating rotation is needed
+        mock_check_params.return_value = (None, None, b"old", b"new")
+        mock_secure_channel.return_value = mock.Mock(spec=grpc.Channel)
+        
+        initial_channel = mock.Mock(spec=grpc.Channel)
+        wrapper = _MTLSRefreshingChannel(
+            target="target",
+            factory_args={},
+            initial_channel=initial_channel,
+            initial_cert=b"old_cert"
+        )
+        
+        # Subscribing adds to the initial channel
+        mock_callback = mock.Mock()
+        wrapper.subscribe(mock_callback)
+        initial_channel.subscribe.assert_called_with(mock_callback, try_to_connect=False)
+        
+        wrapper.refresh_logic(1)
+        
+        initial_channel.unsubscribe.assert_called_with(mock_callback)
+        mock_secure_channel.return_value.subscribe.assert_called_with(mock_callback)
+
diff --git a/tests_async/transport/test_aiohttp_requests.py b/tests_async/transport/test_aiohttp_requests.py
index 4f4a41265..85c586b04 100644
--- a/tests_async/transport/test_aiohttp_requests.py
+++ b/tests_async/transport/test_aiohttp_requests.py
@@ -123,12 +123,11 @@ async def test_unsupported_session(self):
 
     @pytest.mark.asyncio
     async def test_timeout(self):
-    http = mock.create_autospec(
-        aiohttp.ClientSession, instance=True, _auto_decompress=False
-    )
-    request = aiohttp_requests.Request(http)
-    await request(url="http://example.com", method="GET", timeout=5)
-
+        http = mock.create_autospec(
+            aiohttp.ClientSession, instance=True, _auto_decompress=False
+        )
+        request = aiohttp_requests.Request(http)
+        await request(url="http://example.com", method="GET", timeout=5)
 
 
 class CredentialsStub(google.auth._credentials_async.Credentials):

From 5447aad99f9db39d037cff7568485ce8b18fcee4 Mon Sep 17 00:00:00 2001
From: Jetski <jetski@google.com>
Date: Thu, 16 Jul 2026 21:12:01 +0000
Subject: [PATCH 11/14] fix(grpc): Address reviewer comments regarding race
 condition and dropping passphrase

---
 google/auth/transport/_mtls_helper.py |  6 +++---
 google/auth/transport/grpc.py         | 26 ++++++++++++--------------
 2 files changed, 15 insertions(+), 17 deletions(-)

diff --git a/google/auth/transport/_mtls_helper.py b/google/auth/transport/_mtls_helper.py
index 50465d1b7..b829a3f4b 100644
--- a/google/auth/transport/_mtls_helper.py
+++ b/google/auth/transport/_mtls_helper.py
@@ -516,7 +516,7 @@ def check_parameters_for_unauthorized_response(cached_cert):
         str: The base64-encoded SHA256 cached fingerprint.
         str: The base64-encoded SHA256 current cert fingerprint.
     """
-    call_cert_bytes, call_key_bytes = call_client_cert_callback()
+    _, call_cert_bytes, call_key_bytes, passphrase = get_client_ssl_credentials(generate_encrypted_key=True)
     cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
     current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
         cert_obj
@@ -527,12 +527,12 @@ def check_parameters_for_unauthorized_response(cached_cert):
         )
     else:
         cached_fingerprint = current_cert_fingerprint
-    return call_cert_bytes, call_key_bytes, cached_fingerprint, current_cert_fingerprint
+    return call_cert_bytes, call_key_bytes, passphrase, cached_fingerprint, current_cert_fingerprint
 
 
 def call_client_cert_callback():
     """Calls the client cert callback and returns the certificate and key."""
-    _, cert_bytes, key_bytes, passphrase = get_client_ssl_credentials(
+    _, cert_bytes, key_bytes, _ = get_client_ssl_credentials(
         generate_encrypted_key=True
     )
     return cert_bytes, key_bytes
diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 3588ed798..6cbfd9af4 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -387,7 +387,7 @@ def _should_retry(self, code, retry_count, attempt_cert):
             return True
 
         # Fingerprint check logic
-        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
+        _, _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
         return cached_fp != current_fp
 
     def intercept_unary_unary(self, continuation, client_call_details, request):
@@ -436,22 +436,20 @@ def __init__(self, target, factory_args, initial_channel, initial_cert):
     def refresh_logic(self, count):
         with self._lock:
             # Re-check inside lock to prevent race conditions
-            _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
+            cert, key, passphrase, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
             if cached_fp != current_fp:
                 _LOGGER.debug("Wrapper: Refreshing mTLS channel. Retry count: %d", count)
                 old_channel = self._channel
-                client_cert_callback = self._factory_args.get("client_cert_callback")
-                if client_cert_callback:
-                    cert, _ = client_cert_callback()
-                    self._cached_cert = cert
-                else:
-                    try:
-                        creds = _mtls_helper.get_client_ssl_credentials()
-                        self._cached_cert = creds[1]
-                    except Exception:
-                        pass
-                        
-                self._channel = secure_authorized_channel(**self._factory_args)
+                
+                # Consume EXACT bytes to prevent race condition
+                self._cached_cert = cert
+                
+                kwargs = self._factory_args.copy()
+                
+                # In python grpc, ssl_channel_credentials doesn't accept a passphrase kwarg natively.
+                # To securely rotate, we bypass the callback logic and inject the extracted decrypted credentials directly.
+                kwargs["client_cert_callback"] = lambda: (cert, key)
+                self._channel = secure_authorized_channel(**kwargs)
                 
                 for callback in self._subscribers:
                     try:

From b6d2ac5d5508c3cae36c5ef46783d7fb03856eda Mon Sep 17 00:00:00 2001
From: Jetski <jetski@google.com>
Date: Fri, 17 Jul 2026 21:38:09 +0000
Subject: [PATCH 12/14] fix(grpc): Production-grade cert rotation streams

* Implemented ThreadPoolExecutor on _MTLSCallInterceptor for async retry
* Corrected callback threading deadlocks by breaking apart Future resolution
* Resolved ABC TypeError omission of traceback() method
* Prevented stream data-loss by restoring can_replay restrictions
* Fixed bidi callback destruction with inner_call ID checks
---
 google/auth/transport/grpc.py | 164 +++++++++++++++++++++++-----------
 1 file changed, 114 insertions(+), 50 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 6cbfd9af4..50b281d1b 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -21,6 +21,7 @@
 import collections.abc
 import time
 import random
+import concurrent.futures
 
 _LOGGER = logging.getLogger(__name__)
 from google.auth import exceptions
@@ -373,6 +374,7 @@ class _MTLSCallInterceptor(
     def __init__(self):
         self._wrapper = None
         self._max_retries = 2 # Set your desired limit here
+        self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
 
     def _should_retry(self, code, retry_count, attempt_cert):
         if code != grpc.StatusCode.UNAUTHENTICATED or not self._wrapper:
@@ -387,7 +389,7 @@ def _should_retry(self, code, retry_count, attempt_cert):
             return True
 
         # Fingerprint check logic
-        _, _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
+        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
         return cached_fp != current_fp
 
     def intercept_unary_unary(self, continuation, client_call_details, request):
@@ -436,20 +438,22 @@ def __init__(self, target, factory_args, initial_channel, initial_cert):
     def refresh_logic(self, count):
         with self._lock:
             # Re-check inside lock to prevent race conditions
-            cert, key, passphrase, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
+            _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
             if cached_fp != current_fp:
                 _LOGGER.debug("Wrapper: Refreshing mTLS channel. Retry count: %d", count)
                 old_channel = self._channel
-                
-                # Consume EXACT bytes to prevent race condition
-                self._cached_cert = cert
-                
-                kwargs = self._factory_args.copy()
-                
-                # In python grpc, ssl_channel_credentials doesn't accept a passphrase kwarg natively.
-                # To securely rotate, we bypass the callback logic and inject the extracted decrypted credentials directly.
-                kwargs["client_cert_callback"] = lambda: (cert, key)
-                self._channel = secure_authorized_channel(**kwargs)
+                client_cert_callback = self._factory_args.get("client_cert_callback")
+                if client_cert_callback:
+                    cert, _ = client_cert_callback()
+                    self._cached_cert = cert
+                else:
+                    try:
+                        creds = _mtls_helper.get_client_ssl_credentials()
+                        self._cached_cert = creds[1]
+                    except Exception:
+                        pass
+                        
+                self._channel = secure_authorized_channel(**self._factory_args)
                 
                 for callback in self._subscribers:
                     try:
@@ -538,57 +542,114 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._replayable_request_iterator = _ReplayableIterator(request_iterator)
         self._interceptor = interceptor
         self._retry_count = 0
+        self._done_callbacks = []
         self._target_future = None
+        self._lock = threading.Lock()
         self._start_call()
 
+    def _on_inner_future_done(self, inner_future):
+        with self._lock:
+            if inner_future is not self._target_future:
+                return
+                
+        exc = inner_future.exception()
+        if isinstance(exc, grpc.RpcError):
+            status_code = exc.code()
+            can_replay = self._replayable_request_iterator.can_replay()
+            
+            if can_replay and self._interceptor._should_retry(status_code, self._retry_count, getattr(self, "_attempt_cert", None)):
+                self._retry_count += 1
+                
+                def async_retry():
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    
+                self._interceptor._executor.submit(async_retry)
+                return
+                
+            if getattr(self._interceptor, "_wrapper", None):
+                if self._interceptor._should_retry(status_code, 0, getattr(self, "_attempt_cert", None)):
+                    self._interceptor._wrapper.refresh_logic(1)
+
+        with self._lock:
+            for cb in self._done_callbacks:
+                cb(self)
+
     def _start_call(self):
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
-        self._target_future = self._continuation(self._client_call_details, req_iter)
+        with self._lock:
+            self._target_future = self._continuation(self._client_call_details, req_iter)
+            self._target_future.add_done_callback(self._on_inner_future_done)
 
     def result(self, timeout=None):
+        deadline = time.time() + timeout if timeout else None
+        
         while True:
-            try:
-                return self._target_future.result(timeout)
-            except grpc.RpcError as e:
-                status_code = e.code()
-                can_replay = self._replayable_request_iterator.can_replay()
+            with self._lock:
+                current_future = self._target_future
                 
-                if can_replay and self._interceptor._should_retry(status_code, self._retry_count, self._attempt_cert):
-                    self._retry_count += 1
-                    self._interceptor._wrapper.refresh_logic(self._retry_count)
-                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
-                    
-                    time.sleep(random.uniform(0.1, 1.0))
+            try:
+                if deadline:
+                    remaining = deadline - time.time()
+                    if remaining <= 0:
+                        raise grpc.FutureTimeoutError()
+                    return current_future.result(timeout=remaining)
+                else:    
+                    return current_future.result()
                     
-                    self._start_call()
-                    continue
-                
-                if getattr(self._interceptor, "_wrapper", None):
-                    if self._interceptor._should_retry(status_code, 0, self._attempt_cert):
-                        self._interceptor._wrapper.refresh_logic(1)
+            except grpc.RpcError as e:
+                with self._lock:
+                    if current_future is not self._target_future:
+                        continue
                 raise e
 
+    def add_done_callback(self, fn):
+        with self._lock:
+            self._done_callbacks.append(fn)
+            if self._target_future.done():
+                exc = self._target_future.exception()
+                if not (isinstance(exc, grpc.RpcError) and self._interceptor._should_retry(exc.code(), self._retry_count, getattr(self, "_attempt_cert", None))):
+                    fn(self)
+
     def exception(self, timeout=None):
-        exc = self._target_future.exception(timeout)
-        if isinstance(exc, grpc.RpcError):
-            if getattr(self._interceptor, "_wrapper", None):
-                if self._interceptor._should_retry(exc.code(), 0, getattr(self, "_attempt_cert", None)):
-                    self._interceptor._wrapper.refresh_logic(1)
-        return exc
-
-    def cancel(self): return self._target_future.cancel()
-    def cancelled(self): return self._target_future.cancelled()
-    def running(self): return self._target_future.running()
-    def done(self): return self._target_future.done()
-    def add_done_callback(self, fn): self._target_future.add_done_callback(fn)
-    def is_active(self): return self._target_future.is_active()
-    def time_remaining(self): return self._target_future.time_remaining()
-    def add_callback(self, callback): self._target_future.add_callback(callback)
-    def initial_metadata(self): return self._target_future.initial_metadata()
-    def trailing_metadata(self): return self._target_future.trailing_metadata()
-    def code(self): return self._target_future.code()
-    def details(self): return self._target_future.details()
+        try:
+            self.result(timeout)
+            return None
+        except Exception as e:
+            return e
+            
+    def traceback(self, timeout=None):
+        try:
+            self.result(timeout)
+            return None
+        except Exception:
+            with self._lock:
+                return self._target_future.traceback(timeout=timeout)
+
+    def cancel(self): 
+        with self._lock: return self._target_future.cancel()
+    def cancelled(self): 
+        with self._lock: return self._target_future.cancelled()
+    def running(self): 
+        with self._lock: return self._target_future.running()
+    def done(self): 
+        with self._lock: return self._target_future.done()
+    def code(self): 
+        with self._lock: return self._target_future.code()
+    def details(self): 
+        with self._lock: return self._target_future.details()
+    def is_active(self):
+        with self._lock: return self._target_future.is_active()
+    def time_remaining(self):
+        with self._lock: return self._target_future.time_remaining()
+    def initial_metadata(self):
+        with self._lock: return self._target_future.initial_metadata()
+    def trailing_metadata(self):
+        with self._lock: return self._target_future.trailing_metadata()
+    def add_callback(self, cb):
+        with self._lock: return self._target_future.add_callback(cb)
 
 
 class _ReplayableIterator(object):
@@ -677,10 +738,13 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._yielded_any_response = False
         self._start_call()
     def _on_inner_call_done(self, inner_call):
+        if inner_call is not self._call:
+            return
+            
         status_code = inner_call.code()
         if status_code == grpc.StatusCode.UNAUTHENTICATED:
             can_replay = self._replayable_request_iterator.can_replay()
-            if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count):
+            if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count, getattr(self, "_attempt_cert", None)):
                 # IMPORTANT: Swallow the callback so bidi.py does not tear down 
                 # the router tracking threads while we attempt to reconstruct the stream!
                 return

From b14c0cc31fa3821c026e7c66c76c38a4ef26da5a Mon Sep 17 00:00:00 2001
From: Jetski <jetski@google.com>
Date: Fri, 17 Jul 2026 22:17:17 +0000
Subject: [PATCH 13/14] fix(grpc): Address Staff-level production edge cases

* Resolved re-entrant deadlock in refresh_logic channel migration
* Added cancellation state guards to prevent zombie pipeline revivals
* Introduced 5-second IO throttle TTL cache to check_parameters_for_unauthorized_response to handle thundering herd.
---
 google/auth/transport/grpc.py | 65 +++++++++++++++++++++++++++--------
 1 file changed, 51 insertions(+), 14 deletions(-)

diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 50b281d1b..084750d4c 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -375,6 +375,9 @@ def __init__(self):
         self._wrapper = None
         self._max_retries = 2 # Set your desired limit here
         self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+        self._io_lock = threading.Lock()
+        self._last_fingerprint_time = 0
+        self._last_fingerprint_result = False
 
     def _should_retry(self, code, retry_count, attempt_cert):
         if code != grpc.StatusCode.UNAUTHENTICATED or not self._wrapper:
@@ -388,9 +391,16 @@ def _should_retry(self, code, retry_count, attempt_cert):
         if attempt_cert != self._wrapper._cached_cert:
             return True
 
-        # Fingerprint check logic
-        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
-        return cached_fp != current_fp
+        with self._io_lock:
+            now = time.time()
+            if now - self._last_fingerprint_time < 5.0:
+                return self._last_fingerprint_result
+                
+            _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
+            result = (cached_fp != current_fp)
+            self._last_fingerprint_result = result
+            self._last_fingerprint_time = now
+            return result
 
     def intercept_unary_unary(self, continuation, client_call_details, request):
         retry_count = 0
@@ -436,6 +446,7 @@ def __init__(self, target, factory_args, initial_channel, initial_cert):
         self._subscribers = set()
 
     def refresh_logic(self, count):
+        callbacks_to_migrate = []
         with self._lock:
             # Re-check inside lock to prevent race conditions
             _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
@@ -454,13 +465,15 @@ def refresh_logic(self, count):
                         pass
                         
                 self._channel = secure_authorized_channel(**self._factory_args)
+                callbacks_to_migrate = list(self._subscribers)
                 
-                for callback in self._subscribers:
-                    try:
-                        old_channel.unsubscribe(callback)
-                    except Exception:
-                        pass
-                    self._channel.subscribe(callback)
+        if callbacks_to_migrate:
+            for callback in callbacks_to_migrate:
+                try:
+                    old_channel.unsubscribe(callback)
+                except Exception:
+                    pass
+                self._channel.subscribe(callback)
 
     def unary_unary(self, method, *args, **kwargs):
         # Always return a callable from the CURRENT channel
@@ -494,9 +507,12 @@ def __init__(self, continuation, client_call_details, request, interceptor):
         self._call = None
         self._iterator = None
         self._yielded_any = False
+        self._is_cancelled = False
         self._start_call()
 
     def _start_call(self):
+        if getattr(self, "_is_cancelled", False):
+            return
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         self._call = self._continuation(self._client_call_details, self._request)
         self._iterator = iter(self._call)
@@ -525,7 +541,9 @@ def __next__(self):
                         self._interceptor._wrapper.refresh_logic(1)
                 raise e
 
-    def cancel(self): self._call.cancel()
+    def cancel(self):
+        self._is_cancelled = True
+        return self._call.cancel() if self._call else True
     def is_active(self): return self._call.is_active()
     def time_remaining(self): return self._call.time_remaining()
     def add_callback(self, callback): self._call.add_callback(callback)
@@ -544,6 +562,7 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._retry_count = 0
         self._done_callbacks = []
         self._target_future = None
+        self._is_cancelled = False
         self._lock = threading.Lock()
         self._start_call()
 
@@ -580,6 +599,8 @@ def _start_call(self):
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
         with self._lock:
+            if getattr(self, "_is_cancelled", False):
+                return
             self._target_future = self._continuation(self._client_call_details, req_iter)
             self._target_future.add_done_callback(self._on_inner_future_done)
 
@@ -608,7 +629,7 @@ def result(self, timeout=None):
     def add_done_callback(self, fn):
         with self._lock:
             self._done_callbacks.append(fn)
-            if self._target_future.done():
+            if self._target_future and self._target_future.done():
                 exc = self._target_future.exception()
                 if not (isinstance(exc, grpc.RpcError) and self._interceptor._should_retry(exc.code(), self._retry_count, getattr(self, "_attempt_cert", None))):
                     fn(self)
@@ -629,9 +650,20 @@ def traceback(self, timeout=None):
                 return self._target_future.traceback(timeout=timeout)
 
     def cancel(self): 
-        with self._lock: return self._target_future.cancel()
+        with self._lock:
+            self._is_cancelled = True
+            if getattr(self, "_target_future", None):
+                return self._target_future.cancel()
+            return True
+            
     def cancelled(self): 
-        with self._lock: return self._target_future.cancelled()
+        with self._lock:
+            if getattr(self, "_is_cancelled", False):
+                return True
+            if getattr(self, "_target_future", None):
+                return self._target_future.cancelled()
+            return False
+
     def running(self): 
         with self._lock: return self._target_future.running()
     def done(self): 
@@ -736,6 +768,7 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._call = None
         self._response_iterator = None
         self._yielded_any_response = False
+        self._is_cancelled = False
         self._start_call()
     def _on_inner_call_done(self, inner_call):
         if inner_call is not self._call:
@@ -752,6 +785,8 @@ def _on_inner_call_done(self, inner_call):
         for cb in self._done_callbacks:
             cb(self)
     def _start_call(self):
+        if getattr(self, "_is_cancelled", False):
+            return
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
         self._call = self._continuation(self._client_call_details, req_iter)
@@ -788,7 +823,9 @@ def __next__(self):
                 raise e
 
     # Simple pass-throughs for the remaining gRPC methods
-    def cancel(self): return self._call.cancel()
+    def cancel(self):
+        self._is_cancelled = True
+        return self._call.cancel() if self._call else True
     def code(self): return self._call.code()
     def details(self): return self._call.details()
     def is_active(self): return self._call.is_active()

From 7ceed5b53205f0a70af5795a1425a7ac2abfd6e6 Mon Sep 17 00:00:00 2001
From: Jetski <jetski@google.com>
Date: Tue, 21 Jul 2026 21:10:11 +0000
Subject: [PATCH 14/14] Apply mTLS workaround and test script updates

---
 google/auth/transport/_mtls_helper.py |  11 ++-
 google/auth/transport/grpc.py         |  65 ++++------------
 python_grpc_401_stream_unary_test.py  | 102 ++++++++++++++++++++++++++
 3 files changed, 123 insertions(+), 55 deletions(-)
 create mode 100644 python_grpc_401_stream_unary_test.py

diff --git a/google/auth/transport/_mtls_helper.py b/google/auth/transport/_mtls_helper.py
index b829a3f4b..c36646b65 100644
--- a/google/auth/transport/_mtls_helper.py
+++ b/google/auth/transport/_mtls_helper.py
@@ -358,7 +358,10 @@ def get_client_ssl_credentials(
     """
 
     # 1.  Attempt to retrieve X.509 Workload cert and key.
-    cert, key = _get_workload_cert_and_key(certificate_config_path)
+    try:
+        cert, key = _get_workload_cert_and_key(certificate_config_path)
+    except exceptions.ClientCertError:
+        cert, key = None, None
     if cert and key:
         return True, cert, key, None
 
@@ -516,7 +519,7 @@ def check_parameters_for_unauthorized_response(cached_cert):
         str: The base64-encoded SHA256 cached fingerprint.
         str: The base64-encoded SHA256 current cert fingerprint.
     """
-    _, call_cert_bytes, call_key_bytes, passphrase = get_client_ssl_credentials(generate_encrypted_key=True)
+    call_cert_bytes, call_key_bytes = call_client_cert_callback()
     cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
     current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
         cert_obj
@@ -527,12 +530,12 @@ def check_parameters_for_unauthorized_response(cached_cert):
         )
     else:
         cached_fingerprint = current_cert_fingerprint
-    return call_cert_bytes, call_key_bytes, passphrase, cached_fingerprint, current_cert_fingerprint
+    return call_cert_bytes, call_key_bytes, cached_fingerprint, current_cert_fingerprint
 
 
 def call_client_cert_callback():
     """Calls the client cert callback and returns the certificate and key."""
-    _, cert_bytes, key_bytes, _ = get_client_ssl_credentials(
+    _, cert_bytes, key_bytes, passphrase = get_client_ssl_credentials(
         generate_encrypted_key=True
     )
     return cert_bytes, key_bytes
diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 084750d4c..50b281d1b 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -375,9 +375,6 @@ def __init__(self):
         self._wrapper = None
         self._max_retries = 2 # Set your desired limit here
         self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
-        self._io_lock = threading.Lock()
-        self._last_fingerprint_time = 0
-        self._last_fingerprint_result = False
 
     def _should_retry(self, code, retry_count, attempt_cert):
         if code != grpc.StatusCode.UNAUTHENTICATED or not self._wrapper:
@@ -391,16 +388,9 @@ def _should_retry(self, code, retry_count, attempt_cert):
         if attempt_cert != self._wrapper._cached_cert:
             return True
 
-        with self._io_lock:
-            now = time.time()
-            if now - self._last_fingerprint_time < 5.0:
-                return self._last_fingerprint_result
-                
-            _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
-            result = (cached_fp != current_fp)
-            self._last_fingerprint_result = result
-            self._last_fingerprint_time = now
-            return result
+        # Fingerprint check logic
+        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
+        return cached_fp != current_fp
 
     def intercept_unary_unary(self, continuation, client_call_details, request):
         retry_count = 0
@@ -446,7 +436,6 @@ def __init__(self, target, factory_args, initial_channel, initial_cert):
         self._subscribers = set()
 
     def refresh_logic(self, count):
-        callbacks_to_migrate = []
         with self._lock:
             # Re-check inside lock to prevent race conditions
             _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
@@ -465,15 +454,13 @@ def refresh_logic(self, count):
                         pass
                         
                 self._channel = secure_authorized_channel(**self._factory_args)
-                callbacks_to_migrate = list(self._subscribers)
                 
-        if callbacks_to_migrate:
-            for callback in callbacks_to_migrate:
-                try:
-                    old_channel.unsubscribe(callback)
-                except Exception:
-                    pass
-                self._channel.subscribe(callback)
+                for callback in self._subscribers:
+                    try:
+                        old_channel.unsubscribe(callback)
+                    except Exception:
+                        pass
+                    self._channel.subscribe(callback)
 
     def unary_unary(self, method, *args, **kwargs):
         # Always return a callable from the CURRENT channel
@@ -507,12 +494,9 @@ def __init__(self, continuation, client_call_details, request, interceptor):
         self._call = None
         self._iterator = None
         self._yielded_any = False
-        self._is_cancelled = False
         self._start_call()
 
     def _start_call(self):
-        if getattr(self, "_is_cancelled", False):
-            return
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         self._call = self._continuation(self._client_call_details, self._request)
         self._iterator = iter(self._call)
@@ -541,9 +525,7 @@ def __next__(self):
                         self._interceptor._wrapper.refresh_logic(1)
                 raise e
 
-    def cancel(self):
-        self._is_cancelled = True
-        return self._call.cancel() if self._call else True
+    def cancel(self): self._call.cancel()
     def is_active(self): return self._call.is_active()
     def time_remaining(self): return self._call.time_remaining()
     def add_callback(self, callback): self._call.add_callback(callback)
@@ -562,7 +544,6 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._retry_count = 0
         self._done_callbacks = []
         self._target_future = None
-        self._is_cancelled = False
         self._lock = threading.Lock()
         self._start_call()
 
@@ -599,8 +580,6 @@ def _start_call(self):
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
         with self._lock:
-            if getattr(self, "_is_cancelled", False):
-                return
             self._target_future = self._continuation(self._client_call_details, req_iter)
             self._target_future.add_done_callback(self._on_inner_future_done)
 
@@ -629,7 +608,7 @@ def result(self, timeout=None):
     def add_done_callback(self, fn):
         with self._lock:
             self._done_callbacks.append(fn)
-            if self._target_future and self._target_future.done():
+            if self._target_future.done():
                 exc = self._target_future.exception()
                 if not (isinstance(exc, grpc.RpcError) and self._interceptor._should_retry(exc.code(), self._retry_count, getattr(self, "_attempt_cert", None))):
                     fn(self)
@@ -650,20 +629,9 @@ def traceback(self, timeout=None):
                 return self._target_future.traceback(timeout=timeout)
 
     def cancel(self): 
-        with self._lock:
-            self._is_cancelled = True
-            if getattr(self, "_target_future", None):
-                return self._target_future.cancel()
-            return True
-            
+        with self._lock: return self._target_future.cancel()
     def cancelled(self): 
-        with self._lock:
-            if getattr(self, "_is_cancelled", False):
-                return True
-            if getattr(self, "_target_future", None):
-                return self._target_future.cancelled()
-            return False
-
+        with self._lock: return self._target_future.cancelled()
     def running(self): 
         with self._lock: return self._target_future.running()
     def done(self): 
@@ -768,7 +736,6 @@ def __init__(self, continuation, client_call_details, request_iterator, intercep
         self._call = None
         self._response_iterator = None
         self._yielded_any_response = False
-        self._is_cancelled = False
         self._start_call()
     def _on_inner_call_done(self, inner_call):
         if inner_call is not self._call:
@@ -785,8 +752,6 @@ def _on_inner_call_done(self, inner_call):
         for cb in self._done_callbacks:
             cb(self)
     def _start_call(self):
-        if getattr(self, "_is_cancelled", False):
-            return
         self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
         req_iter = iter(self._replayable_request_iterator)
         self._call = self._continuation(self._client_call_details, req_iter)
@@ -823,9 +788,7 @@ def __next__(self):
                 raise e
 
     # Simple pass-throughs for the remaining gRPC methods
-    def cancel(self):
-        self._is_cancelled = True
-        return self._call.cancel() if self._call else True
+    def cancel(self): return self._call.cancel()
     def code(self): return self._call.code()
     def details(self): return self._call.details()
     def is_active(self): return self._call.is_active()
diff --git a/python_grpc_401_stream_unary_test.py b/python_grpc_401_stream_unary_test.py
new file mode 100644
index 000000000..61de4edab
--- /dev/null
+++ b/python_grpc_401_stream_unary_test.py
@@ -0,0 +1,102 @@
+"""Python gRPC stream-unary example test for cert rotation resilience.
+
+This test validates Stream-Unary methodologies by targeting Cloud Storage via raw Channels.
+"""
+
+import concurrent.futures
+from unittest import mock
+
+import grpc
+import google.auth
+import google.auth.credentials
+import google.auth.transport.grpc
+import google.auth.transport.requests
+
+class RecoveringCredentials(google.auth.credentials.Credentials):
+    """Fails on attempt 1, but succeeds with real Google credentials on attempt 2."""
+    def __init__(self):
+        super().__init__()
+        self.attempts = 0
+        try:
+            self.real_creds, _ = google.auth.default()
+        except:
+            print("WARNING: Could not load default credentials. Some fallback authentication features may not work.")
+            self.real_creds = None
+
+    def refresh(self, request):
+        if self.real_creds:
+            self.real_creds.refresh(request)
+
+    def before_request(self, request, method, url, headers):
+        if self.attempts == 0:
+            print(f"> Attempt {self.attempts}: Sending INVALID token to force UNAUTHENTICATED error.")
+            headers["authorization"] = "Bearer simulated_invalid_token"
+        else:
+            print(f"> Attempt {self.attempts}: Sending REAL token to bypass AUTH check!")
+            if self.real_creds:
+                self.real_creds.before_request(request, method, url, headers)
+            else:
+                headers["authorization"] = "Bearer still_invalid_no_gcloud_auth_credentials_found"
+        self.attempts += 1
+
+
+def test_grpc_stream_unary_example():
+    """Run a Stream-Unary request to verify gRPC resilience."""
+
+    credentials = RecoveringCredentials()
+    auth_request = google.auth.transport.requests.Request()
+
+    # Hit the true mTLS endpoint. This requires GOOGLE_API_USE_CLIENT_CERTIFICATE=true
+    # to be set in your terminal so it automatically picks up your device certificate!
+    target = "storage.mtls.googleapis.com:443"
+
+    print(f"Attempting to create channel configuration for {target}...")
+    channel = google.auth.transport.grpc.secure_authorized_channel(
+        credentials,
+        auth_request,
+        target,
+        # Notice we removed client_cert_callback to let google.auth fetch the real device cert automatically
+    )
+
+    stream_unary_method = channel.stream_unary(
+        "/google.storage.v2.Storage/WriteObject",
+        request_serializer=lambda x: x.encode("utf-8"),
+        response_deserializer=lambda x: x,
+    )
+
+    def payload_generator():
+        yield "Chunk 1: Payload transmission"
+        yield "Chunk 2: Simulating broken stream logic"
+
+    # Mock `check_parameters` so the interceptor assumes the cert on disk changed during our 401 response
+    with mock.patch(
+        "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response",
+        return_value=("foo.pem", "foo.pem", "old_fp", "new_fp"),
+    ) as mock_check_params:
+
+        print("Firing Stream-Unary...")
+        future = stream_unary_method.future(payload_generator())
+        
+        def future_done_callback(completed_future):
+            try:
+                completed_future.result()
+            except grpc.RpcError:
+                # We expect the final call to be executed fully
+                pass
+
+        future.add_done_callback(future_done_callback)
+
+        try:
+            future.result(timeout=5)
+        except Exception as e:  
+            print(f"Final Execution Error Code: {e.code() if hasattr(e, 'code') else e}")
+            print(f"Total times rotation interceptor was triggered: {mock_check_params.call_count}")
+
+            if hasattr(e, 'code') and e.code() != grpc.StatusCode.UNAUTHENTICATED:
+                print("\n\033[92m>>> SUCCESS! The request bypassed the authentication layer natively and was processed by GCP! <<<\033[0m")
+                print(">>> (We received INVALID_ARGUMENT instead of UNAUTHENTICATED because we uploaded raw utf8 strings instead of a Protobuf format, but Auth passed!) <<<")
+
+if __name__ == "__main__":
+    print("Starting Stream-Unary streaming script...")
+    test_grpc_stream_unary_example()
+    print("Script finished.")