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 d9b2a406c054..acf145127e18 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 @@ 2.12.0-SNAPSHOT provided + + org.apache.nifi + nifi-dbcp-service-api + provided + org.apache.nifi nifi-resource-transfer 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 000000000000..fb886529f9db --- /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 PROPERTY_DESCRIPTORS = List.of( + GCP_CREDENTIALS_PROVIDER_SERVICE + ); + + private volatile GoogleCredentials scopedCredentials; + + @Override + protected List 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 verify(final ConfigurationContext context, final ComponentLog verificationLogger, + final Map attributes) { + final List 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 c0c0f8014911..24a81b118d90 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 @@ private 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 dfe261e895a5..b25e790517f5 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.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 GoogleCredentials getGoogleCredentials(final ConfigurationContext context 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 scopes = parseScopes(scopeValue); final IdentityPoolSubjectTokenSupplier tokenSupplier = createSubjectTokenSupplier(subjectTokenProvider); @@ -77,7 +79,20 @@ public GoogleCredentials getGoogleCredentials(final ConfigurationContext context 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 25cda97bc5de..c90388ef8f95 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 @@ 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.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 GoogleCredentials getGoogleCredentials() throws ProcessException { protected Collection customValidate(final ValidationContext validationContext) { final List 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 List verify(final ConfigurationContext context, 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 f4fa86f59e35..29d279ca0609 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 000000000000..6898fb5d2d94 --- /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 @@ + + +## 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://:5432/?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: `nifi-sa@my-project.iam.gserviceaccount.com` +- DBCP **Database User**: `nifi-sa@my-project.iam` + +## 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://:3306/?sslMode=REQUIRED` | +| Database User | service-account identifier before `@` | + +Example service-account mapping: + +- IAM database user created in Cloud SQL: `nifi-sa@my-project.iam.gserviceaccount.com` +- 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 fc8d0b6acf55..25fdbd89923e 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 \ + "" \ + --role="roles/iam.workloadIdentityUser" \ + --member="principal://iam.googleapis.com/projects//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 000000000000..024e4b2a6eb2 --- /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 = "service-account@test-project.iam"; + 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 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 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 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 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 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 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 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 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 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 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 first = executorService.submit(() -> getPasswordAfterStart(provider, startLatch)); + final Future 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 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 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 getSupportedPropertyDescriptors(final GcpCloudSqlIamDatabasePasswordProvider provider) + throws ReflectiveOperationException { + final Method method = GcpCloudSqlIamDatabasePasswordProvider.class.getDeclaredMethod("getSupportedPropertyDescriptors"); + method.setAccessible(true); + return (List) 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 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 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 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 000000000000..be4d13ff010a --- /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 = "target-account@test-project.iam.gserviceaccount.com"; + 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 538374668ad6..890d41757d64 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 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 = "target-account@test-project.iam.gserviceaccount.com"; + @Test public void testToString() throws Exception { // toString method shouldn't cause an exception @@ -121,12 +137,7 @@ public void testWorkloadIdentityFederationCredentials() throws Exception { 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 void testWorkloadIdentityFederationCredentials() throws Exception { 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 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 testRawJsonCredentials() throws Exception { final String jsonRead = new String( @@ -194,6 +293,28 @@ public void testRawJsonCredentials() throws Exception { "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 validationResults, final String explanationFragment) { + final List 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;