fix: enforce consistency of contract negotiation request and transfer…

… request

DEPENDENCIES

@@ -80,7 +80,7 @@ maven/mavencentral/com.jcraft/jzlib/1.1.3, BSD-2-Clause, approved, CQ6218
 maven/mavencentral/com.lmax/disruptor/3.4.4, Apache-2.0, approved, clearlydefined
 maven/mavencentral/com.networknt/json-schema-validator/1.0.76, Apache-2.0, approved, CQ22638
 maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.28, Apache-2.0, approved, clearlydefined
-maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.40, , restricted, clearlydefined
+maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.40, Apache-2.0, approved, #15156
 maven/mavencentral/com.puppycrawl.tools/checkstyle/10.17.0, LGPL-2.1-or-later AND (Apache-2.0 AND LGPL-2.1-or-later) AND Apache-2.0, approved, #15077
 maven/mavencentral/com.samskivert/jmustache/1.15, BSD-2-Clause, approved, clearlydefined
 maven/mavencentral/com.squareup.okhttp3/okhttp-dnsoverhttps/4.12.0, Apache-2.0, approved, #11159

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

@@ -137,22 +137,22 @@ void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService se
 
         when(identityService.verifyJwtToken(eq(tokenRepresentation), isA(VerificationContext.class))).thenReturn(Result.success(token));
 
+        var transferRequest = createTransferRequest();
         var agent = mock(ParticipantAgent.class);
         var agreement = mock(ContractAgreement.class);
         var providerId = "ProviderId";
 
+        when(agreement.getAssetId()).thenReturn(transferRequest.getAssetId());
         when(agreement.getProviderId()).thenReturn(providerId);
         when(agreement.getPolicy()).thenReturn(Policy.Builder.newInstance().build());
         when(agent.getIdentity()).thenReturn(providerId);
 
         dispatcherRegistry.register(getTestDispatcher());
-        when(policyArchive.findPolicyForContract(matches("contractId"))).thenReturn(mock(Policy.class));
-        when(negotiationStore.findContractAgreement("contractId")).thenReturn(agreement);
+        when(policyArchive.findPolicyForContract(matches(transferRequest.getContractId()))).thenReturn(Policy.Builder.newInstance().target(transferRequest.getAssetId()).build());
+        when(negotiationStore.findContractAgreement(transferRequest.getContractId())).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(() -> {
@@ -196,10 +196,13 @@ void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService se
     }
 
     @Test
-    void shouldTerminateOnInvalidPolicy(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry) {
+    void shouldTerminateOnInvalidPolicy(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry, ContractNegotiationStore negotiationStore) {
         dispatcherRegistry.register(getTestDispatcher());
         eventRouter.register(TransferProcessEvent.class, eventSubscriber);
         var transferRequest = createTransferRequest();
+        var agreement = mock(ContractAgreement.class);
+        when(agreement.getAssetId()).thenReturn(transferRequest.getAssetId());
+        when(negotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(agreement);
 
         service.initiateTransfer(transferRequest);
 
@@ -213,12 +216,16 @@ void shouldTerminateOnInvalidPolicy(TransferProcessService service, EventRouter
     void shouldDispatchEventOnTransferProcessTerminated(TransferProcessService service,
                                                         EventRouter eventRouter,
                                                         RemoteMessageDispatcherRegistry dispatcherRegistry,
-                                                        PolicyArchive policyArchive) {
+                                                        PolicyArchive policyArchive,
+                                                        ContractNegotiationStore negotiationStore) {
 
-        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());
+        var agreement = mock(ContractAgreement.class);
+        when(agreement.getAssetId()).thenReturn(transferRequest.getAssetId());
+        when(negotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(agreement);
         dispatcherRegistry.register(getTestDispatcher());
         eventRouter.register(TransferProcessEvent.class, eventSubscriber);
-        var transferRequest = createTransferRequest();
 
         var initiateResult = service.initiateTransfer(transferRequest);
 
@@ -234,10 +241,13 @@ void shouldDispatchEventOnTransferProcessTerminated(TransferProcessService servi
     }
 
     @Test
-    void shouldDispatchEventOnTransferProcessFailure(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry) {
+    void shouldDispatchEventOnTransferProcessFailure(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry, ContractNegotiationStore negotiationStore) {
         dispatcherRegistry.register(getTestDispatcher());
         eventRouter.register(TransferProcessEvent.class, eventSubscriber);
         var transferRequest = createTransferRequest();
+        var agreement = mock(ContractAgreement.class);
+        when(agreement.getAssetId()).thenReturn(transferRequest.getAssetId());
+        when(negotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(agreement);
 
         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());
     }
 
 }