Módulo 5: Secrets Management

7. Least Privilege e Integración con Python

Siguiente →

Descripción

Has construido las piezas de secrets management: almacenamiento seguro (03, 05), rotación automática (04), lifecycle management (06), y audit trails (06). Ahora necesitas integrar todo en tu aplicación Python de forma que cada servicio solo acceda a los secrets que necesita (least privilege), la integración con FastAPI sea limpia (dependency injection), y el sistema sea resiliente cuando el secrets service falla (fallback y caching).

El principio de least privilege es simple de entender pero difícil de implementar consistentemente: cada componente de tu sistema debe tener acceso solo a los secrets que necesita para funcionar — y nada más. Tu API service necesita la API key de OpenAI, pero no necesita la password de la base de datos del analytics. Tu migration script necesita acceso a la base de datos, pero no necesita la API key de Anthropic. Violar least privilege significa que comprometer un componente expone secrets que ese componente no debería conocer.

En esta cápsula vas a implementar scoped access, integrar secrets management con FastAPI usando dependency injection, construir patrones de fallback robustos, y configurar secrets por environment (dev/staging/prod). Todo el código de esta cápsula se integra directamente en el proyecto de la cápsula 08.

Least Privilege: el principio

Sin Least Privilege                    Con Least Privilege
──────────────────                     ──────────────────────

┌─────────────────┐                    ┌─────────────────┐
│   API Service   │                    │   API Service   │
│                 │                    │                 │
│ OPENAI_KEY   ✅ │                    │ OPENAI_KEY   ✅ │
│ ANTHRO_KEY   ✅ │                    │ ANTHRO_KEY   ❌ │
│ DB_PASSWORD  ✅ │ ← Todos ven       │ DB_PASSWORD  ❌ │ ← Solo lo que
│ REDIS_URL    ✅ │   todo             │ REDIS_URL    ❌ │   necesita
│ STRIPE_KEY   ✅ │                    │ STRIPE_KEY   ❌ │
│ ADMIN_PASS   ✅ │                    │ ADMIN_PASS   ❌ │
└─────────────────┘                    └─────────────────┘

Implementación de scoped access

import json
from dataclasses import dataclass, field
from typing import Optional, 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:
    """Enforce least privilege policies para acceso a secrets."""

    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 defined for service '{service_name}'"

        if not policy.can_access(secret_name, permission):
            if secret_name not in policy.allowed_secrets:
                return False, f"Service '{service_name}' is not allowed to access '{secret_name}'"
            return False, f"Service '{service_name}' lacks '{permission.value}' permission for '{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_policies(self) -> dict:
        report = {}
        for name, policy in self._policies.items():
            report[name] = {
                "allowed_secrets": sorted(policy.allowed_secrets),
                "permissions": [p.value for p in policy.permissions],
                "secret_count": len(policy.allowed_secrets),
            }
        return report


enforcer = PolicyEnforcer()

enforcer.register_policy(ServicePolicy(
    service_name="api-service",
    allowed_secrets={"openai-api-key", "app-config"},
    permissions={Permission.READ},
    description="Main API service — needs LLM key and app config",
))

enforcer.register_policy(ServicePolicy(
    service_name="worker-service",
    allowed_secrets={"database-url", "redis-url"},
    permissions={Permission.READ},
    description="Background worker — needs DB and cache access",
))

enforcer.register_policy(ServicePolicy(
    service_name="migration-service",
    allowed_secrets={"database-url"},
    permissions={Permission.READ},
    description="Database migrations — needs DB access only",
))

enforcer.register_policy(ServicePolicy(
    service_name="rotation-scheduler",
    allowed_secrets={"openai-api-key", "anthropic-api-key", "database-url"},
    permissions={Permission.READ, Permission.WRITE, Permission.ROTATE},
    description="Rotation service — needs read/write/rotate for managed secrets",
))

