"""FastAPI application serving QualPipe frontend routes.
For API endpoint details and schemas, use the autogenerated Swagger UI at
``/docs`` (OpenAPI at ``/openapi.json``).
Auth is delegated entirely to the BFF (mock_bff in dev, web-bff in prod).
The frontend holds no session state of its own -- it reads the BFF session
cookie and calls /auth/session + /auth/me on every protected request.
"""
import logging
import os
from pathlib import Path
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from starlette.middleware.base import BaseHTTPMiddleware
from .page_config import (
build_nav_pages,
build_plot_state,
load_frontend_config,
)
# Configure logging
logging.basicConfig(level=logging.INFO)
# ---------------------------------------------------------------------------
# BFF configuration
# ---------------------------------------------------------------------------
BFF_URL = os.getenv("BFF_URL", "http://localhost:8080")
BFF_SESSION_COOKIE = os.getenv("SESSION_COOKIE_NAME", "ctao_session")
DEFAULT_HOME_PATH = "/home"
PROTECTED_PATHS = (DEFAULT_HOME_PATH, "/LSTs", "/MSTs", "/SSTs", "/Auxiliary")
# ---------------------------------------------------------------------------
# BFF helpers
# ---------------------------------------------------------------------------
async def _bff_session(request: Request) -> dict:
"""Call BFF /auth/session. Always returns a dict -- never raises."""
session_cookie = request.cookies.get(BFF_SESSION_COOKIE, "")
cookies = {BFF_SESSION_COOKIE: session_cookie} if session_cookie else None
async with httpx.AsyncClient(cookies=cookies) as client:
try:
resp = await client.get(f"{BFF_URL}/auth/session")
return resp.json()
except httpx.TransportError:
return {
"authenticated": False,
"sub": None,
"username": None,
"expiresAt": None,
"loginUrl": "/auth/login",
}
async def _bff_me(request: Request) -> dict | None:
"""
Call BFF /auth/me.
Return the profile dict or None if unauthenticated.
Result is cached on request.state so the middleware and route handlers
share one HTTP call per request.
"""
if hasattr(request.state, "current_user"):
return request.state.current_user
session_cookie = request.cookies.get(BFF_SESSION_COOKIE, "")
cookies = {BFF_SESSION_COOKIE: session_cookie} if session_cookie else None
async with httpx.AsyncClient(cookies=cookies) as client:
try:
resp = await client.get(f"{BFF_URL}/auth/me")
user = resp.json() if resp.status_code == 200 else None
except httpx.TransportError:
user = None
if user:
# Keep compatibility with templates expecting username/role keys.
user.setdefault("username", user.get("preferred_username") or user.get("name"))
user.setdefault("role", "user")
request.state.current_user = user
return user
# ---------------------------------------------------------------------------
# Auth middleware -- replaces the old SessionMiddleware + AuthMiddleware stack
# ---------------------------------------------------------------------------
[docs]
class BffAuthMiddleware(BaseHTTPMiddleware):
"""Redirect unauthenticated users to the BFF login page."""
[docs]
async def dispatch(self, request: Request, call_next):
"""Enforce BFF authentication for protected frontend paths."""
path = request.url.path
if any(path == p or path.startswith(f"{p}/") for p in PROTECTED_PATHS):
session = await _bff_session(request)
if not session.get("authenticated"):
# post_login_redirect_uri tells the BFF where to send the user
# after a successful login -- same role as the old ?next= param.
bff_login = f"/auth/login?post_login_redirect_uri={path}"
return RedirectResponse(url=bff_login, status_code=303)
return await call_next(request)
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastAPI()
# BffAuthMiddleware is the only middleware needed.
# CORSMiddleware -- removed: BFF owns CORS; browser never calls this app directly.
app.add_middleware(BffAuthMiddleware)
project_root = Path(__file__).resolve().parents[0]
static_path = project_root / "static"
view_path = project_root / "view"
app.mount("/static", StaticFiles(directory=static_path), name="static")
view = Jinja2Templates(directory=view_path)
ARRAY_ELEMENT_TYPES = ["LSTs", "MSTs", "SSTs", "Auxiliary"]
FRONTEND_CONFIG = load_frontend_config()
view.env.globals["nav_pages"] = build_nav_pages(FRONTEND_CONFIG)
view.env.globals["nav_auxiliary"] = [
{"name": name, "label": name} for name in FRONTEND_CONFIG.auxiliary_subitems
]
template_404 = "templates/404.html"
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
[docs]
@app.get("/health")
async def health_check():
"""Health check endpoint for Kubernetes probes."""
return {"status": "ok", "service": "frontend"}
# ---------------------------------------------------------------------------
# Login / logout -- now just thin redirects into the BFF
# ---------------------------------------------------------------------------
[docs]
@app.get("/login")
async def login_page(request: Request, next: str | None = None): # noqa: A002
"""
Redirect to the BFF login page.
Keeps the /login URL stable so existing links and templates don't break,
but the BFF now owns the login UI and session creation.
"""
post_login = next or DEFAULT_HOME_PATH
return RedirectResponse(
url=f"/auth/login?post_login_redirect_uri={post_login}",
status_code=302,
)
[docs]
@app.post("/logout")
async def logout(request: Request):
"""POST logout via BFF with CSRF token, then forward its redirect."""
session_cookie = request.cookies.get(BFF_SESSION_COOKIE, "")
xsrf_cookie = request.cookies.get("XSRF-TOKEN", "")
cookies: dict[str, str] = {}
if session_cookie:
cookies[BFF_SESSION_COOKIE] = session_cookie
if xsrf_cookie:
cookies["XSRF-TOKEN"] = xsrf_cookie
headers = {"X-XSRF-TOKEN": xsrf_cookie} if xsrf_cookie else None
try:
async with httpx.AsyncClient(cookies=cookies or None) as client:
response = await client.post(f"{BFF_URL}/auth/logout", headers=headers)
except httpx.TransportError:
return RedirectResponse(url="/login", status_code=302)
location = response.headers.get("location")
if location:
return RedirectResponse(url=location, status_code=302)
return RedirectResponse(url="/login", status_code=302)
# ---------------------------------------------------------------------------
# Frontend HTML routes -- unchanged except current_user source
# ---------------------------------------------------------------------------
[docs]
@app.get("/", include_in_schema=False)
async def read_root():
"""Redirect the root path to the home page."""
return RedirectResponse(url=DEFAULT_HOME_PATH)
[docs]
@app.get("/home", response_class=HTMLResponse)
async def read_home(request: Request):
"""Render the home page."""
config = load_frontend_config()
return view.TemplateResponse(
request=request,
name="pages/home.html",
context={
"array_element": "home",
"user_index_error": config.user_index_error,
"current_user": await _bff_me(request),
},
)
[docs]
@app.get("/LSTs", response_class=HTMLResponse)
async def read_lst(request: Request): # noqa: N802
"""Render the LST summary page."""
return view.TemplateResponse(
request=request,
name="pages/array_element_type/LSTs-summary.html",
context={"array_element": "LSTs", "current_user": await _bff_me(request)},
)
[docs]
@app.get("/MSTs", response_class=HTMLResponse)
async def read_mst(request: Request): # noqa: N802
"""Render the MST summary page."""
return view.TemplateResponse(
request=request,
name="pages/array_element_type/MSTs-summary.html",
context={"array_element": "MSTs", "current_user": await _bff_me(request)},
)
[docs]
@app.get("/SSTs", response_class=HTMLResponse)
async def read_sst(request: Request): # noqa: N802
"""Render the SST summary page."""
return view.TemplateResponse(
request=request,
name="pages/array_element_type/SSTs-summary.html",
context={"array_element": "SSTs", "current_user": await _bff_me(request)},
)
[docs]
@app.get("/Auxiliary", response_class=HTMLResponse)
async def read_auxiliary(request: Request): # noqa: N802
"""Render the auxiliary placeholder page."""
return view.TemplateResponse(
request=request,
name="templates/501.html",
context={
"array_element": "Auxiliary",
"current_user": await _bff_me(request),
},
status_code=501,
)
[docs]
@app.get("/{array_element_type}/{subitem}", response_class=HTMLResponse)
async def read_array_element_type_subitem(
array_element_type: str, subitem: str, request: Request
):
"""Render a configured subitem page for a valid array element type."""
if array_element_type not in ARRAY_ELEMENT_TYPES:
return view.TemplateResponse(
request=request, name=template_404, status_code=404
)
current_user = await _bff_me(request)
if array_element_type == "Auxiliary":
config = load_frontend_config()
if subitem not in config.auxiliary_subitems:
return view.TemplateResponse(
request=request, name=template_404, status_code=404
)
return view.TemplateResponse(
request=request,
name="templates/501.html",
context={
"array_element": array_element_type,
"active_subitem": subitem,
"subitem": subitem.replace("_", " "),
"current_user": current_user,
},
status_code=501,
)
config = load_frontend_config()
page = config.page_by_name(subitem)
if page is None:
page_error = config.page_error(subitem)
if page_error:
return view.TemplateResponse(
request=request,
name="pages/array_element_type/page_error.html",
context={
"array_element": array_element_type,
"active_subitem": subitem,
"subitem": config.page_label(subitem),
"page_error": page_error,
"current_user": current_user,
},
status_code=500,
)
return view.TemplateResponse(
request=request, name=template_404, status_code=404
)
plot_configurations, plot_errors = build_plot_state(page)
return view.TemplateResponse(
request=request,
name=page.template,
context={
"array_element": array_element_type,
"active_subitem": page.name,
"subitem": page.label,
"plot_titles": page.plot_titles(),
"plot_keys": page.plot_keys(),
"plot_metrics": page.plot_metrics(),
"plot_css": page.css,
"plot_js": page.js,
"layout": page.layout,
"plot_configurations": plot_configurations,
"plot_errors": plot_errors,
"current_user": current_user,
},
)
[docs]
@app.exception_handler(404)
async def not_found(request: Request, exc):
"""Render the custom 404 page."""
return view.TemplateResponse(request=request, name=template_404, status_code=404)
