Managing Airflow Secrets Securely with Azure Key Vault

Learn how to securely manage your Airflow secrets using Azure Key Vault for enhanced protection of sensitive credentials. 🔐

Note: If you’re not a medium member,** CLICK HERE**

In modern data pipelines, securely managing secrets like API keys, database credentials, and authentication tokens is crucial. Apache Airflow provides several ways to handle secrets, but storing them in a secure, centralized location is best practice.

This blog will walk you through integrating Azure Key Vault with Airflow to fetch and use secrets securely in your DAGs.

Why Use Azure Key Vault in Airflow?

✅ Centralized Secret Management — Store all sensitive information in one place.

✅ Access Control — Use Azure Active Directory (AAD) for role-based access.

✅ Automatic Secret Rotation — Update secrets dynamically without modifying DAGs.

✅ Improved Security — Keep secrets out of Airflow variables and configurations.

1. Configuring Airflow Connection for Azure Key Vault

Step 1: Add Azure Key Vault Connection in Airflow

1. Open Airflow UI (http://localhost:8080)
2. Navigate to Admin → Connections → + Add a new connection
3. Enter the following details:

• Connection Id: azure_keyvault
• Connection Type: Azure
• Azure Client ID: <client_id> (from service principal)
• Azure Secret: <client_secret>
• Extra (JSON format):{“tenantId”: “<tenant_id>”,“vaultUrl”: “https://<your-keyvault-name>.vault.azure.net”}

4. Click Save.

2. Implementing an Azure Key Vault Client in Airflow

We will create a reusable azure_keyvault_client.py module to fetch secrets.

Step 1: Install Dependencies

Ensure you have the required Azure SDKs installed:

pip install azure-identity azure-keyvault-secrets

Step 2: Create the Azure Key Vault Client in Airflow

from azure.identity import ClientSecretCredential
from azure.keyvault.secrets import SecretClient
from airflow.hooks.base import BaseHook

# Module-level variable to hold the client instance
_secret_client = None

def get_secret_client():
    """
    Initializes and returns an Azure Key Vault client using Airflow connection details.
    The connection is retrieved from Airflow's connection system to ensure security and maintainability.
    """
    global _secret_client

    if _secret_client is None:
        azure_connection = BaseHook.get_connection("azure_keyvault")

        # Extract connection details
        tenant_id = azure_connection.extra_dejson.get("tenantId")
        client_id = azure_connection.login
        client_secret = azure_connection.password
        vault_url = azure_connection.extra_dejson.get("vaultUrl")

        if not all([tenant_id, client_id, client_secret, vault_url]):
            raise ValueError("Missing required Azure connection details")

        # Set up Azure client
        credential = ClientSecretCredential(tenant_id=tenant_id, client_id=client_id, client_secret=client_secret)
        _secret_client = SecretClient(vault_url=vault_url, credential=credential)

    return _secret_client

def fetch_secret(secret_name):
    """
    Fetches a secret value from Azure Key Vault.

    Args:
        secret_name (str): The name of the secret to retrieve.

    Returns:
        str: The retrieved secret value.
    """
    client = get_secret_client()
    secret = client.get_secret(secret_name)
    return secret.value

• get_secret_client(): Retrieves the Azure Key Vault connection from Airflow, extracts authentication details, and initializes a SecretClient to interact with Key Vault.
• fetch_secret(secret_name): Uses the client to fetch and return a specific secret from Azure Key Vault.

3. Fetching Secrets in an Airflow DAG

With our *azure_keyvault_client.py* ready, we can now securely fetch secrets inside our DAGs.

Example: Using a Secret in a DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from azure_keyvault_client import fetch_secret

def print_secret():
    secret_value = fetch_secret("db-password")
    print(f"Fetched Secret: {secret_value}")

with DAG("azure_keyvault_example",
         schedule_interval="@daily",
         start_date=datetime(2024, 1, 1),
         catchup=False) as dag:

    get_secret_task = PythonOperator(
        task_id="fetch_secret",
        python_callable=print_secret
    )

• Defines an Airflow DAG named***azure_keyvault_example*.**
• Uses PythonOperator to fetch and print the secret from Azure Key Vault.
• The fetch_secret function retrieves the stored secret and prints it.

Conclusion

By integrating Azure Key Vault with Airflow, we ensure that sensitive credentials are never hard-coded in DAGs or Airflow variables. This improves security, enables seamless secret rotation, and aligns with best practices for managing cloud-based data workflows.

Need Help?

Feel free to reach out if you have any questions or need assistance with the setup.

📩 Email: Vishal Chandra 🔗 LinkedIn: Vishal Chandra |Siddharth Maurya

Happy coding! 🚀