Update grpc.py for cert rotation handling

Author: agrawalradhika-cell

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

@@ -11,17 +11,17 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-        
+
 """Authorization support for gRPC."""
-            
+    
 from __future__ import absolute_import
 
 import logging
 import threading
 from google.auth import exceptions
 from google.auth.transport import _mtls_helper
 from google.oauth2 import service_account
-    
+
 try:
     import grpc  # type: ignore
 except ImportError as caught_exc:  # pragma: NO COVER
@@ -30,17 +30,16 @@
     ) from caught_exc
 
 _LOGGER = logging.getLogger(__name__)
-    
 
 class AuthMetadataPlugin(grpc.AuthMetadataPlugin):
     """A `gRPC AuthMetadataPlugin`_ that inserts the credentials into each
     request.
-            
+
     .. _gRPC AuthMetadataPlugin:
         http://www.grpc.io/grpc/python/grpc.html#grpc.AuthMetadataPlugin
-    
+
     Args:
-        credentials (google.auth.credentials.Credentials): The credentials to  
+        credentials (google.auth.credentials.Credentials): The credentials to
             add to requests.
         request (google.auth.transport.Request): A HTTP transport request
             object used to refresh credentials as needed.
@@ -57,7 +56,7 @@ def __init__(self, credentials, request, default_host=None):
         self._credentials = credentials
         self._request = request
         self._default_host = default_host
-    
+
     def _get_authorization_headers(self, context):
         """Gets the authorization headers for a request.
     
@@ -66,7 +65,7 @@ def _get_authorization_headers(self, context):
                 to add to the request.
         """
         headers = {}
-            
+ 
         # https://google.aip.dev/auth/4111
         # Attempt to use self-signed JWTs when a service account is used.
         # A default host must be explicitly provided since it cannot always
@@ -75,24 +74,24 @@ def _get_authorization_headers(self, context):
             self._credentials._create_self_signed_jwt(
                 "https://{}/".format(self._default_host) if self._default_host else None
             )
-            
+
         self._credentials.before_request(
             self._request, context.method_name, context.service_url, headers
         )
-    
+
         return list(headers.items())
-    
+
     def __call__(self, context, callback):
         """Passes authorization metadata into the given callback.
-    
+
         Args:
             context (grpc.AuthMetadataContext): The RPC context.
             callback (grpc.AuthMetadataPluginCallback): The callback that will
                 be invoked to pass in the authorization metadata.
         """
         callback(self._get_authorization_headers(context), None)
-    
-    
+
+
 def secure_authorized_channel(
     credentials,
     request,
@@ -102,20 +101,20 @@ def secure_authorized_channel(
     **kwargs
 ):
     """Creates a secure authorized gRPC channel.
-            
+
     This creates a channel with SSL and :class:`AuthMetadataPlugin`. This
     channel can be used to create a stub that can make authorized requests.
     Users can configure client certificate or rely on device certificates to
     establish a mutual TLS channel, if the `GOOGLE_API_USE_CLIENT_CERTIFICATE`
     variable is explicitly set to `true`.
             
     Example::
-            
+  
         import google.auth
         import google.auth.transport.grpc
         import google.auth.transport.requests
         from google.cloud.speech.v1 import cloud_speech_pb2
-            
+   
         # Get credentials.
         credentials, _ = google.auth.default()
             
