Middleware

Architecture

quadkit-web’s middleware follows the ASGI model — each middleware wraps the next in a chain:

Request → RequestBodySizeLimitMiddleware
        → WebHooksMiddleware
        → SecurityHeadersMiddleware
        → CORSMiddleware
        → CSRFProtectionMiddleware
        → RequestContextMiddleware
        → DIScopeMiddleware
        → (user middleware)
        → Router → Controller

Middleware is applied via MiddlewareSetup.configure() during WebProvider.boot(). The stack is composed once and frozen — it is never rebuilt per-request.

Ordering

The MiddlewareSetup class applies middleware in security-first order:

1. Auto-configure CSP for API docs (mutates config for Swagger/ReDoc)
2. Security headers (HSTS, CSP, X-Frame-Options, cross-origin isolation)
3. CORS (cross-origin resource sharing)
4. CSRF (opt-in, configurable via SecurityConfig.csrf)
5. Request context (wires shared Context to the request scope)
6. DI scope (creates a request-scoped child container)
7. Request body size limit (protects against oversized payloads)
8. Web hooks (outermost — emits WebRequestReceivedHook / WebResponsePreparedHook)

User-supplied middleware is prepended to DefaultMiddlewareStack.build() before the internal stack, so it runs outermost.

Built-in Middleware

Middleware	Module	Purpose
SecurityHeadersMiddleware	quadkit.web.middleware.security	HSTS, CSP, X-Frame-Options, cross-origin isolation
CORSMiddleware	quadkit.web.middleware.cors	Cross-origin resource sharing
CSRFProtectionMiddleware	quadkit.web.security.csrf.middleware	CSRF token validation
RateLimitMiddleware	quadkit.web.middleware.rate_limit	Sliding-window rate limiting
InputSanitizationMiddleware	quadkit.web.middleware.sanitization	Strips null bytes and script injection from query params
RequestBodySizeLimitMiddleware	quadkit.web.middleware.body_limit	Enforces max_body_size (default 10 MiB)
AccessLogMiddleware	quadkit.web.middleware.access_log	Structured request logging
RequestContextMiddleware	quadkit.web.middleware.request_context	Assign and propagate X-Request-ID, bind the request context
TimingMiddleware	quadkit.web.middleware.timing	X-Response-Time header
CompressionMiddleware	quadkit.web.middleware.compression	Gzip/brotli response compression
StaticFilesMiddleware	quadkit.web.middleware.static	Serve static files

Custom Middleware

Write any ASGI callable — no base class required:

from starlette.types import ASGIApp, Scope, Receive, Send


class ServerTimingMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        scope["state"]["started"] = time.perf_counter()
        await self.app(scope, receive, send)

MiddlewareProtocol

For middleware that needs DI-resolved services, implement MiddlewareProtocol from quadkit.web.middleware.base:

from quadkit.web.middleware import MiddlewareChain


class TimingMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        import time
        start = time.perf_counter()
        await self.app(scope, receive, send)
        elapsed = time.perf_counter() - start
        # Send wrapper needed to inject response headers at the ASGI level

Registering Middleware

Via WebProvider

from quadkit.web import WebProvider
from my_app.middleware import ServerTimingMiddleware, TimingMiddleware

provider = WebProvider(
    middleware=[TimingMiddleware, ServerTimingMiddleware],
)

Via WebProvider.middleware_manager

provider.middleware_manager.add(TimingMiddleware)

Via WebConfig (entry-point discovery)

Packages can register middleware via the quadkit.web.contributors entry point. The WebContributorRegistry merges them at boot.

Via application.yaml

Middleware configuration is driven by WebConfig fields — security headers, CORS, CSRF, rate limiting, body size, and static files are all configured declaratively:

web:
  security:
    csrf:
      enabled: true
  cors:
    enabled: true
    allowed_origins: ["https://myapp.com"]
  rate_limit:
    enabled: true
    default_limit: 100
  max_body_size: 5242880  # 5 MiB

The local body-limit middleware rejects an oversized declared length before calling the app and counts actual bytes as they are read, even with a missing, invalid or inaccurate length. An over-limit read aborts downstream processing rather than presenting a truncated body as valid; it returns 413 if the response has not started. Applications should finish reading and validating their body before making changes: the middleware cannot undo earlier effects, measure a body an app never reads, or replace an already-started response. An ASGI server or proxy may already buffer a large chunk, so configure limits and timeouts at the edge too. Catching BaseException inside an app can defeat the processing abort and must not be used to continue a request.

RequestBodySizeLimitMiddleware.supports_streaming_body_abort_v1 declares this implementation’s streaming-abort capability for consumers that must reject incompatible versions. The marker does not mean the middleware is installed, enabled, ordered correctly, or protected by an edge limit.

Testing Middleware

Use TestClient from quadkit.testing:

from quadkit.testing import WebTestClient
from quadkit.web import WebProvider


class TestRateLimitMiddleware:
    async def test_rate_limit_applied(
        self, test_client: TestClient, web_provider: WebProvider
    ) -> None:
        responses = await asyncio.gather(
            *[test_client.get("/api/test") for _ in range(110)]
        )
        last = responses[-1]
        assert last.status_code == 429  # Too Many Requests
        assert "Retry-After" in last.headers

Testing custom middleware

from starlette.testclient import TestClient
from quadkit.web.middleware.stack import DefaultMiddlewareStack


class TestServerTimingMiddleware:
    def test_server_timing_added(self) -> None:
        stack = DefaultMiddlewareStack(
            extra_middlewares=[ServerTimingMiddleware],
        )
        middlewares = stack.build()
        # Assert middleware is in the stack
        assert any(
            "ServerTimingMiddleware" in str(mw.cls.__name__) for mw in middlewares
        )