Module 5: Secrets Management

5. Cloud KMS: AWS, GCP, Azure

Next →

Overview

In capsule 03 you met HashiCorp Vault as the enterprise reference for secrets management. Vault is powerful but requires infrastructure and expertise to operate. For most teams — startups, mid-size teams, and projects that already run on a cloud provider — managed secrets management services are the most pragmatic option: zero infrastructure to operate, native IAM integration, and predictable costs.

This capsule covers the three main ones: AWS Secrets Manager (boto3), GCP Secret Manager (google-cloud-secret-manager), and Azure Key Vault (azure-keyvault-secrets). You'll see working Python code for each, a detailed comparison, and a unified interface pattern that abstracts the provider — so your application doesn't depend on a specific vendor.

By the end you'll be able to choose the right KMS for your setup, implement access from Python, and have a clear migration path from .env.

AWS Secrets Manager

AWS Secrets Manager is the most widely adopted service for secrets management in the cloud. It offers encrypted storage, automatic rotation with Lambda, and access control via IAM.

Setup

pip install boto3

For local development, configure AWS credentials:

aws configure
# Or use environment variables:
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-east-1"

Operations with boto3

import boto3
import json
from datetime import datetime
from typing import Optional
from botocore.exceptions import ClientError


class AWSSecretsClient:
    """Client for AWS Secrets Manager with common operations."""

    def __init__(self, region_name: str = "us-east-1"):
        self.client = boto3.client("secretsmanager", region_name=region_name)
        self.region = region_name

    def create_secret(self, name: str, value: dict, description: str = "") -> dict:
        try:
            response = self.client.create_secret(
                Name=name,
                Description=description,
                SecretString=json.dumps(value),
                Tags=[
                    {"Key": "managed-by", "Value": "secrets-management-module"},
                    {"Key": "created-at", "Value": datetime.utcnow().isoformat()},
                ],
            )
            return {
                "status": "created",
                "arn": response["ARN"],
                "name": response["Name"],
                "version_id": response["VersionId"],
            }
        except ClientError as e:
            if e.response["Error"]["Code"] == "ResourceExistsException":
                return self.update_secret(name, value)
            raise

    def get_secret(self, name: str) -> Optional[dict]:
        try:
            response = self.client.get_secret_value(SecretId=name)
            secret_data = json.loads(response["SecretString"])
            return {
                "data": secret_data,
                "version_id": response["VersionId"],
                "created_date": str(response.get("CreatedDate", "")),
            }
        except ClientError as e:
            error_code = e.response["Error"]["Code"]
            if error_code in ("ResourceNotFoundException", "DecryptionFailureException"):
                return None
            raise

    def update_secret(self, name: str, value: dict) -> dict:
        response = self.client.update_secret(
            SecretId=name,
            SecretString=json.dumps(value),
        )
        return {
            "status": "updated",
            "arn": response["ARN"],
            "version_id": response["VersionId"],
        }

    def delete_secret(self, name: str, force: bool = False) -> dict:
        kwargs = {"SecretId": name}
        if force:
            kwargs["ForceDeleteWithoutRecovery"] = True
        else:
            kwargs["RecoveryWindowInDays"] = 7

        response = self.client.delete_secret(**kwargs)
        return {
            "status": "scheduled_deletion" if not force else "deleted",
            "name": response["Name"],
            "deletion_date": str(response.get("DeletionDate", "")),
        }

    def list_secrets(self, prefix: str = "") -> list[dict]:
        paginator = self.client.get_paginator("list_secrets")
        secrets = []
        for page in paginator.paginate():
            for secret in page["SecretList"]:
                if prefix and not secret["Name"].startswith(prefix):
                    continue
                secrets.append({
                    "name": secret["Name"],
                    "arn": secret["ARN"],
                    "last_changed": str(secret.get("LastChangedDate", "")),
                    "last_accessed": str(secret.get("LastAccessedDate", "")),
                })
        return secrets

    def configure_rotation(
        self, name: str, lambda_arn: str, days: int = 30
    ) -> dict:
        response = self.client.rotate_secret(
            SecretId=name,
            RotationLambdaARN=lambda_arn,
            RotationRules={"AutomaticallyAfterDays": days},
        )
        return {
            "status": "rotation_configured",
            "version_id": response["VersionId"],
            "rotation_days": days,
        }


