Spaces:
Sleeping
Sleeping
File size: 1,563 Bytes
dc1b199 faa8fb3 dc1b199 7d37f11 dc1b199 7d37f11 dc1b199 7d37f11 dc1b199 7d37f11 dc1b199 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | """Middleware that enforces a non-empty X-Tenant-ID header on every request."""
from collections.abc import Awaitable, Callable
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
# Paths that do not require a tenant identity (exact match)
_EXEMPT: frozenset[str] = frozenset(
{"/", "/health", "/docs", "/openapi.json", "/redoc", "/favicon.ico"}
)
# Path prefixes that do not require a tenant identity
_EXEMPT_PREFIXES: tuple[str, ...] = ("/static/",)
class TenantAuthMiddleware(BaseHTTPMiddleware):
"""Reject requests that carry no ``X-Tenant-ID`` header.
On success the header value is stored on ``request.state.tenant_id``
so downstream handlers can read it without re-parsing headers.
Example::
curl -H "X-Tenant-ID: tenant_abc" http://localhost:8000/upload ...
"""
async def dispatch(
self,
request: Request,
call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
path = request.url.path
if path in _EXEMPT or path.startswith(_EXEMPT_PREFIXES):
return await call_next(request)
tenant_id: str | None = request.headers.get("X-Tenant-ID")
if not tenant_id or not tenant_id.strip():
return JSONResponse(
status_code=401,
content={"detail": "Missing or empty X-Tenant-ID header"},
)
request.state.tenant_id = tenant_id.strip()
return await call_next(request)
|