"""
admin.py — LibBee v3.8
Admin dashboard: session management, config store, RAG rebuild, metrics, system status.
Changes over v3.2:
- Login lockout: 5 failed attempts → 15 minute lockout per IP
- Remaining attempts shown in error message
- secure=True on session cookie
- Login page JS shows server error message (lockout, remaining attempts)
"""
# ── Imports ────────────────────────────────────────────────────────────────────
import json
import os
import secrets
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
from src.config import get_settings, LIBBEE_VERSION
from src.services.runtime_store import JsonRuntimeStore
router = APIRouter()
_START_TIME = time.time()
_CONFIG_DEFAULTS: Dict[str, Any] = {
"max_results": 5,
"maintenance_mode": False,
"welcome_message": "Hi! I'm LibBee, the Khalifa University Library AI Assistant.",
"custom_instructions": "",
"announcement": "",
"maintenance_message": "LibBee is currently under maintenance. Please try again shortly.",
}
# ── Session Persistence ────────────────────────────────────────────────────────
admin_sessions: Dict[str, float] = {}
# ── Admin login lockout ────────────────────────────────────────────────────────
# Tracks failed login attempts per IP.
# After _MAX_LOGIN_ATTEMPTS failures within _LOCKOUT_WINDOW seconds,
# that IP is locked out for _LOCKOUT_DURATION seconds.
_login_attempts: Dict[str, list] = {} # ip → [timestamp, ...]
_MAX_LOGIN_ATTEMPTS = 5
_LOCKOUT_WINDOW = 300 # 5 minutes — window in which attempts are counted
_LOCKOUT_DURATION = 900 # 15 minutes — lockout duration after max attempts
def _sessions_file() -> str:
cfg_path = os.environ.get("CONFIG_STORE_PATH", "")
if cfg_path and "/data" in cfg_path:
return os.path.join(os.path.dirname(cfg_path), "admin_sessions.json")
return "/tmp/libbee_admin_sessions.json"
def _load_sessions() -> None:
global admin_sessions
try:
with open(_sessions_file(), "r", encoding="utf-8") as f:
data = json.load(f)
now = time.time()
valid = {t: exp for t, exp in data.items()
if isinstance(exp, (int, float)) and exp > now}
admin_sessions.update(valid)
except Exception:
pass
def _save_sessions() -> None:
try:
now = time.time()
valid = {t: exp for t, exp in admin_sessions.items() if exp > now}
with open(_sessions_file(), "w", encoding="utf-8") as f:
json.dump(valid, f)
except Exception:
pass
def _cleanup_sessions() -> None:
now = time.time()
expired = [t for t, exp in admin_sessions.items() if exp <= now]
for t in expired:
admin_sessions.pop(t, None)
try:
_load_sessions()
except Exception:
pass
def _require_session(request: Request) -> None:
_cleanup_sessions()
token = request.cookies.get("admin_session")
if not token or token not in admin_sessions:
raise HTTPException(status_code=401, detail="Not authenticated")
def _has_session(request: Request) -> bool:
_cleanup_sessions()
token = request.cookies.get("admin_session")
return bool(token and token in admin_sessions)
# ── Config Store ───────────────────────────────────────────────────────────────
def _config_store() -> JsonRuntimeStore:
settings = get_settings()
return JsonRuntimeStore(settings.config_path, default=_CONFIG_DEFAULTS)
def _load_config() -> Dict[str, Any]:
data = _config_store().load()
for k, v in _CONFIG_DEFAULTS.items():
if k not in data:
data[k] = v
return data
def _esc(s: Any) -> str:
return (
str(s or "")
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
)
# ── Pydantic Models ────────────────────────────────────────────────────────────
class LoginRequest(BaseModel):
password: str
class ConfigUpdate(BaseModel):
key: str
value: str
class ConfigBulkUpdate(BaseModel):
settings: Dict[str, Any]
# ── Auth Endpoints ─────────────────────────────────────────────────────────────
@router.post("/auth")
async def admin_auth(req: LoginRequest, request: Request):
settings = get_settings()
client_ip = request.client.host if request.client else "unknown"
now = time.time()
# ── Lockout check ─────────────────────────────────────────────────────────
attempts = _login_attempts.get(client_ip, [])
attempts = [t for t in attempts if now - t < _LOCKOUT_WINDOW]
if len(attempts) >= _MAX_LOGIN_ATTEMPTS:
wait = int(_LOCKOUT_DURATION - (now - attempts[0]))
raise HTTPException(
status_code=429,
detail=f"Too many failed login attempts. Try again in {wait} seconds."
)
# ── Password check ────────────────────────────────────────────────────────
if not settings.admin_password or not secrets.compare_digest(
req.password.encode(), settings.admin_password.encode()
):
attempts.append(now)
_login_attempts[client_ip] = attempts
remaining = _MAX_LOGIN_ATTEMPTS - len(attempts)
if remaining <= 0:
detail = f"Too many failed login attempts. Try again in {int(_LOCKOUT_DURATION / 60)} minutes."
else:
detail = f"Invalid password. {remaining} attempt{'s' if remaining != 1 else ''} remaining before lockout."
raise HTTPException(status_code=401, detail=detail)
# ── Success — clear attempts and issue session ─────────────────────────────
_login_attempts.pop(client_ip, None)
token = secrets.token_hex(32)
admin_sessions[token] = time.time() + 86400
_save_sessions()
resp = Response(content='{"status":"ok"}', media_type="application/json")
resp.set_cookie(key="admin_session", value=token, httponly=True,
max_age=86400, samesite="lax", secure=True)
return resp
@router.post("/logout")
async def admin_logout_post(request: Request):
token = request.cookies.get("admin_session")
if token:
admin_sessions.pop(token, None)
_save_sessions()
resp = Response(content='{"status":"ok"}', media_type="application/json")
resp.delete_cookie("admin_session")
return resp
@router.get("/logout")
async def admin_logout_get(request: Request):
token = request.cookies.get("admin_session")
if token:
admin_sessions.pop(token, None)
resp = RedirectResponse(url="/admin/login")
resp.delete_cookie("admin_session")
return resp
# ── Login Page ─────────────────────────────────────────────────────────────────
@router.get("/login", response_class=HTMLResponse)
async def admin_login_page(request: Request):
if _has_session(request):
return RedirectResponse(url="/admin")
return HTMLResponse("""
KU Library AI — Admin Login
🔐 Admin Login
KU Library AI Dashboard
Incorrect password. Try again.
""")
# ── Admin Dashboard ────────────────────────────────────────────────────────────
@router.get("", response_class=HTMLResponse)
async def admin_dashboard(request: Request):
if not _has_session(request):
return RedirectResponse(url="/admin/login")
try:
cfg = _load_config()
except Exception:
cfg = dict(_CONFIG_DEFAULTS)
mm_on = "selected" if cfg.get("maintenance_mode") else ""
mm_off = "" if cfg.get("maintenance_mode") else "selected"
# Worker credentials injected into this page only. The dashboard tries the
# server-side proxy first and falls back to a direct browser -> Worker call,
# because some hosting environments (HF Spaces) cannot reach workers.dev
# while the admin's own browser can. This page is served exclusively to an
# authenticated admin session, so the token is not publicly exposed.
_s = get_settings()
cf_url = (_s.cloudflare_worker_url or "").strip().strip('"').strip("'").rstrip("/")
cf_token = (_s.cloudflare_worker_token or "").strip()
return HTMLResponse(f"""
KU Library AI — Admin
Rebuild the FAISS + BM25 hybrid index from knowledge files in the KB directory.
Loading RAG status…
📋 Request Metrics
Loading…
📈 Cloudflare Analytics (Persistent)
Loading…
📋 Recent Queries (Cloudflare D1)
Loading…
🔌 System Status
Loading…
""")
# ── API Endpoints ──────────────────────────────────────────────────────────────
@router.get("/public-config")
async def public_config():
"""No auth — returns public fields for the chat frontend."""
try:
cfg = _load_config()
except Exception:
cfg = {}
return {
"welcome_message": cfg.get("welcome_message", ""),
"announcement": cfg.get("announcement", ""),
"maintenance_mode": cfg.get("maintenance_mode", False),
"maintenance_message": cfg.get("maintenance_message", ""),
}
@router.get("/config")
async def get_config(request: Request):
_require_session(request)
return _load_config()
@router.post("/config")
async def update_config(req: ConfigUpdate, request: Request):
_require_session(request)
allowed = {"max_results", "maintenance_mode", "welcome_message",
"custom_instructions", "announcement", "maintenance_message"}
if req.key not in allowed:
raise HTTPException(status_code=400,
detail=f"Unsupported key. Allowed: {sorted(allowed)}")
data = _config_store().load()
value: Any = req.value
if req.key == "max_results":
value = int(req.value)
elif req.key == "maintenance_mode":
value = str(req.value).lower() in {"1", "true", "yes", "on"}
data[req.key] = value
_config_store().save(data)
return {"status": "ok", "updated": req.key, "value": value}
@router.post("/config-bulk")
async def update_config_bulk(req: ConfigBulkUpdate, request: Request):
_require_session(request)
data = _config_store().load()
for key, value in req.settings.items():
if key == "max_results":
try:
data[key] = int(value)
except (ValueError, TypeError):
pass
elif key == "maintenance_mode":
data[key] = str(value).lower() in {"1", "true", "yes", "on"}
else:
data[key] = str(value)
_config_store().save(data)
return {"status": "ok", "updated": list(req.settings.keys())}
# ── RAG Management Endpoints ───────────────────────────────────────────────────
@router.get("/rag-status")
async def rag_status(request: Request):
_require_session(request)
try:
from app import get_rag_service
return get_rag_service().get_stats()
except Exception as e:
return {"error": str(e), "ready": False}
@router.post("/rebuild")
async def rebuild_rag(request: Request):
_require_session(request)
try:
from app import get_rag_service
svc = get_rag_service()
settings = get_settings()
if not settings.openai_api_key:
return {"status": "error", "error": "OPENAI_API_KEY not set"}
svc._kb_hash = "" # force fresh build
await svc.initialize(openai_api_key=settings.openai_api_key)
return {"status": "ok", "chunks": len(svc.bm25_corpus), "ready": svc.is_ready()}
except Exception as e:
return {"status": "error", "error": str(e)}
# ── Metrics & Status Endpoints ─────────────────────────────────────────────────
@router.get("/metrics")
async def metrics(request: Request):
_require_session(request)
try:
from app import get_metrics_service
return get_metrics_service().snapshot()
except Exception as e:
return {"error": str(e)}
@router.get("/status")
async def system_status(request: Request):
_require_session(request)
settings = get_settings()
mem_used_mb = mem_pct = 0
try:
with open("/sys/fs/cgroup/memory.current") as f:
used = int(f.read().strip())
with open("/sys/fs/cgroup/memory.max") as f:
val = f.read().strip()
total = int(val) if val != "max" else 16 * 1024 ** 3
mem_used_mb = round(used / 1024 / 1024)
mem_pct = round(used / total * 100)
except Exception:
try:
with open("/sys/fs/cgroup/memory/memory.usage_in_bytes") as f:
used = int(f.read().strip())
with open("/sys/fs/cgroup/memory/memory.limit_in_bytes") as f:
total = int(f.read().strip())
mem_used_mb = round(used / 1024 / 1024)
mem_pct = round(used / total * 100)
except Exception:
pass
rag_ready = False
rag_chunks = 0
kb_files = 0
try:
from app import get_rag_service
svc = get_rag_service()
rag_ready = svc.is_ready()
rag_chunks = len(svc.bm25_corpus)
kb_dir = settings.kb_dir
kb_files = len(list(kb_dir.glob("*.txt"))) if kb_dir.exists() else 0
except Exception:
pass
cfg = _load_config()
return {
"status": "ok",
"rag_ready": rag_ready,
"rag_chunks": rag_chunks,
"kb_files": kb_files,
"openai": bool(settings.openai_api_key),
"anthropic": bool(settings.anthropic_api_key),
"primo": bool(settings.primo_api_key),
"mem_used_mb": mem_used_mb,
"mem_pct": mem_pct,
"uptime_secs": round(time.time() - _START_TIME),
"maintenance": bool(cfg.get("maintenance_mode")),
"version": LIBBEE_VERSION,
}
# ── Daily KB-gap / relevance reports (generated by the Cloudflare Worker) ─────
# The Worker's daily cron writes one report row per day; these endpoints proxy
# them into the admin dashboard. Session-protected like every admin route, and
# the Worker itself additionally requires the shared bearer token.
@router.get("/reports")
async def admin_reports(request: Request, days: int = 14):
_require_session(request)
return await _worker_get("/reports", {"days": max(1, min(days, 60))})
@router.post("/reports/{report_id}/reviewed")
async def admin_report_reviewed(report_id: int, request: Request):
_require_session(request)
import httpx as _hx
from src.config import get_settings as _gs
settings = _gs()
worker = (settings.cloudflare_worker_url or "").rstrip("/")
if not worker:
raise HTTPException(status_code=503, detail="CLOUDFLARE_WORKER_URL not configured")
headers = {}
if settings.cloudflare_worker_token:
headers["Authorization"] = f"Bearer {settings.cloudflare_worker_token}"
try:
async with _hx.AsyncClient(timeout=10) as client:
r = await client.post(f"{worker}/reports/{int(report_id)}/reviewed", headers=headers)
return r.json()
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Worker unreachable: {exc}")
# ── Analytics proxies ────────────────────────────────────────────────────────
# The dashboard fetches Worker analytics through these session-protected
# proxies so the Worker URL and ANALYTICS_TOKEN never reach the browser.
async def _worker_get(path: str, params: Optional[dict] = None):
from src.services.http_client import make_client as _mk
from src.config import get_settings as _gs
settings = _gs()
worker = (settings.cloudflare_worker_url or "").rstrip("/")
if not worker:
raise HTTPException(status_code=503, detail="CLOUDFLARE_WORKER_URL not configured")
headers = {}
if settings.cloudflare_worker_token:
headers["Authorization"] = f"Bearer {settings.cloudflare_worker_token}"
try:
async with _mk(timeout=8) as client:
r = await client.get(f"{worker}{path}", params=params or {}, headers=headers)
return r.json()
except HTTPException:
raise
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Worker unreachable: {exc}")
@router.get("/analytics")
async def admin_analytics(request: Request):
_require_session(request)
return await _worker_get("/analytics")
@router.get("/analytics/recent")
async def admin_analytics_recent(request: Request):
_require_session(request)
return await _worker_get("/analytics/recent")
@router.get("/worker-health")
async def admin_worker_health(request: Request):
"""Diagnose the backend -> Worker -> D1 chain.
Every probe runs independently and records its own outcome, including the
exception type, so one failure never hides the rest. Also reports the exact
URL being used (whitespace or quotes in the secret are a common cause).
"""
_require_session(request)
import asyncio as _aio
from src.services.http_client import make_client as _mk
from src.config import get_settings as _gs, LIBBEE_VERSION as _ver
settings = _gs()
raw = settings.cloudflare_worker_url or ""
worker = raw.strip().strip('"').strip("'").rstrip("/")
token = (settings.cloudflare_worker_token or "").strip()
out = {
"backend_version": _ver,
"worker_url_configured": bool(worker),
"worker_url_used": worker,
"worker_url_had_whitespace_or_quotes": raw != worker + ("/" if raw.endswith("/") else ""),
"worker_token_configured": bool(token),
"worker_token_length": len(token),
"probes": {},
"verdict": "",
}
if not worker:
out["verdict"] = "CLOUDFLARE_WORKER_URL is not set in the Space secrets."
return out
headers = {"Authorization": f"Bearer {token}"} if token else {}
async def probe(name, path, use_auth=True):
"""One probe, its own short timeout, never raises."""
entry = {"path": path}
try:
async with _mk(timeout=8) as client:
r = await client.get(f"{worker}{path}",
headers=headers if use_auth else {})
entry["status"] = r.status_code
entry["body"] = r.text[:300]
except Exception as exc:
entry["error_type"] = type(exc).__name__
entry["error"] = repr(exc)[:200] or type(exc).__name__
out["probes"][name] = entry
return entry
# Run all probes concurrently so the endpoint returns in ~6s, not ~48s.
ver, _hours, ana, sch = await _aio.gather(
probe("version", "/version", use_auth=False),
probe("hours", "/hours", use_auth=False),
probe("analytics", "/analytics"),
probe("schema", "/analytics/schema"),
)
# Optional write probe: ?test_write=1 posts one marker row so a POST can be
# compared against the GETs above. It inserts a row tagged 'healthcheck'.
if request.query_params.get("test_write") == "1":
entry = {"path": "/log", "method": "POST"}
try:
async with _mk(timeout=10) as client:
r = await client.post(
f"{worker}/log", headers=headers,
json={"question": "healthcheck", "tool": "healthcheck:probe",
"model": "n/a", "response_time": 0, "result_count": 0},
)
entry["status"] = r.status_code
entry["body"] = r.text[:200]
except Exception as exc:
entry["error_type"] = type(exc).__name__
entry["error"] = repr(exc)[:200] or type(exc).__name__
out["probes"]["log_write"] = entry
# Interpretation
if "error_type" in ver and "error_type" in out["probes"]["hours"]:
et = ver.get("error_type")
if et in ("ConnectTimeout", "ConnectError", "ReadTimeout"):
out["verdict"] = (
f"{et}: the Space cannot open a connection to workers.dev, even "
f"on unauthenticated routes. Other outbound calls (OpenAI, "
f"OpenAlex) work, so this is host-specific - typically an "
f"unroutable IPv6 path or Cloudflare throttling shared "
f"workers.dev addresses from datacentre IPs. IPv4 is now forced; "
f"if this persists, put the Worker behind a custom domain "
f"(Worker > Settings > Domains & Routes > Add custom domain) and "
f"point CLOUDFLARE_WORKER_URL at that instead."
)
else:
out["verdict"] = (
f"No HTTP response from the Worker ({et}). Check that "
f"worker_url_used above is exactly your workers.dev address."
)
elif ver.get("status") == 404:
out["verdict"] = (
"Worker responds but has no /version route: the deployed script is "
"still the pre-3.8 version. Deploy cloudflare-worker-v3.8.1.js."
)
elif ana.get("status") == 401:
out["verdict"] = (
"Analytics returns 401. CLOUDFLARE_WORKER_TOKEN does not match the "
"Worker's ANALYTICS_TOKEN (check for trailing spaces in either)."
)
elif ana.get("status") == 200:
migrated = None
if sch.get("status") == 200:
try:
import json as _json
migrated = (_json.loads(sch["body"]) or {}).get("migration_applied")
except Exception:
migrated = None
if migrated is False:
out["verdict"] = ("Connected, but the D1 migration is not applied "
"(answer_excerpt/source_ids missing).")
else:
out["verdict"] = "OK - backend, Worker and D1 analytics are connected."
else:
out["verdict"] = (f"Unexpected analytics status {ana.get('status')}. "
f"See probes for details.")
return out
@router.get("/log-status")
async def admin_log_status(request: Request):
"""Outcome of recent analytics writes, without needing container logs.
'recent_attempts' is newest-first. Read it like this:
ok -> the write reached the Worker (HTTP 2xx)
throttled -> HTTP 403/429: the host is being rate-limited by Cloudflare
unreachable -> no HTTP response; exception type is in 'detail'
rejected_401 -> CLOUDFLARE_WORKER_TOKEN does not match ANALYTICS_TOKEN
Buffered rows are retried automatically on later queries.
"""
_require_session(request)
from src.agentcore.orchestrator import analytics_log_status
from src.config import get_settings as _gs
st = _gs()
data = analytics_log_status()
data["worker_url_configured"] = bool(st.cloudflare_worker_url)
data["worker_token_configured"] = bool(st.cloudflare_worker_token)
if not data["recent_attempts"]:
data["hint"] = ("No write attempts recorded since the last restart. "
"Send a query through LibBee, then reload this page.")
return data