# Usage example (requires AWS credentials configured):
# aws_client = AWSSecretsClient(region_name="us-east-1")
#
# aws_client.create_secret(
#     "prod/llm/openai",
#     {"api_key": "sk-proj-abc123", "org_id": "org-xyz"},
#     description="OpenAI API key for production"
# )
#
# secret = aws_client.get_secret("prod/llm/openai")
# print(f"API Key: {secret['data']['api_key']}")

AWS Secrets Manager cost

aws_pricing = {
    "per_secret_per_month": 0.40,
    "per_10k_api_calls": 0.05,
    "examples": [
        {"secrets": 5, "calls_per_month": 10000, "cost": 5 * 0.40 + 1 * 0.05},
        {"secrets": 20, "calls_per_month": 100000, "cost": 20 * 0.40 + 10 * 0.05},
        {"secrets": 50, "calls_per_month": 500000, "cost": 50 * 0.40 + 50 * 0.05},
    ],
}

print("AWS Secrets Manager Pricing:")
print(f"  ${aws_pricing['per_secret_per_month']}/secret/month")
print(f"  ${aws_pricing['per_10k_api_calls']}/10K API calls")
print("\nEstimated monthly costs:")
for ex in aws_pricing["examples"]:
    print(f"  {ex['secrets']} secrets, {ex['calls_per_month']:,} calls → ${ex['cost']:.2f}/month")
# Expected output:
# AWS Secrets Manager Pricing:
#   $0.4/secret/month
#   $0.05/10K API calls
#
# Estimated monthly costs:
#   5 secrets, 10,000 calls → $2.05/month
#   20 secrets, 100,000 calls → $8.50/month
#   50 secrets, 500,000 calls → $22.50/month

GCP Secret Manager

Google Cloud Secret Manager is GCP's native option with direct integration with IAM and Cloud Audit Logs.

Setup

pip install google-cloud-secret-manager

For local development:

export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-key.json"
# Or use gcloud auth:
gcloud auth application-default login

Operations with google-cloud-secret-manager

from typing import Optional
import json


class GCPSecretsClient:
    """Client for GCP Secret Manager with common operations."""

    def __init__(self, project_id: str):
        from google.cloud import secretmanager
        self.client = secretmanager.SecretManagerServiceClient()
        self.project_id = project_id
        self.parent = f"projects/{project_id}"

    def create_secret(self, secret_id: str, value: dict) -> dict:
        from google.cloud import secretmanager
        from google.api_core.exceptions import AlreadyExists

        try:
            secret = self.client.create_secret(
                request={
                    "parent": self.parent,
                    "secret_id": secret_id,
                    "secret": {
                        "replication": {"automatic": {}},
                        "labels": {"managed-by": "secrets-management-module"},
                    },
                }
            )
        except AlreadyExists:
            pass

        payload = json.dumps(value).encode("UTF-8")
        version = self.client.add_secret_version(
            request={
                "parent": f"{self.parent}/secrets/{secret_id}",
                "payload": {"data": payload},
            }
        )

        return {
            "status": "created",
            "name": version.name,
            "state": version.state.name,
        }

    def get_secret(self, secret_id: str, version: str = "latest") -> Optional[dict]:
        from google.api_core.exceptions import NotFound

        try:
            name = f"{self.parent}/secrets/{secret_id}/versions/{version}"
            response = self.client.access_secret_version(request={"name": name})
            secret_data = json.loads(response.payload.data.decode("UTF-8"))
            return {
                "data": secret_data,
                "version": response.name.split("/")[-1],
            }
        except NotFound:
            return None

    def delete_secret(self, secret_id: str) -> bool:
        from google.api_core.exceptions import NotFound

        try:
            name = f"{self.parent}/secrets/{secret_id}"
            self.client.delete_secret(request={"name": name})
            return True
        except NotFound:
            return False

    def list_secrets(self, prefix: str = "") -> list[dict]:
        secrets = []
        for secret in self.client.list_secrets(request={"parent": self.parent}):
            name = secret.name.split("/")[-1]
            if prefix and not name.startswith(prefix):
                continue
            secrets.append({
                "name": name,
                "full_name": secret.name,
                "created": str(secret.create_time),
                "labels": dict(secret.labels),
            })
        return secrets


