Module 5: Secrets Management

8. Project: Secrets Management Setup

Project overview

This project closes Module 5 with the implementation of a complete Secrets Management Setup — an integrated secrets management system for your AI application that includes a secrets client with provider abstraction, a rotation scheduler, an audit logger, and FastAPI integration. It's not a partial exercise — it's the infrastructure artifact that changes how your system obtains and manages credentials.

In the previous capsules you built the individual pieces: you understood why .env isn't enough (02), met Vault as a reference (03), implemented rotation patterns (04), explored cloud KMS (05), designed lifecycle and audit trails (06), and integrated with FastAPI using least privilege (07). Now everything consolidates into a system that:

Abstracts the provider: A unified interface that works with Vault, AWS, GCP, Azure, or environment variables
Rotates secrets: An automatic scheduler with a dual-key strategy and zero-downtime
Records audit trails: Every access to every secret is logged with timestamps, actor, and source
Integrates with FastAPI: Dependency injection to obtain secrets in endpoints
Resists failures: Cache with TTL, circuit breaker, and fallback levels when the provider isn't available

By the end you'll have a directory with executable Python code, unit tests, and a FastAPI server that demonstrates the system in action. This is the guide's fifth artifact and it integrates into Module 8 (Secured AI System) as the credential management layer.

Project goal

Build a secrets management system for an AI application with: secrets provider abstraction, a zero-downtime rotation scheduler, an audit trail of every access, FastAPI integration with dependency injection, and resilience when the secrets service isn't available — all in a reusable package that works with Vault (dev mode) or local simulation.

Technical specifications

Stack

Python >= 3.10
pydantic >= 2.0
fastapi >= 0.100
uvicorn >= 0.20
cryptography >= 41.0
schedule >= 1.2
pytest >= 7.0
pytest-asyncio >= 0.21
httpx >= 0.24

Deliverable structure

secrets-management-setup/
├── secrets/
│   ├── __init__.py
│   ├── provider.py           # SecretsProvider interface + implementations
│   ├── client.py             # ResilientSecretsClient
│   ├── rotation.py           # DualKeyRotator + RotationScheduler
│   ├── audit.py              # SecretAuditLogger
│   ├── policy.py             # PolicyEnforcer + ServicePolicy
│   ├── cache.py              # SecureSecretCache
│   └── config.py             # Configuration
├── app.py                     # FastAPI application
├── tests/
│   ├── test_provider.py
│   ├── test_client.py
│   ├── test_rotation.py
│   ├── test_audit.py
│   ├── test_policy.py
│   └── test_app.py
├── requirements.txt
└── README.md

Required functionality

1. SecretsProvider + Implementations

✅ Abstract interface with get, set, delete, list
✅ EnvProvider for local development
✅ MockVaultProvider for testing (simulates Vault KV v2)
✅ CachedProvider wrapper with configurable TTL
✅ FallbackProvider that tries multiple providers in order

2. ResilientSecretsClient

✅ Read with fallback: primary → cache → env
✅ Circuit breaker for the primary provider
✅ Health check endpoint
✅ Manual cache invalidation

3. DualKeyRotator + RotationScheduler

✅ Rotation with dual-key (overlap period)
✅ Post-rotation verification
✅ Scheduler with multiple secrets
✅ Rotation status dashboard

4. SecretAuditLogger

✅ Log of each operation (create, read, update, delete, rotate)
✅ Query by secret_name, actor, action
✅ Basic anomaly detection
✅ Audit summary for dashboards

5. PolicyEnforcer

✅ Service policies with allowed_secrets and permissions
✅ Access check before delivering secrets
✅ Policy audit report

6. FastAPI Integration

✅ Dependency injection for secrets
✅ Health endpoint with secrets status
✅ Audit middleware (optional)

Implementation code

Step 1: Project setup

mkdir secrets-management-setup && cd secrets-management-setup
mkdir secrets tests

Create requirements.txt:

pydantic>=2.0
fastapi>=0.100
uvicorn>=0.20
cryptography>=41.0
schedule>=1.2
pytest>=7.0
pytest-asyncio>=0.21
httpx>=0.24

pip install -r requirements.txt

Step 2: Configuration

Create secrets/config.py:

from pydantic import BaseModel, Field


class CacheConfig(BaseModel):
    ttl_seconds: int = 300
    max_entries: int = 100


class RotationConfig(BaseModel):
    default_interval_days: int = 30
    overlap_seconds: int = 300
    verify_after_rotation: bool = True


class AuditConfig(BaseModel):
    enabled: bool = True
    log_reads: bool = True
    log_file: str = "secrets_audit.jsonl"
    anomaly_threshold_per_hour: int = 50


class CircuitBreakerConfig(BaseModel):
    failure_threshold: int = 3
    timeout_seconds: int = 60


class SecretsConfig(BaseModel):
    environment: str = "development"
    service_name: str = "ai-api-service"
    provider_type: str = "mock_vault"
    cache: CacheConfig = Field(default_factory=CacheConfig)
    rotation: RotationConfig = Field(default_factory=RotationConfig)
    audit: AuditConfig = Field(default_factory=AuditConfig)
    circuit_breaker: CircuitBreakerConfig = Field(default_factory=CircuitBreakerConfig)
    fallback_to_env: bool = True

