Módulo 5: Secrets Management

3. HashiCorp Vault: Conceptos y Setup

Siguiente →

Descripción

En la cápsula anterior entendiste por qué .env no es suficiente para producción. Ahora necesitas conocer la solución que define el estándar de la industria para secrets management enterprise: HashiCorp Vault. No porque todos necesiten Vault — muchos equipos resolverán con AWS Secrets Manager o GCP Secret Manager (cápsula 05) — sino porque Vault encarna los conceptos fundamentales que aplican a cualquier solución: secrets engines, auth methods, policies, dynamic secrets, y transit encryption.

Vault es open-source, maduro (lanzado en 2015), y es la referencia contra la que se comparan todas las alternativas. Entender Vault significa entender secrets management — y eso te permite evaluar cualquier solución cloud con criterio informado.

En esta cápsula vas a entender la arquitectura de Vault, levantar un server en dev mode, y usar el cliente Python hvac para almacenar, recuperar, y gestionar secrets. El código que escribas aquí es la base del secrets client que usarás en el proyecto de la cápsula 08.

Arquitectura de Vault

Vault tiene una arquitectura modular basada en cuatro conceptos principales:

┌──────────────────────────────────────────────┐
│                 HashiCorp Vault                │
│                                                │
│  ┌──────────────┐  ┌──────────────────────┐   │
│  │ Auth Methods  │  │ Secrets Engines      │   │
│  │              │  │                      │   │
│  │ - Token      │  │ - KV (key/value)     │   │
│  │ - AppRole    │  │ - Transit (encrypt)  │   │
│  │ - LDAP       │  │ - Database (dynamic) │   │
│  │ - AWS IAM    │  │ - PKI (certificates) │   │
│  │ - Kubernetes │  │ - AWS (dynamic IAM)  │   │
│  └──────┬───────┘  └──────────┬───────────┘   │
│         │                     │               │
│  ┌──────▼─────────────────────▼───────────┐   │
│  │             Policies                    │   │
│  │  "quién puede acceder a qué"           │   │
│  │  - Path-based access control           │   │
│  │  - Read / Write / Delete / List         │   │
│  └──────────────────┬─────────────────────┘   │
│                     │                         │
│  ┌──────────────────▼─────────────────────┐   │
│  │             Audit Backend               │   │
│  │  "registrar cada operación"            │   │
│  │  - File audit log                      │   │
│  │  - Syslog                              │   │
│  │  - Socket                              │   │
│  └────────────────────────────────────────┘   │
│                                                │
│  ┌────────────────────────────────────────┐   │
│  │             Storage Backend             │   │
│  │  - Integrated Raft (recomendado)       │   │
│  │  - Consul                              │   │
│  │  - File (dev only)                     │   │
│  └────────────────────────────────────────┘   │
└──────────────────────────────────────────────┘

Secrets Engines

Los secrets engines son los módulos que almacenan, generan, o encriptan datos:

secrets_engines = {
    "kv": {
        "description": "Key-Value store para secrets estáticos",
        "use_case": "Almacenar API keys, passwords, connection strings",
        "versions": ["v1 (sin versioning)", "v2 (con versioning)"],
        "example_path": "secret/data/openai",
    },
    "transit": {
        "description": "Encryption as a Service — encripta/desencripta sin revelar keys",
        "use_case": "Encriptar datos sensibles sin gestionar encryption keys",
        "key_point": "Los datos encriptados nunca se almacenan en Vault",
        "example_path": "transit/encrypt/my-key",
    },
    "database": {
        "description": "Genera credenciales dinámicas de base de datos",
        "use_case": "Cada servicio recibe credentials temporales únicas",
        "key_point": "Credentials se revocan automáticamente después del TTL",
        "example_path": "database/creds/readonly",
    },
    "aws": {
        "description": "Genera IAM credentials dinámicas para AWS",
        "use_case": "Cada deployment recibe AWS credentials temporales",
        "key_point": "Elimina la necesidad de long-lived AWS keys",
    },
    "pki": {
        "description": "Certificate Authority — genera TLS certificates",
        "use_case": "Mutual TLS entre servicios internos",
    },
}

