Secrets Management

Last Updated: 2025-10-30
Status: Production
Implementation: src/secrets.py

Overview

The blog_data pipeline uses a centralized secrets management system that provides a consistent interface for all credential and secret management across the codebase.

Design Principles

  1. Environment Variables First - For local development convenience
  2. AWS Secrets Manager Fallback - For production security
  3. Fail Fast with Clear Errors - No silent failures
  4. Centralized Logic - Single source of truth in src/secrets.py
  5. Type Safety - Pydantic models for validation in src/config.py
  6. Caching - Avoid repeated API calls
  7. Testability - Easy to mock and test

Architecture

Components

src/
├── secrets.py          # Centralized secrets management (NEW)
├── config.py           # Configuration using Pydantic (UPDATED)
└── [other modules]     # Use secrets.py for credentials

How It Works

# Priority order for fetching secrets:
1. Environment variable (e.g., NEO4J_URI)
2. AWS Secrets Manager (e.g., blog-data/neo4j/credentials)
3. Default value (if provided)
4. Raise error (if required=True)

Usage

Basic Usage

from src.secrets import get_secrets_manager

# Get secrets manager instance
sm = get_secrets_manager()

# Get a single secret
secret = sm.get_secret(
    env_var_name="NEO4J_URI",
    secret_name="blog-data/neo4j/credentials",
    required=True
)

print(f"Value: {secret.value}")
print(f"Source: {secret.source}")  # ENVIRONMENT or SECRETS_MANAGER

Convenience Functions

For common credentials, use the convenience functions:

from src.secrets import get_neo4j_credentials, get_cloudinary_credentials

# Get Neo4j credentials
neo4j_creds = get_neo4j_credentials()
# Returns: {"uri": "...", "username": "...", "password": "..."}

# Get Cloudinary credentials
cloudinary_creds = get_cloudinary_credentials()
# Returns: {"cloud_name": "...", "api_key": "...", "api_secret": "..."}

Using with Config

The src/config.py module uses secrets.py internally:

from src.config import get_config

config = get_config()

# Neo4j credentials automatically loaded
print(config.neo4j.uri)
print(config.neo4j.username)

# Cloudinary credentials automatically loaded
print(config.cloudinary.cloud_name)

Configuration

Local Development (.env.local)

Create a .env.local file in the project root:

# Neo4j
NEO4J_URI=neo4j+s://xxxxx.databases.neo4j.io
NEO4J_USER=neo4j
NEO4J_PASSWORD=your-password

# Cloudinary
CLOUDINARY_CLOUD_NAME=your-cloud-name
CLOUDINARY_API_KEY=your-api-key
CLOUDINARY_API_SECRET=your-api-secret

# AWS (optional for local development)
AWS_REGION=eu-west-2
AWS_PROFILE=default

Production (AWS Secrets Manager)

Secrets are stored in AWS Secrets Manager with the following structure:

Neo4j Credentials

Secret Name: blog-data/neo4j/credentials

{
  "uri": "neo4j+s://xxxxx.databases.neo4j.io",
  "username": "neo4j",
  "password": "your-password"
}

Cloudinary Credentials

Secret Name: blog-data/cloudinary/credentials

{
  "cloud_name": "your-cloud-name",
  "api_key": "your-api-key",
  "api_secret": "your-api-secret"
}

ECS Task Execution

In ECS, the task execution role automatically has permissions to access these secrets. No additional configuration needed.

API Reference

SecretsManager Class

class SecretsManager:
    def __init__(self, region: str = "eu-west-2", profile: Optional[str] = None)
    
    def get_secret(
        self,
        env_var_name: str,
        secret_name: Optional[str] = None,
        required: bool = True,
        default: Optional[str] = None,
    ) -> SecretValue
    
    def get_secret_dict(
        self,
        secret_name: str,
        env_var_mapping: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]

Helper Functions

def get_secrets_manager() -> SecretsManager
def get_neo4j_credentials() -> Dict[str, str]
def get_cloudinary_credentials() -> Dict[str, str]

Migration Guide

From Direct Environment Variables

Before:

import os

neo4j_uri = os.getenv("NEO4J_URI")
if not neo4j_uri:
    raise ValueError("NEO4J_URI not set")

After:

from src.secrets import get_secrets_manager

sm = get_secrets_manager()
neo4j_uri = sm.get_secret("NEO4J_URI", "blog-data/neo4j/credentials").value

From Config Module

Before:

from src.config import get_config

config = get_config()
neo4j_uri = config.neo4j.uri  # Still works!

After:

# No change needed! Config module uses secrets.py internally
from src.config import get_config

config = get_config()
neo4j_uri = config.neo4j.uri

Testing

Mocking Secrets

from unittest.mock import patch
from src.secrets import SecretValue, SecretSource

def test_with_mocked_secrets():
    with patch('src.secrets.get_secrets_manager') as mock_sm:
        mock_sm.return_value.get_secret.return_value = SecretValue(
            value="test-value",
            source=SecretSource.ENVIRONMENT
        )
        
        # Your test code here

Using Environment Variables in Tests

import os
import pytest

@pytest.fixture
def neo4j_env():
    os.environ["NEO4J_URI"] = "neo4j://localhost:7687"
    os.environ["NEO4J_USER"] = "neo4j"
    os.environ["NEO4J_PASSWORD"] = "test"
    yield
    del os.environ["NEO4J_URI"]
    del os.environ["NEO4J_USER"]
    del os.environ["NEO4J_PASSWORD"]

def test_with_env_vars(neo4j_env):
    from src.secrets import get_neo4j_credentials
    creds = get_neo4j_credentials()
    assert creds["uri"] == "neo4j://localhost:7687"

Troubleshooting

Secret Not Found

Error: ValueError: NEO4J_URI not found in environment or Secrets Manager

Solutions:

  1. Check .env.local file exists and contains the variable
  2. Verify AWS credentials are configured (aws configure)
  3. Check IAM permissions for Secrets Manager access
  4. Verify secret exists in AWS Secrets Manager console

Access Denied

Error: Access denied to secret: blog-data/neo4j/credentials

Solutions:

  1. Check IAM role/user has secretsmanager:GetSecretValue permission
  2. Verify the secret ARN in IAM policy matches the secret name
  3. Check if secret is in the correct AWS region

Wrong Region

Error: Secret not found but exists in AWS console

Solutions:

  1. Set AWS_REGION environment variable
  2. Pass region parameter to SecretsManager(region="eu-west-2")
  3. Check AWS CLI default region: aws configure get region

Security Best Practices

  1. Never commit .env.local - Already in .gitignore
  2. Use IAM roles in ECS - No hardcoded credentials
  3. Rotate secrets regularly - Use AWS Secrets Manager rotation
  4. Limit secret access - Use least-privilege IAM policies
  5. Audit secret access - Enable CloudTrail logging

See Also

Up next

S3 Setup

S3 bucket configuration for data storage.