# Usage example (requires GCP credentials configured):
# gcp_client = GCPSecretsClient(project_id="my-ai-project")
#
# gcp_client.create_secret("openai-api-key", {"api_key": "sk-proj-abc123"})
#
# secret = gcp_client.get_secret("openai-api-key")
# print(f"API Key: {secret['data']['api_key']}")

Azure Key Vault

Azure Key Vault is Azure's native option with integration with Azure Active Directory and RBAC.

Setup

pip install azure-keyvault-secrets azure-identity

Operations with azure-keyvault-secrets

import json
from typing import Optional


class AzureSecretsClient:
    """Client for Azure Key Vault with common operations."""

    def __init__(self, vault_url: str):
        from azure.identity import DefaultAzureCredential
        from azure.keyvault.secrets import SecretClient

        credential = DefaultAzureCredential()
        self.client = SecretClient(vault_url=vault_url, credential=credential)
        self.vault_url = vault_url

    def create_secret(self, name: str, value: dict, content_type: str = "application/json") -> dict:
        secret = self.client.set_secret(
            name=name,
            value=json.dumps(value),
            content_type=content_type,
            tags={"managed-by": "secrets-management-module"},
        )
        return {
            "status": "created",
            "name": secret.name,
            "version": secret.properties.version,
            "created_on": str(secret.properties.created_on),
        }

    def get_secret(self, name: str, version: str = "") -> Optional[dict]:
        from azure.core.exceptions import ResourceNotFoundError

        try:
            if version:
                secret = self.client.get_secret(name, version)
            else:
                secret = self.client.get_secret(name)

            secret_data = json.loads(secret.value)
            return {
                "data": secret_data,
                "version": secret.properties.version,
                "created_on": str(secret.properties.created_on),
            }
        except ResourceNotFoundError:
            return None

    def delete_secret(self, name: str) -> bool:
        from azure.core.exceptions import ResourceNotFoundError

        try:
            self.client.begin_delete_secret(name).wait()
            return True
        except ResourceNotFoundError:
            return False

    def list_secrets(self, prefix: str = "") -> list[dict]:
        secrets = []
        for prop in self.client.list_properties_of_secrets():
            if prefix and not prop.name.startswith(prefix):
                continue
            secrets.append({
                "name": prop.name,
                "enabled": prop.enabled,
                "created_on": str(prop.created_on),
                "updated_on": str(prop.updated_on),
                "tags": dict(prop.tags) if prop.tags else {},
            })
        return secrets


# Usage example (requires Azure credentials configured):
# azure_client = AzureSecretsClient(
#     vault_url="https://my-vault.vault.azure.net"
# )
#
# azure_client.create_secret("openai-api-key", {"api_key": "sk-proj-abc123"})
#
# secret = azure_client.get_secret("openai-api-key")
# print(f"API Key: {secret['data']['api_key']}")

Detailed comparison: AWS vs GCP vs Azure

comparison = {
    "headers": ["Feature", "AWS Secrets Manager", "GCP Secret Manager", "Azure Key Vault"],
    "rows": [
        ["Encryption", "AES-256 (AWS KMS)", "AES-256 (Google KMS)", "AES-256 (Azure KMS)"],
        ["Versioning", "Staging labels", "Numbered versions", "Named versions"],
        ["Rotation", "Lambda-based auto", "Cloud Functions (manual setup)", "Managed identity rotation"],
        ["Audit", "CloudTrail", "Cloud Audit Logs", "Azure Monitor"],
        ["Access Control", "IAM policies", "IAM + conditions", "Azure RBAC"],
        ["Pricing (per secret/mo)", "$0.40", "$0.06", "$0.03/10K ops"],
        ["API Calls pricing", "$0.05/10K", "$0.03/10K", "Included in ops"],
        ["Max secret size", "64 KB", "64 KB", "25 KB (secret), 128 KB (cert)"],
        ["Cross-region", "Replication available", "Automatic replication", "Geo-replication"],
        ["Python SDK", "boto3", "google-cloud-secret-manager", "azure-keyvault-secrets"],
        ["Free tier", "No (30-day trial)", "6 active versions free", "10K ops free/month"],
        ["Best for", "AWS-native apps", "GCP-native, cost-sensitive", "Azure/Microsoft shops"],
    ],
}

