feat: Azure DataLake compatibility + metadata addition (#77)

* feat: Azure DataLake compatibility + metadata addition

* style: Azure DataLake compatibility + metadata addition

Incorporation of review comments (with the exception of one open query/counter-suggestion regarding the creation of a builder):

Essentially
- Removal of "final" for local variables
- Removal of explicit paramaters for mock() where possible
- "var" where possible
The existing code, insofar as it is relevant for the Azure Blob extension, I have also adjusted.

* style: Azure DataLake compatibility + metadata addition

* style: Azure DataLake compatibility + metadata addition

* chore: update dependencies

Author: ank19

DEPENDENCIES

@@ -101,7 +101,7 @@ maven/mavencentral/com.nimbusds/content-type/2.2, Apache-2.0, approved, clearlyd
 maven/mavencentral/com.nimbusds/lang-tag/1.7, Apache-2.0, approved, clearlydefined
 maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.25, Apache-2.0, approved, clearlydefined
 maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.30.2, Apache-2.0, approved, clearlydefined
-maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.37, , restricted, clearlydefined
+maven/mavencentral/com.nimbusds/nimbus-jose-jwt/9.37, Apache-2.0, approved, #11086
 maven/mavencentral/com.nimbusds/oauth2-oidc-sdk/10.7.1, Apache-2.0, approved, clearlydefined
 maven/mavencentral/com.puppycrawl.tools/checkstyle/10.0, LGPL-2.1-or-later, approved, #7936
 maven/mavencentral/com.squareup.okhttp3/logging-interceptor/3.12.12, Apache-2.0, approved, clearlydefined
@@ -148,9 +148,9 @@ maven/mavencentral/io.netty/netty-transport-native-epoll/4.1.94.Final, Apache-2.
 maven/mavencentral/io.netty/netty-transport-native-kqueue/4.1.94.Final, Apache-2.0 AND BSD-3-Clause AND MIT, approved, CQ20926
 maven/mavencentral/io.netty/netty-transport-native-unix-common/4.1.94.Final, Apache-2.0 AND BSD-3-Clause AND MIT, approved, CQ20926
 maven/mavencentral/io.netty/netty-transport/4.1.94.Final, Apache-2.0 AND BSD-3-Clause AND MIT, approved, CQ20926
-maven/mavencentral/io.opentelemetry.instrumentation/opentelemetry-instrumentation-annotations/1.31.0, , restricted, clearlydefined
-maven/mavencentral/io.opentelemetry/opentelemetry-api/1.31.0, , restricted, clearlydefined
-maven/mavencentral/io.opentelemetry/opentelemetry-context/1.31.0, , restricted, clearlydefined
+maven/mavencentral/io.opentelemetry.instrumentation/opentelemetry-instrumentation-annotations/1.31.0, Apache-2.0, approved, #11085
+maven/mavencentral/io.opentelemetry/opentelemetry-api/1.31.0, Apache-2.0, approved, #11087
+maven/mavencentral/io.opentelemetry/opentelemetry-context/1.31.0, Apache-2.0, approved, #11088
 maven/mavencentral/io.projectreactor.netty/reactor-netty-core/1.0.34, Apache-2.0, approved, #9687
 maven/mavencentral/io.projectreactor.netty/reactor-netty-http/1.0.34, Apache-2.0, approved, clearlydefined
 maven/mavencentral/io.projectreactor/reactor-core/3.4.31, Apache-2.0, approved, #7517

extensions/common/azure/azure-blob-core/build.gradle.kts

@@ -20,6 +20,7 @@ plugins {
 dependencies {
     api(libs.edc.controlplane.spi)
     implementation(libs.azure.storageblob)
+    implementation(libs.azure.identity)
     implementation(libs.edc.util)
 
     testFixturesApi(libs.edc.core.dataPlane.util)

extensions/common/azure/azure-blob-core/src/main/java/org/eclipse/edc/azure/blob/AzureBlobStoreSchema.java

@@ -26,5 +26,7 @@ private AzureBlobStoreSchema() {
     public static final String TYPE = "AzureStorage";
     public static final String CONTAINER_NAME = "container";
     public static final String ACCOUNT_NAME = "account";
-    public static final String BLOB_NAME = "blobname";
+    public static final String BLOB_NAME = "blobName";
+    public static final String FOLDER_NAME = "folderName";
+    public static final String CORRELATION_ID = "correlationId";
 }

extensions/common/azure/azure-blob-core/src/main/java/org/eclipse/edc/azure/blob/adapter/BlobAdapter.java

@@ -18,6 +18,7 @@
 
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.Map;
 
 /**
  * Adapter over {@link BlockBlobClient} in order to support mocking.
@@ -30,4 +31,6 @@ public interface BlobAdapter {
     String getBlobName();
 
     long getBlobSize();
+
+    void setMetadata(Map<String, String> metadata);
 }

extensions/common/azure/azure-blob-core/src/main/java/org/eclipse/edc/azure/blob/adapter/DefaultBlobAdapter.java

@@ -18,6 +18,7 @@
 
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.Map;
 
 /**
  * Implementation of {@link BlobAdapter} using a {@link BlockBlobClient}.
@@ -48,4 +49,9 @@ public String getBlobName() {
     public long getBlobSize() {
         return client.getProperties().getBlobSize();
     }
+
+    @Override
+    public void setMetadata(Map<String, String> metadata) {
+        client.setMetadata(metadata);
+    }
 }

extensions/common/azure/azure-blob-core/src/main/java/org/eclipse/edc/azure/blob/api/BlobStoreApi.java

@@ -41,7 +41,42 @@ public interface BlobStoreApi {
 
     byte[] getBlob(String account, String container, String blobName);
 
+    /**
+     * Get a blob adapter containing convenience methods for working on blob objects on a storage account.
+     * This method accepts storage account key credential, and it is used in a context, where unlimited access to
+     * a storage account is required.
+     *
+     * @param accountName The name of the storage account
+     * @param containerName The name of the container within the storage account
+     * @param blobName The name of the blob within the container of the storage account
+     * @param sharedKey The storage account key credential
+     * @return The blob adapter corresponding to the blob specified by the input parameters
+     */
     BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName, String sharedKey);
 
+    /**
+     * Get a blob adapter containing convenience methods for working on blob objects on a storage account.
+     * This method accepts a SAS (Shared Access Signature) token as a credential, and it's typically used for accessing
+     * a storage account with limited sets of privileges. Pls. refer to the Azure SAS documentation for further details.
+     *
+     * @param accountName The name of the storage account
+     * @param containerName The name of the container within the storage account
+     * @param blobName The name of the blob within the container of the storage account
+     * @param credential A valid SAS token
+     * @return The blob adapter corresponding to the blob specified by the input parameters
+     */
     BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName, AzureSasCredential credential);
+
+    /**
+     * Get a blob adapter containing convenience methods for working on blob objects on a storage account.
+     * This method doesn't require any credentials; it uses {@link com.azure.identity.DefaultAzureCredentialBuilder},
+     * which contains an authentication flow consisting of several authentication mechanisms to be tried in a specific
+     * order. Pls. refer to the official documentation for further details, i.e. the list of mechanisms tried.
+     *
+     * @param accountName The name of the storage account
+     * @param containerName The name of the container within the storage account
+     * @param blobName The name of the blob within the container of the storage account
+     * @return The blob adapter corresponding to the blob specified by the input parameters
+     */
+    BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName);
 }

