from __future__ import annotations import hmac import os import time from datetime import timezone from typing import Any, Callable from fastapi import HTTPException, Request from fastapi.responses import JSONResponse from core import PLANS, STORE, human_bytes, parse_dt from features_v23 import plan_features, update_plan_features, send_admin_notification from resource_monitor import RESOURCE_MONITOR ADMIN_ID = "cloudvault-admin" def _configured_key() -> str: return os.getenv("ADMIN_API_KEY", "").strip() def _authorize(request: Request) -> None: expected = _configured_key() if not expected: raise HTTPException(status_code=503, detail="ADMIN_API_KEY is not configured.") provided = request.headers.get("X-CloudVault-Admin-Key", "").strip() if not provided: auth = request.headers.get("Authorization", "") if auth.lower().startswith("bearer "): provided = auth[7:].strip() if not provided or not hmac.compare_digest(provided, expected): raise HTTPException(status_code=401, detail="Invalid admin API key.") def _user_status(user: dict[str, Any]) -> str: if user.get("is_banned"): until = parse_dt(user.get("ban_until")) if until: return f"Banned until {until.astimezone(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}" return "Banned permanently" if user.get("delete_on_expiry"): return "Deletes on package expiry" return "Active" def _system_metrics() -> dict[str, Any]: """Return real CloudVault container usage without percentage conversion.""" snapshot = RESOURCE_MONITOR.snapshot() try: import psutil # type: ignore process = psutil.Process(os.getpid()) process_threads = int(process.num_threads()) uptime_seconds = max(0, int(time.time() - process.create_time())) except Exception: process_threads = 0 uptime_seconds = 0 return { # Average CPU cores used during the latest sample. Example: 0.26 vCPU. "cpu_used_vcpu": snapshot.cpu_used_vcpu, # Current memory used by the whole CloudVault Linux container. "ram_used_bytes": snapshot.ram_used_bytes, "ram_used_gb": snapshot.ram_used_gb, "effective_vcpu": snapshot.effective_vcpu, "logical_cpu_visible": snapshot.logical_cpu_visible, "sample_seconds": snapshot.sample_seconds, "metrics_updated_at": snapshot.updated_at_unix, "cpu_source": snapshot.cpu_source, "ram_source": snapshot.ram_source, "process_threads": process_threads, "uptime_seconds": uptime_seconds, "metrics_scope": "cloudvault_container", } def register_admin_api(demo: Any, active_users_provider: Callable[[], int] | None = None) -> None: """Register secure CloudVault desktop-admin endpoints on a Gradio app.""" def active_count() -> int: try: return max(0, int(active_users_provider() if active_users_provider else 0)) except Exception: return 0 async def overview(request: Request) -> JSONResponse: _authorize(request) overview_data = STORE.admin_overview(ADMIN_ID) users = [] for user in STORE.list_users(ADMIN_ID): plan_key = str(user.get("plan", "free")) used = STORE.used_bytes(user["id"]) expires = parse_dt(user.get("plan_expires_at")) users.append({ "id": user["id"], "username": user.get("username", "User"), "plan": plan_key, "plan_name": PLANS.get(plan_key, PLANS["free"]).name, "used_bytes": used, "used_text": human_bytes(used), "status": _user_status(user), "is_banned": bool(user.get("is_banned")), "ban_until": user.get("ban_until"), "expires_at": expires.isoformat() if expires else None, "created_at": user.get("created_at"), }) room_state = STORE.chat_room_state(ADMIN_ID) response = { "ok": True, "timestamp": time.time(), "active_users": active_count(), "system": _system_metrics(), "storage": { "limit_bytes": overview_data["limit"], "used_bytes": overview_data["total_used"], "remaining_bytes": overview_data["remaining"], "reserved_bytes": overview_data["reserved"], "used_by_plan": overview_data["used_by_plan"], }, "accounts": { "registered": overview_data["registered"], "counts": overview_data["counts"], "users": users, }, "config": overview_data["config"], "plan_features": {k: plan_features(k) for k in ("free","plus","pro")}, "support": {"runtime": STORE.support_runtime(), "tickets": STORE.admin_support_tickets(ADMIN_ID)}, "chat": { "messages": overview_data.get("chat_messages", 0), "mode": room_state.get("mode", "single"), "capacity": room_state.get("capacity", 200), "rooms": room_state.get("rooms", []), }, } return JSONResponse(response, headers={"Cache-Control": "no-store"}) async def chat(request: Request) -> JSONResponse: _authorize(request) room_id = str(request.query_params.get("room_id", "global-1")).strip() or "global-1" messages = STORE.chat_messages(ADMIN_ID, room_id) output = [] for message in messages: created = parse_dt(message.get("created_at")) output.append({ "id": message.get("id"), "room_id": message.get("room_id", "global-1"), "sender_name": message.get("sender_name", "User"), "sender_plan": message.get("sender_plan", "plus"), "sender_label": "👑 admin" if message.get("sender_plan") == "admin" else str(message.get("sender_name", "Kullanıcı")), "text": message.get("text", ""), "created_at": message.get("created_at"), "time_text": created.astimezone(timezone.utc).strftime("%d.%m.%Y %H:%M UTC") if created else "", }) return JSONResponse( {"ok": True, "room_id": room_id, "messages": output}, headers={"Cache-Control": "no-store"}, ) async def support(request: Request) -> JSONResponse: _authorize(request) ticket_id = str(request.query_params.get("ticket_id", "")).strip() if ticket_id: ticket = STORE.get_support_ticket(ADMIN_ID, ticket_id) return JSONResponse({"ok": bool(ticket), "ticket": ticket}, headers={"Cache-Control": "no-store"}) return JSONResponse({"ok": True, "tickets": STORE.admin_support_tickets(ADMIN_ID), "runtime": STORE.support_runtime()}, headers={"Cache-Control": "no-store"}) async def action(request: Request) -> JSONResponse: _authorize(request) payload = await request.json() name = str(payload.get("action", "")).strip() result: Any = None if name == "ban_permanent": STORE.ban_user(ADMIN_ID, str(payload.get("user_id", "")), None) elif name == "ban_temporary": STORE.ban_user(ADMIN_ID, str(payload.get("user_id", "")), float(payload.get("hours", 24))) elif name == "unban": STORE.unban_user(ADMIN_ID, str(payload.get("user_id", ""))) elif name == "delete_user": STORE.admin_delete_user(ADMIN_ID, str(payload.get("user_id", ""))) elif name == "delete_all_free": result = STORE.delete_all_free(ADMIN_ID) elif name == "delete_all_users": result = STORE.delete_all_users(ADMIN_ID) elif name == "toggle_system": cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, system_enabled=not cfg.get("system_enabled", True)) elif name == "toggle_chat": cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, chat_enabled=not cfg.get("chat_enabled", True)) elif name == "toggle_plus_chat": cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, plus_chat_enabled=not cfg.get("plus_chat_enabled", True)) elif name == "toggle_pro_chat": cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, pro_chat_enabled=not cfg.get("pro_chat_enabled", True)) elif name == "toggle_purchases": cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, purchases_enabled=not cfg.get("purchases_enabled", True)) elif name == "toggle_registrations": cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, registrations_enabled=not cfg.get("registrations_enabled", True)) elif name == "clear_all_chat": STORE.clear_chat(ADMIN_ID) elif name == "begin_shutdown": result = STORE.begin_shutdown(ADMIN_ID) elif name == "cancel_shutdown": STORE.cancel_shutdown(ADMIN_ID) elif name == "configure_rooms": result = STORE.set_chat_configuration( ADMIN_ID, str(payload.get("mode", "single")), int(payload.get("capacity", 200)), ) elif name == "toggle_room": result = STORE.toggle_chat_room(ADMIN_ID, str(payload.get("room_id", ""))) elif name == "clear_room": result = STORE.clear_chat_room(ADMIN_ID, str(payload.get("room_id", ""))) elif name == "send_admin_chat": text = str(payload.get("text", "")).strip() room_id = str(payload.get("room_id", "global-1")).strip() or "global-1" result = STORE.send_chat(ADMIN_ID, text, room_id) elif name == "support_reply": result = STORE.reply_support_ticket( ADMIN_ID, str(payload.get("ticket_id", "")), str(payload.get("reply", "")), ) elif name == "support_toggle": cfg = STORE.config() result = STORE.set_config( ADMIN_ID, support_enabled=not cfg.get("support_enabled", True), ) elif name == "support_force_on": result = STORE.set_config( ADMIN_ID, support_enabled=True, support_force_open=True, ) elif name == "support_force_off": result = STORE.set_config(ADMIN_ID, support_force_open=False) elif name == "send_notification": result = send_admin_notification(ADMIN_ID, str(payload.get("title", "")), str(payload.get("message", "")), list(payload.get("audience", []))) elif name == "update_plan_features": result = update_plan_features(ADMIN_ID, str(payload.get("plan", "")), dict(payload.get("features", {}))) else: raise HTTPException(status_code=400, detail="Unknown admin action.") return JSONResponse({"ok": True, "result": result}, headers={"Cache-Control": "no-store"}) existing = {getattr(route, "path", "") for route in demo.app.routes} overview_path = "/gradio_api/cloudvault-admin/overview" action_path = "/gradio_api/cloudvault-admin/action" support_path = "/gradio_api/cloudvault-admin/support" chat_path = "/gradio_api/cloudvault-admin/chat" if overview_path not in existing: demo.app.add_api_route(overview_path, overview, methods=["GET"], include_in_schema=False) if action_path not in existing: demo.app.add_api_route(action_path, action, methods=["POST"], include_in_schema=False) if support_path not in existing: demo.app.add_api_route(support_path, support, methods=["GET"], include_in_schema=False) if chat_path not in existing: demo.app.add_api_route(chat_path, chat, methods=["GET"], include_in_schema=False)