"""
Authentication middleware for web pages
Protects web UI routes while allowing API endpoints to use API Key auth
"""
from fastapi import Request
from fastapi.responses import RedirectResponse
from starlette.middleware.base import BaseHTTPMiddleware
from app.core.config import settings
import logging
logger = logging.getLogger(__name__)
class WebAuthenticationMiddleware(BaseHTTPMiddleware):
"""
Middleware to enforce Keycloak authentication for web pages
Protected routes (require user login):
- / (landing page)
- /docs (API documentation)
- /data-management/* (data management pages)
Excluded routes (no auth required or use different auth):
- /auth/* (authentication endpoints)
- /api/v1/* (API endpoints - use API Key auth)
- /admin/* (SQLAdmin - has its own auth)
- /apiservice/admin/statics/* (admin static files)
- /admin/statics/* (admin static files)
"""
# Routes that require user authentication
PROTECTED_PATHS = [
"/",
"/docs",
"/redoc",
"/openapi.json",
"/data-management"
]
# Routes that are excluded from user authentication
EXCLUDED_PATHS = [
"/auth", # Authentication endpoints
"/api/v1", # API endpoints (use API Key)
"/admin", # SQLAdmin (has own auth)
]
async def dispatch(self, request: Request, call_next):
"""Process request and check authentication if needed"""
# Get the path without root_path prefix
path = request.url.path
# Remove root_path prefix if it exists
if settings.ROOT_PATH and path.startswith(settings.ROOT_PATH):
path = path[len(settings.ROOT_PATH):]
# Ensure path starts with /
if not path.startswith("/"):
path = "/" + path
# Check if path is excluded from authentication
is_excluded = any(path.startswith(excluded) for excluded in self.EXCLUDED_PATHS)
if is_excluded:
# Skip authentication for excluded paths (including /api/v1/*)
return await call_next(request)
# Check if path requires authentication
requires_auth = any(
path == protected or path.startswith(protected + "/")
for protected in self.PROTECTED_PATHS
)
if requires_auth:
# Check if user is authenticated
user = request.session.get("user")
if not user:
# User not authenticated - redirect to login
# Preserve the original path for redirect after login
original_path = request.url.path
# Build login URL with redirect
login_url = f"{settings.ROOT_PATH}/auth/login?redirect_to={original_path}"
logger.info(f"Unauthenticated access to {original_path}, redirecting to login")
return RedirectResponse(url=login_url, status_code=302)
# Role-based access control for specific paths
user_roles = user.get("roles", [])
# /data-management/* requires 'operation' or 'admin' role
if path.startswith("/data-management"):
if not any(role in user_roles for role in ["admin", "operation"]):
logger.warning(
f"Access denied: User {user.get('username')} "
f"(roles: {user_roles}) tried to access {path}"
)
# Return 403 Forbidden page
from fastapi.responses import HTMLResponse
return HTMLResponse(
content=f"""
Access Denied
🚫
Access Denied
You don't have permission to access this page.
Required role: operation or admin
Your roles: {', '.join(user_roles) if user_roles else 'None'}
Please contact your administrator if you need access.