print(f"| {' | '.join(comparison['headers'])} |")
print(f"| {' | '.join(['---'] * len(comparison['headers']))} |")
for row in comparison["rows"]:
    print(f"| {' | '.join(row)} |")

Unified Interface Pattern

The most important pattern in this capsule: a unified interface that abstracts the secrets provider. Your application shouldn't know whether it uses AWS, GCP, Azure, or Vault:

import os
import json
import time
import logging
from abc import ABC, abstractmethod
from typing import Optional
from dataclasses import dataclass, field

logger = logging.getLogger("secrets_client")


@dataclass
class SecretResult:
    key: str
    value: Optional[str]
    found: bool
    provider: str
    version: str = ""
    cached: bool = False
    access_time_ms: float = 0.0


class SecretsProvider(ABC):
    """Unified interface for any secrets provider."""

    @abstractmethod
    def get(self, key: str) -> SecretResult:
        pass

    @abstractmethod
    def set(self, key: str, value: dict) -> bool:
        pass

    @abstractmethod
    def delete(self, key: str) -> bool:
        pass

    @abstractmethod
    def list(self, prefix: str = "") -> list[str]:
        pass

    @property
    @abstractmethod
    def provider_name(self) -> str:
        pass


class EnvProvider(SecretsProvider):
    """Provider for local development using environment variables."""

    def get(self, key: str) -> SecretResult:
        start = time.perf_counter()
        value = os.environ.get(key)
        elapsed = (time.perf_counter() - start) * 1000
        return SecretResult(
            key=key, value=value, found=value is not None,
            provider="env", access_time_ms=elapsed,
        )

    def set(self, key: str, value: dict) -> bool:
        os.environ[key] = json.dumps(value) if isinstance(value, dict) else str(value)
        return True

    def delete(self, key: str) -> bool:
        if key in os.environ:
            del os.environ[key]
            return True
        return False

    def list(self, prefix: str = "") -> list[str]:
        return [k for k in os.environ if k.startswith(prefix)]

    @property
    def provider_name(self) -> str:
        return "environment"


class AWSProvider(SecretsProvider):
    """Provider for AWS Secrets Manager."""

    def __init__(self, region: str = "us-east-1"):
        import boto3
        self.client = boto3.client("secretsmanager", region_name=region)

    def get(self, key: str) -> SecretResult:
        start = time.perf_counter()
        try:
            response = self.client.get_secret_value(SecretId=key)
            data = json.loads(response["SecretString"])
            elapsed = (time.perf_counter() - start) * 1000
            return SecretResult(
                key=key,
                value=data.get("value", json.dumps(data)),
                found=True,
                provider="aws",
                version=response.get("VersionId", ""),
                access_time_ms=elapsed,
            )
        except Exception:
            elapsed = (time.perf_counter() - start) * 1000
            return SecretResult(
                key=key, value=None, found=False,
                provider="aws", access_time_ms=elapsed,
            )

    def set(self, key: str, value: dict) -> bool:
        try:
            self.client.create_secret(Name=key, SecretString=json.dumps(value))
            return True
        except self.client.exceptions.ResourceExistsException:
            self.client.update_secret(SecretId=key, SecretString=json.dumps(value))
            return True
        except Exception:
            return False

    def delete(self, key: str) -> bool:
        try:
            self.client.delete_secret(SecretId=key, ForceDeleteWithoutRecovery=True)
            return True
        except Exception:
            return False

    def list(self, prefix: str = "") -> list[str]:
        secrets = []
        paginator = self.client.get_paginator("list_secrets")
        for page in paginator.paginate():
            for s in page["SecretList"]:
                if not prefix or s["Name"].startswith(prefix):
                    secrets.append(s["Name"])
        return secrets

    @property
    def provider_name(self) -> str:
        return "aws_secrets_manager"


