"""Admin Panel — co-op-wide member overview, spend, and balance management.
Security model:
- Authentication is done by Cloudron's proxyAuth wall (SSO). Cloudron injects
the authenticated user's username via X-Remote-User.
- Access is further restricted to a small ADMIN_USERNAMES allowlist (checked
against the injected username), so only designated admins see this panel.
- This app calls the member portal's /admin/overview and /admin/set-balance
endpoints, authenticated with the portal's WEBHOOK_TOKEN (ADMIN_TOKEN).
Environment (Cloudron env vars):
PORTAL_BASE — base URL of the member portal
ADMIN_TOKEN — the portal's admin/webhook token (for /admin/* endpoints)
ADMIN_USERNAMES — comma-separated list of allowed Cloudron usernames
"""
import os
import logging
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("admin-panel")
PORTAL_BASE = os.environ.get("PORTAL_BASE", "").rstrip("/")
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "")
ADMIN_USERNAMES = {
u.strip().lower()
for u in os.environ.get("ADMIN_USERNAMES", "").split(",")
if u.strip()
}
app = FastAPI()
def get_identity(request: Request) -> str:
for header in (
"x-remote-user",
"x-forwarded-user",
"x-auth-request-user",
"x-auth-request-email",
"x-forwarded-email",
):
val = request.headers.get(header)
if val:
return val.strip()
return ""
def is_admin(request: Request) -> bool:
identity = get_identity(request)
if not identity:
return False
return identity.lower() in ADMIN_USERNAMES
async def portal_get(path: str) -> dict:
async with httpx.AsyncClient(timeout=20.0) as client:
r = await client.get(f"{PORTAL_BASE}{path}", headers={"X-Admin-Token": ADMIN_TOKEN})
if r.status_code != 200:
raise HTTPException(502, f"Portal error {r.status_code}")
return r.json()
async def portal_post(path: str, payload: dict) -> dict:
async with httpx.AsyncClient(timeout=20.0) as client:
r = await client.post(f"{PORTAL_BASE}{path}", json=payload, headers={"X-Admin-Token": ADMIN_TOKEN})
if r.status_code not in (200, 201):
try:
detail = r.json().get("detail", r.text)
except Exception:
detail = r.text
raise HTTPException(502, f"Portal error {r.status_code}: {detail}")
return r.json()
@app.get("/healthz")
async def healthz():
return {"status": "ok"}
@app.get("/")
async def index(request: Request):
if not is_admin(request):
return HTMLResponse(
"Forbidden
This panel is restricted to administrators.
",
status_code=403,
)
return HTMLResponse(render_page())
@app.get("/api/overview")
async def api_overview(request: Request):
if not is_admin(request):
return JSONResponse({"error": "forbidden"}, status_code=403)
return await portal_get("/admin/overview")
@app.post("/api/set-balance")
async def api_set_balance(request: Request):
if not is_admin(request):
return JSONResponse({"error": "forbidden"}, status_code=403)
body = await request.json()
email = (body.get("email") or "").strip().lower()
if not email:
return JSONResponse({"error": "Missing email"}, status_code=400)
payload = {"email": email}
if "add" in body:
payload["add"] = float(body["add"])
elif "balance" in body:
payload["balance"] = float(body["balance"])
else:
return JSONResponse({"error": "Provide 'balance' or 'add'"}, status_code=400)
try:
return await portal_post("/admin/set-balance", payload)
except HTTPException as e:
return JSONResponse({"error": e.detail}, status_code=e.status_code)
def _esc(s: str) -> str:
return s.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
def render_page() -> str:
return PAGE_HTML
PAGE_HTML = """
Inference Cooperative · Admin
Inference Cooperative
Admin panel
| Member | Status | Activated | Balance | Spend | Remaining | Reset | Adjust |
| Loading… |
"""
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse({"error": exc.detail}, status_code=exc.status_code)