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

  1. HashiCorp Vault Documentation — Complete official Vault documentation
  2. Vault KV Secrets Engine v2 — Reference for the most-used secrets engine (versioned key/value)
  3. Vault Transit Engine — Encryption-as-a-service documentation
  4. Vault Policies — Guide to path-based policies for access control
  5. hvac Python Client — Documentation for the official Python client for Vault
  6. Vault Dynamic Secrets — Tutorial on dynamic database credentials
  7. HashiCorp Cloud Platform (HCP) Vault — Managed Vault as a service (alternative to self-hosted)
  8. Vault AppRole Auth — Recommended auth method for applications in production

Created: March 2026 Version: 1.0