NIFI-14547 MiNiFi - Resolve asset references in flows coming from C2 server

minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/service/MiNiFiConfigurationChangeListener.java

@@ -88,9 +88,9 @@ public void handleChange(InputStream flowConfigInputStream) throws Configuration
             backup(currentRawFlowConfigFile, backupRawFlowConfigFile);
 
             byte[] rawFlow = toByteArray(flowConfigInputStream);
-            VersionedDataflow rawDataFlow = flowSerDeService.deserialize(rawFlow);
-            VersionedDataflow enrichedFlow = flowEnrichService.enrichFlow(rawDataFlow);
-            byte[] serializedEnrichedFlow = flowSerDeService.serialize(enrichedFlow);
+            VersionedDataflow dataFlow = flowSerDeService.deserialize(rawFlow);
+            flowEnrichService.enrichFlow(dataFlow);
+            byte[] serializedEnrichedFlow = flowSerDeService.serialize(dataFlow);
             persist(serializedEnrichedFlow, currentFlowConfigFile, true);
             restartInstance();
             persist(rawFlow, currentRawFlowConfigFile, false);

minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java

@@ -29,7 +29,6 @@ public interface FlowEnrichService {
      * Responsible for enriching a VersionedDataflow instance
      *
      * @param versionedDataflow a VersionedDataflow instance
-     * @return VersionedDataflow the enriched flow instance
      */
-    VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow);
+    void enrichFlow(VersionedDataflow versionedDataflow);
 }

minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyAssetReferenceResolver.java

@@ -0,0 +1,29 @@
+/*
+ * 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.minifi.commons.service;
+
+import org.apache.nifi.controller.flow.VersionedDataflow;
+
+public interface FlowPropertyAssetReferenceResolver {
+    /**
+     * Responsible for resolving asset reference properties in a VersionedDataflow instance
+     *
+     * @param flow a VersionedDataflow instance to resolve its asset reference properties
+     */
+    void resolveAssetReferenceProperties(VersionedDataflow flow);
+}

minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyEncryptor.java

@@ -29,7 +29,6 @@ public interface FlowPropertyEncryptor {
      * Responsible for encrypting sensitive properties in a VersionedDataflow instance
      *
      * @param flow a VersionedDataflow instance to encrypt its sensitive properties
-     * @return VersionedDataflow the flow instance with encrypted sensitive properties
      */
-    VersionedDataflow encryptSensitiveProperties(VersionedDataflow flow);
+    void encryptSensitiveProperties(VersionedDataflow flow);
 }

minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichService.java

@@ -97,7 +97,7 @@ public StandardFlowEnrichService(ReadableProperties minifiProperties) {
     }
 
     @Override
