Spaces:
Sleeping
Sleeping
| """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 = '<p class="err">Wrong email or password.</p>' if error else "" | |
| return f"""<!DOCTYPE html> | |
| <html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"> | |
| <title>Tracker login</title> | |
| <style> | |
| body {{ font-family: system-ui, sans-serif; max-width: 22rem; margin: 3rem auto; padding: 0 1rem; }} | |
| label {{ display: block; margin: 0.75rem 0 0.25rem; }} | |
| input {{ width: 100%; padding: 0.5rem; font-size: 1rem; box-sizing: border-box; }} | |
| button {{ margin-top: 1rem; padding: 0.5rem 1rem; font-size: 1rem; cursor: pointer; }} | |
| .err {{ color: #c62828; }} | |
| </style></head><body> | |
| <h1>Tracker</h1> | |
| {err} | |
| <form method="post" action="/login"> | |
| <label for="email">Email</label> | |
| <input id="email" name="email" type="email" autocomplete="username" required> | |
| <label for="pw">Password</label> | |
| <input id="pw" name="password" type="password" autocomplete="current-password" required> | |
| <button type="submit">Sign in</button> | |
| </form> | |
| </body></html>""" | |
| def _single_password_login_html(error: bool) -> str: | |
| err = '<p class="err">Wrong password.</p>' if error else "" | |
| return f"""<!DOCTYPE html> | |
| <html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"> | |
| <title>Tracker login</title> | |
| <style> | |
| body {{ font-family: system-ui, sans-serif; max-width: 22rem; margin: 3rem auto; padding: 0 1rem; }} | |
| label {{ display: block; margin: 0.75rem 0 0.25rem; }} | |
| input[type="password"] {{ width: 100%; padding: 0.5rem; font-size: 1rem; box-sizing: border-box; }} | |
| button {{ margin-top: 1rem; padding: 0.5rem 1rem; font-size: 1rem; cursor: pointer; }} | |
| .err {{ color: #c62828; }} | |
| </style></head><body> | |
| <h1>Tracker</h1> | |
| {err} | |
| <form method="post" action="/login"> | |
| <label for="pw">Password</label> | |
| <input id="pw" name="password" type="password" autocomplete="current-password" required> | |
| <button type="submit">Sign in</button> | |
| </form> | |
| </body></html>""" | |
| _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("<p>Login disabled.</p>", 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) | |