INNER CODE UNIT · Python

SecurityHeadersMiddleware

caura-ai/caura · core-api/src/core_api/app.py:110

class SecurityHeadersMiddleware:
    """Pure ASGI middleware — compatible with mounted raw ASGI apps like MCP."""

    def __init__(self, app: ASGIApplication) -> None:
        self.app = app

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

        async def send_with_headers(message):
            if message["type"] == "http.response.start":
                existing = [(k, v) for k, v in message.get("headers", []) if k not in _SECURITY_HEADER_KEYS]
                message = {**message, "headers": [*existing, *_SECURITY_HEADERS_ENCODED]}
            await send(message)

        await self.app(scope, receive, send_with_headers)

View source record →