Módulo 5: Secrets Management

8. Proyecto: Secrets Management Setup

Descripción del proyecto

Este proyecto cierra el Módulo 5 con la implementación de un Secrets Management Setup completo — un sistema integrado de gestión de secrets para tu aplicación AI que incluye un secrets client con abstracción sobre el provider, rotation scheduler, audit logger, y FastAPI integration. No es un ejercicio parcial — es el artefacto de infraestructura que cambia cómo tu sistema obtiene y gestiona credenciales.

En las cápsulas anteriores construiste las piezas individuales: entendiste por qué .env no basta (02), conociste Vault como referencia (03), implementaste rotation patterns (04), exploraste cloud KMS (05), diseñaste lifecycle y audit trails (06), e integraste con FastAPI usando least privilege (07). Ahora todo se consolida en un sistema que:

Abstrae el provider: Interface unificada que funciona con Vault, AWS, GCP, Azure, o environment variables
Rota secrets: Scheduler automático con dual-key strategy y zero-downtime
Registra auditoría: Cada acceso a cada secret queda logueado con timestamps, actor, y source
Integra con FastAPI: Dependency injection para obtener secrets en endpoints
Resiste fallos: Cache con TTL, circuit breaker, y fallback levels cuando el provider no está disponible

Al terminar vas a tener un directorio con código Python ejecutable, tests unitarios, y un servidor FastAPI que demuestra el sistema en acción. Este es el quinto artefacto de la guía y se integra en el Módulo 8 (Secured AI System) como la capa de gestión de credenciales.

Objetivo del proyecto

Construir un sistema de secrets management para una aplicación AI con: abstracción del provider de secrets, rotation scheduler con zero-downtime, audit trail de cada acceso, FastAPI integration con dependency injection, y resilience cuando el secrets service no está disponible — todo en un paquete reutilizable que funcione con Vault (dev mode) o simulación local.

Especificaciones técnicas

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

Estructura del entregable

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

Funcionalidades obligatorias

1. SecretsProvider + Implementations

✅ Abstract interface con get, set, delete, list
✅ EnvProvider para desarrollo local
✅ MockVaultProvider para testing (simula Vault KV v2)
✅ CachedProvider wrapper con TTL configurable
✅ FallbackProvider que intenta múltiples providers en orden

2. ResilientSecretsClient

✅ Lectura con fallback: primary → cache → env
✅ Circuit breaker para primary provider
✅ Health check endpoint
✅ Cache invalidation manual

3. DualKeyRotator + RotationScheduler

✅ Rotation con dual-key (overlap period)
✅ Verification post-rotación
✅ Scheduler con múltiples secrets
✅ Dashboard de estado de rotación

4. SecretAuditLogger

✅ Log de cada operación (create, read, update, delete, rotate)
✅ Query por secret_name, actor, action
✅ Anomaly detection básica
✅ Audit summary para dashboards

5. PolicyEnforcer

✅ Service policies con allowed_secrets y permissions
✅ Access check antes de entregar secrets
✅ Policy audit report

6. FastAPI Integration

✅ Dependency injection para secrets
✅ Health endpoint con secrets status
✅ Audit middleware (opcional)

Código de implementación

Paso 1: Setup del proyecto

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

Crea 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

Paso 2: Configuración

Crea 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

Paso 3: Secrets Provider