test_cases = [
    ("api-service", "openai-api-key", Permission.READ),
    ("api-service", "database-url", Permission.READ),
    ("worker-service", "database-url", Permission.READ),
    ("worker-service", "openai-api-key", Permission.READ),
    ("migration-service", "database-url", Permission.READ),
    ("migration-service", "redis-url", Permission.READ),
    ("rotation-scheduler", "openai-api-key", Permission.ROTATE),
    ("unknown-service", "openai-api-key", Permission.READ),
]

print("Access Control Tests:")
for service, secret, perm in test_cases:
    allowed, reason = enforcer.check_access(service, secret, perm)
    status = "✅ ALLOW" if allowed else "❌ DENY"
    print(f"  {status} | {service} → {secret} ({perm.value})")
    if not allowed:
        print(f"         Reason: {reason}")

print(f"\nPolicy Audit:")
print(json.dumps(enforcer.audit_policies(), indent=2))
# Output esperado:
# Access Control Tests:
#   ✅ ALLOW | api-service → openai-api-key (read)
#   ❌ DENY  | api-service → database-url (read)
#          Reason: Service 'api-service' is not allowed to access 'database-url'
#   ✅ ALLOW | worker-service → database-url (read)
#   ❌ DENY  | worker-service → openai-api-key (read)
#   ...

Per-Service Keys: un patrón clave

En lugar de una API key de OpenAI compartida por todos los servicios, crea keys separadas por servicio:

per_service_keys = {
    "concept": "Cada servicio tiene su propia API key para el mismo proveedor",
    "benefits": [
        "Si un servicio se compromete, solo se revoca su key",
        "Audit trail muestra qué servicio hizo cada llamada",
        "Rate limits y costos son atribuibles por servicio",
        "Rotación de una key no afecta a otros servicios",
    ],
    "implementation": {
        "api-service": {
            "openai_key": "sk-proj-api-service-abc123",
            "scope": "chat completions, embeddings",
            "rate_limit": "500 RPM",
        },
        "batch-processor": {
            "openai_key": "sk-proj-batch-proc-def456",
            "scope": "embeddings only",
            "rate_limit": "1000 RPM",
        },
        "evaluation-service": {
            "openai_key": "sk-proj-eval-svc-ghi789",
            "scope": "chat completions (eval model)",
            "rate_limit": "100 RPM",
        },
    },
}

print("Per-Service Keys:")
for service, config in per_service_keys["implementation"].items():
    print(f"\n  {service}:")
    print(f"    Key: {config['openai_key'][:20]}...")
    print(f"    Scope: {config['scope']}")
    print(f"    Rate limit: {config['rate_limit']}")

print("\nBenefits:")
for benefit in per_service_keys["benefits"]:
    print(f"  ✅ {benefit}")

Integración con FastAPI: Dependency Injection

La forma más limpia de integrar secrets management con FastAPI es usando dependency injection. Los secrets se inyectan como dependencias en los endpoints que los necesitan:

import os
import json
import time
import logging
from typing import Optional, Annotated
from dataclasses import dataclass
from functools import lru_cache

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

logger = logging.getLogger("secrets_di")


@dataclass
class AppSecrets:
    """Container de secrets para la aplicación."""
    openai_api_key: str
    provider: str
    cached: bool = False


class SecretsService:
    """Servicio de secrets con cache y fallback."""

    def __init__(
        self,
        provider=None,
        cache_ttl: int = 300,
        service_name: str = "api-service",
    ):
        self._provider = provider
        self._cache_ttl = cache_ttl
        self._service_name = service_name
        self._cache: dict[str, tuple[str, float]] = {}

    def get_secret(self, key: str) -> Optional[str]:
        cached = self._get_cached(key)
        if cached is not None:
            return cached

        if self._provider:
            try:
                result = self._provider.get(key)
                if result.found:
                    self._set_cache(key, result.value)
                    return result.value
            except Exception as e:
                logger.warning(f"Provider failed for '{key}': {e}")

        env_value = os.environ.get(key)
        if env_value:
            logger.info(f"Fallback to env for '{key}'")
            return env_value

        return None

    def _get_cached(self, key: str) -> Optional[str]:
        if key in self._cache:
            value, cached_at = self._cache[key]
            if time.time() - cached_at < self._cache_ttl:
                return value
            del self._cache[key]
        return None

    def _set_cache(self, key: str, value: str):
        self._cache[key] = (value, time.time())

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


