feat: add DataAddressValidatorRegistry

core/common/connector-core/build.gradle.kts

@@ -31,6 +31,7 @@ dependencies {
     implementation(project(":core:common:state-machine"))
     implementation(project(":core:common:transform-core"))
     implementation(project(":core:common:util"))
+    implementation(project(":core:common:validator-core"))
 
     implementation(libs.dnsOverHttps)
     implementation(libs.bouncyCastle.bcpkixJdk18on)

core/common/connector-core/src/main/java/org/eclipse/edc/connector/core/CoreDefaultServicesExtension.java

@@ -71,7 +71,6 @@ public TransactionContext defaultTransactionContext(ServiceExtensionContext cont
     public DataSourceRegistry dataSourceRegistry(ServiceExtensionContext context) {
         context.getMonitor().warning("No DataSourceRegistry registered, DefaultDataSourceRegistry will be used, not suitable for production environments");
         return new DefaultDataSourceRegistry();
-
     }
 
     @Provider(isDefault = true)

core/common/connector-core/src/main/java/org/eclipse/edc/connector/core/CoreServicesExtension.java

@@ -23,6 +23,7 @@
 import org.eclipse.edc.connector.core.health.HealthCheckServiceImpl;
 import org.eclipse.edc.connector.core.security.DefaultPrivateKeyParseFunction;
 import org.eclipse.edc.connector.core.security.KeyPairFactoryImpl;
+import org.eclipse.edc.connector.core.validator.DataAddressValidatorRegistryImpl;
 import org.eclipse.edc.connector.core.validator.JsonObjectValidatorRegistryImpl;
 import org.eclipse.edc.core.transform.TypeTransformerRegistryImpl;
 import org.eclipse.edc.policy.engine.PolicyEngineImpl;
@@ -50,6 +51,7 @@
 import org.eclipse.edc.spi.system.health.HealthCheckService;
 import org.eclipse.edc.spi.types.TypeManager;
 import org.eclipse.edc.transform.spi.TypeTransformerRegistry;
+import org.eclipse.edc.validator.spi.DataAddressValidatorRegistry;
 import org.eclipse.edc.validator.spi.JsonObjectValidatorRegistry;
 
 import java.security.PrivateKey;
@@ -193,6 +195,11 @@ public JsonObjectValidatorRegistry jsonObjectValidator() {
         return new JsonObjectValidatorRegistryImpl();
     }
 
+    @Provider
+    public DataAddressValidatorRegistry dataAddressValidatorRegistry(ServiceExtensionContext context) {
+        return new DataAddressValidatorRegistryImpl(context.getMonitor());
+    }
+
     private HealthCheckServiceConfiguration getHealthCheckConfig(ServiceExtensionContext context) {
         return HealthCheckServiceConfiguration.Builder.newInstance()
                 .livenessPeriod(Duration.ofSeconds(context.getSetting(LIVENESS_PERIOD_SECONDS_SETTING, DEFAULT_DURATION)))

core/common/connector-core/src/main/java/org/eclipse/edc/connector/core/validator/DataAddressValidatorRegistryImpl.java

@@ -0,0 +1,62 @@
+/*
+ *  Copyright (c) 2023 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.core.validator;
+
+import org.eclipse.edc.spi.monitor.Monitor;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.eclipse.edc.validator.spi.DataAddressValidatorRegistry;
+import org.eclipse.edc.validator.spi.ValidationResult;
+import org.eclipse.edc.validator.spi.Validator;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class DataAddressValidatorRegistryImpl implements DataAddressValidatorRegistry {
+
+    private final Map<String, Validator<DataAddress>> sourceValidators = new HashMap<>();
+    private final Map<String, Validator<DataAddress>> destinationValidators = new HashMap<>();
+    private final Monitor monitor;
+
+    public DataAddressValidatorRegistryImpl(Monitor monitor) {
+        this.monitor = monitor;
+    }
+
+    @Override
+    public void registerSourceValidator(String type, Validator<DataAddress> validator) {
+        sourceValidators.put(type, validator);
+    }
+
+    @Override
+    public void registerDestinationValidator(String type, Validator<DataAddress> validator) {
+        destinationValidators.put(type, validator);
+    }
+
+    @Override
+    public ValidationResult validateSource(DataAddress dataAddress) {
+        return sourceValidators.getOrDefault(dataAddress.getType(), d -> warning("source")).validate(dataAddress);
+    }
+
+    @Override
+    public ValidationResult validateDestination(DataAddress dataAddress) {
+        return destinationValidators.getOrDefault(dataAddress.getType(), d -> warning("destination")).validate(dataAddress);
+    }
+
+    @NotNull
+    private ValidationResult warning(String type) {
+        monitor.warning("No %s DataAddress validator has been registered, please register one as it is strongly recommended.".formatted(type));
+        return ValidationResult.success();
+    }
+}

core/common/connector-core/src/test/java/org/eclipse/edc/connector/core/validator/DataAddressValidatorRegistryImplTest.java

@@ -0,0 +1,97 @@
+/*
+ *  Copyright (c) 2023 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.core.validator;
+
+import org.eclipse.edc.spi.monitor.Monitor;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.eclipse.edc.validator.spi.DataAddressValidatorRegistry;
+import org.eclipse.edc.validator.spi.ValidationResult;
+import org.eclipse.edc.validator.spi.Validator;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import static org.eclipse.edc.junit.assertions.AbstractResultAssert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class DataAddressValidatorRegistryImplTest {
+
+    private final Monitor monitor = mock();
+    private final DataAddressValidatorRegistry validator = new DataAddressValidatorRegistryImpl(monitor);
+
+    @Nested
+    class Source {
+        @Test
+        void shouldCallRegisteredValidator() {
+            Validator<DataAddress> typeValidator = mock();
+            when(typeValidator.validate(any())).thenReturn(ValidationResult.success());
+            validator.registerSourceValidator("type", typeValidator);
+            var dataAddress = DataAddress.Builder.newInstance()
+                    .property("type", "type")
+                    .build();
+
+            var result = validator.validateSource(dataAddress);
+
+            assertThat(result).isSucceeded();
+            verify(typeValidator).validate(dataAddress);
+        }
+
+        @Test
+        void shouldSucceedWithWarning_whenTypeIsNotRegistered() {
+            var dataAddress = DataAddress.Builder.newInstance()
+                    .property("type", "not-registered")
+                    .build();
+
+            var result = validator.validateSource(dataAddress);
+
+            assertThat(result).isSucceeded();
+            verify(monitor).warning(anyString());
+        }
+    }
+
+    @Nested
+    class Destination {
+        @Test
+        void shouldCallRegisteredValidator() {
+            Validator<DataAddress> typeValidator = mock();
+            when(typeValidator.validate(any())).thenReturn(ValidationResult.success());
+            validator.registerDestinationValidator("type", typeValidator);
+            var dataAddress = DataAddress.Builder.newInstance()
+                    .property("type", "type")
+                    .build();
+
+            var result = validator.validateDestination(dataAddress);
+
+            assertThat(result).isSucceeded();
+            verify(typeValidator).validate(dataAddress);
+        }
+
+        @Test
+        void shouldSucceedWithWarning_whenTypeIsNotRegistered() {
+            var dataAddress = DataAddress.Builder.newInstance()
+                    .property("type", "not-registered")
+                    .build();
+
+            var result = validator.validateDestination(dataAddress);
+
+            assertThat(result).isSucceeded();
+            verify(monitor).warning(anyString());
+        }
+    }
+
+}

core/control-plane/control-plane-aggregate-services/src/main/java/org/eclipse/edc/connector/service/ControlPlaneServicesExtension.java

@@ -34,7 +34,6 @@
 import org.eclipse.edc.connector.service.contractdefinition.ContractDefinitionServiceImpl;
 import org.eclipse.edc.connector.service.contractnegotiation.ContractNegotiationProtocolServiceImpl;
 import org.eclipse.edc.connector.service.contractnegotiation.ContractNegotiationServiceImpl;
-import org.eclipse.edc.connector.service.dataaddress.DataAddressValidatorImpl;
 import org.eclipse.edc.connector.service.policydefinition.PolicyDefinitionEventListener;
 import org.eclipse.edc.connector.service.policydefinition.PolicyDefinitionServiceImpl;
 import org.eclipse.edc.connector.service.transferprocess.TransferProcessProtocolServiceImpl;
@@ -65,7 +64,7 @@
 import org.eclipse.edc.spi.system.ServiceExtensionContext;
 import org.eclipse.edc.spi.telemetry.Telemetry;
 import org.eclipse.edc.transaction.spi.TransactionContext;
-import org.eclipse.edc.validator.spi.DataAddressValidator;
+import org.eclipse.edc.validator.spi.DataAddressValidatorRegistry;
 
 import java.time.Clock;
 
@@ -74,8 +73,6 @@ public class ControlPlaneServicesExtension implements ServiceExtension {
 
     public static final String NAME = "Control Plane Services";
 
-    private final DataAddressValidator dataAddressValidator = new DataAddressValidatorImpl();
-
     @Inject
     private Clock clock;
 
@@ -136,6 +133,9 @@ public class ControlPlaneServicesExtension implements ServiceExtension {
     @Inject
     private CommandHandlerRegistry commandHandlerRegistry;
 
+    @Inject
+    private DataAddressValidatorRegistry dataAddressValidator;
+
     @Override
     public String name() {
         return NAME;

core/control-plane/control-plane-aggregate-services/src/main/java/org/eclipse/edc/connector/service/asset/AssetServiceImpl.java

@@ -24,7 +24,7 @@
 import org.eclipse.edc.spi.query.QuerySpec;
 import org.eclipse.edc.spi.types.domain.asset.Asset;
 import org.eclipse.edc.transaction.spi.TransactionContext;
-import org.eclipse.edc.validator.spi.DataAddressValidator;
+import org.eclipse.edc.validator.spi.DataAddressValidatorRegistry;
 
 import java.util.List;
 import java.util.stream.Stream;
@@ -39,12 +39,12 @@ public class AssetServiceImpl implements AssetService {
     private final ContractNegotiationStore contractNegotiationStore;
     private final TransactionContext transactionContext;
     private final AssetObservable observable;
-    private final DataAddressValidator dataAddressValidator;
+    private final DataAddressValidatorRegistry dataAddressValidator;
     private final QueryValidator queryValidator;
 
     public AssetServiceImpl(AssetIndex index, ContractNegotiationStore contractNegotiationStore,
                             TransactionContext transactionContext, AssetObservable observable,
-                            DataAddressValidator dataAddressValidator) {
+                            DataAddressValidatorRegistry dataAddressValidator) {
         this.index = index;
         this.contractNegotiationStore = contractNegotiationStore;
         this.transactionContext = transactionContext;
@@ -75,7 +75,7 @@ public ServiceResult<Asset> create(Asset asset) {
             return ServiceResult.badRequest(DUPLICATED_KEYS_MESSAGE);
         }
 
-        var validDataAddress = dataAddressValidator.validate(asset.getDataAddress());
+        var validDataAddress = dataAddressValidator.validateSource(asset.getDataAddress());
         if (validDataAddress.failed()) {
             return ServiceResult.badRequest(validDataAddress.getFailureMessages());
         }
@@ -116,6 +116,11 @@ public ServiceResult<Asset> update(Asset asset) {
             return ServiceResult.badRequest(DUPLICATED_KEYS_MESSAGE);
         }
 
+        var validDataAddress = dataAddressValidator.validateSource(asset.getDataAddress());
+        if (validDataAddress.failed()) {
+            return ServiceResult.badRequest(validDataAddress.getFailureMessages());
+        }
+
         return transactionContext.execute(() -> {
             var updatedAsset = index.updateAsset(asset);
             updatedAsset.onSuccess(a -> observable.invokeForEach(l -> l.updated(a)));

core/control-plane/control-plane-aggregate-services/src/main/java/org/eclipse/edc/connector/service/transferprocess/TransferProcessProtocolServiceImpl.java

@@ -37,7 +37,7 @@
 import org.eclipse.edc.spi.telemetry.Telemetry;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.transaction.spi.TransactionContext;
-import org.eclipse.edc.validator.spi.DataAddressValidator;
+import org.eclipse.edc.validator.spi.DataAddressValidatorRegistry;
 import org.jetbrains.annotations.NotNull;
 
 import java.time.Clock;
@@ -56,7 +56,7 @@ public class TransferProcessProtocolServiceImpl implements TransferProcessProtoc
     private final TransactionContext transactionContext;
     private final ContractNegotiationStore negotiationStore;
     private final ContractValidationService contractValidationService;
-    private final DataAddressValidator dataAddressValidator;
+    private final DataAddressValidatorRegistry dataAddressValidator;
     private final TransferProcessObservable observable;
     private final Clock clock;
     private final Monitor monitor;
@@ -65,7 +65,7 @@ public class TransferProcessProtocolServiceImpl implements TransferProcessProtoc
     public TransferProcessProtocolServiceImpl(TransferProcessStore transferProcessStore,
                                               TransactionContext transactionContext, ContractNegotiationStore negotiationStore,
                                               ContractValidationService contractValidationService,
-                                              DataAddressValidator dataAddressValidator, TransferProcessObservable observable,
+                                              DataAddressValidatorRegistry dataAddressValidator, TransferProcessObservable observable,
                                               Clock clock, Monitor monitor, Telemetry telemetry) {
         this.transferProcessStore = transferProcessStore;
         this.transactionContext = transactionContext;
@@ -84,7 +84,7 @@ public TransferProcessProtocolServiceImpl(TransferProcessStore transferProcessSt
     public ServiceResult<TransferProcess> notifyRequested(TransferRequestMessage message, ClaimToken claimToken) {
         var destination = message.getDataDestination();
         if (destination != null) {
-            var validDestination = dataAddressValidator.validate(destination);
+            var validDestination = dataAddressValidator.validateDestination(destination);
             if (validDestination.failed()) {
                 return ServiceResult.badRequest(validDestination.getFailureMessages());
             }