Logan CusanoandClaude Opus 5.5
a976464e72
A server linked to an org now shows in the node picker while a radio bot the org may use is in it, even when the DRB server bot is absent or left. Its voice-channel list is read from that radio bot's cached REST snapshot (discord_rest.guild_directory, the same snapshot the permission check reads, so no extra Discord request): inline when no list is stored, in the background when the list is older than discord_directory_refresh_seconds. The server bot's own fresh list still wins (back-compat) and its directory push now drops the server's cached snapshots. Guards: a server acting only through the unlinked-server fallback org still needs the server bot; REST never creates a discord_guilds doc, never writes org_id or bot_present, and never writes onto a server unlinked meanwhile. Also fixes server-26#251: an interactive start (/join, admin page, web picker, node discord_join) re-reads a cached "denied" snapshot older than 10 s once before refusing, so "fix it in Discord, then try again" works on the next try. The reconciler never re-reads (no REST loop). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
755 lines
32 KiB
Python
755 lines
32 KiB
Python
"""
|
|
Discord REST client for C2 (server-26#225 slice S2).
|
|
|
|
C2 calls Discord's REST API with the radio-bot tokens it holds: to check a
|
|
token when an admin adds it, to re-read which servers a bot is in, to make
|
|
bots leave a server, and (from S3 on) to read a server's roles, the bot's own
|
|
member record and the channel list, so C2 can tell whether a bot may join and
|
|
speak in a voice channel before it hands the bot to a node.
|
|
|
|
Everything here is built so that a Discord problem can never become a C2
|
|
problem:
|
|
|
|
* **One shared ``httpx.AsyncClient``** with the User-Agent Discord asks bots
|
|
to send (``DiscordBot (url, version)``). The token travels per request in
|
|
the ``Authorization`` header and is never logged, never put in an error
|
|
string and never used as a dict key (``_token_key`` hashes it).
|
|
* **Rate limits per token.** Buckets are keyed by the ``X-RateLimit-Bucket``
|
|
Discord returns plus the route's major parameter (the guild or channel id).
|
|
A request that would have to wait more than
|
|
``discord_rest_max_retry_after_seconds`` is not sent: its result is
|
|
*unknown*. A 429 is retried once, only when ``retry_after`` is within that
|
|
limit. A global 429 pauses that token.
|
|
* **A process-wide breaker on invalid requests.** Discord bans the egress IP
|
|
for an hour after 10,000 responses of 401, 403 or 429 (except 429s with
|
|
``X-RateLimit-Scope: shared``) in 10 minutes. The main DRB bot shares that
|
|
IP. Once ``discord_rest_invalid_budget`` (default 4000) is reached inside
|
|
``discord_rest_invalid_window_seconds``, nothing more is sent and every
|
|
result is unknown until the window drains.
|
|
* **Unknown is not bad.** Callers fail open on an unknown result: a Discord
|
|
outage must never block a stream start or starve the reconciler.
|
|
|
|
Status mapping for the per-server snapshot (``guild_snapshot``):
|
|
|
|
401 token invalid: ``bot_tokens.valid = False``; never retried
|
|
404 10004 / 10007 not a member: the server is dropped from ``guild_ids``
|
|
403 50001 unknown, then confirmed with ONE read of the bot's
|
|
server list; ``guild_ids`` is never changed from the
|
|
403 alone
|
|
anything else unknown
|
|
|
|
Snapshots are cached in-process per (token_id, guild_id), one c2-core
|
|
process being the same assumption ``_RateLimiter`` and ``_publish_locks``
|
|
already make. Results that let a bot in live ``discord_access_ttl_seconds``;
|
|
not-member and invalid results live ``discord_access_negative_ttl_seconds``;
|
|
unknown results live ``discord_access_unknown_ttl_seconds``, so the
|
|
reconciler's 30 s loop can't turn into a REST loop against Discord.
|
|
Concurrent loads of one key share one request (single flight).
|
|
``invalidate(token_id=, guild_id=)`` drops entries.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import re
|
|
import time
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Awaitable, Callable, Optional
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
from app.internal import firestore as fstore
|
|
from app.internal.logger import logger
|
|
|
|
DISCORD_API = "https://discord.com/api/v10"
|
|
USER_AGENT = "DiscordBot (https://drb.cusano.net, 1)"
|
|
TOKENS = "bot_tokens"
|
|
|
|
# GET /users/@me/guilds returns at most 200 per page. 50 pages is 10,000
|
|
# servers, well past what an unsharded bot can be in; the cap only stops a
|
|
# misbehaving response from looping forever.
|
|
GUILDS_PAGE = 200
|
|
MAX_GUILD_PAGES = 50
|
|
|
|
# Discord JSON error codes this module acts on.
|
|
UNKNOWN_GUILD = 10004
|
|
UNKNOWN_MEMBER = 10007
|
|
MISSING_ACCESS = 50001
|
|
|
|
# Snapshot / access outcomes. ``denied`` is not produced here; the permission
|
|
# check built on these snapshots (S3) uses it, with the same negative TTL.
|
|
OK = "ok"
|
|
DENIED = "denied"
|
|
NOT_MEMBER = "not_member"
|
|
INVALID = "invalid"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class TokenRejected(Exception):
|
|
"""Discord answered 401: the string is not a working bot token."""
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _token_key(token: str) -> str:
|
|
"""A stable, non-reversible handle for a token, for rate-limit state."""
|
|
return hashlib.sha256(token.encode()).hexdigest()[:16]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Results
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class RestResult:
|
|
"""One Discord REST answer. ``status`` is None when nothing came back
|
|
(not sent, network error): the result is unknown."""
|
|
status: Optional[int]
|
|
data: Any = None
|
|
code: Optional[int] = None # Discord's JSON error code, on an error
|
|
error: Optional[str] = None # short reason; never holds the token
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.status is not None and 200 <= self.status < 300
|
|
|
|
@property
|
|
def unknown(self) -> bool:
|
|
"""No usable answer: not sent, network error, rate limited or 5xx."""
|
|
return self.status is None or self.status == 429 or self.status >= 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rate limits and the invalid-request breaker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ID_RE = re.compile(r"\d{5,}")
|
|
_MAJOR_RE = re.compile(r"^/(?:guilds|channels|webhooks)/(\d+)")
|
|
|
|
|
|
def _route(method: str, path: str) -> tuple[str, str]:
|
|
"""(route key, major parameter) for a path. Snowflakes in the path are
|
|
collapsed so ``/guilds/1/channels`` and ``/guilds/2/channels`` share a
|
|
route; the major parameter keeps their buckets apart."""
|
|
m = _MAJOR_RE.match(path)
|
|
return f"{method.upper()} {_ID_RE.sub('{id}', path)}", (m.group(1) if m else "")
|
|
|
|
|
|
class _RateLimits:
|
|
def __init__(self) -> None:
|
|
self.route_bucket: dict[tuple[str, str], str] = {}
|
|
self.buckets: dict[tuple[str, str, str], tuple[int, float]] = {}
|
|
self.paused_until: dict[str, float] = {}
|
|
|
|
def wait_for(self, tok: str, method: str, path: str, now: float) -> float:
|
|
"""Seconds to wait before this request may be sent (0 = go now)."""
|
|
wait = max(0.0, self.paused_until.get(tok, 0.0) - now)
|
|
route, major = _route(method, path)
|
|
bucket = self.route_bucket.get((tok, route))
|
|
if bucket:
|
|
remaining, reset_at = self.buckets.get((tok, bucket, major), (1, 0.0))
|
|
if remaining <= 0 and reset_at > now:
|
|
wait = max(wait, reset_at - now)
|
|
return wait
|
|
|
|
def update(self, tok: str, method: str, path: str, headers: httpx.Headers, now: float) -> None:
|
|
bucket = headers.get("x-ratelimit-bucket")
|
|
if not bucket:
|
|
return
|
|
route, major = _route(method, path)
|
|
self.route_bucket[(tok, route)] = bucket
|
|
try:
|
|
remaining = int(headers.get("x-ratelimit-remaining", "1"))
|
|
reset_after = float(headers.get("x-ratelimit-reset-after", "0"))
|
|
except ValueError:
|
|
return
|
|
self.buckets[(tok, bucket, major)] = (remaining, now + reset_after)
|
|
|
|
def pause(self, tok: str, seconds: float, now: float) -> None:
|
|
self.paused_until[tok] = max(self.paused_until.get(tok, 0.0), now + seconds)
|
|
|
|
|
|
class _Breaker:
|
|
"""Counts 401 / 403 / non-shared 429 responses in a sliding window."""
|
|
|
|
def __init__(self) -> None:
|
|
self.hits: deque[float] = deque()
|
|
self.open_logged = False
|
|
|
|
def _prune(self, now: float) -> None:
|
|
horizon = now - settings.discord_rest_invalid_window_seconds
|
|
while self.hits and self.hits[0] <= horizon:
|
|
self.hits.popleft()
|
|
|
|
def record(self, now: float) -> None:
|
|
self.hits.append(now)
|
|
|
|
def tripped(self, now: float) -> bool:
|
|
self._prune(now)
|
|
tripped = len(self.hits) >= settings.discord_rest_invalid_budget
|
|
if tripped and not self.open_logged:
|
|
logger.warning(
|
|
f"Discord REST breaker open: {len(self.hits)} invalid responses in "
|
|
f"{settings.discord_rest_invalid_window_seconds}s; REST checks report unknown "
|
|
"until the window drains (protects the shared egress IP from a Discord ban)")
|
|
self.open_logged = True
|
|
elif not tripped:
|
|
self.open_logged = False
|
|
return tripped
|
|
|
|
|
|
_limits = _RateLimits()
|
|
_breaker = _Breaker()
|
|
_sleep = asyncio.sleep # tests replace this, not asyncio.sleep itself
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The shared HTTP client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_client_obj: Optional[httpx.AsyncClient] = None
|
|
_client_loop: Optional[asyncio.AbstractEventLoop] = None
|
|
_transport: Optional[httpx.AsyncBaseTransport] = None
|
|
|
|
|
|
def _client() -> httpx.AsyncClient:
|
|
"""The one shared client, recreated only if the event loop changed (the
|
|
connection pool belongs to a loop; tests run one loop per test)."""
|
|
global _client_obj, _client_loop
|
|
loop = asyncio.get_running_loop()
|
|
if _client_obj is None or _client_loop is not loop or _client_obj.is_closed:
|
|
_client_obj = httpx.AsyncClient(
|
|
base_url=DISCORD_API,
|
|
timeout=settings.discord_rest_timeout_seconds,
|
|
headers={"User-Agent": USER_AGENT},
|
|
transport=_transport,
|
|
)
|
|
_client_loop = loop
|
|
return _client_obj
|
|
|
|
|
|
def set_transport(transport: Optional[httpx.AsyncBaseTransport]) -> None:
|
|
"""Tests: route every request through ``transport`` (e.g. httpx.MockTransport)."""
|
|
global _transport, _client_obj
|
|
_transport = transport
|
|
_client_obj = None
|
|
|
|
|
|
def reset() -> None:
|
|
"""Forget all in-process state: client, rate limits, breaker, cache."""
|
|
global _limits, _breaker, _client_obj
|
|
_limits = _RateLimits()
|
|
_breaker = _Breaker()
|
|
_client_obj = None
|
|
access_cache.clear()
|
|
|
|
|
|
def _retry_after(r: httpx.Response) -> tuple[float, bool]:
|
|
"""(seconds, is_global) from a 429's body or headers."""
|
|
body: Any = None
|
|
try:
|
|
body = r.json()
|
|
except Exception:
|
|
body = None
|
|
seconds: Optional[float] = None
|
|
is_global = r.headers.get("x-ratelimit-global", "").lower() == "true"
|
|
if isinstance(body, dict):
|
|
if isinstance(body.get("retry_after"), (int, float)):
|
|
seconds = float(body["retry_after"])
|
|
is_global = is_global or bool(body.get("global"))
|
|
if seconds is None:
|
|
try:
|
|
seconds = float(r.headers.get("retry-after", "1"))
|
|
except ValueError:
|
|
seconds = 1.0
|
|
return max(0.0, seconds), is_global
|
|
|
|
|
|
async def request(token: str, method: str, path: str, *,
|
|
params: Optional[dict] = None) -> RestResult:
|
|
"""Send one request as the bot ``token``. Never raises for a Discord or
|
|
network failure: those come back as an unknown RestResult."""
|
|
tok = _token_key(token)
|
|
max_wait = settings.discord_rest_max_retry_after_seconds
|
|
for attempt in (0, 1):
|
|
now = time.monotonic()
|
|
if _breaker.tripped(now):
|
|
return RestResult(None, error="Discord REST paused (invalid-request budget reached)")
|
|
wait = _limits.wait_for(tok, method, path, now)
|
|
if wait > max_wait:
|
|
return RestResult(None, error=f"rate limited for {wait:.0f}s")
|
|
if wait > 0:
|
|
await _sleep(wait)
|
|
try:
|
|
r = await _client().request(method, path, params=params,
|
|
headers={"Authorization": f"Bot {token}"})
|
|
except httpx.HTTPError as e:
|
|
# The exception text holds at most the URL, never the header.
|
|
return RestResult(None, error=f"{type(e).__name__}: {str(e)[:120]}")
|
|
now = time.monotonic()
|
|
_limits.update(tok, method, path, r.headers, now)
|
|
status = r.status_code
|
|
if status in (401, 403) or (status == 429 and r.headers.get("x-ratelimit-scope") != "shared"):
|
|
_breaker.record(now)
|
|
if status == 429:
|
|
seconds, is_global = _retry_after(r)
|
|
if is_global:
|
|
_limits.pause(tok, seconds, now)
|
|
if attempt == 0 and seconds <= max_wait:
|
|
await _sleep(seconds)
|
|
continue
|
|
return RestResult(429, error=f"rate limited (retry after {seconds:.1f}s)")
|
|
data: Any = None
|
|
if r.content:
|
|
try:
|
|
data = r.json()
|
|
except ValueError:
|
|
data = None
|
|
code = data.get("code") if (status >= 400 and isinstance(data, dict)) else None
|
|
err = None if 200 <= status < 300 else f"Discord answered {status}" + (f" ({code})" if code else "")
|
|
return RestResult(status, data, code if isinstance(code, int) else None, err)
|
|
return RestResult(429, error="rate limited") # pragma: no cover - loop always returns
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cache (in-process, single flight)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def ttl_for(outcome: str) -> float:
|
|
"""How long a result with this outcome stays cached."""
|
|
if outcome == OK:
|
|
return float(settings.discord_access_ttl_seconds)
|
|
if outcome in (DENIED, NOT_MEMBER, INVALID):
|
|
return float(settings.discord_access_negative_ttl_seconds)
|
|
return float(settings.discord_access_unknown_ttl_seconds)
|
|
|
|
|
|
class AccessCache:
|
|
"""TTL cache keyed by tuples that start ``(token_id, guild_id, ...)``.
|
|
|
|
``get_or_load`` runs one load per key at a time; concurrent callers await
|
|
the same load. A load that overlaps an ``invalidate`` touching its key is
|
|
returned to its callers but not stored, so an invalidation is never undone
|
|
by a load that started before it."""
|
|
|
|
def __init__(self) -> None:
|
|
self._entries: dict[tuple, tuple[float, Any]] = {}
|
|
self._inflight: dict[tuple, asyncio.Task] = {}
|
|
self._gen: dict[tuple, int] = {}
|
|
|
|
def clear(self) -> None:
|
|
self._entries.clear()
|
|
self._inflight.clear()
|
|
self._gen.clear()
|
|
|
|
def peek(self, key: tuple) -> Any:
|
|
hit = self._entries.get(key)
|
|
if hit and hit[0] > time.monotonic():
|
|
return hit[1]
|
|
return None
|
|
|
|
async def get_or_load(self, key: tuple, loader: Callable[[], Awaitable[Any]],
|
|
ttl_of: Callable[[Any], float], *, refresh: bool = False) -> Any:
|
|
if not refresh:
|
|
hit = self._entries.get(key)
|
|
if hit and hit[0] > time.monotonic():
|
|
return hit[1]
|
|
task = self._inflight.get(key)
|
|
loop = asyncio.get_running_loop()
|
|
if task is None or task.done() or task.get_loop() is not loop:
|
|
gen = self._gen.get(key, 0)
|
|
|
|
async def _load() -> Any:
|
|
try:
|
|
value = await loader()
|
|
if self._gen.get(key, 0) == gen:
|
|
self._entries[key] = (time.monotonic() + ttl_of(value), value)
|
|
return value
|
|
finally:
|
|
if self._inflight.get(key) is task_ref[0]:
|
|
self._inflight.pop(key, None)
|
|
|
|
task_ref: list = [None]
|
|
task = loop.create_task(_load())
|
|
task_ref[0] = task
|
|
self._inflight[key] = task
|
|
return await asyncio.shield(task)
|
|
|
|
def invalidate(self, token_id: Optional[str] = None, guild_id: Optional[str] = None) -> int:
|
|
"""Drop entries for a token, a guild, both, or (no args) everything.
|
|
Returns how many cached entries were dropped."""
|
|
def match(k: tuple) -> bool:
|
|
return ((token_id is None or k[0] == token_id)
|
|
and (guild_id is None or (len(k) > 1 and k[1] == str(guild_id))))
|
|
dropped = [k for k in self._entries if match(k)]
|
|
for k in dropped:
|
|
del self._entries[k]
|
|
for k in set(self._inflight) | set(self._entries) | set(dropped):
|
|
if match(k):
|
|
self._gen[k] = self._gen.get(k, 0) + 1
|
|
for k in [k for k in self._inflight if match(k)]:
|
|
self._inflight.pop(k, None)
|
|
return len(dropped)
|
|
|
|
|
|
access_cache = AccessCache()
|
|
|
|
|
|
def invalidate(token_id: Optional[str] = None, guild_id: Optional[str] = None) -> int:
|
|
return access_cache.invalidate(token_id=token_id, guild_id=guild_id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token inspection / membership (moved from discord_leases, server-26#204)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def list_guild_ids(token: str) -> RestResult:
|
|
"""Every server this bot is in, following ``after=`` pages (the old code
|
|
read one page of 200 and silently dropped the rest). On success ``data``
|
|
is the list of guild id strings; any failed page makes the whole result
|
|
that page's failure, never a partial list."""
|
|
out: list[str] = []
|
|
after: Optional[str] = None
|
|
for _ in range(MAX_GUILD_PAGES):
|
|
params: dict = {"limit": GUILDS_PAGE}
|
|
if after:
|
|
params["after"] = after
|
|
r = await request(token, "GET", "/users/@me/guilds", params=params)
|
|
if not r.ok:
|
|
return r
|
|
page = r.data if isinstance(r.data, list) else []
|
|
ids = [str(g["id"]) for g in page if isinstance(g, dict) and g.get("id") is not None]
|
|
out.extend(ids)
|
|
if len(page) < GUILDS_PAGE or not ids:
|
|
return RestResult(200, out)
|
|
after = max(ids, key=int)
|
|
logger.warning(f"Discord /users/@me/guilds: stopped after {MAX_GUILD_PAGES} pages")
|
|
return RestResult(None, error=f"more than {MAX_GUILD_PAGES * GUILDS_PAGE} servers")
|
|
|
|
|
|
async def inspect_token(token: str) -> dict:
|
|
"""Ask Discord who this token is and which servers it is in.
|
|
|
|
Returns ``{bot_user_id, bot_name, application_id, guild_ids, valid,
|
|
validated_at, check_error}``. Raises TokenRejected on a 401. Any other
|
|
failure (Discord down, rate limit) is reported in ``check_error`` with
|
|
``valid``/``guild_ids`` = None — unknown, not bad: a token is never
|
|
rejected because Discord had a bad minute.
|
|
"""
|
|
out: dict = {"valid": None, "guild_ids": None, "check_error": None,
|
|
"validated_at": _now().isoformat()}
|
|
me = await request(token, "GET", "/users/@me")
|
|
if me.status == 401:
|
|
raise TokenRejected("Discord rejected this bot token (401).")
|
|
if not me.ok or not isinstance(me.data, dict):
|
|
out["check_error"] = (me.error or "unexpected answer from Discord")[:200]
|
|
return out
|
|
out["bot_user_id"] = str(me.data.get("id"))
|
|
out["bot_name"] = me.data.get("username")
|
|
app = await request(token, "GET", "/oauth2/applications/@me")
|
|
if app.ok and isinstance(app.data, dict) and app.data.get("id") is not None:
|
|
out["application_id"] = str(app.data["id"])
|
|
guilds = await list_guild_ids(token)
|
|
if guilds.status == 401:
|
|
raise TokenRejected("Discord rejected this bot token (401).")
|
|
if not guilds.ok:
|
|
out["check_error"] = (guilds.error or "could not list servers")[:200]
|
|
return out
|
|
out["guild_ids"] = guilds.data
|
|
out["valid"] = True
|
|
return out
|
|
|
|
|
|
async def refresh_membership(token_ids: list[str]) -> int:
|
|
"""Re-read guild membership for these tokens. Returns how many changed.
|
|
Every token re-read has its cached snapshots dropped."""
|
|
changed = 0
|
|
for tok_id in token_ids:
|
|
tok = await fstore.doc_get(TOKENS, tok_id)
|
|
if not tok or not tok.get("token"):
|
|
continue
|
|
try:
|
|
info = await inspect_token(tok["token"])
|
|
except TokenRejected as e:
|
|
await fstore.doc_update(TOKENS, tok_id, {"valid": False, "check_error": str(e),
|
|
"validated_at": _now().isoformat()})
|
|
invalidate(token_id=tok_id)
|
|
changed += 1
|
|
continue
|
|
updates = {k: v for k, v in info.items() if v is not None or k == "check_error"}
|
|
if updates.get("guild_ids") != tok.get("guild_ids") or updates.get("valid") != tok.get("valid"):
|
|
changed += 1
|
|
await fstore.doc_update(TOKENS, tok_id, updates)
|
|
invalidate(token_id=tok_id)
|
|
return changed
|
|
|
|
|
|
async def leave_guild(guild_id: str, org_id: Optional[str]) -> list[dict]:
|
|
"""Make the radio bots DRB holds tokens for leave one Discord server
|
|
(Discord REST ``DELETE /users/@me/guilds/{id}``, the bot's own token).
|
|
|
|
Only platform-pool bots and ``org_id``'s own BYO bots — never another
|
|
org's branded bot. Bots whose cached membership says they are not in the
|
|
server are skipped; unknown membership is tried (404 = not a member).
|
|
The DRB *server* bot (the discord-bot service) is not in this pool, so it
|
|
stays in the server until someone kicks it. Returns one row per bot tried:
|
|
``{token_id, name, left, error}``. Never raises for a Discord failure."""
|
|
guild_id = str(guild_id)
|
|
out: list[dict] = []
|
|
for tok in await fstore.collection_list(TOKENS):
|
|
tok_id = tok.get("token_id")
|
|
if not tok_id or not tok.get("token"):
|
|
continue
|
|
if tok.get("org_id") is not None and tok.get("org_id") != org_id:
|
|
continue
|
|
members = tok.get("guild_ids")
|
|
if members is not None and guild_id not in members:
|
|
continue
|
|
row = {"token_id": tok_id, "name": tok.get("name") or tok_id, "left": False, "error": None}
|
|
r = await request(tok["token"], "DELETE", f"/users/@me/guilds/{guild_id}")
|
|
if r.status in (200, 204, 404): # 404: Unknown Guild — not a member
|
|
row["left"] = True
|
|
else:
|
|
row["error"] = r.error or f"Discord answered {r.status}"
|
|
if row["left"]:
|
|
invalidate(token_id=tok_id, guild_id=guild_id)
|
|
if members is not None:
|
|
await fstore.doc_update(TOKENS, tok_id, {"guild_ids": [g for g in members if g != guild_id]})
|
|
out.append(row)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-(token, guild) snapshot — what S3's permission check reads
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class GuildSnapshot:
|
|
"""What one bot can see of one server. ``status`` is OK, NOT_MEMBER,
|
|
INVALID or UNKNOWN; ``guild``/``member``/``channels`` are set only on OK.
|
|
``guild`` is Discord's guild object (``owner_id``, ``roles``), ``member``
|
|
the bot's own member object (``roles``, ``communication_disabled_until``),
|
|
``channels`` the server's channel list."""
|
|
token_id: str
|
|
guild_id: str
|
|
status: str
|
|
bot_user_id: Optional[str] = None
|
|
guild: Optional[dict] = None
|
|
member: Optional[dict] = None
|
|
channels: Optional[list] = None
|
|
error: Optional[str] = None
|
|
fetched_at: float = field(default_factory=time.time)
|
|
|
|
|
|
async def _mark_invalid(token_id: str) -> None:
|
|
await fstore.doc_update(TOKENS, token_id, {
|
|
"valid": False, "check_error": "Discord rejected this bot token (401).",
|
|
"validated_at": _now().isoformat()})
|
|
|
|
|
|
async def _drop_membership(token_id: str, guild_id: str) -> None:
|
|
tok = await fstore.doc_get(TOKENS, token_id) or {}
|
|
members = tok.get("guild_ids")
|
|
if isinstance(members, list) and guild_id in members:
|
|
await fstore.doc_update(TOKENS, token_id, {"guild_ids": [g for g in members if g != guild_id]})
|
|
|
|
|
|
async def _confirm_membership(token_id: str, token: str, guild_id: str) -> Optional[bool]:
|
|
"""After a 403 Missing Access: one read of the bot's server list decides.
|
|
True/False = member or not (``guild_ids`` written from the full list);
|
|
None = still unknown."""
|
|
r = await list_guild_ids(token)
|
|
if r.status == 401:
|
|
await _mark_invalid(token_id)
|
|
return None
|
|
if not r.ok:
|
|
return None
|
|
await fstore.doc_update(TOKENS, token_id, {"guild_ids": r.data, "validated_at": _now().isoformat()})
|
|
return guild_id in r.data
|
|
|
|
|
|
async def _classify(snap: GuildSnapshot, r: RestResult, token: str) -> Optional[GuildSnapshot]:
|
|
"""None when ``r`` is a success; otherwise the finished snapshot."""
|
|
if r.ok:
|
|
return None
|
|
snap.error = r.error
|
|
if r.status == 401:
|
|
await _mark_invalid(snap.token_id)
|
|
snap.status = INVALID
|
|
elif r.status == 404 and r.code in (UNKNOWN_GUILD, UNKNOWN_MEMBER):
|
|
await _drop_membership(snap.token_id, snap.guild_id)
|
|
snap.status = NOT_MEMBER
|
|
elif r.status == 403 and r.code == MISSING_ACCESS:
|
|
member = await _confirm_membership(snap.token_id, token, snap.guild_id)
|
|
snap.status = NOT_MEMBER if member is False else UNKNOWN
|
|
else:
|
|
snap.status = UNKNOWN
|
|
return snap
|
|
|
|
|
|
async def _load_snapshot(token_doc: dict, guild_id: str) -> GuildSnapshot:
|
|
token_id = str(token_doc.get("token_id"))
|
|
snap = GuildSnapshot(token_id=token_id, guild_id=guild_id, status=UNKNOWN,
|
|
bot_user_id=token_doc.get("bot_user_id"))
|
|
token = token_doc.get("token")
|
|
if not token:
|
|
snap.error = "no token stored"
|
|
return snap
|
|
if token_doc.get("valid") is False:
|
|
snap.status, snap.error = INVALID, "token marked invalid"
|
|
return snap
|
|
if not snap.bot_user_id:
|
|
me = await request(token, "GET", "/users/@me")
|
|
done = await _classify(snap, me, token)
|
|
if done:
|
|
return done
|
|
snap.bot_user_id = str((me.data or {}).get("id") or "") or None
|
|
if not snap.bot_user_id:
|
|
snap.error = "Discord did not say who this bot is"
|
|
return snap
|
|
await fstore.doc_update(TOKENS, token_id, {"bot_user_id": snap.bot_user_id})
|
|
|
|
g = await request(token, "GET", f"/guilds/{guild_id}")
|
|
done = await _classify(snap, g, token)
|
|
if done:
|
|
return done
|
|
m, c = await asyncio.gather(
|
|
request(token, "GET", f"/guilds/{guild_id}/members/{snap.bot_user_id}"),
|
|
request(token, "GET", f"/guilds/{guild_id}/channels"),
|
|
)
|
|
for r in (m, c):
|
|
done = await _classify(snap, r, token)
|
|
if done:
|
|
return done
|
|
if not isinstance(g.data, dict) or not isinstance(m.data, dict) or not isinstance(c.data, list):
|
|
snap.error = "unexpected answer from Discord"
|
|
return snap
|
|
snap.guild, snap.member, snap.channels = g.data, m.data, c.data
|
|
snap.status, snap.error = OK, None
|
|
return snap
|
|
|
|
|
|
async def guild_snapshot(token_doc: dict, guild_id: str, *, refresh: bool = False) -> GuildSnapshot:
|
|
"""Cached snapshot of one server as one bot sees it: three GETs
|
|
(``/guilds/{g}``, ``/guilds/{g}/members/{bot}`` — no privileged intent
|
|
needed — and ``/guilds/{g}/channels``). ``refresh=True`` skips the cache
|
|
(a concurrent load is still shared). Never raises for a Discord failure."""
|
|
guild_id = str(guild_id)
|
|
token_id = str(token_doc.get("token_id"))
|
|
key = (token_id, guild_id, "snapshot")
|
|
|
|
async def _load() -> GuildSnapshot:
|
|
# Fail open on anything else too (a Firestore write inside the
|
|
# classification, a malformed body): the caller gets UNKNOWN, never
|
|
# an exception, and a shielded load never dies unretrieved.
|
|
try:
|
|
return await _load_snapshot(token_doc, guild_id)
|
|
except Exception as e:
|
|
logger.warning(f"Discord snapshot for bot {token_id} in {guild_id} failed: {type(e).__name__}")
|
|
return GuildSnapshot(token_id=token_id, guild_id=guild_id, status=UNKNOWN,
|
|
bot_user_id=token_doc.get("bot_user_id"),
|
|
error=f"check failed ({type(e).__name__})")
|
|
|
|
return await access_cache.get_or_load(key, _load, lambda s: ttl_for(s.status), refresh=refresh)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Voice-channel directory from a radio bot's view (server-26#225 S5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DIRECTORY_MAX_CHANNELS = 200 # same cap as PUT /discord/guilds/{id}/directory
|
|
_DIRECTORY_NAME_MAX = 100
|
|
VOICE_CHANNEL = 2
|
|
CATEGORY_CHANNEL = 4
|
|
|
|
|
|
def _int_or(value: Any, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def directory_from_snapshot(snap: Optional[GuildSnapshot]) -> Optional[dict]:
|
|
"""A server's name and voice channels as one radio bot's snapshot sees
|
|
them, in the shape the DRB server bot pushes (``PUT
|
|
/discord/guilds/{id}/directory``): ``{name, voice_channels: [{channel_id,
|
|
name, category, position}]}``, in Discord's sidebar order (channels with
|
|
no category first, then each category by position; channels by position,
|
|
then id). None unless the snapshot is OK.
|
|
|
|
Every voice channel in the list is kept, including any this bot can't
|
|
View, exactly like the server bot's gateway list: whether a bot may join
|
|
is the permission check's answer (discord_streams.evaluate_access), and
|
|
the picker shows it per channel. Stage channels are not voice channels
|
|
here either."""
|
|
if snap is None or snap.status != OK or not isinstance(snap.channels, list):
|
|
return None
|
|
chans = [c for c in snap.channels if isinstance(c, dict) and c.get("id") is not None]
|
|
cats = {str(c["id"]): c for c in chans if c.get("type") == CATEGORY_CHANNEL}
|
|
|
|
def parent(c: dict) -> Optional[dict]:
|
|
return cats.get(str(c.get("parent_id") or ""))
|
|
|
|
def order(c: dict) -> tuple:
|
|
p = parent(c)
|
|
top = (0, 0, 0) if p is None else (1, _int_or(p.get("position")), _int_or(p.get("id")))
|
|
return (*top, _int_or(c.get("position")), _int_or(c.get("id")))
|
|
|
|
out: list[dict] = []
|
|
for c in sorted((c for c in chans if c.get("type") == VOICE_CHANNEL), key=order)[:DIRECTORY_MAX_CHANNELS]:
|
|
p = parent(c)
|
|
category = str(p.get("name") or "")[:_DIRECTORY_NAME_MAX] if p else ""
|
|
out.append({
|
|
"channel_id": str(c["id"]),
|
|
"name": str(c.get("name") or "")[:_DIRECTORY_NAME_MAX],
|
|
"category": category or None,
|
|
"position": len(out),
|
|
})
|
|
name = snap.guild.get("name") if isinstance(snap.guild, dict) else None
|
|
return {"name": str(name)[:_DIRECTORY_NAME_MAX] if name else None, "voice_channels": out}
|
|
|
|
|
|
@dataclass
|
|
class DirectoryResult:
|
|
"""``guild_directory``'s answer. ``status``: OK (``name`` and
|
|
``voice_channels`` from ``token_id``'s snapshot), NOT_MEMBER (every bot
|
|
asked turned out not to be in the server, or its token is invalid) or
|
|
UNKNOWN (no usable answer — keep what is stored)."""
|
|
guild_id: str
|
|
status: str
|
|
name: Optional[str] = None
|
|
voice_channels: Optional[list] = None
|
|
token_id: Optional[str] = None
|
|
fetched_at: Optional[float] = None
|
|
|
|
|
|
async def guild_directory(token_docs: list[dict], guild_id: str, *, refresh: bool = False,
|
|
max_bots: int = 3) -> DirectoryResult:
|
|
"""The voice-channel list of one server, read from the first of
|
|
``token_docs`` (radio bots believed to be in it, best first) whose cached
|
|
snapshot is OK. It is the same snapshot the permission check reads, so
|
|
listing channels costs no extra Discord request. At most ``max_bots``
|
|
bots are asked, one after another. Never raises for a Discord failure.
|
|
|
|
The caller decides which server and which bots: this never looks a
|
|
server up by itself, so it can't discover one (a server DRB doesn't
|
|
already know is never listed from a radio bot's membership)."""
|
|
guild_id = str(guild_id)
|
|
statuses: list[str] = []
|
|
for doc in token_docs[:max(1, max_bots)]:
|
|
snap = await guild_snapshot(doc, guild_id, refresh=refresh)
|
|
statuses.append(snap.status)
|
|
d = directory_from_snapshot(snap)
|
|
if d is not None:
|
|
return DirectoryResult(guild_id=guild_id, status=OK, name=d["name"],
|
|
voice_channels=d["voice_channels"], token_id=snap.token_id,
|
|
fetched_at=snap.fetched_at)
|
|
if statuses and all(s in (NOT_MEMBER, INVALID) for s in statuses):
|
|
return DirectoryResult(guild_id=guild_id, status=NOT_MEMBER)
|
|
return DirectoryResult(guild_id=guild_id, status=UNKNOWN)
|