secrets_service: Optional[SecretsService] = None


def get_secrets_service() -> SecretsService:
    global secrets_service
    if secrets_service is None:
        secrets_service = SecretsService(service_name="api-service")
    return secrets_service


def get_openai_key(
    svc: Annotated[SecretsService, Depends(get_secrets_service)],
) -> str:
    key = svc.get_secret("OPENAI_API_KEY")
    if not key:
        raise HTTPException(status_code=503, detail="Service temporarily unavailable")
    return key


def get_app_secrets(
    svc: Annotated[SecretsService, Depends(get_secrets_service)],
) -> AppSecrets:
    openai_key = svc.get_secret("OPENAI_API_KEY")
    if not openai_key:
        raise HTTPException(status_code=503, detail="Service temporarily unavailable")
    return AppSecrets(
        openai_api_key=openai_key,
        provider="env_fallback",
    )


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


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


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


@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    openai_key: Annotated[str, Depends(get_openai_key)],
):
    # openai_key se inyecta automáticamente desde el secrets service
    # No necesitas os.getenv() ni acceder al .env directamente
    return ChatResponse(
        answer=f"Response to: {request.message[:50]}...",
        model="gpt-4o-mini",
    )


@app.post("/chat-v2", response_model=ChatResponse)
async def chat_v2(
    request: ChatRequest,
    secrets: Annotated[AppSecrets, Depends(get_app_secrets)],
):
    # secrets contiene todos los secrets que este endpoint necesita
    # Cada endpoint declara explícitamente qué secrets usa
    return ChatResponse(
        answer=f"Response using {secrets.provider}",
        model="gpt-4o-mini",
    )


@app.get("/health")
async def health(
    svc: Annotated[SecretsService, Depends(get_secrets_service)],
):
    openai_key = svc.get_secret("OPENAI_API_KEY")
    return {
        "status": "ok" if openai_key else "degraded",
        "secrets_available": openai_key is not None,
    }

Middleware de audit para secrets access

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response


class SecretsAuditMiddleware(BaseHTTPMiddleware):
    """Middleware que registra acceso a secrets por request."""

    def __init__(self, app, audit_logger=None):
        super().__init__(app)
        self.audit_logger = audit_logger

    async def dispatch(self, request: Request, call_next):
        request.state.secrets_accessed = []
        request.state.request_id = f"req-{time.time_ns() % 1000000:06d}"

        response = await call_next(request)

        if self.audit_logger and request.state.secrets_accessed:
            for secret_name in request.state.secrets_accessed:
                self.audit_logger.log(
                    action="read",
                    secret_name=secret_name,
                    actor=f"endpoint:{request.url.path}",
                    source_ip=request.client.host if request.client else "unknown",
                    details={"request_id": request.state.request_id},
                )

        return response

Fallback y Resilience

¿Qué pasa cuando el secrets service no está disponible? Tu app necesita un plan B:

import time
import logging
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum

logger = logging.getLogger("resilience")


class FallbackLevel(Enum):
    PRIMARY = "primary"
    CACHE = "cache"
    ENCRYPTED_LOCAL = "encrypted_local"
    ENV_VARS = "env_vars"
    UNAVAILABLE = "unavailable"


@dataclass
class FallbackResult:
    value: Optional[str]
    level: FallbackLevel
    message: str


