Módulo 5: Secrets Management

5. Cloud KMS: AWS, GCP, Azure

Siguiente →

Descripción

En la cápsula 03 conociste HashiCorp Vault como la referencia enterprise para secrets management. Vault es poderoso pero requiere infraestructura y expertise para operar. Para la mayoría de equipos — startups, equipos medianos, y proyectos que ya corren en un cloud provider — los servicios managed de secrets management son la opción más pragmática: cero infraestructura que operar, integración nativa con IAM, y costos predecibles.

Esta cápsula cubre los tres principales: AWS Secrets Manager (boto3), GCP Secret Manager (google-cloud-secret-manager), y Azure Key Vault (azure-keyvault-secrets). Vas a ver código Python funcional para cada uno, una comparación detallada, y un patrón de interface unificada que abstrae el proveedor — para que tu aplicación no dependa de un vendor específico.

Al terminar vas a poder elegir el KMS correcto para tu setup, implementar el acceso desde Python, y tener un path claro de migración desde .env.

AWS Secrets Manager

AWS Secrets Manager es el servicio más adoptado para secrets management en la nube. Ofrece almacenamiento encriptado, rotación automática con Lambda, y control de acceso vía IAM.

Setup

pip install boto3

Para desarrollo local, configura credenciales de AWS:

aws configure
# O usa variables de entorno:
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-east-1"

Operaciones con boto3

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


class AWSSecretsClient:
    """Cliente para AWS Secrets Manager con operaciones comunes."""

    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,
        }


# Ejemplo de uso (requiere credenciales AWS configuradas):
# 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']}")

Costo de AWS Secrets Manager

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")
# Output esperado:
# 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 es la opción nativa de GCP con integración directa con IAM y Cloud Audit Logs.

Setup

pip install google-cloud-secret-manager

Para desarrollo local:

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

Operaciones con google-cloud-secret-manager

from typing import Optional
import json


class GCPSecretsClient:
    """Cliente para GCP Secret Manager con operaciones comunes."""

    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


# Ejemplo de uso (requiere credenciales GCP configuradas):
# 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 es la opción nativa de Azure con integración con Azure Active Directory y RBAC.

Setup

pip install azure-keyvault-secrets azure-identity

Operaciones con azure-keyvault-secrets

import json
from typing import Optional


class AzureSecretsClient:
    """Cliente para Azure Key Vault con operaciones comunes."""

    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


# Ejemplo de uso (requiere credenciales Azure configuradas):
# 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']}")

Comparación detallada: 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)} |")

Patrón de Interface Unificada

El patrón más importante de esta cápsula: una interface unificada que abstrae el proveedor de secrets. Tu aplicación no debería saber si usa AWS, GCP, Azure, o 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):
    """Interface unificada para cualquier 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 para desarrollo local usando variables de entorno."""

    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 para 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 que agrega caching a cualquier 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):
    """Intenta múltiples providers en orden — para migración gradual."""

    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 para crear el provider apropiado."""
    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}")
# Output esperado:
# Provider: cached(environment)
# Key: OPENAI_API_KEY, Value: sk-proj-from-en..., Found: True
# Cached: True

Migración de .env a Cloud KMS

Plan de migración paso a paso

migration_plan = {
    "phase_1_preparation": {
        "duration": "1-2 days",
        "steps": [
            "Inventariar todos los secrets en .env",
            "Clasificar por riesgo (CRITICAL, HIGH, MEDIUM, LOW)",
            "Elegir provider (AWS, GCP, Azure) según tu infraestructura",
            "Crear cuenta/proyecto en el provider",
            "Configurar IAM roles/policies",
        ],
    },
    "phase_2_abstraction": {
        "duration": "1 day",
        "steps": [
            "Implementar SecretsProvider interface en tu código",
            "Reemplazar os.getenv() con secrets_provider.get()",
            "Usar FallbackProvider: cloud → env (para backward compatibility)",
            "Verificar que la app funciona igual con el nuevo abstraction layer",
        ],
    },
    "phase_3_migration": {
        "duration": "1-2 days",
        "steps": [
            "Crear secrets en el cloud provider (empezar con CRITICAL)",
            "Deploy con FallbackProvider: cloud primero, env como backup",
            "Monitorear: verificar que secrets se leen del cloud",
            "Migrar HIGH y MEDIUM secrets",
            "Verificar audit trail en el cloud provider",
        ],
    },
    "phase_4_cleanup": {
        "duration": "1 day",
        "steps": [
            "Remover secrets de .env (dejar solo LOW/non-sensitive config)",
            "Remover FallbackProvider → usar solo cloud provider",
            "Configurar rotation schedules",
            "Documentar la nueva arquitectura de secrets",
            "Configurar alertas de rotación",
        ],
    },
}

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}")

Script de migración

import os
import json
from typing import Optional


def parse_env_file(filepath: str) -> dict[str, str]:
    """Parsea un archivo .env y retorna un diccionario de 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]]:
    """Clasifica secrets por nivel de riesgo basado en el nombre."""
    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")
