Clientseitige Entschlüsselung von Azure Blob Storage mit Python ⇐ Python

[Guest]

Ich möchte mit clientseitiger Verschlüsselung verschlüsselte Blobs mit dem folgenden C#-Code entschlüsseln, der in der Produktion ausgeführt wird und den Microsoft C#-Beispielen folgt:

Code: Select all

public static IServiceCollection AddEncryptedFileStorage(this IServiceCollection services, string configurationSectionName) where T : class, IFileStorage
{
string clientName = configurationSectionName;
services.AddAzureClients(clientBuilder =>
{
clientBuilder.AddClient((options, credential, serviceProvider) =>
{
var configuration = serviceProvider.GetRequiredService();
var configurationSection = configuration.GetSection(configurationSectionName);

var storageAccountUri = FileStorageHelper.GetStorageAccountUri(configurationSection.GetRequiredValue("StorageAccountName"));
string encryptionKeyId = configurationSection.GetValue("EncryptionKeyId")!;

var keyClient = serviceProvider.GetRequiredService();

KeyVaultKey key = keyClient.GetKey(encryptionKeyId, cancellationToken: CancellationToken.None);

var encryptionOptions = new ClientSideEncryptionOptions(ClientSideEncryptionVersion.V2_0)
{
KeyEncryptionKey = new CryptographyClient(key.Id, credential),
KeyResolver = new KeyResolver(credential),
KeyWrapAlgorithm = "RSA-OAEP"
};

var blobClientOptions = new SpecializedBlobClientOptions
{
ClientSideEncryption = encryptionOptions,
Retry = { Delay = options.Retry.NetworkTimeout, MaxRetries = options.Retry.MaxRetries }
};

return new BlobServiceClient(storageAccountUri, credential, blobClientOptions);
}).WithName(clientName);
});
services.TryAddSingleton(typeof(T), sp => FileStorageFactory(sp, clientName, configurationSectionName));

services.AddHealthChecks().AddBlobStorage(clientName);

return services;
}

Den Microsoft Python-Dokumenten folgend habe ich versucht, den folgenden Code auszuführen, der Daten ausgibt, wenn Verschlüsselungsoptionen nicht festgelegt sind, ansonsten aber diesen Fehler ausgibt:

Code: Select all

HttpResponseError: Decryption failed.
File /local_disk0/.ephemeral_nfs/envs/pythonEnv-c5de91e5-86a9-4892-ae41-922218e4545e/lib/python3.10/site-packages/azure/storage/blob/_download.py:70, in process_content(data, start_offset, end_offset, encryption)
69 try:
---> 70     return decrypt_blob(
71         encryption.get("required") or False,
72         encryption.get("key"),
73         encryption.get("resolver"),
74         content,
75         start_offset,
76         end_offset,
77         data.response.headers,
78     )
79 except Exception as error:
File /local_disk0/.ephemeral_nfs/envs/pythonEnv-c5de91e5-86a9-4892-ae41-922218e4545e/lib/python3.10/site-packages/azure/storage/blob/_encryption.py:905, in decrypt_blob(require_encryption, key_encryption_key, key_resolver, content, start_offset, end_offset, response_headers)
903     raise ValueError('Specified encryption version is not supported.')
--> 905 content_encryption_key = _validate_and_unwrap_cek(encryption_data, key_encryption_key, key_resolver)
907 if version == _ENCRYPTION_PROTOCOL_V1:
File /local_disk0/.ephemeral_nfs/envs/pythonEnv-c5de91e5-86a9-4892-ae41-922218e4545e/lib/python3.10/site-packages/azure/storage/blob/_encryption.py:658, in _validate_and_unwrap_cek(encryption_data, key_encryption_key, key_resolver)
657 if not hasattr(key_encryption_key, 'get_kid') or not callable(key_encryption_key.get_kid):
--> 658     raise AttributeError(_ERROR_OBJECT_INVALID.format('key encryption key', 'get_kid'))
659 if not hasattr(key_encryption_key, 'unwrap_key') or not callable(key_encryption_key.unwrap_key):
AttributeError: key encryption key does not define a complete interface.  Value of get_kid is either missing or invalid.

Code: Select all

from azure.storage.blob import BlobServiceClient
from azure.identity import DefaultAzureCredential, DeviceCodeCredential
from azure.keyvault.keys import KeyClient
from azure.keyvault.keys.crypto import CryptographyClient
from azure.keyvault.keys.crypto import KeyWrapAlgorithm
from azure.core.exceptions import ResourceNotFoundError

credential = DeviceCodeCredential(additionally_allowed_tenants =[""])

def configure_blob_service_client(key_id, cred):
storage_account_name = ""
storage_account_uri = f"https://{storage_account_name}.blob.core.windows.net"

encryption_options = {
#"require_encryption": True,
"key_encryption_key": CryptographyClient(key_id, credential),#ExtendedCryptographyClient(key.id , credential),
"encryption_version": '2.0',
"key_wrap_algorithm": KeyWrapAlgorithm.rsa_oaep
}

blob_service_client = BlobServiceClient(account_url=storage_account_uri, credential=cred, **encryption_options)
return blob_service_client

key_vault_name = f""
key_vault_uri = f"https://{key_vault_name}.vault.azure.net"

key_client = KeyClient(vault_url=key_vault_uri, credential=credential)

key_name = "/"
key = key_client.get_key(key_name)
display(key.id)

blob_service_client = configure_blob_service_client(key.id, credential)

blob_client = blob_service_client.get_blob_client(container="", blob="")

downloaded = blob_client.download_blob()
blob_data = downloaded.readall()

blob_data

Das SDK erwartet also, dass der Schlüsselverschlüsselungsschlüssel eine get_kid()-Methode (SDK-Quellcode) hat, der CryptographyClient des SDK verfügt jedoch nicht über eine solche Methode.
Ich habe versucht, die get_kid()-Methode selbst zu implementieren, was zu anderen Problemen führt, also wahrscheinlich nicht der richtige Weg