diff --git a/.github/workflows/verify.yaml b/.github/workflows/verify.yaml
index 6d88ae3c9..6596f4f0b 100644
--- a/.github/workflows/verify.yaml
+++ b/.github/workflows/verify.yaml
@@ -149,3 +149,8 @@ jobs:
         run: |
           ./gradlew compileJava compileTestJava
           ./gradlew -p edc-tests/edc-dataplane test -DincludeTags="CloudTransferTest"
+
+      - name: Run Azure/S3 End-To-End tests
+        run: |
+          ./gradlew compileJava compileTestJava
+          ./gradlew -p edc-tests/edc-end2end test -DincludeTags="EndToEndTest"
diff --git a/DEPENDENCIES b/DEPENDENCIES
index f35fb2ed8..9973e5265 100644
--- a/DEPENDENCIES
+++ b/DEPENDENCIES
@@ -341,6 +341,7 @@ maven/mavencentral/org.eclipse.edc.aws/aws-spi/0.8.2-20240902-SNAPSHOT, Apache-2
 maven/mavencentral/org.eclipse.edc.aws/data-plane-aws-s3/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc.aws/validator-data-address-s3/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc.azure/azure-blob-core/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc.azure/azure-test/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc.azure/data-plane-azure-storage/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc.azure/vault-azure/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/accesstoken-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -354,12 +355,15 @@ maven/mavencentral/org.eclipse.edc/asset-spi/0.8.2-SNAPSHOT, Apache-2.0, approve
 maven/mavencentral/org.eclipse.edc/auth-configuration/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/auth-delegated/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/auth-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/auth-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/auth-tokenbased/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/autodoc-processor/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/boot-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/boot-lib/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/boot-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/boot-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/boot/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/boot/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/callback-event-dispatcher/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/callback-http-dispatcher/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/callback-static-endpoint/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -447,7 +451,9 @@ maven/mavencentral/org.eclipse.edc/federated-catalog-cache-sql/0.8.2-20240902-SN
 maven/mavencentral/org.eclipse.edc/federated-catalog-core/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/federated-catalog-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/http-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/http-lib/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/http-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/http-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/http/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/iam-mock/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/identity-did-core/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -480,8 +486,11 @@ maven/mavencentral/org.eclipse.edc/json-ld-lib/0.8.2-20240902-SNAPSHOT, Apache-2
 maven/mavencentral/org.eclipse.edc/json-ld-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/json-ld/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/json-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/json-lib/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/junit-base/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/junit-base/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/junit/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/junit/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/jws2020-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/jwt-signer-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/jwt-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -511,6 +520,7 @@ maven/mavencentral/org.eclipse.edc/policy-monitor-core/0.8.2-20240902-SNAPSHOT,
 maven/mavencentral/org.eclipse.edc/policy-monitor-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/policy-monitor-store-sql/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/policy-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/policy-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/presentation-api/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/query-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/runtime-metamodel/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -518,6 +528,7 @@ maven/mavencentral/org.eclipse.edc/runtime-metamodel/0.8.2-SNAPSHOT, Apache-2.0,
 maven/mavencentral/org.eclipse.edc/secrets-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/sql-bootstrapper/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/sql-core/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/sql-core/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/sql-lease/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/sql-pool-apache-commons/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/state-machine-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -525,8 +536,10 @@ maven/mavencentral/org.eclipse.edc/store-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0
 maven/mavencentral/org.eclipse.edc/token-core/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/token-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/transaction-datasource-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/transaction-datasource-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/transaction-local/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/transaction-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/transaction-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/transfer-data-plane-signaling/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/transfer-data-plane-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/transfer-process-api/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
@@ -548,6 +561,7 @@ maven/mavencentral/org.eclipse.edc/verifiable-credentials-spi/0.8.2-20240902-SNA
 maven/mavencentral/org.eclipse.edc/verifiable-credentials/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/verifiable-presentation-lib/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.edc/web-spi/0.8.2-20240902-SNAPSHOT, Apache-2.0, approved, technology.edc
+maven/mavencentral/org.eclipse.edc/web-spi/0.8.2-SNAPSHOT, Apache-2.0, approved, technology.edc
 maven/mavencentral/org.eclipse.jetty.toolchain/jetty-jakarta-servlet-api/5.0.2, EPL-2.0 OR Apache-2.0, approved, rt.jetty
 maven/mavencentral/org.eclipse.jetty.toolchain/jetty-jakarta-websocket-api/2.0.0, EPL-2.0 OR Apache-2.0, approved, rt.jetty
 maven/mavencentral/org.eclipse.jetty.websocket/websocket-core-client/11.0.23, EPL-2.0 OR Apache-2.0, approved, rt.jetty
@@ -607,6 +621,7 @@ maven/mavencentral/org.jetbrains.kotlin/kotlin-stdlib/1.9.10, Apache-2.0, approv
 maven/mavencentral/org.jetbrains/annotations/13.0, Apache-2.0, approved, clearlydefined
 maven/mavencentral/org.jetbrains/annotations/17.0.0, Apache-2.0, approved, clearlydefined
 maven/mavencentral/org.jetbrains/annotations/24.1.0, Apache-2.0, approved, clearlydefined
+maven/mavencentral/org.junit.jupiter/junit-jupiter-api/5.10.1, EPL-2.0, approved, #9714
 maven/mavencentral/org.junit.jupiter/junit-jupiter-api/5.11.0, EPL-2.0, approved, #15935
 maven/mavencentral/org.junit.jupiter/junit-jupiter-engine/5.11.0, EPL-2.0, approved, #15939
 maven/mavencentral/org.junit.jupiter/junit-jupiter-params/5.11.0, EPL-2.0, approved, #15940
@@ -631,6 +646,7 @@ maven/mavencentral/org.ow2.asm/asm/9.1, BSD-3-Clause, approved, CQ23029
 maven/mavencentral/org.ow2.asm/asm/9.5, BSD-3-Clause, approved, #7554
 maven/mavencentral/org.ow2.asm/asm/9.6, BSD-3-Clause, approved, #10776
 maven/mavencentral/org.ow2.asm/asm/9.7, BSD-3-Clause, approved, #14076
+maven/mavencentral/org.postgresql/postgresql/42.7.2, BSD-2-Clause AND Apache-2.0, approved, #11681
 maven/mavencentral/org.postgresql/postgresql/42.7.4, BSD-2-Clause AND Apache-2.0, approved, #11681
 maven/mavencentral/org.reactivestreams/reactive-streams/1.0.4, CC0-1.0, approved, CQ16332
 maven/mavencentral/org.reflections/reflections/0.10.2, Apache-2.0 AND WTFPL, approved, clearlydefined
