Logan CusanoandClaude Opus 5.5
2c2cf9d9be
start_stream now asks Discord (cached REST snapshots, S2) whether each bot
it could pick may View, Connect and Speak in the voice channel, before the
lease. Covers all four entry points: bot POST /discord/streams, web picker,
nodes discord_join, and the reconciler's re-place.
- pick_token: denied bots skipped; confirmed-ok sorts before unknown;
unknown is allowed (fail open); all denied -> missing_permissions naming
each bot, missing permission, trace reason and fix. The reuse branch
falls through when its token is denied.
- Bounded: discord_perm_check_concurrency at once, within
discord_perm_check_budget_seconds; unanswered = unknown.
- Non-voice channel -> 400 bad_channel (not missing_permissions).
- systems/{id}.preferred_token_id is now read (plan finding 1), as a hint.
- Pool bots at discord_pool_bot_guild_cap are never offered as invites;
pool_full when nothing else is left (owner Q3).
- handle_actual drops cached access on missing_permissions, channel_full,
not_in_guild, bad_channel.
- New GET /discord/nodes/{node_id}/guilds/{guild_id}/voice-check (org
member, names only) and GET /discord/guilds/{guild_id}/voice-check
(service key or admin).
- PUT /tokens/{id}/prefer/{system_id}: org owners for their own systems
and own BYO bots; shared systems and pool bots stay admin-only (owner Q2).
Keeps both pointers single-valued; audited.
- Kill switch: DISCORD_PERM_CHECK_ENABLED=false.
- tests/conftest.py: autouse fixture so no test ever reaches discord.com.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
359 lines
18 KiB
Python
359 lines
18 KiB
Python
"""
|
|
Discord radio channels — the HTTP face of internal/discord_streams.py.
|
|
|
|
Every bot-facing route is scoped to ONE Discord server (``guild_id`` in the
|
|
path or body) and acts for the org that server is linked to
|
|
(server-26#203): /leave in server A can no longer touch server B, and the
|
|
system/node/bot lists a server sees are its own org's.
|
|
|
|
Auth:
|
|
* bot routes — the Discord bot's service key, or a platform admin.
|
|
* PUT /discord/guilds/{id}/link — platform admin only.
|
|
* DELETE /discord/streams/{stream_id}, DELETE /discord/guilds/{id} — platform
|
|
admin only: permanently remove a radio channel / a whole server.
|
|
* GET /discord/node/{node_id}/tokens/{token_id} — that node's own API key;
|
|
the token string is handed only to the node currently holding its lease.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Security
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.config import settings
|
|
from app.internal import discord_leases as leases
|
|
from app.internal import discord_rest
|
|
from app.internal import discord_streams as streams
|
|
from app.internal import firestore as fstore
|
|
from app.internal.audit import write_audit
|
|
from app.internal.auth import describe_actor, require_admin_token, require_service_key_or_admin
|
|
from app.internal.logger import logger
|
|
|
|
router = APIRouter(prefix="/discord", tags=["discord"])
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
# guild:channel[:suffix] (leases.STREAM_ID_PATTERN) — old one-per-channel ids
|
|
# and new per-stream ids. Anything else is a 422, never a Firestore path.
|
|
STREAM_ID = leases.STREAM_ID_PATTERN
|
|
|
|
|
|
def _error(e: streams.StreamError) -> JSONResponse:
|
|
return JSONResponse(status_code=e.status,
|
|
content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite})
|
|
|
|
|
|
async def _org_or_403(guild_id: str) -> str:
|
|
org_id = await streams.resolve_guild_org(guild_id)
|
|
if not org_id:
|
|
raise HTTPException(403, streams.ERROR_TEXT["no_org"])
|
|
return org_id
|
|
|
|
|
|
def _public_stream(s: dict) -> dict:
|
|
keep = ("stream_id", "guild_id", "channel_id", "channel_name", "system_id", "system_name", "mode",
|
|
"desired", "node_id", "token_id", "token_name", "actual", "stop_reason", "updated_at", "delivered",
|
|
"reused", "already_on")
|
|
return {k: s.get(k) for k in keep if k in s}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bot routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/guilds/{guild_id}/overview")
|
|
async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or_admin)):
|
|
"""Everything one Discord server may see: its org's systems and nodes, its
|
|
own streams, and which radio bots could serve it. No token strings."""
|
|
org_id = await streams.resolve_guild_org(guild_id)
|
|
if not org_id:
|
|
return {"guild_id": guild_id, "org_id": None, "linked": False,
|
|
"systems": [], "nodes": [], "streams": [], "bots": []}
|
|
link = await fstore.doc_get(streams.GUILDS, guild_id)
|
|
systems = await fstore.collection_list("systems", org_id=org_id)
|
|
nodes = await fstore.collection_list("nodes", org_id=org_id)
|
|
gstreams = [s for s in await fstore.collection_list(streams.STREAMS, guild_id=guild_id)
|
|
if s.get("desired") == "on"]
|
|
all_tokens = await fstore.collection_list(leases.TOKENS)
|
|
names = {t.get("token_id"): t.get("name") for t in all_tokens}
|
|
# Several streams can share a voice channel; the bot names each by its bot.
|
|
gstreams = [{**s, "token_name": names.get(s.get("token_id"))} for s in gstreams]
|
|
gstreams.sort(key=lambda s: (s.get("channel_name") or "", s.get("channel_id") or "", s.get("created_at") or ""))
|
|
bots = []
|
|
for t in all_tokens:
|
|
if t.get("org_id") not in (None, org_id) or t.get("valid") is False:
|
|
continue
|
|
lease = t.get("lease") or {}
|
|
member = t.get("guild_ids")
|
|
bots.append({
|
|
"token_id": t.get("token_id"),
|
|
"name": t.get("name"),
|
|
"busy_here": any(leases.guild_of(s) == guild_id for s in lease.get("streams") or []),
|
|
"member": None if member is None else guild_id in member,
|
|
# A pool bot at discord_pool_bot_guild_cap is never offered (#225, owner Q3).
|
|
"invite_url": (leases.invite_url(t.get("application_id"), guild_id)
|
|
if member or leases.invitable(t) else None),
|
|
})
|
|
return {
|
|
"guild_id": guild_id,
|
|
"org_id": org_id,
|
|
"linked": bool(link and link.get("org_id")),
|
|
"systems": [{"system_id": s.get("system_id"), "name": s.get("name")} for s in systems],
|
|
"nodes": [{"node_id": n.get("node_id"), "name": n.get("name"), "status": n.get("status"),
|
|
"system_id": streams._effective_system(n)} for n in nodes],
|
|
"streams": [_public_stream(s) for s in gstreams],
|
|
"bots": bots,
|
|
}
|
|
|
|
|
|
class StreamStart(BaseModel):
|
|
guild_id: str
|
|
channel_id: str
|
|
system_id: str
|
|
# Set = restart / edit THAT stream. Unset = add a stream to the channel
|
|
# (never overwrites another system's stream there — several may share it).
|
|
stream_id: Optional[str] = Field(None, pattern=STREAM_ID)
|
|
mode: str = "adhoc"
|
|
preferred_token_id: Optional[str] = None
|
|
channel_name: Optional[str] = None
|
|
guild_name: Optional[str] = None
|
|
created_by: Optional[str] = None
|
|
|
|
|
|
@router.post("/streams")
|
|
async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_or_admin)):
|
|
# The bot always sends snowflakes; the web admin page types them in.
|
|
if not (str(body.guild_id).isdigit() and str(body.channel_id).isdigit()):
|
|
raise HTTPException(400, "guild_id and channel_id must be Discord ids (digits only).")
|
|
org_id = await _org_or_403(body.guild_id)
|
|
system = await fstore.doc_get_cached("systems", body.system_id)
|
|
if not system or system.get("org_id") not in (None, org_id):
|
|
raise HTTPException(404, "System not found.")
|
|
try:
|
|
doc = await streams.start_stream(
|
|
org_id=org_id, guild_id=body.guild_id, channel_id=body.channel_id, system_id=body.system_id,
|
|
mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name,
|
|
created_by=body.created_by, preferred_token_id=body.preferred_token_id,
|
|
max_streams=settings.discord_guild_max_streams, stream_id=body.stream_id,
|
|
)
|
|
except streams.StreamError as e:
|
|
return _error(e)
|
|
return _public_stream(doc)
|
|
|
|
|
|
@router.get("/streams/{stream_id}")
|
|
async def get_stream(stream_id: str = Path(..., pattern=STREAM_ID),
|
|
_: dict = Depends(require_service_key_or_admin)):
|
|
doc = await fstore.doc_get(streams.STREAMS, stream_id)
|
|
if not doc:
|
|
raise HTTPException(404, "Stream not found.")
|
|
return _public_stream(doc)
|
|
|
|
|
|
class GuildStop(BaseModel):
|
|
# Neither: every live stream in the server. channel_id: every stream in
|
|
# that voice channel. stream_id: just that one (it must be this server's).
|
|
channel_id: Optional[str] = None
|
|
stream_id: Optional[str] = Field(None, pattern=STREAM_ID)
|
|
reason: str = "stopped"
|
|
|
|
|
|
@router.post("/guilds/{guild_id}/stop")
|
|
async def stop_guild(guild_id: str, body: GuildStop, _: dict = Depends(require_service_key_or_admin)):
|
|
stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id,
|
|
stream_id=body.stream_id)
|
|
return {"stopped": stopped}
|
|
|
|
|
|
class GuildLink(BaseModel):
|
|
org_id: str
|
|
# Display label only (the admin page); Discord never reads it.
|
|
guild_name: Optional[str] = None
|
|
|
|
|
|
@router.put("/guilds/{guild_id}/link")
|
|
async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(require_admin_token)):
|
|
if not guild_id.isdigit():
|
|
raise HTTPException(400, "guild_id must be a Discord server id (digits only).")
|
|
org = await fstore.doc_get("organizations", body.org_id)
|
|
if not org:
|
|
raise HTTPException(404, "Organization not found.")
|
|
previous_org = await streams.resolve_guild_org(guild_id)
|
|
doc = {
|
|
"guild_id": guild_id, "org_id": body.org_id,
|
|
"linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(),
|
|
# Re-linking a server that was unlinked or removed brings it back.
|
|
"unlinked_at": None, "unlinked_by": None, "removed_at": None, "removed_by": None,
|
|
}
|
|
if body.guild_name and body.guild_name.strip():
|
|
doc["guild_name"] = body.guild_name.strip()[:100]
|
|
await fstore.doc_set(streams.GUILDS, guild_id, doc)
|
|
# Live channels in this server were placed on the previous org's nodes;
|
|
# left running, the reconciler would keep re-placing them there, so one
|
|
# org's equipment would keep serving a server that now belongs to another.
|
|
stopped: list[str] = []
|
|
if previous_org and previous_org != body.org_id:
|
|
stopped = await streams.stop_guild(guild_id, "server linked to another organization")
|
|
return {"ok": True, "guild_id": guild_id, "org_id": body.org_id, "stopped": stopped}
|
|
|
|
|
|
@router.delete("/streams/{stream_id}")
|
|
async def delete_stream(stream_id: str = Path(..., pattern=STREAM_ID),
|
|
decoded: dict = Depends(require_admin_token)):
|
|
"""Permanently remove one radio channel: stopped through the desired-state
|
|
path first (node logs the bot out, lease released on its ack), then the
|
|
doc is deleted. Platform admin only; org owners use the node page route."""
|
|
try:
|
|
doc = await streams.delete_stream(stream_id, "removed_from_web_admin")
|
|
except streams.StreamError as e:
|
|
return _error(e)
|
|
if not doc:
|
|
raise HTTPException(404, "Radio channel not found.")
|
|
uid, email = describe_actor(decoded)
|
|
await write_audit(uid, email, "discord.stream.delete", details={
|
|
"stream_id": stream_id, "guild_id": doc.get("guild_id"), "channel_id": doc.get("channel_id"),
|
|
"org_id": doc.get("org_id"), "node_id": doc.get("node_id"), "was_desired": doc.get("desired"),
|
|
"via": "settings_discord",
|
|
})
|
|
return {"ok": True, "stream_id": stream_id}
|
|
|
|
|
|
@router.delete("/guilds/{guild_id}")
|
|
async def remove_guild(guild_id: str, leave_radio_bots: bool = False,
|
|
decoded: dict = Depends(require_admin_token)):
|
|
"""Remove a Discord server from DRB: delete its radio channels (each
|
|
stopped first), then unlink it with no fallback org (streams.remove_guild).
|
|
``leave_radio_bots`` also makes the pool radio bots (and the server's org's
|
|
own BYO bots) leave it via Discord's REST API. The DRB server bot is not
|
|
one of those and stays until someone kicks it in Discord."""
|
|
if not guild_id.isdigit():
|
|
raise HTTPException(400, "guild_id must be a Discord server id (digits only).")
|
|
known = await fstore.doc_get(streams.GUILDS, guild_id)
|
|
if not known and not await fstore.collection_list(streams.STREAMS, guild_id=guild_id):
|
|
raise HTTPException(404, "Discord server not found.")
|
|
uid, email = describe_actor(decoded)
|
|
acting_org = await streams.resolve_guild_org(guild_id)
|
|
try:
|
|
out = await streams.remove_guild(guild_id, uid)
|
|
except streams.StreamError as e:
|
|
# Channels deleted before the race are gone for good — audit them.
|
|
await write_audit(uid, email, "discord.guild.remove", details={
|
|
"guild_id": guild_id, "acting_org_id": acting_org, "outcome": "incomplete", "error": e.code,
|
|
"streams_removed": getattr(e, "removed", []), "leave_radio_bots": leave_radio_bots,
|
|
})
|
|
return _error(e)
|
|
left: list = []
|
|
if leave_radio_bots:
|
|
left = await discord_rest.leave_guild(guild_id, out["previous_org_id"] or acting_org)
|
|
await write_audit(uid, email, "discord.guild.remove", details={
|
|
"guild_id": guild_id, "previous_org_id": out["previous_org_id"], "acting_org_id": acting_org,
|
|
"streams_removed": out["streams_removed"], "leave_radio_bots": leave_radio_bots,
|
|
"radio_bots_left": [r["token_id"] for r in left if r["left"]],
|
|
"radio_bots_failed": [r["token_id"] for r in left if not r["left"]],
|
|
})
|
|
return {"ok": True, "guild_id": guild_id, "streams_removed": out["streams_removed"],
|
|
"radio_bots": [{"name": r["name"], "left": r["left"], "error": r["error"]} for r in left]}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin view — the web Discord page (server-26#213 item 2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/admin/overview")
|
|
async def admin_overview(_: dict = Depends(require_admin_token)):
|
|
"""Every Discord server DRB knows about, every radio channel (desired vs
|
|
actual) and what the page needs to act on them. Platform admin only.
|
|
|
|
A server is "known" when it is linked (``discord_guilds``) or when it has
|
|
ever had a radio channel — an unlinked server that ran ``/join`` acts for
|
|
``DISCORD_UNLINKED_GUILD_ORG_ID`` and should be visible so it can be linked.
|
|
Radio bots come from ``GET /tokens``; no token string is returned here.
|
|
Discord user ids (``created_by_discord_id``) and the linking admin's uid
|
|
are deliberately left out.
|
|
"""
|
|
links = {str(g.get("guild_id")): g for g in await fstore.collection_list(streams.GUILDS)
|
|
if g.get("guild_id")}
|
|
all_streams = await fstore.collection_list(streams.STREAMS)
|
|
orgs = {o.get("org_id"): o for o in await fstore.collection_list("organizations") if o.get("org_id")}
|
|
nodes = {n.get("node_id"): n for n in await fstore.collection_list("nodes") if n.get("node_id")}
|
|
tokens = {t.get("token_id"): t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
|
|
fallback = settings.discord_unlinked_guild_org_id or None
|
|
|
|
guilds: dict[str, dict] = {}
|
|
for gid, link in links.items():
|
|
if link.get("removed_at"):
|
|
continue # removed from DRB (DELETE /discord/guilds/{id}); re-listed if it gets a channel again
|
|
guilds[gid] = {"guild_id": gid, "guild_name": link.get("guild_name"), "linked": bool(link.get("org_id")),
|
|
"org_id": link.get("org_id"), "linked_at": link.get("linked_at"),
|
|
# Explicitly unlinked servers get no fallback org (mirrors
|
|
# streams.resolve_guild_org once server-26#218 lands).
|
|
"_no_fallback": bool(link.get("unlinked_at"))}
|
|
rows = []
|
|
for s in all_streams:
|
|
gid = str(s.get("guild_id") or leases.guild_of(s.get("stream_id") or ""))
|
|
g = guilds.setdefault(gid, {"guild_id": gid, "guild_name": None, "linked": False,
|
|
"org_id": None, "linked_at": None})
|
|
if s.get("guild_name") and not g.get("guild_name"):
|
|
g["guild_name"] = s["guild_name"]
|
|
node = nodes.get(s.get("node_id")) or {}
|
|
tok = tokens.get(s.get("token_id")) or {}
|
|
rows.append({**_public_stream(s), "guild_id": gid, "org_id": s.get("org_id"), "guild_name": s.get("guild_name"),
|
|
"created_at": s.get("created_at"), "node_name": node.get("name"),
|
|
"node_status": node.get("status"), "token_name": tok.get("name")})
|
|
for g in guilds.values():
|
|
no_fallback = g.pop("_no_fallback", False)
|
|
acting = g["org_id"] if g["linked"] else (None if no_fallback else fallback)
|
|
g["acting_org_id"] = acting
|
|
g["acting_org_name"] = (orgs.get(acting) or {}).get("name") if acting else None
|
|
g["live_streams"] = sum(1 for r in rows if r.get("guild_id") == g["guild_id"] and r.get("desired") == "on")
|
|
|
|
# Streams sharing a voice channel sit together, oldest first.
|
|
rows.sort(key=lambda r: (0 if r.get("desired") == "on" else 1, r.get("guild_name") or r.get("guild_id") or "",
|
|
r.get("channel_id") or "", r.get("created_at") or "", r.get("stream_id") or ""))
|
|
systems = await fstore.collection_list("systems")
|
|
# Names for every node a stream or a radio-bot lease points at.
|
|
referenced = {r.get("node_id") for r in rows} | {(t.get("lease") or {}).get("node_id") for t in tokens.values()}
|
|
node_names = {nid: nodes[nid].get("name") or nid for nid in referenced if nid in nodes}
|
|
return {
|
|
"guilds": sorted(guilds.values(), key=lambda g: (not g["linked"], g.get("guild_name") or g["guild_id"])),
|
|
"streams": rows,
|
|
"orgs": sorted(({"org_id": oid, "name": o.get("name") or oid} for oid, o in orgs.items()),
|
|
key=lambda o: o["name"].lower()),
|
|
"node_names": node_names,
|
|
"systems": [{"system_id": s.get("system_id"), "name": s.get("name"), "org_id": s.get("org_id")}
|
|
for s in systems if s.get("system_id")],
|
|
"limits": {"guild_max_streams": settings.discord_guild_max_streams,
|
|
"node_max_streams": settings.discord_node_max_streams,
|
|
"unlinked_guild_org_id": fallback,
|
|
"unlinked_guild_org_name": (orgs.get(fallback) or {}).get("name") if fallback else None},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Node route — token pickup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/node/{node_id}/tokens/{token_id}")
|
|
async def node_token(
|
|
node_id: str,
|
|
token_id: str,
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer),
|
|
):
|
|
"""The bot token string, for the node that holds its lease and nobody else."""
|
|
if not credentials:
|
|
raise HTTPException(401, "Missing authorization")
|
|
key_doc = await fstore.doc_get("node_keys", node_id)
|
|
stored = (key_doc or {}).get("api_key") or ""
|
|
if not stored or not secrets.compare_digest(stored, credentials.credentials):
|
|
raise HTTPException(401, "Invalid node API key")
|
|
tok = await fstore.doc_get(leases.TOKENS, token_id)
|
|
lease = (tok or {}).get("lease") or {}
|
|
if not tok or lease.get("node_id") != node_id or lease.get("releasing"):
|
|
logger.warning(f"Node {node_id} asked for token {token_id} it does not hold a lease on")
|
|
raise HTTPException(403, "This node does not hold a lease on that token.")
|
|
return {"token_id": token_id, "token": tok["token"], "lease_id": lease.get("lease_id"),
|
|
"name": tok.get("name")}
|