Step 3: Secrets Provider

Create secrets/provider.py:

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


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


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

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

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

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

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


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

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

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

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

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


class MockVaultProvider(SecretsProvider):
    """Simulates HashiCorp Vault KV v2 for development and testing."""

    def __init__(self):
        self._store: dict[str, dict] = {}
        self._versions: dict[str, int] = {}

    def get(self, key: str) -> SecretResult:
        start = time.perf_counter()
        if key in self._store:
            elapsed = (time.perf_counter() - start) * 1000
            data = self._store[key]
            value = data.get("value", json.dumps(data))
            return SecretResult(
                key=key, value=value, found=True,
                provider="mock_vault",
                version=str(self._versions.get(key, 1)),
                access_time_ms=elapsed,
            )
        elapsed = (time.perf_counter() - start) * 1000
        return SecretResult(
            key=key, value=None, found=False,
            provider="mock_vault", access_time_ms=elapsed,
        )

    def set(self, key: str, value: dict) -> bool:
        self._store[key] = value
        self._versions[key] = self._versions.get(key, 0) + 1
        return True

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

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

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


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

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

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

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

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

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

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

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


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

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

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

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

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

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

Step 4: Secrets Cache

Create secrets/cache.py:

import time
import threading
from typing import Optional
from dataclasses import dataclass


@dataclass
class CachedEntry:
    key: str
    value: str
    cached_at: float
    ttl: int
    access_count: int = 0

    @property
    def is_expired(self) -> bool:
        return time.time() - self.cached_at > self.ttl


class SecureSecretCache:
    def __init__(self, default_ttl: int = 300, max_entries: int = 100):
        self._cache: dict[str, CachedEntry] = {}
        self._default_ttl = default_ttl
        self._max_entries = max_entries
        self._lock = threading.Lock()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> Optional[str]:
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                self._misses += 1
                return None
            if entry.is_expired:
                del self._cache[key]
                self._misses += 1
                return None
            entry.access_count += 1
            self._hits += 1
            return entry.value

    def set(self, key: str, value: str, ttl: Optional[int] = None):
        with self._lock:
            if len(self._cache) >= self._max_entries:
                self._evict_oldest()
            self._cache[key] = CachedEntry(
                key=key, value=value,
                cached_at=time.time(),
                ttl=ttl or self._default_ttl,
            )

    def invalidate(self, key: str):
        with self._lock:
            self._cache.pop(key, None)

    def clear(self):
        with self._lock:
            self._cache.clear()

    def _evict_oldest(self):
        if not self._cache:
            return
        oldest = min(self._cache, key=lambda k: self._cache[k].cached_at)
        del self._cache[oldest]

    def stats(self) -> dict:
        total = self._hits + self._misses
        return {
            "entries": len(self._cache),
            "max_entries": self._max_entries,
            "hits": self._hits,
            "misses": self._misses,
            "hit_rate": f"{(self._hits/total*100):.1f}%" if total > 0 else "0%",
        }

Step 5: Audit Logger

Create secrets/audit.py:

import json
import uuid
import hashlib
import logging
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Optional
from enum import Enum

logger = logging.getLogger("secrets.audit")


class AuditAction(Enum):
    CREATE = "create"
    READ = "read"
    UPDATE = "update"
    DELETE = "delete"
    ROTATE = "rotate"
    REVOKE = "revoke"
    FAILED_ACCESS = "failed_access"


@dataclass
class AuditEntry:
    entry_id: str
    timestamp: str
    action: str
    secret_name: str
    actor: str
    success: bool
    source_ip: str = "unknown"
    service: str = "unknown"
    details: dict = field(default_factory=dict)
    checksum: str = ""

    def __post_init__(self):
        if not self.checksum:
            data = f"{self.timestamp}:{self.action}:{self.secret_name}:{self.actor}"
            self.checksum = hashlib.sha256(data.encode()).hexdigest()[:16]


