FastAPI middleware — CORS, logging, authentication og rate limiting

FastAPI middleware intercepter alle requests og responses. Her er opsætningen til en produktionsklar SaaS-applikation.

---

Middleware execution order

Request flow (top → bottom):
  1. TrustedHostMiddleware     (blokér ukendte hosts)
  2. CORSMiddleware            (preflight + headers)
  3. LoggingMiddleware         (request-ID + timing)
  4. RateLimitMiddleware       (429 ved overskridelse)
  5. AuthenticationMiddleware  (JWT validation — valgfri)
  6. Endpoint handler

Response flow (bottom → top, samme liste)

# server/src/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI(...)

# Orden er vigtig — tilføjes i omvendt rækkefølge (LIFO):
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["api.mbus-gateway.dk", "localhost"])
app.add_middleware(CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
)
app.add_middleware(LoggingMiddleware)
app.add_middleware(RateLimitMiddleware)

---

CORS-konfiguration

# server/src/core/config.py

from pydantic_settings import BaseSettings
from typing import list

class Settings(BaseSettings):
    # Produktion: Præcise origins (ingen wildcard)
    CORS_ORIGINS: list[str] = [
        "https://app.mbus-gateway.dk",    # Udlejer-portal
        "https://mbus-gateway.dk",         # Hjemmeside
        "https://tech.mbus-gateway.dk",    # Tech PWA
    ]
    # Udvikling: Tilføj localhost
    # CORS_ORIGINS: ["http://localhost:3000", "http://localhost:5173"]

# CORS fejl-debugging:
# Fejl: "No 'Access-Control-Allow-Origin' header"
# Årsag 1: Origin ikke i allow_origins → Tilføj den
# Årsag 2: allow_credentials=True + allow_origins=["*"] → Ikke tilladt
# Løsning: Præcise origins, aldrig wildcard med credentials

---

Custom middleware klasse

# server/src/middleware/logging.py

import uuid
import time
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

logger = structlog.get_logger("http")

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = str(uuid.uuid4())
        start = time.perf_counter()

        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            method=request.method,
            path=request.url.path,
        )

        response = await call_next(request)
        duration_ms = round((time.perf_counter() - start) * 1000, 2)

        logger.info(
            "request",
            status=response.status_code,
            duration_ms=duration_ms,
        )

        response.headers["X-Request-ID"] = request_id
        structlog.contextvars.clear_contextvars()
        return response

---

Rate limiting middleware

# server/src/middleware/rate_limit.py

import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

class RateLimitMiddleware(BaseHTTPMiddleware):
    """
    Sliding window rate limiter — Redis-backed.
    Bruger IP + path som key for granulær kontrol.
    """

    # Endpoint-specifikke limits (path prefix → limit/window):
    LIMITS = {
        "/api/v1/auth/": (10, 60),       # 10 req/60s — brute force beskyttelse
        "/api/v1/settlements/generate": (5, 60),  # 5 req/60s — dyr operation
        "/api/v1/": (200, 60),            # 200 req/60s — standard API
    }
    DEFAULT_LIMIT = (100, 60)

    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting for health check:
        if request.url.path == "/health":
            return await call_next(request)

        redis = request.app.state.redis
        ip = request.client.host if request.client else "unknown"

        # Find applicable limit:
        limit, window = self.DEFAULT_LIMIT
        for prefix, lim in self.LIMITS.items():
            if request.url.path.startswith(prefix):
                limit, window = lim
                break

        key = f"rate:{ip}:{request.url.path[:50]}"
        now = time.time()

        pipe = redis.pipeline()
        pipe.zremrangebyscore(key, 0, now - window)
        pipe.zadd(key, {str(now): now})
        pipe.zcard(key)
        pipe.expire(key, window)
        results = await pipe.execute()

        if results[2] > limit:
            return JSONResponse(
                status_code=429,
                content={"detail": f"Rate limit: {limit} requests/{window}s"},
                headers={"Retry-After": str(window)},
            )

        return await call_next(request)

---

Security headers middleware

# server/src/middleware/security.py

from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)

        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Permissions-Policy"] = "geolocation=(), microphone=()"

        # Kun HTTPS i produktion:
        if settings.ENVIRONMENT == "production":
            response.headers["Strict-Transport-Security"] = (
                "max-age=31536000; includeSubDomains"
            )

        return response

---

Tenant-isolering via middleware

# server/src/middleware/tenant.py

from starlette.middleware.base import BaseHTTPMiddleware

class TenantContextMiddleware(BaseHTTPMiddleware):
    """
    Sætter tenant-context fra JWT — undgår gentagen auth-parsing.
    Starlette state er tilgængeligt i alle downstream dependencies.
    """
    async def dispatch(self, request: Request, call_next):
        # Public endpoints har ingen auth:
        PUBLIC_PATHS = {"/health", "/api/v1/auth/login", "/api/v1/auth/register"}
        if request.url.path in PUBLIC_PATHS:
            return await call_next(request)

        token = request.headers.get("Authorization", "").removeprefix("Bearer ")
        if token:
            try:
                payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
                request.state.tenant_id = payload.get("tenant_id")
                request.state.user_role = payload.get("role")
                request.state.user_id = payload.get("sub")
            except Exception:
                pass  # Auth fejl håndteres i Depends()

        return await call_next(request)

---

Performance: middleware overhead

Benchmark (100.000 requests):

Ingen middleware:         0.8ms gennemsnitlig respons
+ LoggingMiddleware:      0.9ms (+0.1ms — structlog JSON)
+ RateLimitMiddleware:    1.2ms (+0.3ms — Redis pipeline)
+ CORSMiddleware:         1.3ms (+0.1ms — header check)
+ SecurityHeaders:        1.35ms (+0.05ms — header tilføjelse)
+ TrustedHost:            1.36ms (+0.01ms — string match)

Total overhead: ~0.5ms — acceptabelt for SaaS API

Optimering:
  → Redis pipeline (ikke separate kald) er afgørende for rate limit
  → structlog's cache_logger_on_first_use=True reducerer overhead
  → TrustedHostMiddleware: Brug list (O(n)) — hold den kort

---

Konklusion

Middleware-orden er LIFO ved tilføjelse — TrustedHost og CORS tilføjes til sidst men kører først. Rate limiting med Redis sliding window giver granulær kontrol pr. endpoint. Security headers tilføjes automatisk i produktion. Samlet middleware-overhead på under 1ms er acceptabelt for en SaaS API med 200 req/s.

Se FastAPI dependency injection guide eller Redis caching guide.