"""Admin console: serves the ``ui/`` SPA and signs the operator in. Everything the console shows comes from ``/api/*`` (``ApiFrontend``) with a bearer. This app, mounted at ``/admin``, owns three JSON routes under ``/admin/auth`` - login (``ADMIN_USER`` / ``ADMIN_PASS`` from env, session cookie signed with ``SESSION_SECRET``, 8 h), logout, and ``session``, which hands a signed-in browser the process-lifetime admin bearer - and the static build under ``/admin/``. The bearer is minted at startup, registered in the token store with scope ``*`` and never persisted; a gateway restart rotates it, and the SPA refetches ``session`` on a 401. """ from __future__ import annotations import hmac import json import logging import secrets import time from collections import deque from pathlib import Path from typing import TYPE_CHECKING, Any import itsdangerous from fastapi import FastAPI, HTTPException, Request, status from fastapi.responses import FileResponse, JSONResponse, Response from beaver_gateway.core import audit from beaver_gateway.frontends.base import Frontend if TYPE_CHECKING: from beaver_gateway.frontends.base import GatewayRuntime _log = logging.getLogger("beaver_gateway.frontends.admin") BASE = "/admin" SESSION_COOKIE = "beaver_admin_session" SESSION_MAX_AGE = 8 * 3600 SESSION_SALT = "beaver-gateway.admin.session.v2" ADMIN_TOKEN_NAME = "admin-ui" # noqa: S105 LOGIN_MAX_ATTEMPTS = 5 LOGIN_WINDOW_SECONDS = 300.0 DEFAULT_UI_DIR = Path(__file__).resolve().parents[4] / "ui" / "build" __all__ = ["AdminFrontend"] class AdminFrontend(Frontend): path = BASE landing = True def __init__(self, *, ui_dir: Path | None = None) -> None: self.ui_dir = (ui_dir or DEFAULT_UI_DIR).resolve() self._app: FastAPI | None = None def configure(self, runtime: GatewayRuntime) -> None: if not runtime.session_secret: msg = "AdminFrontend requires SESSION_SECRET in env" raise RuntimeError(msg) if not runtime.admin_user or not runtime.admin_pass: msg = "AdminFrontend requires ADMIN_USER and ADMIN_PASS in env" raise RuntimeError(msg) token = secrets.token_urlsafe(32) runtime.token_store.grant(ADMIN_TOKEN_NAME, token) self._app = build_app(runtime, token=token, ui_dir=self.ui_dir) if not (self.ui_dir / "index.html").is_file(): _log.warning( "admin UI build missing at %s (run `make ui-build`)", self.ui_dir ) def app(self) -> FastAPI | None: return self._app def build_app(runtime: GatewayRuntime, *, token: str, ui_dir: Path) -> FastAPI: app = FastAPI(title="beaver-gateway / admin", docs_url=None, redoc_url=None) signer = itsdangerous.URLSafeTimedSerializer( runtime.session_secret, salt=SESSION_SALT ) login_limit = _LoginRateLimit( max_attempts=LOGIN_MAX_ATTEMPTS, window=LOGIN_WINDOW_SECONDS ) index = ui_dir / "index.html" @app.post("/auth/login") async def login(request: Request) -> Response: try: data = await request.json() except json.JSONDecodeError as exc: raise HTTPException(status.HTTP_400_BAD_REQUEST, "invalid JSON") from exc username = str(data.get("username") or "") password = str(data.get("password") or "") ip = _client_ip(request) if not login_limit.check(ip): _log.warning("admin login rate-limited: ip=%s user=%r", ip, username) await audit.log( runtime, actor=f"admin:{username}", kind="login_failed", reason="rate_limited", ip=ip, ) raise HTTPException( status.HTTP_429_TOO_MANY_REQUESTS, "too many attempts; try again in a few minutes", headers={"Retry-After": str(int(LOGIN_WINDOW_SECONDS))}, ) ok = hmac.compare_digest(username, runtime.admin_user) and hmac.compare_digest( password, runtime.admin_pass ) if not ok: login_limit.record_failure(ip) _log.info("admin login failed: user=%r ip=%s", username, ip) await audit.log( runtime, actor=f"admin:{username}", kind="login_failed", reason="bad_credentials", ip=ip, ) raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials") login_limit.clear(ip) response = JSONResponse({"user": username}) response.set_cookie( SESSION_COOKIE, signer.dumps({"user": username}), max_age=SESSION_MAX_AGE, httponly=True, samesite="lax", path=BASE, ) _log.info("admin login ok: user=%s ip=%s", username, ip) await audit.log(runtime, actor=f"admin:{username}", kind="login_ok", ip=ip) return response @app.post("/auth/logout", status_code=status.HTTP_204_NO_CONTENT) async def logout(request: Request) -> Response: user = _current_user(request, signer) response = Response(status_code=status.HTTP_204_NO_CONTENT) response.delete_cookie(SESSION_COOKIE, path=BASE) if user: await audit.log(runtime, actor=f"admin:{user}", kind="logout") return response @app.get("/auth/session") async def session(request: Request) -> dict[str, Any]: user = _current_user(request, signer) if user is None: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not signed in") return {"user": user, "token": token} @app.get("/") @app.get("/{path:path}") async def spa(path: str = "") -> Response: if path: target = (ui_dir / path).resolve() if target.is_relative_to(ui_dir) and target.is_file(): headers = ( {"Cache-Control": "public, max-age=31536000, immutable"} if "/immutable/" in f"/{path}" else {} ) return FileResponse(target, headers=headers) if index.is_file(): return FileResponse(index, headers={"Cache-Control": "no-store"}) return Response( "ui/build is missing - run `make ui-build`.", status_code=status.HTTP_503_SERVICE_UNAVAILABLE, media_type="text/plain", ) @app.exception_handler(HTTPException) async def http_error(_request: Request, exc: HTTPException) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content={"error": exc.detail}, headers=exc.headers, ) return app def _current_user( request: Request, signer: itsdangerous.URLSafeTimedSerializer ) -> str | None: raw = request.cookies.get(SESSION_COOKIE) if not raw: return None try: payload = signer.loads(raw, max_age=SESSION_MAX_AGE) except itsdangerous.BadData: return None user = payload.get("user") if isinstance(payload, dict) else None return user if isinstance(user, str) else None def _client_ip(request: Request) -> str: cf = request.headers.get("cf-connecting-ip") if cf: return cf.strip() xff = request.headers.get("x-forwarded-for") if xff: first = xff.split(",", 1)[0].strip() if first: return first return request.client.host if request.client else "unknown" class _LoginRateLimit: """Sliding-window failure counter per source IP for ``POST /admin/auth/login``.""" __slots__ = ("_failures", "_max_attempts", "_window") def __init__(self, *, max_attempts: int, window: float) -> None: self._failures: dict[str, deque[float]] = {} self._max_attempts = max_attempts self._window = window def check(self, ip: str) -> bool: bucket = self._failures.get(ip) if bucket is None: return True cutoff = time.monotonic() - self._window while bucket and bucket[0] < cutoff: bucket.popleft() if not bucket: self._failures.pop(ip, None) return True return len(bucket) < self._max_attempts def record_failure(self, ip: str) -> None: bucket = self._failures.setdefault(ip, deque()) cutoff = time.monotonic() - self._window while bucket and bucket[0] < cutoff: bucket.popleft() bucket.append(time.monotonic()) def clear(self, ip: str) -> None: self._failures.pop(ip, None)