class CachedProvider(SecretsProvider):
    """Wrapper that adds caching to any provider."""

    def __init__(self, provider: SecretsProvider, ttl_seconds: int = 300):
        self._provider = provider
        self._ttl = ttl_seconds
        self._cache: dict[str, tuple[SecretResult, float]] = {}

    def get(self, key: str) -> SecretResult:
        if key in self._cache:
            result, cached_at = self._cache[key]
            if time.time() - cached_at < self._ttl:
                result.cached = True
                return result
            del self._cache[key]

        result = self._provider.get(key)
        if result.found:
            self._cache[key] = (result, time.time())
        return result

    def set(self, key: str, value: dict) -> bool:
        success = self._provider.set(key, value)
        if success:
            self._cache.pop(key, None)
        return success

    def delete(self, key: str) -> bool:
        success = self._provider.delete(key)
        self._cache.pop(key, None)
        return success

    def list(self, prefix: str = "") -> list[str]:
        return self._provider.list(prefix)

    @property
    def provider_name(self) -> str:
        return f"cached({self._provider.provider_name})"

    def clear_cache(self):
        self._cache.clear()


class FallbackProvider(SecretsProvider):
    """Tries multiple providers in order — for gradual migration."""

    def __init__(self, providers: list[SecretsProvider]):
        self._providers = providers

    def get(self, key: str) -> SecretResult:
        for provider in self._providers:
            result = provider.get(key)
            if result.found:
                return result
        return SecretResult(
            key=key, value=None, found=False,
            provider="fallback_exhausted",
        )

    def set(self, key: str, value: dict) -> bool:
        return self._providers[0].set(key, value)

    def delete(self, key: str) -> bool:
        return self._providers[0].delete(key)

    def list(self, prefix: str = "") -> list[str]:
        all_keys = set()
        for provider in self._providers:
            all_keys.update(provider.list(prefix))
        return sorted(all_keys)

    @property
    def provider_name(self) -> str:
        names = [p.provider_name for p in self._providers]
        return f"fallback({', '.join(names)})"


def create_secrets_provider(
    provider_type: str = "env",
    cache_ttl: int = 300,
    **kwargs,
) -> SecretsProvider:
    """Factory function to create the appropriate provider."""
    providers = {
        "env": lambda: EnvProvider(),
        "aws": lambda: AWSProvider(region=kwargs.get("region", "us-east-1")),
    }

    if provider_type not in providers:
        raise ValueError(f"Unknown provider: {provider_type}. Options: {list(providers.keys())}")

    provider = providers[provider_type]()

    if cache_ttl > 0:
        provider = CachedProvider(provider, ttl_seconds=cache_ttl)

    return provider


os.environ["OPENAI_API_KEY"] = "sk-proj-from-env"

provider = create_secrets_provider("env", cache_ttl=60)
result = provider.get("OPENAI_API_KEY")
print(f"Provider: {provider.provider_name}")
print(f"Key: {result.key}, Value: {result.value[:15]}..., Found: {result.found}")

result2 = provider.get("OPENAI_API_KEY")
print(f"Cached: {result2.cached}")
# Expected output:
# Provider: cached(environment)
# Key: OPENAI_API_KEY, Value: sk-proj-from-en..., Found: True
# Cached: True

Migrating from .env to Cloud KMS

Step-by-step migration plan

migration_plan = {
    "phase_1_preparation": {
        "duration": "1-2 days",
        "steps": [
            "Inventory all secrets in .env",
            "Classify by risk (CRITICAL, HIGH, MEDIUM, LOW)",
            "Choose a provider (AWS, GCP, Azure) based on your infrastructure",
            "Create an account/project in the provider",
            "Configure IAM roles/policies",
        ],
    },
    "phase_2_abstraction": {
        "duration": "1 day",
        "steps": [
            "Implement the SecretsProvider interface in your code",
            "Replace os.getenv() with secrets_provider.get()",
            "Use FallbackProvider: cloud → env (for backward compatibility)",
            "Verify the app works the same with the new abstraction layer",
        ],
    },
    "phase_3_migration": {
        "duration": "1-2 days",
        "steps": [
            "Create secrets in the cloud provider (start with CRITICAL)",
            "Deploy with FallbackProvider: cloud first, env as backup",
            "Monitor: verify secrets are read from the cloud",
            "Migrate HIGH and MEDIUM secrets",
            "Verify the audit trail in the cloud provider",
        ],
    },
    "phase_4_cleanup": {
        "duration": "1 day",
        "steps": [
            "Remove secrets from .env (leave only LOW/non-sensitive config)",
            "Remove FallbackProvider → use only the cloud provider",
            "Configure rotation schedules",
            "Document the new secrets architecture",
            "Configure rotation alerts",
        ],
    },
}