class ResilientSecretsClient:
    """Cliente de secrets con múltiples niveles de fallback."""

    def __init__(
        self,
        primary_provider=None,
        encrypted_local_path: Optional[str] = None,
        cache_ttl_seconds: int = 600,
    ):
        self._primary = primary_provider
        self._local_path = encrypted_local_path
        self._cache_ttl = cache_ttl_seconds
        self._cache: dict[str, tuple[str, float]] = {}
        self._circuit_breaker_open = False
        self._circuit_breaker_until = 0.0
        self._failure_count = 0
        self._failure_threshold = 3
        self._circuit_timeout = 60

    def get(self, key: str) -> FallbackResult:
        if self._primary and not self._is_circuit_open():
            try:
                result = self._primary.get(key)
                if result.found:
                    self._cache[key] = (result.value, time.time())
                    self._reset_circuit()
                    return FallbackResult(result.value, FallbackLevel.PRIMARY, "From primary provider")
            except Exception as e:
                logger.warning(f"Primary provider failed: {e}")
                self._record_failure()

        if key in self._cache:
            value, cached_at = self._cache[key]
            if time.time() - cached_at < self._cache_ttl:
                return FallbackResult(value, FallbackLevel.CACHE, "From cache (primary unavailable)")

        import os
        env_value = os.environ.get(key)
        if env_value:
            return FallbackResult(env_value, FallbackLevel.ENV_VARS, "Fallback to environment variable")

        return FallbackResult(None, FallbackLevel.UNAVAILABLE, f"Secret '{key}' unavailable from all sources")

    def _is_circuit_open(self) -> bool:
        if self._circuit_breaker_open:
            if time.time() > self._circuit_breaker_until:
                self._circuit_breaker_open = False
                self._failure_count = 0
                logger.info("Circuit breaker: half-open, testing primary")
                return False
            return True
        return False

    def _record_failure(self):
        self._failure_count += 1
        if self._failure_count >= self._failure_threshold:
            self._circuit_breaker_open = True
            self._circuit_breaker_until = time.time() + self._circuit_timeout
            logger.warning(
                f"Circuit breaker OPEN — primary disabled for {self._circuit_timeout}s"
            )

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

    def health_check(self) -> dict:
        return {
            "primary_available": self._primary is not None and not self._is_circuit_open(),
            "circuit_breaker_open": self._circuit_breaker_open,
            "failure_count": self._failure_count,
            "cache_size": len(self._cache),
        }


import os
os.environ["OPENAI_API_KEY"] = "sk-proj-fallback-from-env"

client = ResilientSecretsClient(primary_provider=None, cache_ttl_seconds=300)

result = client.get("OPENAI_API_KEY")
print(f"Value: {result.value[:20]}...")
print(f"Level: {result.level.value}")
print(f"Message: {result.message}")
print(f"Health: {json.dumps(client.health_check())}")
# Output esperado:
# Value: sk-proj-fallback-f...
# Level: env_vars
# Message: Fallback to environment variable
# Health: {"primary_available": false, "circuit_breaker_open": false, ...}

Environment-Specific Configuration

Diferentes environments necesitan diferentes sources de secrets y diferentes niveles de seguridad:

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


