feat: enable self signed jwt

google/cloud/kms_v1/services/key_management_service/transports/base.py

@@ -24,6 +24,7 @@
 from google.api_core import gapic_v1  # type: ignore
 from google.api_core import retry as retries  # type: ignore
 from google.auth import credentials as ga_credentials  # type: ignore
+from google.oauth2 import service_account  # type: ignore
 
 from google.cloud.kms_v1.types import resources
 from google.cloud.kms_v1.types import service
@@ -68,6 +69,7 @@ def __init__(
         scopes: Optional[Sequence[str]] = None,
         quota_project_id: Optional[str] = None,
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
         **kwargs,
     ) -> None:
         """Instantiate the transport.
@@ -91,6 +93,8 @@ def __init__(
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT
+                should be used for service account credentials.
         """
         # Save the hostname. Default to port 443 (HTTPS) if none is specified.
         if ":" not in host:
@@ -119,6 +123,14 @@ def __init__(
                 **scopes_kwargs, quota_project_id=quota_project_id
             )
 
+        # If the credentials is service account credentials, then always try to use self signed JWT.
+        if (
+            always_use_jwt_access
+            and isinstance(credentials, service_account.Credentials)
+            and hasattr(service_account.Credentials, "with_always_use_jwt_access")
+        ):
+            credentials = credentials.with_always_use_jwt_access(True)
+
         # Save the credentials.
         self._credentials = credentials
 

google/cloud/kms_v1/services/key_management_service/transports/grpc.py

@@ -163,6 +163,7 @@ def __init__(
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
+            always_use_jwt_access=True,
         )
 
         if not self._grpc_channel:

google/cloud/kms_v1/services/key_management_service/transports/grpc_asyncio.py

@@ -209,6 +209,7 @@ def __init__(
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
+            always_use_jwt_access=True,
         )
 
         if not self._grpc_channel:

tests/unit/gapic/kms_v1/test_key_management_service.py

@@ -126,6 +126,17 @@ def test__get_default_mtls_endpoint():
     )
 
 
+@pytest.mark.parametrize(
+    "client_class", [KeyManagementServiceClient, KeyManagementServiceAsyncClient,]
+)
+def test_key_management_service_client_service_account_always_use_jwt(client_class):
+    creds = service_account.Credentials(None, None, None)
+    if hasattr(service_account.Credentials, "with_always_use_jwt_access"):
+        assert not creds._always_use_jwt_access
+        client = client_class(credentials=creds)
+        assert client.transport._credentials._always_use_jwt_access
+
+
 @pytest.mark.parametrize(
     "client_class", [KeyManagementServiceClient, KeyManagementServiceAsyncClient,]
 )