fix: enforce consistency of contract negotiation request and transfer…

… request

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

@@ -237,7 +237,7 @@ public PolicyDefinitionService policyDefinitionService() {
     @Provider
     public TransferProcessService transferProcessService() {
         return new TransferProcessServiceImpl(transferProcessStore, transferProcessManager, transactionContext,
-                dataAddressValidator, commandHandlerRegistry, flowTypeExtractor);
+                dataAddressValidator, commandHandlerRegistry, flowTypeExtractor, contractNegotiationStore);
     }
 
     @Provider

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

@@ -15,6 +15,7 @@
 
 package org.eclipse.edc.connector.controlplane.services.transferprocess;
 
+import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.store.ContractNegotiationStore;
 import org.eclipse.edc.connector.controlplane.services.query.QueryValidator;
 import org.eclipse.edc.connector.controlplane.services.spi.transferprocess.TransferProcessService;
 import org.eclipse.edc.connector.controlplane.transfer.spi.TransferProcessManager;
@@ -61,16 +62,19 @@ public class TransferProcessServiceImpl implements TransferProcessService {
     private final DataAddressValidatorRegistry dataAddressValidator;
     private final CommandHandlerRegistry commandHandlerRegistry;
     private final FlowTypeExtractor flowTypeExtractor;
+    private final ContractNegotiationStore contractNegotiationStore;
 
     public TransferProcessServiceImpl(TransferProcessStore transferProcessStore, TransferProcessManager manager,
                                       TransactionContext transactionContext, DataAddressValidatorRegistry dataAddressValidator,
-                                      CommandHandlerRegistry commandHandlerRegistry, FlowTypeExtractor flowTypeExtractor) {
+                                      CommandHandlerRegistry commandHandlerRegistry, FlowTypeExtractor flowTypeExtractor,
+                                      ContractNegotiationStore contractNegotiationStore) {
         this.transferProcessStore = transferProcessStore;
         this.manager = manager;
         this.transactionContext = transactionContext;
         this.dataAddressValidator = dataAddressValidator;
         this.commandHandlerRegistry = commandHandlerRegistry;
         this.flowTypeExtractor = flowTypeExtractor;
+        this.contractNegotiationStore = contractNegotiationStore;
         queryValidator = new QueryValidator(TransferProcess.class, getSubtypes());
     }
 
@@ -123,6 +127,15 @@ public ServiceResult<List<TransferProcess>> search(QuerySpec query) {
 
     @Override
     public @NotNull ServiceResult<TransferProcess> initiateTransfer(TransferRequest request) {
+        var agreement = contractNegotiationStore.findContractAgreement(request.getContractId());
+        if (agreement == null) {
+            return ServiceResult.badRequest("Contract agreement with id %s not found".formatted(request.getContractId()));
+        }
+
+        if (!agreement.getAssetId().equals(request.getAssetId())) {
+            return ServiceResult.badRequest("Asset id %s in contract agreement does not match asset id in transfer request %s".formatted(agreement.getAssetId(), request.getAssetId()));
+        }
+
         var flowType = flowTypeExtractor.extract(request.getTransferType()).getContent();
 
         if (flowType == FlowType.PUSH) {

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/transferprocess/TransferProcessEventDispatchTest.java

@@ -145,14 +145,13 @@ void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService se
         when(agreement.getPolicy()).thenReturn(Policy.Builder.newInstance().build());
         when(agent.getIdentity()).thenReturn(providerId);
 
+        var transferRequest = createTransferRequest();
         dispatcherRegistry.register(getTestDispatcher());
-        when(policyArchive.findPolicyForContract(matches("contractId"))).thenReturn(mock(Policy.class));
+        when(policyArchive.findPolicyForContract(matches("contractId"))).thenReturn(Policy.Builder.newInstance().target(transferRequest.getAssetId()).build());
         when(negotiationStore.findContractAgreement("contractId")).thenReturn(agreement);
         when(agentService.createFor(token)).thenReturn(agent);
         eventRouter.register(TransferProcessEvent.class, eventSubscriber);
 
-        var transferRequest = createTransferRequest();
-
         var initiateResult = service.initiateTransfer(transferRequest);
 
         await().atMost(TIMEOUT).untilAsserted(() -> {
@@ -215,10 +214,10 @@ void shouldDispatchEventOnTransferProcessTerminated(TransferProcessService servi
                                                         RemoteMessageDispatcherRegistry dispatcherRegistry,
                                                         PolicyArchive policyArchive) {
 
-        when(policyArchive.findPolicyForContract(matches("contractId"))).thenReturn(mock(Policy.class));
+        var transferRequest = createTransferRequest();
+        when(policyArchive.findPolicyForContract(matches("contractId"))).thenReturn(Policy.Builder.newInstance().target(transferRequest.getAssetId()).build());
         dispatcherRegistry.register(getTestDispatcher());
         eventRouter.register(TransferProcessEvent.class, eventSubscriber);
-        var transferRequest = createTransferRequest();
 
         var initiateResult = service.initiateTransfer(transferRequest);
 

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/transferprocess/TransferProcessServiceImplTest.java

@@ -15,6 +15,8 @@
 
 package org.eclipse.edc.connector.controlplane.services.transferprocess;
 
+import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.store.ContractNegotiationStore;
+import org.eclipse.edc.connector.controlplane.contract.spi.types.agreement.ContractAgreement;
 import org.eclipse.edc.connector.controlplane.services.spi.transferprocess.TransferProcessService;
 import org.eclipse.edc.connector.controlplane.transfer.spi.TransferProcessManager;
 import org.eclipse.edc.connector.controlplane.transfer.spi.flow.FlowTypeExtractor;
@@ -26,6 +28,7 @@
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.command.ResumeTransferCommand;
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.command.SuspendTransferCommand;
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.command.TerminateTransferCommand;
+import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.spi.command.CommandHandlerRegistry;
 import org.eclipse.edc.spi.command.CommandResult;
 import org.eclipse.edc.spi.query.Criterion;
@@ -77,9 +80,10 @@ class TransferProcessServiceImplTest {
     private final DataAddressValidatorRegistry dataAddressValidator = mock();
     private final CommandHandlerRegistry commandHandlerRegistry = mock();
     private final FlowTypeExtractor flowTypeExtractor = mock();
+    private final ContractNegotiationStore contractNegotiationStore = mock();
 
     private final TransferProcessService service = new TransferProcessServiceImpl(store, manager, transactionContext,
-            dataAddressValidator, commandHandlerRegistry, flowTypeExtractor);
+            dataAddressValidator, commandHandlerRegistry, flowTypeExtractor, contractNegotiationStore);
 
     @Test
     void findById_whenFound() {
@@ -144,6 +148,7 @@ class InitiateTransfer {
         void shouldInitiateTransfer() {
             var transferRequest = transferRequest();
             var transferProcess = transferProcess();
+            when(contractNegotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(createContractAgreement(transferProcess.getContractId(), transferRequest.getAssetId()));
             when(flowTypeExtractor.extract(any())).thenReturn(StatusResult.success(FlowType.PUSH));
             when(dataAddressValidator.validateDestination(any())).thenReturn(ValidationResult.success());
             when(manager.initiateConsumerRequest(transferRequest)).thenReturn(StatusResult.success(transferProcess));
@@ -155,7 +160,7 @@ void shouldInitiateTransfer() {
         }
 
         @Test
-        void shouldFail_whenDestinationIsNotValid() {
+        void shouldFail_whenContractAgreementNotFound() {
             when(flowTypeExtractor.extract(any())).thenReturn(StatusResult.success(FlowType.PUSH));
             when(dataAddressValidator.validateDestination(any())).thenReturn(ValidationResult.failure(violation("invalid data address", "path")));
 
@@ -164,19 +169,57 @@ void shouldFail_whenDestinationIsNotValid() {
             assertThat(result).isFailed()
                     .extracting(ServiceFailure::getReason)
                     .isEqualTo(BAD_REQUEST);
+            assertThat(result.getFailureMessages()).containsExactly("Contract agreement with id %s not found".formatted(transferRequest().getContractId()));
             verifyNoInteractions(manager);
         }
 
         @Test
-        void shouldFail_whenDataDestinationNotPassedAndFlowTypeIsPush() {
+        void shouldFail_whenTpAssetIdNotEqualToAgreementAssetId() {
+            var transferRequest = transferRequest();
+            var transferProcess = transferProcess();
+            when(contractNegotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(createContractAgreement(transferProcess.getContractId(), "other-asset-id"));
+            when(flowTypeExtractor.extract(any())).thenReturn(StatusResult.success(FlowType.PUSH));
+            when(dataAddressValidator.validateDestination(any())).thenReturn(ValidationResult.success());
+            when(manager.initiateConsumerRequest(transferRequest)).thenReturn(StatusResult.success(transferProcess));
+
+            var result = service.initiateTransfer(transferRequest);
+
+            assertThat(result).isFailed()
+                    .extracting(ServiceFailure::getReason)
+                    .isEqualTo(BAD_REQUEST);
+            assertThat(result.getFailureMessages()).containsExactly("Asset id %s in contract agreement does not match asset id in transfer request %s".formatted("other-asset-id", transferRequest.getAssetId()));
+            verifyNoInteractions(manager);
+        }
+
+        @Test
+        void shouldFail_whenDestinationIsNotValid() {
+            var transferRequest = transferRequest();
+            when(contractNegotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(createContractAgreement(transferRequest.getContractId(), transferRequest.getAssetId()));
             when(flowTypeExtractor.extract(any())).thenReturn(StatusResult.success(FlowType.PUSH));
-            var request = TransferRequest.Builder.newInstance()
+            when(dataAddressValidator.validateDestination(any())).thenReturn(ValidationResult.failure(violation("invalid data address", "path")));
+
+            var result = service.initiateTransfer(transferRequest);
+
+            assertThat(result).isFailed()
+                    .extracting(ServiceFailure::getReason)
+                    .isEqualTo(BAD_REQUEST);
+            assertThat(result.getFailureMessages()).containsExactly("invalid data address");
+            verifyNoInteractions(manager);
+        }
+
+        @Test
+        void shouldFail_whenDataDestinationNotPassedAndFlowTypeIsPush() {
+            var transferRequest = TransferRequest.Builder.newInstance()
                     .transferType("any")
+                    .assetId(UUID.randomUUID().toString())
                     .build();
+            when(contractNegotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(createContractAgreement(transferRequest.getContractId(), transferRequest.getAssetId()));
+            when(flowTypeExtractor.extract(any())).thenReturn(StatusResult.success(FlowType.PUSH));
 
-            var result = service.initiateTransfer(request);
+            var result = service.initiateTransfer(transferRequest);
 
             assertThat(result).isFailed().extracting(ServiceFailure::getReason).isEqualTo(BAD_REQUEST);
+            assertThat(result.getFailureMessages()).containsExactly("For PUSH transfers dataDestination must be defined");
             verifyNoInteractions(manager);
         }
     }
@@ -312,6 +355,17 @@ private TransferProcess transferProcess(TransferProcessStates state, String id)
     private TransferRequest transferRequest() {
         return TransferRequest.Builder.newInstance()
                 .dataDestination(DataAddress.Builder.newInstance().type("type").build())
+                .assetId(UUID.randomUUID().toString())
+                .build();
+    }
+
+    private ContractAgreement createContractAgreement(String agreementId, String assetId) {
+        return ContractAgreement.Builder.newInstance()
+                .id(agreementId)
+                .providerId(UUID.randomUUID().toString())
+                .consumerId(UUID.randomUUID().toString())
+                .assetId(assetId)
+                .policy(Policy.Builder.newInstance().build())
                 .build();
     }
 

core/control-plane/control-plane-contract/src/main/java/org/eclipse/edc/connector/controlplane/contract/validation/ContractValidationServiceImpl.java

@@ -64,10 +64,8 @@ public ContractValidationServiceImpl(AssetIndex assetIndex,
     @Override
     public @NotNull Result<ValidatedConsumerOffer> validateInitialOffer(ParticipantAgent agent, ValidatableConsumerOffer consumerOffer) {
         return validateInitialOffer(consumerOffer, agent)
-                .map(sanitizedPolicy -> {
-                    var offer = createContractOffer(sanitizedPolicy, consumerOffer.getOfferId());
-                    return new ValidatedConsumerOffer(agent.getIdentity(), offer);
-                });
+                .compose(policy -> createContractOffer(policy, consumerOffer.getOfferId()))
+                .map(contractOffer -> new ValidatedConsumerOffer(agent.getIdentity(), contractOffer));
     }
 
     @Override
@@ -163,12 +161,15 @@ private Result<Policy> evaluatePolicy(Policy policy, String scope, ParticipantAg
     }
 
     @NotNull
-    private ContractOffer createContractOffer(Policy policy, ContractOfferId contractOfferId) {
-        return ContractOffer.Builder.newInstance()
+    private Result<ContractOffer> createContractOffer(Policy policy, ContractOfferId contractOfferId) {
+        if (!contractOfferId.assetIdPart().equals(policy.getTarget())) {
+            return Result.failure("Policy target %s does not match the asset ID in the contract offer %s".formatted(policy.getTarget(), contractOfferId.assetIdPart()));
+        }
+        return Result.success(ContractOffer.Builder.newInstance()
                 .id(contractOfferId.toString())
                 .policy(policy)
                 .assetId(contractOfferId.assetIdPart())
-                .build();
+                .build());
     }
 
 }

system-tests/management-api/management-api-test-runner/src/test/java/org/eclipse/edc/test/e2e/managementapi/TransferProcessApiEndToEndTest.java

@@ -19,11 +19,15 @@
 import jakarta.json.JsonArray;
 import jakarta.json.JsonArrayBuilder;
 import jakarta.json.JsonObject;
+import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.store.ContractNegotiationStore;
+import org.eclipse.edc.connector.controlplane.contract.spi.types.agreement.ContractAgreement;
+import org.eclipse.edc.connector.controlplane.contract.spi.types.negotiation.ContractNegotiation;
 import org.eclipse.edc.connector.controlplane.transfer.spi.store.TransferProcessStore;
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.TransferProcess;
 import org.eclipse.edc.jsonld.util.JacksonJsonLd;
 import org.eclipse.edc.junit.annotations.EndToEndTest;
 import org.eclipse.edc.junit.annotations.PostgresqlIntegrationTest;
+import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.callback.CallbackAddress;
 import org.junit.jupiter.api.Nested;
@@ -106,7 +110,18 @@ void getState(ManagementEndToEndTestContext context, TransferProcessStore store)
         }
 
         @Test
-        void create(ManagementEndToEndTestContext context, TransferProcessStore store) {
+        void create(ManagementEndToEndTestContext context, TransferProcessStore transferProcessStore, ContractNegotiationStore contractNegotiationStore) {
+            var assetId = UUID.randomUUID().toString();
+            var contractId = UUID.randomUUID().toString();
+            var contractNegotiation = ContractNegotiation.Builder.newInstance()
+                    .id(UUID.randomUUID().toString())
+                    .counterPartyId("counterPartyId")
+                    .counterPartyAddress("http://counterparty")
+                    .protocol("dataspace-protocol-http")
+                    .contractAgreement(createContractAgreement(contractId, assetId).build())
+                    .build();
+            contractNegotiationStore.save(contractNegotiation);
+
             var requestBody = createObjectBuilder()
                     .add(CONTEXT, createObjectBuilder().add(VOCAB, EDC_NAMESPACE))
                     .add(TYPE, "TransferRequest")
@@ -122,8 +137,8 @@ void create(ManagementEndToEndTestContext context, TransferProcessStore store) {
                     .add("callbackAddresses", createCallbackAddress())
                     .add("protocol", "dataspace-protocol-http")
                     .add("counterPartyAddress", "http://connector-address")
-                    .add("contractId", "contractId")
-                    .add("assetId", "assetId")
+                    .add("contractId", contractId)
+                    .add("assetId", assetId)
                     .build();
 
             var id = context.baseRequest()
@@ -135,7 +150,7 @@ void create(ManagementEndToEndTestContext context, TransferProcessStore store) {
                     .statusCode(200)
                     .extract().jsonPath().getString(ID);
 
-            assertThat(store.findById(id)).isNotNull();
+            assertThat(transferProcessStore.findById(id)).isNotNull();
         }
 
         @Test
@@ -277,16 +292,27 @@ private JsonArrayBuilder createCallbackAddress() {
                     .add(URI, "http://test.local/")
                     .add(EVENTS, Json.createArrayBuilder().build()));
         }
+
+        private ContractAgreement.Builder createContractAgreement(String contractId, String assetId) {
+            return ContractAgreement.Builder.newInstance()
+                    .id(contractId)
+                    .providerId("providerId")
+                    .consumerId("consumerId")
+                    .policy(Policy.Builder.newInstance().target(assetId).build())
+                    .assetId(assetId);
+        }
     }
 
     @Nested
     @EndToEndTest
     @ExtendWith(ManagementEndToEndExtension.InMemory.class)
-    class InMemory extends Tests { }
+    class InMemory extends Tests {
+    }
 
     @Nested
     @PostgresqlIntegrationTest
     @ExtendWith(ManagementEndToEndExtension.Postgres.class)
-    class Postgres extends Tests { }
+    class Postgres extends Tests {
+    }
 
 }