Technical Tutorials

Managing credentials in data pipelines is a critical security concern. Hardcoding secrets or relying on static environment variables often leads to configuration drift and security vulnerabilities. With Airflow 2.4 and later, we can leverage dynamic variable injection to pull sensitive data directly from AWS Secrets Manager, ensuring your DAGs remain secure, flexible, and compliant.

Why Dynamic Injection Over Static Configuration

Traditional approaches often involve placing secrets in airflow.cfg or using environment variables that require container restarts to update. This is problematic for frequent rotation or multi-environment deployments. Dynamic injection allows Airflow to fetch secrets at runtime. By combining custom operators or the Connection object with AWS Boto3, you can ensure that credentials are encrypted in transit and stored at rest by AWS, while only being exposed to specific tasks that need them.

Prerequisites and Setup

To get started, ensure your Airflow environment has the necessary permissions. The IAM role or user associated with your Airflow workers must have secretsmanager:GetSecretValue permissions for the specific secret ARNs.

First, install the required dependencies in your Airflow requirements file:


boto3>=1.26.0

Implementing the Secret Fetcher

Creating a reusable utility function is best practice. This keeps your DAGs clean and centralizes error handling. Below is a robust implementation of a secret retrieval function that integrates with Boto3.


import boto3
from botocore.exceptions import ClientError
from airflow.models import Variable

def get_aws_secret(secret_id: str, region_name: str = 'us-east-1') -> dict:
    """
    Retrieve a secret from AWS Secrets Manager.
    
    Args:
        secret_id: The name or ARN of the secret.
        region_name: The AWS region where the secret resides.
        
    Returns:
        A dictionary containing the secret string parsed as JSON.
    """
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )
    
    try:
        get_secret_value_response = client.get_secret_value(SecretId=secret_id)
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == "ResourceNotFoundException":
            raise Exception(f"Secret with ID {secret_id} not found")
        else:
            raise e

    # Handle JSON vs String secrets
    if 'SecretString' in get_secret_value_response:
        return eval(get_secret_value_response['SecretString'])
    elif 'SecretBinary' in get_secret_value_response:
        return get_secret_value_response['SecretBinary'].decode()

# Usage in DAG
import os
# This would typically be wrapped in a Task or Operator
DB_PASSWORD = get_aws_secret('prod/db-credentials')['password']

Integrating with Airflow Connections

Instead of passing raw passwords to individual tasks, it is better to leverage Airflow's built-in connection management. You can use a custom Connection or a plugin that resolves the connection extra from Secrets Manager upon instantiation.

For a more advanced approach, consider using the @task decorator in Airflow 2.4+ (TaskFlow API) to inject secrets directly into task logic without polluting the global namespace.


from airflow.decorators import dag, task

@dag(schedule=None, tags=['secrets-demo'], start_date=datetime(2024, 1, 1))
def secure_data_pipeline():
    
    @task
    def fetch_and_process():
        # Secrets are fetched only when the task runs
        credentials = get_aws_secret('app/api-key')
        api_key = credentials['api_key']
        
        # Perform API call with secret
        # requests.post(url, headers={'Authorization': f'Bearer {api_key}'})
        return f"API call made with key ending in {api_key[-4:]}"

    fetch_and_process()

secure_data_pipeline()

Security Best Practices

Least Privilege: Create separate IAM roles for each environment (dev, staging, prod) with access only to specific secret paths.
Secret Rotation: Configure AWS Secrets Manager to auto-rotate your secrets. Since Airflow fetches them dynamically, you benefit immediately without redeploying.
Logging Masking: Never log the full secret value. In the example above, we only log the last 4 characters to aid debugging without compromising security.
Encryption at Rest: Ensure you use KMS Customer-Managed Keys (CMKs) for your secrets to meet strict compliance requirements.

Conclusion

By adopting dynamic variable injection with AWS Secrets Manager in Airflow 2.4+, you eliminate the risk of stale credentials and reduce the attack surface of your infrastructure. This pattern scales well for complex microservices architectures and ensures that your data engineering teams can focus on building logic, not managing keys. Implement these practices today to future-proof your pipeline security.