for engine, info in secrets_engines.items():
    print(f"\n{engine}:")
    print(f"  {info['description']}")
    print(f"  Use case: {info['use_case']}")

Auth Methods

Los auth methods determinan cómo los clientes se autentican con Vault:

auth_methods = {
    "token": {
        "description": "Token directo — el más simple",
        "ideal_for": "Development, scripts manuales",
        "security": "Media — tokens son long-lived por default",
    },
    "approle": {
        "description": "Role ID + Secret ID — para aplicaciones",
        "ideal_for": "Servicios que necesitan autenticarse programáticamente",
        "security": "Alta — Secret ID puede ser de uso único",
    },
    "kubernetes": {
        "description": "Service account tokens de Kubernetes",
        "ideal_for": "Pods en Kubernetes que necesitan secrets",
        "security": "Alta — integrado con el control plane de K8s",
    },
    "aws_iam": {
        "description": "Usa IAM roles/users de AWS",
        "ideal_for": "EC2, Lambda, ECS tasks en AWS",
        "security": "Alta — delegada a AWS IAM",
    },
    "ldap": {
        "description": "Autenticación contra LDAP/Active Directory",
        "ideal_for": "Enterprise con directorio centralizado",
        "security": "Depende de la configuración LDAP",
    },
}

for method, info in auth_methods.items():
    print(f"\n{method}:")
    print(f"  {info['description']}")
    print(f"  Ideal para: {info['ideal_for']}")

Policies

Las policies definen quién puede acceder a qué. Son path-based y usan HCL (HashiCorp Configuration Language):

example_policies = {
    "ai-api-service": {
        "description": "Policy para el servicio de API AI",
        "hcl": '''
# Leer API keys de LLM
path "secret/data/llm/*" {
  capabilities = ["read", "list"]
}

# Leer config de la aplicación
path "secret/data/app/config" {
  capabilities = ["read"]
}

# NO puede escribir ni borrar secrets
# NO puede acceder a database credentials
# NO puede acceder a admin secrets
''',
    },
    "migration-service": {
        "description": "Policy para el servicio de migraciones",
        "hcl": '''
# Solo leer database credentials
path "database/creds/migration" {
  capabilities = ["read"]
}

# NO puede acceder a API keys
# NO puede acceder a otros secrets
''',
    },
    "admin": {
        "description": "Policy para admins (restringida)",
        "hcl": '''
# Gestionar secrets de LLM
path "secret/data/llm/*" {
  capabilities = ["create", "read", "update", "delete", "list"]
}

# Gestionar database roles
path "database/roles/*" {
  capabilities = ["create", "read", "update", "delete", "list"]
}
''',
    },
}

for policy_name, info in example_policies.items():
    print(f"\n=== Policy: {policy_name} ===")
    print(f"Descripción: {info['description']}")
    print(info["hcl"])

Setup: Vault en Dev Mode

Para aprender y prototipar, Vault tiene un dev mode que no necesita configuración. En producción usarías un cluster — para este módulo, dev mode es suficiente.

Opción 1: Docker (recomendada)

docker run -d \
  --name vault-dev \
  -p 8200:8200 \
  -e 'VAULT_DEV_ROOT_TOKEN_ID=dev-token-12345' \
  -e 'VAULT_DEV_LISTEN_ADDRESS=0.0.0.0:8200' \
  hashicorp/vault:latest

echo "Vault dev server running at http://localhost:8200"
echo "Root token: dev-token-12345"

Opción 2: Binario local

# macOS
brew install vault

# Linux
curl -fsSL https://releases.hashicorp.com/vault/1.15.4/vault_1.15.4_linux_amd64.zip -o vault.zip
unzip vault.zip && sudo mv vault /usr/local/bin/

vault server -dev -dev-root-token-id="dev-token-12345"

Verificación

export VAULT_ADDR='http://127.0.0.1:8200'
export VAULT_TOKEN='dev-token-12345'

vault status
vault secrets list

Opción 3: Sin Vault (simulación con Python)

Si no puedes instalar Vault, puedes simular su comportamiento con Python para seguir el módulo:

import json
import time
from typing import Optional
from dataclasses import dataclass, field


@dataclass
class VaultSecret:
    data: dict
    metadata: dict = field(default_factory=dict)