class SecretAuditLogger:
    def __init__(self, log_file: Optional[str] = None, enabled: bool = True):
        self._entries: list[AuditEntry] = []
        self._log_file = log_file
        self._enabled = enabled

    def log(
        self,
        action: AuditAction,
        secret_name: str,
        actor: str,
        success: bool = True,
        source_ip: str = "unknown",
        service: str = "unknown",
        details: Optional[dict] = None,
    ) -> Optional[AuditEntry]:
        if not self._enabled:
            return None

        entry = AuditEntry(
            entry_id=f"aud-{uuid.uuid4().hex[:8]}",
            timestamp=datetime.utcnow().isoformat(),
            action=action.value,
            secret_name=secret_name,
            actor=actor,
            success=success,
            source_ip=source_ip,
            service=service,
            details=details or {},
        )

        self._entries.append(entry)
        log_line = json.dumps(asdict(entry), ensure_ascii=False)
        logger.info(log_line)

        if self._log_file:
            with open(self._log_file, "a") as f:
                f.write(log_line + "\n")

        return entry

    def query(
        self,
        secret_name: Optional[str] = None,
        actor: Optional[str] = None,
        action: Optional[AuditAction] = None,
        limit: int = 100,
    ) -> list[AuditEntry]:
        results = self._entries
        if secret_name:
            results = [e for e in results if e.secret_name == secret_name]
        if actor:
            results = [e for e in results if e.actor == actor]
        if action:
            results = [e for e in results if e.action == action.value]
        return results[-limit:]

    def detect_anomalies(self, window_seconds: int = 3600) -> list[dict]:
        anomalies = []
        import time
        now = time.time()
        actor_counts: dict[str, int] = {}
        failed_counts: dict[str, int] = {}

        for entry in self._entries:
            entry_time = datetime.fromisoformat(entry.timestamp).timestamp()
            if now - entry_time > window_seconds:
                continue
            actor_counts[entry.actor] = actor_counts.get(entry.actor, 0) + 1
            if not entry.success:
                key = f"{entry.actor}:{entry.secret_name}"
                failed_counts[key] = failed_counts.get(key, 0) + 1

        for actor, count in actor_counts.items():
            if count > 50:
                anomalies.append({
                    "type": "excessive_access",
                    "actor": actor,
                    "count": count,
                    "severity": "WARNING",
                })

        for key, count in failed_counts.items():
            if count > 3:
                actor, secret = key.split(":", 1)
                anomalies.append({
                    "type": "repeated_failures",
                    "actor": actor,
                    "secret": secret,
                    "count": count,
                    "severity": "CRITICAL",
                })

        return anomalies

    def summary(self) -> dict:
        total = len(self._entries)
        by_action = {}
        failures = 0
        for entry in self._entries:
            by_action[entry.action] = by_action.get(entry.action, 0) + 1
            if not entry.success:
                failures += 1
        return {
            "total_entries": total,
            "by_action": by_action,
            "failures": failures,
            "failure_rate": f"{(failures/total*100):.1f}%" if total > 0 else "0%",
        }

Step 6: Policy Enforcer

Create secrets/policy.py:

from dataclasses import dataclass, field
from typing import Set
from enum import Enum


class Permission(Enum):
    READ = "read"
    WRITE = "write"
    ROTATE = "rotate"
    DELETE = "delete"


@dataclass
class ServicePolicy:
    service_name: str
    allowed_secrets: Set[str]
    permissions: Set[Permission] = field(default_factory=lambda: {Permission.READ})
    description: str = ""

    def can_access(self, secret_name: str, permission: Permission = Permission.READ) -> bool:
        return secret_name in self.allowed_secrets and permission in self.permissions


class PolicyEnforcer:
    def __init__(self):
        self._policies: dict[str, ServicePolicy] = {}

    def register_policy(self, policy: ServicePolicy):
        self._policies[policy.service_name] = policy

    def check_access(
        self,
        service_name: str,
        secret_name: str,
        permission: Permission = Permission.READ,
    ) -> tuple[bool, str]:
        policy = self._policies.get(service_name)
        if policy is None:
            return False, f"No policy for service '{service_name}'"
        if not policy.can_access(secret_name, permission):
            return False, f"Access denied: '{service_name}' cannot {permission.value} '{secret_name}'"
        return True, "Access granted"

    def get_service_secrets(self, service_name: str) -> list[str]:
        policy = self._policies.get(service_name)
        return sorted(policy.allowed_secrets) if policy else []

    def audit(self) -> dict:
        return {
            name: {
                "allowed": sorted(p.allowed_secrets),
                "permissions": [perm.value for perm in p.permissions],
            }
            for name, p in self._policies.items()
        }

Step 7: Rotation

Create secrets/rotation.py:

import uuid
import time
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Callable
from enum import Enum

logger = logging.getLogger("secrets.rotation")


class KeyStatus(Enum):
    ACTIVE = "active"
    DEPRECATED = "deprecated"
    REVOKED = "revoked"


@dataclass
class ManagedKey:
    key_id: str
    value: str
    status: KeyStatus
    created_at: datetime
    expires_at: Optional[datetime] = None

    @property
    def is_expired(self) -> bool:
        return self.expires_at is not None and datetime.utcnow() >= self.expires_at


@dataclass
class RotationResult:
    success: bool
    old_key_id: Optional[str]
    new_key_id: Optional[str]
    message: str
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())