Crea 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):
    """Simula HashiCorp Vault KV v2 para desarrollo y 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)})"

Paso 4: Secrets Cache

Crea 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%",
        }

Paso 5: Audit Logger

Crea 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%",
        }

Paso 6: Policy Enforcer

Crea 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()
        }

Paso 7: Rotation

Crea 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

Paso 8: Resilient Client

Crea 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

Paso 9: Package init

Crea 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

Paso 10: FastAPI Application

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

Paso 11: Tests

Crea 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

Crea 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

Crea 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

Crea 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

Crea 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

Crea 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()

Ejecución

Correr los tests

cd secrets-management-setup
pytest tests/ -v

# Output esperado:
# 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!

Correr el servidor

uvicorn app:app --reload --port 8001

Probar con curl

# Chat (usa secret del mock vault)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "¿Cuánto cuesta el iPhone 15?"}'

# Health check con estado de secrets
curl http://localhost:8001/health

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

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

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

Rúbrica de evaluación

Total: 100 puntos

Categoría	Puntos	Criterios clave
SecretsProvider	15	Interface abstracta (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)

Distribución de notas

Rango	Calificación
90-100	Excelente — Setup production-ready
80-89	Muy bien — Setup sólido con mejoras menores
70-79	Bien — Cubre lo básico pero necesita más robustez
60-69	Aceptable — Faltan componentes o profundidad
< 60	Necesita revisión — Brechas en el sistema

Errores comunes

1. No usar interface abstracta para el provider

Sin SecretsProvider como interface, tu código queda acoplado a un provider específico. Si mañana migras de MockVault a AWS, tienes que cambiar código en múltiples archivos.

2. Circuit breaker que nunca se cierra

Si el timeout del circuit breaker es muy largo (e.g., 1 hora), tu app estará degradada por demasiado tiempo. 30-60 segundos es un buen default. Verifica que el half-open state funciona correctamente.

3. Cache sin TTL o con TTL infinito

Un cache sin TTL significa que un secret rotado nunca se actualiza en la app hasta el próximo restart. TTL de 300-600 segundos es un buen balance entre latencia y actualización.

4. Audit log que revela valores de secrets

El audit log debe registrar el nombre del secret, no su valor. Nunca loguees secret_value, solo secret_name y metadata.

5. Policies que no se verifican en cada acceso

Si la verificación de policies solo ocurre al startup, un cambio de policy no aplica hasta el restart. Verifica en cada get() que el servicio tiene acceso.

6. Fallback a env vars en producción

En producción, fallback_to_env debe ser False. Si el provider cloud no está disponible, la app debe responder 503 y alertar, no caer silenciosamente a env vars que podrían estar desactualizados.

7. Tests que dependen de servicios externos

Todos los tests deben funcionar sin Vault, AWS, o GCP corriendo. Usa MockVaultProvider y EnvProvider para testing. Los tests de integración con providers reales son separados.

8. No limpiar secrets del cache en rotación

Cuando rotas un secret, debes invalidar el cache. Si no, la app sigue usando la key vieja cacheada hasta que expire el TTL.

Conexión con los módulos siguientes

Tu Secrets Management Setup es el quinto artefacto. Cuando avances:

Módulo	Cómo se conecta
Módulo 6: PII Protection	Tu audit logger se extiende para registrar accesos a datos de PII — no solo a secrets
Módulo 7: Security Testing	Testeas tu secrets management con pen testing: ¿puede un servicio acceder a secrets que no le corresponden?
Módulo 8: Integración	Tu secrets client provee las credenciales para Injection Defense (M3), Sanitization (M4), y PII Protection (M6)

El cambio más importante: donde antes tenías os.getenv("OPENAI_API_KEY"), ahora tienes secrets_client.get("OPENAI_API_KEY") con caching, audit, policies, y fallback. El código de aplicación apenas cambia — la infraestructura de seguridad detrás cambia completamente.

Resumen

El Secrets Management Setup es el artefacto central del Módulo 5 — integra secrets provider, rotation, audit, policies, y FastAPI en un sistema unificado
6 componentes trabajan juntos: SecretsProvider → ResilientClient → PolicyEnforcer → SecureCache → AuditLogger → RotationScheduler
MockVaultProvider simula Vault KV v2 para desarrollo y testing sin infraestructura
ResilientSecretsClient implementa fallback por niveles (primary → cache → env) con circuit breaker
DualKeyRotator ejecuta rotation con zero-downtime usando overlap periods
SecretAuditLogger registra cada operación con checksums para tamper-evidence
PolicyEnforcer implementa least privilege controlando qué servicio accede a qué secret
Tests unitarios verifican cada componente de forma aislada y la app integrada
Este setup se integra con los pipelines de M3, M4, y M6 en el Módulo 8

Recursos para el proyecto

HashiCorp Vault Documentation — Referencia para entender la API que MockVault simula
AWS Secrets Manager Best Practices — Mejores prácticas aplicables a cualquier setup
FastAPI Dependencies — Documentación de DI en FastAPI para la integración
Pytest Documentation — Framework de testing para los tests del proyecto
Circuit Breaker Pattern — Patrón de resilience implementado en el client
OWASP Secrets Management Cheat Sheet — Mejores prácticas de la industria
Python Cryptography Library — Referencia para encryption de secrets
12-Factor App — Config — Principios de configuración que el setup implementa

Creado: Marzo 2026 Versión: 1.0

← Anterior7. Least Privilege e Integración con Python Siguiente →1. Introducción: Data Privacy & PII Protection