@@ -659,38 +675,38 @@ maven/mavencentral/org.yaml/snakeyaml/2.3, Apache-2.0 AND (Apache-2.0 AND BSD-3-
 maven/mavencentral/software.amazon.awssdk/annotations/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/annotations/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/apache-client/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/apache-client/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/apache-client/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/arns/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/arns/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/arns/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/auth/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/auth/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/auth/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/aws-core/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/aws-core/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/aws-core/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/aws-query-protocol/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/aws-query-protocol/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/aws-xml-protocol/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/aws-xml-protocol/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/aws-xml-protocol/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/checksums-spi/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/checksums-spi/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/checksums-spi/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/checksums/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/checksums/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/checksums/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/crt-core/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/crt-core/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/crt-core/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/endpoints-spi/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/endpoints-spi/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/endpoints-spi/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/http-auth-aws-eventstream/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/http-auth-aws-eventstream/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/http-auth-aws-eventstream/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/http-auth-aws/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/http-auth-aws/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/http-auth-aws/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/http-auth-spi/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/http-auth-spi/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/http-auth-spi/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/http-auth/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/http-auth/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/http-auth/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/http-client-spi/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/http-client-spi/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/iam/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/identity-spi/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/identity-spi/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/identity-spi/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/json-utils/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/json-utils/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/metrics-spi/2.26.27, Apache-2.0, approved, clearlydefined
@@ -702,19 +718,19 @@ maven/mavencentral/software.amazon.awssdk/profiles/2.27.18, , restricted, clearl
 maven/mavencentral/software.amazon.awssdk/protocol-core/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/protocol-core/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/regions/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/regions/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/regions/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/retries-spi/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/retries-spi/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/retries/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/retries/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/s3-transfer-manager/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/s3/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/s3/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/s3/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/sdk-core/2.26.27, Apache-2.0, approved, #15695
-maven/mavencentral/software.amazon.awssdk/sdk-core/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/sdk-core/2.27.18, Apache-2.0, restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/sts/2.26.27, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.awssdk/third-party-jackson-core/2.26.27, Apache-2.0, approved, #15693
 maven/mavencentral/software.amazon.awssdk/third-party-jackson-core/2.27.18, , restricted, clearlydefined
 maven/mavencentral/software.amazon.awssdk/utils/2.26.27, Apache-2.0, approved, clearlydefined
-maven/mavencentral/software.amazon.awssdk/utils/2.27.18, , restricted, clearlydefined
+maven/mavencentral/software.amazon.awssdk/utils/2.27.18, Apache-2.0, approved, clearlydefined
 maven/mavencentral/software.amazon.eventstream/eventstream/1.0.1, Apache-2.0, approved, clearlydefined
diff --git a/edc-controlplane/edc-controlplane-base/build.gradle.kts b/edc-controlplane/edc-controlplane-base/build.gradle.kts
index 62c9d388f..d1d85969c 100644
--- a/edc-controlplane/edc-controlplane-base/build.gradle.kts
+++ b/edc-controlplane/edc-controlplane-base/build.gradle.kts
@@ -79,6 +79,9 @@ dependencies {
     runtimeOnly(libs.edc.controlplane.callback.dispatcher.event)
     runtimeOnly(libs.edc.controlplane.callback.dispatcher.http)
 
+    // cloud provisioner extensions
+    runtimeOnly(project(":edc-extensions:backport:azblob-provisioner"))
+
     // Federated Catalog Crawler + Query API
     runtimeOnly(project(":edc-extensions:federated-catalog"))
     runtimeOnly(libs.edc.fc.core)
diff --git a/edc-extensions/backport/README.md b/edc-extensions/backport/README.md
new file mode 100644
index 000000000..43df91b98
--- /dev/null
+++ b/edc-extensions/backport/README.md
@@ -0,0 +1,2 @@
+All modules in this directory are only temporary and they should be replaced with their upstream counterparts at the
+earliest opportunity. changes made here should be upstreamed ASAP, lest we run the risk of feature divergence!
\ No newline at end of file
diff --git a/edc-extensions/backport/azblob-provisioner/build.gradle.kts b/edc-extensions/backport/azblob-provisioner/build.gradle.kts
new file mode 100644
index 000000000..45db9bccb
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/build.gradle.kts
@@ -0,0 +1,29 @@
+/*
+ *  Copyright (c) 2020, 2021 Microsoft Corporation
+ *
+ *  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:
+ *       Microsoft Corporation - initial API and implementation
+ *
+ */
+
+plugins {
+    `java-library`
+}
+
+dependencies {
+    api(libs.edc.spi.core)
+    api(libs.edc.azure.blob.core)
+
+    implementation(libs.azure.storage.blob)
+//    implementation(libs.failsafe.core)
+
+    testImplementation(testFixtures(libs.edc.azure.test))
+}
+
+
diff --git a/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/AzureProvisionExtension.java b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/AzureProvisionExtension.java
new file mode 100644
index 000000000..bed0067fe
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/AzureProvisionExtension.java
@@ -0,0 +1,77 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure;
+
+import dev.failsafe.RetryPolicy;
+import org.eclipse.edc.azure.blob.AzureSasToken;
+import org.eclipse.edc.azure.blob.api.BlobStoreApi;
+import org.eclipse.edc.connector.controlplane.transfer.spi.provision.ProvisionManager;
+import org.eclipse.edc.connector.controlplane.transfer.spi.provision.Provisioner;
+import org.eclipse.edc.connector.controlplane.transfer.spi.provision.ResourceManifestGenerator;
+import org.eclipse.edc.connector.provision.azure.blob.ObjectContainerProvisionedResource;
+import org.eclipse.edc.connector.provision.azure.blob.ObjectStorageConsumerResourceDefinitionGenerator;
+import org.eclipse.edc.connector.provision.azure.blob.ObjectStorageProvisioner;
+import org.eclipse.edc.connector.provision.azure.blob.ObjectStorageResourceDefinition;
+import org.eclipse.edc.runtime.metamodel.annotation.Inject;
+import org.eclipse.edc.spi.system.ServiceExtension;
+import org.eclipse.edc.spi.system.ServiceExtensionContext;
+import org.eclipse.edc.spi.types.TypeManager;
+
+/**
+ * Provides data transfer {@link Provisioner}s backed by Azure services.
+ */
+public class AzureProvisionExtension implements ServiceExtension {
+
+    @Inject
+    private BlobStoreApi blobStoreApi;
+
+    @Inject
+    private RetryPolicy<Object> retryPolicy;
+
+    @Inject
+    private ResourceManifestGenerator manifestGenerator;
+
+    @Inject
+    private TypeManager typeManager;
+
+    @Override
+    public String name() {
+        return "Azure Provision";
+    }
+
+    @Override
+    public void initialize(ServiceExtensionContext context) {
+
+        var monitor = context.getMonitor();
+        var provisionManager = context.getService(ProvisionManager.class);
+
+        provisionManager.register(new ObjectStorageProvisioner(retryPolicy, monitor, blobStoreApi));
+
+        // register the generator
+        manifestGenerator.registerGenerator(new ObjectStorageConsumerResourceDefinitionGenerator());
+
+        registerTypes(typeManager);
+    }
+
+    private void registerTypes(TypeManager typeManager) {
+        typeManager.registerTypes(ObjectContainerProvisionedResource.class, ObjectStorageResourceDefinition.class, AzureSasToken.class);
+    }
+
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectContainerProvisionedResource.java b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectContainerProvisionedResource.java
new file mode 100644
index 000000000..9aa1dd24e
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectContainerProvisionedResource.java
@@ -0,0 +1,86 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonPOJOBuilder;
+import org.eclipse.edc.azure.blob.AzureBlobStoreSchema;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ProvisionedDataDestinationResource;
+
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.ACCOUNT_NAME;
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.CONTAINER_NAME;
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.FOLDER_NAME;
+import static org.eclipse.edc.spi.constants.CoreConstants.EDC_NAMESPACE;
+
+@JsonDeserialize(builder = ObjectContainerProvisionedResource.Builder.class)
+@JsonTypeName("dataspaceconnector:objectcontainerprovisionedresource")
+public class ObjectContainerProvisionedResource extends ProvisionedDataDestinationResource {
+
+    private ObjectContainerProvisionedResource() {
+    }
+
+    public String getAccountName() {
+        return getDataAddress().getStringProperty(ACCOUNT_NAME);
+    }
+
+    public String getContainerName() {
+        return getDataAddress().getStringProperty(CONTAINER_NAME);
+    }
+
+    @JsonPOJOBuilder(withPrefix = "")
+    public static class Builder extends ProvisionedDataDestinationResource.Builder<ObjectContainerProvisionedResource, Builder> {
+
+        private Builder() {
+            super(new ObjectContainerProvisionedResource());
+            dataAddressBuilder.type(AzureBlobStoreSchema.TYPE);
+        }
+
+        @JsonCreator
+        public static Builder newInstance() {
+            return new Builder();
+        }
+
+        public Builder accountName(String accountName) {
+            dataAddressBuilder.property(EDC_NAMESPACE + ACCOUNT_NAME, accountName);
+            return this;
+        }
+
+        public Builder containerName(String containerName) {
+            dataAddressBuilder.property(EDC_NAMESPACE + CONTAINER_NAME, containerName);
+            return this;
+        }
+
+        @Override
+        public Builder resourceName(String name) {
+            dataAddressBuilder.keyName(name);
+            super.resourceName(name);
+            return this;
+        }
+
+        public Builder folderName(String folderName) {
+            if (folderName != null) {
+                dataAddressBuilder.property(EDC_NAMESPACE + FOLDER_NAME, folderName);
+            }
+            return this;
+        }
+    }
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGenerator.java b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGenerator.java
new file mode 100644
index 000000000..d06e47cfa
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGenerator.java
@@ -0,0 +1,59 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import org.eclipse.edc.azure.blob.AzureBlobStoreSchema;
+import org.eclipse.edc.connector.controlplane.transfer.spi.provision.ConsumerResourceDefinitionGenerator;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ResourceDefinition;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.TransferProcess;
+import org.eclipse.edc.policy.model.Policy;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.jetbrains.annotations.Nullable;
+
+import static java.util.Optional.ofNullable;
+import static java.util.UUID.randomUUID;
+
+public class ObjectStorageConsumerResourceDefinitionGenerator implements ConsumerResourceDefinitionGenerator {
+
+    @Override
+    public @Nullable ResourceDefinition generate(TransferProcess transferProcess, Policy policy) {
+        var destination = transferProcess.getDataDestination();
+        var id = randomUUID().toString();
+        var account = destination.getStringProperty(AzureBlobStoreSchema.ACCOUNT_NAME);
+        var container = destination.getStringProperty(AzureBlobStoreSchema.CONTAINER_NAME);
+        var folderName = destination.getStringProperty(AzureBlobStoreSchema.FOLDER_NAME);
+
+        if (container == null) {
+            container = randomUUID().toString();
+        }
+        return ObjectStorageResourceDefinition.Builder.newInstance()
+                .id(id)
+                .accountName(account)
+                .containerName(container)
+                .folderName(folderName)
+                .build();
+    }
+
+    @Override
+    public boolean canGenerate(TransferProcess dataRequest, Policy policy) {
+        var type = ofNullable(dataRequest.getDataDestination()).map(DataAddress::getType).orElse(null);
+        return AzureBlobStoreSchema.TYPE.equals(type);
+    }
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageProvisioner.java b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageProvisioner.java
new file mode 100644
index 000000000..5b5807dba
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageProvisioner.java
@@ -0,0 +1,131 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import dev.failsafe.RetryPolicy;
+import org.eclipse.edc.azure.blob.AzureSasToken;
+import org.eclipse.edc.azure.blob.api.BlobStoreApi;
+import org.eclipse.edc.connector.controlplane.transfer.spi.provision.Provisioner;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.DeprovisionedResource;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ProvisionResponse;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ProvisionedResource;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ResourceDefinition;
+import org.eclipse.edc.policy.model.Policy;
+import org.eclipse.edc.spi.monitor.Monitor;
+import org.eclipse.edc.spi.response.StatusResult;
+import org.jetbrains.annotations.NotNull;
+
+import java.time.OffsetDateTime;
+import java.util.concurrent.CompletableFuture;
+
+import static dev.failsafe.Failsafe.with;
+
+public class ObjectStorageProvisioner implements Provisioner<ObjectStorageResourceDefinition, ObjectContainerProvisionedResource> {
+    private final RetryPolicy<Object> retryPolicy;
+    private final Monitor monitor;
+    private final BlobStoreApi blobStoreApi;
+
+    public ObjectStorageProvisioner(RetryPolicy<Object> retryPolicy, Monitor monitor, BlobStoreApi blobStoreApi) {
+        this.retryPolicy = retryPolicy;
+        this.monitor = monitor;
+        this.blobStoreApi = blobStoreApi;
+    }
+
+    @Override
+    public boolean canProvision(ResourceDefinition resourceDefinition) {
+        return resourceDefinition instanceof ObjectStorageResourceDefinition;
+    }
+
+    @Override
+    public boolean canDeprovision(ProvisionedResource resourceDefinition) {
+        return resourceDefinition instanceof ObjectContainerProvisionedResource;
+    }
+
+    @Override
+    public CompletableFuture<StatusResult<ProvisionResponse>> provision(ObjectStorageResourceDefinition resourceDefinition, Policy policy) {
+        String containerName = resourceDefinition.getContainerName();
+        String accountName = resourceDefinition.getAccountName();
+        String folderName = resourceDefinition.getFolderName();
+
+        monitor.debug("Azure Storage Container request submitted: " + containerName);
+
+        OffsetDateTime expiryTime = OffsetDateTime.now().plusHours(1);
+
+        return with(retryPolicy).getAsync(() -> blobStoreApi.exists(accountName, containerName))
+                .thenCompose(exists -> {
+                    if (exists) {
+                        return reusingExistingContainer(containerName);
+                    } else {
+                        return createContainer(containerName, accountName);
+                    }
+                })
+                .thenCompose(empty -> createContainerSasToken(containerName, accountName, expiryTime))
+                .thenApply(writeOnlySas -> {
+                    // Ensure resource name is unique to avoid key collisions in local and remote vaults
+                    String resourceName = resourceDefinition.getId() + "-container";
+                    var resource = ObjectContainerProvisionedResource.Builder.newInstance()
+                            .id(containerName)
+                            .accountName(accountName)
+                            .containerName(containerName)
+                            .folderName(folderName)
+                            .resourceDefinitionId(resourceDefinition.getId())
+                            .transferProcessId(resourceDefinition.getTransferProcessId())
+                            .resourceName(resourceName)
+                            .hasToken(true)
+                            .build();
+
+                    var secretToken = new AzureSasToken("?" + writeOnlySas, expiryTime.toInstant().toEpochMilli());
+
+                    var response = ProvisionResponse.Builder.newInstance().resource(resource).secretToken(secretToken).build();
+                    return StatusResult.success(response);
+                });
+    }
+
+    @Override
+    public CompletableFuture<StatusResult<DeprovisionedResource>> deprovision(ObjectContainerProvisionedResource provisionedResource, Policy policy) {
+        return with(retryPolicy).runAsync(() -> blobStoreApi.deleteContainer(provisionedResource.getAccountName(), provisionedResource.getContainerName()))
+                //the sas token will expire automatically. there is no way of revoking them other than a stored access policy
+                .thenApply(empty -> StatusResult.success(DeprovisionedResource.Builder.newInstance().provisionedResourceId(provisionedResource.getId()).build()));
+    }
+
+    @NotNull
+    private CompletableFuture<Void> reusingExistingContainer(String containerName) {
+        monitor.debug("ObjectStorageProvisioner: re-use existing container " + containerName);
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @NotNull
+    private CompletableFuture<Void> createContainer(String containerName, String accountName) {
+        return with(retryPolicy)
+                .runAsync(() -> {
+                    blobStoreApi.createContainer(accountName, containerName);
+                    monitor.debug("ObjectStorageProvisioner: created a new container " + containerName);
+                });
+    }
+
+    @NotNull
+    private CompletableFuture<String> createContainerSasToken(String containerName, String accountName, OffsetDateTime expiryTime) {
+        return with(retryPolicy)
+                .getAsync(() -> {
+                    monitor.debug("ObjectStorageProvisioner: obtained temporary SAS token (write-only)");
+                    return blobStoreApi.createContainerSasToken(accountName, containerName, "w", expiryTime);
+                });
+    }
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageResourceDefinition.java b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageResourceDefinition.java
new file mode 100644
index 000000000..abec2cc4f
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageResourceDefinition.java
@@ -0,0 +1,86 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ResourceDefinition;
+
+import java.util.Objects;
+
+public class ObjectStorageResourceDefinition extends ResourceDefinition {
+
+    private String containerName;
+    private String accountName;
+    private String folderName;
+
+    public String getContainerName() {
+        return containerName;
+    }
+
+    public String getAccountName() {
+        return accountName;
+    }
+
+    @Override
+    public Builder toBuilder() {
+        return initializeBuilder(new Builder())
+                .containerName(containerName)
+                .folderName(folderName)
+                .accountName(accountName);
+    }
+
+    public String getFolderName() {
+        return folderName;
+    }
+
+    public static class Builder extends ResourceDefinition.Builder<ObjectStorageResourceDefinition, Builder> {
+
+        private Builder() {
+            super(new ObjectStorageResourceDefinition());
+        }
+
+        public static Builder newInstance() {
+            return new Builder();
+        }
+
+        public Builder containerName(String id) {
+            resourceDefinition.containerName = id;
+            return this;
+        }
+
+        public Builder accountName(String accountName) {
+            resourceDefinition.accountName = accountName;
+            return this;
+        }
+
+        public Builder folderName(String folderName) {
+            resourceDefinition.folderName = folderName;
+            return this;
+        }
+
+        @Override
+        protected void verify() {
+            super.verify();
+            Objects.requireNonNull(resourceDefinition.containerName, "containerName");
+            Objects.requireNonNull(resourceDefinition.accountName, "accountName");
+        }
+    }
+
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/main/resources/META-INF/services/org.eclipse.edc.spi.system.ServiceExtension b/edc-extensions/backport/azblob-provisioner/src/main/resources/META-INF/services/org.eclipse.edc.spi.system.ServiceExtension
new file mode 100644
index 000000000..13293b18a
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/main/resources/META-INF/services/org.eclipse.edc.spi.system.ServiceExtension
@@ -0,0 +1,21 @@
+#################################################################################
+#  Copyright (c) 2020,2021 Microsoft Corporation
+#
+#  See the NOTICE file(s) distributed with this work for additional
+#  information regarding copyright ownership.
+#
+#  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.
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+#
+#  SPDX-License-Identifier: Apache-2.0
+#################################################################################
+
+org.eclipse.edc.connector.provision.azure.AzureProvisionExtension
+
diff --git a/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectContainerProvisionedResourceTest.java b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectContainerProvisionedResourceTest.java
new file mode 100644
index 000000000..40d94dfea
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectContainerProvisionedResourceTest.java
@@ -0,0 +1,109 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import org.eclipse.edc.azure.blob.AzureBlobStoreSchema;
+import org.eclipse.edc.json.JacksonTypeManager;
+import org.eclipse.edc.spi.types.TypeManager;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.ACCOUNT_NAME;
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.CONTAINER_NAME;
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.FOLDER_NAME;
+
+class ObjectContainerProvisionedResourceTest {
+
+    private final TypeManager typeManager = new JacksonTypeManager();
+    private ObjectContainerProvisionedResource.Builder builder;
+
+    @BeforeEach
+    void setUp() {
+        typeManager.registerTypes(ObjectContainerProvisionedResource.class);
+
+        builder = ObjectContainerProvisionedResource.Builder.newInstance()
+                .containerName("test-container")
+                .accountName("test-account")
+                .transferProcessId("test-process-id")
+                .resourceDefinitionId("test-resdef-id")
+                .resourceName("test-container")
+                .id("test-id");
+    }
+
+    @Test
+    void createDataDestination() {
+        var dest = builder.build().getDataAddress();
+
+        assertThat(dest.getType()).isEqualTo(AzureBlobStoreSchema.TYPE);
+        assertThat(dest.getKeyName()).isEqualTo("test-container");
+        assertThat(dest.getStringProperty(CONTAINER_NAME)).isEqualTo("test-container");
+        assertThat(dest.getStringProperty(ACCOUNT_NAME)).isEqualTo("test-account");
+        assertThat(dest.getProperties()).doesNotContainKey(FOLDER_NAME);
+    }
+
+    @Test
+    void createDataDestination_withFolder() {
+        var dest = builder.folderName("testfolder").build().getDataAddress();
+
+        assertThat(dest.getType()).isEqualTo(AzureBlobStoreSchema.TYPE);
+        assertThat(dest.getKeyName()).isEqualTo("test-container");
+        assertThat(dest.getStringProperty(CONTAINER_NAME)).isEqualTo("test-container");
+        assertThat(dest.getStringProperty(ACCOUNT_NAME)).isEqualTo("test-account");
+        assertThat(dest.getStringProperty(FOLDER_NAME)).isEqualTo("testfolder");
+    }
+
+
+    @Test
+    void getResourceName() {
+        assertThat(builder.build().getResourceName()).isEqualTo("test-container");
+    }
+
+    @Test
+    void verifySerialization() {
+        var json = typeManager.writeValueAsString(builder.build());
+
+        assertThat(json).isNotNull()
+                .contains("accountName")
+                .contains("containerName");
+    }
+
+    @Test
+    void verifyDeserialization() {
+        var serialized = Map.of(
+                "id", "test-id",
+                "edctype", "dataspaceconnector:objectcontainerprovisionedresource",
+                "transferProcessId", "test-process-id",
+                "resourceDefinitionId", "test-resdef-id",
+                "accountName", "test-account",
+                "containerName", "test-container",
+                "resourceName", "test-container"
+        );
+
+        var res = typeManager.readValue(typeManager.writeValueAsBytes(serialized), ObjectContainerProvisionedResource.class);
+
+        assertThat(res).isNotNull();
+        assertThat(res.getContainerName()).isEqualTo("test-container");
+        assertThat(res.getAccountName()).isEqualTo("test-account");
+        assertThat(res).usingRecursiveComparison().isEqualTo(builder.build());
+    }
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGeneratorTest.java b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGeneratorTest.java
new file mode 100644
index 000000000..c1f5c476a
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGeneratorTest.java
@@ -0,0 +1,132 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import org.eclipse.edc.azure.blob.AzureBlobStoreSchema;
+import org.eclipse.edc.connector.controlplane.asset.spi.domain.Asset;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.TransferProcess;
+import org.eclipse.edc.policy.model.Policy;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+class ObjectStorageConsumerResourceDefinitionGeneratorTest {
+
+    private final ObjectStorageConsumerResourceDefinitionGenerator generator =
+            new ObjectStorageConsumerResourceDefinitionGenerator();
+
+    @Test
+    void generate_withContainerName() {
+        var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)
+                .property(AzureBlobStoreSchema.CONTAINER_NAME, "test-container")
+                .property(AzureBlobStoreSchema.ACCOUNT_NAME, "test-account")
+                .build();
+        var asset = Asset.Builder.newInstance().build();
+        var transferProcess = TransferProcess.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
+        var policy = Policy.Builder.newInstance().build();
+
+        var definition = generator.generate(transferProcess, policy);
+
+        assertThat(definition).isInstanceOf(ObjectStorageResourceDefinition.class);
+        var objectDef = (ObjectStorageResourceDefinition) definition;
+        assertThat(objectDef.getAccountName()).isEqualTo("test-account");
+        assertThat(objectDef.getContainerName()).isEqualTo("test-container");
+        assertThat(objectDef.getId()).satisfies(UUID::fromString);
+        assertThat(objectDef.getFolderName()).isNull();
+    }
+
+    @Test
+    void generate_withContainerName_andFolder() {
+        var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)
+                .property(AzureBlobStoreSchema.CONTAINER_NAME, "test-container")
+                .property(AzureBlobStoreSchema.ACCOUNT_NAME, "test-account")
+                .property(AzureBlobStoreSchema.FOLDER_NAME, "test-folder")
+                .build();
+        var asset = Asset.Builder.newInstance().build();
+        var transferProcess = TransferProcess.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
+        var policy = Policy.Builder.newInstance().build();
+
+        var definition = generator.generate(transferProcess, policy);
+
+        assertThat(definition).isInstanceOf(ObjectStorageResourceDefinition.class);
+        var objectDef = (ObjectStorageResourceDefinition) definition;
+        assertThat(objectDef.getAccountName()).isEqualTo("test-account");
+        assertThat(objectDef.getContainerName()).isEqualTo("test-container");
+        assertThat(objectDef.getId()).satisfies(UUID::fromString);
+        assertThat(objectDef.getFolderName()).isEqualTo("test-folder");
+    }
+
+    @Test
+    void generate_noDataRequestAsParameter() {
+        var policy = Policy.Builder.newInstance().build();
+        assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> generator.generate(null, policy));
+    }
+
+    @Test
+    void generate_withoutContainerName() {
+        var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)
+                .property(AzureBlobStoreSchema.ACCOUNT_NAME, "test-account")
+                .build();
+        var asset = Asset.Builder.newInstance().build();
+        var transferProcess = TransferProcess.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
+        var policy = Policy.Builder.newInstance().build();
+
+        var definition = generator.generate(transferProcess, policy);
+
+        assertThat(definition).isInstanceOf(ObjectStorageResourceDefinition.class);
+        var objectDef = (ObjectStorageResourceDefinition) definition;
+        assertThat(objectDef.getAccountName()).isEqualTo("test-account");
+        assertThat(objectDef.getContainerName()).satisfies(UUID::fromString);
+        assertThat(objectDef.getId()).satisfies(UUID::fromString);
+    }
+
+    @Test
+    void canGenerate() {
+        var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)
+                .property(AzureBlobStoreSchema.ACCOUNT_NAME, "test-account")
+                .build();
+        var asset = Asset.Builder.newInstance().build();
+        var transferProcess = TransferProcess.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
+        var policy = Policy.Builder.newInstance().build();
+
+        var definition = generator.canGenerate(transferProcess, policy);
+
+        assertThat(definition).isTrue();
+    }
+
+    @Test
+    void canGenerate_isNotTypeAzureBlobStoreSchema() {
+        var destination = DataAddress.Builder.newInstance().type("aNonGoogleCloudStorage")
+                .property(AzureBlobStoreSchema.ACCOUNT_NAME, "test-account")
+                .build();
+        var asset = Asset.Builder.newInstance().build();
+        var transferProcess = TransferProcess.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
+        var policy = Policy.Builder.newInstance().build();
+
+        var definition = generator.canGenerate(transferProcess, policy);
+
+        assertThat(definition).isFalse();
+    }
+
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageProvisionerTest.java b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageProvisionerTest.java
new file mode 100644
index 000000000..5a06f8356
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageProvisionerTest.java
@@ -0,0 +1,206 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import com.azure.storage.blob.models.BlobStorageException;
+import dev.failsafe.RetryPolicy;
+import org.eclipse.edc.azure.blob.AzureSasToken;
+import org.eclipse.edc.azure.blob.api.BlobStoreApi;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ProvisionedResource;
+import org.eclipse.edc.connector.controlplane.transfer.spi.types.ResourceDefinition;
+import org.eclipse.edc.policy.model.Policy;
+import org.eclipse.edc.spi.monitor.Monitor;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.eclipse.edc.azure.blob.AzureBlobStoreSchema.FOLDER_NAME;
+import static org.eclipse.edc.spi.constants.CoreConstants.EDC_NAMESPACE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ObjectStorageProvisionerTest {
+
+    private final BlobStoreApi blobStoreApiMock = mock(BlobStoreApi.class);
+    private ObjectStorageProvisioner provisioner;
+    private Policy policy;
+
+    @BeforeEach
+    void setup() {
+        RetryPolicy<Object> retryPolicy = RetryPolicy.builder().withMaxRetries(0).build();
+        provisioner = new ObjectStorageProvisioner(retryPolicy, mock(Monitor.class), blobStoreApiMock);
+        policy = Policy.Builder.newInstance().build();
+    }
+
+    @Test
+    void canProvision() {
+        assertThat(provisioner.canProvision(new ObjectStorageResourceDefinition())).isTrue();
+        assertThat(provisioner.canProvision(new ResourceDefinition() {
+            @Override
+            public <RD extends ResourceDefinition, B extends Builder<RD, B>> B toBuilder() {
+                return null;
+            }
+        })).isFalse();
+    }
+
+    @Test
+    void canDeprovision() {
+        var resource = createProvisionedResource();
+        assertThat(provisioner.canDeprovision(resource)).isTrue();
+        assertThat(provisioner.canDeprovision(new ProvisionedResource() {
+        })).isFalse();
+    }
+
+    @Test
+    void deprovision_should_not_do_anything() {
+        ObjectContainerProvisionedResource resource = createProvisionedResource();
+        var result = provisioner.deprovision(resource, policy);
+
+        assertThat(result).succeedsWithin(1, SECONDS);
+    }
+
+    @Test
+    void provision_withFolder_success() {
+        var resourceDef = createResourceDefinitionBuilder().transferProcessId("tpId").folderName("test-folder").build();
+        String accountName = resourceDef.getAccountName();
+        String containerName = resourceDef.getContainerName();
+        when(blobStoreApiMock.exists(anyString(), anyString())).thenReturn(false);
+        when(blobStoreApiMock.createContainerSasToken(eq(accountName), eq(containerName), eq("w"), any())).thenReturn("some-sas");
+
+        var response = provisioner.provision(resourceDef, policy).join().getContent();
+
+        assertThat(response.getResource()).isInstanceOfSatisfying(ObjectContainerProvisionedResource.class, resource -> {
+            assertThat(resource.getTransferProcessId()).isEqualTo("tpId");
+            assertThat(resource.getDataAddress().getStringProperty(EDC_NAMESPACE + FOLDER_NAME)).isEqualTo("test-folder");
+        });
+        assertThat(response.getSecretToken()).isInstanceOfSatisfying(AzureSasToken.class, secretToken -> {
+            assertThat(secretToken.getSas()).isEqualTo("?some-sas");
+        });
+
+        verify(blobStoreApiMock).exists(anyString(), anyString());
+        verify(blobStoreApiMock).createContainer(accountName, containerName);
+    }
+
+    @Test
+    void provision_success() {
+        var resourceDef = createResourceDefinitionBuilder().transferProcessId("tpId").build();
+        String accountName = resourceDef.getAccountName();
+        String containerName = resourceDef.getContainerName();
+        when(blobStoreApiMock.exists(anyString(), anyString())).thenReturn(false);
+        when(blobStoreApiMock.createContainerSasToken(eq(accountName), eq(containerName), eq("w"), any())).thenReturn("some-sas");
+
+        var response = provisioner.provision(resourceDef, policy).join().getContent();
+
+        assertThat(response.getResource()).isInstanceOfSatisfying(ObjectContainerProvisionedResource.class, resource -> {
+            assertThat(resource.getTransferProcessId()).isEqualTo("tpId");
+            assertThat(resource.getDataAddress().getStringProperty(EDC_NAMESPACE + FOLDER_NAME)).isNull();
+        });
+        assertThat(response.getSecretToken()).isInstanceOfSatisfying(AzureSasToken.class, secretToken -> {
+            assertThat(secretToken.getSas()).isEqualTo("?some-sas");
+        });
+
+        verify(blobStoreApiMock).exists(anyString(), anyString());
+        verify(blobStoreApiMock).createContainer(accountName, containerName);
+    }
+
+    @Test
+    void provision_unique_name() {
+        var resourceDef = createResourceDefinitionBuilder().id("id").transferProcessId("tpId").build();
+        String accountName = resourceDef.getAccountName();
+        String containerName = resourceDef.getContainerName();
+        when(blobStoreApiMock.exists(accountName, containerName)).thenReturn(true);
+        when(blobStoreApiMock.createContainerSasToken(eq(accountName), eq(containerName), eq("w"), any())).thenReturn("some-sas");
+
+        var response = provisioner.provision(resourceDef, policy).join().getContent();
+
+        var resourceDef2 = createResourceDefinitionBuilder().id("id2").transferProcessId("tpId2").build();
+        var response2 = provisioner.provision(resourceDef2, policy).join().getContent();
+        var resource1 = (ObjectContainerProvisionedResource) response.getResource();
+        var resource2 = (ObjectContainerProvisionedResource) response2.getResource();
+        assertThat(resource2.getResourceName()).isNotEqualTo(resource1.getResourceName());
+    }
+
+    @Test
+    void provision_container_already_exists() {
+        var resourceDef = createResourceDefinitionBuilder().transferProcessId("tpId").build();
+        String accountName = resourceDef.getAccountName();
+        String containerName = resourceDef.getContainerName();
+        when(blobStoreApiMock.exists(accountName, containerName)).thenReturn(true);
+        when(blobStoreApiMock.createContainerSasToken(eq(accountName), eq(containerName), eq("w"), any())).thenReturn("some-sas");
+
+        var response = provisioner.provision(resourceDef, policy).join().getContent();
+
+        assertThat(response.getResource()).isInstanceOfSatisfying(ObjectContainerProvisionedResource.class, resource -> {
+            assertThat(resource.getTransferProcessId()).isEqualTo("tpId");
+        });
+        assertThat(response.getSecretToken()).isInstanceOfSatisfying(AzureSasToken.class, secretToken -> {
+            assertThat(secretToken.getSas()).isEqualTo("?some-sas");
+        });
+        verify(blobStoreApiMock).exists(anyString(), anyString());
+        verify(blobStoreApiMock).createContainerSasToken(eq(accountName), eq(containerName), eq("w"), any());
+    }
+
+    @Test
+    void provision_no_key_found_in_vault() {
+        var resourceDefinition = createResourceDefinitionBuilder().build();
+        when(blobStoreApiMock.exists(any(), anyString()))
+                .thenThrow(new IllegalArgumentException("No Object Storage credential found in vault"));
+
+        assertThatThrownBy(() -> provisioner.provision(resourceDefinition, policy).join()).hasCauseInstanceOf(IllegalArgumentException.class);
+        verify(blobStoreApiMock).exists(any(), any());
+    }
+
+    @Test
+    void provision_key_not_authorized() {
+        var resourceDef = createResourceDefinitionBuilder().build();
+        when(blobStoreApiMock.exists(anyString(), anyString())).thenReturn(false);
+        doThrow(new BlobStorageException("not authorized", null, null))
+                .when(blobStoreApiMock).createContainer(resourceDef.getAccountName(), resourceDef.getContainerName());
+
+        assertThatThrownBy(() -> provisioner.provision(resourceDef, policy).join()).hasCauseInstanceOf(BlobStorageException.class);
+        verify(blobStoreApiMock).exists(anyString(), anyString());
+    }
+
+    private ObjectStorageResourceDefinition.Builder createResourceDefinitionBuilder() {
+        return ObjectStorageResourceDefinition.Builder
+                .newInstance()
+                .accountName("test-account-name")
+                .containerName("test-container-name")
+                .transferProcessId("test-process-id")
+                .id("test-id");
+    }
+
+    private ObjectContainerProvisionedResource createProvisionedResource() {
+        return ObjectContainerProvisionedResource.Builder.newInstance()
+                .id("1")
+                .transferProcessId("2")
+                .resourceDefinitionId("3")
+                .resourceName("resource")
+                .build();
+    }
+
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageResourceDefinitionTest.java b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageResourceDefinitionTest.java
new file mode 100644
index 000000000..433f93d4f
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageResourceDefinitionTest.java
@@ -0,0 +1,47 @@
+/********************************************************************************
+ * Copyright (c) 2020,2021 Microsoft Corporation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.edc.connector.provision.azure.blob;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ObjectStorageResourceDefinitionTest {
+
+    @ParameterizedTest
+    @ValueSource(strings = { "test-folder" })
+    @NullAndEmptySource
+    void toBuilder_verifyEqualResourceDefinition(String folder) {
+        var definition = ObjectStorageResourceDefinition.Builder.newInstance()
+                .id("id")
+                .transferProcessId("tp-id")
+                .accountName("account")
+                .containerName("container")
+                .folderName(folder)
+                .build();
+        var builder = definition.toBuilder();
+        var rebuiltDefinition = builder.build();
+
+        assertThat(rebuiltDefinition).usingRecursiveComparison().isEqualTo(definition);
+    }
+
+}
diff --git a/edc-extensions/backport/azblob-provisioner/src/test/resources/hello.txt b/edc-extensions/backport/azblob-provisioner/src/test/resources/hello.txt
new file mode 100644
index 000000000..bc7774a7b
--- /dev/null
+++ b/edc-extensions/backport/azblob-provisioner/src/test/resources/hello.txt
@@ -0,0 +1 @@
+hello world!
\ No newline at end of file
diff --git a/edc-tests/edc-controlplane/catalog-tests/src/test/java/org/eclipse/tractusx/edc/tests/catalog/FederatedCatalogTest.java b/edc-tests/edc-controlplane/catalog-tests/src/test/java/org/eclipse/tractusx/edc/tests/catalog/FederatedCatalogTest.java
index eaedc2e81..4f17996c0 100644
--- a/edc-tests/edc-controlplane/catalog-tests/src/test/java/org/eclipse/tractusx/edc/tests/catalog/FederatedCatalogTest.java
+++ b/edc-tests/edc-controlplane/catalog-tests/src/test/java/org/eclipse/tractusx/edc/tests/catalog/FederatedCatalogTest.java
@@ -59,7 +59,7 @@ public class FederatedCatalogTest {
             .name(PROVIDER_NAME)
             .id(PROVIDER_BPN)
             .build();
-    
+
     abstract static class Tests {
 
         @Test
@@ -77,9 +77,9 @@ void requestCatalog_fulfillsPolicy_shouldReturnOffer() {
                     .atMost(ASYNC_TIMEOUT)
                     .untilAsserted(() -> {
                         CONSUMER.getFederatedCatalog()
-                                .log().ifValidationFails()
                                 .statusCode(200)
                                 .contentType(JSON)
+                                .log().ifValidationFails()
                                 .body("size()", is(1))
                                 .body("[0].'dcat:dataset'.'@id'", equalTo("test-asset"));
                     });
diff --git a/edc-tests/edc-controlplane/fixtures/src/testFixtures/java/org/eclipse/tractusx/edc/tests/runtimes/ParticipantRuntime.java b/edc-tests/edc-controlplane/fixtures/src/testFixtures/java/org/eclipse/tractusx/edc/tests/runtimes/ParticipantRuntime.java
index 1aa1667d0..49d0f5fcc 100644
--- a/edc-tests/edc-controlplane/fixtures/src/testFixtures/java/org/eclipse/tractusx/edc/tests/runtimes/ParticipantRuntime.java
+++ b/edc-tests/edc-controlplane/fixtures/src/testFixtures/java/org/eclipse/tractusx/edc/tests/runtimes/ParticipantRuntime.java
@@ -70,8 +70,10 @@ public ParticipantRuntime(String moduleName, String runtimeName, String bpn, Map
             runtimeKeyPair = new ECKeyGenerator(Curve.P_256).keyID(kid).generate();
             KeyPool.register(kid, runtimeKeyPair.toKeyPair());
             var privateKey = runtimeKeyPair.toPrivateKey();
-            
-            registerServiceMock(SecureTokenService.class, new EmbeddedSecureTokenService(new JwtGenerationService(new DefaultJwsSignerProvider((k) -> Result.success(privateKey))), () -> privateAlias, () -> kid, Clock.systemUTC(), Duration.ofMinutes(10).toMillis()));
+
+            registerServiceMock(SecureTokenService.class, new EmbeddedSecureTokenService(new JwtGenerationService(new DefaultJwsSignerProvider((k) -> Result.success(privateKey))),
+                    () -> privateAlias,
+                    () -> kid, Clock.systemUTC(), Duration.ofMinutes(10).toMillis()));
             registerServiceMock(DidPublicKeyResolver.class, keyId -> Result.success(KeyPool.forId(keyId).getPublic()));
         } catch (JOSEException e) {
             throw new RuntimeException(e);
diff --git a/edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/AzureToAzureTest.java b/edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/AzureToAzureTest.java
index 242af0ecb..430debaa6 100644
--- a/edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/AzureToAzureTest.java
+++ b/edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/AzureToAzureTest.java
@@ -87,7 +87,7 @@ public class AzureToAzureTest {
             "AzureBlob-Dataplane",
             RuntimeConfig.Azure.blobstoreDataplaneConfig("/control", PROVIDER_CONTROL_PORT, AZURITE_HOST_PORT),
             ":edc-tests:runtime:dataplane-cloud"
-    )).registerServiceMock(Monitor.class, spy(new ConsoleMonitor("AwsS3-Dataplane", ConsoleMonitor.Level.DEBUG)));
+    )).registerServiceMock(Monitor.class, spy(new ConsoleMonitor("AzureBlob-Dataplane", ConsoleMonitor.Level.DEBUG)));
     /**
      * Currently we have to use one container to host both consumer and provider accounts, because we cannot handle
      * two different endpoint templates for provider and consumer. Endpoint templates are configured globally.
@@ -232,6 +232,61 @@ void transferFile_largeFile(long sizeBytes, Vault vault) throws IOException {
 
     }
 
+    @Test
+    void transferFolder_targetFolderNotExists_shouldCreate(Vault vault) {
+
+        vault.storeSecret(AZBLOB_PROVIDER_KEY_ALIAS, AZBLOB_PROVIDER_ACCOUNT_KEY);
+        var sas = consumerBlobHelper.generateAccountSas(AZBLOB_CONSUMER_CONTAINER_NAME);
+        vault.storeSecret(AZBLOB_CONSUMER_KEY_ALIAS, """
+                {"sas": "%s","edctype":"dataspaceconnector:azuretoken"}
+                """.formatted(sas));
+
+        // create container in consumer's blob store
+        consumerBlobHelper.createContainer(AZBLOB_CONSUMER_CONTAINER_NAME);
+
+        var sourceContainer = providerBlobHelper.createContainer(AZBLOB_PROVIDER_CONTAINER_NAME);
+        var fileData = BinaryData.fromString(TestUtils.getResourceFileContentAsString(TESTFILE_NAME));
+
+        providerBlobHelper.uploadBlob(sourceContainer, fileData, "folder/blob.bin");
+        providerBlobHelper.uploadBlob(sourceContainer, fileData, "folder/blob2.bin");
+        providerBlobHelper.uploadBlob(sourceContainer, fileData, "folder/blob3.bin");
+
+        var request = createFlowRequestBuilder(TESTFILE_NAME)
+                .sourceDataAddress(DataAddress.Builder.newInstance()
+                        .type("AzureStorage")
+                        .property("container", AZBLOB_PROVIDER_CONTAINER_NAME)
+                        .property("account", AZBLOB_PROVIDER_ACCOUNT_NAME)
+                        .property("keyName", AZBLOB_PROVIDER_KEY_ALIAS)
+                        .property("blobPrefix", "folder/")
+                        .build())
+                .destinationDataAddress(DataAddress.Builder.newInstance()
+                        .type("AzureStorage")
+                        .property("container", AZBLOB_CONSUMER_CONTAINER_NAME)
+                        .property("account", AZBLOB_CONSUMER_ACCOUNT_NAME)
+                        .property("keyName", AZBLOB_CONSUMER_KEY_ALIAS)
+                        .property("folderName", "destfolder")
+                        .build())
+                .build();
+
+        var url = "http://localhost:%s/control/transfer".formatted(PROVIDER_CONTROL_PORT);
+
+        given().when()
+                .baseUri(url)
+                .contentType(ContentType.JSON)
+                .body(request)
+                .post()
+                .then()
+                .log().ifError()
+                .statusCode(200);
+
+        await().pollInterval(Duration.ofSeconds(2))
+                .atMost(Duration.ofSeconds(60))
+                .untilAsserted(() -> assertThat(consumerBlobHelper.listBlobs(AZBLOB_CONSUMER_CONTAINER_NAME))
+                        .isNotEmpty()
+                        .contains("destfolder/folder/blob.bin", "destfolder/folder/blob2.bin", "destfolder/folder/blob3.bin"));
+    }
+
+
     @Test
     void transferFile_targetContainerNotExist_shouldFail(Vault vault) {
         var sourceContainer = providerBlobHelper.createContainer(AZBLOB_PROVIDER_CONTAINER_NAME);
@@ -264,12 +319,7 @@ void transferFile_targetContainerNotExist_shouldFail(Vault vault) {
     }
 
     private DataFlowStartMessage createFlowRequest(String blobName) {
-        return DataFlowStartMessage.Builder.newInstance()
-                .id("test-request")
-                .sourceDataAddress(blobSourceAddress(blobName))
-                .destinationDataAddress(blobDestinationAddress(blobName))
-                .processId("test-process-id")
-                .flowType(FlowType.PUSH)
+        return createFlowRequestBuilder(blobName)
                 .build();
     }
 
@@ -289,4 +339,13 @@ private DataFlowStartMessage createMultipleFileFlowRequest(String blobPrefix) {
                 .flowType(FlowType.PUSH)
                 .build();
     }
+
+    private DataFlowStartMessage.Builder createFlowRequestBuilder(String blobName) {
+        return DataFlowStartMessage.Builder.newInstance()
+                .id("test-request")
+                .sourceDataAddress(blobSourceAddress(blobName))
+                .destinationDataAddress(blobDestinationAddress(blobName))
+                .processId("test-process-id")
+                .flowType(FlowType.PUSH);
+    }
 }
diff --git a/edc-tests/edc-end2end/end2end-transfer-cloud/build.gradle.kts b/edc-tests/edc-end2end/end2end-transfer-cloud/build.gradle.kts
new file mode 100644
index 000000000..0f5f65a03
--- /dev/null
+++ b/edc-tests/edc-end2end/end2end-transfer-cloud/build.gradle.kts
@@ -0,0 +1,42 @@
+/********************************************************************************
+ * Copyright (c) 2023 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+plugins {
+    `java-library`
+    `java-test-fixtures`
+}
+
+dependencies {
+    testImplementation(testFixtures(project(":edc-tests:edc-controlplane:fixtures")))
+
+    testImplementation(libs.edc.junit)
+    testImplementation(libs.restAssured)
+
+    // test runtime config
+    testImplementation(libs.testcontainers.junit)
+    testImplementation(libs.edc.aws.s3.core)
+    testImplementation(libs.aws.s3)
+    testImplementation(libs.aws.s3transfer)
+    testImplementation(libs.azure.storage.blob)
+}
+
+// do not publish
+edcBuild {
+    publish.set(false)
+}
diff --git a/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AzureBlobHelper.java b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AzureBlobHelper.java
new file mode 100644
index 000000000..81da3a164
--- /dev/null
+++ b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AzureBlobHelper.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.eclipse.tractusx.edc.tests.transfer;
+
+import com.azure.core.util.BinaryData;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.blob.models.BlobItem;
+import com.azure.storage.common.StorageSharedKeyCredential;
+
+import java.util.List;
+
+
+/**
+ * Helper class that internally uses Azure SDK classes to create containers, upload blobs, generate SAS tokens, etc.
+ */
+class AzureBlobHelper {
+    private final String accountName;
+    private final String key;
+    private final String host;
+    private final int port;
+    private BlobServiceClient blobServiceClient;
+
+    AzureBlobHelper(String accountName, String key, String host, int port) {
+        this.accountName = accountName;
+        this.key = key;
+        this.host = host;
+        this.port = port;
+    }
+
+    public BlobContainerClient createContainer(String containerName) {
+        return blobClient().createBlobContainer(containerName);
+    }
+
+    public void uploadBlob(BlobContainerClient client, BinaryData data, String targetBlobName) {
+        client.getBlobClient(targetBlobName).upload(data, true);
+    }
+
+    public List<String> listBlobs(String container) {
+        if (blobClient().listBlobContainers().stream().noneMatch(bci -> bci.getName().equalsIgnoreCase(container))) {
+            return List.of();
+        }
+        return blobClient()
+                .getBlobContainerClient(container)
+                .listBlobs()
+                .stream().map(BlobItem::getName)
+                .toList();
+    }
+
+    private BlobServiceClient blobClient() {
+        if (blobServiceClient == null) {
+            var endpoint = "http://%s:%s/%s".formatted(host, port, accountName);
+            blobServiceClient = new BlobServiceClientBuilder()
+                    .credential(new StorageSharedKeyCredential(accountName, key))
+                    .endpoint(endpoint)
+                    .buildClient();
+        }
+        return blobServiceClient;
+    }
+}
diff --git a/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AzureToAzureEndToEndTest.java b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AzureToAzureEndToEndTest.java
new file mode 100644
index 000000000..37a48f18a
--- /dev/null
+++ b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/AzureToAzureEndToEndTest.java
@@ -0,0 +1,275 @@
+/*
+ * Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.eclipse.tractusx.edc.tests.transfer;
+
+import com.azure.core.util.BinaryData;
+import org.apache.commons.codec.binary.Base64;
+import org.eclipse.edc.junit.annotations.EndToEndTest;
+import org.eclipse.edc.junit.extensions.RuntimeExtension;
+import org.eclipse.edc.junit.testfixtures.TestUtils;
+import org.eclipse.edc.spi.security.Vault;
+import org.eclipse.tractusx.edc.tests.participant.TractusxParticipantBase;
+import org.eclipse.tractusx.edc.tests.participant.TransferParticipant;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.testcontainers.containers.FixedHostPortGenericContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static jakarta.json.Json.createObjectBuilder;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.eclipse.edc.connector.controlplane.transfer.spi.types.TransferProcessStates.COMPLETED;
+import static org.eclipse.edc.jsonld.spi.JsonLdKeywords.TYPE;
+import static org.eclipse.edc.spi.constants.CoreConstants.EDC_NAMESPACE;
+import static org.eclipse.edc.util.io.Ports.getFreePort;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.CONSUMER_BPN;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.CONSUMER_NAME;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.PROVIDER_BPN;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.PROVIDER_NAME;
+import static org.eclipse.tractusx.edc.tests.helpers.PolicyHelperFunctions.bnpPolicy;
+import static org.eclipse.tractusx.edc.tests.participant.TractusxParticipantBase.ASYNC_TIMEOUT;
+import static org.eclipse.tractusx.edc.tests.runtimes.Runtimes.memoryRuntime;
+
+/**
+ * This test runs through a contract negotiation and transfer process phase, then transfers files from an Az Blob container
+ * to another Az blob container
+ */
+@Testcontainers
+@EndToEndTest
+public class AzureToAzureEndToEndTest {
+
+    public static final String AZURITE_DOCKER_IMAGE = "mcr.microsoft.com/azure-storage/azurite";
+    public static final String CONSUMER_CONTAINER_NAME = "consumer-container";
+    public static final String PROVIDER_CONTAINER_NAME = "provider-container";
+    public static final String PROVIDER_KEY_ALIAS = "provider-key-alias";
+    protected static final TransferParticipant CONSUMER = TransferParticipant.Builder.newInstance()
+            .name(CONSUMER_NAME)
+            .id(CONSUMER_BPN)
+            .build();
+    protected static final TransferParticipant PROVIDER = TransferParticipant.Builder.newInstance()
+            .name(PROVIDER_NAME)
+            .id(PROVIDER_BPN)
+            .build();
+    private static final int AZURITE_HOST_PORT = getFreePort();
+    @RegisterExtension
+    protected static final RuntimeExtension PROVIDER_RUNTIME = memoryRuntime(PROVIDER.getName(), PROVIDER.getBpn(), with(PROVIDER.getConfiguration(), AZURITE_HOST_PORT));
+    @RegisterExtension
+    protected static final RuntimeExtension CONSUMER_RUNTIME = memoryRuntime(CONSUMER.getName(), CONSUMER.getBpn(), with(CONSUMER.getConfiguration(), AZURITE_HOST_PORT));
+    private static final String AZBLOB_PROVIDER_ACCOUNT_NAME = "provider";
+    private static final String AZBLOB_CONSUMER_ACCOUNT_NAME = "consumer";
+    private static final String AZBLOB_PROVIDER_ACCOUNT_KEY = Base64.encodeBase64String("provider-key".getBytes());
+    private static final String AZBLOB_CONSUMER_ACCOUNT_KEY = Base64.encodeBase64String("provider-key".getBytes());
+    private static final int AZURITE_CONTAINER_PORT = 10000;
+    private static final String TESTFILE_NAME = "hello.txt";
+    @Container
+    private final FixedHostPortGenericContainer<?> azuriteContainer = new FixedHostPortGenericContainer<>(AZURITE_DOCKER_IMAGE)
+            .withFixedExposedPort(AZURITE_HOST_PORT, AZURITE_CONTAINER_PORT)
+            .withEnv("AZURITE_ACCOUNTS", AZBLOB_PROVIDER_ACCOUNT_NAME + ":" + AZBLOB_PROVIDER_ACCOUNT_KEY + ";" + AZBLOB_CONSUMER_ACCOUNT_NAME + ":" + AZBLOB_CONSUMER_ACCOUNT_KEY);
+    private AzureBlobHelper providerBlobHelper;
+    private AzureBlobHelper consumerBlobHelper;
+
+    private static Map<String, String> with(Map<String, String> configuration, int port) {
+        configuration.putAll(new HashMap<>() {
+            {
+                put("edc.blobstore.endpoint.template", "http://127.0.0.1:" + port + "/%s"); // set the Azure Blob endpoint template
+            }
+        });
+        return configuration;
+    }
+
+    public TractusxParticipantBase provider() {
+        return PROVIDER;
+    }
+
+    public TractusxParticipantBase consumer() {
+        return CONSUMER;
+    }
+
+    @BeforeEach
+    void setup() {
+        PROVIDER_RUNTIME.getService(Vault.class)
+                .storeSecret(PROVIDER_KEY_ALIAS, AZBLOB_PROVIDER_ACCOUNT_KEY);
+
+        CONSUMER_RUNTIME.getService(Vault.class)
+                .storeSecret("%s-key1".formatted(AZBLOB_CONSUMER_ACCOUNT_NAME), AZBLOB_CONSUMER_ACCOUNT_KEY);
+
+        providerBlobHelper = new AzureBlobHelper(AZBLOB_PROVIDER_ACCOUNT_NAME, AZBLOB_PROVIDER_ACCOUNT_KEY, azuriteContainer.getHost(), azuriteContainer.getMappedPort(AZURITE_CONTAINER_PORT));
+        consumerBlobHelper = new AzureBlobHelper(AZBLOB_CONSUMER_ACCOUNT_NAME, AZBLOB_CONSUMER_ACCOUNT_KEY, azuriteContainer.getHost(), azuriteContainer.getMappedPort(AZURITE_CONTAINER_PORT));
+    }
+
+    @Test
+    void azureBlobPush_withDestFolder() {
+        var assetId = "felix-blob-test-asset";
+
+        Map<String, Object> dataAddress = Map.of(
+                "name", "transfer-test",
+                "@type", "DataAddress",
+                "type", "AzureStorage",
+                "container", PROVIDER_CONTAINER_NAME,
+                "account", AZBLOB_PROVIDER_ACCOUNT_NAME,
+                "blobPrefix", "folder/",
+                "keyName", PROVIDER_KEY_ALIAS
+        );
+
+        // upload file to provider's blob store
+        var sourceContainer = providerBlobHelper.createContainer(PROVIDER_CONTAINER_NAME);
+        var fileData = BinaryData.fromString(TestUtils.getResourceFileContentAsString(TESTFILE_NAME));
+        providerBlobHelper.uploadBlob(sourceContainer, fileData, "folder/" + TESTFILE_NAME);
+        consumerBlobHelper.createContainer(CONSUMER_CONTAINER_NAME);
+
+        // create objects in EDC
+        provider().createAsset(assetId, Map.of(), dataAddress);
+        var policyId = provider().createPolicyDefinition(bnpPolicy(consumer().getBpn()));
+        provider().createContractDefinition(assetId, "def-1", policyId, policyId);
+
+        var destfolder = "destfolder";
+        var destination = createObjectBuilder()
+                .add(TYPE, EDC_NAMESPACE + "DataAddress")
+                .add(EDC_NAMESPACE + "type", "AzureStorage")
+                .add(EDC_NAMESPACE + "properties", createObjectBuilder()
+                        .add(EDC_NAMESPACE + "type", "AzureStorage")
+                        .add(EDC_NAMESPACE + "account", AZBLOB_CONSUMER_ACCOUNT_NAME)
+                        .add(EDC_NAMESPACE + "container", CONSUMER_CONTAINER_NAME)
+                        .add(EDC_NAMESPACE + "folderName", destfolder)
+                        .build())
+                .build();
+
+        // perform contract negotiation and transfer process
+        var transferProcessId = consumer()
+                .requestAssetFrom(assetId, provider())
+                .withTransferType("AzureStorage-PUSH")
+                .withDestination(destination)
+                .execute();
+
+        await().atMost(ASYNC_TIMEOUT).untilAsserted(() -> {
+            var state = consumer().getTransferProcessState(transferProcessId);
+            assertThat(state).isEqualTo(COMPLETED.name());
+            assertThat(consumerBlobHelper.listBlobs(CONSUMER_CONTAINER_NAME))
+                    .contains("%s/folder/%s".formatted(destfolder, TESTFILE_NAME));
+        });
+    }
+
+    @Test
+    void azureBlobPush_withoutDestFolder() {
+        var assetId = "felix-blob-test-asset";
+
+        Map<String, Object> dataAddress = Map.of(
+                "name", "transfer-test",
+                "@type", "DataAddress",
+                "type", "AzureStorage",
+                "container", PROVIDER_CONTAINER_NAME,
+                "account", AZBLOB_PROVIDER_ACCOUNT_NAME,
+                "blobPrefix", "folder/",
+                "keyName", PROVIDER_KEY_ALIAS
+        );
+
+        // upload file to provider's blob store
+        var sourceContainer = providerBlobHelper.createContainer(PROVIDER_CONTAINER_NAME);
+        var fileData = BinaryData.fromString(TestUtils.getResourceFileContentAsString(TESTFILE_NAME));
+        providerBlobHelper.uploadBlob(sourceContainer, fileData, "folder/" + TESTFILE_NAME);
+        consumerBlobHelper.createContainer(CONSUMER_CONTAINER_NAME);
+
+        // create objects in EDC
+        provider().createAsset(assetId, Map.of(), dataAddress);
+        var policyId = provider().createPolicyDefinition(bnpPolicy(consumer().getBpn()));
+        provider().createContractDefinition(assetId, "def-1", policyId, policyId);
+
+        var destination = createObjectBuilder()
+                .add(TYPE, EDC_NAMESPACE + "DataAddress")
+                .add(EDC_NAMESPACE + "type", "AzureStorage")
+                .add(EDC_NAMESPACE + "properties", createObjectBuilder()
+                        .add(EDC_NAMESPACE + "type", "AzureStorage")
+                        .add(EDC_NAMESPACE + "account", AZBLOB_CONSUMER_ACCOUNT_NAME)
+                        .add(EDC_NAMESPACE + "container", CONSUMER_CONTAINER_NAME)
+                        // .add(EDC_NAMESPACE + "folderName", destfolder) <-- no dest folder
+                        .build())
+                .build();
+
+        // perform contract negotiation and transfer process
+        var transferProcessId = consumer()
+                .requestAssetFrom(assetId, provider())
+                .withTransferType("AzureStorage-PUSH")
+                .withDestination(destination)
+                .execute();
+
+        await().atMost(ASYNC_TIMEOUT).untilAsserted(() -> {
+            var state = consumer().getTransferProcessState(transferProcessId);
+            assertThat(state).isEqualTo(COMPLETED.name());
+            assertThat(consumerBlobHelper.listBlobs(CONSUMER_CONTAINER_NAME))
+                    .contains("folder/%s".formatted(TESTFILE_NAME));
+        });
+    }
+
+    @Test
+    void azureBlobPush_containerNotExist() {
+        var assetId = "blob-test-asset";
+
+        Map<String, Object> dataAddress = Map.of(
+                "name", "transfer-test",
+                "@type", "DataAddress",
+                "type", "AzureStorage",
+                "container", PROVIDER_CONTAINER_NAME,
+                "account", AZBLOB_PROVIDER_ACCOUNT_NAME,
+                "blobPrefix", "folder/",
+                "keyName", PROVIDER_KEY_ALIAS
+        );
+
+        // upload file to provider's blob store
+        var sourceContainer = providerBlobHelper.createContainer(PROVIDER_CONTAINER_NAME);
+        var fileData = BinaryData.fromString(TestUtils.getResourceFileContentAsString(TESTFILE_NAME));
+        providerBlobHelper.uploadBlob(sourceContainer, fileData, "folder/" + TESTFILE_NAME);
+        // consumerBlobHelper.createContainer(CONSUMER_CONTAINER_NAME); <-- container is not created
+
+        // create objects in EDC
+        provider().createAsset(assetId, Map.of(), dataAddress);
+        var policyId = provider().createPolicyDefinition(bnpPolicy(consumer().getBpn()));
+        provider().createContractDefinition(assetId, "def-1", policyId, policyId);
+
+        var destination = createObjectBuilder()
+                .add(TYPE, EDC_NAMESPACE + "DataAddress")
+                .add(EDC_NAMESPACE + "type", "AzureStorage")
+                .add(EDC_NAMESPACE + "properties", createObjectBuilder()
+                        .add(EDC_NAMESPACE + "type", "AzureStorage")
+                        .add(EDC_NAMESPACE + "account", AZBLOB_CONSUMER_ACCOUNT_NAME)
+                        .add(EDC_NAMESPACE + "container", CONSUMER_CONTAINER_NAME)
+                        .build())
+                .build();
+
+        // perform contract negotiation and transfer process
+        var transferProcessId = consumer()
+                .requestAssetFrom(assetId, provider())
+                .withTransferType("AzureStorage-PUSH")
+                .withDestination(destination)
+                .execute();
+
+        await().atMost(ASYNC_TIMEOUT).untilAsserted(() -> {
+            var state = consumer().getTransferProcessState(transferProcessId);
+            assertThat(state).isEqualTo(COMPLETED.name());
+            assertThat(consumerBlobHelper.listBlobs(CONSUMER_CONTAINER_NAME))
+                    .contains("folder/%s".formatted(TESTFILE_NAME));
+        });
+    }
+
+}
\ No newline at end of file
diff --git a/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/MinioContainer.java b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/MinioContainer.java
new file mode 100644
index 000000000..329a69c4e
--- /dev/null
+++ b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/MinioContainer.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.eclipse.tractusx.edc.tests.transfer;
+
+import org.testcontainers.containers.GenericContainer;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+
+import java.util.UUID;
+
+public class MinioContainer extends GenericContainer<MinioContainer> {
+
+    private final String accessKeyId = "test-access-key";
+    private final String secretAccessKey = UUID.randomUUID().toString();
+
+    public MinioContainer() {
+        super("bitnami/minio");
+        addEnv("MINIO_ROOT_USER", accessKeyId);
+        addEnv("MINIO_ROOT_PASSWORD", secretAccessKey);
+        addExposedPort(9000);
+    }
+
+    public AwsBasicCredentials getCredentials() {
+        return AwsBasicCredentials.create(accessKeyId, secretAccessKey);
+    }
+}
diff --git a/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/S3ToS3EndToEndTest.java b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/S3ToS3EndToEndTest.java
new file mode 100644
index 000000000..e195ea437
--- /dev/null
+++ b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/java/org/eclipse/tractusx/edc/tests/transfer/S3ToS3EndToEndTest.java
@@ -0,0 +1,184 @@
+/********************************************************************************
+ * Copyright (c) 2023 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * 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.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.tractusx.edc.tests.transfer;
+
+import jakarta.json.Json;
+import org.eclipse.edc.aws.s3.AwsClientProviderConfiguration;
+import org.eclipse.edc.aws.s3.AwsClientProviderImpl;
+import org.eclipse.edc.aws.s3.S3ClientRequest;
+import org.eclipse.edc.junit.annotations.EndToEndTest;
+import org.eclipse.edc.junit.extensions.RuntimeExtension;
+import org.eclipse.edc.junit.testfixtures.TestUtils;
+import org.eclipse.tractusx.edc.tests.participant.TractusxParticipantBase;
+import org.eclipse.tractusx.edc.tests.participant.TransferParticipant;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.eclipse.edc.connector.controlplane.transfer.spi.types.TransferProcessStates.COMPLETED;
+import static org.eclipse.edc.jsonld.spi.JsonLdKeywords.TYPE;
+import static org.eclipse.edc.spi.constants.CoreConstants.EDC_NAMESPACE;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.CONSUMER_BPN;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.CONSUMER_NAME;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.PROVIDER_BPN;
+import static org.eclipse.tractusx.edc.tests.TestRuntimeConfiguration.PROVIDER_NAME;
+import static org.eclipse.tractusx.edc.tests.helpers.PolicyHelperFunctions.bnpPolicy;
+import static org.eclipse.tractusx.edc.tests.participant.TractusxParticipantBase.ASYNC_TIMEOUT;
+import static org.eclipse.tractusx.edc.tests.runtimes.Runtimes.memoryRuntime;
+
+/**
+ * This test is intended to verify transfers within the same cloud provider, i.e. S3-to-S3.
+ * It spins up a fully-fledged dataplane and issues the DataFlowStartMessage via the data plane's Control API
+ */
+@Testcontainers
+@EndToEndTest
+public class S3ToS3EndToEndTest {
+    protected static final TransferParticipant CONSUMER = TransferParticipant.Builder.newInstance()
+            .name(CONSUMER_NAME)
+            .id(CONSUMER_BPN)
+            .build();
+    protected static final TransferParticipant PROVIDER = TransferParticipant.Builder.newInstance()
+            .name(PROVIDER_NAME)
+            .id(PROVIDER_BPN)
+            .build();
+    @RegisterExtension
+    protected static final RuntimeExtension PROVIDER_RUNTIME = memoryRuntime(PROVIDER.getName(), PROVIDER.getBpn(), PROVIDER.getConfiguration());
+    @RegisterExtension
+    protected static final RuntimeExtension CONSUMER_RUNTIME = memoryRuntime(CONSUMER.getName(), CONSUMER.getBpn(), CONSUMER.getConfiguration());
+    private static final String S3_REGION = Region.US_WEST_2.id();
+    private static final String S3_PROVIDER_BUCKET_NAME = "provider-bucket";
+    private static final String S3_CONSUMER_BUCKET_NAME = "consumer-bucket" + System.currentTimeMillis();
+    private static final String TESTFILE_NAME = "hello.txt";
+    @Container
+    private final MinioContainer providerContainer = new MinioContainer();
+    @Container
+    private final MinioContainer consumerContainer = new MinioContainer();
+
+    private S3Client providerClient;
+    private S3Client consumerClient;
+    private String providerEndpointOverride;
+    private String consumerEndpointOverride;
+
+    @BeforeEach
+    void setup() {
+        providerEndpointOverride = "http://localhost:%s/".formatted(providerContainer.getFirstMappedPort());
+        var providerConfig = AwsClientProviderConfiguration.Builder.newInstance()
+                .endpointOverride(URI.create(providerEndpointOverride))
+                .credentialsProvider(providerContainer::getCredentials)
+                .build();
+        providerClient = new AwsClientProviderImpl(providerConfig).s3Client(S3ClientRequest.from(S3_REGION, providerEndpointOverride));
+
+        consumerEndpointOverride = "http://localhost:%s".formatted(consumerContainer.getFirstMappedPort());
+        var consumerConfig = AwsClientProviderConfiguration.Builder.newInstance()
+                .endpointOverride(URI.create(consumerEndpointOverride))
+                .credentialsProvider(consumerContainer::getCredentials)
+                .build();
+        consumerClient = new AwsClientProviderImpl(consumerConfig).s3Client(S3ClientRequest.from(S3_REGION, consumerEndpointOverride));
+    }
+
+    @Test
+    void transferFile_success() {
+        var assetId = "s3-test-asset";
+
+        // create bucket in provider
+        var b1 = providerClient.createBucket(CreateBucketRequest.builder().bucket(S3_PROVIDER_BUCKET_NAME).build());
+        assertThat(b1.sdkHttpResponse().isSuccessful()).isTrue();
+        // upload test file in provider
+        var putResponse = providerClient.putObject(PutObjectRequest.builder().bucket(S3_PROVIDER_BUCKET_NAME).key(TESTFILE_NAME).build(), TestUtils.getFileFromResourceName(TESTFILE_NAME).toPath());
+        assertThat(putResponse.sdkHttpResponse().isSuccessful()).isTrue();
+
+        // create bucket in consumer
+        var b2 = consumerClient.createBucket(CreateBucketRequest.builder().bucket(S3_CONSUMER_BUCKET_NAME).build());
+        assertThat(b2.sdkHttpResponse().isSuccessful()).isTrue();
+
+        Map<String, Object> dataAddress = Map.of(
+                "name", "transfer-test",
+                "@type", "DataAddress",
+                "type", "AmazonS3",
+                "objectName", TESTFILE_NAME,
+                "region", S3_REGION,
+                "bucketName", S3_PROVIDER_BUCKET_NAME,
+                "accessKeyId", providerContainer.getCredentials().accessKeyId(),
+                "secretAccessKey", providerContainer.getCredentials().secretAccessKey(),
+                "endpointOverride", providerEndpointOverride
+        );
+
+        // create objects in EDC
+        provider().createAsset(assetId, Map.of(), dataAddress);
+        var policyId = provider().createPolicyDefinition(bnpPolicy(consumer().getBpn()));
+        provider().createContractDefinition(assetId, "def-1", policyId, policyId);
+
+
+        var destination = Json.createObjectBuilder()
+                .add(TYPE, EDC_NAMESPACE + "DataAddress")
+                .add(EDC_NAMESPACE + "type", "AmazonS3")
+                .add(EDC_NAMESPACE + "properties", Json.createObjectBuilder()
+                        .add(EDC_NAMESPACE + "type", "AmazonS3")
+                        .add(EDC_NAMESPACE + "objectName", TESTFILE_NAME)
+                        .add(EDC_NAMESPACE + "region", S3_REGION)
+                        .add(EDC_NAMESPACE + "bucketName", S3_CONSUMER_BUCKET_NAME)
+                        .add(EDC_NAMESPACE + "endpointOverride", consumerEndpointOverride)
+                        .add(EDC_NAMESPACE + "accessKeyId", consumerContainer.getCredentials().accessKeyId())
+                        .add(EDC_NAMESPACE + "secretAccessKey", consumerContainer.getCredentials().secretAccessKey())
+                        .build()
+                ).build();
+
+        var transferProcessId = consumer()
+                .requestAssetFrom(assetId, provider())
+                .withTransferType("AmazonS3-PUSH")
+                .withDestination(destination)
+                .execute();
+
+        await().atMost(ASYNC_TIMEOUT).untilAsserted(() -> {
+            var state = consumer().getTransferProcessState(transferProcessId);
+            assertThat(state).isEqualTo(COMPLETED.name());
+            var rq = ListObjectsRequest.builder().bucket(S3_CONSUMER_BUCKET_NAME).build();
+            assertThat(consumerClient.listObjects(rq).contents()).isNotEmpty();
+        });
+    }
+
+    public TractusxParticipantBase provider() {
+        return PROVIDER;
+    }
+
+    public TractusxParticipantBase consumer() {
+        return CONSUMER;
+    }
+
+    private List<String> listObjects(S3Client consumerClient, String bucketName) {
+        var response = consumerClient.listObjects(ListObjectsRequest.builder().bucket(bucketName).build());
+        return response.contents().stream().map(S3Object::key).toList();
+    }
+
+}
diff --git a/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/resources/hello.txt b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/resources/hello.txt
new file mode 100644
index 000000000..bc7774a7b
--- /dev/null
+++ b/edc-tests/edc-end2end/end2end-transfer-cloud/src/test/resources/hello.txt
@@ -0,0 +1 @@
+hello world!
\ No newline at end of file
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 22f02b2a8..4ddd7865e 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -126,6 +126,11 @@ edc-azure-vault = { module = "org.eclipse.edc.azure:vault-azure", version.ref =
 azure-storage-blob = { module = "com.azure:azure-storage-blob", version.ref = "azure-storage-blob" }
 edc-azure-identity = { module = "com.azure:azure-identity", version.ref = "azure-identity" }
 edc-dpf-azblob = { module = "org.eclipse.edc.azure:data-plane-azure-storage", version.ref = "edc" }
+edc-azure-blob-core = { module = "org.eclipse.edc.azure:azure-blob-core", version.ref = "edc" }
+edc-azure-test = { module = "org.eclipse.edc.azure:azure-test", version.ref = "edc" }
+
+# commented, because this module has been backported temporarily
+#edc-azure-blob-provision = { module = "org.eclipse.edc.azure:provision-blob", version.ref = "edc" }
 
 # EDC aws s3 stuff
 edc-aws-s3-core = { module = "org.eclipse.edc.aws:aws-s3-core", version.ref = "edc" }
diff --git a/settings.gradle.kts b/settings.gradle.kts
index e1d6ecc1b..fe245c0d1 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -51,6 +51,8 @@ include(":edc-extensions:cx-policy")
 include(":edc-extensions:dcp:tx-dcp")
 include(":edc-extensions:dcp:tx-dcp-sts-dim")
 include(":edc-extensions:data-flow-properties-provider")
+include(":edc-extensions:backport:azblob-provisioner")
+
 
 // extensions - data plane
 include(":edc-extensions:dataplane:dataplane-proxy:edc-dataplane-proxy-consumer-api")
@@ -77,6 +79,7 @@ include(":edc-tests:runtime:iatp:runtime-memory-sts")
 include(":edc-tests:runtime:iatp:iatp-extensions")
 include(":edc-tests:edc-dataplane:edc-dataplane-tokenrefresh-tests")
 include(":edc-tests:edc-dataplane:cloud-transfer-tests")
+include(":edc-tests:edc-end2end:end2end-transfer-cloud")
 
 // modules for controlplane artifacts
 include(":edc-controlplane")