class DualKeyRotator:
    def __init__(
        self,
        secret_name: str,
        overlap_seconds: int = 300,
        key_generator: Optional[Callable[[], str]] = None,
    ):
        self.secret_name = secret_name
        self.overlap_seconds = overlap_seconds
        self._key_generator = key_generator or (lambda: f"sk-{uuid.uuid4().hex}")
        self._keys: list[ManagedKey] = []
        self._history: list[RotationResult] = []

    @property
    def active_key(self) -> Optional[ManagedKey]:
        for key in self._keys:
            if key.status == KeyStatus.ACTIVE:
                return key
        return None

    @property
    def all_valid_keys(self) -> list[ManagedKey]:
        return [
            k for k in self._keys
            if k.status in (KeyStatus.ACTIVE, KeyStatus.DEPRECATED)
            and not k.is_expired
        ]

    def initialize(self, initial_value: str) -> ManagedKey:
        key = ManagedKey(
            key_id=f"key-{uuid.uuid4().hex[:8]}",
            value=initial_value,
            status=KeyStatus.ACTIVE,
            created_at=datetime.utcnow(),
        )
        self._keys.append(key)
        logger.info(f"[{self.secret_name}] Initialized: {key.key_id}")
        return key

    def rotate(self, new_value: Optional[str] = None) -> RotationResult:
        old_key = self.active_key
        if not old_key:
            return RotationResult(False, None, None, "No active key to rotate")

        new_value = new_value or self._key_generator()
        new_key = ManagedKey(
            key_id=f"key-{uuid.uuid4().hex[:8]}",
            value=new_value,
            status=KeyStatus.ACTIVE,
            created_at=datetime.utcnow(),
        )

        old_key.status = KeyStatus.DEPRECATED
        old_key.expires_at = datetime.utcnow() + timedelta(seconds=self.overlap_seconds)
        self._keys.append(new_key)

        result = RotationResult(
            True, old_key.key_id, new_key.key_id,
            f"Rotated: {old_key.key_id} → {new_key.key_id}",
        )
        self._history.append(result)
        logger.info(f"[{self.secret_name}] {result.message}")
        return result

    def cleanup_expired(self) -> int:
        revoked = 0
        for key in self._keys:
            if key.status == KeyStatus.DEPRECATED and key.is_expired:
                key.status = KeyStatus.REVOKED
                revoked += 1
        return revoked

    def status(self) -> dict:
        return {
            "secret_name": self.secret_name,
            "active_key": self.active_key.key_id if self.active_key else None,
            "valid_keys": len(self.all_valid_keys),
            "total_rotations": len(self._history),
        }


@dataclass
class ScheduledRotation:
    secret_name: str
    interval_days: int
    last_rotated: datetime
    rotator: DualKeyRotator

    @property
    def next_rotation(self) -> datetime:
        return self.last_rotated + timedelta(days=self.interval_days)

    @property
    def is_due(self) -> bool:
        return datetime.utcnow() >= self.next_rotation


class RotationScheduler:
    def __init__(self):
        self._schedules: dict[str, ScheduledRotation] = {}

    def add(self, schedule: ScheduledRotation):
        self._schedules[schedule.secret_name] = schedule

    def check_and_rotate(self) -> list[dict]:
        results = []
        for name, schedule in self._schedules.items():
            if schedule.is_due:
                result = schedule.rotator.rotate()
                if result.success:
                    schedule.last_rotated = datetime.utcnow()
                results.append({
                    "secret": name,
                    "status": "rotated" if result.success else "failed",
                    "message": result.message,
                })
        return results

    def dashboard(self) -> list[dict]:
        items = []
        for name, schedule in self._schedules.items():
            days_until = max(0, (schedule.next_rotation - datetime.utcnow()).days)
            items.append({
                "secret": name,
                "interval_days": schedule.interval_days,
                "last_rotated": schedule.last_rotated.isoformat(),
                "next_rotation": schedule.next_rotation.isoformat(),
                "is_due": schedule.is_due,
                "days_until": days_until,
            })
        return items

Step 8: Resilient Client

Create secrets/client.py:

import os
import time
import logging
from typing import Optional
from dataclasses import dataclass
from enum import Enum

from .provider import SecretsProvider, SecretResult
from .audit import SecretAuditLogger, AuditAction
from .policy import PolicyEnforcer, Permission
from .cache import SecureSecretCache
from .config import SecretsConfig

logger = logging.getLogger("secrets.client")


class FallbackLevel(Enum):
    PRIMARY = "primary"
    CACHE = "cache"
    ENV = "env"
    UNAVAILABLE = "unavailable"


@dataclass
class SecretResponse:
    key: str
    value: Optional[str]
    found: bool
    level: FallbackLevel
    cached: bool = False


