NIFI-15161 - Add support for Azure Federated Identity Credentials in Storage components

Author: pvillard31

nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/pom.xml

@@ -26,6 +26,20 @@
             <artifactId>nifi-utils</artifactId>
         </dependency>
 
+        <!-- OAuth2 Access Token Provider API for Web Identity (OIDC) support -->
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-oauth2-provider-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-web-client-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-web-client-provider-api</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-service-utils</artifactId>

nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureStorageUtils.java

@@ -25,6 +25,7 @@
 import org.apache.nifi.context.PropertyContext;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.services.azure.AzureIdentityFederationTokenProvider;
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.proxy.ProxyConfiguration;
@@ -85,7 +86,8 @@ public final class AzureStorageUtils {
                     AzureStorageCredentialsType.ACCOUNT_KEY,
                     AzureStorageCredentialsType.SAS_TOKEN,
                     AzureStorageCredentialsType.MANAGED_IDENTITY,
-                    AzureStorageCredentialsType.SERVICE_PRINCIPAL))
+                    AzureStorageCredentialsType.SERVICE_PRINCIPAL,
+                    AzureStorageCredentialsType.ACCESS_TOKEN))
             .defaultValue(AzureStorageCredentialsType.SAS_TOKEN)
             .build();
 
@@ -252,6 +254,14 @@ public final class AzureStorageUtils {
             .dependsOn(CREDENTIALS_TYPE, AzureStorageCredentialsType.SERVICE_PRINCIPAL)
             .build();
 
+    public static final PropertyDescriptor OAUTH2_ACCESS_TOKEN_PROVIDER = new PropertyDescriptor.Builder()
+            .name("Azure Identity Federation Token Provider")
+            .description("Controller Service that exchanges workload identity tokens for Azure AD access tokens.")
+            .identifiesControllerService(AzureIdentityFederationTokenProvider.class)
+            .required(true)
+            .dependsOn(CREDENTIALS_TYPE, AzureStorageCredentialsType.ACCESS_TOKEN)
+            .build();
+
     private AzureStorageUtils() {
         // do not instantiate
     }

nifi-extension-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/services/azure/StandardAzureCredentialsControllerService.java

@@ -30,6 +30,8 @@
 import org.apache.nifi.migration.PropertyConfiguration;
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.util.OAuth2AccessTokenAdapter;
+import reactor.core.publisher.Mono;
 
 import java.util.List;
 
@@ -47,12 +49,15 @@ public class StandardAzureCredentialsControllerService extends AbstractControlle
     public static AllowableValue MANAGED_IDENTITY = new AllowableValue("managed-identity",
             "Managed Identity",
             "Azure Virtual Machine Managed Identity (it can only be used when NiFi is running on Azure)");
+    public static AllowableValue OAUTH2 = new AllowableValue("oauth2-access-token",
+            "OAuth2 Access Token",
+            "Uses an OAuth2 Access Token Provider controller service to obtain access tokens for Azure clients.");
     public static final PropertyDescriptor CREDENTIAL_CONFIGURATION_STRATEGY = new PropertyDescriptor.Builder()
             .name("Credential Configuration Strategy")
             .expressionLanguageSupported(ExpressionLanguageScope.NONE)
             .required(true)
             .sensitive(false)
-            .allowableValues(DEFAULT_CREDENTIAL, MANAGED_IDENTITY)
+            .allowableValues(DEFAULT_CREDENTIAL, MANAGED_IDENTITY, OAUTH2)
             .defaultValue(DEFAULT_CREDENTIAL)
             .build();
 
@@ -67,9 +72,18 @@ public class StandardAzureCredentialsControllerService extends AbstractControlle
             .dependsOn(CREDENTIAL_CONFIGURATION_STRATEGY, MANAGED_IDENTITY)
             .build();
 
+    public static final PropertyDescriptor OAUTH2_ACCESS_TOKEN_PROVIDER = new PropertyDescriptor.Builder()
+            .name("Azure Identity Federation Token Provider")
+            .description("Controller Service used to obtain Azure access tokens via workload identity federation.")
+            .identifiesControllerService(AzureIdentityFederationTokenProvider.class)
+            .required(true)
+            .dependsOn(CREDENTIAL_CONFIGURATION_STRATEGY, OAUTH2)
+            .build();
+
     private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(
             CREDENTIAL_CONFIGURATION_STRATEGY,
-            MANAGED_IDENTITY_CLIENT_ID
+            MANAGED_IDENTITY_CLIENT_ID,
+            OAUTH2_ACCESS_TOKEN_PROVIDER
     );
 
     private TokenCredential credentials;
@@ -92,6 +106,8 @@ public void onConfigured(final ConfigurationContext context) {
             credentials = getDefaultAzureCredential();
         } else if (MANAGED_IDENTITY.getValue().equals(configurationStrategy)) {
             credentials = getManagedIdentityCredential(context);
+        } else if (OAUTH2.getValue().equals(configurationStrategy)) {
+            credentials = getOAuth2Credential(context);
         } else {
             final String errorMsg = String.format("Configuration Strategy [%s] not recognized", configurationStrategy);
             getLogger().error(errorMsg);
@@ -117,6 +133,13 @@ private TokenCredential getManagedIdentityCredential(final ConfigurationContext
                 .build();
     }
 
+    private TokenCredential getOAuth2Credential(final ConfigurationContext context) {
+        final AzureIdentityFederationTokenProvider oauth2AccessTokenProvider = context.getProperty(OAUTH2_ACCESS_TOKEN_PROVIDER)
+                .asControllerService(AzureIdentityFederationTokenProvider.class);
+        return tokenRequestContext -> Mono.fromSupplier(() ->
+                OAuth2AccessTokenAdapter.toAzureAccessToken(oauth2AccessTokenProvider.getAccessDetails()));
+    }
+
     @Override
     public String toString() {
         return "StandardAzureCredentialsControllerService[id=" + getIdentifier() + "]";