print("Migration Plan: .env → Cloud KMS")
for phase, info in migration_plan.items():
    print(f"\n📋 {phase.replace('_', ' ').title()} ({info['duration']})")
    for i, step in enumerate(info["steps"], 1):
        print(f"   {i}. {step}")

Migration script

import os
import json
from typing import Optional


def parse_env_file(filepath: str) -> dict[str, str]:
    """Parses a .env file and returns a dictionary of secrets."""
    secrets = {}
    try:
        with open(filepath, "r") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                if "=" in line:
                    key, value = line.split("=", 1)
                    secrets[key.strip()] = value.strip().strip('"').strip("'")
    except FileNotFoundError:
        pass
    return secrets


def classify_secrets(secrets: dict[str, str]) -> dict[str, list[str]]:
    """Classifies secrets by risk level based on the name."""
    critical_patterns = ["api_key", "secret_key", "password", "token"]
    high_patterns = ["database", "redis", "url"]
    medium_patterns = ["webhook", "smtp", "sentry"]

    classified = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": []}

    for key in secrets:
        key_lower = key.lower()
        if any(p in key_lower for p in critical_patterns):
            classified["CRITICAL"].append(key)
        elif any(p in key_lower for p in high_patterns):
            classified["HIGH"].append(key)
        elif any(p in key_lower for p in medium_patterns):
            classified["MEDIUM"].append(key)
        else:
            classified["LOW"].append(key)

    return classified


env_content = """
OPENAI_API_KEY=sk-proj-abc123
ANTHROPIC_API_KEY=sk-ant-xyz789
DATABASE_URL=postgresql://user:pass@host:5432/db
DATABASE_PASSWORD=super-secret
REDIS_URL=redis://:pass@host:6379
SENTRY_DSN=https://key@sentry.io/123
APP_ENV=production
LOG_LEVEL=info
PORT=8000
""".strip()

with open(".env.example", "w") as f:
    f.write(env_content)

secrets = parse_env_file(".env.example")
classified = classify_secrets(secrets)

print("Secret Classification:")
for level, keys in classified.items():
    if keys:
        print(f"\n  {level}:")
        for key in keys:
            print(f"    - {key}")

print(f"\nMigration priority:")
print(f"  1. CRITICAL ({len(classified['CRITICAL'])} secrets) → Migrate first")
print(f"  2. HIGH ({len(classified['HIGH'])} secrets) → Migrate second")
print(f"  3. MEDIUM ({len(classified['MEDIUM'])} secrets) → Migrate third")
print(f"  4. LOW ({len(classified['LOW'])} secrets) → Keep in env vars")

os.remove(".env.example")
# Expected output:
# Secret Classification:
#   CRITICAL:
#     - OPENAI_API_KEY
#     - ANTHROPIC_API_KEY
#     - DATABASE_PASSWORD
#   HIGH:
#     - DATABASE_URL
#     - REDIS_URL
#   MEDIUM:
#     - SENTRY_DSN
#   LOW:
#     - APP_ENV
#     - LOG_LEVEL
#     - PORT

Choosing the right KMS

decision_tree = {
    "question_1": {
        "q": "Which cloud provider do you run your application on?",
        "aws": "→ AWS Secrets Manager (native integration, IAM)",
        "gcp": "→ GCP Secret Manager (cheaper, Cloud Audit Logs)",
        "azure": "→ Azure Key Vault (Azure AD, native RBAC)",
        "multi_cloud": "→ HashiCorp Vault (provider-agnostic)",
        "on_premises": "→ HashiCorp Vault (self-hosted)",
    },
    "question_2": {
        "q": "How many secrets do you manage?",
        "less_than_10": "Cloud KMS is more than enough (~$2-4/month)",
        "10_to_50": "Cloud KMS is still practical (~$5-20/month)",
        "50_plus": "Evaluate Vault if you need dynamic secrets",
    },
    "question_3": {
        "q": "Do you need dynamic secrets (temporary credentials)?",
        "yes": "→ HashiCorp Vault (the only one that supports this natively)",
        "no": "→ Cloud KMS with automatic rotation is enough",
    },
}