class MockVault:
    """Simulación de Vault KV v2 para aprendizaje sin instalación."""

    def __init__(self, token: str = "dev-token-12345"):
        self._token = token
        self._store: dict[str, VaultSecret] = {}
        self._audit_log: list[dict] = []
        self._version_counter: dict[str, int] = {}

    def _audit(self, operation: str, path: str, success: bool = True):
        self._audit_log.append({
            "timestamp": time.time(),
            "operation": operation,
            "path": path,
            "success": success,
        })

    def write_secret(self, path: str, data: dict) -> dict:
        version = self._version_counter.get(path, 0) + 1
        self._version_counter[path] = version

        metadata = {
            "created_time": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "version": version,
            "destroyed": False,
        }
        self._store[path] = VaultSecret(data=data, metadata=metadata)
        self._audit("write", path)
        return metadata

    def read_secret(self, path: str) -> Optional[dict]:
        secret = self._store.get(path)
        if secret is None:
            self._audit("read", path, success=False)
            return None
        self._audit("read", path)
        return {
            "data": secret.data,
            "metadata": secret.metadata,
        }

    def delete_secret(self, path: str) -> bool:
        if path in self._store:
            del self._store[path]
            self._audit("delete", path)
            return True
        self._audit("delete", path, success=False)
        return False

    def list_secrets(self, prefix: str = "") -> list[str]:
        self._audit("list", prefix)
        return [k for k in self._store.keys() if k.startswith(prefix)]


vault = MockVault()

vault.write_secret("secret/llm/openai", {
    "api_key": "sk-proj-abc123",
    "org_id": "org-xyz789",
})
vault.write_secret("secret/llm/anthropic", {
    "api_key": "sk-ant-def456",
})
vault.write_secret("secret/database/main", {
    "url": "postgresql://user:pass@host:5432/db",
})

result = vault.read_secret("secret/llm/openai")
print(f"OpenAI secret: {json.dumps(result, indent=2)}")

keys = vault.list_secrets("secret/llm/")
print(f"\nLLM secrets: {keys}")
# Output esperado:
# OpenAI secret: {
#   "data": {
#     "api_key": "sk-proj-abc123",
#     "org_id": "org-xyz789"
#   },
#   "metadata": {
#     "created_time": "2026-03-13T...",
#     "version": 1,
#     "destroyed": false
#   }
# }
#
# LLM secrets: ['secret/llm/openai', 'secret/llm/anthropic']

Python + Vault: el cliente hvac

hvac es el cliente Python oficial para Vault. Proporciona una interface limpia para todas las operaciones:

pip install hvac

Conexión y operaciones básicas

import hvac
import json

client = hvac.Client(
    url="http://127.0.0.1:8200",
    token="dev-token-12345",
)

print(f"Connected: {client.is_authenticated()}")
# Output esperado:
# Connected: True

Escribir y leer secrets (KV v2)

import hvac

client = hvac.Client(url="http://127.0.0.1:8200", token="dev-token-12345")

client.secrets.kv.v2.create_or_update_secret(
    path="llm/openai",
    secret={"api_key": "sk-proj-abc123", "org_id": "org-xyz"},
)
print("Secret written: llm/openai")

response = client.secrets.kv.v2.read_secret_version(path="llm/openai")
secret_data = response["data"]["data"]
metadata = response["data"]["metadata"]

print(f"API Key: {secret_data['api_key']}")
print(f"Version: {metadata['version']}")
print(f"Created: {metadata['created_time']}")
# Output esperado:
# Secret written: llm/openai
# API Key: sk-proj-abc123
# Version: 1
# Created: 2026-03-13T...

Actualizar (nueva versión)

client.secrets.kv.v2.create_or_update_secret(
    path="llm/openai",
    secret={"api_key": "sk-proj-NEW-KEY-789", "org_id": "org-xyz"},
)

response = client.secrets.kv.v2.read_secret_version(path="llm/openai")
print(f"New key: {response['data']['data']['api_key']}")
print(f"Version: {response['data']['metadata']['version']}")