class ResilientSecretsClient:
    def __init__(
        self,
        provider: Optional[SecretsProvider] = None,
        config: Optional[SecretsConfig] = None,
        audit_logger: Optional[SecretAuditLogger] = None,
        policy_enforcer: Optional[PolicyEnforcer] = None,
    ):
        self._provider = provider
        self._config = config or SecretsConfig()
        self._audit = audit_logger
        self._policy = policy_enforcer
        self._cache = SecureSecretCache(
            default_ttl=self._config.cache.ttl_seconds,
            max_entries=self._config.cache.max_entries,
        )
        self._circuit_open = False
        self._circuit_until = 0.0
        self._failure_count = 0

    def get(
        self,
        key: str,
        service_name: str = "default",
        actor: str = "unknown",
    ) -> SecretResponse:
        if self._policy:
            allowed, reason = self._policy.check_access(service_name, key)
            if not allowed:
                if self._audit:
                    self._audit.log(AuditAction.FAILED_ACCESS, key, actor,
                                   success=False, service=service_name,
                                   details={"reason": reason})
                return SecretResponse(key=key, value=None, found=False,
                                     level=FallbackLevel.UNAVAILABLE)

        cached = self._cache.get(key)
        if cached is not None:
            if self._audit:
                self._audit.log(AuditAction.READ, key, actor,
                               service=service_name, details={"source": "cache"})
            return SecretResponse(key=key, value=cached, found=True,
                                 level=FallbackLevel.CACHE, cached=True)

        if self._provider and not self._is_circuit_open():
            try:
                result = self._provider.get(key)
                if result.found:
                    self._cache.set(key, result.value)
                    self._reset_circuit()
                    if self._audit:
                        self._audit.log(AuditAction.READ, key, actor,
                                       service=service_name,
                                       details={"source": "primary"})
                    return SecretResponse(key=key, value=result.value,
                                         found=True, level=FallbackLevel.PRIMARY)
            except Exception as e:
                logger.warning(f"Primary failed for '{key}': {e}")
                self._record_failure()

        if self._config.fallback_to_env:
            env_value = os.environ.get(key)
            if env_value:
                if self._audit:
                    self._audit.log(AuditAction.READ, key, actor,
                                   service=service_name,
                                   details={"source": "env_fallback"})
                return SecretResponse(key=key, value=env_value,
                                     found=True, level=FallbackLevel.ENV)

        if self._audit:
            self._audit.log(AuditAction.FAILED_ACCESS, key, actor,
                           success=False, service=service_name,
                           details={"reason": "not found in any source"})
        return SecretResponse(key=key, value=None, found=False,
                             level=FallbackLevel.UNAVAILABLE)

    def set(self, key: str, value: dict, actor: str = "system") -> bool:
        if not self._provider:
            return False
        success = self._provider.set(key, value)
        if success:
            self._cache.invalidate(key)
            if self._audit:
                self._audit.log(AuditAction.UPDATE, key, actor)
        return success

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

    def health(self) -> dict:
        return {
            "provider": self._provider.provider_name if self._provider else "none",
            "circuit_breaker_open": self._circuit_open,
            "failure_count": self._failure_count,
            "cache_stats": self._cache.stats(),
        }

    def _is_circuit_open(self) -> bool:
        if self._circuit_open and time.time() > self._circuit_until:
            self._circuit_open = False
            self._failure_count = 0
        return self._circuit_open

    def _record_failure(self):
        self._failure_count += 1
        threshold = self._config.circuit_breaker.failure_threshold
        if self._failure_count >= threshold:
            self._circuit_open = True
            self._circuit_until = time.time() + self._config.circuit_breaker.timeout_seconds

    def _reset_circuit(self):
        self._failure_count = 0
        self._circuit_open = False

Step 9: Package init

Create secrets/__init__.py:

from .provider import (
    SecretsProvider, SecretResult,
    EnvProvider, MockVaultProvider,
    CachedProvider, FallbackProvider,
)
from .client import ResilientSecretsClient, SecretResponse, FallbackLevel
from .rotation import (
    DualKeyRotator, RotationScheduler, ScheduledRotation,
    RotationResult, ManagedKey, KeyStatus,
)
from .audit import SecretAuditLogger, AuditEntry, AuditAction
from .policy import PolicyEnforcer, ServicePolicy, Permission
from .cache import SecureSecretCache
from .config import SecretsConfig

Step 10: FastAPI Application

Create app.py:

import os
import logging
from typing import Annotated

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field

from secrets import (
    ResilientSecretsClient, MockVaultProvider, CachedProvider,
    SecretAuditLogger, AuditAction, PolicyEnforcer, ServicePolicy,
    Permission, SecretsConfig, RotationScheduler, DualKeyRotator,
    ScheduledRotation,
)
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="AI API with Secrets Management", version="1.0")

config = SecretsConfig()
vault = MockVaultProvider()
vault.set("OPENAI_API_KEY", {"value": "sk-proj-mock-openai-key-12345"})
vault.set("APP_CONFIG", {"value": "production-config"})

cached_vault = CachedProvider(vault, ttl_seconds=config.cache.ttl_seconds)
audit = SecretAuditLogger(enabled=config.audit.enabled)

policy = PolicyEnforcer()
policy.register_policy(ServicePolicy(
    service_name="api-service",
    allowed_secrets={"OPENAI_API_KEY", "APP_CONFIG"},
    permissions={Permission.READ},
))

client = ResilientSecretsClient(
    provider=cached_vault,
    config=config,
    audit_logger=audit,
    policy_enforcer=policy,
)

openai_rotator = DualKeyRotator("OPENAI_API_KEY", overlap_seconds=300)
openai_rotator.initialize("sk-proj-mock-openai-key-12345")

scheduler = RotationScheduler()
scheduler.add(ScheduledRotation(
    secret_name="OPENAI_API_KEY",
    interval_days=30,
    last_rotated=datetime.utcnow(),
    rotator=openai_rotator,
))


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=5000)


class ChatResponse(BaseModel):
    answer: str
    model: str = "gpt-4o-mini"


def get_client() -> ResilientSecretsClient:
    return client


@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    secrets: Annotated[ResilientSecretsClient, Depends(get_client)],
):
    result = secrets.get("OPENAI_API_KEY", service_name="api-service", actor="chat-endpoint")
    if not result.found:
        raise HTTPException(503, "AI service temporarily unavailable")

    return ChatResponse(
        answer=f"Processed: {request.message[:50]}...",
        model="gpt-4o-mini",
    )


@app.get("/health")
async def health(secrets: Annotated[ResilientSecretsClient, Depends(get_client)]):
    openai = secrets.get("OPENAI_API_KEY", service_name="api-service", actor="health-check")
    return {
        "status": "healthy" if openai.found else "degraded",
        "secrets_health": secrets.health(),
        "openai_available": openai.found,
        "openai_source": openai.level.value,
    }


