Spaces:
Runtime error
Runtime error
File size: 3,092 Bytes
865bc90 dc1b199 faa8fb3 dc1b199 865bc90 dc1b199 865bc90 dc1b199 865bc90 7d37f11 dc1b199 865bc90 dc1b199 865bc90 dc1b199 7d37f11 dc1b199 865bc90 dc1b199 865bc90 dc1b199 865bc90 dc1b199 865bc90 aad7814 865bc90 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Middleware that authenticates every request and pins ``request.state.tenant_id``.
Identity is resolved from a signed bearer token (``Authorization: Bearer <token>``)
whose subject is the tenant_id. Because the token is HMAC-signed server-side, a
client cannot claim another tenant's data — this is what enforces isolation.
For local development (``DEV_MODE=true``) a raw ``X-Tenant-ID`` header is still
accepted as a fallback so existing tooling and tests keep working. In production
a valid token is mandatory.
"""
from collections.abc import Awaitable, Callable
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from app.api.auth import verify_token
from app.config import settings
# Paths that do not require authentication (exact match)
_EXEMPT: frozenset[str] = frozenset(
{
"/",
"/health",
"/docs",
"/openapi.json",
"/redoc",
"/favicon.ico",
"/auth/login",
"/auth/register",
}
)
# Path prefixes that do not require authentication
_EXEMPT_PREFIXES: tuple[str, ...] = ("/static/",)
class TenantAuthMiddleware(BaseHTTPMiddleware):
"""Authenticate the caller and expose the verified tenant on request state.
Example::
# Obtain a token, then call protected endpoints with it:
curl -H "Authorization: Bearer <token>" http://localhost:8000/documents
"""
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 = self._resolve_tenant(request)
if tenant_id is None:
return JSONResponse(
status_code=401,
content={"detail": "Authentication required. Log in to obtain a token."},
)
request.state.tenant_id = tenant_id
return await call_next(request)
@staticmethod
def _resolve_tenant(request: Request) -> str | None:
auth = request.headers.get("Authorization", "")
if auth.lower().startswith("bearer "):
token = auth[7:].strip()
sub = verify_token(token)
if sub:
return sub
# A present-but-invalid token must never silently fall through to
# the dev header path.
return None
# ``<img src>`` and direct links cannot send Authorization headers; the UI
# passes the same signed token as ``access_token`` (scoped to this origin).
query_token = (request.query_params.get("access_token") or "").strip()
if query_token:
sub = verify_token(query_token)
if sub:
return sub
return None
if settings.dev_mode:
raw = request.headers.get("X-Tenant-ID")
if raw and raw.strip():
return raw.strip()
return None
|