print("Decision Tree: Which KMS to use?")
for name, tree in decision_tree.items():
    print(f"\n{tree['q']}")
    for option, recommendation in tree.items():
        if option != "q":
            print(f"  {option}: {recommendation}")

Troubleshooting

"boto3 throws NoCredentialsError"

Verify that you have AWS credentials configured: aws configure or the environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY. For local development, use AWS SSO or profiles.

"GCP throws PermissionDenied"

The service account needs the roles/secretmanager.secretAccessor role to read secrets and roles/secretmanager.admin to create/delete. Check in IAM & Admin.

"Azure throws AuthenticationError"

DefaultAzureCredential tries multiple authentication methods in order. For local development, use az login. For production, use Managed Identity.

"The migration broke my app in production"

Use FallbackProvider during migration: the cloud provider as the first option and env vars as backup. That way, if the cloud provider fails or a secret wasn't migrated, the app keeps working with the value from .env.

"I can't test locally without cloud credentials"

Use EnvProvider for local development and AWSProvider/GCPProvider for staging/production. The unified interface makes the switch transparent.

Exercises

Exercise 1: Implement a GCPProvider

Following the AWSProvider pattern, implement a GCPProvider that implements the SecretsProvider interface:

See solution 
class GCPProvider(SecretsProvider):
    def __init__(self, project_id: str):
        from google.cloud import secretmanager
        self.client = secretmanager.SecretManagerServiceClient()
        self.project_id = project_id
        self.parent = f"projects/{project_id}"

    def get(self, key: str) -> SecretResult:
        start = time.perf_counter()
        try:
            name = f"{self.parent}/secrets/{key}/versions/latest"
            response = self.client.access_secret_version(request={"name": name})
            value = response.payload.data.decode("UTF-8")
            elapsed = (time.perf_counter() - start) * 1000
            return SecretResult(key=key, value=value, found=True, provider="gcp", access_time_ms=elapsed)
        except Exception:
            elapsed = (time.perf_counter() - start) * 1000
            return SecretResult(key=key, value=None, found=False, provider="gcp", access_time_ms=elapsed)

    def set(self, key: str, value: dict) -> bool:
        try:
            from google.api_core.exceptions import AlreadyExists
            try:
                self.client.create_secret(request={
                    "parent": self.parent,
                    "secret_id": key,
                    "secret": {"replication": {"automatic": {}}},
                })
            except AlreadyExists:
                pass
            self.client.add_secret_version(request={
                "parent": f"{self.parent}/secrets/{key}",
                "payload": {"data": json.dumps(value).encode("UTF-8")},
            })
            return True
        except Exception:
            return False

    def delete(self, key: str) -> bool:
        try:
            self.client.delete_secret(request={"name": f"{self.parent}/secrets/{key}"})
            return True
        except Exception:
            return False

    def list(self, prefix: str = "") -> list[str]:
        secrets = []
        for s in self.client.list_secrets(request={"parent": self.parent}):
            name = s.name.split("/")[-1]
            if not prefix or name.startswith(prefix):
                secrets.append(name)
        return secrets

    @property
    def provider_name(self) -> str:
        return "gcp_secret_manager"

Exercise 2: Migrate 5 secrets from .env to a provider

Write a script that reads a .env, classifies the secrets, and migrates them to the provider you choose:

See solution 
def migrate_env_to_provider(env_path: str, provider: SecretsProvider) -> dict:
    secrets = parse_env_file(env_path)
    classified = classify_secrets(secrets)
    results = {"migrated": [], "skipped": [], "failed": []}

    for level in ["CRITICAL", "HIGH", "MEDIUM"]:
        for key in classified.get(level, []):
            try:
                success = provider.set(key, {"value": secrets[key]})
                if success:
                    results["migrated"].append({"key": key, "level": level})
                else:
                    results["failed"].append({"key": key, "level": level})
            except Exception as e:
                results["failed"].append({"key": key, "error": str(e)})

    for key in classified.get("LOW", []):
        results["skipped"].append({"key": key, "reason": "LOW risk — keep in env"})

    return results

