[UNOMI-973] - Confine recurrent import/export file endpoints to configurable base directories

extensions/router/router-api/src/main/java/org/apache/unomi/router/api/RouterConstants.java

@@ -32,6 +32,13 @@ enum CONFIG_CAMEL_REFRESH {
     String CONFIG_STATUS_COMPLETE_ERRORS = "ERRORS";
     String CONFIG_STATUS_COMPLETE_SUCCESS = "SUCCESS";
     String CONFIG_STATUS_COMPLETE_WITH_ERRORS = "WITH_ERRORS";
+    /**
+     * The configuration names an endpoint that cannot be honoured, so no route carries it. Kept apart
+     * from the execution statuses above: those report on a run that happened, this one says no run can.
+     * It is set and cleared by the route builders alone, so that restoring the deployment's permitted
+     * directories brings the configuration back on its own.
+     */
+    String CONFIG_STATUS_INVALID_ENDPOINT = "INVALID_ENDPOINT";
 
     String IMPORT_EXPORT_CONFIG_TYPE_RECURRENT = "recurrent";
     String IMPORT_EXPORT_CONFIG_TYPE_ONESHOT = "oneshot";
@@ -51,6 +58,10 @@ enum CONFIG_CAMEL_REFRESH {
     String IMPORT_ONESHOT_ROUTE_ID = "ONE_SHOT_ROUTE";
     String IMPORT_ONESHOT_UPLOAD_DIR = "oneshotImportUploadDir";
 
+    String CONFIG_ALLOWED_ENDPOINTS = "routerAllowedEndpoints";
+    String CONFIG_IMPORT_BASE_DIRS = "routerImportBaseDirs";
+    String CONFIG_EXPORT_BASE_DIRS = "routerExportBaseDirs";
+
     String DEFAULT_FILE_COLUMN_SEPARATOR = ",";
     String DEFAULT_FILE_LINE_SEPARATOR = "\n";
     String KEY_HISTORY_SIZE = "historySize";

extensions/router/router-core/pom.xml

@@ -148,6 +148,13 @@
             <artifactId>slf4j-api</artifactId>
             <scope>provided</scope>
         </dependency>
+
+        <!-- tests -->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java

@@ -71,6 +71,8 @@ public class RouterCamelContext implements IRouterCamelContext {
     private Map<String, String> kafkaProps;
     private String configType;
     private String allowedEndpoints;
+    private String permittedImportBaseDirs;
+    private String permittedExportBaseDirs;
     private BundleContext bundleContext;
     private ConfigSharingService configSharingService;
 
@@ -108,6 +110,10 @@ public void init() throws Exception {
         scheduler = Executors.newSingleThreadScheduledExecutor();
 
         configSharingService.setProperty(RouterConstants.IMPORT_ONESHOT_UPLOAD_DIR, uploadDir);
+        // shared with router-rest, which validates a configuration's endpoint before it is stored
+        configSharingService.setProperty(RouterConstants.CONFIG_ALLOWED_ENDPOINTS, allowedEndpoints);
+        configSharingService.setProperty(RouterConstants.CONFIG_IMPORT_BASE_DIRS, permittedImportBaseDirs);
+        configSharingService.setProperty(RouterConstants.CONFIG_EXPORT_BASE_DIRS, permittedExportBaseDirs);
         configSharingService.setProperty(RouterConstants.KEY_HISTORY_SIZE, execHistorySize);
 
         initCamel();
@@ -179,6 +185,7 @@ private void initCamel() throws Exception {
         builderReader.setImportConfigurationService(importConfigurationService);
         builderReader.setJacksonDataFormat(jacksonDataFormat);
         builderReader.setAllowedEndpoints(allowedEndpoints);
+        builderReader.setPermittedImportBaseDirs(permittedImportBaseDirs);
         builderReader.setContext(camelContext);
         camelContext.addRoutes(builderReader);
 
@@ -204,8 +211,10 @@ private void initCamel() throws Exception {
         //Profiles collect
         ProfileExportCollectRouteBuilder profileExportCollectRouteBuilder = new ProfileExportCollectRouteBuilder(kafkaProps, configType);
         profileExportCollectRouteBuilder.setExportConfigurationList(exportConfigurationService.getAll());
+        profileExportCollectRouteBuilder.setExportConfigurationService(exportConfigurationService);
         profileExportCollectRouteBuilder.setPersistenceService(persistenceService);
         profileExportCollectRouteBuilder.setAllowedEndpoints(allowedEndpoints);
+        profileExportCollectRouteBuilder.setPermittedExportBaseDirs(permittedExportBaseDirs);
         profileExportCollectRouteBuilder.setJacksonDataFormat(jacksonDataFormat);
         profileExportCollectRouteBuilder.setContext(camelContext);
         camelContext.addRoutes(profileExportCollectRouteBuilder);
@@ -249,6 +258,7 @@ public void updateProfileImportReaderRoute(String configId, boolean fireEvent) t
             builder.setImportConfigurationService(importConfigurationService);
             builder.setProfileService(profileService);
             builder.setAllowedEndpoints(allowedEndpoints);
+            builder.setPermittedImportBaseDirs(permittedImportBaseDirs);
             builder.setJacksonDataFormat(jacksonDataFormat);
             builder.setContext(camelContext);
             camelContext.addRoutes(builder);
@@ -267,8 +277,10 @@ public void updateProfileExportReaderRoute(String configId, boolean fireEvent) t
         if (RouterConstants.IMPORT_EXPORT_CONFIG_TYPE_RECURRENT.equals(exportConfiguration.getConfigType())) {
             ProfileExportCollectRouteBuilder profileExportCollectRouteBuilder = new ProfileExportCollectRouteBuilder(kafkaProps, configType);
             profileExportCollectRouteBuilder.setExportConfigurationList(Collections.singletonList(exportConfiguration));
+            profileExportCollectRouteBuilder.setExportConfigurationService(exportConfigurationService);
             profileExportCollectRouteBuilder.setPersistenceService(persistenceService);
             profileExportCollectRouteBuilder.setAllowedEndpoints(allowedEndpoints);
+            profileExportCollectRouteBuilder.setPermittedExportBaseDirs(permittedExportBaseDirs);
             profileExportCollectRouteBuilder.setJacksonDataFormat(jacksonDataFormat);
             profileExportCollectRouteBuilder.setContext(camelContext);
             camelContext.addRoutes(profileExportCollectRouteBuilder);
@@ -334,4 +346,12 @@ public void setConfigType(String configType) {
     public void setAllowedEndpoints(String allowedEndpoints) {
         this.allowedEndpoints = allowedEndpoints;
     }
+
+    public void setPermittedImportBaseDirs(String permittedImportBaseDirs) {
+        this.permittedImportBaseDirs = permittedImportBaseDirs;
+    }
+
+    public void setPermittedExportBaseDirs(String permittedExportBaseDirs) {
+        this.permittedExportBaseDirs = permittedExportBaseDirs;
+    }
 }

extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileExportCollectRouteBuilder.java

@@ -19,10 +19,11 @@
 import org.apache.camel.LoggingLevel;
 import org.apache.camel.component.kafka.KafkaEndpoint;
 import org.apache.camel.model.ProcessorDefinition;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.unomi.persistence.spi.PersistenceService;
+import org.apache.unomi.router.api.EndpointValidator;
 import org.apache.unomi.router.api.ExportConfiguration;
 import org.apache.unomi.router.api.RouterConstants;
+import org.apache.unomi.router.api.services.ImportExportConfigurationService;
 import org.apache.unomi.router.core.bean.CollectProfileBean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,6 +39,7 @@ public class ProfileExportCollectRouteBuilder extends RouterAbstractRouteBuilder
     private static final Logger LOGGER = LoggerFactory.getLogger(ProfileExportCollectRouteBuilder.class);
 
     private List<ExportConfiguration> exportConfigurationList;
+    private ImportExportConfigurationService<ExportConfiguration> exportConfigurationService;
     private PersistenceService persistenceService;
 
     public ProfileExportCollectRouteBuilder(Map<String, String> kafkaProps, String configType) {
@@ -62,7 +64,9 @@ public void configure() throws Exception {
                     exportConfiguration.getProperties() != null && exportConfiguration.getProperties().size() > 0) {
                 if ((Map<String, String>) exportConfiguration.getProperties().get("mapping") != null) {
                     String destinationEndpoint = (String) exportConfiguration.getProperties().get("destination");
-                    if (StringUtils.isNotBlank(destinationEndpoint) && allowedEndpoints.contains(destinationEndpoint.substring(0, destinationEndpoint.indexOf(':')))) {
+                    String refusal = EndpointValidator.validate(destinationEndpoint, allowedEndpoints, permittedBaseDirs);
+                    recordEndpointOutcome(exportConfiguration, exportConfigurationService, refusal);
+                    if (refusal == null) {
                         String timerString = "timer://collectProfile?fixedRate=true&period=" + (String) exportConfiguration.getProperties().get("period");
                         if ((String) exportConfiguration.getProperties().get("delay") != null) {
                             timerString += "&delay=" + (String) exportConfiguration.getProperties().get("delay");
@@ -82,7 +86,7 @@ public void configure() throws Exception {
                             prDef.to((String) getEndpointURI(RouterConstants.DIRECTION_FROM, RouterConstants.DIRECT_EXPORT_DEPOSIT_BUFFER));
                         }
                     } else {
-                        LOGGER.error("Endpoint scheme {} is not allowed, route {} will be skipped.", destinationEndpoint.substring(0, destinationEndpoint.indexOf(':')), exportConfiguration.getItemId());
+                        LOGGER.error("Destination endpoint is refused ({}), route {} will be skipped.", refusal, exportConfiguration.getItemId());
                     }
                 } else {
                     LOGGER.warn("Mapping is null in export configuration, route {} will be skipped!", exportConfiguration.getItemId());
@@ -93,6 +97,19 @@ public void configure() throws Exception {
         }
     }
 
+    /**
+     * Sets the comma-separated list of base directories an export {@code file} endpoint may resolve into.
+     *
+     * @param permittedExportBaseDirs the permitted base directories
+     */
+    public void setPermittedExportBaseDirs(String permittedExportBaseDirs) {
+        this.permittedBaseDirs = permittedExportBaseDirs;
+    }
+
+    public void setExportConfigurationService(ImportExportConfigurationService<ExportConfiguration> exportConfigurationService) {
+        this.exportConfigurationService = exportConfigurationService;
+    }
+
     public void setExportConfigurationList(List<ExportConfiguration> exportConfigurationList) {
         this.exportConfigurationList = exportConfigurationList;
     }

extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportFromSourceRouteBuilder.java

@@ -23,6 +23,7 @@
 import org.apache.camel.component.kafka.KafkaEndpoint;
 import org.apache.camel.model.ProcessorDefinition;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.unomi.router.api.EndpointValidator;
 import org.apache.unomi.router.api.ImportConfiguration;
 import org.apache.unomi.router.api.RouterConstants;
 import org.apache.unomi.router.api.services.ImportExportConfigurationService;
@@ -92,9 +93,13 @@ public void configure() throws Exception {
                 lineSplitProcessor.setProfilePropertyTypes(profileService.getTargetPropertyTypes("profiles"));
 
                 String endpoint = (String) importConfiguration.getProperties().get("source");
-                endpoint += "&moveFailed=.error";
+                if (StringUtils.isNotBlank(endpoint)) {
+                    endpoint += "&moveFailed=.error";
+                }
 
-                if (StringUtils.isNotBlank(endpoint) && allowedEndpoints.contains(endpoint.substring(0, endpoint.indexOf(':')))) {
+                String refusal = EndpointValidator.validate(endpoint, allowedEndpoints, permittedBaseDirs);
+                recordEndpointOutcome(importConfiguration, importConfigurationService, refusal);
+                if (refusal == null) {
                     ProcessorDefinition prDef = from(endpoint)
                             .routeId(importConfiguration.getItemId())// This allow identification of the route for manual start/stop
                             .autoStartup(importConfiguration.isActive())// Auto-start if the import configuration is set active
@@ -126,12 +131,21 @@ public void process(Exchange exchange) throws Exception {
                         prDef.to((String) getEndpointURI(RouterConstants.DIRECTION_FROM, RouterConstants.DIRECT_IMPORT_DEPOSIT_BUFFER));
                     }
                 } else {
-                    LOGGER.error("Endpoint scheme {} is not allowed, route {} will be skipped.", endpoint.substring(0, endpoint.indexOf(':')), importConfiguration.getItemId());
+                    LOGGER.error("Source endpoint is refused ({}), route {} will be skipped.", refusal, importConfiguration.getItemId());
                 }
             }
         }
     }
 
+    /**
+     * Sets the comma-separated list of base directories an import {@code file} endpoint may resolve into.
+     *
+     * @param permittedImportBaseDirs the permitted base directories
+     */
+    public void setPermittedImportBaseDirs(String permittedImportBaseDirs) {
+        this.permittedBaseDirs = permittedImportBaseDirs;
+    }
+
     public void setImportConfigurationList(List<ImportConfiguration> importConfigurationList) {
         this.importConfigurationList = importConfigurationList;
     }

extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/RouterAbstractRouteBuilder.java

@@ -23,7 +23,9 @@
 import org.apache.camel.component.kafka.KafkaEndpoint;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.unomi.api.services.ProfileService;
+import org.apache.unomi.router.api.ImportExportConfiguration;
 import org.apache.unomi.router.api.RouterConstants;
+import org.apache.unomi.router.api.services.ImportExportConfigurationService;
 
 import java.util.Map;
 
@@ -45,6 +47,7 @@ public abstract class RouterAbstractRouteBuilder extends RouteBuilder {
 
     protected String configType;
     protected String allowedEndpoints;
+    protected String permittedBaseDirs;
 
     protected ProfileService profileService;
 
@@ -60,6 +63,37 @@ public RouterAbstractRouteBuilder(Map<String, String> kafkaProps, String configT
         this.configType = configType;
     }
 
+    /**
+     * Records, on the configuration itself, whether the endpoint it names can be honoured.
+     *
+     * <p>The permitted directories are an operational setting and the configurations are user data, so
+     * the two drift apart: a configuration that was legitimate when it was created can be refused after
+     * the deployment is reconfigured. Refusing it silently leaves the owner with a configuration that
+     * looks fine and does nothing, so the refusal is written where they will see it. It is theirs to
+     * correct or remove — nothing is deleted here.
+     *
+     * <p>The other way round matters just as much: restoring the permitted directories must bring the
+     * configuration back on its own, without anyone having to touch it. Only the status this method
+     * sets is cleared, so the record of a run that genuinely failed survives.
+     *
+     * <p>The configuration is saved without asking for its running route to be refreshed: the refresh
+     * would rebuild the route, refuse it again and save it again, without end.
+     *
+     * @param configuration the configuration whose endpoint was examined
+     * @param service       the service holding that kind of configuration
+     * @param refusal       the reason the endpoint was refused, or {@code null} if it can be honoured
+     */
+    protected <T extends ImportExportConfiguration> void recordEndpointOutcome(
+            T configuration, ImportExportConfigurationService<T> service, String refusal) {
+        if (refusal != null) {
+            configuration.setStatus(RouterConstants.CONFIG_STATUS_INVALID_ENDPOINT);
+            service.save(configuration, false);
+        } else if (RouterConstants.CONFIG_STATUS_INVALID_ENDPOINT.equals(configuration.getStatus())) {
+            configuration.setStatus(null);
+            service.save(configuration, false);
+        }
+    }
+
     public Object getEndpointURI(String direction, String operationDepositBuffer) {
         Object endpoint;
         if (RouterConstants.CONFIG_TYPE_KAFKA.equals(configType)) {

extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml

@@ -26,6 +26,8 @@
         <cm:default-properties>
             <cm:property name="router.config.type" value="nobroker"/>
             <cm:property name="config.allowedEndpoints" value="file,ftp"/>
+            <cm:property name="config.import.baseDir" value="${karaf.data}/router/import/"/>
+            <cm:property name="config.export.baseDir" value="${karaf.data}/router/export/"/>
             <cm:property name="kafka.host" value="localhost"/>
             <cm:property name="kafka.port" value="9092"/>
             <cm:property name="kafka.import.topic" value="import-deposit"/>
@@ -84,6 +86,8 @@
           init-method="init" destroy-method="destroy">
         <property name="configType" value="${router.config.type}"/>
         <property name="allowedEndpoints" value="${config.allowedEndpoints}"/>
+        <property name="permittedImportBaseDirs" value="${config.import.baseDir}"/>
+        <property name="permittedExportBaseDirs" value="${config.export.baseDir}"/>
         <property name="uploadDir" value="${import.oneshot.uploadDir}"/>
         <property name="execHistorySize" value="${executionsHistory.size}"/>
         <property name="execErrReportSize" value="${executions.error.report.size}"/>

extensions/router/router-core/src/main/resources/org.apache.unomi.router.cfg

@@ -38,4 +38,11 @@ executionsHistory.size=${org.apache.unomi.router.executionsHistory.size:-5}
 executions.error.report.size=${org.apache.unomi.router.executions.error.report.size:-200}
 
 #Allowed source endpoints
-config.allowedEndpoints=${org.apache.unomi.router.config.allowedEndpoints:-file,ftp,sftp,ftps}
+config.allowedEndpoints=${org.apache.unomi.router.config.allowedEndpoints:-file,ftp,sftp,ftps}
+
+#Base directories a file endpoint may resolve into, comma-separated. A recurrent import source or
+#export destination using the file scheme is refused unless it resolves inside one of them, at any
+#depth. Import and export are kept apart so that an export cannot write into a directory an import
+#route is polling; point them at the same directory only if that is what you mean.
+config.import.baseDir=${org.apache.unomi.router.config.import.baseDir:-${karaf.data}/router/import/}
+config.export.baseDir=${org.apache.unomi.router.config.export.baseDir:-${karaf.data}/router/export/}