feat: add BPN as header in the call from dataplane to http source

edc-extensions/provision-additional-headers/README.md

@@ -8,3 +8,4 @@ This gives for example the provider backend service the possibility to audit the
 The following headers are added to the `HttpDataAddress`:
 
 - `Edc-Contract-Agreement-Id`: the id of the contract agreement
+- `Edc-Bpn`: the BPN of the consumer

edc-extensions/provision-additional-headers/build.gradle.kts

@@ -23,15 +23,9 @@ plugins {
 }
 
 dependencies {
+    implementation(libs.edc.spi.controlplane)
     implementation(libs.edc.spi.core)
     implementation(libs.edc.spi.transfer)
 
-    testImplementation(libs.awaitility)
     testImplementation(libs.edc.junit)
-
-    testImplementation(libs.edc.core.controlplane)
-    testImplementation(libs.edc.dpf.selector.core)
-    testImplementation(libs.edc.dsp)
-    testImplementation(libs.edc.iam.mock)
-    testImplementation(libs.mockito.inline)
 }

edc-extensions/provision-additional-headers/src/main/java/org/eclipse/tractusx/edc/provision/additionalheaders/AdditionalHeadersProvisioner.java

@@ -32,9 +32,7 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 
-public class AdditionalHeadersProvisioner
-        implements Provisioner<
-        AdditionalHeadersResourceDefinition, AdditionalHeadersProvisionedResource> {
+public class AdditionalHeadersProvisioner implements Provisioner<AdditionalHeadersResourceDefinition, AdditionalHeadersProvisionedResource> {
 
     @Override
     public boolean canProvision(ResourceDefinition resourceDefinition) {
@@ -53,6 +51,7 @@ public CompletableFuture<StatusResult<ProvisionResponse>> provision(AdditionalHe
                 HttpDataAddress.Builder.newInstance()
                         .copyFrom(resourceDefinition.getDataAddress())
                         .addAdditionalHeader("Edc-Contract-Agreement-Id", resourceDefinition.getContractId())
+                        .addAdditionalHeader("Edc-Bpn", resourceDefinition.getBpn())
                         .build();
 
         var provisioned =

edc-extensions/provision-additional-headers/src/main/java/org/eclipse/tractusx/edc/provision/additionalheaders/AdditionalHeadersResourceDefinition.java

@@ -33,6 +33,7 @@ class AdditionalHeadersResourceDefinition extends ResourceDefinition {
 
     private String contractId;
     private DataAddress dataAddress;
+    private String bpn;
 
     @Override
     public Builder toBuilder() {
@@ -47,6 +48,10 @@ public String getContractId() {
         return contractId;
     }
 
+    public String getBpn() {
+        return bpn;
+    }
+
     @JsonPOJOBuilder(withPrefix = "")
     public static class Builder
             extends ResourceDefinition.Builder<AdditionalHeadersResourceDefinition, Builder> {
@@ -61,12 +66,17 @@ public static Builder newInstance() {
         }
 
         public Builder contractId(String contractId) {
-            this.resourceDefinition.contractId = contractId;
+            resourceDefinition.contractId = contractId;
             return this;
         }
 
         public Builder dataAddress(DataAddress dataAddress) {
-            this.resourceDefinition.dataAddress = dataAddress;
+            resourceDefinition.dataAddress = dataAddress;
+            return this;
+        }
+
+        public Builder bpn(String bpn) {
+            resourceDefinition.bpn = bpn;
             return this;
         }
     }

edc-extensions/provision-additional-headers/src/main/java/org/eclipse/tractusx/edc/provision/additionalheaders/AdditionalHeadersResourceDefinitionGenerator.java

@@ -20,17 +20,26 @@
 
 package org.eclipse.tractusx.edc.provision.additionalheaders;
 
+import org.eclipse.edc.connector.contract.spi.types.agreement.ContractAgreement;
+import org.eclipse.edc.connector.spi.contractagreement.ContractAgreementService;
 import org.eclipse.edc.connector.transfer.spi.provision.ProviderResourceDefinitionGenerator;
 import org.eclipse.edc.connector.transfer.spi.types.DataRequest;
 import org.eclipse.edc.connector.transfer.spi.types.ResourceDefinition;
 import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.jetbrains.annotations.Nullable;
 
+import java.util.Optional;
 import java.util.UUID;
 
 class AdditionalHeadersResourceDefinitionGenerator implements ProviderResourceDefinitionGenerator {
 
+    private final ContractAgreementService contractAgreementService;
+
+    AdditionalHeadersResourceDefinitionGenerator(ContractAgreementService contractAgreementService) {
+        this.contractAgreementService = contractAgreementService;
+    }
+
     @Override
     public boolean canGenerate(DataRequest dataRequest, DataAddress dataAddress, Policy policy) {
         return "HttpData".equals(dataAddress.getType());
@@ -39,10 +48,16 @@ public boolean canGenerate(DataRequest dataRequest, DataAddress dataAddress, Pol
     @Override
     public @Nullable ResourceDefinition generate(
             DataRequest dataRequest, DataAddress dataAddress, Policy policy) {
+        var bpn = Optional.of(dataRequest.getContractId())
+                .map(contractAgreementService::findById)
+                .map(ContractAgreement::getConsumerId)
+                .orElse(null);
+
         return AdditionalHeadersResourceDefinition.Builder.newInstance()
                 .id(UUID.randomUUID().toString())
                 .dataAddress(dataAddress)
                 .contractId(dataRequest.getContractId())
+                .bpn(bpn)
                 .build();
     }
 }

edc-extensions/provision-additional-headers/src/main/java/org/eclipse/tractusx/edc/provision/additionalheaders/ProvisionAdditionalHeadersExtension.java

@@ -20,6 +20,7 @@
 
 package org.eclipse.tractusx.edc.provision.additionalheaders;
 
+import org.eclipse.edc.connector.spi.contractagreement.ContractAgreementService;
 import org.eclipse.edc.connector.transfer.spi.provision.ProvisionManager;
 import org.eclipse.edc.connector.transfer.spi.provision.ResourceManifestGenerator;
 import org.eclipse.edc.runtime.metamodel.annotation.Inject;
@@ -38,10 +39,13 @@ public class ProvisionAdditionalHeadersExtension implements ServiceExtension {
     @Inject
     private TypeManager typeManager;
 
+    @Inject
+    private ContractAgreementService contractAgreementService;
+
     @Override
     public void initialize(ServiceExtensionContext context) {
         typeManager.registerTypes(AdditionalHeadersResourceDefinition.class, AdditionalHeadersProvisionedResource.class);
-        resourceManifestGenerator.registerGenerator(new AdditionalHeadersResourceDefinitionGenerator());
+        resourceManifestGenerator.registerGenerator(new AdditionalHeadersResourceDefinitionGenerator(contractAgreementService));
         provisionManager.register(new AdditionalHeadersProvisioner());
     }
 }

edc-extensions/provision-additional-headers/src/test/java/org/eclipse/tractusx/edc/provision/additionalheaders/AdditionalHeadersProvisionerTest.java

@@ -49,19 +49,19 @@ void canProvisionAdditionalHeadersResourceDefinition() {
 
     @Test
     void cannotDeprovisionAdditionalHeadersResourceDefinition() {
-        assertThat(provisioner.canDeprovision(mock(AdditionalHeadersProvisionedResource.class)))
-                .isFalse();
+        assertThat(provisioner.canDeprovision(mock(AdditionalHeadersProvisionedResource.class))).isFalse();
         assertThat(provisioner.canDeprovision(mock(ProvisionedResource.class))).isFalse();
     }
 
     @Test
-    void shouldAddContractIdAdditionalHeader() {
+    void shouldAddAdditionalHeaders() {
         var address = HttpDataAddress.Builder.newInstance().baseUrl("http://any").build();
         var resourceDefinition =
                 AdditionalHeadersResourceDefinition.Builder.newInstance()
                         .id(UUID.randomUUID().toString())
                         .transferProcessId(UUID.randomUUID().toString())
                         .contractId("contractId")
+                        .bpn("bpn")
                         .dataAddress(address)
                         .build();
 
@@ -77,6 +77,7 @@ void shouldAddContractIdAdditionalHeader() {
                 .extracting(a -> HttpDataAddress.Builder.newInstance().copyFrom(a).build())
                 .extracting(HttpDataAddress::getAdditionalHeaders)
                 .asInstanceOf(map(String.class, String.class))
-                .containsEntry("Edc-Contract-Agreement-Id", "contractId");
+                .containsEntry("Edc-Contract-Agreement-Id", "contractId")
+                .containsEntry("Edc-Bpn", "bpn");
     }
 }

edc-extensions/provision-additional-headers/src/test/java/org/eclipse/tractusx/edc/provision/additionalheaders/AdditionalHeadersResourceDefinitionGeneratorTest.java

@@ -20,6 +20,10 @@
 
 package org.eclipse.tractusx.edc.provision.additionalheaders;
 
+import org.eclipse.edc.connector.contract.spi.ContractId;
+import org.eclipse.edc.connector.contract.spi.types.agreement.ContractAgreement;
+import org.eclipse.edc.connector.spi.contractagreement.ContractAgreementService;
+import org.eclipse.edc.connector.transfer.spi.provision.ProviderResourceDefinitionGenerator;
 import org.eclipse.edc.connector.transfer.spi.types.DataRequest;
 import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.spi.types.domain.DataAddress;
@@ -30,11 +34,15 @@
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 import static org.assertj.core.api.InstanceOfAssertFactories.type;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 class AdditionalHeadersResourceDefinitionGeneratorTest {
 
-    private final AdditionalHeadersResourceDefinitionGenerator generator =
-            new AdditionalHeadersResourceDefinitionGenerator();
+    private final ContractAgreementService contractAgreementService = mock();
+    private final ProviderResourceDefinitionGenerator generator = new AdditionalHeadersResourceDefinitionGenerator(contractAgreementService);
 
     @Test
     void canGenerate_shouldReturnFalseForNotHttpDataAddresses() {
@@ -73,16 +81,33 @@ void shouldCreateResourceDefinitionWithDataAddress() {
                 DataRequest.Builder.newInstance()
                         .id(UUID.randomUUID().toString())
                         .dataDestination(dataAddress)
+                        .contractId("contractId")
                         .build();
         var build = Policy.Builder.newInstance().build();
+        when(contractAgreementService.findById(any())).thenReturn(contractAgreementWithBpn("bpn"));
 
         var result = generator.generate(dataRequest, dataAddress, build);
 
         assertThat(result)
                 .asInstanceOf(type(AdditionalHeadersResourceDefinition.class))
-                .extracting(AdditionalHeadersResourceDefinition::getDataAddress)
-                .extracting(address -> HttpDataAddress.Builder.newInstance().copyFrom(address).build())
-                .extracting(HttpDataAddress::getBaseUrl)
-                .isEqualTo("http://any");
+                .satisfies(resourceDefinition -> {
+                    assertThat(resourceDefinition.getDataAddress())
+                            .extracting(address -> HttpDataAddress.Builder.newInstance().copyFrom(address).build())
+                            .extracting(HttpDataAddress::getBaseUrl)
+                            .isEqualTo("http://any");
+                    assertThat(resourceDefinition.getContractId()).isEqualTo("contractId");
+                    assertThat(resourceDefinition.getBpn()).isEqualTo("bpn");
+                });
+        verify(contractAgreementService).findById("contractId");
+    }
+
+    private static ContractAgreement contractAgreementWithBpn(String bpn) {
+        return ContractAgreement.Builder.newInstance()
+                .id(UUID.randomUUID().toString())
+                .consumerId(bpn)
+                .providerId("providerId")
+                .assetId("assetId")
+                .policy(Policy.Builder.newInstance().build())
+                .build();
     }
 }

edc-extensions/provision-additional-headers/src/test/java/org/eclipse/tractusx/edc/provision/additionalheaders/ProvisionAdditionalHeadersExtensionTest.java

@@ -20,97 +20,35 @@
 
 package org.eclipse.tractusx.edc.provision.additionalheaders;
 
-import org.eclipse.edc.connector.contract.spi.negotiation.store.ContractNegotiationStore;
-import org.eclipse.edc.connector.contract.spi.types.agreement.ContractAgreement;
-import org.eclipse.edc.connector.contract.spi.validation.ContractValidationService;
-import org.eclipse.edc.connector.spi.transferprocess.TransferProcessProtocolService;
-import org.eclipse.edc.connector.transfer.spi.flow.DataFlowController;
-import org.eclipse.edc.connector.transfer.spi.flow.DataFlowManager;
-import org.eclipse.edc.connector.transfer.spi.types.DataFlowResponse;
-import org.eclipse.edc.connector.transfer.spi.types.protocol.TransferRequestMessage;
-import org.eclipse.edc.junit.annotations.ComponentTest;
-import org.eclipse.edc.junit.extensions.EdcExtension;
-import org.eclipse.edc.policy.model.Policy;
-import org.eclipse.edc.service.spi.result.ServiceResult;
-import org.eclipse.edc.spi.asset.AssetIndex;
-import org.eclipse.edc.spi.iam.ClaimToken;
-import org.eclipse.edc.spi.message.RemoteMessageDispatcherRegistry;
-import org.eclipse.edc.spi.response.StatusResult;
-import org.eclipse.edc.spi.result.Result;
-import org.eclipse.edc.spi.types.domain.DataAddress;
-import org.eclipse.edc.spi.types.domain.asset.Asset;
+import org.eclipse.edc.connector.transfer.spi.provision.ProvisionManager;
+import org.eclipse.edc.connector.transfer.spi.provision.ResourceManifestGenerator;
+import org.eclipse.edc.junit.extensions.DependencyInjectionExtension;
+import org.eclipse.edc.spi.system.ServiceExtensionContext;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-import java.time.Duration;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.awaitility.Awaitility.await;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.argThat;
+import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
-@ComponentTest
-@ExtendWith(EdcExtension.class)
+@ExtendWith(DependencyInjectionExtension.class)
 class ProvisionAdditionalHeadersExtensionTest {
 
-    private final DataFlowController dataFlowController = mock(DataFlowController.class);
-    private final RemoteMessageDispatcherRegistry dispatcherRegistry = mock(RemoteMessageDispatcherRegistry.class);
-
-    private final ContractNegotiationStore contractNegotiationStore = mock(ContractNegotiationStore.class);
-    private final ContractValidationService contractValidationService = mock(ContractValidationService.class);
+    private final ResourceManifestGenerator resourceManifestGenerator = mock();
+    private final ProvisionManager provisionManager = mock();
 
     @BeforeEach
-    void setUp(EdcExtension extension) {
-        extension.setConfiguration(Map.of("edc.ids.id", "urn:connector:test"));
-        when(dataFlowController.canHandle(any(), any())).thenReturn(true);
-        when(dataFlowController.initiateFlow(any(), any(), any())).thenReturn(StatusResult.success(DataFlowResponse.Builder.newInstance().build()));
-        extension.registerServiceMock(RemoteMessageDispatcherRegistry.class, dispatcherRegistry);
-        extension.registerServiceMock(ContractNegotiationStore.class, contractNegotiationStore);
-        extension.registerServiceMock(ContractValidationService.class, contractValidationService);
+    void setUp(ServiceExtensionContext context) {
+        context.registerService(ResourceManifestGenerator.class, resourceManifestGenerator);
+        context.registerService(ProvisionManager.class, provisionManager);
     }
 
     @Test
-    void shouldPutContractIdAsHeaderInDataAddress(
-            TransferProcessProtocolService transferProcessProtocolService,
-            AssetIndex assetIndex,
-            DataFlowManager dataFlowManager) {
-
-        var agreement = ContractAgreement.Builder.newInstance()
-                .id("aContractId")
-                .providerId("provider")
-                .consumerId("consumer")
-                .policy(Policy.Builder.newInstance().build())
-                .assetId("assetId")
-                .build();
-
-        when(dispatcherRegistry.send(any(), any())).thenReturn(CompletableFuture.completedFuture(null));
-        when(contractNegotiationStore.findContractAgreement(any())).thenReturn(agreement);
-        when(contractValidationService.validateAgreement(any(), any())).thenReturn(Result.success(agreement));
-
-        dataFlowManager.register(dataFlowController);
-        var asset = Asset.Builder.newInstance().id("assetId").build();
-        var dataAddress = DataAddress.Builder.newInstance().type("HttpData").build();
-        assetIndex.create(asset, dataAddress);
-
-        var transferMessage = TransferRequestMessage.Builder.newInstance()
-                .id("id")
-                .protocol("protocol")
-                .contractId("1:assetId:aContractId")
-                .dataDestination(DataAddress.Builder.newInstance().type("HttpProxy").build())
-                .callbackAddress("callbackAddress")
-                .build();
-
-        var result = transferProcessProtocolService.notifyRequested(transferMessage, ClaimToken.Builder.newInstance().build());
-
-        assertThat(result).matches(ServiceResult::succeeded);
+    void initializeShouldRegisterProvisioner(ProvisionAdditionalHeadersExtension extension, ServiceExtensionContext context) {
+        extension.initialize(context);
 
-        await().atMost(Duration.ofSeconds(5))
-                .untilAsserted(() -> verify(dataFlowController).initiateFlow(any(), argThat(it -> "1:assetId:aContractId".equals(it.getProperty("header:Edc-Contract-Agreement-Id"))), any()));
+        verify(resourceManifestGenerator).registerGenerator(isA(AdditionalHeadersResourceDefinitionGenerator.class));
+        verify(provisionManager).register(isA(AdditionalHeadersProvisioner.class));
     }
 }

edc-tests/e2e-tests/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AbstractHttpConsumerPullWithProxyTest.java

@@ -128,6 +128,7 @@ void transferData_privateBackend() throws IOException, InterruptedException {
         var rq = server.takeRequest();
         assertThat(rq.getHeader(authCodeHeaderName)).isEqualTo(authCode);
         assertThat(rq.getHeader("Edc-Contract-Agreement-Id")).isEqualTo(contractAgreementId.get());
+        assertThat(rq.getHeader("Edc-Bpn")).isEqualTo(SOKRATES.getBpn());
         assertThat(rq.getMethod()).isEqualToIgnoringCase("GET");
     }