@app.get("/internal/audit")
async def audit_summary():
    return {
        "summary": audit.summary(),
        "anomalies": audit.detect_anomalies(),
    }


@app.get("/internal/rotation")
async def rotation_status():
    return {
        "dashboard": scheduler.dashboard(),
    }


@app.post("/internal/rotation/run")
async def run_rotation():
    results = scheduler.check_and_rotate()
    return {"results": results}


@app.post("/internal/cache/clear")
async def clear_cache(secrets: Annotated[ResilientSecretsClient, Depends(get_client)]):
    secrets.clear_cache()
    return {"status": "cache cleared"}

Step 11: Tests

Create tests/test_provider.py:

import os
from secrets.provider import EnvProvider, MockVaultProvider, CachedProvider, FallbackProvider


def test_env_provider_reads_env():
    os.environ["TEST_SECRET"] = "test-value"
    provider = EnvProvider()
    result = provider.get("TEST_SECRET")
    assert result.found
    assert result.value == "test-value"
    del os.environ["TEST_SECRET"]


def test_env_provider_missing_key():
    provider = EnvProvider()
    result = provider.get("NONEXISTENT_KEY_12345")
    assert not result.found
    assert result.value is None


def test_mock_vault_crud():
    vault = MockVaultProvider()
    vault.set("test-key", {"value": "test-val"})
    result = vault.get("test-key")
    assert result.found
    assert "test-val" in result.value
    assert vault.delete("test-key")
    assert not vault.get("test-key").found


def test_cached_provider():
    vault = MockVaultProvider()
    vault.set("key1", {"value": "val1"})
    cached = CachedProvider(vault, ttl_seconds=60)

    r1 = cached.get("key1")
    assert r1.found and not r1.cached

    r2 = cached.get("key1")
    assert r2.found and r2.cached


def test_fallback_provider():
    empty = MockVaultProvider()
    full = MockVaultProvider()
    full.set("key1", {"value": "from-full"})

    fallback = FallbackProvider([empty, full])
    result = fallback.get("key1")
    assert result.found
    assert "from-full" in result.value

Create tests/test_client.py:

import os
import pytest
from secrets.client import ResilientSecretsClient, FallbackLevel
from secrets.provider import MockVaultProvider
from secrets.config import SecretsConfig


def test_client_reads_from_provider():
    vault = MockVaultProvider()
    vault.set("API_KEY", {"value": "sk-test"})
    client = ResilientSecretsClient(provider=vault)
    result = client.get("API_KEY")
    assert result.found
    assert result.level == FallbackLevel.PRIMARY


def test_client_falls_back_to_env():
    os.environ["FALLBACK_KEY"] = "from-env"
    client = ResilientSecretsClient(provider=None)
    result = client.get("FALLBACK_KEY")
    assert result.found
    assert result.level == FallbackLevel.ENV
    del os.environ["FALLBACK_KEY"]


def test_client_returns_unavailable():
    config = SecretsConfig(fallback_to_env=False)
    client = ResilientSecretsClient(provider=None, config=config)
    result = client.get("NONEXISTENT")
    assert not result.found
    assert result.level == FallbackLevel.UNAVAILABLE


def test_client_caches_values():
    vault = MockVaultProvider()
    vault.set("CACHED_KEY", {"value": "cached-val"})
    client = ResilientSecretsClient(provider=vault)

    r1 = client.get("CACHED_KEY")
    assert not r1.cached

    r2 = client.get("CACHED_KEY")
    assert r2.cached
    assert r2.level == FallbackLevel.CACHE

Create tests/test_rotation.py:

from datetime import datetime, timedelta
from secrets.rotation import DualKeyRotator, RotationScheduler, ScheduledRotation, KeyStatus


def test_rotation_creates_new_active_key():
    rotator = DualKeyRotator("test-secret", overlap_seconds=10)
    rotator.initialize("original-key")
    result = rotator.rotate("new-key")
    assert result.success
    assert rotator.active_key.value == "new-key"


def test_rotation_deprecates_old_key():
    rotator = DualKeyRotator("test-secret", overlap_seconds=10)
    rotator.initialize("original")
    rotator.rotate("new")
    deprecated = [k for k in rotator._keys if k.status == KeyStatus.DEPRECATED]
    assert len(deprecated) == 1
    assert deprecated[0].value == "original"


def test_dual_key_overlap():
    rotator = DualKeyRotator("test-secret", overlap_seconds=60)
    rotator.initialize("key-a")
    rotator.rotate("key-b")
    assert len(rotator.all_valid_keys) == 2


def test_scheduler_rotates_due_secrets():
    rotator = DualKeyRotator("due-secret", overlap_seconds=5)
    rotator.initialize("init-val")

    scheduler = RotationScheduler()
    scheduler.add(ScheduledRotation(
        secret_name="due-secret",
        interval_days=30,
        last_rotated=datetime.utcnow() - timedelta(days=31),
        rotator=rotator,
    ))

    results = scheduler.check_and_rotate()
    assert len(results) == 1
    assert results[0]["status"] == "rotated"


