Secrets Management & Key Rotation

# vault_client.py
# Requires: vault server -dev (for testing)
import hvac
import os

VAULT_URL = os.getenv("VAULT_ADDR", "http://localhost:8200")
VAULT_TOKEN = os.getenv("VAULT_TOKEN", "dev-token")   # Use AppRole/AWS auth in production

def get_vault_client() -> hvac.Client:
    """Create an authenticated Vault client."""
    client = hvac.Client(url=VAULT_URL, token=VAULT_TOKEN)
    assert client.is_authenticated(), "Vault authentication failed"
    return client

client = get_vault_client()

# ── Write secrets to KV v2 store ──────────────────────────────
client.secrets.kv.v2.create_or_update_secret(
    path="myapp/database",
    secret={
        "url": "postgresql://user:pass@db:5432/mydb",
        "pool_size": "10",
    },
)
print("✅ Secret written to vault: myapp/database")

# ── Read secrets ─────────────────────────────────────────────
secret = client.secrets.kv.v2.read_secret_version(path="myapp/database")
data = secret["data"]["data"]
print(f"Database URL: {data['url'][:30]}...")   # Never log full secret

# ── Dynamic secrets, Vault generates DB credentials ─────────
# (Requires database secrets engine configured)
# lease = client.secrets.database.generate_credentials(name="myapp-role")
# temp_username = lease["data"]["username"]   # Expires automatically!
# temp_password = lease["data"]["password"]

# ── Rotate a secret ──────────────────────────────────────────
client.secrets.kv.v2.create_or_update_secret(
    path="myapp/database",
    secret={"url": "postgresql://newuser:newpass@db:5432/mydb", "pool_size": "10"},
)

# KV v2 keeps history, you can access previous versions
prev = client.secrets.kv.v2.read_secret_version(path="myapp/database", version=1)
print(f"Previous version retained: {prev['data']['metadata']['version']}")

# aws_secrets.py
import boto3
import json
from functools import lru_cache
from datetime import datetime, timezone, timedelta

# ── Retrieve a secret ─────────────────────────────────────────
def get_secret(secret_name: str, region_name: str = "us-east-1") -> dict:
    """Retrieve a secret from AWS Secrets Manager."""
    client = boto3.client("secretsmanager", region_name=region_name)
    response = client.get_secret_value(SecretId=secret_name)

    if "SecretString" in response:
        return json.loads(response["SecretString"])
    else:
        # Binary secret
        return {"binary": response["SecretBinary"]}

# ── Cache with TTL, don't call AWS on every request ─────────
_secret_cache: dict[str, tuple[dict, datetime]] = {}
CACHE_TTL = timedelta(minutes=5)

def get_secret_cached(secret_name: str) -> dict:
    """Get secret with 5-minute cache to reduce AWS API calls."""
    cached = _secret_cache.get(secret_name)
    if cached:
        value, fetched_at = cached
        if datetime.now(timezone.utc) - fetched_at < CACHE_TTL:
            return value

    value = get_secret(secret_name)
    _secret_cache[secret_name] = (value, datetime.now(timezone.utc))
    return value

# ── Key rotation pattern ──────────────────────────────────────
def rotate_database_secret(secret_name: str, new_credentials: dict) -> None:
    """
    Rotate a database secret with zero downtime.
    Steps:
    1. Create new credentials in the database
    2. Update the secret with new + old credentials (AWSPENDING)
    3. Test new credentials
    4. Finalize rotation (AWSCURRENT = new)
    """
    client = boto3.client("secretsmanager", region_name="us-east-1")

    # AWS Secrets Manager can automate this with Lambda rotation functions
    client.rotate_secret(
        SecretId=secret_name,
        RotationRules={"AutomaticallyAfterDays": 30},
        RotateImmediately=True,
    )

print("AWS Secrets Manager patterns:")
print("  • get_secret(), retrieve and parse JSON secret")
print("  • Cache with TTL, avoid API throttling (10,000 req/month free tier)")
print("  • Automatic rotation, Lambda function updates DB credentials")
print("  • IAM roles, EC2/Lambda instances access secrets via role, no credentials needed")