Module 5: Secrets Management
3. HashiCorp Vault: Concepts and Setup
Overview
In the previous capsule you understood why .env isn't enough for production. Now you need to learn the solution that defines the industry standard for enterprise secrets management: HashiCorp Vault. Not because everyone needs Vault — many teams will solve it with AWS Secrets Manager or GCP Secret Manager (capsule 05) — but because Vault embodies the fundamental concepts that apply to any solution: secrets engines, auth methods, policies, dynamic secrets, and transit encryption.
Vault is open-source, mature (released in 2015), and is the reference that all alternatives are compared against. Understanding Vault means understanding secrets management — and that lets you evaluate any cloud solution with informed judgment.
In this capsule you'll understand Vault's architecture, spin up a server in dev mode, and use the Python hvac client to store, retrieve, and manage secrets. The code you write here is the foundation of the secrets client you'll use in the capsule 08 project.
Vault architecture
Vault has a modular architecture based on four main concepts:
┌──────────────────────────────────────────────┐
│ 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 │ │
│ │ "who can access what" │ │
│ │ - Path-based access control │ │
│ │ - Read / Write / Delete / List │ │
│ └──────────────────┬─────────────────────┘ │
│ │ │
│ ┌──────────────────▼─────────────────────┐ │
│ │ Audit Backend │ │
│ │ "record every operation" │ │
│ │ - File audit log │ │
│ │ - Syslog │ │
│ │ - Socket │ │
│ └────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────┐ │
│ │ Storage Backend │ │
│ │ - Integrated Raft (recommended) │ │
│ │ - Consul │ │
│ │ - File (dev only) │ │
│ └────────────────────────────────────────┘ │
└──────────────────────────────────────────────┘
Secrets Engines
Secrets engines are the modules that store, generate, or encrypt data:
secrets_engines = {
"kv": {
"description": "Key-Value store for static secrets",
"use_case": "Store API keys, passwords, connection strings",
"versions": ["v1 (no versioning)", "v2 (with versioning)"],
"example_path": "secret/data/openai",
},
"transit": {
"description": "Encryption as a Service — encrypts/decrypts without revealing keys",
"use_case": "Encrypt sensitive data without managing encryption keys",
"key_point": "Encrypted data is never stored in Vault",
"example_path": "transit/encrypt/my-key",
},
"database": {
"description": "Generates dynamic database credentials",
"use_case": "Each service gets unique temporary credentials",
"key_point": "Credentials are automatically revoked after the TTL",
"example_path": "database/creds/readonly",
},
"aws": {
"description": "Generates dynamic IAM credentials for AWS",
"use_case": "Each deployment gets temporary AWS credentials",
"key_point": "Eliminates the need for long-lived AWS keys",
},
"pki": {
"description": "Certificate Authority — generates TLS certificates",
"use_case": "Mutual TLS between internal services",
},
}
for engine, info in secrets_engines.items():
print(f"\n{engine}:")
print(f" {info['description']}")
print(f" Use case: {info['use_case']}")
Auth Methods
Auth methods determine how clients authenticate with Vault:
auth_methods = {
"token": {
"description": "Direct token — the simplest",
"ideal_for": "Development, manual scripts",
"security": "Medium — tokens are long-lived by default",
},
"approle": {
"description": "Role ID + Secret ID — for applications",
"ideal_for": "Services that need to authenticate programmatically",
"security": "High — Secret ID can be single-use",
},
"kubernetes": {
"description": "Kubernetes service account tokens",
"ideal_for": "Pods in Kubernetes that need secrets",
"security": "High — integrated with the K8s control plane",
},
"aws_iam": {
"description": "Uses AWS IAM roles/users",
"ideal_for": "EC2, Lambda, ECS tasks on AWS",
"security": "High — delegated to AWS IAM",
},
"ldap": {
"description": "Authentication against LDAP/Active Directory",
"ideal_for": "Enterprise with a centralized directory",
"security": "Depends on the LDAP configuration",
},
}
for method, info in auth_methods.items():
print(f"\n{method}:")
print(f" {info['description']}")
print(f" Ideal for: {info['ideal_for']}")
Policies
Policies define who can access what. They're path-based and use HCL (HashiCorp Configuration Language):
example_policies = {
"ai-api-service": {
"description": "Policy for the AI API service",
"hcl": '''
# Read LLM API keys
path "secret/data/llm/*" {
capabilities = ["read", "list"]
}
# Read the application config
path "secret/data/app/config" {
capabilities = ["read"]
}
# CANNOT write or delete secrets
# CANNOT access database credentials
# CANNOT access admin secrets
''',
},
"migration-service": {
"description": "Policy for the migrations service",
"hcl": '''
# Only read database credentials
path "database/creds/migration" {
capabilities = ["read"]
}
# CANNOT access API keys
# CANNOT access other secrets
''',
},
"admin": {
"description": "Policy for admins (restricted)",
"hcl": '''
# Manage LLM secrets
path "secret/data/llm/*" {
capabilities = ["create", "read", "update", "delete", "list"]
}
# Manage 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"Description: {info['description']}")
print(info["hcl"])
Setup: Vault in Dev Mode
To learn and prototype, Vault has a dev mode that needs no configuration. In production you'd use a cluster — for this module, dev mode is enough.
Option 1: Docker (recommended)
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"
Option 2: Local binary
# 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"
Verification
export VAULT_ADDR='http://127.0.0.1:8200'
export VAULT_TOKEN='dev-token-12345'
vault status
vault secrets list
Option 3: Without Vault (Python simulation)
If you can't install Vault, you can simulate its behavior with Python to follow along with the module:
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:
"""Vault KV v2 simulation for learning without installation."""
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}")
# Expected output:
# 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: the hvac client
hvac is the official Python client for Vault. It provides a clean interface for all operations:
pip install hvac
Connection and basic operations
import hvac
import json
client = hvac.Client(
url="http://127.0.0.1:8200",
token="dev-token-12345",
)
print(f"Connected: {client.is_authenticated()}")
# Expected output:
# Connected: True
Write and read 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']}")
# Expected output:
# Secret written: llm/openai
# API Key: sk-proj-abc123
# Version: 1
# Created: 2026-03-13T...
Update (new version)
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']}")
# Expected output:
# New key: sk-proj-NEW-KEY-789
# Version: 2
# Old key (v1): sk-proj-abc123
List secrets
response = client.secrets.kv.v2.list_secrets(path="llm")
print(f"LLM secrets: {response['data']['keys']}")
# Expected output:
# LLM secrets: ['openai']
Vault Client wrapper for your project
To integrate Vault into your AI application, create a wrapper that encapsulates the common operations:
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:
"""Vault client with cache, fallback, and 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()
Using the 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 — use MockVault for development")
# Expected output (with Vault):
# Key: sk-proj-abc, Source: vault, Time: 15.3ms
# Key: sk-proj-abc, Source: cache, Cached: True
Dynamic Secrets: credentials that self-destruct
Vault's most powerful concept is dynamic secrets: credentials generated on-demand that are automatically revoked after a TTL:
dynamic_secrets_concept = {
"static_secret": {
"description": "An API key that lives indefinitely",
"lifecycle": "Created once → used forever → rotated manually (or never)",
"risk": "If it leaks, the attacker has access until someone revokes it",
},
"dynamic_secret": {
"description": "Credentials generated on-demand with a TTL",
"lifecycle": "Requested → generated → used → auto-revoked after the TTL",
"risk": "If it leaks, the attacker has access only until it expires (e.g., 1 hour)",
"examples": [
"Database credentials with a 1-hour TTL",
"Temporary AWS IAM credentials",
"Short-lived TLS certificates",
],
},
}
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}")
Dynamic secrets simulation
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:
"""Simulates Vault's dynamic secrets behavior."""
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")
# Expected output:
# 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
Vault's transit engine lets you encrypt and decrypt data without exposing the encryption keys. The application sends data to Vault, Vault encrypts it, and returns the ciphertext. The key never leaves Vault:
import base64
import json
from cryptography.fernet import Fernet
class TransitEngine:
"""Simulates Vault's Transit Engine for 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] + "...")
# Expected output:
# 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: when to use each
comparison = {
"vault": {
"pros": [
"Open-source, multi-cloud, on-premises",
"Dynamic secrets (temporary credentials)",
"Transit engine (encryption as a service)",
"Granular path-based policies",
"Full control over the infrastructure",
],
"cons": [
"Complex operation (HA, unsealing, upgrades)",
"Requires DevOps/SRE expertise",
"Infrastructure cost (servers, storage)",
"Overkill for teams < 20 people",
],
"ideal_for": "Enterprise, multi-cloud, on-premises, teams with SRE",
},
"cloud_kms": {
"pros": [
"Managed — no operations",
"Native integration with the cloud provider",
"Pay-per-use — predictable cost",
"Setup in minutes, not hours",
"Native IAM for access control",
],
"cons": [
"Partial vendor lock-in",
"No native dynamic secrets (in most)",
"Less flexibility in policies",
"Features vary by provider",
],
"ideal_for": "Startups, mid-size teams, single-cloud, pragmatism",
},
}
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 for: {info['ideal_for']}")
Organizing secrets for AI systems
A good structure of paths in Vault (or its equivalent in cloud KMS) makes management easier:
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("Secrets structure for an AI system:")
for path, data in ai_secrets_structure.items():
keys = list(data.keys())
print(f" {path}: [{', '.join(keys)}]")
Troubleshooting
"The Vault dev server stopped and I lost all my secrets"
Dev mode stores everything in memory. When you stop the server, the data is lost. This is intentional — dev mode is for development. For persistence, use a storage backend like Raft or Consul.
"hvac throws a ConnectionError when connecting"
Check that: (1) Vault is running (vault status), (2) VAULT_ADDR points to the right host, (3) there's no firewall blocking port 8200, (4) if you're using Docker, the port mapping is correct (-p 8200:8200).
"Permission denied when reading a secret"
Your token doesn't have the required policy. In dev mode with the root token, this shouldn't happen. In production, verify that the token's policy includes the read capability on the secret's path.
"I can't install Vault and I'm not sure it's worth it"
Use this capsule's MockVault to learn the concepts. If your team is < 20 people and you use a single cloud provider, AWS Secrets Manager or GCP Secret Manager (capsule 05) is probably more practical.
Exercises
Exercise 1: Store and retrieve 5 secrets for your AI system
Using hvac or MockVault, create a complete secrets structure for an AI system with OpenAI, database, Redis, and webhook secrets:
See solution
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}")
Exercise 2: Implement a decorator that gets secrets from Vault
Create a @requires_secret decorator that injects a secret as an argument to a function:
See solution
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)
Exercise 3: Simulate dynamic secrets with a TTL
Extend DynamicSecretsEngine to support renewing leases before they expire:
See solution
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")
Exercise 4: Create policies for 3 different services
Define the minimal policies that each service needs:
See solution
policies = {
"api-service": {
"description": "Main API service — needs LLM keys and database",
"allowed_paths": ["secret/data/llm/*", "secret/data/database/main"],
"capabilities": ["read"],
},
"worker-service": {
"description": "Background worker — only database and Redis",
"allowed_paths": ["secret/data/database/main", "secret/data/cache/redis"],
"capabilities": ["read"],
},
"admin-cli": {
"description": "Admin CLI — full secrets management",
"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']}")
Summary
- HashiCorp Vault is the open-source reference for enterprise secrets management with secrets engines, auth methods, policies, and audit backends
- The secrets engines most relevant to AI are KV (store API keys), Transit (encryption as a service), and Database (dynamic credentials)
- Auth methods determine how your services authenticate: token for dev, AppRole for production, Kubernetes for pods
- Policies implement least privilege with path-based access control — each service only sees the secrets it needs
- Dynamic secrets are the most powerful concept: temporary credentials that auto-revoke, eliminating the risk of long-lived credentials
- The Transit engine lets you encrypt data without managing encryption keys — Vault handles the keys internally
- The VaultSecretsClient wrapper with cache and audit logging is the foundation of the project's secrets client
- Not everyone needs Vault — cloud KMS (capsule 05) is more practical for most. Vault's concepts apply to any solution
Next capsule: In capsule 04 you'll implement API key rotation — the most critical operation in secrets management. You'll see zero-downtime rotation strategies, the dual-key pattern, and you'll build a rotation scheduler in Python that automatically rotates your LLM API keys without affecting service availability.
Resources
- HashiCorp Vault Documentation — Complete official Vault documentation
- Vault KV Secrets Engine v2 — Reference for the most-used secrets engine (versioned key/value)
- Vault Transit Engine — Encryption-as-a-service documentation
- Vault Policies — Guide to path-based policies for access control
- hvac Python Client — Documentation for the official Python client for Vault
- Vault Dynamic Secrets — Tutorial on dynamic database credentials
- HashiCorp Cloud Platform (HCP) Vault — Managed Vault as a service (alternative to self-hosted)
- Vault AppRole Auth — Recommended auth method for applications in production
Created: March 2026 Version: 1.0