def test_scheduler_skips_not_due():
    rotator = DualKeyRotator("not-due", overlap_seconds=5)
    rotator.initialize("init")

    scheduler = RotationScheduler()
    scheduler.add(ScheduledRotation(
        secret_name="not-due",
        interval_days=30,
        last_rotated=datetime.utcnow(),
        rotator=rotator,
    ))

    results = scheduler.check_and_rotate()
    assert len(results) == 0

Create tests/test_audit.py:

from secrets.audit import SecretAuditLogger, AuditAction


def test_audit_logs_access():
    audit = SecretAuditLogger()
    entry = audit.log(AuditAction.READ, "test-key", "test-actor")
    assert entry is not None
    assert entry.action == "read"
    assert entry.success


def test_audit_logs_failure():
    audit = SecretAuditLogger()
    audit.log(AuditAction.FAILED_ACCESS, "key", "actor", success=False)
    summary = audit.summary()
    assert summary["failures"] == 1


def test_audit_query_by_secret():
    audit = SecretAuditLogger()
    audit.log(AuditAction.READ, "key-a", "actor1")
    audit.log(AuditAction.READ, "key-b", "actor2")
    audit.log(AuditAction.READ, "key-a", "actor3")
    results = audit.query(secret_name="key-a")
    assert len(results) == 2


def test_audit_disabled():
    audit = SecretAuditLogger(enabled=False)
    entry = audit.log(AuditAction.READ, "key", "actor")
    assert entry is None
    assert audit.summary()["total_entries"] == 0

Create tests/test_policy.py:

from secrets.policy import PolicyEnforcer, ServicePolicy, Permission


def test_policy_allows_access():
    enforcer = PolicyEnforcer()
    enforcer.register_policy(ServicePolicy(
        service_name="api",
        allowed_secrets={"openai-key"},
        permissions={Permission.READ},
    ))
    allowed, _ = enforcer.check_access("api", "openai-key")
    assert allowed


def test_policy_denies_unauthorized_secret():
    enforcer = PolicyEnforcer()
    enforcer.register_policy(ServicePolicy(
        service_name="api",
        allowed_secrets={"openai-key"},
    ))
    allowed, reason = enforcer.check_access("api", "database-password")
    assert not allowed
    assert "denied" in reason.lower() or "cannot" in reason.lower()


def test_policy_denies_unknown_service():
    enforcer = PolicyEnforcer()
    allowed, _ = enforcer.check_access("unknown", "any-key")
    assert not allowed


def test_policy_denies_wrong_permission():
    enforcer = PolicyEnforcer()
    enforcer.register_policy(ServicePolicy(
        service_name="readonly",
        allowed_secrets={"key"},
        permissions={Permission.READ},
    ))
    allowed, _ = enforcer.check_access("readonly", "key", Permission.WRITE)
    assert not allowed

Create tests/test_app.py:

import pytest
from httpx import AsyncClient, ASGITransport
from app import app


@pytest.mark.asyncio
async def test_health_endpoint():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] in ("healthy", "degraded")


@pytest.mark.asyncio
async def test_chat_endpoint():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/chat",
            json={"message": "Hello, test message"},
        )
        assert response.status_code == 200
        data = response.json()
        assert "answer" in data


@pytest.mark.asyncio
async def test_audit_endpoint():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/internal/audit")
        assert response.status_code == 200
        data = response.json()
        assert "summary" in data


@pytest.mark.asyncio
async def test_rotation_endpoint():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/internal/rotation")
        assert response.status_code == 200
        assert "dashboard" in response.json()

Running it

Run the tests

cd secrets-management-setup
pytest tests/ -v

# Expected output:
# tests/test_provider.py::test_env_provider_reads_env PASSED
# tests/test_provider.py::test_env_provider_missing_key PASSED
# tests/test_provider.py::test_mock_vault_crud PASSED
# tests/test_provider.py::test_cached_provider PASSED
# tests/test_provider.py::test_fallback_provider PASSED
# tests/test_client.py::test_client_reads_from_provider PASSED
# tests/test_client.py::test_client_falls_back_to_env PASSED
# tests/test_client.py::test_client_returns_unavailable PASSED
# tests/test_client.py::test_client_caches_values PASSED
# tests/test_rotation.py::test_rotation_creates_new_active_key PASSED
# tests/test_rotation.py::test_rotation_deprecates_old_key PASSED
# tests/test_rotation.py::test_dual_key_overlap PASSED
# tests/test_rotation.py::test_scheduler_rotates_due_secrets PASSED
# tests/test_rotation.py::test_scheduler_skips_not_due PASSED
# tests/test_audit.py::test_audit_logs_access PASSED
# tests/test_audit.py::test_audit_logs_failure PASSED
# tests/test_audit.py::test_audit_query_by_secret PASSED
# tests/test_audit.py::test_audit_disabled PASSED
# tests/test_policy.py::test_policy_allows_access PASSED
# tests/test_policy.py::test_policy_denies_unauthorized_secret PASSED
# tests/test_policy.py::test_policy_denies_unknown_service PASSED
# tests/test_policy.py::test_policy_denies_wrong_permission PASSED
# tests/test_app.py::test_health_endpoint PASSED
# tests/test_app.py::test_chat_endpoint PASSED
# tests/test_app.py::test_audit_endpoint PASSED
# tests/test_app.py::test_rotation_endpoint PASSED
# All 26 tests passed!