-    public VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow) {
+    public void enrichFlow(VersionedDataflow versionedDataflow) {
         versionedDataflow.setReportingTasks(ofNullable(versionedDataflow.getReportingTasks()).orElseGet(ArrayList::new));
         versionedDataflow.setRegistries(ofNullable(versionedDataflow.getRegistries()).orElseGet(ArrayList::new));
         versionedDataflow.setControllerServices(ofNullable(versionedDataflow.getControllerServices()).orElseGet(ArrayList::new));
@@ -147,8 +147,6 @@ public VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow) {
             Map<String, String> idToInstanceIdMap = createIdToInstanceIdMap(rootGroup);
             setConnectableComponentsInstanceId(rootGroup, idToInstanceIdMap);
         }
-
-        return versionedDataflow;
     }
 
     private void createDefaultParameterContext(VersionedDataflow versionedDataflow) {

minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverService.java

@@ -0,0 +1,87 @@
+/*
+ * 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.minifi.commons.service;
+
+import org.apache.nifi.controller.flow.VersionedDataflow;
+import org.apache.nifi.flow.VersionedConfigurableExtension;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import static java.util.Optional.ofNullable;
+import static java.util.stream.Stream.concat;
+
+public class StandardFlowPropertyAssetReferenceResolverService implements FlowPropertyAssetReferenceResolver {
+
+    private static final String ASSET_REFERENCE_PREFIX = "@{asset-id:";
+    private static final String ASSET_REFERENCE_SUFFIX = "}";
+    private static final String EMPTY_STRING = "";
+
+    private final Function<String, Optional<Path>> assetPathResolver;
+
+    public StandardFlowPropertyAssetReferenceResolverService(Function<String, Optional<Path>> assetPathResolver) {
+        this.assetPathResolver = assetPathResolver;
+    }
+
+    @Override
+    public void resolveAssetReferenceProperties(VersionedDataflow flow) {
+        fetchFlowComponents(flow).forEach(component -> {
+            component.getProperties().entrySet().stream()
+                .filter(e -> isAssetReference(e.getValue()))
+                .forEach(entry -> entry.setValue(getAssetAbsolutePathOrThrowIllegalStateException(entry.getValue())));
+       });
+    }
+
+    private boolean isAssetReference(String value) {
+        return value != null
+                && value.startsWith(ASSET_REFERENCE_PREFIX)
+                && value.endsWith(ASSET_REFERENCE_SUFFIX);
+    }
+
+    private Stream<? extends VersionedConfigurableExtension> fetchFlowComponents(VersionedDataflow flow) {
+        return concat(
+                ofNullable(flow.getControllerServices()).orElse(List.of()).stream(),
+                fetchComponentsRecursively(flow.getRootGroup())
+        );
+    }
+
+    private Stream<? extends VersionedConfigurableExtension> fetchComponentsRecursively(VersionedProcessGroup processGroup) {
+        return concat(
+                Stream.of(
+                        ofNullable(processGroup.getProcessors()).orElse(Set.of()),
+                        ofNullable(processGroup.getControllerServices()).orElse(Set.of())
+                        )
+                .flatMap(Set::stream),
+                ofNullable(processGroup.getProcessGroups()).orElse(Set.of()).stream()
+                        .flatMap(this::fetchComponentsRecursively)
+        );
+    }
+
+    private String getAssetAbsolutePathOrThrowIllegalStateException(String assetReference) {
+        String resourceId = assetReference.replace(ASSET_REFERENCE_PREFIX, EMPTY_STRING)
+                .replace(ASSET_REFERENCE_SUFFIX, EMPTY_STRING);
+        return assetPathResolver.apply(resourceId)
+                .map(Path::toString)
+                .orElseThrow(() -> new IllegalStateException("Resource '" + resourceId + "' not found"));
+    }
+}

minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptor.java

@@ -55,16 +55,14 @@ public StandardFlowPropertyEncryptor(PropertyEncryptor propertyEncryptor, Runtim
     }
 
     @Override
-    public VersionedDataflow encryptSensitiveProperties(VersionedDataflow flow) {
+    public void encryptSensitiveProperties(VersionedDataflow flow) {
         encryptParameterContextsProperties(flow);
 
         Map<String, Set<String>> sensitivePropertiesByComponentType = Optional.of(flowProvidedSensitiveProperties(flow))
             .filter(not(Map::isEmpty))
             .orElseGet(this::runtimeManifestSensitiveProperties);
 
         encryptFlowComponentsProperties(flow, sensitivePropertiesByComponentType);
-
-        return flow;
     }
 
     private void encryptParameterContextsProperties(VersionedDataflow flow) {

minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichServiceTest.java

@@ -61,10 +61,10 @@ public void testFlowIsLeftIntactIfEnrichingIsNotNecessary() {
         VersionedDataflow testFlow = loadDefaultFlow();
 
         FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
-        VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+        testFlowEnrichService.enrichFlow(testFlow);
 
         byte[] testFlowBytes = flowToString(testFlow).getBytes(UTF_8);
-        byte[] enrichedFlowBytes = flowToString(enrichedFlow).getBytes(UTF_8);
+        byte[] enrichedFlowBytes = flowToString(testFlow).getBytes(UTF_8);
         assertArrayEquals(testFlowBytes, enrichedFlowBytes);
     }
 
@@ -80,10 +80,10 @@ public void testMissingRootGroupIdsAreFilledIn() {
             uuid.when(UUID::randomUUID).thenReturn(expectedIdentifier);
 
             FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
-            VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+            testFlowEnrichService.enrichFlow(testFlow);
 
-            assertEquals(expectedIdentifier.toString(), enrichedFlow.getRootGroup().getIdentifier());
-            assertEquals(expectedIdentifier.toString(), enrichedFlow.getRootGroup().getInstanceIdentifier());
+            assertEquals(expectedIdentifier.toString(), testFlow.getRootGroup().getIdentifier());
+            assertEquals(expectedIdentifier.toString(), testFlow.getRootGroup().getInstanceIdentifier());
         }
     }
 
@@ -100,13 +100,13 @@ public void testCommonSslControllerServiceIsAddedWithBundleVersionAndProcessorCo
             ));
 
         FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
-        VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+        testFlowEnrichService.enrichFlow(testFlow);
 
-        assertEquals(1, enrichedFlow.getRootGroup().getControllerServices().size());
-        VersionedControllerService sslControllerService = enrichedFlow.getRootGroup().getControllerServices().iterator().next();
+        assertEquals(1, testFlow.getRootGroup().getControllerServices().size());
+        VersionedControllerService sslControllerService = testFlow.getRootGroup().getControllerServices().iterator().next();
         assertEquals(PARENT_SSL_CONTEXT_SERVICE_NAME, sslControllerService.getName());
         assertEquals(StringUtils.EMPTY, sslControllerService.getBundle().getVersion());
-        Set<VersionedProcessor> processors = enrichedFlow.getRootGroup().getProcessors();
+        Set<VersionedProcessor> processors = testFlow.getRootGroup().getProcessors();
         assertEquals(2, processors.size());
         assertTrue(
             processors.stream()
@@ -132,9 +132,9 @@ public void testProvenanceReportingTaskIsAdded() {
         VersionedDataflow testFlow = loadDefaultFlow();
 
         FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
-        VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+        testFlowEnrichService.enrichFlow(testFlow);
 
-        List<VersionedReportingTask> reportingTasks = enrichedFlow.getReportingTasks();
+        List<VersionedReportingTask> reportingTasks = testFlow.getReportingTasks();
         assertEquals(1, reportingTasks.size());
         VersionedReportingTask provenanceReportingTask = reportingTasks.get(0);
         assertEquals(SITE_TO_SITE_PROVENANCE_REPORTING_TASK_NAME, provenanceReportingTask.getName());

minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverServiceTest.java

@@ -0,0 +1,136 @@
+/*
+ * 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.minifi.commons.service;
+
+import org.apache.nifi.controller.flow.VersionedDataflow;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class StandardFlowPropertyAssetReferenceResolverServiceTest {
+
+    private Map<String, String> processorProperties;
+    private Map<String, String> controllerServiceProperties;
+
+    @Mock
+    private Function<String, Optional<Path>> assetPathResolver;
+
+    @InjectMocks
+    private StandardFlowPropertyAssetReferenceResolverService victim;
+
+    @Test
+    public void testResolveAssetReferenceProperties() {
+        initProperties();
+        VersionedDataflow dataFlow = aVersionedDataflow();
+
+        when(assetPathResolver.apply("asset1")).thenReturn(Optional.of(Path.of("asset1Path")));
+        when(assetPathResolver.apply("asset2")).thenReturn(Optional.of(Path.of("asset2Path")));
+
+        victim.resolveAssetReferenceProperties(dataFlow);
+
+        verifyProperties();
+    }
+
+    @Test
+    public void testResolveNestedAssetReferenceProperties() {
+        initProperties();
+        VersionedDataflow dataFlow = aVersionedDataflowWithNestedProcessGroup();
+
+        when(assetPathResolver.apply("asset1")).thenReturn(Optional.of(Path.of("asset1Path")));
+        when(assetPathResolver.apply("asset2")).thenReturn(Optional.of(Path.of("asset2Path")));
+
+        victim.resolveAssetReferenceProperties(dataFlow);
+
+        verifyProperties();
+    }
+
+    @Test
+    public void testResolveAssetReferencePropertiesThrowIllegalStateException() {
+        initProperties();
+        VersionedDataflow dataFlow = aVersionedDataflow();
+
+        when(assetPathResolver.apply("asset1")).thenReturn(Optional.empty());
+
+        assertThrows(IllegalStateException.class, () -> {
+            victim.resolveAssetReferenceProperties(dataFlow);
+        });
+    }
+
+    private void initProperties() {
+        processorProperties = new HashMap<>();
+        processorProperties.put("assetReferenceProperty", "@{asset-id:asset1}");
+        processorProperties.put("notAssetReferenceProperty", "some value1");
+
+        controllerServiceProperties = new HashMap<>();
+        controllerServiceProperties.put("assetReferenceProperty", "@{asset-id:asset2}");
+        controllerServiceProperties.put("notAssetReferenceProperty", "some value2");
+    }
+
+    private void verifyProperties() {
+        assertEquals(processorProperties.get("assetReferenceProperty"), "asset1Path");
+        assertEquals(processorProperties.get("notAssetReferenceProperty"), "some value1");
+        assertEquals(controllerServiceProperties.get("assetReferenceProperty"), "asset2Path");
+        assertEquals(controllerServiceProperties.get("notAssetReferenceProperty"), "some value2");
+    }
+
+    private VersionedDataflow aVersionedDataflow() {
+        VersionedDataflow versionedDataflow = new VersionedDataflow();
+        versionedDataflow.setRootGroup(aVersionedProcessGroup());
+        return versionedDataflow;
+    }
+
+    private VersionedDataflow aVersionedDataflowWithNestedProcessGroup() {
+        VersionedDataflow versionedDataflow = new VersionedDataflow();
+        VersionedProcessGroup versionedProcessGroup = new VersionedProcessGroup();
+        versionedProcessGroup.setProcessGroups(Set.of(aVersionedProcessGroup()));
+        versionedDataflow.setRootGroup(versionedProcessGroup);
+
+        return versionedDataflow;
+    }
+
+    private VersionedProcessGroup aVersionedProcessGroup() {
+        VersionedProcessGroup versionedProcessGroup = new VersionedProcessGroup();
+        VersionedProcessor versionedProcessor = new VersionedProcessor();
+        VersionedControllerService versionedControllerService = new VersionedControllerService();
+
+        versionedControllerService.setProperties(controllerServiceProperties);
+        versionedProcessor.setProperties(processorProperties);
+
+        versionedProcessGroup.setProcessors(Set.of(versionedProcessor));
+        versionedProcessGroup.setControllerServices(Set.of(versionedControllerService));
+
+        return versionedProcessGroup;
+    }
+}