"""Browser session login and API key gates for HTTP (HF and remote agents).""" from __future__ import annotations from typing import Annotated from fastapi import APIRouter, FastAPI, Form from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import Response from tracker.config import DashboardLogin, TrackerConfig SESSION_KEY = "tracker_dashboard_ok" SESSION_USER_KEY = "tracker_dashboard_user" def _request_config(request: Request) -> TrackerConfig: return request.app.state.tracker_config def _auth_enabled(cfg: TrackerConfig) -> bool: return bool(cfg.dashboard_password or cfg.dashboard_logins) def _find_login(cfg: TrackerConfig, email: str, password: str) -> DashboardLogin | None: normalized = email.strip().lower() for login in cfg.dashboard_logins: if login.email.strip().lower() == normalized and login.password == password: return login return None def _email_password_login_html(error: bool) -> str: err = '

Wrong email or password.

' if error else "" return f""" Tracker login

Tracker

{err}
""" def _single_password_login_html(error: bool) -> str: err = '

Wrong password.

' if error else "" return f""" Tracker login

Tracker

{err}
""" _LOGIN_NO_STORE = { "Cache-Control": "no-store, max-age=0", "Pragma": "no-cache", } async def login_get(request: Request) -> HTMLResponse: cfg = _request_config(request) if not _auth_enabled(cfg): return HTMLResponse("

Login disabled.

", status_code=404) err = request.query_params.get("error") == "1" if cfg.dashboard_logins: return HTMLResponse( _email_password_login_html(error=err), headers=_LOGIN_NO_STORE, ) return HTMLResponse( _single_password_login_html(error=err), headers=_LOGIN_NO_STORE, ) async def login_post( request: Request, password: Annotated[str, Form(min_length=1)], email: Annotated[str | None, Form()] = None, ) -> Response: cfg = _request_config(request) if not _auth_enabled(cfg): return JSONResponse(status_code=404, content={"detail": "Login disabled"}) if cfg.dashboard_logins: if email is None: return RedirectResponse("/login?error=1", status_code=303) login = _find_login(cfg, email, password) if login is None: return RedirectResponse("/login?error=1", status_code=303) request.session[SESSION_KEY] = True request.session[SESSION_USER_KEY] = login.email return RedirectResponse("/", status_code=303) if password != cfg.dashboard_password: return RedirectResponse("/login?error=1", status_code=303) request.session[SESSION_KEY] = True return RedirectResponse("/", status_code=303) async def logout_post(request: Request) -> Response: session = request.scope.get("session") if session is not None: session.clear() return RedirectResponse("/login", status_code=303) def register_auth_routes(app: FastAPI) -> None: router = APIRouter(tags=["auth"]) router.add_api_route("/login", login_get, methods=["GET"]) router.add_api_route("/login", login_post, methods=["POST"]) router.add_api_route("/logout", logout_post, methods=["POST"]) router.add_api_route("/api/meta/demo_accounts", demo_accounts, methods=["GET"]) app.include_router(router) async def demo_accounts(request: Request) -> JSONResponse: """Lists configured dashboard labels (no secrets; use only with X-API-Key on /api).""" cfg = _request_config(request) accounts = [ { "label": login.label, "email": login.email, "role": login.role, } for login in cfg.dashboard_logins ] return JSONResponse({"accounts": accounts}) class UnifiedAuthMiddleware(BaseHTTPMiddleware): """Require X-API-Key for /api when configured; optional session for browser UI.""" async def dispatch(self, request: Request, call_next): cfg = _request_config(request) path = request.url.path if path in ("/health", "/api/health", "/api/meta/display"): return await call_next(request) if path.startswith("/api"): api_key = cfg.agent_api_key if api_key: if request.headers.get("x-api-key") == api_key: return await call_next(request) session = request.scope.get("session") if session is not None and session.get(SESSION_KEY) is True: return await call_next(request) return JSONResponse(status_code=401, content={"detail": "Unauthorized"}) return await call_next(request) if not _auth_enabled(cfg): return await call_next(request) if path.startswith("/login"): return await call_next(request) session = request.scope.get("session") if session is None: return RedirectResponse("/login", status_code=303) if session.get(SESSION_KEY) is True: return await call_next(request) return RedirectResponse("/login", status_code=303) def apply_session_middleware(app: FastAPI, cfg: TrackerConfig) -> None: if not _auth_enabled(cfg): return secret = cfg.session_secret if not secret: msg = "session_secret is required when dashboard_password is set" raise ValueError(msg) from starlette.middleware.sessions import SessionMiddleware app.add_middleware( SessionMiddleware, secret_key=secret, session_cookie="tracker_session", same_site="lax", https_only=False, max_age=14 * 24 * 3600, ) def apply_auth_middleware(app: FastAPI) -> None: app.add_middleware(UnifiedAuthMiddleware)