old = client.secrets.kv.v2.read_secret_version(path="llm/openai", version=1)
print(f"Old key (v1): {old['data']['data']['api_key']}")
# Output esperado:
# New key: sk-proj-NEW-KEY-789
# Version: 2
# Old key (v1): sk-proj-abc123

Listar secrets

response = client.secrets.kv.v2.list_secrets(path="llm")
print(f"LLM secrets: {response['data']['keys']}")
# Output esperado:
# LLM secrets: ['openai']

Vault Client wrapper para tu proyecto

Para integrar Vault en tu aplicación AI, crea un wrapper que encapsule las operaciones comunes:

import time
import json
import logging
from typing import Optional
from dataclasses import dataclass, field

logger = logging.getLogger("vault_client")


@dataclass
class SecretResponse:
    key: str
    value: Optional[str]
    version: int = 0
    found: bool = True
    source: str = "vault"
    cached: bool = False
    access_time_ms: float = 0.0


class VaultSecretsClient:
    """Cliente de Vault con cache, fallback, y audit logging."""

    def __init__(
        self,
        vault_url: str = "http://127.0.0.1:8200",
        token: str = "",
        cache_ttl_seconds: int = 300,
        mount_point: str = "secret",
    ):
        self.vault_url = vault_url
        self.token = token
        self.mount_point = mount_point
        self.cache_ttl = cache_ttl_seconds
        self._cache: dict[str, tuple[dict, float]] = {}

        try:
            import hvac
            self._client = hvac.Client(url=vault_url, token=token)
            self._connected = self._client.is_authenticated()
        except Exception:
            self._client = None
            self._connected = False

    @property
    def is_connected(self) -> bool:
        return self._connected

    def get_secret(self, path: str, key: str = "value") -> SecretResponse:
        start = time.perf_counter()

        cached = self._get_from_cache(path)
        if cached is not None:
            elapsed = (time.perf_counter() - start) * 1000
            return SecretResponse(
                key=path, value=cached.get(key),
                found=True, cached=True,
                source="cache", access_time_ms=elapsed,
            )

        if not self._connected or self._client is None:
            elapsed = (time.perf_counter() - start) * 1000
            logger.warning(f"Vault not connected, cannot read: {path}")
            return SecretResponse(
                key=path, value=None, found=False,
                source="vault_unavailable", access_time_ms=elapsed,
            )

        try:
            response = self._client.secrets.kv.v2.read_secret_version(
                path=path, mount_point=self.mount_point,
            )
            data = response["data"]["data"]
            version = response["data"]["metadata"]["version"]
            self._set_cache(path, data)

            elapsed = (time.perf_counter() - start) * 1000
            logger.info(f"Secret read: {path} (v{version}, {elapsed:.1f}ms)")
            return SecretResponse(
                key=path, value=data.get(key),
                version=version, found=True,
                source="vault", access_time_ms=elapsed,
            )
        except Exception as e:
            elapsed = (time.perf_counter() - start) * 1000
            logger.error(f"Failed to read secret {path}: {e}")
            return SecretResponse(
                key=path, value=None, found=False,
                source="vault_error", access_time_ms=elapsed,
            )

    def set_secret(self, path: str, data: dict) -> bool:
        if not self._connected or self._client is None:
            logger.error("Vault not connected, cannot write")
            return False

        try:
            self._client.secrets.kv.v2.create_or_update_secret(
                path=path, secret=data, mount_point=self.mount_point,
            )
            self._invalidate_cache(path)
            logger.info(f"Secret written: {path}")
            return True
        except Exception as e:
            logger.error(f"Failed to write secret {path}: {e}")
            return False

    def _get_from_cache(self, path: str) -> Optional[dict]:
        if path in self._cache:
            data, cached_at = self._cache[path]
            if time.time() - cached_at < self.cache_ttl:
                return data
            del self._cache[path]
        return None

    def _set_cache(self, path: str, data: dict):
        self._cache[path] = (data, time.time())

    def _invalidate_cache(self, path: str):
        self._cache.pop(path, None)

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

Uso del wrapper

secrets = VaultSecretsClient(
    vault_url="http://127.0.0.1:8200",
    token="dev-token-12345",
    cache_ttl_seconds=300,
)