@@ -126,34 +125,34 @@ def secure_authorized_channel(
         channel = google.auth.transport.grpc.secure_authorized_channel(
             credentials, regular_endpoint, request,
             ssl_credentials=grpc.ssl_channel_credentials())
-        
+
         # Use the channel to create a stub.
         cloud_speech.create_Speech_stub(channel)
-    
+
     Usage:
 
     There are actually a couple of options to create a channel, depending on if
     you want to create a regular or mutual TLS channel.
 
     First let's list the endpoints (regular vs mutual TLS) to choose from::
-        
+
         regular_endpoint = 'speech.googleapis.com:443'
         mtls_endpoint = 'speech.mtls.googleapis.com:443'
-         
+
     Option 1: create a regular (non-mutual) TLS channel by explicitly setting
     the ssl_credentials::
-    
+
         regular_ssl_credentials = grpc.ssl_channel_credentials()
-        
+
         channel = google.auth.transport.grpc.secure_authorized_channel(
             credentials, request, regular_endpoint,
             ssl_credentials=regular_ssl_credentials)
-            
+
     Option 2: create a mutual TLS channel by calling a callback which returns
     the client side certificate and the key (Note that
     `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable must be explicitly
     set to `true`)::
-            
+
         def my_client_cert_callback():
             code_to_load_client_cert_and_key()
             if loaded:
@@ -164,16 +163,16 @@ def my_client_cert_callback():
             channel = google.auth.transport.grpc.secure_authorized_channel(
                 credentials, request, mtls_endpoint,
                 client_cert_callback=my_client_cert_callback)
-        except MyClientCertFailureException: 
+        except MyClientCertFailureException:
             # handle the exception
-    
+
     Option 3: use application default SSL credentials. It searches and uses
     the command in a context aware metadata file, which is available on devices
     with endpoint verification support (Note that
     `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable must be explicitly
     set to `true`).
     See https://cloud.google.com/endpoint-verification/docs/overview::
-            
+
         try:
             default_ssl_credentials = SslCredentials()
         except:
@@ -194,21 +193,21 @@ def my_client_cert_callback():
     environment variable is not `true`, a regular TLS channel is created;
     otherwise, a mutual TLS channel is created, however, the call should be
     wrapped in a try/except block in case of malformed context aware metadata.
-    
+
     The following code uses regular_endpoint, it works the same no matter the
     created channle is regular or mutual TLS. Regular endpoint ignores client
     certificate and key::
-       
+
         channel = google.auth.transport.grpc.secure_authorized_channel(
             credentials, request, regular_endpoint)
-        
+
     The following code uses mtls_endpoint, if the created channle is regular,
     and API mtls_endpoint is confgured to require client SSL credentials, API
     calls using this channel will be rejected::
-        
+
         channel = google.auth.transport.grpc.secure_authorized_channel(
             credentials, request, mtls_endpoint)
-                
+      
     Args:
         credentials (google.auth.credentials.Credentials): The credentials to
             add to requests.
@@ -232,10 +231,10 @@ def my_client_cert_callback():
             This argument does nothing unless `GOOGLE_API_USE_CLIENT_CERTIFICATE`
             environment variable is explicitly set to `true`.
         kwargs: Additional arguments to pass to :func:`grpc.secure_channel`.
-                    
+     
     Returns:
         grpc.Channel: The created gRPC channel.
-            
+
     Raises:
         google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
             creation failed for any reason.
@@ -245,13 +244,13 @@ def my_client_cert_callback():
 
     # Create a set of grpc.CallCredentials using the metadata plugin.
     google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
-        
+   
     if ssl_credentials and client_cert_callback:
         raise exceptions.MalformedError(
             "Received both ssl_credentials and client_cert_callback; "
             "these are mutually exclusive."
         )
-        
+
     # If SSL credentials are not explicitly set, try client_cert_callback and ADC.
     cached_cert = None
     if not ssl_credentials:
@@ -270,7 +269,7 @@ def my_client_cert_callback():
             cached_cert = adc_ssl_credentils._cached_cert
         else:
             ssl_credentials = grpc.ssl_channel_credentials()
-            
+
     # Combine the ssl credentials and the authorization credentials.
     composite_credentials = grpc.composite_channel_credentials(
         ssl_credentials, google_auth_credentials
@@ -292,19 +291,19 @@ def my_client_cert_callback():
         interceptor = _MTLSCallInterceptor(cached_cert)
 
         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.
-        
+
     The behavior is controlled by `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment
     variable whose default value is `false`. Client certificate will not be used
     unless the environment variable is explicitly set to `true`. See
     https://google.aip.dev/auth/4114
-                
+
     If the environment variable is `true`, then for devices with endpoint verification
     support, a device certificate will be automatically loaded and mutual TLS will
     be established.
@@ -433,19 +432,3 @@ def stream_stream(self, method, *args, **kwargs): return self._channel.stream_st
     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()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-