provider = EnvProvider()
# In production you'd use: provider = AWSProvider() or GCPProvider()

Exercise 3: Implement a health check for the secrets provider

Create an endpoint that verifies connectivity with the secrets provider:

See solution 
def health_check(provider: SecretsProvider) -> dict:
    start = time.perf_counter()
    try:
        result = provider.get("__health_check__")
        elapsed = (time.perf_counter() - start) * 1000
        return {
            "provider": provider.provider_name,
            "status": "healthy",
            "response_time_ms": round(elapsed, 2),
            "note": "Secret not found is expected for health check",
        }
    except Exception as e:
        elapsed = (time.perf_counter() - start) * 1000
        return {
            "provider": provider.provider_name,
            "status": "unhealthy",
            "response_time_ms": round(elapsed, 2),
            "error": str(e),
        }

check = health_check(EnvProvider())
print(json.dumps(check, indent=2))

Exercise 4: Compare latency between providers

Write a benchmark that compares read latency between EnvProvider and CachedProvider:

See solution 
def benchmark_provider(provider: SecretsProvider, key: str, iterations: int = 100) -> dict:
    times = []
    for _ in range(iterations):
        start = time.perf_counter()
        provider.get(key)
        elapsed = (time.perf_counter() - start) * 1000
        times.append(elapsed)

    return {
        "provider": provider.provider_name,
        "iterations": iterations,
        "avg_ms": round(sum(times) / len(times), 4),
        "min_ms": round(min(times), 4),
        "max_ms": round(max(times), 4),
        "p99_ms": round(sorted(times)[int(len(times) * 0.99)], 4),
    }

os.environ["BENCH_KEY"] = "test-value"
env = EnvProvider()
cached = CachedProvider(env, ttl_seconds=60)

cached.get("BENCH_KEY")

print("Benchmark Results:")
for result in [benchmark_provider(env, "BENCH_KEY"), benchmark_provider(cached, "BENCH_KEY")]:
    print(f"  {result['provider']}: avg={result['avg_ms']}ms, p99={result['p99_ms']}ms")

Summary

  • The three main cloud KMS (AWS Secrets Manager, GCP Secret Manager, Azure Key Vault) offer encrypted storage, versioning, and audit — with no infrastructure to operate
  • AWS Secrets Manager is the most mature with automatic rotation via Lambda and the widest adoption
  • GCP Secret Manager is the most affordable and has direct integration with Cloud Audit Logs
  • Azure Key Vault integrates natively with Azure AD and RBAC for Microsoft organizations
  • The unified interface pattern (SecretsProvider) abstracts the provider so your app doesn't depend on a specific vendor
  • CachedProvider adds caching with a TTL to reduce latency and calls to the cloud service
  • FallbackProvider enables gradual migration: cloud as primary, env as backup
  • The migration from .env to cloud KMS is incremental: classify → abstract → migrate → clean up
  • Monthly cost for most AI projects: $2-20/month — significantly less than the cost of a leaked key

Next capsule: In capsule 06 you'll learn about token lifecycle management and audit trails — how to manage the complete lifecycle of your secrets (creation, distribution, rotation, revocation) and how to record every access for compliance and incident detection.

Resources

  1. AWS Secrets Manager Documentation — Official documentation for AWS's secrets service
  2. GCP Secret Manager Documentation — Official documentation for Google Cloud's secrets service
  3. Azure Key Vault Documentation — Official documentation for Azure's secrets service
  4. boto3 Secrets Manager Reference — Python SDK reference for AWS
  5. google-cloud-secret-manager PyPI — Python package for GCP Secret Manager
  6. azure-keyvault-secrets PyPI — Python package for Azure Key Vault
  7. AWS Secrets Manager Pricing — AWS pricing details for cost planning
  8. Multi-Cloud Secrets Management (CNCF) — CNCF guide for managing secrets across multi-cloud environments

Created: March 2026 Version: 1.0

← Previous4. API Key Rotation StrategiesNext →6. Token Lifecycle and Audit Trails