if secrets.is_connected:
    secrets.set_secret("llm/openai", {"api_key": "sk-proj-abc", "org_id": "org-1"})

    result = secrets.get_secret("llm/openai", key="api_key")
    print(f"Key: {result.value}, Source: {result.source}, Time: {result.access_time_ms:.1f}ms")

    result2 = secrets.get_secret("llm/openai", key="api_key")
    print(f"Key: {result2.value}, Source: {result2.source}, Cached: {result2.cached}")
else:
    print("Vault not available — usa MockVault para desarrollo")
# Output esperado (con Vault):
# Key: sk-proj-abc, Source: vault, Time: 15.3ms
# Key: sk-proj-abc, Source: cache, Cached: True

Dynamic Secrets: credenciales que se auto-destruyen

El concepto más poderoso de Vault son los dynamic secrets: credenciales generadas on-demand que se revocan automáticamente después de un TTL:

dynamic_secrets_concept = {
    "static_secret": {
        "description": "Una API key que vive indefinidamente",
        "lifecycle": "Creada una vez → usada siempre → rotada manualmente (o nunca)",
        "risk": "Si se filtra, el atacante tiene acceso hasta que alguien la revoque",
    },
    "dynamic_secret": {
        "description": "Credenciales generadas on-demand con TTL",
        "lifecycle": "Solicitada → generada → usada → auto-revocada después del TTL",
        "risk": "Si se filtra, el atacante tiene acceso solo hasta que expire (e.g., 1 hora)",
        "examples": [
            "Database credentials con TTL de 1 hora",
            "AWS IAM credentials temporales",
            "TLS certificates de corta duración",
        ],
    },
}

print("Static Secret:")
for k, v in dynamic_secrets_concept["static_secret"].items():
    print(f"  {k}: {v}")

print("\nDynamic Secret:")
for k, v in dynamic_secrets_concept["dynamic_secret"].items():
    if isinstance(v, list):
        print(f"  {k}:")
        for item in v:
            print(f"    - {item}")
    else:
        print(f"  {k}: {v}")

Simulación de dynamic secrets

import time
import uuid
from dataclasses import dataclass
from typing import Optional


@dataclass
class DynamicCredential:
    username: str
    password: str
    created_at: float
    ttl_seconds: int
    lease_id: str

    @property
    def is_expired(self) -> bool:
        return time.time() - self.created_at > self.ttl_seconds

    @property
    def remaining_seconds(self) -> float:
        remaining = self.ttl_seconds - (time.time() - self.created_at)
        return max(0, remaining)


class DynamicSecretsEngine:
    """Simula el comportamiento de dynamic secrets de Vault."""

    def __init__(self):
        self._active_leases: dict[str, DynamicCredential] = {}

    def generate_database_credential(
        self, role: str, ttl_seconds: int = 3600
    ) -> DynamicCredential:
        cred = DynamicCredential(
            username=f"v-{role}-{uuid.uuid4().hex[:8]}",
            password=uuid.uuid4().hex,
            created_at=time.time(),
            ttl_seconds=ttl_seconds,
            lease_id=f"database/creds/{role}/{uuid.uuid4().hex[:8]}",
        )
        self._active_leases[cred.lease_id] = cred
        return cred

    def revoke_credential(self, lease_id: str) -> bool:
        if lease_id in self._active_leases:
            del self._active_leases[lease_id]
            return True
        return False

    def cleanup_expired(self) -> int:
        expired = [
            lid for lid, cred in self._active_leases.items()
            if cred.is_expired
        ]
        for lid in expired:
            del self._active_leases[lid]
        return len(expired)

    @property
    def active_count(self) -> int:
        return len(self._active_leases)


engine = DynamicSecretsEngine()

api_cred = engine.generate_database_credential("api-readonly", ttl_seconds=5)
print(f"Generated credential:")
print(f"  Username: {api_cred.username}")
print(f"  Password: {api_cred.password[:8]}...")
print(f"  TTL: {api_cred.ttl_seconds}s")
print(f"  Expired: {api_cred.is_expired}")
print(f"  Remaining: {api_cred.remaining_seconds:.0f}s")

print(f"\nActive leases: {engine.active_count}")

