37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""Bearer-token auth middleware for the MCP sub-app.
|
|
|
|
Mounted on the FastMCP ``streamable_http_app()`` (NOT the console FastAPI),
|
|
so cookie-session auth on console routes is unaffected.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse, Response
|
|
|
|
from host_agent.mcp_token import McpTokenStore
|
|
|
|
|
|
class BearerAuthMiddleware(BaseHTTPMiddleware):
|
|
def __init__(self, app, token_store: McpTokenStore) -> None:
|
|
super().__init__(app)
|
|
self._store = token_store
|
|
|
|
async def dispatch(self, request: Request, call_next) -> Response: # type: ignore[no-untyped-def]
|
|
header = request.headers.get("Authorization")
|
|
if not header or not header.lower().startswith("bearer "):
|
|
return _unauthorized()
|
|
presented = header.split(" ", 1)[1].strip()
|
|
if not self._store.verify(presented):
|
|
return _unauthorized()
|
|
return await call_next(request)
|
|
|
|
|
|
def _unauthorized() -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={"error": "invalid token"},
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|