class Environment(Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"


@dataclass
class EnvironmentConfig:
    environment: Environment
    secrets_provider: str
    cache_ttl: int
    rotation_enabled: bool
    audit_enabled: bool
    fallback_to_env: bool
    description: str


ENV_CONFIGS = {
    Environment.DEVELOPMENT: EnvironmentConfig(
        environment=Environment.DEVELOPMENT,
        secrets_provider="env",
        cache_ttl=0,
        rotation_enabled=False,
        audit_enabled=False,
        fallback_to_env=True,
        description="Dev: .env file, no rotation, no audit",
    ),
    Environment.STAGING: EnvironmentConfig(
        environment=Environment.STAGING,
        secrets_provider="aws",
        cache_ttl=300,
        rotation_enabled=True,
        audit_enabled=True,
        fallback_to_env=True,
        description="Staging: Cloud KMS with env fallback",
    ),
    Environment.PRODUCTION: EnvironmentConfig(
        environment=Environment.PRODUCTION,
        secrets_provider="aws",
        cache_ttl=600,
        rotation_enabled=True,
        audit_enabled=True,
        fallback_to_env=False,
        description="Prod: Cloud KMS only, no env fallback",
    ),
}


def get_environment() -> Environment:
    env_str = os.environ.get("APP_ENV", "development").lower()
    try:
        return Environment(env_str)
    except ValueError:
        return Environment.DEVELOPMENT


def create_configured_client(env: Optional[Environment] = None) -> ResilientSecretsClient:
    """Factory que crea el secrets client apropiado para el environment."""
    if env is None:
        env = get_environment()

    config = ENV_CONFIGS[env]

    primary = None
    if config.secrets_provider == "aws":
        try:
            primary = None  # AWSProvider(region="us-east-1") en producción real
        except Exception:
            logger.warning("AWS provider unavailable, will use fallback")

    return ResilientSecretsClient(
        primary_provider=primary,
        cache_ttl_seconds=config.cache_ttl,
    )


for env, config in ENV_CONFIGS.items():
    print(f"\n{env.value}:")
    print(f"  Provider: {config.secrets_provider}")
    print(f"  Cache TTL: {config.cache_ttl}s")
    print(f"  Rotation: {'✅' if config.rotation_enabled else '❌'}")
    print(f"  Audit: {'✅' if config.audit_enabled else '❌'}")
    print(f"  Env fallback: {'✅' if config.fallback_to_env else '❌'}")

Caching secrets de forma segura

Cachear secrets reduce latencia y dependencia del provider, pero introduce riesgos si no se hace correctamente:

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


@dataclass
class CachedSecret:
    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

    @property
    def age_seconds(self) -> float:
        return time.time() - self.cached_at


class SecureSecretCache:
    """Cache de secrets con TTL, limits, y cleanup automático."""

    def __init__(self, default_ttl: int = 300, max_entries: int = 100):
        self._cache: dict[str, CachedSecret] = {}
        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] = CachedSecret(
                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()
            self._hits = 0
            self._misses = 0

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

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


cache = SecureSecretCache(default_ttl=10, max_entries=5)

cache.set("openai-key", "sk-proj-abc123")
cache.set("anthropic-key", "sk-ant-def456")

print(f"Get openai-key: {cache.get('openai-key')[:15]}...")
print(f"Get missing: {cache.get('nonexistent')}")
print(f"Stats: {cache.stats()}")
# Output esperado:
# Get openai-key: sk-proj-abc123...
# Get missing: None
# Stats: {'entries': 2, 'max_entries': 5, 'hits': 1, 'misses': 1, 'hit_rate': '50.0%'}

IAM Policies para LLM Access

Configura IAM policies que restrinjan el acceso a secrets por servicio:

iam_policies = {
    "aws_iam_for_api_service": {
        "description": "IAM policy para el API service — solo lectura de LLM keys",
        "policy_json": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": ["secretsmanager:GetSecretValue"],
                    "Resource": [
                        "arn:aws:secretsmanager:*:*:secret:prod/llm/openai-*",
                        "arn:aws:secretsmanager:*:*:secret:prod/app/config-*",
                    ],
                },
                {
                    "Effect": "Deny",
                    "Action": ["secretsmanager:*"],
                    "Resource": [
                        "arn:aws:secretsmanager:*:*:secret:prod/database/*",
                        "arn:aws:secretsmanager:*:*:secret:prod/admin/*",
                    ],
                },
            ],
        },
    },
    "aws_iam_for_rotation_service": {
        "description": "IAM policy para rotation — read + write de secrets gestionados",
        "policy_json": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "secretsmanager:GetSecretValue",
                        "secretsmanager:PutSecretValue",
                        "secretsmanager:UpdateSecret",
                        "secretsmanager:DescribeSecret",
                    ],
                    "Resource": [
                        "arn:aws:secretsmanager:*:*:secret:prod/llm/*",
                    ],
                },
            ],
        },
    },
}

for policy_name, info in iam_policies.items():
    print(f"\n=== {policy_name} ===")
    print(f"Description: {info['description']}")
    print(f"Policy:\n{json.dumps(info['policy_json'], indent=2)}")

Patrón completo: FastAPI + Secrets + Least Privilege

Aquí está el patrón completo que combina todo lo anterior:

import os
import json
import time
import logging
from typing import Annotated, Optional
from functools import lru_cache

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

logger = logging.getLogger("app")


class AppConfig(BaseModel):
    environment: str = "development"
    secrets_cache_ttl: int = 300
    service_name: str = "ai-api-service"


