exceptionfactory commented on code in PR #11604:
URL: https://github.com/apache/nifi/pull/11604#discussion_r3941386795
##########
nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java:
##########
@@ -166,6 +218,69 @@ public void testMultipleCredentialSourcesRemainValid()
throws Exception {
runner.assertValid(serviceImpl);
}
+ @Test
+ public void testComputeEngineCredentials() throws Exception {
+ final TestRunner runner =
TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new
GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ runner.setProperty(serviceImpl, AUTHENTICATION_STRATEGY,
AuthenticationStrategy.COMPUTE_ENGINE.getValue());
+ runner.enableControllerService(serviceImpl);
+
+ runner.assertValid(serviceImpl);
+
+ final GCPCredentialsService service = (GCPCredentialsService)
runner.getProcessContext()
+
.getControllerServiceLookup().getControllerService("gcpCredentialsProvider");
+
+ assertNotNull(service);
+ final GoogleCredentials credentials = service.getGoogleCredentials();
+ assertNotNull(credentials);
+
+ assertEquals(ComputeEngineCredentials.class, credentials.getClass(),
+ "Credentials class should be equal");
+ }
+
+ @Test
+ public void testTargetServiceAccountBlankRejected() throws Exception {
+ final TestRunner runner =
TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new
GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ final MockOAuth2AccessTokenProvider subjectTokenProvider = new
MockOAuth2AccessTokenProvider();
+ runner.addControllerService("subjectTokenProvider",
subjectTokenProvider);
+ runner.enableControllerService(subjectTokenProvider);
+
+ configureWorkloadIdentityFederation(runner, serviceImpl);
+ runner.setProperty(serviceImpl, TARGET_SERVICE_ACCOUNT, " ");
+
+ assertHasInvalidResult(runner.validate(serviceImpl), "must contain at
least one character that is not white space");
+ }
+
+ @Test
+ public void testTargetServiceAccountRequiresWorkloadIdentityFederation()
throws Exception {
+ final TestRunner runner =
TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new
GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ runner.setProperty(serviceImpl, AUTHENTICATION_STRATEGY,
AuthenticationStrategy.APPLICATION_DEFAULT.getValue());
+ runner.setProperty(serviceImpl, TARGET_SERVICE_ACCOUNT,
TARGET_SERVICE_ACCOUNT_VALUE);
+
+ assertHasInvalidResult(runner.validate(serviceImpl), "Target Service
Account requires Workload Identity Federation");
+ }
+
+ @Test
+ public void
testAdditionalDetailsDocumentTargetServiceAccountVerificationSemantics() throws
IOException {
Review Comment:
This test should also be removed
##########
nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProvider.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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 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,
+ final Map<String, String>
attributes) {
+ final List<ConfigVerificationResult> results = new ArrayList<>(2);
+ final GoogleCredentials scopedVerificationCredentials =
resolveVerificationCredentials(context);
+ if (scopedVerificationCredentials == null) {
+ results.add(buildVerificationResult(VERIFY_SCOPE_STEP,
Outcome.FAILED, VERIFY_CREDENTIALS_UNAVAILABLE));
+ } else {
+ final ConfigVerificationResult scopedCredentialResult =
describeScopedCredential(scopedVerificationCredentials);
+ results.add(scopedCredentialResult);
+ if (scopedCredentialResult.getOutcome() == Outcome.SUCCESSFUL) {
+ results.add(verifyAccessToken(scopedVerificationCredentials));
+ }
+ }
+ return results;
+ }
+
+ private GoogleCredentials resolveVerificationCredentials(final
ConfigurationContext context) {
+ try {
+ return
createSqlLoginScopedCredentials(resolveGoogleCredentials(context));
+ } catch (final RuntimeException e) {
+ return null;
+ }
+ }
+
+ private ConfigVerificationResult verifyAccessToken(final GoogleCredentials
credentials) {
+ try {
+ final AccessToken accessToken = credentials.refreshAccessToken();
+ return hasTokenValue(accessToken)
+ ? buildTokenVerificationResult()
+ : buildVerificationResult(VERIFY_TOKEN_STEP,
Outcome.FAILED, VERIFY_TOKEN_ACQUISITION_FAILED);
+ } catch (final IOException | RuntimeException e) {
+ return buildVerificationResult(VERIFY_TOKEN_STEP, Outcome.FAILED,
VERIFY_TOKEN_ACQUISITION_FAILED);
+ }
+ }
+
+ private GoogleCredentials requireScopedCredentials(final
ConfigurationContext context) throws InitializationException {
+ final GoogleCredentials googleCredentials;
+ try {
+ googleCredentials = resolveGoogleCredentials(context);
+ } catch (final RuntimeException e) {
+ throw new InitializationException(VERIFY_CREDENTIALS_UNAVAILABLE,
e);
+ }
+
+ if (googleCredentials == null) {
+ throw new InitializationException(VERIFY_CREDENTIALS_UNAVAILABLE);
+ }
+
+ final GoogleCredentials credentials;
+ try {
+ credentials = createSqlLoginScopedCredentials(googleCredentials);
+ } catch (final RuntimeException e) {
+ throw new
InitializationException(VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE, e);
+ }
+
+ if (credentials == null) {
+ throw new
InitializationException(VERIFY_SCOPED_CREDENTIALS_UNAVAILABLE);
+ }
+
+ rejectIdentityPoolCredentialsOnEnable(credentials);
+ return credentials;
+ }
+
+ private AccessToken refreshAccessToken(final GoogleCredentials
credentials) {
+ try {
+ credentials.refreshIfExpired();
+ } catch (final IOException | RuntimeException e) {
+ throw new ProcessException(FAILED_PASSWORD_MESSAGE);
+ }
+
+ return credentials.getAccessToken();
+ }
+
+ private GoogleCredentials resolveGoogleCredentials(final
ConfigurationContext context) {
+ final GCPCredentialsService credentialsService =
context.getProperty(GCP_CREDENTIALS_PROVIDER_SERVICE)
+ .asControllerService(GCPCredentialsService.class);
+ if (credentialsService == null) {
+ return null;
+ }
+
+ return credentialsService.getGoogleCredentials();
+ }
+
+ private GoogleCredentials createSqlLoginScopedCredentials(final
GoogleCredentials googleCredentials) {
+ if (googleCredentials == null) {
+ return null;
+ }
+
+ return googleCredentials.createScoped(List.of(SQLSERVICE_LOGIN_SCOPE));
+ }
+
+ private void rejectIdentityPoolCredentialsOnEnable(final GoogleCredentials
credentials) throws InitializationException {
+ if (credentials instanceof IdentityPoolCredentials) {
Review Comment:
It looks like this rejects other kinds of credentials like
`ExternalAccountCredentials`, is that intentional?
##########
nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProvider.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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 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:
With the `verificationLogger` passed in but not used, it would be helpful to
pass it down to some of the nested methods and log the exceptions thrown,
instead of just returning `null` for verification.
##########
nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProviderTest.java:
##########
@@ -0,0 +1,680 @@
+/*
+ * 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_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_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 static final String CLOUD_SQL_IAM = "Cloud SQL IAM";
+
+ 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);
+ 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_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_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_ACQUISITION_FAILED);
+ }
+
+ @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_ACQUISITION_FAILED);
+ }
+
+ @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()));
+ }
+ }
+
+ private TestRunner configureRunner(final GoogleCredentials
rootCredentials) throws Exception {
+ return configureRunner(rootCredentials, true);
+ }
+
+ private TestRunner configureRunner(final GoogleCredentials
rootCredentials, final boolean enableProvider) throws Exception {
+ final TestRunner runner =
TestRunners.newTestRunner(NoOpProcessor.class);
+
+ final TestGCPCredentialsService credentialsService = new
TestGCPCredentialsService(rootCredentials);
+ runner.addControllerService(CREDENTIALS_SERVICE_ID,
credentialsService);
+ runner.enableControllerService(credentialsService);
+
+ final GcpCloudSqlIamDatabasePasswordProvider provider = new
GcpCloudSqlIamDatabasePasswordProvider();
+ runner.addControllerService(PASSWORD_PROVIDER_ID, provider);
+ runner.setProperty(provider, GCP_CREDENTIALS_PROVIDER_SERVICE,
CREDENTIALS_SERVICE_ID);
+ if (enableProvider) {
+ runner.enableControllerService(provider);
+ runner.assertValid(provider);
+ }
+
+ return runner;
+ }
+
+ private DatabasePasswordProvider getProvider(final TestRunner runner) {
+ return (DatabasePasswordProvider) runner.getProcessContext()
+ .getControllerServiceLookup()
+ .getControllerService(PASSWORD_PROVIDER_ID);
+ }
+
+ private GcpCloudSqlIamDatabasePasswordProvider
getProviderImplementation(final TestRunner runner) {
+ return (GcpCloudSqlIamDatabasePasswordProvider) getProvider(runner);
+ }
+
+ private DatabasePasswordRequestContext requestContext() {
+ return DatabasePasswordRequestContext.builder()
+ .jdbcUrl(JDBC_URL)
+ .databaseUser(DATABASE_USER)
+ .driverClassName(DRIVER_CLASS)
+ .connectionProperties(Map.of())
+ .build();
+ }
+
+ private char[] getPasswordAfterStart(final DatabasePasswordProvider
provider, final CountDownLatch startLatch) throws InterruptedException {
+ startLatch.await(5, TimeUnit.SECONDS);
+ return provider.getPassword(requestContext());
+ }
+
+ private static AccessToken accessToken(final String tokenValue, final long
offsetMinutes) {
+ return tokenValue == null ? null : new AccessToken(tokenValue,
java.util.Date.from(Instant.now().plusSeconds(offsetMinutes * 60)));
+ }
+
+ private static IOException ioException(final String message) {
+ return new IOException(message);
+ }
+
+ private static void assertVerificationResult(final
ConfigVerificationResult result, final String stepName,
+ final
ConfigVerificationResult.Outcome outcome, final String explanationFragment) {
+ assertEquals(stepName, result.getVerificationStepName());
+ assertEquals(outcome, result.getOutcome());
+ assertTrue(result.getExplanation().contains(explanationFragment),
result::getExplanation);
+ }
+
+ private static void assertNoLogMessagesContain(final MockComponentLog
logger, final String value) {
+ final List<LogMessage> logMessages = new ArrayList<>();
+ logMessages.addAll(logger.getInfoMessages());
+ logMessages.addAll(logger.getWarnMessages());
+ logMessages.addAll(logger.getErrorMessages());
+
+ for (final LogMessage logMessage : logMessages) {
+ final String rawMessage = logMessage.getMsg();
+ assertFalse(rawMessage != null && rawMessage.contains(value));
+ final Object[] args = logMessage.getArgs();
+ final String formattedMessage =
MessageFormatter.arrayFormat(rawMessage, args == null ? new Object[0] :
args).getMessage();
+ assertFalse(formattedMessage != null &&
formattedMessage.contains(value));
+ if (args != null) {
+ for (final Object arg : args) {
+ final String argValue = arg == null ? null :
arg.toString();
+ assertFalse(argValue != null && argValue.contains(value));
+ }
+ }
+ assertThrowableChainDoesNotContain(logMessage.getThrowable(),
value, java.util.Collections.newSetFromMap(new IdentityHashMap<>()));
Review Comment:
The qualified `java.util.Collections` should be replaced with an import,
along with other fully-qualified class references in this test class
--
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]
