cabbage
009fd039a2
Disabled by default (empty API_TOKEN). When set: - All /api/* and /mcp requests require: Authorization: Bearer <token> - Public exemptions: /, /healthz, /static/*, /auth-check - Web UI: pre-flight /auth-check on load; shows token modal if required - Token stored in sessionStorage, sent on every API request - Mid-session 401s re-trigger the token modal - MCP clients must pass the header: Authorization: Bearer <token> Files changed: - app/config.py: api_token field + API_TOKEN env var - app/api/auth.py: Starlette BaseHTTPMiddleware for token enforcement - main.py: register middleware + /auth-check public endpoint - static/js/api.js: token storage, auth header, 401 handler hook - static/js/app.js: auth pre-flight, showTokenModal(), bootstrap() - static/css/components.css: .auth-overlay / .auth-card styles - README.md: API_TOKEN env var docs + MCP client header example
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
"""
|
|
app/api/auth.py
|
|
Optional Bearer-token authentication middleware.
|
|
|
|
When API_TOKEN is set in the environment, every request to a protected path
|
|
must include the header:
|
|
|
|
Authorization: Bearer <token>
|
|
|
|
Public paths (no auth required even when a token is configured):
|
|
GET / (serves index.html — UI bootstrap)
|
|
GET /healthz (health check)
|
|
GET /static/* (JS / CSS / fonts — UI assets)
|
|
GET /auth-check (lets the UI know whether auth is required)
|
|
|
|
All other paths — /api/*, /mcp, /docs, /openapi.json — are protected.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Paths that are always accessible without a token
|
|
_PUBLIC_EXACT = frozenset({"/", "/healthz", "/auth-check"})
|
|
_PUBLIC_PREFIXES = ("/static/",)
|
|
|
|
|
|
class TokenAuthMiddleware(BaseHTTPMiddleware):
|
|
def __init__(self, app, token: str) -> None:
|
|
super().__init__(app)
|
|
self._token = token
|
|
if token:
|
|
logger.info("Token authentication ENABLED")
|
|
else:
|
|
logger.info("Token authentication disabled (set API_TOKEN to enable)")
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
# Auth disabled → pass through
|
|
if not self._token:
|
|
return await call_next(request)
|
|
|
|
path = request.url.path
|
|
|
|
# Always-public paths
|
|
if path in _PUBLIC_EXACT:
|
|
return await call_next(request)
|
|
for prefix in _PUBLIC_PREFIXES:
|
|
if path.startswith(prefix):
|
|
return await call_next(request)
|
|
|
|
# Validate Bearer token
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if auth_header == f"Bearer {self._token}":
|
|
return await call_next(request)
|
|
|
|
logger.warning("Unauthorized request: %s %s", request.method, path)
|
|
return JSONResponse(
|
|
{"detail": "Unauthorized — provide a valid Bearer token"},
|
|
status_code=401,
|
|
)
|
|
|