time.sleep(2)
print(f"\nAfter 2s — Expired: {api_cred.is_expired}, Remaining: {api_cred.remaining_seconds:.0f}s")
# Output esperado:
# Generated credential:
#   Username: v-api-readonly-a1b2c3d4
#   Password: e5f6g7h8...
#   TTL: 5s
#   Expired: False
#   Remaining: 5s
#
# Active leases: 1
#
# After 2s — Expired: False, Remaining: 3s

Transit Engine: Encryption as a Service

El transit engine de Vault permite encriptar y desencriptar datos sin exponer las encryption keys. La aplicación envía datos a Vault, Vault los encripta, y devuelve el ciphertext. La key nunca sale de Vault:

import base64
import json
from cryptography.fernet import Fernet


class TransitEngine:
    """Simula el Transit Engine de Vault para encryption as a service."""

    def __init__(self):
        self._keys: dict[str, bytes] = {}

    def create_key(self, name: str):
        self._keys[name] = Fernet.generate_key()

    def encrypt(self, key_name: str, plaintext: str) -> str:
        if key_name not in self._keys:
            raise ValueError(f"Key not found: {key_name}")

        cipher = Fernet(self._keys[key_name])
        encrypted = cipher.encrypt(plaintext.encode())
        return f"vault:v1:{base64.b64encode(encrypted).decode()}"

    def decrypt(self, key_name: str, ciphertext: str) -> str:
        if key_name not in self._keys:
            raise ValueError(f"Key not found: {key_name}")

        prefix = "vault:v1:"
        if not ciphertext.startswith(prefix):
            raise ValueError("Invalid ciphertext format")

        encrypted = base64.b64decode(ciphertext[len(prefix):])
        cipher = Fernet(self._keys[key_name])
        return cipher.decrypt(encrypted).decode()


transit = TransitEngine()
transit.create_key("pii-encryption")

sensitive_data = "usuario@email.com"
encrypted = transit.encrypt("pii-encryption", sensitive_data)
print(f"Original:  {sensitive_data}")
print(f"Encrypted: {encrypted[:50]}...")

decrypted = transit.decrypt("pii-encryption", encrypted)
print(f"Decrypted: {decrypted}")
print(f"Match: {sensitive_data == decrypted}")

db_record = {
    "user_id": "usr-123",
    "email": transit.encrypt("pii-encryption", "usuario@email.com"),
    "name": transit.encrypt("pii-encryption", "Juan García"),
    "plan": "pro",
}
print(f"\nDB record (encrypted PII):")
print(json.dumps(db_record, indent=2)[:200] + "...")
# Output esperado:
# Original:  usuario@email.com
# Encrypted: vault:v1:Z0FBQUFB...
# Decrypted: usuario@email.com
# Match: True
#
# DB record (encrypted PII):
# {
#   "user_id": "usr-123",
#   "email": "vault:v1:...",
#   ...
# }

Vault vs Cloud KMS: cuándo usar cada uno

comparison = {
    "vault": {
        "pros": [
            "Open-source, multi-cloud, on-premises",
            "Dynamic secrets (credenciales temporales)",
            "Transit engine (encryption as a service)",
            "Policies granulares path-based",
            "Control total sobre la infraestructura",
        ],
        "cons": [
            "Operación compleja (HA, unsealing, upgrades)",
            "Requiere expertise DevOps/SRE",
            "Costo de infraestructura (servers, storage)",
            "Overkill para equipos < 20 personas",
        ],
        "ideal_for": "Enterprise, multi-cloud, on-premises, equipos con SRE",
    },
    "cloud_kms": {
        "pros": [
            "Managed — sin operación",
            "Integración nativa con el cloud provider",
            "Pay-per-use — costo predecible",
            "Setup en minutos, no horas",
            "IAM nativo para access control",
        ],
        "cons": [
            "Vendor lock-in parcial",
            "Sin dynamic secrets nativos (en la mayoría)",
            "Menos flexibilidad en policies",
            "Features varían por proveedor",
        ],
        "ideal_for": "Startups, equipos medianos, single-cloud, pragmatismo",
    },
}

