Module 5: Secrets Management

7. Least Privilege and Python Integration

Next →

Overview

You've built the pieces of secrets management: secure storage (03, 05), automatic rotation (04), lifecycle management (06), and audit trails (06). Now you need to integrate everything into your Python application so that each service only accesses the secrets it needs (least privilege), the FastAPI integration is clean (dependency injection), and the system is resilient when the secrets service fails (fallback and caching).

The least privilege principle is simple to understand but hard to implement consistently: every component of your system should have access only to the secrets it needs to function — and nothing more. Your API service needs the OpenAI API key, but it doesn't need the analytics database password. Your migration script needs database access, but it doesn't need the Anthropic API key. Violating least privilege means that compromising one component exposes secrets that component should never have known.

In this capsule you'll implement scoped access, integrate secrets management with FastAPI using dependency injection, build robust fallback patterns, and configure secrets per environment (dev/staging/prod). All the code in this capsule integrates directly into the capsule 08 project.

Least Privilege: the principle

Without Least Privilege                With Least Privilege
───────────────────────                ────────────────────

┌─────────────────┐                    ┌─────────────────┐
│   API Service   │                    │   API Service   │
│                 │                    │                 │
│ OPENAI_KEY   ✅ │                    │ OPENAI_KEY   ✅ │
│ ANTHRO_KEY   ✅ │                    │ ANTHRO_KEY   ❌ │
│ DB_PASSWORD  ✅ │ ← Everyone sees    │ DB_PASSWORD  ❌ │ ← Only what
│ REDIS_URL    ✅ │   everything       │ REDIS_URL    ❌ │   it needs
│ STRIPE_KEY   ✅ │                    │ STRIPE_KEY   ❌ │
│ ADMIN_PASS   ✅ │                    │ ADMIN_PASS   ❌ │
└─────────────────┘                    └─────────────────┘

Scoped access implementation

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:
    """Enforces least privilege policies for access to 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))
# Expected output:
# 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: a key pattern

Instead of one OpenAI API key shared by all services, create separate keys per service:

per_service_keys = {
    "concept": "Each service has its own API key for the same provider",
    "benefits": [
        "If a service is compromised, only its key is revoked",
        "Audit trail shows which service made each call",
        "Rate limits and costs are attributable per service",
        "Rotating one key doesn't affect other services",
    ],
    "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}")

FastAPI Integration: Dependency Injection

The cleanest way to integrate secrets management with FastAPI is using dependency injection. Secrets are injected as dependencies into the endpoints that need them:

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 of secrets for the application."""
    openai_api_key: str
    provider: str
    cached: bool = False


class SecretsService:
    """Secrets service with cache and 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 is injected automatically from the secrets service
    # You don't need os.getenv() or to access .env directly
    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 contains all the secrets this endpoint needs
    # Each endpoint explicitly declares which secrets it uses
    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,
    }

Audit middleware for secrets access

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


class SecretsAuditMiddleware(BaseHTTPMiddleware):
    """Middleware that records secret access per 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 and Resilience

What happens when the secrets service isn't available? Your app needs a 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:
    """Secrets client with multiple fallback levels."""

    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())}")
# Expected output:
# Value: sk-proj-fallback-fro...
# Level: env_vars
# Message: Fallback to environment variable
# Health: {"primary_available": false, "circuit_breaker_open": false, ...}

Environment-Specific Configuration

Different environments need different secret sources and different security levels:

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 that creates the appropriate secrets client for the 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") in real production
        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 securely

Caching secrets reduces latency and provider dependency, but introduces risks if not done correctly:

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:
    """Secrets cache with TTL, limits, and automatic cleanup."""

    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()}")
# Expected output:
# 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 for LLM Access

Configure IAM policies that restrict access to secrets per service:

iam_policies = {
    "aws_iam_for_api_service": {
        "description": "IAM policy for the API service — read-only for 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 for rotation — read + write for managed secrets",
        "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)}")

Complete pattern: FastAPI + Secrets + Least Privilege

Here's the complete pattern that combines everything above:

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:
    """FastAPI application with integrated secrets management."""

    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

"Dependency injection makes every request read the secret"

Use caching in the SecretsService. With a TTL of 300 seconds, only one in every N requests reads from the real provider. The cache absorbs the repeated traffic.

"The circuit breaker opens and never closes"

Check the _circuit_timeout. When the circuit breaker is open, it waits until the timeout passes before trying again (half-open state). If the provider keeps failing, it opens again.

"I don't know what permissions to give each service"

Start with READ only for all services. Only add WRITE and ROTATE to services that explicitly need them (rotation scheduler, admin CLI). Less is more.

"Fallback to env vars in production is insecure"

Correct. In production, fallback_to_env should be False. Fallback to env vars is only for development and staging. If the cloud provider isn't available in production, the app must respond with 503 and alert the team.

"The secrets cache uses too much memory"

SecureSecretCache has a configurable max_entries. With 100 entries and secrets of ~100 bytes each, consumption is ~10KB. It's not a problem in practice.

Exercises

Exercise 1: Create policies for 4 different services

Define least privilege policies for: API gateway, ML inference service, data pipeline, and admin dashboard:

See solution 
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))

Exercise 2: Implement a health endpoint with secrets status

Create an endpoint that shows the status of each secret the service needs:

See solution 
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))

Exercise 3: Implement rate limiting for secret accesses

Prevent a service from making too many secret accesses in a short time:

See solution 
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)

Exercise 4: Create an @inject_secret decorator for FastAPI

Create a reusable decorator that injects a specific secret into an endpoint:

See solution 
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

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

Summary

  • Least privilege means each service only accesses the secrets it needs — implemented with PolicyEnforcer and ServicePolicy
  • Per-service keys eliminate the shared-credentials problem: each service has its own API key for the same provider
  • The FastAPI integration uses dependency injection: secrets are injected as dependencies, not read with os.getenv()
  • Fallback and resilience with ResilientSecretsClient: primary provider → cache → env vars, with a circuit breaker to avoid cascading failures
  • Environment-specific config distinguishes between dev (.env), staging (cloud + fallback), and prod (cloud only, no fallback)
  • Secure caching with SecureSecretCache: configurable TTL, max entries, thread-safe, automatic eviction
  • IAM policies (AWS, GCP, Azure) implement least privilege at the infrastructure level — they complement application policies
  • The complete pattern combines: config per environment → secrets service with cache → dependency injection → audit middleware

Next capsule: In capsule 08 you'll build the complete Secrets Management Setup project: an integrated system with secrets client abstraction, rotation scheduler, audit logger, FastAPI integration, and testing with a mock vault — all as a reusable, production-ready artifact.

Resources

  1. OWASP Least Privilege Principle — OWASP guide to implementing least privilege in applications
  2. FastAPI Dependencies — Official documentation on dependency injection in FastAPI
  3. AWS IAM Best Practices — IAM best practices for least privilege on AWS
  4. Circuit Breaker Pattern — Resilience pattern for external services
  5. Python functools.lru_cache — Python's built-in caching for singletons
  6. GCP IAM Conditions — Advanced IAM conditions for granular access control
  7. 12-Factor App — Backing Services — Principle of backing services as attached resources
  8. OpenAI API Key Management — OpenAI's official security practices for API keys

Created: March 2026 Version: 1.0

← Previous6. Token Lifecycle and Audit TrailsNext →8. Project: Secrets Management Setup