This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 760f9d44ac2 NIFI-16260 Add a Database Password Provider for GCP
CloudSQL (#11604)
760f9d44ac2 is described below
commit 760f9d44ac2726cfdc51f687c355b03b346c7a7e
Author: Pierre Villard <[email protected]>
AuthorDate: Mon Sep 7 17:05:51 2026 +0200
NIFI-16260 Add a Database Password Provider for GCP CloudSQL (#11604)
Signed-off-by: David Handermann <[email protected]>
---
.../nifi-gcp-bundle/nifi-gcp-processors/pom.xml | 5 +
.../GcpCloudSqlIamDatabasePasswordProvider.java | 234 +++++++
.../factory/CredentialPropertyDescriptors.java | 7 +
...kloadIdentityFederationCredentialsStrategy.java | 17 +-
.../service/GCPCredentialsControllerService.java | 19 +-
.../org.apache.nifi.controller.ControllerService | 1 +
.../additionalDetails.md | 91 +++
.../additionalDetails.md | 38 +-
...GcpCloudSqlIamDatabasePasswordProviderTest.java | 669 +++++++++++++++++++++
...dIdentityFederationCredentialsStrategyTest.java | 131 ++++
.../service/GCPCredentialsServiceTest.java | 133 +++-
11 files changed, 1324 insertions(+), 21 deletions(-)
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/pom.xml
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/pom.xml
index 5d6ebd6bc80..e47e00fc485 100644
--- a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/pom.xml
+++ b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/pom.xml
@@ -72,6 +72,11 @@
<version>2.12.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-dbcp-service-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-resource-transfer</artifactId>
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProvider.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProvider.java
new file mode 100644
index 00000000000..fb886529f9d
--- /dev/null
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProvider.java
@@ -0,0 +1,234 @@
+/*
+ * 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.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_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);
+ }
+
+ 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, verificationLogger);
+ 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,
verificationLogger));
+ }
+ }
+ return results;
+ }
+
+ private GoogleCredentials resolveVerificationCredentials(final
ConfigurationContext context, final ComponentLog verificationLogger) {
+ try {
+ return
createSqlLoginScopedCredentials(resolveGoogleCredentials(context));
+ } catch (final RuntimeException e) {
+ verificationLogger.error("Failed to resolve scoped Google
credentials", e);
+ return null;
+ }
+ }
+
+ private ConfigVerificationResult verifyAccessToken(final GoogleCredentials
credentials, final ComponentLog verificationLogger) {
+ 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) {
+ verificationLogger.error("Failed to acquire Cloud SQL IAM access
token", 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);
+ }
+
+ 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 ConfigVerificationResult describeScopedCredential(final
GoogleCredentials scopedVerificationCredentials) {
+ if (scopedVerificationCredentials instanceof ImpersonatedCredentials) {
+ return buildVerificationResult(
+ VERIFY_SCOPE_STEP,
+ Outcome.SUCCESSFUL,
+ "Resolved GCP credentials and Cloud SQL login scope.
Target service account impersonation is active."
+ );
+ }
+
+ return buildVerificationResult(
+ VERIFY_SCOPE_STEP,
+ Outcome.SUCCESSFUL,
+ "Resolved GCP credentials and Cloud SQL login scope."
+ );
+ }
+
+ private ConfigVerificationResult buildTokenVerificationResult() {
+ return buildVerificationResult(
+ VERIFY_TOKEN_STEP,
+ Outcome.SUCCESSFUL,
+ "Acquired a Cloud SQL IAM access token. Use DBCP Verify to
validate the database connection."
+ );
+ }
+
+ private boolean hasTokenValue(final AccessToken accessToken) {
+ return accessToken != null &&
StringUtils.isNotBlank(accessToken.getTokenValue());
+ }
+
+ private ConfigVerificationResult buildVerificationResult(final String
stepName, final Outcome outcome, final String explanation) {
+ return new ConfigVerificationResult.Builder()
+ .verificationStepName(stepName)
+ .outcome(outcome)
+ .explanation(explanation)
+ .build();
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialPropertyDescriptors.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialPropertyDescriptors.java
index c0c0f801491..24a81b118d9 100644
---
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialPropertyDescriptors.java
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialPropertyDescriptors.java
@@ -132,6 +132,13 @@ public final class CredentialPropertyDescriptors {
.dependsOn(AUTHENTICATION_STRATEGY,
AuthenticationStrategy.WORKLOAD_IDENTITY_FEDERATION.getValue())
.build();
+ public static final PropertyDescriptor TARGET_SERVICE_ACCOUNT = new
PropertyDescriptor.Builder()
+ .name("Target Service Account")
+ .description("Target Google service account email to impersonate
after Workload Identity Federation token exchange.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .dependsOn(AUTHENTICATION_STRATEGY,
AuthenticationStrategy.WORKLOAD_IDENTITY_FEDERATION.getValue())
+ .build();
+
public static final PropertyDescriptor DELEGATION_STRATEGY = new
PropertyDescriptor.Builder()
.name("Delegation Strategy")
.required(true)
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategy.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategy.java
index dfe261e895a..b25e790517f 100644
---
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategy.java
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategy.java
@@ -20,6 +20,7 @@ import com.google.auth.http.HttpTransportFactory;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.auth.oauth2.IdentityPoolCredentials;
import com.google.auth.oauth2.IdentityPoolSubjectTokenSupplier;
+import com.google.auth.oauth2.ImpersonatedCredentials;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ConfigurationContext;
@@ -60,6 +61,7 @@ public class WorkloadIdentityFederationCredentialsStrategy
extends AbstractCrede
final String scopeValue =
context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SCOPE).getValue();
final String tokenEndpoint =
context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_TOKEN_ENDPOINT).getValue();
final String subjectTokenType =
context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE).getValue();
+ final String targetServiceAccount =
context.getProperty(CredentialPropertyDescriptors.TARGET_SERVICE_ACCOUNT).getValue();
final List<String> scopes = parseScopes(scopeValue);
final IdentityPoolSubjectTokenSupplier tokenSupplier =
createSubjectTokenSupplier(subjectTokenProvider);
@@ -77,7 +79,20 @@ public class WorkloadIdentityFederationCredentialsStrategy
extends AbstractCrede
builder.setHttpTransportFactory(transportFactory);
}
- return builder.build();
+ final IdentityPoolCredentials sourceCredentials = builder.build();
+ if (StringUtils.isBlank(targetServiceAccount)) {
+ return sourceCredentials;
+ }
+
+ final ImpersonatedCredentials.Builder impersonatedCredentialsBuilder =
ImpersonatedCredentials.newBuilder()
+ .setSourceCredentials(sourceCredentials)
+ .setTargetPrincipal(targetServiceAccount)
+ .setScopes(Collections.emptyList());
+ if (transportFactory != null) {
+
impersonatedCredentialsBuilder.setHttpTransportFactory(transportFactory);
+ }
+
+ return impersonatedCredentialsBuilder.build();
}
private IdentityPoolSubjectTokenSupplier createSubjectTokenSupplier(final
OAuth2AccessTokenProvider tokenProvider) {
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsControllerService.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsControllerService.java
index 25cda97bc5d..c90388ef8f9 100644
---
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsControllerService.java
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsControllerService.java
@@ -18,6 +18,7 @@ package org.apache.nifi.processors.gcp.credentials.service;
import com.google.auth.http.HttpTransportFactory;
import com.google.auth.oauth2.GoogleCredentials;
+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.OnEnabled;
@@ -55,6 +56,7 @@ import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPrope
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.LEGACY_USE_COMPUTE_ENGINE_CREDENTIALS;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE;
+import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.TARGET_SERVICE_ACCOUNT;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_AUDIENCE;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SCOPE;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SUBJECT_TOKEN_PROVIDER;
@@ -84,6 +86,7 @@ public class GCPCredentialsControllerService extends
AbstractControllerService i
WORKLOAD_IDENTITY_TOKEN_ENDPOINT,
WORKLOAD_IDENTITY_SUBJECT_TOKEN_PROVIDER,
WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE,
+ TARGET_SERVICE_ACCOUNT,
ProxyConfiguration.createProxyConfigPropertyDescriptor(ProxyAwareTransportFactory.PROXY_SPECS),
DELEGATION_STRATEGY,
DELEGATION_USER
@@ -106,6 +109,20 @@ public class GCPCredentialsControllerService extends
AbstractControllerService i
protected Collection<ValidationResult> customValidate(final
ValidationContext validationContext) {
final List<ValidationResult> results = new ArrayList<>();
ProxyConfiguration.validateProxySpec(validationContext, results,
ProxyAwareTransportFactory.PROXY_SPECS);
+
+ final String targetServiceAccount =
validationContext.getProperty(TARGET_SERVICE_ACCOUNT).getValue();
+ final AuthenticationStrategy authenticationStrategy =
validationContext.getProperty(AUTHENTICATION_STRATEGY)
+ .asAllowableValue(AuthenticationStrategy.class);
+ if (StringUtils.isNotBlank(targetServiceAccount)
+ && authenticationStrategy !=
AuthenticationStrategy.WORKLOAD_IDENTITY_FEDERATION) {
+ results.add(new ValidationResult.Builder()
+ .subject(TARGET_SERVICE_ACCOUNT.getDisplayName())
+ .input(targetServiceAccount)
+ .valid(false)
+ .explanation("Target Service Account requires Workload
Identity Federation")
+ .build());
+ }
+
return results;
}
@@ -117,7 +134,7 @@ public class GCPCredentialsControllerService extends
AbstractControllerService i
result = new ConfigVerificationResult.Builder()
.verificationStepName("Provide Google Credentials")
.outcome(Outcome.SUCCESSFUL)
- .explanation(String.format("Successfully provided [%s] as
Google Credentials", credentials.getClass().getSimpleName()))
+ .explanation("Successfully provided [%s] as Google
Credentials".formatted(credentials.getClass().getSimpleName()))
.build();
} catch (final IOException e) {
result = new ConfigVerificationResult.Builder()
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
index f4fa86f59e3..29d279ca060 100644
---
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -12,5 +12,6 @@
# 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.
+org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider
org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService
org.apache.nifi.processors.gcp.storage.GCSFileResourceService
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider/additionalDetails.md
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider/additionalDetails.md
new file mode 100644
index 00000000000..6898fb5d2d9
--- /dev/null
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.cloudsql.GcpCloudSqlIamDatabasePasswordProvider/additionalDetails.md
@@ -0,0 +1,91 @@
+<!--
+ 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.
+-->
+
+## Summary
+
+`GcpCloudSqlIamDatabasePasswordProvider` generates a short-lived Cloud SQL IAM
login token and supplies it as the
+database password for a DBCP service. Use it when you want NiFi to connect to
Cloud SQL without storing a long-lived
+database password in NiFi.
+
+The provider works with Cloud SQL for PostgreSQL and Cloud SQL for MySQL. When
a DBCP service references this provider,
+the static DBCP password property is ignored.
+
+## Usage
+
+1. Configure `GCPCredentialsControllerService` so NiFi can obtain Google
credentials.
+2. Create and enable `GcpCloudSqlIamDatabasePasswordProvider`.
+3. Set **GCP Credentials Provider Service** to the credentials service.
+4. Configure the DBCP service with the JDBC URL, driver, database user, and
**Database Password Provider** set to
+ `GcpCloudSqlIamDatabasePasswordProvider`.
+5. Run **Verify** on the provider, then run **Verify** on the DBCP service.
+
+Create the Cloud SQL IAM database user separately and grant the database
privileges required by your application.
+
+## Workload Identity Federation
+
+For the Snowflake Workload Identity Federation configuration tested with this
provider, configure
+`GCPCredentialsControllerService` with **Target Service Account**. The
workload identity principal must also have
+`roles/iam.workloadIdentityUser` on that target service account. Other Google
credential configurations can be used
+when they can acquire a token accepted by Cloud SQL for the configured
database user.
+
+## PostgreSQL Configuration
+
+Cloud SQL for PostgreSQL expects the DBCP **Database User** to match the IAM
identity used for login. Configure the
+DBCP service with a PostgreSQL JDBC driver and PostgreSQL JDBC URL. Configure
TLS in the DBCP service as required for
+your environment.
+
+| Setting | Value |
+|---|---|
+| Driver Class Name | `org.postgresql.Driver` |
+| JDBC URL | `jdbc:postgresql://<HOST>:5432/<DATABASE>?sslmode=require` |
+| Database User for Google user | full email address |
+| Database User for service account | service-account email without
`.gserviceaccount.com` |
+
+Example service-account mapping:
+
+- Target service account: `[email protected]`
+- DBCP **Database User**: `[email protected]`
+
+## MySQL Configuration
+
+Cloud SQL for MySQL uses the full service-account email when the IAM database
user is created, but the JDBC login name
+must be only the portion before `@`. Configure the DBCP service with a
compatible MySQL Connector/J driver and MySQL
+JDBC URL. Configure TLS in the DBCP service as required for your environment.
+
+| Setting | Value |
+|---|---|
+| Driver Class Name | `com.mysql.cj.jdbc.Driver` |
+| Driver Location(s) | compatible MySQL Connector/J driver jar provided to the
DBCP service |
+| JDBC URL | `jdbc:mysql://<HOST>:3306/<DATABASE>?sslMode=REQUIRED` |
+| Database User | service-account identifier before `@` |
+
+Example service-account mapping:
+
+- IAM database user created in Cloud SQL:
`[email protected]`
+- DBCP **Database User**: `nifi-sa`
+
+## Verify and Troubleshooting
+
+`GcpCloudSqlIamDatabasePasswordProvider` **Verify** checks that NiFi can
obtain a token. DBCP **Verify** checks the
+actual database connection using the configured URL, driver, TLS settings,
database user, and password provider.
+
+If provider **Verify** fails:
+
+- Confirm the referenced `GCPCredentialsControllerService` is enabled.
+- For Workload Identity Federation using service-account impersonation,
confirm **Target Service Account** is set and
+ the workload identity principal has `roles/iam.workloadIdentityUser` on that
service account.
+
+If provider **Verify** succeeds but DBCP **Verify** fails, token acquisition
is working and the problem is in the JDBC
+connection configuration, network path, TLS settings, driver setup, database
user, or database privileges.
\ No newline at end of file
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService/additionalDetails.md
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService/additionalDetails.md
index fc8d0b6acf5..25fdbd89923 100644
---
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService/additionalDetails.md
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/resources/docs/org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService/additionalDetails.md
@@ -108,8 +108,8 @@ Record the audience string printed by the command; it must
be copied into NiFi
### 2. Authorize the workload identity principal for Google Cloud resources
-The STS-issued access token represents the workload identity principal itself.
Grant IAM roles to that identity on
-projects or specific resources:
+When **Target Service Account** is left blank, the STS-issued access token
represents the workload identity principal
+itself. Grant IAM roles to that identity on projects or specific resources:
```bash
# Project scoped
@@ -123,8 +123,17 @@ gcloud storage buckets add-iam-policy-binding
gs://MY_BUCKET \
--role="roles/storage.objectViewer"
```
-`IDENTITY_SUBJECT` must match the claim you mapped in the provider (for
example `assertion.sub`). Service-account
-impersonation is not yet supported, so grant roles directly to the workload
identity principal.
+`IDENTITY_SUBJECT` must match the claim you mapped in the provider (for
example `assertion.sub`).
+
+When you set **Target Service Account**, also grant
`roles/iam.workloadIdentityUser` on that service account to the
+same workload identity subject so Google can perform service-account
impersonation:
+
+```bash
+gcloud iam service-accounts add-iam-policy-binding \
+ "<TARGET_SERVICE_ACCOUNT_EMAIL>" \
+ --role="roles/iam.workloadIdentityUser" \
+
--member="principal://iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/nifi-pool/subject/IDENTITY_SUBJECT"
+```
### 3. Configure NiFi properties (Workload Identity strategy selected)
@@ -135,26 +144,29 @@ impersonation is not yet supported, so grant roles
directly to the workload iden
| **STS Token Endpoint** | Optional override for the Google STS endpoint;
leave blank to use `https://sts.googleapis.com/v1/token`. |
| **Subject Token Provider** | Controller Service that retrieves the upstream
workload identity token (JWT or access token). The token must contain the
claims referenced by your attribute mapping. |
| **Subject Token Type** | Defaults to `urn:ietf:params:oauth:token-type:jwt`.
Choose the alternate access-token type only when the upstream provider issues
OAuth access tokens instead of JWTs. |
+| **Target Service Account** | Optional. Set this only when downstream
components must impersonate a Google service account. When configured, the
controller service returns generic impersonated credentials and dependent
components apply any service-specific scopes when they first use the
credential. |
| **Proxy Configuration Service** | Optional controller service allowing NiFi
to reach STS through HTTP/SOCKS proxies. |
-Once these properties are set, enable GCPCredentialsControllerService.
Processors referencing it immediately obtain
-`IdentityPoolCredentials`, and Google’s libraries refresh access tokens
automatically using the configured subject
--token provider.
+Once these properties are set, enable GCPCredentialsControllerService. When
**Target Service Account** is blank,
+processors and controller services reference direct `IdentityPoolCredentials`.
When **Target Service Account** is set,
+they reference impersonated credentials layered over the workload identity
source credential.
### Verification workflow
1. Enable or refresh the Subject Token Provider controller service.
-2. Use the **Verify** action on GCPCredentialsControllerService. Successful
verification confirms that NiFi can
- exchange the subject token with Google STS using the configured proxy,
audience, and scopes.
-3. Enable dependent processors. No additional controller services are required.
+2. Use the **Verify** action on GCPCredentialsControllerService. Successful
verification confirms only that NiFi can
+ construct the configured credential object from the current properties.
+3. Verify does not perform STS exchange, service-account impersonation, or
network reachability checks. The first token
+ refresh happens when a dependent processor or controller service actually
uses the credential.
+4. Enable the dependent components that reference this controller service.
Product-specific validation happens there.
### Troubleshooting
| Symptom | Guidance |
| --- | --- |
-| `403 Caller does not have storage.objects.list` | Confirm the workload
identity principal has the required IAM role: `gcloud projects get-iam-policy`
/ `gcloud storage buckets get-iam-policy`. Ensure the attribute mapping emits
the same subject referenced in IAM. |
-| STS errors during verification | Double-check the **Audience** string and
**STS Token Endpoint**. Use DEBUG logs or the Verify dialog output to inspect
the STS response. Ensure the subject token includes the mapped claims. |
-| Access token rejected by Google APIs | Call the API directly with the
federated token (for example, `curl -H "Authorization: Bearer TOKEN"
https://storage.googleapis.com/...`). If it still fails, revisit IAM bindings
or scope selection. |
+| `403 Caller does not have storage.objects.list` | Confirm the active
identity has the required IAM role. Without **Target Service Account**, grant
roles directly to the workload identity principal. With **Target Service
Account**, confirm both the impersonated service account permissions and the
`roles/iam.workloadIdentityUser` binding for the workload identity subject. |
+| Verify succeeds but the dependent component fails on first use | Verify only
constructs credentials. Recheck the **Audience**, **STS Token Endpoint**,
upstream subject token, proxy reachability, and any optional **Target Service
Account** setting at the dependent component that triggers the first refresh. |
+| Access token rejected by Google APIs | Revisit IAM bindings and scope
selection for the dependent component. If the component requires a
service-specific scope or impersonated identity, confirm that **Target Service
Account** is set and that the downstream component supports the needed scope. |
| Need to rotate upstream tokens | The controller service requests a fresh
subject token 60 seconds before expiry. Trigger **Refresh** on the Subject
Token Provider to invalidate cached tokens immediately. |
---
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProviderTest.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProviderTest.java
new file mode 100644
index 00000000000..024e4b2a6eb
--- /dev/null
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/cloudsql/GcpCloudSqlIamDatabasePasswordProviderTest.java
@@ -0,0 +1,669 @@
+/*
+ * 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.Collection;
+import java.util.Collections;
+import java.util.Date;
+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_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 testVerifyIdentityPoolCredentialsAcquireToken() 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(2, results.size());
+ assertVerificationResult(results.get(0), VERIFY_SCOPE_STEP,
SUCCESSFUL, "Cloud SQL login scope");
+ assertVerificationResult(results.get(1), VERIFY_TOKEN_STEP,
SUCCESSFUL, CLOUD_SQL_IAM);
+ Mockito.verify(scopedCredentials).refreshAccessToken();
+ }
+
+ @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));
+ assertVerificationExceptionLogged(
+ runner.getControllerServiceLogger(PASSWORD_PROVIDER_ID),
+ "Failed to resolve scoped Google credentials",
+ 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));
+ assertVerificationExceptionLogged(
+ runner.getControllerServiceLogger(PASSWORD_PROVIDER_ID),
+ "Failed to acquire Cloud SQL IAM access token",
+ 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 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,
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, Collections.newSetFromMap(new IdentityHashMap<>()));
+ }
+ }
+
+ private static void assertVerificationExceptionLogged(final
MockComponentLog logger, final String message, final String exceptionMessage) {
+ assertTrue(logger.getErrorMessages().stream().anyMatch(logMessage -> {
+ final Object[] arguments = logMessage.getArgs();
+ final Throwable throwable = logMessage.getThrowable();
+ final boolean throwableMatched = throwable != null &&
exceptionMessage.equals(throwable.getMessage());
+ final boolean argumentMatched = arguments != null
+ && List.of(arguments).stream()
+ .filter(Throwable.class::isInstance)
+ .map(Throwable.class::cast)
+ .anyMatch(argument ->
exceptionMessage.equals(argument.getMessage()));
+ return logMessage.getMsg().contains(message) && (throwableMatched
|| argumentMatched);
+ }));
+ }
+
+ private static void assertThrowableChainDoesNotContain(final Throwable
throwable, final String value, final Set<Throwable> visited) {
+ if (throwable == null || !visited.add(throwable)) {
+ return;
+ }
+
+ final String message = throwable.getMessage();
+ assertFalse(message != null && message.contains(value));
+
+ for (final Throwable suppressed : throwable.getSuppressed()) {
+ assertThrowableChainDoesNotContain(suppressed, value, visited);
+ }
+
+ assertThrowableChainDoesNotContain(throwable.getCause(), value,
visited);
+ }
+
+ private static GoogleCredentials getScopedCredentials(final
GcpCloudSqlIamDatabasePasswordProvider provider) throws
ReflectiveOperationException {
+ final Field field =
GcpCloudSqlIamDatabasePasswordProvider.class.getDeclaredField("scopedCredentials");
+ field.setAccessible(true);
+ return (GoogleCredentials) field.get(provider);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<PropertyDescriptor>
getSupportedPropertyDescriptors(final GcpCloudSqlIamDatabasePasswordProvider
provider)
+ throws ReflectiveOperationException {
+ final Method method =
GcpCloudSqlIamDatabasePasswordProvider.class.getDeclaredMethod("getSupportedPropertyDescriptors");
+ method.setAccessible(true);
+ return (List<PropertyDescriptor>) method.invoke(provider);
+ }
+
+ private static final class TestGCPCredentialsService extends
AbstractControllerService implements GCPCredentialsService {
+ private final GoogleCredentials googleCredentials;
+
+ private TestGCPCredentialsService(final GoogleCredentials
googleCredentials) {
+ this.googleCredentials = googleCredentials;
+ }
+
+ @Override
+ public GoogleCredentials getGoogleCredentials() {
+ return googleCredentials;
+ }
+ }
+
+ private static final class RootGoogleCredentials extends GoogleCredentials
{
+ private final AtomicInteger createScopedCount = new AtomicInteger();
+ private final GoogleCredentials[] scopedCredentials;
+ private final RuntimeException createScopedException;
+ private final AtomicInteger scopedCredentialIndex = new
AtomicInteger();
+ private volatile List<String> lastRequestedScopes = List.of();
+
+ private RootGoogleCredentials(final GoogleCredentials
scopedCredentials) {
+ this.scopedCredentials = new
GoogleCredentials[]{scopedCredentials};
+ this.createScopedException = null;
+ }
+
+ private RootGoogleCredentials(final RuntimeException
createScopedException) {
+ this.scopedCredentials = new GoogleCredentials[0];
+ this.createScopedException = createScopedException;
+ }
+
+ private RootGoogleCredentials(final GoogleCredentials
firstScopedCredentials, final GoogleCredentials secondScopedCredentials) {
+ this.scopedCredentials = new
GoogleCredentials[]{firstScopedCredentials, secondScopedCredentials};
+ this.createScopedException = null;
+ }
+
+ @Override
+ public GoogleCredentials createScoped(final Collection<String> scopes)
{
+ createScopedCount.incrementAndGet();
+ lastRequestedScopes = List.copyOf(scopes);
+ if (createScopedException != null) {
+ throw createScopedException;
+ }
+
+ final int index =
Math.min(scopedCredentialIndex.getAndIncrement(), scopedCredentials.length - 1);
+ return scopedCredentials.length == 0 ? null :
scopedCredentials[index];
+ }
+
+ private int getCreateScopedCount() {
+ return createScopedCount.get();
+ }
+
+ private List<String> getLastRequestedScopes() {
+ return lastRequestedScopes;
+ }
+ }
+
+ private static class TestScopedGoogleCredentials extends GoogleCredentials
{
+ private final AtomicInteger refreshAccessTokenCount = new
AtomicInteger();
+ private volatile AccessToken refreshedAccessToken;
+ private volatile IOException refreshException;
+
+ private TestScopedGoogleCredentials(final AccessToken
initialAccessToken) {
+ super(initialAccessToken);
+ }
+
+ @Override
+ public AccessToken refreshAccessToken() throws IOException {
+ refreshAccessTokenCount.incrementAndGet();
+ if (refreshException != null) {
+ throw refreshException;
+ }
+ return refreshedAccessToken;
+ }
+
+ protected void setRefreshedAccessToken(final AccessToken
refreshedAccessToken) {
+ this.refreshedAccessToken = refreshedAccessToken;
+ }
+
+ protected void setRefreshException(final IOException refreshException)
{
+ this.refreshException = refreshException;
+ }
+
+ protected int getRefreshAccessTokenCount() {
+ return refreshAccessTokenCount.get();
+ }
+ }
+
+ private static ImpersonatedCredentials impersonatedCredentials(final
AccessToken accessToken) throws IOException {
+ final ImpersonatedCredentials credentials =
mock(ImpersonatedCredentials.class);
+ when(credentials.refreshAccessToken()).thenReturn(accessToken);
+ return credentials;
+ }
+
+ private static ImpersonatedCredentials impersonatedCredentials(final
IOException exception) throws IOException {
+ final ImpersonatedCredentials credentials =
mock(ImpersonatedCredentials.class);
+ when(credentials.refreshAccessToken()).thenThrow(exception);
+ return credentials;
+ }
+
+ private static IdentityPoolCredentials identityPoolCredentials(final
AccessToken accessToken) throws IOException {
+ final IdentityPoolCredentials credentials =
mock(IdentityPoolCredentials.class);
+ when(credentials.refreshAccessToken()).thenReturn(accessToken);
+ return credentials;
+ }
+
+ private static final class BlockingScopedGoogleCredentials extends
TestScopedGoogleCredentials {
+ private final CountDownLatch refreshEnteredLatch = new
CountDownLatch(1);
+ private final CountDownLatch releaseRefreshLatch = new
CountDownLatch(1);
+
+ private BlockingScopedGoogleCredentials() {
+ super(null);
+ }
+
+ @Override
+ public AccessToken refreshAccessToken() throws IOException {
+ refreshEnteredLatch.countDown();
+ try {
+ if (!releaseRefreshLatch.await(5, TimeUnit.SECONDS)) {
+ throw new IOException("Timed out waiting for refresh
release");
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while waiting for refresh
release", e);
+ }
+ return super.refreshAccessToken();
+ }
+
+ private boolean awaitRefreshEntry() throws InterruptedException {
+ return refreshEnteredLatch.await(5, TimeUnit.SECONDS);
+ }
+
+ private void releaseRefresh() {
+ releaseRefreshLatch.countDown();
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategyTest.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategyTest.java
new file mode 100644
index 00000000000..be4d13ff010
--- /dev/null
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/strategies/WorkloadIdentityFederationCredentialsStrategyTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.credentials.factory.strategies;
+
+import com.google.api.client.http.HttpTransport;
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.auth.http.HttpTransportFactory;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.IdentityPoolCredentials;
+import com.google.auth.oauth2.ImpersonatedCredentials;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.oauth2.AccessToken;
+import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
+import
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class WorkloadIdentityFederationCredentialsStrategyTest {
+ private static final String AUDIENCE =
"projects/123456789/locations/global/workloadIdentityPools/pool/providers/provider";
+ private static final String SCOPE =
"https://www.googleapis.com/auth/cloud-platform";;
+ private static final String TOKEN_ENDPOINT =
"https://sts.googleapis.com/v1/token";;
+ private static final String SUBJECT_TOKEN_TYPE =
"urn:ietf:params:oauth:token-type:jwt";
+ private static final String TARGET_SERVICE_ACCOUNT =
"[email protected]";
+ private static final String SQLSERVICE_LOGIN_SCOPE =
"https://www.googleapis.com/auth/sqlservice.login";;
+
+ private final WorkloadIdentityFederationCredentialsStrategy strategy = new
WorkloadIdentityFederationCredentialsStrategy();
+
+ @Test
+ void
testReturnsIdentityPoolCredentialsWhenTargetServiceAccountNotConfigured()
throws IOException {
+ final ConfigurationContext context = mockConfigurationContext(null);
+
+ final GoogleCredentials credentials =
strategy.getGoogleCredentials(context, transportFactory());
+
+ assertInstanceOf(IdentityPoolCredentials.class, credentials);
+ }
+
+ @Test
+ void
testReturnsImpersonatedCredentialsWithEmptyScopesAndPreservedTransportFactory()
throws IOException {
+ final HttpTransportFactory transportFactory = transportFactory();
+ final ConfigurationContext context =
mockConfigurationContext(TARGET_SERVICE_ACCOUNT);
+
+ final GoogleCredentials credentials =
strategy.getGoogleCredentials(context, transportFactory);
+
+ final ImpersonatedCredentials impersonatedCredentials =
assertInstanceOf(ImpersonatedCredentials.class, credentials);
+ assertEquals(TARGET_SERVICE_ACCOUNT,
impersonatedCredentials.getAccount());
+ assertInstanceOf(IdentityPoolCredentials.class,
impersonatedCredentials.getSourceCredentials());
+ assertTrue(impersonatedCredentials.toBuilder().getScopes().isEmpty());
+ assertTrue(impersonatedCredentials.createScopedRequired());
+ assertSame(transportFactory,
impersonatedCredentials.toBuilder().getHttpTransportFactory());
+
+ final GoogleCredentials scopedCredentials =
credentials.createScoped(List.of(SQLSERVICE_LOGIN_SCOPE));
+
+ final ImpersonatedCredentials scopedImpersonatedCredentials =
assertInstanceOf(ImpersonatedCredentials.class, scopedCredentials);
+ assertEquals(TARGET_SERVICE_ACCOUNT,
scopedImpersonatedCredentials.getAccount());
+ assertSame(impersonatedCredentials.getSourceCredentials(),
scopedImpersonatedCredentials.getSourceCredentials());
+ assertEquals(List.of(SQLSERVICE_LOGIN_SCOPE),
scopedImpersonatedCredentials.toBuilder().getScopes());
+ assertFalse(scopedImpersonatedCredentials.createScopedRequired());
+ assertSame(transportFactory,
scopedImpersonatedCredentials.toBuilder().getHttpTransportFactory());
+ }
+
+ private ConfigurationContext mockConfigurationContext(final String
targetServiceAccount) {
+ final ConfigurationContext context = mock(ConfigurationContext.class);
+ final PropertyValue audiencePropertyValue =
stringPropertyValue(AUDIENCE);
+ final PropertyValue scopePropertyValue = stringPropertyValue(SCOPE);
+ final PropertyValue tokenEndpointPropertyValue =
stringPropertyValue(TOKEN_ENDPOINT);
+ final PropertyValue subjectTokenTypePropertyValue =
stringPropertyValue(SUBJECT_TOKEN_TYPE);
+ final PropertyValue targetServiceAccountPropertyValue =
stringPropertyValue(targetServiceAccount);
+
+
when(context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_AUDIENCE)).thenReturn(audiencePropertyValue);
+
when(context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SCOPE)).thenReturn(scopePropertyValue);
+
when(context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_TOKEN_ENDPOINT)).thenReturn(tokenEndpointPropertyValue);
+
when(context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE)).thenReturn(subjectTokenTypePropertyValue);
+
when(context.getProperty(CredentialPropertyDescriptors.TARGET_SERVICE_ACCOUNT)).thenReturn(targetServiceAccountPropertyValue);
+
+ final PropertyValue subjectTokenProviderProperty =
mock(PropertyValue.class);
+
when(subjectTokenProviderProperty.asControllerService(OAuth2AccessTokenProvider.class)).thenReturn(new
MockOAuth2AccessTokenProvider());
+
when(context.getProperty(CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SUBJECT_TOKEN_PROVIDER)).thenReturn(subjectTokenProviderProperty);
+ return context;
+ }
+
+ private PropertyValue stringPropertyValue(final String value) {
+ final PropertyValue propertyValue = mock(PropertyValue.class);
+ when(propertyValue.getValue()).thenReturn(value);
+ return propertyValue;
+ }
+
+ private HttpTransportFactory transportFactory() {
+ final HttpTransport transport = new NetHttpTransport();
+ return () -> transport;
+ }
+
+ private static final class MockOAuth2AccessTokenProvider extends
AbstractControllerService implements OAuth2AccessTokenProvider {
+ @Override
+ public AccessToken getAccessDetails() {
+ final AccessToken accessToken = new AccessToken();
+ accessToken.setAccessToken("subject-token");
+ accessToken.setExpiresIn(3600L);
+ return accessToken;
+ }
+
+ @Override
+ public void refreshAccessDetails() {
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
index 538374668ad..890d41757d6 100644
---
a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
+++
b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
@@ -16,9 +16,12 @@
*/
package org.apache.nifi.processors.gcp.credentials.service;
+import com.google.auth.oauth2.ComputeEngineCredentials;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.auth.oauth2.IdentityPoolCredentials;
+import com.google.auth.oauth2.ImpersonatedCredentials;
import com.google.auth.oauth2.ServiceAccountCredentials;
+import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.gcp.credentials.service.GCPCredentialsService;
import org.apache.nifi.oauth2.AccessToken;
@@ -30,19 +33,32 @@ import org.junit.jupiter.api.Test;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.AUTHENTICATION_STRATEGY;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE;
+import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.TARGET_SERVICE_ACCOUNT;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_AUDIENCE;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SCOPE;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SUBJECT_TOKEN_PROVIDER;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE;
import static
org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.WORKLOAD_IDENTITY_TOKEN_ENDPOINT;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class GCPCredentialsServiceTest {
+ private static final String WORKLOAD_IDENTITY_AUDIENCE_VALUE =
"projects/123456789/locations/global/workloadIdentityPools/pool/providers/provider";
+ private static final String WORKLOAD_IDENTITY_SCOPE_VALUE =
"https://www.googleapis.com/auth/cloud-platform";;
+ private static final String WORKLOAD_IDENTITY_TOKEN_ENDPOINT_VALUE =
"https://sts.googleapis.com/v1/token";;
+ private static final String WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE_VALUE =
"urn:ietf:params:oauth:token-type:jwt";
+ private static final String TARGET_SERVICE_ACCOUNT_VALUE =
"[email protected]";
+
@Test
public void testToString() throws Exception {
// toString method shouldn't cause an exception
@@ -121,12 +137,7 @@ public class GCPCredentialsServiceTest {
runner.addControllerService("subjectTokenProvider",
subjectTokenProvider);
runner.enableControllerService(subjectTokenProvider);
- runner.setProperty(serviceImpl, AUTHENTICATION_STRATEGY,
AuthenticationStrategy.WORKLOAD_IDENTITY_FEDERATION.getValue());
- runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_AUDIENCE,
"projects/123456789/locations/global/workloadIdentityPools/pool/providers/provider");
- runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_SCOPE,
"https://www.googleapis.com/auth/cloud-platform";);
- runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_TOKEN_ENDPOINT,
"https://sts.googleapis.com/v1/token";);
- runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE,
"urn:ietf:params:oauth:token-type:jwt");
- runner.setProperty(serviceImpl,
WORKLOAD_IDENTITY_SUBJECT_TOKEN_PROVIDER, "subjectTokenProvider");
+ configureWorkloadIdentityFederation(runner, serviceImpl);
runner.enableControllerService(serviceImpl);
runner.assertValid(serviceImpl);
@@ -140,6 +151,44 @@ public class GCPCredentialsServiceTest {
assertEquals(IdentityPoolCredentials.class, credentials.getClass());
}
+ @Test
+ public void testWorkloadIdentityFederationImpersonationCredentials()
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,
TARGET_SERVICE_ACCOUNT_VALUE);
+
+ runner.enableControllerService(serviceImpl);
+ runner.assertValid(serviceImpl);
+
+ final GCPCredentialsService service = (GCPCredentialsService)
runner.getProcessContext()
+
.getControllerServiceLookup().getControllerService("gcpCredentialsProvider");
+
+ assertNotNull(service);
+ final GoogleCredentials credentials = service.getGoogleCredentials();
+ assertNotNull(credentials);
+ final ImpersonatedCredentials impersonatedCredentials =
assertInstanceOf(ImpersonatedCredentials.class, credentials);
+ assertEquals(TARGET_SERVICE_ACCOUNT_VALUE,
impersonatedCredentials.getAccount());
+ assertInstanceOf(IdentityPoolCredentials.class,
impersonatedCredentials.getSourceCredentials());
+ assertTrue(impersonatedCredentials.createScopedRequired());
+
+ final GoogleCredentials scopedCredentials =
credentials.createScoped(List.of("https://www.googleapis.com/auth/sqlservice.login";));
+
+ final ImpersonatedCredentials scopedImpersonatedCredentials =
assertInstanceOf(ImpersonatedCredentials.class, scopedCredentials);
+ assertEquals(TARGET_SERVICE_ACCOUNT_VALUE,
scopedImpersonatedCredentials.getAccount());
+ assertEquals(impersonatedCredentials.getSourceCredentials(),
scopedImpersonatedCredentials.getSourceCredentials());
+
assertEquals(List.of("https://www.googleapis.com/auth/sqlservice.login";),
scopedImpersonatedCredentials.toBuilder().getScopes());
+ assertFalse(scopedImpersonatedCredentials.createScopedRequired());
+
assertNotNull(impersonatedCredentials.toBuilder().getHttpTransportFactory());
+
assertEquals(impersonatedCredentials.toBuilder().getHttpTransportFactory(),
scopedImpersonatedCredentials.toBuilder().getHttpTransportFactory());
+ }
+
@Test
public void testBadFileCredentials() throws Exception {
final TestRunner runner =
TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
@@ -166,6 +215,56 @@ public class GCPCredentialsServiceTest {
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 testRawJsonCredentials() throws Exception {
final String jsonRead = new String(
@@ -194,6 +293,28 @@ public class GCPCredentialsServiceTest {
"Credentials class should be equal");
}
+ private static void configureWorkloadIdentityFederation(final TestRunner
runner, final GCPCredentialsControllerService serviceImpl) {
+ runner.setProperty(serviceImpl, AUTHENTICATION_STRATEGY,
AuthenticationStrategy.WORKLOAD_IDENTITY_FEDERATION.getValue());
+ runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_AUDIENCE,
WORKLOAD_IDENTITY_AUDIENCE_VALUE);
+ runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_SCOPE,
WORKLOAD_IDENTITY_SCOPE_VALUE);
+ runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_TOKEN_ENDPOINT,
WORKLOAD_IDENTITY_TOKEN_ENDPOINT_VALUE);
+ runner.setProperty(serviceImpl, WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE,
WORKLOAD_IDENTITY_SUBJECT_TOKEN_TYPE_VALUE);
+ runner.setProperty(serviceImpl,
WORKLOAD_IDENTITY_SUBJECT_TOKEN_PROVIDER, "subjectTokenProvider");
+ }
+
+ private static void assertHasInvalidResult(final
Collection<ValidationResult> validationResults, final String
explanationFragment) {
+ final List<String> explanations = new
ArrayList<>(validationResults.size());
+ for (final ValidationResult validationResult : validationResults) {
+ if (!validationResult.isValid()) {
+ explanations.add(validationResult.getExplanation());
+ }
+ }
+
+ final boolean explanationFound = explanations.stream()
+ .anyMatch(explanation -> explanation != null &&
explanation.contains(explanationFragment));
+ assertTrue(explanationFound, () -> "Expected invalid result containing
[%s] but found %s".formatted(explanationFragment, explanations));
+ }
+
private static final class MockOAuth2AccessTokenProvider extends
AbstractControllerService implements OAuth2AccessTokenProvider {
private static final String ACCESS_TOKEN_VALUE =
"federated-access-token";
private static final long EXPIRES_IN_SECONDS = 3600;