for solution, info in comparison.items():
    print(f"\n=== {solution.upper()} ===")
    print("Pros:")
    for pro in info["pros"]:
        print(f"  ✅ {pro}")
    print("Cons:")
    for con in info["cons"]:
        print(f"  ❌ {con}")
    print(f"Ideal para: {info['ideal_for']}")

Organizando secrets para sistemas AI

Una buena estructura de paths en Vault (o su equivalente en cloud KMS) facilita la gestión:

ai_secrets_structure = {
    "secret/llm/openai": {
        "api_key": "sk-proj-...",
        "org_id": "org-...",
        "project_id": "proj-...",
    },
    "secret/llm/anthropic": {
        "api_key": "sk-ant-...",
    },
    "secret/database/main": {
        "url": "postgresql://...",
        "readonly_url": "postgresql://...",
    },
    "secret/database/vector": {
        "api_key": "pcsk-...",
        "environment": "us-east-1",
        "index_name": "production",
    },
    "secret/cache/redis": {
        "url": "redis://...",
    },
    "secret/app/config": {
        "jwt_secret": "...",
        "webhook_secret": "...",
    },
}

print("Estructura de secrets para sistema AI:")
for path, data in ai_secrets_structure.items():
    keys = list(data.keys())
    print(f"  {path}: [{', '.join(keys)}]")

Troubleshooting

"Vault dev server se detuvo y perdí todos los secrets"

Dev mode almacena todo en memoria. Al detener el server, los datos se pierden. Esto es intencional — dev mode es para desarrollo. Para persistencia, usa un storage backend como Raft o Consul.

"hvac da ConnectionError al conectar"

Verifica que: (1) Vault está corriendo (vault status), (2) VAULT_ADDR apunta al host correcto, (3) no hay firewall bloqueando el puerto 8200, (4) si usas Docker, el port mapping es correcto (-p 8200:8200).

"Permission denied al leer un secret"

Tu token no tiene la policy necesaria. En dev mode con el root token, no debería pasar. En producción, verifica que la policy del token incluye read capability en el path del secret.

"No puedo instalar Vault y no sé si vale la pena"

Usa MockVault de esta cápsula para aprender los conceptos. Si tu equipo es < 20 personas y usas un solo cloud provider, probablemente AWS Secrets Manager o GCP Secret Manager (cápsula 05) sea más práctico.

Ejercicios

Ejercicio 1: Almacena y recupera 5 secrets para tu sistema AI

Usando hvac o MockVault, crea una estructura completa de secrets para un sistema AI con OpenAI, database, Redis, y webhook secrets:

Ver solución 
vault = MockVault()

secrets_to_store = {
    "llm/openai": {"api_key": "sk-proj-abc", "org_id": "org-1", "model": "gpt-4o"},
    "llm/anthropic": {"api_key": "sk-ant-xyz"},
    "database/main": {"url": "postgresql://user:pass@host/db", "pool_size": "10"},
    "cache/redis": {"url": "redis://:pass@host:6379/0"},
    "app/webhooks": {"slack_url": "https://hooks.slack.com/...", "secret": "whsec-123"},
}

for path, data in secrets_to_store.items():
    vault.write_secret(f"secret/{path}", data)
    print(f"✅ Written: secret/{path}")

for path in secrets_to_store:
    result = vault.read_secret(f"secret/{path}")
    keys = list(result["data"].keys())
    print(f"📖 Read: secret/{path} → keys: {keys}")

Ejercicio 2: Implementa un decorator que obtiene secrets de Vault

Crea un decorator @requires_secret que inyecta un secret como argumento a una función:

Ver solución 
from functools import wraps

def requires_secret(secret_path: str, key: str = "value"):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, vault_client=None, **kwargs):
            if vault_client is None:
                raise ValueError("vault_client is required")
            result = vault_client.read_secret(secret_path)
            if result is None:
                raise ValueError(f"Secret not found: {secret_path}")
            secret_value = result["data"].get(key)
            return func(*args, secret=secret_value, **kwargs)
        return wrapper
    return decorator


vault = MockVault()
vault.write_secret("secret/llm/openai", {"api_key": "sk-proj-test"})

@requires_secret("secret/llm/openai", key="api_key")
def call_openai(prompt: str, secret: str = ""):
    print(f"Calling OpenAI with key: {secret[:10]}... | Prompt: {prompt}")

