exceptionfactory commented on code in PR #11604: URL: https://github.com/apache/nifi/pull/11604#discussion_r3918770063
########## nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProvider.java: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * 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. + */ +package org.apache.nifi.processors.gcp.cloudsql; + +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auth.oauth2.IdentityPoolCredentials; +import com.google.auth.oauth2.ImpersonatedCredentials; +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.ConfigVerificationResult.Outcome; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.controller.VerifiableControllerService; +import org.apache.nifi.dbcp.api.DatabasePasswordProvider; +import org.apache.nifi.dbcp.api.DatabasePasswordRequestContext; +import org.apache.nifi.gcp.credentials.service.GCPCredentialsService; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.reporting.InitializationException; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +@Tags({"gcp", "cloud sql", "postgresql", "mysql", "iam", "jdbc", "password"}) +@CapabilityDescription(""" + Generates Google Cloud SQL IAM authentication tokens for Cloud SQL database connections. + PostgreSQL and MySQL are supported. + The generated access token replaces the database user password so that NiFi does not need to store long-lived credentials inside DBCP services. + """) +public class GcpCloudSqlIamDatabasePasswordProvider extends AbstractControllerService implements DatabasePasswordProvider, VerifiableControllerService { + + static final String SQLSERVICE_LOGIN_SCOPE = "https://www.googleapis.com/auth/sqlservice.login";; + static final String FAILED_PASSWORD_MESSAGE = "Failed to generate Cloud SQL IAM database password"; + static final String VERIFY_SCOPE_STEP = "Resolve GCP credentials"; + static final String VERIFY_TOKEN_STEP = "Acquire Cloud SQL IAM access token"; + static final String VERIFY_CREDENTIALS_UNAVAILABLE = "Configured GCP Credentials Provider Service did not return Google credentials."; + static final String VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE = "Failed to apply the Cloud SQL login scope to the configured Google credentials."; + static final String VERIFY_IMPERSONATION_REQUIRED = "Target service account impersonation is required for Workload Identity Federation Cloud SQL authentication."; + static final String VERIFY_TOKEN_ACQUISITION_FAILED = "Failed to acquire a Cloud SQL IAM access token."; + static final String VERIFY_TOKEN_MISSING = "Cloud SQL IAM access token was empty."; + + static final PropertyDescriptor GCP_CREDENTIALS_PROVIDER_SERVICE = new PropertyDescriptor.Builder() + .name("GCP Credentials Provider Service") + .description("Controller Service that provides the Google credentials used to request Cloud SQL IAM authentication tokens.") + .identifiesControllerService(GCPCredentialsService.class) + .required(true) + .build(); + + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of( + GCP_CREDENTIALS_PROVIDER_SERVICE + ); + + private volatile GoogleCredentials scopedCredentials; + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @OnEnabled + public void onEnabled(final ConfigurationContext context) throws InitializationException { + scopedCredentials = requireScopedCredentials(context); + } + + @OnDisabled + public void onDisabled() { + scopedCredentials = null; + } + + @Override + public char[] getPassword(final DatabasePasswordRequestContext requestContext) { + Objects.requireNonNull(requestContext, "Database Password Request Context required"); + + final GoogleCredentials credentials = scopedCredentials; + if (credentials == null) { + throw new ProcessException(FAILED_PASSWORD_MESSAGE); + } + + rejectIdentityPoolCredentialsOnPasswordGeneration(credentials); + + final AccessToken accessToken = refreshAccessToken(credentials); + if (!hasTokenValue(accessToken)) { + throw new ProcessException(FAILED_PASSWORD_MESSAGE); + } + + return accessToken.getTokenValue().toCharArray(); + } + + @Override + public List<ConfigVerificationResult> verify(final ConfigurationContext context, final ComponentLog verificationLogger, Review Comment: This method has a number of try-catch blocks and a large number of returns. It would be helpful to consolidate to a single return. It may be helpful to collapse multiple exception conditions into a single one, especially since the messages can be the same. ########## nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProviderTest.java: ########## @@ -0,0 +1,699 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * 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. + */ +package org.apache.nifi.processors.gcp.cloudsql; + +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auth.oauth2.IdentityPoolCredentials; +import com.google.auth.oauth2.ImpersonatedCredentials; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.dbcp.api.DatabasePasswordProvider; +import org.apache.nifi.dbcp.api.DatabasePasswordRequestContext; +import org.apache.nifi.gcp.credentials.service.GCPCredentialsService; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.LogMessage; +import org.apache.nifi.util.MockComponentLog; +import org.apache.nifi.util.NoOpProcessor; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import org.slf4j.helpers.MessageFormatter; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.nifi.components.ConfigVerificationResult.Outcome.FAILED; +import static org.apache.nifi.components.ConfigVerificationResult.Outcome.SUCCESSFUL; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.GCP_CREDENTIALS_PROVIDER_SERVICE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.SQLSERVICE_LOGIN_SCOPE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_CREDENTIALS_UNAVAILABLE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_IMPERSONATION_REQUIRED; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_SCOPE_STEP; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_TOKEN_ACQUISITION_FAILED; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_TOKEN_MISSING; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_TOKEN_STEP; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class GcpCloudSqlIamDatabasePasswordProviderTest { + + private static final String CREDENTIALS_SERVICE_ID = "gcpCredentials"; + private static final String PASSWORD_PROVIDER_ID = "cloudSqlIamProvider"; + private static final String DRIVER_CLASS = "org.postgresql.Driver"; + private static final String DATABASE_USER = "[email protected]"; + private static final String JDBC_URL = "jdbc:postgresql://example:5432/database"; + private static final String TOKEN_VALUE = "cloud-sql-token"; + private static final String REFRESHED_TOKEN_VALUE = "refreshed-cloud-sql-token"; + private static final String LEAK_SENTINEL = "sentinel-token-value"; + + private ExecutorService executorService; + + @AfterEach + void tearDown() { + if (executorService != null) { + executorService.shutdownNow(); + } + } + + @Test + void testSupportedPropertyDescriptorsContainOnlyCredentialsService() throws Exception { + final List<PropertyDescriptor> descriptors = getSupportedPropertyDescriptors(new GcpCloudSqlIamDatabasePasswordProvider()); + + assertEquals(1, descriptors.size()); + assertEquals(GCP_CREDENTIALS_PROVIDER_SERVICE, descriptors.get(0)); + assertTrue(descriptors.get(0).isRequired()); + } + + @Test + void testOnEnabledCachesScopedCredentialAndReusesIt() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15)); + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials(scopedCredentials); + final TestRunner runner = configureRunner(rootCredentials); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + assertEquals(1, rootCredentials.getCreateScopedCount()); + assertEquals(List.of(SQLSERVICE_LOGIN_SCOPE), rootCredentials.getLastRequestedScopes()); + assertSame(scopedCredentials, getScopedCredentials(provider)); + assertEquals(TOKEN_VALUE, new String(provider.getPassword(requestContext()))); + assertEquals(TOKEN_VALUE, new String(provider.getPassword(requestContext()))); + assertEquals(0, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testOnDisabledClearsCachedCredential() throws Exception { + final TestRunner runner = configureRunner(new RootGoogleCredentials(new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15))), true); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + runner.disableControllerService(provider); + + assertNull(getScopedCredentials(provider)); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + assertNull(exception.getCause()); + } + + @Test + void testVerifyImpersonatedCredentialsAcquireLiveToken() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials(accessToken(TOKEN_VALUE, 15)); + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials(scopedCredentials); + final TestRunner runner = configureRunner(rootCredentials); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, SUCCESSFUL, "impersonation"); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, SUCCESSFUL, "DBCP Verify"); + assertEquals(2, rootCredentials.getCreateScopedCount()); + Mockito.verify(scopedCredentials).refreshAccessToken(); + } + + @Test + void testVerifyUsesFreshScopedCredentialWithoutMutatingEnabledState() throws Exception { + final TestScopedGoogleCredentials enabledScopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15)); + final TestScopedGoogleCredentials verificationScopedCredentials = new TestScopedGoogleCredentials(accessToken(REFRESHED_TOKEN_VALUE, -15)); + verificationScopedCredentials.setRefreshedAccessToken(accessToken(REFRESHED_TOKEN_VALUE, 15)); + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials(enabledScopedCredentials, verificationScopedCredentials); + final TestRunner runner = configureRunner(rootCredentials); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + final char[] password = provider.getPassword(requestContext()); + + assertEquals(2, rootCredentials.getCreateScopedCount()); + assertEquals(0, enabledScopedCredentials.getRefreshAccessTokenCount()); + assertEquals(1, verificationScopedCredentials.getRefreshAccessTokenCount()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, SUCCESSFUL, "Cloud SQL IAM access token"); + assertArrayEquals(TOKEN_VALUE.toCharArray(), password); + } + + @Test + void testVerifyIdentityPoolCredentialsRequiresImpersonation() throws Exception { + final IdentityPoolCredentials scopedCredentials = identityPoolCredentials(accessToken(TOKEN_VALUE, 15)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials), false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_IMPERSONATION_REQUIRED); + Mockito.verify(scopedCredentials, Mockito.never()).refreshAccessToken(); + } + + @Test + void testOnEnabledRejectsIdentityPoolCredentialsBeforePublishingState() throws Exception { + final IdentityPoolCredentials scopedCredentials = identityPoolCredentials(accessToken(TOKEN_VALUE, 15)); + final GcpCloudSqlIamDatabasePasswordProvider provider = new GcpCloudSqlIamDatabasePasswordProvider(); + final ConfigurationContext context = mock(ConfigurationContext.class); + final PropertyValue credentialsPropertyValue = mock(PropertyValue.class); + final GCPCredentialsService credentialsService = mock(GCPCredentialsService.class); + + when(context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)).thenReturn(credentialsPropertyValue); + when(credentialsPropertyValue.asControllerService(GCPCredentialsService.class)).thenReturn(credentialsService); + when(credentialsService.getGoogleCredentials()).thenReturn(new RootGoogleCredentials(scopedCredentials)); + + final InitializationException exception = assertThrows(InitializationException.class, () -> provider.onEnabled(context)); + + assertTrue(exception.getMessage().contains("impersonation")); + assertNull(getScopedCredentials(provider)); + } + + @Test + void testVerifyNullCredentialsFails() throws Exception { + final TestRunner runner = configureRunner(null, false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_CREDENTIALS_UNAVAILABLE); + } + + @Test + void testOnEnabledNullCredentialsFails() throws Exception { + final GcpCloudSqlIamDatabasePasswordProvider provider = new GcpCloudSqlIamDatabasePasswordProvider(); + final ConfigurationContext context = mock(ConfigurationContext.class); + final PropertyValue credentialsPropertyValue = mock(PropertyValue.class); + + when(context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)).thenReturn(credentialsPropertyValue); + when(credentialsPropertyValue.asControllerService(GCPCredentialsService.class)).thenReturn(null); + + final InitializationException exception = assertThrows(InitializationException.class, () -> provider.onEnabled(context)); + + assertTrue(exception.getMessage().contains("credentials")); + assertNull(getScopedCredentials(provider)); + } + + @Test + void testVerifyScopedCredentialCreationReturningNullFails() throws Exception { + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials((GoogleCredentials) null); + final TestRunner runner = configureRunner(rootCredentials, false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE); + assertEquals(List.of(SQLSERVICE_LOGIN_SCOPE), rootCredentials.getLastRequestedScopes()); + } + + @Test + void testVerifyScopedCredentialCreationFailureIsReported() throws Exception { + final TestRunner runner = configureRunner(new RootGoogleCredentials(new IllegalStateException(LEAK_SENTINEL)), false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE); + assertFalse(results.get(0).getExplanation().contains(LEAK_SENTINEL)); + } + + @Test + void testOnEnabledScopedCredentialCreationReturningNullFails() throws Exception { + final GcpCloudSqlIamDatabasePasswordProvider provider = new GcpCloudSqlIamDatabasePasswordProvider(); + final ConfigurationContext context = mock(ConfigurationContext.class); + final PropertyValue credentialsPropertyValue = mock(PropertyValue.class); + final GCPCredentialsService credentialsService = mock(GCPCredentialsService.class); + + when(context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)).thenReturn(credentialsPropertyValue); + when(credentialsPropertyValue.asControllerService(GCPCredentialsService.class)).thenReturn(credentialsService); + when(credentialsService.getGoogleCredentials()).thenReturn(new RootGoogleCredentials((GoogleCredentials) null)); + + final InitializationException exception = assertThrows(InitializationException.class, () -> provider.onEnabled(context)); + + assertTrue(exception.getMessage().contains("scope")); + assertNull(getScopedCredentials(provider)); + } + + @Test + void testVerifyRefreshFailureIsSanitized() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials(ioException(LEAK_SENTINEL)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, FAILED, VERIFY_TOKEN_ACQUISITION_FAILED); + assertFalse(results.get(1).getExplanation().contains(LEAK_SENTINEL)); + assertNoLogMessagesContain(runner.getControllerServiceLogger(PASSWORD_PROVIDER_ID), LEAK_SENTINEL); + } + + @Test + void testVerifyNullAccessTokenFails() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials((AccessToken) null); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + + final List<ConfigVerificationResult> results = runner.verify(getProviderImplementation(runner), Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, FAILED, VERIFY_TOKEN_MISSING); + } + + @Test + void testVerifyBlankAccessTokenFails() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials(accessToken(" ", 15)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + + final List<ConfigVerificationResult> results = runner.verify(getProviderImplementation(runner), Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, FAILED, VERIFY_TOKEN_MISSING); + } + + @Test + void testFreshTokenDoesNotRefresh() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final char[] password = provider.getPassword(requestContext()); + + assertArrayEquals(TOKEN_VALUE.toCharArray(), password); + assertEquals(0, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testExpiredTokenRefreshes() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, -15)); + scopedCredentials.setRefreshedAccessToken(accessToken(REFRESHED_TOKEN_VALUE, 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final char[] password = provider.getPassword(requestContext()); + + assertArrayEquals(REFRESHED_TOKEN_VALUE.toCharArray(), password); + assertEquals(1, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testConcurrentGetPasswordPerformsSingleRefresh() throws Exception { + final BlockingScopedGoogleCredentials scopedCredentials = new BlockingScopedGoogleCredentials(); + scopedCredentials.setRefreshedAccessToken(accessToken(REFRESHED_TOKEN_VALUE, 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + executorService = Executors.newFixedThreadPool(2); + final CountDownLatch startLatch = new CountDownLatch(1); + final Future<char[]> first = executorService.submit(() -> getPasswordAfterStart(provider, startLatch)); + final Future<char[]> second = executorService.submit(() -> getPasswordAfterStart(provider, startLatch)); + + startLatch.countDown(); + assertTrue(scopedCredentials.awaitRefreshEntry()); + scopedCredentials.releaseRefresh(); + + assertArrayEquals(REFRESHED_TOKEN_VALUE.toCharArray(), first.get(5, TimeUnit.SECONDS)); + assertArrayEquals(REFRESHED_TOKEN_VALUE.toCharArray(), second.get(5, TimeUnit.SECONDS)); + assertEquals(1, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testNullAccessTokenRejectedForPasswordGeneration() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(null); + scopedCredentials.setRefreshedAccessToken(null); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + } + + @Test + void testBlankAccessTokenRejectedForPasswordGeneration() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(null); + scopedCredentials.setRefreshedAccessToken(accessToken(" ", 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + } + + @Test + void testRefreshFailureIsSanitizedForPasswordGeneration() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(null); + scopedCredentials.setRefreshException(ioException(LEAK_SENTINEL)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + final DatabasePasswordProvider provider = getProvider(runner); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + assertNull(exception.getCause()); + assertNoLogMessagesContain(runner.getControllerServiceLogger(PASSWORD_PROVIDER_ID), LEAK_SENTINEL); + } + + @Test + void testIdentityPoolCredentialsFailClosedAtRuntime() throws Exception { + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation( + configureRunner(new RootGoogleCredentials(new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15))))); + + setScopedCredentials(provider, identityPoolCredentials(accessToken(TOKEN_VALUE, 15))); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); Review Comment: It would be helpful to declare `Cloud SQL IAM` as a static and reuse the reference across multiple methods. ########## nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProviderTest.java: ########## @@ -0,0 +1,699 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * 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. + */ +package org.apache.nifi.processors.gcp.cloudsql; + +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auth.oauth2.IdentityPoolCredentials; +import com.google.auth.oauth2.ImpersonatedCredentials; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.dbcp.api.DatabasePasswordProvider; +import org.apache.nifi.dbcp.api.DatabasePasswordRequestContext; +import org.apache.nifi.gcp.credentials.service.GCPCredentialsService; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.LogMessage; +import org.apache.nifi.util.MockComponentLog; +import org.apache.nifi.util.NoOpProcessor; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import org.slf4j.helpers.MessageFormatter; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.nifi.components.ConfigVerificationResult.Outcome.FAILED; +import static org.apache.nifi.components.ConfigVerificationResult.Outcome.SUCCESSFUL; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.GCP_CREDENTIALS_PROVIDER_SERVICE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.SQLSERVICE_LOGIN_SCOPE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_CREDENTIALS_UNAVAILABLE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_IMPERSONATION_REQUIRED; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_SCOPE_STEP; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_TOKEN_ACQUISITION_FAILED; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_TOKEN_MISSING; +import static org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider.VERIFY_TOKEN_STEP; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class GcpCloudSqlIamDatabasePasswordProviderTest { + + private static final String CREDENTIALS_SERVICE_ID = "gcpCredentials"; + private static final String PASSWORD_PROVIDER_ID = "cloudSqlIamProvider"; + private static final String DRIVER_CLASS = "org.postgresql.Driver"; + private static final String DATABASE_USER = "[email protected]"; + private static final String JDBC_URL = "jdbc:postgresql://example:5432/database"; + private static final String TOKEN_VALUE = "cloud-sql-token"; + private static final String REFRESHED_TOKEN_VALUE = "refreshed-cloud-sql-token"; + private static final String LEAK_SENTINEL = "sentinel-token-value"; + + private ExecutorService executorService; + + @AfterEach + void tearDown() { + if (executorService != null) { + executorService.shutdownNow(); + } + } + + @Test + void testSupportedPropertyDescriptorsContainOnlyCredentialsService() throws Exception { + final List<PropertyDescriptor> descriptors = getSupportedPropertyDescriptors(new GcpCloudSqlIamDatabasePasswordProvider()); + + assertEquals(1, descriptors.size()); + assertEquals(GCP_CREDENTIALS_PROVIDER_SERVICE, descriptors.get(0)); + assertTrue(descriptors.get(0).isRequired()); + } + + @Test + void testOnEnabledCachesScopedCredentialAndReusesIt() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15)); + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials(scopedCredentials); + final TestRunner runner = configureRunner(rootCredentials); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + assertEquals(1, rootCredentials.getCreateScopedCount()); + assertEquals(List.of(SQLSERVICE_LOGIN_SCOPE), rootCredentials.getLastRequestedScopes()); + assertSame(scopedCredentials, getScopedCredentials(provider)); + assertEquals(TOKEN_VALUE, new String(provider.getPassword(requestContext()))); + assertEquals(TOKEN_VALUE, new String(provider.getPassword(requestContext()))); + assertEquals(0, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testOnDisabledClearsCachedCredential() throws Exception { + final TestRunner runner = configureRunner(new RootGoogleCredentials(new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15))), true); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + runner.disableControllerService(provider); + + assertNull(getScopedCredentials(provider)); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + assertNull(exception.getCause()); + } + + @Test + void testVerifyImpersonatedCredentialsAcquireLiveToken() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials(accessToken(TOKEN_VALUE, 15)); + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials(scopedCredentials); + final TestRunner runner = configureRunner(rootCredentials); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, SUCCESSFUL, "impersonation"); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, SUCCESSFUL, "DBCP Verify"); + assertEquals(2, rootCredentials.getCreateScopedCount()); + Mockito.verify(scopedCredentials).refreshAccessToken(); + } + + @Test + void testVerifyUsesFreshScopedCredentialWithoutMutatingEnabledState() throws Exception { + final TestScopedGoogleCredentials enabledScopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15)); + final TestScopedGoogleCredentials verificationScopedCredentials = new TestScopedGoogleCredentials(accessToken(REFRESHED_TOKEN_VALUE, -15)); + verificationScopedCredentials.setRefreshedAccessToken(accessToken(REFRESHED_TOKEN_VALUE, 15)); + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials(enabledScopedCredentials, verificationScopedCredentials); + final TestRunner runner = configureRunner(rootCredentials); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + final char[] password = provider.getPassword(requestContext()); + + assertEquals(2, rootCredentials.getCreateScopedCount()); + assertEquals(0, enabledScopedCredentials.getRefreshAccessTokenCount()); + assertEquals(1, verificationScopedCredentials.getRefreshAccessTokenCount()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, SUCCESSFUL, "Cloud SQL IAM access token"); + assertArrayEquals(TOKEN_VALUE.toCharArray(), password); + } + + @Test + void testVerifyIdentityPoolCredentialsRequiresImpersonation() throws Exception { + final IdentityPoolCredentials scopedCredentials = identityPoolCredentials(accessToken(TOKEN_VALUE, 15)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials), false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_IMPERSONATION_REQUIRED); + Mockito.verify(scopedCredentials, Mockito.never()).refreshAccessToken(); + } + + @Test + void testOnEnabledRejectsIdentityPoolCredentialsBeforePublishingState() throws Exception { + final IdentityPoolCredentials scopedCredentials = identityPoolCredentials(accessToken(TOKEN_VALUE, 15)); + final GcpCloudSqlIamDatabasePasswordProvider provider = new GcpCloudSqlIamDatabasePasswordProvider(); + final ConfigurationContext context = mock(ConfigurationContext.class); + final PropertyValue credentialsPropertyValue = mock(PropertyValue.class); + final GCPCredentialsService credentialsService = mock(GCPCredentialsService.class); + + when(context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)).thenReturn(credentialsPropertyValue); + when(credentialsPropertyValue.asControllerService(GCPCredentialsService.class)).thenReturn(credentialsService); + when(credentialsService.getGoogleCredentials()).thenReturn(new RootGoogleCredentials(scopedCredentials)); + + final InitializationException exception = assertThrows(InitializationException.class, () -> provider.onEnabled(context)); + + assertTrue(exception.getMessage().contains("impersonation")); + assertNull(getScopedCredentials(provider)); + } + + @Test + void testVerifyNullCredentialsFails() throws Exception { + final TestRunner runner = configureRunner(null, false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_CREDENTIALS_UNAVAILABLE); + } + + @Test + void testOnEnabledNullCredentialsFails() throws Exception { + final GcpCloudSqlIamDatabasePasswordProvider provider = new GcpCloudSqlIamDatabasePasswordProvider(); + final ConfigurationContext context = mock(ConfigurationContext.class); + final PropertyValue credentialsPropertyValue = mock(PropertyValue.class); + + when(context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)).thenReturn(credentialsPropertyValue); + when(credentialsPropertyValue.asControllerService(GCPCredentialsService.class)).thenReturn(null); + + final InitializationException exception = assertThrows(InitializationException.class, () -> provider.onEnabled(context)); + + assertTrue(exception.getMessage().contains("credentials")); + assertNull(getScopedCredentials(provider)); + } + + @Test + void testVerifyScopedCredentialCreationReturningNullFails() throws Exception { + final RootGoogleCredentials rootCredentials = new RootGoogleCredentials((GoogleCredentials) null); + final TestRunner runner = configureRunner(rootCredentials, false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE); + assertEquals(List.of(SQLSERVICE_LOGIN_SCOPE), rootCredentials.getLastRequestedScopes()); + } + + @Test + void testVerifyScopedCredentialCreationFailureIsReported() throws Exception { + final TestRunner runner = configureRunner(new RootGoogleCredentials(new IllegalStateException(LEAK_SENTINEL)), false); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(1, results.size()); + assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP, FAILED, VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE); + assertFalse(results.get(0).getExplanation().contains(LEAK_SENTINEL)); + } + + @Test + void testOnEnabledScopedCredentialCreationReturningNullFails() throws Exception { + final GcpCloudSqlIamDatabasePasswordProvider provider = new GcpCloudSqlIamDatabasePasswordProvider(); + final ConfigurationContext context = mock(ConfigurationContext.class); + final PropertyValue credentialsPropertyValue = mock(PropertyValue.class); + final GCPCredentialsService credentialsService = mock(GCPCredentialsService.class); + + when(context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)).thenReturn(credentialsPropertyValue); + when(credentialsPropertyValue.asControllerService(GCPCredentialsService.class)).thenReturn(credentialsService); + when(credentialsService.getGoogleCredentials()).thenReturn(new RootGoogleCredentials((GoogleCredentials) null)); + + final InitializationException exception = assertThrows(InitializationException.class, () -> provider.onEnabled(context)); + + assertTrue(exception.getMessage().contains("scope")); + assertNull(getScopedCredentials(provider)); + } + + @Test + void testVerifyRefreshFailureIsSanitized() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials(ioException(LEAK_SENTINEL)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation(runner); + + final List<ConfigVerificationResult> results = runner.verify(provider, Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, FAILED, VERIFY_TOKEN_ACQUISITION_FAILED); + assertFalse(results.get(1).getExplanation().contains(LEAK_SENTINEL)); + assertNoLogMessagesContain(runner.getControllerServiceLogger(PASSWORD_PROVIDER_ID), LEAK_SENTINEL); + } + + @Test + void testVerifyNullAccessTokenFails() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials((AccessToken) null); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + + final List<ConfigVerificationResult> results = runner.verify(getProviderImplementation(runner), Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, FAILED, VERIFY_TOKEN_MISSING); + } + + @Test + void testVerifyBlankAccessTokenFails() throws Exception { + final ImpersonatedCredentials scopedCredentials = impersonatedCredentials(accessToken(" ", 15)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + + final List<ConfigVerificationResult> results = runner.verify(getProviderImplementation(runner), Map.of()); + + assertEquals(2, results.size()); + assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP, FAILED, VERIFY_TOKEN_MISSING); + } + + @Test + void testFreshTokenDoesNotRefresh() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final char[] password = provider.getPassword(requestContext()); + + assertArrayEquals(TOKEN_VALUE.toCharArray(), password); + assertEquals(0, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testExpiredTokenRefreshes() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, -15)); + scopedCredentials.setRefreshedAccessToken(accessToken(REFRESHED_TOKEN_VALUE, 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final char[] password = provider.getPassword(requestContext()); + + assertArrayEquals(REFRESHED_TOKEN_VALUE.toCharArray(), password); + assertEquals(1, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testConcurrentGetPasswordPerformsSingleRefresh() throws Exception { + final BlockingScopedGoogleCredentials scopedCredentials = new BlockingScopedGoogleCredentials(); + scopedCredentials.setRefreshedAccessToken(accessToken(REFRESHED_TOKEN_VALUE, 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + executorService = Executors.newFixedThreadPool(2); + final CountDownLatch startLatch = new CountDownLatch(1); + final Future<char[]> first = executorService.submit(() -> getPasswordAfterStart(provider, startLatch)); + final Future<char[]> second = executorService.submit(() -> getPasswordAfterStart(provider, startLatch)); + + startLatch.countDown(); + assertTrue(scopedCredentials.awaitRefreshEntry()); + scopedCredentials.releaseRefresh(); + + assertArrayEquals(REFRESHED_TOKEN_VALUE.toCharArray(), first.get(5, TimeUnit.SECONDS)); + assertArrayEquals(REFRESHED_TOKEN_VALUE.toCharArray(), second.get(5, TimeUnit.SECONDS)); + assertEquals(1, scopedCredentials.getRefreshAccessTokenCount()); + } + + @Test + void testNullAccessTokenRejectedForPasswordGeneration() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(null); + scopedCredentials.setRefreshedAccessToken(null); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + } + + @Test + void testBlankAccessTokenRejectedForPasswordGeneration() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(null); + scopedCredentials.setRefreshedAccessToken(accessToken(" ", 15)); + final DatabasePasswordProvider provider = getProvider(configureRunner(new RootGoogleCredentials(scopedCredentials))); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + } + + @Test + void testRefreshFailureIsSanitizedForPasswordGeneration() throws Exception { + final TestScopedGoogleCredentials scopedCredentials = new TestScopedGoogleCredentials(null); + scopedCredentials.setRefreshException(ioException(LEAK_SENTINEL)); + final TestRunner runner = configureRunner(new RootGoogleCredentials(scopedCredentials)); + final DatabasePasswordProvider provider = getProvider(runner); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + assertNull(exception.getCause()); + assertNoLogMessagesContain(runner.getControllerServiceLogger(PASSWORD_PROVIDER_ID), LEAK_SENTINEL); + } + + @Test + void testIdentityPoolCredentialsFailClosedAtRuntime() throws Exception { + final GcpCloudSqlIamDatabasePasswordProvider provider = getProviderImplementation( + configureRunner(new RootGoogleCredentials(new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15))))); + + setScopedCredentials(provider, identityPoolCredentials(accessToken(TOKEN_VALUE, 15))); + + final ProcessException exception = assertThrows(ProcessException.class, () -> provider.getPassword(requestContext())); + + assertTrue(exception.getMessage().contains("Cloud SQL IAM")); + } + + @Test + void testGetPasswordReturnsFreshCharacterArrayEachCall() throws Exception { + final DatabasePasswordProvider provider = getProvider(configureRunner( + new RootGoogleCredentials(new TestScopedGoogleCredentials(accessToken(TOKEN_VALUE, 15))))); + + final char[] firstPassword = provider.getPassword(requestContext()); + firstPassword[0] = 'X'; + final char[] secondPassword = provider.getPassword(requestContext()); + + assertNotSame(firstPassword, secondPassword); + assertArrayEquals(TOKEN_VALUE.toCharArray(), secondPassword); + } + + @Test + void testControllerServiceRegistrationContainsProvider() throws IOException { + final String resourcePath = "META-INF/services/org.apache.nifi.controller.ControllerService"; + try (InputStream inputStream = GcpCloudSqlIamDatabasePasswordProvider.class.getClassLoader().getResourceAsStream(resourcePath)) { + assertNotNull(inputStream); + final String registeredServices = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8); + assertTrue(registeredServices.contains(GcpCloudSqlIamDatabasePasswordProvider.class.getName())); + } + } + + @Test + void testAdditionalDetailsResourceDocumentsSupportedPath() throws IOException { Review Comment: I recommend removing this test method since it is purely about documentation -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
