fix: use of resolved shared key, when available, in transfer requests for … (#189)

* Use of resolved shared key, when available, in transfer requests for several files.

* Add Account Cache and UT.

* Transfer Account cache logic to own class.

* Added Unit Tests to Account Cache.

* Added Override annotation.

* Update DEPENDENCIES file.

* Update DEPENDENCIES file.

* Fix incorrect implementation.

* Update DEPENDENCIES file.

* Update DEPENDENCIES file.

Author: bmg13

DEPENDENCIES

@@ -41,9 +41,10 @@ public interface BlobStoreApi {
      * @param accountName The name of the storage account
      * @param containerName The name of the container within the storage account
      * @param directory The name of the folder within the container of the storage account
+     * @param accountKey The key of the storage account
      * @return Lazy loaded list of blobs from folder specified by the input parameters
      */
-    List<BlobItem> listContainerFolder(String accountName, String containerName, String directory);
+    List<BlobItem> listContainerFolder(String accountName, String containerName, String directory, String accountKey);
 
     void putBlob(String accountName, String containerName, String blobName, byte[] data);
 

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

@@ -17,7 +17,6 @@
 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;
 import com.azure.storage.blob.models.ListBlobsOptions;
@@ -30,61 +29,61 @@
 import com.azure.storage.common.sas.AccountSasSignatureValues;
 import org.eclipse.edc.azure.blob.adapter.BlobAdapter;
 import org.eclipse.edc.azure.blob.adapter.DefaultBlobAdapter;
+import org.eclipse.edc.azure.blob.cache.AccountCache;
+import org.eclipse.edc.azure.blob.cache.AccountCacheImpl;
 import org.eclipse.edc.spi.security.Vault;
 
 import java.time.OffsetDateTime;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+
+import static org.eclipse.edc.azure.blob.utils.BlobStoreUtils.createEndpoint;
 
 public class BlobStoreApiImpl implements BlobStoreApi {
 
-    private final Vault vault;
     private final String blobstoreEndpointTemplate;
-    private final Map<String, BlobServiceClient> cache = new HashMap<>();
+    private final AccountCache accountCache;
 
     public BlobStoreApiImpl(Vault vault, String blobstoreEndpointTemplate) {
-        this.vault = vault;
         this.blobstoreEndpointTemplate = blobstoreEndpointTemplate;
+        this.accountCache = new AccountCacheImpl(vault, blobstoreEndpointTemplate);
     }
 
     @Override
     public void createContainer(String accountName, String containerName) {
-        getBlobServiceClient(accountName).createBlobContainer(containerName);
+        accountCache.getBlobServiceClient(accountName).createBlobContainer(containerName);
     }
 
     @Override
     public void deleteContainer(String accountName, String containerName) {
-        getBlobServiceClient(accountName).deleteBlobContainer(containerName);
+        accountCache.getBlobServiceClient(accountName).deleteBlobContainer(containerName);
     }
 
     @Override
     public boolean exists(String accountName, String containerName) {
-        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).exists();
+        return accountCache.getBlobServiceClient(accountName).getBlobContainerClient(containerName).exists();
     }
 
     @Override
     public String createContainerSasToken(String accountName, String containerName, String permissionSpec, OffsetDateTime expiry) {
         var permissions = BlobContainerSasPermission.parse(permissionSpec);
         var values = new BlobServiceSasSignatureValues(expiry, permissions);
-        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).generateSas(values);
+        return accountCache.getBlobServiceClient(accountName).getBlobContainerClient(containerName).generateSas(values);
     }
 
     @Override
     public List<BlobItem> listContainer(String accountName, String containerName) {
-        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).listBlobs().stream().toList();
+        return accountCache.getBlobServiceClient(accountName).getBlobContainerClient(containerName).listBlobs().stream().toList();
     }
 
     @Override
