""" admin.py — ResearchBee admin dashboard Ported from LibBee v3.8's admin module so the two dashboards behave identically: same cookie-session login, same 5-attempt IP lockout, same server-rendered single page, same visual language. Someone who administers LibBee should not have to learn a second interface. What is deliberately different ------------------------------ LibBee's dashboard shows recent questions. This one cannot, because ResearchBee analytics contain counts, not question text — see usage.py. The equivalent panels here answer "how much of each tool is being used" and "how often could we not answer", both from counters. Mount with: from admin import router as admin_router app.include_router(admin_router, prefix="/admin") Environment: RESEARCHBEE_ADMIN_PASSWORD required — the dashboard is unreachable without it CLOUDFLARE_WORKER_URL the LibBee Worker (shared) CLOUDFLARE_WORKER_TOKEN must equal the Worker's ANALYTICS_TOKEN """ from __future__ import annotations import json import hashlib import hmac import logging import math import stat import tempfile from pathlib import Path import os import secrets import time from typing import Any, Dict, Optional from fastapi import APIRouter, HTTPException, Request, Response from fastapi.responses import HTMLResponse, RedirectResponse from pydantic import BaseModel, Field from runtime_security import get_client_ip router = APIRouter() logger = logging.getLogger(__name__) _NO_STORE = {"Cache-Control": "no-store", "Pragma": "no-cache"} _START_TIME = time.time() ADMIN_PASSWORD = (os.environ.get("RESEARCHBEE_ADMIN_PASSWORD") or "").strip() WORKER_URL = (os.environ.get("CLOUDFLARE_WORKER_URL") or "").strip().strip('"').strip("'").rstrip("/") WORKER_TOKEN = (os.environ.get("CLOUDFLARE_WORKER_TOKEN") or "").strip() _SESSIONS_FILE = os.getenv("RB_ADMIN_SESSIONS_FILE", "/tmp/researchbee_admin_sessions.json") _MAX_SESSIONS = 256 _SESSION_TTL = 86400 admin_sessions: Dict[str, float] = {} # Failed logins per IP. Five misses inside the window locks that IP out # for fifteen minutes — enough to make guessing a shared password # impractical without locking out a librarian who fat-fingers it twice. _login_attempts: Dict[str, list] = {} _locked_until: Dict[str, float] = {} _MAX_LOGIN_IDENTITIES = 4096 _MAX_LOGIN_ATTEMPTS = 5 _LOCKOUT_WINDOW = 300 _LOCKOUT_DURATION = 900 # ── Sessions ────────────────────────────────────────────────────────── def _session_signature(sessions: dict) -> str: payload = json.dumps(sessions, sort_keys=True, separators=(",", ":")).encode() return hmac.new(ADMIN_PASSWORD.encode(), payload, hashlib.sha256).hexdigest() def _load_sessions() -> None: """Read the password-bound v2 envelope; old/rotated sessions fail closed.""" admin_sessions.clear() if not ADMIN_PASSWORD: return try: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) fd = os.open(_SESSIONS_FILE, flags) with os.fdopen(fd, "r", encoding="utf-8") as handle: info = os.fstat(handle.fileno()) if not stat.S_ISREG(info.st_mode) or info.st_size > 131072: return data = json.load(handle) if not isinstance(data, dict) or data.get("version") != 2: return sessions = data.get("sessions") signature = data.get("signature") if not isinstance(sessions, dict) or len(sessions) > _MAX_SESSIONS: return if not isinstance(signature, str) or not hmac.compare_digest( signature.encode(), _session_signature(sessions).encode() ): return now = time.time() admin_sessions.update({ token: expiry for token, expiry in sessions.items() if isinstance(token, str) and len(token) == 64 and all(c in "0123456789abcdef" for c in token) and isinstance(expiry, (int, float)) and not isinstance(expiry, bool) and math.isfinite(expiry) and now < expiry <= now + _SESSION_TTL }) except (OSError, ValueError, TypeError): # Missing, malformed or inaccessible session state never grants access. return def _save_sessions() -> bool: """Atomic replacement in the same directory; bearer tokens are mode 0600.""" temp_path = None try: now = time.time() sessions = {token: expiry for token, expiry in admin_sessions.items() if expiry > now} envelope = {"version": 2, "sessions": sessions, "signature": _session_signature(sessions)} target = Path(_SESSIONS_FILE) fd, temp_path = tempfile.mkstemp(prefix=target.name + ".", dir=str(target.parent)) with os.fdopen(fd, "w", encoding="utf-8") as handle: os.fchmod(handle.fileno(), 0o600) json.dump(envelope, handle) handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, target) return True except (OSError, TypeError, ValueError) as exc: logger.warning("Admin session persistence failed (%s)", type(exc).__name__) return False finally: if temp_path: try: os.unlink(temp_path) except OSError: pass def _cleanup_sessions() -> None: # Reload authoritative state so logout is not undone by a stale copy. # This envelope verifies with the CURRENT configured password, so a Space # restart following password rotation invalidates all previous sessions. _load_sessions() def _cleanup_login_attempts(now: float) -> None: for ip, deadline in list(_locked_until.items()): if deadline <= now: _locked_until.pop(ip, None) _login_attempts.pop(ip, None) for ip, attempts in list(_login_attempts.items()): recent = [value for value in attempts if now - value < _LOCKOUT_WINDOW] if recent: _login_attempts[ip] = recent elif ip not in _locked_until: _login_attempts.pop(ip, None) def _has_session(request: Request) -> bool: _cleanup_sessions() token = request.cookies.get("rb_admin_session") return bool(token and token in admin_sessions) def _require_session(request: Request) -> None: if not _has_session(request): raise HTTPException(status_code=401, detail="Not authenticated") class LoginRequest(BaseModel): password: str = Field(..., min_length=1, max_length=1024) # ── Auth ────────────────────────────────────────────────────────────── @router.post("/auth") async def admin_auth(req: LoginRequest, request: Request): client_ip = get_client_ip(request) now = time.time() _cleanup_login_attempts(now) deadline = _locked_until.get(client_ip, 0) if deadline > now: wait = max(1, math.ceil(deadline - now)) raise HTTPException(429, "Too many failed attempts. Please try again later.", headers={"Retry-After": str(wait), **_NO_STORE}) if client_ip not in _login_attempts and len(_login_attempts) >= _MAX_LOGIN_IDENTITIES: raise HTTPException(429, "Login is busy. Please try again shortly.", headers={"Retry-After": "60", **_NO_STORE}) attempts = _login_attempts.get(client_ip, []) if not ADMIN_PASSWORD or not secrets.compare_digest( req.password.encode(), ADMIN_PASSWORD.encode() ): attempts.append(now) _login_attempts[client_ip] = attempts if len(attempts) >= _MAX_LOGIN_ATTEMPTS: _locked_until[client_ip] = now + _LOCKOUT_DURATION raise HTTPException(429, "Too many failed attempts. Try again in 15 minutes.", headers={"Retry-After": str(_LOCKOUT_DURATION), **_NO_STORE}) raise HTTPException(401, "Sign-in failed. Check your password or contact the administrator.", headers=_NO_STORE) _login_attempts.pop(client_ip, None) _locked_until.pop(client_ip, None) _cleanup_sessions() if len(admin_sessions) >= _MAX_SESSIONS: admin_sessions.pop(min(admin_sessions, key=admin_sessions.get), None) token = secrets.token_hex(32) admin_sessions[token] = now + _SESSION_TTL if not _save_sessions(): admin_sessions.pop(token, None) raise HTTPException(503, "Sign-in is temporarily unavailable. Please try again later.", headers=_NO_STORE) resp = Response(content='{"status":"ok"}', media_type="application/json", headers=_NO_STORE) resp.set_cookie("rb_admin_session", token, httponly=True, max_age=_SESSION_TTL, samesite="lax", secure=True) return resp @router.get("/logout") async def admin_logout(request: Request): _cleanup_sessions() token = request.cookies.get("rb_admin_session") if token: admin_sessions.pop(token, None) if not _save_sessions(): raise HTTPException(503, "Sign-out could not be completed. Please try again.", headers=_NO_STORE) resp = RedirectResponse(url="/admin/login", headers=_NO_STORE) resp.delete_cookie("rb_admin_session") return resp # ── Worker proxies ──────────────────────────────────────────────────── # The Worker URL and token stay server-side. The browser only ever talks # to this Space, so neither secret reaches the page. async def _worker_get(path: str, params: Optional[dict] = None): if not WORKER_URL: raise HTTPException(503, "CLOUDFLARE_WORKER_URL is not set on this Space") import httpx headers = {"Authorization": f"Bearer {WORKER_TOKEN}"} if WORKER_TOKEN else {} try: async with httpx.AsyncClient(timeout=15) as client: r = await client.get(f"{WORKER_URL}{path}", params=params or {}, headers=headers) if not r.is_success: raise HTTPException(502, "Analytics service is temporarily unavailable.") data = r.json() if not isinstance(data, dict): raise HTTPException(502, "Analytics service returned an invalid response.") return data except HTTPException: raise except Exception as exc: logger.warning("Analytics proxy failed (%s)", type(exc).__name__) raise HTTPException(502, "Analytics service is temporarily unavailable. Counts remain buffered.") @router.get("/usage") async def admin_usage(request: Request, days: int = 30): _require_session(request) return await _worker_get("/rb/usage", {"days": max(0, min(days, 3650))}) @router.get("/canary-status") async def admin_canary_status(request: Request): _require_session(request) return await _worker_get("/rb/canary/status") @router.get("/local-usage") async def admin_local_usage(request: Request): """Counts held in this container since it last started. Useful when the Worker is unreachable, and as a cross-check that counting is actually happening before blaming the Worker.""" _require_session(request) from usage import usage return usage.snapshot() # ── Pull endpoints (called by the Cloudflare Worker, not by a browser) ─ # Authenticated with the shared token rather than an admin session, # because the caller is a machine. This is the direction that works: # Cloudflare can reach this Space even though this Space cannot reach # Cloudflare. def _require_worker_token(request: Request) -> None: if not WORKER_TOKEN: raise HTTPException(503, "CLOUDFLARE_WORKER_TOKEN is not set on this Space") supplied = request.headers.get("authorization", "") if not secrets.compare_digest(supplied.encode(), f"Bearer {WORKER_TOKEN}".encode()): raise HTTPException(401, "Unauthorized") def _worker_or_session(request: Request) -> None: """Accept either the Worker's token or a signed-in admin, so the same endpoints back both the cron pull and the dashboard's manual send.""" if _has_session(request): return _require_worker_token(request) @router.get("/usage-export") async def usage_export(request: Request): """Hand over accumulated counts. Counts are held, not deleted, until the caller confirms with /usage-ack — so a caller that reads them and then fails to store them cannot destroy them.""" _worker_or_session(request) from usage import usage return usage.export_snapshot() @router.post("/usage-ack") async def usage_ack(request: Request, snapshot_id: str): """Confirm a snapshot reached storage. Only now is it discarded.""" _worker_or_session(request) from usage import usage return {"acknowledged": usage.ack_snapshot(snapshot_id)} @router.get("/worker-probe") async def admin_worker_probe(request: Request): """Is this Space's route to the Worker broken, or just slow? HF Spaces lose connectivity to workers.dev for stretches at a time — LibBee hit the same thing and moved its analytics writes into the browser because of it. Three attempts with timings tell you which case you are in: three identical failures is a blocked route, a mix is the intermittent case, which the flush buffer already absorbs. """ _require_session(request) import httpx, time as _t if not WORKER_URL: return {"error": "CLOUDFLARE_WORKER_URL is not set on this Space"} attempts = [] for i in range(3): t0 = _t.time() try: async with httpx.AsyncClient(timeout=20) as client: r = await client.get(f"{WORKER_URL}/version") attempts.append({"attempt": i + 1, "ok": r.status_code == 200, "status": r.status_code, "ms": int((_t.time() - t0) * 1000)}) except Exception as exc: attempts.append({"attempt": i + 1, "ok": False, "error": type(exc).__name__, "ms": int((_t.time() - t0) * 1000)}) ok = sum(1 for a in attempts if a.get("ok")) return { "worker_url": WORKER_URL, "attempts": attempts, "verdict": ("reachable" if ok == 3 else "intermittent" if ok else "unreachable from this Space"), "note": ("Counts are buffered and retried every 60 seconds, so patchy " "egress delays the dashboard. Counts remain in this container until " "a server-side send or scheduled Worker pull succeeds; a container " "restart can lose buffered counts."), } @router.get("/status") async def admin_status(request: Request): _require_session(request) import app as rb_app out: Dict[str, Any] = { "uptime_seconds": int(time.time() - _START_TIME), "worker_url_configured": bool(WORKER_URL), "worker_token_configured": bool(WORKER_TOKEN), "admin_password_set": bool(ADMIN_PASSWORD), "openai_key_set": bool(os.environ.get("OPENAI_API_KEY")), } try: out["health"] = await rb_app.health() except Exception as exc: out["health"] = {"error": "Service status is temporarily unavailable."} return out @router.post("/reload-khazna") async def admin_reload_khazna(request: Request): """Session-protected wrapper around the existing reload endpoint, so the button on this page works without exposing an unauthenticated admin path.""" _require_session(request) import asyncio from app import khazna_index ok = await asyncio.to_thread(khazna_index.reload) return {"reloaded": ok, "khazna": khazna_index.status} @router.post("/usage-flush") async def admin_flush_usage(request: Request): """Send via the backend; never deliver a machine credential to a browser.""" _require_session(request) if not WORKER_URL or not WORKER_TOKEN: raise HTTPException(503, "Analytics service is not configured.") from usage import usage ok = await usage.flush() if not ok: raise HTTPException(502, "Counts remain buffered. The server cannot reach analytics yet; a scheduled Worker pull can collect them.") return {"sent": True, "message": "Pending counts sent, or no counts were waiting."} # ── Login page ──────────────────────────────────────────────────────── @router.get("/login", response_class=HTMLResponse) async def admin_login_page(request: Request): if _has_session(request): return RedirectResponse(url="/admin", headers=_NO_STORE) return HTMLResponse("""