call_openai("Hello!", vault_client=vault)

Ejercicio 3: Simula dynamic secrets con TTL

Extiende DynamicSecretsEngine para soportar renovación de leases antes de que expiren:

Ver solución 
class ExtendedDynamicEngine(DynamicSecretsEngine):
    def renew_lease(self, lease_id: str, extend_seconds: int = 3600) -> bool:
        if lease_id not in self._active_leases:
            return False
        cred = self._active_leases[lease_id]
        if cred.is_expired:
            return False
        cred.ttl_seconds += extend_seconds
        print(f"Renewed: {lease_id}, new remaining: {cred.remaining_seconds:.0f}s")
        return True


engine = ExtendedDynamicEngine()
cred = engine.generate_database_credential("api", ttl_seconds=10)
print(f"Initial TTL: {cred.remaining_seconds:.0f}s")

engine.renew_lease(cred.lease_id, extend_seconds=60)
print(f"After renewal: {cred.remaining_seconds:.0f}s")

Ejercicio 4: Crea policies para 3 servicios diferentes

Define las policies mínimas que cada servicio necesita:

Ver solución 
policies = {
    "api-service": {
        "description": "Servicio API principal — necesita LLM keys y database",
        "allowed_paths": ["secret/data/llm/*", "secret/data/database/main"],
        "capabilities": ["read"],
    },
    "worker-service": {
        "description": "Worker de background — solo database y Redis",
        "allowed_paths": ["secret/data/database/main", "secret/data/cache/redis"],
        "capabilities": ["read"],
    },
    "admin-cli": {
        "description": "CLI de admin — gestión completa de secrets",
        "allowed_paths": ["secret/data/*"],
        "capabilities": ["create", "read", "update", "delete", "list"],
    },
}

for name, policy in policies.items():
    print(f"\n=== {name} ===")
    print(f"  {policy['description']}")
    print(f"  Paths: {policy['allowed_paths']}")
    print(f"  Capabilities: {policy['capabilities']}")

Resumen

  • HashiCorp Vault es la referencia open-source para secrets management enterprise con secrets engines, auth methods, policies, y audit backends
  • Los secrets engines más relevantes para AI son KV (almacenar API keys), Transit (encryption as a service), y Database (dynamic credentials)
  • Auth methods determinan cómo tus servicios se autentican: token para dev, AppRole para producción, Kubernetes para pods
  • Policies implementan least privilege con path-based access control — cada servicio solo ve los secrets que necesita
  • Dynamic secrets son el concepto más poderoso: credenciales temporales que se auto-revocan, eliminando el riesgo de long-lived credentials
  • Transit engine permite encriptar datos sin gestionar encryption keys — Vault maneja las keys internamente
  • El VaultSecretsClient wrapper con cache y audit logging es la base del secrets client del proyecto
  • No todos necesitan Vault — cloud KMS (cápsula 05) es más práctico para la mayoría. Los conceptos de Vault aplican a cualquier solución

Próxima cápsula: En la cápsula 04 vas a implementar API key rotation — la operación más crítica de secrets management. Verás estrategias de zero-downtime rotation, el patrón de dual-key, y construirás un rotation scheduler en Python que rota automáticamente tus API keys de LLM sin afectar la disponibilidad del servicio.

Recursos

  1. HashiCorp Vault Documentation — Documentación oficial completa de Vault
  2. Vault KV Secrets Engine v2 — Referencia del secrets engine más usado (versioned key/value)
  3. Vault Transit Engine — Documentación de encryption as a service
  4. Vault Policies — Guía de policies path-based para access control
  5. hvac Python Client — Documentación del cliente Python oficial para Vault
  6. Vault Dynamic Secrets — Tutorial de credenciales dinámicas de database
  7. HashiCorp Cloud Platform (HCP) Vault — Vault managed como servicio (alternativa a self-hosted)
  8. Vault AppRole Auth — Auth method recomendado para aplicaciones en producción

Creado: Marzo 2026 Versión: 1.0

← Anterior2. Más Allá de .env: Por Qué No Es SuficienteSiguiente →4. API Key Rotation Strategies