diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 2ed7c1e..31d2eae 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -258,8 +258,10 @@ class Settings(BaseSettings): # --- Discord permission checks, REST client, OAuth linking (server-26#225) # Every setting the #225 slices use lives here, added once (S2). - # Kill switch (S3): False restores today's token picking exactly — no - # REST permission check before a bot is handed to a node. + # Kill switch (S3): False restores token picking without the REST + # permission check — no Discord call before a bot is handed to a node. + # (The system's preferred bot is still tried first either way: that is + # a separate fix, plan finding 1, and makes no Discord call.) discord_perm_check_enabled: bool = True # Cache lifetimes for per-(bot, server) results (app/internal/discord_rest.py): # a result that lets a bot in; denied / not-a-member / invalid token; and diff --git a/drb-c2-core/app/internal/discord_leases.py b/drb-c2-core/app/internal/discord_leases.py index 869be06..250dfd5 100644 --- a/drb-c2-core/app/internal/discord_leases.py +++ b/drb-c2-core/app/internal/discord_leases.py @@ -49,6 +49,7 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional +from app.config import settings from app.internal import firestore as fstore TOKENS = "bot_tokens" @@ -92,6 +93,25 @@ def guild_of(stream_id: str) -> str: return stream_id.split(":", 1)[0] +def invitable(tok: dict) -> bool: + """May DRB suggest inviting this bot to one more server? A pool bot + (``org_id`` None) at ``discord_pool_bot_guild_cap`` servers is full: an + unverified Discord bot can't join more than 100, and the cap leaves room + (owner decision Q3, 2026-10-05: cap 90, add bots when full). An org's own + BYO bot is the org's business, never capped here.""" + if tok.get("org_id") is not None: + return True + members = tok.get("guild_ids") + return members is None or len(members) < settings.discord_pool_bot_guild_cap + + +# Permission-check outcomes the lease code acts on (server-26#225 S3). The +# full per-bot result is built by discord_streams.check_bots; pick_token reads +# only ``status`` (plus name/missing/reason/fix for the refusal). +PERM_OK = "ok" +PERM_DENIED = "denied" + + def _now() -> datetime: return datetime.now(timezone.utc) @@ -105,7 +125,11 @@ class LeaseResult: # error codes: pool_empty | no_bot_in_guild | pool_exhausted | # preferred_busy (the chosen bot already has a stream in this guild) | # preferred_other_node (the chosen bot is logged in on another node) | - # preferred_unavailable (unknown, invalid, or another org's bot) + # preferred_unavailable (unknown, invalid, or another org's bot) | + # missing_permissions (every bot that could take it lacks a Discord + # permission in the channel — ``detail["denied"]`` names each) | + # pool_full (no bot is in this server and every pool bot that could be + # invited is at discord_pool_bot_guild_cap) error: Optional[str] = None # bots that would work if invited to this guild: [{name, invite_url}] invite: list = field(default_factory=list) @@ -128,12 +152,23 @@ def _legacy_mirror(lease: Optional[dict]) -> dict: } +def _perm_status(perms: Optional[dict], tok_id: str) -> Optional[str]: + return ((perms or {}).get(tok_id) or {}).get("status") + + +def _denied_row(tok_id: str, tok: dict, perms: dict) -> dict: + p = perms.get(tok_id) or {} + return {"name": tok.get("name") or tok_id, "missing": list(p.get("missing") or []), + "reason": p.get("reason"), "fix": p.get("fix")} + + def _candidate_order(tok_id: str, tok: dict, node_id: str, org_id: Optional[str], - preferred: Optional[str]) -> tuple: + preferred: Optional[str], perms: Optional[dict] = None) -> tuple: lease = tok.get("lease") or {} return ( - 0 if tok_id == preferred else 1, + 0 if tok_id == preferred else 1, # the system's preferred bot 0 if org_id and tok.get("org_id") == org_id else 1, # BYO before pool + 0 if _perm_status(perms, tok_id) == PERM_OK else 1, # confirmed-ok before unknown 0 if lease.get("node_id") == node_id else 1, # reuse a session 0 if tok.get("guild_ids") is not None else 1, # known member first str(tok.get("name") or ""), @@ -149,18 +184,32 @@ def pick_token( org_id: Optional[str], preferred_token_id: Optional[str] = None, exclude_stream_id: Optional[str] = None, + soft_preferred_token_id: Optional[str] = None, + perms: Optional[dict] = None, ) -> LeaseResult: """Pure choice over a snapshot of bot_tokens. No I/O — unit tested directly. A ``preferred_token_id`` is a demand, not a hint: if that bot cannot take - the stream the result says why instead of quietly using another bot.""" + the stream the result says why instead of quietly using another bot. + ``soft_preferred_token_id`` (the system's preferred bot, + ``systems/{id}.preferred_token_id``) is a hint: tried first, but any other + bot that fits is used when it can't. + + ``perms`` ({token_id: {status, missing, reason, fix}}, server-26#225) is + the Discord permission check for this voice channel. A ``denied`` bot is + skipped; a confirmed ``ok`` bot sorts before one whose check was + ``unknown`` (or absent), and unknown is allowed (fail open: a Discord + outage must never block a start). None = no check (kill switch). + """ if not tokens: return LeaseResult(ok=False, error="pool_empty") if preferred_token_id: return _pick_preferred(tokens, preferred_token_id, node_id=node_id, guild_id=guild_id, - org_id=org_id, exclude_stream_id=exclude_stream_id) + org_id=org_id, exclude_stream_id=exclude_stream_id, perms=perms) usable: list[tuple] = [] not_member: list[dict] = [] + denied: list[dict] = [] + pool_full = False for tok_id, tok in tokens.items(): if tok.get("valid") is False: continue @@ -174,24 +223,41 @@ def pick_token( if guild_id in busy_guilds: continue # this bot already has a voice connection in this guild members = tok.get("guild_ids") - if members is not None and guild_id not in members: - not_member.append({"name": tok.get("name") or tok_id, - "invite_url": invite_url(tok.get("application_id"), guild_id)}) + if (members is not None and guild_id not in members) or _perm_status(perms, tok_id) == "not_member": + if invitable(tok): + not_member.append({"name": tok.get("name") or tok_id, + "invite_url": invite_url(tok.get("application_id"), guild_id)}) + else: + pool_full = True continue - usable.append((_candidate_order(tok_id, tok, node_id, org_id, preferred_token_id), tok_id)) + if _perm_status(perms, tok_id) == "invalid": + continue # Discord answered 401 during the check + if _perm_status(perms, tok_id) == PERM_DENIED: + denied.append(_denied_row(tok_id, tok, perms or {})) + continue + usable.append((_candidate_order(tok_id, tok, node_id, org_id, soft_preferred_token_id, perms), tok_id)) if usable: usable.sort() return LeaseResult(ok=True, token_id=usable[0][1]) + if denied: + # Fixing a permission is the shortest way to radio; inviting another + # bot is the other, so the invite links ride along. + return LeaseResult(ok=False, error="missing_permissions", invite=not_member, + detail={"denied": denied}) if not_member: return LeaseResult(ok=False, error="no_bot_in_guild", invite=not_member) + if pool_full: + return LeaseResult(ok=False, error="pool_full") return LeaseResult(ok=False, error="pool_exhausted") def _pick_preferred(tokens: dict[str, dict], tok_id: str, *, node_id: str, guild_id: str, - org_id: Optional[str], exclude_stream_id: Optional[str]) -> LeaseResult: + org_id: Optional[str], exclude_stream_id: Optional[str], + perms: Optional[dict] = None) -> LeaseResult: tok = tokens.get(tok_id) owner = (tok or {}).get("org_id") - if not tok or tok.get("valid") is False or (owner is not None and owner != org_id): + if (not tok or tok.get("valid") is False or (owner is not None and owner != org_id) + or _perm_status(perms, tok_id) == "invalid"): return LeaseResult(ok=False, error="preferred_unavailable", detail={"token_name": (tok or {}).get("name") or tok_id}) name = tok.get("name") or tok_id @@ -206,9 +272,14 @@ def _pick_preferred(tokens: dict[str, dict], tok_id: str, *, node_id: str, guild return LeaseResult(ok=False, error="preferred_other_node", detail={"token_name": name, "node_id": lease["node_id"]}) members = tok.get("guild_ids") - if members is not None and guild_id not in members: + if (members is not None and guild_id not in members) or _perm_status(perms, tok_id) == "not_member": + if not invitable(tok): + return LeaseResult(ok=False, error="pool_full", detail={"token_name": name}) return LeaseResult(ok=False, error="no_bot_in_guild", invite=[{"name": name, "invite_url": invite_url(tok.get("application_id"), guild_id)}]) + if _perm_status(perms, tok_id) == PERM_DENIED: + return LeaseResult(ok=False, error="missing_permissions", + detail={"token_name": name, "denied": [_denied_row(tok_id, tok, perms or {})]}) return LeaseResult(ok=True, token_id=tok_id) @@ -219,12 +290,18 @@ async def lease_for_stream( org_id: Optional[str], preferred_token_id: Optional[str] = None, stream_fields: Optional[dict] = None, + soft_preferred_token_id: Optional[str] = None, + perms: Optional[dict] = None, ) -> LeaseResult: """Bind ``stream_id`` to a token leased to ``node_id``, atomically. Also writes ``stream_fields`` (desired=on, node_id, token_id, lease_id...) onto the stream doc in the same transaction, so a stream doc can never name a token whose lease does not list it. + + ``perms`` / ``soft_preferred_token_id``: see ``pick_token``. A stream's + current token is reused only while its permission check is not + ``denied`` — otherwise the stream moves to a bot that can speak there. """ guild_id = guild_of(stream_id) db = fstore.db @@ -243,6 +320,7 @@ async def lease_for_stream( and (prev_tok.get("lease") or {}).get("node_id") == node_id and stream_id in _lease_streams(prev_tok) and prev_tok.get("valid") is not False + and _perm_status(perms, prev_tok_id) not in (PERM_DENIED, "invalid", "not_member") # An explicit edit to a different bot moves the stream to it. and (not preferred_token_id or preferred_token_id == prev_tok_id) ): @@ -255,7 +333,8 @@ async def lease_for_stream( return LeaseResult(ok=True, token_id=prev_tok_id, lease_id=lease["lease_id"], reused=True) result = pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id, - preferred_token_id=preferred_token_id, exclude_stream_id=stream_id) + preferred_token_id=preferred_token_id, exclude_stream_id=stream_id, + soft_preferred_token_id=soft_preferred_token_id, perms=perms) if not result.ok: return result diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index 13fa8d3..1f7a907 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -23,10 +23,20 @@ the same system in the same channel (idempotent re-add / restart). Fields: state: pending | connecting | connected | error | stopping | node_offline | paused_locally code: the node's machine-readable reason on error (login_failed, - connect_failed, not_in_guild, bad_channel, no_audio) or None + connect_failed, not_in_guild, bad_channel, no_audio, + missing_permissions, channel_full) or None tg_filter: None — phase-2 seam (per-channel talkgroup gate); always None now now_playing_message_id, created_by_discord_id, created_at, updated_at +Before a token is leased, ``start_stream`` asks Discord (REST, cached — +discord_rest.guild_snapshot) whether each bot it could pick may View, Connect +and Speak in the voice channel (server-26#225 S3). A bot that is denied is +never handed to a node; when every bot is denied the start is refused with +the bot, the channel, the missing permission and what removed it. A check +that can't be completed is "unknown" and allowed (fail open). Kill switch: +``DISCORD_PERM_CHECK_ENABLED=false``. The system's preferred bot +(``systems/{id}.preferred_token_id``) is tried first when nobody named one. + Flow: ``start_stream`` picks a node + leases a token -> ``publish_desired`` retains ``nodes/{node}/discord/desired`` (full list for that node, with a monotonically increasing ``seq``) -> the edge reconciles and publishes @@ -49,6 +59,7 @@ from typing import Optional from app.config import settings from app.internal import firestore as fstore from app.internal import discord_leases as leases +from app.internal import discord_perms from app.internal import discord_rest from app.internal.logger import logger @@ -203,6 +214,178 @@ async def publish_desired(node_id: str) -> bool: return bool(ok) +# --------------------------------------------------------------------------- +# Discord permission check (server-26#225 S3) +# --------------------------------------------------------------------------- + +# Trace sentences (discord_perms) that only say a permission went with +# another one. The fix is for the other one, so they are not what we report. +_IMPLIED = ("without View Channel", "without Connect") + + +def _join_labels(labels: list) -> str: + if len(labels) <= 1: + return "".join(labels) + return ", ".join(labels[:-1]) + " or " + labels[-1] + + +def _fix_text(bot: str, missing: list, reason: str) -> str: + if "timed out" in reason: + return (f"In Discord: remove {bot}'s timeout (right-click it in the member list → Remove Timeout), " + "then try again.") + allow = " and ".join(missing) if len(missing) <= 2 else ", ".join(missing[:-1]) + " and " + missing[-1] + return f"In Discord: Edit Channel → Permissions → add {bot} → allow {allow}, then try again." + + +def evaluate_access(snap: "discord_rest.GuildSnapshot", channel_id: str, *, bot: str) -> dict: + """One bot's verdict for one voice channel, from its cached snapshot. + + ``status``: ok | denied | unknown | not_member | invalid | bad_channel + (the channel exists but is not a voice channel). ``missing`` lists the + edge's labels ("View Channel", "Connect", "Speak"); ``reason`` is the + trace sentence for the first one that isn't just implied by another; + ``fix`` is what to change in Discord. Never raises.""" + out: dict = {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None, + "user_limit": 0, "limit_bypass": False, "channel_name": None} + if snap.status != discord_rest.OK: + out["status"] = snap.status + out["error"] = snap.error + return out + channel = next((c for c in snap.channels or [] + if isinstance(c, dict) and str(c.get("id")) == str(channel_id)), None) + if channel is None: + # Deleted, or hidden from the bot's list: the edge's join is the judge. + out["error"] = "channel not in the server's channel list" + return out + out["channel_name"] = channel.get("name") + try: + bits, trace = discord_perms.effective_permissions(snap.guild or {}, snap.member or {}, channel) + except (ValueError, TypeError, AttributeError) as e: + out["error"] = f"could not read permissions ({type(e).__name__})" + return out + verdict = discord_perms.check_voice(bits, channel, trace) + out["user_limit"] = verdict["user_limit"] + out["limit_bypass"] = verdict["limit_bypass"] + if not verdict["is_voice"]: + out["status"] = "bad_channel" + return out + if verdict["ok"]: + out["status"] = leases.PERM_OK + return out + missing = verdict["missing"] + reasons = verdict["reasons"] + primary = [m for m in missing if not any(x in (reasons.get(m) or "") for x in _IMPLIED)] or missing + out.update(status=leases.PERM_DENIED, missing=missing, reasons=reasons, + reason=reasons.get(primary[0]) or "not granted", + fix=_fix_text(bot, primary, reasons.get(primary[0]) or "")) + return out + + +def tokens_to_check(tokens: dict[str, dict], *, guild_id: str, org_id: Optional[str], + node_ids: Optional[set] = None, exclude_stream_id: Optional[str] = None, + preferred_token_id: Optional[str] = None) -> list[str]: + """The bots ``pick_token`` could hand this stream to, best first — the + only ones worth asking Discord about. ``node_ids`` None = any node.""" + out = [] + for tok_id, tok in tokens.items(): + if tok.get("valid") is False or not tok.get("token"): + continue + if tok.get("org_id") is not None and tok.get("org_id") != org_id: + continue # another org's bot is never asked about this org's server + members = tok.get("guild_ids") + if members is not None and guild_id not in members: + continue + if preferred_token_id: + if tok_id == preferred_token_id: + out.append(tok_id) + continue + lease_node = (tok.get("lease") or {}).get("node_id") + if lease_node and node_ids is not None and lease_node not in node_ids: + continue + if any(leases.guild_of(s) == guild_id + for s in leases._lease_streams(tok) if s != exclude_stream_id): + continue + out.append(tok_id) + out.sort(key=lambda t: leases._candidate_order(t, tokens[t], "", org_id, None)) + return out + + +def _unanswered() -> dict: + return {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None, + "error": "Discord did not answer in time"} + + +async def gather_snapshots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, *, + refresh: bool = False, budget: Optional[float] = None) -> dict: + """``discord_rest.guild_snapshot`` for each bot, at most + ``discord_perm_check_concurrency`` at a time, all within + ``discord_perm_check_budget_seconds``. A bot not answered in time maps + to None — unknown (its load keeps running and fills the cache for next + time: the cache shields it from this cancellation).""" + out: dict = {} + if not token_ids: + return out + sem = asyncio.Semaphore(max(1, settings.discord_perm_check_concurrency)) + + async def _one(tok_id: str) -> None: + async with sem: + out[tok_id] = await discord_rest.guild_snapshot({**tokens[tok_id], "token_id": tok_id}, guild_id, + refresh=refresh) + + tasks = [asyncio.create_task(_one(t)) for t in token_ids] + _, pending = await asyncio.wait(tasks, timeout=budget if budget is not None + else settings.discord_perm_check_budget_seconds) + for t in pending: + t.cancel() + for t in tasks: + if t.done() and not t.cancelled() and t.exception() is not None: + logger.warning(f"Discord permission check failed: {type(t.exception()).__name__}") + return {t: out.get(t) for t in token_ids} + + +async def check_bots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, channel_id: str, + *, refresh: bool = False, budget: Optional[float] = None) -> dict[str, dict]: + """``evaluate_access`` for each bot in one voice channel. Never raises.""" + snaps = await gather_snapshots(tokens, token_ids, guild_id, refresh=refresh, budget=budget) + return {t: (evaluate_access(snaps[t], channel_id, bot=tokens[t].get("name") or t) if snaps.get(t) + else _unanswered()) + for t in token_ids} + + +async def _perm_check(*, org_id: str, guild_id: str, channel_id: str, stream_id: str, + preferred_token_id: Optional[str], node_ids: set) -> Optional[dict]: + if not settings.discord_perm_check_enabled: + return None + tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")} + ids = tokens_to_check(tokens, guild_id=guild_id, org_id=org_id, node_ids=node_ids, + exclude_stream_id=stream_id, preferred_token_id=preferred_token_id) + return await check_bots(tokens, ids, guild_id, channel_id) + + +def _channel_ref(perms: Optional[dict], channel_name: Optional[str]) -> str: + name = next((p.get("channel_name") for p in (perms or {}).values() if p.get("channel_name")), None) + name = name or channel_name + return f"#{name}" if name else "that voice channel" + + +def _missing_permissions_error(denied: list, invite: list, chan: str) -> "StreamError": + def line(d: dict) -> str: + return f"{d['name']} can't {_join_labels(d.get('missing') or ['Speak'])} in {chan}: {d.get('reason')}." + if len(denied) == 1: + d = denied[0] + text = f"{line(d)} {d.get('fix') or ''}".strip() + else: + shown = denied[:3] + text = f"No radio bot can play in {chan}:\n" + "\n".join(f"- {line(d)}" for d in shown) + if len(denied) > len(shown): + text += f"\n- …and {len(denied) - len(shown)} more." + text += "\nIn Discord: Edit Channel → Permissions → add one of these bots → allow the missing " \ + "permission, then try again." + if invite: + text += " Or invite another radio bot with a link below." + return StreamError("missing_permissions", text, invite=invite, extra={"denied": denied}) + + # --------------------------------------------------------------------------- # Start / stop # --------------------------------------------------------------------------- @@ -229,6 +412,10 @@ ERROR_TEXT = { "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.", "preferred_unavailable": "That radio bot can't be used here (it was removed, failed validation, or belongs to " "another organization). Pick another bot, or let DRB choose any free one.", + "pool_full": ("No radio bot is in this server, and every DRB radio bot that could be invited is already in its " + "maximum number of servers. A platform admin needs to add another radio bot."), + "missing_permissions": "No radio bot has the Discord permissions it needs in that voice channel.", + "bad_channel": "That is not a voice channel. Radio can only play in a voice channel.", "not_found": "That radio channel no longer exists.", } @@ -264,10 +451,12 @@ async def find_channel_stream(guild_id: str, channel_id: str, system_id: str) -> return same[0] if same else None -async def _lease_error(res: leases.LeaseResult) -> "StreamError": +async def _lease_error(res: leases.LeaseResult, chan: str = "that voice channel") -> "StreamError": code = res.error or "no_node" d = res.detail or {} bot = d.get("token_name") or "That radio bot" + if code == "missing_permissions" and d.get("denied"): + return _missing_permissions_error(d["denied"], res.invite, chan) if code == "preferred_busy": busy = (await fstore.doc_get(STREAMS, d["busy_stream_id"]) if d.get("busy_stream_id") else None) or {} what = _stream_label(busy) if busy else "another radio channel" @@ -307,7 +496,14 @@ async def start_stream( the same channel is never touched. ``preferred_token_id`` is strict: a bot already in this server, on another - node, or unusable is refused with a reason, never silently swapped. + node, or unusable is refused with a reason, never silently swapped. With + none given, the system's ``preferred_token_id`` is tried first (a hint: + another bot is used when it can't take the stream). + + Every bot that could be picked is permission-checked against the voice + channel first (``_perm_check``; server-26#225). Every entry point — the + bot's /join, the web picker, the node command and the reconciler — + comes through here, so none can hand a node a bot Discord will refuse. """ if mode not in ("persistent", "adhoc"): raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400) @@ -362,6 +558,14 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i raise StreamError("limit", ERROR_TEXT["limit"]) system = await fstore.doc_get_cached("systems", system_id) or {} + # Finding 1 (#225 plan): "preferred for this system" was written by + # PUT /tokens/{id}/prefer/{system} and never read. It is a hint, not a + # demand — an explicit preferred_token_id still wins and stays strict. + # Read uncached: the systems cache lives 5 min, and a preference the + # owner just set should apply to the next start. + soft_pref = None + if not preferred_token_id: + soft_pref = ((await fstore.doc_get("systems", system_id)) or {}).get("preferred_token_id") or None if node_id: node = await fstore.doc_get("nodes", node_id) @@ -372,16 +576,18 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i candidates = [node] else: candidates = await candidate_nodes(org_id, system_id, exclude_stream_id=stream_id) - # Keep the current node when it is still a valid choice (reuse the lease). - cur = existing.get("node_id") - if cur and existing.get("system_id") == system_id: - candidates.sort(key=lambda n: 0 if n["node_id"] == cur else 1) - if preferred_token_id: - # A chosen bot can only run on the node it is logged in on. - tok = await fstore.doc_get(leases.TOKENS, preferred_token_id) or {} + cur = existing.get("node_id") if existing.get("system_id") == system_id else None + held_on = None + if preferred_token_id or soft_pref: + # A bot can only run on the node it is logged in on. + tok = await fstore.doc_get(leases.TOKENS, preferred_token_id or soft_pref) or {} held_on = (tok.get("lease") or {}).get("node_id") - if held_on: - candidates.sort(key=lambda n: 0 if n["node_id"] == held_on else 1) + # Keep the current node when it is still a valid choice (reuse the + # lease) — unless a bot was demanded, whose node then comes first. + order = [held_on, cur] if preferred_token_id else [cur, held_on] + for nid in reversed(order): + if nid: + candidates.sort(key=lambda n, nid=nid: 0 if n["node_id"] == nid else 1) if not candidates: raise StreamError("no_node", ERROR_TEXT["no_node"]) @@ -404,12 +610,22 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i fields["created_by_discord_id"] = created_by fields["now_playing_message_id"] = None + node_ids = {n["node_id"] for n in candidates} + perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id, + preferred_token_id=preferred_token_id, node_ids=node_ids) + if perms and any(p.get("status") == "bad_channel" for p in perms.values()): + raise StreamError("bad_channel", f"{_channel_ref(perms, channel_name)} is not a voice channel. " + "Radio can only play in a voice channel — pick one from the list.", + status=400) + last: Optional[leases.LeaseResult] = None + perm_refusal: Optional[leases.LeaseResult] = None prev_node = existing.get("node_id") refreshed = False for node in candidates: res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, - preferred_token_id=preferred_token_id, stream_fields=fields) + preferred_token_id=preferred_token_id, stream_fields=fields, + soft_preferred_token_id=soft_pref, perms=perms) if not res.ok and res.error == "no_bot_in_guild" and not refreshed: # Membership is cached on the token doc; someone may have just # invited a bot. Re-read it from Discord once, then retry. @@ -418,8 +634,14 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i stale = [t["token_id"] for t in tokens if t.get("token_id") and t.get("guild_ids") is not None and guild_id not in t["guild_ids"]] if await discord_rest.refresh_membership(stale): + if perms is not None: + # Bots that just turned out to be members get checked too. + perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, + stream_id=stream_id, preferred_token_id=preferred_token_id, + node_ids=node_ids) res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, - preferred_token_id=preferred_token_id, stream_fields=fields) + preferred_token_id=preferred_token_id, stream_fields=fields, + soft_preferred_token_id=soft_pref, perms=perms) if res.ok: delivered = await publish_desired(node["node_id"]) if prev_node and prev_node != node["node_id"]: @@ -429,12 +651,17 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i return {**existing, **fields, "node_id": node["node_id"], "node_name": node.get("name"), "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused, "delivered": delivered, "already_on": already_on} - if res.error in ("preferred_busy", "preferred_unavailable"): - raise await _lease_error(res) # true on every node — no point trying the rest + chan = _channel_ref(perms, channel_name or existing.get("channel_name")) + if res.error in ("preferred_busy", "preferred_unavailable") or ( + preferred_token_id and res.error == "missing_permissions"): + raise await _lease_error(res, chan) # true on every node — no point trying the rest + if res.error == "missing_permissions": + perm_refusal = res # more useful than another node's "all busy" last = res if last is None: raise StreamError("no_node", ERROR_TEXT["no_node"]) - raise await _lease_error(last) + raise await _lease_error(perm_refusal or last, + _channel_ref(perms, channel_name or existing.get("channel_name"))) async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: @@ -550,6 +777,8 @@ async def stop_node(node_id: str, reason: str) -> list[str]: # --------------------------------------------------------------------------- _VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"} +# Edge error codes that mean the cached permission/membership view is stale. +_INVALIDATE_ON = {"missing_permissions", "channel_full", "not_in_guild", "bad_channel"} def _log_transition(node_id: str, doc: dict, prev_state: Optional[str], state: str, @@ -595,6 +824,10 @@ async def handle_actual(node_id: str, payload: dict) -> None: code = str(code)[:40] if code else None if code == "not_in_guild" and doc.get("token_id"): recheck.add(doc["token_id"]) + if code in _INVALIDATE_ON and doc.get("token_id"): + # The edge saw Discord refuse; a cached "ok" must not keep + # handing this bot the same channel (server-26#225). + discord_rest.invalidate(token_id=doc["token_id"], guild_id=str(doc.get("guild_id"))) prev = doc.get("actual") or {} if prev.get("state") == state and prev.get("error") == error and prev.get("code") == code: since = prev.get("since") or now diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index 35c631a..8ebe83a 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -24,6 +24,7 @@ from app.routers import discord_link as discord_link_router from app.internal import dynsec from app.internal import firestore as fstore from app.routers import discord_web +from app.routers import discord_checks async def _clear_legacy_token_flags(): @@ -164,6 +165,7 @@ app.include_router(discord_web.router) # the token-fetch route takes the calling node's own API key. app.include_router(discord_router.router) app.include_router(discord_link_router.router) # auth per-endpoint (org owner / bot service key) — server-26#213 +app.include_router(discord_checks.router) # auth per-endpoint (org member / service key or admin) — server-26#225 # NOTE: there used to be an app.routers.mqtt_auth router here (an HTTP # backend for the mosquitto-go-auth plugin). That plugin's upstream project # is archived (no CVE patches) and was rejected for an internet-facing diff --git a/drb-c2-core/app/routers/discord.py b/drb-c2-core/app/routers/discord.py index 040c4c8..4e8a069 100644 --- a/drb-c2-core/app/routers/discord.py +++ b/drb-c2-core/app/routers/discord.py @@ -93,7 +93,9 @@ async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or "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, - "invite_url": leases.invite_url(t.get("application_id"), guild_id), + # 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, diff --git a/drb-c2-core/app/routers/discord_checks.py b/drb-c2-core/app/routers/discord_checks.py new file mode 100644 index 0000000..39f5131 --- /dev/null +++ b/drb-c2-core/app/routers/discord_checks.py @@ -0,0 +1,215 @@ +""" +Discord voice-channel permission status for the picker (server-26#225 S3). + +Read-only. For one Discord server and its voice channels, says for each radio +bot that could serve it whether Discord lets that bot View, Connect and Speak +there, and when not, which permission is missing, what removed it and what to +change. Built on the same cached REST snapshots ``start_stream`` checks before +it hands a bot to a node (discord_streams.gather_snapshots / evaluate_access), +so the badge the UI shows and the refusal a start gets agree. + +Auth: + * GET /discord/nodes/{node_id}/guilds/{guild_id}/voice-check + any signed-in member of the node's org, or a platform admin + (``discord_web._node_for_caller``); the server must be one of the org's + (``discord_web._org_guilds``). Anything else is a 404. Bot NAMES only — + no token ids for non-admins, never a token string. + * GET /discord/guilds/{guild_id}/voice-check + the Discord bot's service key or a platform admin; rows carry token ids. + +``refresh=1`` skips the cache, at most once per 10 s per caller and server; +past that the cached answer is served with ``refreshed: false``. + +User limits are advisory: REST can't see who is connected, so the edge's +``channel_full`` stays the enforcement. +""" +from __future__ import annotations + +import time +from datetime import datetime, timezone +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query + +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.auth import get_role, require_firebase_token, require_service_key_or_admin +from app.routers.discord_web import _guild_name, _node_for_caller, _org_guilds + +router = APIRouter(prefix="/discord", tags=["discord"]) + +REFRESH_EVERY_SECONDS = 10.0 +_SNOWFLAKE = r"^[0-9]{1,25}$" +_last_refresh: dict[tuple[str, str], float] = {} + + +def _refresh_allowed(caller: str, guild_id: str) -> bool: + now = time.monotonic() + key = (caller, guild_id) + if now - _last_refresh.get(key, -1e9) < REFRESH_EVERY_SECONDS: + return False + _last_refresh[key] = now + if len(_last_refresh) > 5000: # bounded: drop entries past the window + for k in [k for k, t in _last_refresh.items() if now - t >= REFRESH_EVERY_SECONDS]: + _last_refresh.pop(k, None) + return True + + +def _availability(tok: dict, *, guild_id: str, node_id: Optional[str]) -> Optional[str]: + """None when the bot could take a new stream here, else why not.""" + if any(leases.guild_of(s) == guild_id for s in leases._lease_streams(tok)): + return "busy_here" # one voice connection per bot per server + lease_node = (tok.get("lease") or {}).get("node_id") + if node_id and lease_node and lease_node != node_id: + return "other_node" # logged in on another node + return None + + +def _channel_state(rows: list[dict], enabled: bool) -> str: + if not enabled: + return "unchecked" + if any(r["status"] == "bad_channel" for r in rows): + return "bad_channel" + if not rows: + return "no_bot" + free = [r for r in rows if r["available"]] + if any(r["ok"] for r in free): + return "ready" + if any(r["status"] == discord_rest.UNKNOWN for r in free): + return "unknown" # fail open: a start is still tried + if any(r["status"] == leases.PERM_DENIED for r in free): + return "needs_fix" # a start is refused with missing_permissions + if any(r["ok"] for r in rows): + return "busy" + return "no_bot" + + +async def voice_check(*, guild_id: str, org_id: Optional[str], node_id: Optional[str], + channels: list[dict], channel_id: Optional[str], refresh: bool, + with_token_ids: bool) -> dict: + """The JSON contract (see the PR for server-26#225 S3). ``channels`` is + the server's directory (``[{channel_id, name}]``); when it is empty the + voice channels a bot's snapshot lists are used instead.""" + enabled = settings.discord_perm_check_enabled + tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")} + eligible = {tid: t for tid, t in tokens.items() + if t.get("valid") is not False and t.get("org_id") in (None, org_id)} + members = [tid for tid, t in eligible.items() + if t.get("guild_ids") is None or guild_id in t["guild_ids"]] + members.sort(key=lambda t: leases._candidate_order(t, eligible[t], node_id or "", org_id, None)) + invite = [{"name": t.get("name") or tid, "invite_url": leases.invite_url(t.get("application_id"), guild_id)} + for tid, t in sorted(eligible.items(), key=lambda kv: str(kv[1].get("name") or kv[0])) + if tid not in members and leases.invitable(t) and t.get("application_id")][:5] + + snaps = (await streams.gather_snapshots(tokens, [m for m in members if tokens[m].get("token")], guild_id, + refresh=refresh) + if enabled else {}) + + picked = [{"channel_id": str(c.get("channel_id")), "name": c.get("name")} for c in channels] + if not picked: + seen = next((s for s in snaps.values() if s is not None and s.status == discord_rest.OK), None) + picked = [{"channel_id": str(c.get("id")), "name": c.get("name")} + for c in sorted((seen.channels if seen else []) or [], + key=lambda c: (c.get("position") or 0, str(c.get("name") or ""))) + if isinstance(c, dict) and c.get("type") == 2] + if channel_id: + picked = [c for c in picked if c["channel_id"] == channel_id] or [{"channel_id": channel_id, "name": None}] + + # Bots whose rows say "not a member" after the check move to the invite list. + newly_out = {t for t, s in snaps.items() if s is not None and s.status == discord_rest.NOT_MEMBER} + out_channels = [] + for c in picked: + rows = [] + for tid in members: + if tid in newly_out: + continue + tok = tokens[tid] + name = tok.get("name") or tid + if not enabled: + res = {"status": "unchecked", "missing": [], "reason": None, "fix": None} + elif snaps.get(tid) is None: + res = streams._unanswered() + else: + res = streams.evaluate_access(snaps[tid], c["channel_id"], bot=name) + why = _availability(tok, guild_id=guild_id, node_id=node_id) + row = { + "name": name, + "ok": res["status"] == leases.PERM_OK, + "status": res["status"], + "missing": res.get("missing") or [], + "reason": res.get("reason") or (res.get("error") if res["status"] == discord_rest.UNKNOWN else None), + "fix": res.get("fix"), + "available": why is None, + "unavailable_reason": why, + "user_limit": res.get("user_limit") or 0, + "limit_bypass": bool(res.get("limit_bypass")), + } + if with_token_ids: + row["token_id"] = tid + rows.append(row) + if not c["name"] and res.get("channel_name"): + c["name"] = res["channel_name"] + limit = max((r["user_limit"] for r in rows), default=0) + out_channels.append({ + "channel_id": c["channel_id"], + "name": c["name"], + "state": _channel_state(rows, enabled), + "any_ok": any(r["ok"] for r in rows), + "user_limit": limit, + "limit_bypass": any(r["ok"] and r["limit_bypass"] for r in rows), + "bots": [{k: v for k, v in r.items() if k not in ("user_limit", "limit_bypass")} for r in rows], + }) + for tid in sorted(newly_out): + t = tokens[tid] + if leases.invitable(t) and t.get("application_id") and len(invite) < 5: + invite.append({"name": t.get("name") or tid, + "invite_url": leases.invite_url(t.get("application_id"), guild_id)}) + return { + "guild_id": guild_id, + "node_id": node_id, + "enabled": enabled, + "checked_at": datetime.now(timezone.utc).isoformat(), + "refreshed": refresh, + "channels": out_channels, + "invite": invite, + } + + +@router.get("/nodes/{node_id}/guilds/{guild_id}/voice-check") +async def node_voice_check(node_id: str, guild_id: str, + refresh: int = Query(0, ge=0, le=1), + channel_id: Optional[str] = Query(None, pattern=_SNOWFLAKE), + decoded: dict = Depends(require_firebase_token)): + node, _ = await _node_for_caller(node_id, decoded) + org_id = node.get("org_id") + guild = next((g for g in (await _org_guilds(org_id) if org_id else []) if g["guild_id"] == str(guild_id)), None) + if not guild: + raise HTTPException(404, "That Discord server is not linked to this organization.") + do_refresh = bool(refresh) and _refresh_allowed(f"uid:{decoded.get('uid') or '?'}", guild["guild_id"]) + out = await voice_check(guild_id=guild["guild_id"], org_id=org_id, node_id=node_id, + channels=guild.get("voice_channels") or [], channel_id=channel_id, + refresh=do_refresh, with_token_ids=get_role(decoded) == "admin") + out["guild_name"] = _guild_name(guild) + return out + + +@router.get("/guilds/{guild_id}/voice-check") +async def guild_voice_check(guild_id: str, + channel_id: Optional[str] = Query(None, pattern=_SNOWFLAKE), + refresh: int = Query(0, ge=0, le=1), + decoded: dict = Depends(require_service_key_or_admin)): + if not guild_id.isdigit(): + raise HTTPException(400, "guild_id must be a Discord server id (digits only).") + org_id = await streams.resolve_guild_org(guild_id) + link = await fstore.doc_get(streams.GUILDS, guild_id) or {} + caller = "service" if decoded.get("service") else f"uid:{decoded.get('uid') or '?'}" + do_refresh = bool(refresh) and _refresh_allowed(caller, guild_id) + out = await voice_check(guild_id=guild_id, org_id=org_id, node_id=None, + channels=link.get("voice_channels") or [], channel_id=channel_id, + refresh=do_refresh, with_token_ids=True) + out["guild_name"] = _guild_name(link) + out["org_id"] = org_id + return out diff --git a/drb-c2-core/app/routers/tokens.py b/drb-c2-core/app/routers/tokens.py index d3310a1..bfb6f50 100644 --- a/drb-c2-core/app/routers/tokens.py +++ b/drb-c2-core/app/routers/tokens.py @@ -5,7 +5,14 @@ from typing import Optional from app.internal import firestore as fstore from app.internal import discord_leases as leases from app.internal import discord_rest -from app.internal.auth import require_admin_token +from app.internal.audit import write_audit +from app.internal.auth import ( + describe_actor, + get_org_role, + get_role, + require_admin_token, + require_firebase_token, +) from app.internal.logger import logger router = APIRouter(prefix="/tokens", tags=["tokens"]) @@ -94,35 +101,99 @@ async def flush_tokens(_: dict = Depends(require_admin_token)): return {"released": released} +def _may_edit_system(decoded: dict, system: Optional[dict]) -> bool: + """Platform admin: any system. Org owner: only their own org's systems — + a shared system (no org_id) serves every org, so a preference there is + platform-admin only (owner decision Q2, 2026-10-05).""" + if get_role(decoded) == "admin": + return True + return bool(system and system.get("org_id") and system.get("org_id") == decoded.get("org_id")) + + +def _check_prefer_access(decoded: dict, token: Optional[dict], system: Optional[dict], + system_id: Optional[str]) -> None: + """Raise unless the caller may point ``system`` at ``token``. Another + org's bot or system is a 404, never a 403 (no existence leak).""" + if get_role(decoded) == "admin": + if not token: + raise HTTPException(404, "Token not found.") + if system_id is not None and not system: + raise HTTPException(404, "System not found.") + return + org_id = decoded.get("org_id") + if not org_id or get_org_role(decoded) != "owner": + raise HTTPException(403, "Only the organization owner or a platform admin can choose a system's radio bot.") + if not token or (token.get("org_id") is not None and token.get("org_id") != org_id): + raise HTTPException(404, "Token not found.") + if token.get("org_id") is None: + raise HTTPException(403, "Only a platform admin can make a shared DRB radio bot preferred. " + "Organization owners can choose their own organization's radio bots.") + if system_id is None: + return + if not system or (system.get("org_id") and system.get("org_id") != org_id): + raise HTTPException(404, "System not found.") + if not system.get("org_id"): + raise HTTPException(403, "That system is shared by every organization; only a platform admin can " + "choose its radio bot.") + + @router.put("/{token_id}/prefer/{system_id}", status_code=200) async def set_preferred_system( token_id: str, system_id: str, - _: dict = Depends(require_admin_token), + decoded: dict = Depends(require_firebase_token), ): """ - Mark this token as the preferred bot for a system. - When a discord_join is issued for any node in that system, this token - is tried first before falling back to the general pool. + Mark this token as the preferred bot for a system: ``start_stream`` tries + it first for that system's radio channels when nobody named a bot + (server-26#225 S3), and uses another bot when it can't take the stream. Pass system_id="_none" to clear the preference. + + Who (owner decision Q2, 2026-10-05): a platform admin for any bot and any + system; an org owner for their OWN org's systems and their OWN + bring-your-own bots. Shared systems (no org_id) and pool bots stay + platform-admin only. """ + if "/" in token_id or "/" in system_id: + raise HTTPException(400, "Bad id.") existing = await fstore.doc_get("bot_tokens", token_id) - if not existing: - raise HTTPException(404, "Token not found.") + uid, email = describe_actor(decoded) if system_id == "_none": - # Clear any existing preference on the system that pointed to this token. - system_doc = await fstore.doc_get("systems", existing.get("preferred_for_system_id", "")) - if system_doc: - await fstore.doc_set("systems", existing["preferred_for_system_id"], {"preferred_token_id": None}) + _check_prefer_access(decoded, existing, None, None) + old_id = existing.get("preferred_for_system_id") or "" + system_doc = await fstore.doc_get("systems", old_id) if old_id else None + if system_doc and system_doc.get("preferred_token_id") == token_id: + if not _may_edit_system(decoded, system_doc): + raise HTTPException(403, "That system is shared by every organization; only a platform admin " + "can change its radio bot.") + await fstore.doc_set("systems", old_id, {"preferred_token_id": None}) await fstore.doc_set("bot_tokens", token_id, {"preferred_for_system_id": None}) + await write_audit(uid, email, "discord.token.prefer", details={ + "token_id": token_id, "system_id": None, "previous_system_id": old_id or None}) return {"ok": True, "preferred_for_system_id": None} system_doc = await fstore.doc_get("systems", system_id) - if not system_doc: - raise HTTPException(404, "System not found.") - # Set preference on both sides for easy lookup in either direction. + _check_prefer_access(decoded, existing, system_doc, system_id) + + # Keep both pointers single-valued: this bot stops being preferred for the + # system it was preferred for (when the caller may edit that system), and + # the bot this system preferred before loses its back-pointer. + old_sys_id = existing.get("preferred_for_system_id") + if old_sys_id and old_sys_id != system_id: + old_sys = await fstore.doc_get("systems", old_sys_id) + if old_sys and old_sys.get("preferred_token_id") == token_id and _may_edit_system(decoded, old_sys): + await fstore.doc_set("systems", old_sys_id, {"preferred_token_id": None}) + old_tok_id = system_doc.get("preferred_token_id") + if old_tok_id and old_tok_id != token_id: + old_tok = await fstore.doc_get("bot_tokens", old_tok_id) + if old_tok and old_tok.get("preferred_for_system_id") == system_id: + await fstore.doc_set("bot_tokens", old_tok_id, {"preferred_for_system_id": None}) + await fstore.doc_set("systems", system_id, {"preferred_token_id": token_id}) await fstore.doc_set("bot_tokens", token_id, {"preferred_for_system_id": system_id}) + await write_audit(uid, email, "discord.token.prefer", details={ + "token_id": token_id, "system_id": system_id, "previous_token_id": old_tok_id, + "org_id": system_doc.get("org_id")}) return {"ok": True, "preferred_for_system_id": system_id} diff --git a/drb-c2-core/tests/conftest.py b/drb-c2-core/tests/conftest.py index 41c939f..05a31b1 100644 --- a/drb-c2-core/tests/conftest.py +++ b/drb-c2-core/tests/conftest.py @@ -72,3 +72,27 @@ except ModuleNotFoundError: ): sys.modules.setdefault(_name, ModuleType(_name)) sys.modules["google.cloud.firestore_v1.base_query"].FieldFilter = MagicMock() + + +import pytest # noqa: E402 + + +@pytest.fixture(autouse=True) +def _no_real_discord(monkeypatch): + """No test reaches discord.com (server-26#225 S3). + + ``start_stream`` now asks Discord, over REST, whether a bot may speak in + the channel before it hands the bot to a node. Every Discord call in a + test is answered by a transport that fails to connect, so the check is + "unknown" and fails open — today's token picking — unless the test + installs its own transport or stubs ``discord_rest.guild_snapshot``.""" + import httpx + from app.internal import discord_rest + + def _offline(request): + raise httpx.ConnectError("tests never reach Discord", request=request) + + discord_rest.reset() + monkeypatch.setattr(discord_rest, "_transport", httpx.MockTransport(_offline)) + yield + discord_rest.reset() diff --git a/drb-c2-core/tests/test_discord_perm_check.py b/drb-c2-core/tests/test_discord_perm_check.py new file mode 100644 index 0000000..8324f0b --- /dev/null +++ b/drb-c2-core/tests/test_discord_perm_check.py @@ -0,0 +1,557 @@ +""" +Discord permission check wired into token picking (server-26#225 S3). + +Before a radio bot is handed to a node, C2 asks Discord (cached REST +snapshots, discord_rest.guild_snapshot — stubbed here) whether the bot may +View, Connect and Speak in the voice channel. These tests pin: + +* pick_token: a denied bot is skipped, confirmed-ok sorts before unknown, + unknown is allowed (fail open), all-denied is ``missing_permissions`` + naming each bot; a full pool bot (discord_pool_bot_guild_cap) is never + offered as an invite. +* start_stream: the check runs for every entry point (bot POST + /discord/streams, web picker, node discord_join, reconciler), inside a time + budget, never when the kill switch is off; a non-voice channel is + ``bad_channel``; the system's preferred bot is tried first. +* GET voice-check: org-scoped, bot names only for non-admins. +* PUT /tokens/{id}/prefer/{system}: owner decision Q2. +""" +import asyncio + +import pytest +from fastapi import HTTPException + +from app.config import settings +from app.internal import discord_leases as leases +from app.internal import discord_perms as perms_mod +from app.internal import discord_rest +from app.internal import discord_streams as streams +from app.models import CommandPayload +from app.routers import discord as discord_router +from app.routers import discord_checks as checks +from app.routers import discord_web as web +from app.routers import nodes as nodes_router +from app.routers import tokens as tokens_router +from tests.fake_firestore import FakeFirestore + +ORG, OTHER, SYS = "org-a", "org-b", "sys-1" +GUILD = "100" +ADMIN = {"uid": "u-admin", "role": "admin"} +OWNER = {"uid": "u-own", "org_id": ORG, "org_role": "owner"} +MEMBER = {"uid": "u-mem", "org_id": ORG, "org_role": "member"} +STRANGER = {"uid": "u-x", "org_id": OTHER, "org_role": "owner"} + +VOICE = perms_mod.VIEW_CHANNEL | perms_mod.CONNECT | perms_mod.SPEAK + + +def _guild(): + return {"id": GUILD, "owner_id": "owner-user", + "roles": [{"id": GUILD, "name": "@everyone", "permissions": str(VOICE)}, + {"id": "r-muted", "name": "Muted", "permissions": "0"}]} + + +def _channels(): + return [ + {"id": "1001", "name": "dispatch", "type": 2, "user_limit": 0, + "permission_overwrites": [{"id": "r-muted", "type": 0, "allow": "0", "deny": str(perms_mod.SPEAK)}]}, + {"id": "1002", "name": "fireground", "type": 2, "user_limit": 5, "permission_overwrites": []}, + {"id": "1003", "name": "chat", "type": 0, "permission_overwrites": []}, + ] + + +def _snap(tok_id, *, muted=False, status=discord_rest.OK): + if status != discord_rest.OK: + return discord_rest.GuildSnapshot(token_id=tok_id, guild_id=GUILD, status=status, error="x") + return discord_rest.GuildSnapshot( + token_id=tok_id, guild_id=GUILD, status=discord_rest.OK, bot_user_id=f"u-{tok_id}", + guild=_guild(), member={"user": {"id": f"u-{tok_id}"}, "roles": ["r-muted"] if muted else []}, + channels=_channels()) + + +@pytest.fixture +def fs(monkeypatch): + fake = FakeFirestore.install(monkeypatch) + published: list = [] + from app.internal.mqtt_handler import mqtt_handler + monkeypatch.setattr(mqtt_handler, "publish_discord_desired", lambda n, p: published.append((n, p)) or True) + streams._publish_locks.clear() + streams._guild_locks.clear() + checks._last_refresh.clear() + + async def _no_discord(token_ids): + return 0 + monkeypatch.setattr(discord_rest, "refresh_membership", _no_discord) + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "") + monkeypatch.setattr(settings, "discord_perm_check_enabled", True) + + # Per-token snapshot behaviour: "ok" | "muted" | "unknown" | "slow" | "not_member" + fake.snap_mode = {} + fake.snap_calls = [] + + async def _snapshot(token_doc, guild_id, *, refresh=False): + tid = token_doc["token_id"] + fake.snap_calls.append((tid, guild_id, refresh)) + assert "token" in token_doc # the real call needs the string; it never leaves C2 + mode = fake.snap_mode.get(tid, "ok") + if mode == "slow": + await asyncio.sleep(5) + if mode == "unknown": + return _snap(tid, status=discord_rest.UNKNOWN) + if mode == "not_member": + return _snap(tid, status=discord_rest.NOT_MEMBER) + return _snap(tid, muted=(mode == "muted")) + monkeypatch.setattr(discord_rest, "guild_snapshot", _snapshot) + + for nid, org in (("n1", ORG), ("n2", ORG), ("nx", OTHER)): + fake.put("nodes", nid, {"node_id": nid, "name": f"Node {nid}", "org_id": org, "status": "online", + "approval_status": "approved", "assigned_system_id": SYS}) + fake.put("systems", SYS, {"system_id": SYS, "name": "County P25", "org_id": ORG}) + fake.put("systems", "shared", {"system_id": "shared", "name": "State Net", "org_id": None}) + fake.put("systems", "sys-b", {"system_id": "sys-b", "name": "Theirs", "org_id": OTHER}) + fake.put("discord_guilds", GUILD, {"guild_id": GUILD, "org_id": ORG, "name": "Fire Co. 7", "bot_present": True, + "voice_channels": [{"channel_id": "1001", "name": "dispatch"}, + {"channel_id": "1002", "name": "fireground"}]}) + fake.put("discord_guilds", "200", {"guild_id": "200", "org_id": OTHER, "name": "Other", "bot_present": True, + "voice_channels": [{"channel_id": "2001", "name": "theirs"}]}) + fake.published = published + return fake + + +def _token(fake, tid, *, name=None, org=None, guilds=(GUILD,), **extra): + fake.put("bot_tokens", tid, {"token_id": tid, "name": name or f"DRB Radio {tid}", "token": f"secret-{tid}", + "org_id": org, "guild_ids": list(guilds) if guilds is not None else None, + "valid": True, "application_id": f"app-{tid}", "lease": None, **extra}) + + +def run(coro): + return asyncio.run(coro) + + +async def _start(channel="1001", **kw): + return await streams.start_stream(org_id=kw.pop("org_id", ORG), guild_id=GUILD, channel_id=channel, + system_id=kw.pop("system_id", SYS), **kw) + + +# --------------------------------------------------------------------------- +# pick_token — pure +# --------------------------------------------------------------------------- + +def _toks(*ids, **kw): + return {t: {"name": f"Radio {t}", "guild_ids": [GUILD], "application_id": f"a{t}", **kw} for t in ids} + + +def test_pick_skips_denied_and_prefers_confirmed_ok_over_unknown(): + toks = _toks("a", "b", "c") + perms = {"a": {"status": "denied", "missing": ["Speak"], "reason": "r", "fix": "f"}, + "b": {"status": "unknown"}, "c": {"status": "ok"}} + assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, perms=perms).token_id == "c" + # Unknown is allowed: fail open. + perms["c"] = {"status": "denied", "missing": ["Speak"]} + assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, perms=perms).token_id == "b" + + +def test_pick_all_denied_is_missing_permissions_naming_each_bot(): + toks = {**_toks("a", "b"), "out": {"name": "Radio out", "guild_ids": [], "application_id": "9"}} + perms = {t: {"status": "denied", "missing": ["Speak"], "reason": f"why {t}", "fix": "fix"} for t in ("a", "b")} + res = leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, perms=perms) + assert res.error == "missing_permissions" + assert [d["name"] for d in res.detail["denied"]] == ["Radio a", "Radio b"] + assert res.detail["denied"][0] == {"name": "Radio a", "missing": ["Speak"], "reason": "why a", "fix": "fix"} + assert res.invite and res.invite[0]["name"] == "Radio out" # the other way out rides along + + +def test_pick_without_perms_is_unchanged(): + toks = _toks("a", "b") + assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG).token_id == "a" + + +def test_preferred_bot_denied_is_refused_by_name_not_swapped(): + toks = _toks("a", "b") + perms = {"a": {"status": "denied", "missing": ["Connect", "Speak"], "reason": "r", "fix": "f"}} + res = leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, preferred_token_id="a", perms=perms) + assert res.error == "missing_permissions" and res.detail["token_name"] == "Radio a" + + +def test_soft_preferred_goes_first_but_falls_back(): + toks = _toks("a", "z") + assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, + soft_preferred_token_id="z").token_id == "z" + toks["z"]["lease"] = {"node_id": "n1", "streams": [f"{GUILD}:9:abcdef012345"]} # busy in this server + assert leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG, + soft_preferred_token_id="z").token_id == "a" + + +def test_full_pool_bot_is_never_offered_as_an_invite(monkeypatch): + monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 3) + toks = {"full": {"name": "Full", "guild_ids": ["1", "2", "3"], "application_id": "1"}, + "room": {"name": "Room", "guild_ids": ["1"], "application_id": "2"}, + "byo": {"name": "Ours", "org_id": ORG, "guild_ids": ["1", "2", "3", "4"], "application_id": "3"}} + res = leases.pick_token(toks, node_id="n1", guild_id=GUILD, org_id=ORG) + assert res.error == "no_bot_in_guild" and {i["name"] for i in res.invite} == {"Room", "Ours"} + res = leases.pick_token({"full": toks["full"]}, node_id="n1", guild_id=GUILD, org_id=ORG) + assert res.error == "pool_full" and not res.invite + # A full pool bot already IN the server is still used: the cap is about joining more. + toks["full"]["guild_ids"].append(GUILD) + assert leases.pick_token({"full": toks["full"]}, node_id="n1", guild_id=GUILD, org_id=ORG).ok + + +# --------------------------------------------------------------------------- +# evaluate_access — trace -> reason/fix text +# --------------------------------------------------------------------------- + +def test_evaluate_names_missing_permission_reason_and_fix(): + r = streams.evaluate_access(_snap("t", muted=True), "1001", bot="DRB Radio 2") + assert r["status"] == "denied" and r["missing"] == ["Speak"] + assert r["reason"] == "denied by #dispatch's override for @Muted" + assert "DRB Radio 2" in r["fix"] and "allow Speak" in r["fix"] + assert streams.evaluate_access(_snap("t"), "1001", bot="x")["status"] == "ok" + assert streams.evaluate_access(_snap("t"), "1003", bot="x")["status"] == "bad_channel" + assert streams.evaluate_access(_snap("t"), "9999", bot="x")["status"] == "unknown" # not in list: fail open + bad = _snap("t") + bad.guild = {"id": GUILD, "roles": []} # partial payload + assert streams.evaluate_access(bad, "1001", bot="x")["status"] == "unknown" + + +# --------------------------------------------------------------------------- +# start_stream +# --------------------------------------------------------------------------- + +async def test_denied_bot_is_skipped_for_one_that_can_speak(fs): + _token(fs, "t1"); _token(fs, "t2") + fs.snap_mode["t1"] = "muted" + doc = await _start(node_id="n1") + assert doc["token_id"] == "t2" + + +async def test_every_bot_denied_is_a_409_naming_bot_channel_permission_and_reason(fs): + _token(fs, "t1", name="DRB Radio 2") + fs.snap_mode["t1"] = "muted" + with pytest.raises(streams.StreamError) as e: + await _start(node_id="n1") + err = e.value + assert err.code == "missing_permissions" and err.status == 409 + assert err.message.startswith("DRB Radio 2 can't Speak in #dispatch: denied by #dispatch's override for @Muted.") + assert "Edit Channel → Permissions" in err.message + assert err.extra["denied"][0]["name"] == "DRB Radio 2" and "token" not in str(err.extra) + assert not fs.get("bot_tokens", "t1")["lease"] # nothing handed to a node + assert not fs.data.get("discord_streams") + + +async def test_unknown_fails_open(fs): + _token(fs, "t1") + fs.snap_mode["t1"] = "unknown" + assert (await _start(node_id="n1"))["token_id"] == "t1" + + +async def test_check_is_bounded_by_the_time_budget(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_perm_check_budget_seconds", 0.05) + _token(fs, "t1") + fs.snap_mode["t1"] = "slow" + loop = asyncio.get_running_loop() + t0 = loop.time() + doc = await _start(node_id="n1") + assert doc["token_id"] == "t1" and loop.time() - t0 < 2 + + +async def test_concurrency_is_bounded(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_perm_check_concurrency", 2) + for i in range(6): + _token(fs, f"t{i}") + live = {"now": 0, "max": 0} + + async def _snapshot(token_doc, guild_id, *, refresh=False): + live["now"] += 1 + live["max"] = max(live["max"], live["now"]) + await asyncio.sleep(0.01) + live["now"] -= 1 + return _snap(token_doc["token_id"]) + monkeypatch.setattr(discord_rest, "guild_snapshot", _snapshot) + await _start(node_id="n1") + assert live["max"] == 2 + + +async def test_kill_switch_restores_unchecked_picking(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_perm_check_enabled", False) + _token(fs, "t1") + fs.snap_mode["t1"] = "muted" + assert (await _start(node_id="n1"))["token_id"] == "t1" + assert fs.snap_calls == [] + + +async def test_text_channel_is_bad_channel_not_missing_permissions(fs): + _token(fs, "t1") + with pytest.raises(streams.StreamError) as e: + await _start(channel="1003", node_id="n1") + assert e.value.code == "bad_channel" and e.value.status == 400 and "#chat" in e.value.message + + +async def test_only_this_orgs_and_member_bots_are_asked(fs): + _token(fs, "pool"); _token(fs, "theirs", org=OTHER); _token(fs, "elsewhere", guilds=("999",)) + _token(fs, "dead", valid=False) + await _start(node_id="n1") + assert {c[0] for c in fs.snap_calls} == {"pool"} + + +async def test_system_preferred_bot_is_used_first(fs): + _token(fs, "a"); _token(fs, "z") + fs.put("systems", SYS, {**fs.get("systems", SYS), "preferred_token_id": "z"}) + assert (await _start(node_id="n1"))["token_id"] == "z" + + +async def test_system_preferred_bot_is_a_hint_not_a_demand(fs): + _token(fs, "a"); _token(fs, "z") + fs.put("systems", SYS, {**fs.get("systems", SYS), "preferred_token_id": "z"}) + fs.snap_mode["z"] = "muted" + assert (await _start(node_id="n1"))["token_id"] == "a" + + +async def test_one_bot_setup_invite_mark_preferred_stream(fs): + """Owner requirement 1: only the system's own BYO radio bot is in the server.""" + _token(fs, "byo", org=ORG) + _token(fs, "pool", guilds=("555",)) + fs.put("systems", SYS, {**fs.get("systems", SYS), "preferred_token_id": "byo"}) + assert (await _start())["token_id"] == "byo" + + +async def test_running_stream_moves_off_a_bot_that_lost_its_permission(fs): + _token(fs, "t1"); _token(fs, "t2") + first = await _start(node_id="n1") + assert first["token_id"] == "t1" + discord_rest.invalidate() + fs.snap_mode["t1"] = "muted" + again = await streams.start_stream(org_id=ORG, stream_id=first["stream_id"], system_id=SYS, node_id="n1") + assert again["token_id"] == "t2" and again["reused"] is False + assert first["stream_id"] not in (fs.get("bot_tokens", "t1")["lease"] or {}).get("streams", []) + + +async def test_handle_actual_drops_cached_access_on_edge_refusals(fs, monkeypatch): + _token(fs, "t1") + doc = await _start(node_id="n1") + dropped = [] + monkeypatch.setattr(discord_rest, "invalidate", lambda token_id=None, guild_id=None: dropped.append( + (token_id, guild_id)) or 1) + for code in ("missing_permissions", "channel_full", "bad_channel", "no_audio"): + await streams.handle_actual("n1", {"streams": {doc["stream_id"]: {"state": "error", "code": code}}}) + assert dropped == [("t1", GUILD)] * 3 + + +# --------------------------------------------------------------------------- +# Every entry point gets the same refusal +# --------------------------------------------------------------------------- + +def _all_denied(fs): + _token(fs, "t1", name="DRB Radio 2") + fs.snap_mode["t1"] = "muted" + + +def test_bot_join_route_returns_409_with_text(fs): + _all_denied(fs) + r = run(discord_router.start_stream(discord_router.StreamStart(guild_id=GUILD, channel_id="1001", + system_id=SYS), {"service": True})) + assert r.status_code == 409 + import json + body = json.loads(r.body) + assert body["code"] == "missing_permissions" and "DRB Radio 2 can't Speak in #dispatch" in body["detail"] + assert body["denied"][0]["missing"] == ["Speak"] and "secret" not in r.body.decode() + + +def test_web_picker_returns_409_with_text(fs): + _all_denied(fs) + r = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id=GUILD, channel_id="1001"), OWNER)) + assert r.status_code == 409 and b"missing_permissions" in r.body and b"#dispatch" in r.body + + +def test_node_discord_join_command_is_refused(fs): + _all_denied(fs) + with pytest.raises(HTTPException) as e: + run(nodes_router.send_command("n1", CommandPayload(action="discord_join", guild_id=GUILD, + channel_id="1001"), ADMIN)) + assert e.value.status_code == 409 and "can't Speak in #dispatch" in e.value.detail + + +async def test_reconciler_marks_the_stream_with_the_refusal(fs): + _token(fs, "t1", name="DRB Radio 2") + doc = await _start(node_id="n1", mode="persistent") + # Node goes away; the bot meanwhile lost Speak. + await streams.release_offline_node("n1") + fs.put("nodes", "n1", {**fs.get("nodes", "n1"), "status": "offline", "last_seen": "2000-01-01T00:00:00+00:00"}) + discord_rest.invalidate() + fs.snap_mode["t1"] = "muted" + await streams.reconcile_once() + s = fs.get("discord_streams", doc["stream_id"]) + assert s["actual"]["state"] == "error" and "DRB Radio 2 can't Speak in #dispatch" in s["actual"]["error"] + assert not s.get("token_id") + + +# --------------------------------------------------------------------------- +# GET voice-check +# --------------------------------------------------------------------------- + +def test_voice_check_reports_each_channel_for_an_org_member(fs): + _token(fs, "t1", name="DRB Radio 1"); _token(fs, "t2", name="DRB Radio 2") + fs.snap_mode["t2"] = "muted" + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=MEMBER)) + assert out["enabled"] is True and out["guild_name"] == "Fire Co. 7" + by = {c["channel_id"]: c for c in out["channels"]} + assert by["1001"]["state"] == "ready" and by["1001"]["any_ok"] is True + bots = {b["name"]: b for b in by["1001"]["bots"]} + assert bots["DRB Radio 2"]["ok"] is False and bots["DRB Radio 2"]["missing"] == ["Speak"] + assert bots["DRB Radio 2"]["reason"] == "denied by #dispatch's override for @Muted" + assert all("token_id" not in b for c in out["channels"] for b in c["bots"]) # names only + assert "secret" not in str(out) + assert by["1002"]["user_limit"] == 5 + + +def test_voice_check_needs_fix_when_every_free_bot_is_denied(fs): + _token(fs, "t1") + fs.snap_mode["t1"] = "muted" + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id="1001", decoded=OWNER)) + assert [c["state"] for c in out["channels"]] == ["needs_fix"] + fs.snap_mode["t1"] = "unknown" + discord_rest.invalidate() + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id="1001", decoded=OWNER)) + assert out["channels"][0]["state"] == "unknown" # can't tell -> start still allowed + + +def test_voice_check_is_org_scoped(fs): + _token(fs, "t1") + for node, guild, who in (("n1", "200", OWNER), ("nx", GUILD, OWNER), ("n1", GUILD, STRANGER)): + with pytest.raises(HTTPException) as e: + run(checks.node_voice_check(node, guild, refresh=0, channel_id=None, decoded=who)) + assert e.value.status_code == 404 + + +def test_voice_check_admin_sees_token_ids_and_service_route(fs): + _token(fs, "t1") + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=ADMIN)) + assert out["channels"][0]["bots"][0]["token_id"] == "t1" + out = run(checks.guild_voice_check(GUILD, channel_id="1001", refresh=0, decoded={"service": True})) + assert out["org_id"] == ORG and out["channels"][0]["bots"][0]["token_id"] == "t1" + + +def test_voice_check_refresh_is_rate_limited_per_caller_and_server(fs): + _token(fs, "t1") + first = run(checks.node_voice_check("n1", GUILD, refresh=1, channel_id=None, decoded=OWNER)) + second = run(checks.node_voice_check("n1", GUILD, refresh=1, channel_id=None, decoded=OWNER)) + assert first["refreshed"] is True and second["refreshed"] is False + assert [c[2] for c in fs.snap_calls] == [True, False] + + +def test_voice_check_lists_invites_within_the_pool_cap(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 2) + _token(fs, "in") + _token(fs, "room", guilds=("1",)) + _token(fs, "full", guilds=("1", "2")) + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=OWNER)) + assert [i["name"] for i in out["invite"]] == ["DRB Radio room"] + + +def test_voice_check_busy_and_other_node(fs): + _token(fs, "t1", lease={"node_id": "n2", "streams": [], "lease_id": "L"}) + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id="1001", decoded=OWNER)) + b = out["channels"][0]["bots"][0] + assert b["ok"] is True and b["available"] is False and b["unavailable_reason"] == "other_node" + assert out["channels"][0]["state"] == "busy" + + +def test_voice_check_kill_switch(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_perm_check_enabled", False) + _token(fs, "t1") + out = run(checks.node_voice_check("n1", GUILD, refresh=0, channel_id=None, decoded=OWNER)) + assert out["enabled"] is False and {c["state"] for c in out["channels"]} == {"unchecked"} + assert fs.snap_calls == [] + + +# --------------------------------------------------------------------------- +# PUT /tokens/{id}/prefer/{system_id} — owner decision Q2 +# --------------------------------------------------------------------------- + +def _prefer(tid, sid, who): + return run(tokens_router.set_preferred_system(tid, sid, who)) + + +def test_owner_may_prefer_own_byo_bot_for_own_system(fs): + _token(fs, "byo", org=ORG) + assert _prefer("byo", SYS, OWNER)["preferred_for_system_id"] == SYS + assert fs.get("systems", SYS)["preferred_token_id"] == "byo" + assert fs.get("bot_tokens", "byo")["preferred_for_system_id"] == SYS + assert any(a.get("action") == "discord.token.prefer" for a in (fs.data.get("audit_log") or {}).values()) + assert _prefer("byo", "_none", OWNER)["preferred_for_system_id"] is None + assert fs.get("systems", SYS)["preferred_token_id"] is None + + +@pytest.mark.parametrize("tid,sid,who,status", [ + ("pool", SYS, OWNER, 403), # pool bots: platform admin only + ("byo", "shared", OWNER, 403), # shared system: platform admin only + ("theirs", SYS, OWNER, 404), # another org's bot: not found + ("byo", "sys-b", OWNER, 404), # another org's system: not found + ("byo", SYS, MEMBER, 403), # members can't + ("byo", SYS, STRANGER, 404), + ("nope", SYS, OWNER, 404), +]) +def test_prefer_refusals(fs, tid, sid, who, status): + _token(fs, "pool"); _token(fs, "byo", org=ORG); _token(fs, "theirs", org=OTHER) + with pytest.raises(HTTPException) as e: + _prefer(tid, sid, who) + assert e.value.status_code == status + assert not (fs.get("systems", sid) or {}).get("preferred_token_id") + + +def test_admin_may_prefer_any_bot_for_any_system(fs): + _token(fs, "pool") + assert _prefer("pool", "shared", ADMIN)["preferred_for_system_id"] == "shared" + + +def test_prefer_keeps_both_pointers_single_valued(fs): + _token(fs, "byo", org=ORG); _token(fs, "byo2", org=ORG) + fs.put("systems", "sys-2", {"system_id": "sys-2", "name": "City", "org_id": ORG}) + _prefer("byo", SYS, OWNER) + _prefer("byo", "sys-2", OWNER) # moved: the old system no longer points at it + assert fs.get("systems", SYS)["preferred_token_id"] is None + _prefer("byo2", "sys-2", OWNER) # replaced: the old bot loses its back-pointer + assert fs.get("bot_tokens", "byo")["preferred_for_system_id"] is None + assert fs.get("systems", "sys-2")["preferred_token_id"] == "byo2" + + +# --------------------------------------------------------------------------- +# End to end through the real REST client (Discord answered by MockTransport) +# --------------------------------------------------------------------------- + +async def test_real_snapshot_path_denies_and_never_logs_the_token(fs, monkeypatch, caplog): + import httpx + from app.internal import discord_rest as rest + monkeypatch.undo() # drop the fixture's guild_snapshot stub ... + fs = FakeFirestore.install(monkeypatch) # ... and re-install only what this test needs + from app.internal.mqtt_handler import mqtt_handler + monkeypatch.setattr(mqtt_handler, "publish_discord_desired", lambda n, p: True) + monkeypatch.setattr(settings, "discord_perm_check_enabled", True) + streams._guild_locks.clear() + fs.put("nodes", "n1", {"node_id": "n1", "name": "Node n1", "org_id": ORG, "status": "online", + "approval_status": "approved", "assigned_system_id": SYS}) + fs.put("systems", SYS, {"system_id": SYS, "name": "County P25", "org_id": ORG}) + _token(fs, "t1", name="DRB Radio 2", bot_user_id="u-t1") + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append((request.url.path, request.headers.get("authorization"))) + path = request.url.path.removeprefix("/api/v10") + if path == f"/guilds/{GUILD}": + return httpx.Response(200, json=_guild()) + if path == f"/guilds/{GUILD}/members/u-t1": + return httpx.Response(200, json={"user": {"id": "u-t1"}, "roles": ["r-muted"]}) + if path == f"/guilds/{GUILD}/channels": + return httpx.Response(200, json=_channels()) + return httpx.Response(404, json={"code": 0}) + rest.reset() + monkeypatch.setattr(rest, "_transport", httpx.MockTransport(handler)) + with pytest.raises(streams.StreamError) as e: + await _start(node_id="n1") + assert e.value.code == "missing_permissions" and "can't Speak in #dispatch" in e.value.message + assert {p for p, _ in seen} >= {f"/api/v10/guilds/{GUILD}", f"/api/v10/guilds/{GUILD}/channels"} + assert all(a == "Bot secret-t1" for _, a in seen) + assert "secret-t1" not in caplog.text and "secret-t1" not in e.value.message + # Cached: a second start asks Discord nothing new. + n = len(seen) + with pytest.raises(streams.StreamError): + await _start(node_id="n1") + assert len(seen) == n