extensions/common/azure/azure-blob-core/src/main/java/org/eclipse/edc/azure/blob/api/BlobStoreApiImpl.java

@@ -16,6 +16,7 @@
 
 import com.azure.core.credential.AzureSasCredential;
 import com.azure.core.util.BinaryData;
+import com.azure.identity.DefaultAzureCredentialBuilder;
 import com.azure.storage.blob.BlobServiceClient;
 import com.azure.storage.blob.BlobServiceClientBuilder;
 import com.azure.storage.blob.models.BlobItem;
@@ -35,7 +36,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.stream.Collectors;
 
 public class BlobStoreApiImpl implements BlobStoreApi {
 
@@ -65,30 +65,30 @@ public boolean exists(String accountName, String containerName) {
 
     @Override
     public String createContainerSasToken(String accountName, String containerName, String permissionSpec, OffsetDateTime expiry) {
-        BlobContainerSasPermission permissions = BlobContainerSasPermission.parse(permissionSpec);
-        BlobServiceSasSignatureValues vals = new BlobServiceSasSignatureValues(expiry, permissions);
-        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).generateSas(vals);
+        var permissions = BlobContainerSasPermission.parse(permissionSpec);
+        var values = new BlobServiceSasSignatureValues(expiry, permissions);
+        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).generateSas(values);
     }
 
     @Override
     public List<BlobItem> listContainer(String accountName, String containerName) {
-        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).listBlobs().stream().collect(Collectors.toList());
+        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).listBlobs().stream().toList();
     }
 
     @Override
     public void putBlob(String accountName, String containerName, String blobName, byte[] data) {
-        BlobServiceClient blobServiceClient = getBlobServiceClient(accountName);
+        var blobServiceClient = getBlobServiceClient(accountName);
         blobServiceClient.getBlobContainerClient(containerName).getBlobClient(blobName).upload(BinaryData.fromBytes(data), true);
     }
 
     @Override
     public String createAccountSas(String accountName, String containerName, String permissionSpec, OffsetDateTime expiry) {
-        AccountSasPermission permissions = AccountSasPermission.parse(permissionSpec);
+        var permissions = AccountSasPermission.parse(permissionSpec);
 
-        AccountSasService services = AccountSasService.parse("b");
-        AccountSasResourceType resourceTypes = AccountSasResourceType.parse("co");
-        AccountSasSignatureValues vals = new AccountSasSignatureValues(expiry, permissions, services, resourceTypes);
-        return getBlobServiceClient(accountName).generateAccountSas(vals);
+        var services = AccountSasService.parse("b");
+        var resourceTypes = AccountSasResourceType.parse("co");
+        var values = new AccountSasSignatureValues(expiry, permissions, services, resourceTypes);
+        return getBlobServiceClient(accountName).generateAccountSas(values);
     }
 
     @Override