# Output esperado:
# 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": "¿En qué cloud provider corres tu aplicación?",
        "aws": "→ AWS Secrets Manager (integración nativa, IAM)",
        "gcp": "→ GCP Secret Manager (más barato, Cloud Audit Logs)",
        "azure": "→ Azure Key Vault (Azure AD, RBAC nativo)",
        "multi_cloud": "→ HashiCorp Vault (provider-agnostic)",
        "on_premises": "→ HashiCorp Vault (self-hosted)",
    },
    "question_2": {
        "q": "¿Cuántos secrets gestionas?",
        "less_than_10": "Cloud KMS es más que suficiente (~$2-4/mes)",
        "10_to_50": "Cloud KMS sigue siendo práctico (~$5-20/mes)",
        "50_plus": "Evalúa Vault si necesitas dynamic secrets",
    },
    "question_3": {
        "q": "¿Necesitas dynamic secrets (credenciales temporales)?",
        "yes": "→ HashiCorp Vault (único que soporta esto nativamente)",
        "no": "→ Cloud KMS con rotation automática es suficiente",
    },
}

print("Decision Tree: ¿Qué KMS usar?")
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 da NoCredentialsError"

Verifica que tienes credenciales AWS configuradas: aws configure o variables de entorno AWS_ACCESS_KEY_ID y AWS_SECRET_ACCESS_KEY. Para desarrollo local, usa AWS SSO o profiles.

"GCP da PermissionDenied"

El service account necesita el rol roles/secretmanager.secretAccessor para leer secrets y roles/secretmanager.admin para crear/eliminar. Verifica en IAM & Admin.

"Azure da AuthenticationError"

DefaultAzureCredential intenta múltiples métodos de autenticación en orden. Para desarrollo local, usa az login. Para producción, usa Managed Identity.

"La migración rompió mi app en producción"

Usa FallbackProvider durante la migración: el cloud provider como primera opción y env vars como backup. Así, si el cloud provider falla o un secret no se migró, la app sigue funcionando con el valor de .env.

"No puedo testear localmente sin credenciales cloud"

Usa EnvProvider para desarrollo local y AWSProvider/GCPProvider para staging/producción. La interface unificada hace que el switch sea transparente.

Ejercicios

Ejercicio 1: Implementa un GCPProvider

Siguiendo el patrón de AWSProvider, implementa un GCPProvider que implemente la interface SecretsProvider:

Ver solución 
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"

Ejercicio 2: Migra 5 secrets de .env a un provider

Escribe un script que lea un .env, clasifique los secrets, y los migre al provider que elijas:

Ver solución 
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()
# En producción usarías: provider = AWSProvider() o GCPProvider()

Ejercicio 3: Implementa un health check para el secrets provider

Crea un endpoint que verifique la conectividad con el secrets provider:

Ver solución 
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))

Ejercicio 4: Compara latencia entre providers

Escribe un benchmark que compare la latencia de lectura entre EnvProvider y CachedProvider:

Ver solución 
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")

Resumen

  • Los tres principales cloud KMS (AWS Secrets Manager, GCP Secret Manager, Azure Key Vault) ofrecen almacenamiento encriptado, versioning, y audit — sin infraestructura que operar
  • AWS Secrets Manager es el más maduro con rotation automática vía Lambda y la mayor adopción
  • GCP Secret Manager es el más económico y tiene integración directa con Cloud Audit Logs
  • Azure Key Vault se integra nativamente con Azure AD y RBAC para organizaciones Microsoft
  • El patrón de interface unificada (SecretsProvider) abstrae el proveedor para que tu app no dependa de un vendor específico
  • CachedProvider agrega caching con TTL para reducir latencia y llamadas al servicio cloud
  • FallbackProvider permite migración gradual: cloud como primario, env como backup
  • La migración de .env a cloud KMS es incremental: clasificar → abstraer → migrar → limpiar
  • Costo mensual para la mayoría de proyectos AI: $2-20/mes — significativamente menor que el costo de una key filtrada

Próxima cápsula: En la cápsula 06 vas a aprender sobre token lifecycle management y audit trails — cómo gestionar el ciclo de vida completo de tus secrets (creación, distribución, rotación, revocación) y cómo registrar cada acceso para compliance y detección de incidentes.

Recursos

  1. AWS Secrets Manager Documentation — Documentación oficial del servicio de secrets de AWS
  2. GCP Secret Manager Documentation — Documentación oficial del servicio de secrets de Google Cloud
  3. Azure Key Vault Documentation — Documentación oficial del servicio de secrets de Azure
  4. boto3 Secrets Manager Reference — Referencia del SDK Python para AWS
  5. google-cloud-secret-manager PyPI — Paquete Python para GCP Secret Manager
  6. azure-keyvault-secrets PyPI — Paquete Python para Azure Key Vault
  7. AWS Secrets Manager Pricing — Detalles de precios de AWS para planificación de costos
  8. Multi-Cloud Secrets Management (CNCF) — Guía CNCF para gestión de secrets en entornos multi-cloud

Creado: Marzo 2026 Versión: 1.0

← Anterior4. API Key Rotation StrategiesSiguiente →6. Token Lifecycle y Audit Trails