# ╔══════════════════════════════════════════════════════════════════╗ # ║ ║ # ║ +-+-+-+-+-+-+-+-+ ║ # ║ |H|e|x|a|H|o|s|t| ║ # ║ +-+-+-+-+-+-+-+-+ ║ # ║ ║ # ║ © 2026 HexaHost — All Rights Reserved ║ # ║ ║ # ║ discord ── https://discord.gg/hexahost ║ # ║ github ── https://github.com/theoneandonlymace ║ # ║ ║ # ╚══════════════════════════════════════════════════════════════════╝ from __future__ import annotations import asyncio import re import time from typing import Any import aiohttp from fastapi import HTTPException, Request from utils.config import OWNER_IDS_STR MANAGE_GUILD = 0x20 ADMINISTRATOR = 0x8 GUILD_PATH_RE = re.compile(r"^/api/v1/guilds/(\d+)(?:/|$)") # Short-lived process caches — cut Discord 429s from dashboard reloads _USER_CACHE_TTL = 60.0 _GUILDS_CACHE_TTL = 60.0 _user_cache: dict[str, tuple[float, dict[str, Any]]] = {} _guilds_cache: dict[str, tuple[float, list[dict[str, Any]]]] = {} _guilds_inflight: dict[str, asyncio.Future[list[dict[str, Any]]]] = {} def _token_key(access_token: str) -> str: return f"{access_token[:12]}:{len(access_token)}:{access_token[-8:]}" def _parse_admin_ids() -> list[str]: import os raw = os.getenv("DASHBOARD_ADMIN_IDS", "").strip() if raw: return [p.strip() for p in raw.split(",") if p.strip().isdigit()] return list(OWNER_IDS_STR) def can_manage_guild(guild: dict[str, Any]) -> bool: if guild.get("owner"): return True try: perms = int(guild.get("permissions", "0")) except (TypeError, ValueError): return False return bool(perms & ADMINISTRATOR) or bool(perms & MANAGE_GUILD) def _discord_auth_error(status: int) -> HTTPException: if status == 429: return HTTPException( status_code=429, detail="Discord rate limit — please wait a few seconds and retry.", ) if status in (401, 403): return HTTPException( status_code=401, detail="Invalid or expired Discord access token. Please sign out and sign in again.", ) return HTTPException( status_code=403, detail=f"Discord API error ({status}).", ) async def fetch_user_guilds(access_token: str) -> list[dict[str, Any]]: key = _token_key(access_token) cached = _guilds_cache.get(key) if cached and cached[0] > time.monotonic(): return cached[1] existing = _guilds_inflight.get(key) if existing is not None: return await existing loop = asyncio.get_running_loop() fut: asyncio.Future[list[dict[str, Any]]] = loop.create_future() _guilds_inflight[key] = fut try: guilds = await _fetch_user_guilds_uncached(access_token) _guilds_cache[key] = (time.monotonic() + _GUILDS_CACHE_TTL, guilds) fut.set_result(guilds) return guilds except Exception as exc: fut.set_exception(exc) raise finally: _guilds_inflight.pop(key, None) async def _fetch_user_guilds_uncached(access_token: str) -> list[dict[str, Any]]: max_attempts = 3 async with aiohttp.ClientSession() as session: for attempt in range(1, max_attempts + 1): async with session.get( "https://discord.com/api/users/@me/guilds", headers={"Authorization": f"Bearer {access_token}"}, ) as resp: if resp.status == 200: data = await resp.json() return data if isinstance(data, list) else [] if resp.status == 429 and attempt < max_attempts: retry_after = 1.0 try: body = await resp.json() retry_after = float(body.get("retry_after", 1)) except Exception: retry_after = float(resp.headers.get("Retry-After", "1") or 1) await asyncio.sleep(min(max(retry_after, 0.3), 5.0)) continue raise _discord_auth_error(resp.status) raise _discord_auth_error(429) async def verify_discord_token(access_token: str) -> dict[str, Any]: key = _token_key(access_token) cached = _user_cache.get(key) if cached and cached[0] > time.monotonic(): return cached[1] async with aiohttp.ClientSession() as session: async with session.get( "https://discord.com/api/users/@me", headers={"Authorization": f"Bearer {access_token}"}, ) as resp: if resp.status != 200: raise _discord_auth_error(resp.status) user = await resp.json() _user_cache[key] = (time.monotonic() + _USER_CACHE_TTL, user) return user async def user_can_manage_guild(access_token: str, guild_id: int) -> bool: guilds = await fetch_user_guilds(access_token) for guild in guilds: if str(guild.get("id")) == str(guild_id): return can_manage_guild(guild) return False async def get_manageable_guild_ids(access_token: str) -> set[str]: guilds = await fetch_user_guilds(access_token) return {str(g["id"]) for g in guilds if can_manage_guild(g)} def get_discord_headers(request: Request) -> tuple[str | None, str | None]: token = request.headers.get("X-Discord-Access-Token") user_id = request.headers.get("X-Discord-User-Id") return token, user_id async def require_discord_session(request: Request) -> tuple[str, str]: token, user_id = get_discord_headers(request) if not token: raise HTTPException( status_code=403, detail="X-Discord-Access-Token header is required.", ) user = await verify_discord_token(token) resolved_id = str(user.get("id", "")) if not resolved_id: raise HTTPException(status_code=403, detail="Could not resolve Discord user.") if user_id and user_id != resolved_id: raise HTTPException(status_code=403, detail="Discord user ID mismatch.") return token, resolved_id async def require_guild_access(request: Request, guild_id: int) -> None: token, _ = await require_discord_session(request) if not await user_can_manage_guild(token, guild_id): raise HTTPException( status_code=403, detail="You do not have permission to manage this guild.", ) async def require_dashboard_admin(request: Request) -> None: token, user_id = await require_discord_session(request) admin_ids = _parse_admin_ids() if user_id not in admin_ids: raise HTTPException( status_code=403, detail="Dashboard admin access required.", ) def extract_guild_id_from_path(path: str) -> int | None: match = GUILD_PATH_RE.match(path) if not match: return None return int(match.group(1))