@@ -110,8 +110,10 @@ private BlobServiceClient getBlobServiceClient(String accountName) {
             throw new IllegalArgumentException("No Object Storage credential found in vault!");
         }
 
-        BlobServiceClient blobServiceClient = new BlobServiceClientBuilder().credential(createCredential(accountKey, accountName))
-                .endpoint(createEndpoint(accountName))
+        var endpoint = createEndpoint(accountName);
+        var blobServiceClient = new BlobServiceClientBuilder()
+                .credential(createCredential(accountKey, accountName))
+                .endpoint(endpoint)
                 .buildClient();
 
         cache.put(accountName, blobServiceClient);
@@ -124,13 +126,19 @@ private StorageSharedKeyCredential createCredential(String accountKey, String ac
 
     @Override
     public BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName, String sharedKey) {
-        BlobServiceClientBuilder builder = new BlobServiceClientBuilder().credential(new StorageSharedKeyCredential(accountName, sharedKey));
+        var builder = new BlobServiceClientBuilder().credential(new StorageSharedKeyCredential(accountName, sharedKey));
         return getBlobAdapter(accountName, containerName, blobName, builder);
     }
 
     @Override
     public BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName, AzureSasCredential credential) {
-        BlobServiceClientBuilder builder = new BlobServiceClientBuilder().credential(credential);
+        var builder = new BlobServiceClientBuilder().credential(credential);
+        return getBlobAdapter(accountName, containerName, blobName, builder);
+    }
+
+    @Override
+    public BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName) {
+        var builder = new BlobServiceClientBuilder().credential(new DefaultAzureCredentialBuilder().build());
         return getBlobAdapter(accountName, containerName, blobName, builder);
     }
 

extensions/common/azure/azure-blob-core/src/main/java/org/eclipse/edc/azure/blob/validator/AzureStorageValidator.java

@@ -33,13 +33,17 @@ public class AzureStorageValidator {
     private static final int CONTAINER_MAX_LENGTH = 63;
     private static final int BLOB_MIN_LENGTH = 1;
     private static final int BLOB_MAX_LENGTH = 1024;
+    private static final int METADATA_MIN_LENGTH = 1;
+    private static final int METADATA_MAX_LENGTH = 4096;
     private static final Pattern ACCOUNT_REGEX = Pattern.compile("^[a-z0-9]+$");
     private static final Pattern CONTAINER_REGEX = Pattern.compile("^[a-z0-9]+(-[a-z0-9]+)*$");
+    private static final Pattern METADATA_REGEX = Pattern.compile("^[ -~]*$"); // US-ASCII
 
     private static final String ACCOUNT = "account";
     private static final String BLOB = "blob";
     private static final String CONTAINER = "container";
     private static final String KEY_NAME = "keyName";
+    private static final String METADATA = "metadata";
     private static final String INVALID_RESOURCE_NAME = "Invalid %s name";
     private static final String INVALID_RESOURCE_NAME_LENGTH = "Invalid %s name length, the name must be between %s and %s characters long";
     private static final String RESOURCE_NAME_EMPTY = "Invalid %s name, the name may not be null, empty or blank";
@@ -91,6 +95,23 @@ public static void validateBlobName(String blobName) {
         }
     }
 
