Add test using default credentials, and allow credential_path to be None.

PiperOrigin-RevId: 520027764

2 files changed, 29 insertions, 4 deletions

diff --git a/python/tink/integration/gcpkms/_gcp_kms_client.py b/python/tink/integration/gcpkms/_gcp_kms_client.py
index 554c0c6d8..c463bab74 100644
--- a/python/tink/integration/gcpkms/_gcp_kms_client.py
+++ b/python/tink/integration/gcpkms/_gcp_kms_client.py
@@ -26,7 +26,9 @@ GCP_KEYURI_PREFIX = 'gcp-kms://'
 class GcpKmsClient(_kms_aead_key_manager.KmsClient):
   """Basic GCP client for AEAD."""
 
-  def __init__(self, key_uri: Optional[str], credentials_path: str):
+  def __init__(
+      self, key_uri: Optional[str], credentials_path: Optional[str]
+  ) -> None:
     """Creates a new GcpKmsClient that is bound to the key specified in 'key_uri'.
 
     Uses the specified credentials when communicating with the KMS.
@@ -35,7 +37,7 @@ class GcpKmsClient(_kms_aead_key_manager.KmsClient):
       key_uri: The URI of the key the client should be bound to. If it is None
           or empty, then the client is not bound to any particular key.
       credentials_path: Path to the file with the access credentials. If it is
-          empty, then default credentials will be used.
+          None or empty, then default credentials will be used.
 
     Raises:
       ValueError: If the path or filename of the credentials is invalid.
@@ -48,7 +50,8 @@ class GcpKmsClient(_kms_aead_key_manager.KmsClient):
       self._key_uri = key_uri
     else:
       raise core.TinkError('Invalid key_uri.')
-
+    if not credentials_path:
+      credentials_path = ''
     # Use the C++ GCP KMS client
     self.cc_client = tink_bindings.GcpKmsClient(self._key_uri, credentials_path)
 
@@ -77,7 +80,9 @@ class GcpKmsClient(_kms_aead_key_manager.KmsClient):
     return aead.AeadCcToPyWrapper(self.cc_client.get_aead(key_uri))
 
   @classmethod
-  def register_client(cls, key_uri, credentials_path) -> None:
+  def register_client(
+      cls, key_uri: Optional[str], credentials_path: Optional[str]
+  ) -> None:
     """Registers the KMS client internally."""
     _kms_aead_key_manager.register_kms_client(  # pylint: disable=protected-access
         GcpKmsClient(key_uri, credentials_path)
diff --git a/python/tink/integration/gcpkms/_gcp_kms_client_integration_test.py b/python/tink/integration/gcpkms/_gcp_kms_client_integration_test.py
index a8c4185f5..12e67c73c 100644
--- a/python/tink/integration/gcpkms/_gcp_kms_client_integration_test.py
+++ b/python/tink/integration/gcpkms/_gcp_kms_client_integration_test.py
@@ -29,6 +29,8 @@ KEY_URI = 'gcp-kms://projects/tink-test-infrastructure/locations/global/keyRings
 LOCAL_KEY_URI = 'gcp-kms://projects/tink-test-infrastructure/locations/europe-west1/keyRings/unit-and-integration-test/cryptoKeys/aead-key'
 BAD_KEY_URI = 'aws-kms://arn:aws:kms:us-east-2:235739564943:key/3ee50705-5a82-4f5b-9753-05c4f473922f'
 
+KEY2_URI = 'gcp-kms://projects/tink-test-infrastructure/locations/global/keyRings/unit-and-integration-testing/cryptoKeys/aead2-key'
+
 if 'TEST_SRCDIR' in os.environ:
   # Set root certificates for gRPC in Bazel Test which are needed on MacOS
   os.environ['GRPC_DEFAULT_SSL_ROOTS_FILE_PATH'] = os.path.join(
@@ -91,6 +93,9 @@ class GcpKmsAeadTest(absltest.TestCase):
         gcp_aead.decrypt(corrupted_ciphertext, b'')
 
   def test_registration_client_bound_to_uri_works(self):
+    # Make sure default credentials are not set.
+    os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = ''
+
     # Register GCP KMS Client bound to KEY_URI.
     gcpkms.GcpKmsClient.register_client(KEY_URI, CREDENTIAL_PATH)
 
@@ -112,6 +117,21 @@ class GcpKmsAeadTest(absltest.TestCase):
       gcp_aead = handle2.primitive(aead.Aead)
       gcp_aead.encrypt(b'plaintext', b'associated_data')
 
+  def test_registration_client_with_default_credentials_works(self):
+    # Set default credentials, see
+    # https://cloud.google.com/docs/authentication/application-default-credentials
+    os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = CREDENTIAL_PATH
+
+    gcpkms.GcpKmsClient.register_client(KEY2_URI, None)
+
+    handle = tink.new_keyset_handle(
+        aead.aead_key_templates.create_kms_aead_key_template(KEY2_URI)
+    )
+    gcp_aead = handle.primitive(aead.Aead)
+    ciphertext = gcp_aead.encrypt(b'plaintext', b'associated_data')
+    self.assertEqual(
+        b'plaintext', gcp_aead.decrypt(ciphertext, b'associated_data')
+    )
 
 if __name__ == '__main__':
   absltest.main()