Run the server

uvicorn app:app --reload --port 8001

Test with curl

# Chat (uses secret from the mock vault)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "How much does the iPhone 15 cost?"}'

# Health check with secrets status
curl http://localhost:8001/health

# Audit trail
curl http://localhost:8001/internal/audit

# Rotation dashboard
curl http://localhost:8001/internal/rotation

# Clear cache
curl -X POST http://localhost:8001/internal/cache/clear

Evaluation rubric

Total: 100 points

Category	Points	Key criteria
SecretsProvider	15	Abstract interface (3), EnvProvider (3), MockVaultProvider (3), CachedProvider (3), FallbackProvider (3)
ResilientClient	20	Primary → cache → env fallback (5), circuit breaker (5), policy check (5), health check (5)
Rotation	15	DualKeyRotator (5), overlap/deprecation (3), RotationScheduler (4), dashboard (3)
AuditLogger	15	Log operations (4), query/filter (3), anomaly detection (4), summary (2), checksums (2)
PolicyEnforcer	10	Service policies (3), access check (3), audit report (2), permissions enum (2)
FastAPI Integration	10	DI for secrets (3), health endpoint (2), internal endpoints (3), error handling (2)
Tests	10	Provider tests (2), client tests (2), rotation tests (2), audit tests (2), app tests (2)
Code Quality	5	Clean structure (2), config centralized (1), typing (1), no hardcoded values (1)

Grade distribution

Range	Grade
90-100	Excellent — Production-ready setup
80-89	Very good — Solid setup with minor improvements
70-79	Good — Covers the basics but needs more robustness
60-69	Acceptable — Missing components or depth
< 60	Needs revision — Gaps in the system

Common mistakes

1. Not using an abstract interface for the provider

Without SecretsProvider as an interface, your code stays coupled to a specific provider. If you migrate from MockVault to AWS tomorrow, you have to change code in multiple files.

2. Circuit breaker that never closes

If the circuit breaker's timeout is too long (e.g., 1 hour), your app will be degraded for too long. 30-60 seconds is a good default. Verify that the half-open state works correctly.

3. Cache with no TTL or an infinite TTL

A cache with no TTL means a rotated secret is never updated in the app until the next restart. A TTL of 300-600 seconds is a good balance between latency and freshness.

4. Audit log that reveals secret values

The audit log should record the secret's name, not its value. Never log secret_value, only secret_name and metadata.

5. Policies that aren't checked on every access

If policy verification only happens at startup, a policy change doesn't apply until the restart. Verify on every get() that the service has access.

6. Fallback to env vars in production

In production, fallback_to_env should be False. If the cloud provider isn't available, the app should respond with 503 and alert, not silently fall back to env vars that might be out of date.

7. Tests that depend on external services

All tests should work without Vault, AWS, or GCP running. Use MockVaultProvider and EnvProvider for testing. Integration tests with real providers are separate.

8. Not clearing secrets from the cache on rotation

When you rotate a secret, you must invalidate the cache. Otherwise, the app keeps using the old cached key until the TTL expires.

Connection with the following modules

Your Secrets Management Setup is the fifth artifact. As you advance:

Module	How it connects
Module 6: PII Protection	Your audit logger extends to record accesses to PII data — not just secrets
Module 7: Security Testing	You test your secrets management with pen testing: can a service access secrets it shouldn't?
Module 8: Integration	Your secrets client provides the credentials for Injection Defense (M3), Sanitization (M4), and PII Protection (M6)

The most important change: where you used to have os.getenv("OPENAI_API_KEY"), you now have secrets_client.get("OPENAI_API_KEY") with caching, audit, policies, and fallback. The application code barely changes — the security infrastructure behind it changes completely.

Summary

The Secrets Management Setup is the central artifact of Module 5 — it integrates secrets provider, rotation, audit, policies, and FastAPI into a unified system
6 components work together: SecretsProvider → ResilientClient → PolicyEnforcer → SecureCache → AuditLogger → RotationScheduler
MockVaultProvider simulates Vault KV v2 for development and testing without infrastructure
ResilientSecretsClient implements tiered fallback (primary → cache → env) with a circuit breaker
DualKeyRotator runs zero-downtime rotation using overlap periods
SecretAuditLogger records every operation with checksums for tamper-evidence
PolicyEnforcer implements least privilege by controlling which service accesses which secret
Unit tests verify each component in isolation and the integrated app
This setup integrates with the M3, M4, and M6 pipelines in Module 8

Project resources

HashiCorp Vault Documentation — Reference to understand the API that MockVault simulates
AWS Secrets Manager Best Practices — Best practices applicable to any setup
FastAPI Dependencies — DI documentation in FastAPI for the integration
Pytest Documentation — Testing framework for the project's tests
Circuit Breaker Pattern — Resilience pattern implemented in the client
OWASP Secrets Management Cheat Sheet — Industry best practices
Python Cryptography Library — Reference for secrets encryption
12-Factor App — Config — Configuration principles the setup implements

Created: March 2026 Version: 1.0

← Previous7. Least Privilege and Python Integration Next →1. Introduction: Data Privacy & PII Protection