+    /**
+     * Checks if a metadata value is valid.
+     * The restriction is based on allowed characters for HTTP header values. As there is no length restriction per
+     * header field, a reasonable restriction of 4096 character is assumed to leave some space, considering an
+     * overall length restriction for HTTP headers of approximately 8KB.
+     *
+     * @param metadata A String representing the metadata value to validate.
+     * @throws IllegalArgumentException if the string does not represent a valid metadata value.
+     */
+    public static void validateMetadata(String metadata) {
+        checkLength(metadata, METADATA, METADATA_MIN_LENGTH, METADATA_MAX_LENGTH);
+
+        if (!METADATA_REGEX.matcher(metadata).matches()) {
+            throw new IllegalArgumentException(String.format(INVALID_RESOURCE_NAME, METADATA));
+        }
+    }
+
     /**
      * Checks if key name is valid.
      *

extensions/common/azure/azure-blob-core/src/test/java/org/eclipse/edc/azure/blob/validator/AzureStorageValidatorTest.java

@@ -90,6 +90,20 @@ void validateBlobName_fail(String input) {
                 .isThrownBy(() -> AzureStorageValidator.validateBlobName(input));
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {  "abcdefghijklmnop", "-", "a/%!_- $K1~"})
+    void validateMetadata_success(String input) {
+        AzureStorageValidator.validateMetadata(input);
+    }
+
+    @ParameterizedTest
+    @ArgumentsSource(InvalidMetadataProvider.class)
+    @NullAndEmptySource
+    void validateMetadata_fail(String input) {
+        assertThatExceptionOfType(IllegalArgumentException.class)
+                .isThrownBy(() -> AzureStorageValidator.validateMetadata(input));
+    }
+
     private static class InvalidBlobNameProvider implements ArgumentsProvider {
         @Override
         public Stream<? extends Arguments> provideArguments(ExtensionContext context) throws Exception {
@@ -114,4 +128,13 @@ public Stream<? extends Arguments> provideArguments(ExtensionContext context) th
                     Arguments.of("a/b".repeat(253)));
         }
     }
-}
+
+    private static class InvalidMetadataProvider implements ArgumentsProvider {
+        @Override
+        public Stream<? extends Arguments> provideArguments(ExtensionContext context) throws Exception {
+            return Stream.of(
+                    Arguments.of("abcdefghijklmnop".repeat(256) + " a"),
+                    Arguments.of("a/%!_- $KÄ".repeat(64)));
+        }
+    }
+}

extensions/common/azure/azure-test/src/testFixtures/java/org/eclipse/edc/azure/testfixtures/TestFunctions.java

@@ -16,13 +16,28 @@
 
 import com.azure.storage.blob.BlobServiceClient;
 import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.blob.specialized.BlockBlobClient;
 import com.azure.storage.common.StorageSharedKeyCredential;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.ArrayList;
+import java.util.List;
+
 public final class TestFunctions {
     private TestFunctions() {
     }
 
+    @NotNull
+    public static BlobServiceClient getBlobServiceClient(String accountName, String key) {
+        var connectionString = new LocalConnectionStringBuilder()
+                .http().account(accountName).key(key).endpoints(accountName).build();
+        var client = new BlobServiceClientBuilder()
+                .connectionString(connectionString)
+                .buildClient();
+        client.getAccountInfo();
+        return client;
+    }
+
     @NotNull
     public static BlobServiceClient getBlobServiceClient(String accountName, String key, String endpoint) {
         var client = new BlobServiceClientBuilder()
@@ -35,7 +50,55 @@ public static BlobServiceClient getBlobServiceClient(String accountName, String
     }
 
     @NotNull
+    public static BlockBlobClient getBlobClient(String accountName, String containerName, String blobName, String token) {
+        var connectionString = new LocalConnectionStringBuilder()
+                .http().account(accountName).sharedAccessSignature(token).endpoints(accountName).build();
+        var serviceClient = new BlobServiceClientBuilder()
+                .connectionString(connectionString)
+                .buildClient();
+        return serviceClient.getBlobContainerClient(containerName).getBlobClient(blobName).getBlockBlobClient();
+    }
+
     public static String getBlobServiceTestEndpoint(String accountName) {
         return "http://127.0.0.1:10000/" + accountName;
     }
+
+    private static class LocalConnectionStringBuilder {
+
+        private final List<String> elements = new ArrayList<>();
+
+        public String build() {
+            return String.join(";", elements) + ";";
+        }
+
+        public LocalConnectionStringBuilder endpoints(String accountName) {
+            elements.add("BlobEndpoint=http://127.0.0.1:10000/" + accountName);
+
+            // Even though not used, a malformed connection string exception is thrown if not present
+            elements.add("QueueEndpoint=http://127.0.0.1:10001/" + accountName);
+            elements.add("TableEndpoint=http://127.0.0.1:10002/" + accountName);
+            elements.add("FileEndpoint=http://127.0.0.1:10003/" + accountName);
+            return this;
+        }
+
+        public LocalConnectionStringBuilder http() {
+            elements.add("DefaultEndpointsProtocol=http");
+            return this;
+        }
+
+        public LocalConnectionStringBuilder account(String accountName) {
+            elements.add("AccountName=" + accountName);
+            return this;
+        }
+
+        public LocalConnectionStringBuilder key(String accountKey) {
+            elements.add("AccountKey=" + accountKey);
+            return this;
+        }
+
+        public LocalConnectionStringBuilder sharedAccessSignature(String sharedAccessSignature) {
+            elements.add("SharedAccessSignature=" + sharedAccessSignature);
+            return this;
+        }
+    }
 }