Access to Azure blobs from Azure Function by using Managed Identity


How to grant access to the blob files available on the Azure Storage Account without providing a SAS token nor Access Key?

In some cases, it might be helpful, especially when access to the Azure Storage Account is available via Managed Instance.  In that case, we don't want to use any secrets provided explicitly. 

In fact, we still want to use the SAS token behind the scene, but generated on-the-fly, with a very short lifetime.  Each SAS token requires to be built based on the Access Key. We may want to use one of 2 access keys provided by the Storage Accounts, but it's not a good option as we don't want to deal with them.  Azure lets us generate a temporary access key, based on our credentials, and then use it to generate a SAS token.


In that case, we need to:
  1. Create a Storage Account
  2. Create Azure Function App 
  3. Assign the Storage Blob Delegator (or Storage Blob Data Owner) role to the Manage Identity of the Function App. The action needs to be performed against the Storage Account (Access Control (IAM))

The function sample may look like below. Once deployed, you will be able to get the payload of the sample.txt, without providing any extra tokes (SAS/Access key).

https://{function app name}.azurewebsites.net/api/{function name}?code={function code}

Useful links:

import logging
import azure.functions as func
from datetime
import datetime, timedelta
from azure.identity import DefaultAzureCredential
from azure.storage.blob  import (
    BlobClient,
    BlobSasPermissions,
    BlobServiceClient,
    generate_blob_sas
)

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')


    storage_acct_name = 'whaisa'
    container_name = 'mycontainer'
    blob_name  = 'sample.txt'

    account_url = f'https://{storage_acct_name}.blob.core.windows.net'
       
    credential = DefaultAzureCredential()
   
    client = BlobServiceClient(account_url, credential=credential)

    key_start_time = datetime.utcnow() - timedelta(hours=1)
    key_expiry_time = datetime.utcnow() + timedelta(hours=1)
    udk = client.get_user_delegation_key(key_start_time, key_expiry_time)

    sas_token = generate_blob_sas(account_name=storage_acct_name,
                                    container_name=container_name,
                                    blob_name=blob_name,
                                    account_key= udk,
                                    permission=BlobSasPermissions(read=True),
                                    expiry=key_expiry_time)

    sas_url = (
        f'{account_url}/{container_name}/{blob_name}?{sas_token}'
    )

    blob_client = BlobClient.from_blob_url(sas_url)
    blob_data = blob_client.download_blob(encoding='UTF-8')
    data = blob_data.readall()
   
    return func.HttpResponse(
            data,
            status_code=200
    )


Comments

Popular posts from this blog

Podman using Ubuntu WSL2

Automatization of test cases in SCRUM

Azure Synapse - first contact