Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@
<artifactId>nifi-oauth2-provider-api</artifactId>
</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-service-utils</artifactId>
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,215 @@
/*
* 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.services.azure;

import com.azure.core.credential.AccessToken;
import com.azure.core.credential.TokenCredential;
import com.azure.core.credential.TokenRequestContext;
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.logging.ComponentLog;
import org.apache.nifi.processor.exception.ProcessException;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Tags({"azure", "microsoft entra", "jdbc", "database password", "postgresql", "mysql", "authentication"})
@CapabilityDescription("""
Acquires short-lived Microsoft Entra database passwords for JDBC authentication to Azure Database for PostgreSQL Flexible Server and
Azure Database for MySQL Flexible Server. Only Azure Database for PostgreSQL Flexible Server and Azure Database for MySQL Flexible
Server are supported.
""")
public class AzureEntraDatabasePasswordProvider extends AbstractControllerService
implements DatabasePasswordProvider, VerifiableControllerService {

static final String OSS_RDBMS_SCOPE = "https://ossrdbms-aad.database.windows.net/.default";
static final String FAILED_CREDENTIAL_RESOLUTION_MESSAGE = "Failed to resolve Azure credentials for Microsoft Entra database password.";
static final String FAILED_TOKEN_ACQUISITION_MESSAGE = "Failed to acquire a valid Microsoft Entra database access token.";
static final String VERIFY_CREDENTIALS_STEP = "Resolve Azure credentials";
static final String VERIFY_TOKEN_STEP = "Acquire Microsoft Entra database access token";
static final String VERIFY_CREDENTIALS_UNAVAILABLE = "Configured Azure Credentials Service did not return Azure credentials.";
static final String VERIFY_TOKEN_ACQUISITION_FAILED = "Failed to acquire a valid Microsoft Entra database access token.";
static final Duration DEFAULT_TOKEN_ACQUISITION_TIMEOUT = Duration.ofMinutes(2);

static final PropertyDescriptor AZURE_CREDENTIALS_SERVICE = new PropertyDescriptor.Builder()
.name("Azure Credentials Service")
.description("Controller Service that provides the Azure credentials used to request Microsoft Entra database access tokens.")
.identifiesControllerService(AzureCredentialsService.class)
.required(true)
.build();

private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(
AZURE_CREDENTIALS_SERVICE
);

private final Duration tokenAcquisitionTimeout;
private volatile AzureCredentialsService azureCredentialsService;

public AzureEntraDatabasePasswordProvider() {
this(DEFAULT_TOKEN_ACQUISITION_TIMEOUT);
}

AzureEntraDatabasePasswordProvider(final Duration tokenAcquisitionTimeout) {
this.tokenAcquisitionTimeout = requirePositiveDuration(tokenAcquisitionTimeout);
}

@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return PROPERTY_DESCRIPTORS;
}

@OnEnabled
public void onEnabled(final ConfigurationContext context) {
azureCredentialsService = resolveAzureCredentialsService(context);
}

@OnDisabled
public void onDisabled() {
azureCredentialsService = null;
}

@Override
public char[] getPassword(final DatabasePasswordRequestContext requestContext) {
Objects.requireNonNull(requestContext, "Database Password Request Context required");

final AzureCredentialsService configuredCredentialsService = azureCredentialsService;
if (configuredCredentialsService == null) {
throw new ProcessException(FAILED_CREDENTIAL_RESOLUTION_MESSAGE);
}

final TokenCredential credential;
try {
credential = configuredCredentialsService.getCredentials();
} catch (final RuntimeException e) {
throw new ProcessException(FAILED_CREDENTIAL_RESOLUTION_MESSAGE);
}

if (credential == null) {
throw new ProcessException(FAILED_CREDENTIAL_RESOLUTION_MESSAGE);
}

final AccessToken accessToken;
try {
accessToken = credential.getToken(createTokenRequestContext()).block(tokenAcquisitionTimeout);
} catch (final RuntimeException e) {
throw new ProcessException(FAILED_TOKEN_ACQUISITION_MESSAGE);
}

if (!isValidAccessToken(accessToken)) {
throw new ProcessException(FAILED_TOKEN_ACQUISITION_MESSAGE);
}
Comment on lines +103 to +128

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All of these failures result in the same message, at minimum, it would be helpful to distinguish between credentials retrieval issues and Access Token issues.


return accessToken.getToken().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 TokenCredential verificationCredential = resolveVerificationCredential(context, verificationLogger);
if (verificationCredential == null) {
results.add(buildVerificationResult(VERIFY_CREDENTIALS_STEP, Outcome.FAILED, VERIFY_CREDENTIALS_UNAVAILABLE));
return results;
}

results.add(buildVerificationResult(
VERIFY_CREDENTIALS_STEP,
Outcome.SUCCESSFUL,
"Resolved Azure credentials service and current TokenCredential."
));
results.add(verifyAccessToken(verificationCredential, verificationLogger));
return results;
}

private TokenCredential resolveVerificationCredential(final ConfigurationContext context, final ComponentLog verificationLogger) {
final AzureCredentialsService verificationCredentialsService = resolveAzureCredentialsService(context);
if (verificationCredentialsService == null) {
return null;
}

try {
return verificationCredentialsService.getCredentials();
} catch (final RuntimeException e) {
verificationLogger.error(VERIFY_CREDENTIALS_UNAVAILABLE);
return null;
}
}

private ConfigVerificationResult verifyAccessToken(final TokenCredential credential, final ComponentLog verificationLogger) {
final AccessToken accessToken;
try {
accessToken = credential.getToken(createTokenRequestContext()).block(tokenAcquisitionTimeout);
} catch (final RuntimeException e) {
verificationLogger.error(VERIFY_TOKEN_ACQUISITION_FAILED);
return buildVerificationResult(VERIFY_TOKEN_STEP, Outcome.FAILED, VERIFY_TOKEN_ACQUISITION_FAILED);
}

if (!isValidAccessToken(accessToken)) {
verificationLogger.error(VERIFY_TOKEN_ACQUISITION_FAILED);
return buildVerificationResult(VERIFY_TOKEN_STEP, Outcome.FAILED, VERIFY_TOKEN_ACQUISITION_FAILED);
}

return buildVerificationResult(
VERIFY_TOKEN_STEP,
Outcome.SUCCESSFUL,
"Acquired a Microsoft Entra database access token. Use DBCP Verify to validate database connectivity."
);
}

private AzureCredentialsService resolveAzureCredentialsService(final ConfigurationContext context) {
return context.getProperty(AZURE_CREDENTIALS_SERVICE).asControllerService(AzureCredentialsService.class);
}

private TokenRequestContext createTokenRequestContext() {
return new TokenRequestContext().addScopes(OSS_RDBMS_SCOPE);
}

private Duration requirePositiveDuration(final Duration tokenAcquisitionTimeout) {
final Duration configuredTimeout = Objects.requireNonNull(tokenAcquisitionTimeout, "Token acquisition timeout required");
if (configuredTimeout.isZero() || configuredTimeout.isNegative()) {
throw new IllegalArgumentException("Token acquisition timeout must be positive");
}
return configuredTimeout;
}

private boolean isValidAccessToken(final AccessToken accessToken) {
return accessToken != null && StringUtils.isNotBlank(accessToken.getToken());
}

private ConfigVerificationResult buildVerificationResult(final String stepName, final Outcome outcome, final String explanation) {
return new ConfigVerificationResult.Builder()
.verificationStepName(stepName)
.outcome(outcome)
.explanation(explanation)
.build();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -22,5 +22,6 @@ org.apache.nifi.services.azure.storage.AzureStorageCredentialsControllerService_
org.apache.nifi.services.azure.storage.AzureStorageCredentialsControllerServiceLookup_v12
org.apache.nifi.services.azure.StandardAzureCredentialsControllerService
org.apache.nifi.services.azure.StandardAzureIdentityFederationTokenProvider
org.apache.nifi.services.azure.AzureEntraDatabasePasswordProvider
org.apache.nifi.services.azure.storage.AzureBlobStorageFileResourceService
org.apache.nifi.services.azure.storage.AzureDataLakeStorageFileResourceService
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
<!--
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

`AzureEntraDatabasePasswordProvider` acquires a short-lived Microsoft Entra access token and supplies it as the
database password for a DBCP service. Use it to connect NiFi to Azure Database for PostgreSQL Flexible Server or Azure
Database for MySQL Flexible Server without storing a long-lived database password in NiFi. Other database products and
sovereign Azure clouds are not supported.

The provider supplies a password when DBCP creates a new physical connection. Existing pooled connections are not
reauthenticated when the token expires.

## Usage

1. Configure an Azure Credentials Service that can obtain credentials for the public Azure cloud.
2. Create and enable `AzureEntraDatabasePasswordProvider`.
3. Set **Azure Credentials Service** to the configured credentials service.
4. Configure the DBCP service with the JDBC URL, driver, database user, and **Database Password Provider** set to
`AzureEntraDatabasePasswordProvider`.
5. Run **Verify** on the provider, then run **Verify** on the DBCP service.

Create the Microsoft Entra principal in the database separately and grant the database privileges required by NiFi.

## Workload Identity Federation

For Workload Identity Federation, configure `StandardAzureCredentialsControllerService` with **Credentials Strategy**
set to **Identity Federation** and select a `StandardAzureIdentityFederationTokenProvider`. The federation token
provider accepts an OAuth2 access token provider that supplies the external client assertion. The federated identity
credential in Microsoft Entra must match the assertion issuer, subject, and audience.

The password provider requests the public-cloud Azure OSS RDBMS resource. It does not require the Azure JDBC
authentication plugins and does not depend on a particular external assertion issuer.

## PostgreSQL Configuration

Configure Microsoft Entra authentication on Azure Database for PostgreSQL Flexible Server and create the database
role that corresponds to the Entra principal. Set the DBCP **Database User** to that mapped role name.

The token is sent using PostgreSQL cleartext-password authentication and must be protected by TLS. Use direct port
5432 as the baseline and set `sslmode=verify-full`, with the client configured to trust the server certificate chain.

| Setting | Value |
|---|---|
| Driver Class Name | `org.postgresql.Driver` |
| JDBC URL | `jdbc:postgresql://<SERVER>.postgres.database.azure.com:5432/<DATABASE>?sslmode=verify-full` |
| Database User | Microsoft Entra principal's mapped PostgreSQL role name |

Azure's built-in PgBouncer on port 6432 must be validated separately. Large Entra tokens containing many group claims
can exceed intermediary protocol buffers even when the same token works against direct port 5432. This provider does
not claim unrestricted PgBouncer compatibility.

## MySQL Configuration

Configure Microsoft Entra authentication on Azure Database for MySQL Flexible Server and create an Entra user or alias
for the principal. Set the DBCP **Database User** to the mapped alias. MySQL user names are limited to 32 characters, so
an alias can be required for a longer Entra principal name.

MySQL Connector/J must use TLS, verify the server certificate, and permit the `mysql_clear_password` authentication
mechanism used to transmit the token. Configure Connector/J with a trust store containing the issuing CA certificate,
use `sslMode=VERIFY_IDENTITY`, and set `allowCleartextPasswords=true`. The Azure MySQL JDBC authentication plugin is not
required when this provider supplies the token.

| 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://<SERVER>.mysql.database.azure.com:3306/<DATABASE>?sslMode=VERIFY_IDENTITY&allowCleartextPasswords=true` |
| Database User | Microsoft Entra principal's mapped MySQL user or alias |

## Verify and Troubleshooting

`AzureEntraDatabasePasswordProvider` **Verify** checks that the configured Azure Credentials Service can acquire a
nonblank token for the Azure OSS RDBMS resource. It does not validate the JDBC URL, database server, network
path, TLS configuration, driver, principal mapping, or database grants. DBCP **Verify** checks the actual database
connection using those settings.

If provider **Verify** fails, confirm that the Azure Credentials Service and any upstream identity-federation services
are enabled and configured for the intended Entra application and tenant. If provider **Verify** succeeds but DBCP
**Verify** fails, investigate the JDBC connection, network, TLS, driver, database principal mapping, and grants.
Loading
Loading