@lru_cache()
def get_config() -> AppConfig:
    return AppConfig(environment=os.getenv("APP_ENV", "development"))


class SecureAPIApp:
    """Aplicación FastAPI con secrets management integrado."""

    def __init__(self, config: AppConfig):
        self.config = config
        self.app = FastAPI(title="Secure AI API")
        self._secrets_svc = SecretsService(
            service_name=config.service_name,
            cache_ttl=config.secrets_cache_ttl,
        )
        self._setup_routes()

    def _setup_routes(self):
        app = self.app
        secrets_svc = self._secrets_svc

        def get_secrets() -> SecretsService:
            return secrets_svc

        @app.post("/api/chat")
        async def chat(
            message: str,
            svc: Annotated[SecretsService, Depends(get_secrets)],
        ):
            api_key = svc.get_secret("OPENAI_API_KEY")
            if not api_key:
                raise HTTPException(503, "AI service temporarily unavailable")

            return {"response": f"Processed: {message[:30]}...", "service": "ai-api"}

        @app.get("/api/health")
        async def health(svc: Annotated[SecretsService, Depends(get_secrets)]):
            checks = {
                "openai_key": svc.get_secret("OPENAI_API_KEY") is not None,
                "service": self.config.service_name,
                "environment": self.config.environment,
            }
            status = "healthy" if all(v for k, v in checks.items() if k != "service" and k != "environment") else "degraded"
            return {"status": status, **checks}

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


config = get_config()
secure_app = SecureAPIApp(config)
# uvicorn:  app = secure_app.app

Troubleshooting

"La dependency injection hace que cada request lea el secret"

Usa caching en el SecretsService. Con un TTL de 300 segundos, solo una de cada N requests lee del provider real. El cache absorbe el tráfico repetido.

"El circuit breaker se abre y nunca se cierra"

Verifica el _circuit_timeout. Cuando el circuit breaker está open, espera hasta que pase el timeout para intentar de nuevo (half-open state). Si el provider sigue fallando, se abre de nuevo.

"No sé qué permissions dar a cada servicio"

Empieza con READ only para todos los servicios. Solo agrega WRITE y ROTATE a los servicios que explícitamente lo necesitan (rotation scheduler, admin CLI). Menos es más.

"El fallback a env vars en producción es inseguro"

Correcto. En producción, fallback_to_env debe ser False. El fallback a env vars es solo para desarrollo y staging. Si el provider cloud no está disponible en producción, la app debe responder con 503 y alertar al equipo.

"El cache de secrets ocupa mucha memoria"

SecureSecretCache tiene max_entries configurable. Con 100 entries y secrets de ~100 bytes cada uno, el consumo es ~10KB. No es un problema en la práctica.

Ejercicios

Ejercicio 1: Crea policies para 4 servicios diferentes

Define policies de least privilege para: API gateway, ML inference service, data pipeline, y admin dashboard:

Ver solución 
enforcer = PolicyEnforcer()

services = [
    ServicePolicy("api-gateway", {"openai-api-key", "rate-limit-config"}, {Permission.READ},
                  "Gateway: LLM key + config"),
    ServicePolicy("ml-inference", {"openai-api-key", "model-config"}, {Permission.READ},
                  "ML: LLM key + model config"),
    ServicePolicy("data-pipeline", {"database-url", "s3-credentials"}, {Permission.READ},
                  "Pipeline: DB + storage"),
    ServicePolicy("admin-dashboard", {"database-url", "admin-token"}, {Permission.READ, Permission.WRITE},
                  "Admin: DB + admin token with write"),
]

for policy in services:
    enforcer.register_policy(policy)

print(json.dumps(enforcer.audit_policies(), indent=2))

Ejercicio 2: Implementa un health endpoint con secrets status

Crea un endpoint que muestre el estado de cada secret que el servicio necesita:

