How to use Azure Key Vault in Airflow

Datacoves integrates with the Airflow Secrets Backend Interface, offering support for its native Datacoves Secrets Backend, AWS Secrets Manager, and Azure Key Vault. For other Airflow-compatible Secrets Managers, please reach out to us.

Secrets backends can be configured at the project level, at the environment level, or both. See configure your Azure Key Vault for details.

When Datacoves runs in your Azure subscription, Airflow can authenticate to Key Vault with the cluster's Managed Identity, so no credentials need to be stored anywhere. Alternatively, a service principal with a client secret can be used on any Datacoves deployment.

Read variable from Azure Key Vault

Airflow's Variable.get searches multiple places:

1. Azure Key Vault (if configured)
2. Datacoves Secrets Manager
3. Airflow variables and environment variables

Once a variable is found, Airflow stops searching.

Secret naming

Variable keys and connection ids must start with datacoves- for the lookup to be sent to Azure Key Vault. The Key Vault secret name is the variable key with the airflow-variables- prefix (or airflow-connections- for connections), and underscores are translated to dashes because Key Vault secret names only allow letters, numbers and dashes:

In your DAG	Key Vault secret name
Variable.get("datacoves-my-secret")	airflow-variables-datacoves-my-secret
Variable.get("datacoves-my_secret")	airflow-variables-datacoves-my-secret
Connection datacoves-warehouse	airflow-connections-datacoves-warehouse

Best practices

1. Call Variable.get from within an Airflow/Datacoves decorator to fetch at runtime only. Fetching at the top level of a DAG file would query Key Vault on every DAG parse.
2. Keep the datacoves- prefix on everything you store in Key Vault for Airflow; lookups without it never reach the vault.

Example DAG using Azure Key Vault

try:
    # Airflow 3
    from airflow.sdk import Variable, dag, task
except ImportError:
    # Airflow 2
    from airflow.decorators import dag, task
    from airflow.models import Variable

from pendulum import datetime

@dag(
    catchup=False,
    default_args={
        "start_date": datetime(2024, 1, 1),
        "owner": "Mayra Pena",
        "email": "mayra@example.com",
        "email_on_failure": True,
    },
    tags=["version_1"],
    description="Read a variable from Azure Key Vault",
    schedule="0 0 1 */12 *",
)
def azure_key_vault_example():

    @task
    def read_secret_from_key_vault():
        # Fetch at runtime (inside the task), never at the top level of the
        # DAG file, so Key Vault is only called when the task runs.
        my_var = Variable.get("datacoves-my-secret")
        print(f"Fetched a {len(my_var)} character value from Azure Key Vault")

    read_secret_from_key_vault()

dag = azure_key_vault_example()

tip

To auto mask your secret you can use secret or password in the variable name since this will honor hide_sensitive_var_conn_fields. eg datacoves-my-password. Please see this documentation for a full list of masking words.

Using Azure Key Vault directly from Airflow

While not recommended, you can bypass the Datacoves secrets manager integration by configuring an Airflow connection and reading secrets with the Azure Key Vault SDK. The SDK (azure-identity and azure-keyvault-secrets) is already installed in Datacoves Airflow images as part of the Microsoft Azure provider.

When reading secrets this way, the secret naming rules above do not apply: you fetch any Key Vault secret by its exact name, with no airflow-variables- or datacoves- prefix required.

Configure an Airflow Connection

Create a new Airflow Connection with the service principal credentials:

Connection Id: azure_key_vault
Connection Type: Generic
Login: <client id>
Password: <client secret>

Extra:

{
  "tenant_id": "<tenant id>",
  "vault_url": "https://<your-vault>.vault.azure.net/"
}

Example DAG reading Key Vault directly

try:
    # Airflow 3
    from airflow.sdk import dag, task
except ImportError:
    # Airflow 2
    from airflow.decorators import dag, task

from airflow.hooks.base import BaseHook
from pendulum import datetime

@dag(
    catchup=False,
    default_args={
        "start_date": datetime(2024, 1, 1),
        "owner": "Noel Gomez",
        "email": "noel@example.com",
        "email_on_failure": True,
    },
    tags=["sample"],
    description="Read a secret directly from Azure Key Vault",
    schedule="0 0 1 */12 *",
)
def key_vault_direct_usage():

    @task
    def azure_secret():
        from azure.identity import ClientSecretCredential
        from azure.keyvault.secrets import SecretClient

        conn = BaseHook.get_connection("azure_key_vault")
        credential = ClientSecretCredential(
            tenant_id=conn.extra_dejson["tenant_id"],
            client_id=conn.login,
            client_secret=conn.password,
        )
        client = SecretClient(
            vault_url=conn.extra_dejson["vault_url"], credential=credential
        )
        secret = client.get_secret("my-secret-name")
        print(f"Fetched a {len(secret.value)} character value from Azure Key Vault")

    azure_secret()

dag = key_vault_direct_usage()

tip

When Datacoves runs in your Azure subscription with Managed Identity, no connection or credentials are needed at all: replace ClientSecretCredential with DefaultAzureCredential() from azure.identity and pass your vault URL to SecretClient directly.

Check when a secret is being fetched from Azure

It is a good idea to verify that secrets are only being fetched when expected. To do this, enable diagnostic logging on your Key Vault:

1. In the Azure Portal, go to your Key Vault
2. Click Diagnostic settings and send the Audit (AuditEvent) category to a Log Analytics workspace
3. In Log Analytics, query for SecretGet operations:

AzureDiagnostics
| where ResourceType == "VAULTS" and OperationName == "SecretGet"
| project TimeGenerated, requestUri_s, identity_claim_appid_g, ResultSignature
| order by TimeGenerated desc

Review the request URI (which contains the secret name) and the timestamp. Note: it may take a few minutes for events to show up in Log Analytics.