By Richie Lee

Hello!

There are a couple of ways to secure access to Azure Storage, and one of the mechanisms I’ve used in an app recently was SAS or shared access signatures. One of the reasons I chose this over Storage Access Keys is because their granularity in terms of what they can control access to and how much they can do with that access (ie read/write/delete). This is all very useful, however they are temporary and have no way of extending the expiry date. This can be a pain in terms of refreshing them, so you can instead create a shared access policy and link the sas key to that. This means that you can extend the life of the key. In theory you can set a key that needs to be used every day for the next 3 years to have an expiry date of tomorrow, and then extend it by one day every day. If the key is then compromised then it’s usefullness is short, and more importantly if a sas key is found to be compromised then the access can be revoked through the shared access policy.

Anyway, I wrote a python script that is executed as a Azure Databricks job via an ADF pipeline; it sets the expiry date three days into the future. So now I don’t have to think about expiry dates at all, as long as the container is used then the sas key will be valid, and if it ever becomes dormant then I don’t need to worry about the sas key in the keyvault being compromised. There’s 6 vars that need to be set. They’re surrounded by <>. These vars include secrets, so keep that in mind when modifying the script.

import os
import sys
from datetime import datetime, timedelta
from azure.storage.blob import AccessPolicy, BlobServiceClient, ContainerSasPermissions
from azure.identity import ClientSecretCredential
from azure.mgmt.resource import ResourceManagementClient
from azure.mgmt.storage import StorageManagementClient
from azure.mgmt.storage.models import StorageAccountCreateParameters

os.environ["AZURE_TENANT_ID"] = <enter_tenant_id>
os.environ["AZURE_CLIENT_ID"] = <enter_client_id>
os.environ["AZURE_CLIENT_SECRET"] = <enter_client_secret>

subscription_id = <enter_subscription_id>

credentials = ClientSecretCredential(
    client_id=os.environ["AZURE_CLIENT_ID"],
    client_secret=os.environ["AZURE_CLIENT_SECRET"],
    tenant_id=os.environ["AZURE_TENANT_ID"],
)
resource_client = ResourceManagementClient(credentials, subscription_id)
storage_client = StorageManagementClient(credentials, subscription_id)

storage_keys = storage_client.storage_accounts.list_keys(
    "<enter_resource_group>", "<enterstorageaccountname>"
)
storage_keys = {v.key_name: v.value for v in storage_keys.keys}

service_client = BlobServiceClient(
    account_url="https://<enterstorageaccountname>.blob.core.windows.net",
    credential=(storage_keys["key1"]),
)
container_client = service_client.get_container_client(str(sys.argv[1]))

print("\n..Getting container access policy")
access_policy = container_client.get_container_access_policy()
print(f"Blob Access Type: {access_policy['public_access']}")
for identifier in access_policy["signed_identifiers"]:
    print(
        f"Identifier '{identifier.id}' has permissions '{identifier.access_policy.permission}'' and expires {identifier.access_policy.expiry}"
    )
    print(identifier)
    access_policy = AccessPolicy(
        permission=identifier.access_policy.permission,
        expiry=datetime.utcnow() + timedelta(days=3),
        start=identifier.access_policy.start,
    )
    identifiers = {identifier.id: access_policy}
    container_client.set_container_access_policy(signed_identifiers=identifiers)

print("\n..Getting container access policy")
access_policy = container_client.get_container_access_policy()
print(f"Blob Access Type: {access_policy['public_access']}")
for identifier in access_policy["signed_identifiers"]:
    print(
        f"Identifier '{identifier.id}' has permissions '{identifier.access_policy.permission}'' and now expires {identifier.access_policy.expiry}"
    )
    print(identifier)