Ver solución 
def detailed_health(secrets_svc: SecretsService, required_secrets: list[str]) -> dict:
    checks = {}
    all_ok = True
    for key in required_secrets:
        value = secrets_svc.get_secret(key)
        available = value is not None
        checks[key] = {
            "available": available,
            "key_preview": f"{value[:4]}..." if value else None,
        }
        if not available:
            all_ok = False

    return {
        "status": "healthy" if all_ok else "degraded",
        "checks": checks,
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }

svc = SecretsService()
os.environ["OPENAI_API_KEY"] = "sk-test"
result = detailed_health(svc, ["OPENAI_API_KEY", "MISSING_KEY"])
print(json.dumps(result, indent=2))

Ejercicio 3: Implementa rate limiting para accesos a secrets

Previene que un servicio haga demasiados accesos a secrets en poco tiempo:

Ver solución 
class RateLimitedSecretsService(SecretsService):
    def __init__(self, max_reads_per_minute: int = 60, **kwargs):
        super().__init__(**kwargs)
        self._max_rpm = max_reads_per_minute
        self._access_times: list[float] = []

    def get_secret(self, key: str) -> Optional[str]:
        now = time.time()
        self._access_times = [t for t in self._access_times if now - t < 60]

        if len(self._access_times) >= self._max_rpm:
            logger.warning(f"Rate limit exceeded: {len(self._access_times)} reads/min")
            raise Exception("Secret access rate limit exceeded")

        self._access_times.append(now)
        return super().get_secret(key)

Ejercicio 4: Crea un decorator @inject_secret para FastAPI

Crea un decorator reutilizable que inyecte un secret específico en un endpoint:

Ver solución 
from functools import wraps

def inject_secret(secret_key: str, param_name: str = "api_key"):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            svc = get_secrets_service()
            value = svc.get_secret(secret_key)
            if not value:
                raise HTTPException(503, "Required secret unavailable")
            kwargs[param_name] = value
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Uso:
# @app.post("/generate")
# @inject_secret("OPENAI_API_KEY", "api_key")
# async def generate(prompt: str, api_key: str = ""):
#     ...

Resumen

  • Least privilege significa que cada servicio solo accede a los secrets que necesita — implementado con PolicyEnforcer y ServicePolicy
  • Per-service keys eliminan el problema de credenciales compartidas: cada servicio tiene su propia API key del mismo proveedor
  • La integración con FastAPI usa dependency injection: los secrets se inyectan como dependencias, no se leen con os.getenv()
  • Fallback y resilience con ResilientSecretsClient: primary provider → cache → env vars, con circuit breaker para evitar cascading failures
  • Environment-specific config distingue entre dev (.env), staging (cloud + fallback), y prod (cloud only, no fallback)
  • Caching seguro con SecureSecretCache: TTL configurable, max entries, thread-safe, eviction automática
  • IAM policies (AWS, GCP, Azure) implementan least privilege a nivel de infraestructura — complementan las policies de aplicación
  • El patrón completo combina: config por environment → secrets service con cache → dependency injection → audit middleware

Próxima cápsula: En la cápsula 08 vas a construir el proyecto Secrets Management Setup completo: un sistema integrado con secrets client abstraction, rotation scheduler, audit logger, FastAPI integration, y testing con mock vault — todo como un artefacto reutilizable y production-ready.

Recursos

  1. OWASP Least Privilege Principle — Guía de OWASP para implementar least privilege en aplicaciones
  2. FastAPI Dependencies — Documentación oficial de dependency injection en FastAPI
  3. AWS IAM Best Practices — Mejores prácticas de IAM para least privilege en AWS
  4. Circuit Breaker Pattern — Patrón de resilience para servicios externos
  5. Python functools.lru_cache — Caching built-in de Python para singletons
  6. GCP IAM Conditions — Condiciones avanzadas de IAM para access control granular
  7. 12-Factor App — Backing Services — Principio de servicios de respaldo como recursos adjuntos
  8. OpenAI API Key Management — Prácticas de seguridad oficiales de OpenAI para API keys

Creado: Marzo 2026 Versión: 1.0

← Anterior6. Token Lifecycle y Audit TrailsSiguiente →8. Proyecto: Secrets Management Setup