-    public List<BlobItem> listContainerFolder(String accountName, String containerName, String directory) {
+    public List<BlobItem> listContainerFolder(String accountName, String containerName, String directory, String accountKey) {
         var options = new ListBlobsOptions().setPrefix(directory);
-        return getBlobServiceClient(accountName).getBlobContainerClient(containerName).listBlobs(options, null).stream().toList();
+        return accountCache.getBlobServiceClient(accountName, accountKey).getBlobContainerClient(containerName).listBlobs(options, null).stream().toList();
     }
 
     @Override
     public void putBlob(String accountName, String containerName, String blobName, byte[] data) {
-        var blobServiceClient = getBlobServiceClient(accountName);
+        var blobServiceClient = accountCache.getBlobServiceClient(accountName);
         blobServiceClient.getBlobContainerClient(containerName).getBlobClient(blobName).upload(BinaryData.fromBytes(data), true);
     }
 
@@ -95,41 +94,15 @@ public String createAccountSas(String accountName, String containerName, String
         var services = AccountSasService.parse("b");
         var resourceTypes = AccountSasResourceType.parse("co");
         var values = new AccountSasSignatureValues(expiry, permissions, services, resourceTypes);
-        return getBlobServiceClient(accountName).generateAccountSas(values);
+        return accountCache.getBlobServiceClient(accountName).generateAccountSas(values);
     }
 
     @Override
     public byte[] getBlob(String account, String container, String blobName) {
-        var client = getBlobServiceClient(account);
+        var client = accountCache.getBlobServiceClient(account);
         return client.getBlobContainerClient(container).getBlobClient(blobName).downloadContent().toBytes();
     }
 
-    private BlobServiceClient getBlobServiceClient(String accountName) {
-        Objects.requireNonNull(accountName, "accountName");
-
-        if (cache.containsKey(accountName)) {
-            return cache.get(accountName);
-        }
-
-        var accountKey = vault.resolveSecret(accountName + "-key1");
-        var endpoint = createEndpoint(accountName);
-
-        var blobServiceClient = accountKey == null ?
-                new BlobServiceClientBuilder().credential(new DefaultAzureCredentialBuilder().build())
-                        .endpoint(endpoint)
-                        .buildClient() :
-                new BlobServiceClientBuilder().credential(createCredential(accountKey, accountName))
-                        .endpoint(endpoint)
-                        .buildClient();
-
-        cache.put(accountName, blobServiceClient);
-        return blobServiceClient;
-    }
-
-    private StorageSharedKeyCredential createCredential(String accountKey, String accountName) {
-        return new StorageSharedKeyCredential(accountName, accountKey);
-    }
-
     @Override
     public BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName, String sharedKey) {
         var builder = new BlobServiceClientBuilder().credential(new StorageSharedKeyCredential(accountName, sharedKey));
@@ -150,7 +123,7 @@ public BlobAdapter getBlobAdapter(String accountName, String containerName, Stri
 
     private BlobAdapter getBlobAdapter(String accountName, String containerName, String blobName, BlobServiceClientBuilder builder) {
         var blobServiceClient = builder
-                .endpoint(createEndpoint(accountName))
+                .endpoint(createEndpoint(blobstoreEndpointTemplate, accountName))
                 .buildClient();
 
         var blockBlobClient = blobServiceClient
@@ -160,8 +133,4 @@ private BlobAdapter getBlobAdapter(String accountName, String containerName, Str
 
         return new DefaultBlobAdapter(blockBlobClient);
     }
-
-    private String createEndpoint(String accountName) {
-        return String.format(blobstoreEndpointTemplate, accountName);
-    }
 }

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

@@ -0,0 +1,38 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  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:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.azure.blob.cache;
+
+import com.azure.storage.blob.BlobServiceClient;
+
+public interface AccountCache {
+
+
+    /**
+     * Initially, confirms if account is stored in cache and, if so, returns it. If not, saves the account in cache and retrieves it.
+     *
+     * @param accountName The name of the storage account.
+     * @param accountKey  The key of the storage account
+     * @return The blob service client corresponding to the client to a storage account.
+     */
+    BlobServiceClient getBlobServiceClient(String accountName, String accountKey);
+
+    /**
+     * Initially, confirms if account is stored in cache and, if so, returns it. If not, resolved the account key and saves the account in cache and retrieves it.
+     *
+     * @param accountName The name of the storage account.
+     * @return The blob service client corresponding to the client to a storage account.
+     */
+    BlobServiceClient getBlobServiceClient(String accountName);
+}

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

@@ -0,0 +1,88 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  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:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.azure.blob.cache;
+
+import com.azure.identity.DefaultAzureCredentialBuilder;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
+import org.eclipse.edc.spi.security.Vault;
+import org.eclipse.edc.util.string.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.eclipse.edc.azure.blob.utils.BlobStoreUtils.createEndpoint;
+
+public class AccountCacheImpl implements AccountCache {
+
+    private final Vault vault;
+    private final Map<String, BlobServiceClient> accountCache = new HashMap<>();
+    private final String endpointTemplate;
+
+    public AccountCacheImpl(Vault vault, String endpointTemplate) {
+        this.vault = vault;
+        this.endpointTemplate = endpointTemplate;
+    }
+
+    @Override
+    public BlobServiceClient getBlobServiceClient(String accountName) {
+        if (isAccountInCache(accountName)) {
+            return getAccount(accountName);
+        }
+
+        var accountKey = vault.resolveSecret(accountName + "-key1");
+
+        return saveAccount(accountName, accountKey);
+    }
+
+    @Override
+    public BlobServiceClient getBlobServiceClient(String accountName, String accountKey) {
+        if (StringUtils.isNullOrBlank(accountKey)) {
+            return getBlobServiceClient(accountName);
+        }
+
+        if (isAccountInCache(accountName)) {
+            return getAccount(accountName);
+        }
+
+        return saveAccount(accountName, accountKey);
+    }
+
+    private BlobServiceClient getAccount(String accountName) {
+        return accountCache.get(accountName);
+    }
+
+    private boolean isAccountInCache(String accountName) {
+        Objects.requireNonNull(accountName, "accountName");
+        return accountCache.containsKey(accountName);
+    }
+
+    private BlobServiceClient saveAccount(String accountName, String accountKey) {
+        var endpoint = createEndpoint(endpointTemplate, accountName);
+
+        var blobServiceClient = accountKey == null ?
+                new BlobServiceClientBuilder().credential(new DefaultAzureCredentialBuilder().build())
+                        .endpoint(endpoint)
+                        .buildClient() :
+                new BlobServiceClientBuilder().credential(new StorageSharedKeyCredential(accountName, accountKey))
+                        .endpoint(endpoint)
+                        .buildClient();
+
+        accountCache.put(accountName, blobServiceClient);
+        return blobServiceClient;
+    }
+}

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

@@ -0,0 +1,22 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  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:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.azure.blob.utils;
+
+public class BlobStoreUtils {
+
+    public static String createEndpoint(String endpointTemplate, String accountName) {
+        return String.format(endpointTemplate, accountName);
+    }
+}

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

@@ -0,0 +1,74 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  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:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.azure.blob.cache;
+
+import org.eclipse.edc.spi.security.Vault;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.edc.azure.blob.utils.BlobStoreUtils.createEndpoint;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+class AccountCacheImplTest {
+
+    private static final String ENDPOINT_TEMPLATE = "https://%s.blob.core.windows.net";
+    private static final String ACCOUNT_NAME = "test_account";
+    private static final String OTHER_ACCOUNT_NAME = "other_test_account";
+    private static final String ACCOUNT_KEY = "test_key";
+
+    private final AccountCache accountCache = new AccountCacheImpl(mock(Vault.class), ENDPOINT_TEMPLATE);
+
+    @Test
+    void getAccount_withKey_succeeds() {
+        var result = accountCache.getBlobServiceClient(ACCOUNT_NAME, ACCOUNT_KEY);
+        assertThat(result.getAccountName()).isEqualTo(ACCOUNT_NAME);
+        assertEquals(result.getAccountUrl(), createEndpoint(ENDPOINT_TEMPLATE, ACCOUNT_NAME));
+    }
+
+    @Test
+    void getAccount_succeeds() {
+        var result = accountCache.getBlobServiceClient(ACCOUNT_NAME);
+        assertThat(result.getAccountName()).isEqualTo(ACCOUNT_NAME);
+        assertEquals(result.getAccountUrl(), createEndpoint(ENDPOINT_TEMPLATE, ACCOUNT_NAME));
+    }
+
+    @Test
+    void getAccount_sameAccount_succeeds() {
+        var firstAccount = accountCache.getBlobServiceClient(ACCOUNT_NAME);
+        var secondAccount = accountCache.getBlobServiceClient(ACCOUNT_NAME);
+        assertThat(firstAccount).isEqualTo(secondAccount);
+    }
+
+    @Test
+    void getAccount_differentAccounts_succeeds() {
+        var firstAccount = accountCache.getBlobServiceClient(ACCOUNT_NAME);
+        var secondAccount = accountCache.getBlobServiceClient(OTHER_ACCOUNT_NAME);
+        assertThat(firstAccount).isNotEqualTo(secondAccount);
+    }
+
+    @Test
+    void getAccount_sameAccountWithKey_succeeds() {
+        var firstAccount = accountCache.getBlobServiceClient(ACCOUNT_NAME, ACCOUNT_KEY);
+        var secondAccount = accountCache.getBlobServiceClient(ACCOUNT_NAME, ACCOUNT_KEY);
+        assertThat(firstAccount).isEqualTo(secondAccount);
+    }
+
+    @Test
+    void getAccount_WithNullKey_succeeds() {
+        var result = accountCache.getBlobServiceClient(ACCOUNT_NAME, null);
+        assertThat(result.getAccountName()).isEqualTo(ACCOUNT_NAME);
+    }
+}

extensions/common/azure/azure-blob-core/src/test/java/org/eclipse/edc/azure/blob/cache/AccountCacheImplTest.java

@@ -56,7 +56,7 @@ private AzureStorageDataSource() {
     public StreamResult<Stream<Part>> openPartStream() {
 
         if (!StringUtils.isNullOrBlank(blobPrefix)) {
-            var folderBlobs = blobStoreApi.listContainerFolder(accountName, containerName, blobPrefix);
+            var folderBlobs = blobStoreApi.listContainerFolder(accountName, containerName, blobPrefix, sharedKey);
             if (folderBlobs.isEmpty()) {
                 monitor.severe(format("Error listing blobs in the container %s with prefix %s", containerName, blobPrefix));
                 return failure(new StreamFailure(List.of(format("Error listing blobs in the container %s with prefix %s", containerName, blobPrefix)), GENERAL_ERROR));